novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

indexer: kept while named, the state named of the wants and running builds that name a fingerprint

Browse Source 
The indexer consumes the seat mesh-wants' wanted (on wanted.<node>.<module>, kept last per subject: the seat's definition gains
the two tokens and the keep) and the node-engine seat's running, each read whole at its start and followed; it keeps in the
old state section named (max-mib 4) one key per fingerprint with its set of <node>.<module>, written by compare-and-set. An
entry object is removed only when no kept revision lists it and its set is empty, and not before both streams were read and the
sets reconciled; its entry-index key goes with it. indexer_entries reads the set, indexer_named_drop removes one member by
hand (class act, said in the hand-act log first).

Builds: design 47 G §675, §695, A-61
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-11 22:59:59 +01:00
parent 0f62358fb0
commit 86bae053b6
23 changed files with 842 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contracts/go/seats/meshwants/contract.go
+10 -2
View File
@@ -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.<node>.<module>.
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"
contracts/go/seats/meshwants/contract_test.go
+4 -4
View File
@@ -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\"}",
contracts/seats/mesh-wants/fixtures/events.json
+2 -2
View File
@@ -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",
contracts/seats/mesh-wants/seat.json
+5
View File
@@ -303,6 +303,11 @@
},
{
"name": "wanted",
"subject": "wanted.<node>.<module>",
"keep": {
"max-age-days": 0,
"last-per-subject": true
},
"body": {
"type": "object",
"description": "an entry written as a want, from a trunk revision only",
contracts/ts/seats/meshwants/index.d.ts
Vendored
+5 -1
View File
@@ -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";
contracts/ts/seats/meshwants/index.js
+7 -1
View File
@@ -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";
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go
+5 -1
View File
@@ -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
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go
+52 -1
View File
@@ -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.<node>.<module> 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)
}
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+5 -1
View File
@@ -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.
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_events.go
+55
View File
@@ -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.<node>.<module>`
// and the node-engine seat's `running.<node>.<module>`, 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
})
}
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go
+4
View File
@@ -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 {
modules/mesh-delivery-indexer/go.mod
+2
View File
@@ -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
modules/mesh-delivery-indexer/internal/index/candidates.go
+3
View File
@@ -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
modules/mesh-delivery-indexer/internal/index/catalog_revisions.go
+3
View File
@@ -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})
}
modules/mesh-delivery-indexer/internal/index/entry_verb.go
+13 -6
View File
@@ -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 `<node>.<module>` 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 `<node>.<module>` 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
}}
}
modules/mesh-delivery-indexer/internal/index/named.go
+383
View File
@@ -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 `<node>.<module>` 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.<node>.<module>: 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 (`<node>.<module>`) 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, <node>.<module>, 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
}}}
}
modules/mesh-delivery-indexer/internal/index/pending.go
+1 -1
View File
@@ -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...) }
modules/mesh-delivery-indexer/internal/index/pending_test.go
+1 -1
View File
@@ -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{})
modules/mesh-delivery-indexer/internal/index/revision_tools.go
+70 -4
View File
@@ -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 {
modules/mesh-delivery-indexer/internal/index/takein.go
+5 -1
View File
@@ -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
modules/mesh-delivery-indexer/internal/index/wanted_names_test.go
+199
View File
@@ -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
// `<node>.<module>` 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
}
modules/mesh-delivery-indexer/manifest_test.go
+4 -4
View File
@@ -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.
modules/mesh-delivery-indexer/module.json
+4
View File
@@ -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 }
],