From d9066aa1c6038d0c42267fdd2ebf416a19530232 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 23:07:45 +0100 Subject: [PATCH] indexer: the release's edge entries, outdated for kinds 1, 2 and 4, toolchain-agreed and version-delivered kept as heard, the verb input and its state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kind 3, kinds 5 and 6, the first fill's edge entries (§377) and the deletion of an input at its ask's step-ended are not built. Builds: design 47 G §374, §375, §354, A-68 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-indexer-tools/main.go | 4 +- .../cmd/mesh-delivery-indexer/main.go | 8 + .../cmd/mesh-delivery-indexer/versions.go | 35 ++++ .../internal/index/candidates.go | 2 + .../internal/index/pending.go | 2 +- .../internal/index/pending_test.go | 2 +- .../internal/index/run_inputs.go | 181 ++++++++++++++++ .../internal/index/versions.go | 93 +++++++-- .../internal/index/versions_edges.go | 197 ++++++++++++++++++ .../internal/index/versions_test.go | 125 +++++++++++ .../mesh-delivery-indexer/manifest_test.go | 12 +- modules/mesh-delivery-indexer/module.json | 12 +- 12 files changed, 640 insertions(+), 33 deletions(-) create mode 100644 modules/mesh-delivery-indexer/internal/index/run_inputs.go create mode 100644 modules/mesh-delivery-indexer/internal/index/versions_edges.go 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 b4bf550f..cb249524 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,7 +288,9 @@ 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)}), index.VersionsTool(keptRead{stdio.State(index.VersionsState)})) + tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.VersionsTool(keptRead{stdio.State(index.VersionsState)}), + index.OutdatedTool(&index.Indexer{Versions: keptRead{stdio.State(index.VersionsState)}, OwnBuildState: keptRead{stdio.State(index.OwnBuildStateName)}}), + index.InputTool(keptRead{stdio.State(index.InputsState)}), index.InputsReadTool(keptRead{stdio.State(index.InputsState)})) 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 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 f5827821..7d261d67 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -24,6 +24,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -445,6 +446,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro SayFill: sayFill(conn, cred), // The catalog's released versions (design 47 G §349): the state versions, bound when it appears, and the event version-released. Versions: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.VersionsBucket, "no released version is kept, so no package-published makes a version", logger.Printf)}, + Inputs: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.InputsBucket, "no input of a run ask is kept, so the verb input answers none", logger.Printf)}, SayVersionReleased: sayVersionReleased(conn, cred), SayVersionsFilled: sayVersionsFilled(conn, cred), // The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear; // until then every merge is taken in at once, as before. @@ -679,6 +681,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hub.Handle(meshbuilds.SubjectPackagePublished, func(m jetstream.Msg) { heardPackagePublished(ctx, ix, m, logger.Printf) }); err != nil { return err } + if err := hub.Handle(meshbuilds.SubjectToolchainAgreed, func(m jetstream.Msg) { heardToolchainAgreed(ctx, ix, m, logger.Printf) }); err != nil { + return err + } + if err := hub.Handle(meshwants.SubjectVersionDelivered, func(m jetstream.Msg) { heardVersionDelivered(ctx, ix, m, logger.Printf) }); err != nil { + return err + } if err := hub.Handle(decidedSubject(cred.Module), func(m jetstream.Msg) { heardDecided(ctx, registry, m, logger.Printf) }); err != nil { return err } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go index f8507c48..ef0bca4d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go @@ -12,6 +12,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-indexer/internal/index" @@ -59,6 +60,40 @@ func heardPackagePublished(ctx context.Context, ix *index.Indexer, m jetstream.M _ = m.Ack() } +// heardToolchainAgreed keeps the newest toolchain-agreed of its language (A-68); one that does not decode is said and +// acknowledged, one that cannot be kept is left, so the bus gives it again. +func heardToolchainAgreed(ctx context.Context, ix *index.Indexer, m jetstream.Msg, logf func(string, ...any)) { + var t meshbuilds.ToolchainAgreed + if err := json.Unmarshal(m.Data(), &t); err != nil { + logf("%s holds what is no toolchain-agreed, left: %v", m.Subject(), err) + _ = m.Ack() + return + } + if err := ix.HearToolchainAgreed(ctx, t); err != nil { + logf("versions: the toolchain-agreed of %s could not be kept, so the bus gives it again: %v", t.Language, err) + _ = m.Nak() + return + } + _ = m.Ack() +} + +// heardVersionDelivered writes the delivered release's entries of kinds 2, 4 and 6 and keeps the newest delivered version of +// its module (design 47 G §374, A-68), as heardToolchainAgreed keeps its event. +func heardVersionDelivered(ctx context.Context, ix *index.Indexer, m jetstream.Msg, logf func(string, ...any)) { + var d meshwants.VersionDeliveredEvent + if err := json.Unmarshal(m.Data(), &d); err != nil { + logf("%s holds what is no version-delivered, left: %v", m.Subject(), err) + _ = m.Ack() + return + } + if err := ix.HearVersionDelivered(ctx, d); err != nil { + logf("versions: the version-delivered of %s %s could not be kept, so the bus gives it again: %v", d.Module, d.Version, err) + _ = m.Nak() + return + } + _ = m.Ack() +} + // sayVersionsFilled says versions-filled (design 47 G §372) once, on the seat's stream that keeps it with no age limit. func sayVersionsFilled(conn *nats.Conn, cred process.Account) func(context.Context, meshcatalog.VersionsFilledEvent, string) error { return func(_ context.Context, e meshcatalog.VersionsFilledEvent, id string) error { diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index a683c83b..9a19d11b 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -113,6 +113,8 @@ type Indexer struct { SayVersionReleased func(ctx context.Context, e meshcatalog.VersionReleasedEvent, id string) error // SayVersionsFilled says the seat mesh-catalog's event versions-filled, once, under the duplicate id given (§372); nil says nothing. SayVersionsFilled func(ctx context.Context, e meshcatalog.VersionsFilledEvent, id string) error + // Inputs is the state inputs (run_inputs.go, design 47 G §354): the inputs of the indexer's run asks; nil keeps none. + Inputs Store // Registry is the registrations and their trusted changes (registrations.go, design 47 G §2 to §4, §667); SeedModules reads // what the first registrations are composed from (§8). Nil: no registration act is answered. Registry *Registry diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index a18efefe..49ef4603 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, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil, + Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil, Inputs: 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 e51ba1dc..73da47fe 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", "Versions", "SayVersionReleased", "SayVersionsFilled"} { + "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/run_inputs.go b/modules/mesh-delivery-indexer/internal/index/run_inputs.go new file mode 100644 index 00000000..647737ae --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/run_inputs.go @@ -0,0 +1,181 @@ +package index + +import ( + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "slices" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// The indexer's inputs (design 47 G §354): the input of the ask `run` the indexer's validation puts, kept in the old state +// section inputs, one key per input named by its write time and its sha256, read by the seat mesh-catalog's verb input (whose +// callers are the holders of the build seats) and by the tool indexer_inputs. A key older than a day is deleted at each new +// ask, judged over at most the 20 oldest keys. +const ( + InputsState = "inputs" + InputsBucket = Module + "_" + InputsState + InputsTool = "indexer_inputs" + // InputsKeptFor is how long an input is kept at the longest. + InputsKeptFor = 24 * time.Hour + // InputsSwept is how many of the oldest keys each new ask judges. + InputsSwept = 20 + inputsTime = "20060102T150405Z" +) + +// InputDigest is the sha256 of an input's exact bytes, as "sha256:" and its hex digits. +func InputDigest(body []byte) string { + sum := sha256.Sum256(body) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// KeepInput keeps the exact bytes of a run ask's input under their write time and sha256, after sweeping the keys more than a +// day old among the 20 oldest; it answers the digest. The same bytes written again in the same second are one key. +func (ix *Indexer) KeepInput(ctx context.Context, body []byte) (string, error) { + if ix.Inputs == nil { + return "", errors.New("the state inputs is not kept here") + } + if err := ix.sweepInputs(ctx); err != nil { + ix.logf("inputs: the keys older than a day are not all deleted: %v", err) + } + digest := InputDigest(body) + key := meshcatalog.InputsByTimeDigestKey(ix.now().UTC().Format(inputsTime), digest) + data, err := json.Marshal(base64.StdEncoding.EncodeToString(body)) + if err != nil { + return "", err + } + if err := ix.Inputs.Create(ctx, key, data); err != nil && !errors.Is(err, ErrConflict) { + return "", err + } + return digest, nil +} + +// sweepInputs deletes, among the 20 oldest keys, those written more than a day ago. +func (ix *Indexer) sweepInputs(ctx context.Context) error { + all, err := ix.Inputs.Keys(ctx) + if err != nil { + return err + } + slices.Sort(all) // the write time leads the key, so the oldest come first + if len(all) > InputsSwept { + all = all[:InputsSwept] + } + var errs []error + for _, k := range all { + t, _, ok := meshcatalog.ParseInputsByTimeDigestKey(k) + at, perr := time.Parse(inputsTime, t) + if !ok || perr != nil || ix.now().Sub(at) < InputsKeptFor { + continue + } + if _, rev, found, err := ix.Inputs.Get(ctx, k); err == nil && found { + if err := ix.Inputs.Delete(ctx, k, rev); err != nil { + errs = append(errs, err) + } + } + } + return errors.Join(errs...) +} + +// InputKey is the key of the input of the digest, or none. +func InputKey(ctx context.Context, st Store, digest string) (string, error) { + all, err := st.Keys(ctx) + if err != nil { + return "", err + } + for _, k := range all { + if _, d, ok := meshcatalog.ParseInputsByTimeDigestKey(k); ok && d == digest { + return k, nil + } + } + return "", nil +} + +// Input answers the verb input: the exact bytes (base64) of the input of the digest, its digest checked. +func Input(ctx context.Context, st Store, in meshcatalog.InputInput) (meshcatalog.InputAnswer, error) { + var out meshcatalog.InputAnswer + if in.Digest == "" { + return out, errors.New("name the input's sha256 (digest)") + } + key, err := InputKey(ctx, st, in.Digest) + if err != nil { + return out, fmt.Errorf("the state %s cannot be read: %w", InputsState, err) + } + raw, _, found, err := st.Get(ctx, key) + if key == "" || err != nil || !found { + return out, fmt.Errorf("no input of digest %s is kept", in.Digest) + } + text, err := meshcatalog.DecodeInputsByTimeDigestValue(raw) + if err != nil { + return out, fmt.Errorf("the input %s cannot be read: %w", key, err) + } + body, err := base64.StdEncoding.DecodeString(text) + if err != nil || InputDigest(body) != in.Digest { + return out, fmt.Errorf("the input kept under %s does not match its digest", key) + } + return meshcatalog.InputAnswer{Digest: in.Digest, Input: text}, nil +} + +// InputTool is the verb input as the tool mesh-catalog.input of the indexer's own runtime. +func InputTool(st Store) stdio.Tool { + return stdio.Tool{Name: meshcatalog.Seat + "." + meshcatalog.VerbInput, Run: func(args map[string]any) (any, error) { + var in meshcatalog.InputInput + if raw, err := json.Marshal(args); err == nil { + _ = json.Unmarshal(raw, &in) + } + return Input(context.Background(), st, in) + }} +} + +// InputsAnswer is what indexer_inputs answers: each key kept with its time, digest and size, or one input's bytes. +type InputsAnswer struct { + Inputs []InputKept `json:"inputs"` +} + +// InputKept is one input kept. +type InputKept struct { + Key string `json:"key"` + Digest string `json:"digest"` + Bytes int `json:"bytes"` + Input string `json:"input,omitempty"` +} + +// InputsReadTool is indexer_inputs (class read): every input the indexer keeps, with its key, digest and size; with a digest, +// that input whole (base64). +func InputsReadTool(st Store) stdio.Tool { + return stdio.Tool{Name: InputsTool, Description: "Read the indexer's inputs of the ask run (design 47 G section 354): each key (write time and sha256) with its size; with a digest, that input's exact bytes, base64.", + Input: map[string]any{"type": "object", "properties": map[string]any{"digest": map[string]any{"type": "string", "description": "one input's sha256; none lists every key"}}}, + Run: func(args map[string]any) (any, error) { + ctx := context.Background() + want, _ := args["digest"].(string) + all, err := st.Keys(ctx) + if err != nil { + return nil, err + } + slices.Sort(all) + out := InputsAnswer{Inputs: []InputKept{}} + for _, k := range all { + _, d, ok := meshcatalog.ParseInputsByTimeDigestKey(k) + if !ok || (want != "" && d != want) { + continue + } + raw, _, found, err := st.Get(ctx, k) + if err != nil || !found { + continue + } + text, _ := meshcatalog.DecodeInputsByTimeDigestValue(raw) + kept := InputKept{Key: k, Digest: d, Bytes: base64.StdEncoding.DecodedLen(len(text))} + if want != "" { + kept.Input = text + } + out.Inputs = append(out.Inputs, kept) + } + return out, nil + }} +} diff --git a/modules/mesh-delivery-indexer/internal/index/versions.go b/modules/mesh-delivery-indexer/internal/index/versions.go index 8163a690..154013e4 100644 --- a/modules/mesh-delivery-indexer/internal/index/versions.go +++ b/modules/mesh-delivery-indexer/internal/index/versions.go @@ -1,6 +1,7 @@ package index import ( + "cmp" "context" "encoding/json" "errors" @@ -32,8 +33,9 @@ const ( // published is what a not yet released version holds: each artifact published at it, by reference. type published struct { - Artifacts map[string]meshcatalog.VersionArtifact `json:"artifacts"` - Source int64 `json:"source,omitempty"` + Artifacts map[string]meshcatalog.VersionArtifact `json:"artifacts"` + Dependencies []meshbuilds.Dependency `json:"dependencies,omitempty"` + Source int64 `json:"source,omitempty"` } // versionParts is MAJOR.MINOR.PATCH, read; false for any other form. @@ -113,10 +115,14 @@ func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.Packag if _, _, found, err := ix.Versions.Get(ctx, meshcatalog.VersionsByModuleVersionKey(p.Module, p.Version)); err != nil || found { return err // released already: heard again } - want, err := ix.artifactsOf(ctx, p.Module) + entry, err := ix.entryOf(ctx, p.Module) if err != nil { return err } + var want []string + for _, a := range entry.Artifacts { + want = append(want, a.Ref) + } key := PendingPrefix + meshcatalog.VersionsByModuleVersionKey(p.Module, p.Version) var have published err = write(ctx, ix.OwnBuildState, key, func(raw []byte, found bool) ([]byte, bool, error) { @@ -131,6 +137,11 @@ func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.Packag } have.Artifacts[p.Artifact] = meshcatalog.VersionArtifact{Artifact: p.Artifact, Digest: p.Digest} have.Source = p.Source + for _, d := range p.Dependencies { + if !slices.ContainsFunc(have.Dependencies, func(x meshbuilds.Dependency) bool { return x.Kind == d.Kind && x.Part == d.Part }) { + have.Dependencies = append(have.Dependencies, d) + } + } data, err := json.Marshal(have) return data, false, err }) @@ -142,41 +153,36 @@ func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.Packag return nil } } - return ix.releaseVersion(ctx, p, key, want, have) + return ix.releaseVersion(ctx, p, key, entry, have) } -// artifactsOf are the artifact references the module's entry in the newest trunk revision names. -func (ix *Indexer) artifactsOf(ctx context.Context, module string) ([]string, error) { +// entryOf is the module's entry in the newest trunk revision. +func (ix *Indexer) entryOf(ctx context.Context, module string) (meshcatalog.CatalogEntry, error) { if ix.Revisions == nil { - return nil, errors.New("the catalog graph is not kept here: no entry names the module's artifacts") + return meshcatalog.CatalogEntry{}, errors.New("the catalog graph is not kept here: no entry names the module's artifacts") } rev, found, err := newestTrunk(ctx, ix.Revisions) if err != nil { - return nil, err + return meshcatalog.CatalogEntry{}, err } if !found { - return nil, errors.New("the indexer keeps no trunk revision yet") + return meshcatalog.CatalogEntry{}, errors.New("the indexer keeps no trunk revision yet") } _, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", rev.Revision)) if err != nil { - return nil, err + return meshcatalog.CatalogEntry{}, err } for _, e := range entries { - if e.Module != module { - continue + if e.Module == module { + return e, nil } - var refs []string - for _, a := range e.Artifacts { - refs = append(refs, a.Ref) - } - return refs, nil } - return nil, fmt.Errorf("the trunk holds no module %s", module) + return meshcatalog.CatalogEntry{}, fmt.Errorf("the trunk holds no module %s", module) } // releaseVersion records the version of p whole, says version-released unless it is late, deletes what exceeds the kept versions, // and forgets the artifacts kept as published. -func (ix *Indexer) releaseVersion(ctx context.Context, p meshbuilds.PackagePublished, pendingKey string, want []string, have published) error { +func (ix *Indexer) releaseVersion(ctx context.Context, p meshbuilds.PackagePublished, pendingKey string, entry meshcatalog.CatalogEntry, have published) error { key := meshcatalog.VersionsByModuleVersionKey(p.Module, p.Version) kept, err := ModuleVersions(ctx, ix.Versions, p.Module) if err != nil { @@ -185,9 +191,11 @@ func (ix *Indexer) releaseVersion(ctx context.Context, p meshbuilds.PackagePubli late := len(kept) > 0 && compareVersions(kept[0].Version, p.Version) > 0 at := ix.now().UTC() v := meshcatalog.VersionEntry{V: int64(keys.Version), Module: p.Module, Version: p.Version, Source: have.Source, At: &at, Late: late} - for _, ref := range slices.Sorted(slices.Values(want)) { - v.Artifacts = append(v.Artifacts, have.Artifacts[ref]) + for _, a := range entry.Artifacts { + v.Artifacts = append(v.Artifacts, have.Artifacts[a.Ref]) } + slices.SortFunc(v.Artifacts, func(a, b meshcatalog.VersionArtifact) int { return strings.Compare(a.Artifact, b.Artifact) }) + v.Edges = releaseEdges(entry, have.Dependencies) data, err := json.Marshal(v) if err != nil { return err @@ -270,3 +278,46 @@ func VersionsTool(st Store) stdio.Tool { return Versions(context.Background(), st, in) }} } + +// notYetDelivered is the version of an edge entry of kind 2, 4 or 6 until its release is first delivered (design 47 G §374). +const notYetDelivered = "not yet delivered" + +// releaseEdges are the edge entries of a release (design 47 G §374): kinds 1 and 3 from the dependencies the build +// resolved, as package-published carried them (stale: the dependent is made from them, §378); kinds 2, 4 and 6 of the entry's +// catalog graph edges, reading "not yet delivered" until the first delivery writes them. Kind 5 is not written here. +func releaseEdges(entry meshcatalog.CatalogEntry, deps []meshbuilds.Dependency) []meshcatalog.VersionEdge { + word := func(kind int64, part string) string { + for _, e := range entry.Edges { + if e.Kind == kind && (e.Part == part || e.From == part) { + return e.Word + } + } + return "" + } + var out []meshcatalog.VersionEdge + for _, d := range deps { + out = append(out, meshcatalog.VersionEdge{Kind: d.Kind, Part: d.Part, Digest: d.Digest, Version: d.Version, Stale: d.Kind == 1 || d.Kind == 3, Word: word(d.Kind, d.Part)}) + } + for _, e := range entry.Edges { + if e.Kind != 2 && e.Kind != 4 && e.Kind != 6 { + continue + } + part := cmp.Or(e.Part, e.From) + if !slices.ContainsFunc(out, func(x meshcatalog.VersionEdge) bool { return x.Kind == e.Kind && x.Part == part && x.Word == e.Word }) { + out = append(out, meshcatalog.VersionEdge{Kind: e.Kind, Part: part, Word: e.Word, Version: notYetDelivered}) + } + } + return out +} + +// OutdatedTool is the verb outdated as the tool mesh-catalog.outdated of the indexer's own runtime: it reads the versions +// state and the heard toolchain and delivered keys of the own state. +func OutdatedTool(ix *Indexer) stdio.Tool { + return stdio.Tool{Name: meshcatalog.Seat + "." + meshcatalog.VerbOutdated, Run: func(args map[string]any) (any, error) { + var in meshcatalog.OutdatedInput + if raw, err := json.Marshal(args); err == nil { + _ = json.Unmarshal(raw, &in) + } + return ix.Outdated(context.Background(), in) + }} +} diff --git a/modules/mesh-delivery-indexer/internal/index/versions_edges.go b/modules/mesh-delivery-indexer/internal/index/versions_edges.go new file mode 100644 index 00000000..5de415e7 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/versions_edges.go @@ -0,0 +1,197 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" +) + +// What outdated compares against is kept as heard, in the indexer's own state (design 47 G A-68): toolchain. the newest +// toolchain-agreed per language, delivered. the newest version-delivered per module. +const ( + ToolchainPrefix = "toolchain." + DeliveredPrefix = "delivered." +) + +// keepNewest writes body at key of the own-build state unless it holds a value whose version is higher; version of the body +// is read by newer, which says whether the heard one is not older than the kept. +func (ix *Indexer) keepNewest(ctx context.Context, key string, body any, newer func(kept []byte) bool) error { + if ix.OwnBuildState == nil { + return errors.New("the own state is not kept here") + } + data, err := json.Marshal(body) + if err != nil { + return err + } + return write(ctx, ix.OwnBuildState, key, func(kept []byte, found bool) ([]byte, bool, error) { + if found && !newer(kept) { + return nil, false, nil + } + return data, false, nil + }) +} + +// HearToolchainAgreed keeps the newest toolchain-agreed of a language (A-68, §375): the one heard last. +func (ix *Indexer) HearToolchainAgreed(ctx context.Context, t meshbuilds.ToolchainAgreed) error { + if t.Language == "" { + return errors.New("a toolchain-agreed names no language") + } + return ix.keepNewest(ctx, ToolchainPrefix+t.Language, t, func([]byte) bool { return true }) +} + +// HearVersionDelivered takes one version-delivered (design 47 G §374, §376, A-68): the entries of kinds 2, 4 and 6 of the +// delivered release, which read "not yet delivered", are written once with each dependency's newest delivered version and +// part digest; then the delivered version is kept as the module's newest. +func (ix *Indexer) HearVersionDelivered(ctx context.Context, d meshwants.VersionDeliveredEvent) error { + if ix.Versions == nil || ix.OwnBuildState == nil { + return errors.New("the state versions is not kept here") + } + if _, ok := versionParts(d.Version); !ok { + return fmt.Errorf("the version-delivered of %s names the version %q, which is no MAJOR.MINOR.PATCH", d.Module, d.Version) + } + key := meshcatalog.VersionsByModuleVersionKey(d.Module, d.Version) + err := write(ctx, ix.Versions, key, func(raw []byte, found bool) ([]byte, bool, error) { + if !found { + return nil, false, nil // a version the catalog does not hold (a build of another branch) has no entries to write + } + v, err := meshcatalog.DecodeVersionsByModuleVersionValue(raw) + if err != nil { + return nil, false, err + } + changed := false + for i, e := range v.Edges { + if e.Version != notYetDelivered { + continue + } + dep, ok := ix.deliveredOf(ctx, strings.SplitN(e.Part, ":", 2)[0]) + if !ok { + continue // the dependency is not delivered yet: the entry reads so until this release's own delivery finds it + } + v.Edges[i].Version, v.Edges[i].Digest = dep.Version, dep.Parts[e.Part] + changed = true + } + if !changed { + return nil, false, nil + } + data, err := json.Marshal(v) + return data, false, err + }) + if err != nil { + return err + } + return ix.keepNewest(ctx, DeliveredPrefix+d.Module, d, func(kept []byte) bool { + var k meshwants.VersionDeliveredEvent + return json.Unmarshal(kept, &k) != nil || compareVersions(d.Version, k.Version) >= 0 + }) +} + +// deliveredOf is the newest version-delivered kept of a module. +func (ix *Indexer) deliveredOf(ctx context.Context, module string) (meshwants.VersionDeliveredEvent, bool) { + var d meshwants.VersionDeliveredEvent + raw, _, found, err := ix.OwnBuildState.Get(ctx, DeliveredPrefix+module) + if err != nil || !found || json.Unmarshal(raw, &d) != nil { + return d, false + } + return d, true +} + +// Outdated answers the verb outdated (design 47 G §375) for the edge kinds 1, 2 and 4: every entry of each module's newest +// released version that is behind by the part's digest: kind 1, the recorded digest differs from the part's digest in the +// dependency's newest released version; kinds 2 and 4, from the dependency's newest version-delivered. A stale edge (kind 1) +// has the act rebuild, an ordering edge is shown and starts nothing. A dependency's newer version that left the digest as it +// was makes nothing behind. +func (ix *Indexer) Outdated(ctx context.Context, in meshcatalog.OutdatedInput) (meshcatalog.OutdatedAnswer, error) { + out := meshcatalog.OutdatedAnswer{Behind: []meshcatalog.Outdated{}} + if ix.Versions == nil || ix.OwnBuildState == nil { + return out, errors.New("the state versions is not kept here") + } + all, err := ix.Versions.Keys(ctx) + if err != nil { + return out, err + } + modules := map[string]bool{} + for _, k := range all { + if m, _, ok := meshcatalog.ParseVersionsByModuleVersionKey(k); ok { + modules[m] = true + } + } + newest := map[string]meshcatalog.VersionEntry{} + for m := range modules { + vs, err := ModuleVersions(ctx, ix.Versions, m) + if err != nil { + return out, err + } + for _, v := range vs { + if !v.Late { + newest[m] = v + break + } + } + } + for _, m := range sortedKeys(newest) { + v := newest[m] + for _, e := range v.Edges { + dep := strings.SplitN(e.Part, ":", 2)[0] + if in.Dependency != "" && in.Dependency != e.Part { + continue + } + var latest, digest string + switch e.Kind { + case 1: + n, ok := newest[dep] + if !ok { + continue + } + for _, a := range n.Artifacts { + if a.Artifact == e.Part { + latest, digest = n.Version, a.Digest + } + } + case 2, 4: + d, ok := ix.deliveredOf(ctx, dep) + if !ok || e.Version == notYetDelivered { + continue + } + latest, digest = d.Version, d.Parts[e.Part] + default: + continue + } + if digest == "" || digest == e.Digest { + continue + } + o := meshcatalog.Outdated{Module: m, Version: v.Version, Kind: e.Kind, Recorded: e.Version + " " + e.Digest, Newest: latest + " " + digest, + Dependency: meshcatalog.OutdatedDependency{Artifact: e.Part, Branch: ix.firstBranch(ctx, dep)}} + if e.Stale { + o.Act = "rebuild" + } + out.Behind = append(out.Behind, o) + } + } + return out, nil +} + +// firstBranch is the first watched branch of a module's registration, which its released versions are of. +func (ix *Indexer) firstBranch(ctx context.Context, module string) string { + if ix.Registry != nil { + if r, found, err := ix.Registry.Registration(ctx, module); err == nil && found && len(r.Branches) > 0 { + return r.Branches[0] + } + } + return "" +} + +func sortedKeys[V any](m map[string]V) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + slices.Sort(out) + return out +} diff --git a/modules/mesh-delivery-indexer/internal/index/versions_test.go b/modules/mesh-delivery-indexer/internal/index/versions_test.go index 3f459836..7b5f1ebf 100644 --- a/modules/mesh-delivery-indexer/internal/index/versions_test.go +++ b/modules/mesh-delivery-indexer/internal/index/versions_test.go @@ -4,11 +4,13 @@ import ( "context" "errors" "fmt" + "strings" "testing" "time" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" ) // Design 47 G §349: a version is released once every artifact the module's entry names is published at it, said once as @@ -157,3 +159,126 @@ func TestTheFirstFillOfTheVersionsComesFromTheBuildersPublished(t *testing.T) { t.Errorf("filled again: %v, said %d", err, len(filledSaid)) } } + +// Design 47 G §374, §375, A-68: a release records, per edge, the part it was built against (kind 1, from package-published's +// dependencies) or, once delivered, the delivered version it was ordered after (kinds 2 and 4); outdated answers an edge behind +// by the part's digest alone, with the act rebuild only for a stale edge; a newer version that left the digest as it was +// makes nothing behind. +func TestOutdatedAnswersTheEdgesBehindByTheirPartsDigest(t *testing.T) { + ctx := context.Background() + ix := &Indexer{Versions: newMem(), OwnBuildState: newMem(), Revisions: newMemObjects(1 << 20)} + entries := map[string]meshcatalog.CatalogEntry{ + "lib": {Module: "lib", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "lib:pkg"}}}, + "dep": {Module: "dep", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "dep:program"}}}, + "app": {Module: "app", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "app:program"}}, + Edges: []meshcatalog.Edge{{Kind: 1, From: "lib:pkg", To: "app:program", Part: "lib:pkg", Stale: true, Word: "imports"}, + {Kind: 4, From: "dep:program", To: "app:program", Part: "dep:program", Word: "provision"}}}, + } + rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: 3, Kind: "trunk"}, entries) + if err != nil { + t.Fatal(err) + } + if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil { + t.Fatal(err) + } + pub := func(module, version, digest string, deps ...meshbuilds.Dependency) { + t.Helper() + if err := ix.HearPackagePublished(ctx, meshbuilds.PackagePublished{Module: module, Artifact: module + ":" + map[string]string{"lib": "pkg", "dep": "program", "app": "program"}[module], Digest: digest, Version: version, Dependencies: deps}); err != nil { + t.Fatal(err) + } + } + behind := func() []meshcatalog.Outdated { + t.Helper() + a, err := ix.Outdated(ctx, meshcatalog.OutdatedInput{}) + if err != nil { + t.Fatal(err) + } + return a.Behind + } + pub("lib", "0.1.1", "d1") + pub("dep", "0.1.1", "x1") + pub("app", "0.1.1", "a1", meshbuilds.Dependency{Kind: 1, Part: "lib:pkg", Version: "0.1.1", Digest: "d1"}) + v, _ := ModuleVersions(ctx, ix.Versions, "app") + if len(v[0].Edges) != 2 || v[0].Edges[0].Kind != 1 || !v[0].Edges[0].Stale || v[0].Edges[0].Word != "imports" || v[0].Edges[1].Version != "not yet delivered" { + t.Fatalf("the release's edges: %+v", v[0].Edges) + } + pub("lib", "0.1.2", "d1") // a newer version, the part's digest as it was + if b := behind(); len(b) != 0 { + t.Fatalf("a newer version that left the digest as it was is behind: %+v", b) + } + pub("lib", "0.1.3", "d3") + if b := behind(); len(b) != 1 || b[0].Module != "app" || b[0].Kind != 1 || b[0].Act != "rebuild" || b[0].Dependency.Artifact != "lib:pkg" || !strings.Contains(b[0].Newest, "d3") || !strings.Contains(b[0].Recorded, "d1") { + t.Fatalf("behind after lib's new digest: %+v", b) + } + // delivery: dep 0.1.1 is delivered, then app 0.1.1 writes its kind 4 entry; dep 0.1.2 delivered later is behind, an ordering edge + deliver := func(module, version string, parts map[string]string) { + t.Helper() + if err := ix.HearVersionDelivered(ctx, meshwants.VersionDeliveredEvent{Module: module, Version: version, Parts: parts}); err != nil { + t.Fatal(err) + } + } + deliver("dep", "0.1.1", map[string]string{"dep:program": "x1"}) + deliver("app", "0.1.1", map[string]string{"app:program": "a1"}) + v, _ = ModuleVersions(ctx, ix.Versions, "app") + if e := v[0].Edges[1]; e.Version != "0.1.1" || e.Digest != "x1" { + t.Fatalf("the kind 4 entry after the first delivery: %+v", e) + } + deliver("dep", "0.1.2", map[string]string{"dep:program": "x2"}) + deliver("app", "0.1.1", map[string]string{"app:program": "a1"}) // delivered again: written once, not changed + v, _ = ModuleVersions(ctx, ix.Versions, "app") + if e := v[0].Edges[1]; e.Version != "0.1.1" { + t.Errorf("an entry written at the first delivery changed: %+v", e) + } + b := behind() + if len(b) != 2 || b[1].Kind != 4 || b[1].Act != "" || b[1].Dependency.Artifact != "dep:program" { + t.Fatalf("behind after dep's second delivery: %+v", b) + } + // the newest toolchain-agreed per language is kept as heard + if err := ix.HearToolchainAgreed(ctx, meshbuilds.ToolchainAgreed{Language: "go", Version: "1.25"}); err != nil { + t.Fatal(err) + } + if err := ix.HearToolchainAgreed(ctx, meshbuilds.ToolchainAgreed{Language: "go", Version: "1.26"}); err != nil { + t.Fatal(err) + } + if raw, _, found, _ := ix.OwnBuildState.Get(ctx, ToolchainPrefix+"go"); !found || !strings.Contains(string(raw), "1.26") { + t.Errorf("toolchain.go holds %s", raw) + } +} + +// Design 47 G §354: an input is kept under its write time and sha256, read back by the verb input with its digest checked; +// the keys older than a day among the 20 oldest are deleted at each new keep. +func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) + ix := &Indexer{Inputs: newMem(), Now: func() time.Time { return now }} + d1, err := ix.KeepInput(ctx, []byte("first")) + if err != nil { + t.Fatal(err) + } + a, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1}) + if err != nil || a.Digest != d1 || a.Input != "Zmlyc3Q=" { + t.Fatalf("the input read back: %+v, %v", a, err) + } + if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: "sha256:none"}); err == nil { + t.Error("an input never kept is answered") + } + now = now.Add(23 * time.Hour) + if _, err := ix.KeepInput(ctx, []byte("second")); err != nil { + t.Fatal(err) + } + if keys, _ := ix.Inputs.Keys(ctx); len(keys) != 2 { + t.Fatalf("inside a day: %v", keys) + } + now = now.Add(2 * time.Hour) + d3, _ := ix.KeepInput(ctx, []byte("third")) + keys, _ := ix.Inputs.Keys(ctx) + if len(keys) != 2 { + t.Fatalf("after a day the first stays: %v", keys) + } + if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1}); err == nil { + t.Error("the input a day old is still answered") + } + if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d3}); err != nil { + t.Error(err) + } +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 7e7c7280..2ac33f79 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -125,7 +125,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build {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.RegistrationsState, History: 1}, {Name: index.ProposalsState, History: 1, MaxMiB: 4}, - {Name: index.VersionsState, History: 1, MaxMiB: 64}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17 + {Name: index.VersionsState, History: 1, MaxMiB: 64}, {Name: index.InputsState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17 t.Errorf("state = %+v, want %+v", m.State, want) } // Milestone 47/G W4.2 (design 47, Two graphs; ADR 0619 §2): the catalog graph revisions are objects in the old section, @@ -159,9 +159,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control // internal/broker/state.go:62-93). // 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-builds.package-published", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", + if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "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.version-delivered", "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" { @@ -196,8 +196,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // 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.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool) // design 47 G §667, §8 - if !reflect.DeepEqual(m.Tools, want) || len(want) != 45 || slices.Contains(m.Tools, "modules") { + index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool, index.InputsTool) // design 47 G §667, §8 + 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. @@ -301,7 +301,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // ADR 0604 §2, ADR 0636 §4: it holds the seat mesh-catalog, serving its four verbs and seat-definitions from its tools // bundle (which stays without a request subject); ADR 0663 §1: and revision, in the same runtime of its own. if len(m.Claims) != 1 || m.Claims[0].Name != index.CatalogSeat || m.Claims[0].Scope != catalogue.ScopeMesh || - !slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents", index.SeatDefinitionsVerb, "revision", "entry", "registrations", "register", "policy", "watch", "unwatch", "versions"}) { + !slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents", index.SeatDefinitionsVerb, "revision", "entry", "registrations", "register", "policy", "watch", "unwatch", "versions", "outdated", "input"}) { t.Errorf("claims = %+v, want mesh-catalog with its seven verbs (entry: design 47 G §675)", m.Claims) } for _, e := range []string{index.SeatDefinitionChangedEvent, index.SeatDefinitionRemovedEvent} { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 093f7de4..f9427b76 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -20,7 +20,9 @@ "policy", "watch", "unwatch", - "versions" + "versions", + "outdated", + "input" ] } ], @@ -69,7 +71,8 @@ "indexer_entries", "indexer_proposals", "indexer_proposal_drop", - "indexer_seed_again" + "indexer_seed_again", + "indexer_inputs" ], "consumes": [ "go-package-registry.published", @@ -81,6 +84,8 @@ "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-builds.package-published", + "mesh-builds.toolchain-agreed", + "mesh-wants.version-delivered", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", "mesh-delivery.source-moved", @@ -110,7 +115,8 @@ { "name": "entry-index", "history": 1, "max-mib": 8 }, { "name": "registrations", "history": 1 }, { "name": "proposals", "history": 1, "max-mib": 4 }, - { "name": "versions", "history": 1, "max-mib": 64 } + { "name": "versions", "history": 1, "max-mib": 64 }, + { "name": "inputs", "history": 1 } ], "streams": [ { "name": "registration-history", "events": ["registration.*"], "retention": "history", "max-mib": 64 },