From fc4ce1cd79c39616c44178bccafd130d0649dade Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:34:01 +0100 Subject: [PATCH 1/6] RN25: the runner's stores declare their caps, its sent store, tools named runner_ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pipelines state is capped at 256 MiB and kept at 128 MiB in the old state section, the sent store sent-objects is declared at 256 MiB, and the first-view tools carry the runner's tool prefix. Builds: design 47 G §469, §85, §477, §533, §539 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-pipeline-tools/main.go | 6 +++--- .../mesh-delivery-pipeline-tools/main_test.go | 4 ++-- modules/mesh-delivery-pipeline/manifest_test.go | 16 +++++++++++++--- modules/mesh-delivery-pipeline/module.json | 14 ++++++++++---- 4 files changed, 28 insertions(+), 12 deletions(-) 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 64776e43..a8aafaef 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 @@ -86,16 +86,16 @@ func Tools(r pipeline.Reads) []stdio.Tool { return out } -// Module and Slug are the runner's module and its tools' prefix, as its manifest names them. +// Module is the runner's module, and Prefix its tools' prefix (design 47 G §539: runner_first_views, runner_health). const ( Module = "mesh-delivery-pipeline" - Slug = "pipeline" + Prefix = "runner" ) // FirstViewTools are the runner's `_first_views` and `_first_view_again` over its state first-views // (design 47 G §77, §200): each view it holds or waits for, since when, and asked again by hand, a hand act. func FirstViewTools(st stdio.KeptState, acts views.HandActs, now func() time.Time) []stdio.Tool { - return views.StatusTools(Module, Slug+"_first_views", Slug+"_first_view_again", st, acts, now) + return views.StatusTools(Module, Prefix+"_first_views", Prefix+"_first_view_again", st, acts, now) } func main() { 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 73d96db0..2638c5d9 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 @@ -17,7 +17,7 @@ func (none) Keys() ([]string, error) { return nil, nil } func (none) Value(string) (json.RawMessage, error) { return nil, nil } // TestTools: the bundle serves the five read verbs as . (§83, §306; RN15/AC1), and the first-view tools -// (§77, §200; RN2/AC4). +// (§77, §200; RN2/AC4), named with the prefix runner_ (§539; RN25). func TestTools(t *testing.T) { var names []string for _, tl := range Tools(pipeline.Reads{Pipelines: none{}, Kept: none{}}) { @@ -26,7 +26,7 @@ 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 pipeline_first_views pipeline_first_view_again" + 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" if got := strings.Join(names, " "); got != want { t.Errorf("tools %q, want %q", got, want) } diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index a1ffc26d..cbc157a5 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -22,15 +22,19 @@ func TestManifest(t *testing.T) { } var states, streams, objects []string keyTTL := map[string]bool{} + stateMiB := map[string]int{} for _, s := range m.State { states = append(states, s.Name) keyTTL[s.Name] = s.KeyTTL + stateMiB[s.Name] = s.MaxMiB } + objectMiB := map[string]int{} for _, s := range m.Streams { streams = append(streams, s.Name) } for _, o := range m.Objects { objects = append(objects, o.Name) + objectMiB[o.Name] = o.MaxMiB } var program catalogue.Artifact if m.Build == nil { @@ -53,6 +57,9 @@ func TestManifest(t *testing.T) { {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views"})}, // §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. + {"pipelines 256 MiB", stateMiB[seat.StatePipelines] == 256}, + {"kept 128 MiB", stateMiB[seat.StateKept] == 128}, // §470 (amending §27, §85): the step log is an old logs section kept 100 days, 128 MiB; RN24 (§383, §387): the finding // stream its findings are said on is the one old stream. {"old streams section", equalSets(streams, []string{"mesh-delivery-pipeline-findings"})}, @@ -62,8 +69,10 @@ func TestManifest(t *testing.T) { equalSets(m.Streams[0].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"})}, // RN24 (§368): the version as MAJOR.MINOR, 0.1 with none published. {"version 0.1", m.Version == "0.1"}, - {"old objects section", equalSets(objects, []string{seat.StateStepOutputs})}, - {"step-outputs 1 GiB", len(m.Objects) == 1 && m.Objects[0].MaxMiB == 1024}, + // RN25 (§477, §533, §59): a foundation part whose seat has verbs and events declares its sent store, 256 MiB. + {"old objects section", equalSets(objects, []string{seat.StateStepOutputs, catalogue.SentObjectsLocal})}, + {"step-outputs 1 GiB", objectMiB[seat.StateStepOutputs] == 1024}, + {"sent-objects 256 MiB", objectMiB[catalogue.SentObjectsLocal] == 256}, // §333, §54, §44, §53, §376, §24, §37, §4, §77: what the runner hears. {"consumes", equalSets(m.Consumes, []string{ "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", @@ -81,7 +90,8 @@ func TestManifest(t *testing.T) { "seat:mesh-builds.step", "seat:mesh-wants.step", "seat:node-build-go.check.step", "seat:node-build-node.check.step", "seat:node-build-image.check.step"})}, {"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})}, - {"first-view tools", equalSets(m.Tools, []string{"pipeline_first_views", "pipeline_first_view_again"})}, + // RN25 (§539): the runner's tools carry the prefix runner_. + {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again"})}, // 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"})}, } diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index b66b07dd..143719c6 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -23,8 +23,8 @@ } ], "tools": [ - "pipeline_first_views", - "pipeline_first_view_again" + "runner_first_views", + "runner_first_view_again" ], "consumes": [ "mesh-delivery-indexer.take-in.taken", @@ -61,11 +61,13 @@ "state": [ { "name": "pipelines", - "history": 1 + "history": 1, + "max-mib": 256 }, { "name": "kept", - "history": 1 + "history": 1, + "max-mib": 128 }, { "name": "aged", @@ -105,6 +107,10 @@ { "name": "step-outputs", "max-mib": 1024 + }, + { + "name": "sent-objects", + "max-mib": 256 } ], "own-secrets": { From 8f09aa223b22b27845dc0bf59df636b1f4d7a246 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:37:55 +0100 Subject: [PATCH 2/6] 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" } } ], From 9199bb20f446447b6ad1b13324a5ab86c63cd516 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:41:03 +0100 Subject: [PATCH 3/6] RN14: the runner serves start and ask-again MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit start starts by hand the pipeline of a watched branch's head from the runner's catalog half, refusing a head already started; ask-again asks a declined approval once more under an ask of its own, its id following from when it was asked again so the ask rebuilt for its warrant is the same. Each is recorded in the hand-act log first; the process answers both beside review, rerun, accept and cancel. Builds: design 47 G §30, §39, §306, §36, §394 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-pipeline/verbs.go | 24 ++++--- .../cmd/mesh-delivery-pipeline/verbs_test.go | 18 +++++ .../internal/pipeline/approval.go | 14 +++- .../internal/pipeline/verbs.go | 65 +++++++++++++++++++ .../internal/pipeline/verbs_test.go | 64 ++++++++++++++++++ 5 files changed, 176 insertions(+), 9 deletions(-) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go index 6ddf5016..1be958c2 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go @@ -32,15 +32,23 @@ func verbOf[In any](do func(caller string, in In) (any, error)) process.SeatHand } } -// answerVerbs answers the seat's write verbs `review`, `rerun`, `accept` and `cancel` over the bus, each writing the state -// pipelines by compare-and-set (design 47 G §22, §70, §482, A-107). The returned stop ends the answers. +// answerVerbs answers the seat's write verbs over the bus, each writing the state pipelines by compare-and-set (design 47 +// G §22, §70, §306, §482, A-107). The returned stop ends the answers. func answerVerbs(conn *process.Conn, v pipeline.Verbs) (func(), error) { - return conn.AnswerSeat(seat.Seat, map[string]process.SeatHandler{ - seat.VerbReview: verbOf(func(c string, in seat.ReviewInput) (any, error) { return v.Review(c, in) }), - seat.VerbRerun: verbOf(func(c string, in seat.RerunInput) (any, error) { return v.Rerun(c, in) }), - seat.VerbAccept: verbOf(func(c string, in seat.AcceptInput) (any, error) { return v.Accept(c, in) }), - seat.VerbCancel: verbOf(func(c string, in seat.CancelInput) (any, error) { return v.Cancel(c, in) }), - }) + return conn.AnswerSeat(seat.Seat, writeHandlers(v)) +} + +// writeHandlers are the seat's write verbs the process answers: review, rerun, accept, cancel, start and ask-again +// (§306; RN14). +func writeHandlers(v pipeline.Verbs) map[string]process.SeatHandler { + return map[string]process.SeatHandler{ + seat.VerbReview: verbOf(func(c string, in seat.ReviewInput) (any, error) { return v.Review(c, in) }), + seat.VerbRerun: verbOf(func(c string, in seat.RerunInput) (any, error) { return v.Rerun(c, in) }), + seat.VerbAccept: verbOf(func(c string, in seat.AcceptInput) (any, error) { return v.Accept(c, in) }), + seat.VerbCancel: verbOf(func(c string, in seat.CancelInput) (any, error) { return v.Cancel(c, in) }), + seat.VerbStart: verbOf(func(c string, in seat.StartInput) (any, error) { return v.Start(c, in) }), + seat.VerbAskAgain: verbOf(func(c string, in seat.AskAgainInput) (any, error) { return v.AskAgain(c, in) }), + } } // dropBound deletes the bound of an ask a cancelled pipeline no longer waits on. diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go index 9b5fb4d6..6b1db4e8 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go @@ -2,11 +2,29 @@ package main import ( "encoding/json" + "slices" "testing" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + + "mesh-delivery-pipeline/internal/pipeline" ) +// RN12, RN14 (§306): the process answers every write verb of the seat, the bundle every read verb, so each verb the +// manifest claims is served. +func TestEveryWriteVerbIsServed(t *testing.T) { + var got []string + for v := range writeHandlers(pipeline.Verbs{}) { + got = append(got, v) + } + slices.Sort(got) + want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbReview, seat.VerbStart} + slices.Sort(want) + if !slices.Equal(got, want) { + t.Errorf("the process answers %v, want %v", got, want) + } +} + // A write verb is answered only for a caller the bus proved, and its input is the verb's own (design 47 G §394, A-107). func TestAVerbIsAnsweredOnlyForAProvedCallerAndItsOwnInput(t *testing.T) { var got seat.CancelInput diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/approval.go b/modules/mesh-delivery-pipeline/internal/pipeline/approval.go index 2756fa75..6497dfe5 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/approval.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/approval.go @@ -51,7 +51,15 @@ func approvalDue(p seat.Pipeline) int { // ApprovalAsk is the ask of a pipeline's approval step. Its id and its expiry follow from the pipeline and from when the // step was asked, so the ask built again for the same pipeline is the same ask, digest included: a warrant is held to it. func ApprovalAsk(p seat.Pipeline, from time.Time) (oc.Ask, error) { - sum := sha256.Sum256([]byte(p.Pipeline + "\n" + p.Commit)) + // An approval asked again (ask-again) is an ask of its own, its id following from when it was asked again, so the + // ask rebuilt from the step is the same and the earlier, decided ask is not its duplicate. + idOf := p.Pipeline + "\n" + p.Commit + for _, st := range p.Steps { + if st.Step == seat.StepResultStepApproval && st.Rerun { + idOf += "\nasked again\n" + from.UTC().Format(time.RFC3339Nano) + } + } + sum := sha256.Sum256([]byte(idOf)) act := func(do string) (string, error) { return operatorchannelclient.ActDigest(operatorchannelclient.Act{"pipeline": p.Pipeline, "commit": p.Commit, "fingerprint": p.Fingerprint, "do": do}) @@ -98,6 +106,10 @@ func askApproval(key string, p *seat.Pipeline, from time.Time, asker Asker) (boo if i < 0 || asker == nil { return false, nil } + // An approval asked again keeps the time ask-again wrote, so an ask published and then not written is the same ask. + if s := p.Steps[i]; s.Rerun && s.Started != nil { + from = *s.Started + } a, err := ApprovalAsk(*p, from) if err != nil { return false, fmt.Errorf("the approval of %s cannot be asked: %w", key, err) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go index 65a924a6..e3caef80 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go @@ -334,3 +334,68 @@ func (v Verbs) askOf(p seat.Pipeline, s seat.StepResult, caller string, now time } return nil, fmt.Errorf("the %s step of %s is placed no ask now", s.Step, p.Pipeline) } + +// Start starts by hand a delivering pipeline for the head of a watched branch the runner's catalog half holds, recorded in +// the hand-act log first (§306, §394): the pipeline of that branch's source revision and the module, as a take-in would +// start it. A head whose pipeline is written already starts nothing and is refused, naming the pipeline. +func (v Verbs) Start(_ string, in seat.StartInput) (seat.StartAnswer, error) { + if in.Branch == "" || in.Module == "" { + return seat.StartAnswer{}, errors.New("start names the watched branch and the module") + } + if v.Starter == nil || v.View == nil { + return seat.StartAnswer{}, errors.New("the runner cannot start a pipeline: it holds no starter or no catalog half") + } + half, ok := v.View(in.Branch) + if !ok { + return seat.StartAnswer{}, fmt.Errorf("the catalog half of %s is not held yet: nothing was started", in.Branch) + } + e, ok := half.Entries[in.Module] + if !ok || half.Source <= 0 { + return seat.StartAnswer{}, fmt.Errorf("the catalog half of %s holds no entry of %s at a source revision: nothing was started", in.Branch, in.Module) + } + key := Key(in.Branch, half.Source, in.Module) + if err := v.record(seat.VerbStart, "started the pipeline "+key+" by hand", in.Why); err != nil { + return seat.StartAnswer{}, err + } + started, err := v.Starter.Start(Begin{Branch: in.Branch, Source: half.Source, Module: in.Module, Commit: e.Commit, Fp: e.Fp, + Repository: e.Repository, Revision: half.Revision, Entries: half.Entries}) + if err != nil { + return seat.StartAnswer{}, err + } + if !started { + return seat.StartAnswer{}, fmt.Errorf("the pipeline %s of the head is started already: a step of it is asked again by rerun", key) + } + return seat.StartAnswer{Pipeline: key}, nil +} + +// AskAgain asks a declined approval step once more (§306, §36), recorded in the hand-act log first: the step waits again, +// marked asked again by the caller the bus proved, and is asked at once when the pipeline's every other required step +// passed, under an ask of its own (the earlier ask's decision ended it). +func (v Verbs) AskAgain(caller string, in seat.AskAgainInput) (seat.AskAgainAnswer, error) { + p, i, err := v.change(in.Pipeline, seat.EventStepStarted, func() error { + return v.record(seat.VerbAskAgain, "asked the approval of "+in.Pipeline+" again", in.Why) + }, func(p *seat.Pipeline, now time.Time) (int, error) { + switch p.Outcome { + case seat.PipelineOutcomeCancelled, seat.PipelineOutcomeSuperseded: + return -1, fmt.Errorf("the pipeline %s is %s: its approval is not asked again", p.Pipeline, p.Outcome) + } + i, err := stepOf(*p, seat.StepResultStepApproval, "") + if err != nil { + return -1, err + } + s := &p.Steps[i] + if s.Outcome != seat.StepResultOutcomeDeclined { + return -1, fmt.Errorf("only a declined approval is asked again: the approval of %s is %s", p.Pipeline, s.Outcome) + } + *s = seat.StepResult{Step: s.Step, Artifact: s.Artifact, Status: s.Status, Fingerprint: s.Fingerprint, + Outcome: seat.StepResultOutcomeWaiting, By: caller, Rerun: true, Started: &now} + if p.Outcome != seat.PipelineOutcomeRunning { + p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeRunning, "", nil + } + return i, nil + }) + if err != nil { + return seat.AskAgainAnswer{}, err + } + return seat.AskAgainAnswer{Result: p.Steps[i]}, nil +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go index 602f33c2..633f120b 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go @@ -170,3 +170,67 @@ func TestAVerbWhoseHandActIsNotRecordedDoesNothing(t *testing.T) { type failingActs struct{} func (failingActs) RecordHandAct(views.HandAct) error { return errors.New("no holder") } + +// RN14 (§306, §394): start starts by hand the pipeline of a watched branch's head, recorded in the hand-act log, and a head +// already started is refused; ask-again asks a declined approval once more under an ask of its own, recorded, and an +// approval not declined is refused. +func TestStartAndAskAgain(t *testing.T) { + v, store, _, _, acts, _ := verbsFixture() + made := created{} + v.Starter.Pipelines, v.Starter.Say = made, func(string, any) error { return nil } + _, entries := placeFixture() + v.View = func(string) (Half, bool) { return Half{Source: 7, Revision: 3, Entries: entries}, true } + a, err := v.Start("operator", seat.StartInput{Branch: "main", Module: "app", Why: "no pipeline for the head"}) + if err != nil || a.Pipeline != "main.7.app" || made["main.7.app"] == nil { + t.Fatalf("start = %+v, %v; written %v", a, err, len(made)) + } + if _, err := v.Start("operator", seat.StartInput{Branch: "main", Module: "app", Why: "again"}); err == nil { + t.Error("a second start of the same head was not refused") + } + if _, err := v.Start("operator", seat.StartInput{Branch: "main", Module: "absent", Why: "x"}); err == nil { + t.Error("a start of a module the half does not hold was not refused") + } + if len(*acts) != 2 || !strings.HasPrefix((*acts)[0], "mesh-pipeline-runner.start: ") { + t.Errorf("hand acts %v, want the two starts recorded", *acts) + } + + // ask-again: the approval of the fixture's pipeline is declined; asked again it is asked under a new ask. + h := v.Hearer + asker := h.Asker.(*askRecorder) + e := store[pkey] + first, err := ApprovalAsk(e.p, time.Unix(1, 0)) + if err != nil { + t.Fatal(err) + } + for i := range e.p.Steps { + s := &e.p.Steps[i] + switch s.Step { + case seat.StepResultStepApproval: + s.Outcome, s.Ask = seat.StepResultOutcomeDeclined, first.ID + default: + s.Outcome = seat.StepResultOutcomePassed + } + } + e.p.Outcome = seat.PipelineOutcomeFailed + store[pkey] = e + if _, err := v.AskAgain("operator", seat.AskAgainInput{Pipeline: pkey}); err == nil { + t.Error("an ask-again saying no why was taken") + } + got, err := v.AskAgain("operator", seat.AskAgainInput{Pipeline: pkey, Why: "asked by mistake"}) + if err != nil || got.Result.Outcome != seat.StepResultOutcomeRunning || got.Result.Ask == "" || got.Result.Ask == first.ID || !got.Result.Rerun { + t.Fatalf("ask-again = %+v, %v", got, err) + } + if len(asker.asked) != 1 || asker.asked[0].ID != got.Result.Ask { + t.Errorf("asked %v, want one ask %s", asker.asked, got.Result.Ask) + } + if store[pkey].p.Outcome != seat.PipelineOutcomeRunning { + t.Errorf("the pipeline is %s after ask-again, want running", store[pkey].p.Outcome) + } + // The ask rebuilt from the step, as a warrant is held to it, is the ask asked. + if again, err := ApprovalAsk(store[pkey].p, *stepNamed(store[pkey].p, "", seat.StepResultStepApproval).Started); err != nil || again.ID != got.Result.Ask { + t.Errorf("the ask rebuilt = %s, %v; want %s", again.ID, err, got.Result.Ask) + } + if _, err := v.AskAgain("operator", seat.AskAgainInput{Pipeline: pkey, Why: "twice"}); err == nil { + t.Error("an approval running was asked again") + } +} From 4b47e883693ffaaf49ad08098ad547d2eec7ca86 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:42:33 +0100 Subject: [PATCH 4/6] RN25: runner_health reads the consumer facts mesh-sdk's ledger writes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The runner's own writer of its consumer's oldest unacknowledged delivery is replaced by mesh-sdk's consumer ledger: each message the hub hands a handler is noted delivered, and applied when the handler acknowledges or ends it; one it asks again stays unacknowledged. The key form is the same, a key per consumer by its name in the state consumers. A runner that heard no event yet is healthy. Builds: design 47 G §76, §468, §495; lead answers A-166, A-176 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-pipeline/consumers.go | 46 ++++----- .../mesh-delivery-pipeline/consumers_test.go | 51 +++++++--- .../cmd/mesh-delivery-pipeline/main.go | 8 +- .../run_integration_test.go | 4 +- .../internal/pipeline/consumers.go | 99 ++----------------- .../internal/pipeline/consumers_test.go | 46 +-------- 6 files changed, 77 insertions(+), 177 deletions(-) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go index 2581c317..288ab266 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go @@ -7,7 +7,7 @@ import ( "github.com/nats-io/nats.go/jetstream" - "mesh-delivery-pipeline/internal/pipeline" + "git.novox.be/novox/mesh-sdk/go/process" ) // handlers is where a part of the runner registers its handler of the events it hears: the event hub, or agedHub over it. @@ -15,43 +15,43 @@ 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). +// agedHub registers each handler on the hub so that every message it is handed is noted delivered on the consumer's +// ledger, and applied once the handler acknowledges 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, A-176). A message +// the handler asks again (a refusal for now) stays unacknowledged. type agedHub struct { - hub handlers - ages *pipeline.ConsumerAges + hub handlers + ledger *process.ConsumerLedger } 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()}) + a.ledger.Delivered(m) + handle(&agedMsg{Msg: m, ledger: a.ledger}) }) } -// agedMsg is a message whose acknowledgement, refusal or end is noted on its consumer's ages, once. +// agedMsg is a message whose acknowledgement or end is noted applied on its consumer's ledger, once. type agedMsg struct { jetstream.Msg - ages *pipeline.ConsumerAges - mark uint64 - once sync.Once + ledger *process.ConsumerLedger + once sync.Once } -func (m *agedMsg) done() { m.once.Do(func() { m.ages.Acknowledged(m.mark) }) } +func (m *agedMsg) applied() { m.once.Do(func() { m.ledger.Applied(m.Msg, time.Time{}) }) } -func (m *agedMsg) Ack() error { defer m.done(); return m.Msg.Ack() } +func (m *agedMsg) Ack() error { defer m.applied(); return m.Msg.Ack() } -func (m *agedMsg) DoubleAck(ctx context.Context) error { defer m.done(); return m.Msg.DoubleAck(ctx) } +func (m *agedMsg) DoubleAck(ctx context.Context) error { + defer m.applied() + return m.Msg.DoubleAck(ctx) +} -func (m *agedMsg) Nak() error { defer m.done(); return m.Msg.Nak() } +func (m *agedMsg) Term() error { defer m.applied(); return m.Msg.Term() } -func (m *agedMsg) NakWithDelay(d time.Duration) error { defer m.done(); return m.Msg.NakWithDelay(d) } +func (m *agedMsg) TermWithReason(r string) error { defer m.applied(); return m.Msg.TermWithReason(r) } -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 } +// consumersSink writes a consumer's facts in the state consumers. +func consumersSink(kv jetstream.KeyValue) process.ConsumerSink { + return func(ctx context.Context, 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 index 8dd8db43..91ddcd8c 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go @@ -1,41 +1,62 @@ package main import ( + "context" "encoding/json" "testing" "github.com/nats-io/nats.go/jetstream" - "mesh-delivery-pipeline/internal/pipeline" + "git.novox.be/novox/mesh-sdk/go/process" ) 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 } +type ackedMsg struct { + jetstream.Msg + seq uint64 +} func (ackedMsg) Ack() error { return nil } +func (ackedMsg) Nak() error { return nil } +func (m ackedMsg) Metadata() (*jetstream.MsgMetadata, error) { + return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: m.seq}}, nil +} +func (ackedMsg) Subject() string { return "mesh.seat.x.event.y" } -// 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). +// TestAgedHub: a message handed to a handler is noted delivered on the consumer's ledger, applied when the handler +// acknowledges it, and left unacknowledged when the handler asks it again (§76, §468; RN25, A-166, A-176). 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) }} + var last process.ConsumerFacts + ledger := process.NewConsumerLedger(process.EventsStream, "n_runner", func(_ context.Context, _ string, v []byte) error { + last = process.ConsumerFacts{} + return json.Unmarshal(v, &last) + }, nil, nil) hub := &oneHub{} - var inside *pipeline.ConsumerAge - if err := (agedHub{hub: hub, ages: ages}).Handle("x.>", func(m jetstream.Msg) { - seen := last - inside = &seen + var inside process.ConsumerFacts + nak := false + if err := (agedHub{hub: hub, ledger: ledger}).Handle("x.>", func(m jetstream.Msg) { + inside = last + if nak { + _ = m.Nak() + return + } _ = 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) + hub.handle(ackedMsg{seq: 1}) + if inside.OldestUnacknowledged.IsZero() { + t.Fatalf("inside the handler the consumer's facts = %+v, want its delivery written", inside) } - if last.Oldest != nil { - t.Errorf("after the handler acknowledged: oldest %v, want none", last.Oldest) + if !last.OldestUnacknowledged.IsZero() || last.Acknowledged != 1 { + t.Errorf("after the handler acknowledged: %+v, want none unacknowledged and 1 acknowledged", last) + } + nak = true + hub.handle(ackedMsg{seq: 2}) + if last.OldestUnacknowledged.IsZero() { + t.Errorf("a message asked again is noted applied: %+v", last) } } 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 4e073a16..9712039e 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -71,15 +71,13 @@ 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. + // The durable consumer's facts, among them when its oldest unacknowledged message was delivered, read by the tool + // runner_health (§76, §468; A-164, A-166, A-176): written when a message is delivered or applied. 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} + heard := agedHub{hub: hub, ledger: process.NewConsumerLedger(process.EventsStream, cred.EventConsumerName(), consumersSink(consumers), nil, logger.Printf)} // 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)] 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 61ce0a1f..feb5d854 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 @@ -167,10 +167,10 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes if err != nil { t.Fatal(err) } - var age pipeline.ConsumerAge + var facts process.ConsumerFacts 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 + return err == nil && json.Unmarshal(e.Value(), &facts) == nil && facts.OldestUnacknowledged.IsZero() && facts.Acknowledged > 0 }) if a := pipeline.Health(kvKept{ctx, pipelines}, kvKept{ctx, consumers}, time.Now()); !a.Healthy { t.Errorf("runner_health over the bus: %+v", a) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go b/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go index 05f8b95a..d7a85fab 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go @@ -5,103 +5,20 @@ import ( "fmt" "sort" "strings" - "sync" "time" "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/process" ) -// 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" +// StateConsumers is the runner's old state section of its durable consumers' facts, a key per consumer, written by +// mesh-sdk's consumer ledger (design 47 G §468, §76, §495; A-164, A-166, A-176). +const StateConsumers = process.ConsumersState // 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. @@ -114,7 +31,7 @@ func Health(pipelines, consumers Kept, now time.Time) health.Answer { 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)) + return health.Healthy(fmt.Sprintf("The runner answers show and has heard no event yet: its state %s holds no consumer.", StateConsumers)) } sort.Strings(keys) var behind []string @@ -123,14 +40,14 @@ func Health(pipelines, consumers Kept, now time.Time) health.Answer { if err != nil { return health.Unhealthy(fmt.Sprintf("The age of the consumer %s cannot be read (%v).", k, err)) } - var a ConsumerAge + var a process.ConsumerFacts 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 { + if !a.OldestUnacknowledged.IsZero() { + if age := now.Sub(a.OldestUnacknowledged); age >= BehindAfter { behind = append(behind, fmt.Sprintf("%s (its oldest unacknowledged message waits %s)", k, age.Round(time.Second))) } } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go index 8cfac785..8decf31c 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go @@ -6,6 +6,8 @@ import ( "strings" "testing" "time" + + "git.novox.be/novox/mesh-sdk/go/process" ) type mapKept map[string]json.RawMessage @@ -24,52 +26,14 @@ 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} + a := process.ConsumerFacts{Name: name} if since > 0 { - t := now.Add(-since) - a.Oldest = &t + a.OldestUnacknowledged = now.Add(-since) } raw, _ := json.Marshal(a) return mapKept{name: raw} @@ -85,7 +49,7 @@ func TestHealth(t *testing.T) { {"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"}, + {"no event heard yet", mapKept{}, mapKept{}, true, "has heard no event yet"}, } for _, r := range rows { a := Health(r.pipelines, r.consumers, now) From 56c45eb7d68956090be5d468685d16d63f5ba8b5 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:01:44 +0100 Subject: [PATCH 5/6] RN8: a newer commit or publish supersedes the older running pipeline, its asks dropped where they stand MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A step keeps the node whose holder said ask-taken; a superseded step's drop is asked of that node's holder, an untaken ask's of any holder, as kept calls; steps on mesh-builds and mesh-wants are only ended superseded. A pipeline a publish started says that publish (published), so a newer publish of the package supersedes it. Builds: design 47 G §31, §119, §350, A-182, A-202 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../go/seats/meshpipelinerunner/contract.go | 16 +- .../seats/mesh-pipeline-runner/seat.json | 26 +++ .../ts/seats/meshpipelinerunner/index.d.ts | 12 ++ .../cmd/mesh-delivery-pipeline/drop.go | 41 +++++ .../cmd/mesh-delivery-pipeline/main.go | 12 +- .../cmd/mesh-delivery-pipeline/steps.go | 12 +- .../internal/pipeline/start.go | 14 +- .../internal/pipeline/start_test.go | 13 +- .../internal/pipeline/supersede.go | 155 ++++++++++++++++++ .../internal/pipeline/supersede_test.go | 93 +++++++++++ .../mesh-delivery-pipeline/manifest_test.go | 2 + modules/mesh-delivery-pipeline/module.json | 5 +- 12 files changed, 389 insertions(+), 12 deletions(-) create mode 100644 modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drop.go create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/supersede.go create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go diff --git a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go index 7b0bfe66..307ffd31 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go @@ -432,8 +432,10 @@ type Pipeline struct { // yes when the module is over its test budget (design 47 G section 419) OverBudget bool `json:"over-budget,omitempty"` // the pipeline's key, .., a branch escaped as a key's name is - Pipeline string `json:"pipeline,omitempty"` - Repository string `json:"repository,omitempty"` + Pipeline string `json:"pipeline,omitempty"` + // for a pipeline a package's publish started, that publish (A-202) + Published *Published `json:"published,omitempty"` + Repository string `json:"repository,omitempty"` // the catalog graph revision of the branch the pipeline runs for Revision int64 `json:"revision,omitempty"` // the run on request's id, for a pipeline of a run; such a pipeline delivers nothing @@ -545,6 +547,8 @@ type PipelineStarted struct { Module string `json:"module"` // the pipeline's key, .., a branch escaped as a key's name is Pipeline string `json:"pipeline"` + // for a pipeline a package's publish started, that publish (A-202) + Published *Published `json:"published,omitempty"` // the catalog graph revision of the branch the pipeline runs for Revision int64 `json:"revision,omitempty"` // the run on request's id, for a pipeline of a run; such a pipeline delivers nothing @@ -556,6 +560,14 @@ type PipelineStarted struct { V int64 `json:"v"` } +// Published is the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202). +type Published struct { + // the package whose publish started the pipeline + Artifact string `json:"artifact"` + // the version that publish named + Version string `json:"version"` +} + // RunStarted is a run on request's start, with its members at their commits (ADR 0666 section 18). type RunStarted struct { At *time.Time `json:"at,omitempty"` diff --git a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json index 01778d80..a644b81e 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json @@ -907,6 +907,24 @@ "name" ] }, + "Published": { + "type": "object", + "properties": { + "artifact": { + "type": "string", + "description": "the package whose publish started the pipeline" + }, + "version": { + "type": "string", + "description": "the version that publish named" + } + }, + "description": "the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202)", + "required": [ + "artifact", + "version" + ] + }, "Pipeline": { "type": "object", "properties": { @@ -1002,6 +1020,10 @@ "version": { "type": "string", "description": "the version package-published names, for a pipeline the runner started from a publication or from outdated; none for any other (design 47 G 550)" + }, + "published": { + "$ref": "#/defs/Published", + "description": "for a pipeline a package's publish started, that publish (A-202)" } }, "description": "one pipeline as the runner keeps it in its state pipelines, by compare-and-set (design 47 *The runner's kept data*)", @@ -1191,6 +1213,10 @@ "at": { "type": "string", "format": "date-time" + }, + "published": { + "$ref": "#/defs/Published", + "description": "for a pipeline a package's publish started, that publish (A-202)" } }, "description": "a pipeline's start", diff --git a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts index 73c3dcb1..7a7cc39a 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts @@ -195,6 +195,8 @@ export interface Pipeline { "over-budget"?: boolean; /** the pipeline's key, .., a branch escaped as a key's name is */ "pipeline"?: string; + /** for a pipeline a package's publish started, that publish (A-202) */ + "published"?: Published; "repository"?: string; /** the catalog graph revision of the branch the pipeline runs for */ "revision"?: number; @@ -268,6 +270,8 @@ export interface PipelineStarted { "module": string; /** the pipeline's key, .., a branch escaped as a key's name is */ "pipeline": string; + /** for a pipeline a package's publish started, that publish (A-202) */ + "published"?: Published; /** the catalog graph revision of the branch the pipeline runs for */ "revision"?: number; /** the run on request's id, for a pipeline of a run; such a pipeline delivers nothing */ @@ -279,6 +283,14 @@ export interface PipelineStarted { "v": number; } +/** the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202) */ +export interface Published { + /** the package whose publish started the pipeline */ + "artifact": string; + /** the version that publish named */ + "version": string; +} + /** a run on request's start, with its members at their commits (ADR 0666 section 18) */ export interface RunStarted { "at"?: string; diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drop.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drop.go new file mode 100644 index 00000000..8a3f1a05 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drop.go @@ -0,0 +1,41 @@ +package main + +import ( + "encoding/json" + "fmt" + + nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-sdk/go/buscalls" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// keptCaller is the runner's kept calls, to any holder of a seat or to one node's (process.Conn). +type keptCaller interface { + CallKept(key string, body any) (json.RawMessage, error) + CallKeptOn(node, key string, body any) (json.RawMessage, error) +} + +// dropAsk asks the build seat's `drop` of a superseded step's ask as a kept call (design 47 G §31, §350; A-182): of the +// holder on the node that took it, or, never taken, of any holder, which records it in the seat's shared drop key. The +// three build seats' drop take one input (nodebuildgo.DropInput; the manifest invokes each). +func dropAsk(calls keptCaller, by string) pipeline.Drop { + return func(buildSeat, node, ask, why string) error { + key := "seat:" + buildSeat + "." + nodebuildgo.VerbDrop + in := nodebuildgo.DropInput{ID: ask, Why: why, By: by} + var raw json.RawMessage + var err error + if node != "" { + raw, err = calls.CallKeptOn(node, key, in) + } else { + raw, err = calls.CallKept(key, in) + } + if err != nil { + return err + } + if _, said, ok := buscalls.Unwrap(raw); !ok { + return fmt.Errorf("%s answered: %s", key, said) + } + return nil + } +} 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 9712039e..1bfb5453 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -119,13 +119,23 @@ 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) + // A newer commit or publish supersedes the older running pipelines it replaces, their asks dropped as kept calls (§31, + // §350; A-182, A-202). + starter.Supersede = func(newer seat.Pipeline) { + n, err := hearer.Supersede(newer, dropAsk(calls, cred.Module), dropBound(ctx, bounds)) + if err != nil { + logger.Printf("the pipelines %s supersedes: %d ended, then %v", newer.Pipeline, n, err) + } else if n > 0 { + logger.Printf("the pipeline %s superseded %d older pipelines", newer.Pipeline, n) + } + } if err := hearStepEnded(heard, hearer, logger.Printf); err != nil { return err } if err := hearToolchain(ctx, heard, toolchains, logger.Printf); err != nil { return err } - if err := hearAskTaken(ctx, heard, bounds, logger.Printf); err != nil { + if err := hearAskTaken(ctx, heard, bounds, hearer.Taken, logger.Printf); err != nil { return err } if err := watchBounds(ctx, bounds, hearer, logger.Printf); 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 7322ee79..c75823dc 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -135,9 +135,9 @@ func hearStepEnded(hub handlers, h *pipeline.Hearer, logf func(string, ...any)) return nil } -// 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 handlers, bounds jetstream.KeyValue, logf func(string, ...any)) error { +// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647) and keeps on its step the node that took +// it (A-182): the ask is answered, so its five minutes end with no finding. +func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, taken func(ask, node string) error, logf func(string, ...any)) error { for _, ts := range pipeline.TakenSeats { ts := ts err := hub.Handle(ts.Subject, func(m jetstream.Msg) { @@ -152,6 +152,12 @@ func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, _ = m.Nak() return } + // The step keeps the node that took its ask, so a drop of it is asked of that node's holder (A-182). + if err := taken(ev.Ask, ev.Node); err != nil { + logf("the node %s that took ask %s is not kept on its step, heard again: %v", ev.Node, ev.Ask, err) + _ = m.Nak() + return + } _ = m.Ack() }) if err != nil { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 9e5737aa..0d8c561c 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -37,6 +37,8 @@ type Begin struct { Revision int64 // Entries are the catalog half's entries of the pipeline's branch: the steps' artifacts and build seats are read from them. Entries map[string]catalog.CatalogEntry + // Published is the package's publish that started the pipeline, nil for any other start (A-202). + Published *seat.Published } // Key is a pipeline's key in the state pipelines, `..`, the branch escaped as a key's @@ -70,6 +72,8 @@ type Starter struct { Toolchain func(language string) (*nodebuildgo.ToolchainUsed, error) // Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none. Arm Arm + // Supersede ends the older running pipelines the pipeline just written supersedes (§31; Hearer.Supersede); nil ends none. + Supersede func(newer seat.Pipeline) } // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second @@ -85,7 +89,7 @@ func (s *Starter) Start(b Begin) (bool, error) { key := Key(b.Branch, b.Source, b.Module) p := seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now, - Steps: []seat.StepResult{}} + Steps: []seat.StepResult{}, Published: b.Published} var asks []Asked if s.Registration != nil { if reg, ok := s.Registration(b.Module); ok { @@ -105,7 +109,7 @@ func (s *Starter) Start(b Begin) (bool, error) { return false, fmt.Errorf("the pipeline %s is not written: %w", key, err) } ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, - Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Revision: b.Revision} + Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Revision: b.Revision, Published: b.Published} commitOrGroup := b.Commit if b.Group != "" { commitOrGroup = b.Group @@ -113,6 +117,9 @@ func (s *Starter) Start(b Begin) (bool, error) { if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, b.Module, b.Branch, commitOrGroup), ev); err != nil { s.logf("pipeline-started of %s is not said: %v", key, err) } + if s.Supersede != nil { + s.Supersede(p) + } if s.Steps != nil && len(asks) > 0 { s.carry(key, asks) if err := askAll(s.Steps, s.Arm, key, asks); err != nil { @@ -252,7 +259,8 @@ func (s *Starter) Published(ev builds.PackagePublished, outdated func(catalog.Ou if o.Kind != outdatedKind || o.Module == "" { continue } - b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision, Entries: half.Entries} + b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision, Entries: half.Entries, + Published: &seat.Published{Artifact: ev.Artifact, Version: ev.Version}} if e, ok := half.Entries[o.Module]; ok { b.Commit, b.Fp, b.Repository = e.Commit, e.Fp, e.Repository } else { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go index 7fcfec71..63fef192 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go @@ -27,11 +27,15 @@ func (c created) Create(key string, value []byte) error { func TestStarts(t *testing.T) { store := created{} var said []string + var superseding []string s := &Starter{Pipelines: store, Say: func(subject string, body any) error { ev := body.(seat.PipelineStarted) said = append(said, subject+" "+ev.Pipeline) + if ev.Published == nil || ev.Published.Artifact != "mesh-sdk:go" || ev.Published.Version != "0.1.16" { + t.Errorf("pipeline-started of %s says the publish %+v (A-202)", ev.Pipeline, ev.Published) + } return nil - }} + }, Supersede: func(newer seat.Pipeline) { superseding = append(superseding, newer.Pipeline) }} view := func(branch string) (Half, bool) { return Half{Revision: 9, Entries: map[string]catalog.CatalogEntry{ "app": {Module: "app", Commit: "c1", Fp: "src2:aa", Repository: "o/app"}}}, true @@ -51,9 +55,13 @@ func TestStarts(t *testing.T) { } var p seat.Pipeline if err := json.Unmarshal(store["main.41.app"], &p); err != nil || p.Commit != "c1" || p.Fingerprint != "src2:aa" || p.Revision != 9 || - p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 { + p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 || p.Published == nil || p.Published.Version != "0.1.16" { t.Errorf("the pipeline main.41.app = %+v (%v)", p, err) } + // Each pipeline written supersedes the older ones it replaces (§31). + if len(superseding) != 2 { + t.Errorf("the publish's pipelines superseding: %v, want both", superseding) + } if _, ok := store["main.41.ordered"]; ok { t.Error("an ordering edge starts no pipeline") } @@ -66,6 +74,7 @@ func TestStarts(t *testing.T) { if !strings.HasPrefix(said[0], "mesh.seat.mesh-pipeline-runner.event.pipeline-started.app.main.c1 main.41.app") { t.Errorf("pipeline-started said on %q", said[0]) } + s.Say = func(string, any) error { return nil } // A branch holding a dot is escaped in the key, so the key reads back; a start naming no source revision is refused. if _, err := s.Start(Begin{Branch: "release.1", Source: 3, Module: "app"}); err != nil { t.Fatal(err) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/supersede.go b/modules/mesh-delivery-pipeline/internal/pipeline/supersede.go new file mode 100644 index 00000000..292f171f --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/supersede.go @@ -0,0 +1,155 @@ +package pipeline + +import ( + "errors" + "fmt" + "time" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// Drop asks a build seat's `drop` of a superseded step's ask as a kept call (design 47 G §31, §350; A-182): of node's holder +// when a holder took it (node the one its ask-taken named), else of any holder, which records it in the seat's shared drop +// key so no holder takes it. why names the newer commit. +type Drop func(buildSeat, node, ask, why string) error + +// Taken notes on the running step asked under ask the node whose holder said ask-taken for it (A-182), so a drop of it is +// asked of that node's holder. A step no pipeline keeps running under ask takes nothing. +func (h *Hearer) Taken(ask, node string) error { + if ask == "" || node == "" { + return nil + } + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + return fmt.Errorf("the pipelines are not listed: %w", err) + } + for _, key := range pkeys { + for { + p, rev, err := h.Store.Get(key) + if err != nil { + break + } + i := stepOfAsk(p, ask) + if i < 0 || p.Steps[i].Outcome != seat.StepResultOutcomeRunning || p.Steps[i].Node == node { + break + } + p.Steps[i].Node = node + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return fmt.Errorf("pipeline %s cannot be written: %w", key, err) + } + return nil + } + } + return nil +} + +// Supersede ends, for the pipeline newer just written, each older running pipeline of the same watched branch and module +// whose fingerprint newer's commit changed, and for a publish's pipeline each an older publish of the package started +// (design 47 G §31, §119; ADR 0666 §9; A-202): the pipeline and each step not ended +// `superseded`, naming the newer commit, and it says pipeline-ended. A module the newer commit left as it was finishes its +// pipeline. The ask of each superseded step on a build seat that serves `drop` is dropped (drop; its bound deleted through +// unbound); a step on mesh-builds or mesh-wants, whose seats serve no drop, is only ended superseded (A-182). A run on +// request's pipeline neither supersedes nor is superseded. It says how many pipelines it ended. +func (h *Hearer) Supersede(newer seat.Pipeline, drop Drop, unbound func(ask string) error) (int, error) { + if newer.Run != "" || newer.Source <= 0 { + return 0, nil + } + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + return 0, fmt.Errorf("the pipelines are not listed: %w", err) + } + why := supersededWhy(newer) + ended := 0 + var first error + for _, key := range pkeys { + if key == newer.Pipeline { + continue + } + done, err := h.supersedeOne(key, newer, why, drop, unbound) + if err != nil && first == nil { + first = err + } + if done { + ended++ + } + } + return ended, first +} + +// superseded says whether old is a running pipeline newer supersedes (§31, A-202): of the same watched branch and an older +// source revision, either of the same module at another fingerprint, or, for newer a package's publish started, one an older +// publish of that package started (its version another, made from an older source revision of the branch). +func superseded(old, newer seat.Pipeline) bool { + if old.Outcome != seat.PipelineOutcomeRunning || old.Run != "" || old.Branch != newer.Branch || old.Source >= newer.Source { + return false + } + if newer.Published != nil && old.Published != nil && old.Published.Artifact == newer.Published.Artifact { + return old.Published.Version != newer.Published.Version + } + return old.Module == newer.Module && old.Fingerprint != newer.Fingerprint +} + +// supersededWhy names the newer commit, or the newer publish, that ended a pipeline. +func supersededWhy(newer seat.Pipeline) string { + if newer.Published != nil { + return fmt.Sprintf("the newer publish %s %s of %s supersedes it", newer.Published.Artifact, newer.Published.Version, newer.Branch) + } + return fmt.Sprintf("the newer commit %s of %s changed %s's fingerprint", newer.Commit, newer.Branch, newer.Module) +} + +func (h *Hearer) supersedeOne(key string, newer seat.Pipeline, why string, drop Drop, unbound func(string) error) (bool, error) { + for { + p, rev, err := h.Store.Get(key) + if err != nil || !superseded(p, newer) { + return false, nil + } + now := h.now() + var drops []seat.StepResult + var changed []int + for i := range p.Steps { + s := &p.Steps[i] + if s.Outcome != seat.StepResultOutcomeRunning && s.Outcome != seat.StepResultOutcomeWaiting { + continue + } + if s.Ask != "" && BoundSeat(s.Seat) { + drops = append(drops, *s) + } + s.Outcome, s.Why, s.Ended = seat.StepResultOutcomeSuperseded, why, ptrTime(now) + changed = append(changed, i) + } + p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeSuperseded, why, ptrTime(now) + p.EndCount++ + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return false, fmt.Errorf("pipeline %s cannot be written: %w", key, err) + } + var first error + for _, i := range changed { + h.logEnd(key, p.Steps[i], now) + if err := h.say(seat.EventStepEnded, p, &p.Steps[i]); err != nil && first == nil { + first = fmt.Errorf("pipeline %s's %s is written but not said: %w", key, seat.EventStepEnded, err) + } + } + for _, s := range drops { + if drop != nil { + if err := drop(s.Seat, s.Node, s.Ask, why); err != nil { + h.logf("the drop of ask %s of the superseded %s is not answered: %v", s.Ask, key, err) + } + } + if unbound != nil { + if err := unbound(s.Ask); err != nil { + h.logf("the bound of ask %s of the superseded %s is not dropped: %v", s.Ask, key, err) + } + } + } + if err := h.said(p, false, true); err != nil && first == nil { + first = err + } + return true, first + } +} + +func ptrTime(t time.Time) *time.Time { return &t } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go new file mode 100644 index 00000000..d6b96b54 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go @@ -0,0 +1,93 @@ +package pipeline + +import ( + "slices" + "strings" + "testing" + + builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" +) + +// RN8 (design 47 G §31, §119, §350; A-182, A-202): a step keeps the node that took its ask; a newer commit changing the +// module's fingerprint, or a newer publish of the package that started a pipeline, ends the older running pipeline +// superseded, its asks dropped of the node that took each (any holder for one never taken), a mesh-builds step only ended; +// an unchanged fingerprint, a run on request and an equal publish supersede nothing. +func TestSupersede(t *testing.T) { + h, store, said, _, _, _ := fixture() + old := store[pkey] + old.p.Steps = append(old.p.Steps, seat.StepResult{Artifact: "program", Step: seat.StepResultStepComposition, Status: seat.StepResultStatusRequired, + Outcome: seat.StepResultOutcomeRunning, Ask: "b1", Seat: builds.Seat}) + store[pkey] = old + if err := h.Taken("a1", "n7"); err != nil { + t.Fatal(err) + } + if p, _, _ := store.Get(pkey); p.Steps[0].Node != "n7" || p.Steps[1].Node != "" { + t.Fatalf("after ask-taken of a1 on n7 the steps name %q and %q", p.Steps[0].Node, p.Steps[1].Node) + } + type dropped struct{ seat, node, ask string } + var drops []dropped + var unbound []string + drop := func(s, node, ask, why string) error { + if !strings.Contains(why, "c2") { + t.Errorf("the drop of %s names no newer commit: %q", ask, why) + } + drops = append(drops, dropped{s, node, ask}) + return nil + } + unbind := func(ask string) error { unbound = append(unbound, ask); return nil } + newer := func(key string, source int64, fp string, run string, pub *seat.Published) seat.Pipeline { + return seat.Pipeline{Pipeline: key, Branch: "main", Source: source, Module: "app", Commit: "c2", Fingerprint: fp, Run: run, Published: pub, + Outcome: seat.PipelineOutcomeRunning} + } + for _, c := range []struct { + name string + newer seat.Pipeline + }{ + {"an unchanged fingerprint", newer("main.6.app", 6, "src2:aa", "", nil)}, + {"a run on request", newer("main.r1.app", 6, "src2:bb", "r1", nil)}, + {"an older source revision", newer("main.4.app", 4, "src2:bb", "", nil)}, + } { + if n, err := h.Supersede(c.newer, drop, unbind); n != 0 || err != nil { + t.Errorf("%s superseded %d (%v)", c.name, n, err) + } + } + *said = nil + if n, err := h.Supersede(newer("main.6.app", 6, "src2:bb", "", nil), drop, unbind); n != 1 || err != nil { + t.Fatalf("a changed fingerprint superseded %d (%v)", n, err) + } + p, _, _ := store.Get(pkey) + if p.Outcome != seat.PipelineOutcomeSuperseded || p.EndCount != 1 || !strings.Contains(p.Why, "c2") { + t.Errorf("the older pipeline is %s (%d ends): %q", p.Outcome, p.EndCount, p.Why) + } + for _, s := range p.Steps { + if s.Outcome != seat.StepResultOutcomeSuperseded { + t.Errorf("the %s step is %s, want superseded", s.Step, s.Outcome) + } + } + want := []dropped{{nodebuildgo.Seat, "n7", "a1"}, {nodebuildgo.Seat, "", "a2"}} + if !slices.Equal(drops, want) || !slices.Equal(unbound, []string{"a1", "a2"}) { + t.Errorf("dropped %+v and unbound %v, want %+v (the mesh-builds step only ended)", drops, unbound, want) + } + if n := len(*said); n != 5 || (*said)[4] != seat.EventPipelineEnded+".app.main.c1" { + t.Errorf("said %v, want four step-ended then pipeline-ended", *said) + } + if n, _ := h.Supersede(newer("main.7.app", 7, "src2:cc", "", nil), drop, unbind); n != 0 { + t.Errorf("a superseded pipeline was superseded again: %d", n) + } + + // A newer publish of a package supersedes the pipelines an older publish of it started, whatever their module. + pub := memStore{"main.5.lib": {p: seat.Pipeline{Pipeline: "main.5.lib", Branch: "main", Source: 5, Module: "lib", Fingerprint: "src2:aa", + Outcome: seat.PipelineOutcomeRunning, Published: &seat.Published{Artifact: "core:pkg", Version: "v1"}}, rev: 1}} + h.Store = pub + if n, _ := h.Supersede(newer("main.6.app", 6, "src2:aa", "", &seat.Published{Artifact: "core:pkg", Version: "v1"}), drop, unbind); n != 0 { + t.Errorf("a publish of the same version superseded %d", n) + } + if n, err := h.Supersede(newer("main.6.app", 6, "src2:aa", "", &seat.Published{Artifact: "core:pkg", Version: "v2"}), drop, unbind); n != 1 || err != nil { + t.Fatalf("a newer publish superseded %d (%v)", n, err) + } + if p, _, _ := pub.Get("main.5.lib"); p.Outcome != seat.PipelineOutcomeSuperseded || !strings.Contains(p.Why, "v2") { + t.Errorf("the older publish's pipeline is %s: %q", p.Outcome, p.Why) + } +} diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index 58693b51..457b0298 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -84,6 +84,8 @@ func TestManifest(t *testing.T) { {"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act", // RN3 (§53, §334): a publication's dependents behind. "seat:mesh-catalog.outdated", + // RN8 (§31, §350; A-182): a superseded step's ask dropped of the build seat that holds it. + "seat:node-build-go.drop", "seat:node-build-node.drop", "seat:node-build-image.drop", // A-74, A-89: who holds the seat operator-channel, whose state asks holds the router's record of the approval ask. "seat:mesh-module-assigner.assignments", // A-102 (§22, §24): the accepted kinds the runner puts its steps on. diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index e185be5c..e16d652f 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -54,7 +54,10 @@ "seat:mesh-wants.step", "seat:node-build-go.check.step", "seat:node-build-node.check.step", - "seat:node-build-image.check.step" + "seat:node-build-image.check.step", + "seat:node-build-go.drop", + "seat:node-build-node.drop", + "seat:node-build-image.drop" ], "uses": [ "operator-channel" From d9a9da921a31a878bf7044c7a6484c9a0af213f3 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:07:38 +0100 Subject: [PATCH 6/6] RN12: the runner runs on request, its pipelines placed from the run's candidate revision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit run writes each member's pipeline at once under the run's random id and its named branch and says run-started; the candidate run-'s catalog-revision, handed to the runner by meshcatalogclient's candidates hook, is read whole through the revision verb and places each member's steps and a pipeline of each entry the members change, saying pipeline-started once each. Builds: design 47 G §64, §373, §420, A-160, A-180, A-185, A-193, A-199 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../clients/meshcatalogclient/candidates.go | 36 +++ .../meshcatalogclient/catalog_view_test.go | 26 ++ .../cmd/mesh-delivery-pipeline/candidate.go | 43 ++++ .../mesh-delivery-pipeline/candidate_test.go | 36 +++ .../cmd/mesh-delivery-pipeline/main.go | 23 ++ .../cmd/mesh-delivery-pipeline/verbs.go | 5 +- .../cmd/mesh-delivery-pipeline/verbs_test.go | 2 +- .../cmd/mesh-delivery-pipeline/views.go | 8 +- .../internal/pipeline/run.go | 239 ++++++++++++++++++ .../internal/pipeline/run_test.go | 124 +++++++++ .../internal/pipeline/start.go | 5 +- .../internal/pipeline/verbs.go | 9 + 12 files changed, 551 insertions(+), 5 deletions(-) create mode 100644 modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidates.go create mode 100644 modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go create mode 100644 modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/run.go create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/run_test.go diff --git a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidates.go b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidates.go new file mode 100644 index 00000000..5ca67701 --- /dev/null +++ b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidates.go @@ -0,0 +1,36 @@ +package meshcatalogclient + +import ( + "context" + "encoding/json" + "fmt" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// WireCandidatesProcess hands each candidate's `catalog-revision` to hook (design 47 G A-185, A-193, A-199): a candidate's +// revision no Catalog keeps, heard once on the event's one subscription of the process's reader, beside every watched +// branch's Catalog and whether or not the part follows any, so the runner places a run's pipelines on it. A trunk revision +// is passed over (each branch's Catalog keeps it); data that is no such event is an error. The part's manifest consumes the +// seat's event catalog-revision. +func WireCandidatesProcess(r *views.ProcessReader, hook func(ctx context.Context, ev seat.CatalogRevisionEvent) error) error { + if hook == nil { + return fmt.Errorf("meshcatalogclient: a candidates hook is a function") + } + return r.AdoptEvent(seat.Seat, seat.EventCatalogRevision, candidateHear(hook)) +} + +// candidateHear is the handler of catalog-revision that hands a candidate's revision to hook. +func candidateHear(hook func(ctx context.Context, ev seat.CatalogRevisionEvent) error) func(context.Context, []byte) error { + return func(ctx context.Context, data []byte) error { + var ev seat.CatalogRevisionEvent + if err := json.Unmarshal(data, &ev); err != nil || ev.Revision <= 0 { + return fmt.Errorf("meshcatalogclient: not a %s event: %s", seat.EventCatalogRevision, clipBytes(data)) + } + if ev.Kind != KindCandidate { + return nil + } + return hook(ctx, ev) + } +} diff --git a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/catalog_view_test.go b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/catalog_view_test.go index 6c74285e..760d8bfc 100644 --- a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/catalog_view_test.go +++ b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/catalog_view_test.go @@ -162,3 +162,29 @@ func TestCatalogIsAViewPerWatchedBranch(t *testing.T) { t.Error("a Catalog of no branch was made") } } + +// A-185, A-193, A-199: a candidate's catalog-revision reaches the hook, adopted on the event's one subscription of the +// reader, also with no watched branch's Catalog; a trunk revision does not, and data that is no such event is said. +func TestCandidatesReachTheHook(t *testing.T) { + var hooked []seat.CatalogRevisionEvent + hook := func(_ context.Context, ev seat.CatalogRevisionEvent) error { hooked = append(hooked, ev); return nil } + r := &views.ProcessReader{Module: "runner"} + if err := WireCandidatesProcess(r, hook); err != nil { + t.Fatal(err) + } + if err := WireCandidatesProcess(r, nil); err == nil { + t.Error("a nil hook was taken") + } + hear := candidateHear(hook) + for _, data := range [][]byte{revisionEvent("main", KindTrunk, 3), revisionEvent("main", KindCandidate, 4)} { + if err := hear(context.Background(), data); err != nil { + t.Fatal(err) + } + } + if err := hear(context.Background(), []byte(`not json`)); err == nil { + t.Error("data that is no catalog-revision was taken") + } + if len(hooked) != 1 || hooked[0].Revision != 4 || hooked[0].Kind != KindCandidate { + t.Errorf("the hook was handed %+v, want the candidate's revision 4 alone", hooked) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go new file mode 100644 index 00000000..81fb00b2 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go @@ -0,0 +1,43 @@ +package main + +import ( + "encoding/json" + "fmt" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/buscalls" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// readCandidate reads a candidate's revision whole through the catalog seat's verb revision, with the candidate's group and +// heads and the revision its event names, page by page (design 47 G A-180). +func readCandidate(call func(key string, body any) (json.RawMessage, error), ev catalog.CatalogRevisionEvent) (pipeline.Half, error) { + half := pipeline.Half{Revision: ev.Revision, Source: ev.Source, Entries: map[string]catalog.CatalogEntry{}} + after := "" + for { + raw, err := call(meshcatalogclient.RevisionVerb, catalog.RevisionInput{Branch: ev.Branch, Group: ev.Group, Heads: ev.Heads, Revision: ev.Revision, After: after}) + if err != nil { + return half, err + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return half, fmt.Errorf("%s answered: %s", meshcatalogclient.RevisionVerb, said) + } + var a catalog.RevisionAnswer + if err := json.Unmarshal(answer, &a); err != nil { + return half, fmt.Errorf("%s answered no revision: %w", meshcatalogclient.RevisionVerb, err) + } + if a.Revision == nil || a.Revision.Revision != ev.Revision || a.Revision.Group != ev.Group { + return half, fmt.Errorf("%s answered %+v for the candidate %s's revision %d", meshcatalogclient.RevisionVerb, a.Revision, ev.Group, ev.Revision) + } + for _, e := range a.Entries { + half.Entries[e.Module] = e + } + if a.Next == "" { + return half, nil + } + after = a.Next + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go new file mode 100644 index 00000000..21efca52 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go @@ -0,0 +1,36 @@ +package main + +import ( + "encoding/json" + "testing" + + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" +) + +// A-180: a run's candidate revision is read whole through the seat's revision verb with the candidate's group, heads and +// revision, page by page; an answer of another revision is refused. +func TestReadCandidateReadsTheRevisionWhole(t *testing.T) { + ev := catalog.CatalogRevisionEvent{Kind: "candidate", Group: "run-0a1b2c3d4e5f", Heads: "h", Revision: 12, Branch: "main"} + var asked []catalog.RevisionInput + answerRev := int64(12) + call := func(key string, body any) (json.RawMessage, error) { + in := body.(catalog.RevisionInput) + asked = append(asked, in) + a := catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: answerRev, Kind: "candidate", Group: in.Group}} + if in.After == "" { + a.Entries, a.Next = []catalog.CatalogEntry{{Module: "app", Fp: "src2:a"}}, "app" + } else { + a.Entries = []catalog.CatalogEntry{{Module: "lib", Fp: "src2:l"}} + } + return json.Marshal(a) + } + half, err := readCandidate(call, ev) + if err != nil || len(half.Entries) != 2 || half.Revision != 12 || len(asked) != 2 || asked[1].After != "app" || + asked[0].Group != ev.Group || asked[0].Heads != "h" || asked[0].Revision != 12 || asked[0].Branch != "main" { + t.Fatalf("read %+v (%v), asked %+v", half, err, asked) + } + answerRev = 13 + if _, err := readCandidate(call, ev); err == nil { + t.Error("an answer of another revision was taken") + } +} 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 1bfb5453..b5b04915 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -11,11 +11,13 @@ import ( "log" "os" "os/signal" + "strings" "sync/atomic" "syscall" "github.com/nats-io/nats.go" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-sdk/go/process" @@ -81,10 +83,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // 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)] + var placer atomic.Pointer[func(catalog.CatalogRevisionEvent) error] rv, err := startViews(ctx, js, conn, cred, hub, calls.State("findings"), calls.Call, func(branch string) { if f := resumer.Load(); f != nil { (*f)(branch) } + }, func(_ context.Context, ev catalog.CatalogRevisionEvent) error { + if f := placer.Load(); f != nil { + return (*f)(ev) + } + return fmt.Errorf("the candidate %s's revision %d is heard before the runner can place it", ev.Group, ev.Revision) }, logger.Printf) if err != nil { return err @@ -119,6 +127,21 @@ 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) + starter.Store = pipelineStore{ctx: ctx, kv: pipelines} + // A run's candidate revision places its pipelines' steps, its entries read through the seat's revision verb (A-180). + place := func(ev catalog.CatalogRevisionEvent) error { + if !strings.HasPrefix(ev.Group, pipeline.RunGroupPrefix) { + return nil + } + half, err := readCandidate(calls.CallKept, ev) + if err != nil { + return fmt.Errorf("the candidate %s's revision %d is not read, so its run is not placed: %w", ev.Group, ev.Revision, err) + } + n, err := starter.PlaceRun(ev, half, rv.Catalog.View) + logger.Printf("the candidate %s's revision %d placed %d pipelines of its run", ev.Group, ev.Revision, n) + return err + } + placer.Store(&place) // A newer commit or publish supersedes the older running pipelines it replaces, their asks dropped as kept calls (§31, // §350; A-182, A-202). starter.Supersede = func(newer seat.Pipeline) { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go index 1be958c2..cecbacbd 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go @@ -38,10 +38,11 @@ func answerVerbs(conn *process.Conn, v pipeline.Verbs) (func(), error) { return conn.AnswerSeat(seat.Seat, writeHandlers(v)) } -// writeHandlers are the seat's write verbs the process answers: review, rerun, accept, cancel, start and ask-again -// (§306; RN14). +// writeHandlers are the seat's write verbs the process answers: run, review, rerun, accept, cancel, start and ask-again +// (§64, §306; RN12, RN14). func writeHandlers(v pipeline.Verbs) map[string]process.SeatHandler { return map[string]process.SeatHandler{ + seat.VerbRun: verbOf(func(c string, in seat.RunInput) (any, error) { return v.Run(c, in) }), seat.VerbReview: verbOf(func(c string, in seat.ReviewInput) (any, error) { return v.Review(c, in) }), seat.VerbRerun: verbOf(func(c string, in seat.RerunInput) (any, error) { return v.Rerun(c, in) }), seat.VerbAccept: verbOf(func(c string, in seat.AcceptInput) (any, error) { return v.Accept(c, in) }), diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go index 6b1db4e8..b535357a 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go @@ -18,7 +18,7 @@ func TestEveryWriteVerbIsServed(t *testing.T) { got = append(got, v) } slices.Sort(got) - want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbReview, seat.VerbStart} + want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbReview, seat.VerbStart, seat.VerbRun} slices.Sort(want) if !slices.Equal(got, want) { t.Errorf("the process answers %v, want %v", got, want) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go index 363c34db..eca36ef1 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go @@ -25,7 +25,8 @@ type runnerViews struct { // startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or // after, then start hub once. func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cred process.Account, hub *process.EventHub, findingsState process.State, - call meshcatalogclient.Call, onReady func(branch string), logf func(string, ...any)) (*runnerViews, error) { + call meshcatalogclient.Call, onReady func(branch string), candidate func(ctx context.Context, ev catalog.CatalogRevisionEvent) error, + logf func(string, ...any)) (*runnerViews, error) { findings, err := findingWriter(ctx, cred, conn, js, findingsState, logf) if err != nil { return nil, err @@ -52,6 +53,11 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr if err != nil { return nil, err } + // A candidate's revision, heard once on the event's one subscription whatever branches are followed, places a run's + // pipelines (A-185, A-193, A-199). + if err := meshcatalogclient.WireCandidatesProcess(reader, candidate); err != nil { + return nil, err + } reader.Start(ctx, js, cred, status) return &runnerViews{Registrations: regs, Catalog: cv}, nil } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/run.go b/modules/mesh-delivery-pipeline/internal/pipeline/run.go new file mode 100644 index 00000000..bd7a68f8 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/run.go @@ -0,0 +1,239 @@ +package pipeline + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "maps" + "slices" + "strings" + "time" + + runnerclient "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// RunGroupPrefix is a run on request's candidate group before its id, `run-` (design 47 G A-160). +const RunGroupPrefix = "run-" + +// RunKey is the key of a run on request's pipeline, `..`: the run's id under the member's named branch +// (design 47 G §420, A-193). +func RunKey(branch, run, module string) string { + return seat.PipelinesByBranchRevisionModuleKey(keys.EscapeName(branch), run, module) +} + +// NewRunID is a run's id, a random 12-hex id, so two runs of one commit are two pipelines (A-180). +func NewRunID() (string, error) { + b := make([]byte, 6) + if _, err := rand.Read(b); err != nil { + return "", err + } + return hex.EncodeToString(b), nil +} + +// Run starts a run on request (design 47 G §64, §420; A-180, A-193): it writes each member's pipeline at once, running and +// with no steps, its `run` the run's new id, and says run-started, from which the indexer builds the run's candidate; the +// candidate's revision places the steps (PlaceRun). A member naming no module, commit or branch is refused, the run then +// writing nothing. caller is the one the bus proved. newID makes the run's id (NewRunID). +func (s *Starter) Run(caller string, in seat.RunInput, newID func() (string, error)) (seat.RunAnswer, error) { + if len(in.Members) == 0 { + return seat.RunAnswer{}, errors.New("a run names one or more members, each a module at a commit on a branch") + } + for _, m := range in.Members { + if m.Module == "" || m.Commit == "" || m.Branch == "" { + return seat.RunAnswer{}, fmt.Errorf("a run's member names its module, commit and branch: %+v", m) + } + } + id, err := newID() + if err != nil { + return seat.RunAnswer{}, fmt.Errorf("the run's id is not made: %w", err) + } + now := s.now() + var written []string + for _, m := range in.Members { + key := RunKey(m.Branch, id, m.Module) + p := seat.Pipeline{V: 1, Pipeline: key, Branch: m.Branch, Module: m.Module, Commit: m.Commit, Run: id, + Outcome: seat.PipelineOutcomeRunning, Started: &now, Steps: []seat.StepResult{}} + raw, err := json.Marshal(p) + if err == nil { + err = s.Pipelines.Create(key, raw) + } + if err != nil { + s.failRun(written, "the run's other pipelines were not written") + return seat.RunAnswer{}, fmt.Errorf("the pipeline %s of the run is not written: %w", key, err) + } + written = append(written, key) + } + first := in.Members[0] + last := first.Commit + if len(in.Members) > 1 { + last = RunGroupPrefix + id + } + ev := seat.RunStarted{V: 1, At: &now, By: caller, Members: in.Members, Run: id} + if err := s.Say(runnerclient.Subject(seat.EventRunStarted, first.Module, first.Branch, last), ev); err != nil { + s.failRun(written, "run-started was not said, so no candidate is built for it") + return seat.RunAnswer{}, fmt.Errorf("the run %s is not started: run-started is not said: %w", id, err) + } + return seat.RunAnswer{Run: id, Pipelines: written}, nil +} + +// failRun ends the run's pipelines written before its start failed, failed and saying why, so none stands running. +func (s *Starter) failRun(written []string, why string) { + if s.Store == nil { + return + } + for _, key := range written { + for { + p, rev, err := s.Store.Get(key) + if err != nil || p.Outcome != seat.PipelineOutcomeRunning { + break + } + now := s.now() + p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeFailed, why, &now + p.EndCount++ + if _, err := s.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + s.logf("the pipeline %s of a run not started stays running: %v", key, err) + } + break + } + } +} + +// PlaceRun places a run's pipelines from its candidate's revision (design 47 G §64, §373, A-180, A-193): ev is the +// catalog-revision of the candidate `run-`, half its entries read whole through the seat's revision verb. Each member's +// pipeline of the run not yet placed takes its fingerprint, the candidate's revision and its steps, by compare-and-set, and +// says pipeline-started; each further entry whose fingerprint the members change (another than in base, the branch's catalog +// half, or absent there) gets a pipeline of the run of its own, placed the same way. Another candidate's revision places +// nothing; a revision heard again places nothing twice. It says how many pipelines it placed. +func (s *Starter) PlaceRun(ev catalog.CatalogRevisionEvent, half Half, base func(branch string) (Half, bool)) (int, error) { + id, ok := strings.CutPrefix(ev.Group, RunGroupPrefix) + if !ok || id == "" { + return 0, nil + } + if s.Store == nil { + return 0, errors.New("the runner holds no pipelines to place a run's in") + } + pkeys, err := s.Store.Keys("*." + id + ".*") + if err != nil { + return 0, fmt.Errorf("the pipelines of the run %s are not listed: %w", id, err) + } + slices.Sort(pkeys) + members := map[string]bool{} + branch := "" + for _, key := range pkeys { + if p, _, err := s.Store.Get(key); err == nil && p.Run == id { + members[p.Module] = true + if branch == "" { + branch = p.Branch + } + } + } + if branch == "" { + s.logf("the candidate %s's revision %d names a run with no pipeline here: nothing is placed", ev.Group, ev.Revision) + return 0, nil + } + placed := 0 + var first error + keep := func(ok bool, err error) { + if err != nil && first == nil { + first = err + } + if ok { + placed++ + } + } + for _, key := range pkeys { + keep(s.placeRunPipeline(key, id, ev.Revision, half)) + } + // The dependents (§373): the candidate's entries the members change, beside the members, on the first member's branch. + trunk, ok := base(branch) + if b, err := keys.UnescapeName(ev.Branch); err == nil && b != "" && b != branch { + if h, held := base(b); held { + trunk, ok = h, true + } + } + if !ok { + s.logf("the run %s's dependents are not known: the catalog half of %s is not held, so its members run alone", id, branch) + return placed, first + } + for _, module := range slices.Sorted(maps.Keys(half.Entries)) { + e := half.Entries[module] + if members[module] { + continue + } + if was, held := trunk.Entries[module]; held && was.Fp == e.Fp { + continue + } + key := RunKey(branch, id, module) + now := s.now() + p := seat.Pipeline{V: 1, Pipeline: key, Branch: branch, Module: module, Commit: e.Commit, Run: id, + Outcome: seat.PipelineOutcomeRunning, Started: &now, Steps: []seat.StepResult{}} + raw, err := json.Marshal(p) + if err == nil { + err = s.Pipelines.Create(key, raw) + } + if err != nil && !errors.Is(err, ErrExists) { + keep(false, fmt.Errorf("the dependent %s of the run %s is not written: %w", module, id, err)) + continue + } + keep(s.placeRunPipeline(key, id, ev.Revision, half)) + } + return placed, first +} + +// placeRunPipeline places one pipeline of the run, written and not yet placed (no revision), from the candidate's half. +func (s *Starter) placeRunPipeline(key, id string, revision int64, half Half) (bool, error) { + for { + p, rev, err := s.Store.Get(key) + if err != nil { + return false, fmt.Errorf("the pipeline %s is not read: %w", key, err) + } + if p.Run != id || p.Outcome != seat.PipelineOutcomeRunning || p.Revision != 0 { + return false, nil + } + now := s.now() + e, ok := half.Entries[p.Module] + if !ok { + s.logf("the candidate of the run %s holds no entry of %s: its pipeline is placed with no fingerprint", id, p.Module) + } + p.Fingerprint, p.Repository, p.Revision = e.Fp, e.Repository, revision + var asks []Asked + if s.Registration != nil { + if reg, ok := s.Registration(p.Module); ok { + p.Steps, asks = Place(p, reg, half.Entries, now, s.Name) + } else { + s.logf("%s has no registration yet: its pipeline is placed with no step", key) + } + } + if _, err := s.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return false, fmt.Errorf("the pipeline %s cannot be written: %w", key, err) + } + ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: p.Branch, Module: p.Module, Commit: p.Commit, + Fingerprint: p.Fingerprint, Revision: p.Revision, Run: id} + if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, p.Module, p.Branch, p.Commit), ev); err != nil { + s.logf("pipeline-started of %s is not said: %v", key, err) + } + if s.Steps != nil && len(asks) > 0 { + s.carry(key, asks) + if err := askAll(s.Steps, s.Arm, key, asks); err != nil { + return true, err + } + } + return true, nil + } +} + +func (s *Starter) now() time.Time { + if s.Now != nil { + return s.Now().UTC() + } + return time.Now().UTC() +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/run_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/run_test.go new file mode 100644 index 00000000..b14ffe56 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/run_test.go @@ -0,0 +1,124 @@ +package pipeline + +import ( + "encoding/json" + "errors" + "strings" + "testing" + + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// memCreated is memStore written by Create too, as the state pipelines is. +type memCreated struct{ memStore } + +func (m memCreated) Create(key string, value []byte) error { + if _, ok := m.memStore[key]; ok { + return ErrExists + } + var p seat.Pipeline + if err := json.Unmarshal(value, &p); err != nil { + return err + } + m.memStore[key] = struct { + p seat.Pipeline + rev uint64 + }{p, 1} + return nil +} + +// RN12 (design 47 G §64, §373, §420; A-180, A-193): `run` writes each member's pipeline at once under the run's id and its +// named branch and says run-started; the candidate run-'s revision places each member's steps and a pipeline of each +// entry the members change, says pipeline-started once each, and places nothing twice; a member naming no branch is refused. +func TestRunOnRequest(t *testing.T) { + reg, entries := placeFixture() + store := memCreated{memStore{}} + var said []string + asks := &stepAsks{} + s := &Starter{Pipelines: store, Store: store, Steps: asks, Name: "the-runner", + Registration: func(m string) (catalog.Registration, bool) { + r := reg + r.Module = m + return r, m == "app" || m == "lib" + }, + Say: func(subject string, body any) error { + said = append(said, subject[strings.Index(subject, ".event.")+7:]) + return nil + }} + id := func() (string, error) { return "0a1b2c3d4e5f", nil } + if _, err := s.Run("sessions", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c1"}}}, id); err == nil { + t.Error("a member naming no branch was taken") + } + if len(store.memStore) != 0 || len(said) != 0 { + t.Fatalf("a refused run wrote %v and said %v", store.memStore, said) + } + a, err := s.Run("sessions", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c1", Branch: "fix/1"}}}, id) + if err != nil || a.Run != "0a1b2c3d4e5f" || len(a.Pipelines) != 1 { + t.Fatalf("run = %+v, %v", a, err) + } + key := a.Pipelines[0] + if key != RunKey("fix/1", "0a1b2c3d4e5f", "app") { + t.Errorf("the member's pipeline is %s, want the run's id under its named branch", key) + } + if p, _, _ := store.Get(key); p.Run != a.Run || p.Outcome != seat.PipelineOutcomeRunning || len(p.Steps) != 0 || p.Branch != "fix/1" || p.Source != 0 { + t.Errorf("the member's pipeline %s = %+v", key, p) + } + if len(said) != 1 || !strings.HasPrefix(said[0], seat.EventRunStarted+".app.") || !strings.HasSuffix(said[0], ".c1") { + t.Errorf("said %v, want run-started on the member's module, branch and commit", said) + } + cand := entries + cand["app"] = catalog.CatalogEntry{Module: "app", Commit: "c1", Fp: "src2:new", Repository: "o/app", Artifacts: entries["app"].Artifacts} + cand["lib"] = catalog.CatalogEntry{Module: "lib", Commit: "c0", Fp: "src2:lib2", Artifacts: entries["app"].Artifacts} + cand["same"] = catalog.CatalogEntry{Module: "same", Fp: "src2:same"} + base := func(string) (Half, bool) { + return Half{Entries: map[string]catalog.CatalogEntry{"lib": {Fp: "src2:lib1"}, "same": {Fp: "src2:same"}, "mesh-build-go": entries["mesh-build-go"]}}, true + } + ev := catalog.CatalogRevisionEvent{Kind: "candidate", Group: "run-" + a.Run, Revision: 12} + if n, err := s.PlaceRun(catalog.CatalogRevisionEvent{Kind: "candidate", Group: "try-3", Revision: 11}, Half{Entries: cand}, base); n != 0 || err != nil { + t.Errorf("a try's candidate placed %d (%v)", n, err) + } + said = nil + n, err := s.PlaceRun(ev, Half{Revision: 12, Entries: cand}, base) + if err != nil || n != 2 { + t.Fatalf("the candidate placed %d (%v), want the member and the dependent lib", n, err) + } + p, _, _ := store.Get(key) + if p.Fingerprint != "src2:new" || p.Revision != 12 || len(p.Steps) == 0 { + t.Errorf("the member's pipeline placed = %+v", p) + } + dep, _, err := store.Get(RunKey("fix/1", a.Run, "lib")) + if err != nil || dep.Run != a.Run || dep.Fingerprint != "src2:lib2" || len(dep.Steps) == 0 { + t.Errorf("the dependent's pipeline = %+v (%v)", dep, err) + } + if _, _, err := store.Get(RunKey("fix/1", a.Run, "same")); err == nil { + t.Error("an entry the members do not change got a pipeline") + } + if len(said) != 2 || !strings.HasPrefix(said[0], seat.EventPipelineStarted+".") || len(asks.ids) == 0 { + t.Errorf("placing said %v and asked %d steps", said, len(asks.ids)) + } + if n, _ := s.PlaceRun(ev, Half{Revision: 12, Entries: cand}, base); n != 0 { + t.Errorf("the revision heard again placed %d", n) + } + // A restart asks a run's steps again from no branch's half: they are its candidate's. + if n, err := s.Resume(store, "fix/1", Half{Entries: entries}); n != 0 || err != nil { + t.Errorf("a restart asked %d steps of the run again (%v)", n, err) + } + + // Several members say run-started on the first member's module and branch and run-; a run-started not said fails the + // pipelines written, so none stands running. + said = nil + two, err := s.Run("operator", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c2", Branch: "main"}, {Module: "lib", Commit: "c3", Branch: "main"}}}, + func() (string, error) { return "ffffffffffff", nil }) + if err != nil || len(two.Pipelines) != 2 || len(said) != 1 || said[0] != seat.EventRunStarted+".app.main.run-ffffffffffff" { + t.Errorf("a run of two: %+v (%v), said %v", two, err, said) + } + s.Say = func(string, any) error { return errors.New("bus away") } + three, err := s.Run("operator", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c4", Branch: "main"}}}, func() (string, error) { return "eeeeeeeeeeee", nil }) + if err == nil || three.Run != "" { + t.Errorf("a run whose run-started was not said answered %+v", three) + } + if p, _, _ := store.Get(RunKey("main", "eeeeeeeeeeee", "app")); p.Outcome != seat.PipelineOutcomeFailed { + t.Errorf("its pipeline is %s, want failed", p.Outcome) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 0d8c561c..e874fce0 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -72,6 +72,8 @@ type Starter struct { Toolchain func(language string) (*nodebuildgo.ToolchainUsed, error) // Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none. Arm Arm + // Store is the state pipelines read and written by compare-and-set, where a run's pipelines are placed (PlaceRun). + Store Updater // Supersede ends the older running pipelines the pipeline just written supersedes (§31; Hearer.Supersede); nil ends none. Supersede func(newer seat.Pipeline) } @@ -156,7 +158,8 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) { } continue } - if p.Branch != branch || p.Outcome != seat.PipelineOutcomeRunning { + // A run on request's steps are placed from its candidate, not from the branch's half (PlaceRun). + if p.Branch != branch || p.Outcome != seat.PipelineOutcomeRunning || p.Run != "" { continue } reg, ok := s.Registration(p.Module) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go index e3caef80..6ede0c1b 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go @@ -335,6 +335,15 @@ func (v Verbs) askOf(p seat.Pipeline, s seat.StepResult, caller string, now time return nil, fmt.Errorf("the %s step of %s is placed no ask now", s.Step, p.Pipeline) } +// Run starts a run on request for the caller the bus proved (design 47 G §64; Starter.Run): its pipelines written, its +// run-started said, its steps placed when its candidate's revision is heard. +func (v Verbs) Run(caller string, in seat.RunInput) (seat.RunAnswer, error) { + if v.Starter == nil { + return seat.RunAnswer{}, errors.New("the runner cannot start a run: it holds no starter") + } + return v.Starter.Run(caller, in, NewRunID) +} + // Start starts by hand a delivering pipeline for the head of a watched branch the runner's catalog half holds, recorded in // the hand-act log first (§306, §394): the pipeline of that branch's source revision and the module, as a take-in would // start it. A head whose pipeline is written already starts nothing and is refused, naming the pipeline.