RN19, RN21: the pipelines of a change to the runner are held until its new version runs healthy

The seat's pipeline record gains `held` (since, the fingerprint awaited). A take-in giving the runner's own module a new
fingerprint holds every pipeline it starts, and a later take-in's while any is held; wait-held (step pipeline) is said, the
30-minute bound is a key of the state bounds. The node-engine seat's running of the runner's module at the awaited
fingerprint, healthy, or the bound's expiry, releases them: steps placed and asked, wait-ended and pipeline-started said. A
runner starting at the awaited fingerprint releases at its start. runner_waits lists the held pipelines.

Builds: design 47 G §65, §266, §276, §534, §19 (lead answer A-195)

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 05:57:05 +01:00
parent 6676b42dab
commit e053605e82
14 changed files with 450 additions and 15 deletions
@@ -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:<hex> 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)
@@ -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": {
@@ -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:<hex> 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) */
@@ -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)
@@ -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 {
@@ -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
}
+1
View File
@@ -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
@@ -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
}
@@ -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")
}
}
@@ -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)
@@ -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)
}
}
@@ -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
}
@@ -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.
+3 -1
View File
@@ -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",