indexer: TestIndexerWhole runs the take-in step whole, a restart and a group (ADR 0633, issue 1585 I8a, I8b)

A revision changing the indexer writes only what the named build needs and is taken in when that build starts; a gathered group stays across the restart and its end makes one take-in. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
2026-11-04 23:22:35 +01:00
parent 9c6e70e555
commit ef13e63268
1 changed files with 250 additions and 9 deletions
@@ -9,7 +9,9 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"

@@ -20,6 +22,10 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"git.novox.be/novox/mesh-delivery/sdk/graph"
+
+	"mesh-delivery-indexer/internal/index"
 )

 // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
@@ -42,11 +48,58 @@ accounts {
 // bus with the stream FORGE and the indexer's durable <node>_<module> on it, the buckets the controller makes from the
 // manifest's state (mesh-delivery-indexer_candidates, history 10) and mesh-controller_builds holding build records, and
 // no mesh-delivery-gate_source bucket.
-func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10
-	js, ctx, candidates, _ := startIndexer(t, func(ctx context.Context, builds buildLog) {
+//
+// And the take-in step (ADR 0633, issue 1585 I8a, I8b): the indexer's states pending, verdicts, graph, taken and packages
+// and its stream take-in are there, mesh-build-go reported its toolchain, and the node-engine's composition answers the
+// build of the indexer that runs. A merge changing the indexer's own tree writes only the indexer's entry, its manifest,
+// its verdict and its key of the graph (and the packages it publishes); the rest is pending, said take-in.pending and
+// named by indexer_pending with the build it waits for. A group's first member is gathered meanwhile. Started again as
+// that build, the indexer takes the revision in and leaves the group gathered; the group's second member and its end
+// then make one take-in, said take-in.taken under the group's key.
+func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
+	var running atomic.Pointer[string]
+	first := "fp-the-indexer-that-runs"
+	running.Store(&first)
+	js, ctx, candidates, _, restart := startIndexerRun(t, func(ctx context.Context, builds buildLog) {
 		// The builder's log entries of two builds' ends (ADR 0571 clause 1), as the builder appends them.
 		builds.end(t, "novox/mesh-delivery", "a", "modules/a")
 		builds.end(t, "novox/mesh-catalog", "c", "modules/c")
+	}, func(ctx context.Context, js jetstream.JetStream) {
+		// The states and the stream the controller makes from the indexer's manifest (old sections, ADR 0595 §11).
+		for bucket, history := range map[string]uint8{index.PendingBucket: 1, delivery.VerdictsBucket: delivery.VerdictsHistory,
+			graph.Bucket: 1, delivery.TakenBucket: 1, delivery.PackagesBucket: 1} {
+			if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: history}); err != nil {
+				t.Fatal(err)
+			}
+		}
+		for name, subject := range map[string]string{"TAKE_IN": "mesh.stream.mesh-delivery-indexer.event.take-in.>",
+			"INDEXER_EVENTS": "mesh.mod.mesh-delivery-indexer.event.>"} {
+			if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: []string{subject}}); err != nil {
+				t.Fatal(err)
+			}
+		}
+		tc, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1})
+		if err != nil {
+			t.Fatal(err)
+		}
+		key, _ := delivery.ToolchainKey("n1", "go")
+		value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "go", Version: "go1.27.1", Recipe: "1", At: time.Now()})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := tc.Put(ctx, key, value); err != nil {
+			t.Fatal(err)
+		}
+		// The node-engine's composition on n1 (process.Conn.Call's subjects): the build of the indexer that runs there.
+		answer := func(m *nats.Msg) {
+			raw, _ := json.Marshal(map[string]any{"result": map[string]any{"runs": map[string]string{"fp": *running.Load()}}, "node": "n1"})
+			_ = m.Respond(raw)
+		}
+		for _, subject := range []string{"mesh.seat.node-engine.call.composition.n1.>", "mesh.seat.node-engine.tool.composition.n1"} {
+			if _, err := js.Conn().Subscribe(subject, answer); err != nil {
+				t.Fatal(err)
+			}
+		}
 	})
 	h1, h2 := "0123456789abcdef0123456789abcdef01234567", "89abcdef0123456789abcdef0123456789abcdef"
 	for _, f := range []delivery.Forge{
@@ -83,11 +136,186 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10
 			t.Errorf("%s#%d modules = %v, want [%s]", p.Repository, p.Number, p.Modules, want)
 		}
 	}
