RN8: a newer commit or publish supersedes the older running pipeline, its asks dropped where they stand
A step keeps the node whose holder said ask-taken; a superseded step's drop is asked of that node's holder, an untaken ask's of any holder, as kept calls; steps on mesh-builds and mesh-wants are only ended superseded. A pipeline a publish started says that publish (published), so a newer publish of the package supersedes it. Builds: design 47 G §31, §119, §350, A-182, A-202 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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, <branch>.<revision>.<module>, 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, <branch>.<revision>.<module>, 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"`
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -195,6 +195,8 @@ export interface Pipeline {
|
||||
"over-budget"?: boolean;
|
||||
/** the pipeline's key, <branch>.<revision>.<module>, 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, <branch>.<revision>.<module>, 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;
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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, `<branch>.<source revision>.<module>`, 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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 }
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user