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 +}