From 69aeb817e229c61e78a481c3ba99db9b2901ae3c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:49:32 +0100 Subject: [PATCH] WM2: the watchman raises build-capacity-waits for a build seat whose oldest queued step has waited 60 minutes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §86 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-watchman/consumes.go | 3 + .../internal/watch/pipelines.go | 78 +++++++++++++++++++ .../internal/watch/pipelines_test.go | 37 +++++++++ .../internal/watch/watch.go | 4 +- .../mesh-delivery-watchman/manifest_test.go | 2 +- modules/mesh-delivery-watchman/module.json | 2 + 6 files changed, 124 insertions(+), 2 deletions(-) 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 fb66bc3d..f2ccce58 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -152,6 +152,9 @@ func pipelineSubject(w *watch.Watchman, subject string, body []byte) bool { w.SourceMoved(body) case strings.HasPrefix(subject, "mesh.seat."+meshpipelinerunner.Seat+".event."+meshpipelinerunner.EventPipelineStarted+"."): w.PipelineStarted(body) + case strings.HasPrefix(subject, "mesh.seat."+meshpipelinerunner.Seat+".event."+meshpipelinerunner.EventStepStarted+"."), + strings.HasPrefix(subject, "mesh.seat."+meshpipelinerunner.Seat+".event."+meshpipelinerunner.EventStepEnded+"."): + w.StepEvent(body) default: return false } diff --git a/modules/mesh-delivery-watchman/internal/watch/pipelines.go b/modules/mesh-delivery-watchman/internal/watch/pipelines.go index 1a04cf0a..ed9e62f2 100644 --- a/modules/mesh-delivery-watchman/internal/watch/pipelines.go +++ b/modules/mesh-delivery-watchman/internal/watch/pipelines.go @@ -89,3 +89,81 @@ func (w *Watchman) PipelineStarted(body []byte) { // headID is the finding's id of a branch's head. func headID(branch string) string { return "head-" + unsafeKey.ReplaceAllString(branch, "_") } + +// KindBuildCapacityWaits is a build seat whose oldest queued step has waited the bound (design 47 G §86). +const KindBuildCapacityWaits = "build-capacity-waits" + +// CapacityBound is how long a step may wait queued on a build seat's stream before the seat is a finding (design 47 G §86). +const CapacityBound = 60 * time.Minute + +// capacityRow is the signals table's row of a build seat's queue. +const capacityRow = "build-capacity" + +// StepEvent hears the runner's `step-started` and `step-ended` (design 47 G §86): a step said waiting for a build node is +// queued on its seat, owed no progress; it leaves the queue when it starts running or ends. A seat's finding is due when +// its oldest queued step has waited CapacityBound, and ends when that seat has none that old. +func (w *Watchman) StepEvent(body []byte) { + var e meshpipelinerunner.StepEvent + if err := json.Unmarshal(body, &e); err != nil || e.Pipeline == "" || e.Result.Step == "" { + w.say("a step event does not read, left: %v", err) + return + } + id := e.Pipeline + "/" + e.Result.Step + w.mu.Lock() + defer w.mu.Unlock() + for seat, q := range w.queued { + if _, ok := q[id]; ok && (e.Result.Outcome != meshpipelinerunner.StepResultOutcomeWaitingForABuildNode || e.Result.Seat != seat) { + delete(q, id) + w.capacity(seat) + } + } + if e.Result.Outcome != meshpipelinerunner.StepResultOutcomeWaitingForABuildNode || e.Result.Seat == "" { + return + } + since := w.now() + switch { + case e.Result.Started != nil: + since = *e.Result.Started + case e.At != nil: + since = *e.At + } + seat := e.Result.Seat + if w.queued[seat] == nil { + w.queued[seat] = map[string]time.Time{} + } + if _, ok := w.queued[seat][id]; !ok { + w.queued[seat][id] = since + } + w.capacity(seat) +} + +// capacity judges a build seat's queue again, w.mu held: one signal for its oldest queued step. +func (w *Watchman) capacity(seat string) { + w.unsignal(capacityRow, seat) + key := FindingKey(ScopeDelivery, seat, KindBuildCapacityWaits) + var oldest time.Time + for _, t := range w.queued[seat] { + if oldest.IsZero() || t.Before(oldest) { + oldest = t + } + } + if oldest.IsZero() { + delete(w.queued, seat) + w.unfind(key) + return + } + f := Finding{Scope: ScopeDelivery, ID: seat, Kind: KindBuildCapacityWaits, Severity: Warning, Source: capacityRow, Since: oldest, + Summary: fmt.Sprintf("the oldest step queued on %s has waited since %s", seat, stamp(oldest)), + Said: "queued for " + CapacityBound.String(), + Headline: "Builds wait for a build node", + Explanation: "The oldest step queued on the build seat " + seat + " has waited " + CapacityBound.String() + + " for a build node to take it.", + Needs: "build capacity: a build node present and not paused, or fewer steps asked"} + left := CapacityBound - w.now().Sub(oldest) + if left > 0 { + w.unfind(key) + w.signal(capacityRow, seat, left, signalDue{Do: dueFind, Finding: f}) + return + } + w.find(f) +} diff --git a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go index 82fa5545..a5aa1afa 100644 --- a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go @@ -74,3 +74,40 @@ func TestAHeadWithNoPipelineStands(t *testing.T) { }) } } + +func queuedStep(pipeline, step, seat, outcome string, started time.Time) []byte { + b, _ := json.Marshal(meshpipelinerunner.StepEvent{Pipeline: pipeline, Branch: "main", Module: "m", Commit: "c", + Result: meshpipelinerunner.StepResult{Step: step, Seat: seat, Outcome: outcome, Started: &started}}) + return b +} + +// design 47 G §86: a build seat whose oldest queued step has waited 60 minutes raises one `build-capacity-waits`, cleared +// when that step starts. +func TestABuildSeatsOldestQueuedStepWaitsAnHour(t *testing.T) { + const seat = "node-build-go" + key := FindingKey(ScopeDelivery, seat, KindBuildCapacityWaits) + wait := meshpipelinerunner.StepResultOutcomeWaitingForABuildNode + w, o, c := setup() + w.StepEvent(queuedStep("p1", "unit", seat, wait, c.t)) + c.t = c.t.Add(10 * time.Minute) + w.StepEvent(queuedStep("p2", "unit", seat, wait, c.t)) + pk := signalKey(capacityRow, seat) + if o.ttls[pk] != 50*time.Minute || len(o.findings) != 0 { + t.Fatalf("within the bound: %v %v", o.ttls, o.findings) + } + expire(w, c, pk, 50*time.Minute) + if _, ok := o.findings[key]; !ok || len(o.findings) != 1 { + t.Fatalf("an hour for the oldest: %v", o.findings) + } + w.StepEvent(queuedStep("p1", "unit", seat, meshpipelinerunner.StepResultOutcomeRunning, c.t)) + if len(o.findings) != 0 { + t.Fatalf("the oldest started, the next has waited 50 minutes: %v", o.findings) + } + if o.ttls[pk] != 10*time.Minute { + t.Fatalf("the next one's hour: %v", o.ttls[pk]) + } + w.StepEvent(queuedStep("p2", "unit", seat, meshpipelinerunner.StepResultOutcomePassed, c.t)) + if _, ok := o.pulses[pk]; ok || len(o.findings) != 0 { + t.Fatalf("the queue is empty: %v %v", o.pulses, o.findings) + } +} diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 45fdf1f2..41083c1d 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -114,6 +114,8 @@ type Watchman struct { awaited map[string]bool // heads is, by watched branch, the head still waiting for its pipeline (pipelines.go). heads map[string]head + // 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 } // New is a watchman writing to out, reading the time from now. @@ -124,7 +126,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{}} + 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{}} } // 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 e312196c..826e749b 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-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-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 9629fa8f..35bf7289 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -38,6 +38,8 @@ "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" ],