diff --git a/cmd/mesh-host/seatlist.go b/cmd/mesh-host/seatlist.go index a480650..7ae8c9f 100644 --- a/cmd/mesh-host/seatlist.go +++ b/cmd/mesh-host/seatlist.go @@ -126,12 +126,18 @@ func (w findingsOfTheNode) Put(f views.Finding) error { return w.session.PutFinding(key, raw) } +// keyOf is the finding's key in the state; a read or a clear needs no more of the finding than Kind and ID. +func (w findingsOfTheNode) keyOf(f views.Finding) string { + return conditions.Key(conditions.ScopeMachine, f.ID, f.Kind) +} + func (w findingsOfTheNode) Delete(f views.Finding) error { - key, _, err := w.encode(f) - if err != nil { - return err - } - return w.session.DeleteFinding(key) + return w.session.DeleteFinding(w.keyOf(f)) +} + +// Has says whether the finding stands in the node's state `findings` now. +func (w findingsOfTheNode) Has(f views.Finding) (bool, error) { + return w.session.HasFinding(w.keyOf(f)) } // startSeatList starts this run's reader: its bootstrapped seats are given to catalogue at once, and every change of what diff --git a/internal/link/findings.go b/internal/link/findings.go index 0b0ba3f..c0356f9 100644 --- a/internal/link/findings.go +++ b/internal/link/findings.go @@ -42,6 +42,21 @@ func (q *Session) PutFinding(key string, value []byte) error { return err } +// HasFinding says whether one finding of this engine stands in its state now. +func (q *Session) HasFinding(key string) (bool, error) { + kv, err := q.findings() + if err != nil { + return false, err + } + if _, err = kv.Get(key); err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return false, nil + } + return false, err + } + return true, nil +} + // DeleteFinding clears one finding of this engine over the link open now; a finding already gone is cleared. func (q *Session) DeleteFinding(key string) error { kv, err := q.findings() diff --git a/internal/seatwait/seatwait.go b/internal/seatwait/seatwait.go index 9c74565..db6c1c2 100644 --- a/internal/seatwait/seatwait.go +++ b/internal/seatwait/seatwait.go @@ -36,20 +36,32 @@ type Clock interface { AfterFunc(d time.Duration, f func()) Timer } +// Findings is the node's own state `findings`: the writer, and the read of whether a finding stands there. The bus state +// is the one truth of whether the finding exists (ADR 0636 §1, issue 1966): the Waiter keeps no copy, so a new process +// of the node-engine clears what an earlier one wrote. +type Findings interface { + views.FindingWriter + Has(f views.Finding) (bool, error) +} + // Waiter writes and clears the node's finding. type Waiter struct { Node string Observer string Reader Reader - Findings views.FindingWriter + Findings Findings Clock Clock // Say tells the node's journal; nil says nothing. Say func(string) - mu sync.Mutex - timer Timer - since time.Time - written *views.Finding + mu sync.Mutex + timer Timer + since time.Time +} + +// key is the finding this node's Waiter writes: Kind and ID name it, the rest is its text. +func (w *Waiter) key() views.Finding { + return views.Finding{Observer: w.Observer, Kind: Kind, ID: w.Node} } func waiting(st seatlist.Status) bool { return !st.Heard || len(st.Waiting) > 0 } @@ -66,18 +78,24 @@ func (w *Waiter) Changed() { w.timer = nil } w.since = time.Time{} - if w.written != nil { - if err := w.Findings.Delete(*w.written); err != nil { + stands, err := w.Findings.Has(w.key()) + if err != nil { + w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be read: %v", err)) + return + } + if stands { + if err := w.Findings.Delete(w.key()); err != nil { w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be cleared: %v", err)) - return } - w.written = nil } return } - if w.timer != nil || w.written != nil { + if w.timer != nil { return } + if stands, err := w.Findings.Has(w.key()); err == nil && stands { + return // an earlier process wrote it and the list is still unread: it stands until the first full read + } w.since = w.Clock.Now() w.timer = w.Clock.AfterFunc(Bound, w.expired) } @@ -86,7 +104,11 @@ func (w *Waiter) Changed() { func (w *Waiter) Waiting() (bool, time.Time) { w.mu.Lock() defer w.mu.Unlock() - return w.timer != nil || w.written != nil, w.since + if w.timer != nil { + return true, w.since + } + stands, err := w.Findings.Has(w.key()) + return err == nil && stands, w.since } func (w *Waiter) expired() { @@ -98,6 +120,9 @@ func (w *Waiter) expired() { w.since = time.Time{} return } + if stands, err := w.Findings.Has(w.key()); err == nil && stands { + return + } f := views.Finding{Observer: w.Observer, Kind: Kind, ID: w.Node, Summary: fmt.Sprintf("%s has read no complete seat list since %s: %s", w.Node, w.since.UTC().Format(time.RFC3339), st.Health()), Headline: fmt.Sprintf("%s waits for the seat list", w.Node), @@ -107,7 +132,6 @@ func (w *Waiter) expired() { w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be written: %v", err)) return } - w.written = &f w.say(f.Summary) } diff --git a/internal/seatwait/seatwait_test.go b/internal/seatwait/seatwait_test.go index c04db1a..f34aa5c 100644 --- a/internal/seatwait/seatwait_test.go +++ b/internal/seatwait/seatwait_test.go @@ -48,16 +48,24 @@ func (c *fakeClock) advance(d time.Duration) { type fakeFindings struct { put, deleted []views.Finding err error + standing bool // the bus state holds the finding, written by this process or an earlier one } +func (f *fakeFindings) Has(views.Finding) (bool, error) { return f.standing, nil } + func (f *fakeFindings) Put(x views.Finding) error { if f.err != nil { return f.err } f.put = append(f.put, x) + f.standing = true + return nil +} +func (f *fakeFindings) Delete(x views.Finding) error { + f.deleted = append(f.deleted, x) + f.standing = false return nil } -func (f *fakeFindings) Delete(x views.Finding) error { f.deleted = append(f.deleted, x); return nil } // ADR 0636 §1: a reader that waited five minutes makes the node write one finding, which the first full read clears. func caseTheNodeWritesOneFindingAfterFiveMinutesAndClearsItAtTheFirstFullRead(t *testing.T) { @@ -137,12 +145,35 @@ func caseAFindingThatCannotBeWrittenIsSaidAndTriedAgain(t *testing.T) { } } +// A finding an earlier process wrote stands in the bus state: the new process clears it at its first full read with no +// write first, and while the list stays unread it neither starts a bound nor writes it again (ADR 0636 §1, issue 1966). +func caseAFindingOfAnEarlierProcessIsClearedAtTheFirstFullRead(t *testing.T) { + reader := &fakeReader{st: seatlist.Status{Heard: false}} + clock := &fakeClock{now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)} + findings := &fakeFindings{standing: true} + w := &Waiter{Node: "ace", Observer: "node-engine", Reader: reader, Findings: findings, Clock: clock} + w.Changed() + clock.advance(Bound) + if len(findings.put) != 0 || len(clock.timers) != 0 { + t.Fatalf("a standing finding was written again (%+v) or a bound started (%d)", findings.put, len(clock.timers)) + } + if on, _ := w.Waiting(); !on { + t.Error("the waiter does not say the finding stands") + } + reader.st = seatlist.Status{Heard: true} + w.Changed() + if len(findings.deleted) != 1 || findings.deleted[0].ID != "ace" || len(findings.put) != 0 { + t.Fatalf("the first full read left the earlier process's finding: deleted %+v, put %+v", findings.deleted, findings.put) + } +} + // ADR 0636 §1, the node's finding of a seat list unread. func TestTheSeatListWait(t *testing.T) { for name, run := range map[string]func(*testing.T){ - "five minutes write one finding and the first full read clears it": caseTheNodeWritesOneFindingAfterFiveMinutesAndClearsItAtTheFirstFullRead, - "a read within the bound writes nothing": caseAReadWithinTheBoundWritesNothing, - "a finding that cannot be written is said and tried again": caseAFindingThatCannotBeWrittenIsSaidAndTriedAgain, + "five minutes write one finding and the first full read clears it": caseTheNodeWritesOneFindingAfterFiveMinutesAndClearsItAtTheFirstFullRead, + "a read within the bound writes nothing": caseAReadWithinTheBoundWritesNothing, + "a finding that cannot be written is said and tried again": caseAFindingThatCannotBeWrittenIsSaidAndTriedAgain, + "a finding of an earlier process is cleared at the first full read": caseAFindingOfAnEarlierProcessIsClearedAtTheFirstFullRead, } { t.Run(name, run) }