diff --git a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go index 7b0bfe66..307ffd31 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go @@ -432,8 +432,10 @@ type Pipeline struct { // yes when the module is over its test budget (design 47 G section 419) OverBudget bool `json:"over-budget,omitempty"` // the pipeline's key, .., a branch escaped as a key's name is - Pipeline string `json:"pipeline,omitempty"` - Repository string `json:"repository,omitempty"` + Pipeline string `json:"pipeline,omitempty"` + // for a pipeline a package's publish started, that publish (A-202) + Published *Published `json:"published,omitempty"` + Repository string `json:"repository,omitempty"` // the catalog graph revision of the branch the pipeline runs for Revision int64 `json:"revision,omitempty"` // the run on request's id, for a pipeline of a run; such a pipeline delivers nothing @@ -545,6 +547,8 @@ type PipelineStarted struct { Module string `json:"module"` // the pipeline's key, .., a branch escaped as a key's name is Pipeline string `json:"pipeline"` + // for a pipeline a package's publish started, that publish (A-202) + Published *Published `json:"published,omitempty"` // the catalog graph revision of the branch the pipeline runs for Revision int64 `json:"revision,omitempty"` // the run on request's id, for a pipeline of a run; such a pipeline delivers nothing @@ -556,6 +560,14 @@ type PipelineStarted struct { V int64 `json:"v"` } +// Published is the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202). +type Published struct { + // the package whose publish started the pipeline + Artifact string `json:"artifact"` + // the version that publish named + Version string `json:"version"` +} + // RunStarted is a run on request's start, with its members at their commits (ADR 0666 section 18). type RunStarted struct { At *time.Time `json:"at,omitempty"` diff --git a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json index 01778d80..a644b81e 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json @@ -907,6 +907,24 @@ "name" ] }, + "Published": { + "type": "object", + "properties": { + "artifact": { + "type": "string", + "description": "the package whose publish started the pipeline" + }, + "version": { + "type": "string", + "description": "the version that publish named" + } + }, + "description": "the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202)", + "required": [ + "artifact", + "version" + ] + }, "Pipeline": { "type": "object", "properties": { @@ -1002,6 +1020,10 @@ "version": { "type": "string", "description": "the version package-published names, for a pipeline the runner started from a publication or from outdated; none for any other (design 47 G 550)" + }, + "published": { + "$ref": "#/defs/Published", + "description": "for a pipeline a package's publish started, that publish (A-202)" } }, "description": "one pipeline as the runner keeps it in its state pipelines, by compare-and-set (design 47 *The runner's kept data*)", @@ -1191,6 +1213,10 @@ "at": { "type": "string", "format": "date-time" + }, + "published": { + "$ref": "#/defs/Published", + "description": "for a pipeline a package's publish started, that publish (A-202)" } }, "description": "a pipeline's start", diff --git a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts index 73c3dcb1..7a7cc39a 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts @@ -195,6 +195,8 @@ export interface Pipeline { "over-budget"?: boolean; /** the pipeline's key, .., a branch escaped as a key's name is */ "pipeline"?: string; + /** for a pipeline a package's publish started, that publish (A-202) */ + "published"?: Published; "repository"?: string; /** the catalog graph revision of the branch the pipeline runs for */ "revision"?: number; @@ -268,6 +270,8 @@ export interface PipelineStarted { "module": string; /** the pipeline's key, .., a branch escaped as a key's name is */ "pipeline": string; + /** for a pipeline a package's publish started, that publish (A-202) */ + "published"?: Published; /** the catalog graph revision of the branch the pipeline runs for */ "revision"?: number; /** the run on request's id, for a pipeline of a run; such a pipeline delivers nothing */ @@ -279,6 +283,14 @@ export interface PipelineStarted { "v": number; } +/** the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202) */ +export interface Published { + /** the package whose publish started the pipeline */ + "artifact": string; + /** the version that publish named */ + "version": string; +} + /** a run on request's start, with its members at their commits (ADR 0666 section 18) */ export interface RunStarted { "at"?: string; diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drop.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drop.go new file mode 100644 index 00000000..8a3f1a05 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drop.go @@ -0,0 +1,41 @@ +package main + +import ( + "encoding/json" + "fmt" + + nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-sdk/go/buscalls" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// keptCaller is the runner's kept calls, to any holder of a seat or to one node's (process.Conn). +type keptCaller interface { + CallKept(key string, body any) (json.RawMessage, error) + CallKeptOn(node, key string, body any) (json.RawMessage, error) +} + +// dropAsk asks the build seat's `drop` of a superseded step's ask as a kept call (design 47 G §31, §350; A-182): of the +// holder on the node that took it, or, never taken, of any holder, which records it in the seat's shared drop key. The +// three build seats' drop take one input (nodebuildgo.DropInput; the manifest invokes each). +func dropAsk(calls keptCaller, by string) pipeline.Drop { + return func(buildSeat, node, ask, why string) error { + key := "seat:" + buildSeat + "." + nodebuildgo.VerbDrop + in := nodebuildgo.DropInput{ID: ask, Why: why, By: by} + var raw json.RawMessage + var err error + if node != "" { + raw, err = calls.CallKeptOn(node, key, in) + } else { + raw, err = calls.CallKept(key, in) + } + if err != nil { + return err + } + if _, said, ok := buscalls.Unwrap(raw); !ok { + return fmt.Errorf("%s answered: %s", key, said) + } + return nil + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index 9712039e..1bfb5453 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -119,13 +119,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro logger.Printf("the steps placed on %s and not yet asked: %d asked again", branch, n) } resumer.Store(&resume) + // A newer commit or publish supersedes the older running pipelines it replaces, their asks dropped as kept calls (§31, + // §350; A-182, A-202). + starter.Supersede = func(newer seat.Pipeline) { + n, err := hearer.Supersede(newer, dropAsk(calls, cred.Module), dropBound(ctx, bounds)) + if err != nil { + logger.Printf("the pipelines %s supersedes: %d ended, then %v", newer.Pipeline, n, err) + } else if n > 0 { + logger.Printf("the pipeline %s superseded %d older pipelines", newer.Pipeline, n) + } + } if err := hearStepEnded(heard, hearer, logger.Printf); err != nil { return err } if err := hearToolchain(ctx, heard, toolchains, logger.Printf); err != nil { return err } - if err := hearAskTaken(ctx, heard, bounds, logger.Printf); err != nil { + if err := hearAskTaken(ctx, heard, bounds, hearer.Taken, logger.Printf); err != nil { return err } if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go index 7322ee79..c75823dc 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -135,9 +135,9 @@ func hearStepEnded(hub handlers, h *pipeline.Hearer, logf func(string, ...any)) return nil } -// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647): the ask is answered, so its five minutes -// end with no finding. -func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, logf func(string, ...any)) error { +// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647) and keeps on its step the node that took +// it (A-182): the ask is answered, so its five minutes end with no finding. +func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, taken func(ask, node string) error, logf func(string, ...any)) error { for _, ts := range pipeline.TakenSeats { ts := ts err := hub.Handle(ts.Subject, func(m jetstream.Msg) { @@ -152,6 +152,12 @@ func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, _ = m.Nak() return } + // The step keeps the node that took its ask, so a drop of it is asked of that node's holder (A-182). + if err := taken(ev.Ask, ev.Node); err != nil { + logf("the node %s that took ask %s is not kept on its step, heard again: %v", ev.Node, ev.Ask, err) + _ = m.Nak() + return + } _ = m.Ack() }) if err != nil { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 9e5737aa..0d8c561c 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -37,6 +37,8 @@ type Begin struct { Revision int64 // Entries are the catalog half's entries of the pipeline's branch: the steps' artifacts and build seats are read from them. Entries map[string]catalog.CatalogEntry + // Published is the package's publish that started the pipeline, nil for any other start (A-202). + Published *seat.Published } // Key is a pipeline's key in the state pipelines, `..`, the branch escaped as a key's @@ -70,6 +72,8 @@ type Starter struct { Toolchain func(language string) (*nodebuildgo.ToolchainUsed, error) // Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none. Arm Arm + // Supersede ends the older running pipelines the pipeline just written supersedes (§31; Hearer.Supersede); nil ends none. + Supersede func(newer seat.Pipeline) } // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second @@ -85,7 +89,7 @@ func (s *Starter) Start(b Begin) (bool, error) { key := Key(b.Branch, b.Source, b.Module) p := seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now, - Steps: []seat.StepResult{}} + Steps: []seat.StepResult{}, Published: b.Published} var asks []Asked if s.Registration != nil { if reg, ok := s.Registration(b.Module); ok { @@ -105,7 +109,7 @@ func (s *Starter) Start(b Begin) (bool, error) { return false, fmt.Errorf("the pipeline %s is not written: %w", key, err) } ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, - Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Revision: b.Revision} + Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Revision: b.Revision, Published: b.Published} commitOrGroup := b.Commit if b.Group != "" { commitOrGroup = b.Group @@ -113,6 +117,9 @@ func (s *Starter) Start(b Begin) (bool, error) { if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, b.Module, b.Branch, commitOrGroup), ev); err != nil { s.logf("pipeline-started of %s is not said: %v", key, err) } + if s.Supersede != nil { + s.Supersede(p) + } if s.Steps != nil && len(asks) > 0 { s.carry(key, asks) if err := askAll(s.Steps, s.Arm, key, asks); err != nil { @@ -252,7 +259,8 @@ func (s *Starter) Published(ev builds.PackagePublished, outdated func(catalog.Ou if o.Kind != outdatedKind || o.Module == "" { continue } - b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision, Entries: half.Entries} + b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision, Entries: half.Entries, + Published: &seat.Published{Artifact: ev.Artifact, Version: ev.Version}} if e, ok := half.Entries[o.Module]; ok { b.Commit, b.Fp, b.Repository = e.Commit, e.Fp, e.Repository } else { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go index 7fcfec71..63fef192 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go @@ -27,11 +27,15 @@ func (c created) Create(key string, value []byte) error { func TestStarts(t *testing.T) { store := created{} var said []string + var superseding []string s := &Starter{Pipelines: store, Say: func(subject string, body any) error { ev := body.(seat.PipelineStarted) said = append(said, subject+" "+ev.Pipeline) + if ev.Published == nil || ev.Published.Artifact != "mesh-sdk:go" || ev.Published.Version != "0.1.16" { + t.Errorf("pipeline-started of %s says the publish %+v (A-202)", ev.Pipeline, ev.Published) + } return nil - }} + }, Supersede: func(newer seat.Pipeline) { superseding = append(superseding, newer.Pipeline) }} view := func(branch string) (Half, bool) { return Half{Revision: 9, Entries: map[string]catalog.CatalogEntry{ "app": {Module: "app", Commit: "c1", Fp: "src2:aa", Repository: "o/app"}}}, true @@ -51,9 +55,13 @@ func TestStarts(t *testing.T) { } var p seat.Pipeline if err := json.Unmarshal(store["main.41.app"], &p); err != nil || p.Commit != "c1" || p.Fingerprint != "src2:aa" || p.Revision != 9 || - p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 { + p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 || p.Published == nil || p.Published.Version != "0.1.16" { t.Errorf("the pipeline main.41.app = %+v (%v)", p, err) } + // Each pipeline written supersedes the older ones it replaces (§31). + if len(superseding) != 2 { + t.Errorf("the publish's pipelines superseding: %v, want both", superseding) + } if _, ok := store["main.41.ordered"]; ok { t.Error("an ordering edge starts no pipeline") } @@ -66,6 +74,7 @@ func TestStarts(t *testing.T) { if !strings.HasPrefix(said[0], "mesh.seat.mesh-pipeline-runner.event.pipeline-started.app.main.c1 main.41.app") { t.Errorf("pipeline-started said on %q", said[0]) } + s.Say = func(string, any) error { return nil } // A branch holding a dot is escaped in the key, so the key reads back; a start naming no source revision is refused. if _, err := s.Start(Begin{Branch: "release.1", Source: 3, Module: "app"}); err != nil { t.Fatal(err) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/supersede.go b/modules/mesh-delivery-pipeline/internal/pipeline/supersede.go new file mode 100644 index 00000000..292f171f --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/supersede.go @@ -0,0 +1,155 @@ +package pipeline + +import ( + "errors" + "fmt" + "time" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// Drop asks a build seat's `drop` of a superseded step's ask as a kept call (design 47 G §31, §350; A-182): of node's holder +// when a holder took it (node the one its ask-taken named), else of any holder, which records it in the seat's shared drop +// key so no holder takes it. why names the newer commit. +type Drop func(buildSeat, node, ask, why string) error + +// Taken notes on the running step asked under ask the node whose holder said ask-taken for it (A-182), so a drop of it is +// asked of that node's holder. A step no pipeline keeps running under ask takes nothing. +func (h *Hearer) Taken(ask, node string) error { + if ask == "" || node == "" { + return nil + } + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + return fmt.Errorf("the pipelines are not listed: %w", err) + } + for _, key := range pkeys { + for { + p, rev, err := h.Store.Get(key) + if err != nil { + break + } + i := stepOfAsk(p, ask) + if i < 0 || p.Steps[i].Outcome != seat.StepResultOutcomeRunning || p.Steps[i].Node == node { + break + } + p.Steps[i].Node = node + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return fmt.Errorf("pipeline %s cannot be written: %w", key, err) + } + return nil + } + } + return nil +} + +// Supersede ends, for the pipeline newer just written, each older running pipeline of the same watched branch and module +// whose fingerprint newer's commit changed, and for a publish's pipeline each an older publish of the package started +// (design 47 G §31, §119; ADR 0666 §9; A-202): the pipeline and each step not ended +// `superseded`, naming the newer commit, and it says pipeline-ended. A module the newer commit left as it was finishes its +// pipeline. The ask of each superseded step on a build seat that serves `drop` is dropped (drop; its bound deleted through +// unbound); a step on mesh-builds or mesh-wants, whose seats serve no drop, is only ended superseded (A-182). A run on +// request's pipeline neither supersedes nor is superseded. It says how many pipelines it ended. +func (h *Hearer) Supersede(newer seat.Pipeline, drop Drop, unbound func(ask string) error) (int, error) { + if newer.Run != "" || newer.Source <= 0 { + return 0, nil + } + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + return 0, fmt.Errorf("the pipelines are not listed: %w", err) + } + why := supersededWhy(newer) + ended := 0 + var first error + for _, key := range pkeys { + if key == newer.Pipeline { + continue + } + done, err := h.supersedeOne(key, newer, why, drop, unbound) + if err != nil && first == nil { + first = err + } + if done { + ended++ + } + } + return ended, first +} + +// superseded says whether old is a running pipeline newer supersedes (§31, A-202): of the same watched branch and an older +// source revision, either of the same module at another fingerprint, or, for newer a package's publish started, one an older +// publish of that package started (its version another, made from an older source revision of the branch). +func superseded(old, newer seat.Pipeline) bool { + if old.Outcome != seat.PipelineOutcomeRunning || old.Run != "" || old.Branch != newer.Branch || old.Source >= newer.Source { + return false + } + if newer.Published != nil && old.Published != nil && old.Published.Artifact == newer.Published.Artifact { + return old.Published.Version != newer.Published.Version + } + return old.Module == newer.Module && old.Fingerprint != newer.Fingerprint +} + +// supersededWhy names the newer commit, or the newer publish, that ended a pipeline. +func supersededWhy(newer seat.Pipeline) string { + if newer.Published != nil { + return fmt.Sprintf("the newer publish %s %s of %s supersedes it", newer.Published.Artifact, newer.Published.Version, newer.Branch) + } + return fmt.Sprintf("the newer commit %s of %s changed %s's fingerprint", newer.Commit, newer.Branch, newer.Module) +} + +func (h *Hearer) supersedeOne(key string, newer seat.Pipeline, why string, drop Drop, unbound func(string) error) (bool, error) { + for { + p, rev, err := h.Store.Get(key) + if err != nil || !superseded(p, newer) { + return false, nil + } + now := h.now() + var drops []seat.StepResult + var changed []int + for i := range p.Steps { + s := &p.Steps[i] + if s.Outcome != seat.StepResultOutcomeRunning && s.Outcome != seat.StepResultOutcomeWaiting { + continue + } + if s.Ask != "" && BoundSeat(s.Seat) { + drops = append(drops, *s) + } + s.Outcome, s.Why, s.Ended = seat.StepResultOutcomeSuperseded, why, ptrTime(now) + changed = append(changed, i) + } + p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeSuperseded, why, ptrTime(now) + p.EndCount++ + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return false, fmt.Errorf("pipeline %s cannot be written: %w", key, err) + } + var first error + for _, i := range changed { + h.logEnd(key, p.Steps[i], now) + if err := h.say(seat.EventStepEnded, p, &p.Steps[i]); err != nil && first == nil { + first = fmt.Errorf("pipeline %s's %s is written but not said: %w", key, seat.EventStepEnded, err) + } + } + for _, s := range drops { + if drop != nil { + if err := drop(s.Seat, s.Node, s.Ask, why); err != nil { + h.logf("the drop of ask %s of the superseded %s is not answered: %v", s.Ask, key, err) + } + } + if unbound != nil { + if err := unbound(s.Ask); err != nil { + h.logf("the bound of ask %s of the superseded %s is not dropped: %v", s.Ask, key, err) + } + } + } + if err := h.said(p, false, true); err != nil && first == nil { + first = err + } + return true, first + } +} + +func ptrTime(t time.Time) *time.Time { return &t } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go new file mode 100644 index 00000000..d6b96b54 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go @@ -0,0 +1,93 @@ +package pipeline + +import ( + "slices" + "strings" + "testing" + + builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" +) + +// RN8 (design 47 G §31, §119, §350; A-182, A-202): a step keeps the node that took its ask; a newer commit changing the +// module's fingerprint, or a newer publish of the package that started a pipeline, ends the older running pipeline +// superseded, its asks dropped of the node that took each (any holder for one never taken), a mesh-builds step only ended; +// an unchanged fingerprint, a run on request and an equal publish supersede nothing. +func TestSupersede(t *testing.T) { + h, store, said, _, _, _ := fixture() + old := store[pkey] + old.p.Steps = append(old.p.Steps, seat.StepResult{Artifact: "program", Step: seat.StepResultStepComposition, Status: seat.StepResultStatusRequired, + Outcome: seat.StepResultOutcomeRunning, Ask: "b1", Seat: builds.Seat}) + store[pkey] = old + if err := h.Taken("a1", "n7"); err != nil { + t.Fatal(err) + } + if p, _, _ := store.Get(pkey); p.Steps[0].Node != "n7" || p.Steps[1].Node != "" { + t.Fatalf("after ask-taken of a1 on n7 the steps name %q and %q", p.Steps[0].Node, p.Steps[1].Node) + } + type dropped struct{ seat, node, ask string } + var drops []dropped + var unbound []string + drop := func(s, node, ask, why string) error { + if !strings.Contains(why, "c2") { + t.Errorf("the drop of %s names no newer commit: %q", ask, why) + } + drops = append(drops, dropped{s, node, ask}) + return nil + } + unbind := func(ask string) error { unbound = append(unbound, ask); return nil } + newer := func(key string, source int64, fp string, run string, pub *seat.Published) seat.Pipeline { + return seat.Pipeline{Pipeline: key, Branch: "main", Source: source, Module: "app", Commit: "c2", Fingerprint: fp, Run: run, Published: pub, + Outcome: seat.PipelineOutcomeRunning} + } + for _, c := range []struct { + name string + newer seat.Pipeline + }{ + {"an unchanged fingerprint", newer("main.6.app", 6, "src2:aa", "", nil)}, + {"a run on request", newer("main.r1.app", 6, "src2:bb", "r1", nil)}, + {"an older source revision", newer("main.4.app", 4, "src2:bb", "", nil)}, + } { + if n, err := h.Supersede(c.newer, drop, unbind); n != 0 || err != nil { + t.Errorf("%s superseded %d (%v)", c.name, n, err) + } + } + *said = nil + if n, err := h.Supersede(newer("main.6.app", 6, "src2:bb", "", nil), drop, unbind); n != 1 || err != nil { + t.Fatalf("a changed fingerprint superseded %d (%v)", n, err) + } + p, _, _ := store.Get(pkey) + if p.Outcome != seat.PipelineOutcomeSuperseded || p.EndCount != 1 || !strings.Contains(p.Why, "c2") { + t.Errorf("the older pipeline is %s (%d ends): %q", p.Outcome, p.EndCount, p.Why) + } + for _, s := range p.Steps { + if s.Outcome != seat.StepResultOutcomeSuperseded { + t.Errorf("the %s step is %s, want superseded", s.Step, s.Outcome) + } + } + want := []dropped{{nodebuildgo.Seat, "n7", "a1"}, {nodebuildgo.Seat, "", "a2"}} + if !slices.Equal(drops, want) || !slices.Equal(unbound, []string{"a1", "a2"}) { + t.Errorf("dropped %+v and unbound %v, want %+v (the mesh-builds step only ended)", drops, unbound, want) + } + if n := len(*said); n != 5 || (*said)[4] != seat.EventPipelineEnded+".app.main.c1" { + t.Errorf("said %v, want four step-ended then pipeline-ended", *said) + } + if n, _ := h.Supersede(newer("main.7.app", 7, "src2:cc", "", nil), drop, unbind); n != 0 { + t.Errorf("a superseded pipeline was superseded again: %d", n) + } + + // A newer publish of a package supersedes the pipelines an older publish of it started, whatever their module. + pub := memStore{"main.5.lib": {p: seat.Pipeline{Pipeline: "main.5.lib", Branch: "main", Source: 5, Module: "lib", Fingerprint: "src2:aa", + Outcome: seat.PipelineOutcomeRunning, Published: &seat.Published{Artifact: "core:pkg", Version: "v1"}}, rev: 1}} + h.Store = pub + if n, _ := h.Supersede(newer("main.6.app", 6, "src2:aa", "", &seat.Published{Artifact: "core:pkg", Version: "v1"}), drop, unbind); n != 0 { + t.Errorf("a publish of the same version superseded %d", n) + } + if n, err := h.Supersede(newer("main.6.app", 6, "src2:aa", "", &seat.Published{Artifact: "core:pkg", Version: "v2"}), drop, unbind); n != 1 || err != nil { + t.Fatalf("a newer publish superseded %d (%v)", n, err) + } + if p, _, _ := pub.Get("main.5.lib"); p.Outcome != seat.PipelineOutcomeSuperseded || !strings.Contains(p.Why, "v2") { + t.Errorf("the older publish's pipeline is %s: %q", p.Outcome, p.Why) + } +} diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index 58693b51..457b0298 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -84,6 +84,8 @@ func TestManifest(t *testing.T) { {"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act", // RN3 (§53, §334): a publication's dependents behind. "seat:mesh-catalog.outdated", + // RN8 (§31, §350; A-182): a superseded step's ask dropped of the build seat that holds it. + "seat:node-build-go.drop", "seat:node-build-node.drop", "seat:node-build-image.drop", // A-74, A-89: who holds the seat operator-channel, whose state asks holds the router's record of the approval ask. "seat:mesh-module-assigner.assignments", // A-102 (§22, §24): the accepted kinds the runner puts its steps on. diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index e185be5c..e16d652f 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -54,7 +54,10 @@ "seat:mesh-wants.step", "seat:node-build-go.check.step", "seat:node-build-node.check.step", - "seat:node-build-image.check.step" + "seat:node-build-image.check.step", + "seat:node-build-go.drop", + "seat:node-build-node.drop", + "seat:node-build-image.drop" ], "uses": [ "operator-channel"