J23 A-292: a node seat event the bus did not take is said again on the same link

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
This commit is contained in:
2026-11-12 16:57:36 +01:00
parent 0c37ac378c
commit a9b9709911
2 changed files with 98 additions and 7 deletions
+42 -1
View File
@@ -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))
+56 -6
View File
@@ -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{}