diff --git a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go index 7b0bfe66..87206148 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go @@ -27,6 +27,7 @@ const ( VerbAccept = "accept" VerbCancel = "cancel" VerbAskAgain = "ask-again" + VerbRelease = "release" ) const ( @@ -42,7 +43,7 @@ const ( // Verbs is every verb the seat serves, in its definition's order: a new copy each call. func Verbs() []string { - return []string{VerbRun, VerbStart, VerbShow, VerbList, VerbStands, VerbWhyCarried, VerbKept, VerbReview, VerbRerun, VerbAccept, VerbCancel, VerbAskAgain} + return []string{VerbRun, VerbStart, VerbShow, VerbList, VerbStands, VerbWhyCarried, VerbKept, VerbReview, VerbRerun, VerbAccept, VerbCancel, VerbAskAgain, VerbRelease} } // Events is every event published on the seat's subject: a new copy each call. @@ -347,6 +348,14 @@ const ( GroupEndedOutcomeCancelled = "cancelled" ) +// Held is a pipeline held for the new version of the runner: its steps are placed and asked when the node-engine seat says that fingerprint runs healthy (design 47 G 65, 266, 534, A-195). +type Held struct { + // the fingerprint of the runner's own module the pipeline awaits running healthy (design 47 G 266, 276, A-195) + Fingerprint string `json:"fingerprint"` + // since when the pipeline is held, RFC 3339 + Since time.Time `json:"since"` +} + // KeptResult is a package's passing result the runner keeps, by the digest of what its tests read (ADR 0666 section 8). type KeptResult struct { // sha256: of what the package's tests read @@ -427,6 +436,7 @@ type Pipeline struct { Fingerprint string `json:"fingerprint,omitempty"` // the gathered group, when the pipeline is its Group string `json:"group,omitempty"` + Held *Held `json:"held,omitempty"` Module string `json:"module"` Outcome string `json:"outcome"` // yes when the module is over its test budget (design 47 G section 419) @@ -691,7 +701,7 @@ type Wait struct { // for its end, how it ended: answered, ran, released, superseded How string `json:"how,omitempty"` Module string `json:"module"` - // what the step waits on (design 47 *Health and going back*) + // what the step waits on (design 47 *Health and going back*); the runner version is what a pipeline held for the runner's new version waits on (design 47 G 534, A-206) On string `json:"on"` // the pipeline's key, .., a branch escaped as a key's name is Pipeline string `json:"pipeline"` @@ -711,9 +721,10 @@ type Wait struct { // The values of Wait's on. const ( - WaitOnPerson = "person" - WaitOnBuildNode = "build node" - WaitOnRevisionView = "revision view" + WaitOnPerson = "person" + WaitOnBuildNode = "build node" + WaitOnRevisionView = "revision view" + WaitOnRunnerVersion = "runner version" ) // The values of Wait's step. @@ -950,6 +961,18 @@ type AskAgainAnswer struct { Result StepResult `json:"result"` } +// ReleaseInput is what release takes. +type ReleaseInput struct { + // why, written with the caller the runtime proves in the hand-act log + Why string `json:"why"` +} + +// ReleaseAnswer is what release answers. +type ReleaseAnswer struct { + // how many held pipelines were started + Released int64 `json:"released"` +} + // PipelineStartedEvent is the body of the event pipeline-started. type PipelineStartedEvent = PipelineStarted 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..60db0ab2 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json @@ -501,6 +501,35 @@ ] }, "class": "act" + }, + { + "name": "release", + "description": "The pipelines the runner holds for its new version started by hand with the running runner, whatever the new version's state; written in the hand-act log (design 47 G 276, A-206).", + "input": { + "type": "object", + "properties": { + "why": { + "type": "string", + "description": "why, written with the caller the runtime proves in the hand-act log" + } + }, + "required": [ + "why" + ] + }, + "answer": { + "type": "object", + "properties": { + "released": { + "type": "integer", + "description": "how many held pipelines were started" + } + }, + "required": [ + "released" + ] + }, + "class": "act" } ], "events": [ @@ -672,6 +701,25 @@ }, "description": "how many tests a step ran and carried, per level (ADR 0666 section 8)" }, + "Held": { + "type": "object", + "properties": { + "since": { + "type": "string", + "format": "date-time", + "description": "since when the pipeline is held, RFC 3339" + }, + "fingerprint": { + "type": "string", + "description": "the fingerprint of the runner's own module the pipeline awaits running healthy (design 47 G 266, 276, A-195)" + } + }, + "required": [ + "since", + "fingerprint" + ], + "description": "a pipeline held for the new version of the runner: its steps are placed and asked when the node-engine seat says that fingerprint runs healthy (design 47 G 65, 266, 534, A-195)" + }, "ToolchainUsed": { "type": "object", "properties": { @@ -985,6 +1033,9 @@ "type": "boolean", "description": "yes when the module is over its test budget (design 47 G section 419)" }, + "held": { + "$ref": "#/defs/Held" + }, "widens": { "type": "array", "items": { @@ -1430,9 +1481,10 @@ "enum": [ "person", "build node", - "revision view" + "revision view", + "runner version" ], - "description": "what the step waits on (design 47 *Health and going back*)" + "description": "what the step waits on (design 47 *Health and going back*); the runner version is what a pipeline held for the runner's new version waits on (design 47 G 534, A-206)" }, "who": { "type": "string", 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..7ca146f9 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts @@ -17,6 +17,7 @@ export declare const verbs: { readonly accept: "accept"; readonly cancel: "cancel"; readonly askAgain: "ask-again"; + readonly release: "release"; }; export declare const events: { @@ -144,6 +145,14 @@ export interface GroupEnded { "why"?: string; } +/** a pipeline held for the new version of the runner: its steps are placed and asked when the node-engine seat says that fingerprint runs healthy (design 47 G 65, 266, 534, A-195) */ +export interface Held { + /** the fingerprint of the runner's own module the pipeline awaits running healthy (design 47 G 266, 276, A-195) */ + "fingerprint": string; + /** since when the pipeline is held, RFC 3339 */ + "since": string; +} + /** a package's passing result the runner keeps, by the digest of what its tests read (ADR 0666 section 8) */ export interface KeptResult { /** sha256: of what the package's tests read */ @@ -189,6 +198,7 @@ export interface Pipeline { "fingerprint"?: string; /** the gathered group, when the pipeline is its */ "group"?: string; + "held"?: Held; "module": string; "outcome": "running" | "passed" | "failed" | "refused" | "superseded" | "cancelled"; /** yes when the module is over its test budget (design 47 G section 419) */ @@ -377,8 +387,8 @@ export interface Wait { /** for its end, how it ended: answered, ran, released, superseded */ "how"?: string; "module": string; - /** what the step waits on (design 47 *Health and going back*) */ - "on": "person" | "build node" | "revision view"; + /** what the step waits on (design 47 *Health and going back*); the runner version is what a pipeline held for the runner's new version waits on (design 47 G 534, A-206) */ + "on": "person" | "build node" | "revision view" | "runner version"; /** the pipeline's key, .., a branch escaped as a key's name is */ "pipeline": string; /** the catalog graph revision of the branch the pipeline runs for */ @@ -569,6 +579,18 @@ export interface AskAgainAnswer { "result": StepResult; } +/** is what release takes. */ +export interface ReleaseInput { + /** why, written with the caller the runtime proves in the hand-act log */ + "why": string; +} + +/** is what release answers. */ +export interface ReleaseAnswer { + /** how many held pipelines were started */ + "released": number; +} + /** is the body of the event pipeline-started. */ export type PipelineStartedEvent = PipelineStarted; diff --git a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js index a36f83ae..5b96596a 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js @@ -17,6 +17,7 @@ export const verbs = { accept: "accept", cancel: "cancel", askAgain: "ask-again", + release: "release", }; export const events = { 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..9e8adee7 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,21 +86,61 @@ 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) +} + +// 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 }} +} + +// WaitsTool is the runner's tool runner_waits: the pipelines held for its new version, with the fingerprint each awaits and since +// when (design 47 G §276, §539, A-195). +func WaitsTool(r pipeline.Reads) stdio.Tool { + return stdio.Tool{Name: Prefix + "_waits", + Description: "The pipelines the pipeline runner holds until its new version runs healthy: each with its pipeline, the " + + "fingerprint of the runner it awaits and since when.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { return r.Waits() }} +} + +// ReleaseTool is the runner's tool runner_release: it starts the pipelines held for the runner's new version with the running +// runner, through the seat's verb release, which the process that holds the pipelines answers and records in the hand-act log +// first (design 47 G §276, A-206). +func ReleaseTool(ask func(key string, body any) (json.RawMessage, error)) stdio.Tool { + return stdio.Tool{Name: Prefix + "_release", + Description: "Start now, with the running pipeline runner, the pipelines it holds until its new version runs healthy " + + "(runner_waits lists them), each wait ending released. Written in the hand-act log (mesh-hand-acts).", + Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{ + "why": map[string]any{"type": "string", "description": "why, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + in, err := as[seat.ReleaseInput](args) + if err != nil { + return nil, err + } + return ask("seat:"+seat.Seat+"."+seat.VerbRelease, in) + }} } 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, WaitsTool(r), ReleaseTool(stdio.Ask)) + 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 73d96db0..ef0bd12b 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 @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "fmt" "strings" "testing" @@ -17,7 +18,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 +27,19 @@ 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" + // RN25 (§468, §76; A-164): the health check is the bundle's runner_health. + names = append(names, HealthTool(none{}, none{}, nil).Name) + // A-206: the tool runner_release asks the seat's verb release with its why. + var asked string + rel := ReleaseTool(func(key string, body any) (json.RawMessage, error) { + asked = key + fmt.Sprintf(" %+v", body) + return json.RawMessage(`{"released":1}`), nil + }) + if _, err := rel.Run(map[string]any{"why": "w"}); err != nil || asked != "seat:mesh-pipeline-runner.release {Why:w}" { + t.Errorf("runner_release asked %q, %v", asked, err) + } + names = append(names, rel.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 runner_release" 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/buildends.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends.go new file mode 100644 index 00000000..fbb5149c --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends.go @@ -0,0 +1,44 @@ +package main + +import ( + "context" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// hearBuildEnds hears the builder's stream of build ends, kept last per subject and read whole at the start, then followed +// (design 47 G A-181, §556): when the build of the runner's new version ended is the time the wait's bound counts from, and +// each end heard for a runner held pipelines await starts their clock (§266, A-206). A stream it cannot hear is said, and the +// bound of such a wait is not armed until the runner starts again. +func hearBuildEnds(ctx context.Context, cred process.Account, js jetstream.JetStream, ends *pipeline.BuildEnds, st *pipeline.Starter, store pipeline.Updater, + logf func(string, ...any)) { + stream := catalogue.StreamBusName(builds.BuildsStream) + cc, err := cred.HearStream(ctx, js, stream, func(m jetstream.Msg) { + defer m.Ack() + md, err := m.Metadata() + if err != nil { + logf("%s has no time the stream stamped: %v", m.Subject(), err) + return + } + module, fp, ok := ends.Heard(m.Data(), md.Timestamp) + if !ok { + return + } + if n, err := st.Clocked(store, module, fp); err != nil { + logf("the clock of the %d pipelines held for %s: %v", n, fp, err) + } else if n > 0 { + logf("the build of %s ended: the bound of the %d pipelines held for it counts from it", fp, n) + } + }, logf) + if err != nil { + logf("the stream %s is not heard, so no wait for the runner's new version has its bound armed until the runner starts again: %v", stream, err) + return + } + go func() { <-ctx.Done(); cc.Stop() }() +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends_integration_test.go new file mode 100644 index 00000000..cb17252d --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends_integration_test.go @@ -0,0 +1,76 @@ +//go:build integration + +package main + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/testbus" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// The runner hears the builder's stream of build ends, kept last per subject, read whole at its start: the end of its new version's +// build, published before it started, starts the clock of the pipelines held for it, counted from the time the stream stamped the end +// (design 47 G A-181, §266, A-206). +func TestTheRunnerHearsTheBuildEndOfItsNewVersionAtItsStart(t *testing.T) { + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + stream, subject := catalogue.StreamBusName(builds.BuildsStream), process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEvent+".>") + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + published := time.Now().UTC() + if _, err := js.Publish(ctx, process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEventName("mesh-delivery-pipeline", "src2:new")), + []byte(`{"module":"mesh-delivery-pipeline","fingerprint":"src2:new","outcome":"done"}`)); err != nil { + t.Fatal(err) + } + cred := process.Account{Node: "n1", Module: "mesh-delivery-pipeline"} + durable := cred.SeatEventConsumerConfig([]string{subject}, true) + durable.DeliverPolicy = jetstream.DeliverNewPolicy + if _, err := js.CreateOrUpdateConsumer(ctx, stream, durable); err != nil { + t.Fatal(err) + } + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "pipelines", History: 1}) + if err != nil { + t.Fatal(err) + } + store := pipelineStore{ctx: ctx, kv: kv} + held, _ := json.Marshal(seat.Pipeline{Pipeline: "main.4.app", Branch: "main", Module: "app", Outcome: seat.PipelineOutcomeRunning, + Held: &seat.Held{Since: published, Fingerprint: "src2:new"}}) + if _, err := kv.Create(ctx, "main.4.app", held); err != nil { + t.Fatal(err) + } + ends := &pipeline.BuildEnds{} + armed := make(chan time.Time, 1) + st := &pipeline.Starter{Self: "mesh-delivery-pipeline", Ends: ends.At, Say: func(string, any) error { return nil }, + Hold: func(fp string, from time.Time) error { armed <- from; return nil }} + hearBuildEnds(ctx, cred, js, ends, st, store, t.Logf) + select { + case from := <-armed: + if from.Before(published.Add(-time.Minute)) || from.After(time.Now().Add(time.Second)) { + t.Errorf("the bound counts from %s, not from the end published at %s", from, published) + } + case <-ctx.Done(): + t.Fatal("the end of the new runner's build was not heard at the start") + } +} 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..288ab266 --- /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" + + "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. +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 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 + ledger *process.ConsumerLedger +} + +func (a agedHub) Handle(pattern string, handle func(jetstream.Msg)) error { + return a.hub.Handle(pattern, func(m jetstream.Msg) { + a.ledger.Delivered(m) + handle(&agedMsg{Msg: m, ledger: a.ledger}) + }) +} + +// agedMsg is a message whose acknowledgement or end is noted applied on its consumer's ledger, once. +type agedMsg struct { + jetstream.Msg + ledger *process.ConsumerLedger + once sync.Once +} + +func (m *agedMsg) applied() { m.once.Do(func() { m.ledger.Applied(m.Msg, time.Time{}) }) } + +func (m *agedMsg) Ack() error { defer m.applied(); return m.Msg.Ack() } + +func (m *agedMsg) DoubleAck(ctx context.Context) error { + defer m.applied() + return m.Msg.DoubleAck(ctx) +} + +func (m *agedMsg) Term() error { defer m.applied(); return m.Msg.Term() } + +func (m *agedMsg) TermWithReason(r string) error { defer m.applied(); return m.Msg.TermWithReason(r) } + +// 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 new file mode 100644 index 00000000..91ddcd8c --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go @@ -0,0 +1,62 @@ +package main + +import ( + "context" + "encoding/json" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + "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 + 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 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 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 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{seq: 1}) + if inside.OldestUnacknowledged.IsZero() { + t.Fatalf("inside the handler the consumer's facts = %+v, want its delivery written", inside) + } + 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/findingwriter.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go index 0fedcad4..b8d2d35e 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go @@ -2,6 +2,8 @@ package main import ( "context" + "fmt" + "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -9,12 +11,32 @@ import ( condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" ) +// findingsWriter is the runner's finding writer as the mesh-sdk makes it. +type findingsWriter = views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent] + +// KindPipelineWaits is the finding of pipelines held for the runner's new version that the wait's bound started with the running +// runner (design 47 G §266, A-206). +const KindPipelineWaits = "pipeline-waits" + +// lateFinding is the finding pipeline-waits of one pipeline the bound's expiry started: scope module, id its key +// `..`, naming the fingerprint of the new runner it waited for and since when. +func lateFinding(p seat.Pipeline, held seat.Held, runner, node string, at time.Time) condseat.Finding { + return condseat.Finding{Scope: views.FindingScope, ID: p.Pipeline, Kind: KindPipelineWaits, Severity: condseat.FindingSeverityWarning, + Headline: "the new pipeline runner did not run healthy in time", + Summary: fmt.Sprintf("the pipeline %s was held since %s for the runner at %s to run healthy; the bound passed and the running runner runs it", + p.Pipeline, held.Since.UTC().Format(time.RFC3339), held.Fingerprint), + Source: "bound of the wait for the new runner", Observer: runner, At: at, + Subject: condseat.Subject{Scope: views.FindingScope, ID: p.Module, Machine: node}} +} + // findingWriter is the runner's mesh-sdk finding writer (design 47 G §461, §689): the runner observes only what its views // reader finds, so its own observe states none and the writer's statement covers the reader's open findings. The writer // forms its subjects itself (design 47 G §383, A-13). -func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) { +func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*findingsWriter, error) { w, err := views.NewWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ Module: cred.Module, State: state, Node: cred.Node, Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf, Observe: func(context.Context) (views.Observed[condseat.Finding], error) { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go new file mode 100644 index 00000000..8d9dd3fc --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go @@ -0,0 +1,29 @@ +package main + +import ( + "strings" + "testing" + "time" + + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/views" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// The bound's expiry's finding is pipeline-waits, scope module, id the pipeline's key, naming the awaited fingerprint and since +// when (design 47 G §266, A-206); it passes the writer's own check of a finding. +func TestLateFindingIsPipelineWaitsOfTheModuleScope(t *testing.T) { + since := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC) + f := lateFinding(seat.Pipeline{Pipeline: "main.4.app", Module: "app", Branch: "main"}, seat.Held{Since: since, Fingerprint: "src2:new"}, + "mesh-delivery-pipeline", "n1", since.Add(30*time.Minute)) + if f.Kind != "pipeline-waits" || f.Scope != views.FindingScope || f.ID != "main.4.app" || f.Severity != condseat.FindingSeverityWarning { + t.Errorf("finding %+v", f) + } + if !strings.Contains(f.Summary, "src2:new") || !strings.Contains(f.Summary, "2026-11-11T12:00:00Z") { + t.Errorf("summary %q", f.Summary) + } + if err := views.CheckFinding(f); err != nil { + t.Error(err) + } +} 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..f6b1d924 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -13,6 +13,7 @@ import ( "os/signal" "sync/atomic" "syscall" + "time" "github.com/nats-io/nats.go" @@ -71,6 +72,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer calls.Close() hub := process.NewEventHub(logger.Printf) + // 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 + } + 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)] @@ -96,14 +104,29 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask) // Its asking half (§22, §24): a pipeline's steps are placed from the module's registration and each build seat is asked // for its step. + ends := &pipeline.BuildEnds{} starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say, Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module, - Arm: armBound(ctx, bounds), Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)} + Arm: armBound(ctx, bounds), Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains), + // The runner's own module and fingerprint (A-149): a take-in giving it another holds the pipelines it starts (§65, A-195). + Self: cred.Module, SelfFp: getenv("MESH_FINGERPRINT"), Awaiting: pipeline.Awaiting(pipelineStore{ctx: ctx, kv: pipelines}), + Ends: ends.At, Hold: armHeld(ctx, bounds), + // The bound's expiry writes the finding pipeline-waits for each pipeline it started (§266, A-206). + Late: func(p seat.Pipeline, held seat.Held) { + module, node := rv.Findings.ModuleNode() + if err := rv.Findings.Write(ctx, lateFinding(p, held, module, node, time.Now().UTC())); err != nil { + logger.Printf("the finding pipeline-waits of %s is not written: %v", p.Pipeline, err) + } + }} resume := func(branch string) { half, ok := rv.Catalog.View(branch) if !ok { return } + // Pipelines held for the fingerprint this runner runs are started now: it is the new runner (§19, A-195). + if n, err := starter.Release(pipelineStore{ctx: ctx, kv: pipelines}, starter.SelfFp, "ran", rv.Catalog.View); err != nil { + logger.Printf("the pipelines held for the runner at %s: %d released, then %v", starter.SelfFp, n, err) + } n, err := starter.Resume(pipelineStore{ctx: ctx, kv: pipelines}, branch, half) if err != nil { logger.Printf("the steps placed on %s and not yet asked: %d asked, then %v", branch, n, err) @@ -112,25 +135,36 @@ 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 { + store := pipelineStore{ctx: ctx, kv: pipelines} + if err := watchBounds(ctx, bounds, hearer, func(key string) { + if n, err := starter.ReleaseExpired(store, key, rv.Catalog.View); err != nil { + logger.Printf("the pipelines held for the runner, whose %s passed: %d released, then %v", key, n, err) + } else { + logger.Printf("the runner's new version did not run healthy within %s: %d held pipelines started with the running runner", pipeline.HeldBound, n) + } + }, logger.Printf); err != nil { return err } - if err := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil { + hearBuildEnds(ctx, cred, js, ends, starter, store, logger.Printf) + if err := hearRunning(ctx, heard, cred.Module, starter, store, bounds, rv.Catalog.View, logger.Printf); err != nil { return err } - if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil { + if err := hearDecided(heard, hearer, cred.Module, logger.Printf); err != nil { return err } - if err := hearTakenIn(hub, starter, rv.Catalog, logger.Printf); err != nil { + if err := hearPublished(ctx, heard, starter, calls, rv.Catalog, logger.Printf); err != nil { + return err + } + 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..cc28c592 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) } @@ -74,7 +77,7 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes } // The seats the runner calls, answered as their holders do. - reg := catalog.Registration{Module: "app", Branches: []string{"main"}, Origin: "o/app", Policy: catalog.Policy{Steps: catalog.StepStatuses{Review: "optional"}}} + reg := catalog.Registration{Module: "app", Branches: []string{"main"}, Origin: "o/app", Policy: catalog.Policy{Steps: catalog.StepStatuses{Review: "optional", Composition: "bypass", Approval: "bypass"}}} answer := func(subject string, body any) { if _, err := nc.Subscribe(subject, func(m *nats.Msg) { raw, _ := json.Marshal(map[string]any{"result": body}) @@ -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 facts process.ConsumerFacts + waitFor(t, ctx, logs, func() bool { + e, err := consumers.Get(ctx, "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) + } +} + +// 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..58f0ad0f 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -6,9 +6,11 @@ import ( "errors" "fmt" "strings" + "time" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" 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" @@ -99,7 +101,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 +139,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) { @@ -172,9 +174,54 @@ func armBound(ctx context.Context, bounds jetstream.KeyValue) pipeline.Arm { } } +// armHeld is the bound of the pipelines held for the runner's new version: a key of the state bounds named for the awaited +// fingerprint, with its own time to live of 30 minutes counted from the end of that runner's build (design 47 G §266, A-195, A-77, +// A-206). +func armHeld(ctx context.Context, bounds jetstream.KeyValue) func(string, time.Time) error { + return func(fp string, from time.Time) error { + _, err := bounds.Create(ctx, pipeline.HeldKey(fp), []byte(`{"fingerprint":"`+fp+`"}`), jetstream.KeyTTL(pipeline.HeldTTL(from, time.Now()))) + if errors.Is(err, jetstream.ErrKeyExists) { + return nil + } + return err + } +} + +// hearRunning releases the pipelines held for the runner's new version when the node-engine seat says the runner's own module +// runs that fingerprint healthy, on any node (§266, §276, A-195); its bound is then deleted. A running with no health reads +// as healthy (§520). Heard again when the release could not be written. +func hearRunning(ctx context.Context, hub handlers, self string, st *pipeline.Starter, store pipeline.Updater, bounds jetstream.KeyValue, + view func(string) (pipeline.Half, bool), logf func(string, ...any)) error { + return hub.Handle(nodeengine.SubjectRunningAll, func(m jetstream.Msg) { + var ev nodeengine.Running + if err := json.Unmarshal(m.Data(), &ev); err != nil { + logf("%s is no running: %v", m.Subject(), err) + _ = m.Term() + return + } + if ev.Module != self || ev.Stopped || ev.Fingerprint == "" || (ev.Healthy != nil && !*ev.Healthy) { + _ = m.Ack() + return + } + n, err := st.Release(store, ev.Fingerprint, "ran", view) + if err != nil { + logf("the pipelines held for %s: %d released, then %v; heard again", ev.Fingerprint, n, err) + _ = m.Nak() + return + } + if n > 0 { + logf("the runner runs %s on %s: %d held pipelines released", ev.Fingerprint, ev.Node, n) + if err := bounds.Delete(ctx, pipeline.HeldKey(ev.Fingerprint)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + logf("the bound of the wait for %s is not deleted: %v", ev.Fingerprint, err) + } + } + _ = m.Ack() + }) +} + // watchBounds hears the end of each ask's bound, the bus's marker of the key's expiry, and ends the step no holder took (§32, // §535, A-77): no timer in the process. -func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hearer, logf func(string, ...any)) error { +func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hearer, release func(key string), logf func(string, ...any)) error { w, err := bounds.WatchAll(ctx, jetstream.UpdatesOnly()) if err != nil { return err @@ -189,6 +236,10 @@ func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hea if !ok { return } + if e != nil && e.Operation() == jetstream.KeyValuePurge && pipeline.IsHeldKey(e.Key()) { + release(e.Key()) + continue + } if e == nil || e.Operation() != jetstream.KeyValuePurge || !pipeline.IsAskKey(e.Key()) { continue } @@ -252,7 +303,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/cmd/mesh-delivery-pipeline/verbs.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go index 6ddf5016..1517865c 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,24 @@ 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, ask-again and +// release (§306; RN14; A-206). +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) }), + seat.VerbRelease: verbOf(func(c string, in seat.ReleaseInput) (any, error) { return v.Release(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..61801856 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.VerbRelease, 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/cmd/mesh-delivery-pipeline/views.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go index 363c34db..1236cf75 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go @@ -20,6 +20,8 @@ import ( type runnerViews struct { Registrations *meshcatalogclient.Registrations Catalog *pipeline.Views + // Findings is the runner's finding writer, which also says the runner's own findings (A-206). + Findings *findingsWriter } // startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or @@ -53,7 +55,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr return nil, err } reader.Start(ctx, js, cred, status) - return &runnerViews{Registrations: regs, Catalog: cv}, nil + return &runnerViews{Registrations: regs, Catalog: cv, Findings: findings}, nil } // catalogView is a branch's meshcatalogclient.Catalog as the runner reads it. diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod index 9bfc48db..028117ac 100644 --- a/modules/mesh-delivery-pipeline/go.mod +++ b/modules/mesh-delivery-pipeline/go.mod @@ -11,9 +11,11 @@ require ( require ( git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildimage v0.0.0-00010101000000-000000000000 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/consumers.go b/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go new file mode 100644 index 00000000..d7a85fab --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go @@ -0,0 +1,59 @@ +package pipeline + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "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' 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 + +// 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.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 + 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 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.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))) + } + } + } + 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..8decf31c --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go @@ -0,0 +1,60 @@ +package pipeline + +import ( + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/process" +) + +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") } + +// 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 := process.ConsumerFacts{Name: name} + if since > 0 { + a.OldestUnacknowledged = now.Add(-since) + } + 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 event heard yet", mapKept{}, mapKept{}, true, "has heard no event yet"}, + } + 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/internal/pipeline/ended_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/ended_test.go new file mode 100644 index 00000000..765cb420 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/ended_test.go @@ -0,0 +1,60 @@ +package pipeline + +import ( + "strings" + "testing" + + 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" +) + +// pipeline-ended carries the module's test counts and says it over budget past 500 unit or 50 integration tests (design 47 G +// §24, §419), and a group's one pipeline ends with group-ended under the group once every member ended (§73, §332). +func TestPipelineEndedCarriesTestCountsAndAGroupEndsOnItsMembers(t *testing.T) { + h, store, _, _, _, _ := fixture() + bodies := map[string]any{} + var order []string + h.Say = func(subject string, body any) error { + ev := subject[strings.Index(subject, ".event.")+7:] + bodies[ev] = body + order = append(order, ev) + return nil + } + e := store[pkey] + e.p.Steps = e.p.Steps[:2] + e.p.Steps[1].Status = seat.StepResultStatusRequired + e.p.Group = "g1" + store[pkey] = e + // A second member of the same group, on the same source revision, still running. + store["main.5.lib"] = struct { + p seat.Pipeline + rev uint64 + }{seat.Pipeline{V: 1, Pipeline: "main.5.lib", Branch: "main", Source: 5, Module: "lib", Commit: "c2", Group: "g1", Outcome: seat.PipelineOutcomeRunning, + Steps: []seat.StepResult{{Step: seat.StepResultStepLint, Status: seat.StepResultStatusRequired, Outcome: seat.StepResultOutcomeRunning, Ask: "b1", Seat: nodebuildgo.Seat}}}, 1} + + lint := ended("a1", "lint", "passed") + unit := ended("a2", "unit", "passed") + unit.Tests = &nodebuildgo.TestCounts{Unit: 501, Integration: 0} + for _, ev := range []StepEnded{lint, unit} { + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ev); err != nil { + t.Fatal(err) + } + } + pe, ok := bodies["pipeline-ended.app.main.g1"].(seat.PipelineEnded) + if !ok || pe.Tests == nil || pe.Tests.Unit != 501 || !pe.OverBudget { + t.Fatalf("pipeline-ended = %+v (said %v)", bodies["pipeline-ended.app.main.g1"], order) + } + if _, said := bodies["group-ended.main.g1"]; said { + t.Fatalf("group-ended said while a member runs: %v", order) + } + lib := ended("b1", "lint", "passed") + lib.Pipeline, lib.Module, lib.Commit, lib.Fingerprint = "main.5.lib", "lib", "c2", "" + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", lib); err != nil { + t.Fatal(err) + } + ge, ok := bodies["group-ended.main.g1"].(seat.GroupEnded) + if !ok || ge.Outcome != seat.GroupEndedOutcomePassed || ge.Group != "g1" || ge.Branch != "main" || ge.Source != 5 || len(ge.Pipelines) != 2 { + t.Fatalf("group-ended = %+v (said %v)", bodies["group-ended.main.g1"], order) + } + // A pipeline of no group says no group-ended. +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/ends.go b/modules/mesh-delivery-pipeline/internal/pipeline/ends.go new file mode 100644 index 00000000..cb7ae9e1 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/ends.go @@ -0,0 +1,46 @@ +package pipeline + +import ( + "encoding/json" + "sync" + "time" +) + +// BuildEnds are the done builds' ends heard from the builder's stream of build ends (event build.ended, kept last per subject +// and read whole at the start, design 47 G A-181): when each build, by module and fingerprint, ended, which is the time a held +// wait's bound counts from (§266, A-206). +type BuildEnds struct { + mu sync.Mutex + at map[string]time.Time +} + +// buildEnd is the part of the body of build.ended the runner reads. +type buildEnd struct { + Module string `json:"module"` + Fingerprint string `json:"fingerprint"` + Outcome string `json:"outcome"` +} + +// Heard keeps the end a message of the stream says, at the time the stream stamped it; a failed end, or a body naming no module +// and fingerprint, is kept for nothing. It says the module and fingerprint it kept. +func (e *BuildEnds) Heard(data []byte, at time.Time) (module, fp string, ok bool) { + var b buildEnd + if err := json.Unmarshal(data, &b); err != nil || b.Module == "" || b.Fingerprint == "" || b.Outcome == "failed" { + return "", "", false + } + e.mu.Lock() + defer e.mu.Unlock() + if e.at == nil { + e.at = map[string]time.Time{} + } + e.at[b.Module+"\x00"+b.Fingerprint] = at.UTC() + return b.Module, b.Fingerprint, true +} + +// At is when the build of the module at the fingerprint ended, false when no end was heard. +func (e *BuildEnds) At(module, fp string) (time.Time, bool) { + e.mu.Lock() + defer e.mu.Unlock() + t, ok := e.at[module+"\x00"+fp] + return t, ok +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go new file mode 100644 index 00000000..b6222c7a --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go @@ -0,0 +1,269 @@ +package pipeline + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "slices" + "time" + + runnerclient "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// HeldBound is how long the pipelines held for the runner's new version wait for it to run healthy (design 47 G §266: 30 +// minutes for the runner's `pipeline` step, A-195): a key of the state `bounds` with its own time to live, never a timer. +const HeldBound = 30 * time.Minute + +// HeldPrefix starts the key of a held wait's bound. +const HeldPrefix = "held-" + +// HeldKey is the bound's key for the awaited fingerprint: a fingerprint holds characters a key's name does not. +func HeldKey(fp string) string { + sum := sha256.Sum256([]byte(fp)) + return HeldPrefix + hex.EncodeToString(sum[:12]) +} + +// IsHeldKey says whether a key of the state `bounds` is a held wait's bound. +func IsHeldKey(key string) bool { + return len(key) > len(HeldPrefix) && key[:len(HeldPrefix)] == HeldPrefix +} + +// Awaiting is the awaited fingerprint of the pipelines held in store, "" and false when none is held (design 47 G §266, A-195): +// a pipeline started while one is held is held for the same version, so it never runs on the old runner. +func Awaiting(store Updater) func() (string, bool) { + return func() (string, bool) { + pkeys, err := store.Keys("*.*.*") + if err != nil { + return "", false + } + for _, key := range pkeys { + if p, _, err := store.Get(key); err == nil && p.Held != nil && p.Outcome == seat.PipelineOutcomeRunning { + return p.Held.Fingerprint, true + } + } + return "", false + } +} + +// changesSelf says whether a take-in's entry is the runner's own module at a fingerprint other than the one that runs: its +// pipelines are held for the new runner (design 47 G §65, §266, A-195). +func (s *Starter) changesSelf(module, fp string) bool { + return s.Self != "" && module == s.Self && fp != "" && fp != s.SelfFp +} + +// startHeld writes b's pipeline held for the runner's version awaiting, its steps neither placed nor asked, and says +// wait-held with step `pipeline`; pipeline-started is said when it is released (§74, §266). +func (s *Starter) startHeld(b Begin, key string, now time.Time, awaiting string) (bool, error) { + p := newPipeline(b, key, now) + p.Held = &seat.Held{Since: now, Fingerprint: awaiting} + if err := s.create(key, p); errors.Is(err, ErrExists) { + s.logf("%s is started already: this start writes nothing", key) + return false, nil + } else if err != nil { + return false, fmt.Errorf("the pipeline %s is not written: %w", key, err) + } + from := s.clockOf(awaiting) + if from != nil && s.Hold != nil { + if err := s.Hold(awaiting, *from); err != nil { + s.logf("the bound of the wait for the runner at %s is not armed: %v", awaiting, err) + } + } + w := s.waitOf(p, now, "") + w.ClockFrom = from + if err := s.Say(runnerclient.Subject(seat.EventWaitHeld, b.Module, b.Branch, commitOrGroup(p)), w); err != nil { + s.logf("wait-held of %s is not said: %v", key, err) + } + return true, nil +} + +// clockOf is when the build of the runner at fp ended, the time the bound counts from, nil while it has not (§266, A-181, A-206). +func (s *Starter) clockOf(fp string) *time.Time { + if s.Ends == nil { + return nil + } + if at, ok := s.Ends(s.Self, fp); ok { + return &at + } + return nil +} + +// Clocked starts the clock of the pipelines held for the runner at fp, the end of whose build was heard: it arms the bound from that +// end and says wait-held again for each with `clock-from`, so the watchman times it by the bound (§269, A-206). It says how many +// pipelines it said, and nothing when none is held for fp or the end is not known. +func (s *Starter) Clocked(store Updater, module, fp string) (int, error) { + if module != s.Self { + return 0, nil + } + from := s.clockOf(fp) + if from == nil { + return 0, nil + } + pkeys, err := store.Keys("*.*.*") + if err != nil { + return 0, fmt.Errorf("the pipelines are not listed: %w", err) + } + said := 0 + var first error + for _, key := range pkeys { + p, _, err := store.Get(key) + if err != nil || p.Held == nil || p.Outcome != seat.PipelineOutcomeRunning || p.Held.Fingerprint != fp { + continue + } + if said == 0 && s.Hold != nil { + if err := s.Hold(fp, *from); err != nil { + first = fmt.Errorf("the bound of the wait for the runner at %s is not armed: %w", fp, err) + } + } + w := s.waitOf(p, p.Held.Since, "") + w.ClockFrom = from + if err := s.Say(runnerclient.Subject(seat.EventWaitHeld, p.Module, p.Branch, commitOrGroup(p)), w); err != nil && first == nil { + first = fmt.Errorf("wait-held of %s is not said: %w", key, err) + } + said++ + } + return said, first +} + +func commitOrGroup(p seat.Pipeline) string { + if p.Group != "" { + return p.Group + } + return p.Commit +} + +func (s *Starter) waitOf(p seat.Pipeline, since time.Time, how string) seat.Wait { + return seat.Wait{V: 1, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit, Fingerprint: p.Fingerprint, Group: p.Group, + Revision: p.Revision, Source: p.Source, Step: seat.WaitStepPipeline, On: seat.WaitOnRunnerVersion, Since: since, How: how} +} + +// Release starts the pipelines held for fingerprint fp ("" for every held one) as the new runner running healthy lets +// them, or the bound's expiry or a hand release (how: `ran` or `released`; §266, §276): each has its steps placed and asked and +// says wait-ended and pipeline-started. A pipeline whose branch has no view yet stays held, and the first error is returned +// so the event is heard again. It says how many it started. +func (s *Starter) Release(store Updater, fp, how string, view func(branch string) (Half, bool)) (int, error) { + pkeys, err := store.Keys("*.*.*") + if err != nil { + return 0, fmt.Errorf("the pipelines are not listed: %w", err) + } + now := time.Now().UTC() + if s.Now != nil { + now = s.Now().UTC() + } + started := 0 + var first error + note := func(err error) { + if first == nil { + first = err + } + } + for _, key := range pkeys { + p, rev, err := store.Get(key) + if err != nil || p.Held == nil || p.Outcome != seat.PipelineOutcomeRunning || (fp != "" && p.Held.Fingerprint != fp) { + continue + } + half, ok := view(p.Branch) + if !ok { + note(fmt.Errorf("the catalog half of %s is not held yet: %s stays held", p.Branch, key)) + continue + } + held := *p.Held + p.Held = nil + 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) + } + } + if _, err := store.Update(key, p, rev); err != nil { + note(fmt.Errorf("the held pipeline %s is not released: %w", key, err)) + continue + } + started++ + if err := s.Say(runnerclient.Subject(seat.EventWaitEnded, p.Module, p.Branch, commitOrGroup(p)), s.waitOf(p, held.Since, how)); err != nil { + s.logf("wait-ended of %s is not said: %v", key, err) + } + ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: p.Branch, Source: p.Source, Module: p.Module, Commit: p.Commit, + Fingerprint: p.Fingerprint, Group: p.Group, Revision: p.Revision} + if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, p.Module, p.Branch, commitOrGroup(p)), 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 { + note(err) + } + } + } + return started, first +} + +// ReleaseExpired starts the pipelines held for the fingerprint whose bound key expired (§266, A-195): the 30 minutes passed +// without the new runner running healthy, so the running runner runs them, and each started is said late through Late (A-206). +func (s *Starter) ReleaseExpired(store Updater, key string, view func(branch string) (Half, bool)) (int, error) { + pkeys, err := store.Keys("*.*.*") + if err != nil { + return 0, fmt.Errorf("the pipelines are not listed: %w", err) + } + fp := "" + held := map[string]seat.Held{} + for _, k := range pkeys { + if p, _, err := store.Get(k); err == nil && p.Held != nil && p.Outcome == seat.PipelineOutcomeRunning && HeldKey(p.Held.Fingerprint) == key { + fp, held[k] = p.Held.Fingerprint, *p.Held + } + } + if fp == "" { + return 0, nil + } + n, err := s.Release(store, fp, "released", view) + for k, h := range held { + if p, _, gerr := store.Get(k); gerr == nil && p.Held == nil && s.Late != nil { + s.Late(p, h) + } + } + return n, err +} + +// heldFingerprints are the fingerprints of the runner the held pipelines in store await, each once. +func heldFingerprints(store Updater) []string { + pkeys, err := store.Keys("*.*.*") + if err != nil { + return nil + } + var fps []string + for _, key := range pkeys { + if p, _, err := store.Get(key); err == nil && p.Held != nil && p.Outcome == seat.PipelineOutcomeRunning && !slices.Contains(fps, p.Held.Fingerprint) { + fps = append(fps, p.Held.Fingerprint) + } + } + return fps +} + +// Release starts by hand the pipelines held for the runner's new version with the running runner, recorded in the hand-act +// log first (§276, A-206): each says wait-ended `released` and pipeline-started, and the bound of each wait is dropped. It +// says how many it started. +func (v Verbs) Release(_ string, in seat.ReleaseInput) (seat.ReleaseAnswer, error) { + if v.Starter == nil || v.View == nil { + return seat.ReleaseAnswer{}, errors.New("the runner cannot release a pipeline: it holds no starter or no catalog half") + } + if err := v.record(seat.VerbRelease, "started the pipelines held for the new runner by hand", in.Why); err != nil { + return seat.ReleaseAnswer{}, err + } + fps := heldFingerprints(v.Hearer.Store) + n, err := v.Starter.Release(v.Hearer.Store, "", "released", v.View) + if n > 0 && v.Drop != nil { + for _, fp := range fps { + if derr := v.Drop(HeldKey(fp)); derr != nil && err == nil { + err = fmt.Errorf("the bound of the wait for %s is not dropped: %w", fp, derr) + } + } + } + return seat.ReleaseAnswer{Released: int64(n)}, err +} + +// HeldTTL is how long the bound of a wait that counts from the build end at from still has to run at now: the 30 minutes less what passed since +// the end, at least a second, the least a key's time to live holds (design 47 G §266, A-206). +func HeldTTL(from, now time.Time) time.Duration { + return max(HeldBound-now.Sub(from), time.Second) +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go new file mode 100644 index 00000000..d150d00a --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go @@ -0,0 +1,234 @@ +package pipeline + +import ( + "encoding/json" + "strings" + "testing" + "time" + + 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 with the Create a start writes through. +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 +} + +// A take-in giving the runner's own module a new fingerprint holds every pipeline it starts, a later take-in's too while one is +// held, with no step asked; the new runner running healthy releases them, placing and asking their steps (design 47 G §65, §266, +// §276, §534; A-195). +func TestPipelinesAreHeldForTheNewRunnerAndReleasedWhenItRuns(t *testing.T) { + reg, entries := placeFixture() + entries["runner"] = catalog.CatalogEntry{Module: "runner"} + store := memCreated{memStore{}} + asks := &stepAsks{} + var said []string + var bounds []string + s := &Starter{Pipelines: store, Steps: asks, Name: "the-runner", Self: "runner", SelfFp: "src2:old", + Awaiting: Awaiting(store.memStore), Ends: func(string, string) (time.Time, bool) { return time.Unix(9, 0), true }, + Hold: func(fp string, _ time.Time) error { bounds = append(bounds, fp); return nil }, + Registration: func(m string) (catalog.Registration, bool) { return reg, m == "app" }, + Say: func(subject string, _ any) error { + said = append(said, subject[strings.Index(subject, ".event.")+7:]) + return nil + }} + view := func(string) (Half, bool) { return Half{Entries: entries}, true } + n, err := s.TakenIn(catalog.TakeInTaken{Key: "m.1", Branch: "main", Source: 4, Entries: []catalog.TakeInEntry{ + {Module: "app", Commit: "c1", Fingerprint: "src2:aa"}, {Module: "runner", Commit: "c2", Fingerprint: "src2:new"}}}, view) + if err != nil || n != 2 { + t.Fatalf("take-in = %d, %v", n, err) + } + p, _, _ := store.Get("main.4.app") + if p.Held == nil || p.Held.Fingerprint != "src2:new" || len(p.Steps) != 0 || len(asks.asks) != 0 { + t.Fatalf("held = %+v, steps %d, asks %d", p.Held, len(p.Steps), len(asks.asks)) + } + if strings.Join(said, "|") != "wait-held.app.main.c1|wait-held.runner.main.c2" || len(bounds) != 2 || bounds[0] != "src2:new" { + t.Errorf("said %v, bounds %v", said, bounds) + } + // A later take-in not naming the runner is held for the same version. + if n, err := s.TakenIn(catalog.TakeInTaken{Key: "m.2", Branch: "main", Source: 5, Entries: []catalog.TakeInEntry{{Module: "app", Commit: "c3", Fingerprint: "src2:bb"}}}, view); err != nil || n != 1 { + t.Fatalf("second take-in = %d, %v", n, err) + } + if p, _, _ := store.Get("main.5.app"); p.Held == nil || len(asks.asks) != 0 { + t.Fatalf("the later pipeline is not held: %+v", p.Held) + } + // Another fingerprint's release starts nothing. + if n, err := s.Release(store.memStore, "src2:other", "ran", view); err != nil || n != 0 { + t.Fatalf("release of another version = %d, %v", n, err) + } + said = nil + n, err = s.Release(store.memStore, "src2:new", "ran", view) + if err != nil || n != 3 { + t.Fatalf("release = %d, %v", n, err) + } + p, _, _ = store.Get("main.4.app") + if p.Held != nil || len(p.Steps) == 0 || len(asks.asks) == 0 { + t.Errorf("after the release: held %+v, steps %d, asks %d", p.Held, len(p.Steps), len(asks.asks)) + } + if !strings.Contains(strings.Join(said, "|"), "wait-ended.app.main.c1|pipeline-started.app.main.c1") { + t.Errorf("said %v", said) + } + if _, held := Awaiting(store.memStore)(); held { + t.Error("a pipeline is still held") + } +} + +// The verb release starts the held pipelines by hand, recorded in the hand-act log first, saying why and ending each wait +// `released`, and drops the bound of the wait; a release with no why does nothing (design 47 G §276, A-206). +func TestReleaseVerbStartsTheHeldPipelinesByHand(t *testing.T) { + reg, entries := placeFixture() + entries["runner"] = catalog.CatalogEntry{Module: "runner"} + store := memCreated{memStore{}} + asks := &stepAsks{} + var said []string + s := &Starter{Pipelines: store, Steps: asks, Name: "the-runner", Self: "runner", SelfFp: "src2:old", Awaiting: Awaiting(store.memStore), + Registration: func(m string) (catalog.Registration, bool) { return reg, m == "app" }, + Say: func(subject string, _ any) error { + said = append(said, subject[strings.Index(subject, ".event.")+7:]) + return nil + }} + view := func(string) (Half, bool) { return Half{Entries: entries}, true } + if _, err := s.TakenIn(catalog.TakeInTaken{Key: "m.1", Branch: "main", Source: 4, Entries: []catalog.TakeInEntry{ + {Module: "app", Commit: "c1", Fingerprint: "src2:aa"}, {Module: "runner", Commit: "c2", Fingerprint: "src2:new"}}}, view); err != nil { + t.Fatal(err) + } + acts, dropped := &actLog{}, []string{} + v := Verbs{Hearer: &Hearer{Store: store.memStore}, Starter: s, View: view, Acts: acts, + Drop: func(k string) error { dropped = append(dropped, k); return nil }} + if _, err := v.Release("operator", seat.ReleaseInput{}); err == nil || len(*acts) != 0 { + t.Fatalf("a release saying no why: %v, acts %v", err, *acts) + } + said = nil + a, err := v.Release("operator", seat.ReleaseInput{Why: "the new runner will not come"}) + if err != nil || a.Released != 2 { + t.Fatalf("release = %+v, %v", a, err) + } + if p, _, _ := store.Get("main.4.app"); p.Held != nil || len(p.Steps) == 0 || len(asks.asks) == 0 { + t.Errorf("after the release: held %+v, steps %d", p.Held, len(p.Steps)) + } + if len(*acts) != 1 || !strings.HasPrefix((*acts)[0], "mesh-pipeline-runner.release: ") || len(dropped) != 1 || dropped[0] != HeldKey("src2:new") { + t.Errorf("acts %v, dropped %v", *acts, dropped) + } + if !strings.Contains(strings.Join(said, "|"), "wait-ended.app.main.c1") { + t.Errorf("said %v", said) + } +} + +// The bound's expiry starts the held pipelines and says each as late through the starter's Late, with the awaited fingerprint and +// since when; a pipeline that stays held (no view) is not said (design 47 G §266, A-206). +func TestExpiredBoundSaysEachReleasedPipelineLate(t *testing.T) { + reg, entries := placeFixture() + store := memCreated{memStore{}} + since := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC) + for _, k := range []string{"main.4.app", "dev.4.app"} { + store.memStore[k] = struct { + p seat.Pipeline + rev uint64 + }{seat.Pipeline{Pipeline: k, Branch: strings.Split(k, ".")[0], Module: "app", Outcome: seat.PipelineOutcomeRunning, + Held: &seat.Held{Since: since, Fingerprint: "src2:new"}}, 1} + } + var late []string + s := &Starter{Pipelines: store, Steps: &stepAsks{}, Name: "the-runner", Say: func(string, any) error { return nil }, + Registration: func(string) (catalog.Registration, bool) { return reg, true }, + Late: func(p seat.Pipeline, h seat.Held) { + late = append(late, p.Pipeline+"@"+h.Fingerprint+"@"+h.Since.Format(time.RFC3339)) + }} + view := func(b string) (Half, bool) { return Half{Entries: entries}, b == "main" } + if n, err := s.ReleaseExpired(store.memStore, HeldKey("src2:new"), view); err == nil || n != 1 { + t.Fatalf("release = %d, %v (the pipeline of a branch with no view stays held, an error)", n, err) + } + if len(late) != 1 || late[0] != "main.4.app@src2:new@2026-11-11T12:00:00Z" { + t.Errorf("late %v", late) + } +} + +// A wait of the step pipeline waits on the runner version (design 47 G §534, A-206). +func TestHeldWaitIsOnTheRunnerVersion(t *testing.T) { + w := (&Starter{}).waitOf(seat.Pipeline{Pipeline: "main.4.app", Branch: "main", Module: "app"}, time.Unix(1, 0), "") + if w.Step != seat.WaitStepPipeline || w.On != "runner version" { + t.Errorf("wait %+v", w) + } +} + +// The wait of a held pipeline counts from the build end of the awaited runner: wait-held says it as `clock-from` and the bound is +// armed from it; while that build has not ended nothing is armed, and hearing its end arms the bound and says wait-held again +// with `clock-from`; a failed end starts no clock (design 47 G §266, §269, A-181, A-206). +func TestHeldWaitCountsFromTheBuildEndOfTheAwaitedRunner(t *testing.T) { + reg, entries := placeFixture() + store := memCreated{memStore{}} + ends := &BuildEnds{} + var waits []seat.Wait + var armed []string + s := &Starter{Pipelines: store, Steps: &stepAsks{}, Name: "the-runner", Self: "runner", SelfFp: "src2:old", Awaiting: Awaiting(store.memStore), + Ends: ends.At, Hold: func(fp string, from time.Time) error { + armed = append(armed, fp+"@"+from.Format(time.RFC3339)) + return nil + }, + Registration: func(string) (catalog.Registration, bool) { return reg, true }, + Say: func(subject string, body any) error { + if w, ok := body.(seat.Wait); ok && strings.Contains(subject, ".wait-held.") { + waits = append(waits, w) + } + return nil + }} + view := func(string) (Half, bool) { return Half{Entries: entries}, true } + taken := catalog.TakeInTaken{Key: "m.1", Branch: "main", Source: 4, Entries: []catalog.TakeInEntry{ + {Module: "app", Commit: "c1", Fingerprint: "src2:aa"}, {Module: "runner", Commit: "c2", Fingerprint: "src2:new"}}} + end := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC) + // The runner's build has not ended: held, no clock, no bound. + if _, err := s.TakenIn(taken, view); err != nil { + t.Fatal(err) + } + if len(waits) != 2 || waits[0].ClockFrom != nil || len(armed) != 0 { + t.Fatalf("waits %+v, armed %v", waits, armed) + } + // A failed end starts no clock; a done end does. + if _, _, ok := ends.Heard([]byte(`{"module":"runner","fingerprint":"src2:new","outcome":"failed"}`), end); ok { + t.Error("a failed end was kept") + } + if m, fp, ok := ends.Heard([]byte(`{"module":"runner","fingerprint":"src2:new","outcome":"done"}`), end); !ok || m != "runner" || fp != "src2:new" { + t.Fatalf("heard %q %q %v", m, fp, ok) + } + waits = nil + if n, err := s.Clocked(store.memStore, "runner", "src2:new"); err != nil || n != 2 { + t.Fatalf("clocked = %d, %v", n, err) + } + if len(waits) != 2 || waits[0].ClockFrom == nil || !waits[0].ClockFrom.Equal(end) || len(armed) != 1 || armed[0] != "src2:new@2026-11-11T12:00:00Z" { + t.Errorf("waits %+v, armed %v", waits, armed) + } + // A pipeline held after the end is known says the clock at once. + waits = nil + taken.Key, taken.Source = "m.2", 5 + if _, err := s.TakenIn(taken, view); err != nil { + t.Fatal(err) + } + if len(waits) == 0 || waits[0].ClockFrom == nil || !waits[0].ClockFrom.Equal(end) { + t.Errorf("waits %+v", waits) + } +} + +// The bound counts from the build end: the 30 minutes less what passed since, never under a second. +func TestHeldTTLCountsFromTheBuildEnd(t *testing.T) { + end := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC) + for _, c := range []struct { + after, want time.Duration + }{{0, 30 * time.Minute}, {10 * time.Minute, 20 * time.Minute}, {45 * time.Minute, time.Second}} { + if got := HeldTTL(end, end.Add(c.after)); got != c.want { + t.Errorf("after %s the bound has %s, want %s", c.after, got, c.want) + } + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/place.go b/modules/mesh-delivery-pipeline/internal/pipeline/place.go index 05eee910..946ac980 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/place.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/place.go @@ -50,7 +50,7 @@ func AskID(pipeline, artifact, step string) string { // Place places a pipeline's steps (design 47 G §21, §23, §323; A-91): only the steps the module's registration policy // names, each artifact's lint, unit and integration steps beside the module's composition, review and approval. A step the -// policy does not name is no step. A `bypass` step is placed bypassed and asked of no one; a build step is placed running, +// policy leaves without a status is `optional` (§553). A `bypass` step is placed bypassed and asked of no one; a build step is placed running, // asked of the build seat the artifact's language names in the catalog half (§323), or, where no seat names it, ended // could not run naming the missing seat; the module's steps wait. It returns the steps and the asks to put on the seats' // work queues, one per running step. @@ -74,9 +74,6 @@ func Place(p seat.Pipeline, reg catalog.Registration, entries map[string]catalog } for _, step := range buildSteps { status := Status(reg, name, step) - if status == "" { - continue - } s := seat.StepResult{Step: step, Artifact: a.Ref, Status: status, Fingerprint: p.Fingerprint} if status == seat.StepResultStatusBypass { s.Outcome, s.Why = seat.StepResultOutcomeBypassed, "bypassed by the module's policy" diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/place_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/place_test.go index f5fb7fd4..67f9b960 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/place_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/place_test.go @@ -49,7 +49,7 @@ func placeFixture() (catalog.Registration, map[string]catalog.CatalogEntry) { // The runner's asking half (design 47 G §22, §24, §323; A-91): a pipeline's steps are the ones the policy names, a bypassed // step asked of no one, an artifact no build seat names ended could not run naming it, and each running step asked of its -// build seat's accepted kind step with the pipeline's commit and fingerprint; a step the policy does not name is no step. +// build seat's accepted kind step with the pipeline's commit and fingerprint; a step the policy leaves without a status is optional (§553). func TestStartPlacesTheStepsThePolicyNamesAndAsksEachBuildSeat(t *testing.T) { reg, entries := placeFixture() store := created{} @@ -67,16 +67,19 @@ func TestStartPlacesTheStepsThePolicyNamesAndAsksEachBuildSeat(t *testing.T) { for _, st := range p.Steps { got = append(got, st.Artifact+" "+st.Step+" "+st.Outcome) } - want := []string{" build running", "app:program lint running", "app:program unit running", "app:tools lint bypassed", "app:tools unit running", - "app:web lint could not run", "app:web unit could not run", " approval waiting"} + // §553: integration, composition and review are named by no policy, so they are optional steps and run. + want := []string{" build running", "app:program lint running", "app:program unit running", "app:program integration running", + "app:tools lint bypassed", "app:tools unit running", "app:tools integration running", + "app:web lint could not run", "app:web unit could not run", "app:web integration could not run", + " composition running", " review waiting", " approval waiting"} if strings.Join(got, "|") != strings.Join(want, "|") { - t.Errorf("steps = %q, want %q (integration and composition are named by no policy: no step)", got, want) + t.Errorf("steps = %q, want %q", got, want) } - if !strings.Contains(p.Steps[5].Why, `"rust"`) { - t.Errorf("the step of an artifact no seat builds says %q, want it to name the language", p.Steps[5].Why) + if !strings.Contains(p.Steps[7].Why, `"rust"`) { + t.Errorf("the step of an artifact no seat builds says %q, want it to name the language", p.Steps[7].Why) } - if len(asks.asks) != 4 { - t.Fatalf("asked %d steps, want the build and the three running ones", len(asks.asks)) + if len(asks.asks) != 7 { + t.Fatalf("asked %d steps, want the build, the five running build steps and the composition", len(asks.asks)) } // A-99: the build is every pipeline's first step, required whatever the policy names, asked of mesh-builds for the // module's own entry only. @@ -92,7 +95,7 @@ func TestStartPlacesTheStepsThePolicyNamesAndAsksEachBuildSeat(t *testing.T) { t.Errorf("the ask = %+v at %s", a, asks.at[1]) } // A second start of the revision asks nothing. - if ok, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Entries: entries}); ok || err != nil || len(asks.asks) != 4 { + if ok, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Entries: entries}); ok || err != nil || len(asks.asks) != 7 { t.Errorf("second start = %v, %v, %d asks", ok, err, len(asks.asks)) } } @@ -198,9 +201,9 @@ func TestResumeAsksTheStepsPlacedAndNotEnded(t *testing.T) { asks := &stepAsks{} s := &Starter{Steps: asks, Name: "the-runner", Registration: func(string) (catalog.Registration, bool) { return reg, true }} n, err := s.Resume(store, "main", Half{Entries: entries}) - // main.4 has the build and three running steps of the fixture; main.5 one fewer, its build ended. - if err != nil || n != 7 || len(asks.asks) != 7 { - t.Fatalf("resume asked %d (%d put), %v; want 7", n, len(asks.asks), err) + // main.4 has the build, five running build steps and the composition of the fixture; main.5 one fewer, its build ended. + if err != nil || n != 13 || len(asks.asks) != 13 { + t.Fatalf("resume asked %d (%d put), %v; want 13", n, len(asks.asks), err) } for _, id := range asks.ids { found := false diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read.go b/modules/mesh-delivery-pipeline/internal/pipeline/read.go index ced5a604..4da65cb7 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/read.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read.go @@ -88,6 +88,25 @@ func (r Reads) Stands() (seat.StandsAnswer, error) { return seat.StandsAnswer{Stands: out}, nil } +// Waits is the pipelines held for the runner's new version, each as the wait of step `pipeline` it says (design 47 G §266, §276, +// §534, A-195): the acting part awaited by its fingerprint, since when, and the pipeline. +func (r Reads) Waits() (seat.StandsAnswer, error) { + all, err := r.pipelines() + if err != nil { + return seat.StandsAnswer{}, err + } + out := []seat.Wait{} + for _, p := range all { + if p.Held == nil || p.Outcome != seat.PipelineOutcomeRunning { + continue + } + out = append(out, seat.Wait{Pipeline: p.Pipeline, Module: p.Module, Branch: p.Branch, Commit: p.Commit, Group: p.Group, Source: p.Source, + Revision: p.Revision, Fingerprint: p.Held.Fingerprint, Step: seat.WaitStepPipeline, On: seat.WaitOnBuildNode, Since: p.Held.Since, V: 1}) + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Since.Before(out[j].Since) }) + return seat.StandsAnswer{Stands: out}, nil +} + // WhyCarried is where a step's result was carried from (§25, §83): the pipeline it names. func (r Reads) WhyCarried(in seat.WhyCarriedInput) (seat.WhyCarriedAnswer, error) { p, ok, err := r.pipeline(in.Pipeline) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go index 601e639c..5dabc4b4 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go @@ -127,3 +127,19 @@ func TestReads(t *testing.T) { } } } + +// runner_waits lists the pipelines held for the runner's new version as waits of step `pipeline` (design 47 G §276, A-195). +func TestWaitsListTheHeldPipelines(t *testing.T) { + since := time.Date(2026, 11, 12, 6, 0, 0, 0, time.UTC) + held := seat.Pipeline{Pipeline: "main.4.app", Module: "app", Branch: "main", Commit: "c1", Source: 4, Outcome: seat.PipelineOutcomeRunning, + Held: &seat.Held{Since: since, Fingerprint: "src2:new"}} + plain := seat.Pipeline{Pipeline: "main.3.app", Module: "app", Branch: "main", Outcome: seat.PipelineOutcomeRunning} + pipes := memKept{} + put(t, pipes, held.Pipeline, held) + put(t, pipes, plain.Pipeline, plain) + r := Reads{Pipelines: pipes, Kept: memKept{}} + got, err := r.Waits() + if err != nil || len(got.Stands) != 1 || got.Stands[0].Step != seat.WaitStepPipeline || got.Stands[0].Fingerprint != "src2:new" || !got.Stands[0].Since.Equal(since) { + t.Fatalf("waits = %+v, %v", got, err) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/registrations.go b/modules/mesh-delivery-pipeline/internal/pipeline/registrations.go index 27bdbe82..91073caf 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/registrations.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/registrations.go @@ -12,14 +12,20 @@ import ( // Status is a step's status for one artifact of a module, read from the module's registration and never from a // manifest (design 47 G §21, §23): the artifact's own status where its policy names one, else the module's. A step -// neither names is "" and is no step of the pipeline (A-91): Place places none for it. +// whose status neither names is `optional` (§553): it runs and a failure holds nothing; an empty approval is no step (A-196). func Status(reg catalog.Registration, artifact, step string) string { if a, ok := reg.Policy.Artifacts[artifact]; ok { if s := statusOf(a, step); s != "" { return s } } - return statusOf(reg.Policy.Steps, step) + if s := statusOf(reg.Policy.Steps, step); s != "" { + return s + } + if step == seat.StepResultStepApproval { + return "" + } + return seat.StepResultStatusOptional } func statusOf(s catalog.StepStatuses, step string) string { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/registrations_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/registrations_test.go index 03eae010..8cc0b8cf 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/registrations_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/registrations_test.go @@ -20,13 +20,20 @@ func TestPolicyAndBuildSeat(t *testing.T) { {"tools", "unit", "bypass"}, {"tools", "integration", "optional"}, {"tools", "approval", "required"}, - {"program", "composition", ""}, + // A step whose status the policy leaves empty is optional (design 47 G §553), never no step. + {"program", "composition", "optional"}, + {"program", "review", "optional"}, } { if got := Status(reg, c.artifact, c.step); got != c.want { t.Errorf("Status(%s, %s) = %q, want %q", c.artifact, c.step, got, c.want) } } + // An empty approval status asks the operator nothing (A-196): no step. + if got := Status(catalog.Registration{}, "", "approval"); got != "" { + t.Errorf("an empty approval status = %q, want none", got) + } + entries := map[string]catalog.CatalogEntry{ "mesh-build-go": {Module: "mesh-build-go", Acts: []catalog.ActingRule{{Seat: "node-build-go", Step: "build", On: "go"}}}, "mesh-build-node": {Module: "mesh-build-node", Acts: []catalog.ActingRule{{Seat: "node-build-node", Step: "build", On: "typescript"}, {Seat: "node-build-node", Step: "check", On: "go"}}}, diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 9e5737aa..3d01f85b 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -35,6 +35,8 @@ type Begin struct { Group string Repository string Revision int64 + // Await is the fingerprint of the runner's own module this pipeline is held for (§266, A-195); empty when it is not held. + Await string // 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 } @@ -70,6 +72,30 @@ 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 + // Self is the runner's own module and SelfFp the fingerprint it runs (A-149); a take-in giving Self another fingerprint + // holds its pipelines for the new runner (§65, §266, A-195). Awaiting is the fingerprint the held pipelines await, while + // any is held; Hold arms the bound of that wait (HeldBound) counting from the time the awaited build ended. All nil: nothing is held. + Self, SelfFp string + Awaiting func() (string, bool) + Hold func(fp string, from time.Time) error + // Ends says when the build of a module at a fingerprint ended (BuildEnds.At): the clock of the wait's bound (A-181, A-206). + Ends func(module, fp string) (time.Time, bool) + // Late says a pipeline the bound's expiry started with the running runner (A-206): the finding pipeline-waits. + Late func(p seat.Pipeline, held seat.Held) +} + +func newPipeline(b Begin, key string, now time.Time) seat.Pipeline { + return 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{}} +} + +func (s *Starter) create(key string, p seat.Pipeline) error { + raw, err := json.Marshal(p) + if err != nil { + return err + } + return s.Pipelines.Create(key, raw) } // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second @@ -83,9 +109,15 @@ func (s *Starter) Start(b Begin) (bool, error) { now = s.Now().UTC() } 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{}} + if b.Await == "" && s.Awaiting != nil { + if fp, ok := s.Awaiting(); ok { + b.Await = fp + } + } + if b.Await != "" { + return s.startHeld(b, key, now, b.Await) + } + p := newPipeline(b, key, now) var asks []Asked if s.Registration != nil { if reg, ok := s.Registration(b.Module); ok { @@ -94,11 +126,7 @@ func (s *Starter) Start(b Begin) (bool, error) { s.logf("%s has no registration yet: its pipeline starts with no step", key) } } - raw, err := json.Marshal(p) - if err != nil { - return false, err - } - if err := s.Pipelines.Create(key, raw); errors.Is(err, ErrExists) { + if err := s.create(key, p); errors.Is(err, ErrExists) { s.logf("%s is started already: this start writes nothing", key) return false, nil } else if err != nil { @@ -288,9 +316,16 @@ func (s *Starter) TakenIn(ev catalog.TakeInTaken, view func(branch string) (Half half, _ := view(ev.Branch) started := 0 var first error + // A take-in giving the runner's own module a new fingerprint holds every pipeline it starts for that runner (§65, A-195). + await := "" + for _, e := range ev.Entries { + if s.changesSelf(e.Module, e.Fingerprint) { + await = e.Fingerprint + } + } for _, e := range ev.Entries { ok, err := s.Start(Begin{Branch: ev.Branch, Source: ev.Source, Module: e.Module, Commit: e.Commit, Fp: e.Fingerprint, - Group: ev.Group, Repository: e.Repository, Revision: half.Revision, Entries: half.Entries}) + Group: ev.Group, Repository: e.Repository, Revision: half.Revision, Entries: half.Entries, Await: await}) if err != nil && first == nil { first = err } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go index fd2c3fa3..a5c60fc5 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "sort" "strings" "time" @@ -207,6 +208,49 @@ func (h *Hearer) said(p seat.Pipeline, asked, ended bool) error { if err := h.say(seat.EventPipelineEnded, p, nil); err != nil { return fmt.Errorf("pipeline %s ended but %s is not said: %w", p.Pipeline, seat.EventPipelineEnded, err) } + return h.groupEnded(p) + } + return nil +} + +// groupEnded says a group's one pipeline's end once every member's pipeline of the same watched branch and source revision +// ended (design 47 G §73, §332): on the subject under the group, no module part, with every member's end. A pipeline of no +// group says none; a member still running says none yet. +func (h *Hearer) groupEnded(p seat.Pipeline) error { + if p.Group == "" { + return nil + } + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + return fmt.Errorf("the members of group %s are not listed: %w", p.Group, err) + } + sort.Strings(pkeys) + var members []seat.PipelineEnded + outcome, why := seat.GroupEndedOutcomePassed, "" + var last *time.Time + for _, key := range pkeys { + m, _, err := h.Store.Get(key) + if err != nil || m.Branch != p.Branch || m.Source != p.Source || m.Group != p.Group { + continue + } + if m.Outcome == seat.PipelineOutcomeRunning { + return nil + } + members = append(members, pipelineEnded(m)) + if m.Ended != nil && (last == nil || m.Ended.After(*last)) { + last = m.Ended + } + if m.Outcome != seat.PipelineOutcomePassed && outcome != seat.GroupEndedOutcomeFailed { + outcome, why = m.Outcome, m.Module+": "+m.Why + if m.Outcome == seat.PipelineOutcomeFailed { + outcome = seat.GroupEndedOutcomeFailed + } + } + } + subject := "mesh.seat." + seat.Seat + ".event." + seat.EventGroupEnded + "." + keys.EscapeName(p.Branch) + "." + keys.EscapeName(p.Group) + if err := h.Say(subject, seat.GroupEnded{V: 1, Branch: p.Branch, Group: p.Group, Source: p.Source, Revision: p.Revision, Ended: last, + Outcome: outcome, Why: why, Pipelines: members}); err != nil { + return fmt.Errorf("group %s ended but %s is not said: %w", p.Group, seat.EventGroupEnded, err) } return nil } @@ -247,8 +291,31 @@ func (h *Hearer) say(event string, p seat.Pipeline, step *seat.StepResult) error } } +// The test budget per catalog module (design 47 G §24, §419; the test budget): at most 500 unit and 50 integration tests. +const ( + budgetUnit = 500 + budgetIntegration = 50 +) + +// testCounts is the module's test counts, the sum of its test steps' own counts; nil when no step counted any. +func testCounts(steps []seat.StepResult) *seat.TestCounts { + var t seat.TestCounts + counted := false + for _, s := range steps { + if s.Tests != nil { + t.Unit, t.Integration, counted = t.Unit+s.Tests.Unit, t.Integration+s.Tests.Integration, true + } + } + if !counted { + return nil + } + return &t +} + func pipelineEnded(p seat.Pipeline) seat.PipelineEnded { - return seat.PipelineEnded{V: 1, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit, Fingerprint: p.Fingerprint, + tests := testCounts(p.Steps) + return seat.PipelineEnded{Tests: tests, OverBudget: tests != nil && (tests.Unit > budgetUnit || tests.Integration > budgetIntegration), + V: 1, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit, Fingerprint: p.Fingerprint, Group: p.Group, Revision: p.Revision, Source: p.Source, Run: p.Run, Outcome: p.Outcome, Why: p.Why, Started: p.Started, Ended: p.Ended, EndCount: p.EndCount, Steps: p.Steps} } 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") + } +} diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index a1ffc26d..bcf8f548 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 { @@ -50,9 +54,12 @@ 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. + {"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,13 +69,15 @@ 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", + "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", "mesh-delivery-builder.build.ended.**", "operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended", - "node-build-image.step-ended", "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", "mesh-wants.step-ended", "mesh-catalog.registration-changed", + "node-build-image.step-ended", "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", "node-engine.running", "mesh-wants.step-ended", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-conditions.findings-asked"})}, // RN2 (§21, §195, §200, §77): the registrations and the catalog half asked by their verbs, the first views read and // asked again by hand (a hand act). @@ -81,7 +90,10 @@ 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", "runner_health", "runner_waits", "runner_release"})}, + // 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"})}, } @@ -117,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 b66b07dd..1bb8273d 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -18,19 +18,24 @@ "rerun", "accept", "cancel", - "ask-again" + "ask-again", + "release" ] } ], "tools": [ - "pipeline_first_views", - "pipeline_first_view_again" + "runner_first_views", + "runner_first_view_again", + "runner_health", + "runner_waits", + "runner_release" ], "consumes": [ "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", + "mesh-delivery-builder.build.ended.**", "operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended", @@ -38,6 +43,7 @@ "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", + "node-engine.running", "mesh-wants.step-ended", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", @@ -61,11 +67,13 @@ "state": [ { "name": "pipelines", - "history": 1 + "history": 1, + "max-mib": 256 }, { "name": "kept", - "history": 1 + "history": 1, + "max-mib": 128 }, { "name": "aged", @@ -87,6 +95,10 @@ { "name": "first-views", "history": 1 + }, + { + "name": "consumers", + "history": 1 } ], "streams": [ @@ -105,6 +117,10 @@ { "name": "step-outputs", "max-mib": 1024 + }, + { + "name": "sent-objects", + "max-mib": 256 } ], "own-secrets": { @@ -140,7 +156,8 @@ }, "user": "mesh-delivery-pipeline", "health": { - "kind": "unit" + "kind": "tool", + "tool": "runner_health" } } ],