diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 8ec5142d..8117b1d9 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -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 _ 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) + } +} diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 40e308b2..67cadbe5 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -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. 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 +} diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index 042a5321..b1409b4a 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -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}, } diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go new file mode 100644 index 00000000..8d9abf23 --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/takein.go @@ -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..`): +// 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}) +} diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 0e8b7925..67b641fe 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -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 diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index a577361e..6a54308e 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -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) } diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index 4c85f5d7..2de43956 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -26,6 +26,9 @@ "mesh-provisioner.provisions", "node-engine.have" ], + "consumes": [ + "mesh-delivery-indexer.take-in.**" + ], "tools": [ "watchman_findings" ],