Merge commit '5dec3dff' into feat/47-G-fix11-b2f

This commit is contained in:
2026-11-12 06:16:03 +01:00
33 changed files with 1482 additions and 48 deletions
@@ -0,0 +1,36 @@
package meshcatalogclient
import (
"context"
"encoding/json"
"fmt"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/views"
)
// WireCandidatesProcess hands each candidate's `catalog-revision` to hook (design 47 G A-185, A-193, A-199): a candidate's
// revision no Catalog keeps, heard once on the event's one subscription of the process's reader, beside every watched
// branch's Catalog and whether or not the part follows any, so the runner places a run's pipelines on it. A trunk revision
// is passed over (each branch's Catalog keeps it); data that is no such event is an error. The part's manifest consumes the
// seat's event catalog-revision.
func WireCandidatesProcess(r *views.ProcessReader, hook func(ctx context.Context, ev seat.CatalogRevisionEvent) error) error {
if hook == nil {
return fmt.Errorf("meshcatalogclient: a candidates hook is a function")
}
return r.AdoptEvent(seat.Seat, seat.EventCatalogRevision, candidateHear(hook))
}
// candidateHear is the handler of catalog-revision that hands a candidate's revision to hook.
func candidateHear(hook func(ctx context.Context, ev seat.CatalogRevisionEvent) error) func(context.Context, []byte) error {
return func(ctx context.Context, data []byte) error {
var ev seat.CatalogRevisionEvent
if err := json.Unmarshal(data, &ev); err != nil || ev.Revision <= 0 {
return fmt.Errorf("meshcatalogclient: not a %s event: %s", seat.EventCatalogRevision, clipBytes(data))
}
if ev.Kind != KindCandidate {
return nil
}
return hook(ctx, ev)
}
}
@@ -162,3 +162,29 @@ func TestCatalogIsAViewPerWatchedBranch(t *testing.T) {
t.Error("a Catalog of no branch was made")
}
}
// A-185, A-193, A-199: a candidate's catalog-revision reaches the hook, adopted on the event's one subscription of the
// reader, also with no watched branch's Catalog; a trunk revision does not, and data that is no such event is said.
func TestCandidatesReachTheHook(t *testing.T) {
var hooked []seat.CatalogRevisionEvent
hook := func(_ context.Context, ev seat.CatalogRevisionEvent) error { hooked = append(hooked, ev); return nil }
r := &views.ProcessReader{Module: "runner"}
if err := WireCandidatesProcess(r, hook); err != nil {
t.Fatal(err)
}
if err := WireCandidatesProcess(r, nil); err == nil {
t.Error("a nil hook was taken")
}
hear := candidateHear(hook)
for _, data := range [][]byte{revisionEvent("main", KindTrunk, 3), revisionEvent("main", KindCandidate, 4)} {
if err := hear(context.Background(), data); err != nil {
t.Fatal(err)
}
}
if err := hear(context.Background(), []byte(`not json`)); err == nil {
t.Error("data that is no catalog-revision was taken")
}
if len(hooked) != 1 || hooked[0].Revision != 4 || hooked[0].Kind != KindCandidate {
t.Errorf("the hook was handed %+v, want the candidate's revision 4 alone", hooked)
}
}
@@ -432,8 +432,10 @@ type Pipeline struct {
// yes when the module is over its test budget (design 47 G section 419)
OverBudget bool `json:"over-budget,omitempty"`
// the pipeline's key, <branch>.<revision>.<module>, a branch escaped as a key's name is
Pipeline string `json:"pipeline,omitempty"`
Repository string `json:"repository,omitempty"`
Pipeline string `json:"pipeline,omitempty"`
// for a pipeline a package's publish started, that publish (A-202)
Published *Published `json:"published,omitempty"`
Repository string `json:"repository,omitempty"`
// the catalog graph revision of the branch the pipeline runs for
Revision int64 `json:"revision,omitempty"`
// the run on request's id, for a pipeline of a run; such a pipeline delivers nothing
@@ -545,6 +547,8 @@ type PipelineStarted struct {
Module string `json:"module"`
// the pipeline's key, <branch>.<revision>.<module>, a branch escaped as a key's name is
Pipeline string `json:"pipeline"`
// for a pipeline a package's publish started, that publish (A-202)
Published *Published `json:"published,omitempty"`
// the catalog graph revision of the branch the pipeline runs for
Revision int64 `json:"revision,omitempty"`
// the run on request's id, for a pipeline of a run; such a pipeline delivers nothing
@@ -556,6 +560,14 @@ type PipelineStarted struct {
V int64 `json:"v"`
}
// Published is the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202).
type Published struct {
// the package whose publish started the pipeline
Artifact string `json:"artifact"`
// the version that publish named
Version string `json:"version"`
}
// RunStarted is a run on request's start, with its members at their commits (ADR 0666 section 18).
type RunStarted struct {
At *time.Time `json:"at,omitempty"`
@@ -907,6 +907,24 @@
"name"
]
},
"Published": {
"type": "object",
"properties": {
"artifact": {
"type": "string",
"description": "the package whose publish started the pipeline"
},
"version": {
"type": "string",
"description": "the version that publish named"
}
},
"description": "the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202)",
"required": [
"artifact",
"version"
]
},
"Pipeline": {
"type": "object",
"properties": {
@@ -1002,6 +1020,10 @@
"version": {
"type": "string",
"description": "the version package-published names, for a pipeline the runner started from a publication or from outdated; none for any other (design 47 G 550)"
},
"published": {
"$ref": "#/defs/Published",
"description": "for a pipeline a package's publish started, that publish (A-202)"
}
},
"description": "one pipeline as the runner keeps it in its state pipelines, by compare-and-set (design 47 *The runner's kept data*)",
@@ -1191,6 +1213,10 @@
"at": {
"type": "string",
"format": "date-time"
},
"published": {
"$ref": "#/defs/Published",
"description": "for a pipeline a package's publish started, that publish (A-202)"
}
},
"description": "a pipeline's start",
@@ -195,6 +195,8 @@ export interface Pipeline {
"over-budget"?: boolean;
/** the pipeline's key, <branch>.<revision>.<module>, a branch escaped as a key's name is */
"pipeline"?: string;
/** for a pipeline a package's publish started, that publish (A-202) */
"published"?: Published;
"repository"?: string;
/** the catalog graph revision of the branch the pipeline runs for */
"revision"?: number;
@@ -268,6 +270,8 @@ export interface PipelineStarted {
"module": string;
/** the pipeline's key, <branch>.<revision>.<module>, a branch escaped as a key's name is */
"pipeline": string;
/** for a pipeline a package's publish started, that publish (A-202) */
"published"?: Published;
/** the catalog graph revision of the branch the pipeline runs for */
"revision"?: number;
/** the run on request's id, for a pipeline of a run; such a pipeline delivers nothing */
@@ -279,6 +283,14 @@ export interface PipelineStarted {
"v": number;
}
/** the publish of a package that started a pipeline: a newer publish of the same artifact on the same branch supersedes it (design 47 G section 31, A-202) */
export interface Published {
/** the package whose publish started the pipeline */
"artifact": string;
/** the version that publish named */
"version": string;
}
/** a run on request's start, with its members at their commits (ADR 0666 section 18) */
export interface RunStarted {
"at"?: string;
@@ -86,21 +86,32 @@ 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 }}
}
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, 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)
@@ -17,7 +17,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 +26,9 @@ 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)
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"
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,43 @@
package main
import (
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"mesh-delivery-pipeline/internal/pipeline"
)
// readCandidate reads a candidate's revision whole through the catalog seat's verb revision, with the candidate's group and
// heads and the revision its event names, page by page (design 47 G A-180).
func readCandidate(call func(key string, body any) (json.RawMessage, error), ev catalog.CatalogRevisionEvent) (pipeline.Half, error) {
half := pipeline.Half{Revision: ev.Revision, Source: ev.Source, Entries: map[string]catalog.CatalogEntry{}}
after := ""
for {
raw, err := call(meshcatalogclient.RevisionVerb, catalog.RevisionInput{Branch: ev.Branch, Group: ev.Group, Heads: ev.Heads, Revision: ev.Revision, After: after})
if err != nil {
return half, err
}
answer, said, ok := buscalls.Unwrap(raw)
if !ok {
return half, fmt.Errorf("%s answered: %s", meshcatalogclient.RevisionVerb, said)
}
var a catalog.RevisionAnswer
if err := json.Unmarshal(answer, &a); err != nil {
return half, fmt.Errorf("%s answered no revision: %w", meshcatalogclient.RevisionVerb, err)
}
if a.Revision == nil || a.Revision.Revision != ev.Revision || a.Revision.Group != ev.Group {
return half, fmt.Errorf("%s answered %+v for the candidate %s's revision %d", meshcatalogclient.RevisionVerb, a.Revision, ev.Group, ev.Revision)
}
for _, e := range a.Entries {
half.Entries[e.Module] = e
}
if a.Next == "" {
return half, nil
}
after = a.Next
}
}
@@ -0,0 +1,36 @@
package main
import (
"encoding/json"
"testing"
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
)
// A-180: a run's candidate revision is read whole through the seat's revision verb with the candidate's group, heads and
// revision, page by page; an answer of another revision is refused.
func TestReadCandidateReadsTheRevisionWhole(t *testing.T) {
ev := catalog.CatalogRevisionEvent{Kind: "candidate", Group: "run-0a1b2c3d4e5f", Heads: "h", Revision: 12, Branch: "main"}
var asked []catalog.RevisionInput
answerRev := int64(12)
call := func(key string, body any) (json.RawMessage, error) {
in := body.(catalog.RevisionInput)
asked = append(asked, in)
a := catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: answerRev, Kind: "candidate", Group: in.Group}}
if in.After == "" {
a.Entries, a.Next = []catalog.CatalogEntry{{Module: "app", Fp: "src2:a"}}, "app"
} else {
a.Entries = []catalog.CatalogEntry{{Module: "lib", Fp: "src2:l"}}
}
return json.Marshal(a)
}
half, err := readCandidate(call, ev)
if err != nil || len(half.Entries) != 2 || half.Revision != 12 || len(asked) != 2 || asked[1].After != "app" ||
asked[0].Group != ev.Group || asked[0].Heads != "h" || asked[0].Revision != 12 || asked[0].Branch != "main" {
t.Fatalf("read %+v (%v), asked %+v", half, err, asked)
}
answerRev = 13
if _, err := readCandidate(call, ev); err == nil {
t.Error("an answer of another revision was taken")
}
}
@@ -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)
}
}
@@ -0,0 +1,41 @@
package main
import (
"encoding/json"
"fmt"
nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"mesh-delivery-pipeline/internal/pipeline"
)
// keptCaller is the runner's kept calls, to any holder of a seat or to one node's (process.Conn).
type keptCaller interface {
CallKept(key string, body any) (json.RawMessage, error)
CallKeptOn(node, key string, body any) (json.RawMessage, error)
}
// dropAsk asks the build seat's `drop` of a superseded step's ask as a kept call (design 47 G §31, §350; A-182): of the
// holder on the node that took it, or, never taken, of any holder, which records it in the seat's shared drop key. The
// three build seats' drop take one input (nodebuildgo.DropInput; the manifest invokes each).
func dropAsk(calls keptCaller, by string) pipeline.Drop {
return func(buildSeat, node, ask, why string) error {
key := "seat:" + buildSeat + "." + nodebuildgo.VerbDrop
in := nodebuildgo.DropInput{ID: ask, Why: why, By: by}
var raw json.RawMessage
var err error
if node != "" {
raw, err = calls.CallKeptOn(node, key, in)
} else {
raw, err = calls.CallKept(key, in)
}
if err != nil {
return err
}
if _, said, ok := buscalls.Unwrap(raw); !ok {
return fmt.Errorf("%s answered: %s", key, said)
}
return nil
}
}
@@ -11,11 +11,13 @@ import (
"log"
"os"
"os/signal"
"strings"
"sync/atomic"
"syscall"
"github.com/nats-io/nats.go"
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"
"git.novox.be/novox/mesh-sdk/go/process"
@@ -71,13 +73,26 @@ 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)]
var placer atomic.Pointer[func(catalog.CatalogRevisionEvent) error]
rv, err := startViews(ctx, js, conn, cred, hub, calls.State("findings"), calls.Call, func(branch string) {
if f := resumer.Load(); f != nil {
(*f)(branch)
}
}, func(_ context.Context, ev catalog.CatalogRevisionEvent) error {
if f := placer.Load(); f != nil {
return (*f)(ev)
}
return fmt.Errorf("the candidate %s's revision %d is heard before the runner can place it", ev.Group, ev.Revision)
}, logger.Printf)
if err != nil {
return err
@@ -112,25 +127,50 @@ 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 {
starter.Store = pipelineStore{ctx: ctx, kv: pipelines}
// A run's candidate revision places its pipelines' steps, its entries read through the seat's revision verb (A-180).
place := func(ev catalog.CatalogRevisionEvent) error {
if !strings.HasPrefix(ev.Group, pipeline.RunGroupPrefix) {
return nil
}
half, err := readCandidate(calls.CallKept, ev)
if err != nil {
return fmt.Errorf("the candidate %s's revision %d is not read, so its run is not placed: %w", ev.Group, ev.Revision, err)
}
n, err := starter.PlaceRun(ev, half, rv.Catalog.View)
logger.Printf("the candidate %s's revision %d placed %d pipelines of its run", ev.Group, ev.Revision, n)
return err
}
if err := hearToolchain(ctx, hub, toolchains, logger.Printf); err != nil {
placer.Store(&place)
// A newer commit or publish supersedes the older running pipelines it replaces, their asks dropped as kept calls (§31,
// §350; A-182, A-202).
starter.Supersede = func(newer seat.Pipeline) {
n, err := hearer.Supersede(newer, dropAsk(calls, cred.Module), dropBound(ctx, bounds))
if err != nil {
logger.Printf("the pipelines %s supersedes: %d ended, then %v", newer.Pipeline, n, err)
} else if n > 0 {
logger.Printf("the pipeline %s superseded %d older pipelines", newer.Pipeline, n)
}
}
if err := hearStepEnded(heard, hearer, logger.Printf); err != nil {
return err
}
if err := hearAskTaken(ctx, hub, bounds, logger.Printf); err != nil {
if err := hearToolchain(ctx, heard, toolchains, logger.Printf); err != nil {
return err
}
if err := hearAskTaken(ctx, heard, bounds, hearer.Taken, logger.Printf); err != nil {
return err
}
if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil {
return err
}
if err := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil {
if err := hearDecided(heard, hearer, cred.Module, logger.Printf); err != nil {
return err
}
if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil {
if err := hearPublished(ctx, heard, starter, calls, rv.Catalog, logger.Printf); err != nil {
return err
}
if err := hearTakenIn(hub, starter, rv.Catalog, logger.Printf); err != nil {
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)
}
@@ -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 {
@@ -99,7 +99,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) {
@@ -135,9 +135,9 @@ func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string,
return nil
}
// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647): the ask is answered, so its five minutes
// end with no finding.
func hearAskTaken(ctx context.Context, hub *process.EventHub, bounds jetstream.KeyValue, logf func(string, ...any)) error {
// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647) and keeps on its step the node that took
// it (A-182): the ask is answered, so its five minutes end with no finding.
func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, taken func(ask, node string) error, logf func(string, ...any)) error {
for _, ts := range pipeline.TakenSeats {
ts := ts
err := hub.Handle(ts.Subject, func(m jetstream.Msg) {
@@ -152,6 +152,12 @@ func hearAskTaken(ctx context.Context, hub *process.EventHub, bounds jetstream.K
_ = m.Nak()
return
}
// The step keeps the node that took its ask, so a drop of it is asked of that node's holder (A-182).
if err := taken(ev.Ask, ev.Node); err != nil {
logf("the node %s that took ask %s is not kept on its step, heard again: %v", ev.Node, ev.Ask, err)
_ = m.Nak()
return
}
_ = m.Ack()
})
if err != nil {
@@ -252,7 +258,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: run, review, rerun, accept, cancel, start and ask-again
// (§64, §306; RN12, RN14).
func writeHandlers(v pipeline.Verbs) map[string]process.SeatHandler {
return map[string]process.SeatHandler{
seat.VerbRun: verbOf(func(c string, in seat.RunInput) (any, error) { return v.Run(c, in) }),
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) }),
}
}
// 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.VerbReview, seat.VerbStart, seat.VerbRun}
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
@@ -25,7 +25,8 @@ type runnerViews struct {
// startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or
// after, then start hub once.
func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cred process.Account, hub *process.EventHub, findingsState process.State,
call meshcatalogclient.Call, onReady func(branch string), logf func(string, ...any)) (*runnerViews, error) {
call meshcatalogclient.Call, onReady func(branch string), candidate func(ctx context.Context, ev catalog.CatalogRevisionEvent) error,
logf func(string, ...any)) (*runnerViews, error) {
findings, err := findingWriter(ctx, cred, conn, js, findingsState, logf)
if err != nil {
return nil, err
@@ -52,6 +53,11 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr
if err != nil {
return nil, err
}
// A candidate's revision, heard once on the event's one subscription whatever branches are followed, places a run's
// pipelines (A-185, A-193, A-199).
if err := meshcatalogclient.WireCandidatesProcess(reader, candidate); err != nil {
return nil, err
}
reader.Start(ctx, js, cred, status)
return &runnerViews{Registrations: regs, Catalog: cv}, nil
}
@@ -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,239 @@
package pipeline
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"maps"
"slices"
"strings"
"time"
runnerclient "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient"
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"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// RunGroupPrefix is a run on request's candidate group before its id, `run-<id>` (design 47 G A-160).
const RunGroupPrefix = "run-"
// RunKey is the key of a run on request's pipeline, `<branch>.<id>.<module>`: the run's id under the member's named branch
// (design 47 G §420, A-193).
func RunKey(branch, run, module string) string {
return seat.PipelinesByBranchRevisionModuleKey(keys.EscapeName(branch), run, module)
}
// NewRunID is a run's id, a random 12-hex id, so two runs of one commit are two pipelines (A-180).
func NewRunID() (string, error) {
b := make([]byte, 6)
if _, err := rand.Read(b); err != nil {
return "", err
}
return hex.EncodeToString(b), nil
}
// Run starts a run on request (design 47 G §64, §420; A-180, A-193): it writes each member's pipeline at once, running and
// with no steps, its `run` the run's new id, and says run-started, from which the indexer builds the run's candidate; the
// candidate's revision places the steps (PlaceRun). A member naming no module, commit or branch is refused, the run then
// writing nothing. caller is the one the bus proved. newID makes the run's id (NewRunID).
func (s *Starter) Run(caller string, in seat.RunInput, newID func() (string, error)) (seat.RunAnswer, error) {
if len(in.Members) == 0 {
return seat.RunAnswer{}, errors.New("a run names one or more members, each a module at a commit on a branch")
}
for _, m := range in.Members {
if m.Module == "" || m.Commit == "" || m.Branch == "" {
return seat.RunAnswer{}, fmt.Errorf("a run's member names its module, commit and branch: %+v", m)
}
}
id, err := newID()
if err != nil {
return seat.RunAnswer{}, fmt.Errorf("the run's id is not made: %w", err)
}
now := s.now()
var written []string
for _, m := range in.Members {
key := RunKey(m.Branch, id, m.Module)
p := seat.Pipeline{V: 1, Pipeline: key, Branch: m.Branch, Module: m.Module, Commit: m.Commit, Run: id,
Outcome: seat.PipelineOutcomeRunning, Started: &now, Steps: []seat.StepResult{}}
raw, err := json.Marshal(p)
if err == nil {
err = s.Pipelines.Create(key, raw)
}
if err != nil {
s.failRun(written, "the run's other pipelines were not written")
return seat.RunAnswer{}, fmt.Errorf("the pipeline %s of the run is not written: %w", key, err)
}
written = append(written, key)
}
first := in.Members[0]
last := first.Commit
if len(in.Members) > 1 {
last = RunGroupPrefix + id
}
ev := seat.RunStarted{V: 1, At: &now, By: caller, Members: in.Members, Run: id}
if err := s.Say(runnerclient.Subject(seat.EventRunStarted, first.Module, first.Branch, last), ev); err != nil {
s.failRun(written, "run-started was not said, so no candidate is built for it")
return seat.RunAnswer{}, fmt.Errorf("the run %s is not started: run-started is not said: %w", id, err)
}
return seat.RunAnswer{Run: id, Pipelines: written}, nil
}
// failRun ends the run's pipelines written before its start failed, failed and saying why, so none stands running.
func (s *Starter) failRun(written []string, why string) {
if s.Store == nil {
return
}
for _, key := range written {
for {
p, rev, err := s.Store.Get(key)
if err != nil || p.Outcome != seat.PipelineOutcomeRunning {
break
}
now := s.now()
p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeFailed, why, &now
p.EndCount++
if _, err := s.Store.Update(key, p, rev); errors.Is(err, ErrChanged) {
continue
} else if err != nil {
s.logf("the pipeline %s of a run not started stays running: %v", key, err)
}
break
}
}
}
// PlaceRun places a run's pipelines from its candidate's revision (design 47 G §64, §373, A-180, A-193): ev is the
// catalog-revision of the candidate `run-<id>`, half its entries read whole through the seat's revision verb. Each member's
// pipeline of the run not yet placed takes its fingerprint, the candidate's revision and its steps, by compare-and-set, and
// says pipeline-started; each further entry whose fingerprint the members change (another than in base, the branch's catalog
// half, or absent there) gets a pipeline of the run of its own, placed the same way. Another candidate's revision places
// nothing; a revision heard again places nothing twice. It says how many pipelines it placed.
func (s *Starter) PlaceRun(ev catalog.CatalogRevisionEvent, half Half, base func(branch string) (Half, bool)) (int, error) {
id, ok := strings.CutPrefix(ev.Group, RunGroupPrefix)
if !ok || id == "" {
return 0, nil
}
if s.Store == nil {
return 0, errors.New("the runner holds no pipelines to place a run's in")
}
pkeys, err := s.Store.Keys("*." + id + ".*")
if err != nil {
return 0, fmt.Errorf("the pipelines of the run %s are not listed: %w", id, err)
}
slices.Sort(pkeys)
members := map[string]bool{}
branch := ""
for _, key := range pkeys {
if p, _, err := s.Store.Get(key); err == nil && p.Run == id {
members[p.Module] = true
if branch == "" {
branch = p.Branch
}
}
}
if branch == "" {
s.logf("the candidate %s's revision %d names a run with no pipeline here: nothing is placed", ev.Group, ev.Revision)
return 0, nil
}
placed := 0
var first error
keep := func(ok bool, err error) {
if err != nil && first == nil {
first = err
}
if ok {
placed++
}
}
for _, key := range pkeys {
keep(s.placeRunPipeline(key, id, ev.Revision, half))
}
// The dependents (§373): the candidate's entries the members change, beside the members, on the first member's branch.
trunk, ok := base(branch)
if b, err := keys.UnescapeName(ev.Branch); err == nil && b != "" && b != branch {
if h, held := base(b); held {
trunk, ok = h, true
}
}
if !ok {
s.logf("the run %s's dependents are not known: the catalog half of %s is not held, so its members run alone", id, branch)
return placed, first
}
for _, module := range slices.Sorted(maps.Keys(half.Entries)) {
e := half.Entries[module]
if members[module] {
continue
}
if was, held := trunk.Entries[module]; held && was.Fp == e.Fp {
continue
}
key := RunKey(branch, id, module)
now := s.now()
p := seat.Pipeline{V: 1, Pipeline: key, Branch: branch, Module: module, Commit: e.Commit, Run: id,
Outcome: seat.PipelineOutcomeRunning, Started: &now, Steps: []seat.StepResult{}}
raw, err := json.Marshal(p)
if err == nil {
err = s.Pipelines.Create(key, raw)
}
if err != nil && !errors.Is(err, ErrExists) {
keep(false, fmt.Errorf("the dependent %s of the run %s is not written: %w", module, id, err))
continue
}
keep(s.placeRunPipeline(key, id, ev.Revision, half))
}
return placed, first
}
// placeRunPipeline places one pipeline of the run, written and not yet placed (no revision), from the candidate's half.
func (s *Starter) placeRunPipeline(key, id string, revision int64, half Half) (bool, error) {
for {
p, rev, err := s.Store.Get(key)
if err != nil {
return false, fmt.Errorf("the pipeline %s is not read: %w", key, err)
}
if p.Run != id || p.Outcome != seat.PipelineOutcomeRunning || p.Revision != 0 {
return false, nil
}
now := s.now()
e, ok := half.Entries[p.Module]
if !ok {
s.logf("the candidate of the run %s holds no entry of %s: its pipeline is placed with no fingerprint", id, p.Module)
}
p.Fingerprint, p.Repository, p.Revision = e.Fp, e.Repository, revision
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)
} else {
s.logf("%s has no registration yet: its pipeline is placed with no step", key)
}
}
if _, err := s.Store.Update(key, p, rev); errors.Is(err, ErrChanged) {
continue
} else if err != nil {
return false, fmt.Errorf("the pipeline %s cannot be written: %w", key, err)
}
ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: p.Branch, Module: p.Module, Commit: p.Commit,
Fingerprint: p.Fingerprint, Revision: p.Revision, Run: id}
if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, p.Module, p.Branch, p.Commit), 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 {
return true, err
}
}
return true, nil
}
}
func (s *Starter) now() time.Time {
if s.Now != nil {
return s.Now().UTC()
}
return time.Now().UTC()
}
@@ -0,0 +1,124 @@
package pipeline
import (
"encoding/json"
"errors"
"strings"
"testing"
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
)
// memCreated is memStore written by Create too, as the state pipelines is.
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
}
// RN12 (design 47 G §64, §373, §420; A-180, A-193): `run` writes each member's pipeline at once under the run's id and its
// named branch and says run-started; the candidate run-<id>'s revision places each member's steps and a pipeline of each
// entry the members change, says pipeline-started once each, and places nothing twice; a member naming no branch is refused.
func TestRunOnRequest(t *testing.T) {
reg, entries := placeFixture()
store := memCreated{memStore{}}
var said []string
asks := &stepAsks{}
s := &Starter{Pipelines: store, Store: store, Steps: asks, Name: "the-runner",
Registration: func(m string) (catalog.Registration, bool) {
r := reg
r.Module = m
return r, m == "app" || m == "lib"
},
Say: func(subject string, body any) error {
said = append(said, subject[strings.Index(subject, ".event.")+7:])
return nil
}}
id := func() (string, error) { return "0a1b2c3d4e5f", nil }
if _, err := s.Run("sessions", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c1"}}}, id); err == nil {
t.Error("a member naming no branch was taken")
}
if len(store.memStore) != 0 || len(said) != 0 {
t.Fatalf("a refused run wrote %v and said %v", store.memStore, said)
}
a, err := s.Run("sessions", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c1", Branch: "fix/1"}}}, id)
if err != nil || a.Run != "0a1b2c3d4e5f" || len(a.Pipelines) != 1 {
t.Fatalf("run = %+v, %v", a, err)
}
key := a.Pipelines[0]
if key != RunKey("fix/1", "0a1b2c3d4e5f", "app") {
t.Errorf("the member's pipeline is %s, want the run's id under its named branch", key)
}
if p, _, _ := store.Get(key); p.Run != a.Run || p.Outcome != seat.PipelineOutcomeRunning || len(p.Steps) != 0 || p.Branch != "fix/1" || p.Source != 0 {
t.Errorf("the member's pipeline %s = %+v", key, p)
}
if len(said) != 1 || !strings.HasPrefix(said[0], seat.EventRunStarted+".app.") || !strings.HasSuffix(said[0], ".c1") {
t.Errorf("said %v, want run-started on the member's module, branch and commit", said)
}
cand := entries
cand["app"] = catalog.CatalogEntry{Module: "app", Commit: "c1", Fp: "src2:new", Repository: "o/app", Artifacts: entries["app"].Artifacts}
cand["lib"] = catalog.CatalogEntry{Module: "lib", Commit: "c0", Fp: "src2:lib2", Artifacts: entries["app"].Artifacts}
cand["same"] = catalog.CatalogEntry{Module: "same", Fp: "src2:same"}
base := func(string) (Half, bool) {
return Half{Entries: map[string]catalog.CatalogEntry{"lib": {Fp: "src2:lib1"}, "same": {Fp: "src2:same"}, "mesh-build-go": entries["mesh-build-go"]}}, true
}
ev := catalog.CatalogRevisionEvent{Kind: "candidate", Group: "run-" + a.Run, Revision: 12}
if n, err := s.PlaceRun(catalog.CatalogRevisionEvent{Kind: "candidate", Group: "try-3", Revision: 11}, Half{Entries: cand}, base); n != 0 || err != nil {
t.Errorf("a try's candidate placed %d (%v)", n, err)
}
said = nil
n, err := s.PlaceRun(ev, Half{Revision: 12, Entries: cand}, base)
if err != nil || n != 2 {
t.Fatalf("the candidate placed %d (%v), want the member and the dependent lib", n, err)
}
p, _, _ := store.Get(key)
if p.Fingerprint != "src2:new" || p.Revision != 12 || len(p.Steps) == 0 {
t.Errorf("the member's pipeline placed = %+v", p)
}
dep, _, err := store.Get(RunKey("fix/1", a.Run, "lib"))
if err != nil || dep.Run != a.Run || dep.Fingerprint != "src2:lib2" || len(dep.Steps) == 0 {
t.Errorf("the dependent's pipeline = %+v (%v)", dep, err)
}
if _, _, err := store.Get(RunKey("fix/1", a.Run, "same")); err == nil {
t.Error("an entry the members do not change got a pipeline")
}
if len(said) != 2 || !strings.HasPrefix(said[0], seat.EventPipelineStarted+".") || len(asks.ids) == 0 {
t.Errorf("placing said %v and asked %d steps", said, len(asks.ids))
}
if n, _ := s.PlaceRun(ev, Half{Revision: 12, Entries: cand}, base); n != 0 {
t.Errorf("the revision heard again placed %d", n)
}
// A restart asks a run's steps again from no branch's half: they are its candidate's.
if n, err := s.Resume(store, "fix/1", Half{Entries: entries}); n != 0 || err != nil {
t.Errorf("a restart asked %d steps of the run again (%v)", n, err)
}
// Several members say run-started on the first member's module and branch and run-<id>; a run-started not said fails the
// pipelines written, so none stands running.
said = nil
two, err := s.Run("operator", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c2", Branch: "main"}, {Module: "lib", Commit: "c3", Branch: "main"}}},
func() (string, error) { return "ffffffffffff", nil })
if err != nil || len(two.Pipelines) != 2 || len(said) != 1 || said[0] != seat.EventRunStarted+".app.main.run-ffffffffffff" {
t.Errorf("a run of two: %+v (%v), said %v", two, err, said)
}
s.Say = func(string, any) error { return errors.New("bus away") }
three, err := s.Run("operator", seat.RunInput{Members: []seat.Member{{Module: "app", Commit: "c4", Branch: "main"}}}, func() (string, error) { return "eeeeeeeeeeee", nil })
if err == nil || three.Run != "" {
t.Errorf("a run whose run-started was not said answered %+v", three)
}
if p, _, _ := store.Get(RunKey("main", "eeeeeeeeeeee", "app")); p.Outcome != seat.PipelineOutcomeFailed {
t.Errorf("its pipeline is %s, want failed", p.Outcome)
}
}
@@ -37,6 +37,8 @@ type Begin struct {
Revision int64
// Entries are the catalog half's entries of the pipeline's branch: the steps' artifacts and build seats are read from them.
Entries map[string]catalog.CatalogEntry
// Published is the package's publish that started the pipeline, nil for any other start (A-202).
Published *seat.Published
}
// Key is a pipeline's key in the state pipelines, `<branch>.<source revision>.<module>`, the branch escaped as a key's
@@ -70,6 +72,10 @@ 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
// Store is the state pipelines read and written by compare-and-set, where a run's pipelines are placed (PlaceRun).
Store Updater
// Supersede ends the older running pipelines the pipeline just written supersedes (§31; Hearer.Supersede); nil ends none.
Supersede func(newer seat.Pipeline)
}
// Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second
@@ -85,7 +91,7 @@ func (s *Starter) Start(b Begin) (bool, error) {
key := Key(b.Branch, b.Source, b.Module)
p := seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fingerprint: b.Fp,
Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now,
Steps: []seat.StepResult{}}
Steps: []seat.StepResult{}, Published: b.Published}
var asks []Asked
if s.Registration != nil {
if reg, ok := s.Registration(b.Module); ok {
@@ -105,7 +111,7 @@ func (s *Starter) Start(b Begin) (bool, error) {
return false, fmt.Errorf("the pipeline %s is not written: %w", key, err)
}
ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module,
Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Revision: b.Revision}
Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Revision: b.Revision, Published: b.Published}
commitOrGroup := b.Commit
if b.Group != "" {
commitOrGroup = b.Group
@@ -113,6 +119,9 @@ func (s *Starter) Start(b Begin) (bool, error) {
if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, b.Module, b.Branch, commitOrGroup), ev); err != nil {
s.logf("pipeline-started of %s is not said: %v", key, err)
}
if s.Supersede != nil {
s.Supersede(p)
}
if s.Steps != nil && len(asks) > 0 {
s.carry(key, asks)
if err := askAll(s.Steps, s.Arm, key, asks); err != nil {
@@ -149,7 +158,8 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) {
}
continue
}
if p.Branch != branch || p.Outcome != seat.PipelineOutcomeRunning {
// A run on request's steps are placed from its candidate, not from the branch's half (PlaceRun).
if p.Branch != branch || p.Outcome != seat.PipelineOutcomeRunning || p.Run != "" {
continue
}
reg, ok := s.Registration(p.Module)
@@ -252,7 +262,8 @@ func (s *Starter) Published(ev builds.PackagePublished, outdated func(catalog.Ou
if o.Kind != outdatedKind || o.Module == "" {
continue
}
b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision, Entries: half.Entries}
b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision, Entries: half.Entries,
Published: &seat.Published{Artifact: ev.Artifact, Version: ev.Version}}
if e, ok := half.Entries[o.Module]; ok {
b.Commit, b.Fp, b.Repository = e.Commit, e.Fp, e.Repository
} else {
@@ -27,11 +27,15 @@ func (c created) Create(key string, value []byte) error {
func TestStarts(t *testing.T) {
store := created{}
var said []string
var superseding []string
s := &Starter{Pipelines: store, Say: func(subject string, body any) error {
ev := body.(seat.PipelineStarted)
said = append(said, subject+" "+ev.Pipeline)
if ev.Published == nil || ev.Published.Artifact != "mesh-sdk:go" || ev.Published.Version != "0.1.16" {
t.Errorf("pipeline-started of %s says the publish %+v (A-202)", ev.Pipeline, ev.Published)
}
return nil
}}
}, Supersede: func(newer seat.Pipeline) { superseding = append(superseding, newer.Pipeline) }}
view := func(branch string) (Half, bool) {
return Half{Revision: 9, Entries: map[string]catalog.CatalogEntry{
"app": {Module: "app", Commit: "c1", Fp: "src2:aa", Repository: "o/app"}}}, true
@@ -51,9 +55,13 @@ func TestStarts(t *testing.T) {
}
var p seat.Pipeline
if err := json.Unmarshal(store["main.41.app"], &p); err != nil || p.Commit != "c1" || p.Fingerprint != "src2:aa" || p.Revision != 9 ||
p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 {
p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 || p.Published == nil || p.Published.Version != "0.1.16" {
t.Errorf("the pipeline main.41.app = %+v (%v)", p, err)
}
// Each pipeline written supersedes the older ones it replaces (§31).
if len(superseding) != 2 {
t.Errorf("the publish's pipelines superseding: %v, want both", superseding)
}
if _, ok := store["main.41.ordered"]; ok {
t.Error("an ordering edge starts no pipeline")
}
@@ -66,6 +74,7 @@ func TestStarts(t *testing.T) {
if !strings.HasPrefix(said[0], "mesh.seat.mesh-pipeline-runner.event.pipeline-started.app.main.c1 main.41.app") {
t.Errorf("pipeline-started said on %q", said[0])
}
s.Say = func(string, any) error { return nil }
// A branch holding a dot is escaped in the key, so the key reads back; a start naming no source revision is refused.
if _, err := s.Start(Begin{Branch: "release.1", Source: 3, Module: "app"}); err != nil {
t.Fatal(err)
@@ -0,0 +1,155 @@
package pipeline
import (
"errors"
"fmt"
"time"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
)
// Drop asks a build seat's `drop` of a superseded step's ask as a kept call (design 47 G §31, §350; A-182): of node's holder
// when a holder took it (node the one its ask-taken named), else of any holder, which records it in the seat's shared drop
// key so no holder takes it. why names the newer commit.
type Drop func(buildSeat, node, ask, why string) error
// Taken notes on the running step asked under ask the node whose holder said ask-taken for it (A-182), so a drop of it is
// asked of that node's holder. A step no pipeline keeps running under ask takes nothing.
func (h *Hearer) Taken(ask, node string) error {
if ask == "" || node == "" {
return nil
}
pkeys, err := h.Store.Keys("*.*.*")
if err != nil {
return fmt.Errorf("the pipelines are not listed: %w", err)
}
for _, key := range pkeys {
for {
p, rev, err := h.Store.Get(key)
if err != nil {
break
}
i := stepOfAsk(p, ask)
if i < 0 || p.Steps[i].Outcome != seat.StepResultOutcomeRunning || p.Steps[i].Node == node {
break
}
p.Steps[i].Node = node
if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) {
continue
} else if err != nil {
return fmt.Errorf("pipeline %s cannot be written: %w", key, err)
}
return nil
}
}
return nil
}
// Supersede ends, for the pipeline newer just written, each older running pipeline of the same watched branch and module
// whose fingerprint newer's commit changed, and for a publish's pipeline each an older publish of the package started
// (design 47 G §31, §119; ADR 0666 §9; A-202): the pipeline and each step not ended
// `superseded`, naming the newer commit, and it says pipeline-ended. A module the newer commit left as it was finishes its
// pipeline. The ask of each superseded step on a build seat that serves `drop` is dropped (drop; its bound deleted through
// unbound); a step on mesh-builds or mesh-wants, whose seats serve no drop, is only ended superseded (A-182). A run on
// request's pipeline neither supersedes nor is superseded. It says how many pipelines it ended.
func (h *Hearer) Supersede(newer seat.Pipeline, drop Drop, unbound func(ask string) error) (int, error) {
if newer.Run != "" || newer.Source <= 0 {
return 0, nil
}
pkeys, err := h.Store.Keys("*.*.*")
if err != nil {
return 0, fmt.Errorf("the pipelines are not listed: %w", err)
}
why := supersededWhy(newer)
ended := 0
var first error
for _, key := range pkeys {
if key == newer.Pipeline {
continue
}
done, err := h.supersedeOne(key, newer, why, drop, unbound)
if err != nil && first == nil {
first = err
}
if done {
ended++
}
}
return ended, first
}
// superseded says whether old is a running pipeline newer supersedes (§31, A-202): of the same watched branch and an older
// source revision, either of the same module at another fingerprint, or, for newer a package's publish started, one an older
// publish of that package started (its version another, made from an older source revision of the branch).
func superseded(old, newer seat.Pipeline) bool {
if old.Outcome != seat.PipelineOutcomeRunning || old.Run != "" || old.Branch != newer.Branch || old.Source >= newer.Source {
return false
}
if newer.Published != nil && old.Published != nil && old.Published.Artifact == newer.Published.Artifact {
return old.Published.Version != newer.Published.Version
}
return old.Module == newer.Module && old.Fingerprint != newer.Fingerprint
}
// supersededWhy names the newer commit, or the newer publish, that ended a pipeline.
func supersededWhy(newer seat.Pipeline) string {
if newer.Published != nil {
return fmt.Sprintf("the newer publish %s %s of %s supersedes it", newer.Published.Artifact, newer.Published.Version, newer.Branch)
}
return fmt.Sprintf("the newer commit %s of %s changed %s's fingerprint", newer.Commit, newer.Branch, newer.Module)
}
func (h *Hearer) supersedeOne(key string, newer seat.Pipeline, why string, drop Drop, unbound func(string) error) (bool, error) {
for {
p, rev, err := h.Store.Get(key)
if err != nil || !superseded(p, newer) {
return false, nil
}
now := h.now()
var drops []seat.StepResult
var changed []int
for i := range p.Steps {
s := &p.Steps[i]
if s.Outcome != seat.StepResultOutcomeRunning && s.Outcome != seat.StepResultOutcomeWaiting {
continue
}
if s.Ask != "" && BoundSeat(s.Seat) {
drops = append(drops, *s)
}
s.Outcome, s.Why, s.Ended = seat.StepResultOutcomeSuperseded, why, ptrTime(now)
changed = append(changed, i)
}
p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeSuperseded, why, ptrTime(now)
p.EndCount++
if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) {
continue
} else if err != nil {
return false, fmt.Errorf("pipeline %s cannot be written: %w", key, err)
}
var first error
for _, i := range changed {
h.logEnd(key, p.Steps[i], now)
if err := h.say(seat.EventStepEnded, p, &p.Steps[i]); err != nil && first == nil {
first = fmt.Errorf("pipeline %s's %s is written but not said: %w", key, seat.EventStepEnded, err)
}
}
for _, s := range drops {
if drop != nil {
if err := drop(s.Seat, s.Node, s.Ask, why); err != nil {
h.logf("the drop of ask %s of the superseded %s is not answered: %v", s.Ask, key, err)
}
}
if unbound != nil {
if err := unbound(s.Ask); err != nil {
h.logf("the bound of ask %s of the superseded %s is not dropped: %v", s.Ask, key, err)
}
}
}
if err := h.said(p, false, true); err != nil && first == nil {
first = err
}
return true, first
}
}
func ptrTime(t time.Time) *time.Time { return &t }
@@ -0,0 +1,93 @@
package pipeline
import (
"slices"
"strings"
"testing"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
)
// RN8 (design 47 G §31, §119, §350; A-182, A-202): a step keeps the node that took its ask; a newer commit changing the
// module's fingerprint, or a newer publish of the package that started a pipeline, ends the older running pipeline
// superseded, its asks dropped of the node that took each (any holder for one never taken), a mesh-builds step only ended;
// an unchanged fingerprint, a run on request and an equal publish supersede nothing.
func TestSupersede(t *testing.T) {
h, store, said, _, _, _ := fixture()
old := store[pkey]
old.p.Steps = append(old.p.Steps, seat.StepResult{Artifact: "program", Step: seat.StepResultStepComposition, Status: seat.StepResultStatusRequired,
Outcome: seat.StepResultOutcomeRunning, Ask: "b1", Seat: builds.Seat})
store[pkey] = old
if err := h.Taken("a1", "n7"); err != nil {
t.Fatal(err)
}
if p, _, _ := store.Get(pkey); p.Steps[0].Node != "n7" || p.Steps[1].Node != "" {
t.Fatalf("after ask-taken of a1 on n7 the steps name %q and %q", p.Steps[0].Node, p.Steps[1].Node)
}
type dropped struct{ seat, node, ask string }
var drops []dropped
var unbound []string
drop := func(s, node, ask, why string) error {
if !strings.Contains(why, "c2") {
t.Errorf("the drop of %s names no newer commit: %q", ask, why)
}
drops = append(drops, dropped{s, node, ask})
return nil
}
unbind := func(ask string) error { unbound = append(unbound, ask); return nil }
newer := func(key string, source int64, fp string, run string, pub *seat.Published) seat.Pipeline {
return seat.Pipeline{Pipeline: key, Branch: "main", Source: source, Module: "app", Commit: "c2", Fingerprint: fp, Run: run, Published: pub,
Outcome: seat.PipelineOutcomeRunning}
}
for _, c := range []struct {
name string
newer seat.Pipeline
}{
{"an unchanged fingerprint", newer("main.6.app", 6, "src2:aa", "", nil)},
{"a run on request", newer("main.r1.app", 6, "src2:bb", "r1", nil)},
{"an older source revision", newer("main.4.app", 4, "src2:bb", "", nil)},
} {
if n, err := h.Supersede(c.newer, drop, unbind); n != 0 || err != nil {
t.Errorf("%s superseded %d (%v)", c.name, n, err)
}
}
*said = nil
if n, err := h.Supersede(newer("main.6.app", 6, "src2:bb", "", nil), drop, unbind); n != 1 || err != nil {
t.Fatalf("a changed fingerprint superseded %d (%v)", n, err)
}
p, _, _ := store.Get(pkey)
if p.Outcome != seat.PipelineOutcomeSuperseded || p.EndCount != 1 || !strings.Contains(p.Why, "c2") {
t.Errorf("the older pipeline is %s (%d ends): %q", p.Outcome, p.EndCount, p.Why)
}
for _, s := range p.Steps {
if s.Outcome != seat.StepResultOutcomeSuperseded {
t.Errorf("the %s step is %s, want superseded", s.Step, s.Outcome)
}
}
want := []dropped{{nodebuildgo.Seat, "n7", "a1"}, {nodebuildgo.Seat, "", "a2"}}
if !slices.Equal(drops, want) || !slices.Equal(unbound, []string{"a1", "a2"}) {
t.Errorf("dropped %+v and unbound %v, want %+v (the mesh-builds step only ended)", drops, unbound, want)
}
if n := len(*said); n != 5 || (*said)[4] != seat.EventPipelineEnded+".app.main.c1" {
t.Errorf("said %v, want four step-ended then pipeline-ended", *said)
}
if n, _ := h.Supersede(newer("main.7.app", 7, "src2:cc", "", nil), drop, unbind); n != 0 {
t.Errorf("a superseded pipeline was superseded again: %d", n)
}
// A newer publish of a package supersedes the pipelines an older publish of it started, whatever their module.
pub := memStore{"main.5.lib": {p: seat.Pipeline{Pipeline: "main.5.lib", Branch: "main", Source: 5, Module: "lib", Fingerprint: "src2:aa",
Outcome: seat.PipelineOutcomeRunning, Published: &seat.Published{Artifact: "core:pkg", Version: "v1"}}, rev: 1}}
h.Store = pub
if n, _ := h.Supersede(newer("main.6.app", 6, "src2:aa", "", &seat.Published{Artifact: "core:pkg", Version: "v1"}), drop, unbind); n != 0 {
t.Errorf("a publish of the same version superseded %d", n)
}
if n, err := h.Supersede(newer("main.6.app", 6, "src2:aa", "", &seat.Published{Artifact: "core:pkg", Version: "v2"}), drop, unbind); n != 1 || err != nil {
t.Fatalf("a newer publish superseded %d (%v)", n, err)
}
if p, _, _ := pub.Get("main.5.lib"); p.Outcome != seat.PipelineOutcomeSuperseded || !strings.Contains(p.Why, "v2") {
t.Errorf("the older publish's pipeline is %s: %q", p.Outcome, p.Why)
}
}
@@ -334,3 +334,77 @@ 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)
}
// Run starts a run on request for the caller the bus proved (design 47 G §64; Starter.Run): its pipelines written, its
// run-started said, its steps placed when its candidate's revision is heard.
func (v Verbs) Run(caller string, in seat.RunInput) (seat.RunAnswer, error) {
if v.Starter == nil {
return seat.RunAnswer{}, errors.New("the runner cannot start a run: it holds no starter")
}
return v.Starter.Run(caller, in, NewRunID)
}
// 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,8 +69,10 @@ 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",
@@ -75,13 +84,18 @@ func TestManifest(t *testing.T) {
{"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act",
// RN3 (§53, §334): a publication's dependents behind.
"seat:mesh-catalog.outdated",
// RN8 (§31, §350; A-182): a superseded step's ask dropped of the build seat that holds it.
"seat:node-build-go.drop", "seat:node-build-node.drop", "seat:node-build-image.drop",
// A-74, A-89: who holds the seat operator-channel, whose state asks holds the router's record of the approval ask.
"seat:mesh-module-assigner.assignments",
// A-102 (§22, §24): the accepted kinds the runner puts its steps on.
"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"})},
// 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 +131,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)
+21 -6
View File
@@ -23,8 +23,9 @@
}
],
"tools": [
"pipeline_first_views",
"pipeline_first_view_again"
"runner_first_views",
"runner_first_view_again",
"runner_health"
],
"consumes": [
"mesh-delivery-indexer.take-in.taken",
@@ -53,7 +54,10 @@
"seat:mesh-wants.step",
"seat:node-build-go.check.step",
"seat:node-build-node.check.step",
"seat:node-build-image.check.step"
"seat:node-build-image.check.step",
"seat:node-build-go.drop",
"seat:node-build-node.drop",
"seat:node-build-image.drop"
],
"uses": [
"operator-channel"
@@ -61,11 +65,13 @@
"state": [
{
"name": "pipelines",
"history": 1
"history": 1,
"max-mib": 256
},
{
"name": "kept",
"history": 1
"history": 1,
"max-mib": 128
},
{
"name": "aged",
@@ -87,6 +93,10 @@
{
"name": "first-views",
"history": 1
},
{
"name": "consumers",
"history": 1
}
],
"streams": [
@@ -105,6 +115,10 @@
{
"name": "step-outputs",
"max-mib": 1024
},
{
"name": "sent-objects",
"max-mib": 256
}
],
"own-secrets": {
@@ -140,7 +154,8 @@
},
"user": "mesh-delivery-pipeline",
"health": {
"kind": "unit"
"kind": "tool",
"tool": "runner_health"
}
}
],