From 4e9b0764085436414e6f9882791929dc85d749fb Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 10:43:35 +0100 Subject: [PATCH 1/7] Checker: say loudly that a pull request cannot be judged when its repository has no trunk head (issue 1422) A repository never merged through the gate has no entry in source.main, so its pull requests were dropped without a log line, a verdict or a status, and checker_status counted them as new. Now the verdict says it cannot be judged and why, the forge status is error with those words, checker_status names it under cannot_be_judged, the log says it once per head, and the heads are judged whole once the trunk head is known. Quick fix: this. Thorough fix (the gate seeding source.main from the pull message's base head) is still to do: it needs mesh-sdk, the forge bridge and the gate. --- .../internal/check/case.go | 38 ++++++++- .../internal/check/checker.go | 85 ++++++++++++++++++- .../internal/check/checker_test.go | 74 +++++++++++++++- .../internal/check/read.go | 22 ++++- 4 files changed, 209 insertions(+), 10 deletions(-) diff --git a/modules/mesh-delivery-checker/internal/check/case.go b/modules/mesh-delivery-checker/internal/check/case.go index 7252046..2f2d0ff 100644 --- a/modules/mesh-delivery-checker/internal/check/case.go +++ b/modules/mesh-delivery-checker/internal/check/case.go @@ -115,7 +115,7 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case { for _, p := range c.Pulls { switch head, known := trunk[p.Repository]; { case !known || head == "": - cs.Left = append(cs.Left, fmt.Sprintf("%s#%d: the trunk head of %s is not known", p.Repository, p.Number, p.Repository)) + cs.Left = append(cs.Left, fmt.Sprintf("%s#%d: %s", p.Repository, p.Number, UntrunkedWordsOf([]string{p.Repository}))) default: cs.Pulls = append(cs.Pulls, Pull{Repository: p.Repository, Number: p.Number, Head: p.Head.Head, Trunk: head, Modules: p.Modules, Fps: p.Fps}) cs.Trunk[p.Repository] = head @@ -133,6 +133,42 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case { return cs } +// UntrunkedWords open the words of a verdict, and of a status, of a pull request that cannot be judged because the gate's +// source.main names no trunk head of its repository (issue 1422): never a failure of the change, and judged whole once the +// trunk head is known. It is said loudly, never left silently "new". +const UntrunkedWords = "cannot be judged: the trunk head of " + +// UntrunkedWordsOf are those words for the repositories whose trunk head is not known, in order, each named once. +func UntrunkedWordsOf(repos []string) string { + rs := slices.Clone(repos) + slices.Sort(rs) + rs = slices.Compact(rs) + return UntrunkedWords + strings.Join(rs, ", ") + " is not known: the gate's source.main names none, so there is no commit to merge it onto" +} + +// IsUntrunked says whether a verdict says its heads cannot be judged for want of a trunk head (UntrunkedWords). +func IsUntrunked(v delivery.Verdict) bool { + return v.State == delivery.VerdictFail && v.Failed.Machine == "checker" && strings.HasPrefix(v.Failed.Words, UntrunkedWords) +} + +// leftOf are the pull requests of a candidate whose repository has no trunk head known, with Trunk empty, in repository and +// number order; nil when every repository's is known. +func leftOf(c delivery.Candidate, trunk map[string]string) []Pull { + var out []Pull + for _, p := range c.Pulls { + if trunk[p.Repository] == "" { + out = append(out, Pull{Repository: p.Repository, Number: p.Number, Head: p.Head.Head, Modules: p.Modules, Fps: p.Fps}) + } + } + sort.SliceStable(out, func(i, j int) bool { + if out[i].Repository != out[j].Repository { + return out[i].Repository < out[j].Repository + } + return out[i].Number < out[j].Number + }) + return out +} + // GateHeads is the hash the gate matches the members an Approve covers by (mesh-catalog mesh-delivery-gate onestatus.go // headsOf): delivery.HeadsHash over every pull request of the candidate, hq's included, as // `#=`. A lone pull request is a group of one. Empty for a candidate diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 940f434..3e5b1da 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -295,9 +295,13 @@ func (ch *Checker) heads() map[string]string { } func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidate) { - cs := CaseOf(c, ch.heads()) + heads := ch.heads() + cs := CaseOf(c, heads) + if left := leftOf(c, heads); len(left) > 0 { + ch.untrunked(ctx, c, left, cs == nil) + } if cs == nil { - return // nothing to judge: no trunk known + return // nothing to judge: no trunk known, which untrunked said } cs.Queued = ch.now() ch.mu.Lock() @@ -351,6 +355,79 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat }() } +// untrunked says, loudly, that pull requests of a candidate cannot be judged because the gate's source.main names no trunk +// head of their repository (issue 1422): each gets the status error with UntrunkedWords, and when none of the candidate is +// judged (whole) its verdict key holds a verdict that says so, which checker_verdicts and checker_status show. It is logged +// once per head: when a status or the verdict is written, never at a look that finds them already said. A verdict of these +// heads that says anything else is left. Once the trunk head is known the verdict is stale (it names no trunk) and the heads +// are judged whole (prepare, claim). +func (ch *Checker) untrunked(ctx context.Context, c delivery.Candidate, left []Pull, whole bool) { + key, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash) + if err != nil { + return + } + repos := make([]string, 0, len(left)) + for _, p := range left { + repos = append(repos, p.Repository) + } + words := UntrunkedWordsOf(repos) + said := false + if whole { + v := delivery.Verdict{Group: c.Group, Heads: GateHeads(c), Trunk: map[string]string{}, State: delivery.VerdictFail, + Failed: delivery.FailedFirst{Machine: "checker", Words: words}} + data, err := delivery.EncodeVerdict(v) + if err != nil { + ch.logf("%s: %s, and the verdict cannot say so: %v", key, words, err) + return + } + wrote, err := putOnce(ctx, ch.Verdicts, key, data, func(cur []byte) bool { + held, err := delivery.DecodeVerdict(cur) + // The same words stand; a verdict that says anything else (a check that ran, one running) is not this one's to replace. + return err != nil || !IsUntrunked(held) || held.Failed.Words == words + }) + if err != nil { + ch.logf("%s: %s, and the verdict cannot say so: %v", key, words, err) + } + said = said || wrote + } + for _, p := range left { + okey, err := delivery.CheckerOwedKey(p.Repository, p.Number, p.Head) + if err != nil { + ch.logf("%s: %v", key, err) + continue + } + o := delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "error", + Target: strings.TrimSuffix(ch.Link, "/") + "/" + key, Text: clipLine(words, MaxLine)} + data, err := delivery.EncodeCheckerOwed(o) + if err != nil { + ch.logf("%s: the status of %s cannot be written: %v", key, okey, err) + continue + } + wrote, err := putOnce(ctx, ch.Owed, okey, data, func(cur []byte) bool { + held, err := delivery.DecodeCheckerOwed(cur) + o.V = delivery.Version + return err == nil && delivery.SameOwed(held, o) + }) + if err != nil { + ch.logf("%s: the status of %s cannot be written: %v", key, okey, err) + continue + } + if wrote { + said = true + if ch.Announce != nil { + ch.Announce(ctx, EventOwedWritten, okey) + } + } + } + if said { + names := make([]string, 0, len(left)) + for _, p := range left { + names = append(names, fmt.Sprintf("%s#%d@%s", p.Repository, p.Number, short12(p.Head))) + } + ch.logf("%s: %s %s; the status says so (error), and they are judged once the trunk head is known", key, strings.Join(names, ", "), words) + } +} + // The kinds of a check (ADR 0536 §4): a head never judged and a head asked again after a failure of the nodes are asked on // the build seats' check stream; a carry, a verdict moved to a newer trunk or sibling commit by its package digests (ADR 0548 // §5), on the recheck stream, pulled after. @@ -380,8 +457,8 @@ func (ch *Checker) prepare(ctx context.Context, cs *Case) { return case IsConflictFailure(v): return // a head git refused: its next check is whole (ADR 0548 §7) - case IsEnded(v): - return // ended unjudged when its heads went (issue 1890): held again, they are checked whole + case IsEnded(v), IsUntrunked(v): + return // ended unjudged when its heads went (issue 1890), or never judged for want of a trunk head (issue 1422): checked whole } cs.Kind = KindTrunk if v.State == delivery.VerdictPass && ch.stale(v, cs) { diff --git a/modules/mesh-delivery-checker/internal/check/checker_test.go b/modules/mesh-delivery-checker/internal/check/checker_test.go index bc5820b..dd89894 100644 --- a/modules/mesh-delivery-checker/internal/check/checker_test.go +++ b/modules/mesh-delivery-checker/internal/check/checker_test.go @@ -239,14 +239,80 @@ func TestAnUnheardForgeFailsWithTheReasonAndIsJudgedAgainWhenTheNodesChange(t *t } } -func TestARepositoryWithNoTrunkGetsNoVerdict(t *testing.T) { // N6/AC6, ADR 0484 clause 11 +// Issue 1422: a pull request of a repository the gate's source.main names no trunk head of (one never merged through the +// gate) is never dropped silently. It is not judged, its verdict says it cannot be judged and why, its status is error with +// those words, checker_status names it, and the log says it once per head. Once its trunk head is known it is judged whole; +// in a group, the members whose trunk is known are judged and the others' status says why they are not. +func TestARepositoryWithNoTrunkCannotBeJudgedLoudlyUntilItsTrunkIsKnown(t *testing.T) { // N6/AC6, ADR 0484 clause 11, 1422 ctx, j := context.Background(), passJudge() r := newRig(t, j) - r.ch.Candidate(ctx, "feat/y", candidateOf(t, "feat/y", pullOf("novox/unknown", 4, head1))) + var logged []string + r.ch.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) } + const unknown = "novox/unknown" + cand := candidateOf(t, "feat/y", pullOf(unknown, 4, head1)) + key := keyOf(t, "feat/y", cand) + r.ch.Candidate(ctx, "feat/y", cand) r.ch.Wait() - if j.count() != 0 || len(r.verdicts.vals) != 0 || len(r.owed.vals) != 0 { - t.Errorf("judged %d times, %d verdicts, %d statuses, want none", j.count(), len(r.verdicts.vals), len(r.owed.vals)) + r.ch.Reconsider(ctx) // looked at again: nothing is written or said twice + r.ch.NodesChanged(ctx) + r.ch.Wait() + v := r.verdict(t, key) + if j.count() != 0 || !IsUntrunked(v) || !strings.Contains(v.Failed.Words, "the trunk head of "+unknown+" is not known") { + t.Fatalf("judged %d times, verdict %+v, want none judged and a verdict that says it cannot be judged", j.count(), v) } + if s := owedState(t, r, unknown, 4, head1); s != "error" { + t.Errorf("%s#4 owes %q, want error", unknown, s) + } + if n := countWith(logged, "cannot be judged"); n != 1 { + t.Errorf("said %d times in the log, want once for the head: %q", n, logged) + } + reader := func() Reader { + ckey, _ := delivery.CandidatesKey("feat/y") + return Reader{CandidateState: fakeState{ckey: cand}.read(), VerdictState: fakeState(r.verdicts.vals).read(), + Trunks: func() (map[string]string, error) { return r.trunks.Heads(), nil }} + } + st, err := reader().Status() + if err != nil { + t.Fatal(err) + } + if why := st.CannotBeJudged["feat/y"]; !strings.Contains(why, unknown) || len(st.Waiting[KindNew]) != 0 || len(st.Unjudged) != 0 { + t.Errorf("cannot_be_judged %v, waiting %v, unjudged %v: want feat/y named as cannot be judged, not as new", st.CannotBeJudged, st.Waiting, st.Unjudged) + } + if rows, _, err := reader().Verdicts("feat/y"); err != nil || len(rows) != 1 || !strings.Contains(rows[0].Reports[unknown+"#4"], "cannot be judged") { + t.Errorf("checker_verdicts %+v (%v), want the report of %s#4 to say it cannot be judged", rows, err, unknown) + } + + // Its trunk head becomes known: the heads are judged whole, the same key rewritten, the status success. + r.setTrunks(t, map[string]string{repoA: trunkA, repoSDK: trunkS, unknown: trunkA2}) + r.ch.TrunkMoved(ctx) + r.ch.Wait() + if v := r.verdict(t, key); j.count() != 1 || v.State != delivery.VerdictPass || v.Trunk[unknown] != trunkA2 { + t.Errorf("judged %d times, verdict %+v, want one pass on %s", j.count(), v, trunkA2) + } + if s := owedState(t, r, unknown, 4, head1); s != "success" { + t.Errorf("%s#4 owes %q once its trunk is known, want success", unknown, s) + } + + // A group with one member of a repository whose trunk is not known: the others are judged, its status says why it is not. + group := candidateOf(t, "feat/z", pullOf(repoA, 9, head2), pullOf("novox/other", 2, head2)) + r.ch.Candidate(ctx, "feat/z", group) + r.ch.Wait() + if s := owedState(t, r, repoA, 9, head2); s != "success" { + t.Errorf("%s#9 owes %q, want success", repoA, s) + } + if s := owedState(t, r, "novox/other", 2, head2); s != "error" { + t.Errorf("novox/other#2 owes %q, want error", s) + } +} + +func countWith(lines []string, words string) int { + n := 0 + for _, l := range lines { + if strings.Contains(l, words) { + n++ + } + } + return n } // 1887/A10, ADR 0527 §2: hq is judged as any repository, its checks Go tests, and its pull request gets a verdict and a diff --git a/modules/mesh-delivery-checker/internal/check/read.go b/modules/mesh-delivery-checker/internal/check/read.go index 28ed623..cbe4581 100644 --- a/modules/mesh-delivery-checker/internal/check/read.go +++ b/modules/mesh-delivery-checker/internal/check/read.go @@ -86,6 +86,11 @@ func (r Reader) read(key string, raw json.RawMessage) (VerdictRead, error) { cs := CaseOf(c, v.Trunk) out.Reports = map[string]string{} if cs == nil { + if IsUntrunked(v) { + for _, p := range leftOf(c, v.Trunk) { + out.Reports[pullKey(p)] = v.Failed.Words + } + } return out, nil } for _, p := range cs.Pulls { @@ -323,6 +328,9 @@ type Status struct { // HeldByTurn are the candidates whose check an approved merge's turn holds off (ADR 0548 §3), by group: the repository and // the group of the merge that waits. Their check is claimed when that turn goes. HeldByTurn map[string]string `json:"held_by_turn"` + // CannotBeJudged are the candidates no pull request of which can be judged, by group, with why: the gate's source.main + // names no trunk head of their repositories (issue 1422). They are judged once it does; until then their status is error. + CannotBeJudged map[string]string `json:"cannot_be_judged"` // DeadLetters says whether the checker hears the asks the build seats give up on (issue 734): deaf, a try or check // whose ask is given up waits until its bound. DeadLetters DeadLettersHeard `json:"dead_letters"` @@ -452,7 +460,7 @@ func (r Reader) Status() (Status, error) { } st.Unjudged = []string{} st.Waiting = map[string][]string{KindNew: {}, KindNodes: {}, KindTrunk: {}} - st.HeldByTurn = map[string]string{} + st.HeldByTurn, st.CannotBeJudged = map[string]string{}, map[string]string{} turns := r.turns(&st) var heads map[string]string if r.Trunks == nil { @@ -503,6 +511,15 @@ func (r Reader) Status() (Status, error) { continue } c, raw := row.c, row.raw + if heads != nil && CaseOf(c, heads) == nil { + // Nothing of it can be judged now, whatever its verdict says (issue 1422): never counted as waiting. + var repos []string + for _, p := range leftOf(c, heads) { + repos = append(repos, p.Repository) + } + st.CannotBeJudged[c.Group] = UntrunkedWordsOf(repos) + continue + } if len(raw) == 0 { st.Unjudged = append(st.Unjudged, c.Group) st.Waiting[KindNew] = append(st.Waiting[KindNew], c.Group) @@ -516,6 +533,9 @@ func (r Reader) Status() (Status, error) { case err != nil || v.State == delivery.VerdictRunning: case IsNodesFailure(v): st.Waiting[KindNodes] = append(st.Waiting[KindNodes], c.Group) + case IsUntrunked(v): + // Its trunk head is known now: it waits for its whole check. + st.Waiting[KindNew] = append(st.Waiting[KindNew], c.Group) case heads != nil: if cs := CaseOf(c, heads); cs != nil && (!sameTrunk(v.Trunk, cs.Trunk) || besideMoved(v.Beside, besideAt(heads, r.Runs))) { st.Waiting[KindTrunk] = append(st.Waiting[KindTrunk], c.Group) From 93c0ae19dd2a65b3ae27c367b4b100cafc453e06 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 10:50:39 +0100 Subject: [PATCH 2/7] Forge bridge: write a head pushed to a pull request into another base as closed, and say so (issue 1422) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The checker judges pull requests into main only and source.main names main's head only (ADR 0484 §11), but a live update into another base (de-spiegel#6 into nox-mesh) became a candidate, unlike at the start (OpenPulls) and at its merge. Once the gate seeds a repository's trunk head, such a candidate would be judged against main's head: a verdict that lies. --- .../cmd/mesh-delivery-forge-bridge/main.go | 3 +++ .../internal/bridge/bridge.go | 13 ++++++++++++- .../internal/bridge/bridge_test.go | 15 +++++++++++++++ 3 files changed, 30 insertions(+), 1 deletion(-) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index 9bfb202..a11626c 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -96,6 +96,9 @@ func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, logger.Printf("%s: %s could not be written on FORGE, left for redelivery: %v", m.Subject(), msg.Event, err) return } + if msg.Said != "" { + logger.Printf("%s: %s", m.Subject(), msg.Said) + } _ = m.Ack() } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index 45f4334..80e9b24 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -54,6 +54,8 @@ var LockFiles = []string{"go.sum", "package-lock.json"} type Message struct { Event string Data []byte + // Said is what the bridge did that the log says, beside writing the message; empty for none. + Said string } // The subjects of the forge module's events the bridge writes pull messages for (mesh-catalog modules/gitea, @@ -148,6 +150,15 @@ func ConvertWith(subject string, data []byte, trees Trees, files Files, list Lis if subject == MergedSubject && p.Base == Trunk { return merged(subject, p, trees, files, list) } + said := "" + if action == delivery.PullUpdated && p.Base != "" && p.Base != Trunk { + // The checker judges pull requests into main only, and source.main names main's head only (novox/hq ADR 0484 §11): a + // pull request into another base is no candidate, as it is none at the start (OpenPulls) and at its merge. It is + // written closed, so a candidate an older head made is taken out, and said (issue 1422). + action = delivery.PullClosed + said = fmt.Sprintf("%s/%s#%d at %s is into %s, not %s: written closed, so it is no candidate; the checker judges pull requests into %s only (novox/hq ADR 0484 §11)", + p.Owner, p.Repo, p.Number, p.HeadSHA[:12], p.Base, Trunk, Trunk) + } f := delivery.Forge{Kind: delivery.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number, Head: p.HeadSHA, Action: action} // The directories are the forge's word only when it said it could read them (module_dirs_said). @@ -173,7 +184,7 @@ func ConvertWith(subject string, data []byte, trees Trees, files Files, list Lis if err != nil { return Message{}, unreadable(subject, "%v", err) } - return Message{Event: event, Data: body}, nil + return Message{Event: event, Data: body, Said: said}, nil } // MaxReadsInFlight is how many files of the forge inputsOf asks for at once. A merge into a repository of many diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go index c2424a5..9536738 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go @@ -110,6 +110,21 @@ func TestClosedBecomesAClosedPullMessage(t *testing.T) { // 1235/AC4 if !delivery.SameForge(f, want) { t.Errorf("message = %+v, want %+v", f, want) } + // Issue 1422, ADR 0484 §11: a head pushed to a pull request into another base is written closed, read nothing for, and said; + // one into main is no closed message, and an event naming no base is written as before. + msg, err = Convert(updatedSubject, updated(map[string]any{"base": "nox-mesh"}), nil, nil) + if err != nil { + t.Fatal(err) + } + if f, err = delivery.DecodeForge(msg.Data); err != nil || !delivery.SameForge(f, want) || !strings.Contains(msg.Said, "is into nox-mesh, not main") { + t.Errorf("an update into nox-mesh is %+v (%v), said %q; want it closed and said", f, err, msg.Said) + } + for _, base := range []any{"main", nil} { + msg, err = Convert(updatedSubject, updated(map[string]any{"base": base}), nil, anyFiles) + if f, _ = delivery.DecodeForge(msg.Data); err != nil || f.Action != delivery.PullUpdated || msg.Said != "" { + t.Errorf("an update with base %v is %+v (%v), said %q; want it updated", base, f, err, msg.Said) + } + } } func TestUnreadableEventsWriteNothing(t *testing.T) { // 1235/AC5 From 458e403a049775502143463141e8691a4f2bcd55 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 11:05:25 +0100 Subject: [PATCH 3/7] The indexer's taken key moves only by a merge it applied, never by the gate's source head (issue 1126) A drained forge stream does not say a merge was applied: the forge bridge was still retrying mesh-catalog#535's merged message when the indexer started, and the source's head moved taken.mesh-catalog to 9fd9569, so the view and show said the merge was on the machines while mesh-vault's record stood at 78d7ef7 (ADR 0556 clause 1). --- .../cmd/mesh-delivery-indexer/main.go | 11 ++---- .../internal/index/candidates.go | 3 -- .../internal/index/modules_test.go | 30 ++++++++-------- .../internal/index/taken.go | 34 ++++--------------- 4 files changed, 23 insertions(+), 55 deletions(-) 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 e444c15..01060f1 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -404,15 +404,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Essential: the forge's merges are what the indexer indexes. return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the indexer hears no merge without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, forgeStream, err) } - // A merge applied while the taken state was missing, and the gate's heads once the forge stream is applied whole, get - // their taken key (issue 1280): when the state appears (Ready runs once, now or then), and after each applied message. - ix.Drained = func(ctx context.Context) (bool, error) { - info, err := cons.Info(ctx) - if err != nil { - return false, err - } - return info.NumPending == 0 && info.NumAckPending == 0, nil - } + // A merge applied while the taken state was missing gets its taken key (issue 1280): when the state appears (Ready runs + // once, now or then), and after each applied message. The gate's source heads never move a key (issue 1126). taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) }) cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, logger); ix.KickFill(ctx) }, logger.Printf) if err != nil { diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index c6c82e6..2a2b6aa 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -79,9 +79,6 @@ type Indexer struct { ReadOpen func(ctx context.Context) ([]delivery.OpenPull, error) // Say says one of the indexer's events (placement-changed); nil says none. Say func(event string, body any) error - // Drained says whether the indexer has applied every message of the forge stream (the consumer has none pending and none - // unacknowledged); FillTaken writes the source's heads only then. Nil fills none from the source (taken.go). - Drained func(ctx context.Context) (bool, error) mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach source sourceNow diff --git a/modules/mesh-delivery-indexer/internal/index/modules_test.go b/modules/mesh-delivery-indexer/internal/index/modules_test.go index 8369a2e..1936672 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules_test.go +++ b/modules/mesh-delivery-indexer/internal/index/modules_test.go @@ -139,31 +139,29 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value {name: "a source head whose merge was not applied is not a taken key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: "", then: func(ix *Indexer) error { // the forge stream still holds the merge: the consumer is not drained delete(ix.Taken.(*memStore).vals, takenKey(repo)) - ix.Drained = func(context.Context) (bool, error) { return false, nil } if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil { return err } return ix.FillTaken(context.Background()) }}, - {name: "a source head of a drained forge stream fills a missing key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: next, - then: func(ix *Indexer) error { - delete(ix.Taken.(*memStore).vals, takenKey(repo)) - ix.Drained = func(context.Context) (bool, error) { return true, nil } - if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil { - return err - } - return ix.FillTaken(context.Background()) - }}, - {name: "filling from an older source head leaves a later key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: next, + // Issue 1126 (mesh-catalog#535): the forge bridge never wrote the merged message of 9fd9569, so the forge stream was + // drained with the merge not applied; the source head moved the key all the same, and the view said the merge was + // on the machines. A key moves only by a merge applied (ADR 0556 clause 1), never by the source's word. + {name: "a source head of a drained forge stream whose merge was never heard leaves the key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge, then: func(ix *Indexer) error { ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == merge && in == next, nil } - m := mergedMsg(repo, three) - m.Merge = next - if err := ix.Apply(context.Background(), m); err != nil { + if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil { return err } - ix.Drained = func(context.Context) (bool, error) { return true, nil } - return ix.FillTaken(context.Background()) // the source still names merge + return ix.FillTaken(context.Background()) + }}, + {name: "a source head of a drained forge stream fills no missing key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: "", + then: func(ix *Indexer) error { + delete(ix.Taken.(*memStore).vals, takenKey(repo)) + if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil { + return err + } + return ix.FillTaken(context.Background()) }}, {name: "an earlier merge heard again after a later one leaves taken", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge, then: func(ix *Indexer) error { diff --git a/modules/mesh-delivery-indexer/internal/index/taken.go b/modules/mesh-delivery-indexer/internal/index/taken.go index e8082a0..a6d97e7 100644 --- a/modules/mesh-delivery-indexer/internal/index/taken.go +++ b/modules/mesh-delivery-indexer/internal/index/taken.go @@ -68,13 +68,13 @@ func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error { return err } -// FillTaken writes the taken keys that were left unwritten, never a merge the indexer has not applied whole (a key names -// a merge only after it was applied: ADR 0556 clause 1, and the view reads it as the upper bound of what is reached). Two -// sources, both through writeTaken as a live merge does (issue 1280): -// - the merges applied while the state was not on the bus (kept by writeTaken), whenever the state is there; -// - the gate's source heads, only when Drained says the indexer has applied every message of the forge stream that was -// there before the heads were read. Otherwise it says so, and the next kick of the fill (after a message is applied) -// tries again. +// FillTaken writes the taken keys of the merges applied while the state was not on the bus (kept by writeTaken), through +// writeTaken as a live merge does (issue 1280), whenever the state is there. It never writes a merge the indexer has not +// applied whole (a key names a merge only after it was applied: ADR 0556 clause 1, and the view reads it as the upper bound +// of what is reached). The gate's source heads are not such merges: a drained forge stream does not say a merge was applied, +// since the forge bridge may not have written its merged message yet (issue 1126: mesh-catalog#535's merged message was +// still being retried by the bridge when the indexer started, and the source's head moved the key, so the view said the +// merge was on the machines while the indexer held mesh-vault at the merge before). // // It is idempotent: a key naming the merge or a later one is left. A state not on the bus yet writes nothing and says // nothing (the fill runs again when it appears). The first error is returned after every repository was tried. @@ -87,7 +87,6 @@ func (ix *Indexer) FillTaken(ctx context.Context) error { } ix.mu.Lock() pending := maps.Clone(ix.untaken) - heads := maps.Clone(ix.source.heads) ix.mu.Unlock() var first error note := func(repository string, err error) { @@ -106,25 +105,6 @@ func (ix *Indexer) FillTaken(ctx context.Context) error { } note(repository, err) } - if ix.Drained == nil { - return first - } - if len(heads) == 0 { - ix.logf("the source names no repository head, so no taken key is filled from it") - return first - } - done, err := ix.Drained(ctx) - if err != nil { - ix.logf("whether the forge stream is applied cannot be read, so no taken key is filled from the source: %v", err) - return first - } - if !done { - ix.logf("the forge stream is not applied whole yet, so no taken key is filled from the source; the next applied message tries again") - return first - } - for _, repository := range slices.Sorted(maps.Keys(heads)) { - note(repository, ix.writeTaken(ctx, delivery.Forge{Repository: repository, Merge: heads[repository]})) - } return first } From 704fc8d4678c787a555b81c0636dc485d5cc2f50 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 11:05:25 +0100 Subject: [PATCH 4/7] The forge bridge reads a merge's module directories only where their tree changed (issue 1126) mesh-catalog's 140 module directories are 420 forge reads per merge, past the 30 s bound of a handled event while the forge is busy: the merged message of #535 timed out on every try, and #3fb7951 needed seven (issue 1458 met the same kind at 131). A tree hash names its bytes, so what was read at a tree is kept in the new state inputs and in memory, and a merge reads only the directories it changed. --- .../integration_test.go | 8 +- .../cmd/mesh-delivery-forge-bridge/kept.go | 81 +++++++++++++++++++ .../cmd/mesh-delivery-forge-bridge/main.go | 8 +- .../mesh-delivery-forge-bridge/main_test.go | 10 +-- .../internal/bridge/bridge.go | 26 ++++-- .../internal/bridge/files_test.go | 57 +++++++++++++ .../internal/bridge/kept.go | 57 +++++++++++++ .../manifest_test.go | 11 +-- .../mesh-delivery-forge-bridge/module.json | 3 +- 9 files changed, 241 insertions(+), 20 deletions(-) create mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/kept.go create mode 100644 modules/mesh-delivery-forge-bridge/internal/bridge/kept.go diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index 59ed05e..77d84fe 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go @@ -72,11 +72,17 @@ accounts { "mesh.mod.gitea.tool.gitea_open_pulls" "mesh.mod.gitea.tool.gitea_open_pulls.*" "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_performed.>" + "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_inputs.>" "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_performed.>" + "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_inputs.>" "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_performed.>" + "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_inputs.>" "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_performed" + "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_inputs" "$JS.FC.KV_mesh-delivery-forge-bridge_performed.>" + "$JS.FC.KV_mesh-delivery-forge-bridge_inputs.>" "$KV.mesh-delivery-forge-bridge_performed.>" + "$KV.mesh-delivery-forge-bridge_inputs.>" "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-gate_merges.>" "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-gate_merges.>" "$JS.API.DIRECT.GET.KV_mesh-delivery-gate_merges.>" @@ -219,7 +225,7 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, } // Its other state bucket, and the gate's buckets the bridge reads, made by the controller from the gate's manifest // (mesh-sdk go/delivery MergesBucket, OwedBucket with their histories). - for name, history := range map[string]uint8{performedBucket: 1, delivery.MergesBucket: delivery.MergesHistory, delivery.OwedBucket: delivery.OwedHistory} { + for name, history := range map[string]uint8{performedBucket: 1, inputsBucket: 1, delivery.MergesBucket: delivery.MergesHistory, delivery.OwedBucket: delivery.OwedHistory} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/kept.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/kept.go new file mode 100644 index 0000000..ceaec66 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/kept.go @@ -0,0 +1,81 @@ +package main + +import ( + "context" + "encoding/json" + "log" + "regexp" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-forge-bridge/internal/bridge" +) + +// inputsBucket is the module's state bucket inputs (manifest state entry inputs): what the bridge read of a module +// directory, its manifest and lock-file hashes, by the directory's git tree hash, so a merge reads the forge only for the +// directories it changed (issue 1126; internal/bridge kept.go). A tree hash names its bytes, so a value is never stale; +// a merge adds a value only for each directory it changed (a few kilobytes each). +const inputsBucket = delivery.ForgeBridgeModule + "_inputs" + +// keptTimeout bounds one read or write of the bucket: the bucket is a saving, never a reason to fail a merge. +const keptTimeout = 2 * time.Second + +var treeHash = regexp.MustCompile(`^[0-9a-f]{40}$|^[0-9a-f]{64}$`) + +// kept is the bridge's Kept: in memory always, and in the bucket inputs when it is on the bus, so the first merge after a +// start reads only what changed too. A value the bucket cannot give or take is read from the forge, and said. +type kept struct { + kv jetstream.KeyValue // nil when the bucket is not on the bus + logger *log.Logger + mu sync.Mutex + mem map[string]bridge.Inputs +} + +func newKept(kv jetstream.KeyValue, logger *log.Logger) *kept { + return &kept{kv: kv, logger: logger, mem: map[string]bridge.Inputs{}} +} + +func (k *kept) Get(tree string) (bridge.Inputs, bool) { + k.mu.Lock() + in, ok := k.mem[tree] + k.mu.Unlock() + if ok || k.kv == nil || !treeHash.MatchString(tree) { + return in, ok + } + ctx, cancel := context.WithTimeout(context.Background(), keptTimeout) + defer cancel() + e, err := k.kv.Get(ctx, tree) + if err != nil { + return bridge.Inputs{}, false // not kept (or not readable): read from the forge + } + if err := json.Unmarshal(e.Value(), &in); err != nil || in.Manifest == "" { + k.logger.Printf("%s.%s cannot be read, so its directory is read from the forge: %v", inputsBucket, tree, err) + return bridge.Inputs{}, false + } + k.mu.Lock() + k.mem[tree] = in + k.mu.Unlock() + return in, true +} + +func (k *kept) Put(tree string, in bridge.Inputs) { + k.mu.Lock() + k.mem[tree] = in + k.mu.Unlock() + if k.kv == nil || !treeHash.MatchString(tree) { + return + } + data, err := json.Marshal(in) + if err != nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), keptTimeout) + defer cancel() + if _, err := k.kv.Put(ctx, tree, data); err != nil { + k.logger.Printf("%s.%s is not kept, so the next merge reads its directory from the forge again: %v", inputsBucket, tree, err) + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index a11626c..df27efa 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -67,8 +67,8 @@ type publisher interface { // handle turns one event into its message on FORGE and acknowledges it. An event it cannot read is terminated, never // redelivered, and the log names its subject and why. A message it could not write stays unacknowledged: the server // delivers it again (the server's ack wait is the only clock). -func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, list bridge.Listed, logger *log.Logger) { - msg, err := bridge.ConvertWith(m.Subject(), m.Data(), trees, files, list) +func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, list bridge.Listed, kept bridge.Kept, logger *log.Logger) { + msg, err := bridge.ConvertKept(m.Subject(), m.Data(), trees, files, list, kept) switch { case errors.Is(err, bridge.ErrRetry): logger.Printf("%v; heard again in %s", err, retryDelay) @@ -163,6 +163,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } go onReconnect(ctx, again, said, logger) + // What a merge read of each module directory, by its tree (issue 1126): a merge reads only the directories it changed. + inputs := newKept(process.OptionalBucket(ctx, js, inputsBucket, "a merge's module directories are kept in memory only, so the first merge after a start reads every one", logger.Printf), logger) cons, err := cred.EventConsumer(ctx, js, eventsStream) if err != nil { // Essential: the forge's events are what the bridge carries. @@ -173,7 +175,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Bounded: the turn is held, and no sync or other event goes on without it. hctx, cancel := context.WithTimeout(ctx, syncDeadline) defer cancel() - handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), logger) + handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), inputs, logger) }) }, logger.Printf) if err != nil { diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go index 85df582..881a1ae 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go @@ -81,7 +81,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit }) var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{} - handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, someFiles, nil, nil, log.New(&logs, "", 0)) if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull" { t.Fatalf("published on %v", pub.subjects) } @@ -97,7 +97,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(map[string]any{"head_sha": "short"})}, &fakePub{} - handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, someFiles, nil, nil, log.New(&logs, "", 0)) if len(pub.subjects) != 0 { t.Errorf("a message was written: %v", pub.subjects) } @@ -112,7 +112,7 @@ func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 func TestUnwrittenMessageIsNotAcked(t *testing.T) { // 1235 (a failed write is redelivered by the server, not retried here) var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{err: errors.New("no stream")} - handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, someFiles, nil, nil, log.New(&logs, "", 0)) if m.acked != 0 || m.termd != 0 { t.Errorf("acked %d terminated %d, want neither", m.acked, m.termd) } @@ -134,7 +134,7 @@ func TestTreesRefusedLeavesTheEventForTenSecondsAndLogs(t *testing.T) { // 1236/ call := func(context.Context, string, any) (json.RawMessage, error) { return nil, errors.New("the tree listing of novox/mesh-delivery was cut") } - handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, nil, log.New(&logs, "", 0)) if len(pub.subjects) != 0 { t.Errorf("a message was written: %v", pub.subjects) } @@ -163,7 +163,7 @@ func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { // 1236/ } return json.RawMessage(`{"root":"r0","trees":{"":"t0","modules/a":"t1"}}`), nil } - handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, nil, log.New(&logs, "", 0)) if len(keys) == 0 || keys[0] != "gitea.gitea_commit_trees" || slices.Index(keys[1:], "gitea.gitea_commit_trees") >= 0 { t.Fatalf("calls %v, want the trees once and first", keys) } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index 80e9b24..a8669bb 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -109,7 +109,7 @@ func unreadable(subject, why string, args ...any) error { // the trunk asks trees once and becomes a merged message; a merge into any other branch is a closed pull message, as the // pull request left its group. A forge that cannot say is ErrRetry; a merge with no module.json is ErrNoModules. func Convert(subject string, data []byte, trees Trees, files Files) (Message, error) { - return ConvertWith(subject, data, trees, files, nil) + return ConvertKept(subject, data, trees, files, nil, nil) } // Listed asks the forge for the files directly under one directory of a commit, by name to blob id (gitea_commit_trees @@ -122,6 +122,12 @@ type Listed func(owner, repo, sha, dir string) (map[string]string, error) // updated pull request of the catalogue that touches placement/ (touchesPlacement), with every placement file at its // head, which the indexer writes as placement candidates (ADR 0549 §4: the nodes it places answer it before it merges). func ConvertWith(subject string, data []byte, trees Trees, files Files, list Listed) (Message, error) { + return ConvertKept(subject, data, trees, files, list, nil) +} + +// ConvertKept is ConvertWith; with kept, a merge reads the forge only for the module directories whose tree hash kept +// does not hold, and keeps what it read (kept.go). +func ConvertKept(subject string, data []byte, trees Trees, files Files, list Listed, kept Kept) (Message, error) { var action string switch subject { case UpdatedSubject: @@ -148,7 +154,7 @@ func ConvertWith(subject string, data []byte, trees Trees, files Files, list Lis return Message{}, unreadable(subject, "no branch") } if subject == MergedSubject && p.Base == Trunk { - return merged(subject, p, trees, files, list) + return merged(subject, p, trees, files, list, kept) } said := "" if action == delivery.PullUpdated && p.Base != "" && p.Base != Trunk { @@ -277,7 +283,7 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma } // merged is the merged message of a merge into the trunk: the trees of the merge commit, asked once. -func merged(subject string, p pull, trees Trees, files Files, list Listed) (Message, error) { +func merged(subject string, p pull, trees Trees, files Files, list Listed, kept Kept) (Message, error) { if !sha40.MatchString(p.MergeSHA) { return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", p.MergeSHA) } @@ -294,9 +300,19 @@ func merged(subject string, p pull, trees Trees, files Files, list Listed) (Mess names = append(names, dir) } slices.Sort(names) - manifests, locks, err := inputsOf(p.Owner, p.Repo, p.MergeSHA, names, files) + manifests, locks, toRead := keptInputs(names, dirs, kept) + read, readLocks, err := inputsOf(p.Owner, p.Repo, p.MergeSHA, toRead, files) if err != nil { - return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s: %v", ErrRetry, subject, repository, p.MergeSHA, err) + return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s (%d of %d module directories read from the forge, the others kept by their tree): %v", ErrRetry, subject, repository, p.MergeSHA, len(toRead), len(names), err) + } + keep(toRead, dirs, read, readLocks, kept) + maps.Copy(manifests, read) + maps.Copy(locks, readLocks) + if len(manifests) == 0 { + manifests = nil + } + if len(locks) == 0 { + locks = nil } event, err := delivery.ForgeEvent(repository, delivery.ForgeMerged) if err != nil { diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go index 3d07f17..a963a19 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go @@ -369,3 +369,60 @@ func TestOneFailingFileIsRetryAndEveryCallHasReturnedBeforeTheReadDoes(t *testin t.Errorf("err = %v, want ErrRetry", err) } } + +// memKept is a Kept in memory. +type memKept map[string]Inputs + +func (k memKept) Get(tree string) (Inputs, bool) { in, ok := k[tree]; return in, ok } +func (k memKept) Put(tree string, in Inputs) { k[tree] = in } + +// Issue 1126 (mesh-catalog#535; issue 1458 met the kind first): a merge reads the forge only for the module directories +// whose tree changed. Its message is the same as one read whole; a directory whose tree is kept costs no read. +func TestAMergeReadsOnlyTheDirectoriesWhoseTreeChanged(t *testing.T) { + manifest, sum := real(t, "module.json"), real(t, "go.sum") + forge := &forgeFiles{have: map[string][]byte{"modules/a/module.json": manifest, "modules/a/go.sum": sum, "modules/b/module.json": []byte(`{"module":"b"}`)}} + kept := memKept{} + var calls []call + first, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1", "modules/b": "t2"}, nil), forge.get, nil, kept) + if err != nil { + t.Fatal(err) + } + if len(forge.calls) != 6 { + t.Fatalf("the first merge read %d files, want 6 (a manifest and two lock files for each directory)", len(forge.calls)) + } + // The next merge changes modules/b alone: only its three files are read, and the message equals one read whole. + forge.calls = nil + forge.have["modules/b/module.json"] = []byte(`{"module":"b","v":2}`) + trees := map[string]string{"modules/a": "t1", "modules/b": "t3"} + next, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get, nil, kept) + if err != nil { + t.Fatal(err) + } + for _, c := range forge.calls { + if !strings.HasPrefix(c.path, "modules/b/") { + t.Errorf("%s was read although its directory's tree is kept", c.path) + } + } + if len(forge.calls) != 3 { + t.Errorf("the second merge read %d files, want 3", len(forge.calls)) + } + whole, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get) + if err != nil { + t.Fatal(err) + } + if string(next.Data) != string(whole.Data) { + t.Errorf("the message from kept inputs differs from one read whole:\n%s\n%s", next.Data, whole.Data) + } + if string(first.Data) == string(next.Data) { + t.Error("the second merge's message did not change") + } + // A failed read keeps nothing of the directories it read, so the next try reads them again. + kept2 := memKept{} + failing := &forgeFiles{have: forge.have, errs: map[string]error{"modules/b/module.json": errors.New("context deadline exceeded")}} + if _, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, nil, kept2); !errors.Is(err, ErrRetry) { + t.Fatalf("err = %v, want ErrRetry", err) + } + if len(kept2) != 0 { + t.Errorf("a failed read kept %v", kept2) + } +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/kept.go b/modules/mesh-delivery-forge-bridge/internal/bridge/kept.go new file mode 100644 index 0000000..1e8c87b --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/kept.go @@ -0,0 +1,57 @@ +package bridge + +// A merge of a repository of many modules read three files of the forge for every module, whatever the merge changed: +// the catalogue's 140 modules are 420 reads, which no longer fit the 30 s bound of a handled event while the forge is +// busy after a merge, so the merged message of mesh-catalog#535 was never written and the indexer never saw it (issue +// 1126; issue 1458 met the same kind at 131 modules and raised the reads in flight). A directory's git tree hash names +// every byte under it, so the manifest and the lock-file hashes read once at a tree hash are those of every later commit +// whose directory has the same tree: a merge reads the forge only for the directories it changed. + +// Inputs is what the bridge read of one module directory at one tree hash: its manifest and its lock files' hex sha256 +// by name (none when it has none). +type Inputs struct { + Manifest string `json:"manifest"` + Locks map[string]string `json:"locks,omitempty"` +} + +// Kept holds Inputs by a directory's git tree hash (the bridge's state bucket inputs). Get answers false for a tree it +// does not hold or cannot read; Put is best effort: a value it could not keep is read from the forge next time. +type Kept interface { + Get(tree string) (Inputs, bool) + Put(tree string, in Inputs) +} + +// keptInputs splits dirs into the inputs kept for their tree hash and the directories still to read from the forge. +func keptInputs(dirs []string, trees map[string]string, kept Kept) (manifests map[string]string, locks map[string]map[string]string, read []string) { + manifests, locks = map[string]string{}, map[string]map[string]string{} + for _, dir := range dirs { + tree := trees[dir] + if kept == nil || tree == "" { + read = append(read, dir) + continue + } + in, ok := kept.Get(tree) + if !ok { + read = append(read, dir) + continue + } + manifests[dir] = in.Manifest + if len(in.Locks) > 0 { + locks[dir] = in.Locks + } + } + return manifests, locks, read +} + +// keep puts what was read of each directory under its tree hash. +func keep(dirs []string, trees map[string]string, manifests map[string]string, locks map[string]map[string]string, kept Kept) { + if kept == nil { + return + } + for _, dir := range dirs { + raw, ok := manifests[dir] + if tree := trees[dir]; ok && tree != "" { + kept.Put(tree, Inputs{Manifest: raw, Locks: locks[dir]}) + } + } +} diff --git a/modules/mesh-delivery-forge-bridge/manifest_test.go b/modules/mesh-delivery-forge-bridge/manifest_test.go index 75e356b..19686ac 100644 --- a/modules/mesh-delivery-forge-bridge/manifest_test.go +++ b/modules/mesh-delivery-forge-bridge/manifest_test.go @@ -103,11 +103,12 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1 if m.SecretsOwner != "mesh-delivery-forge-bridge" { t.Errorf("secrets-owner = %q", m.SecretsOwner) } - // Two buckets: the head last written on FORGE for each pull request, so a sync writes only what differs (ADR 0443 - // §5), and the revision of each key of the gate's buckets the bridge performed, so a key is performed once per - // revision (ADR 0437 §8; issue 1341). - if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}}) { // 1279/AC6, 1341 - t.Errorf("state = %+v, want the buckets heads and performed", m.State) + // Three buckets: the head last written on FORGE for each pull request, so a sync writes only what differs (ADR 0443 + // §5), the revision of each key of the gate's buckets the bridge performed, so a key is performed once per + // revision (ADR 0437 §8; issue 1341), and what a merge read of each module directory by its tree, so a merge reads + // the forge only for the directories it changed (issue 1126). + if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}, {Name: "inputs", History: 1}}) { // 1279/AC6, 1341, 1126 + t.Errorf("state = %+v, want the buckets heads, performed and inputs", m.State) } // No other state, no bus, no seat, no tool, no event (ADR 0437 §4; ADR 0521: no check.status). for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} { diff --git a/modules/mesh-delivery-forge-bridge/module.json b/modules/mesh-delivery-forge-bridge/module.json index 25b68be..f152cad 100644 --- a/modules/mesh-delivery-forge-bridge/module.json +++ b/modules/mesh-delivery-forge-bridge/module.json @@ -12,7 +12,8 @@ ], "state": [ { "name": "heads", "history": 1 }, - { "name": "performed", "history": 1 } + { "name": "performed", "history": 1 }, + { "name": "inputs", "history": 1 } ], "reads": [ "mesh-delivery-gate.merges", From 89be76bb6acc3793dde1f6ff437b4323d1a3e164 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 11:07:05 +0100 Subject: [PATCH 5/7] The forge bridge keeps the directories a failed try read whole, so a large merge converges over its tries (issue 1126) Without it the first merge after the bridge starts, with nothing kept, still reads all 140 catalogue directories in one 30 s try and times out as before. --- .../internal/bridge/bridge.go | 35 +++++++++++++++++-- .../internal/bridge/files_test.go | 19 ++++++++-- 2 files changed, 48 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index a8669bb..19fd944 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -203,6 +203,7 @@ const MaxReadsInFlight = 8 type read struct { dir, path string manifest bool + asked bool raw []byte err error } @@ -216,6 +217,14 @@ type read struct { // order the calls finish in, and the error is that of the first failing file in the order of dirs and LockFiles. After // a failure no further file is asked for, and inputsOf returns only when every call it started has returned. func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests map[string]string, locks map[string]map[string]string, err error) { + manifests, locks, _, err = inputsRead(owner, repo, ref, dirs, files) + return manifests, locks, err +} + +// inputsRead is inputsOf, and also answers, on a failure too, the directories whose manifest and lock files were all +// read, with their manifests and lock hashes: a merge keeps them by their tree, so its next try reads only the rest +// (issue 1126). +func inputsRead(owner, repo, ref string, dirs []string, files Files) (manifests map[string]string, locks map[string]map[string]string, whole []string, err error) { at := func(dir, name string) string { if dir == "" { return name @@ -244,6 +253,7 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma <-slots break } + r.asked = true wg.Add(1) go func(r *read) { defer func() { <-slots; wg.Done() }() @@ -258,10 +268,16 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma wg.Wait() manifests = map[string]string{} locks = map[string]map[string]string{} + broken := map[string]bool{} // a directory with a file not read (failed, or never asked after a failure) for _, r := range reads { switch { case r.err != nil && (r.manifest || !errors.Is(r.err, ErrNotFound)): - return nil, nil, fmt.Errorf("%s: %w", r.path, r.err) + broken[r.dir] = true + if err == nil { + err = fmt.Errorf("%s: %w", r.path, r.err) + } + case r.err == nil && r.raw == nil && !r.asked: + broken[r.dir] = true case r.err != nil: // a lock file the forge does not have case r.manifest: manifests[r.dir] = string(r.raw) @@ -273,13 +289,25 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma locks[r.dir][path.Base(r.path)] = hex.EncodeToString(sum[:]) } } + for _, dir := range dirs { + if !broken[dir] { + whole = append(whole, dir) + } + } + if err != nil { + for dir := range broken { + delete(manifests, dir) + delete(locks, dir) + } + return manifests, locks, whole, err + } if len(manifests) == 0 { manifests = nil } if len(locks) == 0 { locks = nil } - return manifests, locks, nil + return manifests, locks, whole, nil } // merged is the merged message of a merge into the trunk: the trees of the merge commit, asked once. @@ -301,8 +329,9 @@ func merged(subject string, p pull, trees Trees, files Files, list Listed, kept } slices.Sort(names) manifests, locks, toRead := keptInputs(names, dirs, kept) - read, readLocks, err := inputsOf(p.Owner, p.Repo, p.MergeSHA, toRead, files) + read, readLocks, whole, err := inputsRead(p.Owner, p.Repo, p.MergeSHA, toRead, files) if err != nil { + keep(whole, dirs, read, readLocks, kept) // what was read whole is not asked again at the next try return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s (%d of %d module directories read from the forge, the others kept by their tree): %v", ErrRetry, subject, repository, p.MergeSHA, len(toRead), len(names), err) } keep(toRead, dirs, read, readLocks, kept) diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go index a963a19..aaecb3d 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go @@ -416,13 +416,26 @@ func TestAMergeReadsOnlyTheDirectoriesWhoseTreeChanged(t *testing.T) { if string(first.Data) == string(next.Data) { t.Error("the second merge's message did not change") } - // A failed read keeps nothing of the directories it read, so the next try reads them again. + // A failed read keeps the directories it read whole and nothing of the one that failed, so the next try reads only the + // rest: a merge too large for one try's bound converges over its tries. kept2 := memKept{} failing := &forgeFiles{have: forge.have, errs: map[string]error{"modules/b/module.json": errors.New("context deadline exceeded")}} if _, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, nil, kept2); !errors.Is(err, ErrRetry) { t.Fatalf("err = %v, want ErrRetry", err) } - if len(kept2) != 0 { - t.Errorf("a failed read kept %v", kept2) + if _, ok := kept2["t1"]; !ok || len(kept2) != 1 { // modules/a's files are asked first, before the failure is seen + t.Errorf("a failed read kept %v, want modules/a's tree t1 alone", kept2) + } + failing.calls = nil + failing.errs = nil + again, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, nil, kept2) + if err != nil { + t.Fatal(err) + } + if string(again.Data) != string(whole.Data) { + t.Errorf("the message of the try after a failure differs from one read whole") + } + if len(failing.calls) != 3 { + t.Errorf("the try after a failure read %d files, want only modules/b's 3", len(failing.calls)) } } From 790653dfde9be67cf38e3a55928c88387ae517f7 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 11:08:56 +0100 Subject: [PATCH 6/7] indexer_take_in reads a held module again at the source's head when its record stands behind it (issue 1126) A merge whose merged message never reached the indexer had no recovery: take-in answered held for mesh-vault, whose record stayed at 78d7ef7 after mesh-catalog#535. --- .../cmd/mesh-delivery-indexer-tools/main.go | 4 +- .../internal/index/registered.go | 60 ++++++++++++++++++- .../internal/index/registered_test.go | 29 +++++++++ 3 files changed, 91 insertions(+), 2 deletions(-) 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 2640989..6904086 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 @@ -29,7 +29,9 @@ func Tools(ask asker, now func() time.Time) []stdio.Tool { Description: "Take a registered module in now: the indexer reads its manifest from the repository and commit the controller's " + "build records register it from (at the head the gate's source names, when it names one), whatever it tried before, and " + "writes it to its modules state as a merge would. For a module registered from outside the indexed repositories whose " + - "repository the forge could not read (a 404 before the forge's account could read it). Answers stored, from which " + + "repository the forge could not read (a 404 before the forge's account could read it). A module the indexer holds " + + "whose record stands behind the head the gate's source names (a merge whose merged message never reached the " + + "indexer) is read again at that head and written when it changed. Answers stored, from which " + "repository, commit and path, or why not (held already, no registration, the forge's error, no manifest there). " + "The indexer reads such a module again by itself on the next source revision and merged message; this asks now. " + "indexer_module (mesh-delivery-view) reads the result.", diff --git a/modules/mesh-delivery-indexer/internal/index/registered.go b/modules/mesh-delivery-indexer/internal/index/registered.go index e83fb4f..0a78a95 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered.go +++ b/modules/mesh-delivery-indexer/internal/index/registered.go @@ -45,6 +45,12 @@ type taken struct { } func (ix *Indexer) indexRegistered(ctx context.Context, r Registration, force bool) (taken, error) { + return ix.indexAt(ctx, r, force, false) +} + +// indexAt is indexRegistered; reread also reads a module the modules state holds (a person's take-in of a held module +// whose record stands behind the head the gate's source names, issue 1126). +func (ix *Indexer) indexAt(ctx context.Context, r Registration, force, reread bool) (taken, error) { if ix.Modules == nil || ix.Read == nil { return taken{why: "the indexer reads no forge or writes no modules"}, nil } @@ -55,7 +61,7 @@ func (ix *Indexer) indexRegistered(ctx context.Context, r Registration, force bo } if _, _, found, err := ix.Modules.Get(ctx, key); err != nil { return taken{}, fmt.Errorf("modules.%s: %w", key, err) - } else if found { + } else if found && !reread { ix.markRead(r.Module) return taken{held: true, why: "the modules state holds it already: a merge to its repository writes it from now on"}, nil } @@ -204,6 +210,9 @@ type TakenIn struct { // controller's records, or whose every registration could not be read or holds no manifest of it, is answered with why. func (ix *Indexer) TakeIn(ctx context.Context, module string) TakenIn { out := TakenIn{Module: module} + if again, ok := ix.takeInHeld(ctx, module); ok { + return again + } if ix.Builds == nil { out.Answer = "the indexer reads no controller's build records, so it knows no registration" return out @@ -273,3 +282,52 @@ func (ix *Indexer) IndexRegisteredAll(ctx context.Context) error { } return first } + +// takeInHeld reads again, at the head the gate's source names for its repository, a module the modules state holds whose +// record stands at another commit, and writes it as a merge would (only when it changed). It is the recovery for a merge +// whose merged message never reached the indexer (issue 1126: mesh-catalog#535's message was given up by the forge bridge +// after its reads timed out, and mesh-vault's record stayed at the merge before). It answers false where it does not +// apply: no such module held, or no head named, or the record at that head already. +func (ix *Indexer) takeInHeld(ctx context.Context, module string) (TakenIn, bool) { + out := TakenIn{Module: module} + key, err := delivery.ModulesKey(module) + if err != nil || ix.Modules == nil { + return out, false + } + value, _, found, err := ix.Modules.Get(ctx, key) + if err != nil || !found { + return out, false + } + cur, err := delivery.DecodeModule(value) + if err != nil || cur.Repository == "" || cur.Path == "" { + return out, false + } + head := "" + ix.mu.Lock() + for repo, h := range ix.source.heads { + if delivery.SameRepository(repo, cur.Repository) && h != "" { + head = h + } + } + ix.mu.Unlock() + if head == "" || head == cur.Commit { + return out, false + } + t, err := ix.indexAt(ctx, Registration{Module: module, Repository: cur.Repository, Path: cur.Path, Commit: head}, true, true) + out.Held, out.Repository, out.Commit, out.Path = true, t.repository, t.commit, pathOf(t.dir) + switch { + case err != nil: + out.Answer = "held, and not read again at the source's head: " + err.Error() + case !t.stored: + out.Answer = "held, and not read again at the source's head: " + t.why + default: + after, _, _, _ := ix.Modules.Get(ctx, key) + out.Stored = true + if string(after) == string(value) { + out.Answer = fmt.Sprintf("held, and read again from %s at %s: unchanged there, so its record stands", t.repository, t.commit) + } else { + out.Answer = fmt.Sprintf("held, read again from %s at %s and written to the modules state: it changed since %.12s", t.repository, t.commit, cur.Commit) + } + } + return out, true +} diff --git a/modules/mesh-delivery-indexer/internal/index/registered_test.go b/modules/mesh-delivery-indexer/internal/index/registered_test.go index 13eca05..6e3ac1e 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered_test.go +++ b/modules/mesh-delivery-indexer/internal/index/registered_test.go @@ -256,3 +256,32 @@ func TestARecordOfAPullRequestDoesNotReplaceTheRegisteredRef(t *testing.T) { // t.Fatalf("registrations = %+v", got) } } + +// Issue 1126: mesh-catalog#535 changed mesh-vault, but its merged message never reached the indexer, so the record stood at +// the merge before. A take-in of the held module reads it again at the head the gate's source names and writes it. +func TestATakeInOfAHeldModuleBehindTheSourceHeadReadsItAgain(t *testing.T) { + ctx := context.Background() + ix, mods := modulesIndexer(t) + ix.Manifests = newMem() + if err := ix.Apply(ctx, mergedMsg("novox/mesh-catalog", map[string]string{"modules/a": "treeA"})); err != nil { + t.Fatal(err) + } + ix.source = sourceNow{rev: 2, heads: map[string]string{"novox/mesh-catalog": h2}} + var asked []string + ix.Read = func(_ context.Context, _, sha, dir string) (RepoRead, error) { + asked = append(asked, sha+":"+dir) + return RepoRead{Root: "r1", Trees: map[string]string{"modules/a": "treeA2"}, Manifests: map[string]string{"modules/a": `{"module":"a","version":"2"}`}}, nil + } + got := ix.TakeIn(ctx, "a") + if !got.Stored || !got.Held || got.Commit != h2 || got.Path != "modules/a" || !strings.Contains(got.Answer, "written") { + t.Errorf("take-in answered %+v", got) + } + if m, ok := moduleOf(t, mods, "a"); !ok || m.Commit != h2 || m.Inputs2 == nil || m.Inputs2.Tree != "treeA2" { + t.Errorf("modules.a = %+v (%v), want it at %.8s with tree treeA2 (asked %v)", m, ok, h2, asked) + } + // At the head now: a take-in answers held and reads nothing. + n := len(asked) + if again := ix.TakeIn(ctx, "a"); again.Stored || !again.Held || len(asked) != n { + t.Errorf("a take-in of a module at the source's head answered %+v and asked %v", again, asked) + } +} From 2b801f5bd49a710203944af416a8bf641231f7f6 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 11:18:01 +0100 Subject: [PATCH 7/7] indexer_take_in reads a held module again only when its record is behind the source's head, and the two new tests are rows (issue 1126, review) A head older than the record, or one whose order the forge cannot say, would have written an older manifest back. The new tests become rows of existing ones: the repository is over its 500 unit tests. --- .../internal/bridge/files_test.go | 4 +- .../internal/index/registered.go | 16 ++++ .../internal/index/registered_test.go | 81 ++++++++++++------- 3 files changed, 71 insertions(+), 30 deletions(-) diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go index aaecb3d..3e88564 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go @@ -120,6 +120,8 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1 if _, err := ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, failing); !errors.Is(err, ErrRetry) { t.Errorf("a listing the forge cannot give: %v", err) } + // Issue 1126: a merge reads the forge only for the module directories whose tree changed. + t.Run("a merge reads only the directories whose tree changed", mergeReadsOnlyChangedDirs) } func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2 @@ -378,7 +380,7 @@ func (k memKept) Put(tree string, in Inputs) { k[tree] = in } // Issue 1126 (mesh-catalog#535; issue 1458 met the kind first): a merge reads the forge only for the module directories // whose tree changed. Its message is the same as one read whole; a directory whose tree is kept costs no read. -func TestAMergeReadsOnlyTheDirectoriesWhoseTreeChanged(t *testing.T) { +func mergeReadsOnlyChangedDirs(t *testing.T) { manifest, sum := real(t, "module.json"), real(t, "go.sum") forge := &forgeFiles{have: map[string][]byte{"modules/a/module.json": manifest, "modules/a/go.sum": sum, "modules/b/module.json": []byte(`{"module":"b"}`)}} kept := memKept{} diff --git a/modules/mesh-delivery-indexer/internal/index/registered.go b/modules/mesh-delivery-indexer/internal/index/registered.go index 0a78a95..7fe4c28 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered.go +++ b/modules/mesh-delivery-indexer/internal/index/registered.go @@ -313,6 +313,22 @@ func (ix *Indexer) takeInHeld(ctx context.Context, module string) (TakenIn, bool if head == "" || head == cur.Commit { return out, false } + // Only forward: a record at a merge the source does not name yet (a merge applied before the source moved) is never + // written back to the source's older head. Where the order cannot be known, nothing is read. + behind, why := false, "the forge cannot be asked" + if ix.Contains != nil { + var err error + if behind, err = ix.Contains(ctx, cur.Repository, cur.Commit, head); err != nil { + why = err.Error() + } else if !behind { + why = fmt.Sprintf("its record at %.12s is not in %.12s", cur.Commit, head) + } + } + if !behind { + out.Held, out.Repository, out.Commit, out.Path = true, cur.Repository, cur.Commit, cur.Path + out.Answer = fmt.Sprintf("held; the source names another head, %.12s, and not one its record is behind (%s), so it is not read again", head, why) + return out, true + } t, err := ix.indexAt(ctx, Registration{Module: module, Repository: cur.Repository, Path: cur.Path, Commit: head}, true, true) out.Held, out.Repository, out.Commit, out.Path = true, t.repository, t.commit, pathOf(t.dir) switch { diff --git a/modules/mesh-delivery-indexer/internal/index/registered_test.go b/modules/mesh-delivery-indexer/internal/index/registered_test.go index 6e3ac1e..879c7e5 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered_test.go +++ b/modules/mesh-delivery-indexer/internal/index/registered_test.go @@ -111,6 +111,58 @@ func TestARegisteredModuleIsIndexedFromItsRepository(t *testing.T) { // the gran } }) } + + // Issue 1126: mesh-catalog#535 changed mesh-vault, but its merged message never reached the indexer, so the record + // stood at the merge before. A take-in of the held module reads it again at the source's head, only when the record is + // behind that head: a head older than the record (a merge applied before the source moved), or one whose order the forge + // cannot say, is never written back. + for _, tc := range []struct { + name string + contains func(context.Context, string, string, string) (bool, error) + moved bool + }{ + {name: "a held module behind the source's head is read again", moved: true, + contains: func(_ context.Context, _, commit, in string) (bool, error) { return commit == merge && in == h2, nil }}, + {name: "a source head older than the record is not read", contains: func(context.Context, string, string, string) (bool, error) { return false, nil }}, + {name: "a forge that cannot say the order reads nothing", contains: func(context.Context, string, string, string) (bool, error) { + return false, errors.New("timeout") + }}, + {name: "no forge to ask reads nothing"}, + } { + t.Run("take-in: "+tc.name, func(t *testing.T) { + ctx := context.Background() + ix, mods := modulesIndexer(t) + ix.Manifests = newMem() + if err := ix.Apply(ctx, mergedMsg("novox/mesh-catalog", map[string]string{"modules/a": "treeA"})); err != nil { + t.Fatal(err) + } + ix.source = sourceNow{rev: 2, heads: map[string]string{"novox/mesh-catalog": h2}} + ix.Contains = tc.contains + var asked []string + ix.Read = func(_ context.Context, _, sha, dir string) (RepoRead, error) { + asked = append(asked, sha+":"+dir) + return RepoRead{Root: "r1", Trees: map[string]string{"modules/a": "treeA2"}, Manifests: map[string]string{"modules/a": `{"module":"a","version":"2"}`}}, nil + } + got := ix.TakeIn(ctx, "a") + m, ok := moduleOf(t, mods, "a") + if !tc.moved { + if got.Stored || !got.Held || len(asked) != 0 || !strings.Contains(got.Answer, "another head") || !ok || m.Commit != merge { + t.Errorf("take-in answered %+v, asked %v, modules.a at %.8s", got, asked, m.Commit) + } + return + } + if !got.Stored || !got.Held || got.Commit != h2 || got.Path != "modules/a" || !strings.Contains(got.Answer, "written") { + t.Errorf("take-in answered %+v", got) + } + if !ok || m.Commit != h2 || m.Inputs2 == nil || m.Inputs2.Tree != "treeA2" { + t.Errorf("modules.a = %+v (%v), want it at %.8s with tree treeA2 (asked %v)", m, ok, h2, asked) + } + n := len(asked) // at the head now: a take-in answers held and reads nothing + if again := ix.TakeIn(ctx, "a"); again.Stored || !again.Held || len(asked) != n { + t.Errorf("a take-in of a module at the source's head answered %+v and asked %v", again, asked) + } + }) + } } func mustModule(t *testing.T, m delivery.Module) []byte { @@ -256,32 +308,3 @@ func TestARecordOfAPullRequestDoesNotReplaceTheRegisteredRef(t *testing.T) { // t.Fatalf("registrations = %+v", got) } } - -// Issue 1126: mesh-catalog#535 changed mesh-vault, but its merged message never reached the indexer, so the record stood at -// the merge before. A take-in of the held module reads it again at the head the gate's source names and writes it. -func TestATakeInOfAHeldModuleBehindTheSourceHeadReadsItAgain(t *testing.T) { - ctx := context.Background() - ix, mods := modulesIndexer(t) - ix.Manifests = newMem() - if err := ix.Apply(ctx, mergedMsg("novox/mesh-catalog", map[string]string{"modules/a": "treeA"})); err != nil { - t.Fatal(err) - } - ix.source = sourceNow{rev: 2, heads: map[string]string{"novox/mesh-catalog": h2}} - var asked []string - ix.Read = func(_ context.Context, _, sha, dir string) (RepoRead, error) { - asked = append(asked, sha+":"+dir) - return RepoRead{Root: "r1", Trees: map[string]string{"modules/a": "treeA2"}, Manifests: map[string]string{"modules/a": `{"module":"a","version":"2"}`}}, nil - } - got := ix.TakeIn(ctx, "a") - if !got.Stored || !got.Held || got.Commit != h2 || got.Path != "modules/a" || !strings.Contains(got.Answer, "written") { - t.Errorf("take-in answered %+v", got) - } - if m, ok := moduleOf(t, mods, "a"); !ok || m.Commit != h2 || m.Inputs2 == nil || m.Inputs2.Tree != "treeA2" { - t.Errorf("modules.a = %+v (%v), want it at %.8s with tree treeA2 (asked %v)", m, ok, h2, asked) - } - // At the head now: a take-in answers held and reads nothing. - n := len(asked) - if again := ix.TakeIn(ctx, "a"); again.Stored || !again.Held || len(asked) != n { - t.Errorf("a take-in of a module at the source's head answered %+v and asked %v", again, asked) - } -}