iop/apps/node/internal/node/node_test_support_test.go

package node_test

import (
	"context"
	"errors"
	"fmt"
	"io"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"go.uber.org/zap"
	"iop/apps/node/internal/node"
	"iop/apps/node/internal/store"
	runtime "iop/packages/go/execution"
)

// fixedRouter dispatches to a pre-built adapter map. It satisfies runtime.Router.
type fixedRouter struct {
	adapterName  string
	adapter      runtime.Provider
	adapters     map[string]runtime.Provider
	lookupErrors map[string]error // optional per-adapter errors for LookupAdapter
}

func (r *fixedRouter) Resolve(_ context.Context, req runtime.RunRequest) (runtime.ExecutionSpec, error) {
	return runtime.ExecutionSpec{
		RunID:      req.RunID,
		Adapter:    r.adapterName,
		Target:     req.Target,
		SessionID:  req.SessionID,
		Background: req.Background,
		Policy:     req.Policy,
		Input:      req.Input,
		TimeoutSec: req.TimeoutSec,
		Metadata:   req.Metadata,
	}, nil
}

func (r *fixedRouter) ResolveAdapter(ctx context.Context, req runtime.RunRequest) (runtime.ExecutionSpec, runtime.Provider, error) {
	spec, err := r.Resolve(ctx, req)
	if err != nil {
		return runtime.ExecutionSpec{}, nil, err
	}
	a, ok := r.adapters[spec.Adapter]
	if !ok {
		return runtime.ExecutionSpec{}, nil, fmt.Errorf("router: adapter %q not found", spec.Adapter)
	}
	return spec, a, nil
}

func (r *fixedRouter) LookupAdapter(adapterName string) (runtime.Provider, error) {
	if r.lookupErrors != nil {
		if err, ok := r.lookupErrors[adapterName]; ok {
			return nil, err
		}
	}
	a, ok := r.adapters[adapterName]
	if !ok {
		return nil, fmt.Errorf("adapter %q not found", adapterName)
	}
	return a, nil
}

func (r *fixedRouter) GetAdapter(adapterName string) (runtime.Provider, bool) {
	a, ok := r.adapters[adapterName]
	return a, ok
}

// errorRouter always returns the configured error.
type errorRouter struct{ err error }

func (r *errorRouter) Resolve(_ context.Context, _ runtime.RunRequest) (runtime.ExecutionSpec, error) {
	return runtime.ExecutionSpec{}, r.err
}

func (r *errorRouter) ResolveAdapter(_ context.Context, _ runtime.RunRequest) (runtime.ExecutionSpec, runtime.Provider, error) {
	return runtime.ExecutionSpec{}, nil, r.err
}

func (r *errorRouter) LookupAdapter(_ string) (runtime.Provider, error) {
	return nil, r.err
}

func (r *errorRouter) GetAdapter(_ string) (runtime.Provider, bool) {
	return nil, false
}

// makeNode builds a node with a default store (in-memory) and no concurrency gate.
func makeNode(t *testing.T, rtr runtime.Router) (*node.Node, *store.Store) {
	t.Helper()
	return makeNodeWithConcurrency(t, rtr, 0)
}

// makeNodeWithConcurrency builds a node with a default store (in-memory) and a
// global concurrency limit (0 = disabled).
func makeNodeWithConcurrency(t *testing.T, rtr runtime.Router, concurrency int) (*node.Node, *store.Store) {
	t.Helper()
	st, err := store.New(":memory:", zap.NewNop())
	if err != nil {
		t.Fatalf("store: %v", err)
	}
	t.Cleanup(func() { _ = st.Close() })
	return node.New("test-node", rtr, st, concurrency, io.Discard, zap.NewNop(), nil), st
}

// --- shared concurrency helpers (slow adapter + synchronization) ---

// slowAdapterUnlimited reports MaxConcurrency=0 (unlimited) and blocks until
// released, exposing a start channel so tests can synchronize.
type slowAdapterUnlimited struct {
	started chan struct{}
	release chan struct{}
}

func newSlowAdapterUnlimited() *slowAdapterUnlimited {
	return &slowAdapterUnlimited{started: make(chan struct{}, 1), release: make(chan struct{})}
}

func (a *slowAdapterUnlimited) Name() string { return "slow" }
func (a *slowAdapterUnlimited) Capabilities(_ context.Context) (runtime.Capabilities, error) {
	return runtime.Capabilities{AdapterName: "slow", MaxConcurrency: 0}, nil
}
func (a *slowAdapterUnlimited) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error {
	select {
	case a.started <- struct{}{}:
	default:
	}
	select {
	case <-a.release:
	case <-ctx.Done():
		return runtime.ErrRunCancelled
	}
	return nil
}

// queuedSlowAdapter is a configurable adapter used by concurrency safety tests.
// It advertises MaxConcurrency / MaxQueue / QueueTimeout via Capabilities, and
// gives per-run control: each Execute blocks until the test releases (or fails)
// that specific run id, so tests can observe execution ordering deterministically.
type queuedSlowAdapter struct {
	name           string
	maxConcurrency int
	maxQueue       int
	queueTimeout   time.Duration

	mu       sync.Mutex
	gates    map[string]chan error // run_id → release signal carrying exec result
	startSeq chan string           // run_id pushed when Execute begins
}

