From d827d1abf0ca8299440d658a2bd827f0aefbe22e Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 13 Nov 2026 02:41:39 +0100 Subject: [PATCH] Runner (A-336): a take-in heard before its branch's half starts with no steps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A take-in heard before its branch's catalog half was held placed steps from empty entries, acting on a branch §200 says nothing new of is acted on. Its pipelines now start with no steps, and the half becoming held places them by A-327's path, also when the first view was none yet and the first half comes from a later catalog-revision. Builds: design 47 G §200, A-327, A-336 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../run_integration_test.go | 31 +++++++++++++++++-- .../cmd/mesh-delivery-pipeline/views.go | 18 ++++++++++- .../internal/pipeline/start.go | 12 +++++-- .../internal/pipeline/start_test.go | 17 ++++++++++ 4 files changed, 71 insertions(+), 7 deletions(-) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go index 1f3f9581..25489f6a 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go @@ -11,6 +11,7 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "testing" "time" @@ -94,7 +95,21 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes } } answer("mesh.seat.mesh-catalog.call.registrations.*", catalog.RegistrationsAnswer{Registrations: []catalog.Registration{reg}}) - answer("mesh.seat.mesh-catalog.call.revision.*", catalog.RevisionAnswer{NoneYet: true}) + // The branch main has no revision yet at the runner's start (none-yet), then revision 1 holding app (A-336). + var revised atomic.Bool + if _, err := nc.Subscribe("mesh.seat.mesh-catalog.call.revision.*", func(m *nats.Msg) { + var a catalog.RevisionAnswer + if revised.Load() { + a = catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: 1, Kind: "trunk", Branch: "main", Source: 3}, + Entries: []catalog.CatalogEntry{{Module: "app", Commit: "c1", Fp: "src2:aa"}}} + } else { + a.NoneYet = true + } + raw, _ := json.Marshal(map[string]any{"result": a}) + _ = m.Respond(raw) + }); err != nil { + t.Fatal(err) + } answer("mesh.seat.mesh-hand-acts.call.hand-act.*", map[string]any{}) // The seat list (A-186): the seat mesh-catalog's seat-definitions names the runner's seat, whose definition its write // verbs' callers are checked against, as the contracts' file holds it. @@ -148,12 +163,22 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes const key = "main.3.app" var p seat.Pipeline waitFor(t, ctx, logs, func() bool { - // Said until the runner holds the registration, so the pipeline is placed with its policy's steps. + // Heard before main's half is held, the pipeline starts with no steps, never placed from empty entries (A-336). if _, err := js.Publish(ctx, "mesh.mod.mesh-delivery-indexer.event.take-in.taken", taken); err != nil { t.Fatal(err) } - return readPipeline(ctx, pipelines, key, &p) && len(p.Steps) > 1 + return readPipeline(ctx, pipelines, key, &p) }) + if len(p.Steps) != 0 { + t.Fatalf("a take-in before its branch's half placed %d steps, want none (A-336)", len(p.Steps)) + } + // The revision said is main's first half: held, it places the pipeline's steps from the registration (A-327). + revised.Store(true) + rev, _ := json.Marshal(catalog.CatalogRevisionEvent{V: 1, Revision: 1, Kind: "trunk", Branch: "main", Source: 3}) + if _, err := js.Publish(ctx, catalog.SubjectCatalogRevision("main"), rev); err != nil { + t.Fatal(err) + } + waitFor(t, ctx, logs, func() bool { return readPipeline(ctx, pipelines, key, &p) && len(p.Steps) > 1 }) build := stepOf(p, seat.StepResultStepBuild) if build.Outcome != seat.StepResultOutcomeRunning || build.Ask == "" { t.Fatalf("the build step = %+v", build) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go index 55bbe780..4a2ed358 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go @@ -2,6 +2,9 @@ package main import ( "context" + "encoding/json" + "sync/atomic" + "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -42,10 +45,23 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr // 47 G §532, A-26): the reader keeps its status, raises and clears its findings through the module's writer, hears the // seat's catalog-revision on the module's consumer and asks its first view at once. cv := &pipeline.Views{Logf: logf, OnReady: onReady, Open: func(branch string) (pipeline.BranchView, error) { - c, _, err := meshcatalogclient.WireCatalogProcess(reader, call, branch, nil) + // A branch whose first view was none yet is ready with no half; the first half held later, from a catalog-revision, + // is picked up as readiness is: its pipelines started with no steps are placed then (A-327, A-336). + var wired atomic.Pointer[meshcatalogclient.Catalog] + var held atomic.Bool + c, _, err := meshcatalogclient.WireCatalogProcess(reader, call, branch, func(string, json.RawMessage, bool, time.Time) { + c := wired.Load() + if c == nil || onReady == nil { + return + } + if _, ok := c.View(); ok && !held.Swap(true) { + go onReady(branch) + } + }) if err != nil { return nil, err } + wired.Store(c) return catalogView{c}, nil }} regs, _, err := meshcatalogclient.WireRegistrationsProcess(reader, call, func(reg catalog.Registration, gone bool) { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index dcd0f0e6..7d209e4d 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -41,6 +41,8 @@ type Begin struct { 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 + // unheld says the branch's catalog half is not held yet: the pipeline starts with no steps, placed once it is (A-336). + unheld bool } // Key is a pipeline's key in the state pipelines, `..`, the branch escaped as a key's @@ -137,7 +139,9 @@ func (s *Starter) Start(b Begin) (bool, error) { } p := newPipeline(b, key, now) var asks []Asked - if s.Registration != nil { + if b.unheld { + s.logf("%s starts with no step: the catalog half of %s is not held yet, and its steps are placed once it is (A-336)", key, b.Branch) + } else if s.Registration != nil { if reg, ok := s.Registration(b.Module); ok { p.Steps, asks = Place(p, reg, b.Entries, now, s.Name) } else { @@ -433,7 +437,9 @@ func (s *Starter) TakenIn(ev catalog.TakeInTaken, view func(branch string) (Half if ev.Branch == "" || ev.Source <= 0 { return 0, fmt.Errorf("a take-in.taken names its watched branch and source revision: key %q branch %q source %d", ev.Key, ev.Branch, ev.Source) } - half, _ := view(ev.Branch) + // A take-in heard before its branch's half is held starts its pipelines with no steps, never placed from empty entries + // (§200, A-336); Waiting places them once the half is held. + half, held := view(ev.Branch) started := 0 var first error // A take-in giving the runner's own module a new fingerprint holds every other pipeline it starts for that runner (§65, @@ -446,7 +452,7 @@ func (s *Starter) TakenIn(ev catalog.TakeInTaken, view func(branch string) (Half } for _, e := range ev.Entries { ok, err := s.Start(Begin{Branch: ev.Branch, Source: ev.Source, Module: e.Module, Commit: e.Commit, Fp: e.Fingerprint, - Group: ev.Group, Repository: e.Repository, Revision: half.Revision, Entries: half.Entries, Await: await}) + Group: ev.Group, Repository: e.Repository, Revision: half.Revision, Entries: half.Entries, Await: await, unheld: !held}) if err != nil && first == nil { first = err } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go index 95d13a76..d084fadf 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start_test.go @@ -178,4 +178,21 @@ func TestAPipelineStartedBeforeItsRegistrationIsPlacedWhenItArrives(t *testing.T if n, err := s.Waiting(store.memStore, "", late); n != 0 || err != nil || len(asks.asks) != before { t.Errorf("the registrations' first view after it placed %d (%v), asked %d more, want nothing", n, err, len(asks.asks)-before) } + + // A-336 (§200): a take-in heard before its branch's half is held starts its pipelines with no steps, never placed from + // empty entries, the registration held or not; the half becoming held places them (Waiting). + held = false + taken = catalog.TakeInTaken{Key: "m.3", Branch: "next", Source: 6, Entries: []catalog.TakeInEntry{{Module: "app", Commit: "c3", Fingerprint: "src2:cc"}}} + if n, err := s.TakenIn(taken, late); n != 1 || err != nil { + t.Fatalf("the take-in before the half started %d (%v)", n, err) + } + if p, _, _ := store.Get("next.6.app"); len(p.Steps) != 0 { + t.Fatalf("a take-in before its branch's half placed %d steps from empty entries, want none", len(p.Steps)) + } + held = true + n, err = s.Waiting(store.memStore, "next", late) + p, _, _ = store.Get("next.6.app") + if n != 1 || err != nil || len(p.Steps) == 0 { + t.Errorf("the half of next placed %d (%v): %d steps, want the take-in's pipeline placed", n, err, len(p.Steps)) + } }