compose-candidate is answered off its subscription, so its bound counts from the ask

An endpoint's subscription hands its asks over one at a time: two checks asking
one node at once queued, and the second answered at about 100 s, past the
checker's bound and the bus's one-minute reply window. compose-candidate is now
answered in a goroutine of its own, at most four at once, one more refused in
words. The new cases are rows of the existing tests (test budget).

Issue 1126, review round 1.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL

internal/link/engineseat.go

@@ -116,11 +116,10 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
 	if err != nil {
 		return
 	}
+	apart := make(chan struct{}, apartAtOnce)
 	for _, v := range catalogue.EngineVerbs() {
 		verb := v.Name
-		answer := micro.HandlerFunc(func(r micro.Request) {
+		answer := EngineAnswer(handle, verb, l.node, apart)
-			_ = r.Respond(AnswerEngineCall(handle, verb, l.node, r.Subject(), r.Data()))
-		})
 		meta := EngineEndpointMetadata(v, l.node)
 		_ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)),
 			micro.WithEndpointMetadata(meta))
@@ -130,6 +129,40 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
 	l.engine = svc
 }
 
+// answeredApart are the verbs answered off the subscription that delivered them: each endpoint's subscription hands its
+// asks over one at a time, so a second ask of a verb that takes long waited for the first, and its bound counted from
+// when the first ended (issue 1126, review of 2026-11-03: two checks asking compose-candidate of one node at once, the
+// second answered at about 100 s, past the bus's one-minute reply window). Each is answered in a goroutine of its own,
+// its bound counted from its arrival.
+var answeredApart = map[string]bool{ActComposeCandidate: true}
+
+// apartAtOnce bounds the asks answered apart at once on this node; one more is refused in words at once.
+const apartAtOnce = 4
+
+// EngineAnswer is the handler of one verb: answered on the subscription's goroutine, or, for a verb answered apart, in
+// its own while fewer than apart's capacity run, and refused at once past it.
+func EngineAnswer(handle EngineFunc, verb, node string, apart chan struct{}) micro.HandlerFunc {
+	return func(r micro.Request) {
+		if !answeredApart[verb] {
+			_ = r.Respond(AnswerEngineCall(handle, verb, node, r.Subject(), r.Data()))
+			return
+		}
+		select {
+		case apart <- struct{}{}:
+		default:
+			raw, _ := json.Marshal(map[string]any{"node": node, "error": fmt.Sprintf(
+				"this node answers %d asks of %s at once now: ask again when one is answered", cap(apart), verb)})
+			_ = r.Respond(raw)
+			return
+		}
+		subject, data := r.Subject(), r.Data()
+		go func() {
+			defer func() { <-apart }()
+			_ = r.Respond(AnswerEngineCall(handle, verb, node, subject, data))
+		}()
+	}
+}
+
 // EngineModule is the module whose code answers the seat, as discovery names it: the node-engine's manifest's.
 const EngineModule = "mesh-host"
 

internal/link/handover_test.go

@@ -7,11 +7,13 @@ import (
 	"errors"
 	"sort"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/micro"
 )
 
 // The hand-over's ask and answer hold these field names; the controller's side holds the same list
@@ -265,8 +267,63 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
 			t.Errorf("the judgements are recorded as %+v", got)
 		}
 	})
+	// Issue 1126 (review of 2026-11-03): two asks of compose-candidate at once, handed over one at a time as an
+	// endpoint's subscription hands them, are both answered while the other runs: neither waits for the other, so each
+	// bound counts from its own arrival. A verb answered on the subscription answers in order, as before.
+	t.Run("two asks of compose-candidate at once are answered side by side", func(t *testing.T) {
+		var mu sync.Mutex
+		arrived, both := 0, make(chan struct{})
+		handle := func(c EngineCall) (any, error) {
+			mu.Lock()
+			if arrived++; arrived == 2 {
+				close(both)
+			}
+			mu.Unlock()
+			select {
+			case <-both:
+				return "together", nil
+			case <-time.After(2 * time.Second): // only when the asks are answered one after the other
+				return "alone", nil
+			}
+		}
+		answered := make(chan string, 2)
+		answer := EngineAnswer(handle, ActComposeCandidate, "n1", make(chan struct{}, 4))
+		for range 2 {
+			answer(fakeRequest{subject: EngineToolSubject(ActComposeCandidate, "n1"), answered: answered})
+		}
+		for range 2 {
+			if got := <-answered; !strings.Contains(got, `"result":"together"`) {
+				t.Errorf("an ask answered %s; want both answered side by side", got)
+			}
+		}
+		full := make(chan struct{}, 1)
+		full <- struct{}{}
+		EngineAnswer(handle, ActComposeCandidate, "n1", full)(fakeRequest{subject: "s", answered: answered})
+		if got := <-answered; !strings.Contains(got, "asks of compose-candidate at once") {
+			t.Errorf("an ask past the bound answered %s", got)
+		}
+	})
 }
 
