carlos / platform Public

console: the seeds sweep (Plan 8 Task 2)

StartSeedsPass, a claim worker every five minutes (spec §3.1, §5.6):
snapshots listed before records (M2), records strictly, any listing error
ends the pass; unreadable hosts and hosts two records hold are kept whole.
Stale reserving --from (1h) and awaiting-upload/uploaded imports (7d) are
abandoned by provision.SeedAbortStale, a CAS from the exact state read
(M5); a snapshot is deleted when its own seed is placed, committed,
aborted or rolled-back (M6), when another placed/committed seed superseded
it, or when no record names it and it is over an hour old; rolled-back
seeds are finished after the decisions. Off on a host-claim console and in
every claim inventory (M3). finishSeed's sharing check reads strictly and
counts an unreadable record as a sharer (M7).

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

claude 2 weeks ago 931b14c042b9ca723400dd89bdff02849ba0d8a7 parent 580cd4a

11 files changed, +829 −14

cmd/console/main.go +5 −0
internal/console/claim_invariant_fixtures_test.go +1 −0
internal/console/claim_invariant_probes_worker_test.go +2 −1
internal/console/claim_workers.go +1 −0
internal/console/seedfinish.go +33 −13
internal/console/seedfinish_test.go +21 −0
internal/console/seedsweep.go +283 −0
internal/console/seedsweep_test.go +405 −0
internal/console/testdata/claim-routes.txt +1 −0
internal/provision/seed.go +22 −0
internal/provision/seed_test.go +55 −0

diff --git a/cmd/console/main.go b/cmd/console/main.go

index 4be3a03f7..61deef85c 100644

--- a/cmd/console/main.go

+++ b/cmd/console/main.go

