B1g-2: the runner at its start places each run's lost pipelines from its candidate's newest revision

Builds: design 47 G A-222, A-193, A-180, A-205

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go

@@ -7,12 +7,14 @@ import (
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
 	catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
 	"git.novox.be/novox/mesh-sdk/go/buscalls"
+	"git.novox.be/novox/mesh-sdk/go/keys"
 
 	"mesh-delivery-pipeline/internal/pipeline"
 )
 
 // readCandidate reads a candidate's revision whole through the catalog seat's verb revision, with the candidate's group and
-// heads and the revision its event names, page by page (design 47 G A-180).
+// heads and the revision its event names, page by page (design 47 G A-180); an event naming no revision reads the candidate's
+// newest, the one its answer names (A-205, A-222).
 func readCandidate(call func(key string, body any) (json.RawMessage, error), ev catalog.CatalogRevisionEvent) (pipeline.Half, error) {
 	half := pipeline.Half{Revision: ev.Revision, Source: ev.Source, Entries: map[string]catalog.CatalogEntry{}}
 	after := ""
@@ -29,6 +31,10 @@ func readCandidate(call func(key string, body any) (json.RawMessage, error), ev
 		if err := json.Unmarshal(answer, &a); err != nil {
 			return half, fmt.Errorf("%s answered no revision: %w", meshcatalogclient.RevisionVerb, err)
 		}
+		if a.Revision != nil && ev.Revision == 0 && after == "" {
+			half.Revision, half.Source = a.Revision.Revision, a.Revision.Source
+			ev.Revision = a.Revision.Revision
+		}
 		if a.Revision == nil || a.Revision.Revision != ev.Revision || a.Revision.Group != ev.Group {
 			return half, fmt.Errorf("%s answered %+v for the candidate %s's revision %d", meshcatalogclient.RevisionVerb, a.Revision, ev.Group, ev.Revision)
 		}
@@ -41,3 +47,31 @@ func readCandidate(call func(key string, body any) (json.RawMessage, error), ev
 		after = a.Next
 	}
 }
+
+// placeLostRuns places, at the runner's start, each run whose pipelines have no steps yet (A-222): the candidate's
+// newest revision is read through the seat's revision verb with the run's group and placed as its catalog-revision event
+// would place it, so a placing lost while the runner was down is taken up. A run whose candidate cannot be read is said and
+// the others go on; its first failure is returned, to be tried at the next start.
+func placeLostRuns(unplaced func() ([]pipeline.UnplacedRun, error), call func(key string, body any) (json.RawMessage, error),
+	place func(ev catalog.CatalogRevisionEvent, half pipeline.Half) error, logf func(string, ...any)) error {
+	runs, err := unplaced()
+	if err != nil {
+		return err
+	}
+	var first error
+	for _, u := range runs {
+		ev := catalog.CatalogRevisionEvent{Kind: "candidate", Group: pipeline.RunGroupPrefix + u.Run, Branch: keys.EscapeName(u.Branch)}
+		half, err := readCandidate(call, ev)
+		if err == nil {
+			ev.Revision, ev.Source = half.Revision, half.Source
+			err = place(ev, half)
+		}
+		if err != nil {
+			logf("the run %s of %s is not placed at the start, its candidate's newest revision: %v", u.Run, u.Branch, err)
+			if first == nil {
+				first = err
+			}
+		}
+	}
+	return first
+}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go

@@ -2,9 +2,12 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
 	"testing"
 
 	catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
+
+	"mesh-delivery-pipeline/internal/pipeline"
 )
 
 // A-180: a run's candidate revision is read whole through the seat's revision verb with the candidate's group, heads and
@@ -34,3 +37,37 @@ func TestReadCandidateReadsTheRevisionWhole(t *testing.T) {
 		t.Error("an answer of another revision was taken")
 	}
 }
+
+// A-222: at its start the runner reads the newest revision of each unplaced run's candidate through the seat's revision verb
+// (its group, no heads, no revision) and places it; a run whose candidate cannot be read is said and the others go on.
+func TestPlaceLostRunsReadsEachCandidateNewest(t *testing.T) {
+	var asked []catalog.RevisionInput
+	call := func(key string, body any) (json.RawMessage, error) {
+		in := body.(catalog.RevisionInput)
+		asked = append(asked, in)
+		if in.Group == "run-bad" {
+			return nil, errors.New("no responders")
+		}
+		return json.Marshal(catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: 7, Source: 3, Kind: "candidate", Group: in.Group},
+			Entries: []catalog.CatalogEntry{{Module: "app", Fp: "src2:a"}}})
+	}
+	var placed []catalog.CatalogRevisionEvent
+	var said []string
+	err := placeLostRuns(func() ([]pipeline.UnplacedRun, error) {
+		return []pipeline.UnplacedRun{{Run: "bad", Branch: "fix/1"}, {Run: "good", Branch: "fix/1"}}, nil
+	}, call,
+		func(ev catalog.CatalogRevisionEvent, half pipeline.Half) error {
+			placed = append(placed, ev)
+			return nil
+		},
+		func(f string, a ...any) { said = append(said, f) })
+	if err == nil || len(said) == 0 {
+		t.Errorf("an unreadable candidate was not said: %v, %v", err, said)
+	}
+	if len(placed) != 1 || placed[0].Group != "run-good" || placed[0].Revision != 7 || placed[0].Source != 3 || placed[0].Branch != "fix/1" {
+		t.Fatalf("placed %+v", placed)
+	}
+	if asked[1].Heads != "" || asked[1].Revision != 0 || asked[1].Group != "run-good" {
+		t.Errorf("asked %+v, want the candidate's group alone", asked[1])
+	}
+}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go

