From cbb678f42ebb35ecb346626733d5fea5f664fce9 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:15:38 +0100 Subject: [PATCH] B1g-3: a run's rerun asks with the candidate's catalog half, not the branch's Builds: design 47 G A-222, A-193, A-205 --- .../cmd/mesh-delivery-pipeline/verbs.go | 6 ++- .../internal/pipeline/verbs.go | 20 +++++-- .../internal/pipeline/verbs_test.go | 53 +++++++++++++++++++ 3 files changed, 75 insertions(+), 4 deletions(-) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go index cecbacbd..44671341 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go @@ -9,6 +9,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + 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" @@ -65,5 +66,8 @@ func dropBound(ctx context.Context, bounds jetstream.KeyValue) func(string) erro // writeVerbs are the verbs over the hearer and the starter, their hand acts recorded through the seat mesh-hand-acts. func writeVerbs(ctx context.Context, h *pipeline.Hearer, s *pipeline.Starter, view func(string) (pipeline.Half, bool), calls *process.Conn, bounds jetstream.KeyValue) pipeline.Verbs { - return pipeline.Verbs{Hearer: h, Starter: s, View: view, Acts: meshhandactsclient.Recorder{Ask: calls.Call}, Drop: dropBound(ctx, bounds)} + candidate := func(group, branch string, revision int64) (pipeline.Half, error) { + return readCandidate(calls.CallKept, catalog.CatalogRevisionEvent{Kind: "candidate", Group: group, Branch: branch, Revision: revision}) + } + return pipeline.Verbs{Hearer: h, Starter: s, View: view, Candidate: candidate, Acts: meshhandactsclient.Recorder{Ask: calls.Call}, Drop: dropBound(ctx, bounds)} } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go index 6ede0c1b..03bdf718 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go @@ -10,6 +10,7 @@ import ( seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" wants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -21,6 +22,9 @@ type Verbs struct { // Starter asks a step again (rerun), as it asks the steps of a start; View reads a branch's catalog half for the ask. Starter *Starter View func(branch string) (Half, bool) + // Candidate reads a run's candidate's catalog half, its group `run-`, on branch at the revision its pipeline was placed + // from, for a rerun of a run's step (A-222): a run's steps are the candidate's, never the branch's. + Candidate func(group, branch string, revision int64) (Half, error) // Acts records a hand act before the verb does it (rerun, accept, cancel); a nil Acts refuses the verb, nothing recorded. Acts views.HandActs // Drop drops the bound of an ask a cancelled pipeline no longer waits on. @@ -296,15 +300,25 @@ func (v Verbs) Rerun(caller string, in seat.RerunInput) (seat.RerunAnswer, error // askOf is the ask to put on the serving seat for step s again: the one Place makes for it, under a new id (an ask put twice // under one id is kept once), saying it is a rerun and by whom. func (v Verbs) askOf(p seat.Pipeline, s seat.StepResult, caller string, now time.Time) (*Asked, error) { - if v.Starter == nil || v.Starter.Registration == nil || v.Starter.Steps == nil || v.View == nil { + if v.Starter == nil || v.Starter.Registration == nil || v.Starter.Steps == nil || (v.View == nil && p.Run == "") { return nil, errors.New("the runner cannot ask a step: it holds no registration or no ask") } reg, ok := v.Starter.Registration(p.Module) if !ok { return nil, fmt.Errorf("%s has no registration", p.Module) } - half, ok := v.View(p.Branch) - if !ok { + var half Half + if p.Run != "" { + if v.Candidate == nil { + return nil, errors.New("the runner cannot read a run's candidate") + } + var err error + if half, err = v.Candidate(RunGroupPrefix+p.Run, keys.EscapeName(p.Branch), p.Revision); err != nil { + return nil, fmt.Errorf("the candidate %s%s's revision %d is not read, so its step is not asked: %w", RunGroupPrefix, p.Run, p.Revision, err) + } + } else if h, ok := v.View(p.Branch); ok { + half = h + } else { return nil, fmt.Errorf("the catalog half of %s is not held yet: the step is asked when it is", p.Branch) } _, placed := Place(p, reg, half.Entries, now, v.Starter.Name) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go index 633f120b..6b6ed3df 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go @@ -155,6 +155,59 @@ func TestAcceptRerunAndCancelWriteByCompareAndSetAndRecordTheirHandAct(t *testin } } +// A-222: a rerun of a run's step asks with the candidate's catalog half (its entries, here the directory), never the branch's, +// and is refused, nothing asked, when the candidate cannot be read. +func TestRerunOfARunAsksWithItsCandidatesHalf(t *testing.T) { + v, store, _, asks, _, _ := verbsFixture() + p, rev, _ := store.Get(pkey) + p.Run, p.Branch, p.Revision = "0a1b2c3d4e5f", "fix/1", 12 + for i := range p.Steps { + if p.Steps[i].Step == "lint" && p.Steps[i].Artifact == "app:program" { + p.Steps[i].Outcome = seat.StepResultOutcomeFailed + } + } + store.Update(pkey, p, rev) + _, cand := placeFixture() + e := cand["app"] + e.Path = "candidate/app" + cand["app"] = e + var read struct { + group, branch string + revision int64 + } + v.Candidate = func(group, branch string, revision int64) (Half, error) { + read.group, read.branch, read.revision = group, branch, revision + return Half{Revision: revision, Entries: cand}, nil + } + in := seat.RerunInput{Pipeline: pkey, Step: "lint", Artifact: "app:program", Why: "retry"} + if _, err := v.Rerun("operator", in); err != nil { + t.Fatal(err) + } + if b, ok := asks.asks[0].(nodebuildgo.StepAsk); !ok || b.Step == nil || b.Step.Fetch.Dir != "candidate/app" { + t.Errorf("the ask = %+v, want the candidate's directory", asks.asks[0]) + } + if read.group != "run-0a1b2c3d4e5f" || read.branch != "fix/1" || read.revision != 12 { + t.Errorf("the candidate read = %+v", read) + } + p, rev, _ = store.Get(pkey) + p.Steps[stepIndex(p, "lint", "app:program")].Outcome = seat.StepResultOutcomeFailed + store.Update(pkey, p, rev) + v.Candidate = func(string, string, int64) (Half, error) { return Half{}, errors.New("no answer") } + n := len(asks.asks) + if _, err := v.Rerun("operator", in); err == nil || len(asks.asks) != n { + t.Errorf("a rerun with no candidate read: %v, asked %d more", err, len(asks.asks)-n) + } +} + +func stepIndex(p seat.Pipeline, step, artifact string) int { + for i, s := range p.Steps { + if s.Step == step && s.Artifact == artifact { + return i + } + } + return -1 +} + // A hand act the log does not take is a verb that does nothing (ADR 0565 clause 11). func TestAVerbWhoseHandActIsNotRecordedDoesNothing(t *testing.T) { v, store, _, _, _, _ := verbsFixture()