From 8f09aa223b22b27845dc0bf59df636b1f4d7a246 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:37:55 +0100 Subject: [PATCH] RN25: the runner's health check is its tool runner_health MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The process writes, in its state consumers, when its durable consumer's oldest unacknowledged message was delivered, at each delivery and acknowledgement; the bundle's runner_health reads it and the pipelines state, and names a consumer five minutes or more behind. The process resource's health is that tool. Builds: design 47 G §76, §468, §495; lead answers A-164, A-166 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-pipeline-tools/main.go | 11 ++ .../mesh-delivery-pipeline-tools/main_test.go | 4 +- .../cmd/mesh-delivery-pipeline/approval.go | 2 +- .../cmd/mesh-delivery-pipeline/consumers.go | 57 +++++++ .../mesh-delivery-pipeline/consumers_test.go | 41 +++++ .../cmd/mesh-delivery-pipeline/main.go | 21 ++- .../run_integration_test.go | 42 +++++- .../cmd/mesh-delivery-pipeline/start.go | 4 +- .../cmd/mesh-delivery-pipeline/steps.go | 6 +- .../internal/pipeline/consumers.go | 142 ++++++++++++++++++ .../internal/pipeline/consumers_test.go | 96 ++++++++++++ .../mesh-delivery-pipeline/manifest_test.go | 28 +++- modules/mesh-delivery-pipeline/module.json | 10 +- 13 files changed, 446 insertions(+), 18 deletions(-) create mode 100644 modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go create mode 100644 modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/consumers.go create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go index a8aafaef..4b0559b7 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go @@ -98,9 +98,20 @@ func FirstViewTools(st stdio.KeptState, acts views.HandActs, now func() time.Tim return views.StatusTools(Module, Prefix+"_first_views", Prefix+"_first_view_again", st, acts, now) } +// HealthTool is the runner's health check, run outside its process (design 47 G §76, §468, §495; A-164): it reads the +// runner's state pipelines as show does and the ages its process writes in the state consumers, and asks no seat verb. +func HealthTool(pipelines, consumers pipeline.Kept, now func() time.Time) stdio.Tool { + return stdio.Tool{Name: Prefix + "_health", + Description: "The pipeline runner's health: healthy when it answers show and each of its durable consumers' oldest " + + "unacknowledged message waits less than five minutes; otherwise it names each consumer behind.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { return pipeline.Health(pipelines, consumers, now()), nil }} +} + func main() { r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}} tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) + tools = append(tools, HealthTool(r.Pipelines, kept{stdio.State(pipeline.StateConsumers)}, time.Now)) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go index 2638c5d9..bb3dd146 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go @@ -26,7 +26,9 @@ func TestTools(t *testing.T) { for _, tl := range FirstViewTools(stdio.State(views.StatusState), nil, nil) { names = append(names, tl.Name) } - want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again" + // RN25 (§468, §76; A-164): the health check is the bundle's runner_health. + names = append(names, HealthTool(none{}, none{}, nil).Name) + want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again runner_health" if got := strings.Join(names, " "); got != want { t.Errorf("tools %q, want %q", got, want) } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go index 0927551a..67ec25b7 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go @@ -56,7 +56,7 @@ var _ pipeline.Asker = busAsker{} // hearDecided wakes the runner on the operator-channel's decided for its own asks (§37, A-89): the warrant is read from the // router's record, so an event the record does not back yet is heard again. -func hearDecided(hub *process.EventHub, h *pipeline.Hearer, module string, logf func(string, ...any)) error { +func hearDecided(hub handlers, h *pipeline.Hearer, module string, logf func(string, ...any)) error { return hub.Handle(oc.SubjectDecided(module), func(m jetstream.Msg) { var ev oc.Warrant if err := json.Unmarshal(m.Data(), &ev); err != nil { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go new file mode 100644 index 00000000..2581c317 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go @@ -0,0 +1,57 @@ +package main + +import ( + "context" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// handlers is where a part of the runner registers its handler of the events it hears: the event hub, or agedHub over it. +type handlers interface { + Handle(pattern string, handle func(jetstream.Msg)) error +} + +// agedHub registers each handler on the hub so that every message it is handed is noted delivered, and acknowledged +// once the handler acknowledges, refuses or ends it, so the runner's state consumers holds when the consumer's oldest +// unacknowledged message was delivered (design 47 G §76, §468, §495; A-164, A-166). +type agedHub struct { + hub handlers + ages *pipeline.ConsumerAges +} + +func (a agedHub) Handle(pattern string, handle func(jetstream.Msg)) error { + return a.hub.Handle(pattern, func(m jetstream.Msg) { + handle(&agedMsg{Msg: m, ages: a.ages, mark: a.ages.Delivered()}) + }) +} + +// agedMsg is a message whose acknowledgement, refusal or end is noted on its consumer's ages, once. +type agedMsg struct { + jetstream.Msg + ages *pipeline.ConsumerAges + mark uint64 + once sync.Once +} + +func (m *agedMsg) done() { m.once.Do(func() { m.ages.Acknowledged(m.mark) }) } + +func (m *agedMsg) Ack() error { defer m.done(); return m.Msg.Ack() } + +func (m *agedMsg) DoubleAck(ctx context.Context) error { defer m.done(); return m.Msg.DoubleAck(ctx) } + +func (m *agedMsg) Nak() error { defer m.done(); return m.Msg.Nak() } + +func (m *agedMsg) NakWithDelay(d time.Duration) error { defer m.done(); return m.Msg.NakWithDelay(d) } + +func (m *agedMsg) Term() error { defer m.done(); return m.Msg.Term() } + +func (m *agedMsg) TermWithReason(r string) error { defer m.done(); return m.Msg.TermWithReason(r) } + +// consumersPut writes a consumer's age in the state consumers. +func consumersPut(ctx context.Context, kv jetstream.KeyValue) func(string, []byte) error { + return func(k string, v []byte) error { _, err := kv.Put(ctx, k, v); return err } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go new file mode 100644 index 00000000..8dd8db43 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go @@ -0,0 +1,41 @@ +package main + +import ( + "encoding/json" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-delivery-pipeline/internal/pipeline" +) + +type oneHub struct{ handle func(jetstream.Msg) } + +func (h *oneHub) Handle(_ string, handle func(jetstream.Msg)) error { h.handle = handle; return nil } + +type ackedMsg struct{ jetstream.Msg } + +func (ackedMsg) Ack() error { return nil } + +// TestAgedHub: a message handed to a handler is noted delivered, and acknowledged when the handler acknowledges it, so +// the state consumers holds the consumer's oldest unacknowledged delivery (§76, §468; RN25, A-166). +func TestAgedHub(t *testing.T) { + var last pipeline.ConsumerAge + ages := &pipeline.ConsumerAges{Consumer: "n_runner", Put: func(_ string, v []byte) error { last = pipeline.ConsumerAge{}; return json.Unmarshal(v, &last) }} + hub := &oneHub{} + var inside *pipeline.ConsumerAge + if err := (agedHub{hub: hub, ages: ages}).Handle("x.>", func(m jetstream.Msg) { + seen := last + inside = &seen + _ = m.Ack() + }); err != nil { + t.Fatal(err) + } + hub.handle(ackedMsg{}) + if inside == nil || inside.Oldest == nil { + t.Fatalf("inside the handler the consumer's oldest unacknowledged delivery = %+v, want it written", inside) + } + if last.Oldest != nil { + t.Errorf("after the handler acknowledged: oldest %v, want none", last.Oldest) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index 2190fdfb..4e073a16 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -71,6 +71,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer calls.Close() hub := process.NewEventHub(logger.Printf) + // The age of the durable consumer's oldest unacknowledged message, read by the tool runner_health (§76, §468; A-164, + // A-166): written when a message is delivered or acknowledged. + consumers, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, pipeline.StateConsumers), "say how far its consumer is behind") + if err != nil { + return err + } + ages := &pipeline.ConsumerAges{Consumer: cred.EventConsumerName(), Put: consumersPut(ctx, consumers), Logf: logger.Printf} + ages.Begin() + heard := agedHub{hub: hub, ages: ages} // A branch's view holding its first half asks again the steps placed in its running pipelines that were never asked // (§88); the starter is made after the views, so it is reached through a pointer set once it is. var resumer atomic.Pointer[func(string)] @@ -112,25 +121,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro logger.Printf("the steps placed on %s and not yet asked: %d asked again", branch, n) } resumer.Store(&resume) - if err := hearStepEnded(hub, hearer, logger.Printf); err != nil { + if err := hearStepEnded(heard, hearer, logger.Printf); err != nil { return err } - if err := hearToolchain(ctx, hub, toolchains, logger.Printf); err != nil { + if err := hearToolchain(ctx, heard, toolchains, logger.Printf); err != nil { return err } - if err := hearAskTaken(ctx, hub, bounds, logger.Printf); err != nil { + if err := hearAskTaken(ctx, heard, bounds, logger.Printf); err != nil { return err } if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil { return err } - if err := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil { + if err := hearDecided(heard, hearer, cred.Module, logger.Printf); err != nil { return err } - if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil { + if err := hearPublished(ctx, heard, starter, calls, rv.Catalog, logger.Printf); err != nil { return err } - if err := hearTakenIn(hub, starter, rv.Catalog, logger.Printf); err != nil { + if err := hearTakenIn(heard, starter, rv.Catalog, logger.Printf); err != nil { return err } if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go index 5637d20c..61ce0a1f 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go @@ -6,6 +6,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "os" "path/filepath" "sync" @@ -20,6 +21,8 @@ import ( seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/testbus" + + "mesh-delivery-pipeline/internal/pipeline" ) type syncBuffer struct { @@ -53,7 +56,7 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes defer cancel() const module = "mesh-delivery-pipeline" - for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views"} { + for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views", "consumers"} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_" + state, History: 1}); err != nil { t.Fatal(err) } @@ -157,6 +160,43 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes }) waitFor(t, ctx, logs, func() bool { return readPipeline(ctx, pipelines, key, &p) && p.Outcome == seat.PipelineOutcomePassed }) + + // RN25 (§76, §468; A-166): the durable consumer's age is in the state consumers, none waiting once every message heard + // is acknowledged, and runner_health reads it healthy. + consumers, err := js.KeyValue(ctx, module+"_consumers") + if err != nil { + t.Fatal(err) + } + var age pipeline.ConsumerAge + waitFor(t, ctx, logs, func() bool { + e, err := consumers.Get(ctx, "n1_"+module) + return err == nil && json.Unmarshal(e.Value(), &age) == nil && age.Oldest == nil && age.Consumer == "n1_"+module + }) + if a := pipeline.Health(kvKept{ctx, pipelines}, kvKept{ctx, consumers}, time.Now()); !a.Healthy { + t.Errorf("runner_health over the bus: %+v", a) + } +} + +// kvKept reads a bucket as the tools bundle's states do. +type kvKept struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (k kvKept) Keys() ([]string, error) { + keys, err := k.kv.Keys(k.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + +func (k kvKept) Value(key string) (json.RawMessage, error) { + e, err := k.kv.Get(k.ctx, key) + if err != nil { + return nil, err + } + return e.Value(), nil } func stepOf(p seat.Pipeline, step string) seat.StepResult { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go index 5c933297..f75f738f 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/start.go @@ -36,7 +36,7 @@ func (c createdKV) Create(key string, value []byte) error { // hearPublished starts the pipelines of each package-published heard (§53, §334): the dependents the catalog's outdated // names behind, each asked as a kept call. The message is acknowledged once its starts are written, on a goroutine of its // own, so a slow catalog holds no other event. -func hearPublished(ctx context.Context, hub *process.EventHub, starter *pipeline.Starter, calls *process.Conn, v *pipeline.Views, logf func(string, ...any)) error { +func hearPublished(ctx context.Context, hub handlers, starter *pipeline.Starter, calls *process.Conn, v *pipeline.Views, logf func(string, ...any)) error { outdated := func(in catalog.OutdatedInput) (catalog.OutdatedAnswer, error) { raw, err := calls.CallKept(outdatedVerb, in) if err != nil { @@ -72,7 +72,7 @@ func hearPublished(ctx context.Context, hub *process.EventHub, starter *pipeline // hearTakenIn starts the pipelines of each take-in.taken heard (§44, §54): one per entry, the message acknowledged once // they are written, and heard again when a write failed, so the keys already written start nothing twice. -func hearTakenIn(hub *process.EventHub, starter *pipeline.Starter, v *pipeline.Views, logf func(string, ...any)) error { +func hearTakenIn(hub handlers, starter *pipeline.Starter, v *pipeline.Views, logf func(string, ...any)) error { return hub.Handle(pipeline.TakenInSubject, func(m jetstream.Msg) { var ev catalog.TakeInTaken if err := json.Unmarshal(m.Data(), &ev); err != nil { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go index 5e4aa9e7..7322ee79 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -99,7 +99,7 @@ func newHearer(ctx context.Context, js jetstream.JetStream, cred process.Account // hearStepEnded takes each build seat's step-ended into its pipeline (§24; ADR 0666 §21): refused and said, never taken, // when it is not as the runner asked it, and heard again when a write failed. -func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string, ...any)) error { +func hearStepEnded(hub handlers, h *pipeline.Hearer, logf func(string, ...any)) error { for _, bs := range pipeline.StepSeats { bs := bs err := hub.Handle(bs.Prefix+"*", func(m jetstream.Msg) { @@ -137,7 +137,7 @@ func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string, // hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647): the ask is answered, so its five minutes // end with no finding. -func hearAskTaken(ctx context.Context, hub *process.EventHub, bounds jetstream.KeyValue, logf func(string, ...any)) error { +func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, logf func(string, ...any)) error { for _, ts := range pipeline.TakenSeats { ts := ts err := hub.Handle(ts.Subject, func(m jetstream.Msg) { @@ -252,7 +252,7 @@ func toolchainReader(ctx context.Context, toolchains jetstream.KeyValue) func(st } // hearToolchain keeps each toolchain-agreed heard as `` in the state toolchains (A-104): the newest of a language. -func hearToolchain(ctx context.Context, hub *process.EventHub, toolchains jetstream.KeyValue, logf func(string, ...any)) error { +func hearToolchain(ctx context.Context, hub handlers, toolchains jetstream.KeyValue, logf func(string, ...any)) error { return hub.Handle(builds.SubjectToolchainAgreed, func(m jetstream.Msg) { t, err := pipeline.ToolchainOf(m.Data()) if err != nil || t == nil { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go b/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go new file mode 100644 index 00000000..05f8b95a --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go @@ -0,0 +1,142 @@ +package pipeline + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +// StateConsumers is the runner's old state section of its durable consumers' ages, a key per consumer (design 47 G +// §468, §76, §495; A-164, A-166). +const StateConsumers = "consumers" + +// BehindAfter is how long a consumer's oldest unacknowledged message may wait before the runner's health check names it +// behind (§76). +const BehindAfter = 5 * time.Minute + +// ConsumerAge is one durable consumer's value in the state consumers: when its oldest unacknowledged message was +// delivered, none when every delivered message is acknowledged (A-166). +type ConsumerAge struct { + Consumer string `json:"consumer"` + Oldest *time.Time `json:"oldest-unacknowledged,omitempty"` +} + +// ConsumerAges follows one durable consumer's delivered and acknowledged messages and writes its ConsumerAge through +// Put each time the oldest unacknowledged delivery changes (A-166). +type ConsumerAges struct { + Consumer string + Put func(key string, value []byte) error + Now func() time.Time + Logf func(string, ...any) + + mu sync.Mutex + next uint64 + pending map[uint64]time.Time + written *time.Time + wrote bool +} + +// Begin writes the consumer's value at the runner's start, before any message is delivered, so its health check reads +// every consumer the process holds. +func (c *ConsumerAges) Begin() { + c.mu.Lock() + defer c.mu.Unlock() + c.writeLocked() +} + +// Delivered notes a message delivered now and answers its mark, given to Acknowledged once the message is acknowledged. +func (c *ConsumerAges) Delivered() uint64 { + c.mu.Lock() + defer c.mu.Unlock() + if c.pending == nil { + c.pending = map[uint64]time.Time{} + } + c.next++ + now := time.Now().UTC() + if c.Now != nil { + now = c.Now().UTC() + } + c.pending[c.next] = now + c.writeLocked() + return c.next +} + +// Acknowledged notes the message of mark acknowledged (or refused for good): it waits no more. +func (c *ConsumerAges) Acknowledged(mark uint64) { + c.mu.Lock() + defer c.mu.Unlock() + if _, ok := c.pending[mark]; !ok { + return + } + delete(c.pending, mark) + c.writeLocked() +} + +// writeLocked writes the consumer's oldest unacknowledged delivery when it differs from the one last written. +func (c *ConsumerAges) writeLocked() { + var oldest *time.Time + for _, at := range c.pending { + if oldest == nil || at.Before(*oldest) { + at := at + oldest = &at + } + } + if c.wrote && ((oldest == nil && c.written == nil) || (oldest != nil && c.written != nil && oldest.Equal(*c.written))) { + return + } + raw, err := json.Marshal(ConsumerAge{Consumer: c.Consumer, Oldest: oldest}) + if err == nil { + err = c.Put(c.Consumer, raw) + } + if err != nil { + if c.Logf != nil { + c.Logf("the age of the consumer %s is not written in %s: %v", c.Consumer, StateConsumers, err) + } + return + } + c.written, c.wrote = oldest, true +} + +// Health is the runner's health as its tool runner_health answers it (§468, §76, §495; A-164): healthy when its state +// pipelines answers, as show reads it, and each durable consumer's oldest unacknowledged message waits less than +// BehindAfter; otherwise unhealthy naming each consumer behind and how long. +func Health(pipelines, consumers Kept, now time.Time) health.Answer { + if _, err := pipelines.Keys(); err != nil { + return health.Unhealthy(fmt.Sprintf("The runner does not answer show: its state pipelines cannot be listed (%v).", err)) + } + keys, err := consumers.Keys() + if err != nil { + return health.Unhealthy(fmt.Sprintf("The ages of the runner's consumers cannot be read from its state %s (%v).", StateConsumers, err)) + } + if len(keys) == 0 { + return health.Unhealthy(fmt.Sprintf("The runner has written no consumer's age in its state %s: it has heard no event yet.", StateConsumers)) + } + sort.Strings(keys) + var behind []string + for _, k := range keys { + raw, err := consumers.Value(k) + if err != nil { + return health.Unhealthy(fmt.Sprintf("The age of the consumer %s cannot be read (%v).", k, err)) + } + var a ConsumerAge + if raw != nil { + if err := json.Unmarshal(raw, &a); err != nil { + return health.Unhealthy(fmt.Sprintf("The age of the consumer %s does not read (%v).", k, err)) + } + } + if a.Oldest != nil { + if age := now.Sub(*a.Oldest); age >= BehindAfter { + behind = append(behind, fmt.Sprintf("%s (its oldest unacknowledged message waits %s)", k, age.Round(time.Second))) + } + } + } + if len(behind) > 0 { + return health.Unhealthy(fmt.Sprintf("The runner's consumers behind by %s or more: %s.", BehindAfter, strings.Join(behind, ", "))) + } + return health.Healthy(fmt.Sprintf("The runner answers show, and each of its %d consumers is less than %s behind.", len(keys), BehindAfter)) +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go new file mode 100644 index 00000000..8cfac785 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go @@ -0,0 +1,96 @@ +package pipeline + +import ( + "encoding/json" + "errors" + "strings" + "testing" + "time" +) + +type mapKept map[string]json.RawMessage + +func (m mapKept) Keys() ([]string, error) { + var out []string + for k := range m { + out = append(out, k) + } + return out, nil +} +func (m mapKept) Value(k string) (json.RawMessage, error) { return m[k], nil } + +type failingKept struct{} + +func (failingKept) Keys() ([]string, error) { return nil, errors.New("no bucket") } +func (failingKept) Value(string) (json.RawMessage, error) { return nil, errors.New("no bucket") } + +// TestConsumerAges: the process writes when its consumer's oldest unacknowledged message was delivered, each time it +// changes, and none once all are acknowledged (§468, §495; A-164, A-166). +func TestConsumerAges(t *testing.T) { + at := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC) + now := at + st := mapKept{} + writes := 0 + c := &ConsumerAges{Consumer: "n_mesh-delivery-pipeline", Now: func() time.Time { return now }, + Put: func(k string, v []byte) error { writes++; st[k] = v; return nil }} + oldest := func() *time.Time { + var a ConsumerAge + if err := json.Unmarshal(st["n_mesh-delivery-pipeline"], &a); err != nil { + t.Fatal(err) + } + return a.Oldest + } + c.Begin() + if writes != 1 || oldest() != nil { + t.Fatalf("at start: %d writes, oldest %v; want one write of none", writes, oldest()) + } + first := c.Delivered() + now = at.Add(time.Minute) + second := c.Delivered() + if writes != 2 || !oldest().Equal(at) { + t.Errorf("two delivered: %d writes, oldest %v; want 2 writes, oldest %v (a later delivery leaves the oldest)", writes, oldest(), at) + } + c.Acknowledged(first) + if !oldest().Equal(at.Add(time.Minute)) { + t.Errorf("the first acknowledged: oldest %v, want %v", oldest(), at.Add(time.Minute)) + } + c.Acknowledged(second) + c.Acknowledged(second) + if oldest() != nil || writes != 4 { + t.Errorf("all acknowledged: oldest %v after %d writes, want none after 4", oldest(), writes) + } +} + +// TestHealth: runner_health is healthy when the runner answers show and each consumer is less than five minutes behind, +// and otherwise names each consumer behind (§76, §468, §495; RN25, A-164). +func TestHealth(t *testing.T) { + now := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC) + age := func(name string, since time.Duration) mapKept { + a := ConsumerAge{Consumer: name} + if since > 0 { + t := now.Add(-since) + a.Oldest = &t + } + raw, _ := json.Marshal(a) + return mapKept{name: raw} + } + rows := []struct { + name string + pipelines Kept + consumers Kept + healthy bool + says string + }{ + {"none waiting", mapKept{}, age("n_runner", 0), true, "less than 5m0s behind"}, + {"four minutes", mapKept{}, age("n_runner", 4*time.Minute), true, ""}, + {"six minutes names the consumer", mapKept{}, age("n_runner", 6*time.Minute), false, "n_runner (its oldest unacknowledged message waits 6m0s)"}, + {"show does not answer", failingKept{}, age("n_runner", 0), false, "does not answer show"}, + {"no consumer written", mapKept{}, mapKept{}, false, "no consumer's age"}, + } + for _, r := range rows { + a := Health(r.pipelines, r.consumers, now) + if a.Healthy != r.healthy || !strings.Contains(a.Why, r.says) { + t.Errorf("%s: healthy %v %q; want %v saying %q", r.name, a.Healthy, a.Why, r.healthy, r.says) + } + } +} diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index cbc157a5..58693b51 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -54,7 +54,7 @@ func TestManifest(t *testing.T) { {"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat && equalSets(m.Claims[0].Serves, seat.Verbs())}, // §27, §85, §330, §355, §77: the old state sections. - {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views"})}, + {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views", "consumers"})}, // §535, §647, A-77, A-104: an ask no holder takes is bound by a key with its own time to live. {"bounds is key-ttl", keyTTL["bounds"]}, // RN25 (§469, §85): an old state section declares its cap, pipelines 256 MiB and kept 128 MiB. @@ -91,7 +91,9 @@ func TestManifest(t *testing.T) { "seat:node-build-image.check.step"})}, {"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})}, // RN25 (§539): the runner's tools carry the prefix runner_. - {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again"})}, + {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health"})}, + // RN25 (§468, §76): the health check, run outside the process, is the tool runner_health. + {"health is runner_health", healthTool(raw) == "tool runner_health"}, // RN1/AC5 (§95, §130): the new assigner is delivered before the runner. {"program needs the assigner's program", slices.Equal(program.Needs, []string{"mesh-delivery-assigner:program"})}, } @@ -127,6 +129,28 @@ func TestNoVerbNamesItsCallers(t *testing.T) { } } +// healthTool is the process resource's health kind and tool. +func healthTool(raw []byte) string { + var m struct { + Resources []struct { + Type string `json:"type"` + Health struct { + Kind string `json:"kind"` + Tool string `json:"tool"` + } `json:"health"` + } `json:"resources"` + } + if err := json.Unmarshal(raw, &m); err != nil { + return err.Error() + } + for _, r := range m.Resources { + if r.Type == "process" { + return r.Health.Kind + " " + r.Health.Tool + } + } + return "" +} + func equalSets(a, b []string) bool { x, y := slices.Clone(a), slices.Clone(b) slices.Sort(x) diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index 143719c6..e185be5c 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -24,7 +24,8 @@ ], "tools": [ "runner_first_views", - "runner_first_view_again" + "runner_first_view_again", + "runner_health" ], "consumes": [ "mesh-delivery-indexer.take-in.taken", @@ -89,6 +90,10 @@ { "name": "first-views", "history": 1 + }, + { + "name": "consumers", + "history": 1 } ], "streams": [ @@ -146,7 +151,8 @@ }, "user": "mesh-delivery-pipeline", "health": { - "kind": "unit" + "kind": "tool", + "tool": "runner_health" } } ],