From a9b970991198ef0984cb8cf1620f04534b6befc2 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 16:57:36 +0100 Subject: [PATCH] J23 A-292: a node seat event the bus did not take is said again on the same link MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After a restart, g14's engine restated its reads three seconds after its first, bootstrapped statement, and the assigner never heard that event: a say that failed past the bus's retry bound waited for the next link. Now a say the bus did not take (no responders, a deadline) is said again on the link until the bus takes it, off the caller, each failed attempt written to the engine's journal with the event's name; only a lost link, or a refusal the bus answered, keeps it for the next link. Builds: design 47 G §590, A-292 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- internal/link/have_test.go | 43 +++++++++++++++++++++++++- internal/link/session.go | 62 ++++++++++++++++++++++++++++++++++---- 2 files changed, 98 insertions(+), 7 deletions(-) diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 88195a60..ed29f89c 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -45,6 +45,8 @@ type haveLink struct { 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 + // sayFault is what a failed say answers; a bus answering a refusal when nil. + sayFault error said []string // onSaid are the events said on a subject of their own (running), by subject, in order. onSaid []string @@ -77,7 +79,11 @@ func (l *haveLink) SayEngine(event string, body any) error { l.hmu.Lock() if l.sayFails > 0 { l.sayFails-- + fault := l.sayFault l.hmu.Unlock() + if fault != nil { + return fault + } return errors.New("nats: timeout") } l.said = append(l.said, event) @@ -320,7 +326,7 @@ func TestAModulesHealthFollowsTheLook(t *testing.T) { // 1132/AC6: the node's key holds its windows, busy, strays, filters and profile. func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) { l := newHaveLink() - q, _ := aHaveQueue(l) + q, journal := aHaveQueue(l) rep := appliedModules() rep.Windows = []Window{{Step: "backup", Holds: []string{"db"}}} rep.Busy = []Busy{{ID: "x", Container: "x", State: "waiting"}} @@ -465,6 +471,41 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) { catalogue.UseSeats(all) q.RestateReads(context.Background()) + // A-292 (§590, issue 1975): a restatement the bus did not take, past its retry bound, is said again on the same link + // until the bus takes it, never held for the next link, and each failed attempt is in the engine's journal by name. + saidBefore = len(l.readsSaid) + l.hmu.Lock() + l.sayFails, l.sayFault = 2, fmt.Errorf("saying: %w", nats.ErrTimeout) + l.hmu.Unlock() + catalogue.UseSeats(slices.DeleteFunc(slices.Clone(all), func(s catalogue.Seat) bool { return s.Name == "mesh-vault" })) + q.RestateReads(context.Background()) + q.retrying.Wait() + l.hmu.Lock() + again := slices.Clone(l.readsSaid[saidBefore:]) + l.sayFault = nil + l.hmu.Unlock() + if len(again) != 1 || slices.Contains(again[0].ManifestReads, "seat:mesh-vault") { + t.Errorf("a restatement the bus did not take, said on this link: %+v, want it said once with the new statement", again) + } + failed := 0 + for _, line := range *journal { + if strings.Contains(line, "could not say the node seat's "+nodeengine.EventManifestReadsChanged) && + strings.Contains(line, "on this link") { + failed++ + } + } + if failed != 2 { + t.Errorf("the journal names %d failed attempts of %s, want 2: %q", failed, nodeengine.EventManifestReadsChanged, *journal) + } + q.reads.Lock() + _, held := q.reads.unsaid[nodeengine.EventManifestReadsChanged] + q.reads.Unlock() + if held { + t.Errorf("a restatement said on this link is also held for the next link") + } + catalogue.UseSeats(all) + q.RestateReads(context.Background()) + // 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)) diff --git a/internal/link/session.go b/internal/link/session.go index ee9dc0ab..8e687287 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -16,6 +16,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" occ "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/health" "git.novox.be/novox/mesh-sdk/go/operatorask" ) @@ -112,6 +113,8 @@ type Session struct { } // restating takes RestateReads one at a time. restating sync.Mutex + // retrying counts the says being said again on the link open (sayAgain). + retrying sync.WaitGroup // haves is what this node-engine writes beside its facts (novox/hq ADR 0429 §4, issue 1132). haves haveState @@ -524,22 +527,69 @@ func (q *Session) sayEngineOn(subject, event string, body any) { } } -// 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. +// SayEngine says one of the node seat's events over the link open now. A publish the bus did not take is said again on +// that link until it takes it (sayAgain; design 47 G A-292, §590); with no link open, or a link lost meanwhile, or a +// refusal the bus answered, 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() - sayer, ok := bus.(interface{ SayEngine(string, any) error }) - var err error + sayer, ok := bus.(engineSayer) if ok { - if err = sayer.SayEngine(event, body); err == nil { + err := sayer.SayEngine(event, body) + if err == nil { + return + } + if sayableAgain(err) { + // A say the bus did not take is said again on this link until the bus takes it (design 47 G A-292, §590): + // off the caller, so a link's setup and a restatement never wait on it. + q.Say("could not say the node seat's " + event + ", said again on this link until the bus takes it: " + err.Error()) + q.retrying.Add(1) + go q.sayAgain(bus, sayer, event, body) return } q.Say("could not say the node seat's " + event + ", said again at the next link: " + err.Error()) } + q.holdUnsaid(event, body) +} + +// engineSayer is a link that says the node seat's events. +type engineSayer interface{ SayEngine(string, any) error } + +// sayableAgain is an error the SDK's bus retry asks again on: the bus took nothing (no responders, a deadline). An error +// the bus answered is not asked again on this link. +func sayableAgain(err error) bool { return busretry.Undelivered(err) || busretry.TimedOut(err) } + +// sayAgain says event again through the link's bus retry for as long as bus is the link open and the bus did not take it, +// writing each failed attempt to the engine's journal with the event's name (A-292). A link lost meanwhile keeps the +// event for the next link (holdUnsaid). +func (q *Session) sayAgain(bus Bus, sayer engineSayer, event string, body any) { + defer q.retrying.Done() + for q.linkedTo(bus) { + err := sayer.SayEngine(event, body) + if err == nil { + return + } + if !sayableAgain(err) { + q.Say("could not say the node seat's " + event + ", said again at the next link: " + err.Error()) + break + } + q.Say("could not say the node seat's " + event + " again, said again on this link until the bus takes it: " + err.Error()) + } + q.holdUnsaid(event, body) +} + +// linkedTo says bus is the link open now. +func (q *Session) linkedTo(bus Bus) bool { + q.mu.Lock() + defer q.mu.Unlock() + return q.bus != nil && q.bus == bus +} + +// holdUnsaid keeps event for the next link's SayReads: no link is open, or the bus answered it with a refusal. +func (q *Session) holdUnsaid(event string, body any) { q.reads.Lock() if q.reads.unsaid == nil { q.reads.unsaid = map[string]any{}