indexer: the first fill of the versions from builder_versions, then versions-filled
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
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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...) }
|
||||
|
||||
@@ -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{})
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
// <module>/<artifact> (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)
|
||||
}
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user