From e9d29977603dec15d5ed6fbaedfbf8d5724d9083 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 23:04:14 +0100 Subject: [PATCH] indexer: the first fill of the versions from builder_versions, then versions-filled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Filled once, as released, with each module's highest published version; an unanswered builder_versions holds it and, after five minutes, is call-unanswered. The edge entries of the first fill (§377) wait for answers. Builds: design 47 G §372 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-indexer/main.go | 9 +- .../cmd/mesh-delivery-indexer/versions.go | 52 ++++++ .../internal/index/candidates.go | 45 ++--- .../internal/index/pending.go | 2 +- .../internal/index/pending_test.go | 2 +- .../internal/index/seed_steps.go | 32 ++-- .../internal/index/versions_fill.go | 174 ++++++++++++++++++ .../internal/index/versions_test.go | 79 ++++++++ .../mesh-delivery-indexer/manifest_test.go | 2 +- modules/mesh-delivery-indexer/module.json | 3 +- 10 files changed, 361 insertions(+), 39 deletions(-) create mode 100644 modules/mesh-delivery-indexer/internal/index/versions_fill.go 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 9795d4f5..f5827821 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -431,6 +431,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } }) readWhenBound(seatReader, seatDefs, marks) + var fillFollow atomic.Pointer[versionsFill] 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, 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, @@ -440,11 +441,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, Message: messageVia(calls.CallKept), Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.EntryIndexBucket, "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1) // Each catalog revision may bring another holder of a build seat: the build modules are found again (buildmodules.go). - SayCatalogRevision: refreshingBuildModules(sayCatalogRevision(conn, cred), &buildFollow), + SayCatalogRevision: afterRevision(refreshingBuildModules(sayCatalogRevision(conn, cred), &buildFollow), &fillFollow), 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)}, - SayVersionReleased: sayVersionReleased(conn, cred), + 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. Pending: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)}, @@ -456,6 +457,10 @@ 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 + // The first fill of the versions (design 47 G §372): at the start, and again at each catalog revision kept until it is made. + fill := &versionsFill{ix: ix, ctx: ctx, ask: versionsVia(calls.CallKept), logf: logger.Printf} + fillFollow.Store(fill) + fill.kick() ix.SeedModules = func(ctx context.Context) ([]index.SeedModule, error) { r := seatReader.Reader() if r == nil { 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 a9822e1b..f8507c48 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go @@ -3,6 +3,8 @@ package main import ( "context" "encoding/json" + "fmt" + "sync/atomic" "time" "github.com/nats-io/nats.go" @@ -56,3 +58,53 @@ func heardPackagePublished(ctx context.Context, ix *index.Indexer, m jetstream.M } _ = 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 { + return saySeatEvent(conn, cred, meshcatalog.EventVersionsFilled, id, e) + } +} + +// versionsVia asks the builder's tool builder_versions for one name as a kept call (ADR 0668 §1, §2) and reads its answer. +func versionsVia(call toolCaller) index.VersionsAsker { + return func(_ context.Context, name string) (index.BuilderVersionsAnswer, error) { + var a index.BuilderVersionsAnswer + raw, err := call(index.BuilderVersionsCall, map[string]any{"name": name}) + if err != nil { + return a, err + } + if err := json.Unmarshal(raw, &a); err != nil { + return a, fmt.Errorf("the answer of %s is not a name's versions: %w", index.BuilderVersionsCall, err) + } + return a, nil + } +} + +// versionsFill is the first fill of the versions in the background (design 47 G §372): started at the start and at each +// catalog revision kept, until it is made; one runs at a time, and a fill that fails is said and made again at the next. +type versionsFill struct { + ix *index.Indexer + ctx context.Context + ask index.VersionsAsker + logf func(string, ...any) +} + +func (f *versionsFill) kick() { + go func() { + if err := f.ix.FillVersions(f.ctx, f.ask); err != nil { + f.logf("versions: the first fill waits: %v", err) + } + }() +} + +// afterRevision runs the fill's kick after each catalog revision kept, once the fill is set. +func afterRevision(say func(context.Context, meshcatalog.CatalogRevisionEvent) error, fill *atomic.Pointer[versionsFill]) func(context.Context, meshcatalog.CatalogRevisionEvent) error { + return func(ctx context.Context, e meshcatalog.CatalogRevisionEvent) error { + err := say(ctx, e) + if f := fill.Load(); f != nil { + f.kick() + } + return err + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 9c86e19f..a683c83b 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -111,6 +111,8 @@ type Indexer struct { Versions Store // SayVersionReleased says the seat mesh-catalog's event version-released under the duplicate id given (§349); nil says nothing. 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 // 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 @@ -148,27 +150,28 @@ type Indexer struct { // closed message the indexer missed heals at the next one (issue 1126); nil asks none. ReadOpen func(ctx context.Context) ([]forgemsg.OpenPull, error) - mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach - source sourceNow - previews map[string]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue // by previews key (previews.go) - toolchains map[string]map[string]nodebuildgo.Toolchain // language, then node (toolchains.go) - buildModules map[string]BuildModule // by language: the build modules whose toolchains are taken in (buildmodules.go) - stepsSince map[string]time.Time // by build seat: since when its steps is unanswered (seed_steps.go) - tried map[string]bool // repository, commit and directory a registered module's manifest was tried at (registered.go) - unread map[string]bool // registered modules whose last read the forge could not answer, read again by RetryUnread - moving map[string]forgemsg.Forge // by modules key: a merge naming a module another repository holds, waiting for that trunk to drop it (moves.go) - retrying atomic.Bool // a RetryUnread runs in the background (kickRetry) - retryAgain atomic.Bool // an event asked for one more RetryUnread after the running one - retryWG sync.WaitGroup // the background RetryUnread, for tests to wait on - fresh map[string]bool // modules whose import form is computed again by hand (FingerprintsAgain), "*" for every one; under freshMu - freshMu sync.Mutex - recompute sync.Mutex // one Recompute at a time - untaken map[string]string // repository to the latest merge applied while the taken state was missing (taken.go) - filling atomic.Bool // a FillTaken runs in the background (KickFill) - fillAgain atomic.Bool // a kick arrived while it ran - atHeads map[string]map[string]meshcatalog.PackagesByModuleValue // a member head's published Go modules, by repository@head (candidatePublished) - saidAt map[string]string // the head of a repository whose failed read was said, by repository: said once per head - failedAt map[string]string // the head of a repository whose tree lacks a directory asked, by repository: not asked again until the head changes + mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach + source sourceNow + previews map[string]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue // by previews key (previews.go) + toolchains map[string]map[string]nodebuildgo.Toolchain // language, then node (toolchains.go) + buildModules map[string]BuildModule // by language: the build modules whose toolchains are taken in (buildmodules.go) + versionsFilling atomic.Bool // a FillVersions runs (versions_fill.go) + stepsSince map[string]time.Time // by build seat: since when its steps is unanswered (seed_steps.go) + tried map[string]bool // repository, commit and directory a registered module's manifest was tried at (registered.go) + unread map[string]bool // registered modules whose last read the forge could not answer, read again by RetryUnread + moving map[string]forgemsg.Forge // by modules key: a merge naming a module another repository holds, waiting for that trunk to drop it (moves.go) + retrying atomic.Bool // a RetryUnread runs in the background (kickRetry) + retryAgain atomic.Bool // an event asked for one more RetryUnread after the running one + retryWG sync.WaitGroup // the background RetryUnread, for tests to wait on + fresh map[string]bool // modules whose import form is computed again by hand (FingerprintsAgain), "*" for every one; under freshMu + freshMu sync.Mutex + recompute sync.Mutex // one Recompute at a time + untaken map[string]string // repository to the latest merge applied while the taken state was missing (taken.go) + filling atomic.Bool // a FillTaken runs in the background (KickFill) + fillAgain atomic.Bool // a kick arrived while it ran + atHeads map[string]map[string]meshcatalog.PackagesByModuleValue // a member head's published Go modules, by repository@head (candidatePublished) + saidAt map[string]string // the head of a repository whose failed read was said, by repository: said once per head + failedAt map[string]string // the head of a repository whose tree lacks a directory asked, by repository: not asked again until the head changes // OwnBuild answers the fingerprint of the build this program runs, from the node-engine's seat; nil knows none. OwnBuild func(ctx context.Context) (string, error) underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 36f24c49..a18efefe 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, + Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: 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 58e4d7b3..e51ba1dc 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"} { + "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/seed_steps.go b/modules/mesh-delivery-indexer/internal/index/seed_steps.go index 587261ae..1453433f 100644 --- a/modules/mesh-delivery-indexer/internal/index/seed_steps.go +++ b/modules/mesh-delivery-indexer/internal/index/seed_steps.go @@ -175,44 +175,52 @@ func namedPaths(e meshcatalog.CatalogEntry, name string) []string { // five minutes. const StepsSincePrefix = "steps-since." -// stepsUnansweredSince is the time the seat first failed to answer, kept in the indexer's own state; the first failure -// writes now. Where no state is kept it is held in memory. +// stepsUnansweredSince is the time the seat first failed to answer. func (ix *Indexer) stepsUnansweredSince(ctx context.Context, seat string, now time.Time) time.Time { + return ix.unansweredSince(ctx, StepsSincePrefix+seat, now) +} + +// stepsAnswered forgets the time a seat that answers was unanswered since. +func (ix *Indexer) stepsAnswered(ctx context.Context, seat string) { + ix.answered(ctx, StepsSincePrefix+seat) +} + +// unansweredSince is the time the call kept at key first failed to answer, kept in the indexer's own state (the first +// failure writes now); where no state is kept it is held in memory. +func (ix *Indexer) unansweredSince(ctx context.Context, key string, now time.Time) time.Time { if ix.OwnBuildState == nil { ix.mu.Lock() defer ix.mu.Unlock() if ix.stepsSince == nil { ix.stepsSince = map[string]time.Time{} } - if _, ok := ix.stepsSince[seat]; !ok { - ix.stepsSince[seat] = now + if _, ok := ix.stepsSince[key]; !ok { + ix.stepsSince[key] = now } - return ix.stepsSince[seat] + return ix.stepsSince[key] } - key := StepsSincePrefix + seat if v, _, found, err := ix.OwnBuildState.Get(ctx, key); err == nil && found { if t, err := time.Parse(time.RFC3339Nano, string(v)); err == nil { return t } } if err := ix.OwnBuildState.Create(ctx, key, []byte(now.UTC().Format(time.RFC3339Nano))); err != nil { - ix.logf("seed: the time since %s.%s is unanswered is not kept: %v", seat, StepsVerb, err) + ix.logf("the time since %s is unanswered is not kept: %v", key, err) } return now } -// stepsAnswered forgets the time a seat that answers was unanswered since. -func (ix *Indexer) stepsAnswered(ctx context.Context, seat string) { +// answered forgets the time the call kept at key was unanswered since. +func (ix *Indexer) answered(ctx context.Context, key string) { if ix.OwnBuildState == nil { ix.mu.Lock() - delete(ix.stepsSince, seat) + delete(ix.stepsSince, key) ix.mu.Unlock() return } - key := StepsSincePrefix + seat if _, rev, found, err := ix.OwnBuildState.Get(ctx, key); err == nil && found { if err := ix.OwnBuildState.Delete(ctx, key, rev); err != nil { - ix.logf("seed: the time since %s.%s is unanswered is not forgotten: %v", seat, StepsVerb, err) + ix.logf("the time since %s is unanswered is not forgotten: %v", key, err) } } } diff --git a/modules/mesh-delivery-indexer/internal/index/versions_fill.go b/modules/mesh-delivery-indexer/internal/index/versions_fill.go new file mode 100644 index 00000000..b52537f4 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/versions_fill.go @@ -0,0 +1,174 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "time" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// The first fill of the versions (design 47 G §372): at the first start under G the state versions is filled once, as +// released, for each module from the highest published version over every artifact of the module that the builder's state +// versions held, read by the builder's tool builder_versions as a kept call; then versions-filled is said once, and the +// builder reserves nothing before it (§369). +const ( + // BuilderVersionsCall is the tool the manifest names in its invokes (until that tool goes). + BuilderVersionsCall = "mesh-delivery-builder.builder_versions" + // VersionsFilledKey is the key of the own-build state that says the fill is done, so it is not made again. + VersionsFilledKey = "versions-filled" + // versionsFillSince is the own-build key of the time builder_versions is first unanswered. + versionsFillSince = "call-since." + BuilderVersionsCall +) + +// BuilderVersion is one version of a name in the answer of builder_versions (its fields the fill reads). +type BuilderVersion struct { + Version string `json:"version"` + State string `json:"state"` + Digest string `json:"digest"` + Revision string `json:"revision"` + At time.Time `json:"at"` +} + +// BuilderVersionsAnswer is what builder_versions answers for a name. +type BuilderVersionsAnswer struct { + Name string `json:"name"` + Versions []BuilderVersion `json:"versions"` +} + +// VersionsAsker asks builder_versions for a name's versions (a kept call, ADR 0668 §1, §2). +type VersionsAsker func(ctx context.Context, name string) (BuilderVersionsAnswer, error) + +// versionNameOf is the name the builder numbers an artifact under: its package's name in its ecosystem, else +// / (images, bundles and archives). +func versionNameOf(module string, a meshcatalog.Artifact) string { + if a.Package != nil && a.Package.Name != "" { + return a.Package.Name + } + return module + "/" + artifactName(a.Ref) +} + +// FillVersions fills the state versions once from builder_versions and says versions-filled (§372). A fill already made +// is not made again; a builder_versions that does not answer holds the fill and, after five minutes, is the finding +// call-unanswered (scope call, id mesh-delivery-builder.builder_versions), cleared when it answers. +func (ix *Indexer) FillVersions(ctx context.Context, ask VersionsAsker) error { + if ix.Versions == nil || ix.OwnBuildState == nil { + return errors.New("the state versions is not kept here") + } + if !ix.versionsFilling.CompareAndSwap(false, true) { + return nil // one fill runs at a time + } + defer ix.versionsFilling.Store(false) + if _, _, done, err := ix.OwnBuildState.Get(ctx, VersionsFilledKey); err != nil || done { + return err + } + if ix.Revisions == nil { + return errors.New("the catalog graph is not kept here: no entry names the modules' artifacts") + } + rev, found, err := newestTrunk(ctx, ix.Revisions) + if err != nil { + return err + } + if !found { + return errors.New("the indexer keeps no trunk revision yet: the first fill of the versions waits for its first") + } + _, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", rev.Revision)) + if err != nil { + return err + } + type found1 struct { + v BuilderVersion + name string + ref string + } + best := map[string][]found1{} // by module: each artifact's highest published version + var unanswered error + for _, e := range entries { + var top string // the module's highest published version + per := []found1{} + for _, a := range e.Artifacts { + name := versionNameOf(e.Module, a) + ans, err := ask(ctx, name) + if err != nil { + unanswered = err + continue + } + for _, v := range ans.Versions { + if _, ok := versionParts(v.Version); !ok || v.State != "published" { + continue + } + per = append(per, found1{v: v, name: name, ref: a.Ref}) + if top == "" || compareVersions(v.Version, top) > 0 { + top = v.Version + } + } + } + for _, f := range per { + if f.v.Version == top { + best[e.Module] = append(best[e.Module], f) + } + } + } + if unanswered != nil { + ix.fillUnanswered(ctx, unanswered) + return fmt.Errorf("builder_versions is not answered: %w", unanswered) + } + ix.answered(ctx, versionsFillSince) + if ix.Findings != nil { + if err := ix.clearFinding(ctx, mcc.ScopeCall, BuilderVersionsCall, mcc.KindCallUnanswered, "builder_versions answers"); err != nil { + ix.logf("versions fill: the finding %s is not cleared: %v", BuilderVersionsCall, err) + } + } + now := ix.now().UTC() + filled := 0 + for module, fs := range best { + v := meshcatalog.VersionEntry{V: int64(keys.Version), Module: module, Version: fs[0].v.Version, At: &now} + for _, f := range fs { + v.Artifacts = append(v.Artifacts, meshcatalog.VersionArtifact{Artifact: f.ref, Digest: f.v.Digest}) + if n, err := strconv.ParseInt(f.v.Revision, 10, 64); err == nil && n > v.Source { + v.Source = n + } + } + data, err := json.Marshal(v) + if err != nil { + return err + } + if err := ix.Versions.Create(ctx, meshcatalog.VersionsByModuleVersionKey(module, v.Version), data); err != nil && !errors.Is(err, ErrConflict) { + return fmt.Errorf("the version %s %s is not kept: %w", module, v.Version, err) + } + filled++ + } + if ix.SayVersionsFilled != nil { + ev := meshcatalog.VersionsFilledEvent{V: int64(keys.Version), At: now, Modules: int64(filled)} + if err := ix.SayVersionsFilled(ctx, ev, meshcatalog.EventVersionsFilled); err != nil { + return fmt.Errorf("versions-filled is not said: %w", err) + } + } + if err := ix.OwnBuildState.Create(ctx, VersionsFilledKey, []byte(now.Format(time.RFC3339Nano))); err != nil && !errors.Is(err, ErrConflict) { + return err + } + ix.logf("versions: filled for %d modules from builder_versions, versions-filled said", filled) + return nil +} + +// fillUnanswered writes call-unanswered once builder_versions is unanswered five minutes (the time kept in the own state). +func (ix *Indexer) fillUnanswered(ctx context.Context, cause error) { + now := ix.now() + since := ix.unansweredSince(ctx, versionsFillSince, now) + if ix.Findings == nil || now.Sub(since) < mcc.CallUnansweredAfter { + return + } + f := ix.observed(mcc.ScopeCall, BuilderVersionsCall, mcc.KindCallUnanswered, + fmt.Sprintf("the call %s is unanswered since %s", BuilderVersionsCall, since.UTC().Format(time.RFC3339)), + "A call whose answer is kept is not answered", + fmt.Sprintf("The first fill of the catalog's versions asks %s of every artifact; it is not answered since %s (%v), so no version is released and the builder reserves nothing.", BuilderVersionsCall, since.UTC().Format(time.RFC3339), cause), + "indexer."+mcc.KindCallUnanswered) + if err := ix.Findings.Write(ctx, f); err != nil { + ix.logf("versions fill: the finding %s is not written: %v", BuilderVersionsCall, err) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/versions_test.go b/modules/mesh-delivery-indexer/internal/index/versions_test.go index 5fbf0fe0..3f459836 100644 --- a/modules/mesh-delivery-indexer/internal/index/versions_test.go +++ b/modules/mesh-delivery-indexer/internal/index/versions_test.go @@ -2,6 +2,7 @@ package index import ( "context" + "errors" "fmt" "testing" "time" @@ -78,3 +79,81 @@ func TestAVersionIsReleasedOnceEveryArtifactIsPublished(t *testing.T) { t.Errorf("the oldest kept is %s: the oldest were not the ones deleted", oldest) } } + +// Design 47 G §372: the first fill records, once, each module's highest published version over its artifacts as released +// and says versions-filled once; a builder_versions that does not answer holds it and, after five minutes, is the finding +// call-unanswered; a fill made is not made again. +func TestTheFirstFillOfTheVersionsComesFromTheBuildersPublished(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) + w := newFindings() + var filledSaid []meshcatalog.VersionsFilledEvent + ix := &Indexer{Versions: newMem(), OwnBuildState: newMem(), Findings: w, Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now }, + SayVersionsFilled: func(_ context.Context, e meshcatalog.VersionsFilledEvent, id string) error { + filledSaid = append(filledSaid, e) + return nil + }} + entries := map[string]meshcatalog.CatalogEntry{ + "mesh-sdk": {Module: "mesh-sdk", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{ + {Ref: "mesh-sdk:go", Package: &meshcatalog.ArtifactPackage{Ecosystem: "go", Name: "git.novox.be/novox/mesh-sdk/go"}}, + {Ref: "mesh-sdk:ts", Package: &meshcatalog.ArtifactPackage{Ecosystem: "npm", Name: "@novox/mesh-sdk"}}}}, + "avahi": {Module: "avahi", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "avahi:image", Kind: "image"}}}, + "fresh": {Module: "fresh", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "fresh:image", Kind: "image"}}}, + } + 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) + } + up := false + ask := func(_ context.Context, name string) (BuilderVersionsAnswer, error) { + if !up { + return BuilderVersionsAnswer{}, errors.New("no responders") + } + pub := func(v, d string) BuilderVersion { + return BuilderVersion{Version: v, State: "published", Digest: d, Revision: "9"} + } + switch name { + case "git.novox.be/novox/mesh-sdk/go": + return BuilderVersionsAnswer{Name: name, Versions: []BuilderVersion{pub("0.1.14", "sha256:go14"), {Version: "0.1.15", State: "refused"}}}, nil + case "@novox/mesh-sdk": + return BuilderVersionsAnswer{Name: name, Versions: []BuilderVersion{pub("0.1.18", "sha256:ts18")}}, nil + case "avahi/image": + return BuilderVersionsAnswer{Name: name, Versions: []BuilderVersion{pub("0.2.3", "sha256:av")}}, nil + } + return BuilderVersionsAnswer{Name: name}, nil + } + if err := ix.FillVersions(ctx, ask); err == nil || len(filledSaid) != 0 { + t.Fatalf("a builder_versions unanswered filled: %v", err) + } + if len(w.Held()) != 0 { + t.Fatal("a call unanswered for no time wrote its finding") + } + now = now.Add(5 * time.Minute) + _ = ix.FillVersions(ctx, ask) + if h := w.Held(); len(h) != 1 || h[0].ID != BuilderVersionsCall { + t.Fatalf("the findings after five minutes: %+v", h) + } + up = true + if err := ix.FillVersions(ctx, ask); err != nil { + t.Fatal(err) + } + if len(w.Held()) != 0 { + t.Errorf("the finding stands after builder_versions answered: %+v", w.Held()) + } + if len(filledSaid) != 1 || filledSaid[0].Modules != 2 { + t.Fatalf("versions-filled said %+v", filledSaid) + } + sdk, _ := ModuleVersions(ctx, ix.Versions, "mesh-sdk") + if len(sdk) != 1 || sdk[0].Version != "0.1.18" || len(sdk[0].Artifacts) != 1 || sdk[0].Artifacts[0].Artifact != "mesh-sdk:ts" || sdk[0].Source != 9 { + t.Errorf("mesh-sdk's filled version: %+v (the higher of Go's 0.1.14 and npm's 0.1.18)", sdk) + } + if fresh, _ := ModuleVersions(ctx, ix.Versions, "fresh"); len(fresh) != 0 { + t.Errorf("a module with none published is filled: %+v", fresh) + } + if err := ix.FillVersions(ctx, ask); err != nil || len(filledSaid) != 1 { + t.Errorf("filled again: %v, said %d", err, len(filledSaid)) + } +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 9b42127a..7e7c7280 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -173,7 +173,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths: // ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a // commit is in another (issue 1900). - if want := []string{"mesh-delivery-assigner.assignments", "seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", "seat:node-build-node.steps", "seat:node-build-image.steps"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"mesh-delivery-assigner.assignments", "seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", "seat:node-build-node.steps", "seat:node-build-image.steps", "mesh-delivery-builder.builder_versions"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } // 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 33863ef4..093f7de4 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -155,7 +155,8 @@ "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", "seat:node-build-node.steps", - "seat:node-build-image.steps" + "seat:node-build-image.steps", + "mesh-delivery-builder.builder_versions" ], "own-secrets": { "broker": "${dir:state}/broker"