Merge branch 'feat/47-G-fix11-b1e' into feat/47-G-join11

This commit is contained in:
2026-11-12 10:28:34 +01:00
40 changed files with 1750 additions and 83 deletions
@@ -27,6 +27,7 @@ const (
VerbAccept = "accept"
VerbCancel = "cancel"
VerbAskAgain = "ask-again"
VerbRelease = "release"
)
const (
@@ -42,7 +43,7 @@ const (
// Verbs is every verb the seat serves, in its definition's order: a new copy each call.
func Verbs() []string {
return []string{VerbRun, VerbStart, VerbShow, VerbList, VerbStands, VerbWhyCarried, VerbKept, VerbReview, VerbRerun, VerbAccept, VerbCancel, VerbAskAgain}
return []string{VerbRun, VerbStart, VerbShow, VerbList, VerbStands, VerbWhyCarried, VerbKept, VerbReview, VerbRerun, VerbAccept, VerbCancel, VerbAskAgain, VerbRelease}
}
// Events is every event published on the seat's subject: a new copy each call.
@@ -347,6 +348,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 +436,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)
@@ -691,7 +701,7 @@ type Wait struct {
// for its end, how it ended: answered, ran, released, superseded
How string `json:"how,omitempty"`
Module string `json:"module"`
// what the step waits on (design 47 *Health and going back*)
// what the step waits on (design 47 *Health and going back*); the runner version is what a pipeline held for the runner's new version waits on (design 47 G 534, A-206)
On string `json:"on"`
// the pipeline's key, <branch>.<revision>.<module>, a branch escaped as a key's name is
Pipeline string `json:"pipeline"`
@@ -711,9 +721,10 @@ type Wait struct {
// The values of Wait's on.
const (
WaitOnPerson = "person"
WaitOnBuildNode = "build node"
WaitOnRevisionView = "revision view"
WaitOnPerson = "person"
WaitOnBuildNode = "build node"
WaitOnRevisionView = "revision view"
WaitOnRunnerVersion = "runner version"
)
// The values of Wait's step.
@@ -950,6 +961,18 @@ type AskAgainAnswer struct {
Result StepResult `json:"result"`
}
// ReleaseInput is what release takes.
type ReleaseInput struct {
// why, written with the caller the runtime proves in the hand-act log
Why string `json:"why"`
}
// ReleaseAnswer is what release answers.
type ReleaseAnswer struct {
// how many held pipelines were started
Released int64 `json:"released"`
}
// PipelineStartedEvent is the body of the event pipeline-started.
type PipelineStartedEvent = PipelineStarted
@@ -501,6 +501,35 @@
]
},
"class": "act"
},
{
"name": "release",
"description": "The pipelines the runner holds for its new version started by hand with the running runner, whatever the new version's state; written in the hand-act log (design 47 G 276, A-206).",
"input": {
"type": "object",
"properties": {
"why": {
"type": "string",
"description": "why, written with the caller the runtime proves in the hand-act log"
}
},
"required": [
"why"
]
},
"answer": {
"type": "object",
"properties": {
"released": {
"type": "integer",
"description": "how many held pipelines were started"
}
},
"required": [
"released"
]
},
"class": "act"
}
],
"events": [
@@ -672,6 +701,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 +1033,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": {
@@ -1430,9 +1481,10 @@
"enum": [
"person",
"build node",
"revision view"
"revision view",
"runner version"
],
"description": "what the step waits on (design 47 *Health and going back*)"
"description": "what the step waits on (design 47 *Health and going back*); the runner version is what a pipeline held for the runner's new version waits on (design 47 G 534, A-206)"
},
"who": {
"type": "string",
@@ -17,6 +17,7 @@ export declare const verbs: {
readonly accept: "accept";
readonly cancel: "cancel";
readonly askAgain: "ask-again";
readonly release: "release";
};
export declare const events: {
@@ -144,6 +145,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 +198,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) */
@@ -377,8 +387,8 @@ export interface Wait {
/** for its end, how it ended: answered, ran, released, superseded */
"how"?: string;
"module": string;
/** what the step waits on (design 47 *Health and going back*) */
"on": "person" | "build node" | "revision view";
/** what the step waits on (design 47 *Health and going back*); the runner version is what a pipeline held for the runner's new version waits on (design 47 G 534, A-206) */
"on": "person" | "build node" | "revision view" | "runner version";
/** the pipeline's key, <branch>.<revision>.<module>, a branch escaped as a key's name is */
"pipeline": string;
/** the catalog graph revision of the branch the pipeline runs for */
@@ -569,6 +579,18 @@ export interface AskAgainAnswer {
"result": StepResult;
}
/** is what release takes. */
export interface ReleaseInput {
/** why, written with the caller the runtime proves in the hand-act log */
"why": string;
}
/** is what release answers. */
export interface ReleaseAnswer {
/** how many held pipelines were started */
"released": number;
}
/** is the body of the event pipeline-started. */
export type PipelineStartedEvent = PipelineStarted;
@@ -17,6 +17,7 @@ export const verbs = {
accept: "accept",
cancel: "cancel",
askAgain: "ask-again",
release: "release",
};
export const events = {
@@ -86,21 +86,61 @@ func Tools(r pipeline.Reads) []stdio.Tool {
return out
}
// Module and Slug are the runner's module and its tools' prefix, as its manifest names them.
// Module is the runner's module, and Prefix its tools' prefix (design 47 G §539: runner_first_views, runner_health).
const (
Module = "mesh-delivery-pipeline"
Slug = "pipeline"
Prefix = "runner"
)
// FirstViewTools are the runner's `<module>_first_views` and `<module>_first_view_again` over its state first-views
// (design 47 G §77, §200): each view it holds or waits for, since when, and asked again by hand, a hand act.
func FirstViewTools(st stdio.KeptState, acts views.HandActs, now func() time.Time) []stdio.Tool {
return views.StatusTools(Module, Slug+"_first_views", Slug+"_first_view_again", st, acts, now)
return views.StatusTools(Module, Prefix+"_first_views", Prefix+"_first_view_again", st, acts, now)
}
// HealthTool is the runner's health check, run outside its process (design 47 G §76, §468, §495; A-164): it reads the
// runner's state pipelines as show does and the ages its process writes in the state consumers, and asks no seat verb.
func HealthTool(pipelines, consumers pipeline.Kept, now func() time.Time) stdio.Tool {
return stdio.Tool{Name: Prefix + "_health",
Description: "The pipeline runner's health: healthy when it answers show and each of its durable consumers' oldest " +
"unacknowledged message waits less than five minutes; otherwise it names each consumer behind.",
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) { return pipeline.Health(pipelines, consumers, now()), nil }}
}
// 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() }}
}
// ReleaseTool is the runner's tool runner_release: it starts the pipelines held for the runner's new version with the running
// runner, through the seat's verb release, which the process that holds the pipelines answers and records in the hand-act log
// first (design 47 G §276, A-206).
func ReleaseTool(ask func(key string, body any) (json.RawMessage, error)) stdio.Tool {
return stdio.Tool{Name: Prefix + "_release",
Description: "Start now, with the running pipeline runner, the pipelines it holds until its new version runs healthy " +
"(runner_waits lists them), each wait ending released. Written in the hand-act log (mesh-hand-acts).",
Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{
"why": map[string]any{"type": "string", "description": "why, in plain words"}}},
Run: func(args map[string]any) (any, error) {
in, err := as[seat.ReleaseInput](args)
if err != nil {
return nil, err
}
return ask("seat:"+seat.Seat+"."+seat.VerbRelease, in)
}}
}
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), ReleaseTool(stdio.Ask))
tools = append(tools, HealthTool(r.Pipelines, kept{stdio.State(pipeline.StateConsumers)}, time.Now))
if err := stdio.Serve("", tools); err != nil {
fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] %v\n", err)
os.Exit(1)
@@ -2,6 +2,7 @@ package main
import (
"encoding/json"
"fmt"
"strings"
"testing"
@@ -17,7 +18,7 @@ func (none) Keys() ([]string, error) { return nil, nil }
func (none) Value(string) (json.RawMessage, error) { return nil, nil }
// TestTools: the bundle serves the five read verbs as <seat>.<verb> (§83, §306; RN15/AC1), and the first-view tools
// (§77, §200; RN2/AC4).
// (§77, §200; RN2/AC4), named with the prefix runner_ (§539; RN25).
func TestTools(t *testing.T) {
var names []string
for _, tl := range Tools(pipeline.Reads{Pipelines: none{}, Kept: none{}}) {
@@ -26,7 +27,19 @@ func TestTools(t *testing.T) {
for _, tl := range FirstViewTools(stdio.State(views.StatusState), nil, nil) {
names = append(names, tl.Name)
}
want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept pipeline_first_views pipeline_first_view_again"
// RN25 (§468, §76; A-164): the health check is the bundle's runner_health.
names = append(names, HealthTool(none{}, none{}, nil).Name)
// A-206: the tool runner_release asks the seat's verb release with its why.
var asked string
rel := ReleaseTool(func(key string, body any) (json.RawMessage, error) {
asked = key + fmt.Sprintf(" %+v", body)
return json.RawMessage(`{"released":1}`), nil
})
if _, err := rel.Run(map[string]any{"why": "w"}); err != nil || asked != "seat:mesh-pipeline-runner.release {Why:w}" {
t.Errorf("runner_release asked %q, %v", asked, err)
}
names = append(names, rel.Name)
want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again runner_health runner_release"
if got := strings.Join(names, " "); got != want {
t.Errorf("tools %q, want %q", got, want)
}
@@ -56,7 +56,7 @@ var _ pipeline.Asker = busAsker{}
// hearDecided wakes the runner on the operator-channel's decided for its own asks (§37, A-89): the warrant is read from the
// router's record, so an event the record does not back yet is heard again.
func hearDecided(hub *process.EventHub, h *pipeline.Hearer, module string, logf func(string, ...any)) error {
func hearDecided(hub handlers, h *pipeline.Hearer, module string, logf func(string, ...any)) error {
return hub.Handle(oc.SubjectDecided(module), func(m jetstream.Msg) {
var ev oc.Warrant
if err := json.Unmarshal(m.Data(), &ev); err != nil {
@@ -0,0 +1,44 @@
package main
import (
"context"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-pipeline/internal/pipeline"
)
// hearBuildEnds hears the builder's stream of build ends, kept last per subject and read whole at the start, then followed
// (design 47 G A-181, §556): when the build of the runner's new version ended is the time the wait's bound counts from, and
// each end heard for a runner held pipelines await starts their clock (§266, A-206). A stream it cannot hear is said, and the
// bound of such a wait is not armed until the runner starts again.
func hearBuildEnds(ctx context.Context, cred process.Account, js jetstream.JetStream, ends *pipeline.BuildEnds, st *pipeline.Starter, store pipeline.Updater,
logf func(string, ...any)) {
stream := catalogue.StreamBusName(builds.BuildsStream)
cc, err := cred.HearStream(ctx, js, stream, func(m jetstream.Msg) {
defer m.Ack()
md, err := m.Metadata()
if err != nil {
logf("%s has no time the stream stamped: %v", m.Subject(), err)
return
}
module, fp, ok := ends.Heard(m.Data(), md.Timestamp)
if !ok {
return
}
if n, err := st.Clocked(store, module, fp); err != nil {
logf("the clock of the %d pipelines held for %s: %v", n, fp, err)
} else if n > 0 {
logf("the build of %s ended: the bound of the %d pipelines held for it counts from it", fp, n)
}
}, logf)
if err != nil {
logf("the stream %s is not heard, so no wait for the runner's new version has its bound armed until the runner starts again: %v", stream, err)
return
}
go func() { <-ctx.Done(); cc.Stop() }()
}
@@ -0,0 +1,76 @@
//go:build integration
package main
import (
"context"
"encoding/json"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/testbus"
"mesh-delivery-pipeline/internal/pipeline"
)
// The runner hears the builder's stream of build ends, kept last per subject, read whole at its start: the end of its new version's
// build, published before it started, starts the clock of the pipelines held for it, counted from the time the stream stamped the end
// (design 47 G A-181, §266, A-206).
func TestTheRunnerHearsTheBuildEndOfItsNewVersionAtItsStart(t *testing.T) {
nc, err := nats.Connect(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
defer nc.Close()
js, err := jetstream.New(nc)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
stream, subject := catalogue.StreamBusName(builds.BuildsStream), process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEvent+".>")
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil {
t.Fatal(err)
}
published := time.Now().UTC()
if _, err := js.Publish(ctx, process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEventName("mesh-delivery-pipeline", "src2:new")),
[]byte(`{"module":"mesh-delivery-pipeline","fingerprint":"src2:new","outcome":"done"}`)); err != nil {
t.Fatal(err)
}
cred := process.Account{Node: "n1", Module: "mesh-delivery-pipeline"}
durable := cred.SeatEventConsumerConfig([]string{subject}, true)
durable.DeliverPolicy = jetstream.DeliverNewPolicy
if _, err := js.CreateOrUpdateConsumer(ctx, stream, durable); err != nil {
t.Fatal(err)
}
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "pipelines", History: 1})
if err != nil {
t.Fatal(err)
}
store := pipelineStore{ctx: ctx, kv: kv}
held, _ := json.Marshal(seat.Pipeline{Pipeline: "main.4.app", Branch: "main", Module: "app", Outcome: seat.PipelineOutcomeRunning,
Held: &seat.Held{Since: published, Fingerprint: "src2:new"}})
if _, err := kv.Create(ctx, "main.4.app", held); err != nil {
t.Fatal(err)
}
ends := &pipeline.BuildEnds{}
armed := make(chan time.Time, 1)
st := &pipeline.Starter{Self: "mesh-delivery-pipeline", Ends: ends.At, Say: func(string, any) error { return nil },
Hold: func(fp string, from time.Time) error { armed <- from; return nil }}
hearBuildEnds(ctx, cred, js, ends, st, store, t.Logf)
select {
case from := <-armed:
if from.Before(published.Add(-time.Minute)) || from.After(time.Now().Add(time.Second)) {
t.Errorf("the bound counts from %s, not from the end published at %s", from, published)
}
case <-ctx.Done():
t.Fatal("the end of the new runner's build was not heard at the start")
}
}
@@ -0,0 +1,57 @@
package main
import (
"context"
"sync"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/process"
)
// handlers is where a part of the runner registers its handler of the events it hears: the event hub, or agedHub over it.
type handlers interface {
Handle(pattern string, handle func(jetstream.Msg)) error
}
// agedHub registers each handler on the hub so that every message it is handed is noted delivered on the consumer's
// ledger, and applied once the handler acknowledges or ends it, so the runner's state consumers holds when the
// consumer's oldest unacknowledged message was delivered (design 47 G §76, §468, §495; A-164, A-166, A-176). A message
// the handler asks again (a refusal for now) stays unacknowledged.
type agedHub struct {
hub handlers
ledger *process.ConsumerLedger
}
func (a agedHub) Handle(pattern string, handle func(jetstream.Msg)) error {
return a.hub.Handle(pattern, func(m jetstream.Msg) {
a.ledger.Delivered(m)
handle(&agedMsg{Msg: m, ledger: a.ledger})
})
}
// agedMsg is a message whose acknowledgement or end is noted applied on its consumer's ledger, once.
type agedMsg struct {
jetstream.Msg
ledger *process.ConsumerLedger
once sync.Once
}
func (m *agedMsg) applied() { m.once.Do(func() { m.ledger.Applied(m.Msg, time.Time{}) }) }
func (m *agedMsg) Ack() error { defer m.applied(); return m.Msg.Ack() }
func (m *agedMsg) DoubleAck(ctx context.Context) error {
defer m.applied()
return m.Msg.DoubleAck(ctx)
}
func (m *agedMsg) Term() error { defer m.applied(); return m.Msg.Term() }
func (m *agedMsg) TermWithReason(r string) error { defer m.applied(); return m.Msg.TermWithReason(r) }
// consumersSink writes a consumer's facts in the state consumers.
func consumersSink(kv jetstream.KeyValue) process.ConsumerSink {
return func(ctx context.Context, k string, v []byte) error { _, err := kv.Put(ctx, k, v); return err }
}
@@ -0,0 +1,62 @@
package main
import (
"context"
"encoding/json"
"testing"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/process"
)
type oneHub struct{ handle func(jetstream.Msg) }
func (h *oneHub) Handle(_ string, handle func(jetstream.Msg)) error { h.handle = handle; return nil }
type ackedMsg struct {
jetstream.Msg
seq uint64
}
func (ackedMsg) Ack() error { return nil }
func (ackedMsg) Nak() error { return nil }
func (m ackedMsg) Metadata() (*jetstream.MsgMetadata, error) {
return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: m.seq}}, nil
}
func (ackedMsg) Subject() string { return "mesh.seat.x.event.y" }
// TestAgedHub: a message handed to a handler is noted delivered on the consumer's ledger, applied when the handler
// acknowledges it, and left unacknowledged when the handler asks it again (§76, §468; RN25, A-166, A-176).
func TestAgedHub(t *testing.T) {
var last process.ConsumerFacts
ledger := process.NewConsumerLedger(process.EventsStream, "n_runner", func(_ context.Context, _ string, v []byte) error {
last = process.ConsumerFacts{}
return json.Unmarshal(v, &last)
}, nil, nil)
hub := &oneHub{}
var inside process.ConsumerFacts
nak := false
if err := (agedHub{hub: hub, ledger: ledger}).Handle("x.>", func(m jetstream.Msg) {
inside = last
if nak {
_ = m.Nak()
return
}
_ = m.Ack()
}); err != nil {
t.Fatal(err)
}
hub.handle(ackedMsg{seq: 1})
if inside.OldestUnacknowledged.IsZero() {
t.Fatalf("inside the handler the consumer's facts = %+v, want its delivery written", inside)
}
if !last.OldestUnacknowledged.IsZero() || last.Acknowledged != 1 {
t.Errorf("after the handler acknowledged: %+v, want none unacknowledged and 1 acknowledged", last)
}
nak = true
hub.handle(ackedMsg{seq: 2})
if last.OldestUnacknowledged.IsZero() {
t.Errorf("a message asked again is noted applied: %+v", last)
}
}
@@ -2,6 +2,8 @@ package main
import (
"context"
"fmt"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -9,12 +11,32 @@ import (
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/views"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
)
// findingsWriter is the runner's finding writer as the mesh-sdk makes it.
type findingsWriter = views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]
// KindPipelineWaits is the finding of pipelines held for the runner's new version that the wait's bound started with the running
// runner (design 47 G §266, A-206).
const KindPipelineWaits = "pipeline-waits"
// lateFinding is the finding pipeline-waits of one pipeline the bound's expiry started: scope module, id its key
// `<branch>.<id>.<module>`, naming the fingerprint of the new runner it waited for and since when.
func lateFinding(p seat.Pipeline, held seat.Held, runner, node string, at time.Time) condseat.Finding {
return condseat.Finding{Scope: views.FindingScope, ID: p.Pipeline, Kind: KindPipelineWaits, Severity: condseat.FindingSeverityWarning,
Headline: "the new pipeline runner did not run healthy in time",
Summary: fmt.Sprintf("the pipeline %s was held since %s for the runner at %s to run healthy; the bound passed and the running runner runs it",
p.Pipeline, held.Since.UTC().Format(time.RFC3339), held.Fingerprint),
Source: "bound of the wait for the new runner", Observer: runner, At: at,
Subject: condseat.Subject{Scope: views.FindingScope, ID: p.Module, Machine: node}}
}
// findingWriter is the runner's mesh-sdk finding writer (design 47 G §461, §689): the runner observes only what its views
// reader finds, so its own observe states none and the writer's statement covers the reader's open findings. The writer
// forms its subjects itself (design 47 G §383, A-13).
func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) {
func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*findingsWriter, error) {
w, err := views.NewWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{
Module: cred.Module, State: state, Node: cred.Node, Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf,
Observe: func(context.Context) (views.Observed[condseat.Finding], error) {
@@ -0,0 +1,29 @@
package main
import (
"strings"
"testing"
"time"
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-sdk/go/views"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
)
// The bound's expiry's finding is pipeline-waits, scope module, id the pipeline's key, naming the awaited fingerprint and since
// when (design 47 G §266, A-206); it passes the writer's own check of a finding.
func TestLateFindingIsPipelineWaitsOfTheModuleScope(t *testing.T) {
since := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)
f := lateFinding(seat.Pipeline{Pipeline: "main.4.app", Module: "app", Branch: "main"}, seat.Held{Since: since, Fingerprint: "src2:new"},
"mesh-delivery-pipeline", "n1", since.Add(30*time.Minute))
if f.Kind != "pipeline-waits" || f.Scope != views.FindingScope || f.ID != "main.4.app" || f.Severity != condseat.FindingSeverityWarning {
t.Errorf("finding %+v", f)
}
if !strings.Contains(f.Summary, "src2:new") || !strings.Contains(f.Summary, "2026-11-11T12:00:00Z") {
t.Errorf("summary %q", f.Summary)
}
if err := views.CheckFinding(f); err != nil {
t.Error(err)
}
}
@@ -13,6 +13,7 @@ import (
"os/signal"
"sync/atomic"
"syscall"
"time"
"github.com/nats-io/nats.go"
@@ -71,6 +72,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
defer calls.Close()
hub := process.NewEventHub(logger.Printf)
// The durable consumer's facts, among them when its oldest unacknowledged message was delivered, read by the tool
// runner_health (§76, §468; A-164, A-166, A-176): written when a message is delivered or applied.
consumers, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, pipeline.StateConsumers), "say how far its consumer is behind")
if err != nil {
return err
}
heard := agedHub{hub: hub, ledger: process.NewConsumerLedger(process.EventsStream, cred.EventConsumerName(), consumersSink(consumers), nil, logger.Printf)}
// A branch's view holding its first half asks again the steps placed in its running pipelines that were never asked
// (§88); the starter is made after the views, so it is reached through a pointer set once it is.
var resumer atomic.Pointer[func(string)]
@@ -96,14 +104,29 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask)
// Its asking half (§22, §24): a pipeline's steps are placed from the module's registration and each build seat is asked
// for its step.
ends := &pipeline.BuildEnds{}
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}),
Ends: ends.At, Hold: armHeld(ctx, bounds),
// The bound's expiry writes the finding pipeline-waits for each pipeline it started (§266, A-206).
Late: func(p seat.Pipeline, held seat.Held) {
module, node := rv.Findings.ModuleNode()
if err := rv.Findings.Write(ctx, lateFinding(p, held, module, node, time.Now().UTC())); err != nil {
logger.Printf("the finding pipeline-waits of %s is not written: %v", p.Pipeline, err)
}
}}
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)
@@ -112,25 +135,36 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
logger.Printf("the steps placed on %s and not yet asked: %d asked again", branch, n)
}
resumer.Store(&resume)
if err := hearStepEnded(hub, hearer, logger.Printf); err != nil {
if err := hearStepEnded(heard, hearer, logger.Printf); err != nil {
return err
}
if err := hearToolchain(ctx, hub, toolchains, logger.Printf); err != nil {
if err := hearToolchain(ctx, heard, toolchains, logger.Printf); err != nil {
return err
}
if err := hearAskTaken(ctx, hub, bounds, logger.Printf); err != nil {
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 := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil {
hearBuildEnds(ctx, cred, js, ends, starter, store, logger.Printf)
if err := hearRunning(ctx, heard, cred.Module, starter, store, bounds, rv.Catalog.View, logger.Printf); err != nil {
return err
}
if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil {
if err := hearDecided(heard, hearer, cred.Module, logger.Printf); err != nil {
return err
}
if err := hearTakenIn(hub, starter, rv.Catalog, logger.Printf); err != nil {
if err := hearPublished(ctx, heard, starter, calls, rv.Catalog, logger.Printf); err != nil {
return err
}
if err := hearTakenIn(heard, starter, rv.Catalog, logger.Printf); err != nil {
return err
}
if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil {
@@ -6,6 +6,7 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
"sync"
@@ -20,6 +21,8 @@ import (
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/testbus"
"mesh-delivery-pipeline/internal/pipeline"
)
type syncBuffer struct {
@@ -53,7 +56,7 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes
defer cancel()
const module = "mesh-delivery-pipeline"
for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views"} {
for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views", "consumers"} {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_" + state, History: 1}); err != nil {
t.Fatal(err)
}
@@ -74,7 +77,7 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes
}
// The seats the runner calls, answered as their holders do.
reg := catalog.Registration{Module: "app", Branches: []string{"main"}, Origin: "o/app", Policy: catalog.Policy{Steps: catalog.StepStatuses{Review: "optional"}}}
reg := catalog.Registration{Module: "app", Branches: []string{"main"}, Origin: "o/app", Policy: catalog.Policy{Steps: catalog.StepStatuses{Review: "optional", Composition: "bypass", Approval: "bypass"}}}
answer := func(subject string, body any) {
if _, err := nc.Subscribe(subject, func(m *nats.Msg) {
raw, _ := json.Marshal(map[string]any{"result": body})
@@ -157,6 +160,43 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes
})
waitFor(t, ctx, logs, func() bool { return readPipeline(ctx, pipelines, key, &p) && p.Outcome == seat.PipelineOutcomePassed })
// RN25 (§76, §468; A-166): the durable consumer's age is in the state consumers, none waiting once every message heard
// is acknowledged, and runner_health reads it healthy.
consumers, err := js.KeyValue(ctx, module+"_consumers")
if err != nil {
t.Fatal(err)
}
var facts process.ConsumerFacts
waitFor(t, ctx, logs, func() bool {
e, err := consumers.Get(ctx, "n1_"+module)
return err == nil && json.Unmarshal(e.Value(), &facts) == nil && facts.OldestUnacknowledged.IsZero() && facts.Acknowledged > 0
})
if a := pipeline.Health(kvKept{ctx, pipelines}, kvKept{ctx, consumers}, time.Now()); !a.Healthy {
t.Errorf("runner_health over the bus: %+v", a)
}
}
// kvKept reads a bucket as the tools bundle's states do.
type kvKept struct {
ctx context.Context
kv jetstream.KeyValue
}
func (k kvKept) Keys() ([]string, error) {
keys, err := k.kv.Keys(k.ctx)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
}
return keys, err
}
func (k kvKept) Value(key string) (json.RawMessage, error) {
e, err := k.kv.Get(k.ctx, key)
if err != nil {
return nil, err
}
return e.Value(), nil
}
func stepOf(p seat.Pipeline, step string) seat.StepResult {
@@ -36,7 +36,7 @@ func (c createdKV) Create(key string, value []byte) error {
// hearPublished starts the pipelines of each package-published heard (§53, §334): the dependents the catalog's outdated
// names behind, each asked as a kept call. The message is acknowledged once its starts are written, on a goroutine of its
// own, so a slow catalog holds no other event.
func hearPublished(ctx context.Context, hub *process.EventHub, starter *pipeline.Starter, calls *process.Conn, v *pipeline.Views, logf func(string, ...any)) error {
func hearPublished(ctx context.Context, hub handlers, starter *pipeline.Starter, calls *process.Conn, v *pipeline.Views, logf func(string, ...any)) error {
outdated := func(in catalog.OutdatedInput) (catalog.OutdatedAnswer, error) {
raw, err := calls.CallKept(outdatedVerb, in)
if err != nil {
@@ -72,7 +72,7 @@ func hearPublished(ctx context.Context, hub *process.EventHub, starter *pipeline
// hearTakenIn starts the pipelines of each take-in.taken heard (§44, §54): one per entry, the message acknowledged once
// they are written, and heard again when a write failed, so the keys already written start nothing twice.
func hearTakenIn(hub *process.EventHub, starter *pipeline.Starter, v *pipeline.Views, logf func(string, ...any)) error {
func hearTakenIn(hub handlers, starter *pipeline.Starter, v *pipeline.Views, logf func(string, ...any)) error {
return hub.Handle(pipeline.TakenInSubject, func(m jetstream.Msg) {
var ev catalog.TakeInTaken
if err := json.Unmarshal(m.Data(), &ev); err != nil {
@@ -6,9 +6,11 @@ import (
"errors"
"fmt"
"strings"
"time"
"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"
@@ -99,7 +101,7 @@ func newHearer(ctx context.Context, js jetstream.JetStream, cred process.Account
// hearStepEnded takes each build seat's step-ended into its pipeline (§24; ADR 0666 §21): refused and said, never taken,
// when it is not as the runner asked it, and heard again when a write failed.
func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string, ...any)) error {
func hearStepEnded(hub handlers, h *pipeline.Hearer, logf func(string, ...any)) error {
for _, bs := range pipeline.StepSeats {
bs := bs
err := hub.Handle(bs.Prefix+"*", func(m jetstream.Msg) {
@@ -137,7 +139,7 @@ func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string,
// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647): the ask is answered, so its five minutes
// end with no finding.
func hearAskTaken(ctx context.Context, hub *process.EventHub, bounds jetstream.KeyValue, logf func(string, ...any)) error {
func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, logf func(string, ...any)) error {
for _, ts := range pipeline.TakenSeats {
ts := ts
err := hub.Handle(ts.Subject, func(m jetstream.Msg) {
@@ -172,9 +174,54 @@ 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 counted from the end of that runner's build (design 47 G §266, A-195, A-77,
// A-206).
func armHeld(ctx context.Context, bounds jetstream.KeyValue) func(string, time.Time) error {
return func(fp string, from time.Time) error {
_, err := bounds.Create(ctx, pipeline.HeldKey(fp), []byte(`{"fingerprint":"`+fp+`"}`), jetstream.KeyTTL(pipeline.HeldTTL(from, time.Now())))
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 +236,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
}
@@ -252,7 +303,7 @@ func toolchainReader(ctx context.Context, toolchains jetstream.KeyValue) func(st
}
// hearToolchain keeps each toolchain-agreed heard as `<language>` in the state toolchains (A-104): the newest of a language.
func hearToolchain(ctx context.Context, hub *process.EventHub, toolchains jetstream.KeyValue, logf func(string, ...any)) error {
func hearToolchain(ctx context.Context, hub handlers, toolchains jetstream.KeyValue, logf func(string, ...any)) error {
return hub.Handle(builds.SubjectToolchainAgreed, func(m jetstream.Msg) {
t, err := pipeline.ToolchainOf(m.Data())
if err != nil || t == nil {
@@ -32,15 +32,24 @@ func verbOf[In any](do func(caller string, in In) (any, error)) process.SeatHand
}
}
// answerVerbs answers the seat's write verbs `review`, `rerun`, `accept` and `cancel` over the bus, each writing the state
// pipelines by compare-and-set (design 47 G §22, §70, §482, A-107). The returned stop ends the answers.
// answerVerbs answers the seat's write verbs over the bus, each writing the state pipelines by compare-and-set (design 47
// G §22, §70, §306, §482, A-107). The returned stop ends the answers.
func answerVerbs(conn *process.Conn, v pipeline.Verbs) (func(), error) {
return conn.AnswerSeat(seat.Seat, map[string]process.SeatHandler{
seat.VerbReview: verbOf(func(c string, in seat.ReviewInput) (any, error) { return v.Review(c, in) }),
seat.VerbRerun: verbOf(func(c string, in seat.RerunInput) (any, error) { return v.Rerun(c, in) }),
seat.VerbAccept: verbOf(func(c string, in seat.AcceptInput) (any, error) { return v.Accept(c, in) }),
seat.VerbCancel: verbOf(func(c string, in seat.CancelInput) (any, error) { return v.Cancel(c, in) }),
})
return conn.AnswerSeat(seat.Seat, writeHandlers(v))
}
// writeHandlers are the seat's write verbs the process answers: review, rerun, accept, cancel, start, ask-again and
// release (§306; RN14; A-206).
func writeHandlers(v pipeline.Verbs) map[string]process.SeatHandler {
return map[string]process.SeatHandler{
seat.VerbReview: verbOf(func(c string, in seat.ReviewInput) (any, error) { return v.Review(c, in) }),
seat.VerbRerun: verbOf(func(c string, in seat.RerunInput) (any, error) { return v.Rerun(c, in) }),
seat.VerbAccept: verbOf(func(c string, in seat.AcceptInput) (any, error) { return v.Accept(c, in) }),
seat.VerbCancel: verbOf(func(c string, in seat.CancelInput) (any, error) { return v.Cancel(c, in) }),
seat.VerbStart: verbOf(func(c string, in seat.StartInput) (any, error) { return v.Start(c, in) }),
seat.VerbAskAgain: verbOf(func(c string, in seat.AskAgainInput) (any, error) { return v.AskAgain(c, in) }),
seat.VerbRelease: verbOf(func(c string, in seat.ReleaseInput) (any, error) { return v.Release(c, in) }),
}
}
// dropBound deletes the bound of an ask a cancelled pipeline no longer waits on.
@@ -2,11 +2,29 @@ package main
import (
"encoding/json"
"slices"
"testing"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
"mesh-delivery-pipeline/internal/pipeline"
)
// RN12, RN14 (§306): the process answers every write verb of the seat, the bundle every read verb, so each verb the
// manifest claims is served.
func TestEveryWriteVerbIsServed(t *testing.T) {
var got []string
for v := range writeHandlers(pipeline.Verbs{}) {
got = append(got, v)
}
slices.Sort(got)
want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbRelease, seat.VerbReview, seat.VerbStart}
slices.Sort(want)
if !slices.Equal(got, want) {
t.Errorf("the process answers %v, want %v", got, want)
}
}
// A write verb is answered only for a caller the bus proved, and its input is the verb's own (design 47 G §394, A-107).
func TestAVerbIsAnsweredOnlyForAProvedCallerAndItsOwnInput(t *testing.T) {
var got seat.CancelInput
@@ -20,6 +20,8 @@ import (
type runnerViews struct {
Registrations *meshcatalogclient.Registrations
Catalog *pipeline.Views
// Findings is the runner's finding writer, which also says the runner's own findings (A-206).
Findings *findingsWriter
}
// startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or
@@ -53,7 +55,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr
return nil, err
}
reader.Start(ctx, js, cred, status)
return &runnerViews{Registrations: regs, Catalog: cv}, nil
return &runnerViews{Registrations: regs, Catalog: cv, Findings: findings}, nil
}
// catalogView is a branch's meshcatalogclient.Catalog as the runner reads it.
+2
View File
@@ -11,9 +11,11 @@ require (
require (
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000
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
@@ -51,7 +51,15 @@ func approvalDue(p seat.Pipeline) int {
// ApprovalAsk is the ask of a pipeline's approval step. Its id and its expiry follow from the pipeline and from when the
// step was asked, so the ask built again for the same pipeline is the same ask, digest included: a warrant is held to it.
func ApprovalAsk(p seat.Pipeline, from time.Time) (oc.Ask, error) {
sum := sha256.Sum256([]byte(p.Pipeline + "\n" + p.Commit))
// An approval asked again (ask-again) is an ask of its own, its id following from when it was asked again, so the
// ask rebuilt from the step is the same and the earlier, decided ask is not its duplicate.
idOf := p.Pipeline + "\n" + p.Commit
for _, st := range p.Steps {
if st.Step == seat.StepResultStepApproval && st.Rerun {
idOf += "\nasked again\n" + from.UTC().Format(time.RFC3339Nano)
}
}
sum := sha256.Sum256([]byte(idOf))
act := func(do string) (string, error) {
return operatorchannelclient.ActDigest(operatorchannelclient.Act{"pipeline": p.Pipeline, "commit": p.Commit,
"fingerprint": p.Fingerprint, "do": do})
@@ -98,6 +106,10 @@ func askApproval(key string, p *seat.Pipeline, from time.Time, asker Asker) (boo
if i < 0 || asker == nil {
return false, nil
}
// An approval asked again keeps the time ask-again wrote, so an ask published and then not written is the same ask.
if s := p.Steps[i]; s.Rerun && s.Started != nil {
from = *s.Started
}
a, err := ApprovalAsk(*p, from)
if err != nil {
return false, fmt.Errorf("the approval of %s cannot be asked: %w", key, err)
@@ -0,0 +1,59 @@
package pipeline
import (
"encoding/json"
"fmt"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/process"
)
// StateConsumers is the runner's old state section of its durable consumers' facts, a key per consumer, written by
// mesh-sdk's consumer ledger (design 47 G §468, §76, §495; A-164, A-166, A-176).
const StateConsumers = process.ConsumersState
// BehindAfter is how long a consumer's oldest unacknowledged message may wait before the runner's health check names it
// behind (§76).
const BehindAfter = 5 * time.Minute
// Health is the runner's health as its tool runner_health answers it (§468, §76, §495; A-164): healthy when its state
// pipelines answers, as show reads it, and each durable consumer's oldest unacknowledged message waits less than
// BehindAfter; otherwise unhealthy naming each consumer behind and how long.
func Health(pipelines, consumers Kept, now time.Time) health.Answer {
if _, err := pipelines.Keys(); err != nil {
return health.Unhealthy(fmt.Sprintf("The runner does not answer show: its state pipelines cannot be listed (%v).", err))
}
keys, err := consumers.Keys()
if err != nil {
return health.Unhealthy(fmt.Sprintf("The ages of the runner's consumers cannot be read from its state %s (%v).", StateConsumers, err))
}
if len(keys) == 0 {
return health.Healthy(fmt.Sprintf("The runner answers show and has heard no event yet: its state %s holds no consumer.", StateConsumers))
}
sort.Strings(keys)
var behind []string
for _, k := range keys {
raw, err := consumers.Value(k)
if err != nil {
return health.Unhealthy(fmt.Sprintf("The age of the consumer %s cannot be read (%v).", k, err))
}
var a process.ConsumerFacts
if raw != nil {
if err := json.Unmarshal(raw, &a); err != nil {
return health.Unhealthy(fmt.Sprintf("The age of the consumer %s does not read (%v).", k, err))
}
}
if !a.OldestUnacknowledged.IsZero() {
if age := now.Sub(a.OldestUnacknowledged); age >= BehindAfter {
behind = append(behind, fmt.Sprintf("%s (its oldest unacknowledged message waits %s)", k, age.Round(time.Second)))
}
}
}
if len(behind) > 0 {
return health.Unhealthy(fmt.Sprintf("The runner's consumers behind by %s or more: %s.", BehindAfter, strings.Join(behind, ", ")))
}
return health.Healthy(fmt.Sprintf("The runner answers show, and each of its %d consumers is less than %s behind.", len(keys), BehindAfter))
}
@@ -0,0 +1,60 @@
package pipeline
import (
"encoding/json"
"errors"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/process"
)
type mapKept map[string]json.RawMessage
func (m mapKept) Keys() ([]string, error) {
var out []string
for k := range m {
out = append(out, k)
}
return out, nil
}
func (m mapKept) Value(k string) (json.RawMessage, error) { return m[k], nil }
type failingKept struct{}
func (failingKept) Keys() ([]string, error) { return nil, errors.New("no bucket") }
func (failingKept) Value(string) (json.RawMessage, error) { return nil, errors.New("no bucket") }
// TestHealth: runner_health is healthy when the runner answers show and each consumer is less than five minutes behind,
// and otherwise names each consumer behind (§76, §468, §495; RN25, A-164).
func TestHealth(t *testing.T) {
now := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC)
age := func(name string, since time.Duration) mapKept {
a := process.ConsumerFacts{Name: name}
if since > 0 {
a.OldestUnacknowledged = now.Add(-since)
}
raw, _ := json.Marshal(a)
return mapKept{name: raw}
}
rows := []struct {
name string
pipelines Kept
consumers Kept
healthy bool
says string
}{
{"none waiting", mapKept{}, age("n_runner", 0), true, "less than 5m0s behind"},
{"four minutes", mapKept{}, age("n_runner", 4*time.Minute), true, ""},
{"six minutes names the consumer", mapKept{}, age("n_runner", 6*time.Minute), false, "n_runner (its oldest unacknowledged message waits 6m0s)"},
{"show does not answer", failingKept{}, age("n_runner", 0), false, "does not answer show"},
{"no event heard yet", mapKept{}, mapKept{}, true, "has heard no event yet"},
}
for _, r := range rows {
a := Health(r.pipelines, r.consumers, now)
if a.Healthy != r.healthy || !strings.Contains(a.Why, r.says) {
t.Errorf("%s: healthy %v %q; want %v saying %q", r.name, a.Healthy, a.Why, r.healthy, r.says)
}
}
}
@@ -0,0 +1,60 @@
package pipeline
import (
"strings"
"testing"
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"
)
// pipeline-ended carries the module's test counts and says it over budget past 500 unit or 50 integration tests (design 47 G
// §24, §419), and a group's one pipeline ends with group-ended under the group once every member ended (§73, §332).
func TestPipelineEndedCarriesTestCountsAndAGroupEndsOnItsMembers(t *testing.T) {
h, store, _, _, _, _ := fixture()
bodies := map[string]any{}
var order []string
h.Say = func(subject string, body any) error {
ev := subject[strings.Index(subject, ".event.")+7:]
bodies[ev] = body
order = append(order, ev)
return nil
}
e := store[pkey]
e.p.Steps = e.p.Steps[:2]
e.p.Steps[1].Status = seat.StepResultStatusRequired
e.p.Group = "g1"
store[pkey] = e
// A second member of the same group, on the same source revision, still running.
store["main.5.lib"] = struct {
p seat.Pipeline
rev uint64
}{seat.Pipeline{V: 1, Pipeline: "main.5.lib", Branch: "main", Source: 5, Module: "lib", Commit: "c2", Group: "g1", Outcome: seat.PipelineOutcomeRunning,
Steps: []seat.StepResult{{Step: seat.StepResultStepLint, Status: seat.StepResultStatusRequired, Outcome: seat.StepResultOutcomeRunning, Ask: "b1", Seat: nodebuildgo.Seat}}}, 1}
lint := ended("a1", "lint", "passed")
unit := ended("a2", "unit", "passed")
unit.Tests = &nodebuildgo.TestCounts{Unit: 501, Integration: 0}
for _, ev := range []StepEnded{lint, unit} {
if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ev); err != nil {
t.Fatal(err)
}
}
pe, ok := bodies["pipeline-ended.app.main.g1"].(seat.PipelineEnded)
if !ok || pe.Tests == nil || pe.Tests.Unit != 501 || !pe.OverBudget {
t.Fatalf("pipeline-ended = %+v (said %v)", bodies["pipeline-ended.app.main.g1"], order)
}
if _, said := bodies["group-ended.main.g1"]; said {
t.Fatalf("group-ended said while a member runs: %v", order)
}
lib := ended("b1", "lint", "passed")
lib.Pipeline, lib.Module, lib.Commit, lib.Fingerprint = "main.5.lib", "lib", "c2", ""
if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", lib); err != nil {
t.Fatal(err)
}
ge, ok := bodies["group-ended.main.g1"].(seat.GroupEnded)
if !ok || ge.Outcome != seat.GroupEndedOutcomePassed || ge.Group != "g1" || ge.Branch != "main" || ge.Source != 5 || len(ge.Pipelines) != 2 {
t.Fatalf("group-ended = %+v (said %v)", bodies["group-ended.main.g1"], order)
}
// A pipeline of no group says no group-ended.
}
@@ -0,0 +1,46 @@
package pipeline
import (
"encoding/json"
"sync"
"time"
)
// BuildEnds are the done builds' ends heard from the builder's stream of build ends (event build.ended, kept last per subject
// and read whole at the start, design 47 G A-181): when each build, by module and fingerprint, ended, which is the time a held
// wait's bound counts from (§266, A-206).
type BuildEnds struct {
mu sync.Mutex
at map[string]time.Time
}
// buildEnd is the part of the body of build.ended the runner reads.
type buildEnd struct {
Module string `json:"module"`
Fingerprint string `json:"fingerprint"`
Outcome string `json:"outcome"`
}
// Heard keeps the end a message of the stream says, at the time the stream stamped it; a failed end, or a body naming no module
// and fingerprint, is kept for nothing. It says the module and fingerprint it kept.
func (e *BuildEnds) Heard(data []byte, at time.Time) (module, fp string, ok bool) {
var b buildEnd
if err := json.Unmarshal(data, &b); err != nil || b.Module == "" || b.Fingerprint == "" || b.Outcome == "failed" {
return "", "", false
}
e.mu.Lock()
defer e.mu.Unlock()
if e.at == nil {
e.at = map[string]time.Time{}
}
e.at[b.Module+"\x00"+b.Fingerprint] = at.UTC()
return b.Module, b.Fingerprint, true
}
// At is when the build of the module at the fingerprint ended, false when no end was heard.
func (e *BuildEnds) At(module, fp string) (time.Time, bool) {
e.mu.Lock()
defer e.mu.Unlock()
t, ok := e.at[module+"\x00"+fp]
return t, ok
}
@@ -0,0 +1,269 @@
package pipeline
import (
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"slices"
"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)
}
from := s.clockOf(awaiting)
if from != nil && s.Hold != nil {
if err := s.Hold(awaiting, *from); err != nil {
s.logf("the bound of the wait for the runner at %s is not armed: %v", awaiting, err)
}
}
w := s.waitOf(p, now, "")
w.ClockFrom = from
if err := s.Say(runnerclient.Subject(seat.EventWaitHeld, b.Module, b.Branch, commitOrGroup(p)), w); err != nil {
s.logf("wait-held of %s is not said: %v", key, err)
}
return true, nil
}
// clockOf is when the build of the runner at fp ended, the time the bound counts from, nil while it has not (§266, A-181, A-206).
func (s *Starter) clockOf(fp string) *time.Time {
if s.Ends == nil {
return nil
}
if at, ok := s.Ends(s.Self, fp); ok {
return &at
}
return nil
}
// Clocked starts the clock of the pipelines held for the runner at fp, the end of whose build was heard: it arms the bound from that
// end and says wait-held again for each with `clock-from`, so the watchman times it by the bound (§269, A-206). It says how many
// pipelines it said, and nothing when none is held for fp or the end is not known.
func (s *Starter) Clocked(store Updater, module, fp string) (int, error) {
if module != s.Self {
return 0, nil
}
from := s.clockOf(fp)
if from == nil {
return 0, nil
}
pkeys, err := store.Keys("*.*.*")
if err != nil {
return 0, fmt.Errorf("the pipelines are not listed: %w", err)
}
said := 0
var first error
for _, key := range pkeys {
p, _, err := store.Get(key)
if err != nil || p.Held == nil || p.Outcome != seat.PipelineOutcomeRunning || p.Held.Fingerprint != fp {
continue
}
if said == 0 && s.Hold != nil {
if err := s.Hold(fp, *from); err != nil {
first = fmt.Errorf("the bound of the wait for the runner at %s is not armed: %w", fp, err)
}
}
w := s.waitOf(p, p.Held.Since, "")
w.ClockFrom = from
if err := s.Say(runnerclient.Subject(seat.EventWaitHeld, p.Module, p.Branch, commitOrGroup(p)), w); err != nil && first == nil {
first = fmt.Errorf("wait-held of %s is not said: %w", key, err)
}
said++
}
return said, first
}
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.WaitOnRunnerVersion, 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, and each started is said late through Late (A-206).
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)
}
fp := ""
held := map[string]seat.Held{}
for _, k := range pkeys {
if p, _, err := store.Get(k); err == nil && p.Held != nil && p.Outcome == seat.PipelineOutcomeRunning && HeldKey(p.Held.Fingerprint) == key {
fp, held[k] = p.Held.Fingerprint, *p.Held
}
}
if fp == "" {
return 0, nil
}
n, err := s.Release(store, fp, "released", view)
for k, h := range held {
if p, _, gerr := store.Get(k); gerr == nil && p.Held == nil && s.Late != nil {
s.Late(p, h)
}
}
return n, err
}
// heldFingerprints are the fingerprints of the runner the held pipelines in store await, each once.
func heldFingerprints(store Updater) []string {
pkeys, err := store.Keys("*.*.*")
if err != nil {
return nil
}
var fps []string
for _, key := range pkeys {
if p, _, err := store.Get(key); err == nil && p.Held != nil && p.Outcome == seat.PipelineOutcomeRunning && !slices.Contains(fps, p.Held.Fingerprint) {
fps = append(fps, p.Held.Fingerprint)
}
}
return fps
}
// Release starts by hand the pipelines held for the runner's new version with the running runner, recorded in the hand-act
// log first (§276, A-206): each says wait-ended `released` and pipeline-started, and the bound of each wait is dropped. It
// says how many it started.
func (v Verbs) Release(_ string, in seat.ReleaseInput) (seat.ReleaseAnswer, error) {
if v.Starter == nil || v.View == nil {
return seat.ReleaseAnswer{}, errors.New("the runner cannot release a pipeline: it holds no starter or no catalog half")
}
if err := v.record(seat.VerbRelease, "started the pipelines held for the new runner by hand", in.Why); err != nil {
return seat.ReleaseAnswer{}, err
}
fps := heldFingerprints(v.Hearer.Store)
n, err := v.Starter.Release(v.Hearer.Store, "", "released", v.View)
if n > 0 && v.Drop != nil {
for _, fp := range fps {
if derr := v.Drop(HeldKey(fp)); derr != nil && err == nil {
err = fmt.Errorf("the bound of the wait for %s is not dropped: %w", fp, derr)
}
}
}
return seat.ReleaseAnswer{Released: int64(n)}, err
}
// HeldTTL is how long the bound of a wait that counts from the build end at from still has to run at now: the 30 minutes less what passed since
// the end, at least a second, the least a key's time to live holds (design 47 G §266, A-206).
func HeldTTL(from, now time.Time) time.Duration {
return max(HeldBound-now.Sub(from), time.Second)
}
@@ -0,0 +1,234 @@
package pipeline
import (
"encoding/json"
"strings"
"testing"
"time"
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), Ends: func(string, string) (time.Time, bool) { return time.Unix(9, 0), true },
Hold: func(fp string, _ time.Time) 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")
}
}
// The verb release starts the held pipelines by hand, recorded in the hand-act log first, saying why and ending each wait
// `released`, and drops the bound of the wait; a release with no why does nothing (design 47 G §276, A-206).
func TestReleaseVerbStartsTheHeldPipelinesByHand(t *testing.T) {
reg, entries := placeFixture()
entries["runner"] = catalog.CatalogEntry{Module: "runner"}
store := memCreated{memStore{}}
asks := &stepAsks{}
var said []string
s := &Starter{Pipelines: store, Steps: asks, Name: "the-runner", Self: "runner", SelfFp: "src2:old", Awaiting: Awaiting(store.memStore),
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 }
if _, 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); err != nil {
t.Fatal(err)
}
acts, dropped := &actLog{}, []string{}
v := Verbs{Hearer: &Hearer{Store: store.memStore}, Starter: s, View: view, Acts: acts,
Drop: func(k string) error { dropped = append(dropped, k); return nil }}
if _, err := v.Release("operator", seat.ReleaseInput{}); err == nil || len(*acts) != 0 {
t.Fatalf("a release saying no why: %v, acts %v", err, *acts)
}
said = nil
a, err := v.Release("operator", seat.ReleaseInput{Why: "the new runner will not come"})
if err != nil || a.Released != 2 {
t.Fatalf("release = %+v, %v", a, err)
}
if p, _, _ := store.Get("main.4.app"); p.Held != nil || len(p.Steps) == 0 || len(asks.asks) == 0 {
t.Errorf("after the release: held %+v, steps %d", p.Held, len(p.Steps))
}
if len(*acts) != 1 || !strings.HasPrefix((*acts)[0], "mesh-pipeline-runner.release: ") || len(dropped) != 1 || dropped[0] != HeldKey("src2:new") {
t.Errorf("acts %v, dropped %v", *acts, dropped)
}
if !strings.Contains(strings.Join(said, "|"), "wait-ended.app.main.c1") {
t.Errorf("said %v", said)
}
}
// The bound's expiry starts the held pipelines and says each as late through the starter's Late, with the awaited fingerprint and
// since when; a pipeline that stays held (no view) is not said (design 47 G §266, A-206).
func TestExpiredBoundSaysEachReleasedPipelineLate(t *testing.T) {
reg, entries := placeFixture()
store := memCreated{memStore{}}
since := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)
for _, k := range []string{"main.4.app", "dev.4.app"} {
store.memStore[k] = struct {
p seat.Pipeline
rev uint64
}{seat.Pipeline{Pipeline: k, Branch: strings.Split(k, ".")[0], Module: "app", Outcome: seat.PipelineOutcomeRunning,
Held: &seat.Held{Since: since, Fingerprint: "src2:new"}}, 1}
}
var late []string
s := &Starter{Pipelines: store, Steps: &stepAsks{}, Name: "the-runner", Say: func(string, any) error { return nil },
Registration: func(string) (catalog.Registration, bool) { return reg, true },
Late: func(p seat.Pipeline, h seat.Held) {
late = append(late, p.Pipeline+"@"+h.Fingerprint+"@"+h.Since.Format(time.RFC3339))
}}
view := func(b string) (Half, bool) { return Half{Entries: entries}, b == "main" }
if n, err := s.ReleaseExpired(store.memStore, HeldKey("src2:new"), view); err == nil || n != 1 {
t.Fatalf("release = %d, %v (the pipeline of a branch with no view stays held, an error)", n, err)
}
if len(late) != 1 || late[0] != "main.4.app@src2:new@2026-11-11T12:00:00Z" {
t.Errorf("late %v", late)
}
}
// A wait of the step pipeline waits on the runner version (design 47 G §534, A-206).
func TestHeldWaitIsOnTheRunnerVersion(t *testing.T) {
w := (&Starter{}).waitOf(seat.Pipeline{Pipeline: "main.4.app", Branch: "main", Module: "app"}, time.Unix(1, 0), "")
if w.Step != seat.WaitStepPipeline || w.On != "runner version" {
t.Errorf("wait %+v", w)
}
}
// The wait of a held pipeline counts from the build end of the awaited runner: wait-held says it as `clock-from` and the bound is
// armed from it; while that build has not ended nothing is armed, and hearing its end arms the bound and says wait-held again
// with `clock-from`; a failed end starts no clock (design 47 G §266, §269, A-181, A-206).
func TestHeldWaitCountsFromTheBuildEndOfTheAwaitedRunner(t *testing.T) {
reg, entries := placeFixture()
store := memCreated{memStore{}}
ends := &BuildEnds{}
var waits []seat.Wait
var armed []string
s := &Starter{Pipelines: store, Steps: &stepAsks{}, Name: "the-runner", Self: "runner", SelfFp: "src2:old", Awaiting: Awaiting(store.memStore),
Ends: ends.At, Hold: func(fp string, from time.Time) error {
armed = append(armed, fp+"@"+from.Format(time.RFC3339))
return nil
},
Registration: func(string) (catalog.Registration, bool) { return reg, true },
Say: func(subject string, body any) error {
if w, ok := body.(seat.Wait); ok && strings.Contains(subject, ".wait-held.") {
waits = append(waits, w)
}
return nil
}}
view := func(string) (Half, bool) { return Half{Entries: entries}, true }
taken := 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"}}}
end := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)
// The runner's build has not ended: held, no clock, no bound.
if _, err := s.TakenIn(taken, view); err != nil {
t.Fatal(err)
}
if len(waits) != 2 || waits[0].ClockFrom != nil || len(armed) != 0 {
t.Fatalf("waits %+v, armed %v", waits, armed)
}
// A failed end starts no clock; a done end does.
if _, _, ok := ends.Heard([]byte(`{"module":"runner","fingerprint":"src2:new","outcome":"failed"}`), end); ok {
t.Error("a failed end was kept")
}
if m, fp, ok := ends.Heard([]byte(`{"module":"runner","fingerprint":"src2:new","outcome":"done"}`), end); !ok || m != "runner" || fp != "src2:new" {
t.Fatalf("heard %q %q %v", m, fp, ok)
}
waits = nil
if n, err := s.Clocked(store.memStore, "runner", "src2:new"); err != nil || n != 2 {
t.Fatalf("clocked = %d, %v", n, err)
}
if len(waits) != 2 || waits[0].ClockFrom == nil || !waits[0].ClockFrom.Equal(end) || len(armed) != 1 || armed[0] != "src2:new@2026-11-11T12:00:00Z" {
t.Errorf("waits %+v, armed %v", waits, armed)
}
// A pipeline held after the end is known says the clock at once.
waits = nil
taken.Key, taken.Source = "m.2", 5
if _, err := s.TakenIn(taken, view); err != nil {
t.Fatal(err)
}
if len(waits) == 0 || waits[0].ClockFrom == nil || !waits[0].ClockFrom.Equal(end) {
t.Errorf("waits %+v", waits)
}
}
// The bound counts from the build end: the 30 minutes less what passed since, never under a second.
func TestHeldTTLCountsFromTheBuildEnd(t *testing.T) {
end := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)
for _, c := range []struct {
after, want time.Duration
}{{0, 30 * time.Minute}, {10 * time.Minute, 20 * time.Minute}, {45 * time.Minute, time.Second}} {
if got := HeldTTL(end, end.Add(c.after)); got != c.want {
t.Errorf("after %s the bound has %s, want %s", c.after, got, c.want)
}
}
}
@@ -50,7 +50,7 @@ func AskID(pipeline, artifact, step string) string {
// Place places a pipeline's steps (design 47 G §21, §23, §323; A-91): only the steps the module's registration policy
// names, each artifact's lint, unit and integration steps beside the module's composition, review and approval. A step the
// policy does not name is no step. A `bypass` step is placed bypassed and asked of no one; a build step is placed running,
// policy leaves without a status is `optional` (§553). A `bypass` step is placed bypassed and asked of no one; a build step is placed running,
// asked of the build seat the artifact's language names in the catalog half (§323), or, where no seat names it, ended
// could not run naming the missing seat; the module's steps wait. It returns the steps and the asks to put on the seats'
// work queues, one per running step.
@@ -74,9 +74,6 @@ func Place(p seat.Pipeline, reg catalog.Registration, entries map[string]catalog
}
for _, step := range buildSteps {
status := Status(reg, name, step)
if status == "" {
continue
}
s := seat.StepResult{Step: step, Artifact: a.Ref, Status: status, Fingerprint: p.Fingerprint}
if status == seat.StepResultStatusBypass {
s.Outcome, s.Why = seat.StepResultOutcomeBypassed, "bypassed by the module's policy"
@@ -49,7 +49,7 @@ func placeFixture() (catalog.Registration, map[string]catalog.CatalogEntry) {
// The runner's asking half (design 47 G §22, §24, §323; A-91): a pipeline's steps are the ones the policy names, a bypassed
// step asked of no one, an artifact no build seat names ended could not run naming it, and each running step asked of its
// build seat's accepted kind step with the pipeline's commit and fingerprint; a step the policy does not name is no step.
// build seat's accepted kind step with the pipeline's commit and fingerprint; a step the policy leaves without a status is optional (§553).
func TestStartPlacesTheStepsThePolicyNamesAndAsksEachBuildSeat(t *testing.T) {
reg, entries := placeFixture()
store := created{}
@@ -67,16 +67,19 @@ func TestStartPlacesTheStepsThePolicyNamesAndAsksEachBuildSeat(t *testing.T) {
for _, st := range p.Steps {
got = append(got, st.Artifact+" "+st.Step+" "+st.Outcome)
}
want := []string{" build running", "app:program lint running", "app:program unit running", "app:tools lint bypassed", "app:tools unit running",
"app:web lint could not run", "app:web unit could not run", " approval waiting"}
// §553: integration, composition and review are named by no policy, so they are optional steps and run.
want := []string{" build running", "app:program lint running", "app:program unit running", "app:program integration running",
"app:tools lint bypassed", "app:tools unit running", "app:tools integration running",
"app:web lint could not run", "app:web unit could not run", "app:web integration could not run",
" composition running", " review waiting", " approval waiting"}
if strings.Join(got, "|") != strings.Join(want, "|") {
t.Errorf("steps = %q, want %q (integration and composition are named by no policy: no step)", got, want)
t.Errorf("steps = %q, want %q", got, want)
}
if !strings.Contains(p.Steps[5].Why, `"rust"`) {
t.Errorf("the step of an artifact no seat builds says %q, want it to name the language", p.Steps[5].Why)
if !strings.Contains(p.Steps[7].Why, `"rust"`) {
t.Errorf("the step of an artifact no seat builds says %q, want it to name the language", p.Steps[7].Why)
}
if len(asks.asks) != 4 {
t.Fatalf("asked %d steps, want the build and the three running ones", len(asks.asks))
if len(asks.asks) != 7 {
t.Fatalf("asked %d steps, want the build, the five running build steps and the composition", len(asks.asks))
}
// A-99: the build is every pipeline's first step, required whatever the policy names, asked of mesh-builds for the
// module's own entry only.
@@ -92,7 +95,7 @@ func TestStartPlacesTheStepsThePolicyNamesAndAsksEachBuildSeat(t *testing.T) {
t.Errorf("the ask = %+v at %s", a, asks.at[1])
}
// A second start of the revision asks nothing.
if ok, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Entries: entries}); ok || err != nil || len(asks.asks) != 4 {
if ok, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Entries: entries}); ok || err != nil || len(asks.asks) != 7 {
t.Errorf("second start = %v, %v, %d asks", ok, err, len(asks.asks))
}
}
@@ -198,9 +201,9 @@ func TestResumeAsksTheStepsPlacedAndNotEnded(t *testing.T) {
asks := &stepAsks{}
s := &Starter{Steps: asks, Name: "the-runner", Registration: func(string) (catalog.Registration, bool) { return reg, true }}
n, err := s.Resume(store, "main", Half{Entries: entries})
// main.4 has the build and three running steps of the fixture; main.5 one fewer, its build ended.
if err != nil || n != 7 || len(asks.asks) != 7 {
t.Fatalf("resume asked %d (%d put), %v; want 7", n, len(asks.asks), err)
// main.4 has the build, five running build steps and the composition of the fixture; main.5 one fewer, its build ended.
if err != nil || n != 13 || len(asks.asks) != 13 {
t.Fatalf("resume asked %d (%d put), %v; want 13", n, len(asks.asks), err)
}
for _, id := range asks.ids {
found := false
@@ -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)
}
}
@@ -12,14 +12,20 @@ import (
// Status is a step's status for one artifact of a module, read from the module's registration and never from a
// manifest (design 47 G §21, §23): the artifact's own status where its policy names one, else the module's. A step
// neither names is "" and is no step of the pipeline (A-91): Place places none for it.
// whose status neither names is `optional` (§553): it runs and a failure holds nothing; an empty approval is no step (A-196).
func Status(reg catalog.Registration, artifact, step string) string {
if a, ok := reg.Policy.Artifacts[artifact]; ok {
if s := statusOf(a, step); s != "" {
return s
}
}
return statusOf(reg.Policy.Steps, step)
if s := statusOf(reg.Policy.Steps, step); s != "" {
return s
}
if step == seat.StepResultStepApproval {
return ""
}
return seat.StepResultStatusOptional
}
func statusOf(s catalog.StepStatuses, step string) string {
@@ -20,13 +20,20 @@ func TestPolicyAndBuildSeat(t *testing.T) {
{"tools", "unit", "bypass"},
{"tools", "integration", "optional"},
{"tools", "approval", "required"},
{"program", "composition", ""},
// A step whose status the policy leaves empty is optional (design 47 G §553), never no step.
{"program", "composition", "optional"},
{"program", "review", "optional"},
} {
if got := Status(reg, c.artifact, c.step); got != c.want {
t.Errorf("Status(%s, %s) = %q, want %q", c.artifact, c.step, got, c.want)
}
}
// An empty approval status asks the operator nothing (A-196): no step.
if got := Status(catalog.Registration{}, "", "approval"); got != "" {
t.Errorf("an empty approval status = %q, want none", got)
}
entries := map[string]catalog.CatalogEntry{
"mesh-build-go": {Module: "mesh-build-go", Acts: []catalog.ActingRule{{Seat: "node-build-go", Step: "build", On: "go"}}},
"mesh-build-node": {Module: "mesh-build-node", Acts: []catalog.ActingRule{{Seat: "node-build-node", Step: "build", On: "typescript"}, {Seat: "node-build-node", Step: "check", On: "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,30 @@ 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) counting from the time the awaited build ended. All nil: nothing is held.
Self, SelfFp string
Awaiting func() (string, bool)
Hold func(fp string, from time.Time) error
// Ends says when the build of a module at a fingerprint ended (BuildEnds.At): the clock of the wait's bound (A-181, A-206).
Ends func(module, fp string) (time.Time, bool)
// Late says a pipeline the bound's expiry started with the running runner (A-206): the finding pipeline-waits.
Late func(p seat.Pipeline, held seat.Held)
}
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 +109,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 +126,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 +316,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
}
@@ -4,6 +4,7 @@ import (
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
"time"
@@ -207,6 +208,49 @@ func (h *Hearer) said(p seat.Pipeline, asked, ended bool) error {
if err := h.say(seat.EventPipelineEnded, p, nil); err != nil {
return fmt.Errorf("pipeline %s ended but %s is not said: %w", p.Pipeline, seat.EventPipelineEnded, err)
}
return h.groupEnded(p)
}
return nil
}
// groupEnded says a group's one pipeline's end once every member's pipeline of the same watched branch and source revision
// ended (design 47 G §73, §332): on the subject under the group, no module part, with every member's end. A pipeline of no
// group says none; a member still running says none yet.
func (h *Hearer) groupEnded(p seat.Pipeline) error {
if p.Group == "" {
return nil
}
pkeys, err := h.Store.Keys("*.*.*")
if err != nil {
return fmt.Errorf("the members of group %s are not listed: %w", p.Group, err)
}
sort.Strings(pkeys)
var members []seat.PipelineEnded
outcome, why := seat.GroupEndedOutcomePassed, ""
var last *time.Time
for _, key := range pkeys {
m, _, err := h.Store.Get(key)
if err != nil || m.Branch != p.Branch || m.Source != p.Source || m.Group != p.Group {
continue
}
if m.Outcome == seat.PipelineOutcomeRunning {
return nil
}
members = append(members, pipelineEnded(m))
if m.Ended != nil && (last == nil || m.Ended.After(*last)) {
last = m.Ended
}
if m.Outcome != seat.PipelineOutcomePassed && outcome != seat.GroupEndedOutcomeFailed {
outcome, why = m.Outcome, m.Module+": "+m.Why
if m.Outcome == seat.PipelineOutcomeFailed {
outcome = seat.GroupEndedOutcomeFailed
}
}
}
subject := "mesh.seat." + seat.Seat + ".event." + seat.EventGroupEnded + "." + keys.EscapeName(p.Branch) + "." + keys.EscapeName(p.Group)
if err := h.Say(subject, seat.GroupEnded{V: 1, Branch: p.Branch, Group: p.Group, Source: p.Source, Revision: p.Revision, Ended: last,
Outcome: outcome, Why: why, Pipelines: members}); err != nil {
return fmt.Errorf("group %s ended but %s is not said: %w", p.Group, seat.EventGroupEnded, err)
}
return nil
}
@@ -247,8 +291,31 @@ func (h *Hearer) say(event string, p seat.Pipeline, step *seat.StepResult) error
}
}
// The test budget per catalog module (design 47 G §24, §419; the test budget): at most 500 unit and 50 integration tests.
const (
budgetUnit = 500
budgetIntegration = 50
)
// testCounts is the module's test counts, the sum of its test steps' own counts; nil when no step counted any.
func testCounts(steps []seat.StepResult) *seat.TestCounts {
var t seat.TestCounts
counted := false
for _, s := range steps {
if s.Tests != nil {
t.Unit, t.Integration, counted = t.Unit+s.Tests.Unit, t.Integration+s.Tests.Integration, true
}
}
if !counted {
return nil
}
return &t
}
func pipelineEnded(p seat.Pipeline) seat.PipelineEnded {
return seat.PipelineEnded{V: 1, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit, Fingerprint: p.Fingerprint,
tests := testCounts(p.Steps)
return seat.PipelineEnded{Tests: tests, OverBudget: tests != nil && (tests.Unit > budgetUnit || tests.Integration > budgetIntegration),
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, Run: p.Run, Outcome: p.Outcome, Why: p.Why, Started: p.Started,
Ended: p.Ended, EndCount: p.EndCount, Steps: p.Steps}
}
@@ -334,3 +334,68 @@ func (v Verbs) askOf(p seat.Pipeline, s seat.StepResult, caller string, now time
}
return nil, fmt.Errorf("the %s step of %s is placed no ask now", s.Step, p.Pipeline)
}
// Start starts by hand a delivering pipeline for the head of a watched branch the runner's catalog half holds, recorded in
// the hand-act log first (§306, §394): the pipeline of that branch's source revision and the module, as a take-in would
// start it. A head whose pipeline is written already starts nothing and is refused, naming the pipeline.
func (v Verbs) Start(_ string, in seat.StartInput) (seat.StartAnswer, error) {
if in.Branch == "" || in.Module == "" {
return seat.StartAnswer{}, errors.New("start names the watched branch and the module")
}
if v.Starter == nil || v.View == nil {
return seat.StartAnswer{}, errors.New("the runner cannot start a pipeline: it holds no starter or no catalog half")
}
half, ok := v.View(in.Branch)
if !ok {
return seat.StartAnswer{}, fmt.Errorf("the catalog half of %s is not held yet: nothing was started", in.Branch)
}
e, ok := half.Entries[in.Module]
if !ok || half.Source <= 0 {
return seat.StartAnswer{}, fmt.Errorf("the catalog half of %s holds no entry of %s at a source revision: nothing was started", in.Branch, in.Module)
}
key := Key(in.Branch, half.Source, in.Module)
if err := v.record(seat.VerbStart, "started the pipeline "+key+" by hand", in.Why); err != nil {
return seat.StartAnswer{}, err
}
started, err := v.Starter.Start(Begin{Branch: in.Branch, Source: half.Source, Module: in.Module, Commit: e.Commit, Fp: e.Fp,
Repository: e.Repository, Revision: half.Revision, Entries: half.Entries})
if err != nil {
return seat.StartAnswer{}, err
}
if !started {
return seat.StartAnswer{}, fmt.Errorf("the pipeline %s of the head is started already: a step of it is asked again by rerun", key)
}
return seat.StartAnswer{Pipeline: key}, nil
}
// AskAgain asks a declined approval step once more (§306, §36), recorded in the hand-act log first: the step waits again,
// marked asked again by the caller the bus proved, and is asked at once when the pipeline's every other required step
// passed, under an ask of its own (the earlier ask's decision ended it).
func (v Verbs) AskAgain(caller string, in seat.AskAgainInput) (seat.AskAgainAnswer, error) {
p, i, err := v.change(in.Pipeline, seat.EventStepStarted, func() error {
return v.record(seat.VerbAskAgain, "asked the approval of "+in.Pipeline+" again", in.Why)
}, func(p *seat.Pipeline, now time.Time) (int, error) {
switch p.Outcome {
case seat.PipelineOutcomeCancelled, seat.PipelineOutcomeSuperseded:
return -1, fmt.Errorf("the pipeline %s is %s: its approval is not asked again", p.Pipeline, p.Outcome)
}
i, err := stepOf(*p, seat.StepResultStepApproval, "")
if err != nil {
return -1, err
}
s := &p.Steps[i]
if s.Outcome != seat.StepResultOutcomeDeclined {
return -1, fmt.Errorf("only a declined approval is asked again: the approval of %s is %s", p.Pipeline, s.Outcome)
}
*s = seat.StepResult{Step: s.Step, Artifact: s.Artifact, Status: s.Status, Fingerprint: s.Fingerprint,
Outcome: seat.StepResultOutcomeWaiting, By: caller, Rerun: true, Started: &now}
if p.Outcome != seat.PipelineOutcomeRunning {
p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeRunning, "", nil
}
return i, nil
})
if err != nil {
return seat.AskAgainAnswer{}, err
}
return seat.AskAgainAnswer{Result: p.Steps[i]}, nil
}
@@ -170,3 +170,67 @@ func TestAVerbWhoseHandActIsNotRecordedDoesNothing(t *testing.T) {
type failingActs struct{}
func (failingActs) RecordHandAct(views.HandAct) error { return errors.New("no holder") }
// RN14 (§306, §394): start starts by hand the pipeline of a watched branch's head, recorded in the hand-act log, and a head
// already started is refused; ask-again asks a declined approval once more under an ask of its own, recorded, and an
// approval not declined is refused.
func TestStartAndAskAgain(t *testing.T) {
v, store, _, _, acts, _ := verbsFixture()
made := created{}
v.Starter.Pipelines, v.Starter.Say = made, func(string, any) error { return nil }
_, entries := placeFixture()
v.View = func(string) (Half, bool) { return Half{Source: 7, Revision: 3, Entries: entries}, true }
a, err := v.Start("operator", seat.StartInput{Branch: "main", Module: "app", Why: "no pipeline for the head"})
if err != nil || a.Pipeline != "main.7.app" || made["main.7.app"] == nil {
t.Fatalf("start = %+v, %v; written %v", a, err, len(made))
}
if _, err := v.Start("operator", seat.StartInput{Branch: "main", Module: "app", Why: "again"}); err == nil {
t.Error("a second start of the same head was not refused")
}
if _, err := v.Start("operator", seat.StartInput{Branch: "main", Module: "absent", Why: "x"}); err == nil {
t.Error("a start of a module the half does not hold was not refused")
}
if len(*acts) != 2 || !strings.HasPrefix((*acts)[0], "mesh-pipeline-runner.start: ") {
t.Errorf("hand acts %v, want the two starts recorded", *acts)
}
// ask-again: the approval of the fixture's pipeline is declined; asked again it is asked under a new ask.
h := v.Hearer
asker := h.Asker.(*askRecorder)
e := store[pkey]
first, err := ApprovalAsk(e.p, time.Unix(1, 0))
if err != nil {
t.Fatal(err)
}
for i := range e.p.Steps {
s := &e.p.Steps[i]
switch s.Step {
case seat.StepResultStepApproval:
s.Outcome, s.Ask = seat.StepResultOutcomeDeclined, first.ID
default:
s.Outcome = seat.StepResultOutcomePassed
}
}
e.p.Outcome = seat.PipelineOutcomeFailed
store[pkey] = e
if _, err := v.AskAgain("operator", seat.AskAgainInput{Pipeline: pkey}); err == nil {
t.Error("an ask-again saying no why was taken")
}
got, err := v.AskAgain("operator", seat.AskAgainInput{Pipeline: pkey, Why: "asked by mistake"})
if err != nil || got.Result.Outcome != seat.StepResultOutcomeRunning || got.Result.Ask == "" || got.Result.Ask == first.ID || !got.Result.Rerun {
t.Fatalf("ask-again = %+v, %v", got, err)
}
if len(asker.asked) != 1 || asker.asked[0].ID != got.Result.Ask {
t.Errorf("asked %v, want one ask %s", asker.asked, got.Result.Ask)
}
if store[pkey].p.Outcome != seat.PipelineOutcomeRunning {
t.Errorf("the pipeline is %s after ask-again, want running", store[pkey].p.Outcome)
}
// The ask rebuilt from the step, as a warrant is held to it, is the ask asked.
if again, err := ApprovalAsk(store[pkey].p, *stepNamed(store[pkey].p, "", seat.StepResultStepApproval).Started); err != nil || again.ID != got.Result.Ask {
t.Errorf("the ask rebuilt = %s, %v; want %s", again.ID, err, got.Result.Ask)
}
if _, err := v.AskAgain("operator", seat.AskAgainInput{Pipeline: pkey, Why: "twice"}); err == nil {
t.Error("an approval running was asked again")
}
}
@@ -22,15 +22,19 @@ func TestManifest(t *testing.T) {
}
var states, streams, objects []string
keyTTL := map[string]bool{}
stateMiB := map[string]int{}
for _, s := range m.State {
states = append(states, s.Name)
keyTTL[s.Name] = s.KeyTTL
stateMiB[s.Name] = s.MaxMiB
}
objectMiB := map[string]int{}
for _, s := range m.Streams {
streams = append(streams, s.Name)
}
for _, o := range m.Objects {
objects = append(objects, o.Name)
objectMiB[o.Name] = o.MaxMiB
}
var program catalogue.Artifact
if m.Build == nil {
@@ -50,9 +54,12 @@ func TestManifest(t *testing.T) {
{"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat &&
equalSets(m.Claims[0].Serves, seat.Verbs())},
// §27, §85, §330, §355, §77: the old state sections.
{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views"})},
{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views", "consumers"})},
// §535, §647, A-77, A-104: an ask no holder takes is bound by a key with its own time to live.
{"bounds is key-ttl", keyTTL["bounds"]},
// RN25 (§469, §85): an old state section declares its cap, pipelines 256 MiB and kept 128 MiB.
{"pipelines 256 MiB", stateMiB[seat.StatePipelines] == 256},
{"kept 128 MiB", stateMiB[seat.StateKept] == 128},
// §470 (amending §27, §85): the step log is an old logs section kept 100 days, 128 MiB; RN24 (§383, §387): the finding
// stream its findings are said on is the one old stream.
{"old streams section", equalSets(streams, []string{"mesh-delivery-pipeline-findings"})},
@@ -62,13 +69,15 @@ func TestManifest(t *testing.T) {
equalSets(m.Streams[0].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"})},
// RN24 (§368): the version as MAJOR.MINOR, 0.1 with none published.
{"version 0.1", m.Version == "0.1"},
{"old objects section", equalSets(objects, []string{seat.StateStepOutputs})},
{"step-outputs 1 GiB", len(m.Objects) == 1 && m.Objects[0].MaxMiB == 1024},
// RN25 (§477, §533, §59): a foundation part whose seat has verbs and events declares its sent store, 256 MiB.
{"old objects section", equalSets(objects, []string{seat.StateStepOutputs, catalogue.SentObjectsLocal})},
{"step-outputs 1 GiB", objectMiB[seat.StateStepOutputs] == 1024},
{"sent-objects 256 MiB", objectMiB[catalogue.SentObjectsLocal] == 256},
// §333, §54, §44, §53, §376, §24, §37, §4, §77: what the runner hears.
{"consumes", equalSets(m.Consumes, []string{
"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended",
"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", "mesh-delivery-builder.build.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).
@@ -81,7 +90,10 @@ func TestManifest(t *testing.T) {
"seat:mesh-builds.step", "seat:mesh-wants.step", "seat:node-build-go.check.step", "seat:node-build-node.check.step",
"seat:node-build-image.check.step"})},
{"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})},
{"first-view tools", equalSets(m.Tools, []string{"pipeline_first_views", "pipeline_first_view_again"})},
// RN25 (§539): the runner's tools carry the prefix runner_.
{"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health", "runner_waits", "runner_release"})},
// RN25 (§468, §76): the health check, run outside the process, is the tool runner_health.
{"health is runner_health", healthTool(raw) == "tool runner_health"},
// RN1/AC5 (§95, §130): the new assigner is delivered before the runner.
{"program needs the assigner's program", slices.Equal(program.Needs, []string{"mesh-delivery-assigner:program"})},
}
@@ -117,6 +129,28 @@ func TestNoVerbNamesItsCallers(t *testing.T) {
}
}
// healthTool is the process resource's health kind and tool.
func healthTool(raw []byte) string {
var m struct {
Resources []struct {
Type string `json:"type"`
Health struct {
Kind string `json:"kind"`
Tool string `json:"tool"`
} `json:"health"`
} `json:"resources"`
}
if err := json.Unmarshal(raw, &m); err != nil {
return err.Error()
}
for _, r := range m.Resources {
if r.Type == "process" {
return r.Health.Kind + " " + r.Health.Tool
}
}
return ""
}
func equalSets(a, b []string) bool {
x, y := slices.Clone(a), slices.Clone(b)
slices.Sort(x)
+23 -6
View File
@@ -18,19 +18,24 @@
"rerun",
"accept",
"cancel",
"ask-again"
"ask-again",
"release"
]
}
],
"tools": [
"pipeline_first_views",
"pipeline_first_view_again"
"runner_first_views",
"runner_first_view_again",
"runner_health",
"runner_waits",
"runner_release"
],
"consumes": [
"mesh-delivery-indexer.take-in.taken",
"mesh-builds.package-published",
"mesh-builds.toolchain-agreed",
"mesh-builds.step-ended",
"mesh-delivery-builder.build.ended.**",
"operator-channel.decided",
"node-build-go.step-ended",
"node-build-node.step-ended",
@@ -38,6 +43,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",
@@ -61,11 +67,13 @@
"state": [
{
"name": "pipelines",
"history": 1
"history": 1,
"max-mib": 256
},
{
"name": "kept",
"history": 1
"history": 1,
"max-mib": 128
},
{
"name": "aged",
@@ -87,6 +95,10 @@
{
"name": "first-views",
"history": 1
},
{
"name": "consumers",
"history": 1
}
],
"streams": [
@@ -105,6 +117,10 @@
{
"name": "step-outputs",
"max-mib": 1024
},
{
"name": "sent-objects",
"max-mib": 256
}
],
"own-secrets": {
@@ -140,7 +156,8 @@
},
"user": "mesh-delivery-pipeline",
"health": {
"kind": "unit"
"kind": "tool",
"tool": "runner_health"
}
}
],