diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends.go new file mode 100644 index 00000000..fbb5149c --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends.go @@ -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() }() +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends_integration_test.go new file mode 100644 index 00000000..cb17252d --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/buildends_integration_test.go @@ -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") + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index 0be86b7f..f6b1d924 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -104,12 +104,13 @@ 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), // The runner's own module and fingerprint (A-149): a take-in giving it another holds the pipelines it starts (§65, A-195). Self: cred.Module, SelfFp: getenv("MESH_FINGERPRINT"), Awaiting: pipeline.Awaiting(pipelineStore{ctx: ctx, kv: pipelines}), - Hold: armHeld(ctx, bounds), + 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() @@ -153,6 +154,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }, logger.Printf); err != nil { return err } + 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 } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go index ec0f761d..58f0ad0f 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "strings" + "time" "github.com/nats-io/nats.go/jetstream" @@ -174,10 +175,11 @@ func armBound(ctx context.Context, bounds jetstream.KeyValue) pipeline.Arm { } // armHeld is the bound of the pipelines held for the runner's new version: a key of the state bounds named for the awaited -// fingerprint, with its own time to live of 30 minutes (design 47 G §266, A-195, A-77). -func armHeld(ctx context.Context, bounds jetstream.KeyValue) func(string) error { - return func(fp string) error { - _, err := bounds.Create(ctx, pipeline.HeldKey(fp), []byte(`{"fingerprint":"`+fp+`"}`), jetstream.KeyTTL(pipeline.HeldBound)) +// 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 } diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod index ef4b1695..028117ac 100644 --- a/modules/mesh-delivery-pipeline/go.mod +++ b/modules/mesh-delivery-pipeline/go.mod @@ -11,6 +11,7 @@ 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 diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/ends.go b/modules/mesh-delivery-pipeline/internal/pipeline/ends.go new file mode 100644 index 00000000..cb7ae9e1 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/ends.go @@ -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 +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go index 65404d0f..b6222c7a 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go @@ -64,17 +64,68 @@ func (s *Starter) startHeld(b Begin, key string, now time.Time, awaiting string) } else if err != nil { return false, fmt.Errorf("the pipeline %s is not written: %w", key, err) } - if s.Hold != nil { - if err := s.Hold(awaiting); err != nil { + 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) } } - if err := s.Say(runnerclient.Subject(seat.EventWaitHeld, b.Module, b.Branch, commitOrGroup(p)), s.waitOf(p, now, "")); err != nil { + 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 @@ -210,3 +261,9 @@ func (v Verbs) Release(_ string, in seat.ReleaseInput) (seat.ReleaseAnswer, erro } 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) +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go index a5702164..d150d00a 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go @@ -39,7 +39,8 @@ func TestPipelinesAreHeldForTheNewRunnerAndReleasedWhenItRuns(t *testing.T) { var said []string var bounds []string s := &Starter{Pipelines: store, Steps: asks, Name: "the-runner", Self: "runner", SelfFp: "src2:old", - Awaiting: Awaiting(store.memStore), Hold: func(fp string) error { bounds = append(bounds, fp); return nil }, + 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:]) @@ -162,3 +163,72 @@ func TestHeldWaitIsOnTheRunnerVersion(t *testing.T) { 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) + } + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index f35288ba..3d01f85b 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -74,10 +74,12 @@ type Starter struct { Arm Arm // Self is the runner's own module and SelfFp the fingerprint it runs (A-149); a take-in giving Self another fingerprint // holds its pipelines for the new runner (§65, §266, A-195). Awaiting is the fingerprint the held pipelines await, while - // any is held; Hold arms the bound of that wait (HeldBound). All nil: nothing is held. + // 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) error + 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) } diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index c47f2bfb..bcf8f548 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -75,7 +75,7 @@ func TestManifest(t *testing.T) { {"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", "node-engine.running", "mesh-wants.step-ended", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-conditions.findings-asked"})}, diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index 206245fa..1bb8273d 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -35,6 +35,7 @@ "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",