D6: a take-in waits for the verdicts state, so every manifest it writes is judged in it before take-in.taken; its bind evaluates pending again (issue 1966 note 154, ADR 0668 §4)

A take-in whose judge pass fails is judged again at the next message heard, as a failed start pass is.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-10 03:38:07 +01:00
parent d3f68c5372
commit 9296db5538
4 changed files with 47 additions and 4 deletions
@@ -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)
@@ -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)
@@ -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
@@ -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())
}