carlos / platform Public

settings-4 task 9: fix round 1: unit rows never stop the sweep; sole mode and fleet reports; rotate the refresh

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 1 week ago 27939ab1be327acd25ccef1ae029db54f2110da3 parent 3a2b75f

10 files changed, +456 −82

internal/host/agent.go +16 −8
internal/host/agent_test.go +9 −28
internal/host/buildstate.go +20 −0
internal/host/buildstate_test.go +46 −0
internal/host/housekeeper.go +4 −1
internal/host/retention.go +116 −19
internal/host/retention_settings_test.go +204 −13
internal/host/retention_test.go +3 −3
internal/host/versionsweep.go +12 −3
internal/host/versionsweep_test.go +26 −7

diff --git a/internal/host/agent.go b/internal/host/agent.go

index 9fc30fb6c..a5b6d1b1f 100644

--- a/internal/host/agent.go

+++ b/internal/host/agent.go

@@ -649,7 +649,8 @@ func runAgent(args []string, buildClaims ClaimFactory) {

// keepsBuildState selects, with host claims as the build-state store

// sees them (cfg.HostWrites.Enabled, set by configureWakeClaims).

buildStateClaims := cfg != nil && cfg.HostWrites.Enabled

- go statusLoop(reg, svc, func() (map[string]status.BuildStatePins, error) { return buildStatePins(reg, buildStateClaims) })

+ collectBuildStates := func() (map[string]status.BuildStatePins, error) { return buildStatePins(reg, buildStateClaims) }

+ go statusLoop(reg, svc, collectBuildStates)

// The memory early-warning (memory-pressure spec §2.4): per box, gated

// by CARLOS_HOST_METRICS — one log line and nothing else where unset.

go memwatchLoop()

@@ -661,7 +662,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {

// the process's housekeeper and spawns its loop itself. Its stand-down

// runs in the post-drain block at the end of this function.

stopHousekeeper := startHousekeeper()

- go retentionLoop(reg)

+ go retentionLoop(reg, collectBuildStates)

go registrySnapshotLoop(reg)

go edgesFileLoop(reg)

// Region HA enrollment evidence (spec 1 §5.1): once, at start, and only

@@ -2998,9 +2999,12 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildSta

// within minutes, which is the promise the CLI makes ("boxes prune within

// ~10 minutes"). Thin and untested by convention (like domainsLoop /

// instancesLoop); everything with behavior lives in retentionPass. Missing

-// deployment config or a missing/unparseable release pubkey disables the

+// deployment config, a missing/unparseable release pubkey, or no box

+// reports to read (retentionReports: the status store, or on a sole

+// housekeeper with none this box's own build-state report) disables the

// loop with one log line — fail closed, domainsLoop's auto-purge shape: a

-// box that cannot verify pointers must not delete anything.

+// box that cannot verify pointers, or cannot learn which builds the build

+// states name (settings spec rev 11 §4.4), must not delete anything.

// Since the multi-region writer map (spec

// 2026-08-20-multi-region-ingress-design.md, Part 1) the pass itself is a

@@ -3011,7 +3015,7 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildSta

// silently, until the day it wins the lease (housekeeper.go). On a

// single-box deployment the holder is always this box, so the behaviour is

// unchanged.

-func retentionLoop(reg *registry.Registry) {

+func retentionLoop(reg *registry.Registry, buildStates func() (map[string]status.BuildStatePins, error)) {

ctx := context.Background()

// Platform boxes only (spec F6). Retention prunes the PLATFORM's release

// objects — a customer's box scheduling deletes of them was already wrong

@@ -3043,13 +3047,17 @@ func retentionLoop(reg *registry.Registry) {

}

// The box reports (settings spec rev 11 §4.4: the build-state builds

// every box names) live in the status store, which is not always the

- // deployment store above (CARLOS_STATUS_BUCKET wins). Without it no app

- // can be pruned safely, so the loop does not arm.

- reports, _, err := openStatusStore(ctx, os.Getenv)

+ // deployment store above (CARLOS_STATUS_BUCKET wins); a sole

+ // housekeeper with none reads its own (retentionReports). Without them

+ // no app can be pruned safely, so the loop does not arm.

+ reports, err := retentionReports(ctx, os.Getenv, store, buildStates)

if err != nil {

log.Printf("agent: retention loop disabled: the box reports cannot be read (%v). retention policies will not be enforced by this box", err)

return

}

+ if reports.local != nil {

+ log.Printf("agent: retention: no status store; as the sole housekeeper this box reads its own build-state report")

+ }

log.Printf("agent: retention loop armed (5min + jitter) — prunes only while this box holds the housekeeper lease")

// The pass runs under the HOLDING's context, not this loop's: a sweep

// that is halfway through its deletes when renewal fails must stop, not

diff --git a/internal/host/agent_test.go b/internal/host/agent_test.go

index c29ab77c7..879e75278 100644

--- a/internal/host/agent_test.go

+++ b/internal/host/agent_test.go

@@ -26,6 +26,7 @@ import (

"github.com/carlosframework/platform/internal/provision"

"github.com/carlosframework/platform/internal/registry"

"github.com/carlosframework/platform/internal/release"

+ "github.com/carlosframework/platform/internal/status"

)

// genKit is the shared setup for the generation-poll tests: a directory

@@ -1875,7 +1876,7 @@ func TestRetentionLoopIsAPlatformBoxJob(t *testing.T) {

t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")

t.Setenv("CARLOS_DEPLOYMENT_DIR", "")

- retentionLoop(testRegistry(t)) // must return; a fleet box never reaches the loop

+ retentionLoop(testRegistry(t), noBuildStates) // must return; a fleet box never reaches the loop

if got := logs(); !strings.Contains(got, skipLine) {

t.Fatalf("log = %q, want the F6 skip line verbatim", got)

@@ -1894,10 +1895,10 @@ func TestRetentionLoopIsAPlatformBoxJob(t *testing.T) {

t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")

t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir())

t.Setenv("CARLOS_RELEASE_PUBKEY", base64.StdEncoding.EncodeToString(pub))

- // The box reports retention reads (settings spec rev 11 §4.4) are in

- // the status store. Opening an S3 store makes no call, and the pass

- // runs only under the housekeeper lease, which a test never holds.

- t.Setenv("CARLOS_STATUS_BUCKET", "carlos-status")

+ // A directory-backed deployment publishes no status: it is its own

+ // sole housekeeper and reads this box's own build-state report

+ // (ruling R-T9-SOLE), so it still arms.

+ t.Setenv("CARLOS_STATUS_BUCKET", "")

// retentionLoop never returns once armed, so the arming log line is

// what proves it: run it in its own goroutine (it parks in a

@@ -1918,7 +1919,7 @@ func TestRetentionLoopIsAPlatformBoxJob(t *testing.T) {

// runs retentionPass a minute from now, which lists routes — nil

// would panic the whole test binary long after the test that started

// it has passed.

- go retentionLoop(testRegistry(t))

+ go retentionLoop(testRegistry(t), noBuildStates)

deadline := time.Now().Add(5 * time.Second)

for {

@@ -1939,28 +1940,8 @@ func TestRetentionLoopIsAPlatformBoxJob(t *testing.T) {

})

}

-// Settings spec rev 11 §4.4: without the box reports no app can be pruned

-// safely, so a deployment with no status store (CARLOS_DEPLOYMENT_DIR, which

-// publishes none) does not arm retention at all.

-func TestRetentionLoopNeedsTheBoxReports(t *testing.T) {

- pub, _, err := ed25519.GenerateKey(nil)

- if err != nil {

- t.Fatal(err)

- }

- logs := captureLog(t)

- t.Setenv("CARLOS_FLEET", "")

- t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")

- t.Setenv("CARLOS_STATUS_BUCKET", "")

- t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir())

- t.Setenv("CARLOS_RELEASE_PUBKEY", base64.StdEncoding.EncodeToString(pub))

- retentionLoop(testRegistry(t)) // must return: it does not arm

- got := logs()

- if !strings.Contains(got, "agent: retention loop disabled: the box reports cannot be read") || strings.Contains(got, "retention loop armed") {

- t.Fatalf("log = %q, want the loop disabled for want of box reports", got)

- }

+// noBuildStates is a build-state collector with nothing to report.

+func noBuildStates() (map[string]status.BuildStatePins, error) { return nil, nil }

// writerFunc adapts a func to io.Writer — the log sink above needs its own

// mutex because the loop under test keeps running (and may keep logging)

diff --git a/internal/host/buildstate.go b/internal/host/buildstate.go

index b56a011ff..58f1e4ef8 100644

--- a/internal/host/buildstate.go

+++ b/internal/host/buildstate.go

@@ -7,6 +7,7 @@ import (

"fmt"

"log"

"os"

+ "sort"

"strings"

"sync"

"time"

@@ -55,6 +56,9 @@ type buildStates struct {

mu sync.Mutex

hosts map[string]*sync.Mutex // per-host: a read or update and its column write are one step

+ // resume is the first host the last refreshAll did not reach; the next

+ // one starts there (ruling R-REFRESH-ROTATE). Guarded by mu.

+ resume string

}

// newBuildStates is nil, nil on a box with no replica bucket: it runs no

@@ -366,6 +370,11 @@ const refreshBudget = 20 * time.Second

// names, within the budget. Each host is located on its own: a location a

// caller's context carries names one wake's lease, so it is masked here. It

// returns how many hosts it did not reach.

+//

+// The hosts are taken in host order starting at the first one the last

+// call did not reach, wrapping round (ruling R-REFRESH-ROTATE): a box whose

+// budget runs out every pass reaches every host in turn, rather than

+// leaving the same tail unread (and its app unprunable) forever.

func (b *buildStates) refreshAll(ctx context.Context) int {

budget := b.budget

if budget <= 0 {

@@ -384,8 +393,19 @@ func (b *buildStates) refreshAll(ctx context.Context) int {

hosts = append(hosts, rt.Host)

}

+ sort.Strings(hosts)

+ b.mu.Lock()

+ start := sort.SearchStrings(hosts, b.resume)

+ b.mu.Unlock()

+ if start >= len(hosts) {

+ start = 0

+ }

+ hosts = append(hosts[start:len(hosts):len(hosts)], hosts[:start]...)

for i, host := range hosts {

if ctx.Err() != nil {

+ b.mu.Lock()

+ b.resume = host

+ b.mu.Unlock()

skipped := len(hosts) - i

log.Printf("agent: build state refresh: the %s budget ran out; %d of %d hosts not reached keep their last copy", budget, skipped, len(hosts))

return skipped

diff --git a/internal/host/buildstate_test.go b/internal/host/buildstate_test.go

index 91863757d..346d3809b 100644

--- a/internal/host/buildstate_test.go

+++ b/internal/host/buildstate_test.go

@@ -1028,3 +1028,49 @@ func TestAHeldWakeReadsTheCopyAsRootsTargetDoes(t *testing.T) {

t.Fatalf("a read copy with no last_executed and root's unknown = %+v; want unknown", h)

}

+// budgetEater records which host's build state each read asks for and

+// holds the read until the caller's context ends: the host it is reading

+// spends the whole refresh budget.

+type budgetEater struct {

+ release.Store

+ mu sync.Mutex

+ hosts []string

+func (e *budgetEater) Get(ctx context.Context, key string) ([]byte, string, error) {

+ if host, ok := strings.CutSuffix(strings.TrimPrefix(key, "inst/"), "/build-state.json"); ok {

+ e.mu.Lock()

+ e.hosts = append(e.hosts, host)

+ e.mu.Unlock()

+ <-ctx.Done()

+ return nil, "", ctx.Err()

+ }

+ return e.Store.Get(ctx, key)

+// Ruling R-REFRESH-ROTATE: a box whose refresh budget runs out does not

+// starve the same tail every pass. Each refreshAll starts at the first host

+// the last one did not reach, so every host is reached in turn.

+func TestRefreshAllRotatesPastWhereTheLastOneStopped(t *testing.T) {

+ b, s, reg := testBuildStates(t)

+ for _, h := range []string{"b.example.com", "c.example.com"} {

+ if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", App: "km", Hibernate: true, Socket: "/run/carlos/" + h + ".sock"}); err != nil {

+ t.Fatal(err)

+ }

+ eater := &budgetEater{Store: s}

+ b.storeFor = storesOf(map[string]release.Store{"hib": eater})

+ b.budget = 200 * time.Millisecond

+ for i := 0; i < 4; i++ {

+ if skipped := b.refreshAll(context.Background()); skipped != 2 {

+ t.Fatalf("pass %d skipped %d, want 2 (one host eats the budget)", i, skipped)

+ }

+ want := []string{"a.example.com", "b.example.com", "c.example.com", "a.example.com"}

+ eater.mu.Lock()

+ defer eater.mu.Unlock()

+ if strings.Join(eater.hosts, " ") != strings.Join(want, " ") {

+ t.Fatalf("passes read %v, want %v: each pass starts where the last stopped", eater.hosts, want)

+ }

diff --git a/internal/host/housekeeper.go b/internal/host/housekeeper.go

index cbd4502fb..944c6cd32 100644

--- a/internal/host/housekeeper.go

+++ b/internal/host/housekeeper.go

@@ -141,7 +141,10 @@ const (

// housekeeperSole: a CARLOS_DEPLOYMENT_DIR deployment. There is no

// bucket to hold a lock in, and a directory store is one box by

// construction, so this box IS the housekeeper — anything else would

- // silently stop dev/test deployments pruning at all.

+ // silently stop dev/test deployments pruning at all. Such a deployment

+ // publishes no status, so retention reads this box's own build-state

+ // report in its place (settings spec rev 11 §4.4, ruling R-T9-SOLE):

+ // one box, so its own report is every box's.

housekeeperSole

// housekeeperElect: a bucket-backed platform box. Contend.

housekeeperElect

diff --git a/internal/host/retention.go b/internal/host/retention.go

index 1daeaa424..9fd220e61 100644

--- a/internal/host/retention.go

+++ b/internal/host/retention.go

@@ -29,18 +29,18 @@ import (

// error is logged and skipped, never allowed to stop another app's

// enforcement; a partial delete leaves the remainder for the next tick.

-// reports is the status store the boxes publish to (openStatusStore's

-// bucket, not store's): settings spec rev 11 §4.4's build-state builds come

-// from there. A pass that cannot read the reports prunes nothing at all.

-func retentionPass(ctx context.Context, store, reports release.Store, pub ed25519.PublicKey, reg *registry.Registry) error {

+// reports is where the box reports are read (settings spec rev 11 §4.4's

+// build-state builds; retentionReports). A pass that cannot read them

+// prunes nothing at all.

+func retentionPass(ctx context.Context, store release.Store, reports *boxReports, pub ed25519.PublicKey, reg *registry.Registry) error {

routes, err := reg.List()

if err != nil {

return err

}

if reports == nil {

- return errors.New("no status store to read the box reports from: pruning nothing")

+ return errors.New("no box reports to read: pruning nothing")

}

- boxes, complete, err := loadAllBoxes(ctx, reports)

+ boxes, complete, err := reports.load(ctx)

if err != nil {

return fmt.Errorf("reading the box reports: pruning nothing this pass: %w", err)

}

@@ -158,28 +158,125 @@ func buildStateBuilds(st *release.BuildState) []string {

// hosting one of its instances has been silent longer than this.

const buildStateReportMaxAge = 6 * time.Hour

-// loadAllBoxes reads every box report from the status store: the

-// deployment's (status/boxes/) and every fleet's (fleets/<name>/status/boxes/,

-// enumerated with ListDirs). complete is false if any report could not be

-// read; a listing that fails is an error.

-func loadAllBoxes(ctx context.Context, store release.Store) ([]status.Box, bool, error) {

- boxes, complete, err := status.LoadBoxesComplete(ctx, store)

- if err != nil {

- return nil, false, err

+// boxReports is where a retention pass reads the box reports (settings spec

+// rev 11 §4.4). At least one of status and local is set.

+type boxReports struct {

+ // status is the store the boxes publish to (openStatusStore's bucket);

+ // nil for a sole housekeeper with none.

+ status release.Store

+ // prefix rebases the platform's own reports in status, as

+ // CARLOS_STATUS_PREFIX rebases what the status loop writes

+ // (resolveStatusPrefix). Fleet reports are not rebased: a fleet box

+ // writes at fleets/<fleet>/ from the bucket root.

+ prefix string

+ // deploy is the deployment store, read for fleets/ as well (ruling

+ // R-T9-FLEETS): a fleet box with no bucket publishes through the store

+ // door, which serves the deployment store. nil when it is the status

+ // store's own bucket.

+ deploy release.Store

+ // local is this box's own collector, read as a fresh report at every

+ // pass: set for a sole housekeeper (a directory-backed deployment, one

+ // box by construction, which publishes no status; ruling R-T9-SOLE).

+ local func() (map[string]status.BuildStatePins, error)

+ // now stamps the local report; nil is time.Now.

+ now func() time.Time

+// soleReportLabel names a sole housekeeper's own report in refusals.

+const soleReportLabel = "this box (sole housekeeper)"

+// retentionReports is retentionLoop's report source, from the environment:

+// the status store (rebased by CARLOS_STATUS_PREFIX) and, for fleets, the

+// deployment store too when it is another bucket; or, on a sole

+// housekeeper with no status store, this box's own collector and the

+// deployment store's fleets. Anything else is an error: no app can be

+// pruned without the reports.

+func retentionReports(ctx context.Context, getenv func(string) string, deploy release.Store, local func() (map[string]status.BuildStatePins, error)) (*boxReports, error) {

+ st, _, err := openStatusStore(ctx, getenv)

+ switch {

+ case err == nil:

+ r := &boxReports{status: st, prefix: resolveStatusPrefix(getenv), deploy: deploy}

+ if b := getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" && b == resolveStatusBucket(getenv) {

+ r.deploy = nil // the same bucket: its fleets/ is read once

+ }

+ return r, nil

+ case errors.Is(err, errNoStatusStore) && newHousekeeper(getenv, nil).mode == housekeeperSole && local != nil:

+ return &boxReports{deploy: deploy, local: local}, nil

+ }

+ return nil, err

+// load reads every box report: the platform's (status/boxes/ under prefix,

+// or this box's own on a sole housekeeper), and every fleet's in the status

+// store and in the deployment store, the union (a report in both is read

+// twice; the newest-per-host rule is indifferent to that). complete is

+// false if any report or fleets listing could not be read. Failing to list

+// the platform's own reports is an error.

+func (r *boxReports) load(ctx context.Context) ([]status.Box, bool, error) {

+ if r == nil || (r.status == nil && r.local == nil) {

+ return nil, false, errors.New("no box reports to read")

+ }

+ var boxes []status.Box

+ complete := true

+ if r.status != nil {

+ root, ok, err := status.LoadBoxesComplete(ctx, release.NewPrefixStore(r.status, r.prefix))

+ if err != nil {

+ return nil, false, err

+ }

+ boxes, complete = root, ok

}

+ if r.local != nil {

+ boxes = append(boxes, r.localReport())

+ }

+ for _, s := range []release.Store{r.status, r.deploy} {

+ if s == nil {

+ continue

+ }

+ fb, ok := loadFleetReports(ctx, s)

+ complete = complete && ok

+ boxes = append(boxes, fb...)

+ }

+ return boxes, complete, nil

+// localReport is this box's own report, taken now, as the status loop

+// would publish it: BuildStatesErr when the collector fails.

+func (r *boxReports) localReport() status.Box {

+ now := time.Now

+ if r.now != nil {

+ now = r.now

+ }

+ m := &status.BoxMetrics{}

+ if pins, err := r.local(); err != nil {

+ m.BuildStatesErr = err.Error()

+ } else {

+ m.BuildStates = pins

+ }

+ return status.Box{Label: soleReportLabel, GeneratedAt: now(), Metrics: m}

+// loadFleetReports reads every fleet's reports in store

+// (fleets/<name>/status/boxes/, enumerated with ListDirs). ok is false when

+// the listing or any report could not be read.

+func loadFleetReports(ctx context.Context, store release.Store) ([]status.Box, bool) {

fleets, err := store.ListDirs(ctx, "fleets/")

if err != nil {

- return nil, false, fmt.Errorf("listing fleets: %w", err)

+ log.Printf("agent: retention: listing fleets: %v", err)

+ return nil, false

}

+ var boxes []status.Box

+ ok := true

for _, f := range fleets {

- fb, ok, err := status.LoadBoxesComplete(ctx, release.NewPrefixStore(store, f))

+ fb, complete, err := status.LoadBoxesComplete(ctx, release.NewPrefixStore(store, f))

if err != nil {

- return nil, false, fmt.Errorf("reading the reports of %s: %w", f, err)

+ log.Printf("agent: retention: reading the reports of %s: %v", f, err)

+ ok = false

+ continue

}

- complete = complete && ok

+ ok = ok && complete

boxes = append(boxes, fb...)

}

- return boxes, complete, nil

+ return boxes, ok

}

// appRecords reads every provision record of acct/app, or fails. It lists

diff --git a/internal/host/retention_settings_test.go b/internal/host/retention_settings_test.go

index 93f75f2a2..a838722b7 100644

--- a/internal/host/retention_settings_test.go

+++ b/internal/host/retention_settings_test.go

@@ -342,8 +342,8 @@ func TestTheReportNamesExactlyTheRowsThatKeepABuildState(t *testing.T) {

}

-// loadAllBoxes reads the deployment's reports and every fleet's, and says

-// when any could not be read.

+// The reports are the deployment's and every fleet's, and load says when any

+// could not be read.

func TestLoadAllBoxesReadsEveryFleet(t *testing.T) {

ctx := context.Background()

s := release.NewDirStore(t.TempDir())

@@ -353,9 +353,9 @@ func TestLoadAllBoxesReadsEveryFleet(t *testing.T) {

if err := status.WriteBox(ctx, release.NewPrefixStore(s, "fleets/acme/"), report("acme-1", now0, map[string]status.BuildStatePins{"b.example.com": kmPins("v2")})); err != nil {

t.Fatal(err)

}

- boxes, complete, err := loadAllBoxes(ctx, s)

+ boxes, complete, err := (&boxReports{status: s}).load(ctx)

if err != nil || !complete {

- t.Fatalf("loadAllBoxes = %v, %v", complete, err)

+ t.Fatalf("load = %v, %v", complete, err)

}

var labels []string

for _, b := range boxes {

@@ -369,13 +369,21 @@ func TestLoadAllBoxesReadsEveryFleet(t *testing.T) {

if _, err := s.Put(ctx, "fleets/beta/status/boxes/beta-1.json", []byte("{"), ""); err != nil {

t.Fatal(err)

}

- if _, complete, err := loadAllBoxes(ctx, s); err != nil || complete {

+ if _, complete, err := (&boxReports{status: s}).load(ctx); err != nil || complete {

t.Fatalf("a corrupt fleet report read as complete (%v, %v)", complete, err)

}

- fleetsFail := failingListStore{Store: s, prefix: "fleets/", err: errors.New("AccessDenied")}

- if _, _, err := loadAllBoxes(ctx, fleetsFail); err == nil {

- t.Fatal("a failed fleets listing was not an error")

+ // R-T9-FLEETS: a failed fleets listing is an incomplete read.

+ fleetsFail := failingListStore{Store: release.NewDirStore(t.TempDir()), prefix: "fleets/", err: errors.New("AccessDenied")}

+ if _, complete, err := (&boxReports{status: fleetsFail}).load(ctx); err != nil || complete {

+ t.Fatalf("a failed fleets listing read as complete (%v, %v)", complete, err)

+ }

+ rootFail := failingListStore{Store: release.NewDirStore(t.TempDir()), prefix: "status/boxes/", err: errors.New("AccessDenied")}

+ if _, _, err := (&boxReports{status: rootFail}).load(ctx); err == nil {

+ t.Fatal("a failed listing of the platform's reports was not an error")

+ }

+ if _, _, err := (&boxReports{}).load(ctx); err == nil {

+ t.Fatal("no report source at all was not an error")

}

@@ -430,7 +438,7 @@ func TestRetentionPassProtectsPinsGoodVersionAndReportedBuilds(t *testing.T) {

"kass.example.com": {Account: "bab", App: "kass", Builds: []string{"v03", "v02"}, Complete: true}})); err != nil {

t.Fatal(err)

}

- if err := retentionPass(ctx, s, reports, pub, reg); err != nil {

+ if err := retentionPass(ctx, s, &boxReports{status: reports}, pub, reg); err != nil {

t.Fatal(err)

}

if present("v01") {

@@ -449,7 +457,7 @@ func TestRetentionPassPrunesNothingWithoutTheReports(t *testing.T) {

t.Run("no report names the host", func(t *testing.T) {

ctx, s, reg, present := retentionFixture(t)

_, pub := retentionKeys(t)

- if err := retentionPass(ctx, s, release.NewDirStore(t.TempDir()), pub, reg); err != nil {

+ if err := retentionPass(ctx, s, &boxReports{status: release.NewDirStore(t.TempDir())}, pub, reg); err != nil {

t.Fatal(err)

}

if !present("v01") {

@@ -460,17 +468,200 @@ func TestRetentionPassPrunesNothingWithoutTheReports(t *testing.T) {

ctx, s, reg, present := retentionFixture(t)

_, pub := retentionKeys(t)

reports := failingListStore{Store: release.NewDirStore(t.TempDir()), prefix: "status/boxes/", err: errors.New("AccessDenied")}

- _ = retentionPass(ctx, s, reports, pub, reg)

+ _ = retentionPass(ctx, s, &boxReports{status: reports}, pub, reg)

if !present("v01") {

t.Fatal("v01 was pruned with the box reports unreadable")

}

})

- t.Run("no status store", func(t *testing.T) {

+ t.Run("no report source", func(t *testing.T) {

ctx, s, reg, present := retentionFixture(t)

_, pub := retentionKeys(t)

_ = retentionPass(ctx, s, nil, pub, reg)

+ _ = retentionPass(ctx, s, &boxReports{}, pub, reg)

+ if !present("v01") {

+ t.Fatal("v01 was pruned with no report source")

+ }

+ })

+// kassPins is retentionFixture's host as this box's own collector reports it.

+func kassPins(builds ...string) func() (map[string]status.BuildStatePins, error) {

+ return func() (map[string]status.BuildStatePins, error) {

+ return map[string]status.BuildStatePins{"kass.example.com": {Account: "bab", App: "kass", Builds: builds, Complete: true}}, nil

+ }

+// Ruling R-T9-SOLE: a directory-backed deployment is one box by

+// construction and publishes no status. Its pass takes this box's own

+// collector as a fresh report: what it names is protected, and a collector

+// that fails, or that omits a provisioned host, prunes nothing.

+func TestSoleModeReadsThisBoxsOwnReport(t *testing.T) {

+ t.Run("protects what this box names", func(t *testing.T) {

+ ctx, s, reg, present := retentionFixture(t)

+ _, pub := retentionKeys(t)

+ if err := retentionPass(ctx, s, &boxReports{deploy: s, local: kassPins("v03", "v02")}, pub, reg); err != nil {

+ t.Fatal(err)

+ }

+ if present("v01") {

+ t.Error("v01 survived; nothing names it")

+ }

+ for _, v := range []string{"v02", "v03", "v04", "v05", "v06", "v07"} {

+ if !present(v) {

+ t.Errorf("%s was pruned", v)

+ }

+ })

+ t.Run("a failed collection", func(t *testing.T) {

+ ctx, s, reg, present := retentionFixture(t)

+ _, pub := retentionKeys(t)

+ failing := func() (map[string]status.BuildStatePins, error) { return nil, errors.New("database is locked") }

+ boxes, _, err := (&boxReports{local: failing}).load(ctx)

+ if err != nil || len(boxes) != 1 || boxes[0].Metrics == nil || boxes[0].Metrics.BuildStatesErr != "database is locked" {

+ t.Fatalf("load = %+v, %v; want this box's report saying it could not collect", boxes, err)

+ }

+ _ = retentionPass(ctx, s, &boxReports{deploy: s, local: failing}, pub, reg)

+ if !present("v01") {

+ t.Fatal("v01 was pruned while this box could not collect its build states")

+ }

+ })

+ t.Run("the host is not named", func(t *testing.T) {

+ ctx, s, reg, present := retentionFixture(t)

+ _, pub := retentionKeys(t)

+ none := func() (map[string]status.BuildStatePins, error) { return nil, nil }

+ _ = retentionPass(ctx, s, &boxReports{deploy: s, local: none}, pub, reg)

if !present("v01") {

- t.Fatal("v01 was pruned with no status store to read reports from")

+ t.Fatal("v01 was pruned with kass.example.com named by no report")

+ }

+ })

+ t.Run("fleets come from the deployment store", func(t *testing.T) {

+ ctx := context.Background()

+ deploy := release.NewDirStore(t.TempDir())

+ if err := status.WriteBox(ctx, release.NewPrefixStore(deploy, "fleets/acme/"), report("acme-1", now0, map[string]status.BuildStatePins{"b.example.com": kmPins("v2")})); err != nil {

+ t.Fatal(err)

+ }

+ boxes, complete, err := (&boxReports{deploy: deploy, local: kassPins("v03"), now: func() time.Time { return now0 }}).load(ctx)

+ if err != nil || !complete || len(boxes) != 2 {

+ t.Fatalf("load = %+v, %v, %v; want this box and the fleet's", boxes, complete, err)

+ }

+ var local status.Box

+ for _, b := range boxes {

+ if b.Label != "acme-1" {

+ local = b

+ }

+ if !local.GeneratedAt.Equal(now0) || local.Metrics == nil || local.Metrics.BuildStates["kass.example.com"].App != "kass" {

+ t.Fatalf("this box's report = %+v, want it fresh and naming kass.example.com", local)

}

})

}

+// Ruling R-T9-FLEETS: fleet reports are read from the status store and from

+// the deployment store (a fleet box that publishes through the store door

+// lands in the deployment bucket), and the union is used. A report found in

+// both is read twice, which the newest-per-host rule tolerates. A failed

+// listing or read in either store is an incomplete read.

+func TestFleetReportsComeFromBothStores(t *testing.T) {

+ ctx := context.Background()

+ st, dep := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())

+ both := report("both-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"a.example.com": kmPins("v1")})

+ for _, w := range []struct {

+ s release.Store

+ b status.Box

+ }{

+ {release.NewPrefixStore(st, "fleets/acme/"), report("acme-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"b.example.com": kmPins("v2")})},

+ {release.NewPrefixStore(dep, "fleets/door/"), report("door-1", now0.Add(-time.Minute), map[string]status.BuildStatePins{"c.example.com": kmPins("v3")})},

+ {release.NewPrefixStore(st, "fleets/same/"), both},

+ {release.NewPrefixStore(dep, "fleets/same/"), both},

+ } {

+ if err := status.WriteBox(ctx, w.s, w.b); err != nil {

+ t.Fatal(err)

+ }

+ boxes, complete, err := (&boxReports{status: st, deploy: dep}).load(ctx)

+ if err != nil || !complete {

+ t.Fatalf("load = %v, %v", complete, err)

+ }

+ var labels []string

+ for _, b := range boxes {

+ labels = append(labels, b.Label)

+ }

+ sort.Strings(labels)

+ if !reflect.DeepEqual(labels, []string{"acme-1", "both-1", "both-1", "door-1"}) {

+ t.Fatalf("read %v, want the union of both stores' fleets", labels)

+ }

+ s := recordsStore(t,

+ provision.Record{Schema: 1, Host: "a.example.com", App: "km", Account: "bab", Channel: "stable"},

+ provision.Record{Schema: 1, Host: "b.example.com", App: "km", Account: "bab", Channel: "stable"},

+ provision.Record{Schema: 1, Host: "c.example.com", App: "km", Account: "bab", Channel: "stable"})

+ got, err := settingsProtected(ctx, s, boxes, complete, now0, "bab", "km")

+ if err != nil || !got["v1"] || !got["v2"] || !got["v3"] {

+ t.Fatalf("got %v, %v; want every fleet's builds and no refusal", got, err)

+ }

+ deployListFails := failingListStore{Store: dep, prefix: "fleets/", err: errors.New("AccessDenied")}

+ if _, complete, err := (&boxReports{status: st, deploy: deployListFails}).load(ctx); err != nil || complete {

+ t.Fatalf("a failed fleets listing in the deployment store read as complete (%v, %v)", complete, err)

+ }

+ deployGetFails := failingGetStore{Store: dep, bad: "fleets/door/status/boxes/door-1.json", err: errors.New("SlowDown")}

+ if _, complete, err := (&boxReports{status: st, deploy: deployGetFails}).load(ctx); err != nil || complete {

+ t.Fatalf("a failed fleet report read in the deployment store read as complete (%v, %v)", complete, err)

+ }

+ statusListFails := failingListStore{Store: st, prefix: "fleets/", err: errors.New("AccessDenied")}

+ if _, complete, err := (&boxReports{status: statusListFails, deploy: dep}).load(ctx); err != nil || complete {

+ t.Fatalf("a failed fleets listing in the status store read as complete (%v, %v)", complete, err)

+ }

+// CARLOS_STATUS_PREFIX rebases the platform's own reports, as the status

+// loop writes them; fleet reports stay at fleets/<name>/ from the bucket

+// root, where fleet boxes write them.

+func TestTheStatusPrefixRebasesThePlatformReports(t *testing.T) {

+ ctx := context.Background()

+ s := release.NewDirStore(t.TempDir())

+ if err := status.WriteBox(ctx, release.NewPrefixStore(s, "x/"), report("uk-1", now0, nil)); err != nil {

+ t.Fatal(err)

+ }

+ if err := status.WriteBox(ctx, s, report("someone-else", now0, nil)); err != nil {

+ t.Fatal(err)

+ }

+ if err := status.WriteBox(ctx, release.NewPrefixStore(s, "fleets/acme/"), report("acme-1", now0, nil)); err != nil {

+ t.Fatal(err)

+ }

+ boxes, complete, err := (&boxReports{status: s, prefix: "x/"}).load(ctx)

+ if err != nil || !complete {

+ t.Fatalf("load = %v, %v", complete, err)

+ }

+ var labels []string

+ for _, b := range boxes {

+ labels = append(labels, b.Label)

+ }

+ sort.Strings(labels)

+ if !reflect.DeepEqual(labels, []string{"acme-1", "uk-1"}) {

+ t.Fatalf("read %v, want the prefixed platform report and the fleet's", labels)

+ }

+// retentionReports is retentionLoop's choice of report source, from the

+// environment.

+func TestRetentionReportsFromTheEnvironment(t *testing.T) {

+ ctx := context.Background()

+ deploy := release.NewDirStore(t.TempDir())

+ local := kassPins("v1")

+ env := func(m map[string]string) func(string) string { return func(k string) string { return m[k] } }

+ r, err := retentionReports(ctx, env(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "deploy", "CARLOS_STATUS_BUCKET": "status", "CARLOS_STATUS_PREFIX": "x/"}), deploy, local)

+ if err != nil || r.status == nil || r.prefix != "x/" || r.deploy == nil || r.local != nil {

+ t.Fatalf("separate status bucket: %+v, %v; want status, prefix x/, the deployment store for fleets, no local report", r, err)

+ }

+ r, err = retentionReports(ctx, env(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "deploy"}), deploy, local)

+ if err != nil || r.status == nil || r.deploy != nil || r.local != nil {

+ t.Fatalf("status in the deployment bucket: %+v, %v; want the bucket's fleets read once", r, err)

+ }

+ r, err = retentionReports(ctx, env(map[string]string{"CARLOS_DEPLOYMENT_DIR": "/srv/deploy"}), deploy, local)

+ if err != nil || r.status != nil || r.deploy == nil || r.local == nil {

+ t.Fatalf("directory-backed: %+v, %v; want this box's own report and the deployment store's fleets", r, err)

+ }

+ if _, err := retentionReports(ctx, env(map[string]string{}), deploy, local); err == nil {

+ t.Fatal("no status store and not a sole housekeeper: want an error")

+ }

diff --git a/internal/host/retention_test.go b/internal/host/retention_test.go

index ae9547ae2..775bbc2ed 100644

--- a/internal/host/retention_test.go

+++ b/internal/host/retention_test.go

@@ -70,7 +70,7 @@ func TestRetentionPassPrunesEligibleAndProtectsBoxPins(t *testing.T) {

t.Fatal(err)

}

- if err := retentionPass(ctx, s, release.NewDirStore(t.TempDir()), pub, reg); err != nil {

+ if err := retentionPass(ctx, s, &boxReports{status: release.NewDirStore(t.TempDir())}, pub, reg); err != nil {

t.Fatal(err)

}

@@ -123,7 +123,7 @@ func TestRetentionPassResolvesEmptyAccountRoutesViaRouteAccount(t *testing.T) {

t.Fatal(err)

}

- if err := retentionPass(ctx, s, release.NewDirStore(t.TempDir()), pub, reg); err != nil {

+ if err := retentionPass(ctx, s, &boxReports{status: release.NewDirStore(t.TempDir())}, pub, reg); err != nil {

t.Fatal(err)

}

@@ -164,7 +164,7 @@ func TestRetentionPassSkipsAppsWithoutAPolicy(t *testing.T) {

}); err != nil {

t.Fatal(err)

}

- if err := retentionPass(ctx, s, release.NewDirStore(t.TempDir()), pub, reg); err != nil {

+ if err := retentionPass(ctx, s, &boxReports{status: release.NewDirStore(t.TempDir())}, pub, reg); err != nil {

t.Fatal(err)

}

if _, _, err := s.Get(ctx, release.ManifestKey("bab", "kass", "v01")); err != nil {

diff --git a/internal/host/versionsweep.go b/internal/host/versionsweep.go

index 53a023125..0d5c65a1a 100644

--- a/internal/host/versionsweep.go

+++ b/internal/host/versionsweep.go

@@ -6,7 +6,6 @@ import (

"os"

"time"

- "github.com/carlosframework/platform/internal/activator"

"github.com/carlosframework/platform/internal/registry"

"github.com/carlosframework/platform/internal/release"

)

@@ -134,7 +133,9 @@ func sweepStagedVersionsPass(reg routeLister) {

// failed refresh still names it. A column that does not parse, or a held

// instance (settings other than default) whose build state is not read

// yet, is an incomplete pin set: its last_executed is unknown, so the

-// sweep deletes nothing.

+// sweep deletes nothing. Only a row whose copy the agent refreshes counts:

+// one keepsBuildState never selects (a named unit, the platform's own row)

+// stays unread forever and must not stop the sweep for good.

func settingsPins(routes []registry.Route) (map[string]bool, error) {

out := map[string]bool{}

for _, r := range routes {

@@ -149,7 +150,7 @@ func settingsPins(routes []registry.Route) (map[string]bool, error) {

if err != nil {

return nil, fmt.Errorf("build state of %s: %w", r.Host, err)

}

- if !c.Known && !u.Default() && activator.ActivatorManaged(r) {

+ if !c.Known && !u.Default() && heldWithBuildState(r) {

return nil, fmt.Errorf("the build state of held %s is not read yet", r.Host)

}

for _, v := range buildStateBuilds(c.State) {

@@ -158,3 +159,11 @@ func settingsPins(routes []registry.Route) (map[string]bool, error) {

}

return out, nil

}

+// heldWithBuildState is a row settings can hold whose build state the agent

+// refreshes: keepsBuildState's rows (with host claims off, the wider set:

+// root does not know the agent's claims mode) less workers, which settings

+// never apply to, so no held recovery reads their last_executed.

+func heldWithBuildState(r registry.Route) bool {

+ return keepsBuildState(r, false) && !r.IsSidecar()

diff --git a/internal/host/versionsweep_test.go b/internal/host/versionsweep_test.go

index 6960c0e83..2942f1d7a 100644

--- a/internal/host/versionsweep_test.go

+++ b/internal/host/versionsweep_test.go

@@ -336,12 +336,31 @@ func TestSweepStagedVersionsDeletesNothingWhenSettingsPinsAreUnknown(t *testing.

}

})

}

- // Default settings with an unread copy: nothing is held on its

- // last_executed, so the sweep goes on.

- opt := t.TempDir()

- stageSweepVersion(t, opt, "acct1", "v1", "blog", 48*time.Hour)

- sweepStagedVersions(fakeLister{routes: []registry.Route{settingsRow("a.example.com", "", "")}}, opt, noGenerations(t), 6*time.Hour, time.Now())

- if staged(t, opt, "v1") {

- t.Error("a default-settings row with an unread copy stopped the sweep")

+ // Rows whose unread copy says nothing the sweep needs: default settings

+ // (nothing is held on its last_executed), and rows keepsBuildState never

+ // refreshes, a named unit or the platform's own row, whatever settings

+ // an app policy gives them (their copy stays unread forever). The sweep

+ // goes on.

+ unit := settingsRow("u.example.com", "", manual)

+ unit.Unit = "blog.service"

+ platform := settingsRow("p.example.com", "", manual)

+ platform.Platform = true

+ worker := settingsRow("w.example.com", "", manual)

+ worker.Kind, worker.Hibernate = "sidecar", false

+ goesOn := map[string]registry.Route{

+ "worker, settings set": worker,

+ "default settings": settingsRow("a.example.com", "", ""),

+ "named unit, held": unit,

+ "platform row, held": platform,

+ }

+ for name, rt := range goesOn {

+ t.Run(name, func(t *testing.T) {

+ opt := t.TempDir()

+ stageSweepVersion(t, opt, "acct1", "v1", "blog", 48*time.Hour)

+ sweepStagedVersions(fakeLister{routes: []registry.Route{rt}}, opt, noGenerations(t), 6*time.Hour, time.Now())

+ if staged(t, opt, "v1") {

+ t.Error("a row whose unread copy pins nothing stopped the sweep")

+ }

+ })

}