+// fakeRequest is a seat call as the subscription hands it over, its answer sent to answered.
+type fakeRequest struct {
+	subject  string
+	answered chan string
+}
+
+func (r fakeRequest) Respond(b []byte, _ ...micro.RespondOpt) error {
+	r.answered <- string(b)
+	return nil
+}
+func (r fakeRequest) RespondJSON(any, ...micro.RespondOpt) error { return nil }
+func (r fakeRequest) Error(string, string, []byte, ...micro.RespondOpt) error {
+	return nil
+}
+func (r fakeRequest) Data() []byte           { return []byte(`{}`) }
+func (r fakeRequest) Headers() micro.Headers { return nil }
+func (r fakeRequest) Subject() string        { return r.subject }
+func (r fakeRequest) Reply() string          { return "_INBOX.x" }
+
 // **Against a real server**: a signed request on the node's own subject is answered; one on another node's is
 // not heard; a message with no reply is answered by nobody.
 func TestNatsAHandOverIsAskedOfTheNode(t *testing.T) {

internal/selfcompose/candidate_test.go

@@ -132,122 +132,121 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
 	if len(state.written()) != 0 || applied.Load() != 0 || handed.Load() != 0 {
 		t.Errorf("composing candidates wrote %v, marked %d have(s), handed %d answer(s); want nothing", state.written(), applied.Load(), handed.Load())
 	}
-}
+	// Issue 1126, 2026-11-03: off the control-node an ask of compose-candidate was not answered within a minute, and said
+	// nothing of why. Each of its two compositions read every manifest in the mesh one by one, on copies of the shadow
+	// that kept nothing, so every ask read them all twice again. Each manifest is now read once, before either
+	// composition, into the shadow, and a later ask reads none; an ask whose read does not answer is answered unready at
+	// its bound, naming what the node waits on; and every answer is said in one line with what it waited on.
+	t.Run("each manifest is read once, and the answer comes within its bound", func(t *testing.T) {
+		fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
+		fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
+		lemurs2 := strings.Replace(lemurs, `"version":"1"`, `"version":"2"`, 1)
+		other := strings.NewReplacer(`"module":"lemurs"`, `"module":"other"`, "lemurs.service", "other.service").Replace(lemurs)
+		digest, digest2, digestOther := delivery.ManifestDigest([]byte(lemurs)), delivery.ManifestDigest([]byte(lemurs2)),
+			delivery.ManifestDigest([]byte(other))
+		otherKey, _ := delivery.ManifestsKey(digestOther)
 
-// Issue 1126, 2026-11-03: off the control-node an ask of compose-candidate was not answered within a minute, and said
+		// shadow is a ready shadow of n1 over b: lemurs placed on it, other placed nowhere (its manifest read for the mesh's
-// nothing of why. Each of its two compositions read every manifest in the mesh one by one, on copies of the shadow
+		// catalogue alone), and a candidate laying lemurs at a new build; its clock moves a second at each reading.
-// that kept nothing, so every ask read them all twice again. Each manifest is now read once, before either
+		shadow := func(b *fakeBus) (*Shadow, string) {
-// composition, into the shadow, and a later ask reads none; an ask whose read does not answer is answered unready at
+			for d, body := range map[string]string{digest: lemurs, digest2: lemurs2, digestOther: other} {
-// its bound, naming what the node waits on; and every answer is said in one line with what it waited on.
+				man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "x", JSON: body})
-func TestComposeCandidateReadsOnceAndAnswersWithinItsBound(t *testing.T) {
+				key, _ := delivery.ManifestsKey(d)
-	fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
+				b.kv[delivery.ManifestsBucket+"/"+key] = man
-	fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
-	lemurs2 := strings.Replace(lemurs, `"version":"1"`, `"version":"2"`, 1)
-	other := strings.NewReplacer(`"module":"lemurs"`, `"module":"other"`, "lemurs.service", "other.service").Replace(lemurs)
-	digest, digest2, digestOther := delivery.ManifestDigest([]byte(lemurs)), delivery.ManifestDigest([]byte(lemurs2)),
-		delivery.ManifestDigest([]byte(other))
-	otherKey, _ := delivery.ManifestsKey(digestOther)
-
-	// shadow is a ready shadow of n1 over b: lemurs placed on it, other placed nowhere (its manifest read for the mesh's
-	// catalogue alone), and a candidate laying lemurs at a new build; its clock moves a second at each reading.
-	shadow := func(b *fakeBus) (*Shadow, string) {
-		for d, body := range map[string]string{digest: lemurs, digest2: lemurs2, digestOther: other} {
-			man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "x", JSON: body})
-			key, _ := delivery.ManifestsKey(d)
-			b.kv[delivery.ManifestsBucket+"/"+key] = man
-		}
-		head := delivery.Head{Repository: "novox/x", Number: 7, Head: "c0ffee"}
-		heads, _ := delivery.HeadsHash([]delivery.Head{head})
-		pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"lemurs"},
-			Fps: map[string]delivery.CandidateModule{"lemurs": {FP2: fpNew, Manifest: digest2}}}}
-		raw, _ := delivery.EncodeCandidate(delivery.Candidate{Group: "fix/1126", HeadsHash: heads, Pulls: pulls,
-			FPSet: delivery.FingerprintSet(pulls)})
-		ckey, _ := delivery.CandidatesKey("fix/1126")
-		b.kv[delivery.CandidatesBucket+"/"+ckey] = raw
-		env := b.env()
-		var tick atomic.Int64
-		env.Now = func() time.Time { return time.Unix(0, 0).Add(time.Duration(tick.Add(1)) * time.Second) }
-		s := NewShadow(env, nil)
-		s.c.placements["n1"] = delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
-			Modules: []delivery.PlacedModule{{Module: "lemurs"}}}
-		s.c.records["n1"] = delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}
-		s.c.modules["lemurs"] = delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}
-		s.c.modules["other"] = delivery.Module{FP2: fp, Manifest: digestOther, Repository: "novox/x", Commit: "c"}
-		s.ready.Store(true)
-		key, _ := delivery.CompositionKey("fix/1126", heads)
-		return s, key
-	}
-	readsOf := func(b *fakeBus, key string) (n int) {
-		b.mu.Lock()
-		defer b.mu.Unlock()
-		for _, r := range b.reads {
-			if r == [2]string{delivery.ManifestsBucket, key} {
-				n++
 			}
+			head := delivery.Head{Repository: "novox/x", Number: 7, Head: "c0ffee"}
+			heads, _ := delivery.HeadsHash([]delivery.Head{head})
+			pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"lemurs"},
+				Fps: map[string]delivery.CandidateModule{"lemurs": {FP2: fpNew, Manifest: digest2}}}}
+			raw, _ := delivery.EncodeCandidate(delivery.Candidate{Group: "fix/1126", HeadsHash: heads, Pulls: pulls,
+				FPSet: delivery.FingerprintSet(pulls)})
+			ckey, _ := delivery.CandidatesKey("fix/1126")
+			b.kv[delivery.CandidatesBucket+"/"+ckey] = raw
+			env := b.env()
+			var tick atomic.Int64
+			env.Now = func() time.Time { return time.Unix(0, 0).Add(time.Duration(tick.Add(1)) * time.Second) }
+			s := NewShadow(env, nil)
+			s.c.placements["n1"] = delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
+				Modules: []delivery.PlacedModule{{Module: "lemurs"}}}
+			s.c.records["n1"] = delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}
+			s.c.modules["lemurs"] = delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}
+			s.c.modules["other"] = delivery.Module{FP2: fp, Manifest: digestOther, Repository: "novox/x", Commit: "c"}
+			s.ready.Store(true)
+			key, _ := delivery.CompositionKey("fix/1126", heads)
+			return s, key
 		}