func newQueuedSlowAdapter(name string, maxConcurrency, maxQueue int, queueTimeout time.Duration) *queuedSlowAdapter {
	return &queuedSlowAdapter{
		name:           name,
		maxConcurrency: maxConcurrency,
		maxQueue:       maxQueue,
		queueTimeout:   queueTimeout,
		gates:          make(map[string]chan error),
		startSeq:       make(chan string, 64),
	}
}

func (a *queuedSlowAdapter) Name() string { return a.name }

func (a *queuedSlowAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) {
	return runtime.Capabilities{
		AdapterName:    a.name,
		MaxConcurrency: a.maxConcurrency,
		MaxQueue:       a.maxQueue,
		QueueTimeoutMS: int(a.queueTimeout / time.Millisecond),
	}, nil
}

func (a *queuedSlowAdapter) gateFor(runID string) chan error {
	a.mu.Lock()
	defer a.mu.Unlock()
	ch, ok := a.gates[runID]
	if !ok {
		ch = make(chan error, 1)
		a.gates[runID] = ch
	}
	return ch
}

func (a *queuedSlowAdapter) Execute(ctx context.Context, spec runtime.ExecutionSpec, _ runtime.EventSink) error {
	gate := a.gateFor(spec.RunID)
	a.startSeq <- spec.RunID
	select {
	case res := <-gate:
		return res
	case <-ctx.Done():
		return runtime.ErrRunCancelled
	}
}

// releaseRun lets the given run complete successfully.
func (a *queuedSlowAdapter) releaseRun(runID string) { a.gateFor(runID) <- nil }

// failRun lets the given run terminate with a non-cancel error.
func (a *queuedSlowAdapter) failRun(runID string) { a.gateFor(runID) <- errBoom }

// preRelease pre-arms a run so it completes the instant it starts executing.
func (a *queuedSlowAdapter) preRelease(runID string) { a.gateFor(runID) <- nil }

var errBoom = errors.New("boom")

// waitStarted blocks until the adapter reports the given run id started, or
// fails the test after a timeout.
func waitStarted(t *testing.T, a *queuedSlowAdapter, runID string) {
	t.Helper()
	deadline := time.After(2 * time.Second)
	for {
		select {
		case got := <-a.startSeq:
			if got == runID {
				return
			}
			a.startSeq <- got
			time.Sleep(2 * time.Millisecond)
		case <-deadline:
			t.Fatalf("run %s never started", runID)
		}
	}
}

// requireStatus asserts the run is stored with exactly the given status now.
func requireStatus(t *testing.T, st *store.Store, runID, status string) *store.RunRecord {
	t.Helper()
	return requireStatusEventually(t, st, runID, status)
}

// requireStatusEventually polls until the run reaches the expected status.
func requireStatusEventually(t *testing.T, st *store.Store, runID, status string) *store.RunRecord {
	t.Helper()
	deadline := time.After(2 * time.Second)
	for {
		run, err := st.GetRun(context.Background(), runID)
		if err != nil {
			t.Fatalf("GetRun(%s): %v", runID, err)
		}
		if run != nil && run.Status == status {
			return run
		}
		select {
		case <-deadline:
			got := "<nil>"
			if run != nil {
				got = run.Status
			}
			t.Fatalf("run %s: expected status %q, got %q", runID, status, got)
		case <-time.After(5 * time.Millisecond):
		}
	}
}

// --- lifecycle test doubles (registry-refresh tests) ---

type lifecycleTestAdapter struct {
	name      string
	stopCalls int32
	started   chan struct{}
	blockChan chan struct{}
}

func (a *lifecycleTestAdapter) Name() string { return a.name }

func (a *lifecycleTestAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) {
	return runtime.Capabilities{
		AdapterName:    a.name,
		MaxConcurrency: 1,
	}, nil
}

func (a *lifecycleTestAdapter) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error {
	close(a.started)
	select {
	case <-a.blockChan:
	case <-ctx.Done():
	}
	return nil
}

func (a *lifecycleTestAdapter) Start(_ context.Context) error { return nil }

func (a *lifecycleTestAdapter) Stop(_ context.Context) error {
	atomic.AddInt32(&a.stopCalls, 1)
	return nil
}

type blockingCapsAdapter struct {
	name        string
	capsStarted chan struct{}
	capsBlock   chan struct{}
}

func (a *blockingCapsAdapter) Name() string { return a.name }

func (a *blockingCapsAdapter) Capabilities(ctx context.Context) (runtime.Capabilities, error) {
	select {
	case a.capsStarted <- struct{}{}:
	default:
	}
	<-a.capsBlock
	return runtime.Capabilities{
		AdapterName:    a.name,
		MaxConcurrency: 1,
	}, nil
}

func (a *blockingCapsAdapter) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error {
	return nil
}
func (a *blockingCapsAdapter) Start(_ context.Context) error { return nil }
func (a *blockingCapsAdapter) Stop(_ context.Context) error  { return nil }