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

WM1 (head half): the watchman raises pipeline-stands for a watched branch's head with no pipeline within five minutes

Browse Source 
Builds: design 47 G §12, §75, §112 (the head half; the step half waits on a design question)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 11:48:38 +01:00
parent bcd0888928
commit 0f618f1998
7 changed files with 192 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
+17
View File
@@ -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
}
modules/mesh-delivery-watchman/go.mod
+2
View File
@@ -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
modules/mesh-delivery-watchman/internal/watch/pipelines.go
+91
View File
@@ -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, "_") }
modules/mesh-delivery-watchman/internal/watch/pipelines_test.go
+76
View File
@@ -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)
}
})
}
}
modules/mesh-delivery-watchman/internal/watch/watch.go
+3 -1
View File
@@ -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.
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-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).
modules/mesh-delivery-watchman/module.json
+2
View File
@@ -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"
],