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

Watchman: hear the indexer's take-in stream and raise take-in-waits past 60 minutes (revision) or 30 (group) (issue 1585, ADR 0633 §5)

Browse Source 
Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
jschoubben
2026-11-04 22:54:21 +01:00
parent 2cb3198049
commit 0579e4e74d
7 changed files with 185 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go
+29
View File
@@ -22,6 +22,7 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/provisions"
@@ -375,6 +376,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
dead.ages()
dead.check()
provisioned(ctx, js, w, logger.Printf)
// The indexer's stream take-in (ADR 0633 §5): a revision or a group it holds back is a wait, ended by take-in.taken. The
// durable consumer <node>_<module> is made by the controller from this module's consumes; the whole stream is heard at
// the start, so a wait that began while the watchman was away is read with its own since.
switch tc, err := cred.EventConsumer(ctx, js, catalogue.StreamBusName(delivery.TakeInStream)); {
case err != nil:
logger.Printf("the consumer %s_%s on %s does not exist yet, so no take-in wait is said until the watchman starts again: %v",
cred.Node, cred.Module, catalogue.StreamBusName(delivery.TakeInStream), err)
default:
hc, err := process.HearEvents(tc, func(m jetstream.Msg) { takeInHeard(w, m, logger.Printf) }, logger.Printf)
if err != nil {
return err
}
defer hc.Stop()
}
w.Reconcile()
if err := hear(conn, w, dead, logger.Printf); err != nil {
return err
@@ -439,3 +454,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
}
// takeInHeard hands one event of a take-in stream to the watchman, by its name relative to the stream, and acknowledges it;
// one that is not a take-in event is said and acknowledged, since hearing it again would not read it.
func takeInHeard(w *watch.Watchman, m jetstream.Msg, logf func(string, ...any)) {
_, event, ok := strings.Cut(m.Subject(), ".event.")
if !ok {
logf("the subject %s of the take-in stream is not an event's, left", m.Subject())
} else {
w.TakeIn(event, m.Data())
}
if err := m.Ack(); err != nil {
logf("a take-in event could not be acknowledged: %v", err)
}
}
modules/mesh-delivery-watchman/internal/watch/findings_test.go
+75
View File
@@ -34,6 +34,50 @@ func TestTheWatchmansFindings(t *testing.T) {
// want are the finding keys standing at the end, with their severity.
want map[string]string
}{
{"ADR 0633 §5 (1585/T21): a revision waiting for an indexer build is a finding past 60 minutes from its since, not within; a group past 30", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.TakeIn(takeInEvent(t, delivery.TakeInPending, revKey), takeInBody(t, delivery.TakeInPending, revKey, t0))
if o.ttls[signalKey("take-in", revKey)] != TakeInRevisionBound || len(o.findings) != 0 {
t.Fatalf("a revision's wait: ttls %v, findings %v", o.ttls, o.findings)
}
w.TakeIn(takeInEvent(t, delivery.TakeInPending, groupKey), takeInBody(t, delivery.TakeInPending, groupKey, t0))
if o.ttls[signalKey("take-in", groupKey)] != TakeInGroupBound {
t.Fatalf("a group's wait: ttls %v", o.ttls)
}
expire(w, c, signalKey("take-in", groupKey), TakeInGroupBound)
if len(o.findings) != 1 {
t.Fatalf("findings at 30 minutes: %v", o.findings)
}
expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound-TakeInGroupBound)
}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning, FindingKey(ScopeDelivery, groupKey, KindTakeInWaits): Warning}},
{"ADR 0633 §5 (1585/T21): the take-in.taken of a named indexer, a hand take-in or a drop clears the finding", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
for _, by := range []string{"", "operator@shanks"} {
w.TakeIn(takeInEvent(t, delivery.TakeInPending, revKey), takeInBody(t, delivery.TakeInPending, revKey, t0))
expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound)
if len(o.findings) != 1 {
t.Fatalf("findings past the bound: %v", o.findings)
}
taken, err := delivery.EncodeTakeIn(delivery.TakeIn{Kind: delivery.TakeInTaken, Key: revKey, What: delivery.TakeInRevision, Since: t0,
Merges: []delivery.GroupMerge{{Repository: "novox/a", Merge: "c0ffee"}}, By: by, Why: map[bool]string{true: "dropped", false: ""}[by != ""], Dropped: by != ""})
if err != nil {
t.Fatal(err)
}
w.TakeIn(takeInEvent(t, delivery.TakeInTaken, revKey), taken)
c.t = t0
}
}, map[string]string{}},
{"ADR 0633 §5 (1585/T21): a group's take-in.taken ends its 30 minutes, and the revision then pending under m.<seq> starts its own 60", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.TakeIn(takeInEvent(t, delivery.TakeInPending, groupKey), takeInBody(t, delivery.TakeInPending, groupKey, t0))
expire(w, c, signalKey("take-in", groupKey), 25*time.Minute)
w.TakeIn(takeInEvent(t, delivery.TakeInTaken, groupKey), takeInBody(t, delivery.TakeInTaken, groupKey, t0))
w.TakeIn(takeInEvent(t, delivery.TakeInPending, revKey), takeInBody(t, delivery.TakeInPending, revKey, c.t))
if len(o.findings) != 0 || o.ttls[signalKey("take-in", revKey)] != TakeInRevisionBound {
t.Fatalf("after the hand-over: findings %v, ttls %v", o.findings, o.ttls)
}
expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound)
}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}},
{"ADR 0633 §5 (1585/T21): a pending event heard again at the start, since past its bound, is a finding at once", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.TakeIn(takeInEvent(t, delivery.TakeInPending, revKey), takeInBody(t, delivery.TakeInPending, revKey, t0.Add(-2*time.Hour)))
}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}},
{"ADR 0578 §1: a module whose value carries a refusal of its manifest is a finding until it is accepted", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
value := func(verdict string, problems ...string) []byte {
b, err := delivery.EncodeModule(delivery.Module{FP2: "src2:" + strings.Repeat("a", 64), Manifest: strings.Repeat("b", 64),
@@ -528,3 +572,34 @@ func seedUnhealthy(t *testing.T, w *Watchman, o *fakeOut, sev string, since time
o.findings[f.Key()], _ = DecodeFinding(v)
w.SeedFinding(f.Key(), v)
}
const (
revKey = "m.00000000000000000042"
groupKey = "g.feat/take-in"
)
func takeInEvent(t *testing.T, kind, key string) string {
t.Helper()
e, err := delivery.TakeInEvent(kind, key)
if err != nil {
t.Fatal(err)
}
return e
}
// takeInBody is a take-in event's body for a key: a group's when it starts with g.
func takeInBody(t *testing.T, kind, key string, since time.Time) []byte {
t.Helper()
body := delivery.TakeIn{Kind: kind, Key: key, What: delivery.TakeInRevision, Since: since,
Merges: []delivery.GroupMerge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}}
if key[0] == 'g' {
body.What, body.Group = delivery.TakeInGroup, "feat/take-in"
} else if kind == delivery.TakeInPending {
body.WaitsFor = "src2:abc"
}
b, err := delivery.EncodeTakeIn(body)
if err != nil {
t.Fatal(err)
}
return b
}
modules/mesh-delivery-watchman/internal/watch/signals.go
+1
View File
@@ -919,6 +919,7 @@ var SignalRows = []struct{ Row, Signal, Bound, Kind string }{
{"S11", "a machine's tool runner heartbeat", "3 × its interval (" + (3 * HeartbeatEvery).String() + " when it says none)", KindToolsSilent},
{"S21", "a machine's have failed or its provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing},
{"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy},
{"take-in", "a source revision or a delivery group the indexer holds back (ADR 0633 §5)", TakeInRevisionBound.String() + " a revision, " + TakeInGroupBound.String() + " a group", KindTakeInWaits},
{"provisions", "a provision of mesh-provisioner in waits (ADR 0570 §7)", "at once", KindNeedsOperator},
}
modules/mesh-delivery-watchman/internal/watch/takein.go
+70
View File
@@ -0,0 +1,70 @@
package watch
import (
"fmt"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// KindTakeInWaits is a source revision or a delivery group the indexer holds back and has not taken in within its bound
// (novox/hq ADR 0633 §5, issue 1585): a revision for the indexer build it names, a group for its end. It stands until a
// `take-in.taken` ends the wait, whoever ended it (the named indexer, a revert, a hand take-in or a drop).
const KindTakeInWaits = "take-in-waits"
// The bounds of a take-in wait, from the `take-in.pending` event's own since.
const (
TakeInRevisionBound = 60 * time.Minute
TakeInGroupBound = 30 * time.Minute
)
// takeInRow is the signals table's row of a take-in wait.
const takeInRow = "take-in"
// TakeIn hears one event of the indexer's stream `take-in`, by its name relative to the stream (`take-in.<kind>.<key>`):
// a pending one starts the wait for its key, whose bound runs from the event's since, and a taken one ends it. A group's
// taken event ends the group's wait only: the revision it becomes is its own key and starts its own, longer wait.
func (w *Watchman) TakeIn(event string, body []byte) {
kind, key, err := delivery.ParseTakeInEvent(event)
if err != nil {
w.say("a take-in event is left: %v", err)
return
}
t, err := delivery.DecodeTakeIn(body)
if err != nil {
w.say("the body of %s cannot be read, left: %v", event, err)
return
}
w.mu.Lock()
defer w.mu.Unlock()
finding := FindingKey(ScopeDelivery, key, KindTakeInWaits)
if kind == delivery.TakeInTaken {
w.unsignal(takeInRow, key)
w.unfind(finding)
return
}
bound, what := TakeInRevisionBound, "a source revision waits for the indexer build "+t.WaitsFor
if t.What == delivery.TakeInGroup {
bound, what = TakeInGroupBound, "a delivery group is gathered and its end has not come"
}
var merges []string
for _, m := range t.Merges {
merges = append(merges, m.Repository)
}
f := Finding{Scope: ScopeDelivery, ID: key, Kind: KindTakeInWaits, Severity: Warning, Source: "take-in", Since: t.Since,
Summary: fmt.Sprintf("%s: %s, since %s (%s)", key, what, stamp(t.Since), strings.Join(merges, ", ")),
Said: fmt.Sprintf("held for %s", bound),
Headline: "A merged change is held back by the indexer",
Explanation: "The indexer holds " + what + " and has not taken it in within " + bound.String() + ", so what the merge changes does " +
"not reach the machines.",
Needs: "the indexer build it waits for to run, or the take-in by hand (indexer_take_in_pending), or a drop (indexer_drop_pending)"}
left := bound - w.now().Sub(t.Since)
if left <= 0 {
w.unsignal(takeInRow, key)
w.find(f)
return
}
w.unfind(finding)
w.signal(takeInRow, key, left, signalDue{Do: dueFind, Finding: f})
}
modules/mesh-delivery-watchman/internal/watch/watch.go
+1 -1
View File
@@ -366,7 +366,7 @@ func signalKind(value []byte) bool {
return false
}
switch f.Source {
case "provisioner.failing", "S8", "S9", "S10", "S11":
case "provisioner.failing", "S8", "S9", "S10", "S11", "take-in":
return true
}
return false
modules/mesh-delivery-watchman/manifest_test.go
+6 -1
View File
@@ -51,6 +51,7 @@ type manifest struct {
Version string `json:"version"`
Slug string `json:"slug"`
Reads []string `json:"reads"`
Consumes []string `json:"consumes"`
State []state `json:"state"`
OwnSecrets map[string]string `json:"own-secrets"`
SecretsOwner string `json:"secrets-owner"`
@@ -94,6 +95,10 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
"mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-indexer.modules", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) {
t.Errorf("reads = %v", reads)
}
// 1585/T20: it hears the indexer's stream take-in, whole, through a durable consumer the controller makes from consumes.
if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-indexer.take-in.**"}) {
t.Errorf("consumes = %v", m.Consumes)
}
// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).
if !reflect.DeepEqual(m.Claims, []claim{{Name: "mesh-watchdogs", Scope: "mesh", Serves: []string{"signals", "stuck", "presence", "dead-letters", "resync"}}}) {
t.Errorf("claims = %+v", m.Claims)
@@ -101,7 +106,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
if !reflect.DeepEqual(m.Tools, []string{"watchman_findings"}) {
t.Errorf("tools = %v", m.Tools)
}
for _, k := range []string{"emits", "context", "items", "consumes", "provides"} {
for _, k := range []string{"emits", "context", "items", "provides"} {
if _, ok := top[k]; ok {
t.Errorf("the manifest declares %q: the watchman has none", k)
}
modules/mesh-delivery-watchman/module.json
+3
View File
@@ -26,6 +26,9 @@
"mesh-provisioner.provisions",
"node-engine.have"
],
"consumes": [
"mesh-delivery-indexer.take-in.**"
],
"tools": [
"watchman_findings"
],