-		return n
+		readsOf := func(b *fakeBus, key string) (n int) {
-	}
+			b.mu.Lock()
-	saidLast := func(b *fakeBus) string {
+			defer b.mu.Unlock()
-		b.mu.Lock()
+			for _, r := range b.reads {
-		defer b.mu.Unlock()
+				if r == [2]string{delivery.ManifestsBucket, key} {
-		if len(b.said) == 0 {
+					n++
-			return ""
-		}
-		return b.said[len(b.said)-1]
-	}
-
-	for _, tc := range []struct {
-		name string
-		// stuck: the read of other's manifest does not answer until the ask's bound ends.
-		stuck   bool
-		asks    int
-		verdict string
-		reads   int    // of other's manifest, over every ask
-		words   string // in the answer's words
-		said    string // in the line said of the last answer
-	}{
-		{name: "one ask reads each manifest once", asks: 1, verdict: delivery.ComposeCandidateComposed, reads: 1,
-			said: "compose-candidate fix/1126."},
-		{name: "a later ask reads none again", asks: 2, verdict: delivery.ComposeCandidateComposed, reads: 1,
-			said: "read(s) waited on"},
-		{name: "a read that does not answer is named at the bound", stuck: true, asks: 1, verdict: delivery.ComposeCandidateUnready,
-			words: "waits on", said: "answered unready"},
-	} {
-		t.Run(tc.name, func(t *testing.T) {
-			b := newFakeBus()
-			s, key := shadow(b)
-			ctx, cancel := context.WithCancel(context.Background())
-			defer cancel()
-			if tc.stuck {
-				get := s.c.env.Get
-				s.c.env.Get = func(rctx context.Context, bucket, k string) ([]byte, error) {
-					if bucket == delivery.ManifestsBucket && k == otherKey {
-						b.mu.Lock()
-						b.reads = append(b.reads, [2]string{bucket, k})
-						b.mu.Unlock()
-						cancel() // the ask's bound ends while the read waits
-						<-rctx.Done()
-						return nil, rctx.Err()
-					}
-					return get(rctx, bucket, k)
 				}
 			}
-			var a delivery.ComposeCandidateAnswer
+			return n
-			for range tc.asks {
+		}
-				a = s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: key})
+		saidLast := func(b *fakeBus) string {
+			b.mu.Lock()
+			defer b.mu.Unlock()
+			if len(b.said) == 0 {
+				return ""
 			}
-			if a.Verdict != tc.verdict || !strings.Contains(a.Refused, tc.words) {
+			return b.said[len(b.said)-1]
-				t.Errorf("answered %s %q; want %s with %q", a.Verdict, a.Refused, tc.verdict, tc.words)
+		}
-			}
+
-			if tc.stuck && !strings.Contains(a.Refused, "manifest") {
+		for _, tc := range []struct {
-				t.Errorf("answered %q; want the manifest it waits on named", a.Refused)
+			name string
-			}
+			// stuck: the read of other's manifest does not answer until the ask's bound ends.
-			if got := readsOf(b, otherKey); tc.reads > 0 && got != tc.reads {
+			stuck   bool
-				t.Errorf("other's manifest was read %d time(s); want %d", got, tc.reads)
+			asks    int
-			}
+			verdict string
-			if line := saidLast(b); !strings.Contains(line, tc.said) {
+			reads   int    // of other's manifest, over every ask
-				t.Errorf("said %q; want a line with %q", line, tc.said)
+			words   string // in the answer's words
-			}
+			said    string // in the line said of the last answer
-		})
+		}{
-	}
+			{name: "one ask reads each manifest once", asks: 1, verdict: delivery.ComposeCandidateComposed, reads: 1,
+				said: "compose-candidate fix/1126."},
+			{name: "a later ask reads none again", asks: 2, verdict: delivery.ComposeCandidateComposed, reads: 1,
+				said: "read(s) waited on"},
+			{name: "a read that does not answer is named at the bound", stuck: true, asks: 1, verdict: delivery.ComposeCandidateUnready,
+				words: "waits on", said: "answered unready"},
+		} {
+			t.Run(tc.name, func(t *testing.T) {
+				b := newFakeBus()
+				s, key := shadow(b)
+				ctx, cancel := context.WithCancel(context.Background())
+				defer cancel()
+				if tc.stuck {
+					get := s.c.env.Get
+					s.c.env.Get = func(rctx context.Context, bucket, k string) ([]byte, error) {
+						if bucket == delivery.ManifestsBucket && k == otherKey {
+							b.mu.Lock()
+							b.reads = append(b.reads, [2]string{bucket, k})
+							b.mu.Unlock()
+							cancel() // the ask's bound ends while the read waits
+							<-rctx.Done()
+							return nil, rctx.Err()
+						}
+						return get(rctx, bucket, k)
+					}
+				}
+				var a delivery.ComposeCandidateAnswer
+				for range tc.asks {
+					a = s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: key})
+				}
+				if a.Verdict != tc.verdict || !strings.Contains(a.Refused, tc.words) {
+					t.Errorf("answered %s %q; want %s with %q", a.Verdict, a.Refused, tc.verdict, tc.words)
+				}
+				if tc.stuck && !strings.Contains(a.Refused, "manifest") {
+					t.Errorf("answered %q; want the manifest it waits on named", a.Refused)
+				}
+				if got := readsOf(b, otherKey); tc.reads > 0 && got != tc.reads {
+					t.Errorf("other's manifest was read %d time(s); want %d", got, tc.reads)
+				}
+				if line := saidLast(b); !strings.Contains(line, tc.said) {
+					t.Errorf("said %q; want a line with %q", line, tc.said)
+				}
+			})
+		}
+	})
 }