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
This commit is contained in:
2026-11-05 18:38:11 +01:00
parent c2b8ac3370
commit bd5eb33e9e
14 changed files with 257 additions and 34 deletions
@@ -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)
}
}}
@@ -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
}
@@ -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"},
} {
@@ -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)
@@ -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 {
@@ -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 != "" {
@@ -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) {
@@ -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{})
@@ -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 " +
@@ -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.
@@ -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
@@ -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
}
}
@@ -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
@@ -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 }