The indexer fills every repository's taken key from the source's head at start and when the state appears (issue 1280)

This commit is contained in:
2026-10-27 00:14:41 +01:00
parent 475bb4bb0b
commit 73bf290f58
3 changed files with 57 additions and 0 deletions
@@ -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)
@@ -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 }
@@ -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
}