From b8158affa7af3bc98a50526f7324f71c0573c892 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 01:28:45 +0100 Subject: [PATCH] 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 | 39 ++++- internal/link/handover_test.go | 57 +++++++ internal/selfcompose/candidate_test.go | 223 ++++++++++++------------- 3 files changed, 204 insertions(+), 115 deletions(-) diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go index 7261559..c51d0e9 100644 --- a/internal/link/engineseat.go +++ b/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) { - _ = r.Respond(AnswerEngineCall(handle, verb, l.node, r.Subject(), r.Data())) - }) + answer := EngineAnswer(handle, verb, l.node, apart) 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" diff --git a/internal/link/handover_test.go b/internal/link/handover_test.go index be1b363..2c46bfe 100644 --- a/internal/link/handover_test.go +++ b/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) { diff --git a/internal/selfcompose/candidate_test.go b/internal/selfcompose/candidate_test.go index ad303cd..2efbfbb 100644 --- a/internal/selfcompose/candidate_test.go +++ b/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 -// 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. -func TestComposeCandidateReadsOnceAndAnswersWithinItsBound(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) - - // 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++ + // 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 } - return n - } - saidLast := func(b *fakeBus) string { - b.mu.Lock() - defer b.mu.Unlock() - if len(b.said) == 0 { - 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) + 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++ } } - var a delivery.ComposeCandidateAnswer - for range tc.asks { - a = s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: key}) + return n + } + 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) { - 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) - } - }) - } + 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 + 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) + } + }) + } + }) }