From bd5eb33e9ed99604fbd214227c0ccb0af4a49c44 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 18:38:11 +0100 Subject: [PATCH] A failed recovery of the open tries is said and retried; every open try is listed; a silent node ends a try as error (issue 1921) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-checker/main.go | 2 +- .../internal/check/try.go | 34 ++++-- .../internal/check/try_test.go | 22 ++++ .../cmd/mesh-delivery-indexer-tools/main.go | 2 +- .../cmd/mesh-delivery-indexer/main.go | 8 +- .../cmd/mesh-delivery-indexer/takein.go | 6 +- .../internal/index/candidates.go | 9 +- .../internal/index/pending_test.go | 2 +- .../internal/index/pending_tools.go | 37 +++++- .../internal/index/previews_test.go | 49 +++++++- .../internal/index/takein.go | 6 +- .../internal/index/trycandidates.go | 108 +++++++++++++++++- .../mesh-delivery-indexer/manifest_test.go | 4 +- modules/mesh-delivery-indexer/module.json | 2 + 14 files changed, 257 insertions(+), 34 deletions(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 3f898c41..63bcfcae 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -194,7 +194,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // A whole try's start and end are said on the stream checks, for the indexer to write and delete the candidate of its group. Say: func(ctx context.Context, event string, body tryevent.Event) { data, _ := json.Marshal(body) - if _, err := js.Publish(ctx, eventSubject(event), data); err != nil { + if _, err := js.Publish(ctx, eventSubject(event), data, jetstream.WithRetryAttempts(10), jetstream.WithRetryWait(time.Second)); err != nil { logger.Printf("%s of %s could not be said: %v (the indexer finds the open tries at its start)", event, body.ID, err) } }} diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go index 51bfb026..e794ebc8 100644 --- a/modules/mesh-delivery-checker/internal/check/try.go +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -680,6 +680,11 @@ func (tr *Tryer) run(ctx context.Context, key string) { if IsLostAsk(v) || IsNotTaken(v) { t.Failed = []string{v.Failed.Words} } + case v.State == delivery.VerdictPass && cs.Try.Whole() && compositionMade(v.Composition) && strings.Contains(v.Composition.Words, delivery.ComposeNoAnswer): + // A node that gave no answer passes a group's check with a warning; a try that is to say the group composes says it + // did not, naming the node (ADR 0637 §3). + t.State = TryError + t.Failed = []string{"the composition was not made on every placed node: " + v.Composition.Words} case v.State == delivery.VerdictPass && cs.Try.Whole() && !compositionMade(v.Composition): // A whole try that judged the tests and not the manifests and the composition has not tried what its group's check // judges: it is no pass, and it says why (failure must be loud). It is an error, not the change's failure. @@ -711,8 +716,9 @@ func (tr *Tryer) say(ctx context.Context, event string, body tryevent.Event) { } } -// Open are the tries that wait or run now, newest scan first (TryScan): what checker_tries answers, so a part that missed an -// event (the indexer at its start) finds them without reading the checker's state. +// OpenTries are every try that waits or runs now: what checker_tries answers, so a part that missed an event (the indexer at +// its start) finds them without reading the checker's state. All the keys are read (an old open try is as open as a new one), in +// parallel. func OpenTries(store TryStore) ([]tryevent.Open, error) { keys, err := store.Keys() if err != nil { @@ -724,24 +730,30 @@ func OpenTries(store TryStore) ([]tryevent.Open, error) { } return keys[i] > keys[j] }) - if len(keys) > TryScan { - keys = keys[:TryScan] - } - var out []tryevent.Open - for _, k := range keys { - raw, err := store.Get(k) + found := make([]*tryevent.Open, len(keys)) + if err := parallel(len(keys), func(i int) error { + raw, err := store.Get(keys[i]) if err != nil || len(raw) == 0 { - continue + return nil } t, err := DecodeTry(raw) if err != nil || t.Over() { - continue + return nil } o := tryevent.Open{ID: t.ID, State: t.State, Whole: (&TryOptions{Level: t.Level, Run: t.Tests}).Whole()} for _, m := range t.MembersOf() { o.Members = append(o.Members, tryevent.Member{Repository: m.Repository, Head: m.Head, Trunk: m.Trunk}) } - out = append(out, o) + found[i] = &o + return nil + }); err != nil { + return nil, err + } + var out []tryevent.Open + for _, o := range found { + if o != nil { + out = append(out, *o) + } } return out, nil } diff --git a/modules/mesh-delivery-checker/internal/check/try_test.go b/modules/mesh-delivery-checker/internal/check/try_test.go index 1dc31e87..46133c57 100644 --- a/modules/mesh-delivery-checker/internal/check/try_test.go +++ b/modules/mesh-delivery-checker/internal/check/try_test.go @@ -284,6 +284,26 @@ func TestATryOfOneLevelAndSomeTestsRunsThemAloneAndKeepsNothing(t *testing.T) { } } +func TestATryOfOneLevelKeepsOnlyThatLevel(t *testing.T) { + ctx := context.Background() + kv := newMemKV() + id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Level: delivery.LevelUnit}, time.Now()) + b := newBench(t) + tr := &Tryer{Tries: kv, Judge: b.ex} + raw, _, _, _ := kv.Get(ctx, id) + tr.Put(ctx, id, raw) + tr.Wait() + keys, _ := b.kept.Keys(ctx) + if got := readTry(t, kv, id); got.State != TryPass || len(keys) == 0 { + t.Fatalf("try = %+v, kept %v", got, keys) + } + for _, k := range keys { + if !strings.HasPrefix(k, delivery.LevelUnit+".") { + t.Errorf("a try of the unit level kept %s", k) + } + } +} + func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) { ctx := context.Background() kv := newMemKV() @@ -307,6 +327,8 @@ func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) state, said string }{ {name: "the indexer wrote no candidate", cand: &fakeTryCandidates{err: errors.New("the indexer wrote no candidate within 5m0s")}, state: TryError, said: "the indexer wrote no candidate within 5m0s"}, + {name: "a node that gave no composition answer", cand: &fakeTryCandidates{}, comp: delivery.CompositionVerdict{State: delivery.CompositionPass, SDK: "x", Warnings: []string{"g14: " + delivery.ComposeNoAnswer}}, + state: TryError, said: "g14: " + delivery.ComposeNoAnswer}, {name: "a composition refused on a node", cand: &fakeTryCandidates{}, comp: delivery.CompositionVerdict{State: delivery.CompositionFail, Machine: "g14", Words: "not composed on g14: web: needs a state mail does not keep"}, state: TryFail, said: "not composed on g14: web: needs a state mail does not keep"}, } { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index 754f3b94..2c98c41a 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -217,7 +217,7 @@ func main() { ask := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second) tools := append(Tools(ask, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...) tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, ask, time.Now)...) - tools = append(tools, index.TryCandidateTools(keptRead{stdio.State("candidates")}, ask, time.Now)...) + tools = append(tools, index.TryCandidateTools(keptRead{stdio.State("candidates")}, keptRead{stdio.State(index.TryRecoveryState)}, ask, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 86ba8676..9bf29c33 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -343,7 +343,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro oldSections := process.AwaitBucket(ctx, js, conn, index.OldSectionsBucket, "the count of the old sections is not kept, so the judge refuses no use of one", logger.Printf) // The module graph (ADR 0603 §2): a state this delivery brings, bound when it appears; until then no graph is written. graph := process.AwaitBucket(ctx, js, conn, graph.Bucket, "no module graph is written", logger.Printf) - ix := &index.Indexer{OldSections: awaitedStore{oldSections}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, + tryRecovery := process.AwaitBucket(ctx, js, conn, index.TryRecoveryBucket, "how the recovery of the open tries went is not kept", logger.Printf) + ix := &index.Indexer{TryRecovery: awaitedStore{tryRecovery}, OldSections: awaitedStore{oldSections}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(js, cred, logger.Printf), SayAccepted: sayAccepted(js, cred, logger.Printf), Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), @@ -511,9 +512,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // The open tries are found again through the checker's tool: a try event missed while the indexer was down would leave a // try with no candidate, or a candidate with no try. Not fatal: the events come, and a preview written writes the candidates. - if err := ix.RecoverTries(ctx); err != nil { - logger.Printf("%v", err) - } + // A recovery that fails is said (journal, and indexer_try_candidates) and the events are heard all the same. + _ = ix.RecoverTries(ctx) if perr == nil { pc, err := process.HearEvents(pcons, func(m jetstream.Msg) { previews.heardChecks(m) }, logger.Printf) if err != nil { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index 7a40aabb..fc272890 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -20,6 +20,7 @@ type takerIn interface { DropPending(ctx context.Context, key, by, why string) index.TakenIn TakeInMerge(ctx context.Context, repository, merge string, seq uint64, by, why string) index.TakenIn DropTryCandidate(ctx context.Context, id, by, why string) index.TakenIn + RecoverTriesByHand(ctx context.Context, by, why string) index.TakenIn } // takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the @@ -38,7 +39,10 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream return } asked, r := "take-in", index.TakenIn{} - if a.TryCandidate != "" && a.Drop { + if a.RecoverTries { + asked, r = "recovery of the tries", ix.RecoverTriesByHand(ctx, a.By, a.Why) + a.Module = "tries" + } else if a.TryCandidate != "" && a.Drop { asked, r = "drop of a try's candidate", ix.DropTryCandidate(ctx, a.TryCandidate, a.By, a.Why) a.Module = a.TryCandidate } else if a.Merge != "" { diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index b9ccd9e7..bbae2758 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -136,12 +136,15 @@ type Indexer struct { // Pending is the state of merges held for the indexer their revision names (pending.go, novox/hq ADR 0633 §1); nil // takes every merge in at once, as before the take-in step. Pending Store - // SayTakeIn publishes a take-in event on the indexer's own stream take-in (ADR 0633 §5); nil says none. + // SayTakeIn publishes a take-in event (take-in.pending, take-in.taken; ADR 0633 §5); nil says none. SayTakeIn func(ctx context.Context, t takein.TakeIn) error // OpenTries lists the tries that wait or run, from the checker's tool (checker_tries); nil recovers none (trycandidates.go). OpenTries func(ctx context.Context) ([]tryevent.Open, error) - tries map[string][]TryMember // the whole tries that wait or run, by try id (trycandidates.go); guarded by mu - pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go) + // TryRecovery keeps how the last recovery of the open tries went (trycandidates.go); nil keeps none. + TryRecovery Store + recoveryFailed atomic.Bool // the last recovery of the open tries failed (trycandidates.go) + tries map[string][]TryMember // the whole tries that wait or run, by try id (trycandidates.go); guarded by mu + pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go) } func (ix *Indexer) logf(format string, args ...any) { diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index f87f13fd..b7c1ba71 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -578,7 +578,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) named := map[string]bool{} for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk", - "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", "ReadFile", "OpenTries"} { + "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", "ReadFile", "OpenTries", "TryRecovery"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go index b49a883f..e8811e78 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -100,7 +100,7 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, // TryCandidateTools are the tools of a group try's candidate (issue 1921, trycandidates.go): indexer_try_candidates reads the // candidates the indexer holds for tries (the try, its heads, each member's modules and whether each has its fingerprint), // and indexer_drop_try_candidate drops one by an ask in takeins. candidates is the candidates state as the tools bundle reads it. -func TryCandidateTools(candidates Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool { +func TryCandidateTools(candidates, recovery Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool { return []stdio.Tool{ {Name: TryCandidatesTool, Description: "The candidates the indexer holds for group tries (a try of several repositories at exact commits, mesh-delivery-checker try): " + @@ -127,7 +127,40 @@ func TryCandidateTools(candidates Store, ask func(key string, value []byte) (Tak out = append(out, c) } } - return map[string]any{"tries": out}, nil + // How the recovery of the open tries went: loud when it is not done (since when, why). + said := map[string]any{"done": false, "why": "no recovery has been written yet"} + if v, _, found, err := recovery.Get(ctx, TryRecoveryKey); err == nil && found { + var r TryRecovery + if json.Unmarshal(v, &r) == nil { + said = map[string]any{"done": r.Done} + if !r.Done { + said["why"], said["since"] = r.Why, r.Since + said["says"] = "recovery not done, since " + r.Since.Format(time.RFC3339) + ": " + r.Why + } + } + } + return map[string]any{"tries": out, "recovery": said}, nil + }}, + {Name: RecoverTriesTool, + Description: "Run the recovery of the open tries now: the indexer lists them through the checker's checker_tries, writes the candidate of each " + + "whole try that waits or runs and deletes the candidate of a try no longer listed. It runs by itself at the indexer's start; when it failed " + + "(indexer_try_candidates says recovery not done, since when and why) this runs it again. A hand act: who and why are said in the journal.", + Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{ + "why": map[string]any{"type": "string", "description": "why it is run now, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + why, _ := args["why"].(string) + if strings.TrimSpace(why) == "" { + return nil, fmt.Errorf("say why: a hand act is said in the journal") + } + data, err := json.Marshal(TakeInAsk{V: 1, RecoverTries: true, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()}) + if err != nil { + return nil, err + } + a, err := ask("tries.recover."+now().UTC().Format("20060102T150405.000000000"), data) + if err != nil { + return nil, err + } + return a.Result, nil }}, {Name: DropTryCandidateTool, Description: "Drop a group try's candidate by hand (the indexer deletes it itself when the try ends): for a candidate left for a try that " + diff --git a/modules/mesh-delivery-indexer/internal/index/previews_test.go b/modules/mesh-delivery-indexer/internal/index/previews_test.go index 701fdc8f..6dfe4310 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews_test.go +++ b/modules/mesh-delivery-indexer/internal/index/previews_test.go @@ -3,6 +3,7 @@ package index import ( "context" "encoding/json" + "errors" "os" "slices" "strings" @@ -285,14 +286,16 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW a.Result = &r return a, err } - tools := TryCandidateTools(cands, asks, time.Now) + recovery := newMem() + ix.TryRecovery = recovery + tools := TryCandidateTools(cands, recovery, asks, time.Now) if out, err := tools[0].Run(map[string]any{}); err != nil || len(out.(map[string]any)["tries"].([]delivery.Candidate)) != 1 { t.Errorf("%s = %v, %v, want the try's candidate", tools[0].Name, out, err) } - if _, err := tools[1].Run(map[string]any{"try": "feat/x", "why": "w"}); err == nil { + if _, err := tools[2].Run(map[string]any{"try": "feat/x", "why": "w"}); err == nil { t.Error("a branch was dropped as a try") } - if out, err := tools[1].Run(map[string]any{"try": "try-7", "why": "by hand"}); err != nil || !out.(*TakenIn).Stored || len(held()) != 0 { + if out, err := tools[2].Run(map[string]any{"try": "try-7", "why": "by hand"}); err != nil || !out.(*TakenIn).Stored || len(held()) != 0 { t.Errorf("drop = %v, %v, candidates %v", out, err, held()) } // The start recovered from the checker's tool: an open whole try is written again, one the checker no longer @@ -308,6 +311,46 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW if err := ix.RecoverTries(ctx); err != nil || len(held()) != 0 { t.Errorf("recovered with no open try: %v, candidates %v, want none", err, held()) } + // A recovery whose call fails is loud and deletes nothing: the candidate stays, indexer_try_candidates says recovery not + // done since when and why, and the next try event runs the recovery again, which clears it; the tool runs it by hand. + if err := ix.StartTry(ctx, "try-7", members); err != nil || len(held()) != 1 { + t.Fatalf("started: %v, candidates %v", err, held()) + } + ix.OpenTries = func(context.Context) ([]tryevent.Open, error) { + return nil, errors.New("checker_tries does not answer") + } + if err := ix.RecoverTries(ctx); err == nil || len(held()) != 1 { + t.Fatalf("a failed call: %v, candidates %v, want the error and the candidate kept", err, held()) + } + if out, err := tools[0].Run(map[string]any{}); err != nil || !strings.Contains(out.(map[string]any)["recovery"].(map[string]any)["says"].(string), "recovery not done, since") || + !strings.Contains(out.(map[string]any)["recovery"].(map[string]any)["says"].(string), "checker_tries does not answer") { + t.Errorf("%s = %v, %v, want recovery not done with since and why", tools[0].Name, out, err) + } + ix.OpenTries = func(context.Context) ([]tryevent.Open, error) { return nil, nil } + if err := ix.StartTry(ctx, "try-11", members); err != nil || len(held()) != 1 || held()[0] != "try-11" { + t.Fatalf("a try event after a failed recovery: %v, candidates %v, want the recovery run (try-7 not listed, gone) and try-11 written", err, held()) + } + if out, _ := tools[0].Run(map[string]any{}); out.(map[string]any)["recovery"].(map[string]any)["done"] != true { + t.Errorf("recovery still said not done after a success: %v", out) + } + ix.OpenTries = func(context.Context) ([]tryevent.Open, error) { return nil, errors.New("down") } + recoverAsk := func(key string, value []byte) (TakeInAsk, error) { + a, err := DecodeTakeIn(value) + r := ix.RecoverTriesByHand(ctx, a.By, a.Why) + a.Result = &r + return a, err + } + byHand := TryCandidateTools(cands, recovery, recoverAsk, time.Now) + if out, err := byHand[1].Run(map[string]any{"why": "the checker was replaced"}); err != nil || out.(*TakenIn).Stored || !strings.Contains(out.(*TakenIn).Answer, "recovery not done") { + t.Errorf("%s with the checker down = %v, %v", byHand[1].Name, out, err) + } + ix.OpenTries = func(context.Context) ([]tryevent.Open, error) { return nil, nil } + if out, err := byHand[1].Run(map[string]any{"why": "the checker is back"}); err != nil || !out.(*TakenIn).Stored || len(held()) != 0 { + t.Errorf("%s = %v, %v, candidates %v, want it recovered (try-11 not listed)", byHand[1].Name, out, err, held()) + } + if err := ix.StartTry(ctx, "try-7", members); err != nil { + t.Fatal(err) + } // ADR 0638 §3, §4: the consumer exists before the call, so what happens between the call and the hearing is heard after // it. A try the call listed and whose start is then heard gets one candidate; a try started after the list gets one; a // try that ended during the call, listed as open, leaves none once its end is heard. diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 55ec09a2..e6318147 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -44,7 +44,9 @@ type TakeInAsk struct { MergeRepository string `json:"merge-repository,omitempty"` Sequence uint64 `json:"sequence,omitempty"` // TryCandidate asks the candidate of this try's group to be dropped (indexer_drop_try_candidate), trycandidates.go. - TryCandidate string `json:"try-candidate,omitempty"` + TryCandidate string `json:"try-candidate,omitempty"` + // RecoverTries asks the open tries' candidates to be recovered now (indexer_recover_tries). + RecoverTries bool `json:"recover-tries,omitempty"` Drop bool `json:"drop,omitempty"` By string `json:"by,omitempty"` At time.Time `json:"at"` @@ -64,7 +66,7 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) { if a.V != 1 { return a, fmt.Errorf("a takeins value of version %d, not 1", a.V) } - if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" { + if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries { return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try") } return a, nil diff --git a/modules/mesh-delivery-indexer/internal/index/trycandidates.go b/modules/mesh-delivery-indexer/internal/index/trycandidates.go index ae2b0fb7..75a71b51 100644 --- a/modules/mesh-delivery-indexer/internal/index/trycandidates.go +++ b/modules/mesh-delivery-indexer/internal/index/trycandidates.go @@ -11,22 +11,41 @@ package index import ( "context" + "encoding/json" + "errors" "fmt" "path" "regexp" "slices" "strings" + "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-sdk/go/delivery" ) -// TryCandidateTools: the tools that read and drop a try's candidate. +// TryCandidateTools: the tools that read, drop and recover a try's candidate. const ( TryCandidatesTool = "indexer_try_candidates" DropTryCandidateTool = "indexer_drop_try_candidate" + RecoverTriesTool = "indexer_recover_tries" + // TryRecoveryState is the indexer's state of how the last recovery of the open tries went: the bucket + // mesh-delivery-indexer_try-recovery, one key (TryRecoveryKey). + TryRecoveryState = "try-recovery" + TryRecoveryBucket = delivery.IndexerModule + "_" + TryRecoveryState + TryRecoveryKey = "state" ) +// TryRecovery is the value of the try-recovery state: whether the candidates of the open tries were recovered at the +// indexer's start or by hand, and when it began to fail, why. +type TryRecovery struct { + V int `json:"v"` + Done bool `json:"done"` + Since time.Time `json:"since,omitzero"` + Why string `json:"why,omitempty"` + At time.Time `json:"at"` +} + var tryGroupOK = regexp.MustCompile(`^try-[0-9]{1,20}$`) // IsTryGroup says a candidates key is the group of a try. @@ -38,6 +57,18 @@ type TryMember struct{ Repository, Head, Trunk string } // StartTry takes a whole try in, as the checker said it started (tryevent.Started): its members at their commits are kept, and // its candidate is written once every member's preview is held. The checker's state is never read (ADR 0596). func (ix *Indexer) StartTry(ctx context.Context, id string, members []TryMember) error { + ix.retryRecovery(ctx) + return ix.startTry(ctx, id, members) +} + +// retryRecovery runs the recovery again on a try event when the last one failed: the failure stays said until one succeeds. +func (ix *Indexer) retryRecovery(ctx context.Context) { + if ix.recoveryFailed.Load() && ix.OpenTries != nil { + _ = ix.RecoverTries(ctx) + } +} + +func (ix *Indexer) startTry(ctx context.Context, id string, members []TryMember) error { if !IsTryGroup(id) || len(members) == 0 { return fmt.Errorf("%q with %d members is no try to start", id, len(members)) } @@ -52,6 +83,11 @@ func (ix *Indexer) StartTry(ctx context.Context, id string, members []TryMember) // EndTry takes a try out, as the checker said it ended (tryevent.Ended), and deletes its candidate. func (ix *Indexer) EndTry(ctx context.Context, id string) error { + ix.retryRecovery(ctx) + return ix.endTry(ctx, id) +} + +func (ix *Indexer) endTry(ctx context.Context, id string) error { ix.mu.Lock() delete(ix.tries, id) ix.mu.Unlock() @@ -64,10 +100,76 @@ func (ix *Indexer) EndTry(ctx context.Context, id string) error { // RecoverTries is, at the indexer's start, what the events it missed would have said: the candidates of the tries that wait // or run now are written, and the candidate of a try the checker no longer lists is deleted. The checker's tool lists them // (OpenTries); its bucket is not read. Nothing to recover from when no part is wired to ask. +// +// A recovery that fails is said loudly and kept (the try-recovery state, read by indexer_try_candidates) until one succeeds: +// the events are still heard after it, and indexer_recover_tries runs it again by hand. func (ix *Indexer) RecoverTries(ctx context.Context) error { if ix.OpenTries == nil || ix.Store == nil { return nil } + err := ix.recoverTries(ctx) + ix.recoveryFailed.Store(err != nil) + ix.noteRecovery(ctx, err) + return err +} + +// noteRecovery writes how a recovery went: failing keeps the time it began to fail, and the journal says it each time. +func (ix *Indexer) noteRecovery(ctx context.Context, err error) { + now := ix.now() + r := TryRecovery{V: 1, Done: err == nil, At: now.UTC()} + if err != nil { + r.Why, r.Since = err.Error(), now.UTC() + ix.logf("RECOVERY OF THE TRIES NOT DONE: %v (the try candidates may be stale or missing until indexer_recover_tries or the next start succeeds)", err) + } + if ix.TryRecovery == nil { + return + } + for range writeTries { + cur, rev, found, gerr := ix.TryRecovery.Get(ctx, TryRecoveryKey) + if gerr != nil { + ix.logf("the try-recovery state cannot be read: %v", gerr) + return + } + if err != nil && found { + var was TryRecovery + if json.Unmarshal(cur, &was) == nil && !was.Done && !was.Since.IsZero() { + r.Since = was.Since + } + } + data, _ := json.Marshal(r) + var werr error + if found { + werr = ix.TryRecovery.Update(ctx, TryRecoveryKey, data, rev) + } else { + werr = ix.TryRecovery.Create(ctx, TryRecoveryKey, data) + } + if werr == nil { + return + } + if !errors.Is(werr, ErrConflict) { + ix.logf("the try-recovery state cannot be written: %v", werr) + return + } + } +} + +// RecoverTriesByHand runs the recovery now (indexer_recover_tries), a hand act said in the journal. +func (ix *Indexer) RecoverTriesByHand(ctx context.Context, by, why string) TakenIn { + r := TakenIn{Module: "tries"} + ix.logf("recovery of the tries asked by %s (%s)", by, why) + if ix.OpenTries == nil { + r.Answer = "the indexer is wired to list no open try: nothing to recover from" + return r + } + if err := ix.RecoverTries(ctx); err != nil { + r.Answer = "recovery not done: " + err.Error() + return r + } + r.Stored, r.Answer = true, "the candidates of the open tries are recovered" + return r +} + +func (ix *Indexer) recoverTries(ctx context.Context) error { open, err := ix.OpenTries(ctx) if err != nil { return fmt.Errorf("the open tries cannot be listed, so a missed try event is not recovered: %w", err) @@ -82,7 +184,7 @@ func (ix *Indexer) RecoverTries(ctx context.Context) error { for _, m := range o.Members { ms = append(ms, TryMember{m.Repository, m.Head, m.Trunk}) } - if err := ix.StartTry(ctx, o.ID, ms); err != nil { + if err := ix.startTry(ctx, o.ID, ms); err != nil { return err } } @@ -92,7 +194,7 @@ func (ix *Indexer) RecoverTries(ctx context.Context) error { } for _, k := range keys { if IsTryGroup(k) && !live[k] { - if err := ix.EndTry(ctx, k); err != nil { + if err := ix.endTry(ctx, k); err != nil { return err } } diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index db8524e0..731bfa52 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -108,7 +108,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build t.Errorf("emits = %v, want manifest-refused, manifest-accepted and the two take-in events", m.Emits) } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. - if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, + if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, {Name: graph.State, History: graph.History}, {Name: index.PendingState, History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } @@ -137,7 +137,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // (indexer_judge; issue 774, clause 3). 1966/AC11: indexer_old_sections reads the count of the old sections (state oldsections). // 1585/T12, T13 (ADR 0633 §5): indexer_pending reads the pending state; indexer_take_in_pending and indexer_drop_pending // ask through takeins; indexer_take_in_merge takes one merge in, its message asked of the forge bridge (issue 1585). - if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TryCandidatesTool, index.DropTryCandidateTool, index.TakeInMergeTool}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TryCandidatesTool, index.DropTryCandidateTool, index.RecoverTriesTool, index.TakeInMergeTool}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } // ADR 0604 §2: it holds the seat mesh-catalog, serving its four verbs from its tools bundle (which stays without a diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 4b39a52c..eabac22c 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -24,6 +24,7 @@ "indexer_drop_pending", "indexer_try_candidates", "indexer_drop_try_candidate", + "indexer_recover_tries", "indexer_take_in_merge" ], "consumes": [ @@ -45,6 +46,7 @@ { "name": "takeins", "history": 1 }, { "name": "verdicts", "history": 5 }, { "name": "oldsections", "history": 1 }, + { "name": "try-recovery", "history": 1 }, { "name": "packages", "history": 5 }, { "name": "graph", "history": 5 }, { "name": "pending", "history": 1 }