+
+	// I8a: pull request 12 merged, changing the indexer's own tree and a's.
+	manifest, err := os.ReadFile("../../module.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	const ix = delivery.IndexerModule
+	m1 := delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: h1,
+		Merge: "1111111111111111111111111111111111111111", Root: "r0",
+		Trees:     map[string]string{"modules/" + ix: "tree-indexer-2", "modules/a": "tree-a-2"},
+		Manifests: map[string]string{"modules/" + ix: string(manifest), "modules/a": `{"module":"a","version":"1"}`},
+		Locks:     map[string]map[string]string{"modules/" + ix: {"go.sum": "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381"}}}
+	merged := func(f delivery.Forge) uint64 {
+		t.Helper()
+		data, err := delivery.EncodeForge(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		event, _ := delivery.ForgeEvent(f.Repository, delivery.ForgeMerged)
+		ack, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event."+event, data)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return ack.Sequence
+	}
+	m1Key := delivery.PendingKey(merged(m1))
+	takeIns, err := js.Stream(ctx, "TAKE_IN")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// said is the last take-in event under kind and key, within 20 s.
+	said := func(kind, key string) (delivery.TakeIn, bool) {
+		event, _ := delivery.TakeInEvent(kind, key)
+		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+			if msg, err := takeIns.GetLastMsgForSubject(ctx, process.ModuleStreamSubject(ix, event)); err == nil {
+				if ti, err := delivery.DecodeTakeIn(msg.Data); err == nil {
+					return ti, true
+				}
+			}
+		}
+		return delivery.TakeIn{}, false
+	}
+	held, ok := said(delivery.TakeInPending, m1Key)
+	if !ok || held.What != delivery.TakeInRevision || held.WaitsFor == "" || held.WaitsFor == first || len(held.Merges) != 1 {
+		t.Fatalf("take-in.pending.%s = %+v (said %v) within 20 s, want the revision waiting for another build than %s", m1Key, held, ok, first)
+	}
+	named := held.WaitsFor
+	get := func(bucket, key string) ([]byte, bool) {
+		t.Helper()
+		kv, err := js.KeyValue(ctx, bucket)
+		if err != nil {
+			t.Fatal(err)
+		}
+		e, err := kv.Get(ctx, key)
+		if err != nil {
+			return nil, false
+		}
+		return e.Value(), true
+	}
+	// Only what building and moving the named indexer needs is written.
+	raw, ok := get(delivery.ModulesBucket, ix)
+	own, err := delivery.DecodeModule(raw)
+	if !ok || err != nil || own.FP2 != named || own.Commit != m1.Merge {
+		t.Fatalf("modules.%s = %+v (%v), want the named build %s at the merge", ix, own, err, named)
+	}
+	if _, ok := get(delivery.ManifestsBucket, own.Manifest); !ok {
+		t.Errorf("manifests.%s is not written", own.Manifest)
+	}
+	if _, ok := get(delivery.VerdictsBucket, own.Manifest); !ok {
+		t.Errorf("the verdict on the indexer's manifest %s is not written", own.Manifest)
+	}
+	if gk, _ := graph.Key(ix); !func() bool { _, ok := get(graph.Bucket, gk); return ok }() {
+		t.Errorf("the indexer's key %s of the module graph is not written", gk)
+	}
+	repoKey, _ := delivery.RepositoryKey(m1.Repository)
+	if _, ok := get(delivery.ModulesBucket, "a"); ok {
+		t.Errorf("modules.a is written while the revision is pending")
+	}
+	if _, ok := get(delivery.TakenBucket, repoKey); ok {
+		t.Errorf("taken.%s is written while the revision is pending", repoKey)
+	}
+	if e, err := candidates.Get(ctx, "feat/x"); err != nil {
+		t.Errorf("the candidate feat/x is gone while the merge of #12 is pending: %v", err)
+	} else if c, err := delivery.DecodeCandidate(e.Value()); err != nil || len(c.Pulls) != 2 {
+		t.Errorf("the candidate feat/x = %+v while the merge of #12 is pending, want its two pull requests", c)
+	}
+
+	// A group's first member arrives meanwhile; its end is still to come (the knowledge base's remark: it stays across the restart).
+	members := []delivery.Head{{Repository: "novox/mesh-catalog", Number: 40, Head: "4040404040404040404040404040404040404040"},
+		{Repository: "novox/mesh-controller", Number: 41, Head: "4141414141414141414141414141414141414141"}}
+	g1 := delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/g", Number: 40, Head: members[0].Head,
+		Merge: "4000000000000000000000000000000000000000", Root: "r0", Trees: map[string]string{"modules/c": "tree-c-2"},
+		Manifests: map[string]string{"modules/c": `{"module":"c","version":"1"}`}, Group: "feat/g", Members: members}
+	g2 := delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-controller", Branch: "feat/g", Number: 41, Head: members[1].Head,
+		Merge: "4100000000000000000000000000000000000000", Root: "r0", Trees: map[string]string{"modules/k": "tree-k-1"},
+		Manifests: map[string]string{"modules/k": `{"module":"k","version":"1"}`}, Group: "feat/g", Members: members}
+	g1Key, groupKey := delivery.PendingKey(merged(g1)), delivery.GroupTakeInKey("feat/g")
+	if gathered, ok := said(delivery.TakeInPending, groupKey); !ok || gathered.What != delivery.TakeInGroup || len(gathered.Merges) != 1 {
+		t.Fatalf("take-in.pending.%s = %+v (said %v) within 20 s, want the group with its first member", groupKey, gathered, ok)
+	}
+	// indexer_pending, the tool over the same state, names both waits.
+	pendingKV, err := js.KeyValue(ctx, index.PendingBucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	waits := func() []delivery.PendingWait {
+		t.Helper()
+		answer, err := index.PendingTools(kvStore{pendingKV}, nil, time.Now)[0].Run(map[string]any{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return answer.(delivery.PendingAnswer).Waits
+	}
+	var w []delivery.PendingWait
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		if w = waits(); len(w) == 2 && w[0].WaitsFor == named && w[1].Said != "" {
+			break
+		}
+	}
+	if len(w) != 2 || w[0].Key != m1Key || w[0].WaitsFor != named || !slices.Contains(w[0].Holds, "a") || !slices.Contains(w[0].Holds, ix) ||
+		w[1].Key != groupKey || w[1].What != delivery.TakeInGroup || len(w[1].Merges) != 1 || !strings.HasSuffix(w[1].Said, "pending."+groupKey) {
+		t.Fatalf("indexer_pending = %+v, want the revision %s waiting for %s holding a and the indexer, then the group gathered", w, m1Key, named)
+	}
+
+	// The named indexer starts (the node-engine now runs that build): it takes the revision in, and leaves the group.
+	running.Store(&named)
+	restart()
+	if taken, ok := said(delivery.TakeInTaken, m1Key); !ok || len(taken.Merges) != 1 || taken.Merges[0].Merge != m1.Merge {
+		t.Fatalf("take-in.taken.%s = %+v (said %v) within 20 s of the start of %s", m1Key, taken, ok, named)
+	}
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		if w = waits(); len(w) == 1 {
+			break
+		}
+	}
+	if raw, ok := get(delivery.ModulesBucket, "a"); !ok || !strings.Contains(string(raw), m1.Merge) {
+		t.Errorf("modules.a = %s after the take-in at the start, want it at the merge %s", raw, m1.Merge)
+	}
+	if _, ok := get(delivery.TakenBucket, repoKey); !ok {
+		t.Errorf("taken.%s is not written by the take-in at the start", repoKey)
+	}
+	if len(w) != 1 || w[0].Key != groupKey || len(w[0].Keys) != 1 || w[0].Keys[0] != g1Key {
+		t.Fatalf("indexer_pending after the start = %+v, want only the group gathered, its first member %s", w, g1Key)
+	}
+	if _, ok := get(delivery.ModulesBucket, "c"); ok {
+		t.Errorf("modules.c is written while its group is gathered")
+	}
+
+	// I8b: the group's second member and its end: one take-in of both, said under the group's key.
+	merged(g2)
+	end, err := delivery.EncodeForgeGroupEnd(delivery.ForgeGroupEnd{Group: "feat/g", Members: []delivery.GroupMerge{
+		{Repository: g1.Repository, Number: g1.Number, Merge: g1.Merge}, {Repository: g2.Repository, Number: g2.Number, Merge: g2.Merge}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event."+delivery.ForgeGroupEndEvent, end); err != nil {
+		t.Fatal(err)
+	}
+	if taken, ok := said(delivery.TakeInTaken, groupKey); !ok || taken.What != delivery.TakeInGroup || len(taken.Merges) != 2 {
+		t.Fatalf("take-in.taken.%s = %+v (said %v) within 20 s of the group's end, want both members", groupKey, taken, ok)
+	}
+	var hasC, hasK bool
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && !(hasC && hasK && len(w) == 0); time.Sleep(50 * time.Millisecond) {
+		_, hasC = get(delivery.ModulesBucket, "c")
+		_, hasK = get(delivery.ModulesBucket, "k")
+		w = waits()
+	}
+	if !hasC || !hasK || len(w) != 0 {
+		t.Errorf("after the group's end: modules.c %v, modules.k %v, indexer_pending %+v: want both taken in and nothing waiting", hasC, hasK, w)
+	}
+	// One take-in: neither member was said as a revision of its own.
+	single, _ := delivery.TakeInEvent(delivery.TakeInPending, g1Key)
+	if _, err := takeIns.GetLastMsgForSubject(ctx, process.ModuleStreamSubject(ix, single)); err == nil {
+		t.Errorf("the group's first member %s was said pending as a revision of its own", g1Key)
+	}
 }

-// startIndexer starts the bus, the stream FORGE with the indexer's durable consumer, the buckets, lets put fill
-// mesh-controller_builds, and runs the program. mesh-delivery-gate_source is never created.
-func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) {
+// startIndexerRun starts the bus, the stream FORGE with the indexer's durable consumer, the buckets, lets put fill
+// mesh-controller_builds, and runs the program; restart starts it again. mesh-delivery-gate_source is never created.
+func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue, func()) {
 	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
 	if err != nil {
 		t.Fatal(err)
@@ -166,11 +394,24 @@ func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...f
 		t.Fatal(err)
 	}
 	env := map[string]string{"MESH_BROKER_FILE": broker}
-	runCtx, stop := context.WithCancel(ctx)
-	done := make(chan error, 1)
-	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
-	t.Cleanup(func() { stop(); <-done })
+	var stop func()
+	start := func() {
+		runCtx, cancel := context.WithCancel(ctx)
+		done := make(chan error, 1)
+		go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
+		stop = func() { cancel(); <-done }
+	}
+	start()
+	t.Cleanup(func() { stop() })
+	// restart stops the program and starts it again on the same bus, as the node-engine does with a new build.
+	restart := func() { stop(); start() }

+	return js, ctx, candidates, modules, restart
+}
+
+// startIndexer is startIndexerRun for a test that never starts the program again.
+func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) {
+	js, ctx, candidates, modules, _ := startIndexerRun(t, put, before...)
 	return js, ctx, candidates, modules
 }