WM7: the watchman raises record-open 24 hours after a merge unless another condition covers it, and store-filling from the seats' filling, full and room

Builds: design 47 G §148, §160, §181, §218, §263

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go

@@ -44,6 +44,7 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string,
 		}
 		w.Accepted(a)
 	case pipelineSubject(w, subject, body):
+	case recordSubject(w, subject, body):
 	case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventCreated):
 		if s, err := watch.ReadBrokerSaid(body); err != nil {
 			logf("%v; left", err)
@@ -160,3 +161,24 @@ func pipelineSubject(w *watch.Watchman, subject string, body []byte) bool {
 	}
 	return true
 }
+
+// recordSubject hands the seats' `filling`, `full` and `room` of a store and the seat mesh-delivery's `record-opened` and
+// `record-ended` to the watchman (watch/records.go, design 47 G §148, §181). It says whether the subject was one of them.
+func recordSubject(w *watch.Watchman, subject string, body []byte) bool {
+	parts := strings.Split(subject, ".")
+	if len(parts) != 5 || parts[0] != "mesh" || parts[1] != "seat" || parts[3] != "event" {
+		return false
+	}
+	seat, event := parts[2], parts[4]
+	switch {
+	case event == meshdelivery.EventFilling || event == meshdelivery.EventFull || event == meshdelivery.EventRoom:
+		w.StoreFill(seat, event, body)
+	case seat == meshdelivery.Seat && event == meshdelivery.EventRecordOpened:
+		w.RecordOpen(body)
+	case seat == meshdelivery.Seat && event == meshdelivery.EventRecordEnded:
+		w.RecordEnded(body)
+	default:
+		return false
+	}
+	return true
+}

modules/mesh-delivery-watchman/internal/watch/pipelines_test.go

@@ -137,3 +137,69 @@ func TestAMachinesPresenceIsSaidAsEvents(t *testing.T) {
 		t.Fatalf("said %v, want %v", got, want)
 	}
 }
+
+func fillBody(store string, bytes int64) []byte {
+	b, _ := json.Marshal(meshdelivery.FillingEvent{Store: store, Bytes: bytes, Cap: 100})
+	return b
+}
+
+// design 47 G §181, §263: a store past its share is `store-filling` naming the part and the store, full is urgent, `room` and a
+// fill under 70 % clear it.
+func TestAStoreFillingIsAFinding(t *testing.T) {
+	w, o, _ := setup()
+	key := FindingKey(ScopeDelivery, "mesh-delivery-records", KindStoreFilling)
+	w.StoreFill("mesh-delivery", "filling", fillBody("records", 60))
+	if len(o.findings) != 0 {
+		t.Fatalf("under 70 %%: %v", o.findings)
+	}
+	w.StoreFill("mesh-delivery", "filling", fillBody("records", 85))
+	if o.findings[key].Severity != Warning {
+		t.Fatalf("filling: %v", o.findings)
+	}
+	w.StoreFill("mesh-delivery", "full", fillBody("records", 100))
+	if o.findings[key].Severity != Urgent {
+		t.Fatalf("full: %v", o.findings)
+	}
+	w.StoreFill("mesh-delivery", "room", fillBody("records", 60))
+	if len(o.findings) != 0 {
+		t.Fatalf("room: %v", o.findings)
+	}
+}
+
+func recordBody(id string, at time.Time) []byte {
+	b, _ := json.Marshal(meshdelivery.RecordOpenedEvent{Record: id, Milestone: "47/G", At: &at})
+	return b
+}
+
+// design 47 G §148, §160: a record open 24 hours is `record-open`, cleared by `record-ended`; a record that ends in time
+// raises none; while a store is full, which already tells why, it raises none.
+func TestARecordOpenADayIsAFinding(t *testing.T) {
+	w, o, c := setup()
+	pk := signalKey(recordRow, "r1")
+	w.RecordOpen(recordBody("r1", c.t))
+	if o.ttls[pk] != RecordOpenBound || len(o.findings) != 0 {
+		t.Fatalf("within the bound: %v %v", o.ttls, o.findings)
+	}
+	expire(w, c, pk, RecordOpenBound)
+	key := FindingKey(ScopeDelivery, "r1", KindRecordOpen)
+	if _, ok := o.findings[key]; !ok {
+		t.Fatalf("past 24 hours: %v", o.findings)
+	}
+	w.StoreFill("mesh-delivery", "full", fillBody("records", 100))
+	if _, ok := o.findings[key]; ok {
+		t.Fatalf("a full store covers the record: %v", o.findings)
+	}
+	w.StoreFill("mesh-delivery", "room", fillBody("records", 10))
+	if _, ok := o.findings[key]; !ok {
+		t.Fatalf("room, and the record stands open again: %v", o.findings)
+	}
+	w.RecordEnded(recordBody("r1", c.t))
+	if len(o.findings) != 0 {
+		t.Fatalf("record-ended: %v", o.findings)
+	}
+	w.RecordOpen(recordBody("r2", c.t))
+	w.RecordEnded(recordBody("r2", c.t))
+	if _, ok := o.pulses[signalKey(recordRow, "r2")]; ok {
+		t.Fatal("the pulse of an ended record stands")
+	}
+}

