iop/apps/node/internal/node/concurrency_gate_test.go

package node_test

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"testing"

	"iop/apps/node/internal/node"
	"iop/apps/node/internal/runtime"
	"iop/apps/node/internal/transport"
	iop "iop/proto/gen/iop"
)

// --- concurrency gate tests ---

// TestOnRunRequest_OverDispatchSafetyRejectsWithoutQueue verifies that a run
// is rejected immediately with reason concurrency_unavailable when the concurrency limit is reached (no queue).
func TestOnRunRequest_OverDispatchSafetyRejectsWithoutQueue(t *testing.T) {
	// capacity=1: one running is the ceiling.
	sa := newQueuedSlowAdapter("slow", 1, 0, 0)
	router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
	n, st := makeNodeWithConcurrency(t, router, 0) // global unlimited; adapter cap governs

	hold := make(chan error, 1)
	go func() {
		hold <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
			RunId: "of-hold", Adapter: "slow", Target: "v1", Background: true,
		})
	}()
	waitStarted(t, sa, "of-hold")

	// Second run overflows the capacity and must be rejected immediately.
	err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId: "of-reject", Adapter: "slow", Target: "v1",
	})
	if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
		t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err)
	}
	rec := requireStatus(t, st, "of-reject", "rejected")
	if !strings.Contains(rec.Error, "concurrency unavailable") {
		t.Fatalf("expected concurrency unavailable reason in error, got %q", rec.Error)
	}

	sa.releaseRun("of-hold")
	if err := <-hold; err != nil {
		t.Fatalf("of-hold: %v", err)
	}
}

// TestOnRunRequest_PermitReleasedAfterCompletion verifies that after the first
// run completes, a subsequent run can acquire the slot immediately.
func TestOnRunRequest_PermitReleasedAfterCompletion(t *testing.T) {
	sa := newQueuedSlowAdapter("slow", 1, 4, 0)
	router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
	n, _ := makeNodeWithConcurrency(t, router, 1)

	first := make(chan error, 1)
	go func() {
		first <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
			RunId: "rel-1", Adapter: "slow", Target: "v1",
		})
	}()
	waitStarted(t, sa, "rel-1")
	sa.releaseRun("rel-1")
	if err := <-first; err != nil {
		t.Fatalf("rel-1: %v", err)
	}

	// Second run must run immediately now that the slot is free.
	sa.preRelease("rel-2")
	if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId: "rel-2", Adapter: "slow", Target: "v1",
	}); err != nil {
		t.Fatalf("rel-2 should succeed after slot released, got %v", err)
	}
}

// TestConcurrencyLimit_Unlimited verifies that global=0 + adapter cap=0 imposes no cap.
func TestConcurrencyLimit_Unlimited(t *testing.T) {
	const count = 5
	shared := newSlowAdapterUnlimited() // MaxConcurrency=0 (unlimited)
	router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": shared}}
	nd, _ := makeNodeWithConcurrency(t, router, 0) // global unlimited

	errs := make(chan error, count)
	for i := 0; i < count; i++ {
		runID := fmt.Sprintf("run-unlimited-%d", i)
		go func(id string) {
			errs <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
				RunId: id, Adapter: "slow", Target: "v1",
			})
		}(runID)
	}

	// Release all.
	close(shared.release)

	// All must complete without ErrConcurrencyLimitExceeded.
	for i := 0; i < count; i++ {
		if err := <-errs; errors.Is(err, node.ErrConcurrencyLimitExceeded) {
			t.Fatalf("unexpected concurrency rejection with limit=0: %v", err)
		}
	}
}

// TestConcurrencyLimit_RejectStoreAndEvent verifies that a rejected run
// is stored with terminal status "rejected" and an error message.
func TestConcurrencyLimit_RejectStoreAndEvent(t *testing.T) {
	// capacity=1: any second concurrent run overflows immediately.
	sa := newQueuedSlowAdapter("slow", 1, 0, 0)
	router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}}
	nd, st := makeNodeWithConcurrency(t, router, 0)

	errc := make(chan error, 1)
	go func() {
		errc <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
			RunId: "run-hold", Adapter: "slow", Target: "v1", Background: true,
		})
	}()
	waitStarted(t, sa, "run-hold")

	// Second run must be rejected.
	err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{
		RunId: "run-rejected", Adapter: "slow", Target: "v1",
	})
	if !errors.Is(err, node.ErrConcurrencyLimitExceeded) {
		t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err)
	}

	// Rejected run must be stored with status "rejected".
	run, err := st.GetRun(context.Background(), "run-rejected")
	if err != nil {
		t.Fatalf("GetRun: %v", err)
	}
	if run == nil {
		t.Fatal("expected rejected run to be stored")
	}
	if run.Status != "rejected" {
		t.Fatalf("expected status %q, got %q", "rejected", run.Status)
	}
	if run.Error == "" {
		t.Fatal("expected non-empty error message for rejected run")
	}

	sa.releaseRun("run-hold")
	if err := <-errc; err != nil {
		t.Fatalf("run-hold: %v", err)
	}
}