novox/mesh-host
9
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request 'The seat-list finding is read from the node's own state, so a restart clears it (issue 1966)' (#223) from fix/1966-seat-list-finding-from-the-bus into main

Browse Source
This commit was merged in pull request #223.
This commit is contained in:
mesh-tools
2026-11-11 01:34:11 +00:00
parent 783a2d2207 e64c8ad0ca
commit 14c22b4a45
5 changed files with 102 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-host/seatlist.go
+11 -5
View File
@@ -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
internal/link/findings.go
+15
View File
@@ -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()
internal/nodegrant/testdata/node.json
Vendored
+5 -1
View File
@@ -13,6 +13,7 @@
"$JS.API.CONSUMER.CREATE.KV_mesh-delivery-indexer_verdicts.>",
"$JS.API.CONSUMER.CREATE.KV_mesh-host_acts.*.$KV.mesh-host_acts.NODE.>",
"$JS.API.CONSUMER.CREATE.KV_mesh-host_data.*.$KV.mesh-host_data.NODE.>",
"$JS.API.CONSUMER.CREATE.KV_mesh-host_findings.*.$KV.mesh-host_findings.machine.NODE.>",
"$JS.API.CONSUMER.CREATE.KV_mesh-host_health.*.$KV.mesh-host_health.NODE",
"$JS.API.CONSUMER.CREATE.KV_mesh-host_packages.*.$KV.mesh-host_packages.NODE",
"$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_provisions.*.$KV.mesh-provisioner_provisions.NODE.>",
@@ -31,6 +32,7 @@
"$JS.API.CONSUMER.DELETE.KV_mesh-delivery-indexer_verdicts.>",
"$JS.API.CONSUMER.DELETE.KV_mesh-host_acts.>",
"$JS.API.CONSUMER.DELETE.KV_mesh-host_data.>",
"$JS.API.CONSUMER.DELETE.KV_mesh-host_findings.>",
"$JS.API.CONSUMER.DELETE.KV_mesh-host_health.>",
"$JS.API.CONSUMER.DELETE.KV_mesh-host_packages.>",
"$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_provisions.>",
@@ -49,6 +51,7 @@
"$JS.API.DIRECT.GET.KV_mesh-delivery-indexer_verdicts.>",
"$JS.API.DIRECT.GET.KV_mesh-host_acts.$KV.mesh-host_acts.NODE.>",
"$JS.API.DIRECT.GET.KV_mesh-host_data.$KV.mesh-host_data.NODE.>",
"$JS.API.DIRECT.GET.KV_mesh-host_findings.$KV.mesh-host_findings.machine.NODE.>",
"$JS.API.DIRECT.GET.KV_mesh-host_health.$KV.mesh-host_health.NODE",
"$JS.API.DIRECT.GET.KV_mesh-host_packages.$KV.mesh-host_packages.NODE",
"$JS.API.DIRECT.GET.KV_mesh-provisioner_provisions.$KV.mesh-provisioner_provisions.NODE.>",
@@ -84,6 +87,7 @@
"$JS.FC.KV_mesh-delivery-indexer_verdicts.>",
"$JS.FC.KV_mesh-host_acts.>",
"$JS.FC.KV_mesh-host_data.>",
"$JS.FC.KV_mesh-host_findings.>",
"$JS.FC.KV_mesh-host_health.>",
"$JS.FC.KV_mesh-host_packages.>",
"$JS.FC.KV_mesh-provisioner_provisions.>",
@@ -92,7 +96,7 @@
"$KV.composed-NODE.>",
"$KV.mesh-host_acts.NODE.>",
"$KV.mesh-host_data.NODE.>",
"$KV.mesh-host_findings.>",
"$KV.mesh-host_findings.machine.NODE.>",
"$KV.mesh-host_health.NODE",
"$KV.mesh-host_packages.NODE",
"$KV.node-engine_have.NODE.>",
internal/seatwait/seatwait.go
+36 -12
View File
@@ -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)
}
internal/seatwait/seatwait_test.go
+35 -4
View File
@@ -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)
}