From c9ffed70421a43138eb23daf61debea4f8e4a8aa Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 11:32:56 +0100 Subject: [PATCH] S1-S3: the node-engine seat's client asks a node once more when an ask came while one ran, and shows when the event was heard (issue 1966 note 170) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An event of manifest-reads-changed heard while an ask of the same node ran was dropped without a line, so a restated engine could stay read at its first statement (ace, 2026-11-10 07:36). Every reader goes through Keeper.AskNode, so the assigner, the builder and the indexer are fixed at once (ADR 0668 §4: the caller hears it and calls again; ADR 0634 §1: the seat's client). - AskNode: an ask while one runs is asked once more after it, one follow-up however many came, said in one line (1966/AC170-1, AC170-3). The asks of one node so run one after another, and take keeps the last answer, as before (no rule on the engine's clock: review round 1). - Statement.EventAt (event-at): when the node's event was last heard, read by the parts' engine-reads tools; forgotten with the node's key. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../enginereads/enginereads.go | 53 ++++++++++++++++--- .../enginereads/enginereads_test.go | 43 +++++++++++++++ 2 files changed, 90 insertions(+), 6 deletions(-) diff --git a/go/clients/nodeengineclient/enginereads/enginereads.go b/go/clients/nodeengineclient/enginereads/enginereads.go index 78d02d2..abe620e 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads.go +++ b/go/clients/nodeengineclient/enginereads/enginereads.go @@ -140,6 +140,9 @@ type Statement struct { // UnansweredSince is when the engine first did not answer, of the asks since its last answer; zero once answered // (ADR 0668 §3: a kept call unanswered past five minutes is a finding of its caller). UnansweredSince time.Time `json:"unanswered-since,omitzero"` + // 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"` } // Again is a person's ask again of one node, written by the tool and answered by the part that keeps the state. @@ -243,6 +246,11 @@ type Keeper struct { mu sync.Mutex by map[string]Statement asking map[string]bool + // again is, per node an ask of which runs, the newest why of an ask that came meanwhile: it is asked once more when + // the running ask ends, never dropped (issue 1966 note 170). + again map[string]string + // eventAt is, per node, when its event manifest-reads-changed was last heard (Statement.EventAt). + eventAt map[string]time.Time } func (k *Keeper) logf(f string, a ...any) { @@ -258,14 +266,25 @@ func (k *Keeper) now() time.Time { return time.Now() } -// AskNode asks node's engine, unless an ask of it runs already; why is said in the log. +// AskNode asks node's engine; why is said in the log. When an ask of it runs already, the node is asked once more after +// that ask ends, one follow-up however many asks came meanwhile, so an event heard while an ask runs is never lost (issue +// 1966 note 170; ADR 0668 §4: the caller hears it and calls again). A why that starts with Event is the seat's event, +// whose time the node's statement shows as event-at. func (k *Keeper) AskNode(node, why string) { k.mu.Lock() if k.asking == nil { - k.asking = map[string]bool{} + k.asking, k.again, k.eventAt = map[string]bool{}, map[string]string{}, map[string]time.Time{} + } + if strings.HasPrefix(why, Event) { + k.eventAt[node] = k.now().UTC() } if k.asking[node] { + _, waits := k.again[node] + k.again[node] = why k.mu.Unlock() + if !waits { + k.logf("%s on %s: an ask runs, so it is asked once more after it (%s)", Verb, node, why) + } return } k.asking[node] = true @@ -293,12 +312,25 @@ func (k *Keeper) AtBind(nodes func() []string) func(created time.Time) { } } +// ask asks node, and once more for as long as an ask came while it ran (AskNode); only then is the node no longer asking. func (k *Keeper) ask(node, why string) { - defer func() { + for { + k.askOnce(node, why) k.mu.Lock() - delete(k.asking, node) + next, again := k.again[node] + delete(k.again, node) + if !again { + delete(k.asking, node) + } k.mu.Unlock() - }() + if !again { + return + } + why = "asked again: " + next + } +} + +func (k *Keeper) askOnce(node, why string) { if k.Call != nil && k.Once != nil { first := Ask(k.Once, node, k.now()) if first.Answered { @@ -357,12 +389,19 @@ func (k *Keeper) Asking(node string) bool { } // take keeps s, unless the node did not answer and an earlier statement stands: an away node keeps its last one (ADR 0624 -// §1), marked as not answering. +// §1), marked as not answering. The asks of one node run one after another (AskNode), so the last answer is the newest. +// The statement shows when the node's event was last heard. func (k *Keeper) take(s Statement) { k.mu.Lock() if k.by == nil { k.by = map[string]Statement{} } + if old, ok := k.by[s.Node]; ok && old.EventAt.After(s.EventAt) { + s.EventAt = old.EventAt + } + if at := k.eventAt[s.Node]; at.After(s.EventAt) { + s.EventAt = at + } if !s.Answered { s.UnansweredSince = s.AskedAt if old, ok := k.by[s.Node]; ok && !old.Answered && !old.UnansweredSince.IsZero() { @@ -396,6 +435,7 @@ func (k *Keeper) Heard(node string, raw []byte, deleted bool) { k.mu.Lock() _, had := k.by[node] delete(k.by, node) + delete(k.eventAt, node) k.mu.Unlock() if had && k.Changed != nil { k.Changed() @@ -430,6 +470,7 @@ func (k *Keeper) Forget(node string) { k.mu.Lock() _, had := k.by[node] delete(k.by, node) + delete(k.eventAt, node) k.mu.Unlock() if had && k.Delete != nil { if err := k.Delete(node); err != nil { diff --git a/go/clients/nodeengineclient/enginereads/enginereads_test.go b/go/clients/nodeengineclient/enginereads/enginereads_test.go index 607e3d0..696814b 100644 --- a/go/clients/nodeengineclient/enginereads/enginereads_test.go +++ b/go/clients/nodeengineclient/enginereads/enginereads_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "strings" "testing" "time" @@ -95,6 +96,41 @@ func TestTheAnswerAndTheEventAreJudged(t *testing.T) { if got := u.Unanswered([]string{"ace"}, heard, 5*time.Minute, clock); len(got) != 0 || !u.Statements()["ace"].UnansweredSince.IsZero() { t.Errorf("an answered node is unanswered: %v", got) } + // Issue 1966 note 170, 1966/AC170-1 and AC170-3: an event of a node heard while an ask of it runs is not dropped: the + // node is asked once more after that ask ends (one follow-up however many came), said in one line, and the newer + // statement is kept; the event's time is shown beside the statement (event-at). + const stated26, restated = "2026-11-10T07:36:26Z", "2026-11-10T07:36:30Z" + var lines []string + eclock := time.Date(2026, 11, 10, 7, 36, 26, 0, time.UTC) + asks := 0 + var e *Keeper + e = &Keeper{Sync: true, Now: func() time.Time { return eclock }, + Logf: func(f string, a ...any) { lines = append(lines, fmt.Sprintf(f, a...)) }, + Put: func(_ string, raw []byte) error { _, err := Decode(raw); return err }, + Call: func(node, _ string, _ any) (json.RawMessage, error) { + asks++ + stated := restated + if asks == 1 { + eclock = eclock.Add(4 * time.Second) + e.AskNode(node, Event) + e.AskNode(node, Event) + stated = stated26 + } else if got := e.Statements()[node].StatedAt.Format(time.RFC3339); got != stated26 { + t.Errorf("the second ask began before the first was taken in: held %s", got) + } + return json.RawMessage(`{"v":1,"node":"` + node + `","manifest-reads":["a.b"],"stated-at":"` + stated + `"}`), nil + }} + e.AskNode("ace", Event) + held := e.Statements()["ace"] + if asks != 2 || held.StatedAt.Format(time.RFC3339) != restated || e.Asking("ace") { + t.Errorf("an event during an ask: asked %d times, held stated at %s, asking %v", asks, held.StatedAt, e.Asking("ace")) + } + if again := strings.Count(strings.Join(lines, "\n"), "asked once more"); again != 1 { + t.Errorf("the ask kept for later is said in %d lines, not one: %q", again, lines) + } + if !held.EventAt.Equal(eclock) || !strings.Contains(string(must(Encode(held))), `"event-at":"2026-11-10T07:36:30Z"`) { + t.Errorf("the event's time is not shown: %+v", held) + } if Verb != "manifest-reads" || Event != "manifest-reads-changed" || Consumes != "node-engine.manifest-reads-changed" || Invoke != "seat:node-engine.manifest-reads" { t.Errorf("the names are %s %s %s %s", Verb, Event, Consumes, Invoke) } @@ -167,3 +203,10 @@ func TestReadsCallsSayAnUnansweredNodeHeard(t *testing.T) { t.Errorf("the bound's node is %q %v", n, ok) } } + +func must(b []byte, err error) []byte { + if err != nil { + panic(err) + } + return b +} -- 2.54.0