diff --git a/contracts/go/seats/meshwants/contract.go b/contracts/go/seats/meshwants/contract.go index e86953fe..3ec3c4ea 100644 --- a/contracts/go/seats/meshwants/contract.go +++ b/contracts/go/seats/meshwants/contract.go @@ -67,8 +67,16 @@ const SubjectAssignmentRevision = "mesh.seat.mesh-wants.event.assignment-revisio // SubjectDifference is where difference is published. const SubjectDifference = "mesh.seat.mesh-wants.event.difference" -// SubjectWanted is where wanted is published. -const SubjectWanted = "mesh.seat.mesh-wants.event.wanted" +// StreamWanted is the stream wanted is kept in. +const StreamWanted = "SEAT_mesh-wants_wanted" + +// SubjectWanted is where wanted is published, on mesh.seat.mesh-wants.event.wanted... +func SubjectWanted(node, module string) string { + return "mesh.seat.mesh-wants.event.wanted." + node + "." + module +} + +// SubjectWantedAll hears every wanted. +const SubjectWantedAll = "mesh.seat.mesh-wants.event.wanted.*.*" // SubjectWaitHeld is where wait-held is published. const SubjectWaitHeld = "mesh.seat.mesh-wants.event.wait-held" diff --git a/contracts/go/seats/meshwants/contract_test.go b/contracts/go/seats/meshwants/contract_test.go index d18fcac6..880e34d7 100644 --- a/contracts/go/seats/meshwants/contract_test.go +++ b/contracts/go/seats/meshwants/contract_test.go @@ -25,10 +25,10 @@ func TestFixtures(t *testing.T) { }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.assignment-revision$")}, {name: "a difference worked out", subject: "mesh.seat.mesh-wants.event.difference", given: "{\"changes\":1,\"from\":40,\"nodes\":[\"n1\"],\"source\":3589,\"to\":41,\"v\":1}", wire: "{\"changes\":1,\"from\":40,\"nodes\":[\"n1\"],\"source\":3589,\"to\":41,\"v\":1}", decode: func(b []byte) (any, error) { var v DifferenceEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.difference$")}, - {name: "an entry wanted from a trunk revision", subject: "mesh.seat.mesh-wants.event.wanted", given: "{\"at\":\"2026-11-09T19:00:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"revision\":40,\"source\":3589,\"v\":1}", wire: "{\"at\":\"2026-11-09T19:00:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"revision\":40,\"source\":3589,\"v\":1}", - decode: func(b []byte) (any, error) { var v WantedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wanted$")}, - {name: "a want removed, its node left or its module unplaced", subject: "mesh.seat.mesh-wants.event.wanted", given: "{\"at\":\"2026-11-09T19:05:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"removed\":true,\"revision\":41,\"v\":1}", wire: "{\"at\":\"2026-11-09T19:05:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"removed\":true,\"revision\":41,\"v\":1}", - decode: func(b []byte) (any, error) { var v WantedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wanted$")}, + {name: "an entry wanted from a trunk revision", subject: "mesh.seat.mesh-wants.event.wanted.n1.xclip", given: "{\"at\":\"2026-11-09T19:00:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"revision\":40,\"source\":3589,\"v\":1}", wire: "{\"at\":\"2026-11-09T19:00:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"revision\":40,\"source\":3589,\"v\":1}", + decode: func(b []byte) (any, error) { var v WantedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wanted\\.[^.*> ]+\\.[^.*> ]+$")}, + {name: "a want removed, its node left or its module unplaced", subject: "mesh.seat.mesh-wants.event.wanted.n1.xclip", given: "{\"at\":\"2026-11-09T19:05:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"removed\":true,\"revision\":41,\"v\":1}", wire: "{\"at\":\"2026-11-09T19:05:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"removed\":true,\"revision\":41,\"v\":1}", + decode: func(b []byte) (any, error) { var v WantedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wanted\\.[^.*> ]+\\.[^.*> ]+$")}, {name: "a composing wait held with its bound", subject: "mesh.seat.mesh-wants.event.wait-held", given: "{\"bound\":\"30m\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"part\":\"node-engine\",\"since\":\"2026-11-09T19:00:00Z\",\"step\":\"compose\",\"version\":\"src2:7dd0db8e98224f46121a5ae05c0b9c5a430eff531a5120aa40b1a59237853a3f\"}", wire: "{\"bound\":\"30m\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"part\":\"node-engine\",\"since\":\"2026-11-09T19:00:00Z\",\"step\":\"compose\",\"version\":\"src2:7dd0db8e98224f46121a5ae05c0b9c5a430eff531a5120aa40b1a59237853a3f\"}", decode: func(b []byte) (any, error) { var v WaitHeldEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-held$")}, {name: "a wait ended, released by hand", subject: "mesh.seat.mesh-wants.event.wait-ended", given: "{\"bound\":\"30m\",\"by\":\"lead@shanks: the node is away\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"how\":\"released\",\"module\":\"xclip\",\"node\":\"n1\",\"part\":\"node-engine\",\"since\":\"2026-11-09T19:00:00Z\",\"step\":\"compose\",\"version\":\"src2:7dd0db8e98224f46121a5ae05c0b9c5a430eff531a5120aa40b1a59237853a3f\"}", wire: "{\"bound\":\"30m\",\"by\":\"lead@shanks: the node is away\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"how\":\"released\",\"module\":\"xclip\",\"node\":\"n1\",\"part\":\"node-engine\",\"since\":\"2026-11-09T19:00:00Z\",\"step\":\"compose\",\"version\":\"src2:7dd0db8e98224f46121a5ae05c0b9c5a430eff531a5120aa40b1a59237853a3f\"}", diff --git a/contracts/seats/mesh-wants/fixtures/events.json b/contracts/seats/mesh-wants/fixtures/events.json index 85b2e909..6d7be929 100644 --- a/contracts/seats/mesh-wants/fixtures/events.json +++ b/contracts/seats/mesh-wants/fixtures/events.json @@ -36,7 +36,7 @@ { "name": "an entry wanted from a trunk revision", "of": "event:wanted", - "subject": "mesh.seat.mesh-wants.event.wanted", + "subject": "mesh.seat.mesh-wants.event.wanted.n1.xclip", "given": { "v": 1, "node": "n1", @@ -50,7 +50,7 @@ { "name": "a want removed, its node left or its module unplaced", "of": "event:wanted", - "subject": "mesh.seat.mesh-wants.event.wanted", + "subject": "mesh.seat.mesh-wants.event.wanted.n1.xclip", "given": { "v": 1, "node": "n1", diff --git a/contracts/seats/mesh-wants/seat.json b/contracts/seats/mesh-wants/seat.json index 05966ed3..687862de 100644 --- a/contracts/seats/mesh-wants/seat.json +++ b/contracts/seats/mesh-wants/seat.json @@ -303,6 +303,11 @@ }, { "name": "wanted", + "subject": "wanted..", + "keep": { + "max-age-days": 0, + "last-per-subject": true + }, "body": { "type": "object", "description": "an entry written as a want, from a trunk revision only", diff --git a/contracts/ts/seats/meshwants/index.d.ts b/contracts/ts/seats/meshwants/index.d.ts index 399bdeaa..e52a739e 100644 --- a/contracts/ts/seats/meshwants/index.d.ts +++ b/contracts/ts/seats/meshwants/index.d.ts @@ -40,7 +40,11 @@ export declare const assignmentRevisionSubject: "mesh.seat.mesh-wants.event.assi export declare const differenceSubject: "mesh.seat.mesh-wants.event.difference"; -export declare const wantedSubject: "mesh.seat.mesh-wants.event.wanted"; +export declare const wantedStream: "SEAT_mesh-wants_wanted"; + +export declare function wantedSubject(node: string, module: string): string; + +export declare const wantedSubjectAll: "mesh.seat.mesh-wants.event.wanted.*.*"; export declare const waitHeldSubject: "mesh.seat.mesh-wants.event.wait-held"; diff --git a/contracts/ts/seats/meshwants/index.js b/contracts/ts/seats/meshwants/index.js index df1d0f62..15556f82 100644 --- a/contracts/ts/seats/meshwants/index.js +++ b/contracts/ts/seats/meshwants/index.js @@ -40,7 +40,13 @@ export const assignmentRevisionSubject = "mesh.seat.mesh-wants.event.assignment- export const differenceSubject = "mesh.seat.mesh-wants.event.difference"; -export const wantedSubject = "mesh.seat.mesh-wants.event.wanted"; +export const wantedStream = "SEAT_mesh-wants_wanted"; + +export function wantedSubject(node, module) { + return "mesh.seat.mesh-wants.event.wanted." + node + "." + module; +} + +export const wantedSubjectAll = "mesh.seat.mesh-wants.event.wanted.*.*"; export const waitHeldSubject = "mesh.seat.mesh-wants.event.wait-held"; 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 23c30b89..711bcda8 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 @@ -288,11 +288,15 @@ func main() { tools = append(tools, catalogTools(trunk.State)...) // The seat mesh-catalog's verb revision, a first view of the catalog half answered in this runtime of its own to exactly // the parts whose entry declares it (novox/hq ADR 0663). - tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)})) + tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}, keptRead{stdio.State(index.NamedState)})) tools = append(tools, index.CatalogRevisionTools(revisions, asks, time.Now, func(what, why string) error { _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.DropCandidateGraphTool, Verb: "drop-candidate-graph"}) return err })...) + tools = append(tools, index.NamedTools(asks, time.Now, func(what, why string) error { + _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.NamedDropTool, Verb: "named-drop"}) + return err + })...) tools = append(tools, index.RegistrationTools(keptRead{stdio.State(index.ProposalsState)}, asks, time.Now, func(cause, verb, what, why string) error { _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause, Verb: verb}) return err 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 ef756cfb..1a791453 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 @@ -29,11 +29,13 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" @@ -595,6 +597,20 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before . if _, err := js.CreateOrUpdateConsumer(ctx, "FORGE", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } + // The two seat events that name a fingerprint, each on its stream kept last per subject with the indexer's durable on it, and the + // state named the grants part makes from the manifest (design 47 G §675, §695); both streams empty, so the start's read ends at once. + for stream, subject := range map[string]string{meshwants.StreamWanted: meshwants.SubjectWantedAll, nodeengine.StreamRunning: nodeengine.SubjectRunningAll} { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy, + DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy, FilterSubjects: []string{subject}}); err != nil { + t.Fatal(err) + } + } + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.NamedBucket, History: 1}); err != nil { + t.Fatal(err) + } // The build modules' streams for toolchain.reported (declared by them, heard by the indexer) and its durable consumers. for module, stream := range map[string]string{"mesh-build-go": "TOOLCHAINS_GO", "mesh-build-node": "TOOLCHAINS_NODE"} { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.stream." + module + ".event.toolchain.reported"}}); err != nil { @@ -640,7 +656,9 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before . } t.Cleanup(func() { _ = waiter.Unsubscribe() }) // The shared events stream: the indexer's ordinary events and the assigner's views events (ADR 0635 §5). - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.mesh-catalog.event.>", "mesh.seat.mesh-delivery.event.>", "mesh.seat.mesh-delivery-checker.event.>", + "mesh.seat.mesh-conditions.event.>", "mesh.seat.operator-channel.event.>", "mesh.seat.go-package-registry.event.>", "mesh.seat.npm-package-registry.event.>", + "mesh.seat.mesh-module-assigner.event.>"}}); err != nil { // not the seats kept in streams of their own (wanted and running, above) t.Fatal(err) } // The indexer's own stream of its findings' events (design 47 G §383), which its finding writer says into (A-30). @@ -1414,3 +1432,36 @@ func serveAssignerAssignments(t *testing.T, ctx context.Context, js jetstream.Je } t.Cleanup(func() { _ = sub.Unsubscribe() }) } + +// TestIndexerKeepsWhatAWantAndARunningBuildName (design 47 G §675, §693, A-61): a want said before the start on +// wanted.. and a running build of another node are read whole at the start into the state named. +func TestIndexerKeepsWhatAWantAndARunningBuildName(t *testing.T) { + fp := "src2:" + strings.Repeat("a", 64) + at := time.Now().UTC() + js, ctx, _, _ := startIndexer(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) { + want, _ := json.Marshal(meshwants.WantedEvent{V: 1, Node: "n1", Module: "avahi", Fp: fp, Revision: 1, At: &at}) + run, _ := json.Marshal(nodeengine.Running{V: 1, Node: "n2", Module: "avahi", Fingerprint: fp, At: at}) + for subject, data := range map[string][]byte{meshwants.SubjectWanted("n1", "avahi"): want, nodeengine.SubjectRunning("n2", "avahi"): run} { + if _, err := js.Publish(ctx, subject, data); err != nil { + t.Fatal(err) + } + } + }) + named, err := js.KeyValue(ctx, index.NamedBucket) + if err != nil { + t.Fatal(err) + } + members := func(want int) []string { + var got []string + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + got, _ = index.NamedMembers(ctx, kvStore{named}, fp) + if len(got) == want { + break + } + } + return got + } + if got := members(2); !slices.Equal(got, []string{"n1.avahi", "n2.avahi"}) { + t.Fatalf("named holds %v for the fingerprint within 20 s of the start, want both members", got) + } +} 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 5beaf5bc..ffeb6b6e 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -430,7 +430,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } }) readWhenBound(seatReader, seatDefs, marks) - ix := &index.Indexer{Findings: writer, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known, + // Kept while named (design 47 G §675): the want and the running build that name a fingerprint, heard below. + named := index.NewNamed(awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.NamedBucket, "no entry is kept for a want or a running build that names it", logger.Printf)}, logger.Printf) + ix := &index.Indexer{NamedSets: named, Findings: writer, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known, Verdicts: awaitedStore{b: verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred), Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), @@ -452,6 +454,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.ProposalsBucket, "no change of a registration waits for its warrant", logger.Printf)}, routerRecordUnread, ix, logger.Printf) ix.Registry = registry + named.Changed = ix.RemoveUnlisted ix.SeedModules = func(context.Context) ([]index.SeedModule, error) { return nil, errors.New("the first registrations are not composed by this build: what the build seats answer steps with is not read yet (design question)") } @@ -683,6 +686,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { <-ctx.Done(); hub.Stop() }() return nil }) + hearNamed(ctx, js, cred, notHeard, named, logger.Printf) // The checker's verdicts carry a pull request's out-of-step files (DLV-I20): on the indexer's durable consumer of the // checker's stream asks. Where it is not there yet, no pull request's out-of-step files are kept until the indexer // starts again, said. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_events.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_events.go new file mode 100644 index 00000000..b0442561 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_events.go @@ -0,0 +1,55 @@ +package main + +import ( + "context" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-indexer/internal/index" +) + +// binder binds an observer's consumer of a family and says in a finding when it cannot (heard.Writer). +type binder interface { + Bind(family, consequence string, bind func() error) error +} + +// hearNamed hears the two events that name a fingerprint (design 47 G §675, §693; A-61): the seat mesh-wants' `wanted..` +// and the node-engine seat's `running..`, each on its own stream kept last per subject, read whole at the start +// and then followed on the indexer's durable (§556). The end of each whole read is said to named, which reconciles its sets once +// both are read; a stream the bus does not hold yet is the finding family-not-heard, and no entry is removed meanwhile. +func hearNamed(ctx context.Context, js jetstream.JetStream, cred process.Account, notHeard binder, named *index.Named, logf func(string, ...any)) { + for _, s := range []struct { + stream, kind string + each func(context.Context, []byte) error + }{ + {meshwants.StreamWanted, index.NamedWanted, named.HearWanted}, + {nodeengine.StreamRunning, index.NamedRunning, named.HearRunning}, + } { + _ = notHeard.Bind(s.stream, "no entry is removed, since what a want or a running build names is not heard until the indexer starts again", func() error { + cc, err := cred.HearSeatEvents(ctx, js, s.stream, func(m jetstream.Msg) { + if err := m.Ack(); err != nil { + logf("acknowledging %s: %v", m.Subject(), err) + } + if err := s.each(ctx, m.Data()); err != nil { + logf("%s: %v", m.Subject(), err) + } + }, logf, process.OnRead(func(state string) { + if state != process.ReadDone { + return + } + if err := named.ReadDone(ctx, s.kind); err != nil { + logf("the sets of %s are not reconciled with %s: %v", index.NamedState, s.stream, err) + } + })) + if err != nil { + return err + } + go func() { <-ctx.Done(); cc.Stop() }() + return nil + }) + } +} 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 a1c9d068..552b9afb 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -27,6 +27,7 @@ type takerIn interface { FingerprintsAgain(ctx context.Context, module string) index.TakenIn RecomputeByHand(ctx context.Context, by, why string) index.TakenIn DropCandidateGraph(ctx context.Context, group, heads, by, why string) index.TakenIn + NamedDrop(ctx context.Context, fingerprint, member, by, why string) index.TakenIn RegistryAct(ctx context.Context, a index.TakeInAsk) index.TakenIn } @@ -54,6 +55,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream } else if a.CandidateGroup != "" && a.Drop { asked, r = "drop of a candidate's catalog half", ix.DropCandidateGraph(ctx, a.CandidateGroup, a.CandidateHeads, a.By, a.Why) a.Module = r.Module + } else if a.NamedFingerprint != "" && a.Drop { + asked, r = "drop of a member of a fingerprint's names", ix.NamedDrop(ctx, a.NamedFingerprint, a.NamedMember, a.By, a.Why) + a.Module = r.Module } else if a.Fingerprints { asked, r = "computing of fingerprints again", ix.FingerprintsAgain(ctx, a.Module) } else if a.RecoverTries { diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index af3efc9b..6c8e1e1e 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -5,6 +5,7 @@ go 1.26.0 require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogkey v0.0.0-00010101000000-000000000000 @@ -15,6 +16,7 @@ require ( git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/npmpackageregistry v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index d86201b2..f147a15a 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -102,6 +102,9 @@ type Indexer struct { Revisions Objects // EntryIndex is the state entry-index (entry_verb.go, design 47 G §675); nil keeps none. EntryIndex Store + // NamedSets is the state named: which want or running build names a fingerprint (named.go, §675); nil keeps none, and no + // entry is then kept for a name. + NamedSets *Named // SayCatalogRevision says the seat mesh-catalog's event catalog-revision for a revision kept; nil says nothing. SayCatalogRevision func(ctx context.Context, e meshcatalog.CatalogRevisionEvent) error // SayFill says the seat mesh-catalog's event filling, full or room of the catalog graph revisions, under the duplicate diff --git a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go index d27fa244..78310142 100644 --- a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go +++ b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go @@ -286,6 +286,9 @@ func (ix *Indexer) keepEntries(ctx context.Context, entries map[string]meshcatal ix.logf("catalog graph: revision %d of source revision %d is not kept: %v", rev.Revision, src, err) return err } + if _, err := ix.removeUnlisted(ctx); err != nil { + ix.logf("catalog graph: the entries nothing lists or names any more are not removed: %v", err) // at the first revision after none does (§675) + } if after, capacity, err := ix.Revisions.Fill(ctx); fillErr == nil && err == nil && capacity > 0 && used*5 < capacity*4 && after*5 >= capacity*4 { ix.sayFill(ctx, meshcatalog.EventFilling, fmt.Sprintf("%d", rev.Revision), meshcatalog.Fill{Bytes: after, Cap: capacity}) } diff --git a/modules/mesh-delivery-indexer/internal/index/entry_verb.go b/modules/mesh-delivery-indexer/internal/index/entry_verb.go index 7909c321..3ef54387 100644 --- a/modules/mesh-delivery-indexer/internal/index/entry_verb.go +++ b/modules/mesh-delivery-indexer/internal/index/entry_verb.go @@ -104,16 +104,18 @@ func EntryTool(o Objects, index Store) stdio.Tool { // EntriesAnswer is what indexer_entries answers: for a fingerprint its key and its object's digest, and whether the object // is kept; with none given, every fingerprint the state holds. type EntriesAnswer struct { - Fingerprint string `json:"fingerprint,omitempty"` - Digest string `json:"digest,omitempty"` - Kept bool `json:"kept,omitempty"` - Keys []string `json:"keys,omitempty"` + Fingerprint string `json:"fingerprint,omitempty"` + Digest string `json:"digest,omitempty"` + Kept bool `json:"kept,omitempty"` + // Named is the set of `.` whose want or running build names the fingerprint (named.go, §675). + Named []string `json:"named,omitempty"` + Keys []string `json:"keys,omitempty"` } // EntriesTool is indexer_entries (class read): one fingerprint's key and the digest of its object, or every key of // entry-index with none given. -func EntriesTool(o Objects, index Store) stdio.Tool { - return stdio.Tool{Name: EntryIndexTool, Description: "Read the indexer's entry-index: the digest of the catalog graph entry of a fingerprint and whether its object is kept; with no fingerprint, every key it holds.", +func EntriesTool(o Objects, index, named Store) stdio.Tool { + return stdio.Tool{Name: EntryIndexTool, Description: "Read the indexer's entry-index: the digest of the catalog graph entry of a fingerprint, whether its object is kept and the set of `.` whose want or running build names it (the state named, which keeps the entry while it is not empty); with no fingerprint, every key it holds. indexer_named_drop removes one member.", Input: map[string]any{"type": "object", "properties": map[string]any{"fingerprint": map[string]any{"type": "string", "description": "the entry's fingerprint; none lists every key"}}}, Run: func(args map[string]any) (any, error) { ctx := context.Background() @@ -138,6 +140,11 @@ func EntriesTool(o Objects, index Store) stdio.Tool { out.Kept = true } } + if named != nil { + if out.Named, err = NamedMembers(ctx, named, fp); err != nil { + return nil, err + } + } return out, nil }} } diff --git a/modules/mesh-delivery-indexer/internal/index/named.go b/modules/mesh-delivery-indexer/internal/index/named.go new file mode 100644 index 00000000..af1b4c2a --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/named.go @@ -0,0 +1,383 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// Kept while named (design 47 G §675, §693, §695): the indexer keeps, in its old state section named, one key per +// fingerprint holding the set of `.` whose want (the seat mesh-wants' wanted) or running build (the node-engine +// seat's running) names it, written by compare-and-set and capped at 4 MiB (max-mib, §695). An entry object is removed only +// when no kept revision lists it and its set is empty. Both events are streams kept last per subject, read whole at the +// start and then followed (§556), so the sets are the same whenever the indexer started. +const ( + // NamedState is the old state section of the sets of members by fingerprint. + NamedState = "named" + // NamedBucket is the bucket of the state. + NamedBucket = Module + "_" + NamedState + // NamedDropTool is the tool that removes one member from one fingerprint's set by hand (class act). + NamedDropTool = "indexer_named_drop" + // NamedWanted and NamedRunning name the two streams the sets are read from, for Named.ReadDone. + NamedWanted = "wanted" + NamedRunning = "running" +) + +// NamedValue is one key of named: the fingerprint and the members that name it, sorted. +type NamedValue struct { + V int `json:"v"` + Fp string `json:"fp"` + Members []string `json:"members"` +} + +// NamedKey is the key of a fingerprint in named, the fingerprint escaped as a key's name is (§73). +func NamedKey(fingerprint string) string { return keys.EscapeName(fingerprint) } + +// NamedMember is the member of a node's module in a set. +func NamedMember(node, module string) string { return node + "." + module } + +// namedAt is the fingerprint one source last named for a member, and when it said so. +type namedAt struct { + fp string + at time.Time +} + +// Named keeps the sets. What each member's want and running last named is held in memory only, read again at every start from +// the two streams, so a member whose fingerprint moved while the indexer was away leaves the old one's set at the end of the +// start's read (Reconcile). +type Named struct { + // Store is the state named. + Store Store + // Changed is called, with no lock held, after a set lost a member: an entry object it kept may be removed now (§675); nil + // calls nothing. + Changed func(ctx context.Context) + Logf func(format string, args ...any) + + mu sync.Mutex + want map[string]namedAt + run map[string]namedAt + reads map[string]bool // the streams read whole at the start + ready bool +} + +// NewNamed is the sets over store. +func NewNamed(store Store, logf func(string, ...any)) *Named { + if logf == nil { + logf = func(string, ...any) {} + } + return &Named{Store: store, Logf: logf, want: map[string]namedAt{}, run: map[string]namedAt{}, reads: map[string]bool{}} +} + +// Ready says the two streams were read whole and the sets reconciled with them: until then no entry object is removed, so a +// restart reads named whole before it removes anything (§675). +func (n *Named) Ready() bool { + n.mu.Lock() + defer n.mu.Unlock() + return n.ready +} + +// HearWanted applies one wanted event, said on mesh.seat.mesh-wants.event.wanted..: the want names its fingerprint. +func (n *Named) HearWanted(ctx context.Context, data []byte) error { + var e meshwants.WantedEvent + if err := json.Unmarshal(data, &e); err != nil { + return fmt.Errorf("a wanted that does not read: %w", err) + } + if e.Node == "" || e.Module == "" || e.Fp == "" { + return errors.New("a wanted names no node, module or fingerprint") + } + var at time.Time + if e.At != nil { + at = *e.At + } + return n.apply(ctx, NamedMember(e.Node, e.Module), true, e.Fp, at) +} + +// HearRunning applies one running event: the build it names; a stopped one, or one with no fingerprint, names none (design 47 +// G §365). +func (n *Named) HearRunning(ctx context.Context, data []byte) error { + var r nodeengine.Running + if err := json.Unmarshal(data, &r); err != nil { + return fmt.Errorf("a running that does not read: %w", err) + } + if r.Node == "" || r.Module == "" { + return errors.New("a running names no node or module") + } + fp := r.Fingerprint + if r.Stopped { + fp = "" + } + return n.apply(ctx, NamedMember(r.Node, r.Module), false, fp, r.At) +} + +// apply changes what one source names for the member and writes the sets it moves; one at a time, so the two sources never +// write one set from a stale view of each other. +func (n *Named) apply(ctx context.Context, member string, wanted bool, fp string, at time.Time) error { + n.mu.Lock() + moved, err := n.applyLocked(ctx, member, wanted, fp, at) + n.mu.Unlock() + if moved && n.Changed != nil { + n.Changed(ctx) + } + return err +} + +func (n *Named) applyLocked(ctx context.Context, member string, wanted bool, fp string, at time.Time) (bool, error) { + src := n.run + if wanted { + src = n.want + } + old, had := src[member] + if had && at.Before(old.at) { + return false, nil // an event older than the one heard changes nothing + } + var touched []string + if had { + touched = append(touched, old.fp) + } + switch { + case fp == "": + delete(src, member) + default: + src[member] = namedAt{fp: fp, at: at} + touched = append(touched, fp) + } + return n.writeSets(ctx, member, touched) +} + +// holds says whether either source names fp for the member. +func (n *Named) holds(member, fp string) bool { + return n.want[member].fp == fp || n.run[member].fp == fp +} + +// writeSets makes each fingerprint's set hold the member exactly when a source names it; it answers whether a set lost one. +func (n *Named) writeSets(ctx context.Context, member string, fps []string) (lost bool, err error) { + slices.Sort(fps) + fps = slices.Compact(fps) + var errs []error + for _, fp := range fps { + if fp == "" { + continue + } + in := n.holds(member, fp) + changed := false + err := write(ctx, n.Store, NamedKey(fp), func(old []byte, found bool) ([]byte, bool, error) { + var v NamedValue + if found { + if err := json.Unmarshal(old, &v); err != nil { + return nil, false, fmt.Errorf("%s.%s does not read: %w", NamedState, NamedKey(fp), err) + } + } + has := slices.Contains(v.Members, member) + changed = has != in + switch { + case in && !has: + v.Members = append(v.Members, member) + case !in && has: + v.Members = slices.DeleteFunc(v.Members, func(m string) bool { return m == member }) + default: + return nil, false, nil + } + if len(v.Members) == 0 { + return nil, true, nil + } + slices.Sort(v.Members) + v.V, v.Fp = keys.Version, fp + data, err := json.Marshal(v) + return data, false, err + }) + if err != nil { + errs = append(errs, err) + continue + } + if changed && !in { + lost = true + } + } + return lost, errors.Join(errs...) +} + +// ReadDone says the stream (NamedWanted or NamedRunning) was read whole at the start; with both read, every key is reconciled +// with what the two named, a member whose build moved while the indexer was away leaving its old set, and the sets are ready. +func (n *Named) ReadDone(ctx context.Context, stream string) error { + n.mu.Lock() + n.reads[stream] = true + if n.ready || !n.reads[NamedWanted] || !n.reads[NamedRunning] { + n.mu.Unlock() + return nil + } + lost, err := n.reconcileLocked(ctx) + if err == nil { + n.ready = true + } + n.mu.Unlock() + if (lost || err == nil) && n.Changed != nil { + n.Changed(ctx) // the sets are ready: an entry kept back until now may go + } + return err +} + +func (n *Named) reconcileLocked(ctx context.Context) (bool, error) { + all, err := n.Store.Keys(ctx) + if err != nil { + return false, fmt.Errorf("the state %s cannot be listed: %w", NamedState, err) + } + var errs []error + lost := false + for _, key := range all { + err := write(ctx, n.Store, key, func(old []byte, found bool) ([]byte, bool, error) { + var v NamedValue + if !found || json.Unmarshal(old, &v) != nil { + return nil, false, nil + } + kept := slices.DeleteFunc(slices.Clone(v.Members), func(m string) bool { return !n.holds(m, v.Fp) }) + if len(kept) == len(v.Members) { + return nil, false, nil + } + lost = true + if len(kept) == 0 { + return nil, true, nil + } + v.Members = kept + data, err := json.Marshal(v) + return data, false, err + }) + if err != nil { + errs = append(errs, fmt.Errorf("%s.%s: %w", NamedState, key, err)) + } + } + return lost, errors.Join(errs...) +} + +// Members is the set of a fingerprint, sorted; none is an empty list. +func (n *Named) Members(ctx context.Context, fingerprint string) ([]string, error) { + return NamedMembers(ctx, n.Store, fingerprint) +} + +// NamedMembers reads a fingerprint's set from the state. +func NamedMembers(ctx context.Context, st Store, fingerprint string) ([]string, error) { + value, _, found, err := st.Get(ctx, NamedKey(fingerprint)) + if err != nil { + return nil, fmt.Errorf("the state %s cannot be read: %w", NamedState, err) + } + if !found { + return nil, nil + } + var v NamedValue + if err := json.Unmarshal(value, &v); err != nil { + return nil, fmt.Errorf("%s.%s does not read: %w", NamedState, NamedKey(fingerprint), err) + } + return v.Members, nil +} + +// IsNamed says whether a want or a running build names the fingerprint. An unreadable state names it, so no entry is removed on +// a read that failed. +func (n *Named) IsNamed(ctx context.Context, fingerprint string) (bool, error) { + n.mu.Lock() + defer n.mu.Unlock() + m, err := n.Members(ctx, fingerprint) + return err != nil || len(m) > 0, err +} + +// Drop removes one member from one fingerprint's set by hand (indexer_named_drop), for a member a want or a running build +// named that the mesh no longer holds; the next event of that member is applied as any is. It answers whether the member was in +// the set. +func (n *Named) Drop(ctx context.Context, fingerprint, member string) (bool, error) { + n.mu.Lock() + dropped := false + err := write(ctx, n.Store, NamedKey(fingerprint), func(old []byte, found bool) ([]byte, bool, error) { + var v NamedValue + if !found { + return nil, false, nil + } + if err := json.Unmarshal(old, &v); err != nil { + return nil, false, err + } + if !slices.Contains(v.Members, member) { + return nil, false, nil + } + dropped = true + v.Members = slices.DeleteFunc(v.Members, func(m string) bool { return m == member }) + if len(v.Members) == 0 { + return nil, true, nil + } + data, err := json.Marshal(v) + return data, false, err + }) + if dropped && err == nil { + if n.want[member].fp == fingerprint { + delete(n.want, member) + } + if n.run[member].fp == fingerprint { + delete(n.run, member) + } + } + n.mu.Unlock() + if dropped && err == nil && n.Changed != nil { + n.Changed(ctx) + } + return dropped, err +} + +// NamedDrop is indexer_named_drop answered: the member leaves the fingerprint's set, and an entry nothing names now may go. +func (ix *Indexer) NamedDrop(ctx context.Context, fingerprint, member, by, why string) TakenIn { + r := TakenIn{Module: "named " + fingerprint} + if ix.NamedSets == nil { + r.Answer = "the indexer keeps no state " + NamedState + ": nothing dropped" + return r + } + dropped, err := ix.NamedSets.Drop(ctx, fingerprint, member) + switch { + case err != nil: + r.Answer = member + " could not be dropped from " + fingerprint + ": " + err.Error() + case !dropped: + r.Answer = member + " is not in the set of " + fingerprint + ": nothing dropped" + default: + r.Stored = true + r.Answer = fmt.Sprintf("%s is dropped from the set of %s by %s (%s)", member, fingerprint, by, why) + } + ix.logf("%s", r.Answer) + return r +} + +// NamedTools is indexer_named_drop (class act): the tools bundle writes an ask in takeins and the indexer, the state's owner, +// removes the member; record says the hand act in the hand-act log before it is asked. +func NamedTools(ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(what, why string) error) []stdio.Tool { + return []stdio.Tool{{Name: NamedDropTool, + Description: "Remove one member (`.`) from the set of a fingerprint in the indexer's state named, which keeps the entry object of " + + "the fingerprint while a want or a running build names it (design 47 G §675): for a member the mesh no longer holds, " + + "which indexer_entries shows. A hand act, recorded through mesh-hand-acts before it is asked; an entry nothing lists or names " + + "any more is removed.", + Input: map[string]any{"type": "object", "required": []string{"fingerprint", "member", "why"}, "properties": map[string]any{ + "fingerprint": map[string]any{"type": "string", "description": "the entry's fingerprint"}, + "member": map[string]any{"type": "string", "description": "the member, ., as indexer_entries names it"}, + "why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + str := func(k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + fp, member, why := str("fingerprint"), str("member"), str("why") + if fp == "" || member == "" || why == "" { + return nil, errors.New("name the fingerprint (fingerprint) and the member (member) and say why (why): a drop is a hand act, recorded through mesh-hand-acts") + } + data, err := json.Marshal(TakeInAsk{V: 1, NamedFingerprint: fp, NamedMember: member, Drop: true, Why: why, By: stdio.CallerOf(args), At: now().UTC()}) + if err != nil { + return nil, err + } + if err := record("dropped "+member+" from the names of the entry "+fp+" by hand", why); err != nil { + return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was dropped: %w", err) + } + a, err := ask("named."+NamedKey(fp)+"."+keys.EscapeName(member), data) + if err != nil { + return nil, err + } + return a.Result, nil + }}} +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 36a8b923..e9d2f70d 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -910,7 +910,7 @@ func (ix *Indexer) dry() *dryRun { Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil, OwnBuildState: nil, underSeats: true, // A dry run keeps no catalog graph revision (catalog_revisions.go): its revision is the take-in's to keep. - Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, + Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, NamedSets: nil, // A dry run writes no registration and asks no one. Registry: nil, SeedModules: nil} x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) } diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index fc7481fe..eb60286c 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -1131,7 +1131,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged", "JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk", "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", - "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules"} { + "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/revision_tools.go b/modules/mesh-delivery-indexer/internal/index/revision_tools.go index afe1860f..d746eed0 100644 --- a/modules/mesh-delivery-indexer/internal/index/revision_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/revision_tools.go @@ -132,9 +132,15 @@ func ListRevisions(ctx context.Context, o Objects, candidates bool) (RevisionsAn return out, nil } -// removeUnlisted deletes every entry no kept revision lists any more: an entry is stored once and shared, so it goes -// only with the last revision that lists it. It answers the entries removed. -func removeUnlisted(ctx context.Context, o Objects) ([]string, error) { +// removeUnlisted deletes every entry no kept revision lists any more and no want or running build names (design 47 G §675): an +// entry is stored once and shared, so it goes only with the last revision that lists it, and its entry-index key with it. It +// removes none while the sets of the names are not ready (a restart reads named whole before it removes anything), and keeps +// an entry whose set it cannot read. It answers the entries removed. The caller holds keepMu. +func (ix *Indexer) removeUnlisted(ctx context.Context) ([]string, error) { + if ix.NamedSets != nil && !ix.NamedSets.Ready() { + return nil, nil + } + o := ix.Revisions list, err := ListRevisions(ctx, o, false) if err != nil { return nil, err @@ -145,14 +151,74 @@ func removeUnlisted(ctx context.Context, o Objects) ([]string, error) { } var removed []string for _, e := range list.Unlisted { + fp, digest := ix.entryFingerprint(ctx, e) + if ix.NamedSets != nil && fp != "" { + if named, err := ix.NamedSets.IsNamed(ctx, fp); named { + if err != nil { + ix.logf("catalog graph: %s is kept, the state %s cannot be read: %v", e, NamedState, err) + } + continue + } + } if err := o.Delete(ctx, e); err != nil { return removed, fmt.Errorf("%s: %w", e, err) } removed = append(removed, e) + ix.forgetEntryIndex(ctx, fp, digest) } return removed, nil } +// entryFingerprint is the build-input fingerprint of an entry object and its digest; none when the object does not read. +func (ix *Indexer) entryFingerprint(ctx context.Context, object string) (fp, digest string) { + digest = strings.TrimPrefix(object, "entry.") + data, err := ix.Revisions.Get(ctx, object) + if err != nil { + return "", digest + } + var e meshcatalog.CatalogEntry + if json.Unmarshal(data, &e) != nil { + return "", digest + } + return e.Fp, digest +} + +// forgetEntryIndex deletes the entry-index key of a fingerprint when it still holds the digest of the object just removed (§675). +func (ix *Indexer) forgetEntryIndex(ctx context.Context, fp, digest string) { + if ix.EntryIndex == nil || fp == "" { + return + } + err := write(ctx, ix.EntryIndex, EntryIndexKey(fp), func(old []byte, found bool) ([]byte, bool, error) { + var v EntryIndexValue + if !found || json.Unmarshal(old, &v) != nil || v.Digest != digest { + return nil, false, nil + } + return nil, true, nil + }) + if err != nil { + ix.logf("catalog graph: the key of %s in %s is not removed: %v", fp, EntryIndexState, err) + } +} + +// RemoveUnlisted removes the entries nothing names now (§675): called when a set of the names lost a member. +func (ix *Indexer) RemoveUnlisted(ctx context.Context) { + if ix.Revisions == nil { + return + } + ix.keepMu.Lock() + removed, err := ix.removeUnlisted(ctx) + ix.keepMu.Unlock() + if err != nil { + ix.logf("catalog graph: the entries nothing names are not removed: %v", err) + } + if len(removed) > 0 { + ix.logf("catalog graph: %d entries removed, no revision lists them and no want or running build names them", len(removed)) + if _, err := ix.RoomAgain(ctx); err != nil { + ix.logf("catalog graph: room is not evaluated: %v", err) + } + } +} + // DropCandidateGraph drops a candidate's catalog half by hand (indexer_drop_candidate_graph, design 47: for a candidate // whose gate event never came), the entries no other revision lists with it, and says room when that takes the store // below its filling mark, which takes in a merge held for objects full (RoomAgain). @@ -171,7 +237,7 @@ func (ix *Indexer) DropCandidateGraph(ctx context.Context, group, heads, by, why err := ix.Revisions.Delete(ctx, name) var removed []string if err == nil { - removed, err = removeUnlisted(ctx, ix.Revisions) + removed, err = ix.removeUnlisted(ctx) } ix.keepMu.Unlock() if err != nil { diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 89ca3313..a9c02c1a 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -70,6 +70,10 @@ type TakeInAsk struct { CandidateGroup string `json:"candidate-group,omitempty"` CandidateHeads string `json:"candidate-heads,omitempty"` Drop bool `json:"drop,omitempty"` + // NamedFingerprint and NamedMember, with Drop, ask one member to be removed from that fingerprint's set of the state named + // (indexer_named_drop, named.go, design 47 G §675). + NamedFingerprint string `json:"named-fingerprint,omitempty"` + NamedMember string `json:"named-member,omitempty"` // Registry asks one of the registrations' acts (registrations.go): a trusted change of one, the verb registrations, // indexer_proposal_drop or indexer_seed_again (design 47 G §3, §4, §8, §667). Registry *RegistryAct `json:"registry,omitempty"` @@ -91,7 +95,7 @@ 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 == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.Registry == nil { + if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.NamedFingerprint == "" && a.Registry == nil { return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try") } return a, nil diff --git a/modules/mesh-delivery-indexer/internal/index/wanted_names_test.go b/modules/mesh-delivery-indexer/internal/index/wanted_names_test.go new file mode 100644 index 00000000..d74e2929 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/wanted_names_test.go @@ -0,0 +1,199 @@ +package index + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// Design 47 G §675 (Kept while named), §693 (a removed want): the state named holds, per fingerprint, the set of +// `.` whose want or running build names it. + +var namedBase = time.Date(2026, 11, 11, 20, 0, 0, 0, time.UTC) + +func namedFp(c string) string { return "src2:" + strings.Repeat(c, 64) } + +type namedEvent struct { + wanted bool + node string + module string + fp string + min int + stopped bool +} + +func (e namedEvent) apply(t *testing.T, n *Named) { + t.Helper() + at := namedBase.Add(time.Duration(e.min) * time.Minute) + var data []byte + var err error + if e.wanted { + data, err = json.Marshal(meshwants.WantedEvent{V: 1, Node: e.node, Module: e.module, Fp: e.fp, Revision: 1, At: &at}) + if err == nil { + err = n.HearWanted(context.Background(), data) + } + } else { + data, err = json.Marshal(map[string]any{"v": 1, "node": e.node, "module": e.module, "fingerprint": e.fp, "stopped": e.stopped, "at": at}) + if err == nil { + err = n.HearRunning(context.Background(), data) + } + } + if err != nil { + t.Fatalf("%+v: %v", e, err) + } +} + +func TestNamedHoldsTheMembersWhoseWantOrRunningBuildNamesAFingerprint(t *testing.T) { + a, b := namedFp("a"), namedFp("b") + for _, c := range []struct { + name string + events []namedEvent + want map[string][]string + }{ + {"a want and a running build of one member name one fingerprint once", []namedEvent{ + {wanted: true, node: "n1", module: "m", fp: a, min: 1}, {node: "n1", module: "m", fp: a, min: 2}}, map[string][]string{a: {"n1.m"}, b: nil}}, + {"two nodes of a module are two members", []namedEvent{ + {wanted: true, node: "n1", module: "m", fp: a, min: 1}, {wanted: true, node: "n2", module: "m", fp: a, min: 1}}, map[string][]string{a: {"n1.m", "n2.m"}}}, + {"a want that moves on leaves the old fingerprint's set", []namedEvent{ + {wanted: true, node: "n1", module: "m", fp: a, min: 1}, {wanted: true, node: "n1", module: "m", fp: b, min: 2}}, map[string][]string{a: nil, b: {"n1.m"}}}, + {"a running build still naming the old one keeps it there", []namedEvent{ + {wanted: true, node: "n1", module: "m", fp: a, min: 1}, {node: "n1", module: "m", fp: a, min: 1}, + {wanted: true, node: "n1", module: "m", fp: b, min: 2}}, map[string][]string{a: {"n1.m"}, b: {"n1.m"}}}, + {"a stopped running names nothing", []namedEvent{ + {node: "n1", module: "m", fp: a, min: 1}, {node: "n1", module: "m", stopped: true, min: 2}}, map[string][]string{a: nil}}, + {"an event older than the one heard changes nothing", []namedEvent{ + {wanted: true, node: "n1", module: "m", fp: b, min: 5}, {wanted: true, node: "n1", module: "m", fp: a, min: 4}}, map[string][]string{a: nil, b: {"n1.m"}}}, + } { + t.Run(c.name, func(t *testing.T) { + st := newMem() + n := NewNamed(st, nil) + for _, e := range c.events { + e.apply(t, n) + } + for fp, want := range c.want { + got, err := n.Members(context.Background(), fp) + if err != nil || !slices.Equal(got, want) { + t.Errorf("the set of %.9s = %v, %v; want %v", fp, got, err, want) + } + } + }) + } +} + +// §675: an entry no kept revision lists is removed only when its set is empty, and not before the start's read of both streams +// ended; its entry-index key goes with it. +func TestAnEntryNoRevisionListsStaysWhileAWantNamesItAndGoesAfterTheWantMovesOn(t *testing.T) { + ctx := context.Background() + fp := namedFp("c") + n := NewNamed(newMem(), nil) + ix := &Indexer{Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), NamedSets: n, Now: func() time.Time { return namedBase }} + n.Changed = ix.RemoveUnlisted + if err := ix.keepEntries(ctx, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c", Fp: fp, Manifest: map[string]any{}}}); err != nil { + t.Fatal(err) + } + if err := ix.Revisions.Delete(ctx, "trunk.1"); err != nil { // the revision that listed it is removed + t.Fatal(err) + } + found := func() bool { + _, err := Entry(ctx, ix.Revisions, ix.EntryIndex, meshcatalog.EntryInput{Fingerprint: fp}) + return err == nil + } + + ix.RemoveUnlisted(ctx) + if !found() { + t.Fatal("an entry was removed before the start's read of the wants and the running builds ended") + } + namedEvent{wanted: true, node: "n1", module: "avahi", fp: fp, min: 1}.apply(t, n) + if err := n.ReadDone(ctx, NamedWanted); err != nil { + t.Fatal(err) + } + if !found() { + t.Fatal("an entry was removed with one stream not read") + } + if err := n.ReadDone(ctx, NamedRunning); err != nil { + t.Fatal(err) + } + if !found() { + t.Fatal("an entry a want names was removed") + } + namedEvent{wanted: true, node: "n1", module: "avahi", fp: namedFp("d"), min: 2}.apply(t, n) + if found() { + t.Error("the entry stays after the want moved on") + } + if keys, _ := ix.EntryIndex.Keys(ctx); len(keys) != 0 { + t.Errorf("entry-index still holds %v for the removed entry", keys) + } +} + +// §675: a restart reads named whole before it removes anything, and a member whose build moved while the indexer was away +// leaves its old set when both streams are read. +func TestARestartReconcilesTheSetsWithWhatTheStreamsHoldNow(t *testing.T) { + ctx := context.Background() + st := newMem() + old, now := namedFp("e"), namedFp("f") + before := NewNamed(st, nil) + namedEvent{wanted: true, node: "n1", module: "m", fp: old, min: 1}.apply(t, before) + namedEvent{wanted: true, node: "n2", module: "m", fp: old, min: 1}.apply(t, before) + + after := NewNamed(st, nil) // the next start: the state kept, nothing in memory + namedEvent{wanted: true, node: "n1", module: "m", fp: now, min: 2}.apply(t, after) + if got, _ := after.Members(ctx, old); !slices.Equal(got, []string{"n1.m", "n2.m"}) { + t.Fatalf("a restart removed from %v before it read both streams", got) + } + namedEvent{wanted: true, node: "n2", module: "m", fp: old, min: 1}.apply(t, after) + _ = after.ReadDone(ctx, NamedWanted) + _ = after.ReadDone(ctx, NamedRunning) + if got, _ := after.Members(ctx, old); !slices.Equal(got, []string{"n2.m"}) { + t.Errorf("the old set holds %v after the read, want only the member still naming it", got) + } + if !after.Ready() { + t.Error("the sets are not ready after both streams were read") + } +} + +// §675: indexer_named_drop removes one member by hand, says the hand act first, and indexer_entries reads the set. +func TestNamedDropRemovesAMemberByHandAndEntriesReadsTheSet(t *testing.T) { + ctx := context.Background() + fp := namedFp("9") + st := newMem() + n := NewNamed(st, nil) + namedEvent{wanted: true, node: "n1", module: "m", fp: fp, min: 1}.apply(t, n) + namedEvent{wanted: true, node: "n2", module: "m", fp: fp, min: 1}.apply(t, n) + ix := &Indexer{NamedSets: n, Logf: func(string, ...any) {}} + var acts []string + asked := func(key string, value []byte) (TakeInAsk, error) { + a, err := DecodeTakeIn(value) + if err != nil { + return a, err + } + r := ix.NamedDrop(ctx, a.NamedFingerprint, a.NamedMember, a.By, a.Why) + a.Result = &r + return a, nil + } + tool := NamedTools(asked, func() time.Time { return namedBase }, func(what, why string) error { acts = append(acts, what+" | "+why); return nil })[0] + if _, err := tool.Run(map[string]any{"fingerprint": fp, "member": "n1.m"}); err == nil { + t.Error("a drop with no why was asked") + } + if _, err := tool.Run(map[string]any{"fingerprint": fp, "member": "n1.m", "why": "its node is gone"}); err != nil { + t.Fatal(err) + } + if len(acts) != 1 || !strings.Contains(acts[0], "n1.m") { + t.Errorf("the hand acts = %v, want the drop of n1.m said once", acts) + } + if got, _ := n.Members(ctx, fp); !slices.Equal(got, []string{"n2.m"}) { + t.Errorf("the set holds %v, want n2.m alone", got) + } + read, err := EntriesTool(newMemObjects(1<<10), newMem(), st).Run(map[string]any{"fingerprint": fp}) + if err != nil || fmt.Sprint(read.(EntriesAnswer).Named) != "[n2.m]" { + t.Errorf("indexer_entries = %+v, %v, want the named set [n2.m]", read, err) + } + var _ stdio.Tool = tool +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index ce0067d4..30bbf478 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -123,7 +123,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {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: "try-recovery", History: 1}, {Name: "own-build", History: 1}, {Name: index.PackagesState, History: index.PackagesHistory}, {Name: views.MarksState, History: 1}, {Name: index.FindingsState, History: 5}, {Name: views.StatusState, History: 1}, {Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5}, - {Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}, {Name: index.EntryIndexState, History: 1, MaxMiB: 8}, // design 47 G §695 + {Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}, {Name: index.EntryIndexState, History: 1, MaxMiB: 8}, {Name: index.NamedState, History: 1, MaxMiB: 4}, // design 47 G §695 {Name: index.RegistrationsState, History: 1}, {Name: index.ProposalsState, History: 1, MaxMiB: 4}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17 t.Errorf("state = %+v, want %+v", m.State, want) } @@ -160,7 +160,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // The registry seats' published fill each (§413: no consume of version.numbered) package's series (DLV-I5, design 51 *The versions*). if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", "mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", - "mesh-delivery.source-moved", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + "mesh-delivery.source-moved", "mesh-wants.wanted", "node-engine.running", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { @@ -194,9 +194,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TryCandidatesTool, index.DropTryCandidateTool, index.RecoverTriesTool, index.TakeInMergeTool, index.FingerprintsAgainTool, // issue 1966; 1585/T12, T13 (ADR 0633 §5) // milestone 47/G W4.12 (design 47, the indexer's tools): the catalog graph revisions read and steered. index.CatalogRevisionTool, index.CatalogRevisionsTool, index.CandidateGraphsTool, index.RecomputeTool, index.DropCandidateGraphTool, - index.EntryIndexTool, // design 47 G §675 + index.EntryIndexTool, index.NamedDropTool, // design 47 G §675 index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool) // design 47 G §667, §8 - if !reflect.DeepEqual(m.Tools, want) || len(want) != 45 || slices.Contains(m.Tools, "modules") { + if !reflect.DeepEqual(m.Tools, want) || len(want) != 46 || slices.Contains(m.Tools, "modules") { t.Errorf("tools = %v, want %v", m.Tools, want) } // Design 47 G §2, §9: every version of every registration is kept with no age limit, 64 MiB, on the old streams section. diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 450b07a0..26596f47 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -66,6 +66,7 @@ "indexer_recompute", "indexer_drop_candidate_graph", "indexer_entries", + "indexer_named_drop", "indexer_proposals", "indexer_proposal_drop", "indexer_seed_again" @@ -82,6 +83,8 @@ "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", "mesh-delivery.source-moved", + "mesh-wants.wanted", + "node-engine.running", "operator-channel.decided", "npm-package-registry.published" ], @@ -106,6 +109,7 @@ { "name": "pending", "history": 1 }, { "name": "unjudged", "history": 1 }, { "name": "entry-index", "history": 1, "max-mib": 8 }, + { "name": "named", "history": 1, "max-mib": 4 }, { "name": "registrations", "history": 1 }, { "name": "proposals", "history": 1, "max-mib": 4 } ],