novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Runner (A-336): a take-in heard before its branch's half starts with no steps

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-13 02:41:39 +01:00
parent 8af822e3fd
commit d827d1abf0
4 changed files with 71 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go
+28 -3
View File
@@ -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)
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go
+17 -1
View File
@@ -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) {
modules/mesh-delivery-pipeline/internal/pipeline/start.go
+9 -3
View File
@@ -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, `<branch>.<source revision>.<module>`, 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
}
modules/mesh-delivery-pipeline/internal/pipeline/start_test.go
+17
View File
@@ -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))
}
}