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 0b4258d..43b411e 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -253,6 +253,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + taken, err := process.EssentialBucket(ctx, js, "indexer", delivery.TakenBucket, "say the merge a repository's modules were taken in at") + if err != nil { + return err + } // The controller's build records only name the directory of a module whose key names none (fillPaths): without them // such a module waits for its path, every other is indexed (issue 1945). builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf) @@ -286,7 +290,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } defer calls.Close() - ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Builds: known, Logf: logger.Printf, + ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: kvStore{taken}, Builds: known, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call)} // The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 ยง1): its key // main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 592b309..b3fd257 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -9,6 +9,7 @@ import ( "strings" "sync" "sync/atomic" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -47,7 +48,11 @@ type Indexer struct { Modules Store // Manifests is the manifests bucket; nil writes no manifests (issue 1449). Manifests Store - Builds *Builds + // Taken is the taken bucket: per repository the last merge taken in (taken.go, ADR 0556); nil writes none. + Taken Store + // Now is the clock of a taken value; nil is time.Now. + Now func() time.Time + Builds *Builds // Named asks the forge for the git object of each named path of the module directories given, at a commit // (gitea.gitea_commit_trees with paths; novox/hq ADR 0499 clauses 2 and 3): the indexer is the one reader of a // manifest's paths (ADR 0495 clause 1, the lead's decision L8 of R4e). Nil asks nothing, and a module naming paths @@ -94,6 +99,9 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { if err := ix.applyMerged(ctx, f); err != nil { return err } + if err := ix.writeTaken(ctx, f); err != nil { + return err + } ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945) } var digests map[string]string diff --git a/modules/mesh-delivery-indexer/internal/index/modules_test.go b/modules/mesh-delivery-indexer/internal/index/modules_test.go index 18987ef..c11182b 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules_test.go +++ b/modules/mesh-delivery-indexer/internal/index/modules_test.go @@ -94,22 +94,39 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value // writes is the values written by then; commit is each module's commit after. writes int commit map[string]string + // taken is the merge the repository's taken key names after then (issue 1280, ADR 0556 clause 1). + taken string }{ - {name: "the same merge again", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, + {name: "the same merge again", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge, then: func(ix *Indexer) error { return ix.Apply(context.Background(), mergedMsg(repo, three)) }}, - {name: "a merge touching one of three modules", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, + {name: "a merge touching one of three modules", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next, then: func(ix *Indexer) error { m := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"}) m.Merge = next return ix.Apply(context.Background(), m) }}, - {name: "a later source revision of another repository's merge", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, + {name: "a merge touching none of them is still taken in", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: next, + then: func(ix *Indexer) error { + m := mergedMsg(repo, three) + m.Merge = next + return ix.Apply(context.Background(), m) + }}, + {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 } + m := mergedMsg(repo, three) + m.Merge = next + return ix.Apply(context.Background(), m) + }}, + {name: "a later source revision of another repository's merge", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge, then: func(ix *Indexer) error { return ix.ApplySource(context.Background(), 9, delivery.Source{Group: "g", Heads: map[string]string{repo: merge, "novox/mesh-catalog": otherMerge}}) }}, } { t.Run(tc.name, func(t *testing.T) { ix, mods := modulesIndexer(t) + taken := newMem() + ix.Taken = taken ctx := context.Background() if err := ix.Apply(ctx, mergedMsg(repo, three)); err != nil { t.Fatal(err) @@ -129,6 +146,10 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value t.Errorf("%s = %+v (%v), want commit %.8s", module, m, ok, want) } } + key, _ := delivery.RepositoryKey(repo) + 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) + } }) } } diff --git a/modules/mesh-delivery-indexer/internal/index/taken.go b/modules/mesh-delivery-indexer/internal/index/taken.go new file mode 100644 index 0000000..bcc9a10 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/taken.go @@ -0,0 +1,51 @@ +package index + +import ( + "context" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// What the indexer took in (novox/hq ADR 0556 clause 1, issue 1280): after a merged message is applied whole, the key of +// its repository in the state `taken` names that merge, so a reader of the modules knows they say the repository at that +// merge or later. The indexer is the key's only writer. A merge heard again after a later one (a redelivery) does not put +// the key back: the key moves only to a merge that contains the one it names, as the forge says it; a forge that cannot +// say keeps the key as it is and says so. + +// writeTaken writes the repository's taken key for the merge f carries, unless it names that merge or a later one. +func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error { + if ix.Taken == nil || f.Merge == "" { + return nil + } + key, err := delivery.RepositoryKey(f.Repository) + if err != nil { + ix.logf("%s at %s: no taken key is written: %v", f.Repository, f.Merge, err) + return nil + } + return write(ctx, ix.Taken, key, func(value []byte, found bool) ([]byte, bool, error) { + if found { + if cur, err := delivery.DecodeTaken(value); err == nil { + if cur.Merge == f.Merge { + return nil, false, nil + } + if ix.Contains != nil { + later, err := ix.Contains(ctx, f.Repository, f.Merge, cur.Merge) + if err != nil { + ix.logf("%s: whether %s is in %s cannot be read, so taken.%s stays at %s: %v", f.Repository, f.Merge, cur.Merge, key, cur.Merge, err) + return nil, false, nil + } + if later { + return nil, false, nil // a merge heard again after a later one + } + } + } + } + now := time.Now + if ix.Now != nil { + now = ix.Now + } + data, err := delivery.EncodeTaken(delivery.Taken{Repository: delivery.RepositoryName(f.Repository), Merge: f.Merge, At: now().UTC()}) + return data, false, err + }) +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 46b3589..28f0816 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -80,7 +80,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if m.Module != "mesh-delivery-indexer" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { + if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index a30c24e..a2d545b 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -15,6 +15,7 @@ { "name": "candidates", "history": 10 }, { "name": "manifests", "history": 1 }, { "name": "modules", "history": 20 }, + { "name": "taken", "history": 1 }, { "name": "takeins", "history": 1 } ], "reads": [