WM16: a record is covered only by what is late for the entries of its own revision

record-opened is joined to the take-in.taken of its milestone, <branch>.<source revision>, 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
This commit is contained in:
2026-11-12 05:56:33 +01:00
parent 1abc5ae6d9
commit 161d6e6451
6 changed files with 73 additions and 9 deletions
@@ -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 `<branch>.<source revision>`; 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)
}
}
@@ -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, `<branch>.<source revision>`, 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
}
}
@@ -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
@@ -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
@@ -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
@@ -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 `<branch>.<source revision>`, 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{},