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)

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

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