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 d9cb1aa0..baa62b17 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 @@ -104,12 +104,14 @@ func Tools(ask asker, now func() time.Time, read stdio.KeptState, counts func() Description: "Judge the manifests again now, as the indexer does at its start and after each merge, and answer how the module's " + "verdict stands (waits, refused or accepted, and why). For a manifest that waits for a judge that knows its fields, or " + "after a manifest was mended where the indexer has not heard it. A waiting manifest becomes refused here only when the " + - "running indexer is the newest build of itself the trunk holds. indexer_verdicts reads the result.", + "running indexer is the newest build of itself the trunk holds. With a digest it judges that one manifest of the " + + "module (a candidate's, or one an older judge refused) in the catalogue the trunk holds. indexer_verdicts reads the result.", Input: map[string]any{"type": "object", "required": []string{"module"}, "properties": map[string]any{ "module": map[string]any{"type": "string", "description": "the module's name, as the catalogue registers it"}, + "digest": map[string]any{"type": "string", "description": "one manifest digest of the module to judge (indexer_verdicts names it); none judges every manifest the trunk holds"}, "why": map[string]any{"type": "string", "description": "why it is judged again now, in plain words"}}}, Run: func(args map[string]any) (any, error) { - key, value, err := index.NewJudgeAgain(str(args, "module"), str(args, "why"), stdio.CallerOf(args), now()) + key, value, err := index.NewJudgeAgain(str(args, "module"), str(args, "digest"), str(args, "why"), stdio.CallerOf(args), now()) if err != nil { return nil, err } 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 a32aea21..7a98edb7 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -15,7 +15,7 @@ import ( type takerIn interface { TakeIn(ctx context.Context, module string) index.TakenIn TakeInRepository(ctx context.Context, repository string) index.TakenIn - JudgeAgain(ctx context.Context, module string) index.TakenIn + JudgeAgain(ctx context.Context, module, digest 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 +38,7 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository) a.Module = a.Repository } else if a.Judge { - asked, r = "judging again", ix.JudgeAgain(ctx, a.Module) + asked, r = "judging again", ix.JudgeAgain(ctx, a.Module, a.Digest) } else { r = ix.TakeIn(ctx, a.Module) } diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index b1620ac0..85d1c415 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -186,6 +186,20 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f delivery.Forge) (Applied, if held, err = ix.heldModules(ctx); err != nil { return applied, err } + // Each manifest the pull carries is judged by this judge, unless it holds a verdict already that this judge made or + // that accepts it: a refusal an older judge made never stands for a candidate (or a digest back on the trunk). + over := map[string]string{} + for _, digest := range digests { + if ix.Manifests == nil { + break + } + if value, _, found, err := ix.Manifests.Get(ctx, digest); err == nil && found { + if kept, err := delivery.DecodeManifest(value); err == nil { + over[kept.Module] = digest + } + } + } + ix.judgeDigests(ctx, over, fmt.Sprintf("%s#%d", f.Repository, f.Number)) if err := ix.configCandidateOfPull(ctx, f, held); err != nil { if !errors.Is(err, ErrUnreadable) { return applied, err diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 10866eb0..579ba62b 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -31,7 +31,9 @@ type TakeInAsk struct { Repository string `json:"repository,omitempty"` Why string `json:"why,omitempty"` // Judge asks the manifests to be judged again now (indexer_judge_again) instead of the module to be taken in. - Judge bool `json:"judge,omitempty"` + Judge bool `json:"judge,omitempty"` + // Digest names, with Judge, the one manifest digest of the module to judge (indexer_judge with a digest). + Digest string `json:"digest,omitempty"` By string `json:"by,omitempty"` At time.Time `json:"at"` Result *TakenIn `json:"result,omitempty"` diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts.go b/modules/mesh-delivery-indexer/internal/index/verdicts.go index 50b72e2d..1263a8e1 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts.go @@ -49,6 +49,14 @@ type Judged struct { // watchman raises as manifest-outgrown. A manifest in fresh (the digests a merge just took in) with no verdict yet, and // one already refused, is judged in full. func (ix *Indexer) JudgeAll(ctx context.Context, fresh map[string]bool) (Judged, error) { + return ix.judgeOver(ctx, fresh, nil) +} + +// judgeOver is JudgeAll; with over (module to manifest digest) it judges those digests alone, each in the catalogue the +// modules state holds with its module's manifest replaced by it (a module the state does not hold added), and writes +// their verdicts and nothing else: no event, no count of the old sections, and no module's value but one that names the +// digest judged (carryVerdict). A digest of over with no verdict yet is judged in full, as a fresh one. +func (ix *Indexer) judgeOver(ctx context.Context, fresh map[string]bool, over map[string]string) (Judged, error) { // A merge, the start pass and indexer_judge may all judge at once: one pass at a time, so none writes a verdict from // modules another is rewriting. ix.judgeMu.Lock() @@ -61,6 +69,15 @@ func (ix *Indexer) JudgeAll(ctx context.Context, fresh map[string]bool) (Judged, if err != nil { return out, err } + if over != nil { + fresh = map[string]bool{} + for module, digest := range over { + m := held[module] + m.Manifest = digest + held[module] = m + fresh[digest] = true + } + } raws := map[string][]byte{} digests := map[string]string{} for _, name := range slices.Sorted(maps.Keys(held)) { @@ -99,12 +116,12 @@ func (ix *Indexer) JudgeAll(ctx context.Context, fresh map[string]bool) (Judged, // release's Approve; ADR 0578 §1): it is refused only when this program is the newest build of the indexer the trunk // holds, so no judge that could know the field is still to come. newest := ix.ownBuildIsNewest(held) - id := ix.JudgeID - if id == "" { - id = judge.Identity() - } + id := ix.judgeID() var accepted []catalogue.Manifest for _, name := range slices.Sorted(maps.Keys(digests)) { + if _, judged := over[name]; over != nil && !judged { + continue + } out.Judged++ v := delivery.ManifestVerdict{Module: name, State: delivery.VerdictAccepted, Judge: id, At: ix.now()} if fields := waits[name]; len(fields) > 0 { @@ -143,7 +160,10 @@ func (ix *Indexer) JudgeAll(ctx context.Context, fresh map[string]bool) (Judged, ix.logf("judge: the verdict on %s (digest %s) is not written: %v", name, digests[name], err) continue } - if wrote { + if wrote && over != nil { + out.Written++ + ix.logf("judge: %s judged again by %s: %s", digests[name], id, v.State) + } else if wrote { out.Written++ if v.Waits() { ix.logf("judge: %s", v.Why()) @@ -166,6 +186,9 @@ func (ix *Indexer) JudgeAll(ctx context.Context, fresh map[string]bool) (Judged, ix.logf("judge: modules.%s does not carry its verdict: %v", name, err) } } + if over != nil { + return out, nil + } if err := ix.keepOldSections(ctx, judge.CountOldSections(accepted)); err != nil { ix.logf("judge: the count of the old sections is not kept: %v", err) } @@ -173,14 +196,86 @@ func (ix *Indexer) JudgeAll(ctx context.Context, fresh map[string]bool) (Judged, } // JudgeAtStart is JudgeAll of every manifest held, none fresh, as the indexer starts (a judge that changed judges every -// one again); one that fails is judged again at the next message or source revision heard (kickJudge). +// one again); one that fails is judged again at the next message or source revision heard (kickJudge). Then every other +// digest of the verdicts state that another judge refused or held waiting (one of a candidate, or one back on the trunk +// later) is judged again by this judge (judgeStale), so a refusal an older judge made never stands for good. func (ix *Indexer) JudgeAtStart(ctx context.Context) (Judged, error) { ix.learnOwnBuild(ctx) j, err := ix.JudgeAll(ctx, nil) ix.judgePending.Store(err != nil) + if err == nil { + ix.judgeStale(ctx) + } return j, err } +// judgeID is the judge a verdict of this program names. +func (ix *Indexer) judgeID() string { + if ix.JudgeID != "" { + return ix.JudgeID + } + return judge.Identity() +} + +// staleVerdict says whether the digest's verdict wants this judge's judgement: there is none, or another judge refused +// it or held it waiting. An accepted verdict stays accepted under any later judge (ADR 0579 §1), so judging it again +// changes nothing. +func (ix *Indexer) staleVerdict(ctx context.Context, digest string) bool { + value, _, found, err := ix.Verdicts.Get(ctx, digest) + if err != nil { + return false // unread: left, said by the next pass that reads it + } + if !found { + return true + } + v, err := delivery.DecodeManifestVerdict(value) + return err != nil || (v.Held() && v.Judge != ix.judgeID()) +} + +// judgeDigests judges each digest of over (module to digest) whose verdict is stale (staleVerdict), one at a time in the +// catalogue the modules state holds; a pass that fails is said and the others go on. +func (ix *Indexer) judgeDigests(ctx context.Context, over map[string]string, why string) { + if ix.Verdicts == nil || ix.Manifests == nil || ix.Modules == nil { + return + } + for _, module := range slices.Sorted(maps.Keys(over)) { + if !ix.staleVerdict(ctx, over[module]) { + continue + } + if _, err := ix.judgeOver(ctx, nil, map[string]string{module: over[module]}); err != nil { + ix.logf("judge: the manifest %s of %s (%s) is not judged: %v", over[module], module, why, err) + } + } +} + +// judgeStale judges again every digest of the verdicts state another judge refused or held waiting, each as a manifest +// of the module its verdict names (judgeDigests). Never on a timer: at the start pass. +func (ix *Indexer) judgeStale(ctx context.Context) { + if ix.Verdicts == nil { + return + } + keys, err := ix.Verdicts.Keys(ctx) + if err != nil { + ix.logf("judge: the verdicts other judges made are not judged again: listing the verdicts: %v", err) + return + } + byModule := map[string][]string{} + for _, digest := range keys { + value, _, found, err := ix.Verdicts.Get(ctx, digest) + if err != nil || !found { + continue + } + if v, err := delivery.DecodeManifestVerdict(value); err == nil && v.Held() && v.Judge != ix.judgeID() { + byModule[v.Module] = append(byModule[v.Module], digest) + } + } + for _, module := range slices.Sorted(maps.Keys(byModule)) { + for _, digest := range byModule[module] { + ix.judgeDigests(ctx, map[string]string{module: digest}, "made by another judge") + } + } +} + // kickJudge judges again a start pass that failed, when the indexer hears something: never on a timer. func (ix *Indexer) kickJudge(ctx context.Context) { if !ix.judgePending.Load() { @@ -374,26 +469,55 @@ func ListVerdicts(keys []string, get func(key string) ([]byte, error), module st // JudgeAgain judges the manifests again now, as at the start, and answers how the module's stands: the tool // indexer_judge. A module the modules state does not hold, or whose manifest is not in the manifests state, is -// answered so. -func (ix *Indexer) JudgeAgain(ctx context.Context, module string) TakenIn { +// answered so. With a digest it judges that manifest alone, in the catalogue the trunk holds (judgeOver), whoever made +// its verdict: a candidate's, or one the trunk names again, refused by an older judge. +func (ix *Indexer) JudgeAgain(ctx context.Context, module, digest string) TakenIn { out := TakenIn{Module: module} held, err := ix.heldModules(ctx) if err != nil { out.Answer = "the modules state cannot be read: " + err.Error() return out } + // The build this program runs is asked again, so a start pass that missed the engine's answer is recovered without a restart. + ix.learnOwnBuild(ctx) + if digest != "" { + // One digest (a candidate's, or one the trunk no longer names), judged by this judge in the catalogue the trunk holds. + value, _, found, err := ix.Manifests.Get(ctx, digest) + kept, derr := delivery.DecodeManifest(value) + switch { + case err != nil || !found: + out.Answer = "the indexer keeps no manifest of digest " + digest + return out + case derr != nil: + out.Answer = "the manifest of digest " + digest + " cannot be read: " + derr.Error() + return out + case module != "" && kept.Module != module: + out.Answer = "the manifest of digest " + digest + " is " + kept.Module + "'s, not " + module + "'s" + return out + } + out.Module = kept.Module + if _, err := ix.judgeOver(ctx, nil, map[string]string{kept.Module: digest}); err != nil { + out.Answer = "the manifest could not be judged: " + err.Error() + return out + } + return ix.answerVerdict(ctx, out, digest) + } m, ok := held[module] if !ok || m.Manifest == "" { out.Answer = "the indexer holds no manifest of " + module return out } - // The build this program runs is asked again, so a start pass that missed the engine's answer is recovered without a restart. - ix.learnOwnBuild(ctx) if _, err := ix.JudgeAll(ctx, nil); err != nil { out.Answer = "the manifests could not be judged: " + err.Error() return out } - value, _, found, err := ix.Verdicts.Get(ctx, m.Manifest) + return ix.answerVerdict(ctx, out, m.Manifest) +} + +// answerVerdict is out answering how the verdict under digest stands. +func (ix *Indexer) answerVerdict(ctx context.Context, out TakenIn, digest string) TakenIn { + module := out.Module + value, _, found, err := ix.Verdicts.Get(ctx, digest) if err != nil || !found { out.Answer = "judged, but the verdict on " + module + " cannot be read" return out @@ -411,8 +535,14 @@ func (ix *Indexer) JudgeAgain(ctx context.Context, module string) TakenIn { return out } -// NewJudgeAgain is the value indexer_judge writes for a module: its key in the takeins state and the value. -func NewJudgeAgain(module, why, by string, at time.Time) (string, []byte, error) { +// NewJudgeAgain is the value indexer_judge writes for a module, and one digest of its manifest when given: its key in +// the takeins state and the value. +func NewJudgeAgain(module, digest, why, by string, at time.Time) (string, []byte, error) { + if digest != "" { + if _, err := delivery.ManifestsKey(digest); err != nil { + return "", nil, err + } + } key, data, err := NewTakeIn(module, why, by, at) if err != nil { return "", nil, err @@ -421,7 +551,7 @@ func NewJudgeAgain(module, why, by string, at time.Time) (string, []byte, error) if err := json.Unmarshal(data, &a); err != nil { return "", nil, err } - a.Judge = true + a.Judge, a.Digest = true, digest data, err = json.Marshal(a) return judgeAgainPrefix + key, data, err } diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts_test.go b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go index c19f5143..b8ce4a49 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts_test.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go @@ -276,25 +276,66 @@ func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) { // the indexer is the newer one, so the field waits; asked and answered with that same build it is refused). ix.ownFP = "" ix.OwnBuild = func(context.Context) (string, error) { return "src2:" + strings.Repeat("a", 64), nil } - if got := ix.JudgeAgain(ctx, "later"); !got.Stored || !strings.Contains(got.Answer, "refused") || ix.ownFP == "" { + if got := ix.JudgeAgain(ctx, "later", ""); !got.Stored || !strings.Contains(got.Answer, "refused") || ix.ownFP == "" { t.Fatalf("indexer_judge did not learn the running build: %+v", got) } ix.ownFP = ownFP ix.OwnBuild = nil - if got := ix.JudgeAgain(ctx, "later"); !got.Stored || !strings.Contains(got.Answer, "waits") || !strings.Contains(got.Answer, "a-field-from-later") { + if got := ix.JudgeAgain(ctx, "later", ""); !got.Stored || !strings.Contains(got.Answer, "waits") || !strings.Contains(got.Answer, "a-field-from-later") { t.Fatalf("indexer_judge: %+v", got) } if rows, _ := ListVerdicts(keys, get, "later", false); len(rows) != 1 { t.Fatalf("indexer_verdicts of later after: %+v", rows) } - if got := ix.JudgeAgain(ctx, "nobody"); got.Stored || !strings.Contains(got.Answer, "holds no manifest") { + if got := ix.JudgeAgain(ctx, "nobody", ""); got.Stored || !strings.Contains(got.Answer, "holds no manifest") { t.Fatalf("indexer_judge of nobody: %+v", got) } - if key, value, err := NewJudgeAgain("later", "why", "me", time.Now()); err != nil || key != "judge.later" { + if key, value, err := NewJudgeAgain("later", "", "why", "me", time.Now()); err != nil || key != "judge.later" { t.Fatalf("the ask: %q %v", key, err) } else if a, err := DecodeTakeIn(value); err != nil || !a.Judge || a.Module != "later" { t.Fatalf("the ask's value: %+v %v", a, err) } + // A refusal an older judge made never stands: a digest the trunk no longer names (a candidate's, or one back on the + // trunk later) refused by judge-0 for words it did not know is judged again by this judge at the start, at a pull that + // carries it, and by indexer_judge with its digest; one this judge refused itself is left as it is. + stale := func(module, raw string) string { + digest, err := ix.keepManifest(ctx, module, raw) + if err != nil { + t.Fatal(err) + } + if _, err := ix.writeVerdict(ctx, digest, delivery.ManifestVerdict{Module: module, State: delivery.VerdictRefused, + Problems: []string{module + " says tool-runner, which this judge does not know"}, Judge: "judge-0", At: time.Now()}); err != nil { + t.Fatal(err) + } + return digest + } + atStart := stale("tools", `{"module":"tools","version":"7"}`) + ownRefusal := stale("client", `{"module":"client","version":"7"}`) + if _, err := ix.writeVerdict(ctx, ownRefusal, delivery.ManifestVerdict{Module: "client", State: delivery.VerdictRefused, + Problems: []string{"x"}, Judge: ix.judgeID(), At: time.Now()}); err != nil { + t.Fatal(err) + } + if _, err := ix.JudgeAtStart(ctx); err != nil || verdict(atStart).Refused() || verdict(atStart).Judge != ix.judgeID() || !verdict(ownRefusal).Refused() { + t.Fatalf("the start pass: %+v, this judge's own refusal %+v (%v)", verdict(atStart), verdict(ownRefusal), err) + } + byPull := stale("tools", `{"module":"tools","version":"8"}`) + pull := delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-tools", Number: 3, Branch: "main", + Manifests: map[string]string{"modules/tools": `{"module":"tools","version":"8"}`}} + if _, err := ix.ApplyMessage(ctx, pull); err != nil || verdict(byPull).Refused() || verdict(byPull).Judge != ix.judgeID() { + t.Fatalf("a pull carrying a digest an older judge refused: %+v %v", verdict(byPull), err) + } + byTool := stale("tools", `{"module":"tools","version":"9"}`) + if got := ix.JudgeAgain(ctx, "client", byTool); got.Stored || !strings.Contains(got.Answer, "tools's, not client's") || !verdict(byTool).Refused() { + t.Fatalf("indexer_judge with another module's digest: %+v", got) + } + if got := ix.JudgeAgain(ctx, "tools", byTool); !got.Stored || !strings.Contains(got.Answer, "accepted") || verdict(byTool).Refused() { + t.Fatalf("indexer_judge with a digest: %+v %+v", got, verdict(byTool)) + } + if key, value, err := NewJudgeAgain("tools", byTool, "why", "me", time.Now()); err != nil || key != "judge.tools" { + t.Fatalf("the ask with a digest: %q %v", key, err) + } else if a, err := DecodeTakeIn(value); err != nil || a.Digest != byTool { + t.Fatalf("the ask's digest: %+v %v", a, err) + } // Without the verdicts state the indexer judges nothing. ix.Verdicts = nil if j, err := ix.JudgeAll(ctx, nil); err != nil || j.Judged != 0 {