diff --git a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidates.go b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidates.go new file mode 100644 index 00000000..5ca67701 --- /dev/null +++ b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidates.go @@ -0,0 +1,36 @@ +package meshcatalogclient + +import ( + "context" + "encoding/json" + "fmt" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// WireCandidatesProcess hands each candidate's `catalog-revision` to hook (design 47 G A-185, A-193, A-199): a candidate's +// revision no Catalog keeps, heard once on the event's one subscription of the process's reader, beside every watched +// branch's Catalog and whether or not the part follows any, so the runner places a run's pipelines on it. A trunk revision +// is passed over (each branch's Catalog keeps it); data that is no such event is an error. The part's manifest consumes the +// seat's event catalog-revision. +func WireCandidatesProcess(r *views.ProcessReader, hook func(ctx context.Context, ev seat.CatalogRevisionEvent) error) error { + if hook == nil { + return fmt.Errorf("meshcatalogclient: a candidates hook is a function") + } + return r.AdoptEvent(seat.Seat, seat.EventCatalogRevision, candidateHear(hook)) +} + +// candidateHear is the handler of catalog-revision that hands a candidate's revision to hook. +func candidateHear(hook func(ctx context.Context, ev seat.CatalogRevisionEvent) error) func(context.Context, []byte) error { + return func(ctx context.Context, data []byte) error { + var ev seat.CatalogRevisionEvent + if err := json.Unmarshal(data, &ev); err != nil || ev.Revision <= 0 { + return fmt.Errorf("meshcatalogclient: not a %s event: %s", seat.EventCatalogRevision, clipBytes(data)) + } + if ev.Kind != KindCandidate { + return nil + } + return hook(ctx, ev) + } +} diff --git a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/catalog_view_test.go b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/catalog_view_test.go index 6c74285e..760d8bfc 100644 --- a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/catalog_view_test.go +++ b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/catalog_view_test.go @@ -162,3 +162,29 @@ func TestCatalogIsAViewPerWatchedBranch(t *testing.T) { t.Error("a Catalog of no branch was made") } } + +// A-185, A-193, A-199: a candidate's catalog-revision reaches the hook, adopted on the event's one subscription of the +// reader, also with no watched branch's Catalog; a trunk revision does not, and data that is no such event is said. +func TestCandidatesReachTheHook(t *testing.T) { + var hooked []seat.CatalogRevisionEvent + hook := func(_ context.Context, ev seat.CatalogRevisionEvent) error { hooked = append(hooked, ev); return nil } + r := &views.ProcessReader{Module: "runner"} + if err := WireCandidatesProcess(r, hook); err != nil { + t.Fatal(err) + } + if err := WireCandidatesProcess(r, nil); err == nil { + t.Error("a nil hook was taken") + } + hear := candidateHear(hook) + for _, data := range [][]byte{revisionEvent("main", KindTrunk, 3), revisionEvent("main", KindCandidate, 4)} { + if err := hear(context.Background(), data); err != nil { + t.Fatal(err) + } + } + if err := hear(context.Background(), []byte(`not json`)); err == nil { + t.Error("data that is no catalog-revision was taken") + } + if len(hooked) != 1 || hooked[0].Revision != 4 || hooked[0].Kind != KindCandidate { + t.Errorf("the hook was handed %+v, want the candidate's revision 4 alone", hooked) + } +} diff --git a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go index 7b0bfe66..307ffd31 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go @@ -432,8 +432,10 @@ type Pipeline struct { // yes when the module is over its test budget (design 47 G section 419) OverBudget bool `json:"over-budget,omitempty"` // the pipeline's key, .., a branch escaped as a key's name is - Pipeline string `json:"pipeline,omitempty"` - Repository string `json:"repository,omitempty"` + Pipeline string `json:"pipeline,omitempty"` + // for a pipeline a package's publish started, that publish (A-202) + Published *Published `json:"published,omitempty"` + Repository string `json:"repository,omitempty"` // the catalog graph revision of the branch the pipeline runs for Revision int64 `json:"revision,omitempty"` // the run on request's id, for a pipeline of a run; such a pipeline delivers nothing @@ -545,6 +547,8 @@ type PipelineStarted struct { Module string `json:"module"` // the pipeline's key, .., a branch escaped as a key's name is Pipeline string `json:"pipeline"` + // for a pipeline a package's publish started, that publish (A-202) + Published *Published `json:"published,omitempty"` // the catalog graph revision of the branch the pipeline runs for Revision int64 `json:"revision,omitempty"` // the run on request's id, for a pipeline of a run; such a pipeline delivers nothing @@ -556,6 +560,14 @@ type PipelineStarted struct { V int64 `json:"v"` } +// Published is the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202). +type Published struct { + // the package whose publish started the pipeline + Artifact string `json:"artifact"` + // the version that publish named + Version string `json:"version"` +} + // RunStarted is a run on request's start, with its members at their commits (ADR 0666 section 18). type RunStarted struct { At *time.Time `json:"at,omitempty"` diff --git a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json index 01778d80..a644b81e 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json @@ -907,6 +907,24 @@ "name" ] }, + "Published": { + "type": "object", + "properties": { + "artifact": { + "type": "string", + "description": "the package whose publish started the pipeline" + }, + "version": { + "type": "string", + "description": "the version that publish named" + } + }, + "description": "the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202)", + "required": [ + "artifact", + "version" + ] + }, "Pipeline": { "type": "object", "properties": { @@ -1002,6 +1020,10 @@ "version": { "type": "string", "description": "the version package-published names, for a pipeline the runner started from a publication or from outdated; none for any other (design 47 G 550)" + }, + "published": { + "$ref": "#/defs/Published", + "description": "for a pipeline a package's publish started, that publish (A-202)" } }, "description": "one pipeline as the runner keeps it in its state pipelines, by compare-and-set (design 47 *The runner's kept data*)", @@ -1191,6 +1213,10 @@ "at": { "type": "string", "format": "date-time" + }, + "published": { + "$ref": "#/defs/Published", + "description": "for a pipeline a package's publish started, that publish (A-202)" } }, "description": "a pipeline's start", diff --git a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts index 73c3dcb1..7a7cc39a 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts @@ -195,6 +195,8 @@ export interface Pipeline { "over-budget"?: boolean; /** the pipeline's key, .., a branch escaped as a key's name is */ "pipeline"?: string; + /** for a pipeline a package's publish started, that publish (A-202) */ + "published"?: Published; "repository"?: string; /** the catalog graph revision of the branch the pipeline runs for */ "revision"?: number; @@ -268,6 +270,8 @@ export interface PipelineStarted { "module": string; /** the pipeline's key, .., a branch escaped as a key's name is */ "pipeline": string; + /** for a pipeline a package's publish started, that publish (A-202) */ + "published"?: Published; /** the catalog graph revision of the branch the pipeline runs for */ "revision"?: number; /** the run on request's id, for a pipeline of a run; such a pipeline delivers nothing */ @@ -279,6 +283,14 @@ export interface PipelineStarted { "v": number; } +/** the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202) */ +export interface Published { + /** the package whose publish started the pipeline */ + "artifact": string; + /** the version that publish named */ + "version": string; +} + /** a run on request's start, with its members at their commits (ADR 0666 section 18) */ export interface RunStarted { "at"?: string; diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go index 64776e43..4b0559b7 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go @@ -86,21 +86,32 @@ 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 }} } func main() { r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}} tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) + tools = append(tools, HealthTool(r.Pipelines, kept{stdio.State(pipeline.StateConsumers)}, time.Now)) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go index 73d96db0..bb3dd146 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go @@ -17,7 +17,7 @@ func (none) Keys() ([]string, error) { return nil, nil } func (none) Value(string) (json.RawMessage, error) { return nil, nil } // TestTools: the bundle serves the five read verbs as . (§83, §306; RN15/AC1), and the first-view tools -// (§77, §200; RN2/AC4). +// (§77, §200; RN2/AC4), named with the prefix runner_ (§539; RN25). func TestTools(t *testing.T) { var names []string for _, tl := range Tools(pipeline.Reads{Pipelines: none{}, Kept: none{}}) { @@ -26,7 +26,9 @@ func TestTools(t *testing.T) { for _, tl := range FirstViewTools(stdio.State(views.StatusState), nil, nil) { names = append(names, tl.Name) } - want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept 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) + want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again runner_health" if got := strings.Join(names, " "); got != want { t.Errorf("tools %q, want %q", got, want) } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go index 0927551a..67ec25b7 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/approval.go @@ -56,7 +56,7 @@ var _ pipeline.Asker = busAsker{} // hearDecided wakes the runner on the operator-channel's decided for its own asks (§37, A-89): the warrant is read from the // router's record, so an event the record does not back yet is heard again. -func hearDecided(hub *process.EventHub, h *pipeline.Hearer, module string, logf func(string, ...any)) error { +func hearDecided(hub handlers, h *pipeline.Hearer, module string, logf func(string, ...any)) error { return hub.Handle(oc.SubjectDecided(module), func(m jetstream.Msg) { var ev oc.Warrant if err := json.Unmarshal(m.Data(), &ev); err != nil { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go new file mode 100644 index 00000000..81fb00b2 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go @@ -0,0 +1,43 @@ +package main + +import ( + "encoding/json" + "fmt" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/buscalls" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// readCandidate reads a candidate's revision whole through the catalog seat's verb revision, with the candidate's group and +// heads and the revision its event names, page by page (design 47 G A-180). +func readCandidate(call func(key string, body any) (json.RawMessage, error), ev catalog.CatalogRevisionEvent) (pipeline.Half, error) { + half := pipeline.Half{Revision: ev.Revision, Source: ev.Source, Entries: map[string]catalog.CatalogEntry{}} + after := "" + for { + raw, err := call(meshcatalogclient.RevisionVerb, catalog.RevisionInput{Branch: ev.Branch, Group: ev.Group, Heads: ev.Heads, Revision: ev.Revision, After: after}) + if err != nil { + return half, err + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return half, fmt.Errorf("%s answered: %s", meshcatalogclient.RevisionVerb, said) + } + var a catalog.RevisionAnswer + if err := json.Unmarshal(answer, &a); err != nil { + return half, fmt.Errorf("%s answered no revision: %w", meshcatalogclient.RevisionVerb, err) + } + if a.Revision == nil || a.Revision.Revision != ev.Revision || a.Revision.Group != ev.Group { + return half, fmt.Errorf("%s answered %+v for the candidate %s's revision %d", meshcatalogclient.RevisionVerb, a.Revision, ev.Group, ev.Revision) + } + for _, e := range a.Entries { + half.Entries[e.Module] = e + } + if a.Next == "" { + return half, nil + } + after = a.Next + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go new file mode 100644 index 00000000..21efca52 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go @@ -0,0 +1,36 @@ +package main + +import ( + "encoding/json" + "testing" + + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" +) + +// A-180: a run's candidate revision is read whole through the seat's revision verb with the candidate's group, heads and +// revision, page by page; an answer of another revision is refused. +func TestReadCandidateReadsTheRevisionWhole(t *testing.T) { + ev := catalog.CatalogRevisionEvent{Kind: "candidate", Group: "run-0a1b2c3d4e5f", Heads: "h", Revision: 12, Branch: "main"} + var asked []catalog.RevisionInput + answerRev := int64(12) + call := func(key string, body any) (json.RawMessage, error) { + in := body.(catalog.RevisionInput) + asked = append(asked, in) + a := catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: answerRev, Kind: "candidate", Group: in.Group}} + if in.After == "" { + a.Entries, a.Next = []catalog.CatalogEntry{{Module: "app", Fp: "src2:a"}}, "app" + } else { + a.Entries = []catalog.CatalogEntry{{Module: "lib", Fp: "src2:l"}} + } + return json.Marshal(a) + } + half, err := readCandidate(call, ev) + if err != nil || len(half.Entries) != 2 || half.Revision != 12 || len(asked) != 2 || asked[1].After != "app" || + asked[0].Group != ev.Group || asked[0].Heads != "h" || asked[0].Revision != 12 || asked[0].Branch != "main" { + t.Fatalf("read %+v (%v), asked %+v", half, err, asked) + } + answerRev = 13 + if _, err := readCandidate(call, ev); err == nil { + t.Error("an answer of another revision was taken") + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/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/drop.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drop.go new file mode 100644 index 00000000..8a3f1a05 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drop.go @@ -0,0 +1,41 @@ +package main + +import ( + "encoding/json" + "fmt" + + nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-sdk/go/buscalls" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// keptCaller is the runner's kept calls, to any holder of a seat or to one node's (process.Conn). +type keptCaller interface { + CallKept(key string, body any) (json.RawMessage, error) + CallKeptOn(node, key string, body any) (json.RawMessage, error) +} + +// dropAsk asks the build seat's `drop` of a superseded step's ask as a kept call (design 47 G §31, §350; A-182): of the +// holder on the node that took it, or, never taken, of any holder, which records it in the seat's shared drop key. The +// three build seats' drop take one input (nodebuildgo.DropInput; the manifest invokes each). +func dropAsk(calls keptCaller, by string) pipeline.Drop { + return func(buildSeat, node, ask, why string) error { + key := "seat:" + buildSeat + "." + nodebuildgo.VerbDrop + in := nodebuildgo.DropInput{ID: ask, Why: why, By: by} + var raw json.RawMessage + var err error + if node != "" { + raw, err = calls.CallKeptOn(node, key, in) + } else { + raw, err = calls.CallKept(key, in) + } + if err != nil { + return err + } + if _, said, ok := buscalls.Unwrap(raw); !ok { + return fmt.Errorf("%s answered: %s", key, said) + } + return nil + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index 2190fdfb..b5b04915 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -11,11 +11,13 @@ import ( "log" "os" "os/signal" + "strings" "sync/atomic" "syscall" "github.com/nats-io/nats.go" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-sdk/go/process" @@ -71,13 +73,26 @@ 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)] + var placer atomic.Pointer[func(catalog.CatalogRevisionEvent) error] rv, err := startViews(ctx, js, conn, cred, hub, calls.State("findings"), calls.Call, func(branch string) { if f := resumer.Load(); f != nil { (*f)(branch) } + }, func(_ context.Context, ev catalog.CatalogRevisionEvent) error { + if f := placer.Load(); f != nil { + return (*f)(ev) + } + return fmt.Errorf("the candidate %s's revision %d is heard before the runner can place it", ev.Group, ev.Revision) }, logger.Printf) if err != nil { return err @@ -112,25 +127,50 @@ 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 { + starter.Store = pipelineStore{ctx: ctx, kv: pipelines} + // A run's candidate revision places its pipelines' steps, its entries read through the seat's revision verb (A-180). + place := func(ev catalog.CatalogRevisionEvent) error { + if !strings.HasPrefix(ev.Group, pipeline.RunGroupPrefix) { + return nil + } + half, err := readCandidate(calls.CallKept, ev) + if err != nil { + return fmt.Errorf("the candidate %s's revision %d is not read, so its run is not placed: %w", ev.Group, ev.Revision, err) + } + n, err := starter.PlaceRun(ev, half, rv.Catalog.View) + logger.Printf("the candidate %s's revision %d placed %d pipelines of its run", ev.Group, ev.Revision, n) return err } - if err := hearToolchain(ctx, hub, toolchains, logger.Printf); err != nil { + placer.Store(&place) + // A newer commit or publish supersedes the older running pipelines it replaces, their asks dropped as kept calls (§31, + // §350; A-182, A-202). + starter.Supersede = func(newer seat.Pipeline) { + n, err := hearer.Supersede(newer, dropAsk(calls, cred.Module), dropBound(ctx, bounds)) + if err != nil { + logger.Printf("the pipelines %s supersedes: %d ended, then %v", newer.Pipeline, n, err) + } else if n > 0 { + logger.Printf("the pipeline %s superseded %d older pipelines", newer.Pipeline, n) + } + } + if err := hearStepEnded(heard, hearer, logger.Printf); err != nil { return err } - if err := hearAskTaken(ctx, hub, bounds, logger.Printf); err != nil { + if err := hearToolchain(ctx, heard, toolchains, logger.Printf); err != nil { + return err + } + if err := hearAskTaken(ctx, heard, bounds, hearer.Taken, logger.Printf); err != nil { return err } if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil { return err } - if err := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil { + if err := hearDecided(heard, hearer, cred.Module, logger.Printf); err != nil { return err } - if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil { + if err := hearPublished(ctx, heard, starter, calls, rv.Catalog, logger.Printf); err != nil { return err } - if err := hearTakenIn(hub, starter, rv.Catalog, logger.Printf); err != nil { + if err := hearTakenIn(heard, starter, rv.Catalog, logger.Printf); err != nil { return err } if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go index 5637d20c..feb5d854 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go @@ -6,6 +6,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "os" "path/filepath" "sync" @@ -20,6 +21,8 @@ import ( seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/testbus" + + "mesh-delivery-pipeline/internal/pipeline" ) type syncBuffer struct { @@ -53,7 +56,7 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes defer cancel() const module = "mesh-delivery-pipeline" - for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views"} { + for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views", "consumers"} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_" + state, History: 1}); err != nil { t.Fatal(err) } @@ -157,6 +160,43 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes }) waitFor(t, ctx, logs, func() bool { return readPipeline(ctx, pipelines, key, &p) && p.Outcome == seat.PipelineOutcomePassed }) + + // RN25 (§76, §468; A-166): the durable consumer's age is in the state consumers, none waiting once every message heard + // is acknowledged, and runner_health reads it healthy. + consumers, err := js.KeyValue(ctx, module+"_consumers") + if err != nil { + t.Fatal(err) + } + var 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..c75823dc 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -99,7 +99,7 @@ func newHearer(ctx context.Context, js jetstream.JetStream, cred process.Account // hearStepEnded takes each build seat's step-ended into its pipeline (§24; ADR 0666 §21): refused and said, never taken, // when it is not as the runner asked it, and heard again when a write failed. -func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string, ...any)) error { +func hearStepEnded(hub handlers, h *pipeline.Hearer, logf func(string, ...any)) error { for _, bs := range pipeline.StepSeats { bs := bs err := hub.Handle(bs.Prefix+"*", func(m jetstream.Msg) { @@ -135,9 +135,9 @@ func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string, return nil } -// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647): the ask is answered, so its five minutes -// end with no finding. -func hearAskTaken(ctx context.Context, hub *process.EventHub, bounds jetstream.KeyValue, logf func(string, ...any)) error { +// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647) and keeps on its step the node that took +// it (A-182): the ask is answered, so its five minutes end with no finding. +func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, taken func(ask, node string) error, logf func(string, ...any)) error { for _, ts := range pipeline.TakenSeats { ts := ts err := hub.Handle(ts.Subject, func(m jetstream.Msg) { @@ -152,6 +152,12 @@ func hearAskTaken(ctx context.Context, hub *process.EventHub, bounds jetstream.K _ = m.Nak() return } + // The step keeps the node that took its ask, so a drop of it is asked of that node's holder (A-182). + if err := taken(ev.Ask, ev.Node); err != nil { + logf("the node %s that took ask %s is not kept on its step, heard again: %v", ev.Node, ev.Ask, err) + _ = m.Nak() + return + } _ = m.Ack() }) if err != nil { @@ -252,7 +258,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..cecbacbd 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: run, review, rerun, accept, cancel, start and ask-again +// (§64, §306; RN12, RN14). +func writeHandlers(v pipeline.Verbs) map[string]process.SeatHandler { + return map[string]process.SeatHandler{ + seat.VerbRun: verbOf(func(c string, in seat.RunInput) (any, error) { return v.Run(c, in) }), + seat.VerbReview: verbOf(func(c string, in seat.ReviewInput) (any, error) { return v.Review(c, in) }), + seat.VerbRerun: verbOf(func(c string, in seat.RerunInput) (any, error) { return v.Rerun(c, in) }), + seat.VerbAccept: verbOf(func(c string, in seat.AcceptInput) (any, error) { return v.Accept(c, in) }), + seat.VerbCancel: verbOf(func(c string, in seat.CancelInput) (any, error) { return v.Cancel(c, in) }), + seat.VerbStart: verbOf(func(c string, in seat.StartInput) (any, error) { return v.Start(c, in) }), + seat.VerbAskAgain: verbOf(func(c string, in seat.AskAgainInput) (any, error) { return v.AskAgain(c, in) }), + } } // dropBound deletes the bound of an ask a cancelled pipeline no longer waits on. diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go index 9b5fb4d6..b535357a 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go @@ -2,11 +2,29 @@ package main import ( "encoding/json" + "slices" "testing" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + + "mesh-delivery-pipeline/internal/pipeline" ) +// RN12, RN14 (§306): the process answers every write verb of the seat, the bundle every read verb, so each verb the +// manifest claims is served. +func TestEveryWriteVerbIsServed(t *testing.T) { + var got []string + for v := range writeHandlers(pipeline.Verbs{}) { + got = append(got, v) + } + slices.Sort(got) + want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbReview, seat.VerbStart, seat.VerbRun} + 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..eca36ef1 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go @@ -25,7 +25,8 @@ type runnerViews struct { // startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or // after, then start hub once. func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cred process.Account, hub *process.EventHub, findingsState process.State, - call meshcatalogclient.Call, onReady func(branch string), logf func(string, ...any)) (*runnerViews, error) { + call meshcatalogclient.Call, onReady func(branch string), candidate func(ctx context.Context, ev catalog.CatalogRevisionEvent) error, + logf func(string, ...any)) (*runnerViews, error) { findings, err := findingWriter(ctx, cred, conn, js, findingsState, logf) if err != nil { return nil, err @@ -52,6 +53,11 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr if err != nil { return nil, err } + // A candidate's revision, heard once on the event's one subscription whatever branches are followed, places a run's + // pipelines (A-185, A-193, A-199). + if err := meshcatalogclient.WireCandidatesProcess(reader, candidate); err != nil { + return nil, err + } reader.Start(ctx, js, cred, status) return &runnerViews{Registrations: regs, Catalog: cv}, nil } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/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/run.go b/modules/mesh-delivery-pipeline/internal/pipeline/run.go new file mode 100644 index 00000000..bd7a68f8 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/run.go @@ -0,0 +1,239 @@ +package pipeline + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "maps" + "slices" + "strings" + "time" + + runnerclient "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// RunGroupPrefix is a run on request's candidate group before its id, `run-` (design 47 G A-160). +const RunGroupPrefix = "run-" + +// RunKey is the key of a run on request's pipeline, `..`: the run's id under the member's named branch +// (design 47 G §420, A-193). +func RunKey(branch, run, module string) string { + return seat.PipelinesByBranchRevisionModuleKey(keys.EscapeName(branch), run, module) +} + +// NewRunID is a run's id, a random 12-hex id, so two runs of one commit are two pipelines (A-180). +func NewRunID() (string, error) { + b := make([]byte, 6) + if _, err := rand.Read(b); err != nil { + return "", err + } + return hex.EncodeToString(b), nil +} + +// Run starts a run on request (design 47 G §64, §420; A-180, A-193): it writes each member's pipeline at once, running and +// with no steps, its `run` the run's new id, and says run-started, from which the indexer builds the run's candidate; the +// candidate's revision places the steps (PlaceRun). A member naming no module, commit or branch is refused, the run then +// writing nothing. caller is the one the bus proved. newID makes the run's id (NewRunID). +func (s *Starter) Run(caller string, in seat.RunInput, newID func() (string, error)) (seat.RunAnswer, error) { + if len(in.Members) == 0 { + return seat.RunAnswer{}, errors.New("a run names one or more members, each a module at a commit on a branch") + } + for _, m := range in.Members { + if m.Module == "" || m.Commit == "" || m.Branch == "" { + return seat.RunAnswer{}, fmt.Errorf("a run's member names its module, commit and branch: %+v", m) + } + } + id, err := newID() + if err != nil { + return seat.RunAnswer{}, fmt.Errorf("the run's id is not made: %w", err) + } + now := s.now() + var written []string + for _, m := range in.Members { + key := RunKey(m.Branch, id, m.Module) + p := seat.Pipeline{V: 1, Pipeline: key, Branch: m.Branch, Module: m.Module, Commit: m.Commit, Run: id, + Outcome: seat.PipelineOutcomeRunning, Started: &now, Steps: []seat.StepResult{}} + raw, err := json.Marshal(p) + if err == nil { + err = s.Pipelines.Create(key, raw) + } + if err != nil { + s.failRun(written, "the run's other pipelines were not written") + return seat.RunAnswer{}, fmt.Errorf("the pipeline %s of the run is not written: %w", key, err) + } + written = append(written, key) + } + first := in.Members[0] + last := first.Commit + if len(in.Members) > 1 { + last = RunGroupPrefix + id + } + ev := seat.RunStarted{V: 1, At: &now, By: caller, Members: in.Members, Run: id} + if err := s.Say(runnerclient.Subject(seat.EventRunStarted, first.Module, first.Branch, last), ev); err != nil { + s.failRun(written, "run-started was not said, so no candidate is built for it") + return seat.RunAnswer{}, fmt.Errorf("the run %s is not started: run-started is not said: %w", id, err) + } + return seat.RunAnswer{Run: id, Pipelines: written}, nil +} + +// failRun ends the run's pipelines written before its start failed, failed and saying why, so none stands running. +func (s *Starter) failRun(written []string, why string) { + if s.Store == nil { + return + } + for _, key := range written { + for { + p, rev, err := s.Store.Get(key) + if err != nil || p.Outcome != seat.PipelineOutcomeRunning { + break + } + now := s.now() + p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeFailed, why, &now + p.EndCount++ + if _, err := s.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + s.logf("the pipeline %s of a run not started stays running: %v", key, err) + } + break + } + } +} + +// PlaceRun places a run's pipelines from its candidate's revision (design 47 G §64, §373, A-180, A-193): ev is the +// catalog-revision of the candidate `run-`, half its entries read whole through the seat's revision verb. Each member's +// pipeline of the run not yet placed takes its fingerprint, the candidate's revision and its steps, by compare-and-set, and +// says pipeline-started; each further entry whose fingerprint the members change (another than in base, the branch's catalog +// half, or absent there) gets a pipeline of the run of its own, placed the same way. Another candidate's revision places +// nothing; a revision heard again places nothing twice. It says how many pipelines it placed. +func (s *Starter) PlaceRun(ev catalog.CatalogRevisionEvent, half Half, base func(branch string) (Half, bool)) (int, error) { + id, ok := strings.CutPrefix(ev.Group, RunGroupPrefix) + if !ok || id == "" { + return 0, nil + } + if s.Store == nil { + return 0, errors.New("the runner holds no pipelines to place a run's in") + } + pkeys, err := s.Store.Keys("*." + id + ".*") + if err != nil { + return 0, fmt.Errorf("the pipelines of the run %s are not listed: %w", id, err) + } + slices.Sort(pkeys) + members := map[string]bool{} + branch := "" + for _, key := range pkeys { + if p, _, err := s.Store.Get(key); err == nil && p.Run == id { + members[p.Module] = true + if branch == "" { + branch = p.Branch + } + } + } + if branch == "" { + s.logf("the candidate %s's revision %d names a run with no pipeline here: nothing is placed", ev.Group, ev.Revision) + return 0, nil + } + placed := 0 + var first error + keep := func(ok bool, err error) { + if err != nil && first == nil { + first = err + } + if ok { + placed++ + } + } + for _, key := range pkeys { + keep(s.placeRunPipeline(key, id, ev.Revision, half)) + } + // The dependents (§373): the candidate's entries the members change, beside the members, on the first member's branch. + trunk, ok := base(branch) + if b, err := keys.UnescapeName(ev.Branch); err == nil && b != "" && b != branch { + if h, held := base(b); held { + trunk, ok = h, true + } + } + if !ok { + s.logf("the run %s's dependents are not known: the catalog half of %s is not held, so its members run alone", id, branch) + return placed, first + } + for _, module := range slices.Sorted(maps.Keys(half.Entries)) { + e := half.Entries[module] + if members[module] { + continue + } + if was, held := trunk.Entries[module]; held && was.Fp == e.Fp { + continue + } + key := RunKey(branch, id, module) + now := s.now() + p := seat.Pipeline{V: 1, Pipeline: key, Branch: branch, Module: module, Commit: e.Commit, Run: id, + Outcome: seat.PipelineOutcomeRunning, Started: &now, Steps: []seat.StepResult{}} + raw, err := json.Marshal(p) + if err == nil { + err = s.Pipelines.Create(key, raw) + } + if err != nil && !errors.Is(err, ErrExists) { + keep(false, fmt.Errorf("the dependent %s of the run %s is not written: %w", module, id, err)) + continue + } + keep(s.placeRunPipeline(key, id, ev.Revision, half)) + } + return placed, first +} + +// placeRunPipeline places one pipeline of the run, written and not yet placed (no revision), from the candidate's half. +func (s *Starter) placeRunPipeline(key, id string, revision int64, half Half) (bool, error) { + for { + p, rev, err := s.Store.Get(key) + if err != nil { + return false, fmt.Errorf("the pipeline %s is not read: %w", key, err) + } + if p.Run != id || p.Outcome != seat.PipelineOutcomeRunning || p.Revision != 0 { + return false, nil + } + now := s.now() + e, ok := half.Entries[p.Module] + if !ok { + s.logf("the candidate of the run %s holds no entry of %s: its pipeline is placed with no fingerprint", id, p.Module) + } + p.Fingerprint, p.Repository, p.Revision = e.Fp, e.Repository, revision + var asks []Asked + if s.Registration != nil { + if reg, ok := s.Registration(p.Module); ok { + p.Steps, asks = Place(p, reg, half.Entries, now, s.Name) + } else { + s.logf("%s has no registration yet: its pipeline is placed with no step", key) + } + } + if _, err := s.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return false, fmt.Errorf("the pipeline %s cannot be written: %w", key, err) + } + ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: p.Branch, Module: p.Module, Commit: p.Commit, + Fingerprint: p.Fingerprint, Revision: p.Revision, Run: id} + if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, p.Module, p.Branch, p.Commit), ev); err != nil { + s.logf("pipeline-started of %s is not said: %v", key, err) + } + if s.Steps != nil && len(asks) > 0 { + s.carry(key, asks) + if err := askAll(s.Steps, s.Arm, key, asks); err != nil { + return true, err + } + } + return true, nil + } +} + +func (s *Starter) now() time.Time { + if s.Now != nil { + return s.Now().UTC() + } + return time.Now().UTC() +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/run_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/run_test.go new file mode 100644 index 00000000..b14ffe56 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/run_test.go @@ -0,0 +1,124 @@ +package pipeline + +import ( + "encoding/json" + "errors" + "strings" + "testing" + + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// memCreated is memStore written by Create too, as the state pipelines is. +type memCreated struct{ memStore } + +func (m memCreated) Create(key string, value []byte) error { + if _, ok := m.memStore[key]; ok { + return ErrExists + } + var p seat.Pipeline + if err := json.Unmarshal(value, &p); err != nil { + return err + } + m.memStore[key] = struct { + p seat.Pipeline + rev uint64 + }{p, 1} + return nil +} + +// RN12 (design 47 G §64, §373, §420; A-180, A-193): `run` writes each member's pipeline at once under the run's id and its +// named branch and says run-started; the candidate run-'s revision places each member's steps and a pipeline of each +// entry the members change, says pipeline-started once each, and places nothing twice; a member naming no branch is refused. +func TestRunOnRequest(t *testing.T) { + reg, entries := placeFixture() + store := memCreated{memStore{}} + var said []string + asks := &stepAsks{} + s := &Starter{Pipelines: store, Store: store, Steps: asks, Name: "the-runner", + Registration: func(m string) (catalog.Registration, bool) { + r := reg + r.Module = m + return r, m == "app" || m == "lib" + }, + Say: func(subject string, body any) error { + said = append(said, subject[strings.Index(subject, ".event.")+7:]) + return nil + }} + id := func() (string, error) { return "0a1b2c3d4e5f", nil } + if _, err := s.Run("sessions", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c1"}}}, id); err == nil { + t.Error("a member naming no branch was taken") + } + if len(store.memStore) != 0 || len(said) != 0 { + t.Fatalf("a refused run wrote %v and said %v", store.memStore, said) + } + a, err := s.Run("sessions", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c1", Branch: "fix/1"}}}, id) + if err != nil || a.Run != "0a1b2c3d4e5f" || len(a.Pipelines) != 1 { + t.Fatalf("run = %+v, %v", a, err) + } + key := a.Pipelines[0] + if key != RunKey("fix/1", "0a1b2c3d4e5f", "app") { + t.Errorf("the member's pipeline is %s, want the run's id under its named branch", key) + } + if p, _, _ := store.Get(key); p.Run != a.Run || p.Outcome != seat.PipelineOutcomeRunning || len(p.Steps) != 0 || p.Branch != "fix/1" || p.Source != 0 { + t.Errorf("the member's pipeline %s = %+v", key, p) + } + if len(said) != 1 || !strings.HasPrefix(said[0], seat.EventRunStarted+".app.") || !strings.HasSuffix(said[0], ".c1") { + t.Errorf("said %v, want run-started on the member's module, branch and commit", said) + } + cand := entries + cand["app"] = catalog.CatalogEntry{Module: "app", Commit: "c1", Fp: "src2:new", Repository: "o/app", Artifacts: entries["app"].Artifacts} + cand["lib"] = catalog.CatalogEntry{Module: "lib", Commit: "c0", Fp: "src2:lib2", Artifacts: entries["app"].Artifacts} + cand["same"] = catalog.CatalogEntry{Module: "same", Fp: "src2:same"} + base := func(string) (Half, bool) { + return Half{Entries: map[string]catalog.CatalogEntry{"lib": {Fp: "src2:lib1"}, "same": {Fp: "src2:same"}, "mesh-build-go": entries["mesh-build-go"]}}, true + } + ev := catalog.CatalogRevisionEvent{Kind: "candidate", Group: "run-" + a.Run, Revision: 12} + if n, err := s.PlaceRun(catalog.CatalogRevisionEvent{Kind: "candidate", Group: "try-3", Revision: 11}, Half{Entries: cand}, base); n != 0 || err != nil { + t.Errorf("a try's candidate placed %d (%v)", n, err) + } + said = nil + n, err := s.PlaceRun(ev, Half{Revision: 12, Entries: cand}, base) + if err != nil || n != 2 { + t.Fatalf("the candidate placed %d (%v), want the member and the dependent lib", n, err) + } + p, _, _ := store.Get(key) + if p.Fingerprint != "src2:new" || p.Revision != 12 || len(p.Steps) == 0 { + t.Errorf("the member's pipeline placed = %+v", p) + } + dep, _, err := store.Get(RunKey("fix/1", a.Run, "lib")) + if err != nil || dep.Run != a.Run || dep.Fingerprint != "src2:lib2" || len(dep.Steps) == 0 { + t.Errorf("the dependent's pipeline = %+v (%v)", dep, err) + } + if _, _, err := store.Get(RunKey("fix/1", a.Run, "same")); err == nil { + t.Error("an entry the members do not change got a pipeline") + } + if len(said) != 2 || !strings.HasPrefix(said[0], seat.EventPipelineStarted+".") || len(asks.ids) == 0 { + t.Errorf("placing said %v and asked %d steps", said, len(asks.ids)) + } + if n, _ := s.PlaceRun(ev, Half{Revision: 12, Entries: cand}, base); n != 0 { + t.Errorf("the revision heard again placed %d", n) + } + // A restart asks a run's steps again from no branch's half: they are its candidate's. + if n, err := s.Resume(store, "fix/1", Half{Entries: entries}); n != 0 || err != nil { + t.Errorf("a restart asked %d steps of the run again (%v)", n, err) + } + + // Several members say run-started on the first member's module and branch and run-; a run-started not said fails the + // pipelines written, so none stands running. + said = nil + two, err := s.Run("operator", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c2", Branch: "main"}, {Module: "lib", Commit: "c3", Branch: "main"}}}, + func() (string, error) { return "ffffffffffff", nil }) + if err != nil || len(two.Pipelines) != 2 || len(said) != 1 || said[0] != seat.EventRunStarted+".app.main.run-ffffffffffff" { + t.Errorf("a run of two: %+v (%v), said %v", two, err, said) + } + s.Say = func(string, any) error { return errors.New("bus away") } + three, err := s.Run("operator", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c4", Branch: "main"}}}, func() (string, error) { return "eeeeeeeeeeee", nil }) + if err == nil || three.Run != "" { + t.Errorf("a run whose run-started was not said answered %+v", three) + } + if p, _, _ := store.Get(RunKey("main", "eeeeeeeeeeee", "app")); p.Outcome != seat.PipelineOutcomeFailed { + t.Errorf("its pipeline is %s, want failed", p.Outcome) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 9e5737aa..e874fce0 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -37,6 +37,8 @@ type Begin struct { Revision int64 // Entries are the catalog half's entries of the pipeline's branch: the steps' artifacts and build seats are read from them. Entries map[string]catalog.CatalogEntry + // Published is the package's publish that started the pipeline, nil for any other start (A-202). + Published *seat.Published } // Key is a pipeline's key in the state pipelines, `..`, the branch escaped as a key's @@ -70,6 +72,10 @@ type Starter struct { Toolchain func(language string) (*nodebuildgo.ToolchainUsed, error) // Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none. Arm Arm + // Store is the state pipelines read and written by compare-and-set, where a run's pipelines are placed (PlaceRun). + Store Updater + // Supersede ends the older running pipelines the pipeline just written supersedes (§31; Hearer.Supersede); nil ends none. + Supersede func(newer seat.Pipeline) } // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second @@ -85,7 +91,7 @@ func (s *Starter) Start(b Begin) (bool, error) { key := Key(b.Branch, b.Source, b.Module) p := seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now, - Steps: []seat.StepResult{}} + Steps: []seat.StepResult{}, Published: b.Published} var asks []Asked if s.Registration != nil { if reg, ok := s.Registration(b.Module); ok { @@ -105,7 +111,7 @@ func (s *Starter) Start(b Begin) (bool, error) { return false, fmt.Errorf("the pipeline %s is not written: %w", key, err) } ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, - Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Revision: b.Revision} + Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Revision: b.Revision, Published: b.Published} commitOrGroup := b.Commit if b.Group != "" { commitOrGroup = b.Group @@ -113,6 +119,9 @@ func (s *Starter) Start(b Begin) (bool, error) { if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, b.Module, b.Branch, commitOrGroup), ev); err != nil { s.logf("pipeline-started of %s is not said: %v", key, err) } + if s.Supersede != nil { + s.Supersede(p) + } if s.Steps != nil && len(asks) > 0 { s.carry(key, asks) if err := askAll(s.Steps, s.Arm, key, asks); err != nil { @@ -149,7 +158,8 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) { } continue } - if p.Branch != branch || p.Outcome != seat.PipelineOutcomeRunning { + // A run on request's steps are placed from its candidate, not from the branch's half (PlaceRun). + if p.Branch != branch || p.Outcome != seat.PipelineOutcomeRunning || p.Run != "" { continue } reg, ok := s.Registration(p.Module) @@ -252,7 +262,8 @@ func (s *Starter) Published(ev builds.PackagePublished, outdated func(catalog.Ou if o.Kind != outdatedKind || o.Module == "" { continue } - b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision, Entries: half.Entries} + b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision, Entries: half.Entries, + Published: &seat.Published{Artifact: ev.Artifact, Version: ev.Version}} if e, ok := half.Entries[o.Module]; ok { b.Commit, b.Fp, b.Repository = e.Commit, e.Fp, e.Repository } else { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go index 7fcfec71..63fef192 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go @@ -27,11 +27,15 @@ func (c created) Create(key string, value []byte) error { func TestStarts(t *testing.T) { store := created{} var said []string + var superseding []string s := &Starter{Pipelines: store, Say: func(subject string, body any) error { ev := body.(seat.PipelineStarted) said = append(said, subject+" "+ev.Pipeline) + if ev.Published == nil || ev.Published.Artifact != "mesh-sdk:go" || ev.Published.Version != "0.1.16" { + t.Errorf("pipeline-started of %s says the publish %+v (A-202)", ev.Pipeline, ev.Published) + } return nil - }} + }, Supersede: func(newer seat.Pipeline) { superseding = append(superseding, newer.Pipeline) }} view := func(branch string) (Half, bool) { return Half{Revision: 9, Entries: map[string]catalog.CatalogEntry{ "app": {Module: "app", Commit: "c1", Fp: "src2:aa", Repository: "o/app"}}}, true @@ -51,9 +55,13 @@ func TestStarts(t *testing.T) { } var p seat.Pipeline if err := json.Unmarshal(store["main.41.app"], &p); err != nil || p.Commit != "c1" || p.Fingerprint != "src2:aa" || p.Revision != 9 || - p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 { + p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 || p.Published == nil || p.Published.Version != "0.1.16" { t.Errorf("the pipeline main.41.app = %+v (%v)", p, err) } + // Each pipeline written supersedes the older ones it replaces (§31). + if len(superseding) != 2 { + t.Errorf("the publish's pipelines superseding: %v, want both", superseding) + } if _, ok := store["main.41.ordered"]; ok { t.Error("an ordering edge starts no pipeline") } @@ -66,6 +74,7 @@ func TestStarts(t *testing.T) { if !strings.HasPrefix(said[0], "mesh.seat.mesh-pipeline-runner.event.pipeline-started.app.main.c1 main.41.app") { t.Errorf("pipeline-started said on %q", said[0]) } + s.Say = func(string, any) error { return nil } // A branch holding a dot is escaped in the key, so the key reads back; a start naming no source revision is refused. if _, err := s.Start(Begin{Branch: "release.1", Source: 3, Module: "app"}); err != nil { t.Fatal(err) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/supersede.go b/modules/mesh-delivery-pipeline/internal/pipeline/supersede.go new file mode 100644 index 00000000..292f171f --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/supersede.go @@ -0,0 +1,155 @@ +package pipeline + +import ( + "errors" + "fmt" + "time" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// Drop asks a build seat's `drop` of a superseded step's ask as a kept call (design 47 G §31, §350; A-182): of node's holder +// when a holder took it (node the one its ask-taken named), else of any holder, which records it in the seat's shared drop +// key so no holder takes it. why names the newer commit. +type Drop func(buildSeat, node, ask, why string) error + +// Taken notes on the running step asked under ask the node whose holder said ask-taken for it (A-182), so a drop of it is +// asked of that node's holder. A step no pipeline keeps running under ask takes nothing. +func (h *Hearer) Taken(ask, node string) error { + if ask == "" || node == "" { + return nil + } + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + return fmt.Errorf("the pipelines are not listed: %w", err) + } + for _, key := range pkeys { + for { + p, rev, err := h.Store.Get(key) + if err != nil { + break + } + i := stepOfAsk(p, ask) + if i < 0 || p.Steps[i].Outcome != seat.StepResultOutcomeRunning || p.Steps[i].Node == node { + break + } + p.Steps[i].Node = node + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return fmt.Errorf("pipeline %s cannot be written: %w", key, err) + } + return nil + } + } + return nil +} + +// Supersede ends, for the pipeline newer just written, each older running pipeline of the same watched branch and module +// whose fingerprint newer's commit changed, and for a publish's pipeline each an older publish of the package started +// (design 47 G §31, §119; ADR 0666 §9; A-202): the pipeline and each step not ended +// `superseded`, naming the newer commit, and it says pipeline-ended. A module the newer commit left as it was finishes its +// pipeline. The ask of each superseded step on a build seat that serves `drop` is dropped (drop; its bound deleted through +// unbound); a step on mesh-builds or mesh-wants, whose seats serve no drop, is only ended superseded (A-182). A run on +// request's pipeline neither supersedes nor is superseded. It says how many pipelines it ended. +func (h *Hearer) Supersede(newer seat.Pipeline, drop Drop, unbound func(ask string) error) (int, error) { + if newer.Run != "" || newer.Source <= 0 { + return 0, nil + } + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + return 0, fmt.Errorf("the pipelines are not listed: %w", err) + } + why := supersededWhy(newer) + ended := 0 + var first error + for _, key := range pkeys { + if key == newer.Pipeline { + continue + } + done, err := h.supersedeOne(key, newer, why, drop, unbound) + if err != nil && first == nil { + first = err + } + if done { + ended++ + } + } + return ended, first +} + +// superseded says whether old is a running pipeline newer supersedes (§31, A-202): of the same watched branch and an older +// source revision, either of the same module at another fingerprint, or, for newer a package's publish started, one an older +// publish of that package started (its version another, made from an older source revision of the branch). +func superseded(old, newer seat.Pipeline) bool { + if old.Outcome != seat.PipelineOutcomeRunning || old.Run != "" || old.Branch != newer.Branch || old.Source >= newer.Source { + return false + } + if newer.Published != nil && old.Published != nil && old.Published.Artifact == newer.Published.Artifact { + return old.Published.Version != newer.Published.Version + } + return old.Module == newer.Module && old.Fingerprint != newer.Fingerprint +} + +// supersededWhy names the newer commit, or the newer publish, that ended a pipeline. +func supersededWhy(newer seat.Pipeline) string { + if newer.Published != nil { + return fmt.Sprintf("the newer publish %s %s of %s supersedes it", newer.Published.Artifact, newer.Published.Version, newer.Branch) + } + return fmt.Sprintf("the newer commit %s of %s changed %s's fingerprint", newer.Commit, newer.Branch, newer.Module) +} + +func (h *Hearer) supersedeOne(key string, newer seat.Pipeline, why string, drop Drop, unbound func(string) error) (bool, error) { + for { + p, rev, err := h.Store.Get(key) + if err != nil || !superseded(p, newer) { + return false, nil + } + now := h.now() + var drops []seat.StepResult + var changed []int + for i := range p.Steps { + s := &p.Steps[i] + if s.Outcome != seat.StepResultOutcomeRunning && s.Outcome != seat.StepResultOutcomeWaiting { + continue + } + if s.Ask != "" && BoundSeat(s.Seat) { + drops = append(drops, *s) + } + s.Outcome, s.Why, s.Ended = seat.StepResultOutcomeSuperseded, why, ptrTime(now) + changed = append(changed, i) + } + p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeSuperseded, why, ptrTime(now) + p.EndCount++ + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return false, fmt.Errorf("pipeline %s cannot be written: %w", key, err) + } + var first error + for _, i := range changed { + h.logEnd(key, p.Steps[i], now) + if err := h.say(seat.EventStepEnded, p, &p.Steps[i]); err != nil && first == nil { + first = fmt.Errorf("pipeline %s's %s is written but not said: %w", key, seat.EventStepEnded, err) + } + } + for _, s := range drops { + if drop != nil { + if err := drop(s.Seat, s.Node, s.Ask, why); err != nil { + h.logf("the drop of ask %s of the superseded %s is not answered: %v", s.Ask, key, err) + } + } + if unbound != nil { + if err := unbound(s.Ask); err != nil { + h.logf("the bound of ask %s of the superseded %s is not dropped: %v", s.Ask, key, err) + } + } + } + if err := h.said(p, false, true); err != nil && first == nil { + first = err + } + return true, first + } +} + +func ptrTime(t time.Time) *time.Time { return &t } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go new file mode 100644 index 00000000..d6b96b54 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go @@ -0,0 +1,93 @@ +package pipeline + +import ( + "slices" + "strings" + "testing" + + builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" +) + +// RN8 (design 47 G §31, §119, §350; A-182, A-202): a step keeps the node that took its ask; a newer commit changing the +// module's fingerprint, or a newer publish of the package that started a pipeline, ends the older running pipeline +// superseded, its asks dropped of the node that took each (any holder for one never taken), a mesh-builds step only ended; +// an unchanged fingerprint, a run on request and an equal publish supersede nothing. +func TestSupersede(t *testing.T) { + h, store, said, _, _, _ := fixture() + old := store[pkey] + old.p.Steps = append(old.p.Steps, seat.StepResult{Artifact: "program", Step: seat.StepResultStepComposition, Status: seat.StepResultStatusRequired, + Outcome: seat.StepResultOutcomeRunning, Ask: "b1", Seat: builds.Seat}) + store[pkey] = old + if err := h.Taken("a1", "n7"); err != nil { + t.Fatal(err) + } + if p, _, _ := store.Get(pkey); p.Steps[0].Node != "n7" || p.Steps[1].Node != "" { + t.Fatalf("after ask-taken of a1 on n7 the steps name %q and %q", p.Steps[0].Node, p.Steps[1].Node) + } + type dropped struct{ seat, node, ask string } + var drops []dropped + var unbound []string + drop := func(s, node, ask, why string) error { + if !strings.Contains(why, "c2") { + t.Errorf("the drop of %s names no newer commit: %q", ask, why) + } + drops = append(drops, dropped{s, node, ask}) + return nil + } + unbind := func(ask string) error { unbound = append(unbound, ask); return nil } + newer := func(key string, source int64, fp string, run string, pub *seat.Published) seat.Pipeline { + return seat.Pipeline{Pipeline: key, Branch: "main", Source: source, Module: "app", Commit: "c2", Fingerprint: fp, Run: run, Published: pub, + Outcome: seat.PipelineOutcomeRunning} + } + for _, c := range []struct { + name string + newer seat.Pipeline + }{ + {"an unchanged fingerprint", newer("main.6.app", 6, "src2:aa", "", nil)}, + {"a run on request", newer("main.r1.app", 6, "src2:bb", "r1", nil)}, + {"an older source revision", newer("main.4.app", 4, "src2:bb", "", nil)}, + } { + if n, err := h.Supersede(c.newer, drop, unbind); n != 0 || err != nil { + t.Errorf("%s superseded %d (%v)", c.name, n, err) + } + } + *said = nil + if n, err := h.Supersede(newer("main.6.app", 6, "src2:bb", "", nil), drop, unbind); n != 1 || err != nil { + t.Fatalf("a changed fingerprint superseded %d (%v)", n, err) + } + p, _, _ := store.Get(pkey) + if p.Outcome != seat.PipelineOutcomeSuperseded || p.EndCount != 1 || !strings.Contains(p.Why, "c2") { + t.Errorf("the older pipeline is %s (%d ends): %q", p.Outcome, p.EndCount, p.Why) + } + for _, s := range p.Steps { + if s.Outcome != seat.StepResultOutcomeSuperseded { + t.Errorf("the %s step is %s, want superseded", s.Step, s.Outcome) + } + } + want := []dropped{{nodebuildgo.Seat, "n7", "a1"}, {nodebuildgo.Seat, "", "a2"}} + if !slices.Equal(drops, want) || !slices.Equal(unbound, []string{"a1", "a2"}) { + t.Errorf("dropped %+v and unbound %v, want %+v (the mesh-builds step only ended)", drops, unbound, want) + } + if n := len(*said); n != 5 || (*said)[4] != seat.EventPipelineEnded+".app.main.c1" { + t.Errorf("said %v, want four step-ended then pipeline-ended", *said) + } + if n, _ := h.Supersede(newer("main.7.app", 7, "src2:cc", "", nil), drop, unbind); n != 0 { + t.Errorf("a superseded pipeline was superseded again: %d", n) + } + + // A newer publish of a package supersedes the pipelines an older publish of it started, whatever their module. + pub := memStore{"main.5.lib": {p: seat.Pipeline{Pipeline: "main.5.lib", Branch: "main", Source: 5, Module: "lib", Fingerprint: "src2:aa", + Outcome: seat.PipelineOutcomeRunning, Published: &seat.Published{Artifact: "core:pkg", Version: "v1"}}, rev: 1}} + h.Store = pub + if n, _ := h.Supersede(newer("main.6.app", 6, "src2:aa", "", &seat.Published{Artifact: "core:pkg", Version: "v1"}), drop, unbind); n != 0 { + t.Errorf("a publish of the same version superseded %d", n) + } + if n, err := h.Supersede(newer("main.6.app", 6, "src2:aa", "", &seat.Published{Artifact: "core:pkg", Version: "v2"}), drop, unbind); n != 1 || err != nil { + t.Fatalf("a newer publish superseded %d (%v)", n, err) + } + if p, _, _ := pub.Get("main.5.lib"); p.Outcome != seat.PipelineOutcomeSuperseded || !strings.Contains(p.Why, "v2") { + t.Errorf("the older publish's pipeline is %s: %q", p.Outcome, p.Why) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go index 65a924a6..6ede0c1b 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go @@ -334,3 +334,77 @@ func (v Verbs) askOf(p seat.Pipeline, s seat.StepResult, caller string, now time } return nil, fmt.Errorf("the %s step of %s is placed no ask now", s.Step, p.Pipeline) } + +// Run starts a run on request for the caller the bus proved (design 47 G §64; Starter.Run): its pipelines written, its +// run-started said, its steps placed when its candidate's revision is heard. +func (v Verbs) Run(caller string, in seat.RunInput) (seat.RunAnswer, error) { + if v.Starter == nil { + return seat.RunAnswer{}, errors.New("the runner cannot start a run: it holds no starter") + } + return v.Starter.Run(caller, in, NewRunID) +} + +// Start starts by hand a delivering pipeline for the head of a watched branch the runner's catalog half holds, recorded in +// the hand-act log first (§306, §394): the pipeline of that branch's source revision and the module, as a take-in would +// start it. A head whose pipeline is written already starts nothing and is refused, naming the pipeline. +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..457b0298 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,8 +69,10 @@ func TestManifest(t *testing.T) { equalSets(m.Streams[0].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"})}, // RN24 (§368): the version as MAJOR.MINOR, 0.1 with none published. {"version 0.1", m.Version == "0.1"}, - {"old objects section", equalSets(objects, []string{seat.StateStepOutputs})}, - {"step-outputs 1 GiB", len(m.Objects) == 1 && m.Objects[0].MaxMiB == 1024}, + // RN25 (§477, §533, §59): a foundation part whose seat has verbs and events declares its sent store, 256 MiB. + {"old objects section", equalSets(objects, []string{seat.StateStepOutputs, catalogue.SentObjectsLocal})}, + {"step-outputs 1 GiB", objectMiB[seat.StateStepOutputs] == 1024}, + {"sent-objects 256 MiB", objectMiB[catalogue.SentObjectsLocal] == 256}, // §333, §54, §44, §53, §376, §24, §37, §4, §77: what the runner hears. {"consumes", equalSets(m.Consumes, []string{ "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", @@ -75,13 +84,18 @@ func TestManifest(t *testing.T) { {"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act", // RN3 (§53, §334): a publication's dependents behind. "seat:mesh-catalog.outdated", + // RN8 (§31, §350; A-182): a superseded step's ask dropped of the build seat that holds it. + "seat:node-build-go.drop", "seat:node-build-node.drop", "seat:node-build-image.drop", // A-74, A-89: who holds the seat operator-channel, whose state asks holds the router's record of the approval ask. "seat:mesh-module-assigner.assignments", // A-102 (§22, §24): the accepted kinds the runner puts its steps on. "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"})}, + // 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 +131,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..e16d652f 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -23,8 +23,9 @@ } ], "tools": [ - "pipeline_first_views", - "pipeline_first_view_again" + "runner_first_views", + "runner_first_view_again", + "runner_health" ], "consumes": [ "mesh-delivery-indexer.take-in.taken", @@ -53,7 +54,10 @@ "seat:mesh-wants.step", "seat:node-build-go.check.step", "seat:node-build-node.check.step", - "seat:node-build-image.check.step" + "seat:node-build-image.check.step", + "seat:node-build-go.drop", + "seat:node-build-node.drop", + "seat:node-build-image.drop" ], "uses": [ "operator-channel" @@ -61,11 +65,13 @@ "state": [ { "name": "pipelines", - "history": 1 + "history": 1, + "max-mib": 256 }, { "name": "kept", - "history": 1 + "history": 1, + "max-mib": 128 }, { "name": "aged", @@ -87,6 +93,10 @@ { "name": "first-views", "history": 1 + }, + { + "name": "consumers", + "history": 1 } ], "streams": [ @@ -105,6 +115,10 @@ { "name": "step-outputs", "max-mib": 1024 + }, + { + "name": "sent-objects", + "max-mib": 256 } ], "own-secrets": { @@ -140,7 +154,8 @@ }, "user": "mesh-delivery-pipeline", "health": { - "kind": "unit" + "kind": "tool", + "tool": "runner_health" } } ],