WM1 (step half): a head behind a pending revision of its branch raises no pipeline-stands

The watchman keeps the watched branch each open take-in wait names (take-in.pending's branch) and times a head only
while its branch has none; once the revision is taken in the head is timed again from its own move. The head on a
branch whose branch-unprotected stands (§497's second sentence) is not built: nothing the watchman hears says it
(the forge bridge writes that finding, no event).

Builds: design 47 G §12, §75, §497

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 05:48:30 +01:00
parent 893f425c56
commit 464206d2de
4 changed files with 101 additions and 7 deletions
@@ -24,7 +24,10 @@ const HeadBound = 5 * time.Minute
const headRow = "head-pipeline"
// head is a watched branch's head still waiting for its pipeline.
type head struct{ revision int64 }
type head struct {
revision int64
at time.Time
}
// SourceMoved hears the seat mesh-delivery's `source-moved`: the branch's head is a new source revision, and its pipeline
// is due within HeadBound. A move made by a gathered group is a member governed by the take-in's own bound (ADR 0633 §5),
@@ -48,15 +51,26 @@ func (w *Watchman) SourceMoved(body []byte) {
if m.At != nil && !m.At.IsZero() {
at = *m.At
}
w.heads[m.Branch] = head{revision: m.Revision}
f := Finding{Scope: ScopeDelivery, ID: id, Kind: KindPipelineStands, Severity: Warning, Source: headRow, Since: at,
Summary: fmt.Sprintf("the head of %s (source revision %d) has started no pipeline since %s", m.Branch, m.Revision, stamp(at)),
w.heads[m.Branch] = head{revision: m.Revision, at: at}
w.timeHead(m.Branch)
}
// timeHead times the head of a branch that waits for its pipeline, w.mu held: five minutes from its own time, unless the
// branch has a pending revision, whose take-in wait governs it (design 47 G §497) and which times it again once taken in.
func (w *Watchman) timeHead(branch string) {
h, ok := w.heads[branch]
if !ok || w.branchPending(branch) {
return
}
id := headID(branch)
f := Finding{Scope: ScopeDelivery, ID: id, Kind: KindPipelineStands, Severity: Warning, Source: headRow, Since: h.at,
Summary: fmt.Sprintf("the head of %s (source revision %d) has started no pipeline since %s", branch, h.revision, stamp(h.at)),
Said: fmt.Sprintf("no pipeline in %s", HeadBound),
Headline: "A change has started no pipeline",
Explanation: "The head of the watched branch " + m.Branch + " (source revision " + fmt.Sprint(m.Revision) + ") moved " +
Explanation: "The head of the watched branch " + branch + " (source revision " + fmt.Sprint(h.revision) + ") moved " +
HeadBound.String() + " ago or more and the pipeline runner has said no pipeline started for it.",
Needs: "the runner to start the pipeline (its pipeline_stands and list tools say where it is)"}
left := HeadBound - w.now().Sub(at)
left := HeadBound - w.now().Sub(h.at)
if left > 0 {
w.signal(headRow, id, left, signalDue{Do: dueFind, Finding: f})
return
@@ -64,6 +78,16 @@ func (w *Watchman) SourceMoved(body []byte) {
w.find(f)
}
// branchPending is whether an open take-in wait names the branch (design 47 G §497): its head is behind a pending revision.
func (w *Watchman) branchPending(branch string) bool {
for _, b := range w.pendingBranch {
if b == branch {
return true
}
}
return false
}
// PipelineStarted hears the runner's `pipeline-started`: it ends the wait of the head it is for (the branch's head at that
// source revision or a later one). A run on request is no head's pipeline.
func (w *Watchman) PipelineStarted(body []byte) {
@@ -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-indexer/takein"
)
func moved(branch string, rev int64, group string, at time.Time) []byte {
@@ -14,6 +15,21 @@ func moved(branch string, rev int64, group string, at time.Time) []byte {
return b
}
// takeInBodyOn is a take-in event's body that names its watched branch (design 47 G §493).
func takeInBodyOn(t *testing.T, kind, key, branch string, since time.Time) []byte {
t.Helper()
var m map[string]any
if err := json.Unmarshal(takeInBody(t, kind, key, since), &m); err != nil {
t.Fatal(err)
}
m["branch"] = branch
b, err := json.Marshal(m)
if err != nil {
t.Fatal(err)
}
return b
}
func started(branch string, rev int64, run string) []byte {
b, _ := json.Marshal(meshpipelinerunner.PipelineStartedEvent{Branch: branch, Source: rev, Run: run, Module: "m", Pipeline: "p"})
return b
@@ -58,6 +74,30 @@ func TestAHeadWithNoPipelineStands(t *testing.T) {
t.Fatalf("a pulse for a member: %v", o.pulses)
}
}, false},
{"a head behind a pending revision of its branch is not timed", func(w *Watchman, o *fakeOut, c *clock) {
w.TakeIn(takeInEvent(t, takein.TakeInPending, "m.1"), takeInBodyOn(t, takein.TakeInPending, "m.1", "main", c.t))
w.SourceMoved(moved("main", 7, "", c.t))
if _, ok := o.pulses[signalKey(headRow, headID("main"))]; ok {
t.Fatalf("a pulse for a head behind a pending revision: %v", o.pulses)
}
}, false},
{"a pending revision of another branch does not hold it", func(w *Watchman, o *fakeOut, c *clock) {
w.TakeIn(takeInEvent(t, takein.TakeInPending, "m.1"), takeInBodyOn(t, takein.TakeInPending, "m.1", "other", c.t))
w.SourceMoved(moved("main", 7, "", c.t))
expire(w, c, signalKey(headRow, headID("main")), HeadBound)
}, true},
{"a head whose pending revision is taken in is timed again", func(w *Watchman, o *fakeOut, c *clock) {
w.SourceMoved(moved("main", 7, "", c.t))
w.TakeIn(takeInEvent(t, takein.TakeInPending, "m.1"), takeInBodyOn(t, takein.TakeInPending, "m.1", "main", c.t))
if _, ok := o.pulses[signalKey(headRow, headID("main"))]; ok {
t.Fatalf("the head is timed behind a pending revision: %v", o.pulses)
}
c.t = c.t.Add(HeadBound)
w.TakeIn(takeInEvent(t, takein.TakeInTaken, "m.1"), takeInBodyOn(t, takein.TakeInTaken, "m.1", "main", c.t))
if _, ok := o.findings[key]; !ok {
t.Fatalf("the head, past its five minutes when taken in, stands: %v", o.findings)
}
}, true},
{"a run on request is no head's pipeline", func(w *Watchman, o *fakeOut, c *clock) {
w.SourceMoved(moved("main", 7, "", c.t))
w.PipelineStarted(started("main", 7, "run-1"))
@@ -1,12 +1,14 @@
package watch
import (
"encoding/json"
"fmt"
"maps"
"slices"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
)
@@ -42,10 +44,14 @@ func (w *Watchman) TakeIn(event string, body []byte) {
w.say("the body of %s says the kind %q, left", event, t.Kind)
return
}
// 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)
key := t.Key
w.mu.Lock()
defer w.mu.Unlock()
w.takeIn(kind, key, t)
w.pendingOf(kind, key, named.Branch)
}
// SeedTakeIn starts, at the watchman's start and before its consumer on the take-in stream, the wait of every revision
@@ -88,6 +94,28 @@ func (w *Watchman) SeedTakeIn(a takein.PendingAnswer) {
}
}
// pendingOf records or ends the branch an open take-in wait names (design 47 G §497), w.mu held: the head of a branch behind a
// pending revision raises no `pipeline-stands`, and is timed again, from the move of the head, once the branch has none.
func (w *Watchman) pendingOf(kind, key, branch string) {
if kind == takein.TakeInTaken {
branch = w.pendingBranch[key]
delete(w.pendingBranch, key)
if branch != "" && !w.branchPending(branch) {
w.timeHead(branch)
}
return
}
if branch == "" {
return
}
w.pendingBranch[key] = branch
if _, ok := w.heads[branch]; ok {
id := headID(branch)
w.unsignal(headRow, id)
w.unfind(FindingKey(ScopeDelivery, id, KindPipelineStands))
}
}
// takeIn is one take-in event of key, w.mu held.
func (w *Watchman) takeIn(kind, key string, t takein.TakeIn) {
finding := FindingKey(ScopeDelivery, key, KindTakeInWaits)
@@ -118,6 +118,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
// 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
// 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).
@@ -130,7 +132,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{}, 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{}, 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{},