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

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)}
 }

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-<id>`, 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)

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()