diff --git a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go index 7b0bfe66..ca58da01 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go @@ -347,6 +347,14 @@ const ( GroupEndedOutcomeCancelled = "cancelled" ) +// Held is a pipeline held for the new version of the runner: its steps are placed and asked when the node-engine seat says that fingerprint runs healthy (design 47 G 65, 266, 534, A-195). +type Held struct { + // the fingerprint of the runner's own module the pipeline awaits running healthy (design 47 G 266, 276, A-195) + Fingerprint string `json:"fingerprint"` + // since when the pipeline is held, RFC 3339 + Since time.Time `json:"since"` +} + // KeptResult is a package's passing result the runner keeps, by the digest of what its tests read (ADR 0666 section 8). type KeptResult struct { // sha256: of what the package's tests read @@ -427,6 +435,7 @@ type Pipeline struct { Fingerprint string `json:"fingerprint,omitempty"` // the gathered group, when the pipeline is its Group string `json:"group,omitempty"` + Held *Held `json:"held,omitempty"` Module string `json:"module"` Outcome string `json:"outcome"` // yes when the module is over its test budget (design 47 G section 419) 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..b3509f40 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json @@ -672,6 +672,25 @@ }, "description": "how many tests a step ran and carried, per level (ADR 0666 section 8)" }, + "Held": { + "type": "object", + "properties": { + "since": { + "type": "string", + "format": "date-time", + "description": "since when the pipeline is held, RFC 3339" + }, + "fingerprint": { + "type": "string", + "description": "the fingerprint of the runner's own module the pipeline awaits running healthy (design 47 G 266, 276, A-195)" + } + }, + "required": [ + "since", + "fingerprint" + ], + "description": "a pipeline held for the new version of the runner: its steps are placed and asked when the node-engine seat says that fingerprint runs healthy (design 47 G 65, 266, 534, A-195)" + }, "ToolchainUsed": { "type": "object", "properties": { @@ -985,6 +1004,9 @@ "type": "boolean", "description": "yes when the module is over its test budget (design 47 G section 419)" }, + "held": { + "$ref": "#/defs/Held" + }, "widens": { "type": "array", "items": { 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..40972353 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts @@ -144,6 +144,14 @@ export interface GroupEnded { "why"?: string; } +/** a pipeline held for the new version of the runner: its steps are placed and asked when the node-engine seat says that fingerprint runs healthy (design 47 G 65, 266, 534, A-195) */ +export interface Held { + /** the fingerprint of the runner's own module the pipeline awaits running healthy (design 47 G 266, 276, A-195) */ + "fingerprint": string; + /** since when the pipeline is held, RFC 3339 */ + "since": string; +} + /** a package's passing result the runner keeps, by the digest of what its tests read (ADR 0666 section 8) */ export interface KeptResult { /** sha256: of what the package's tests read */ @@ -189,6 +197,7 @@ export interface Pipeline { "fingerprint"?: string; /** the gathered group, when the pipeline is its */ "group"?: string; + "held"?: Held; "module": string; "outcome": "running" | "passed" | "failed" | "refused" | "superseded" | "cancelled"; /** yes when the module is over its test budget (design 47 G section 419) */ 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 4b0559b7..ed97400a 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 @@ -108,9 +108,20 @@ func HealthTool(pipelines, consumers pipeline.Kept, now func() time.Time) stdio. Run: func(map[string]any) (any, error) { return pipeline.Health(pipelines, consumers, now()), nil }} } +// WaitsTool is the runner's tool runner_waits: the pipelines held for its new version, with the fingerprint each awaits and since +// when (design 47 G §276, §539, A-195). +func WaitsTool(r pipeline.Reads) stdio.Tool { + return stdio.Tool{Name: Prefix + "_waits", + Description: "The pipelines the pipeline runner holds until its new version runs healthy: each with its pipeline, the " + + "fingerprint of the runner it awaits and since when.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { return r.Waits() }} +} + func main() { r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}} tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) + tools = append(tools, WaitsTool(r)) 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) 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..f68f0641 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -105,12 +105,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // for its step. starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say, Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module, - Arm: armBound(ctx, bounds), Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)} + Arm: armBound(ctx, bounds), Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains), + // The runner's own module and fingerprint (A-149): a take-in giving it another holds the pipelines it starts (§65, A-195). + Self: cred.Module, SelfFp: getenv("MESH_FINGERPRINT"), Awaiting: pipeline.Awaiting(pipelineStore{ctx: ctx, kv: pipelines}), + Hold: armHeld(ctx, bounds)} resume := func(branch string) { half, ok := rv.Catalog.View(branch) if !ok { return } + // Pipelines held for the fingerprint this runner runs are started now: it is the new runner (§19, A-195). + if n, err := starter.Release(pipelineStore{ctx: ctx, kv: pipelines}, starter.SelfFp, "ran", rv.Catalog.View); err != nil { + logger.Printf("the pipelines held for the runner at %s: %d released, then %v", starter.SelfFp, n, err) + } n, err := starter.Resume(pipelineStore{ctx: ctx, kv: pipelines}, branch, half) if err != nil { logger.Printf("the steps placed on %s and not yet asked: %d asked, then %v", branch, n, err) @@ -128,7 +135,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hearAskTaken(ctx, heard, bounds, logger.Printf); err != nil { return err } - if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil { + store := pipelineStore{ctx: ctx, kv: pipelines} + if err := watchBounds(ctx, bounds, hearer, func(key string) { + if n, err := starter.ReleaseExpired(store, key, rv.Catalog.View); err != nil { + logger.Printf("the pipelines held for the runner, whose %s passed: %d released, then %v", key, n, err) + } else { + logger.Printf("the runner's new version did not run healthy within %s: %d held pipelines started with the running runner", pipeline.HeldBound, n) + } + }, logger.Printf); err != nil { + return err + } + if err := hearRunning(ctx, heard, cred.Module, starter, store, bounds, rv.Catalog.View, logger.Printf); err != nil { return err } if err := hearDecided(heard, hearer, cred.Module, 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..ec0f761d 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -9,6 +9,7 @@ import ( "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" @@ -172,9 +173,53 @@ func armBound(ctx context.Context, bounds jetstream.KeyValue) pipeline.Arm { } } +// armHeld is the bound of the pipelines held for the runner's new version: a key of the state bounds named for the awaited +// fingerprint, with its own time to live of 30 minutes (design 47 G §266, A-195, A-77). +func armHeld(ctx context.Context, bounds jetstream.KeyValue) func(string) error { + return func(fp string) error { + _, err := bounds.Create(ctx, pipeline.HeldKey(fp), []byte(`{"fingerprint":"`+fp+`"}`), jetstream.KeyTTL(pipeline.HeldBound)) + if errors.Is(err, jetstream.ErrKeyExists) { + return nil + } + return err + } +} + +// hearRunning releases the pipelines held for the runner's new version when the node-engine seat says the runner's own module +// runs that fingerprint healthy, on any node (§266, §276, A-195); its bound is then deleted. A running with no health reads +// as healthy (§520). Heard again when the release could not be written. +func hearRunning(ctx context.Context, hub handlers, self string, st *pipeline.Starter, store pipeline.Updater, bounds jetstream.KeyValue, + view func(string) (pipeline.Half, bool), logf func(string, ...any)) error { + return hub.Handle(nodeengine.SubjectRunningAll, func(m jetstream.Msg) { + var ev nodeengine.Running + if err := json.Unmarshal(m.Data(), &ev); err != nil { + logf("%s is no running: %v", m.Subject(), err) + _ = m.Term() + return + } + if ev.Module != self || ev.Stopped || ev.Fingerprint == "" || (ev.Healthy != nil && !*ev.Healthy) { + _ = m.Ack() + return + } + n, err := st.Release(store, ev.Fingerprint, "ran", view) + if err != nil { + logf("the pipelines held for %s: %d released, then %v; heard again", ev.Fingerprint, n, err) + _ = m.Nak() + return + } + if n > 0 { + logf("the runner runs %s on %s: %d held pipelines released", ev.Fingerprint, ev.Node, n) + if err := bounds.Delete(ctx, pipeline.HeldKey(ev.Fingerprint)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + logf("the bound of the wait for %s is not deleted: %v", ev.Fingerprint, err) + } + } + _ = m.Ack() + }) +} + // watchBounds hears the end of each ask's bound, the bus's marker of the key's expiry, and ends the step no holder took (§32, // §535, A-77): no timer in the process. -func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hearer, logf func(string, ...any)) error { +func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hearer, release func(key string), logf func(string, ...any)) error { w, err := bounds.WatchAll(ctx, jetstream.UpdatesOnly()) if err != nil { return err @@ -189,6 +234,10 @@ func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hea if !ok { return } + if e != nil && e.Operation() == jetstream.KeyValuePurge && pipeline.IsHeldKey(e.Key()) { + release(e.Key()) + continue + } if e == nil || e.Operation() != jetstream.KeyValuePurge || !pipeline.IsAskKey(e.Key()) { continue } diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod index 9bfc48db..ef4b1695 100644 --- a/modules/mesh-delivery-pipeline/go.mod +++ b/modules/mesh-delivery-pipeline/go.mod @@ -14,6 +14,7 @@ require ( git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildimage v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go new file mode 100644 index 00000000..5186d2a6 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go @@ -0,0 +1,163 @@ +package pipeline + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "time" + + runnerclient "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// HeldBound is how long the pipelines held for the runner's new version wait for it to run healthy (design 47 G §266: 30 +// minutes for the runner's `pipeline` step, A-195): a key of the state `bounds` with its own time to live, never a timer. +const HeldBound = 30 * time.Minute + +// HeldPrefix starts the key of a held wait's bound. +const HeldPrefix = "held-" + +// HeldKey is the bound's key for the awaited fingerprint: a fingerprint holds characters a key's name does not. +func HeldKey(fp string) string { + sum := sha256.Sum256([]byte(fp)) + return HeldPrefix + hex.EncodeToString(sum[:12]) +} + +// IsHeldKey says whether a key of the state `bounds` is a held wait's bound. +func IsHeldKey(key string) bool { + return len(key) > len(HeldPrefix) && key[:len(HeldPrefix)] == HeldPrefix +} + +// Awaiting is the awaited fingerprint of the pipelines held in store, "" and false when none is held (design 47 G §266, A-195): +// a pipeline started while one is held is held for the same version, so it never runs on the old runner. +func Awaiting(store Updater) func() (string, bool) { + return func() (string, bool) { + pkeys, err := store.Keys("*.*.*") + if err != nil { + return "", false + } + for _, key := range pkeys { + if p, _, err := store.Get(key); err == nil && p.Held != nil && p.Outcome == seat.PipelineOutcomeRunning { + return p.Held.Fingerprint, true + } + } + return "", false + } +} + +// changesSelf says whether a take-in's entry is the runner's own module at a fingerprint other than the one that runs: its +// pipelines are held for the new runner (design 47 G §65, §266, A-195). +func (s *Starter) changesSelf(module, fp string) bool { + return s.Self != "" && module == s.Self && fp != "" && fp != s.SelfFp +} + +// startHeld writes b's pipeline held for the runner's version awaiting, its steps neither placed nor asked, and says +// wait-held with step `pipeline`; pipeline-started is said when it is released (§74, §266). +func (s *Starter) startHeld(b Begin, key string, now time.Time, awaiting string) (bool, error) { + p := newPipeline(b, key, now) + p.Held = &seat.Held{Since: now, Fingerprint: awaiting} + if err := s.create(key, p); errors.Is(err, ErrExists) { + s.logf("%s is started already: this start writes nothing", key) + return false, nil + } else if err != nil { + return false, fmt.Errorf("the pipeline %s is not written: %w", key, err) + } + if s.Hold != nil { + if err := s.Hold(awaiting); err != nil { + s.logf("the bound of the wait for the runner at %s is not armed: %v", awaiting, err) + } + } + if err := s.Say(runnerclient.Subject(seat.EventWaitHeld, b.Module, b.Branch, commitOrGroup(p)), s.waitOf(p, now, "")); err != nil { + s.logf("wait-held of %s is not said: %v", key, err) + } + return true, nil +} + +func commitOrGroup(p seat.Pipeline) string { + if p.Group != "" { + return p.Group + } + return p.Commit +} + +func (s *Starter) waitOf(p seat.Pipeline, since time.Time, how string) seat.Wait { + return seat.Wait{V: 1, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit, Fingerprint: p.Fingerprint, Group: p.Group, + Revision: p.Revision, Source: p.Source, Step: seat.WaitStepPipeline, On: seat.WaitOnBuildNode, Since: since, How: how} +} + +// Release starts the pipelines held for fingerprint fp ("" for every held one) as the new runner running healthy lets +// them, or the bound's expiry or a hand release (how: `ran` or `released`; §266, §276): each has its steps placed and asked and +// says wait-ended and pipeline-started. A pipeline whose branch has no view yet stays held, and the first error is returned +// so the event is heard again. It says how many it started. +func (s *Starter) Release(store Updater, fp, how string, view func(branch string) (Half, bool)) (int, error) { + pkeys, err := store.Keys("*.*.*") + if err != nil { + return 0, fmt.Errorf("the pipelines are not listed: %w", err) + } + now := time.Now().UTC() + if s.Now != nil { + now = s.Now().UTC() + } + started := 0 + var first error + note := func(err error) { + if first == nil { + first = err + } + } + for _, key := range pkeys { + p, rev, err := store.Get(key) + if err != nil || p.Held == nil || p.Outcome != seat.PipelineOutcomeRunning || (fp != "" && p.Held.Fingerprint != fp) { + continue + } + half, ok := view(p.Branch) + if !ok { + note(fmt.Errorf("the catalog half of %s is not held yet: %s stays held", p.Branch, key)) + continue + } + held := *p.Held + p.Held = nil + var asks []Asked + if s.Registration != nil { + if reg, ok := s.Registration(p.Module); ok { + p.Steps, asks = Place(p, reg, half.Entries, now, s.Name) + } + } + if _, err := store.Update(key, p, rev); err != nil { + note(fmt.Errorf("the held pipeline %s is not released: %w", key, err)) + continue + } + started++ + if err := s.Say(runnerclient.Subject(seat.EventWaitEnded, p.Module, p.Branch, commitOrGroup(p)), s.waitOf(p, held.Since, how)); err != nil { + s.logf("wait-ended of %s is not said: %v", key, err) + } + ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: p.Branch, Source: p.Source, Module: p.Module, Commit: p.Commit, + Fingerprint: p.Fingerprint, Group: p.Group, Revision: p.Revision} + if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, p.Module, p.Branch, commitOrGroup(p)), ev); err != nil { + s.logf("pipeline-started of %s is not said: %v", key, err) + } + if s.Steps != nil && len(asks) > 0 { + s.carry(key, asks) + if err := askAll(s.Steps, s.Arm, key, asks); err != nil { + note(err) + } + } + } + return started, first +} + +// ReleaseExpired starts the pipelines held for the fingerprint whose bound key expired (§266, A-195): the 30 minutes passed +// without the new runner running healthy, so the running runner runs them. +func (s *Starter) ReleaseExpired(store Updater, key string, view func(branch string) (Half, bool)) (int, error) { + pkeys, err := store.Keys("*.*.*") + if err != nil { + return 0, fmt.Errorf("the pipelines are not listed: %w", err) + } + for _, k := range pkeys { + if p, _, err := store.Get(k); err == nil && p.Held != nil && HeldKey(p.Held.Fingerprint) == key { + return s.Release(store, p.Held.Fingerprint, "released", view) + } + } + return 0, nil +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go new file mode 100644 index 00000000..c4911fea --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go @@ -0,0 +1,86 @@ +package pipeline + +import ( + "encoding/json" + "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 with the Create a start writes through. +type memCreated struct{ memStore } + +func (m memCreated) Create(key string, value []byte) error { + if _, ok := m.memStore[key]; ok { + return ErrExists + } + var p seat.Pipeline + if err := json.Unmarshal(value, &p); err != nil { + return err + } + m.memStore[key] = struct { + p seat.Pipeline + rev uint64 + }{p, 1} + return nil +} + +// A take-in giving the runner's own module a new fingerprint holds every pipeline it starts, a later take-in's too while one is +// held, with no step asked; the new runner running healthy releases them, placing and asking their steps (design 47 G §65, §266, +// §276, §534; A-195). +func TestPipelinesAreHeldForTheNewRunnerAndReleasedWhenItRuns(t *testing.T) { + reg, entries := placeFixture() + entries["runner"] = catalog.CatalogEntry{Module: "runner"} + store := memCreated{memStore{}} + asks := &stepAsks{} + var said []string + var bounds []string + s := &Starter{Pipelines: store, Steps: asks, Name: "the-runner", Self: "runner", SelfFp: "src2:old", + Awaiting: Awaiting(store.memStore), Hold: func(fp string) error { bounds = append(bounds, fp); return nil }, + Registration: func(m string) (catalog.Registration, bool) { return reg, m == "app" }, + Say: func(subject string, _ any) error { + said = append(said, subject[strings.Index(subject, ".event.")+7:]) + return nil + }} + view := func(string) (Half, bool) { return Half{Entries: entries}, true } + n, err := s.TakenIn(catalog.TakeInTaken{Key: "m.1", Branch: "main", Source: 4, Entries: []catalog.TakeInEntry{ + {Module: "app", Commit: "c1", Fingerprint: "src2:aa"}, {Module: "runner", Commit: "c2", Fingerprint: "src2:new"}}}, view) + if err != nil || n != 2 { + t.Fatalf("take-in = %d, %v", n, err) + } + p, _, _ := store.Get("main.4.app") + if p.Held == nil || p.Held.Fingerprint != "src2:new" || len(p.Steps) != 0 || len(asks.asks) != 0 { + t.Fatalf("held = %+v, steps %d, asks %d", p.Held, len(p.Steps), len(asks.asks)) + } + if strings.Join(said, "|") != "wait-held.app.main.c1|wait-held.runner.main.c2" || len(bounds) != 2 || bounds[0] != "src2:new" { + t.Errorf("said %v, bounds %v", said, bounds) + } + // A later take-in not naming the runner is held for the same version. + if n, err := s.TakenIn(catalog.TakeInTaken{Key: "m.2", Branch: "main", Source: 5, Entries: []catalog.TakeInEntry{{Module: "app", Commit: "c3", Fingerprint: "src2:bb"}}}, view); err != nil || n != 1 { + t.Fatalf("second take-in = %d, %v", n, err) + } + if p, _, _ := store.Get("main.5.app"); p.Held == nil || len(asks.asks) != 0 { + t.Fatalf("the later pipeline is not held: %+v", p.Held) + } + // Another fingerprint's release starts nothing. + if n, err := s.Release(store.memStore, "src2:other", "ran", view); err != nil || n != 0 { + t.Fatalf("release of another version = %d, %v", n, err) + } + said = nil + n, err = s.Release(store.memStore, "src2:new", "ran", view) + if err != nil || n != 3 { + t.Fatalf("release = %d, %v", n, err) + } + p, _, _ = store.Get("main.4.app") + if p.Held != nil || len(p.Steps) == 0 || len(asks.asks) == 0 { + t.Errorf("after the release: held %+v, steps %d, asks %d", p.Held, len(p.Steps), len(asks.asks)) + } + if !strings.Contains(strings.Join(said, "|"), "wait-ended.app.main.c1|pipeline-started.app.main.c1") { + t.Errorf("said %v", said) + } + if _, held := Awaiting(store.memStore)(); held { + t.Error("a pipeline is still held") + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read.go b/modules/mesh-delivery-pipeline/internal/pipeline/read.go index ced5a604..4da65cb7 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/read.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read.go @@ -88,6 +88,25 @@ func (r Reads) Stands() (seat.StandsAnswer, error) { return seat.StandsAnswer{Stands: out}, nil } +// Waits is the pipelines held for the runner's new version, each as the wait of step `pipeline` it says (design 47 G §266, §276, +// §534, A-195): the acting part awaited by its fingerprint, since when, and the pipeline. +func (r Reads) Waits() (seat.StandsAnswer, error) { + all, err := r.pipelines() + if err != nil { + return seat.StandsAnswer{}, err + } + out := []seat.Wait{} + for _, p := range all { + if p.Held == nil || p.Outcome != seat.PipelineOutcomeRunning { + continue + } + out = append(out, seat.Wait{Pipeline: p.Pipeline, Module: p.Module, Branch: p.Branch, Commit: p.Commit, Group: p.Group, Source: p.Source, + Revision: p.Revision, Fingerprint: p.Held.Fingerprint, Step: seat.WaitStepPipeline, On: seat.WaitOnBuildNode, Since: p.Held.Since, V: 1}) + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Since.Before(out[j].Since) }) + return seat.StandsAnswer{Stands: out}, nil +} + // WhyCarried is where a step's result was carried from (§25, §83): the pipeline it names. func (r Reads) WhyCarried(in seat.WhyCarriedInput) (seat.WhyCarriedAnswer, error) { p, ok, err := r.pipeline(in.Pipeline) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go index 601e639c..5dabc4b4 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go @@ -127,3 +127,19 @@ func TestReads(t *testing.T) { } } } + +// runner_waits lists the pipelines held for the runner's new version as waits of step `pipeline` (design 47 G §276, A-195). +func TestWaitsListTheHeldPipelines(t *testing.T) { + since := time.Date(2026, 11, 12, 6, 0, 0, 0, time.UTC) + held := seat.Pipeline{Pipeline: "main.4.app", Module: "app", Branch: "main", Commit: "c1", Source: 4, Outcome: seat.PipelineOutcomeRunning, + Held: &seat.Held{Since: since, Fingerprint: "src2:new"}} + plain := seat.Pipeline{Pipeline: "main.3.app", Module: "app", Branch: "main", Outcome: seat.PipelineOutcomeRunning} + pipes := memKept{} + put(t, pipes, held.Pipeline, held) + put(t, pipes, plain.Pipeline, plain) + r := Reads{Pipelines: pipes, Kept: memKept{}} + got, err := r.Waits() + if err != nil || len(got.Stands) != 1 || got.Stands[0].Step != seat.WaitStepPipeline || got.Stands[0].Fingerprint != "src2:new" || !got.Stands[0].Since.Equal(since) { + t.Fatalf("waits = %+v, %v", got, err) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 9e5737aa..42f23f01 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -35,6 +35,8 @@ type Begin struct { Group string Repository string Revision int64 + // Await is the fingerprint of the runner's own module this pipeline is held for (§266, A-195); empty when it is not held. + Await string // Entries are the catalog half's entries of the pipeline's branch: the steps' artifacts and build seats are read from them. Entries map[string]catalog.CatalogEntry } @@ -70,6 +72,26 @@ type Starter struct { Toolchain func(language string) (*nodebuildgo.ToolchainUsed, error) // Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none. Arm Arm + // Self is the runner's own module and SelfFp the fingerprint it runs (A-149); a take-in giving Self another fingerprint + // holds its pipelines for the new runner (§65, §266, A-195). Awaiting is the fingerprint the held pipelines await, while + // any is held; Hold arms the bound of that wait (HeldBound). All nil: nothing is held. + Self, SelfFp string + Awaiting func() (string, bool) + Hold func(fp string) error +} + +func newPipeline(b Begin, key string, now time.Time) seat.Pipeline { + return seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fingerprint: b.Fp, + Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now, + Steps: []seat.StepResult{}} +} + +func (s *Starter) create(key string, p seat.Pipeline) error { + raw, err := json.Marshal(p) + if err != nil { + return err + } + return s.Pipelines.Create(key, raw) } // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second @@ -83,9 +105,15 @@ func (s *Starter) Start(b Begin) (bool, error) { now = s.Now().UTC() } key := Key(b.Branch, b.Source, b.Module) - p := seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fingerprint: b.Fp, - Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now, - Steps: []seat.StepResult{}} + if b.Await == "" && s.Awaiting != nil { + if fp, ok := s.Awaiting(); ok { + b.Await = fp + } + } + if b.Await != "" { + return s.startHeld(b, key, now, b.Await) + } + p := newPipeline(b, key, now) var asks []Asked if s.Registration != nil { if reg, ok := s.Registration(b.Module); ok { @@ -94,11 +122,7 @@ func (s *Starter) Start(b Begin) (bool, error) { s.logf("%s has no registration yet: its pipeline starts with no step", key) } } - raw, err := json.Marshal(p) - if err != nil { - return false, err - } - if err := s.Pipelines.Create(key, raw); errors.Is(err, ErrExists) { + if err := s.create(key, p); errors.Is(err, ErrExists) { s.logf("%s is started already: this start writes nothing", key) return false, nil } else if err != nil { @@ -288,9 +312,16 @@ func (s *Starter) TakenIn(ev catalog.TakeInTaken, view func(branch string) (Half half, _ := view(ev.Branch) started := 0 var first error + // A take-in giving the runner's own module a new fingerprint holds every pipeline it starts for that runner (§65, A-195). + await := "" + for _, e := range ev.Entries { + if s.changesSelf(e.Module, e.Fingerprint) { + await = e.Fingerprint + } + } for _, e := range ev.Entries { ok, err := s.Start(Begin{Branch: ev.Branch, Source: ev.Source, Module: e.Module, Commit: e.Commit, Fp: e.Fingerprint, - Group: ev.Group, Repository: e.Repository, Revision: half.Revision, Entries: half.Entries}) + Group: ev.Group, Repository: e.Repository, Revision: half.Revision, Entries: half.Entries, Await: await}) if err != nil && first == nil { first = err } diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index 58693b51..e29f902e 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -77,7 +77,7 @@ func TestManifest(t *testing.T) { {"consumes", equalSets(m.Consumes, []string{ "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", "operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended", - "node-build-image.step-ended", "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", "mesh-wants.step-ended", "mesh-catalog.registration-changed", + "node-build-image.step-ended", "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", "node-engine.running", "mesh-wants.step-ended", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-conditions.findings-asked"})}, // RN2 (§21, §195, §200, §77): the registrations and the catalog half asked by their verbs, the first views read and // asked again by hand (a hand act). @@ -91,7 +91,7 @@ func TestManifest(t *testing.T) { "seat:node-build-image.check.step"})}, {"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})}, // RN25 (§539): the runner's tools carry the prefix runner_. - {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health"})}, + {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health", "runner_waits"})}, // 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. diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index e185be5c..0b100789 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -25,7 +25,8 @@ "tools": [ "runner_first_views", "runner_first_view_again", - "runner_health" + "runner_health", + "runner_waits" ], "consumes": [ "mesh-delivery-indexer.take-in.taken", @@ -39,6 +40,7 @@ "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", + "node-engine.running", "mesh-wants.step-ended", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision",