The indexer says per repository the last merge it took in, so the view knows its modules speak for that merge (issue 1280, ADR 0556 clause 1)

This commit is contained in:
2026-10-26 20:49:00 +01:00
parent 3dd94f714d
commit 6201f58c20
6 changed files with 91 additions and 6 deletions
@@ -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
@@ -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
@@ -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)
}
})
}
}
@@ -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
})
}
@@ -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) {
@@ -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": [