modules/mesh-delivery-watchman/internal/watch/records.go

@@ -0,0 +1,149 @@
+package watch
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
+)
+
+// What the seats say of the delivery's records and stores, heard as the seats' events (design 47 G §148, §181, §218; ADR 0596
+// §2), never read from their state:
+//   - `filling`, `full` and `room` of a store against its cap: `store-filling` while it stands past its cap's share, urgent
+//     when it is full, cleared by `room` or by a `filling` under FillingFrom;
+//   - `record-opened` and `record-ended` of a delivery record: `record-open` when it has stood open for 24 hours, unless
+//     another condition already tells why (`acting-part-late`, or a store `full`).
+// Each is a row of the signals table: its pulse is its bound, written at the event, so a restart loses none.
+
+// The kinds of these findings.
+const (
+	KindActingPartLate = "acting-part-late"
+	KindStoreFilling   = "store-filling"
+	KindRecordOpen     = "record-open"
+)
+
+// The rows of the signals table of these findings, which a restart keeps.
+const (
+	storeRow  = "store-fill"
+	recordRow = "record-open"
+)
+
+// uncovered is the While of a record-open signal: the finding is written at the bound only while no other condition covers
+// the record.
+const uncovered = "uncovered:record-open"
+
+// RecordOpenBound is how long a delivery record may stay open before it is a finding (design 47 G §148).
+const RecordOpenBound = 24 * time.Hour
+
+// FillingFrom is the share of a store's cap under which a `filling` clears its finding (design 47 G §181).
+const FillingFrom = 0.70
+
+// StoreFill hears a seat's `filling`, `full` or `room` of one of its stores. A `filling` is a warning, naming the part and the
+// store, unless its share is under FillingFrom, which clears it; `full` is urgent, naming what it refuses; `room` clears either.
+func (w *Watchman) StoreFill(seat, event string, body []byte) {
+	var f meshdelivery.FillingEvent
+	if err := json.Unmarshal(body, &f); err != nil || f.Store == "" {
+		w.say("the %s of the seat %s does not read, left: %v", event, seat, err)
+		return
+	}
+	id := seat + "-" + f.Store
+	key := FindingKey(ScopeDelivery, id, KindStoreFilling)
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	share := 0.0
+	if f.Cap > 0 {
+		share = float64(f.Bytes) / float64(f.Cap)
+	}
+	since := w.now()
+	if f.At != nil && !f.At.IsZero() {
+		since = *f.At
+	}
+	switch {
+	case event == "room" || (event == "filling" && share < FillingFrom):
+		w.unfind(key)
+	case event == "full":
+		w.find(Finding{Scope: ScopeDelivery, ID: id, Kind: KindStoreFilling, Severity: Urgent, Source: storeRow, Since: since,
+			Summary: fmt.Sprintf("%s's store %s is full (%d of %d bytes)", seat, f.Store, f.Bytes, f.Cap),
+			Said:    "full", Headline: "A store of the delivery is full",
+			Explanation: "The store " + f.Store + " of the seat " + seat + " is at its cap, so it refuses " + orWords(f.Refused, "what it is asked to keep") +
+				" until room is made.",
+			Needs: "room made in the store (a removal, or its cap raised)"})
+	default:
+		w.find(Finding{Scope: ScopeDelivery, ID: id, Kind: KindStoreFilling, Severity: Warning, Source: storeRow, Since: since,
+			Summary: fmt.Sprintf("%s's store %s is at %.0f %% of its cap (%d of %d bytes)", seat, f.Store, share*100, f.Bytes, f.Cap),
+			Said:    "filling", Headline: "A store of the delivery is filling",
+			Explanation: "The store " + f.Store + " of the seat " + seat + " is nearly full; once it is full it refuses new entries.",
+			Needs:       "room made before it is full"})
+	}
+	w.rejudgeRecords()
+}
+
+// RecordOpen hears the seat mesh-delivery's `record-opened`: its finding is due when the record has stood open for 24 hours.
+func (w *Watchman) RecordOpen(body []byte) {
+	var r meshdelivery.RecordOpenedEvent
+	if err := json.Unmarshal(body, &r); err != nil || r.Record == "" {
+		w.say("a record-opened does not read, left: %v", err)
+		return
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	since := w.now()
+	if r.At != nil && !r.At.IsZero() {
+		since = *r.At
+	}
+	f := Finding{Scope: ScopeDelivery, ID: r.Record, Kind: KindRecordOpen, Severity: Warning, Source: recordRow, Since: since,
+		Summary: fmt.Sprintf("the delivery record of %s has stood open since %s", r.Milestone, stamp(since)),
+		Said:    "open for " + RecordOpenBound.String(), Headline: "A delivery is not on every machine yet",
+		Explanation: "The delivery record of " + r.Milestone + " has stood open for " + RecordOpenBound.String() +
+			": a node does not run what it changed, or a step never answered.",
+		Needs: "what it waits for (gate_record names it), or the record ended by hand if a node is gone for good (gate_record_end)"}
+	w.unsignal(recordRow, r.Record)
+	w.unfind(f.Key())
+	if left := RecordOpenBound - w.now().Sub(since); left > 0 {
+		w.signal(recordRow, r.Record, left, signalDue{Do: dueFind, While: uncovered, Finding: f})
+		return
+	}
+	w.opened[r.Record] = f
+	w.rejudgeRecords()
+}
+
+// RecordEnded hears the seat mesh-delivery's `record-ended`: its pulse and its finding end.
+func (w *Watchman) RecordEnded(body []byte) {
+	var r meshdelivery.RecordEndedEvent
+	if err := json.Unmarshal(body, &r); err != nil || r.Record == "" {
+		w.say("a record-ended does not read, left: %v", err)
+		return
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.unsignal(recordRow, r.Record)
+	delete(w.opened, r.Record)
+	w.unfind(FindingKey(ScopeDelivery, r.Record, KindRecordOpen))
+}
+
+// recordCovered says another condition tells the operator why a record is open: a wait past its bound or a full store.
+func (w *Watchman) recordCovered() bool {
+	for _, v := range w.findings {
+		f, err := DecodeFinding(v)
+		if err != nil {
+			continue
+		}
+		if f.Kind == KindActingPartLate || (f.Kind == KindStoreFilling && f.Severity == Urgent) {
+			return true
+		}
+	}
+	return false
+}
+
+// rejudgeRecords writes the finding of each record past its bound that nothing covers, and ends it when something does.
+func (w *Watchman) rejudgeRecords() {
+	for id, f := range w.opened {
+		if w.recordCovered() {
+			w.unfind(f.Key())
+			continue
+		}
+		w.find(f)
+		_ = id
+	}
+}

modules/mesh-delivery-watchman/internal/watch/signals.go

@@ -171,6 +171,11 @@ func (w *Watchman) signalExpired(pk string) {
 	case dueEnd:
 		w.unfind(due.Finding.Key())
 	case dueFind:
+		if due.While == uncovered {
+			w.opened[due.Finding.ID] = due.Finding
+			w.rejudgeRecords()
+			return
+		}
 		if due.While != "" && !w.stillWrong(due.While, due.Finding.Kind) {
 			return
 		}
@@ -181,6 +186,9 @@ func (w *Watchman) signalExpired(pk string) {
 // stillWrong says a have's key still says what a row's finding was waited on: failed for apply-failing, unhealthy for
 // unhealthy.
 func (w *Watchman) stillWrong(haveKey, kind string) bool {
+	if haveKey == uncovered {
+		return !w.recordCovered()
+	}
 	if node, ok := strings.CutPrefix(haveKey, "silent:"); ok {
 		_, open := w.findings[FindingKey(ScopeMachine, node, KindSilent)]
 		return open
@@ -936,6 +944,8 @@ var SignalRows = []struct{ Row, Signal, Bound, Kind string }{
 	{"S21", "a machine's have failed or its provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing},
 	{"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy},
 	{"take-in", "a source revision or a delivery group the indexer holds back (ADR 0633 §5)", TakeInRevisionBound.String() + " a revision, " + TakeInGroupBound.String() + " a group", KindTakeInWaits},
+	{storeRow, "a store of the delivery's parts filling or full (the seats' filling, full and room)", "filling from " + fmt.Sprintf("%.0f", FillingFrom*100) + " %, full urgent", KindStoreFilling},
+	{recordRow, "a delivery record open (the seat mesh-delivery's record-opened, ended by record-ended)", RecordOpenBound.String(), KindRecordOpen},
 	{"provisions", "a provision of mesh-provisioner in waits (ADR 0570 §7)", "at once", KindNeedsOperator},
 }
 

modules/mesh-delivery-watchman/internal/watch/watch.go

@@ -120,6 +120,8 @@ type Watchman struct {
 	queued map[string]map[string]time.Time
 	// awaySaid is each machine whose machine-away was said and not yet its machine-back, with when it was last heard (presence.go).
 	awaySaid map[string]time.Time
+	// opened is each record past its bound and not ended, with its finding (records.go).
+	opened map[string]Finding
 }
 
 // New is a watchman writing to out, reading the time from now.
@@ -130,7 +132,7 @@ func New(out Out, now func() time.Time) *Watchman {
 		haveState: map[string]nodeengine.HaveByNodeModuleValue{}, 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]nodeengine.HealthByNodeValue{}, tunnels: map[string]string{},
-		standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}, heads: map[string]head{}, queued: map[string]map[string]time.Time{}, awaySaid: map[string]time.Time{}}
+		standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}, heads: map[string]head{}, queued: map[string]map[string]time.Time{}, awaySaid: map[string]time.Time{}, opened: map[string]Finding{}}
 }
 
 // Logf sets where a write that failed is said; it is silent until set.

modules/mesh-delivery-watchman/manifest_test.go

@@ -111,7 +111,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
 		t.Errorf("invokes = %v, want %v", m.Invokes, want)
 	}
 	// DLV-M5 its first views, 1585/T20 the indexer's take-in events, and DLV-W1 (ADR 0595 §4): the accepted verdicts and the bus module's created and refused, durably.
-	if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "mesh-delivery.source-moved", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-broker.created", "mesh-broker.refused"}; !reflect.DeepEqual(m.Consumes, want) {
+	if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-broker.created", "mesh-broker.refused"}; !reflect.DeepEqual(m.Consumes, want) {
 		t.Errorf("consumes = %v, want %v", m.Consumes, want)
 	}
 	// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).

modules/mesh-delivery-watchman/module.json

@@ -37,6 +37,20 @@
     "mesh-delivery-indexer.views.modules",
     "mesh-catalog.manifest-accepted",
     "mesh-delivery.source-moved",
+    "mesh-delivery.filling",
+    "mesh-delivery.full",
+    "mesh-delivery.room",
+    "mesh-delivery.record-opened",
+    "mesh-delivery.record-ended",
+    "mesh-catalog.filling",
+    "mesh-catalog.full",
+    "mesh-catalog.room",
+    "mesh-wants.filling",
+    "mesh-wants.full",
+    "mesh-wants.room",
+    "mesh-grants.filling",
+    "mesh-grants.full",
+    "mesh-grants.room",
     "mesh-pipeline-runner.pipeline-started",
     "mesh-pipeline-runner.step-started",
     "mesh-pipeline-runner.step-ended",