From 4685610c198a60f510cc73beb7bd549e16ff204e Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 02:57:34 +0100 Subject: [PATCH 1/4] H1: the node-engine publishes its seat's events again through a transport fault, under one message id (issue 1966, ADR 0668) One publish of manifest-reads-changed with a 10 s deadline was the likely loss behind note 155; the stream takes the id once, so publishing again is safe. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- internal/link/engineseat.go | 26 ++++++++++++++++++++++---- 1 file changed, 22 insertions(+), 4 deletions(-) diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go index 43605eb..045e94e 100644 --- a/internal/link/engineseat.go +++ b/internal/link/engineseat.go @@ -2,6 +2,8 @@ package link import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -12,6 +14,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/seatlist" "github.com/nats-io/nats.go" @@ -198,15 +201,30 @@ func EngineEndpointMetadata(v catalogue.Verb, node string) map[string]string { "tool": v.Name, "node": node, "interchangeable": "false", "description": v.Description, "schema": string(schema)} } -// SayEngine says one of the seat's events about this node, on the events stream. +// SayEngine says one of the seat's events about this node, on the events stream, said again through a transport fault +// (sayOnce). func (l *natsLink) SayEngine(event string, body any) error { raw, err := json.Marshal(body) if err != nil { return err } - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - _, err = l.js.Publish(EngineEventSubject(event, l.node), raw, nats.Context(ctx)) + subject := EngineEventSubject(event, l.node) + return sayOnce(func(id string) error { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err := l.js.Publish(subject, raw, nats.Context(ctx), nats.MsgId(id)) + return err + }, subject, raw) +} + +// sayOnce publishes one event under a message id made of its subject and its body, and publishes it again through a +// transport fault (no responders, a deadline) within the bus's retry bound: the stream takes the id once, so a publish +// that was stored and whose acknowledgement was lost is not stored twice (novox/hq ADR 0668 §1; issue 1966 note 155, a +// manifest-reads-changed that one publish lost was never said, and the assigner kept a bootstrapped statement). +func sayOnce(publish func(id string) error, subject string, raw []byte) error { + sum := sha256.Sum256(raw) + id := subject + "." + hex.EncodeToString(sum[:12]) + _, err := busretry.DoRepeatable("saying "+subject, func() (struct{}, error) { return struct{}{}, publish(id) }, nil) return err } From 5aa6d39a98a6cc35193c3b5557c975a0b58e4c4c Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 02:57:34 +0100 Subject: [PATCH 2/4] =?UTF-8?q?H2:=20an=20engine=20event=20not=20said=20is?= =?UTF-8?q?=20kept=20and=20said=20at=20the=20next=20link's=20opening=20(is?= =?UTF-8?q?sue=201966,=20ADR=200668=20=C2=A74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A failed say was only a line in the engine's log; now it is said again when a link opens, beside the statement made while no link was open (proved by a step of the node key test, with H1's publish). Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- internal/link/have_test.go | 46 ++++++++++++++++++++++++++++++++++++++ internal/link/session.go | 31 +++++++++++++++++++++---- 2 files changed, 73 insertions(+), 4 deletions(-) diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 225cd5b..76cc969 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -15,6 +15,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/catalogue" "github.com/nats-io/nats.go" ) @@ -40,9 +41,20 @@ type haveLink struct { statedErr error // readsSaid are the node seat's manifest-reads-changed events said on the link (ADR 0618 §4). readsSaid []nodeengine.ManifestReadsChangedEvent + // sayFails fails that many of the node seat's says, as a publish past the bus's retry bound; said is every event said. + sayFails int + said []string } func (l *haveLink) SayEngine(event string, body any) error { + l.hmu.Lock() + if l.sayFails > 0 { + l.sayFails-- + l.hmu.Unlock() + return errors.New("nats: timeout") + } + l.said = append(l.said, event) + l.hmu.Unlock() if e, ok := body.(nodeengine.ManifestReadsChangedEvent); ok && event == nodeengine.EventManifestReadsChanged { l.hmu.Lock() l.readsSaid = append(l.readsSaid, e) @@ -372,6 +384,40 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) { if got, _ = nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); !slices.Equal(got.Record.ManifestReads, catalogue.ManifestReads()) { t.Errorf("the list back whole: the record states %v, want its contract's", got.Record.ManifestReads) } + + // Issue 1966 note 155, ADR 0668 §4: a seat list read while no link is open is said at the link's opening with what + // the engine states then; and an event whose publish failed past the bus's retry bound is kept and said there too, + // never only a line in the log. + q.Detach(l) + saidBefore = len(l.readsSaid) + catalogue.UseSeats(slices.DeleteFunc(slices.Clone(all), func(s catalogue.Seat) bool { return s.Name == "mesh-vault" })) + q.RestateReads(context.Background()) + if len(l.readsSaid) != saidBefore { + t.Errorf("a statement made with no link open was said on none") + } + q.Attach(context.Background(), l) + l.sayFails = 1 + q.SayEngine("running", map[string]string{"module": "m"}) + q.SayReads() + if said := l.readsSaid[saidBefore:]; len(said) != 1 || slices.Contains(said[0].ManifestReads, "seat:mesh-vault") || + !slices.Contains(said[0].ManifestReads, "seat:mesh-catalog") { + t.Errorf("at the link's opening said %+v, want the statement made while no link was open", said) + } + if !slices.Contains(l.said, "running") { + t.Errorf("an event whose publish failed was not said at the next link: said %v", l.said) + } + catalogue.UseSeats(all) + q.RestateReads(context.Background()) + + // And the publish itself (sayOnce): through no responders and a deadline it is published again under one message id, + // so the stream takes it once. + t.Cleanup(busretry.Waits(10*time.Second, time.Millisecond, 2*time.Millisecond)) + var ids []string + faults := []error{nats.ErrNoResponders, context.DeadlineExceeded, nil} + if err := sayOnce(func(id string) error { ids = append(ids, id); return faults[len(ids)-1] }, "s", []byte("{}")); err != nil || + len(ids) != 3 || ids[0] != ids[2] || ids[0] == "" { + t.Errorf("a publish through two faults: %v, ids %q", err, ids) + } } // 1132/AC7: a write the bucket refuses is said in the log, and the next apply writes again. diff --git a/internal/link/session.go b/internal/link/session.go index 7068db1..e4cdf90 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -6,6 +6,8 @@ import ( "errors" "fmt" sdkkeys "git.novox.be/novox/mesh-sdk/go/keys" + "maps" + "slices" "strings" "sync" "time" @@ -92,6 +94,9 @@ type Session struct { reads struct { sync.Mutex linked bool + // unsaid is, per event, the last of the node seat's events that could not be said (no link open, or its + // publish failed past the bus's retry bound): said at the next link's SayReads (novox/hq ADR 0668; issue 1966). + unsaid map[string]any } // restating takes RestateReads one at a time. restating sync.Mutex @@ -446,22 +451,40 @@ func (q *Session) SayReads() { why = ReadsReturned } q.reads.linked = true + unsaid := q.reads.unsaid + q.reads.unsaid = nil q.reads.Unlock() + for _, event := range slices.Sorted(maps.Keys(unsaid)) { + if event != nodeengine.EventManifestReadsChanged { // said below with what the engine states now + q.SayEngine(event, unsaid[event]) + } + } q.SayEngine(nodeengine.EventManifestReadsChanged, nodeengine.ManifestReadsChangedEvent{V: sdkkeys.Version, Node: q.Membership.Node, ManifestReads: manifestReads(), At: time.Now().UTC().Format(time.RFC3339Nano), Why: why}) } -// SayEngine says one of the node seat's events over the link open now; with none open it is not said. +// SayEngine says one of the node seat's events over the link open now. With none open, or a publish that failed past the +// bus's retry bound, it is kept and said at the next link's opening (SayReads; novox/hq ADR 0668 §4, issue 1966): an +// event a reader waits on is never only a line in this engine's log. func (q *Session) SayEngine(event string, body any) { q.init() q.mu.Lock() bus := q.bus q.mu.Unlock() - if sayer, ok := bus.(interface{ SayEngine(string, any) error }); ok { - if err := sayer.SayEngine(event, body); err != nil { - q.Say("could not say the node seat's " + event + ": " + err.Error()) + sayer, ok := bus.(interface{ SayEngine(string, any) error }) + var err error + if ok { + if err = sayer.SayEngine(event, body); err == nil { + return } + q.Say("could not say the node seat's " + event + ", said again at the next link: " + err.Error()) } + q.reads.Lock() + if q.reads.unsaid == nil { + q.reads.unsaid = map[string]any{} + } + q.reads.unsaid[event] = body + q.reads.Unlock() } func (q *Session) operatorAsks() (OperatorAsks, error) { From 9cd03a7430cc6b7dca3f5a97a08a5a799a84c673 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 04:12:04 +0100 Subject: [PATCH 3/4] =?UTF-8?q?H3:=20the=20row=20of=20sayOnce=20feeds=20th?= =?UTF-8?q?e=20old=20JetStream=20client's=20own=20no=20responders,=20nats.?= =?UTF-8?q?ErrNoStreamResponse=20(issue=201966,=20ADR=200668=20=C2=A71;=20?= =?UTF-8?q?review=20round=201,=20finding=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- internal/link/have_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 76cc969..18f50d8 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -409,11 +409,11 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) { catalogue.UseSeats(all) q.RestateReads(context.Background()) - // And the publish itself (sayOnce): through no responders and a deadline it is published again under one message id, + // And the publish itself (sayOnce): through no responders, as the old JetStream client says it, and a deadline it is published again under one message id, // so the stream takes it once. t.Cleanup(busretry.Waits(10*time.Second, time.Millisecond, 2*time.Millisecond)) var ids []string - faults := []error{nats.ErrNoResponders, context.DeadlineExceeded, nil} + faults := []error{nats.ErrNoStreamResponse, context.DeadlineExceeded, nil} if err := sayOnce(func(id string) error { ids = append(ids, id); return faults[len(ids)-1] }, "s", []byte("{}")); err != nil || len(ids) != 3 || ids[0] != ids[2] || ids[0] == "" { t.Errorf("a publish through two faults: %v, ids %q", err, ids) From d581acdedae3b52afbb1e2a737b7f1282145bd21 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 05:57:04 +0100 Subject: [PATCH 4/4] =?UTF-8?q?H4:=20the=20builder=20machine=20of=20the=20?= =?UTF-8?q?toolchain=20tests=20has=20the=20tool=20runner's=20seat-settle-d?= =?UTF-8?q?elay=20(issue=201966;=20mesh-catalog's=20issue=201748,=20ADR=20?= =?UTF-8?q?0664=20=C2=A75)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Since mesh-catalog 6b8e7fc6 the build modules ask the capability seat-settle-delay of the machine, so on the stand-in builder without it mesh-build-go and mesh-build-node were left out as unhostable and both toolchain tests failed against the catalogue's main, on mesh-host's trunk as on this group's head. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- internal/compose/resolution/toolchain_manifests_test.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/internal/compose/resolution/toolchain_manifests_test.go b/internal/compose/resolution/toolchain_manifests_test.go index c6d6725..5c2b7f6 100644 --- a/internal/compose/resolution/toolchain_manifests_test.go +++ b/internal/compose/resolution/toolchain_manifests_test.go @@ -10,10 +10,11 @@ import ( // builderNode is a machine that can run a build module: the package manager and user namespaces it asks of // the machine, and the forge account (mesh-scope, provided by the forge) and the artifact store the build // modules publish to (ADR 0513, provided by distribution) stood in for, and node-tools, the tool runner they pull their -// asks through (mesh-catalog issue 1911; mesh-tools' module, not in the catalogue). +// asks through (mesh-catalog issue 1911; mesh-tools' module, not in the catalogue), with the tool runner's +// seat-settle-delay the build modules ask of it since issue 1748 (ADR 0664 §5). func builderNode() mcat.Node { return mcat.Node{Name: "builder", Site: "house", - Capabilities: map[string]bool{"package-manager": true, "user-namespaces": true}} + Capabilities: map[string]bool{"package-manager": true, "user-namespaces": true, "seat-settle-delay": true}} } func toolchainShelf(t *testing.T) map[string]mcat.Manifest {