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
2026-11-04 16:29:21 +01:00
parent 2a32721b80
commit 395a1a0f40
4 changed files with 63 additions and 2 deletions
@@ -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"
@@ -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)
@@ -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
 	}
@@ -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{},