From 9c6e70e555aa3ae1250cc2d3198f3aed7051271c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 23:19:07 +0100 Subject: [PATCH] indexer: before the pending state is on the bus, take a merge in at once and start without asking the own build (ADR 0633, issue 1585) TestIndexerWhole found the start waiting for ever on a build no node-engine answered, so the forge consumer never started, though nothing can be pending. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- .../internal/index/pending.go | 19 +++++++++++++++---- .../internal/index/pending_test.go | 10 ++++++++++ 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index e88b20c0..6e660acc 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -192,6 +192,14 @@ func revisionWait(kind string, ready []PendingEntry, waitsFor string) delivery.T WaitsFor: waitsFor, Since: sinceOf(ready)} } +// 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. +func (ix *Indexer) pendingOnBus(ctx context.Context) bool { + _, _, _, err := ix.Pending.Get(ctx, delivery.PendingKey(0)) + return !errors.Is(err, process.ErrNotYet) +} + // ownBuildNow is the build this program runs: the one learnt, else asked of the node-engine now. func (ix *Indexer) ownBuildNow(ctx context.Context) string { ix.ownMu.Lock() @@ -246,7 +254,9 @@ func (ix *Indexer) markSaid(ctx context.Context, entries []PendingEntry, t *deli // indexer it names. It answers what was taken in, and a line saying what waits. A message is never written while the // build this program runs is not known (ErrOwnBuildUnknown). func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f delivery.Forge) (Applied, string, error) { - if ix.Pending == nil || ix.Modules == nil { + if ix.Pending == nil || ix.Modules == nil || !ix.pendingOnBus(ctx) { + // The state this delivery brings is not on the bus yet (issue 1790): the merge is taken in at once, as before, + // whether or not the build this program runs is known. a, err := ix.ApplyMessage(ctx, f) return a, "", err } @@ -295,7 +305,7 @@ func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f delivery.Forge) (Ap // the member arrived. An end of a group taken in already (every member's merge at or behind its repository's taken // merge) changes nothing and is said: it is never delivered again for ever. func (ix *Indexer) EndGroup(ctx context.Context, end delivery.ForgeGroupEnd) (Applied, string, error) { - if ix.Pending == nil { + if ix.Pending == nil || !ix.pendingOnBus(ctx) { return Applied{}, "no pending state: the group's merges were taken in one by one", nil } own := ix.ownBuildNow(ctx) @@ -388,8 +398,9 @@ func (ix *Indexer) takenAlready(ctx context.Context, m delivery.GroupMerge) (boo // pending, taking in the whole revisions that name this build; a group still being gathered stays, and a revision // naming yet another build is held for that one. A claim left by an indexer that ran before is void here. func (ix *Indexer) TakeInAtStart(ctx context.Context, wait func(context.Context) bool) (Applied, string, error) { - if ix.Pending == nil { - return Applied{}, "", nil + if ix.Pending == nil || !ix.pendingOnBus(ctx) { + // Nothing can wait in a state not on the bus: the start asks no build and does not wait for one. + return Applied{}, "the pending state is not on the bus yet: nothing waits", nil } own := ix.ownBuildNow(ctx) for own == "" { diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 21e6870a..cca2d25e 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -10,6 +10,7 @@ import ( "testing" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" ) // takeInBed is an indexer holding the catalogue's avahi and the indexer's own module, both with a fingerprint, its @@ -112,6 +113,15 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if !errors.Is(err, ErrOwnBuildUnknown) || len(b.keys()) != 0 || b.commit(t, "avahi") == sha('b') { t.Errorf("err %v, pending %v, avahi at %s: want ErrOwnBuildUnknown and nothing written", err, b.keys(), b.commit(t, "avahi")) } + // Before the pending state is on the bus nothing waits: the start asks no build, and the merge is taken in at once, + // as before ADR 0633, though the build is unknown (TestIndexerWhole found the start waiting on it for ever). + b.ix.Pending = &memStore{unbound: process.ErrNotYet} + if _, _, err := b.ix.TakeInAtStart(ctx, func(context.Context) bool { t.Fatal("the start waits for the build"); return false }); err != nil { + t.Fatal(err) + } + if _, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", "")); err != nil || b.commit(t, "avahi") != sha('b') { + t.Errorf("no pending state: err %v, avahi at %s: want the merge taken in at once", err, b.commit(t, "avahi")) + } }) 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)