@@ -463,6 +463,11 @@ func serve() {

// The run sweep (internal/console/runsweep.go): expires queued runs,

// ends lost ones, deletes a day-old run's objects. Same placement rule.

go srv.RunClaimWorker(context.Background(), "StartRunSweep")

+ // The seeds sweep (internal/console/seedsweep.go, seeded-instances spec

+ // §3.1, §5.6): abandons stale reservations and imports, finishes

+ // rolled-back seeds, deletes snapshots no live seed needs. Same

+ // placement rule as every pass above.

+ go srv.RunClaimWorker(context.Background(), "StartSeedsPass")

// The relay's writer lease and its converging reconcile

// (internal/console/relayops.go). Same placement rule as every pass

// above, and it returns immediately when CARLOS_RELAY_REGIONS is empty —

diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go

index ac91e0547..88cf378d8 100644

--- a/internal/console/claim_invariant_fixtures_test.go

+++ b/internal/console/claim_invariant_fixtures_test.go

@@ -1452,6 +1452,7 @@ var claimFixtures = map[string]claimFixture{

"worker PublishMovesConsole": claimNonApp("the moves console record (control/moves-console.json): the platform's moves protocol and key fingerprint; no app, and nothing with the moves switch off"),

"worker ReportApexSquats": claimRead("logs apex squats found in the registry; writes nothing"),

"worker StartRunSweep": claimRead("off on a host-claim console, where every run route refuses: it logs that and writes nothing"),

+ "worker StartSeedsPass": claimRead("off on a host-claim console, where every seed door and finishSeed refuse: it logs that and writes nothing"),

"worker StartAlertWatcher": claimNonApp("box alert state under alerts/boxes and operator mail"),

"worker StartEconomicsPass": claimNonApp("daily usage census, statements and bills: deployment and account economics"),

"worker StartGeoipJobs": claimNonApp("GeoIP orphan sweep, D1 expiry and feed status under control/geoip/: the platform database, no app"),

diff --git a/internal/console/claim_invariant_probes_worker_test.go b/internal/console/claim_invariant_probes_worker_test.go

index 25e0c6b0c..6e01a384a 100644

--- a/internal/console/claim_invariant_probes_worker_test.go

+++ b/internal/console/claim_invariant_probes_worker_test.go

@@ -74,7 +74,8 @@ var claimWorkerProbes = map[string]claimProbe{

cx.addRoute(t, registry.Route{Host: "titogo.bab.oncarlos.com", Kind: "instance", Status: "ready",

App: "kass", Account: "bab", Addr: "127.0.0.1:9301"})

}, "apex squat", "titogo.bab.oncarlos.com"),

- "worker StartRunSweep": claimLogWorkerProbe("StartRunSweep", nil, "run sweep", "host-claim console"),

+ "worker StartRunSweep": claimLogWorkerProbe("StartRunSweep", nil, "run sweep", "host-claim console"),

+ "worker StartSeedsPass": claimLogWorkerProbe("StartSeedsPass", nil, "seeds pass", "host-claim console"),

"worker LogPlacementDiscrepancies": claimLogWorkerProbe("LogPlacementDiscrepancies", claimRegionalApp,

msgPlacementDiscrepancy("drift.go.tito.io", "blog", "uk")),

diff --git a/internal/console/claim_workers.go b/internal/console/claim_workers.go

index 89bedaca3..21983223d 100644

--- a/internal/console/claim_workers.go

+++ b/internal/console/claim_workers.go

@@ -14,6 +14,7 @@ func (s *Server) registerClaimWorkers() {

"StartScanPass": s.StartScanPass,

"StartExpiryPass": s.StartExpiryPass,

"StartRunSweep": s.StartRunSweep,

+ "StartSeedsPass": s.StartSeedsPass,

"StartRelay": s.StartRelay,

"StartMailEventPush": s.StartMailEventPush,

"StartDoorKeyResign": s.StartDoorKeyResign,

diff --git a/internal/console/seedfinish.go b/internal/console/seedfinish.go

index 016350cfe..f1cea9c43 100644

--- a/internal/console/seedfinish.go

+++ b/internal/console/seedfinish.go

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

"fmt"

"log"

"os"

+ "slices"

"time"

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

@@ -29,8 +30,9 @@ import (

// 1. The store of the record's environment, under the store's walk lock

// and re-read there, is torn down only while its Declared.Seed is

// seedID (preflight M9), and only when no other instance names that

-// environment (M10). A store with another seed, or none, is never

-// touched.

+// environment (M10), read strictly: a record of the app that will

+// not decode counts as one (Plan 8 M7). A store with another seed, or

+// none, is never touched.

// 2. The seed's snapshot, seeds/<host>/<id>/snapshot.db.

// 3. For a --from seed, every key under the host's replica prefix in the

// data bucket (M11). It is this seed's: the record is create-only and

@@ -125,10 +127,15 @@ func (s *Server) finishSeedStore(ctx context.Context, sqid, app, host, seedID, e

if store == nil || store.Declared == nil || store.Declared.Seed != seedID {

return nil

}

- other, err := s.otherInstanceInEnvironment(ctx, sqid, app, host, env)

+ other, unread, err := s.otherInstanceInEnvironment(ctx, sqid, app, host, env)

if err != nil {

return err

}

+ if unread {

+ log.Printf("console: finish seed %s of %s/%s/%s: OPERATOR: the record of instance %s could not be read, so it may use environment %q, and the store seed %s made (bucket %s) is left standing; tear it down by hand if it is not wanted",

+ seedID, sqid, app, host, other, env, seedID, storeBucketName(store))

+ return nil

+ }

if other != "" {

log.Printf("console: finish seed %s of %s/%s/%s: OPERATOR: instance %s also uses environment %q, so the store seed %s made (bucket %s) is left standing; tear it down by hand if it is not wanted",

seedID, sqid, app, host, other, env, seedID, storeBucketName(store))

@@ -147,23 +154,36 @@ func (s *Server) finishSeedStore(ctx context.Context, sqid, app, host, seedID, e

}

// otherInstanceInEnvironment names an instance of app other than host

-// whose record names env, or "" when there is none. A listing it cannot

-// complete is an error: without it the finish cannot tell whether the

-// store is shared.

-func (s *Server) otherInstanceInEnvironment(ctx context.Context, sqid, app, host, env string) (string, error) {

- recs, truncated, err := provision.ListRecords(ctx, s.cfg.Releases, sqid, app, instanceListMax)

+// whose record names env, or "" when there is none. It reads strictly

+// (provision.ListRecordsStrict, preflight M7): it runs unattended from the

+// seeds sweep, and its "none" tears a store down. A listing it cannot

+// complete, or one with more than instanceListMax records, is an error:

+// the finish cannot tell whether the store is shared, so the record stays

+// rolled-back for a later finish. A record that reads but will not decode,

+// or disagrees with its path, counts as sharing: it is named, with unread

+// true, since what it says about env is not known.

+func (s *Server) otherInstanceInEnvironment(ctx context.Context, sqid, app, host, env string) (other string, unread bool, err error) {

+ recs, unreadable, err := provision.ListRecordsStrict(ctx, s.cfg.Releases, sqid, app, instanceListMax)

+ if errors.Is(err, provision.ErrRecordsTruncated) {

+ return "", false, fmt.Errorf("list the instances of %s/%s: more than %d, so the finish cannot tell who uses environment %q", sqid, app, instanceListMax, env)

+ }

if err != nil {

- return "", fmt.Errorf("list the instances of %s/%s: %w", sqid, app, err)

+ return "", false, fmt.Errorf("list the instances of %s/%s: %w", sqid, app, err)

}

- if truncated {

- return "", fmt.Errorf("list the instances of %s/%s: more than %d, so the finish cannot tell who uses environment %q", sqid, app, instanceListMax, env)

+ if len(unreadable) > 0 {

+ hosts := make([]string, 0, len(unreadable))

+ for h := range unreadable {

+ hosts = append(hosts, h)

+ }

+ slices.Sort(hosts)

+ return hosts[0], true, nil

}

for _, r := range recs {

if r.Host != host && r.Environment == env {

- return r.Host, nil

+ return r.Host, false, nil

}

- return "", nil

+ return "", false, nil

}

// seedReplicaBatch is how many replica keys are deleted between two

diff --git a/internal/console/seedfinish_test.go b/internal/console/seedfinish_test.go

index b352416ac..8b1340cb7 100644

--- a/internal/console/seedfinish_test.go

+++ b/internal/console/seedfinish_test.go

@@ -385,6 +385,27 @@ func TestFinishSeedSparesAStoreAnotherInstanceUses(t *testing.T) {

f.requireRecordGone(t, fromHost)

}

+// A record of the app that will not decode may name the environment: the

+// finish cannot know it does not, so the store is spared as for a sharer.

+// The sweep runs this finish unattended (Plan 8 M7).

+func TestFinishSeedSparesAStoreWhenARecordIsUnreadable(t *testing.T) {

+ f := finishFixture(t)

+ id := newSeedID(t)

+ f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)

+ if _, err := f.fx.rel.Put(context.Background(), provision.RecordKey("bab", fromApp, "other.shop.example.com"), []byte("{not json"), ""); err != nil {

+ t.Fatal(err)

+ }

+ f.markedStore(t, id)

+ key := objstore.KeyIn("bab", fromApp, "canary")

+ before := f.rawKey(t, f.fx.rel, key)

+ if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil {

+ t.Fatal(err)

+ }

+ if after := f.rawKey(t, f.fx.rel, key); after != before {

+ t.Fatalf("a store an unreadable record may use was touched:\n before %s\n after %s", before, after)

+ }

// failingRecordList fails every listing of the app's records.

type failingRecordList struct {

release.Store

diff --git a/internal/console/seedsweep.go b/internal/console/seedsweep.go

new file mode 100644

index 000000000..cc78d4ffb

--- /dev/null

+++ b/internal/console/seedsweep.go

@@ -0,0 +1,283 @@

+// The seeds sweep (seeded-instances spec §3.1, §5.6): the console's

+// cleanup of abandoned seed state. It abandons, by a guarded CAS, a seed

+// stuck where no box will ever move it; it finishes a rolled-back seed;

+// and it deletes a snapshot under seeds/ that no live seed needs. It runs

+// unattended and deletes customer data, so every deletion rests on

+// positive knowledge: two complete listings, taken in the order that makes

+// "no record names it" exact, and a record set with no gap in it.

+package console

+import (

+ "context"

+ "errors"

+ "log"

+ "os"

+ "sort"

+ "time"

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

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

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

+const (

+ seedsPassInterval = 5 * time.Minute

+ // seedReserveTTL is how long a --from seed may stay reserving. The

+ // door's copy is bounded at about 52 minutes (store walk, a 30-minute

+ // restore, 60s, a 20-minute upload, the publish), so an hour is past

+ // any live copy.

+ seedReserveTTL = time.Hour

+ // seedImportTTL is how long an import may wait for its upload, or for

+ // the app to say its process is stopped.

+ seedImportTTL = 7 * 24 * time.Hour

+ // seedOrphanAge is how old a snapshot no record names must be before

+ // it is deleted.

+ seedOrphanAge = time.Hour

+// The tests' seams, nil in production. seedsPassBetweenListings runs after

+// the snapshot listing and before the record listing.

+// seedsPassBeforeAbort runs inside each abandonment's mutation, after the

+// record is read and before it is written, with the call's number (1, then

+// 2 on SetRecordSeed's retry).

+var (

+ seedsPassBetweenListings func()

+ seedsPassBeforeAbort func(call int)

+// StartSeedsPass runs the seeds sweep now and every five minutes, until

+// ctx is done. A console with host claims armed runs none: its record

+// writes go through the claim, every seed door refuses it, and finishSeed

+// refuses it too.

+func (s *Server) StartSeedsPass(ctx context.Context) {

+ if s.cfg.HostClaim {

+ log.Printf("console: seeds pass: off on a host-claim console, where every seed door refuses")

+ return

+ }

+ if s.cfg.Releases == nil || s.cfg.SeedData == nil {

+ return

+ }

+ if _, ok := s.cfg.SeedData.(release.ModLister); !ok {

+ log.Printf("console: seeds pass: the data bucket cannot list last-modified times, so this sweep deletes no unreferenced snapshot")

+ }

+ t := time.NewTicker(seedsPassInterval)

+ defer t.Stop()

+ for {

+ s.seedsPass(ctx, time.Now())

+ select {

+ case <-ctx.Done():

+ return

+ case <-t.C:

+ }

+// seedStale reports a seed stuck where no box moves it: a --from seed

+// reserving for over seedReserveTTL, or an import waiting for its upload or

+// its stop for over seedImportTTL. Both times are the console's: At is

+// stamped by the door that reserved or began the seed.

+func seedStale(seed *provision.SeedSource, now time.Time) bool {

+ switch {

+ case seed == nil:

+ return false

+ case seed.Kind == seeds.KindReplica && seed.State == seeds.StateReserving:

+ return now.Sub(seed.At) > seedReserveTTL

+ case seed.Import() && (seed.State == seeds.StateAwaitingUpload || seed.State == seeds.StateUploaded):

+ return now.Sub(seed.At) > seedImportTTL

+ }

+ return false

+// seedsPass is one sweep. In order:

+//

+// 1. The snapshots under seeds/, with their last-modified times. Listed

+// first: every door writes its record before its snapshot (spec §4

+// step 7, §5.1), so a snapshot in this listing whose seed is live has

+// its record in the next one. A seed begun after this listing has no

+// snapshot in it, so nothing below can reach it.

+// 2. Every record, strictly. Any listing error, here or in step 1, ends

+// the pass: nothing is decided on a partial set. A host with a record

+// that would not decode is unreadable; a host two records hold is

+// ambiguous. Nothing of either is touched: no snapshot deleted, no seed

+// abandoned or finished.

+// 3. The decisions, one per snapshot, from the two listings alone.

+// 4. Each stale seed abandoned by a CAS from the exact state read

+// (provision.SeedAbortStale); the box rolls back its part.

+// 5. The snapshots deleted: those decided in step 3, and each abandoned

+// seed's own once its CAS has landed.

+// 6. Each rolled-back seed finished (finishSeed re-reads its record).

+//

+// Every delete names one exact seeds/<host>/<id>/snapshot.db, so it never

+// reaches another seed's.

+func (s *Server) seedsPass(ctx context.Context, now time.Time) {

+ if s.cfg.HostClaim || s.cfg.Releases == nil || s.cfg.SeedData == nil {

+ return

+ }

+ var snaps []release.KeyMod

+ ml, walk := s.cfg.SeedData.(release.ModLister)

+ if walk {

+ var err error

+ if snaps, err = ml.ListMod(ctx, seeds.Prefix); err != nil {

+ log.Printf("console: seeds pass: list snapshots: %v (nothing swept this pass)", err)

+ return

+ }

+ if seedsPassBetweenListings != nil {

+ seedsPassBetweenListings()

+ }

+ recs, unreadable, err := provision.ListAllRecordsStrict(ctx, s.cfg.Releases)

+ if err != nil {

+ log.Printf("console: seeds pass: list records: %v (nothing swept this pass)", err)

+ return

+ }

+ byHost := map[string][]provision.Record{}

+ for _, rec := range recs {

+ byHost[rec.Host] = append(byHost[rec.Host], rec)

+ }

+ // held reports a host the pass must not touch, logging why once.

+ logged := map[string]bool{}

+ held := func(host string) bool {

+ switch {

+ case unreadable[host]:

+ if !logged[host] {

+ log.Printf("console: seeds pass: a record of %s could not be read; nothing of %s is swept this pass", host, host)

+ }

+ case len(byHost[host]) > 1:

+ if !logged[host] {

+ log.Printf("console: seeds pass: OPERATOR: %d records hold %s; nothing of %s is swept until one remains", len(byHost[host]), host, host)

+ }

+ default:

+ return false

+ }

+ logged[host] = true

+ return true

+ }

+ // Step 3: the decisions.

+ var condemned []string

+ waiting := map[[2]string]string{} // host, id -> key: deleted only if this pass abandons the seed

+ for _, o := range snaps {

+ host, id, ok := seeds.ParseSnapshotKey(o.Key)

+ if !ok || held(host) {

+ continue

+ }

+ var rec *provision.Record

+ if rs := byHost[host]; len(rs) == 1 {

+ rec = &rs[0]

+ }

+ switch snapshotDecision(rec, id, now, o.Mod) {

+ case snapshotDelete:

+ condemned = append(condemned, o.Key)

+ case snapshotIfAbandoned:

+ waiting[[2]string{host, id}] = o.Key

+ }

+ // Step 4: the abandonments.

+ for _, rec := range recs {

+ if !seedStale(rec.Seed, now) || held(rec.Host) {

+ continue

+ }

+ if s.abandonSeed(ctx, rec) {

+ if key, ok := waiting[[2]string{rec.Host, rec.Seed.ID}]; ok {

+ condemned = append(condemned, key)

+ }

+ // Step 5: the snapshot deletes.

+ sort.Strings(condemned)

+ for _, key := range condemned {

+ if err := s.cfg.SeedData.Delete(ctx, key); err != nil && !errors.Is(err, os.ErrNotExist) {

+ log.Printf("console: seeds pass: delete %s: %v", key, err)

+ continue

+ }

+ log.Printf("console: seeds pass: deleted %s", key)

+ }

+ // Step 6: the finishes.

+ for _, rec := range recs {

+ if rec.Seed == nil || rec.Seed.State != seeds.StateRolledBack || held(rec.Host) {

+ continue

+ }

+ if err := s.finishSeed(ctx, rec.Account, rec.App, rec.Host, rec.Seed.ID); err != nil {

+ log.Printf("console: seeds pass: finish seed %s of %s: %v", rec.Seed.ID, rec.Host, err)

+ }

+type snapshotVerdict int

+const (

+ snapshotKeep snapshotVerdict = iota

+ snapshotDelete

+ snapshotIfAbandoned // keep, unless this pass abandons the seed by CAS

+// snapshotDecision decides one snapshot of seed id from the one record that

+// holds its host (nil for none), and the snapshot's last-modified time. Its

+// age is the console's clock less the bucket's stamp, so skew between

+// the two moves it; it only ever decides a snapshot no record names, which

+// the listing order has already shown is no live seed's, so the hour is a

+// margin and not the proof.

+//

+// - The record names id: deleted when the seed is placed or committed

+// (the host's own replica exists, so the box's delete was lost) or

+// aborted or rolled-back (terminal: nothing places it, and the box's

+// own rollback deletes the same key; this also covers a trashed app's

+// aborted seed, which no box finishes). A seed stale at now waits on

+// its abandonment. Any other state is a seed still wanted: kept.

+// - The record names another seed, placed or committed: that seed

+// superseded this one (spec §3.1).

+// - Otherwise no record names it (spec §3.1): deleted once it is over

+// seedOrphanAge old.

+func snapshotDecision(rec *provision.Record, id string, now, mod time.Time) snapshotVerdict {

+ if rec != nil && rec.Seed != nil {

+ state := rec.Seed.State

+ placed := state == seeds.StatePlaced || state == seeds.StateCommitted

+ if rec.Seed.ID == id {

+ switch {

+ case placed || rec.Seed.Abandoned():

+ return snapshotDelete

+ case seedStale(rec.Seed, now):

+ return snapshotIfAbandoned

+ }

+ return snapshotKeep

+ }

+ if placed {

+ return snapshotDelete

+ }

+ if now.Sub(mod) > seedOrphanAge {

+ return snapshotDelete

+ }

+ return snapshotKeep

+// abandonSeed moves rec's stale seed to aborted by a CAS from exactly the

+// state the pass read, and answers whether that CAS landed. A seed that

+// moved on meanwhile (the door published, the upload landed, the app said

+// stopped) is left alone.

+func (s *Server) abandonSeed(ctx context.Context, rec provision.Record) bool {

+ seed := *rec.Seed

+ abort := provision.SeedAbortStale(seed.ID, seed.State, seed.At)

+ call := 0

+ _, err := provision.SetRecordSeed(ctx, s.cfg.Releases, rec.Account, rec.App, rec.Host, func(r *provision.Record) error {

+ call++

+ if seedsPassBeforeAbort != nil {

+ seedsPassBeforeAbort(call)

+ }

+ return abort(r)

+ })

+ if err != nil {

+ if !errors.Is(err, provision.ErrSeedConflict) {

+ log.Printf("console: seeds pass: abandon seed %s of %s: %v", seed.ID, rec.Host, err)

+ }

+ return false

+ }

+ log.Printf("console: seeds pass: abandoned seed %s of %s/%s/%s (%s since %s)", seed.ID, rec.Account, rec.App, rec.Host,

+ seed.State, seed.At.UTC().Format(time.RFC3339))

+ s.bumpSerial(ctx)

+ return true

diff --git a/internal/console/seedsweep_test.go b/internal/console/seedsweep_test.go

new file mode 100644

index 000000000..8d3ce250e

--- /dev/null

+++ b/internal/console/seedsweep_test.go

@@ -0,0 +1,405 @@

+package console

+import (

+ "context"

+ "errors"

+ "os"

+ "path/filepath"

+ "strings"

+ "testing"

+ "time"

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

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

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

+// sweepFixture is a console with a data bucket for seeds: a DirStore, so

+// a snapshot's last-modified time is its file's mtime and a test can age

+// it. Records are app shop's in the operator account.

+type sweepFixture struct {

+ fx *consoleFixture

+ data release.Store

+ root string

+const sweepApp = "shop"

+func newSweepFixture(t *testing.T) sweepFixture {

+ t.Helper()

+ fx := newConsoleFixture(t)

+ root := t.TempDir()

+ data := release.NewDirStore(root)

+ fx.srv.cfg.SeedData = data

+ return sweepFixture{fx: fx, data: data, root: root}

+// seeded writes a record of shop on host whose seed is a fresh id, of

+// kind, in state, stamped at, and answers the id.

+func (f sweepFixture) seeded(t *testing.T, host, kind, state string, at time.Time) string {

+ t.Helper()

+ id := newSeedID(t)

+ f.recordOf(t, sweepApp, host, &provision.SeedSource{ID: id, Kind: kind, State: state, By: "bob@example.com", At: at})

+ return id

+// recordOf writes a record of app on host with seed (nil for none).

+func (f sweepFixture) recordOf(t *testing.T, app, host string, seed *provision.SeedSource) {

+ t.Helper()

+ rec := provision.Record{Schema: 1, Host: host, Slug: strings.Split(host, ".")[0], App: app, Account: release.OpsAccount,

+ Channel: "stable", CreatedAt: time.Now().UTC(), Seed: seed}

+ if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil {

+ t.Fatal(err)

+ }

+// snapshot writes seeds/<host>/<id>/snapshot.db last modified age ago.

+func (f sweepFixture) snapshot(t *testing.T, host, id string, age time.Duration) string {

+ t.Helper()

+ key := seeds.SnapshotKey(host, id)

+ if _, err := f.data.Put(context.Background(), key, []byte("db"), ""); err != nil {

+ t.Fatal(err)

+ }

+ // DirStore keeps a key at root/key, and its ListMod reports the file's

+ // mtime.

+ when := time.Now().Add(-age)

+ if err := os.Chtimes(filepath.Join(f.root, filepath.FromSlash(key)), when, when); err != nil {

+ t.Fatal(err)

+ }

+ return key

+func (f sweepFixture) snapshotExists(t *testing.T, key string) bool {

+ t.Helper()

+ _, _, err := f.data.Get(context.Background(), key)

+ if err != nil && !errors.Is(err, os.ErrNotExist) {

+ t.Fatal(err)

+ }

+ return err == nil

+// state is the seed state of shop's record on host, or "gone".

+func (f sweepFixture) state(t *testing.T, host string) string {

+ t.Helper()

+ rec, err := provision.GetRecord(context.Background(), f.fx.rel, release.OpsAccount, sweepApp, host)

+ switch {

+ case errors.Is(err, os.ErrNotExist):

+ return "gone"

+ case err != nil:

+ t.Fatal(err)

+ }

+ return rec.Seed.State

+func (f sweepFixture) pass(now time.Time) { f.fx.srv.seedsPass(context.Background(), now) }

+func TestSweepLeavesALiveReservation(t *testing.T) {

+ f := newSweepFixture(t)

+ now := time.Now()

+ id := f.seeded(t, "canary.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-10*time.Minute))

+ key := f.snapshot(t, "canary.shop.example.com", id, 10*time.Minute)

+ f.pass(now)

+ if got := f.state(t, "canary.shop.example.com"); got != seeds.StateReserving || !f.snapshotExists(t, key) {

+ t.Fatalf("a live reservation was touched: %s, snapshot kept %v", got, f.snapshotExists(t, key))

+ }

+ // Past the hour it is abandoned by CAS, and its snapshot goes with it.

+ f.pass(now.Add(2 * time.Hour))

+ if got := f.state(t, "canary.shop.example.com"); got != seeds.StateAborted {

+ t.Fatalf("a stale reservation = %s, want aborted", got)

+ }

+ if f.snapshotExists(t, key) {

+ t.Fatal("the snapshot of a seed this pass abandoned survived")

+ }

+// A --from copy is bounded at about 52 minutes (30m restore, 60s, 20m

+// upload): a reservation younger than the hour is never abandoned.

+func TestSweepReservationThresholdOutlastsTheCopy(t *testing.T) {

+ if seedReserveTTL <= 30*time.Minute+time.Minute+20*time.Minute {

+ t.Fatalf("seedReserveTTL %s does not outlast the --from copy's bound", seedReserveTTL)

+ }

+ f := newSweepFixture(t)

+ now := time.Now()

+ f.seeded(t, "copy.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-55*time.Minute))

+ f.pass(now)

+ if got := f.state(t, "copy.shop.example.com"); got != seeds.StateReserving {

+ t.Fatalf("a 55-minute reservation = %s, want reserving", got)

+ }

+func TestSweepAbandonsOnlyStaleImports(t *testing.T) {

+ f := newSweepFixture(t)

+ now := time.Now()

+ old, young := now.Add(-8*24*time.Hour), now.Add(-6*24*time.Hour)

+ f.seeded(t, "waiting.shop.example.com", seeds.KindImport, seeds.StateAwaitingUpload, old)

+ f.seeded(t, "uploaded.shop.example.com", seeds.KindImport, seeds.StateUploaded, old)

+ f.seeded(t, "recent.shop.example.com", seeds.KindImport, seeds.StateUploaded, young)

+ f.seeded(t, "ready.shop.example.com", seeds.KindImport, seeds.StateReady, old)

+ f.seeded(t, "refused.shop.example.com", seeds.KindImport, seeds.StateRefused, old)

+ f.seeded(t, "fresh.shop.example.com", seeds.KindReplica, seeds.StateReady, old)

+ f.pass(now)

+ for host, want := range map[string]string{

+ "waiting.shop.example.com": seeds.StateAborted,

+ "uploaded.shop.example.com": seeds.StateAborted,

+ "recent.shop.example.com": seeds.StateUploaded,

+ "ready.shop.example.com": seeds.StateReady,

+ "refused.shop.example.com": seeds.StateRefused,

+ "fresh.shop.example.com": seeds.StateReady,

+ } {

+ if got := f.state(t, host); got != want {

+ t.Errorf("%s = %s, want %s", host, got, want)

+ }

+func TestSweepDeletesOnlyUnreferencedOrSupersededSnapshots(t *testing.T) {

+ f := newSweepFixture(t)

+ now := time.Now()

+ keep, gone := map[string]string{}, map[string]string{}

+ live := f.seeded(t, "a.shop.example.com", seeds.KindReplica, seeds.StateReady, now)

+ keep["a snapshot a live record needs"] = f.snapshot(t, "a.shop.example.com", live, 3*time.Hour)

+ gone["an orphan over an hour old beside a live seed"] = f.snapshot(t, "a.shop.example.com", newSeedID(t), 3*time.Hour)

+ keep["a young unreferenced snapshot (its door may be between record and upload)"] =

+ f.snapshot(t, "nobody.shop.example.com", newSeedID(t), 10*time.Minute)

+ gone["an orphan with no record, over an hour old"] = f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour)

+ f.seeded(t, "b.shop.example.com", seeds.KindReplica, seeds.StatePlaced, now)

+ gone["a young snapshot superseded by a placed seed"] = f.snapshot(t, "b.shop.example.com", newSeedID(t), 5*time.Minute)

+ f.seeded(t, "c.shop.example.com", seeds.KindImport, seeds.StateCommitted, now)

+ gone["a young snapshot superseded by a committed seed"] = f.snapshot(t, "c.shop.example.com", newSeedID(t), 5*time.Minute)

+ f.seeded(t, "d.shop.example.com", seeds.KindImport, seeds.StateAborted, now)

+ keep["a young snapshot beside another seed that is not placed"] = f.snapshot(t, "d.shop.example.com", newSeedID(t), 5*time.Minute)

+ own := func(host, kind, state string) string {

+ return f.snapshot(t, host, f.seeded(t, host, kind, state, now), 5*time.Minute)

+ }

+ gone["its own placed seed's (the box's delete was lost)"] = own("e.shop.example.com", seeds.KindReplica, seeds.StatePlaced)

+ gone["its own committed seed's"] = own("f.shop.example.com", seeds.KindImport, seeds.StateCommitted)

+ gone["its own aborted seed's"] = own("g.shop.example.com", seeds.KindImport, seeds.StateAborted)

+ keep["its own refused seed's"] = own("h.shop.example.com", seeds.KindImport, seeds.StateRefused)

+ keep["its own swapped seed's"] = own("i.shop.example.com", seeds.KindImport, seeds.StateSwapped)

+ keep["its own ready import's"] = own("j.shop.example.com", seeds.KindImport, seeds.StateReady)

+ // Not a snapshot: nothing else under seeds/ is the sweep's.

+ stray := seeds.SnapshotPrefix("k.shop.example.com", newSeedID(t)) + "other.db"

+ if _, err := f.data.Put(context.Background(), stray, []byte("x"), ""); err != nil {

+ t.Fatal(err)

+ }

+ keep["an object that is not a snapshot"] = stray

+ f.pass(now)

+ for why, key := range keep {

+ if !f.snapshotExists(t, key) {

+ t.Errorf("deleted %s", why)

+ }

+ for why, key := range gone {

+ if f.snapshotExists(t, key) {

+ t.Errorf("kept %s", why)

+ }

+ if got := f.state(t, "g.shop.example.com"); got != seeds.StateAborted {

+ t.Errorf("the aborted record = %s; the sweep deletes its snapshot and leaves the record to the box", got)

+ }

+func TestSweepKeepsSnapshotsOfAnUnreadableRecord(t *testing.T) {

+ f := newSweepFixture(t)

+ host := "a.shop.example.com"

+ id := f.seeded(t, host, seeds.KindReplica, seeds.StatePlaced, time.Now())

+ key := f.snapshot(t, host, id, 3*time.Hour)

+ orphan := f.snapshot(t, host, newSeedID(t), 3*time.Hour)

+ rk := provision.RecordKey(release.OpsAccount, sweepApp, host)

+ _, etag, err := f.fx.rel.Get(context.Background(), rk)

+ if err != nil {

+ t.Fatal(err)

+ }

+ if _, err := f.fx.rel.Put(context.Background(), rk, []byte("{not json"), etag); err != nil {

+ t.Fatal(err)

+ }

+ f.pass(time.Now())

+ if !f.snapshotExists(t, key) || !f.snapshotExists(t, orphan) {

+ t.Fatalf("an unreadable record's host lost a snapshot: own kept %v, other kept %v", f.snapshotExists(t, key), f.snapshotExists(t, orphan))

+ }

+// Two records holding one host: neither decides for the other, so the

+// sweep keeps everything of the host and abandons nothing on it.

+func TestSweepKeepsAHostTwoRecordsHold(t *testing.T) {

+ f := newSweepFixture(t)

+ now := time.Now()

+ host := "twice.shop.example.com"

+ id := f.seeded(t, host, seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))

+ f.recordOf(t, "blog", host, &provision.SeedSource{ID: newSeedID(t), Kind: seeds.KindReplica, State: seeds.StatePlaced,

+ By: "bob@example.com", At: now})

+ own := f.snapshot(t, host, id, 3*time.Hour)

+ orphan := f.snapshot(t, host, newSeedID(t), 3*time.Hour)

+ f.pass(now)

+ if !f.snapshotExists(t, own) || !f.snapshotExists(t, orphan) {

+ t.Fatalf("a host two records hold lost a snapshot: own kept %v, other kept %v", f.snapshotExists(t, own), f.snapshotExists(t, orphan))

+ }

+ if got := f.state(t, host); got != seeds.StateReserving {

+ t.Fatalf("a host two records hold had its seed moved to %s", got)

+ }

+// The app stops the import between the abort's read and its write: the

+// abort's CAS loses, the retry finds the seed moved on and refuses, and

+// nothing is deleted.

+func TestSweepAbandonsOnlyByCAS(t *testing.T) {

+ f := newSweepFixture(t)

+ host := "imp.shop.example.com"

+ old := time.Now().Add(-8 * 24 * time.Hour)

+ id := f.seeded(t, host, seeds.KindImport, seeds.StateUploaded, old)

+ key := f.snapshot(t, host, id, 8*24*time.Hour)

+ var calls []int

+ seedsPassBeforeAbort = func(call int) {

+ calls = append(calls, call)

+ if call != 1 {

+ return

+ }

+ if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, release.OpsAccount, sweepApp, host, provision.SeedStopped(id)); err != nil {

+ t.Errorf("the app's stop: %v", err)

+ }

+ t.Cleanup(func() { seedsPassBeforeAbort = nil })

+ f.pass(time.Now())

+ if len(calls) != 2 {

+ t.Fatalf("the abort's mutation ran %d times, want 2 (a write that lost, then a refused retry)", len(calls))

+ }

+ if got := f.state(t, host); got != seeds.StateReady || !f.snapshotExists(t, key) {

+ t.Fatalf("state %s, snapshot kept %v; want the app's stop to win and nothing deleted", got, f.snapshotExists(t, key))

+ }

+func TestSweepFinishesARolledBackSeed(t *testing.T) {

+ f := newSweepFixture(t)

+ host := "gone.shop.example.com"

+ id := f.seeded(t, host, seeds.KindImport, seeds.StateRolledBack, time.Now())

+ key := f.snapshot(t, host, id, time.Minute)

+ f.pass(time.Now())

+ if got := f.state(t, host); got != "gone" {

+ t.Fatalf("a rolled-back record after the sweep: %s", got)

+ }

+ if f.snapshotExists(t, key) {

+ t.Fatal("a finished seed's snapshot survived")

+ }

+// The host is deleted and seeded again between the two listings. Listed

+// snapshots first, the new seed's snapshot is not in the listing, so the

+// placed record of the old seed cannot condemn it (the doors write the

+// record before the snapshot).

+func TestSweepListsSnapshotsBeforeRecords(t *testing.T) {

+ f := newSweepFixture(t)

+ now := time.Now()

+ host := "again.shop.example.com"

+ f.seeded(t, host, seeds.KindReplica, seeds.StatePlaced, now.Add(-time.Hour))

+ var fresh string

+ seedsPassBetweenListings = func() {

+ if err := f.fx.rel.Delete(context.Background(), provision.RecordKey(release.OpsAccount, sweepApp, host)); err != nil {

+ t.Error(err)

+ }

+ id := f.seeded(t, host, seeds.KindReplica, seeds.StateReady, now)

+ fresh = f.snapshot(t, host, id, time.Minute)

+ }

+ t.Cleanup(func() { seedsPassBetweenListings = nil })

+ f.pass(now)

+ if fresh == "" {

+ t.Fatal("the seam never ran")

+ }

+ if !f.snapshotExists(t, fresh) {

+ t.Fatal("the new seed's snapshot was deleted as superseded")

+ }

+// failingAccountList fails the record walk's first listing.

+type failingAccountList struct{ release.Store }

+func (s failingAccountList) List(ctx context.Context, prefix string) ([]string, error) {

+ if prefix == "accounts/" {

+ return nil, errors.New("the bucket is unreachable")

+ }

+ return s.Store.List(ctx, prefix)

+// failingModList is a data bucket whose listing fails.

+type failingModList struct{ release.Store }

+func (failingModList) ListMod(context.Context, string) ([]release.KeyMod, error) {

+ return nil, errors.New("the bucket is unreachable")

+// A listing that fails is no knowledge: the pass deletes nothing and

+// abandons nothing.

+func TestSweepDecidesNothingOnAFailedListing(t *testing.T) {

+ for name, broke := range map[string]func(f sweepFixture){

+ "records": func(f sweepFixture) { f.fx.srv.cfg.Releases = failingAccountList{f.fx.rel} },

+ "snapshots": func(f sweepFixture) { f.fx.srv.cfg.SeedData = failingModList{f.data} },

+ } {

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

+ f := newSweepFixture(t)

+ now := time.Now()

+ stale := f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))

+ staleKey := f.snapshot(t, "stale.shop.example.com", stale, 2*time.Hour)

+ orphan := f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour)

+ f.seeded(t, "done.shop.example.com", seeds.KindImport, seeds.StateRolledBack, now)

+ broke(f)

+ f.pass(now)

+ if got := f.state(t, "stale.shop.example.com"); got != seeds.StateReserving {

+ t.Errorf("a stale reservation = %s after a failed listing", got)

+ }

+ if got := f.state(t, "done.shop.example.com"); got != seeds.StateRolledBack {

+ t.Errorf("a rolled-back record = %s after a failed listing", got)

+ }

+ if !f.snapshotExists(t, staleKey) || !f.snapshotExists(t, orphan) {

+ t.Error("a snapshot was deleted after a failed listing")

+ }

+ })

+ }

+// noModList is a data bucket that cannot list last-modified times.

+type noModList struct{ release.Store }

+// Without last-modified times no snapshot is judged, but stale seeds are

+// still abandoned and rolled-back ones finished.

+func TestSweepWithoutListModStillAbandonsAndFinishes(t *testing.T) {

+ f := newSweepFixture(t)

+ now := time.Now()

+ f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))

+ f.seeded(t, "done.shop.example.com", seeds.KindImport, seeds.StateRolledBack, now)

+ orphan := f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour)

+ f.fx.srv.cfg.SeedData = noModList{f.data}

+ f.pass(now)

+ if got := f.state(t, "stale.shop.example.com"); got != seeds.StateAborted {

+ t.Errorf("a stale reservation = %s, want aborted", got)

+ }

+ if got := f.state(t, "done.shop.example.com"); got != "gone" {

+ t.Errorf("a rolled-back record = %s, want gone", got)

+ }

+ if !f.snapshotExists(t, orphan) {

+ t.Error("an orphan was deleted with no last-modified time to judge it by")

+ }

+// A console with host claims armed runs no sweep: its record writes go

+// through the claim, and every seed door refuses it.

+func TestSweepOffOnAHostClaimConsole(t *testing.T) {

+ f := newSweepFixture(t)

+ now := time.Now()

+ f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))

+ orphan := f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour)

+ f.fx.srv.cfg.HostClaim = true

+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)

+ defer cancel()

+ f.fx.srv.StartSeedsPass(ctx) // returns at once

+ if ctx.Err() != nil {

+ t.Fatal("StartSeedsPass ran on a host-claim console")

+ }

+ f.pass(now)

+ if got := f.state(t, "stale.shop.example.com"); got != seeds.StateReserving || !f.snapshotExists(t, orphan) {

+ t.Fatalf("a host-claim console swept: %s, orphan kept %v", got, f.snapshotExists(t, orphan))

+ }

diff --git a/internal/console/testdata/claim-routes.txt b/internal/console/testdata/claim-routes.txt

index 881f2f38c..72b4654b7 100644

--- a/internal/console/testdata/claim-routes.txt

+++ b/internal/console/testdata/claim-routes.txt

@@ -265,3 +265,4 @@ worker StartMailEventPush

worker StartRelay

worker StartRunSweep

worker StartScanPass

+worker StartSeedsPass

diff --git a/internal/provision/seed.go b/internal/provision/seed.go

index f0f82c7ca..5378a0733 100644

--- a/internal/provision/seed.go

+++ b/internal/provision/seed.go

@@ -231,6 +231,28 @@ func SeedAbortUploadBecause(id, reason string) func(*Record) error {

}

+// SeedAbortStale is the console's seeds sweep abandoning a seed it judged

+// stale (spec §3.1, §5.6): SeedAbort, but only while the record's seed is

+// exactly what the sweep read, the same id in the same state since the same

+// At. The sweep judged staleness from that state and At, so a seed that

+// moved on in any way (published, uploaded, stopped, aborted by its own

+// door) is refused with ErrSeedConflict, and SetRecordSeed's retry after a

+// lost race never aborts a seed the sweep did not judge. Every state

+// SeedAbort refuses stays refused.

+func SeedAbortStale(id, state string, at time.Time) func(*Record) error {

+ abort := SeedAbort(id)

+ return func(rec *Record) error {

+ if err := seedIn(rec, id, "", state); err != nil {

+ return err

+ }

+ if !rec.Seed.At.Equal(at) {

+ return fmt.Errorf("%w: seed %s was stamped again (at %s, read at %s)", ErrSeedConflict, id,

+ rec.Seed.At.UTC().Format(time.RFC3339Nano), at.UTC().Format(time.RFC3339Nano))

+ }

+ return abort(rec)

+ }

// SeedRolledBack is the box's last word on an aborted seed (spec §5.6, and

// §3.2's no-journal path). The console finishes from here.

func SeedRolledBack(id string) func(*Record) error {

diff --git a/internal/provision/seed_test.go b/internal/provision/seed_test.go

index c851712e4..3058d3ce1 100644

--- a/internal/provision/seed_test.go

+++ b/internal/provision/seed_test.go

@@ -436,3 +436,58 @@ func TestOrdinaryWritersCannotAddOrDropASeed(t *testing.T) {

t.Fatalf("after the refusal: seed = %+v, %v; want it kept", got.Seed, err)

}

+// The sweep's abort lands only on the seed exactly as the sweep read it:

+// the same id, in the same state, since the same At. A seed that moved on

+// (published, stopped) or was re-stamped is refused, and so is a state

+// SeedAbort itself refuses.

+func TestSeedAbortStaleOnlyFromWhatWasRead(t *testing.T) {

+ ctx := context.Background()

+ for _, c := range []struct{ kind, state string }{

+ {seeds.KindReplica, seeds.StateReserving},

+ {seeds.KindImport, seeds.StateAwaitingUpload},

+ {seeds.KindImport, seeds.StateUploaded},

+ } {

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

+ seed := testSeed(t, c.kind, c.state)

+ rec := seedTestRecord(t, s, seed)

+ got, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbortStale(seed.ID, c.state, seed.At))

+ if err != nil || got.Seed.State != seeds.StateAborted || got.Seed.Reason != "" {

+ t.Fatalf("%s %s: %+v, %v", c.kind, c.state, got, err)

+ }

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

+ seed := testSeed(t, seeds.KindImport, seeds.StateUploaded)

+ rec := seedTestRecord(t, s, seed)

+ other, _ := seeds.NewID()

+ for name, mut := range map[string]func(*Record) error{

+ "another id": SeedAbortStale(other, seeds.StateUploaded, seed.At),

+ "another state": SeedAbortStale(seed.ID, seeds.StateAwaitingUpload, seed.At),

+ "another moment": SeedAbortStale(seed.ID, seeds.StateUploaded, seed.At.Add(time.Second)),

+ } {

+ if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, mut); !errors.Is(err, ErrSeedConflict) {

+ t.Fatalf("%s: err = %v, want ErrSeedConflict", name, err)

+ }

+ // The app stops it: the seed moved on, and the abort read before that

+ // is refused.

+ if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedStopped(seed.ID)); err != nil {

+ t.Fatal(err)

+ }

+ if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbortStale(seed.ID, seeds.StateUploaded, seed.At)); !errors.Is(err, ErrSeedConflict) {

+ t.Fatalf("after the stop: err = %v, want ErrSeedConflict", err)

+ }

+ got, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)

+ if err != nil || got.Seed.State != seeds.StateReady {

+ t.Fatalf("after a refused stale abort: %+v, %v; want ready", got, err)

+ }

+ // A state SeedAbort refuses stays refused: a placed --from seed.

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

+ seed = testSeed(t, seeds.KindReplica, seeds.StatePlaced)

+ rec = seedTestRecord(t, s, seed)

+ if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbortStale(seed.ID, seeds.StatePlaced, seed.At)); !errors.Is(err, ErrSeedConflict) {

+ t.Fatalf("placed --from: err = %v, want ErrSeedConflict", err)

+ }