Merge pull request 'The node-engine seat's client asks a node once more when an event came while an ask ran, never dropping it (issue 1966 note 170; group with mesh-delivery)' (#4) from fix/1966-engine-reads-asked-again into main
This commit was merged in pull request #4.
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user