From 395a1a0f40cddf6051cdb1657bd227baed516ebe Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 16:29:21 +0100 Subject: [PATCH] F-delivery 3: the watchman waits for created only for entries not yet said created The bus module says created once per entry, so every later accepted manifest of the module raised a false bus-state-overdue five minutes on. The watchman keeps a mark per entry said created, on its pulse state with no time to live and read back at a restart; a refused or an entry no longer asked drops it. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../internal/watch/busstate.go | 40 ++++++++++++++++++- .../internal/watch/findings_test.go | 12 ++++++ .../internal/watch/signals.go | 9 +++++ .../internal/watch/watch.go | 4 +- 4 files changed, 63 insertions(+), 2 deletions(-) diff --git a/modules/mesh-delivery-watchman/internal/watch/busstate.go b/modules/mesh-delivery-watchman/internal/watch/busstate.go index 0141ca8e..8828c3e5 100644 --- a/modules/mesh-delivery-watchman/internal/watch/busstate.go +++ b/modules/mesh-delivery-watchman/internal/watch/busstate.go @@ -20,6 +20,11 @@ import ( // The bound is a row of the signals table: a pulse key per entry with OverdueBound to live, whose expiry is the finding, // so no loop looks and a pulse that stood across a restart still acts. Both findings outlive a restart (signalKind), // since what ends them is heard, not read. +// +// The bus module says `created` once per entry, so an entry it said created is not waited on again when a later manifest +// of its module is accepted with the same entry (review round 1): its mark, a pulse key with no time to live under the +// row bus-state-created, is read back at a restart (SeedSignal), and goes when a refused is heard or a manifest no +// longer asks for the entry, so a later ask of it waits again. // OverdueBound is how long an accepted entry may wait for the bus module's created or refused. const OverdueBound = 5 * time.Minute @@ -29,6 +34,7 @@ const ( KindBusStateOverdue = "bus-state-overdue" KindBusStateRefused = "bus-state-refused" rowBusState = "bus-state" + rowBusCreated = "bus-state-created" ) // BrokerSaid is the part of the bus module's created or refused the watchman reads: the module, the entry, and why a @@ -63,7 +69,10 @@ func (w *Watchman) Accepted(a delivery.ManifestAccepted) { continue } asked[e.Name] = true - // One already said created or refused since an earlier verdict waits again: a new verdict asks the bus module again. + if w.busCreated[signalKey(rowBusCreated, busStateID(a.Module, e.Name))] { + continue // said created already, and created is said once per entry: nothing to wait for + } + // One refused since an earlier verdict waits again: a new verdict asks the bus module again. w.signal(rowBusState, busStateID(a.Module, e.Name), OverdueBound, signalDue{Do: dueFind, Finding: Finding{ Scope: ScopeModule, ID: a.Module, Token: KindBusStateOverdue + "-" + e.Name, Kind: KindBusStateOverdue, Severity: Warning, Source: rowBusState, Since: w.now().Add(OverdueBound).UTC().Truncate(time.Second), @@ -83,6 +92,33 @@ func (w *Watchman) Accepted(a delivery.ManifestAccepted) { w.unsignal(rowBusState, busStateID(a.Module, local)) w.unfind(overdueKey(a.Module, local)) w.unfind(refusedKey(a.Module, local)) + w.uncreated(a.Module, local) + } +} + +// markCreated keeps that the bus module said module's local created, on the pulse bucket with no time to live. +func (w *Watchman) markCreated(module, local string) { + pk := signalKey(rowBusCreated, busStateID(module, local)) + owes, _ := json.Marshal(signalDue{Do: dueCreated}) + value, err := delivery.EncodePulse(delivery.Pulse{Bucket: signalBucket, Key: strings.TrimPrefix(pk, signalBucket+"."), Owes: string(owes)}) + if err == nil { + err = w.out.PutPulse(pk, value, 0) + } + if err != nil { + w.say("cannot keep that %s of %s was created (%v): its next accepted manifest waits on it again", local, module, err) + } + w.busCreated[pk] = true +} + +// uncreated drops the mark of module's local said created, so its next ask waits again. +func (w *Watchman) uncreated(module, local string) { + pk := signalKey(rowBusCreated, busStateID(module, local)) + if !w.busCreated[pk] { + return + } + delete(w.busCreated, pk) + if err := w.out.DeletePulse(pk); err != nil { + w.say("cannot delete the mark %s: %v", pk, err) } } @@ -93,6 +129,7 @@ func (w *Watchman) BrokerCreated(s BrokerSaid) { w.unsignal(rowBusState, busStateID(s.Module, s.Local)) w.unfind(overdueKey(s.Module, s.Local)) w.unfind(refusedKey(s.Module, s.Local)) + w.markCreated(s.Module, s.Local) } // BrokerRefused is the bus module's refused heard: the entry's wait ends in the finding bus-state-refused, until a later @@ -102,6 +139,7 @@ func (w *Watchman) BrokerRefused(s BrokerSaid) { defer w.mu.Unlock() w.unsignal(rowBusState, busStateID(s.Module, s.Local)) w.unfind(overdueKey(s.Module, s.Local)) + w.uncreated(s.Module, s.Local) why := strings.TrimSpace(s.Why) if why == "" { why = "it gave no reason" diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index d9c84281..097b249a 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -106,6 +106,18 @@ func TestTheWatchmansFindings(t *testing.T) { if _, open := o.findings[overdueKey("mesh-knowledge", "records")]; open { t.Fatal("created did not clear overdue") } + // Review round 1: created is said once per entry, so a second accepted manifest with the same entries waits on + // none of them, and the mark that says so has no time to live. + delete(o.pulses, pk("records")) // expired above; the fake keeps an expired pulse + w.Accepted(accepted("records", "drafts")) + for _, local := range []string{"records", "drafts"} { + if _, waits := o.pulses[pk(local)]; waits { + t.Fatalf("a second accepted manifest waits again on %s, said created already", local) + } + if mk := signalKey(rowBusCreated, busStateID("mesh-knowledge", local)); o.ttls[mk] != 0 || o.pulses[mk].Owes == "" { + t.Fatalf("the mark of %s created is %v with time to live %v", local, o.pulses[mk], o.ttls[mk]) + } + } w.BrokerRefused(BrokerSaid{Module: "mesh-knowledge", Local: "kept", Why: "the cap is above the mesh's bound"}) if f := o.findings[refusedKey("mesh-knowledge", "kept")]; f.Severity != Urgent || !strings.Contains(f.Summary, "cap is above") { t.Fatalf("findings = %v", o.findings) diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index 042a5321..da693958 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -72,6 +72,8 @@ const ( // dueApply judges a machine's S21 again (applyCheck); dueRecount counts DEAD_LETTERS again (a restricted letter aged out). dueApply = "apply" dueRecount = "recount" + // dueCreated is no due: the mark, with no time to live, that the bus module said an entry created (busstate.go). + dueCreated = "created" ) // signalKey is the pulse key of a row's signal about id. @@ -126,6 +128,13 @@ func (w *Watchman) unsignal(row, id string) { func (w *Watchman) SeedSignal(key string, value []byte) { w.mu.Lock() defer w.mu.Unlock() + if p, err := delivery.DecodePulse(value); err == nil && p.Bucket == signalBucket { + var due signalDue + if json.Unmarshal([]byte(p.Owes), &due) == nil && due.Do == dueCreated { + w.busCreated[key] = true + return + } + } if due, ok := dueOf(value); ok { w.signals[key] = due } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 1e25c8cf..fd910e85 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -74,6 +74,8 @@ type Watchman struct { seededFindings map[string]bool // signals is every signal's pulse standing, by pulse key, with what its expiry does. signals map[string]signalDue + // busCreated is each bus-state entry's mark that the bus module said it created, by its pulse key (busstate.go). + busCreated map[string]bool // away is each machine that said it sleeps or shuts down, until it says it is back. away map[string]bool // haveState is each module's have as last read, unhealthy its writes in a row that said it failed, advised each @@ -111,7 +113,7 @@ type Watchman struct { func New(out Out, now func() time.Time) *Watchman { return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{}, pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: delivery.PulseMoveBound, stepBound: StepBound, - findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, away: map[string]bool{}, + findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, busCreated: map[string]bool{}, away: map[string]bool{}, haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{}, control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{}, health: map[string]delivery.NodeHealth{}, tunnels: map[string]string{},