From 5fcd76a9f936f867eda88f92bdd7e81aec515671 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:56:50 +0100 Subject: [PATCH] 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 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §148, §160, §181, §218, §263 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-watchman/consumes.go | 22 +++ .../internal/watch/pipelines_test.go | 66 ++++++++ .../internal/watch/records.go | 149 ++++++++++++++++++ .../internal/watch/signals.go | 10 ++ .../internal/watch/watch.go | 4 +- .../mesh-delivery-watchman/manifest_test.go | 2 +- modules/mesh-delivery-watchman/module.json | 14 ++ 7 files changed, 265 insertions(+), 2 deletions(-) create mode 100644 modules/mesh-delivery-watchman/internal/watch/records.go diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go index f2ccce58..a0c405d4 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/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 +} diff --git a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go index c9cb3b38..b384cafd 100644 --- a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go +++ b/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") + } +} diff --git a/modules/mesh-delivery-watchman/internal/watch/records.go b/modules/mesh-delivery-watchman/internal/watch/records.go new file mode 100644 index 00000000..f51e2bfa --- /dev/null +++ b/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 + } +} diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index 2d9e7913..4829dc6f 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/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}, } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 9a05ef6e..88fb281e 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/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. diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 826e749b..4926b296 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/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). diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index 35bf7289..08b23559 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/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",