Keep a fresh build's archives from the controller's own sweep (issue 1933)
mesh/delivery delivered

A sweep read what it keeps before what was ever recorded, so a take-in's
build recorded between the two reads, by a concurrent take-in off the
plans' hold, was eligible and let go of seconds after the take-in read
it; a new module's build is unkept until it is registered. Now the sweep
reads the records first and keeps what a build recorded within the hour
or an open plan names, and names every reference it lets go of; a take-in
holds every artifact in the store before recording, failing on one the
store lacks; and RecordBuild clears artifact_collected as RecordMirrored
does.
This commit is contained in:
2026-10-24 02:44:20 +02:00
parent fdf26a1fe2
commit 0741880d31
9 changed files with 431 additions and 12 deletions
+16 -7
View File
@@ -4,7 +4,9 @@ import (
"context"
"errors"
"fmt"
"io"
"os"
"strings"
"time"
"github.com/novox/mesh-controller/internal/artifacts"
@@ -215,27 +217,34 @@ func collect(ctx context.Context, inv *inventory.Inventory) {
}
r := sweep(ctx, inv, artifacts.Store{Address: address}, references, kept, afterBuild)
sayCollected(os.Stderr, r)
}
// sayCollected says what one sweep after a build did.
func sayCollected(w io.Writer, r sweepResult) {
if r.HoldersWritten > 0 {
fmt.Fprintf(os.Stderr, "the artifact store now holds %d more kept archive(s) by a manifest\n", r.HoldersWritten)
fmt.Fprintf(w, "the artifact store now holds %d more kept archive(s) by a manifest\n", r.HoldersWritten)
}
if r.Missing > 0 {
fmt.Fprintf(os.Stderr, "%d archive(s) the mesh keeps are not in the artifact store at all; "+
fmt.Fprintf(w, "%d archive(s) the mesh keeps are not in the artifact store at all; "+
"`collection` lists them\n", r.Missing)
}
if len(r.LetGo) > 0 {
fmt.Fprintf(os.Stderr, "the artifact store let go of %d artifact(s) the mesh no longer keeps\n", len(r.LetGo))
// **Each one named** (novox/hq issue 1933): a count could not say which archive a machine's 404 was let go of.
fmt.Fprintf(w, "the artifact store let go of %d artifact(s) the mesh no longer keeps: %s\n", len(r.LetGo),
strings.Join(r.LetGo, ", "))
}
if r.Stopped != "" && !r.Bounded {
fmt.Fprintln(os.Stderr, r.Stopped)
fmt.Fprintln(w, r.Stopped)
}
if r.Held && r.Left > 0 {
fmt.Fprintf(os.Stderr, "%d more to collect; the next build asks again\n", r.Left)
fmt.Fprintf(w, "%d more to collect; the next build asks again\n", r.Left)
}
if r.Skipped > 0 {
fmt.Fprintf(os.Stderr, "%d artifact(s) the sweep will not address were skipped\n", r.Skipped)
fmt.Fprintf(w, "%d artifact(s) the sweep will not address were skipped\n", r.Skipped)
}
if r.Indexes > 0 {
fmt.Fprintf(os.Stderr, "%d eligible index(es) wait for a collect a person confirms, which takes their platforms too\n", r.Indexes)
fmt.Fprintf(w, "%d eligible index(es) wait for a collect a person confirms, which takes their platforms too\n", r.Indexes)
}
}
+51 -1
View File
@@ -75,6 +75,32 @@ var readArchive = func(ctx context.Context, inv *inventory.Inventory, reference
return artifacts.Store{Address: address}.GetBlob(ctx, reference)
}
// holdArtifact holds what a key's reference names in the artifact store: Gone when the store does not hold it, an
// archive held by its manifest besides. A variable so a test needs no store.
var holdArtifact = func(ctx context.Context, inv *inventory.Inventory, reference string) error {
shelf, err := inv.Catalogue(ctx)
if err != nil {
return err
}
address, err := artifactStoreAddress(ctx, inv, shelf, "")
if err != nil {
return err
}
if address == "" {
// A mesh being raised has no store on its network yet (as the sweep, which collects nothing there): nothing
// can be held, and waiting for the store would keep the store itself from being taken in. Said, never silent.
fmt.Printf("the artifact store is on no network the controller reaches, so %s is taken in without being held\n",
reference)
return nil
}
store := artifacts.Store{Address: address}
if err := store.Holds(ctx, reference); err != nil {
return err
}
_, err = store.Hold(ctx, reference)
return err
}
// predefinedHead is the list of predefined keys at the head of the catalogue repository's trunk, and that head.
type predefinedHead struct {
List catalogue.PredefinedList
@@ -223,7 +249,8 @@ func takeInKey(ctx context.Context, inv *inventory.Inventory, keys keyReads, t k
return catalogue.Manifest{}, inventory.Build{}, err
}
kept, manifest, err := buildOfKey(ctx, keys, t, keyContext{list: head.List, previous: previous,
archive: func(ctx context.Context, reference string) ([]byte, error) { return readArchive(ctx, inv, reference) }})
archive: func(ctx context.Context, reference string) ([]byte, error) { return readArchive(ctx, inv, reference) },
hold: func(ctx context.Context, reference string) error { return holdArtifact(ctx, inv, reference) }})
if err != nil {
return catalogue.Manifest{}, kept, err
}
@@ -251,6 +278,8 @@ type keyContext struct {
list catalogue.PredefinedList
previous []inventory.Build
archive func(ctx context.Context, reference string) ([]byte, error)
// hold holds one artifact the key names in the artifact store, artifacts.Gone when the store does not hold it.
hold func(ctx context.Context, reference string) error
}
// keyPath is a module's directory as the catalogue records it: empty for a module at its repository's root.
@@ -318,6 +347,27 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) (
return inventory.Build{}, catalogue.Manifest{}, err
}
}
// **Every artifact held in the store before anything is recorded** (novox/hq issue 1933): a definition naming an
// archive the store does not hold is sent, and every machine assigned the module fails its apply on the 404. An
// archive is held by its manifest, so the store's own collector keeps it too.
if kc.hold != nil {
for _, a := range k.Build.Artifacts {
reference := catalogue.Recorded(a.Reference)
if _, ours := catalogue.InArtifactStore(reference); !ours {
continue
}
err := kc.hold(ctx, reference)
switch {
case err == nil, errors.Is(err, artifacts.ErrNotOurs):
case errors.Is(err, artifacts.Gone):
return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("the builder's key of %s names %s for its "+
"artifact %q, which is not in the artifact store: nothing was recorded or registered, and the key "+
"must be built again", t.Module, reference, a.Name)
default:
return waits("the artifact store could not hold %s's artifact %q (%s): %v", t.Module, a.Name, reference, err)
}
}
}
built, err := builtOfKey(ctx, kc.archive, manifest, k)
if err != nil {
return waits("%v", err)
@@ -0,0 +1,67 @@
package main
import (
"bytes"
"context"
"errors"
"strings"
"testing"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/artifacts"
)
// 1933/AC2: a take-in holds every artifact of the key in the artifact store before anything is recorded, and an archive
// the store does not hold fails it, "not in the artifact store", with no build to record: on 2026-10-24 00:37 the
// indexer's program blob was missing, the take-in registered it anyway and novox failed every apply on its 404.
func TestATakeInHoldsEveryArtifactAndFailsOnOneTheStoreLacks(t *testing.T) {
program := "artifact-store://x/tools/blobs/" + shaOf("t")
image := "artifact-store://x/img@" + shaOf("i")
keys, trunk := aKeyOf(tsBundle, 3,
sdkdelivery.BuiltArtifact{Name: "tools", Kind: "bundle", Digest: shaOf("t"), Reference: program},
sdkdelivery.BuiltArtifact{Name: "img", Kind: "image", Digest: shaOf("i"), Reference: image})
take := keyTake{Module: "x", Trunk: trunk}
var held []string
holds := func(_ context.Context, reference string) error { held = append(held, reference); return nil }
if _, _, err := buildOfKey(t.Context(), keys, take, keyContext{hold: holds}); err != nil {
t.Fatal(err)
}
if strings.Join(held, " ") != program+" "+image {
t.Errorf("the take-in held %v; want both of the key's artifacts", held)
}
lacks := func(_ context.Context, reference string) error {
if reference == program {
return artifacts.Gone
}
return nil
}
kept, _, err := buildOfKey(t.Context(), keys, take, keyContext{hold: lacks})
if err == nil || errors.Is(err, errKeyWaits) || !strings.Contains(err.Error(), "not in the artifact store") ||
!strings.Contains(err.Error(), program) {
t.Errorf("a take-in whose program archive the store lacks: %v; want a failure naming it, not in the artifact store", err)
}
if kept.ID != "" || len(kept.Made) > 0 {
t.Errorf("a take-in that failed made a build to record: %s %v", kept.ID, kept.Made)
}
unreachable := func(context.Context, string) error { return errors.New("connection refused") }
if _, _, err := buildOfKey(t.Context(), keys, take, keyContext{hold: unreachable}); !errors.Is(err, errKeyWaits) {
t.Errorf("a store that cannot be asked: %v; want a wait", err)
}
}
// 1933/AC1: the sweep after a build names every reference it let go of, not only how many, so a let-go that broke a
// machine is read from the controller's log.
func TestTheSweepSaysEveryReferenceItLetGoOf(t *testing.T) {
var out bytes.Buffer
sayCollected(&out, sweepResult{Held: true, LetGo: []string{
"artifact-store://route-proxy/files/blobs/sha256:58bd41dc", "artifact-store://old/app@sha256:01"}})
for _, reference := range []string{"artifact-store://route-proxy/files/blobs/sha256:58bd41dc",
"artifact-store://old/app@sha256:01"} {
if !strings.Contains(out.String(), reference) {
t.Errorf("the sweep's words do not name %s:\n%s", reference, out.String())
}
}
}
+24 -1
View File
@@ -138,7 +138,30 @@ func (i *Inventory) RecordBuild(ctx context.Context, b Build) error {
if err != nil {
return err
}
return i.chainRecordBuild(ctx, c, b)
if err := i.chainRecordBuild(ctx, c, b); err != nil {
return err
}
if !b.Worked() {
return nil
}
// **What a build says it made is held again** (novox/hq issue 1933), as a copy a build says it holds is
// (RecordMirrored): a build answers only once its artifacts are in the store, so one the records say was collected
// is in the store again. Left collected, it would read collected, not eligible, once nothing keeps it, and stay in the
// store and out of every sweep for ever.
var made []string
for _, a := range b.Made {
if a.Reference != "" {
made = append(made, a.Reference)
if recorded := asRecorded(a.Reference); recorded != a.Reference {
made = append(made, recorded)
}
}
}
if len(made) == 0 {
return nil
}
_, err = i.store.Pool().Exec(ctx, `delete from artifact_collected where reference = any($1)`, made)
return err
}
// RecordMirrored keeps that the mesh copied these bases into the artifact store (novox/hq ADR 0257):
+106 -3
View File
@@ -2,8 +2,10 @@ package inventory
import (
"context"
"slices"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
@@ -40,8 +42,17 @@ const (
// builds of a module the mesh still holds stood on it (novox/hq ADR 0257). A copy has no builds of
// its own; it is kept by the builds that used it.
KeptStoodOn = "stood-on"
// KeptByRecentRecord: a build recorded it within KeptLately, so a take-in or a build may still be registering
// it, and no definition or count of recent builds can speak for it yet (novox/hq issue 1933).
KeptByRecentRecord = "recorded-lately"
// KeptByOpenPlan: a plan that has not ended names the build that made it (novox/hq issue 1933).
KeptByOpenPlan = "open-plan"
)
// KeptLately is how long after a build is recorded its artifacts are kept whatever else the records say (novox/hq
// issue 1933).
const KeptLately = time.Hour
// The states an artifact the mesh recorded making is in (novox/hq ADR 0251).
const (
ArtifactKept = "kept"
@@ -76,15 +87,26 @@ type ArtifactState struct {
// In a stated order, so two runs over the same records ask for the same things in the same
// sequence, which is what makes a failed sweep safe to simply run again.
func (i *Inventory) Artifacts(ctx context.Context) ([]ArtifactState, error) {
keep, err := i.KeepSet(ctx)
return artifactStates(ctx, i.everyReferenceRecorded, i.KeepSet, i.alreadyCollected)
}
// artifactStates is Artifacts from its three reads, which a test gives.
func artifactStates(ctx context.Context, recorded func(context.Context) ([]string, error),
keepSet func(context.Context) (map[string][]string, error),
alreadyCollected func(context.Context) (map[string]bool, error)) ([]ArtifactState, error) {
// **What was recorded is read before what is kept** (novox/hq issue 1933). A build recorded between the two reads
// is then either not among what was recorded, or kept as recorded lately; read the other way round, a take-in's
// build landing between them was recorded and not kept, and the sweep let go of it seconds after the take-in read
// its archive.
all, err := recorded(ctx)
if err != nil {
return nil, err
}
all, err := i.everyReferenceRecorded(ctx)
keep, err := keepSet(ctx)
if err != nil {
return nil, err
}
collected, err := i.alreadyCollected(ctx)
collected, err := alreadyCollected(ctx)
if err != nil {
return nil, err
}
@@ -162,6 +184,16 @@ func (i *Inventory) KeepSet(ctx context.Context) (map[string][]string, error) {
if err != nil {
return nil, err
}
// **What a build recorded lately, or a plan that has not ended names** (novox/hq issue 1933).
lately, err := i.keptLatelyOrPlanned(ctx)
if err != nil {
return nil, err
}
for reference, whys := range lately {
for _, why := range whys {
because(reference, why)
}
}
var stoodOn []string
for _, r := range recent {
for _, a := range r.made {
@@ -388,6 +420,77 @@ func asRecorded(reference string) string {
return catalogue.Recorded(reference)
}
// keptLately is, of these builds, the artifacts kept because their build was recorded within the window before now, or is
// named by a plan that has not ended, each with its reasons.
//
// **Kept until the take-in or the build that recorded it is done** (novox/hq issue 1933). Between a build's record and
// its module's registration, nothing else speaks for it: a new module is not yet held, so no count of its recent builds
// keeps it, and no definition names it yet. A plan that has not ended may still send it, however long it runs.
func keptLately(builds []Build, planned map[string]bool, now time.Time, within time.Duration) map[string][]string {
out := map[string][]string{}
because := func(reference, why string) {
if !slices.Contains(out[reference], why) {
out[reference] = append(out[reference], why)
}
}
for _, b := range builds {
if !b.Worked() {
continue
}
lately := within > 0 && now.Sub(b.At) < within
named := planned[b.ID]
if !lately && !named {
continue
}
for _, a := range b.Made {
reference := asRecorded(a.Reference)
if reference == "" {
continue
}
if lately {
because(reference, KeptByRecentRecord)
}
if named {
because(reference, KeptByOpenPlan)
}
}
}
return out
}
// keptLatelyOrPlanned is keptLately of every build the records hold, against the plans that have not ended.
func (i *Inventory) keptLatelyOrPlanned(ctx context.Context) (map[string][]string, error) {
c, err := i.theChain(ctx)
if err != nil {
return nil, err
}
all, err := chainBuilds(ctx, c)
if err != nil {
return nil, err
}
plans, err := i.OpenPlans(ctx)
if err != nil {
return nil, err
}
return keptLately(all, plannedBuilds(plans), time.Now().UTC(), i.lately()), nil
}
// plannedBuilds is every build id these plans name: a module's build, and the builds a rollout carried.
func plannedBuilds(plans []Plan) map[string]bool {
out := map[string]bool{}
for _, p := range plans {
for _, m := range p.Modules {
if m != nil && m.Build != "" {
out[m.Build] = true
}
}
for _, id := range p.Builds {
out[id] = true
}
}
return out
}
// digestIn is the `sha256:<hex>` a reference names, empty when it names none.
func digestIn(reference string) string {
for _, marker := range []string{"@sha256:", "/sha256:"} {
@@ -0,0 +1,92 @@
package inventory
import (
"context"
"testing"
"time"
)
// 1933/AC1: a build recorded while a sweep reads the records is never offered to let go of. The record lands between the
// sweep's two reads (what it keeps, and what was ever recorded), whichever it reads first: on 2026-10-24 01:51 a
// concurrent take-in's route-proxy build landed between them, read as recorded and not kept, and was let go of seconds
// after the take-in read its archive.
func TestABuildRecordedWhileTheRecordsAreReadIsNotOffered(t *testing.T) {
const old, fresh = "artifact-store://x/app@sha256:01", "artifact-store://y/files/blobs/sha256:02"
reads, landed := 0, false
// read answers what the records hold at the moment of the read; the take-in's RecordBuild lands just after the
// first read, before the second.
read := func(before, after func()) {
if landed {
after()
} else {
before()
}
reads++
if reads == 1 {
landed = true
}
}
recorded := func(context.Context) (out []string, _ error) {
read(func() { out = []string{old} }, func() { out = []string{old, fresh} })
return out, nil
}
keepSet := func(context.Context) (out map[string][]string, _ error) {
// Once recorded, the keep set keeps it as recorded lately.
read(func() { out = map[string][]string{} }, func() { out = map[string][]string{fresh: {KeptByRecentRecord}} })
return out, nil
}
collected := func(context.Context) (map[string]bool, error) { return map[string]bool{}, nil }
states, err := artifactStates(t.Context(), recorded, keepSet, collected)
if err != nil {
t.Fatal(err)
}
for _, s := range states {
if s.Reference == fresh && s.State == ArtifactEligible {
t.Fatalf("a build recorded between the sweep's reads is offered to let go of: %+v", states)
}
}
}
// 1933/AC1: an artifact a build recorded within the last hour, or a build a plan that has not ended names, is kept
// whatever else the records say: a new module's first build is not registered yet, and a definition naming it is not
// written yet, while its take-in runs.
func TestAFreshBuildOrOneAnOpenPlanNamesIsKept(t *testing.T) {
now := time.Date(2026, 10, 24, 1, 52, 0, 0, time.UTC)
made := func(id, reference string, at time.Time) Build {
return Build{ID: id, Module: "route-proxy", At: at,
Made: []Artifact{{Name: "files", Kind: "archive", Reference: reference}}}
}
const (
fresh = "artifact-store://route-proxy/files/blobs/sha256:58bd41dc"
planned = "artifact-store://route-proxy/files/blobs/sha256:0ld91a"
stale = "artifact-store://route-proxy/files/blobs/sha256:5ta1e"
)
builds := []Build{
made("key-ee3448e0-1", fresh, now.Add(-10*time.Second)),
made("key-planned-1", planned, now.Add(-3*time.Hour)),
made("key-stale-1", stale, now.Add(-2*time.Hour)),
{ID: "failed-1", Module: "route-proxy", At: now, Failed: "no", Made: []Artifact{{Reference: "artifact-store://f@sha256:ff"}}},
}
kept := keptLately(builds, map[string]bool{"key-planned-1": true}, now, KeptLately)
if !has(kept[fresh], KeptByRecentRecord) {
t.Errorf("an archive recorded 10 s ago is not kept as recorded lately: %v", kept)
}
if !has(kept[planned], KeptByOpenPlan) || has(kept[planned], KeptByRecentRecord) {
t.Errorf("an archive an open plan names is not kept by that plan alone: %v", kept)
}
if len(kept[stale]) > 0 {
t.Errorf("an archive recorded two hours ago and named by no open plan is kept: %v", kept[stale])
}
if len(kept["artifact-store://f@sha256:ff"]) > 0 {
t.Errorf("a failed build keeps what it names: %v", kept)
}
}
func has(whys []string, why string) bool {
for _, w := range whys {
if w == why {
return true
}
}
return false
}
+55
View File
@@ -288,3 +288,58 @@ func TestKeptArchivesAreTheKeptBlobsOnly(t *testing.T) {
}
}
}
// 1933/AC1: a new module's first build, recorded and not yet registered (its take-in still runs), is not offered to
// let go of while it was recorded within the hour; turned off, the window shows what it guards.
func TestANewModulesBuildIsKeptWhileItsTakeInRuns(t *testing.T) {
inv := fresh(t)
ctx := context.Background()
made := built(t, inv, "key-new-1", "brand-new", 1)
inv.keptLately = 0 // KeptLately, as the controller runs
eligible, err := inv.ToCollect(ctx)
if err != nil {
t.Fatal(err)
}
for _, r := range eligible {
if r == made {
t.Fatalf("a build recorded a moment ago, its module not yet registered, is offered: %v", eligible)
}
}
keep, err := inv.KeepSet(ctx)
if err != nil {
t.Fatal(err)
}
if !has(keep[made], KeptByRecentRecord) {
t.Errorf("the keep set says %v of a build recorded a moment ago", keep[made])
}
inv.keptLately = -1
if eligible, err = inv.ToCollect(ctx); err != nil || len(eligible) != 1 || eligible[0] != made {
t.Errorf("without the window the unregistered module's build is offered (the fault of issue 1933): %v %v", eligible, err)
}
}
// 1933/AC3: a reference the records say was collected, and a build then records again, is no longer collected: it is
// held by the keep set and offered again once nothing keeps it.
func TestABuildRecordingACollectedReferenceHoldsItAgain(t *testing.T) {
inv := fresh(t)
ctx := context.Background()
holding(t, inv, "web")
reference := built(t, inv, "b01", "web", 1)
if err := inv.MarkCollected(ctx, []string{reference}); err != nil {
t.Fatal(err)
}
again := aBuild("b02", "web", "")
again.Made = []Artifact{{Name: "app", Kind: "image", Reference: reference}}
if err := inv.RecordBuild(ctx, again); err != nil {
t.Fatal(err)
}
collected, err := inv.alreadyCollected(ctx)
if err != nil {
t.Fatal(err)
}
if collected[reference] {
t.Errorf("a reference a build recorded again is still collected")
}
}
+6
View File
@@ -71,6 +71,9 @@ func ForTest(t *testing.T) *Inventory {
t.Fatal(err)
}
inv.UseChain(c)
// The keep set's other rules are tested with builds recorded a moment apart; a test of the rule that keeps what was
// recorded lately turns it on (novox/hq issue 1933).
inv.keptLately = -1
testChains.Store(os.Getenv(store.Variable(Name)), c)
t.Cleanup(func() { testChains.Delete(os.Getenv(store.Variable(Name))) })
return inv
@@ -136,6 +139,9 @@ func ForTestBefore(t *testing.T, number int) (*Inventory, func()) {
t.Fatal(err)
}
inv.UseChain(c)
// The keep set's other rules are tested with builds recorded a moment apart; a test of the rule that keeps what was
// recorded lately turns it on (novox/hq issue 1933).
inv.keptLately = -1
testChains.Store(os.Getenv(store.Variable(Name)), c)
t.Cleanup(func() { testChains.Delete(os.Getenv(store.Variable(Name))) })
return inv, func() {
+14
View File
@@ -42,6 +42,20 @@ type Inventory struct {
// throwaway is set on an inventory raised for a moment beside the real mesh (the merge gate's): nothing it
// composes is to be written to the store or the bus.
throwaway bool
// keptLately is how long a build's artifacts are kept after it is recorded (KeptLately, novox/hq issue 1933); zero
// is KeptLately, negative none (ForTest: the tests of the keep set's other rules record their builds a moment apart).
keptLately time.Duration
}
// lately is how long a build's artifacts are kept after it is recorded, whatever else the records say.
func (i *Inventory) lately() time.Duration {
switch {
case i.keptLately == 0:
return KeptLately
case i.keptLately < 0:
return 0
}
return i.keptLately
}
// Open connects to the inventory store.