From 73bf290f582bf50e654803574b2e8dee42be114d Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 27 Oct 2026 00:14:41 +0100 Subject: [PATCH 1/3] The indexer fills every repository's taken key from the source's head at start and when the state appears (issue 1280) --- .../cmd/mesh-delivery-indexer/main.go | 9 +++++++ .../internal/index/modules_test.go | 27 +++++++++++++++++++ .../internal/index/taken.go | 21 +++++++++++++++ 3 files changed, 57 insertions(+) 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 7728190..5fa7dc5 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -363,6 +363,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { sourceErrs <- follow(ctx, sw, each) }() } } + // Every repository the source names gets its taken key at start, and again when the state appears, from the source's + // head and through the path a merge uses (issue 1280): a merge heard before the state existed wrote none. + taken.Ready(func(jetstream.KeyValue) { + go func() { + if err := ix.FillTaken(ctx); err != nil { + logger.Printf("filling the taken keys from the source: %v", err) + } + }() + }) // Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's // assignments are settled first, so the toolchains settle through the gate already knowing them. gate := newToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf) diff --git a/modules/mesh-delivery-indexer/internal/index/modules_test.go b/modules/mesh-delivery-indexer/internal/index/modules_test.go index 71bf565..a61073b 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules_test.go +++ b/modules/mesh-delivery-indexer/internal/index/modules_test.go @@ -119,6 +119,33 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value m.Merge = next return ix.Apply(context.Background(), m) }}, + {name: "a merge heard while the taken state was missing is filled from the source when it appears", writes: 2, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next, + then: func(ix *Indexer) error { // issue 1280: the key was missing, the merge was taken in without it + on := ix.Taken.(*memStore) + key, _ := delivery.RepositoryKey(repo) + delete(on.vals, key) + ix.Taken = &memStore{unbound: process.ErrNotYet} + m := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"}) + m.Merge = next + if err := ix.Apply(context.Background(), m); err != nil { + return err + } + if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil { + return err + } + ix.Taken = on + 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, + 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 { + return err + } + return ix.FillTaken(context.Background()) // the source still names merge + }}, {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 { ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == next && in == merge, nil } diff --git a/modules/mesh-delivery-indexer/internal/index/taken.go b/modules/mesh-delivery-indexer/internal/index/taken.go index 576b2a4..325afde 100644 --- a/modules/mesh-delivery-indexer/internal/index/taken.go +++ b/modules/mesh-delivery-indexer/internal/index/taken.go @@ -3,6 +3,9 @@ package index import ( "context" "errors" + "fmt" + "maps" + "slices" "time" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -58,3 +61,21 @@ func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error { } return err } + +// FillTaken writes the taken key of every repository the gate's source names, at the head it names, through writeTaken as a +// live merge does (issue 1280): a merge heard while the state was not on the bus left its repository without a key until +// that repository's next merge. It runs at start and when the state appears. A repository whose key names that head or a +// later one is left; a repository the source does not name is left; the first error is returned after every repository +// was tried. +func (ix *Indexer) FillTaken(ctx context.Context) error { + ix.mu.Lock() + heads := maps.Clone(ix.source.heads) + ix.mu.Unlock() + var first error + for _, repository := range slices.Sorted(maps.Keys(heads)) { + if err := ix.writeTaken(ctx, delivery.Forge{Repository: repository, Merge: heads[repository]}); err != nil && first == nil { + first = fmt.Errorf("%s: %w", repository, err) + } + } + return first +} -- 2.54.0 From 0b980203fec9122d484e9386e8b5e6537d4d20d0 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 27 Oct 2026 00:18:23 +0100 Subject: [PATCH 2/3] Taken keys are filled only from merges the indexer applied or a drained forge stream's heads (issue 1280) --- .../cmd/mesh-delivery-indexer/main.go | 21 ++--- .../internal/index/candidates.go | 6 ++ .../internal/index/modules_test.go | 29 +++++- .../internal/index/taken.go | 89 +++++++++++++++++-- 4 files changed, 125 insertions(+), 20 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 5fa7dc5..24d94ed 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -363,15 +363,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { sourceErrs <- follow(ctx, sw, each) }() } } - // Every repository the source names gets its taken key at start, and again when the state appears, from the source's - // head and through the path a merge uses (issue 1280): a merge heard before the state existed wrote none. - taken.Ready(func(jetstream.KeyValue) { - go func() { - if err := ix.FillTaken(ctx); err != nil { - logger.Printf("filling the taken keys from the source: %v", err) - } - }() - }) // Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's // assignments are settled first, so the toolchains settle through the gate already knowing them. gate := newToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf) @@ -477,7 +468,17 @@ 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) } - cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, ix, logger) }, + // 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 + } + taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) }) + cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, ix, logger); ix.KickFill(ctx) }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) if err != nil { return err diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 4189fc9..80d9abf 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -69,6 +69,9 @@ type Indexer struct { Placement Store // 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 @@ -80,6 +83,9 @@ type Indexer struct { retryAgain atomic.Bool // an event asked for one more RetryUnread after the running one retryWG sync.WaitGroup // the background RetryUnread, for tests to wait on recompute sync.Mutex // one Recompute at a time + untaken map[string]string // repository to the latest merge applied while the taken state was missing (taken.go) + filling atomic.Bool // a FillTaken runs in the background (KickFill) + fillAgain atomic.Bool // a kick arrived while it ran } func (ix *Indexer) logf(format string, args ...any) { diff --git a/modules/mesh-delivery-indexer/internal/index/modules_test.go b/modules/mesh-delivery-indexer/internal/index/modules_test.go index a61073b..8369a2e 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules_test.go +++ b/modules/mesh-delivery-indexer/internal/index/modules_test.go @@ -119,7 +119,7 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value m.Merge = next return ix.Apply(context.Background(), m) }}, - {name: "a merge heard while the taken state was missing is filled from the source when it appears", writes: 2, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next, + {name: "a merge heard while the taken state was missing is filled when the state appears", writes: 2, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next, then: func(ix *Indexer) error { // issue 1280: the key was missing, the merge was taken in without it on := ix.Taken.(*memStore) key, _ := delivery.RepositoryKey(repo) @@ -134,6 +134,24 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value return err } ix.Taken = on + return ix.FillTaken(context.Background()) // no Drained: only the merge applied is written, not the source's word + }}, + {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, @@ -144,6 +162,7 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value if err := ix.Apply(context.Background(), m); err != nil { return err } + ix.Drained = func(context.Context) (bool, error) { return true, nil } return ix.FillTaken(context.Background()) // the source still names merge }}, {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, @@ -182,6 +201,9 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value } } key, _ := delivery.RepositoryKey(repo) + if _, held := taken.vals[key]; !held && tc.taken == "" { + return + } if got, err := delivery.DecodeTaken(taken.vals[key]); err != nil || got.Merge != tc.taken || got.Repository != repo { t.Errorf("taken.%s = %+v (%v), want merge %.8s", key, got, err, tc.taken) } @@ -308,3 +330,8 @@ func TestTwoDirectoriesNamingOneModuleWriteOneAndSayTheOther(t *testing.T) { // t.Errorf("module = %+v, want the held directory kept", m) } } + +func takenKey(repo string) string { + key, _ := delivery.RepositoryKey(repo) + return key +} diff --git a/modules/mesh-delivery-indexer/internal/index/taken.go b/modules/mesh-delivery-indexer/internal/index/taken.go index 325afde..e8082a0 100644 --- a/modules/mesh-delivery-indexer/internal/index/taken.go +++ b/modules/mesh-delivery-indexer/internal/index/taken.go @@ -55,27 +55,98 @@ func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error { }) if errors.Is(err, process.ErrNotYet) { // The state is not on the bus yet (a delivery that adds it, bound when it appears as issue 1790 says): said, and - // the merge's modules are still taken in; the next merge of the repository writes its key once the state appears. - ix.logf("%s at %s: its taken key is not written: %v", f.Repository, f.Merge, err) + // the merge's modules are still taken in; FillTaken writes its key once the state appears. + ix.logf("%s at %s: its taken key is not written yet, FillTaken writes it when the state appears: %v", f.Repository, f.Merge, err) + ix.mu.Lock() + if ix.untaken == nil { + ix.untaken = map[string]string{} + } + ix.untaken[f.Repository] = f.Merge // the latest heard; FillTaken keeps a later key over an earlier merge + ix.mu.Unlock() return nil } return err } -// FillTaken writes the taken key of every repository the gate's source names, at the head it names, through writeTaken as a -// live merge does (issue 1280): a merge heard while the state was not on the bus left its repository without a key until -// that repository's next merge. It runs at start and when the state appears. A repository whose key names that head or a -// later one is left; a repository the source does not name is left; the first error is returned after every repository -// was tried. +// 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. +// +// 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. func (ix *Indexer) FillTaken(ctx context.Context) error { + if ix.Taken == nil { + return nil + } + if _, err := ix.Taken.Keys(ctx); errors.Is(err, process.ErrNotYet) { + return nil + } ix.mu.Lock() + pending := maps.Clone(ix.untaken) heads := maps.Clone(ix.source.heads) ix.mu.Unlock() var first error - for _, repository := range slices.Sorted(maps.Keys(heads)) { - if err := ix.writeTaken(ctx, delivery.Forge{Repository: repository, Merge: heads[repository]}); err != nil && first == nil { + note := func(repository string, err error) { + if err != nil && first == nil { first = fmt.Errorf("%s: %w", repository, err) } } + for _, repository := range slices.Sorted(maps.Keys(pending)) { + err := ix.writeTaken(ctx, delivery.Forge{Repository: repository, Merge: pending[repository]}) + if err == nil { + ix.mu.Lock() + if ix.untaken[repository] == pending[repository] { + delete(ix.untaken, repository) + } + ix.mu.Unlock() + } + 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 } + +// KickFill runs FillTaken in the background after a message was applied, one at a time; a kick that arrives while one runs +// makes it run once more. +func (ix *Indexer) KickFill(ctx context.Context) { + if ix.filling.Swap(true) { + ix.fillAgain.Store(true) + return + } + go func() { + for { + ix.fillAgain.Store(false) + if err := ix.FillTaken(ctx); err != nil { + ix.logf("filling the taken keys: %v", err) + } + if !ix.fillAgain.Load() { + ix.filling.Store(false) + if !ix.fillAgain.Load() || ix.filling.Swap(true) { + return + } + } + } + }() +} -- 2.54.0 From 6c79e456a100b75b0ec287c85654ee3965d14c2d Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 27 Oct 2026 00:20:16 +0100 Subject: [PATCH 3/3] The indexer confirms a forge message's ack with the server before it fills taken keys (issue 1280) --- .../cmd/mesh-delivery-indexer/main.go | 8 +++++++- .../cmd/mesh-delivery-indexer/main_test.go | 10 ++++++++-- 2 files changed, 15 insertions(+), 3 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 24d94ed..4c2a403 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -192,6 +192,7 @@ type delivered interface { Subject() string Data() []byte Ack() error + DoubleAck(ctx context.Context) error Term() error } @@ -212,7 +213,12 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, logger *log.Log case err != nil: logger.Printf("%s: %s#%d left for redelivery: %v", m.Subject(), f.Repository, f.Number, err) default: - _ = m.Ack() + // Waits for the server, so the consumer's info no longer counts the message as unacknowledged when the fill that + // follows asks whether the stream is applied (issue 1280); a failed ack is redelivered after the ack wait, which + // kicks the fill again. + if err := m.DoubleAck(ctx); err != nil { + logger.Printf("%s: %s#%d applied, its acknowledgement was not confirmed: %v", m.Subject(), f.Repository, f.Number, err) + } } } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index 19b3658..e49c636 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -16,12 +16,18 @@ type fakeMsg struct { subject string data []byte acked, termd int + doubled int // acked by DoubleAck, which waits for the server } func (f *fakeMsg) Subject() string { return f.subject } func (f *fakeMsg) Data() []byte { return f.data } func (f *fakeMsg) Ack() error { f.acked++; return nil } -func (f *fakeMsg) Term() error { f.termd++; return nil } +func (f *fakeMsg) DoubleAck(context.Context) error { + f.acked++ + f.doubled++ + return nil +} +func (f *fakeMsg) Term() error { f.termd++; return nil } type memStore struct{ vals map[string][]byte } @@ -53,7 +59,7 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part) m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull", data: data} handle(context.Background(), m, ix, log.New(&bytes.Buffer{}, "", 0)) c, err := delivery.DecodeCandidate(st.vals["feat/x"]) - if err != nil || len(c.Pulls) != 1 || c.Pulls[0].Modules[0] != "a" || m.acked != 1 || m.termd != 0 { + if err != nil || len(c.Pulls) != 1 || c.Pulls[0].Modules[0] != "a" || m.acked != 1 || m.doubled != 1 || m.termd != 0 { t.Errorf("candidate %+v err %v acked %d terminated %d", c, err, m.acked, m.termd) } } -- 2.54.0