From 161d6e645118ea007ebce412cd48e9a05c3d1368 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:56:33 +0100 Subject: [PATCH] WM16: a record is covered only by what is late for the entries of its own revision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit record-opened is joined to the take-in.taken of its milestone, ., whose entries name the modules; the record-open finding is left unwritten only while one of those modules is held by an acting-part-late wait (or a store says full), no longer while any wait is late. A record whose revision is not known taken is covered by a standing take-in-waits. Builds: design 47 G §502, A-200 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../internal/watch/pipelines_test.go | 36 +++++++++++++++++++ .../internal/watch/records.go | 25 ++++++++++--- .../internal/watch/signals.go | 3 -- .../internal/watch/takein.go | 10 ++++++ .../internal/watch/waits.go | 2 ++ .../internal/watch/watch.go | 6 +++- 6 files changed, 73 insertions(+), 9 deletions(-) diff --git a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go index f44ff602..230a4739 100644 --- a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go @@ -7,6 +7,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" ) @@ -247,3 +248,38 @@ func TestARecordOpenADayIsAFinding(t *testing.T) { t.Fatal("the pulse of an ended record stands") } } + +// design 47 G §502, A-200 (WM16): a record is tied to the entries of the take-in.taken of its milestone `.`; it +// is covered only while one of those entries is held by an `acting-part-late` wait, so another record's late entry covers nothing. +func TestARecordIsCoveredOnlyByTheWaitsOfItsOwnEntries(t *testing.T) { + w, o, c := setup() + taken := func(key string, source int64, module string) { + var m map[string]any + if err := json.Unmarshal(takeInBody(t, takein.TakeInTaken, key, c.t), &m); err != nil { + t.Fatal(err) + } + m["branch"], m["source"] = "main", source + m["entries"] = []map[string]any{{"commit": "c", "fingerprint": "f", "module": module}} + b, _ := json.Marshal(m) + w.TakeIn(takeInEvent(t, takein.TakeInTaken, key), b) + } + taken("m.1", 7, "web") + taken("m.2", 8, "db") + rec := func(id, milestone string) []byte { + b, _ := json.Marshal(meshdelivery.RecordOpenedEvent{Record: id, Milestone: milestone, At: &c.t}) + return b + } + w.RecordOpen(rec("rWeb", "main.7")) + w.RecordOpen(rec("rDb", "main.8")) + late, _ := json.Marshal(meshwants.Wait{Step: "place", Part: "mesh-delivery-indexer", Node: "ace", Module: "web", Since: c.t, Bound: "30m"}) + w.WaitHeld("mesh-wants", late) + expire(w, c, signalKey(actingRow, "mesh-wants-place-mesh-delivery-indexer-ace-web"), 30*time.Minute) + expire(w, c, signalKey(recordRow, "rWeb"), RecordOpenBound) + expire(w, c, signalKey(recordRow, "rDb"), 0) + if _, ok := o.findings[FindingKey(ScopeDelivery, "rWeb", KindRecordOpen)]; ok { + t.Fatalf("a record whose entry is held by acting-part-late is covered: %v", o.findings) + } + if _, ok := o.findings[FindingKey(ScopeDelivery, "rDb", KindRecordOpen)]; !ok { + t.Fatalf("a record of other entries is open at 24 hours whatever else is late: %v", o.findings) + } +} diff --git a/modules/mesh-delivery-watchman/internal/watch/records.go b/modules/mesh-delivery-watchman/internal/watch/records.go index c1ce908d..d08a0e4e 100644 --- a/modules/mesh-delivery-watchman/internal/watch/records.go +++ b/modules/mesh-delivery-watchman/internal/watch/records.go @@ -3,6 +3,7 @@ package watch import ( "encoding/json" "fmt" + "strings" "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" @@ -93,6 +94,7 @@ func (w *Watchman) RecordOpen(body []byte) { if r.At != nil && !r.At.IsZero() { since = *r.At } + w.recordMilestone[r.Record] = r.Milestone 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", @@ -120,17 +122,31 @@ func (w *Watchman) RecordEnded(body []byte) { defer w.mu.Unlock() w.unsignal(recordRow, r.Record) delete(w.opened, r.Record) + delete(w.recordMilestone, 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 { +// recordCovered says another condition tells the operator why a record is open (design 47 G §502, A-200): the record is tied to the +// entries of its source revision's `take-in.taken` by its milestone, `.`, and is covered while one of those +// entries is held by a wait that raised `acting-part-late`, or a store says `full`; a record whose revision is not known to be taken in +// is covered by a standing `take-in-waits`. +func (w *Watchman) recordCovered(record string) bool { + entries, taken := w.takenEntries[w.recordMilestone[record]] for _, v := range w.findings { f, err := DecodeFinding(v) if err != nil { continue } - if f.Kind == KindActingPartLate || (f.Kind == KindStoreFilling && f.Severity == Urgent) { + switch { + case f.Kind == KindStoreFilling && f.Severity == Urgent: + return true + case f.Kind == KindActingPartLate && taken: + for _, module := range entries { + if strings.HasSuffix(f.ID, "-"+module) { + return true + } + } + case f.Kind == KindTakeInWaits && !taken: return true } } @@ -140,11 +156,10 @@ func (w *Watchman) recordCovered() bool { // 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() { + if w.recordCovered(id) { 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 6db0439f..4485ffc1 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -186,9 +186,6 @@ 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 diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go index 4482b385..9120b322 100644 --- a/modules/mesh-delivery-watchman/internal/watch/takein.go +++ b/modules/mesh-delivery-watchman/internal/watch/takein.go @@ -47,11 +47,21 @@ func (w *Watchman) TakeIn(event string, body []byte) { // The wait names the watched branch it is of (design 47 G §493), by the contract's body. var named meshcatalog.TakeInPending _ = json.Unmarshal(body, &named) + var taken meshcatalog.TakeInTaken + _ = json.Unmarshal(body, &taken) key := t.Key w.mu.Lock() defer w.mu.Unlock() w.takeIn(kind, key, t) w.pendingOf(kind, key, named.Branch) + if kind == takein.TakeInTaken && taken.Branch != "" && taken.Source != 0 && len(taken.Entries) > 0 { + var modules []string + for _, e := range taken.Entries { + modules = append(modules, e.Module) + } + w.takenEntries[fmt.Sprintf("%s.%d", taken.Branch, taken.Source)] = modules // a record's milestone (A-200) + w.rejudgeRecords() + } } // SeedTakeIn starts, at the watchman's start and before its consumer on the take-in stream, the wait of every revision diff --git a/modules/mesh-delivery-watchman/internal/watch/waits.go b/modules/mesh-delivery-watchman/internal/watch/waits.go index 235304c9..034bc357 100644 --- a/modules/mesh-delivery-watchman/internal/watch/waits.go +++ b/modules/mesh-delivery-watchman/internal/watch/waits.go @@ -46,6 +46,7 @@ func (w *Watchman) WaitEnded(seat string, body []byte) { w.unsignal(actingRow, id) w.unfind(FindingKey(ScopeDelivery, id, KindActingPartLate)) w.unfind(FindingKey(ScopeDelivery, id, KindPutStands)) + w.rejudgeRecords() } func decodeWait(w *Watchman, seat, event string, body []byte) (meshwants.Wait, bool) { @@ -89,6 +90,7 @@ func (w *Watchman) waitHeld(seat string, a meshwants.Wait) { return } w.find(f) + w.rejudgeRecords() // a record whose entry this wait holds is covered now (§502) } // putStands is a put's wait held (`step: put`, design 47 G §498): a put without hold that stands past its bound is said as the diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index bfe049f7..8bcf9ea9 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -120,6 +120,10 @@ type Watchman struct { heads map[string]head // pendingBranch is, by key, the watched branch of each open take-in wait that names one (takein.go, design 47 G §497). pendingBranch map[string]string + // takenEntries are the modules of each taken-in revision, by `.`, and recordMilestone each open record's + // milestone, which joins a record to them (records.go, design 47 G §502). + takenEntries map[string][]string + recordMilestone map[string]string // queued is, by build seat, each step queued on it for a build node and since when (pipelines.go). 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). @@ -132,7 +136,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]meshwatchdogsclient.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: meshwatchdogsclient.PulseMoveBound, stepBound: StepBound, progressBound: ProgressBound, - findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, busCreated: map[string]bool{}, takeInHeld: map[string]Finding{}, pendingBranch: map[string]string{}, away: map[string]bool{}, + findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, busCreated: map[string]bool{}, takeInHeld: map[string]Finding{}, pendingBranch: map[string]string{}, takenEntries: map[string][]string{}, recordMilestone: map[string]string{}, away: map[string]bool{}, 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{},