From e575659187fd406c90a7070e2fabff3aa57355e6 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 10:46:28 +0100 Subject: [PATCH 1/2] enginereads: a statement older than an event heard does not stand (issue 1975) Builds: issue 1975 note 4 (the lead's handover; no design line) --- .../enginereads/enginereads.go | 76 ++++++++++++++++++- .../enginereads/enginereads_test.go | 63 +++++++++++++++ 2 files changed, 137 insertions(+), 2 deletions(-) diff --git a/go/clients/nodeengineclient/enginereads/enginereads.go b/go/clients/nodeengineclient/enginereads/enginereads.go index abe620e..6f0f27d 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads.go +++ b/go/clients/nodeengineclient/enginereads/enginereads.go @@ -251,8 +251,17 @@ type Keeper struct { again map[string]string // eventAt is, per node, when its event manifest-reads-changed was last heard (Statement.EventAt). eventAt map[string]time.Time + // said is, per node, the newest event heard with what it says the engine states: a statement older than it is behind + // (issue 1975's kind: an answer taken while the engine still stated its bootstrapped seats stood after the full list). + said map[string]nodeengine.ManifestReadsChangedEvent + // catching counts, per node, the asks made since the last event only to catch a statement up to it. + catching map[string]int } +// MaxCatchUp is how many asks one event may cause beyond its own while the engine's answer stays behind it: an engine +// that answers older than its own event is said loudly, never asked in a loop. +const MaxCatchUp = 3 + func (k *Keeper) logf(f string, a ...any) { if k.Logf != nil { k.Logf(f, a...) @@ -296,6 +305,58 @@ func (k *Keeper) AskNode(node, why string) { go k.ask(node, why) } +// HeardChanged keeps the newest event heard of a node: until the node's statement is at least as new, the node reads +// as the event says (the engine's own word), and the node is asked again until its statement is (see ask). Call it before +// AskNode for the same event. +func (k *Keeper) HeardChanged(e nodeengine.ManifestReadsChangedEvent) { + at, err := time.Parse(time.RFC3339Nano, e.At) + if err != nil { + return + } + k.mu.Lock() + defer k.mu.Unlock() + if k.said == nil { + k.said, k.catching = map[string]nodeengine.ManifestReadsChangedEvent{}, map[string]int{} + } + if old, ok := k.said[e.Node]; ok { + if oat, err := time.Parse(time.RFC3339Nano, old.At); err == nil && !at.After(oat) { + return + } + } + e.ManifestReads = slices.Clone(e.ManifestReads) + k.said[e.Node] = e + k.catching[e.Node] = 0 +} + +// behindLocked says node's statement is older than the newest event heard of it: the engine said its reads changed +// after the time the statement was made, and to something else. k.mu is held. +func (k *Keeper) behindLocked(node string) bool { + e, ok := k.said[node] + if !ok { + return false + } + at, err := time.Parse(time.RFC3339Nano, e.At) + if err != nil { + return false + } + s := k.by[node] + return s.StatedAt.Before(at) && !slices.Equal(sortedCopy(s.ManifestReads), sortedCopy(e.ManifestReads)) +} + +func sortedCopy(in []string) []string { + out := slices.Clone(in) + slices.Sort(out) + return out +} + +// readsLocked is what node reads: its statement, or the newest event's reads while the statement is behind it. +func (k *Keeper) readsLocked(node string) []string { + if k.behindLocked(node) { + return slices.Clone(k.said[node].ManifestReads) + } + return slices.Clone(k.by[node].ManifestReads) +} + // AskAll asks every node of nodes once, each as AskNode does. func (k *Keeper) AskAll(nodes []string, why string) { for _, n := range nodes { @@ -319,6 +380,16 @@ func (k *Keeper) ask(node, why string) { k.mu.Lock() next, again := k.again[node] delete(k.again, node) + if !again && k.behindLocked(node) { + // The answer taken is older than an event heard: ask until the statement is at least as new. + if k.catching[node] < MaxCatchUp { + k.catching[node]++ + next, again = "its statement is older than the event heard", true + } else { + defer k.logf("%s on %s: the engine answers older than its own event said (%d asks); the node reads as the event says", + Verb, node, MaxCatchUp) + } + } if !again { delete(k.asking, node) } @@ -471,6 +542,7 @@ func (k *Keeper) Forget(node string) { _, had := k.by[node] delete(k.by, node) delete(k.eventAt, node) + delete(k.said, node) k.mu.Unlock() if had && k.Delete != nil { if err := k.Delete(node); err != nil { @@ -504,7 +576,7 @@ func (k *Keeper) Reads(nodes []string) map[string][]string { defer k.mu.Unlock() out := make(map[string][]string, len(nodes)) for _, n := range nodes { - out[n] = slices.Clone(k.by[n].ManifestReads) + out[n] = k.readsLocked(n) } return out } @@ -516,7 +588,7 @@ func (k *Keeper) Present(nodes []string) map[string][]string { out := map[string][]string{} for _, n := range nodes { if s, ok := k.by[n]; ok && s.Answered { - out[n] = slices.Clone(s.ManifestReads) + out[n] = k.readsLocked(n) } } return out diff --git a/go/clients/nodeengineclient/enginereads/enginereads_test.go b/go/clients/nodeengineclient/enginereads/enginereads_test.go index 696814b..c957e9c 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads_test.go +++ b/go/clients/nodeengineclient/enginereads/enginereads_test.go @@ -8,6 +8,8 @@ import ( "strings" "testing" "time" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" ) // The answer and the event are judged as the contract lays them out and its reader checks: the layout's version, a node @@ -210,3 +212,64 @@ func must(b []byte, err error) []byte { } return b } + +// Issue 1975: an engine states its bootstrapped seats, then, a moment later, the full list, and says so with an event. +// An answer taken before the full statement must not stand: the node reads as the event says at once, and is asked until +// its statement is at least as new; an engine that keeps answering older than its own event is asked a bounded number of +// times and not looped on. +func TestAStatementOlderThanAHeardEventDoesNotStand(t *testing.T) { + const early, late = "2026-11-12T01:14:36.498Z", "2026-11-12T01:14:39.001Z" + event := nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", Why: ReadsStarted, At: late, ManifestReads: []string{"seat:a", "seat:b"}} + answer := func(reads, at string) json.RawMessage { + return json.RawMessage(`{"v":1,"node":"ace","manifest-reads":[` + reads + `],"stated-at":"` + at + `"}`) + } + + // The engine's first answer is the bootstrapped one, taken while the event is heard (the event says the full list). + calls := 0 + k := &Keeper{Sync: true, Call: func(string, string, any) (json.RawMessage, error) { + calls++ + if calls == 1 { + return answer(`"seat:a"`, early), nil + } + return answer(`"seat:a","seat:b"`, late), nil + }} + k.AskNode("ace", "the assigner started") + k.HeardChanged(event) + if got := k.Reads([]string{"ace"})["ace"]; len(got) != 2 { + t.Fatalf("a statement older than the event heard stands: reads %v", got) + } + if got := k.Present([]string{"ace"})["ace"]; len(got) != 2 { + t.Errorf("present reads %v, want the event's", got) + } + k.AskNode("ace", nodeengine.EventManifestReadsChanged) + if s := k.Statements()["ace"]; len(s.ManifestReads) != 2 || calls != 2 { + t.Errorf("after the event's ask: %v, %d calls", s.ManifestReads, calls) + } + + // An engine that answers the old statement to every ask: the event's reads stand, the asks are bounded. + calls = 0 + stuck := &Keeper{Sync: true, Call: func(string, string, any) (json.RawMessage, error) { + calls++ + return answer(`"seat:a"`, early), nil + }} + stuck.HeardChanged(event) + stuck.AskNode("ace", nodeengine.EventManifestReadsChanged) + if calls != 1+MaxCatchUp { + t.Errorf("asked %d times, want %d", calls, 1+MaxCatchUp) + } + if got := stuck.Reads([]string{"ace"})["ace"]; len(got) != 2 { + t.Errorf("reads %v, want the event's", got) + } + + // The statement the event describes does not count as behind it: no ask beyond the one. + calls = 0 + even := &Keeper{Sync: true, Call: func(string, string, any) (json.RawMessage, error) { + calls++ + return answer(`"seat:a","seat:b"`, early), nil + }} + even.HeardChanged(event) + even.AskNode("ace", "x") + if calls != 1 { + t.Errorf("a statement that says what the event says was asked %d times", calls) + } +} From 04c6a1eead98db9f55c6f2609e162ed825abcf18 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 11:00:33 +0100 Subject: [PATCH 2/2] enginereads: the event heard is kept in the node's statement (issue 1975 note 6) Builds: issue 1975 note 6 (the lead's review round 1; no design line). The engine 'not yet' half is dropped there. Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../enginereads/enginereads.go | 24 +++++++++++++++++++ .../enginereads/enginereads_test.go | 16 +++++++++++++ 2 files changed, 40 insertions(+) diff --git a/go/clients/nodeengineclient/enginereads/enginereads.go b/go/clients/nodeengineclient/enginereads/enginereads.go index 6f0f27d..7286477 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads.go +++ b/go/clients/nodeengineclient/enginereads/enginereads.go @@ -143,6 +143,15 @@ type Statement struct { // EventAt is when the part last heard the seat's event manifest-reads-changed for the node, beside when it asked and // what the engine stated then (issue 1966 note 170: was the event heard, and was it asked again after it). EventAt time.Time `json:"event-at,omitzero"` + // Heard is the newest event heard of the node with what it says the engine states, kept so the state shows what the + // rules read while the statement is behind it (issue 1975 note 6), and so a part that starts again holds it. + Heard *HeardEvent `json:"heard,omitempty"` +} + +// HeardEvent is an event manifest-reads-changed as the part kept it: when the engine said it and what it said. +type HeardEvent struct { + At time.Time `json:"at"` + ManifestReads []string `json:"manifest-reads,omitempty"` } // Again is a person's ask again of one node, written by the tool and answered by the part that keeps the state. @@ -473,6 +482,11 @@ func (k *Keeper) take(s Statement) { if at := k.eventAt[s.Node]; at.After(s.EventAt) { s.EventAt = at } + if e, ok := k.said[s.Node]; ok { + if at, err := time.Parse(time.RFC3339Nano, e.At); err == nil { + s.Heard = &HeardEvent{At: at.UTC(), ManifestReads: slices.Clone(e.ManifestReads)} + } + } if !s.Answered { s.UnansweredSince = s.AskedAt if old, ok := k.by[s.Node]; ok && !old.Answered && !old.UnansweredSince.IsZero() { @@ -481,6 +495,7 @@ func (k *Keeper) take(s Statement) { } if old, ok := k.by[s.Node]; ok && !s.Answered && old.From != FromNone { old.Answered, old.AskedAt, old.Words, old.AskAgain, old.UnansweredSince = false, s.AskedAt, s.Words, nil, s.UnansweredSince + old.Heard = s.Heard s = old } k.by[s.Node] = s @@ -529,6 +544,15 @@ func (k *Keeper) Heard(node string, raw []byte, deleted bool) { _, had := k.by[node] if !had { k.by[node] = s + if h := s.Heard; h != nil && !h.At.IsZero() { + if k.said == nil { + k.said, k.catching = map[string]nodeengine.ManifestReadsChangedEvent{}, map[string]int{} + } + if _, ok := k.said[node]; !ok { + k.said[node] = nodeengine.ManifestReadsChangedEvent{V: ContractVersion, Node: node, Why: ReadsStarted, + At: h.At.UTC().Format(time.RFC3339Nano), ManifestReads: slices.Clone(h.ManifestReads)} + } + } } k.mu.Unlock() if !had && k.Changed != nil { diff --git a/go/clients/nodeengineclient/enginereads/enginereads_test.go b/go/clients/nodeengineclient/enginereads/enginereads_test.go index c957e9c..2efe8d6 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads_test.go +++ b/go/clients/nodeengineclient/enginereads/enginereads_test.go @@ -246,6 +246,22 @@ func TestAStatementOlderThanAHeardEventDoesNotStand(t *testing.T) { t.Errorf("after the event's ask: %v, %d calls", s.ManifestReads, calls) } + // The event heard is in the node's statement as the part keeps it, and a part that starts again holds it. + var kept []byte + keeper := &Keeper{Sync: true, Put: func(_ string, v []byte) error { kept = v; return nil }, Call: func(string, string, any) (json.RawMessage, error) { + return answer(`"seat:a"`, early), nil + }} + keeper.HeardChanged(event) + keeper.AskNode("ace", "x") + if st, err := Decode(kept); err != nil || st.Heard == nil || len(st.Heard.ManifestReads) != 2 { + t.Errorf("the kept statement shows the event heard as %+v, %v", st.Heard, err) + } + restarted := &Keeper{Sync: true} + restarted.Heard("ace", kept, false) + if got := restarted.Reads([]string{"ace"})["ace"]; len(got) != 2 { + t.Errorf("a part started again reads %v, want the event's", got) + } + // An engine that answers the old statement to every ask: the event's reads stand, the asks are bounded. calls = 0 stuck := &Keeper{Sync: true, Call: func(string, string, any) (json.RawMessage, error) {