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 a154a11e..fb66bc3d 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -5,6 +5,8 @@ import ( "encoding/json" "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" "strings" "sync" @@ -41,6 +43,7 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string, return } w.Accepted(a) + case pipelineSubject(w, subject, body): case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventCreated): if s, err := watch.ReadBrokerSaid(body); err != nil { logf("%v; left", err) @@ -140,3 +143,17 @@ func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Acco func brokerSubject(subject string) bool { return strings.HasPrefix(subject, "mesh.seat."+meshbroker.Seat+".event.") } + +// pipelineSubject hands the gate's `source-moved` and the runner's `pipeline-started` to the watchman (watch/pipelines.go, +// design 47 G §12): the head of a watched branch and its pipeline. It says whether the subject was one of them. +func pipelineSubject(w *watch.Watchman, subject string, body []byte) bool { + switch { + case strings.HasPrefix(subject, "mesh.seat."+meshdelivery.Seat+".event."+meshdelivery.EventSourceMoved+"."): + w.SourceMoved(body) + case strings.HasPrefix(subject, "mesh.seat."+meshpipelinerunner.Seat+".event."+meshpipelinerunner.EventPipelineStarted+"."): + w.PipelineStarted(body) + default: + return false + } + return true +} diff --git a/modules/mesh-delivery-watchman/go.mod b/modules/mesh-delivery-watchman/go.mod index 6a4e9198..55006f65 100644 --- a/modules/mesh-delivery-watchman/go.mod +++ b/modules/mesh-delivery-watchman/go.mod @@ -14,6 +14,8 @@ require ( git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-watchman/internal/watch/pipelines.go b/modules/mesh-delivery-watchman/internal/watch/pipelines.go new file mode 100644 index 00000000..1a04cf0a --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/pipelines.go @@ -0,0 +1,91 @@ +package watch + +import ( + "encoding/json" + "fmt" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" +) + +// What the runner and the gate say of a watched branch's head, heard as events (design 47 G §12, §75, §112): a head the +// gate says moved (`source-moved`) that starts no pipeline within five minutes is the finding `pipeline-stands`, naming the +// branch and its head, raised by the condition keeper and cleared by the runner's `pipeline-started` for that head. + +// KindPipelineStands is a watched branch's head, or a step, standing still (design 47 G §12, §75). +const KindPipelineStands = "pipeline-stands" + +// HeadBound is how long a watched branch's head may stand with no pipeline started (design 47 G §12, §112: the record's five +// minutes). +const HeadBound = 5 * time.Minute + +// headRow is the signals table's row of a head with no pipeline. +const headRow = "head-pipeline" + +// head is a watched branch's head still waiting for its pipeline. +type head struct{ revision int64 } + +// 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), +// so it is not timed here (design 47 G §12). A later move of the same branch replaces the head that waits. +func (w *Watchman) SourceMoved(body []byte) { + var m meshdelivery.SourceMovedEvent + if err := json.Unmarshal(body, &m); err != nil || m.Branch == "" || m.Revision == 0 { + w.say("a source-moved does not read, left: %v", err) + return + } + w.mu.Lock() + defer w.mu.Unlock() + id := headID(m.Branch) + w.unsignal(headRow, id) + w.unfind(FindingKey(ScopeDelivery, id, KindPipelineStands)) + if m.Group != "" { + delete(w.heads, m.Branch) + return + } + at := w.now() + 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)), + 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 " + + 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) + if left > 0 { + w.signal(headRow, id, left, signalDue{Do: dueFind, Finding: f}) + return + } + w.find(f) +} + +// 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) { + var p meshpipelinerunner.PipelineStartedEvent + if err := json.Unmarshal(body, &p); err != nil || p.Branch == "" { + w.say("a pipeline-started does not read, left: %v", err) + return + } + if p.Run != "" { + return + } + w.mu.Lock() + defer w.mu.Unlock() + h, ok := w.heads[p.Branch] + if !ok || p.Source < h.revision { + return + } + delete(w.heads, p.Branch) + id := headID(p.Branch) + w.unsignal(headRow, id) + w.unfind(FindingKey(ScopeDelivery, id, KindPipelineStands)) +} + +// headID is the finding's id of a branch's head. +func headID(branch string) string { return "head-" + unsafeKey.ReplaceAllString(branch, "_") } diff --git a/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go new file mode 100644 index 00000000..82fa5545 --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/pipelines_test.go @@ -0,0 +1,76 @@ +package watch + +import ( + "encoding/json" + "testing" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" +) + +func moved(branch string, rev int64, group string, at time.Time) []byte { + b, _ := json.Marshal(meshdelivery.SourceMovedEvent{Branch: branch, Revision: rev, Group: group, At: &at}) + 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 +} + +// design 47 G §12, §79 (test 76): a head with no pipeline within five minutes is `pipeline-stands`, cleared by the +// runner's `pipeline-started` for that head; a gathered group's member and a run on request do not. +func TestAHeadWithNoPipelineStands(t *testing.T) { + key := FindingKey(ScopeDelivery, headID("main"), KindPipelineStands) + cases := []struct { + name string + run func(w *Watchman, o *fakeOut, c *clock) + stand bool + }{ + {"five minutes and no pipeline", func(w *Watchman, o *fakeOut, c *clock) { + w.SourceMoved(moved("main", 7, "", c.t)) + if o.ttls[signalKey(headRow, headID("main"))] != HeadBound || len(o.findings) != 0 { + t.Fatalf("within the bound: %v %v", o.ttls, o.findings) + } + expire(w, c, signalKey(headRow, headID("main")), HeadBound) + }, true}, + {"its pipeline-started clears it", func(w *Watchman, o *fakeOut, c *clock) { + w.SourceMoved(moved("main", 7, "", c.t)) + expire(w, c, signalKey(headRow, headID("main")), HeadBound) + w.PipelineStarted(started("main", 7, "")) + }, false}, + {"a pipeline before the bound raises none", func(w *Watchman, o *fakeOut, c *clock) { + w.SourceMoved(moved("main", 7, "", c.t)) + w.PipelineStarted(started("main", 8, "")) + if _, ok := o.pulses[signalKey(headRow, headID("main"))]; ok { + t.Fatal("the pulse stands") + } + }, false}, + {"a pipeline of an older head does not clear a newer head", func(w *Watchman, o *fakeOut, c *clock) { + w.SourceMoved(moved("main", 7, "", c.t)) + w.PipelineStarted(started("main", 6, "")) + expire(w, c, signalKey(headRow, headID("main")), HeadBound) + }, true}, + {"a gathered group's member is not timed", func(w *Watchman, o *fakeOut, c *clock) { + w.SourceMoved(moved("main", 7, "grp", c.t)) + if len(o.pulses) != 0 { + t.Fatalf("a pulse for a member: %v", o.pulses) + } + }, false}, + {"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")) + expire(w, c, signalKey(headRow, headID("main")), HeadBound) + }, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + w, o, c := setup() + tc.run(w, o, c) + if _, ok := o.findings[key]; ok != tc.stand { + t.Fatalf("the finding stands %v, want %v: %v", ok, tc.stand, o.findings) + } + }) + } +} diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index e73bc987..45fdf1f2 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -112,6 +112,8 @@ type Watchman struct { // pulses, stuck keys and findings wait for its own end (ReadLate), so a restart while its owner is down neither drops // nor raises again what it holds. awaited map[string]bool + // heads is, by watched branch, the head still waiting for its pipeline (pipelines.go). + heads map[string]head } // New is a watchman writing to out, reading the time from now. @@ -122,7 +124,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{}} + standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}, heads: map[string]head{}} } // 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 79799276..e312196c 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-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-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 25acacd9..9629fa8f 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -36,6 +36,8 @@ "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" ],