novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

WM2: the watchman raises build-capacity-waits for a build seat whose oldest queued step has waited 60 minutes

Browse Source 
Builds: design 47 G §86

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 11:49:32 +01:00
parent 0f618f1998
commit 69aeb817e2
6 changed files with 124 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go
+3
View File
@@ -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
}
modules/mesh-delivery-watchman/internal/watch/pipelines.go
+78
View File
@@ -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)
}
modules/mesh-delivery-watchman/internal/watch/pipelines_test.go
+37
View File
@@ -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)
}
}
modules/mesh-delivery-watchman/internal/watch/watch.go
+3 -1
View File
@@ -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.
modules/mesh-delivery-watchman/manifest_test.go
+1 -1
View File
@@ -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).
modules/mesh-delivery-watchman/module.json
+2
View File
@@ -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"
],