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 acd2104c..d6bce20d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -591,6 +591,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Every manifest judged at the start, so a judge that changed with this program judges every one again (ADR 0578 // §1), once the verdicts state is there; each merge judges them again (index.ApplyMessage). verdicts.Ready(func(jetstream.KeyValue) { + // A take-in held because the verdicts state was not on the bus yet is evaluated again once the start's pass ended, + // whatever it answered (issue 1966 note 154). + defer func() { + if err := ix.PendingAgain(ctx); err != nil { + logger.Printf("take-in: pending is not evaluated again at the verdicts state's bind: %v", err) + } + }() // A pass that fails (a read the bus did not answer) is judged again at the next message or source revision the // indexer hears (index.kickJudge): no timer. j, err := ix.JudgeAtStart(ctx) diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 0849ba47..191eca68 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -245,7 +245,9 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f forgemsg.Forge) (Applied, // Every manifest judged again: a rule between manifests can turn on the one this merge moved (ADR 0578 §1). // What this take-in newly wrote is fresh: judged in full (ADR 0579 §1). if _, err := ix.JudgeAll(ctx, w.fresh); err != nil { - ix.logf("judge: the manifests are not judged after %s#%d: %v", f.Repository, f.Number, err) + // Judged again at the next message or source revision heard (kickJudge), as a failed start pass is. + ix.judgePending.Store(true) + ix.logf("judge: the manifests are not judged after %s#%d: %v; judged again at the next message heard", f.Repository, f.Number, err) } } ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 08128919..46b42f5c 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -355,6 +355,15 @@ func revisionWait(kind string, ready []PendingEntry, waitsFor string) takein.Tak // pendingOnBus says whether the pending state is on the bus: a state this delivery brings is bound when it appears // (issue 1790), and until then every merge is taken in at once, so the build this program runs is not needed. A read // that fails otherwise counts as on the bus: the step that follows answers the error. +// errVerdictsNotYet is a take-in held because the verdicts state is not on the bus yet (takeInWholeBy). +var errVerdictsNotYet = errors.New("the verdicts state is not on the bus yet: nothing is taken in until it is") + +// verdictsOnBus says the verdicts state is on the bus (a take-in judges what it writes there). +func (ix *Indexer) verdictsOnBus(ctx context.Context) bool { + _, _, _, err := ix.Verdicts.Get(ctx, "on-the-bus") + return !errors.Is(err, process.ErrNotYet) +} + func (ix *Indexer) pendingOnBus(ctx context.Context) bool { _, _, _, err := ix.Pending.Get(ctx, takein.PendingKey(0)) return !errors.Is(err, process.ErrNotYet) @@ -696,7 +705,7 @@ func (ix *Indexer) UnreadAgain(ctx context.Context) error { } // PendingAgain evaluates pending again after something a fingerprint is computed from changed, a toolchain report -// (ADR 0633 §1, §3): a revision that now names the running build is taken in, and one that names another build is held +// (ADR 0633 §1, §3), or the verdicts state bound (issue 1966): a revision that now names the running build is taken in, and one that names another build is held // for that one. It does nothing while the pending state is not on the bus or the build this program runs is not known. func (ix *Indexer) PendingAgain(ctx context.Context) error { if ix.Pending == nil || !ix.pendingOnBus(ctx) { @@ -713,7 +722,7 @@ func (ix *Indexer) PendingAgain(ctx context.Context) error { return nil } if err == nil && words != "" { - ix.logf("take-in: evaluated again after a toolchain report: %s", words) + ix.logf("take-in: evaluated again: %s", words) } return err } @@ -760,6 +769,9 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie if errors.Is(err, errClaimLost) { return a, "claimed by another take-in", nil } + if errors.Is(err, errVerdictsNotYet) { + return a, "the verdicts state is not on the bus yet: the revision waits for it, taken in when it binds", nil + } return a, ix.waitingLine(w), err } return Applied{}, ix.waitingLine(w), ix.holdFor(ctx, ready, names, own, dry) @@ -1002,6 +1014,13 @@ func (ix *Indexer) takeInWhole(ctx context.Context, entries []PendingEntry, own, // takeInWholeBy is takeInWhole; who and why, of a hand act, are carried on the take-in.taken events. func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, own, by, who, why string) (Applied, error) { + if ix.Verdicts != nil && !ix.verdictsOnBus(ctx) && len(entries) > 0 { + // Every manifest a take-in writes is judged in it and its verdict said before take-in.taken (issue 1966 note 154, + // ADR 0668 §4): with the verdicts state not on the bus yet, nothing is taken in; its bind evaluates pending again. + ix.logf("take-in: %d merges wait, oldest %s: the state %s is not on the bus yet, so nothing is taken in until it is, and no manifest is written without its verdict", + len(entries), entries[0].Key, VerdictsBucket) + return Applied{}, errVerdictsNotYet + } claimed, ok := ix.claim(ctx, entries, own, by) if !ok { return Applied{}, errClaimLost diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index e41d3c03..91d401c3 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -206,8 +206,23 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) t.Run("T7 and T3 a merge alone, an indexer change, a merge behind it and a revert", func(t *testing.T) { b := newTakeInBed(t) own, avahi0 := b.fp(t, Module), b.fp(t, "avahi") - // A merge with no change to the indexer and nothing pending is taken in alone, as before. + // Issue 1966 note 154 (ADR 0668 §4): with the verdicts state not on the bus yet, nothing is taken in and no manifest + // is written without its verdict; its bind evaluates pending again, which takes the merge in with its verdicts. + verdictsBound := b.ix.Verdicts + b.ix.Verdicts = &memStore{unbound: process.ErrNotYet} b.arrive(t, 1, b.merge(20, sha('b'), "tree-avahi-2", "")) + if b.fp(t, "avahi") != avahi0 || !reflect.DeepEqual(b.keys(), []string{takein.PendingKey(1)}) { + t.Fatalf("verdicts not on the bus: avahi %s (was %s), pending %v: want the merge waiting, nothing written", b.fp(t, "avahi"), avahi0, b.keys()) + } + b.ix.Verdicts = verdictsBound + judgedBefore := len(verdictsBound.(*memStore).vals) + if err := b.ix.PendingAgain(ctx); err != nil { + t.Fatal(err) + } + if n := len(verdictsBound.(*memStore).vals); n <= judgedBefore { + t.Fatalf("taken in at the bind: %d verdicts (were %d), want the new manifest's", n, judgedBefore) + } + // A merge with no change to the indexer and nothing pending is taken in alone, as before. if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || len(b.said) != 0 { t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in", avahi1, avahi0, b.keys(), b.events()) }