Merge pull request 'enginereads: a reads statement older than a heard reads-changed does not stand (issue 1975)' (#8) from fix/engine-reads-and-compositions into main
This commit was merged in pull request #8.
This commit is contained in:
@@ -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
|
// 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).
|
// 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"`
|
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.
|
// Again is a person's ask again of one node, written by the tool and answered by the part that keeps the state.
|
||||||
@@ -251,8 +260,17 @@ type Keeper struct {
|
|||||||
again map[string]string
|
again map[string]string
|
||||||
// eventAt is, per node, when its event manifest-reads-changed was last heard (Statement.EventAt).
|
// eventAt is, per node, when its event manifest-reads-changed was last heard (Statement.EventAt).
|
||||||
eventAt map[string]time.Time
|
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) {
|
func (k *Keeper) logf(f string, a ...any) {
|
||||||
if k.Logf != nil {
|
if k.Logf != nil {
|
||||||
k.Logf(f, a...)
|
k.Logf(f, a...)
|
||||||
@@ -296,6 +314,58 @@ func (k *Keeper) AskNode(node, why string) {
|
|||||||
go k.ask(node, why)
|
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.
|
// AskAll asks every node of nodes once, each as AskNode does.
|
||||||
func (k *Keeper) AskAll(nodes []string, why string) {
|
func (k *Keeper) AskAll(nodes []string, why string) {
|
||||||
for _, n := range nodes {
|
for _, n := range nodes {
|
||||||
@@ -319,6 +389,16 @@ func (k *Keeper) ask(node, why string) {
|
|||||||
k.mu.Lock()
|
k.mu.Lock()
|
||||||
next, again := k.again[node]
|
next, again := k.again[node]
|
||||||
delete(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 {
|
if !again {
|
||||||
delete(k.asking, node)
|
delete(k.asking, node)
|
||||||
}
|
}
|
||||||
@@ -402,6 +482,11 @@ func (k *Keeper) take(s Statement) {
|
|||||||
if at := k.eventAt[s.Node]; at.After(s.EventAt) {
|
if at := k.eventAt[s.Node]; at.After(s.EventAt) {
|
||||||
s.EventAt = at
|
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 {
|
if !s.Answered {
|
||||||
s.UnansweredSince = s.AskedAt
|
s.UnansweredSince = s.AskedAt
|
||||||
if old, ok := k.by[s.Node]; ok && !old.Answered && !old.UnansweredSince.IsZero() {
|
if old, ok := k.by[s.Node]; ok && !old.Answered && !old.UnansweredSince.IsZero() {
|
||||||
@@ -410,6 +495,7 @@ func (k *Keeper) take(s Statement) {
|
|||||||
}
|
}
|
||||||
if old, ok := k.by[s.Node]; ok && !s.Answered && old.From != FromNone {
|
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.Answered, old.AskedAt, old.Words, old.AskAgain, old.UnansweredSince = false, s.AskedAt, s.Words, nil, s.UnansweredSince
|
||||||
|
old.Heard = s.Heard
|
||||||
s = old
|
s = old
|
||||||
}
|
}
|
||||||
k.by[s.Node] = s
|
k.by[s.Node] = s
|
||||||
@@ -458,6 +544,15 @@ func (k *Keeper) Heard(node string, raw []byte, deleted bool) {
|
|||||||
_, had := k.by[node]
|
_, had := k.by[node]
|
||||||
if !had {
|
if !had {
|
||||||
k.by[node] = s
|
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()
|
k.mu.Unlock()
|
||||||
if !had && k.Changed != nil {
|
if !had && k.Changed != nil {
|
||||||
@@ -471,6 +566,7 @@ func (k *Keeper) Forget(node string) {
|
|||||||
_, had := k.by[node]
|
_, had := k.by[node]
|
||||||
delete(k.by, node)
|
delete(k.by, node)
|
||||||
delete(k.eventAt, node)
|
delete(k.eventAt, node)
|
||||||
|
delete(k.said, node)
|
||||||
k.mu.Unlock()
|
k.mu.Unlock()
|
||||||
if had && k.Delete != nil {
|
if had && k.Delete != nil {
|
||||||
if err := k.Delete(node); err != nil {
|
if err := k.Delete(node); err != nil {
|
||||||
@@ -504,7 +600,7 @@ func (k *Keeper) Reads(nodes []string) map[string][]string {
|
|||||||
defer k.mu.Unlock()
|
defer k.mu.Unlock()
|
||||||
out := make(map[string][]string, len(nodes))
|
out := make(map[string][]string, len(nodes))
|
||||||
for _, n := range nodes {
|
for _, n := range nodes {
|
||||||
out[n] = slices.Clone(k.by[n].ManifestReads)
|
out[n] = k.readsLocked(n)
|
||||||
}
|
}
|
||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
@@ -516,7 +612,7 @@ func (k *Keeper) Present(nodes []string) map[string][]string {
|
|||||||
out := map[string][]string{}
|
out := map[string][]string{}
|
||||||
for _, n := range nodes {
|
for _, n := range nodes {
|
||||||
if s, ok := k.by[n]; ok && s.Answered {
|
if s, ok := k.by[n]; ok && s.Answered {
|
||||||
out[n] = slices.Clone(s.ManifestReads)
|
out[n] = k.readsLocked(n)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return out
|
return out
|
||||||
|
|||||||
@@ -8,6 +8,8 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"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
|
// 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,80 @@ func must(b []byte, err error) []byte {
|
|||||||
}
|
}
|
||||||
return b
|
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)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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) {
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user