diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index baa62b17..37bf9725 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -8,6 +8,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "os" "strings" @@ -191,8 +192,32 @@ func heldCounts(s stdio.KeptState) func() (index.OldSectionsCount, error) { } } +// keptRead is a state the tools bundle reads, as the index package reads a store: Get and Keys; it writes nothing. +type keptRead struct{ s stdio.KeptState } + +func (k keptRead) Get(_ context.Context, key string) ([]byte, uint64, bool, error) { + e, err := k.s.Get(key) + if err != nil || e == nil { + return nil, 0, false, err + } + return e.Value, e.Revision, true, nil +} + +func (k keptRead) Keys(context.Context) ([]string, error) { return k.s.Keys() } + +func (keptRead) Create(context.Context, string, []byte) error { return errReadOnly } + +func (keptRead) Update(context.Context, string, []byte, uint64) error { return errReadOnly } + +func (keptRead) Delete(context.Context, string, uint64) error { return errReadOnly } + +var errReadOnly = errors.New("the tools bundle only reads this state") + func main() { - if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...)); err != nil { + ask := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second) + tools := append(Tools(ask, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...) + tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, ask, time.Now)...) + if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err) os.Exit(1) } 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 } 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 0d896889..ccdbe7ac 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -16,6 +16,7 @@ import ( "log" "os" "os/signal" + "strings" "sync/atomic" "syscall" "time" @@ -153,6 +154,30 @@ type delivered interface { Metadata() (*jetstream.MsgMetadata, error) } +// handleGroupEnd takes a group's end in (index.EndGroup): a body that does not read is terminated and said; an end +// whose members are not all held yet, or while the build this program runs is not known, is left for redelivery. +func handleGroupEnd(ctx context.Context, m delivered, ix *index.Indexer, h *index.Heard, logger *log.Logger) { + end, err := delivery.DecodeForgeGroupEnd(m.Data()) + if err != nil { + logger.Printf("%s: not a group's end, dropped: %v", m.Subject(), err) + h.Outcome, h.Words = index.HeardDropped, err.Error() + _ = m.Term() + return + } + h.Kind, h.Branch = delivery.ForgeGroupEndEvent, end.Group + applied, words, err := ix.EndGroup(ctx, end) + h.Written, h.Late = applied.Written, applied.Late + if err != nil { + logger.Printf("%s: the end of the group %q left for redelivery: %v", m.Subject(), end.Group, err) + h.Outcome, h.Words = index.HeardLeft, err.Error() + return + } + h.Outcome, h.Words = index.HeardApplied, words + if err := m.DoubleAck(ctx); err != nil { + logger.Printf("%s: the end of the group %q applied, its acknowledgement was not confirmed: %v", m.Subject(), end.Group, err) + } +} + // handle takes one message of FORGE into the candidates and acknowledges it. A message it can never read is terminated, // never redelivered, and the log names its subject and why. A write that failed stays unacknowledged: the server // delivers it again (the server's ack wait is the only clock). Each message handled is written to the heard state @@ -171,6 +196,11 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Sto logger.Printf("%s: sequence %d is not written to the heard state: %v", h.Subject, h.Sequence, err) } }() + // A delivery group's end (novox/hq ADR 0633 §4) has a body of its own: told apart by its name, before the decode. + if strings.HasSuffix(m.Subject(), "."+delivery.ForgeGroupEndEvent) { + handleGroupEnd(ctx, m, ix, &h, logger) + return + } f, err := delivery.DecodeForge(m.Data()) if err != nil { logger.Printf("%s: not a forge message, dropped: %v", m.Subject(), err) @@ -182,7 +212,13 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Sto if f.Merge != "" { h.Commit = f.Merge } - applied, err := ix.ApplyMessage(ctx, f) + var applied index.Applied + if f.Kind == delivery.ForgeMerged && h.Sequence > 0 { + // A merge is taken in only by the indexer its revision names (ADR 0633 §1): kept in pending until then. + applied, h.Words, err = ix.Arrive(ctx, h.Sequence, f) + } else { + applied, err = ix.ApplyMessage(ctx, f) + } h.Written, h.Late = applied.Written, applied.Late switch { case errors.Is(err, index.ErrUnreadable): @@ -308,7 +344,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), ReadOpen: openVia(calls.Call), Packages: awaitedStore{packages}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), - OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph}} + OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph}, + // The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear; + // until then every merge is taken in at once, as before. + Pending: awaitedStore{process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)}, + SayTakeIn: sayTakeIn(js, cred)} // The log's entries at the start are read before any fingerprint is computed or registered module indexed (below); each // new one indexes the module it names when that module is registered and the bucket holds none (registered.go). var started atomic.Bool @@ -458,6 +498,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer pc.Stop() } + // The pending revisions that name this build are taken in now, oldest first, before the consumer on the forge bridge's + // stream starts (ADR 0633 §2); the build this program runs is asked until it is known, said every minute. A hand + // take-in (indexer_take_in_pending) is answered meanwhile: the takeins watch runs already. + if _, words, err := ix.TakeInAtStart(ctx, func(ctx context.Context) bool { + // A context's bound between two asks of the node-engine, no timer (ADR 0437 §8), as askTakeIn bounds its wait. + again, cancel := context.WithTimeout(ctx, time.Minute) + defer cancel() + <-again.Done() + return ctx.Err() == nil + }); err != nil { + return fmt.Errorf("the take-in at start: %w", err) + } else if words != "" { + logger.Printf("take-in at start: still waiting: %s", words) + } cons, err := cred.EventConsumer(ctx, js, forgeStream) if err != nil { // Essential: the forge's merges are what the indexer indexes. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index eb70b62f..f42d7972 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -102,6 +102,24 @@ func TestAMessageThatIsNoForgeMessageIsTerminatedAndLogged(t *testing.T) { // 12 if h := heardOf(t, heard); h.Outcome != index.HeardDropped || h.Sequence != 41 || h.Words == "" { t.Errorf("heard = %+v", h) } + // 1585/T9 (ADR 0633 §4): a group's end is told apart by its name before the forge decode: one that reads is acked + // (an indexer with no pending state takes the members in one by one), one that does not is dropped as a group's end. + for _, c := range []struct { + data string + termd int + logged string + }{{`{"v":1,"group":"feat/x","members":[{"repository":"novox/mesh-sdk","number":3,"merge":"` + head + `"}]}`, 0, ""}, + {`{"v":1,"group":"feat/x"}`, 1, "not a group's end"}} { + logs.Reset() + m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event." + delivery.ForgeGroupEndEvent, data: []byte(c.data)} + handle(context.Background(), m, &index.Indexer{Store: st, Builds: index.NewBuilds()}, heard, log.New(&logs, "", 0)) + if m.termd != c.termd || m.acked+m.termd != 1 || !strings.Contains(logs.String(), c.logged) { + t.Errorf("group end %s: terminated %d acked %d log %q", c.data, m.termd, m.acked, logs.String()) + } + if h := heardOf(t, heard); h.Kind != delivery.ForgeGroupEndEvent && c.termd == 0 { + t.Errorf("heard of a group's end = %+v", h) + } + } } func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index 7a98edb7..2ef5b43e 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -16,6 +16,8 @@ type takerIn interface { TakeIn(ctx context.Context, module string) index.TakenIn TakeInRepository(ctx context.Context, repository string) index.TakenIn JudgeAgain(ctx context.Context, module, digest string) index.TakenIn + TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn + DropPending(ctx context.Context, key, by, why string) index.TakenIn } // takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the @@ -34,7 +36,13 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream return } asked, r := "take-in", index.TakenIn{} - if a.Repository != "" { + if a.Pending != "" && a.Drop { + asked, r = "drop of a pending merge", ix.DropPending(ctx, a.Pending, a.By, a.Why) + a.Module = a.Pending + } else if a.Pending != "" { + asked, r = "hand take-in of what waits", ix.TakeInPendingByHand(ctx, a.Pending, a.By, a.Why) + a.Module = a.Pending + } else if a.Repository != "" { asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository) a.Module = a.Repository } else if a.Judge { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go index 79e1b316..495cb684 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go @@ -62,3 +62,28 @@ func sayAccepted(js jetstream.JetStream, cred process.Account, logf func(string, } } } + +// sayTakeIn publishes a take-in event on the indexer's own stream take-in (novox/hq ADR 0633 §5): +// `mesh.stream..event.take-in..`. A publish that fails is the caller's to say. +func sayTakeIn(js jetstream.JetStream, cred process.Account) func(context.Context, delivery.TakeIn) error { + return func(ctx context.Context, t delivery.TakeIn) error { + event, err := delivery.TakeInEvent(t.Kind, t.Key) + if err != nil { + return err + } + data, err := delivery.EncodeTakeIn(t) + if err != nil { + return err + } + msg := nats.NewMsg(process.ModuleStreamSubject(cred.Module, event)) + msg.Header.Set("x-source", cred.Module) + msg.Header.Set("x-node", cred.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + _, err = js.PublishMsg(pub, msg) + return err + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 85d1c415..75b6ea90 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -121,6 +121,12 @@ type Indexer struct { ownMu sync.Mutex // guards ownFP ownFP string // the fingerprint of the build this program runs, asked at the start pass (learnOwnBuild) judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go) + // Pending is the state of merges held for the indexer their revision names (pending.go, novox/hq ADR 0633 §1); nil + // takes every merge in at once, as before the take-in step. + Pending Store + // SayTakeIn publishes a take-in event on the indexer's own stream take-in (ADR 0633 §5); nil says none. + SayTakeIn func(ctx context.Context, t delivery.TakeIn) error + pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go) } func (ix *Indexer) logf(format string, args ...any) { diff --git a/modules/mesh-delivery-indexer/internal/index/overlay.go b/modules/mesh-delivery-indexer/internal/index/overlay.go new file mode 100644 index 00000000..6405f0a4 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/overlay.go @@ -0,0 +1,110 @@ +package index + +import ( + "context" + "maps" + "slices" + "sync" +) + +// overlayStore is a store whose writes stay in memory over the store it reads through (novox/hq ADR 0633 §1): a merge +// applied to it is computed whole, its every fingerprint and verdict, and nothing reaches the bus. A key it wrote reads +// as written, a key it deleted as absent, every other as the store under it holds it. Its compare-and-set holds against +// what it reads: a key it wrote has a revision of its own, counted above every revision under it. A nil store under +// it holds nothing. +type overlayStore struct { + under Store + mu sync.Mutex + vals map[string]overlaid + next uint64 +} + +type overlaid struct { + value []byte + revision uint64 + deleted bool +} + +func newOverlay(under Store) *overlayStore { + return &overlayStore{under: under, vals: map[string]overlaid{}, next: 1 << 62} +} + +func (o *overlayStore) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) { + o.mu.Lock() + v, ok := o.vals[key] + o.mu.Unlock() + if ok { + if v.deleted { + return nil, v.revision, false, nil + } + return slices.Clone(v.value), v.revision, true, nil + } + if o.under == nil { + return nil, 0, false, nil + } + return o.under.Get(ctx, key) +} + +// put checks the revision read against what the key holds now and writes it. +func (o *overlayStore) put(ctx context.Context, key string, value []byte, revision uint64, create, del bool) error { + _, rev, found, err := o.Get(ctx, key) + if err != nil { + return err + } + o.mu.Lock() + defer o.mu.Unlock() + switch { + case create && found, !create && (!found || rev != revision): + return ErrConflict + } + o.next++ + o.vals[key] = overlaid{value: slices.Clone(value), revision: o.next, deleted: del} + return nil +} + +func (o *overlayStore) Create(ctx context.Context, key string, value []byte) error { + return o.put(ctx, key, value, 0, true, false) +} + +func (o *overlayStore) Update(ctx context.Context, key string, value []byte, revision uint64) error { + return o.put(ctx, key, value, revision, false, false) +} + +func (o *overlayStore) Delete(ctx context.Context, key string, revision uint64) error { + return o.put(ctx, key, nil, revision, false, true) +} + +func (o *overlayStore) Keys(ctx context.Context) ([]string, error) { + var keys []string + if o.under != nil { + var err error + if keys, err = o.under.Keys(ctx); err != nil { + return nil, err + } + } + o.mu.Lock() + defer o.mu.Unlock() + keys = slices.DeleteFunc(keys, func(k string) bool { _, ok := o.vals[k]; return ok }) + for k, v := range o.vals { + if !v.deleted { + keys = append(keys, k) + } + } + slices.Sort(keys) + return keys, nil +} + +// written is what the overlay holds of key: its value, and whether it wrote or deleted it at all. +func (o *overlayStore) written(key string) (value []byte, deleted, ok bool) { + o.mu.Lock() + defer o.mu.Unlock() + v, ok := o.vals[key] + return slices.Clone(v.value), v.deleted, ok +} + +// changed is every key the overlay wrote or deleted, in order. +func (o *overlayStore) changed() []string { + o.mu.Lock() + defer o.mu.Unlock() + return slices.Sorted(maps.Keys(o.vals)) +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go new file mode 100644 index 00000000..6e660acc --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -0,0 +1,861 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "slices" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" + + "git.novox.be/novox/mesh-delivery/sdk/graph" +) + +// The take-in step (novox/hq ADR 0633): a source revision is taken in only by the indexer it names. A merged message is +// kept in the indexer's own state pending, one key per merge, until the take-in that ends it; a revision naming another +// build of the indexer writes only what building and moving that indexer needs, and the rest stays pending for it. +const ( + // PendingState is the indexer's state of held merges: the bucket mesh-delivery-indexer_pending, keyed + // delivery.PendingKey, so the key order is the merge order. An old `state` entry (the indexer is foundation, ADR + // 0595 §11), history 1, reached by the indexer alone. + PendingState = "pending" + PendingBucket = delivery.IndexerModule + "_" + PendingState + // PendingTool reads it (ADR 0633 §5); TakeInPendingTool and DropPendingTool steer it. + PendingTool = "indexer_pending" + TakeInPendingTool = "indexer_take_in_pending" + DropPendingTool = "indexer_drop_pending" +) + +// errClaimLost says another take-in claimed a pending merge first: this one took nothing in. +var errClaimLost = errors.New("another take-in claimed the merges first, so this one took nothing in") + +// NoBuild opens what a revision waits for when it gives the indexer no fingerprint: the reason follows. No indexer +// runs such a build, so the revision waits until a later merge gives it one or a hand take-in (indexer_take_in_pending). +const NoBuild = "no build: " + +// ErrOwnBuildUnknown is what a merged message answers while the indexer cannot learn the build it runs (ADR 0633 §1): +// the message is left unacknowledged, so the bus delivers it again, and nothing is written. +var ErrOwnBuildUnknown = errors.New("the build of the indexer this program runs is not known") + +// Pending is one value of the pending state: the merged message whole, as the forge bridge encoded it, the group it +// was merged in (Ended once the group's end was heard), since when it is held, the indexer build its revision waits +// for (absent while it waits for nothing yet, or behind a group being gathered), the build of the indexer that last +// wrote it, the wait it was last said under on the take-in stream (Said, SaidAt), and Taking, set by a take-in that +// claimed it: a second indexer that finds it claimed takes nothing in (ADR 0633 §1: never two take-ins of one merge). +type Pending struct { + V int `json:"v"` + Forge json.RawMessage `json:"forge"` + Group string `json:"group,omitempty"` + Ended bool `json:"ended,omitempty"` + Since time.Time `json:"since"` + WaitsFor string `json:"waits-for,omitempty"` + LastWriter string `json:"last-writer"` + Said string `json:"said,omitempty"` + SaidKind string `json:"said-kind,omitempty"` + SaidAt time.Time `json:"said-at,omitzero"` + Taking string `json:"taking,omitempty"` +} + +// PendingEntry is one key of the pending state as read: its key, revision, value and the merged message decoded. +type PendingEntry struct { + Key string + Revision uint64 + Value Pending + Forge delivery.Forge +} + +// DecodePending reads one pending value and its merged message. +func DecodePending(data []byte) (Pending, delivery.Forge, error) { + var p Pending + if err := json.Unmarshal(data, &p); err != nil { + return p, delivery.Forge{}, fmt.Errorf("not a pending value: %w", err) + } + if p.V != 1 { + return p, delivery.Forge{}, fmt.Errorf("a pending value of version %d, not 1", p.V) + } + f, err := delivery.DecodeForge(p.Forge) + if err != nil { + return p, f, fmt.Errorf("its merged message: %w", err) + } + if f.Kind != delivery.ForgeMerged { + return p, f, fmt.Errorf("a pending value holds a %s message, not a merged one", f.Kind) + } + return p, f, nil +} + +func encodePending(p Pending) ([]byte, error) { + p.V = 1 + return json.Marshal(p) +} + +// ReadPending is every value the pending state holds, in key order (the merge order); a key that does not read is +// said through logf and left. +func ReadPending(ctx context.Context, st Store, logf func(string, ...any)) ([]PendingEntry, error) { + keys, err := st.Keys(ctx) + if err != nil { + return nil, fmt.Errorf("listing %s: %w", PendingState, err) + } + slices.Sort(keys) + var out []PendingEntry + for _, key := range keys { + value, rev, found, err := st.Get(ctx, key) + if err != nil { + return nil, fmt.Errorf("%s.%s: %w", PendingState, key, err) + } + if !found { + continue + } + p, f, err := DecodePending(value) + if err != nil { + if logf != nil { + logf("%s.%s is left: %v", PendingState, key, err) + } + continue + } + out = append(out, PendingEntry{Key: key, Revision: rev, Value: p, Forge: f}) + } + return out, nil +} + +// Waits is the pending state read as what waits (ADR 0633 §3, §4): Ready are the merges no group still being gathered +// holds back, in merge order, which are one revision together; Groups are the groups still being gathered, each with +// its members held so far; Behind are the merges in a gathered group's member repositories that wait behind it, by the +// group's take-in key. +type Waits struct { + Ready []PendingEntry + Groups map[string][]PendingEntry + Behind map[string][]PendingEntry +} + +// WaitsOf works out what waits from the entries, in key order. A member of a group whose end was not heard is +// gathered; from the first such member on, a later merge in one of the group's members' repositories waits behind it, +// because its tree holds that member's commit. Every other merge is ready. +func WaitsOf(entries []PendingEntry) Waits { + w := Waits{Groups: map[string][]PendingEntry{}, Behind: map[string][]PendingEntry{}} + blocked := map[string]string{} // repository, lower case, to the group key it waits behind + for _, e := range entries { + if e.Value.Group != "" && !e.Value.Ended { + w.Groups[e.Value.Group] = append(w.Groups[e.Value.Group], e) + for _, m := range e.Forge.Members { + if _, ok := blocked[strings.ToLower(m.Repository)]; !ok { + blocked[strings.ToLower(m.Repository)] = delivery.GroupTakeInKey(e.Value.Group) + } + } + continue + } + if g, ok := blocked[strings.ToLower(e.Forge.Repository)]; ok { + w.Behind[g] = append(w.Behind[g], e) + continue + } + w.Ready = append(w.Ready, e) + } + return w +} + +// mergesOf are the merges of entries, oldest first. +func mergesOf(entries []PendingEntry) []delivery.GroupMerge { + out := make([]delivery.GroupMerge, len(entries)) + for i, e := range entries { + out[i] = delivery.GroupMerge{Repository: e.Forge.Repository, Number: e.Forge.Number, Merge: e.Forge.Merge} + } + return out +} + +func sinceOf(entries []PendingEntry) time.Time { + var at time.Time + for _, e := range entries { + if at.IsZero() || e.Value.Since.Before(at) { + at = e.Value.Since + } + } + return at +} + +// groupWait is the take-in body of a group being gathered. +func groupWait(kind, group string, held []PendingEntry) delivery.TakeIn { + t := delivery.TakeIn{Kind: kind, Key: delivery.GroupTakeInKey(group), What: delivery.TakeInGroup, Group: group, + Merges: mergesOf(held), Since: sinceOf(held)} + if len(held) > 0 { + t.Members = held[len(held)-1].Forge.Members + } + return t +} + +// revisionWait is the take-in body of a revision: its key is its oldest merge's. +func revisionWait(kind string, ready []PendingEntry, waitsFor string) delivery.TakeIn { + return delivery.TakeIn{Kind: kind, Key: ready[0].Key, What: delivery.TakeInRevision, Merges: mergesOf(ready), + WaitsFor: waitsFor, Since: sinceOf(ready)} +} + +// pendingOnBus says whether the pending state is on the bus: a state this delivery brings is bound when it appears +// (issue 1790), and until then every merge is taken in at once, so the build this program runs is not needed. A read +// that fails otherwise counts as on the bus: the step that follows answers the error. +func (ix *Indexer) pendingOnBus(ctx context.Context) bool { + _, _, _, err := ix.Pending.Get(ctx, delivery.PendingKey(0)) + return !errors.Is(err, process.ErrNotYet) +} + +// ownBuildNow is the build this program runs: the one learnt, else asked of the node-engine now. +func (ix *Indexer) ownBuildNow(ctx context.Context) string { + ix.ownMu.Lock() + fp := ix.ownFP + ix.ownMu.Unlock() + if fp != "" || ix.OwnBuild == nil { + return fp + } + ix.learnOwnBuild(ctx) + ix.ownMu.Lock() + defer ix.ownMu.Unlock() + return ix.ownFP +} + +// sayTakeIn publishes a take-in event and keeps what it said on the entries, so indexer_pending reads the last event of +// each wait from the state. A publish that fails is said in the log: the next evaluation says it again. +func (ix *Indexer) sayTakeIn(ctx context.Context, t delivery.TakeIn) bool { + if ix.SayTakeIn == nil { + return true + } + if err := ix.SayTakeIn(ctx, t); err != nil { + ix.logf("take-in: %s.%s is not said: %v", t.Kind, t.Key, err) + return false + } + return true +} + +// markSaid writes on each entry the wait it was said under, by compare-and-set; a lost one is left (its next +// evaluation says it again). +func (ix *Indexer) markSaid(ctx context.Context, entries []PendingEntry, t *delivery.TakeIn, change func(*Pending)) { + for _, e := range entries { + p := e.Value + if t != nil { + p.Said, p.SaidKind, p.SaidAt = t.Key, t.Kind, ix.now() + } + if change != nil { + change(&p) + } + data, err := encodePending(p) + if err != nil { + continue + } + if err := ix.Pending.Update(ctx, e.Key, data, e.Revision); err != nil { + ix.logf("take-in: %s.%s is not written again: %v", PendingState, e.Key, err) + } + } +} + +// Arrive takes one merged message of the forge bridge's stream in under ADR 0633: seq is its stream sequence. It is +// written to pending first (a redelivered message finds its key there and writes no second), and then what pending +// holds is evaluated (Evaluate): taken in whole when its revision names the build this program runs, or held for the +// indexer it names. It answers what was taken in, and a line saying what waits. A message is never written while the +// build this program runs is not known (ErrOwnBuildUnknown). +func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f delivery.Forge) (Applied, string, error) { + if ix.Pending == nil || ix.Modules == nil || !ix.pendingOnBus(ctx) { + // The state this delivery brings is not on the bus yet (issue 1790): the merge is taken in at once, as before, + // whether or not the build this program runs is known. + a, err := ix.ApplyMessage(ctx, f) + return a, "", err + } + own := ix.ownBuildNow(ctx) + if own == "" { + return Applied{}, "", ErrOwnBuildUnknown + } + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + raw, err := delivery.EncodeForge(f) + if err != nil { + return Applied{}, "", fmt.Errorf("%w: %v", ErrUnreadable, err) + } + key := delivery.PendingKey(seq) + data, err := encodePending(Pending{Forge: raw, Group: f.Group, Since: ix.now(), LastWriter: own}) + if err != nil { + return Applied{}, "", err + } + switch err := ix.Pending.Create(ctx, key, data); { + case errors.Is(err, process.ErrNotYet): + // The state this delivery brings is not on the bus yet (issue 1790): the merge is taken in at once, as before. + ix.logf("take-in: the state %s is not on the bus yet, so %s#%d is taken in at once, as before ADR 0633", PendingBucket, f.Repository, f.Number) + a, err := ix.ApplyMessage(ctx, f) + return a, "", err + case err != nil && !errors.Is(err, ErrConflict): + return Applied{}, "", fmt.Errorf("%s.%s: %w", PendingState, key, err) + } + if f.Group != "" { + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + return Applied{}, "", err + } + if held := WaitsOf(entries).Groups[f.Group]; len(held) > 0 { + t := groupWait(delivery.TakeInPending, f.Group, held) + if ix.sayTakeIn(ctx, t) { + ix.markSaid(ctx, held, &t, nil) + } + } + } + return ix.evaluate(ctx, own, false) +} + +// EndGroup takes a group's end in (ADR 0633 §4): once every member it names is held, the group's members are marked +// ended, the group's wait is said taken (take-in.taken under its group key), and pending is evaluated, the group now +// one revision. A member named and neither held nor taken in yet answers an error, so the end is delivered again once +// the member arrived. An end of a group taken in already (every member's merge at or behind its repository's taken +// merge) changes nothing and is said: it is never delivered again for ever. +func (ix *Indexer) EndGroup(ctx context.Context, end delivery.ForgeGroupEnd) (Applied, string, error) { + if ix.Pending == nil || !ix.pendingOnBus(ctx) { + return Applied{}, "no pending state: the group's merges were taken in one by one", nil + } + own := ix.ownBuildNow(ctx) + if own == "" { + return Applied{}, "", ErrOwnBuildUnknown + } + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if errors.Is(err, process.ErrNotYet) { + return Applied{}, "the pending state is not on the bus yet: the group's merges were taken in one by one", nil + } + if err != nil { + return Applied{}, "", err + } + var held []PendingEntry + for _, e := range entries { + if e.Value.Group == end.Group { + held = append(held, e) + } + } + isHeld := func(m delivery.GroupMerge) bool { + return slices.ContainsFunc(held, func(e PendingEntry) bool { + return delivery.SameRepository(e.Forge.Repository, m.Repository) && e.Forge.Number == m.Number + }) + } + var missing []string + for _, m := range end.Members { + if isHeld(m) { + continue + } + taken, err := ix.takenAlready(ctx, m) + if err != nil { + return Applied{}, "", err + } + if !taken { + missing = append(missing, fmt.Sprintf("%s#%d", m.Repository, m.Number)) + } + } + if len(missing) > 0 { + return Applied{}, "", fmt.Errorf("the end of the group %q names %s, not held yet: it waits for their merged messages", end.Group, strings.Join(missing, ", ")) + } + if len(held) == 0 { + ix.logf("take-in: the end of the group %q comes after its take-in: every member is taken in already, nothing changes", end.Group) + return Applied{}, "the group is taken in already", nil + } + if !slices.ContainsFunc(held, func(e PendingEntry) bool { return !e.Value.Ended }) { + return ix.evaluate(ctx, own, false) + } + // The group's wait ends first (take-in.taken under its group key); a revision it then is, pending for another + // indexer, is said pending under its oldest merge's key by the evaluation that follows. + t := groupWait(delivery.TakeInTaken, end.Group, held) + said := &t + if !ix.sayTakeIn(ctx, t) { + said = nil // said again by the take-in or the hold that follows (saidWaits) + } + ix.markSaid(ctx, held, said, func(p *Pending) { p.Ended = true }) + ix.logf("take-in: the group %q ended with %d members held: one revision now", end.Group, len(held)) + return ix.evaluate(ctx, own, false) +} + +// takenAlready says whether a group member's merge is at or behind its repository's taken merge. +func (ix *Indexer) takenAlready(ctx context.Context, m delivery.GroupMerge) (bool, error) { + if ix.Taken == nil { + return false, nil + } + key, err := delivery.RepositoryKey(m.Repository) + if err != nil { + return false, err + } + value, _, found, err := ix.Taken.Get(ctx, key) + if errors.Is(err, process.ErrNotYet) || (err == nil && !found) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("taken.%s: %w", key, err) + } + cur, err := delivery.DecodeTaken(value) + if err != nil || cur.Merge == "" { + return false, nil + } + if cur.Merge == m.Merge { + return true, nil + } + return ix.older(ctx, m.Repository, m.Merge, cur.Merge) +} + +// TakeInAtStart is the take-in at the start (ADR 0633 §2), before the consumer on the forge bridge's stream starts: +// it asks the build this program runs until it is known (wait says when to ask again, false ends it), then evaluates +// pending, taking in the whole revisions that name this build; a group still being gathered stays, and a revision +// naming yet another build is held for that one. A claim left by an indexer that ran before is void here. +func (ix *Indexer) TakeInAtStart(ctx context.Context, wait func(context.Context) bool) (Applied, string, error) { + if ix.Pending == nil || !ix.pendingOnBus(ctx) { + // Nothing can wait in a state not on the bus: the start asks no build and does not wait for one. + return Applied{}, "the pending state is not on the bus yet: nothing waits", nil + } + own := ix.ownBuildNow(ctx) + for own == "" { + ix.logf("take-in: the build this program runs is not known yet, so nothing pending is taken in; asked again") + if !wait(ctx) { + return Applied{}, "", ErrOwnBuildUnknown + } + own = ix.ownBuildNow(ctx) + } + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + a, words, err := ix.evaluate(ctx, own, true) + if errors.Is(err, process.ErrNotYet) { + return a, "the pending state is not on the bus yet: nothing waits", nil + } + return a, words, err +} + +// evaluate works out what pending's ready merges name, together (ADR 0633 §3), and takes them in or holds them. +func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applied, string, error) { + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + return Applied{}, "", err + } + w := WaitsOf(entries) + if len(w.Ready) == 0 { + return Applied{}, ix.waitingLine(w), nil + } + if !start { + for _, e := range w.Ready { + if e.Value.Taking != "" { + ix.logf("take-in: %s.%s is claimed by the take-in %s, so this indexer takes nothing in", PendingState, e.Key, e.Value.Taking) + return Applied{}, "claimed by another take-in", nil + } + } + } + names, dry, err := ix.computeOver(ctx, w.Ready) + if err != nil { + return Applied{}, "", err + } + if names == own { + a, err := ix.takeInWhole(ctx, w.Ready, own, "") + if errors.Is(err, errClaimLost) { + return a, "claimed by another take-in", nil + } + return a, ix.waitingLine(w), err + } + return Applied{}, ix.waitingLine(w), ix.holdFor(ctx, w.Ready, names, own, dry) +} + +// waitingLine says what waits after an evaluation, for the heard state. +func (ix *Indexer) waitingLine(w Waits) string { + var parts []string + for _, g := range slices.Sorted(maps.Keys(w.Groups)) { + parts = append(parts, fmt.Sprintf("the group %q gathered (%d held)", g, len(w.Groups[g]))) + } + for _, g := range slices.Sorted(maps.Keys(w.Behind)) { + parts = append(parts, fmt.Sprintf("%d merges behind %s", len(w.Behind[g]), g)) + } + return strings.Join(parts, "; ") +} + +// dryRun is an indexer computing a take-in without writing it, and the overlays of what it would have written. +type dryRun struct { + ix *Indexer + modules, graph, verdicts *overlayStore + candidates, taken, predefined, configs *overlayStore + oldSections *overlayStore +} + +// dry is an indexer over overlays of this one's states (overlay.go): what a take-in would write stays in memory and +// nothing reaches the bus but the manifests and the packages the revision publishes, which clause 1 writes all the +// same. It says no event, kicks nothing in the background, and knows what this one knows of the source, the toolchains +// and the previews. Every field of Indexer is named here (TestDryNamesEveryField holds it so). +func (ix *Indexer) dry() *dryRun { + over := func(st Store) (Store, *overlayStore) { + if st == nil { + return nil, nil + } + o := newOverlay(st) + return o, o + } + d := &dryRun{} + x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum, + GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains, + ReadOpen: nil, OwnBuild: nil, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil} + x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) } + x.Store, d.candidates = over(ix.Store) + x.Modules, d.modules = over(ix.Modules) + x.Graph, d.graph = over(ix.Graph) + x.Verdicts, d.verdicts = over(ix.Verdicts) + x.Taken, d.taken = over(ix.Taken) + x.Predefined, d.predefined = over(ix.Predefined) + x.ConfigCandidates, d.configs = over(ix.ConfigCandidates) + x.OldSections, d.oldSections = over(ix.OldSections) + ix.mu.Lock() + x.source = sourceNow{rev: ix.source.rev, heads: maps.Clone(ix.source.heads)} + x.previews = maps.Clone(ix.previews) + x.toolchains = map[string]map[string]delivery.Toolchain{} + for l, by := range ix.toolchains { + x.toolchains[l] = maps.Clone(by) + } + x.moving = maps.Clone(ix.moving) + x.untaken = maps.Clone(ix.untaken) + ix.mu.Unlock() + ix.ownMu.Lock() + x.ownFP = ix.ownFP + ix.ownMu.Unlock() + d.ix = x + return d +} + +// computeOver applies the ready merges, in merge order, to a dry indexer, as one overlay: each later merge's trees +// govern (ADR 0633 §3). It answers the fingerprint the indexer's own module has once they are applied (none is "", +// with its reason), and the dry run, whose overlays hold what the take-in would write. A merge that can never be read +// is left out, as its take-in will drop it. +func (ix *Indexer) computeOver(ctx context.Context, ready []PendingEntry) (string, *dryRun, error) { + d := ix.dry() + for _, e := range ready { + if _, err := d.ix.ApplyMessage(ctx, e.Forge); err != nil { + if errors.Is(err, ErrUnreadable) { + continue + } + return "", d, fmt.Errorf("computing the take-in of %s (%s#%d): %w", e.Key, e.Forge.Repository, e.Forge.Number, err) + } + } + key, _ := delivery.ModulesKey(delivery.IndexerModule) + value, _, found, err := d.ix.Modules.Get(ctx, key) + if err != nil { + return "", d, err + } + if !found { + return NoBuild + "the indexer holds no record of itself", d, nil + } + m, err := delivery.DecodeModule(value) + if err != nil { + return "", d, nil + } + if m.FP2 == "" { + return NoBuild + m.None2, d, nil + } + return m.FP2, d, nil +} + +// claim marks the entries as taken in by this take-in, by compare-and-set on the revisions read; a lost one leaves +// every claim made and answers false, and nothing is taken in. +func (ix *Indexer) claim(ctx context.Context, entries []PendingEntry, own, by string) ([]PendingEntry, bool) { + id := fmt.Sprintf("%s at %s", own, ix.now().Format(time.RFC3339Nano)) + if by != "" { + id = by + ": " + id + } + claimed := make([]PendingEntry, 0, len(entries)) + for _, e := range entries { + p := e.Value + p.Taking, p.LastWriter = id, own + data, err := encodePending(p) + if err == nil { + err = ix.Pending.Update(ctx, e.Key, data, e.Revision) + } + if err != nil { + ix.logf("take-in: the claim of %s.%s is lost (%v): another take-in took it, so this one takes nothing in", PendingState, e.Key, err) + return nil, false + } + _, rev, _, err := ix.Pending.Get(ctx, e.Key) + if err != nil { + return nil, false + } + e.Revision, e.Value = rev, p + claimed = append(claimed, e) + } + return claimed, true +} + +// takeInWhole takes the entries in with this indexer, as merges are taken in (ApplyMessage), oldest first, then +// removes them from pending and says each wait they were said under taken. by names a hand act ("" for none). +func (ix *Indexer) takeInWhole(ctx context.Context, entries []PendingEntry, own, by string) (Applied, error) { + return ix.takeInWholeBy(ctx, entries, own, by, "", "") +} + +// takeInWholeBy is takeInWhole; who and why, of a hand act, are carried on the take-in.taken events. +func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, own, by, who, why string) (Applied, error) { + claimed, ok := ix.claim(ctx, entries, own, by) + if !ok { + return Applied{}, errClaimLost + } + var applied Applied + for _, e := range claimed { + a, err := ix.ApplyMessage(ctx, e.Forge) + applied.Written = append(applied.Written, a.Written...) + if a.Late != "" { + applied.Late = a.Late + } + if errors.Is(err, ErrUnreadable) { + ix.logf("take-in: %s (%s#%d) can never be read, dropped: %v", e.Key, e.Forge.Repository, e.Forge.Number, err) + continue + } + if err != nil { + return applied, fmt.Errorf("taking %s (%s#%d) in: %w", e.Key, e.Forge.Repository, e.Forge.Number, err) + } + } + for _, e := range claimed { + if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil { + ix.logf("take-in: %s.%s is taken in and not removed: %v", PendingState, e.Key, err) + } + } + ix.logf("take-in: %d merges taken in as one revision by %s%s, oldest %s", len(claimed), own, byWords(by), claimed[0].Key) + for _, t := range saidWaits(claimed) { + t.By, t.Why = who, why + ix.sayTakeIn(ctx, t) + } + return applied, nil +} + +func byWords(by string) string { + if by == "" { + return "" + } + return " (a hand act: " + by + ")" +} + +// saidWaits are the taken events of every wait the entries were said under that is not ended yet. +func saidWaits(entries []PendingEntry) []delivery.TakeIn { + by := map[string][]PendingEntry{} + for _, e := range entries { + if e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending { + by[e.Value.Said] = append(by[e.Value.Said], e) + } + } + var out []delivery.TakeIn + for _, key := range slices.Sorted(maps.Keys(by)) { + held := by[key] + if strings.HasPrefix(key, "g.") { + out = append(out, groupWait(delivery.TakeInTaken, held[0].Value.Group, held)) + continue + } + t := revisionWait(delivery.TakeInTaken, held, held[0].Value.WaitsFor) + t.Key = key + out = append(out, t) + } + return out +} + +// holdFor keeps the ready merges pending for the indexer build they name and writes only what building and moving +// that indexer needs (ADR 0633 §1): the indexer's own module entry, its key of the module graph and the verdict on its +// manifest, from the dry run; the manifests and the packages are written by the dry run itself. The revision is said +// pending under its oldest merge's key, and a wait it was said under before by another key is said taken. +func (ix *Indexer) holdFor(ctx context.Context, ready []PendingEntry, names, own string, d *dryRun) error { + key, _ := delivery.ModulesKey(delivery.IndexerModule) + if err := copyKey(ctx, d.modules, ix.Modules, key); err != nil { + return fmt.Errorf("the indexer's own module entry: %w", err) + } + if gk, err := graph.Key(delivery.IndexerModule); err == nil { + if err := copyKey(ctx, d.graph, ix.Graph, gk); err != nil && !errors.Is(err, process.ErrNotYet) { + return fmt.Errorf("the indexer's key of the module graph: %w", err) + } + } + if value, _, found, err := d.ix.Modules.Get(ctx, key); err == nil && found { + if m, err := delivery.DecodeModule(value); err == nil && m.Manifest != "" { + if vk, err := delivery.VerdictsKey(m.Manifest); err == nil { + if err := copyKey(ctx, d.verdicts, ix.Verdicts, vk); err != nil && !errors.Is(err, process.ErrNotYet) { + return fmt.Errorf("the verdict on the indexer's manifest: %w", err) + } + } + } + } + t := revisionWait(delivery.TakeInPending, ready, names) + var ended, unmarked []PendingEntry + said, changed := false, false + for _, e := range ready { + switch { + case e.Value.Said == t.Key && e.Value.SaidKind == delivery.TakeInPending: + said = true + changed = changed || e.Value.WaitsFor != names + if e.Value.WaitsFor != names { + unmarked = append(unmarked, e) + } + case e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending: + ended = append(ended, e) + unmarked = append(unmarked, e) + default: + unmarked = append(unmarked, e) + } + } + for _, old := range saidWaits(ended) { + ix.sayTakeIn(ctx, old) + } + if !said || changed { + ix.logf("take-in: %d merges, oldest %s, name the indexer build %s, not %s, which runs: they wait for it", len(ready), t.Key, names, own) + if !ix.sayTakeIn(ctx, t) { + ix.markSaid(ctx, unmarked, nil, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own }) + return nil + } + unmarked = ready + } + ix.markSaid(ctx, unmarked, &t, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own }) + return nil +} + +// copyKey writes what the overlay holds of key to the store, when it wrote or deleted it. +func copyKey(ctx context.Context, from *overlayStore, to Store, key string) error { + if from == nil || to == nil { + return nil + } + value, deleted, ok := from.written(key) + if !ok { + return nil + } + return write(ctx, to, key, func(cur []byte, found bool) ([]byte, bool, error) { + switch { + case deleted: + return nil, found, nil + case found && string(cur) == string(value): + return nil, false, nil + } + return value, false, nil + }) +} + +// PendingWaits is what indexer_pending answers of the pending state's entries (ADR 0633 §5): the ready merges as one +// revision, each group being gathered, and the merges behind a group, each with the modules it holds back and the +// last take-in event said of it. +func PendingWaits(entries []PendingEntry) []delivery.PendingWait { + w := WaitsOf(entries) + var out []delivery.PendingWait + if len(w.Ready) > 0 { + out = append(out, waitOf(w.Ready, delivery.TakeInRevision, "")) + } + for _, g := range slices.Sorted(maps.Keys(w.Groups)) { + out = append(out, waitOf(w.Groups[g], delivery.TakeInGroup, "")) + } + for _, g := range slices.Sorted(maps.Keys(w.Behind)) { + out = append(out, waitOf(w.Behind[g], delivery.TakeInRevision, g)) + } + slices.SortStableFunc(out, func(a, b delivery.PendingWait) int { return a.Since.Compare(b.Since) }) + return out +} + +func waitOf(entries []PendingEntry, what, behind string) delivery.PendingWait { + pw := delivery.PendingWait{Key: entries[0].Key, What: what, Merges: mergesOf(entries), Behind: behind, Since: sinceOf(entries), + WaitsFor: entries[len(entries)-1].Value.WaitsFor, Holds: []string{}} + if what == delivery.TakeInGroup { + pw.Group = entries[0].Value.Group + pw.Key = delivery.GroupTakeInKey(pw.Group) + pw.Members = entries[len(entries)-1].Forge.Members + pw.WaitsFor = "" + } + for _, e := range entries { + pw.Keys = append(pw.Keys, e.Key) + for _, raw := range e.Forge.Manifests { + if name := manifestModule(raw); name != "" && !slices.Contains(pw.Holds, name) { + pw.Holds = append(pw.Holds, name) + } + } + if e.Value.Said != "" && e.Value.SaidAt.After(pw.SaidAt) { + pw.Said, pw.SaidAt = delivery.TakeInStream+"."+e.Value.SaidKind+"."+e.Value.Said, e.Value.SaidAt + } + } + slices.Sort(pw.Holds) + return pw +} + +// TakeInPendingByHand is the hand take-in (indexer_take_in_pending, ADR 0633 §5): the wait named by its key, the +// ready revision (its m. key) or a group gathered so far (its g. key), is taken in with this indexer and its judge, +// whatever build it names, for when the named indexer can never run or the group's end never comes. by and why are +// said in the journal and on the take-in.taken event. What waited behind it is evaluated again after. +func (ix *Indexer) TakeInPendingByHand(ctx context.Context, key, by, why string) TakenIn { + r := TakenIn{Module: key} + if ix.Pending == nil { + r.Answer = "the pending state is not on the bus: nothing waits" + return r + } + own := ix.ownBuildNow(ctx) + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + r.Answer = err.Error() + return r + } + w := WaitsOf(entries) + var take []PendingEntry + switch { + case strings.HasPrefix(key, "g."): + for g, held := range w.Groups { + if delivery.GroupTakeInKey(g) == key { + take = held + } + } + case len(w.Ready) > 0 && w.Ready[0].Key == key: + take = w.Ready + } + if len(take) == 0 { + r.Answer = fmt.Sprintf("nothing waits under %s: indexer_pending names the waits", key) + return r + } + ix.logf("take-in: a hand take-in of %s (%d merges) asked by %s: %s", key, len(take), by, why) + hand := fmt.Sprintf("%s (%s)", by, why) + if _, err := ix.takeInWholeBy(ctx, take, own, hand, by, why); err != nil { + r.Answer = err.Error() + return r + } + r.Stored = true + r.Answer = fmt.Sprintf("%d merges taken in by the running indexer %s", len(take), own) + if _, _, err := ix.evaluate(ctx, own, false); err != nil { + r.Answer += "; what waited behind it is not evaluated: " + err.Error() + } + return r +} + +// DropPending is the hand drop (indexer_drop_pending, ADR 0633 §5): the one pending merge under key, abandoned +// because a later merge replaced it, is removed and nothing of it is taken in. Its wait is said taken, naming the drop; +// the merges left under that wait are said pending again by the evaluation that follows. +func (ix *Indexer) DropPending(ctx context.Context, key, by, why string) TakenIn { + r := TakenIn{Module: key} + if ix.Pending == nil { + r.Answer = "the pending state is not on the bus: nothing waits" + return r + } + own := ix.ownBuildNow(ctx) + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + r.Answer = err.Error() + return r + } + at := slices.IndexFunc(entries, func(e PendingEntry) bool { return e.Key == key }) + if at < 0 { + r.Answer = fmt.Sprintf("%s holds no merge under %s", PendingState, key) + return r + } + e := entries[at] + if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil { + r.Answer = fmt.Sprintf("%s.%s is not removed: %v", PendingState, key, err) + return r + } + ix.logf("take-in: %s (%s#%d at %s) dropped by hand, asked by %s: %s; nothing of it is taken in", key, e.Forge.Repository, e.Forge.Number, e.Forge.Merge, by, why) + if e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending { + var same []PendingEntry + for _, o := range entries { + if o.Key != key && o.Value.Said == e.Value.Said { + same = append(same, o) + } + } + for _, t := range saidWaits([]PendingEntry{e}) { + t.By, t.Why, t.Dropped = by, why, true + ix.sayTakeIn(ctx, t) + } + ix.markSaid(ctx, same, nil, func(p *Pending) { p.Said, p.SaidKind = "", "" }) + } + r.Stored, r.Answer = true, fmt.Sprintf("%s dropped: %s#%d at %s is not taken in", key, e.Forge.Repository, e.Forge.Number, e.Forge.Merge) + if own != "" { + if _, _, err := ix.evaluate(ctx, own, false); err != nil { + r.Answer += "; what waits is not evaluated again: " + err.Error() + } + } + return r +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go new file mode 100644 index 00000000..cca2d25e --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -0,0 +1,331 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "reflect" + "slices" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// takeInBed is an indexer holding the catalogue's avahi and the indexer's own module, both with a fingerprint, its +// pending state and the take-in events it says; base is the catalogue merge it was built from. +type takeInBed struct { + ix *Indexer + mods *memStore + pending *memStore + said []delivery.TakeIn + base delivery.Forge + logs *strings.Builder +} + +func newTakeInBed(t *testing.T) *takeInBed { + t.Helper() + ix, mods, f := newFormIndexerUnmerged(t, mod{"avahi", catalogueManifest(t, "avahi")}, + mod{delivery.IndexerModule, renamed(t, "avahi", delivery.IndexerModule)}) + if err := ix.Apply(context.Background(), f); err != nil { + t.Fatal(err) + } + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + reportToolchain(t, ix, "mesh-build-go", "ace", "go", "go1.27.1") + b := &takeInBed{ix: ix, mods: mods, pending: newMem(), base: f, logs: &strings.Builder{}} + ix.Verdicts = newMem() + ix.Pending = b.pending + ix.SayTakeIn = func(_ context.Context, t delivery.TakeIn) error { b.said = append(b.said, t); return nil } + logf := ix.Logf + ix.Logf = func(format string, args ...any) { + b.logs.WriteString(format + "\n") + if logf != nil { + logf(format, args...) + } + } + own := b.fp(t, delivery.IndexerModule) + if own == "" || b.fp(t, "avahi") == "" { + t.Fatalf("the bed's modules have no fingerprint") + } + ix.OwnBuild = func(context.Context) (string, error) { return own, nil } + return b +} + +func (b *takeInBed) fp(t *testing.T, module string) string { + t.Helper() + m, _ := moduleOf(t, b.mods, module) + return m.FP2 +} + +func (b *takeInBed) commit(t *testing.T, module string) string { + t.Helper() + m, _ := moduleOf(t, b.mods, module) + return m.Commit +} + +// merge is a catalogue merge at commit c whose trees of avahi and the indexer are given ("" keeps the base's). +func (b *takeInBed) merge(number int, c, avahi, indexer string) delivery.Forge { + f := b.base + f.Number, f.Merge, f.Head = number, c, c + f.Trees = map[string]string{} + for dir, tree := range b.base.Trees { + f.Trees[dir] = tree + } + if avahi != "" { + f.Trees["modules/avahi"] = avahi + } + if indexer != "" { + f.Trees["modules/"+delivery.IndexerModule] = indexer + } + return f +} + +func (b *takeInBed) arrive(t *testing.T, seq uint64, f delivery.Forge) { + t.Helper() + if _, _, err := b.ix.Arrive(context.Background(), seq, f); err != nil { + t.Fatalf("arrive %d: %v", seq, err) + } +} + +func (b *takeInBed) keys() []string { + k, _ := b.pending.Keys(context.Background()) + slices.Sort(k) + return k +} + +func (b *takeInBed) events() []string { + var out []string + for _, s := range b.said { + out = append(out, s.Kind+" "+s.Key) + } + return out +} + +func sha(c byte) string { return strings.Repeat(string(c), 40) } + +func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) + ctx := context.Background() + t.Run("T6 a merge while the own build is unknown writes nothing", func(t *testing.T) { + b := newTakeInBed(t) + b.ix.OwnBuild = func(context.Context) (string, error) { return "", errors.New("the engine does not answer") } + _, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", "")) + if !errors.Is(err, ErrOwnBuildUnknown) || len(b.keys()) != 0 || b.commit(t, "avahi") == sha('b') { + t.Errorf("err %v, pending %v, avahi at %s: want ErrOwnBuildUnknown and nothing written", err, b.keys(), b.commit(t, "avahi")) + } + // Before the pending state is on the bus nothing waits: the start asks no build, and the merge is taken in at once, + // as before ADR 0633, though the build is unknown (TestIndexerWhole found the start waiting on it for ever). + b.ix.Pending = &memStore{unbound: process.ErrNotYet} + if _, _, err := b.ix.TakeInAtStart(ctx, func(context.Context) bool { t.Fatal("the start waits for the build"); return false }); err != nil { + t.Fatal(err) + } + if _, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", "")); err != nil || b.commit(t, "avahi") != sha('b') { + t.Errorf("no pending state: err %v, avahi at %s: want the merge taken in at once", err, b.commit(t, "avahi")) + } + }) + t.Run("T7 and T3 a merge alone, an indexer change, a merge behind it and a revert", func(t *testing.T) { + b := newTakeInBed(t) + own, avahi0 := b.fp(t, delivery.IndexerModule), b.fp(t, "avahi") + // A merge with no change to the indexer and nothing pending is taken in alone, as before. + b.arrive(t, 1, b.merge(20, sha('b'), "tree-avahi-2", "")) + if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || len(b.said) != 0 { + t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in", avahi1, avahi0, b.keys(), b.events()) + } + avahi1, verdicts := b.fp(t, "avahi"), len(b.ix.Verdicts.(*memStore).vals) + // T4: a change to the indexer writes only its entry (with its graph key and verdict); the rest is pending. + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + named := b.fp(t, delivery.IndexerModule) + if named == own || b.fp(t, "avahi") != avahi1 || b.commit(t, "avahi") != sha('b') { + t.Fatalf("indexer change: indexer %s (own %s), avahi %s at %s: want only the indexer written", named, own, b.fp(t, "avahi"), b.commit(t, "avahi")) + } + if g := graphOf(t, b.ix, delivery.IndexerModule); g.FP != named { + t.Errorf("graph of the indexer names %s, want %s", g.FP, named) + } + if n := len(b.ix.Verdicts.(*memStore).vals); n > verdicts+1 { + t.Errorf("verdicts %d, want at most %d: only the new indexer manifest's", n, verdicts+1) + } + if want := []string{delivery.PendingKey(2)}; !reflect.DeepEqual(b.keys(), want) || !slices.Equal(b.events(), []string{"pending " + delivery.PendingKey(2)}) || b.said[0].WaitsFor != named { + t.Fatalf("pending %v, said %+v: want m.2 pending for %s", b.keys(), b.said, named) + } + // T5: the merge delivered again writes no second key and says nothing again. + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + if len(b.keys()) != 1 || len(b.said) != 1 { + t.Errorf("redelivered: pending %v, said %v", b.keys(), b.events()) + } + // A later merge joins: the newest governs and still names the new indexer. + b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")) + if len(b.keys()) != 2 || b.commit(t, "avahi") != sha('b') || len(b.said) != 1 { + t.Fatalf("behind: pending %v, avahi at %s, said %v: want both pending and nothing said again", b.keys(), b.commit(t, "avahi"), b.events()) + } + // A revert brings the indexer back to the running build: everything is taken in, oldest first. + b.arrive(t, 4, b.merge(23, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+delivery.IndexerModule])) + if len(b.keys()) != 0 || b.fp(t, delivery.IndexerModule) != own || b.commit(t, "avahi") != sha('e') { + t.Fatalf("revert: pending %v, indexer %s, avahi at %s: want all taken in", b.keys(), b.fp(t, delivery.IndexerModule), b.commit(t, "avahi")) + } + if want := []string{"pending " + delivery.PendingKey(2), "taken " + delivery.PendingKey(2)}; !slices.Equal(b.events(), want) { + t.Errorf("said %v, want %v", b.events(), want) + } + }) + t.Run("T10 the named indexer takes the revision in at its start; a gathered group stays", func(t *testing.T) { + b := newTakeInBed(t) + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + g := b.merge(30, sha('f'), "", "") + g.Repository, g.Trees, g.Manifests, g.Locks, g.Group = "novox/mesh-sdk", map[string]string{}, nil, nil, "feat/x" + g.Members = []delivery.Head{{Repository: "novox/mesh-sdk", Number: 30, Head: g.Head}, {Repository: "novox/mesh-host", Number: 4, Head: sha('9')}} + b.arrive(t, 5, g) + named := b.fp(t, delivery.IndexerModule) + // The new indexer starts; its build is not known at first: a hand ask is answered meanwhile (R5), and the start waits. + next := &Indexer{} + *next = Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts, + Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum} + next.toolchains = b.ix.toolchains + asked := 0 + next.OwnBuild = func(context.Context) (string, error) { + if asked++; asked < 2 { + return "", errors.New("not yet") + } + return named, nil + } + if r := next.TakeInPendingByHand(ctx, "m.nothing", "lead@shanks", "a test"); r.Stored || !strings.Contains(r.Answer, "nothing waits") { + t.Errorf("a hand ask while the own build is unknown: %+v", r) + } + if _, _, err := next.TakeInAtStart(ctx, func(context.Context) bool { return true }); err != nil { + t.Fatal(err) + } + if asked < 2 || !reflect.DeepEqual(b.keys(), []string{delivery.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') { + t.Errorf("start: asked %d, pending %v, avahi at %s: want the revision taken in, the group left", asked, b.keys(), b.commit(t, "avahi")) + } + }) + t.Run("T8 T9 T11 a group gathered until its end, a merge behind it, one alone, an end again", func(t *testing.T) { + b := newTakeInBed(t) + b.ix.Taken = newMem() + // The commits in the order they were made: one is in another when it comes first. + order := "a756bc" + b.ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { + return strings.IndexByte(order, commit[0]) <= strings.IndexByte(order, in[0]), nil + } + member := func(repo string, number int, c string) delivery.Forge { + f := b.merge(number, c, "", "") + f.Group = "feat/take-in" + f.Members = []delivery.Head{{Repository: catalogueRepo, Number: 40, Head: sha('1')}, {Repository: "novox/mesh-sdk", Number: 41, Head: sha('2')}} + if repo != catalogueRepo { + f.Repository, f.Trees, f.Manifests, f.Locks = repo, map[string]string{}, nil, nil + } + f.Head = map[int]string{40: sha('1'), 41: sha('2')}[number] + return f + } + b.arrive(t, 1, member(catalogueRepo, 40, sha('7'))) + if b.commit(t, "avahi") == sha('7') || !slices.Equal(b.events(), []string{"pending g.feat/take-in"}) { + t.Fatalf("first member: avahi at %s, said %v: want held, the group said pending", b.commit(t, "avahi"), b.events()) + } + // A merge in another repository is taken in alone; one in a member's repository waits behind the group. + alone := b.merge(50, sha('5'), "", "") + alone.Repository, alone.Trees, alone.Manifests, alone.Locks = "novox/hq", map[string]string{}, nil, nil + b.arrive(t, 2, alone) + b.arrive(t, 3, b.merge(51, sha('6'), "tree-avahi-9", "")) + if !reflect.DeepEqual(b.keys(), []string{delivery.PendingKey(1), delivery.PendingKey(3)}) { + t.Fatalf("pending %v: want the member and the merge behind it", b.keys()) + } + end := delivery.ForgeGroupEnd{Group: "feat/take-in", Members: []delivery.GroupMerge{{Repository: catalogueRepo, Number: 40, Merge: sha('7')}, {Repository: "novox/mesh-sdk", Number: 41, Merge: sha('b')}}} + if _, _, err := b.ix.EndGroup(ctx, end); err == nil { + t.Fatal("an end before its last member is taken in") + } + b.arrive(t, 4, member("novox/mesh-sdk", 41, sha('b'))) + if _, _, err := b.ix.EndGroup(ctx, end); err != nil { + t.Fatal(err) + } + if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') { + t.Fatalf("end: pending %v, avahi at %s: want the group and the merge behind it taken in", b.keys(), b.commit(t, "avahi")) + } + if want := []string{"pending g.feat/take-in", "pending g.feat/take-in", "taken g.feat/take-in"}; !slices.Equal(b.events(), want) { + t.Errorf("said %v, want %v", b.events(), want) + } + // R1: the end published again after the take-in changes nothing and is not delivered again. + if _, words, err := b.ix.EndGroup(ctx, end); err != nil || !strings.Contains(words, "taken in already") { + t.Errorf("an end again: %q %v", words, err) + } + // T11 (the lead's row): a group whose revision changes the indexer: the group's wait is taken, then the revision pending. + b.said = nil + g := member(catalogueRepo, 40, sha('c')) + g.Group, g.Trees = "feat/two", b.merge(0, "", "", "tree-indexer-7").Trees + g.Members = g.Members[:1] + b.arrive(t, 9, g) + if _, _, err := b.ix.EndGroup(ctx, delivery.ForgeGroupEnd{Group: "feat/two", Members: []delivery.GroupMerge{{Repository: catalogueRepo, Number: 40, Merge: sha('c')}}}); err != nil { + t.Fatal(err) + } + if want := []string{"pending g.feat/two", "taken g.feat/two", "pending " + delivery.PendingKey(9)}; !slices.Equal(b.events(), want) { + t.Errorf("a group naming another indexer: said %v, want %v", b.events(), want) + } + }) + t.Run("T5 two take-ins of one pending revision: the one that loses the claim writes nothing", func(t *testing.T) { + b := newTakeInBed(t) + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + named := b.fp(t, delivery.IndexerModule) + b.ix.ownMu.Lock() + b.ix.ownFP = named + b.ix.ownMu.Unlock() + // The other indexer claims the merge between this one's read and its claim. + b.pending.beforeWrite = func() { + v := b.pending.vals[delivery.PendingKey(2)] + var p Pending + _ = json.Unmarshal(v, &p) + p.Taking = "the other indexer" + data, _ := json.Marshal(p) + b.pending.put(delivery.PendingKey(2), data) + } + if _, _, err := b.ix.TakeInAtStart(ctx, func(context.Context) bool { return false }); err != nil { + t.Fatal(err) + } + if b.commit(t, "avahi") == sha('c') || len(b.keys()) != 1 || !strings.Contains(b.logs.String(), "is lost") { + t.Errorf("avahi at %s, pending %v: want nothing taken in by the loser, and it said", b.commit(t, "avahi"), b.keys()) + } + }) + t.Run("T12 T13 indexer_pending names the wait; a hand take-in and a drop say what they ended", func(t *testing.T) { + b := newTakeInBed(t) + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")) + var answer delivery.PendingAnswer + for _, tool := range PendingTools(b.pending, nil, nil) { + if tool.Name == PendingTool { + out, err := tool.Run(map[string]any{}) + if err != nil { + t.Fatal(err) + } + answer = out.(delivery.PendingAnswer) + } + } + if len(answer.Waits) != 1 || answer.Waits[0].Key != delivery.PendingKey(2) || !slices.Equal(answer.Waits[0].Holds, []string{"avahi", delivery.IndexerModule}) || + answer.Waits[0].Said != "take-in.pending."+delivery.PendingKey(2) || len(answer.Waits[0].Keys) != 2 { + t.Fatalf("indexer_pending = %+v", answer) + } + b.said = nil + if r := b.ix.DropPending(ctx, delivery.PendingKey(2), "lead@shanks", "replaced by m.3"); !r.Stored { + t.Fatalf("drop: %+v", r) + } + if len(b.said) < 2 || !b.said[0].Dropped || b.said[0].Key != delivery.PendingKey(2) || b.said[0].By != "lead@shanks" || b.said[1].Key != delivery.PendingKey(3) { + t.Fatalf("drop said %+v: want m.2 taken (dropped, by) and m.3 pending", b.said) + } + b.said = nil + if r := b.ix.TakeInPendingByHand(ctx, delivery.PendingKey(3), "lead@shanks", "the named indexer can never run"); !r.Stored { + t.Fatalf("hand take-in: %+v", r) + } + if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || len(b.said) != 1 || b.said[0].Kind != delivery.TakeInTaken || b.said[0].By != "lead@shanks" { + t.Errorf("hand take-in: pending %v, avahi at %s, said %+v", b.keys(), b.commit(t, "avahi"), b.said) + } + }) + t.Run("the dry run names every field of the indexer", func(t *testing.T) { + // dry (pending.go) builds an indexer field by field: a field added to Indexer is named there, or this fails. + named := map[string]bool{} + for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", + "JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk", + "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Pending", "SayTakeIn"} { + named[f] = true + } + ty := reflect.TypeOf(Indexer{}) + for i := range ty.NumField() { + if f := ty.Field(i); f.IsExported() && !named[f.Name] { + t.Errorf("Indexer.%s is not named by dry", f.Name) + } + } + }) +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go new file mode 100644 index 00000000..33c08f4c --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -0,0 +1,69 @@ +package index + +import ( + "context" + "fmt" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// PendingTools are the indexer's tools of what waits for a take-in (novox/hq ADR 0633 §5): indexer_pending reads the +// pending state, and indexer_take_in_pending and indexer_drop_pending steer it by an ask in takeins, as indexer_take_in +// does. pending is the pending state as the tools bundle reads it; ask writes an ask and waits for the indexer's answer. +func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool { + str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s } + hand := func(drop bool) func(map[string]any) (any, error) { + return func(args map[string]any) (any, error) { + key, value, err := NewPendingAsk(str(args, "key"), str(args, "why"), stdio.CallerOf(args), drop, now()) + if err != nil { + return nil, err + } + a, err := ask(key, value) + if err != nil { + return nil, err + } + return a.Result, nil + } + } + return []stdio.Tool{ + {Name: PendingTool, + Description: "What waits for a take-in (ADR 0633): merges the indexer holds in its pending state instead of taking them in. " + + "A revision (what revision, its key its oldest merge's m.) waits for the build of the indexer it names " + + "(waits-for): the running indexer wrote only what building and moving that indexer needs, and the new indexer takes " + + "it in at its start. A group (what group, key g.) is a delivery group whose members are gathered until the " + + "forge bridge says its end; a merge in one of its members' repositories waits behind it (behind). Each names its " + + "merges, since when it waits, the modules it holds back (holds) and the last take-in event said of it. Empty waits: " + + "every merge is taken in. The watchman raises take-in-waits past 60 minutes for a revision, 30 for a group; " + + "indexer_take_in_pending and indexer_drop_pending steer a wait.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { + var unread []string + entries, err := ReadPending(context.Background(), pending, func(format string, args ...any) { + unread = append(unread, fmt.Sprintf(format, args...)) + }) + if err != nil { + return nil, err + } + return delivery.PendingAnswer{Waits: PendingWaits(entries), Unread: unread}, nil + }}, + {Name: TakeInPendingTool, + Description: "Take a wait in by hand with the running indexer and its judge (ADR 0633 §5): the revision under its m. key, " + + "whatever build of the indexer it names, or a delivery group under its g. key with the members gathered so far. For " + + "when the named indexer can never run, or a group's end never comes. A hand act: who and why are said in the " + + "indexer's journal and on the take-in.taken event, which clears take-in-waits. indexer_pending names the keys.", + Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": "the wait's key, m. or g., as indexer_pending names it"}, + "why": map[string]any{"type": "string", "description": "why it is taken in by hand, in plain words"}}}, + Run: hand(false)}, + {Name: DropPendingTool, + Description: "Drop one pending merge by hand, abandoned because a later merge replaced it (ADR 0633 §5): it is removed from " + + "the pending state and nothing of it is taken in; what waited with it is worked out again. A hand act: who and why are " + + "said in the journal and on the take-in.taken event (dropped).", + Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": "the pending merge's key, m., as indexer_pending names its merges' wait"}, + "why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}}, + Run: hand(true)}, + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/source_test.go b/modules/mesh-delivery-indexer/internal/index/source_test.go index 8bb56ebd..31961771 100644 --- a/modules/mesh-delivery-indexer/internal/index/source_test.go +++ b/modules/mesh-delivery-indexer/internal/index/source_test.go @@ -51,6 +51,17 @@ func TestASourceRevisionStampsTheModulesOfTheRepositoriesHeadItNames(t *testing. } func TestAMergeHeardAfterTheSourceNamesItsHeadIsStampedWhenWritten(t *testing.T) { // 1344/AC1 (the source first) + // A source revision older than the one held is left (merged here from a test of its own, for the test budget). + t.Run("an older source revision is left", func(t *testing.T) { + ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs)) + ctx := context.Background() + _ = ix.Apply(ctx, mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"})) + _ = ix.ApplySource(ctx, 9, sourceOf("novox/mesh-delivery", merge)) + _ = ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)) + if m, _ := moduleOf(t, mods, "a"); m.Src != 9 { + t.Errorf("module a Src = %d, want 9", m.Src) + } + }) ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs)) ctx := context.Background() if err := ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)); err != nil { @@ -93,14 +104,3 @@ func TestAMergeThatNoSourceRevisionNamesIsWrittenAsToday(t *testing.T) { // 1344 t.Errorf("module a = %+v, want the new merge with Src 0", m) } } - -func TestASourceRevisionOlderThanTheOneHeldIsLeft(t *testing.T) { - ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs)) - ctx := context.Background() - _ = ix.Apply(ctx, mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"})) - _ = ix.ApplySource(ctx, 9, sourceOf("novox/mesh-delivery", merge)) - _ = ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)) - if m, _ := moduleOf(t, mods, "a"); m.Src != 9 { - t.Errorf("module a Src = %d, want 9", m.Src) - } -} diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 579ba62b..f1982e05 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -33,7 +33,11 @@ type TakeInAsk struct { // Judge asks the manifests to be judged again now (indexer_judge_again) instead of the module to be taken in. Judge bool `json:"judge,omitempty"` // Digest names, with Judge, the one manifest digest of the module to judge (indexer_judge with a digest). - Digest string `json:"digest,omitempty"` + Digest string `json:"digest,omitempty"` + // Pending asks the wait under this take-in key to be taken in by hand (indexer_take_in_pending), or with Drop the + // one pending merge under it to be dropped (indexer_drop_pending), instead of a module (pending.go, ADR 0633 §5). + Pending string `json:"pending,omitempty"` + Drop bool `json:"drop,omitempty"` By string `json:"by,omitempty"` At time.Time `json:"at"` Result *TakenIn `json:"result,omitempty"` @@ -52,8 +56,8 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) { if a.V != 1 { return a, fmt.Errorf("a takeins value of version %d, not 1", a.V) } - if a.Module == "" && a.Repository == "" { - return a, errors.New("a takeins value naming no module or repository") + if a.Module == "" && a.Repository == "" && a.Pending == "" { + return a, errors.New("a takeins value naming no module, repository or pending wait") } return a, nil } @@ -82,3 +86,25 @@ func NewTakeInRepository(repository, why, by string, at time.Time) (string, []by data, err := json.Marshal(TakeInAsk{V: 1, Repository: repository, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) return "repository." + key, data, err } + +// NewPendingAsk is the value indexer_take_in_pending (drop false) or indexer_drop_pending (drop true) writes: its key +// in takeins, `pending..`, and the value. key is a wait's key as indexer_pending names it, for a +// drop one pending merge's. +func NewPendingAsk(key, why, by string, drop bool, at time.Time) (string, []byte, error) { + key = strings.TrimSpace(key) + if !strings.HasPrefix(key, "m.") && !(strings.HasPrefix(key, "g.") && !drop) { + if drop { + return "", nil, errors.New("name the pending merge to drop by its key, m., as indexer_pending names it") + } + return "", nil, errors.New("name the wait to take in by its key, m. or g., as indexer_pending names it") + } + if strings.TrimSpace(why) == "" { + return "", nil, errors.New("say why (why): a hand act on what waits is said in the journal and on its take-in event") + } + act := "take-in" + if drop { + act = "drop" + } + data, err := json.Marshal(TakeInAsk{V: 1, Pending: key, Drop: drop, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) + return "pending." + act + "." + key, data, err +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 4911041b..d8f68d3b 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -16,6 +16,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-delivery/sdk/graph" + + "mesh-delivery-indexer/internal/index" ) // The controller's manifest reader is mesh-controller internal/catalogue (Manifest, manifest.go: Consumes :521, @@ -28,6 +30,12 @@ type state struct { History int `json:"history,omitempty"` } +type streamDecl struct { + Name string `json:"name"` + Events []string `json:"events"` + MaxAgeDays int `json:"max-age-days,omitempty"` +} + type artifact struct { Name string `json:"name"` Kind string `json:"kind"` @@ -51,6 +59,7 @@ type manifest struct { Emits []string `json:"emits"` Consumes []string `json:"consumes"` State []state `json:"state"` + Streams []streamDecl `json:"streams"` Reads []string `json:"reads"` Invokes []string `json:"invokes"` OwnSecrets map[string]string `json:"own-secrets"` @@ -99,7 +108,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, - {Name: graph.State, History: graph.History}}; !reflect.DeepEqual(m.State, want) { + {Name: graph.State, History: graph.History}, {Name: index.PendingState, 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-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { @@ -125,7 +134,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle // reads the verdicts state (indexer_verdicts) and asks the indexer to judge again through the takeins state // (indexer_judge; issue 774, clause 3). 1966/AC11: indexer_old_sections reads the count of the old sections (state oldsections). - if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge"}; !reflect.DeepEqual(m.Tools, want) { + // 1585/T12, T13 (ADR 0633 §5): indexer_pending reads the pending state; indexer_take_in_pending and indexer_drop_pending + // ask through takeins. + if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } // ADR 0604 §2: it holds the seat mesh-catalog, serving its four verbs from its tools bundle (which stays without a @@ -134,8 +145,13 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build !slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents"}) { t.Errorf("claims = %+v, want mesh-catalog with its four verbs", m.Claims) } - // No streams, no bus, no other seat (ADR 0437 §4); it says no event. - for _, k := range []string{"streams", "logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { + // 1585/T11 (ADR 0633 §1, §5): its one stream, take-in, an old streams entry (the indexer is foundation, ADR 0595 §11), + // carries the take-in events. + if want := []streamDecl{{Name: delivery.TakeInStream, Events: []string{"take-in.**"}, MaxAgeDays: 7}}; !reflect.DeepEqual(m.Streams, want) { + t.Errorf("streams = %+v, want %+v", m.Streams, want) + } + // No other bus section, no other seat (ADR 0437 §4). + for _, k := range []string{"logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the indexer has none", k) } diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 3fa9e8a7..fafd9bcc 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -18,7 +18,10 @@ "indexer_take_in", "indexer_old_sections", "indexer_verdicts", - "indexer_judge" + "indexer_judge", + "indexer_pending", + "indexer_take_in_pending", + "indexer_drop_pending" ], "consumes": [ "mesh-build-go.toolchain.reported", @@ -38,7 +41,11 @@ { "name": "verdicts", "history": 5 }, { "name": "oldsections", "history": 1 }, { "name": "packages", "history": 5 }, - { "name": "graph", "history": 5 } + { "name": "graph", "history": 5 }, + { "name": "pending", "history": 1 } + ], + "streams": [ + { "name": "take-in", "events": ["take-in.**"], "max-age-days": 7 } ], "emits": [ "manifest-refused",