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) + } + }) + } + }) }