diff --git a/modules/mesh-delivery-watchman/internal/watch/pipelines.go b/modules/mesh-delivery-watchman/internal/watch/pipelines.go index bdec3fd6..09324691 100644 --- a/modules/mesh-delivery-watchman/internal/watch/pipelines.go +++ b/modules/mesh-delivery-watchman/internal/watch/pipelines.go @@ -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) { diff --git a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go index aa534388..f44ff602 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-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")) diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go index cc6b6c2e..4482b385 100644 --- a/modules/mesh-delivery-watchman/internal/watch/takein.go +++ b/modules/mesh-delivery-watchman/internal/watch/takein.go @@ -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) diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 78d77652..bfe049f7 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -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{},