indexer: what it heard (steps-since, call-since, toolchain, delivered, pending) lives in the state heard, each key part escaped
The own-build state keeps its one key. Builds: design 47 G A-80, §73, §375 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -289,7 +289,7 @@ func main() {
|
||||
// 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)}),
|
||||
index.OutdatedTool(&index.Indexer{Versions: keptRead{stdio.State(index.VersionsState)}, OwnBuildState: keptRead{stdio.State(index.OwnBuildStateName)}}),
|
||||
index.OutdatedTool(&index.Indexer{Versions: keptRead{stdio.State(index.VersionsState)}, Heard: keptRead{stdio.State(index.HeardState)}}),
|
||||
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"})
|
||||
|
||||
@@ -399,6 +399,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// appears; until then indexer_requirements answers none, and the builder's floor waits on its answer.
|
||||
requirements := process.AwaitBucket(ctx, js, conn, index.RequirementsBucket, "what consumers require is not kept", logger.Printf)
|
||||
// The module graph (ADR 0603 §2) is held in memory, read with the view: each revision's entries carry its nodes.
|
||||
// The state heard (A-80): the forge messages handled (key forge) and what the indexer heard and compares later (the time a call is
|
||||
// unanswered since, the newest toolchain-agreed and version-delivered, the artifacts published so far of a version).
|
||||
heardStore := awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.HeardBucket, "what the indexer heard is not kept", logger.Printf)}
|
||||
ownBuildState := process.AwaitBucket(ctx, js, conn, index.OwnBuildBucket, "which build this indexer runs is not kept", logger.Printf)
|
||||
// The manifests a take-in wrote and no judge pass judged yet (issue 1966): a state this delivery brings, bound when it
|
||||
// appears; until then a take-in's new manifests are kept for their judging in its memory alone.
|
||||
@@ -438,7 +441,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf,
|
||||
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
|
||||
ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.PackagesBucket}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call),
|
||||
OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, TryRecovery: awaitedStore{b: tryRecovery}, Graph: graph,
|
||||
OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, TryRecovery: awaitedStore{b: tryRecovery}, Graph: graph,
|
||||
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).
|
||||
@@ -746,7 +749,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) })
|
||||
// What the indexer heard of FORGE (issue 1126 note 76) is bound when its state appears, as every state a delivery adds
|
||||
// (issue 1790): until then no message is written there, and the rest of the indexer runs.
|
||||
heard := awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.HeardBucket, "no FORGE message the indexer handles is written for indexer_heard", logger.Printf)}
|
||||
heard := heardStore
|
||||
// Each change of the states its readers read is said on its views family (ADR 0615 §5), from the mark each state was
|
||||
// said up to; a first view (the tools bundle) answers at that mark.
|
||||
startViews(ctx, js, cred, marks, writer,
|
||||
|
||||
@@ -190,6 +190,10 @@ type Indexer struct {
|
||||
ownWhy string
|
||||
// OwnBuildState keeps which build this process knows it runs, or since when and why not (FixOwnBuild); nil keeps none.
|
||||
OwnBuildState Store
|
||||
// Heard keeps what the indexer heard and compares later (A-80): the time a call is unanswered since (steps-since, call-since),
|
||||
// toolchain.<language>, delivered.<module>, pending.<module>.<version>, acted.<acting part>.<dependent>, each key part escaped as a name
|
||||
// (§73), and the mark versions-filled; nil keeps none. It is the state heard, whose key forge keeps the forge messages handled.
|
||||
Heard Store
|
||||
judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go)
|
||||
// Message asks the forge bridge for the merged message of a merge, repository, merge commit and pull request number in (the number lets the bridge read a merge made on the forge, not through the gate), the message whole out (a
|
||||
// pending value keeps a reference, not a copy: issue 1585); an error is the bridge's words. Nil asks none.
|
||||
|
||||
@@ -908,7 +908,7 @@ func (ix *Indexer) dry() *dryRun {
|
||||
// requirements, findings, seat definitions, holds and their bounds are not overlaid, nothing is said, and no hold is
|
||||
// weighed (Holds and Counted nil hold nothing); ReadFile only reads.
|
||||
Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil,
|
||||
OwnBuildState: nil, underSeats: true,
|
||||
OwnBuildState: nil, Heard: 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, Inputs: nil,
|
||||
// A dry run writes no registration and asks no one.
|
||||
|
||||
@@ -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", "Inputs"} {
|
||||
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Heard", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs"} {
|
||||
named[f] = true
|
||||
}
|
||||
ty := reflect.TypeOf(Indexer{})
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
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-delivery/contracts/go/seats/nodebuildgo"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
)
|
||||
|
||||
@@ -170,25 +171,25 @@ func namedPaths(e meshcatalog.CatalogEntry, name string) []string {
|
||||
return e.Inputs.Paths[ArtifactRefOf(e.Module, name)]
|
||||
}
|
||||
|
||||
// StepsSincePrefix opens the keys of the own-build state (OwnBuildState) that keep, by build seat, since when its steps
|
||||
// StepsSincePrefix opens the keys of the state heard (Heard, A-80) that keep, by build seat, since when its steps
|
||||
// has not answered (design 47 G A-66): a restart keeps the time, and the finding is written at the next seed attempt past
|
||||
// five minutes.
|
||||
const StepsSincePrefix = "steps-since."
|
||||
|
||||
// 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)
|
||||
return ix.unansweredSince(ctx, StepsSincePrefix+keys.EscapeName(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)
|
||||
ix.answered(ctx, StepsSincePrefix+keys.EscapeName(seat))
|
||||
}
|
||||
|
||||
// unansweredSince is the time the call kept at key first failed to answer, kept in the indexer's own state (the first
|
||||
// unansweredSince is the time the call kept at key first failed to answer, kept in the indexer's state heard (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 {
|
||||
if ix.Heard == nil {
|
||||
ix.mu.Lock()
|
||||
defer ix.mu.Unlock()
|
||||
if ix.stepsSince == nil {
|
||||
@@ -199,12 +200,12 @@ func (ix *Indexer) unansweredSince(ctx context.Context, key string, now time.Tim
|
||||
}
|
||||
return ix.stepsSince[key]
|
||||
}
|
||||
if v, _, found, err := ix.OwnBuildState.Get(ctx, key); err == nil && found {
|
||||
if v, _, found, err := ix.Heard.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 {
|
||||
if err := ix.Heard.Create(ctx, key, []byte(now.UTC().Format(time.RFC3339Nano))); err != nil {
|
||||
ix.logf("the time since %s is unanswered is not kept: %v", key, err)
|
||||
}
|
||||
return now
|
||||
@@ -212,14 +213,14 @@ func (ix *Indexer) unansweredSince(ctx context.Context, key string, now time.Tim
|
||||
|
||||
// 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 {
|
||||
if ix.Heard == nil {
|
||||
ix.mu.Lock()
|
||||
delete(ix.stepsSince, key)
|
||||
ix.mu.Unlock()
|
||||
return
|
||||
}
|
||||
if _, rev, found, err := ix.OwnBuildState.Get(ctx, key); err == nil && found {
|
||||
if err := ix.OwnBuildState.Delete(ctx, key, rev); err != nil {
|
||||
if _, rev, found, err := ix.Heard.Get(ctx, key); err == nil && found {
|
||||
if err := ix.Heard.Delete(ctx, key, rev); err != nil {
|
||||
ix.logf("the time since %s is unanswered is not forgotten: %v", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,7 +20,7 @@ func TestTheSeedAsksStepsOfTheBuildSeatsOfTheTrunkEntries(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
|
||||
w := newFindings()
|
||||
ix := &Indexer{Findings: w, OwnBuildState: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now }}
|
||||
ix := &Indexer{Findings: w, Heard: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now }}
|
||||
fp := "src2:" + strings.Repeat("a", 64)
|
||||
entries := map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c1", Path: "mods/avahi", Fp: fp, Manifest: map[string]any{},
|
||||
Artifacts: []meshcatalog.Artifact{{Ref: "avahi:program", Language: "go"}, {Ref: "avahi:image", Language: "go", Kind: "image"}},
|
||||
@@ -78,7 +78,7 @@ func TestTheSeedAsksStepsOfTheBuildSeatsOfTheTrunkEntries(t *testing.T) {
|
||||
t.Fatal("a call unanswered for no time wrote its finding")
|
||||
}
|
||||
now = now.Add(5 * time.Minute)
|
||||
ix = &Indexer{Findings: w, OwnBuildState: ix.OwnBuildState, Revisions: ix.Revisions, Now: func() time.Time { return now }} // a restart keeps the time
|
||||
ix = &Indexer{Findings: w, Heard: ix.Heard, Revisions: ix.Revisions, Now: func() time.Time { return now }} // a restart keeps the time
|
||||
if _, err := ix.SeedModulesFrom(ctx, defs, call); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -26,8 +26,9 @@ const (
|
||||
// VersionsKept is how many versions of a module the state keeps (§349): at each release the module's oldest beyond them
|
||||
// go, never its highest of any MAJOR.MINOR.
|
||||
VersionsKept = 100
|
||||
// PendingPrefix opens the keys of the own-build state (OwnBuildState) that keep, per module and version not yet
|
||||
// released, the artifacts published at it so far (A-68's state, the one of A-66).
|
||||
// PendingPrefix opens the keys of the state heard (Heard) that keep, per module and version not yet released, the
|
||||
// artifacts published at it so far, each key part escaped as a name (A-80, §73).
|
||||
|
||||
PendingPrefix = "pending."
|
||||
)
|
||||
|
||||
@@ -101,7 +102,7 @@ func ModuleVersions(ctx context.Context, st Store, module string) ([]meshcatalog
|
||||
// and said as version-released unless a higher one is released already (then it is recorded marked late and not said). A
|
||||
// build of a watched branch other than the registration's first releases none (§371).
|
||||
func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.PackagePublished) error {
|
||||
if ix.Versions == nil || ix.OwnBuildState == nil {
|
||||
if ix.Versions == nil || ix.Heard == nil {
|
||||
return errors.New("the state versions is not kept here")
|
||||
}
|
||||
if _, ok := versionParts(p.Version); !ok {
|
||||
@@ -123,9 +124,9 @@ func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.Packag
|
||||
for _, a := range entry.Artifacts {
|
||||
want = append(want, a.Ref)
|
||||
}
|
||||
key := PendingPrefix + meshcatalog.VersionsByModuleVersionKey(p.Module, p.Version)
|
||||
key := PendingPrefix + keys.EscapeName(p.Module) + "." + keys.EscapeName(p.Version)
|
||||
var have published
|
||||
err = write(ctx, ix.OwnBuildState, key, func(raw []byte, found bool) ([]byte, bool, error) {
|
||||
err = write(ctx, ix.Heard, key, func(raw []byte, found bool) ([]byte, bool, error) {
|
||||
have = published{}
|
||||
if found {
|
||||
if err := json.Unmarshal(raw, &have); err != nil {
|
||||
@@ -249,10 +250,10 @@ func (ix *Indexer) trimVersions(ctx context.Context, module string) error {
|
||||
|
||||
// forgetPublished removes the artifacts kept as published at a version released.
|
||||
func (ix *Indexer) forgetPublished(ctx context.Context, key string) error {
|
||||
if _, rev, found, err := ix.OwnBuildState.Get(ctx, key); err != nil || !found {
|
||||
if _, rev, found, err := ix.Heard.Get(ctx, key); err != nil || !found {
|
||||
return err
|
||||
} else {
|
||||
return ix.OwnBuildState.Delete(ctx, key, rev)
|
||||
return ix.Heard.Delete(ctx, key, rev)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -11,26 +11,27 @@ 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/keys"
|
||||
)
|
||||
|
||||
// What outdated compares against is kept as heard, in the indexer's own state (design 47 G A-68): toolchain.<language> the newest
|
||||
// What outdated compares against is kept as heard, in the indexer's state heard (design 47 G A-68): toolchain.<language> the newest
|
||||
// toolchain-agreed per language, delivered.<module> 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
|
||||
// keepNewest writes body at key of the state heard 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 {
|
||||
if ix.Heard == 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) {
|
||||
return write(ctx, ix.Heard, key, func(kept []byte, found bool) ([]byte, bool, error) {
|
||||
if found && !newer(kept) {
|
||||
return nil, false, nil
|
||||
}
|
||||
@@ -43,14 +44,14 @@ func (ix *Indexer) HearToolchainAgreed(ctx context.Context, t meshbuilds.Toolcha
|
||||
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 })
|
||||
return ix.keepNewest(ctx, ToolchainPrefix+keys.EscapeName(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 {
|
||||
if ix.Versions == nil || ix.Heard == nil {
|
||||
return errors.New("the state versions is not kept here")
|
||||
}
|
||||
if _, ok := versionParts(d.Version); !ok {
|
||||
@@ -86,7 +87,7 @@ func (ix *Indexer) HearVersionDelivered(ctx context.Context, d meshwants.Version
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return ix.keepNewest(ctx, DeliveredPrefix+d.Module, d, func(kept []byte) bool {
|
||||
return ix.keepNewest(ctx, DeliveredPrefix+keys.EscapeName(d.Module), d, func(kept []byte) bool {
|
||||
var k meshwants.VersionDeliveredEvent
|
||||
return json.Unmarshal(kept, &k) != nil || compareVersions(d.Version, k.Version) >= 0
|
||||
})
|
||||
@@ -95,7 +96,7 @@ func (ix *Indexer) HearVersionDelivered(ctx context.Context, d meshwants.Version
|
||||
// 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)
|
||||
raw, _, found, err := ix.Heard.Get(ctx, DeliveredPrefix+keys.EscapeName(module))
|
||||
if err != nil || !found || json.Unmarshal(raw, &d) != nil {
|
||||
return d, false
|
||||
}
|
||||
@@ -109,7 +110,7 @@ func (ix *Indexer) deliveredOf(ctx context.Context, module string) (meshwants.Ve
|
||||
// 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 {
|
||||
if ix.Versions == nil || ix.Heard == nil {
|
||||
return out, errors.New("the state versions is not kept here")
|
||||
}
|
||||
all, err := ix.Versions.Keys(ctx)
|
||||
|
||||
@@ -20,12 +20,13 @@ import (
|
||||
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 is the key of the state heard 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
|
||||
)
|
||||
|
||||
// versionsFillSince is the key of the state heard of the time builder_versions is first unanswered.
|
||||
var versionsFillSince = "call-since." + keys.EscapeName(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"`
|
||||
@@ -57,14 +58,14 @@ func versionNameOf(module string, a meshcatalog.Artifact) string {
|
||||
// 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 {
|
||||
if ix.Versions == nil || ix.Heard == 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 {
|
||||
if _, _, done, err := ix.Heard.Get(ctx, VersionsFilledKey); err != nil || done {
|
||||
return err
|
||||
}
|
||||
if ix.Revisions == nil {
|
||||
@@ -149,7 +150,7 @@ func (ix *Indexer) FillVersions(ctx context.Context, ask VersionsAsker) error {
|
||||
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) {
|
||||
if err := ix.Heard.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)
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -20,7 +21,7 @@ func TestAVersionIsReleasedOnceEveryArtifactIsPublished(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
|
||||
var said []meshcatalog.VersionReleasedEvent
|
||||
ix := &Indexer{Versions: newMem(), OwnBuildState: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now },
|
||||
ix := &Indexer{Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now },
|
||||
SayVersionReleased: func(_ context.Context, e meshcatalog.VersionReleasedEvent, id string) error {
|
||||
said = append(said, e)
|
||||
return nil
|
||||
@@ -50,7 +51,7 @@ func TestAVersionIsReleasedOnceEveryArtifactIsPublished(t *testing.T) {
|
||||
if len(v) != 1 || len(v[0].Artifacts) != 2 || v[0].Source != 7 || v[0].Late || len(said) != 1 || said[0].Version != "0.1.5" {
|
||||
t.Fatalf("released %+v, said %+v", v, said)
|
||||
}
|
||||
if keys, _ := ix.OwnBuildState.Keys(ctx); len(keys) != 0 {
|
||||
if keys, _ := ix.Heard.Keys(ctx); len(keys) != 0 {
|
||||
t.Errorf("the artifacts kept as published stay after the release: %v", keys)
|
||||
}
|
||||
// a late older version is recorded and not said
|
||||
@@ -90,7 +91,7 @@ func TestTheFirstFillOfTheVersionsComesFromTheBuildersPublished(t *testing.T) {
|
||||
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 },
|
||||
ix := &Indexer{Versions: newMem(), Heard: 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
|
||||
@@ -166,7 +167,7 @@ func TestTheFirstFillOfTheVersionsComesFromTheBuildersPublished(t *testing.T) {
|
||||
// makes nothing behind.
|
||||
func TestOutdatedAnswersTheEdgesBehindByTheirPartsDigest(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
ix := &Indexer{Versions: newMem(), OwnBuildState: newMem(), Revisions: newMemObjects(1 << 20)}
|
||||
ix := &Indexer{Versions: newMem(), Heard: 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"}}},
|
||||
@@ -240,7 +241,7 @@ func TestOutdatedAnswersTheEdgesBehindByTheirPartsDigest(t *testing.T) {
|
||||
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") {
|
||||
if raw, _, found, _ := ix.Heard.Get(ctx, ToolchainPrefix+"go"); !found || !strings.Contains(string(raw), "1.26") {
|
||||
t.Errorf("toolchain.go holds %s", raw)
|
||||
}
|
||||
}
|
||||
@@ -282,3 +283,27 @@ func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G A-80, §73: what the indexer heard lives in the state heard, each key part escaped as a name, so a dotted module
|
||||
// name parses and nothing of it is written to the own-build state.
|
||||
func TestWhatIsHeardIsKeptEscapedInTheStateHeard(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
ix := &Indexer{Versions: newMem(), OwnBuildState: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20)}
|
||||
if err := ix.HearVersionDelivered(ctx, meshwants.VersionDeliveredEvent{Module: "a.b", Version: "1.2.3"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ix.HearToolchainAgreed(ctx, meshbuilds.ToolchainAgreed{Language: "go", Version: "1.26"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := ix.deliveredOf(ctx, "a.b"); !ok {
|
||||
t.Error("the delivered version of a dotted module is not read back")
|
||||
}
|
||||
got, _ := ix.Heard.Keys(ctx)
|
||||
slices.Sort(got)
|
||||
if want := []string{"delivered.a=2eb", "toolchain.go"}; !slices.Equal(got, want) {
|
||||
t.Errorf("the keys of heard are %v, want %v", got, want)
|
||||
}
|
||||
if own, _ := ix.OwnBuildState.Keys(ctx); len(own) != 0 {
|
||||
t.Errorf("the own-build state holds %v", own)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user