diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go index 81fb00b2..7f199d77 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate.go +++ b/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 +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go index 21efca52..26a92e9c 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/candidate_test.go +++ b/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]) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index b5b04915..8a388b9d 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/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, diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/run.go b/modules/mesh-delivery-pipeline/internal/pipeline/run.go index bd7a68f8..b0fbef9b 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/run.go +++ b/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 { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/run_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/run_test.go index b14ffe56..769311ca 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/run_test.go +++ b/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) }