From ef13e63268e46797fbd4a2ed205222ceb958da51 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 23:22:35 +0100 Subject: [PATCH] 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 --- .../mesh-delivery-indexer/integration_test.go | 259 +++++++++++++++++- 1 file changed, 250 insertions(+), 9 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index a28b8db5..e8b23054 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -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 _ 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 }