@@ -114,6 +114,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	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)}
+	placeHalf := func(ev catalog.CatalogRevisionEvent, half pipeline.Half) error {
+		n, err := starter.PlaceRun(ev, half, rv.Catalog.View)
+		logger.Printf("the candidate %s's revision %d placed %d pipelines of its run", ev.Group, ev.Revision, n)
+		return err
+	}
 	resume := func(branch string) {
 		half, ok := rv.Catalog.View(branch)
 		if !ok {
@@ -128,6 +133,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	resumer.Store(&resume)
 	starter.Store = pipelineStore{ctx: ctx, kv: pipelines}
+	// A run's placing lost while the runner was down is taken up from its candidate's newest revision, once the registrations
+	// that place its steps are held (A-222).
+	go func() {
+		select {
+		case <-ctx.Done():
+			return
+		case <-rv.Registrations.Ready():
+		}
+		if err := placeLostRuns(starter.UnplacedRuns, calls.CallKept, placeHalf, logger.Printf); err != nil {
+			logger.Printf("the runs not placed at the start: %v", err)
+		}
+	}()
 	// A run's candidate revision places its pipelines' steps, its entries read through the seat's revision verb (A-180).
 	place := func(ev catalog.CatalogRevisionEvent) error {
 		if !strings.HasPrefix(ev.Group, pipeline.RunGroupPrefix) {
@@ -137,9 +154,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		if err != nil {
 			return fmt.Errorf("the candidate %s's revision %d is not read, so its run is not placed: %w", ev.Group, ev.Revision, err)
 		}
-		n, err := starter.PlaceRun(ev, half, rv.Catalog.View)
-		logger.Printf("the candidate %s's revision %d placed %d pipelines of its run", ev.Group, ev.Revision, n)
-		return err
+		return placeHalf(ev, half)
 	}
 	placer.Store(&place)
 	// A newer commit or publish supersedes the older running pipelines it replaces, their asks dropped as kept calls (§31,

modules/mesh-delivery-pipeline/internal/pipeline/run.go

@@ -187,6 +187,31 @@ func (s *Starter) PlaceRun(ev catalog.CatalogRevisionEvent, half Half, base func
 	return placed, first
 }
 
+// UnplacedRun is a run with a pipeline running and no steps placed yet, on the branch the run names.
+type UnplacedRun struct{ Run, Branch string }
+
+// UnplacedRuns are the runs with a pipeline running and no steps placed yet, by branch and id (design 47 G A-222): a placing
+// lost while the runner was down is taken up from these at its start.
+func (s *Starter) UnplacedRuns() ([]UnplacedRun, error) {
+	if s.Store == nil {
+		return nil, errors.New("the runner holds no pipelines to read a run's from")
+	}
+	pkeys, err := s.Store.Keys("*.*.*")
+	if err != nil {
+		return nil, fmt.Errorf("the pipelines are not listed: %w", err)
+	}
+	var out []UnplacedRun
+	for _, key := range pkeys {
+		if p, _, err := s.Store.Get(key); err == nil && p.Run != "" && p.Outcome == seat.PipelineOutcomeRunning && p.Revision == 0 {
+			if u := (UnplacedRun{Run: p.Run, Branch: p.Branch}); !slices.Contains(out, u) {
+				out = append(out, u)
+			}
+		}
+	}
+	slices.SortFunc(out, func(a, b UnplacedRun) int { return strings.Compare(a.Branch+"\n"+a.Run, b.Branch+"\n"+b.Run) })
+	return out, nil
+}
+
 // placeRunPipeline places one pipeline of the run, written and not yet placed (no revision), from the candidate's half.
 func (s *Starter) placeRunPipeline(key, id string, revision int64, half Half) (bool, error) {
 	for {

modules/mesh-delivery-pipeline/internal/pipeline/run_test.go

@@ -67,6 +67,10 @@ func TestRunOnRequest(t *testing.T) {
 	if len(said) != 1 || !strings.HasPrefix(said[0], seat.EventRunStarted+".app.") || !strings.HasSuffix(said[0], ".c1") {
 		t.Errorf("said %v, want run-started on the member's module, branch and commit", said)
 	}
+	// A-222: a restart finds the runs whose steps were never placed, by their branch.
+	if ids, err := s.UnplacedRuns(); err != nil || len(ids) != 1 || ids[0] != (UnplacedRun{Run: a.Run, Branch: "fix/1"}) {
+		t.Errorf("the unplaced runs = %v (%v), want %s on fix/1", ids, err, a.Run)
+	}
 	cand := entries
 	cand["app"] = catalog.CatalogEntry{Module: "app", Commit: "c1", Fp: "src2:new", Repository: "o/app", Artifacts: entries["app"].Artifacts}
 	cand["lib"] = catalog.CatalogEntry{Module: "lib", Commit: "c0", Fp: "src2:lib2", Artifacts: entries["app"].Artifacts}
@@ -97,6 +101,9 @@ func TestRunOnRequest(t *testing.T) {
 	if len(said) != 2 || !strings.HasPrefix(said[0], seat.EventPipelineStarted+".") || len(asks.ids) == 0 {
 		t.Errorf("placing said %v and asked %d steps", said, len(asks.ids))
 	}
+	if ids, _ := s.UnplacedRuns(); len(ids) != 0 {
+		t.Errorf("a placed run is still unplaced: %v", ids)
+	}
 	if n, _ := s.PlaceRun(ev, Half{Revision: 12, Entries: cand}, base); n != 0 {
 		t.Errorf("the revision heard again placed %d", n)
 	}