amadan

carlos / platform Public

moves: the sealing source box records retired.json at gone for a tracked lifetime, with every real version id in a recorded deletion order and the key's latest named, a normalised prefix and a deterministic cleanup id

offlineSourceGone, with the row gone and this box the move's sealer,
runs recordRetired before the gone report: for a record at
offline-finishing on a box with a judge in the move's source region, an
unreadable lifetime holds the gone report back a pass; any other
untracked answer, a listing holding a "null" id, or a standing reason
writes nothing and logs once a day, and the gone report proceeds as
before. A tracked lifetime gets retired.json (create-only): the seal's
FinalTXID, the prefix HibernatePrefix(OwnPrefix, host) + "/", every
entry ListVersions answers under it except lock.json and build-state.json,
per key oldest first (LastModified ascending, ties by Seq descending)
and the current entry last as Latest, and the live link and programs
file by identity when their directory chain is root-only, else left as
"writable layout". A file already there is scanned (scanRetired) and
never rewritten; a damaged one moves aside to the first free
retired.json.corrupt-<n> behind a corrupt stub whose found_at is
.corrupt-1's mtime. writeRetired replaces with Rev raised by one.

release exports ProgramsPath. The copy scan lists the new host file (log
lines only, no operator sentence).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude e05ca0574646dbf3a1eaf5e4a10fb2fa8c0a4838 parent 0a98db1
6 files changed, +1159 −0
  • approved_moves_copy_test.go +3 −0
  • internal/host/movescleanup_retired.go +430 −0
  • internal/host/movescleanup_retired_test.go +706 −0
  • internal/host/movesoffline.go +10 −0
  • internal/host/movespass.go +5 −0
  • internal/release/programhistory.go +5 −0

CI failed — run details

CI log 
=== gate ===
sigwho: pid 2 ppid 1 pgid 0 sid 0; child pid 3
.amadan/test-gate.sh
gate: 2026-10-24T14:53:14Z on aarch64, 32 cpus, 61413 MB available, go1.26.8 linux/arm64
gate: +0s start launchctx
gate: +0s start vet
gate: +0s start gofmt
go: downloading github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.68.0
go: downloading github.com/aws/aws-sdk-go-v2 v1.44.0
go: downloading github.com/aws/aws-sdk-go-v2/service/costexplorer v1.67.5
go: downloading github.com/aws/aws-sdk-go-v2/service/s3 v1.105.0
go: downloading github.com/aws/smithy-go v1.28.1
go: downloading github.com/superfly/ltx v0.5.1
go: downloading modernc.org/sqlite v1.55.0
go: downloading golang.org/x/net v0.56.0
go: downloading github.com/aws/aws-sdk-go-v2/config v1.32.6
go: downloading github.com/aws/aws-sdk-go-v2/service/sns v1.43.0
go: downloading github.com/aws/aws-sdk-go-v2/service/acm v1.45.0
go: downloading github.com/aws/aws-sdk-go-v2/service/ec2 v1.324.0
go: downloading github.com/aws/aws-sdk-go-v2/service/iam v1.58.2
go: downloading github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi v1.37.0
go: downloading github.com/aws/aws-sdk-go-v2/service/route53 v1.64.1
go: downloading github.com/aws/aws-sdk-go-v2/service/sts v1.45.6
go: downloading golang.org/x/crypto v0.53.0
go: downloading github.com/aws/aws-sdk-go-v2/service/autoscaling v1.74.0
go: downloading github.com/oschwald/maxminddb-golang v1.13.1
go: downloading golang.org/x/sys v0.46.0
go: downloading github.com/aws/aws-sdk-go-v2/credentials v1.19.6
go: downloading github.com/benbjohnson/litestream v0.5.14
go: downloading github.com/pion/logging v0.2.4
go: downloading github.com/pion/stun/v3 v3.0.1
go: downloading github.com/pion/turn/v4 v4.1.4
go: downloading github.com/aws/aws-sdk-go-v2 v1.44.0
go: downloading github.com/aws/aws-sdk-go-v2/service/sns v1.43.0
go: downloading modernc.org/sqlite v1.55.0
go: downloading github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.68.0
go: downloading github.com/aws/aws-sdk-go-v2/service/ec2 v1.324.0
go: downloading github.com/aws/aws-sdk-go-v2/service/route53 v1.64.1
go: downloading github.com/superfly/ltx v0.5.1
go: downloading github.com/aws/aws-sdk-go-v2/service/autoscaling v1.74.0
go: downloading github.com/benbjohnson/litestream v0.5.14
go: downloading github.com/aws/aws-sdk-go-v2/service/ssm v1.74.0
go: downloading github.com/klauspost/compress v1.18.0
go: downloading github.com/aws/aws-sdk-go-v2/service/sesv2 v1.66.6
go: downloading github.com/aws/aws-sdk-go-v2/service/sesv2 v1.66.6
go: downloading github.com/pierrec/lz4/v4 v4.1.22
go: downloading github.com/pion/dtls/v3 v3.0.7
go: downloading github.com/pion/transport/v3 v3.0.8
go: downloading github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16
go: downloading github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4
go: downloading github.com/aws/aws-sdk-go-v2/service/signin v1.0.4
go: downloading github.com/aws/aws-sdk-go-v2/service/sso v1.30.8
go: downloading github.com/pierrec/lz4/v4 v4.1.22
go: downloading github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12
go: downloading github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.40
go: downloading github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.38
go: downloading github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.40
go: downloading github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.40
go: downloading github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.38
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.40
go: downloading github.com/pion/randutil v0.1.0
go: downloading github.com/pion/transport/v4 v4.0.1
go: downloading github.com/pion/dtls/v3 v3.0.7
go: downloading github.com/pion/transport/v3 v3.0.8
go: downloading github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.40
go: downloading github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.23
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.31
go: downloading github.com/pion/transport/v4 v4.0.1
go: downloading github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.40
go: downloading github.com/prometheus/client_golang v1.17.0
go: downloading golang.org/x/sync v0.21.0
go: downloading github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.18
go: downloading github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.23
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.31
go: downloading github.com/prometheus/client_golang v1.17.0
go: downloading github.com/lmittmann/tint v1.1.3
go: downloading github.com/mattn/go-isatty v0.0.20
go: downloading github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.18
go: downloading github.com/wlynxg/anet v0.0.5
go: downloading github.com/wlynxg/anet v0.0.5
go: downloading github.com/prometheus/client_model v0.5.0
go: downloading github.com/prometheus/procfs v0.12.0
go: downloading github.com/beorn7/perks v1.0.1
go: downloading github.com/cespare/xxhash/v2 v2.3.0
go: downloading github.com/prometheus/common v0.45.0
go: downloading google.golang.org/protobuf v1.36.10
go: downloading github.com/beorn7/perks v1.0.1
go: downloading github.com/prometheus/common v0.45.0
go: downloading github.com/cespare/xxhash/v2 v2.3.0
go: downloading github.com/prometheus/client_model v0.5.0
go: downloading google.golang.org/protobuf v1.36.10
go: downloading github.com/prometheus/procfs v0.12.0
go: downloading golang.org/x/text v0.38.0
go: downloading golang.org/x/text v0.38.0
go: downloading github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0
go: downloading github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0
go: downloading modernc.org/libc v1.74.1
go: downloading modernc.org/libc v1.74.1
go: downloading modernc.org/memory v1.11.0
go: downloading github.com/google/uuid v1.6.0
go: downloading modernc.org/mathutil v1.7.1
go: downloading github.com/dustin/go-humanize v1.0.1
go: downloading github.com/google/uuid v1.6.0
go: downloading github.com/dustin/go-humanize v1.0.1
go: downloading modernc.org/mathutil v1.7.1
go: downloading modernc.org/memory v1.11.0
go: downloading github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec
go: downloading github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec
gate: +8s start rest
gate: +49s start internal_console@0
gate: +49s start internal_console@1
gate: +49s start internal_console@2
gate: +49s start internal_console@3
gate: +49s start internal_console@4
gate: +49s start internal_console@5
gate: +49s start internal_console@6
gate: +49s start internal_console@7
gate: +49s waiting for internal_console's shards before the next split package
gate: +228s start root@0
gate: +228s start root@1
gate: +228s start root@2
gate: +228s waiting for root's shards before the next split package

amadan: step gate failed: signal: terminated

amadan: job failed at step gate — its output is above, not at the tail of this log
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index d2d708f5e..26df05443 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -1304,6 +1304,9 @@ var movesScanFiles = []string{
"internal/host/movesoffline_land.go",
"internal/host/movesoffline_stage.go",
"internal/host/movesoffline_reports.go",
+ // Self-serve moves plan 4 Task 1: the cleanup obligation recorded at
+ // gone (its log lines only; no operator sentence).
+ "internal/host/movescleanup_retired.go",
// Self-serve moves plan 2 Task 10a: the exported end-to-end bridge (test
// infrastructure in a non-test file).
"internal/host/e2e_bridge.go",
diff --git a/internal/host/movescleanup_retired.go b/internal/host/movescleanup_retired.go
new file mode 100644
index 000000000..aa413fe45
--- /dev/null
+++ b/internal/host/movescleanup_retired.go
@@ -0,0 +1,430 @@
+package host
+
+// The cleanup obligation and its evidence at gone (self-serve moves plan 4
+// Task 1): the sealing source box of a finished offline move records, once
+// and before its gone report, what the move left behind on this box and in
+// its own bucket, as retired.json in the move's op directory
+// (moves.RetiredFile). Nothing here deletes anything: the later tasks
+// retire the replica from this inventory, version id by version id.
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io/fs"
+ "log"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/offlinecleanup"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// retiredNoticeEvery is how often a standing reason for writing no
+// obligation is logged.
+const retiredNoticeEvery = 24 * time.Hour
+
+// recordRetired writes retired.json for rec's move at offline-finishing,
+// on the sealing source box whose row is gone (offlineSourceGone's caller
+// checks both), when the lifetime is tracked. It answers nil when it
+// writes nothing for a standing reason (no judge, another region, an
+// untracked lifetime, a version id that names no one version), so the gone
+// report is written as before this plan, and an error, which holds the
+// gone report back to the next pass, when a read fails or the evidence is
+// not whole. A file already there is never rewritten: it is scanned
+// (scanRetired), which keeps a damaged one aside behind a corrupt stub.
+func (w *movesWorker) recordRetired(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ m, host := rec.Move, rec.Host
+ none := func(why string, args ...any) error {
+ w.noticeEvery(host+"#retired", retiredNoticeEvery, "moves: %s: offline move %s: no cleanup obligation: "+why, append([]any{host, m.ID}, args...)...)
+ return nil
+ }
+ switch {
+ case m.State != provision.OfflineFinishing:
+ return none("the move is %s, not finishing", m.State)
+ case cfg.Judge == nil:
+ return none("this box judges no lifetime (host claims are on)")
+ case cfg.Region != m.From:
+ return none("this box is in %q, the move left %q", cfg.Region, m.From)
+ case cfg.OwnBucket == "" || cfg.DataDir == "" || cfg.OptDir == "":
+ return none("this box has no hibernate bucket, data directory or version store")
+ }
+ path := moves.RetiredFile(cfg.DataDir, host, m.ID)
+ if path == "" {
+ return none("the host or the move id is not a usable name")
+ }
+ if found, err := scanRetired(path, cfg.BoxID, host, m.ID); err != nil || found.Retired != nil || found.Stub != nil {
+ return err
+ }
+
+ // N5: only a tracked lifetime gets an obligation.
+ obs := cfg.Judge.Observe(ctx, s, rec.Account, rec.App, host)
+ if obs.Verdict == provision.LifetimeUnreadable {
+ return fmt.Errorf("offline move %s: the lifetime of %s cannot be judged: %w", m.ID, host, obs.Err)
+ }
+ if !trackedAtGone(obs) {
+ return none("the lifetime is not tracked (verdict %d)", obs.Verdict)
+ }
+ cur := obs.Rec.Move
+ if cur == nil || cur.ID != m.ID || cur.State != provision.OfflineFinishing || cur.OpenedAt == nil {
+ return none("the judged record no longer carries the move at finishing with its open time")
+ }
+
+ raw, err := w.offlineSealRecord(ctx, s, *obs.Rec)
+ if err != nil {
+ return err
+ }
+ seal, err := decodeSeal(raw, host, m.ID)
+ if err != nil {
+ return fmt.Errorf("offline move %s: the seal: %w", m.ID, err)
+ }
+ if seal == nil {
+ return fmt.Errorf("offline move %s: the seal is missing at finishing", m.ID)
+ }
+
+ prefix := activator.HibernatePrefix(cfg.OwnPrefix, host) + "/"
+ if cfg.Versioned == nil {
+ return errors.New("offline move: no versioned store on this box")
+ }
+ vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
+ if err != nil {
+ return fmt.Errorf("offline move %s: the versioned store for %s: %w", m.ID, cfg.OwnBucket, err)
+ }
+ listing, err := vs.ListVersions(ctx, cfg.OwnBucket, prefix)
+ if err != nil {
+ return fmt.Errorf("offline move %s: list the versions under %s: %w", m.ID, prefix, err)
+ }
+ inventory, unusable, err := inventoryOf(prefix, listing)
+ if err != nil {
+ return fmt.Errorf("offline move %s: %w", m.ID, err)
+ }
+ if unusable != "" {
+ return none("%s lists a version id that names no one version (the bucket wrote it unversioned or suspended)", unusable)
+ }
+ local, left, err := w.localInventory(host)
+ if err != nil {
+ return fmt.Errorf("offline move %s: %w", m.ID, err)
+ }
+ r := cfg.retiredAtGone(obs.Rec, cur, seal, prefix, inventory, local, left)
+ if err := offlinecleanup.ValidateRetired(r); err != nil {
+ return fmt.Errorf("offline move %s: %w", m.ID, err)
+ }
+ if err := createRetired(path, r); err != nil {
+ return fmt.Errorf("offline move %s: write %s: %w", m.ID, path, err)
+ }
+ log.Printf("moves: %s: offline move %s: cleanup obligation %s recorded (%d keys under %s)", host, m.ID, r.ID, len(inventory), prefix)
+ return nil
+}
+
+// retiredAtGone is the obligation as gone records it: rec is the judged
+// record, m its move at finishing, seal the move's seal; the state open at
+// Rev 1.
+func (c moveCfg) retiredAtGone(rec *provision.Record, m *provision.MoveState, seal *moves.OfflineReport, prefix string,
+ inventory []offlinecleanup.InventoryKey, local []offlinecleanup.LocalPath, left []offlinecleanup.KeptPath) offlinecleanup.Retired {
+ return offlinecleanup.Retired{
+ V: offlinecleanup.Version, ID: offlinecleanup.CleanupID(c.BoxID, rec.Host, m.ID), MoveID: m.ID, Host: rec.Host,
+ Account: rec.Account, App: rec.App, Box: c.BoxID, Incarnation: rec.Incarnation, From: m.From, To: m.To,
+ OpenAt: m.OpenedAt.UTC(), FinalTXID: seal.FinalTXID, TombstoneOwner: moves.TombstoneOwner(m.ID),
+ Bucket: c.OwnBucket, BucketRegion: c.OwnBucketRegion, Prefix: prefix,
+ BucketInventory: inventory, LocalInventory: local, LocalLeft: left,
+ WrittenAt: c.now().UTC(), State: offlinecleanup.StateOpen, Rev: 1,
+ }
+}
+
+// trackedAtGone is N5: the judged observation is valid, at a real
+// incarnation, with a live slot at that incarnation. An untracked record
+// (incarnation 0 and no slot), a migrating one, a stale or corrupt verdict
+// is not.
+func trackedAtGone(obs provision.Observation) bool {
+ return obs.Verdict == provision.LifetimeValid && obs.Rec != nil && obs.Rec.Incarnation > 0 && obs.Rec.IncarnationMigrating == "" &&
+ obs.Slot != nil && obs.Slot.State == provision.SlotLive && obs.Slot.Incarnation == obs.Rec.Incarnation
+}
+
+// inventoryOf groups a whole-prefix listing by key into the obligation's
+// inventory: every entry under prefix, object version or delete marker,
+// except the two excluded keys, each with its index in the listing (Seq);
+// per key the entries that are not current, oldest first
+// (offlinecleanup.DeletionLess), then the current one, whose id is the
+// key's Latest. A key with no current entry, or more than one, is an
+// error. The first entry whose id names no one version
+// (lease.UsableVersionID) is returned as unusable, with no inventory.
+func inventoryOf(prefix string, listing []lease.ObjectVersion) (inv []offlinecleanup.InventoryKey, unusable string, err error) {
+ byKey := map[string][]offlinecleanup.InventoryVersion{}
+ for i, v := range listing {
+ if !strings.HasPrefix(v.Key, prefix) || v.Key == prefix+offlinecleanup.LockKey || v.Key == prefix+offlinecleanup.BuildStateKey {
+ continue
+ }
+ if !lease.UsableVersionID(v.VersionID) {
+ return nil, fmt.Sprintf("%s@%q", v.Key, v.VersionID), nil
+ }
+ e := offlinecleanup.InventoryVersion{Seq: i, VersionID: v.VersionID, ETag: v.ETag, Size: v.Size, DeleteMarker: v.DeleteMarker,
+ Latest: v.IsLatest, LastModified: v.LastModified.UTC()}
+ if v.DeleteMarker {
+ e.ETag, e.Size = "", 0
+ }
+ byKey[v.Key] = append(byKey[v.Key], e)
+ }
+ keys := make([]string, 0, len(byKey))
+ for k := range byKey {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ for _, k := range keys {
+ var older []offlinecleanup.InventoryVersion
+ var latest []offlinecleanup.InventoryVersion
+ for _, e := range byKey[k] {
+ if e.Latest {
+ latest = append(latest, e)
+ } else {
+ older = append(older, e)
+ }
+ }
+ if len(latest) != 1 {
+ return nil, "", fmt.Errorf("the listing of %s has %d current entries, want 1", k, len(latest))
+ }
+ sort.SliceStable(older, func(i, j int) bool { return offlinecleanup.DeletionLess(older[i], older[j]) })
+ inv = append(inv, offlinecleanup.InventoryKey{Key: k, Latest: latest[0].VersionID, Versions: append(older, latest[0])})
+ }
+ return inv, "", nil
+}
+
+// localInventory is the live link and the programs file of host: absent,
+// named nowhere; present in a directory whose chain is root-only, in the
+// inventory with its identity (root removes it); present otherwise, left
+// as "writable layout" (on every built box the programs file is here: the
+// agent's unit has ReadWritePaths=/data).
+func (w *movesWorker) localInventory(host string) ([]offlinecleanup.LocalPath, []offlinecleanup.KeptPath, error) {
+ chain := w.cfg.rootOnlyChain
+ if chain == nil {
+ chain = offlinecleanup.RootOnlyChain
+ }
+ var inv []offlinecleanup.LocalPath
+ var left []offlinecleanup.KeptPath
+ for _, p := range []struct{ which, path string }{
+ {offlinecleanup.WhichLiveLink, release.LiveLink(w.cfg.OptDir, host)},
+ {offlinecleanup.WhichProgramsFile, release.ProgramsPath(w.cfg.DataDir, host)},
+ } {
+ id, ok, err := offlinecleanup.Identify(p.path)
+ if err != nil {
+ return nil, nil, fmt.Errorf("the %s: %w", p.which, err)
+ }
+ if !ok {
+ continue
+ }
+ rootOnly, err := chain(filepath.Dir(p.path))
+ if err != nil {
+ return nil, nil, fmt.Errorf("the %s's directory: %w", p.which, err)
+ }
+ if rootOnly {
+ inv = append(inv, offlinecleanup.LocalPath{Which: p.which, Identity: id})
+ } else {
+ left = append(left, offlinecleanup.KeptPath{Which: p.which, Reason: offlinecleanup.ReasonWritableLayout})
+ }
+ }
+ return inv, left, nil
+}
+
+// retiredScan is what scanRetired found at a move's retired.json: nothing
+// (both nil), a valid obligation of this box, or a corrupt stub. MovedTo
+// names where this scan moved a damaged file.
+type retiredScan struct {
+ Retired *offlinecleanup.Retired
+ Stub *offlinecleanup.CorruptStub
+ MovedTo string
+}
+
+// scanRetired is the corrupt scan (N6) of path, the retired.json of box's
+// obligation for move op on host. A valid Retired of that obligation, or a
+// valid CorruptStub of it, is answered as it is and never rewritten.
+// Anything else (bytes that are neither, another obligation's file, a file
+// that is not regular) is damage: it is moved aside create-only, to
+// retired.json.corrupt-1 when that name is free, else to the next free
+// retired.json.corrupt-<n>, so the first damaged file is never overwritten;
+// then the stub is written with FoundAt the mtime of .corrupt-1. An absent
+// path beside a .corrupt-1 (a crash between the move and the stub) gets
+// its stub too. The stub keeps the obligation listed as corrupt until an
+// operator repairs it.
+func scanRetired(path, box, host, op string) (retiredScan, error) {
+ fi, err := os.Lstat(path)
+ switch {
+ case errors.Is(err, fs.ErrNotExist):
+ if _, err := os.Lstat(path + ".corrupt-1"); errors.Is(err, fs.ErrNotExist) {
+ return retiredScan{}, nil
+ } else if err != nil {
+ return retiredScan{}, err
+ }
+ stub, err := writeCorruptStub(path, box, host, op)
+ return retiredScan{Stub: stub}, err
+ case err != nil:
+ return retiredScan{}, err
+ }
+ if fi.Mode().IsRegular() {
+ raw, err := os.ReadFile(path)
+ if err != nil {
+ return retiredScan{}, err
+ }
+ var r offlinecleanup.Retired
+ if json.Unmarshal(raw, &r) == nil && offlinecleanup.ValidateRetired(r) == nil && r.Box == box && r.Host == host && r.MoveID == op {
+ return retiredScan{Retired: &r}, nil
+ }
+ var c offlinecleanup.CorruptStub
+ if json.Unmarshal(raw, &c) == nil && offlinecleanup.ValidateCorruptStub(c, box, host, op) == nil {
+ return retiredScan{Stub: &c}, nil
+ }
+ }
+ moved, err := moveAsideCorrupt(path)
+ if err != nil {
+ return retiredScan{}, fmt.Errorf("move the damaged %s aside: %w", path, err)
+ }
+ log.Printf("moves: %s: offline move %s: retired.json is damaged; kept as %s", host, op, moved)
+ stub, err := writeCorruptStub(path, box, host, op)
+ return retiredScan{Stub: stub, MovedTo: moved}, err
+}
+
+// moveAsideCorrupt links path to the first free retired.json.corrupt-<n>
+// and then removes path. A name already holding this very file (a crash
+// after the link) is where it stays.
+func moveAsideCorrupt(path string) (string, error) {
+ fi, err := os.Lstat(path)
+ if err != nil {
+ return "", err
+ }
+ var dst string
+ for n := 1; ; n++ {
+ if n > 10000 {
+ return "", errors.New("no free corrupt-<n> name")
+ }
+ dst = fmt.Sprintf("%s.corrupt-%d", path, n)
+ err := os.Link(path, dst)
+ if err == nil {
+ break
+ }
+ if !errors.Is(err, fs.ErrExist) {
+ return "", err
+ }
+ if have, err := os.Lstat(dst); err == nil && os.SameFile(fi, have) {
+ break
+ }
+ }
+ if err := os.Remove(path); err != nil {
+ return "", err
+ }
+ return dst, syncDir(filepath.Dir(path))
+}
+
+// writeCorruptStub writes the obligation's stub at path, create-only, with
+// FoundAt the mtime of path's .corrupt-1.
+func writeCorruptStub(path, box, host, op string) (*offlinecleanup.CorruptStub, error) {
+ first, err := os.Lstat(path + ".corrupt-1")
+ if err != nil {
+ return nil, err
+ }
+ stub := offlinecleanup.CorruptStub{Corrupt: true, ID: offlinecleanup.CleanupID(box, host, op), Host: host, FoundAt: first.ModTime().UTC()}
+ if err := offlinecleanup.ValidateCorruptStub(stub, box, host, op); err != nil {
+ return nil, err
+ }
+ body, err := json.MarshalIndent(stub, "", " ")
+ if err != nil {
+ return nil, err
+ }
+ if err := createFile(path, body); err != nil {
+ return nil, err
+ }
+ return &stub, nil
+}
+
+// createRetired writes r at path create-only: a temporary file, fsync, a
+// link into place and a directory fsync. An existing file is never
+// replaced (the link fails, and the next pass scans what is there).
+func createRetired(path string, r offlinecleanup.Retired) error {
+ body, err := json.MarshalIndent(r, "", " ")
+ if err != nil {
+ return err
+ }
+ return createFile(path, body)
+}
+
+func createFile(path string, body []byte) error {
+ dir := filepath.Dir(path)
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ return err
+ }
+ tmp, err := writeSyncedTemp(dir, body)
+ if err != nil {
+ return err
+ }
+ defer os.Remove(tmp)
+ if err := os.Link(tmp, path); err != nil {
+ return err
+ }
+ return syncDir(dir)
+}
+
+// writeRetired replaces path with r, its Rev raised by one: a temporary
+// file, fsync, a rename and a directory fsync. Every write of the
+// obligation after the first goes through it, except Task 5a's
+// markPublished, which sets PublishedRev alone.
+func writeRetired(path string, r *offlinecleanup.Retired) error {
+ next := *r
+ next.Rev++
+ if err := offlinecleanup.ValidateRetired(next); err != nil {
+ return err
+ }
+ body, err := json.MarshalIndent(next, "", " ")
+ if err != nil {
+ return err
+ }
+ dir := filepath.Dir(path)
+ tmp, err := writeSyncedTemp(dir, body)
+ if err != nil {
+ return err
+ }
+ if err := os.Rename(tmp, path); err != nil {
+ os.Remove(tmp)
+ return err
+ }
+ if err := syncDir(dir); err != nil {
+ return err
+ }
+ *r = next
+ return nil
+}
+
+func writeSyncedTemp(dir string, body []byte) (string, error) {
+ f, err := os.CreateTemp(dir, ".retired-*.tmp")
+ if err != nil {
+ return "", err
+ }
+ _, err = f.Write(body)
+ if err == nil {
+ err = f.Sync()
+ }
+ if cerr := f.Close(); err == nil {
+ err = cerr
+ }
+ if err != nil {
+ os.Remove(f.Name())
+ return "", err
+ }
+ return f.Name(), nil
+}
+
+// noticeEvery logs a standing condition at most once per every per key.
+func (w *movesWorker) noticeEvery(key string, every time.Duration, format string, args ...any) {
+ now := time.Now()
+ key = "every:" + key
+ if last, ok := w.notices.Load(key); ok && now.Sub(last.(time.Time)) < every {
+ return
+ }
+ w.notices.Store(key, now)
+ log.Printf(format, args...)
+}
diff --git a/internal/host/movescleanup_retired_test.go b/internal/host/movescleanup_retired_test.go
new file mode 100644
index 000000000..09b56c43d
--- /dev/null
+++ b/internal/host/movescleanup_retired_test.go
@@ -0,0 +1,706 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "strings"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/offlinecleanup"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The cleanup obligation's evidence at gone (self-serve moves plan 4 Task
+// 1): the sealing source box records retired.json once, for a tracked
+// lifetime, before its gone report. Driven through offlineSourceGone, the
+// source step at offline-finishing, over the offline world's store and
+// boxes and leasetest's versioned model as the source's own bucket.
+
+const (
+ retBucket = "data-ie"
+ retRegion = "eu-west-1"
+ retPrefix = "instances/" + offHost + "/"
+)
+
+type retiredWorld struct {
+ *offWorld
+ fake *leasetest.VersionedBucket
+ ie *offBox // the sealer
+}
+
+// newRetiredWorld is offHost's offline move from the home to uk at
+// offline-finishing, sealed by ie-a with FinalTXID 77. tracked puts the
+// lifetime at incarnation 1 with a live slot; otherwise the record keeps
+// incarnation 0 and no slot (untracked, before the lazy migration).
+func newRetiredWorld(t *testing.T, tracked bool) *retiredWorld {
+ t.Helper()
+ wd := newOffWorld(t)
+ wd.judge = newBoxJudge(claimsOffEnv)
+ if tracked {
+ live := provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: wd.acct, App: "app"}
+ if _, err := provision.WriteHostIncarnation(context.Background(), wd.s, offHost, live, ""); err != nil {
+ t.Fatal(err)
+ }
+ wd.reincarnate(1)
+ }
+ fake := leasetest.NewVersionedBucket()
+ fake.AddBucket(retBucket)
+ rw := &retiredWorld{offWorld: wd, fake: fake}
+ rw.ie = rw.sourceBox("ie-a")
+ wd.start()
+ wd.seal()
+ wd.sealed()
+ wd.flip()
+ wd.landed()
+ wd.open()
+ wd.finishing()
+ rw.putSeal(77)
+ return rw
+}
+
+// sourceBox is a home box whose own bucket is the fake, as armMovesPass
+// arms one (CARLOS_HIBERNATE_BUCKET, _REGION and _PREFIX).
+func (rw *retiredWorld) sourceBox(name string) *offBox {
+ rw.t.Helper()
+ b := rw.box(name, "", nil)
+ c := &b.w.cfg
+ c.OwnBucket, c.OwnBucketRegion, c.OwnPrefix = retBucket, retRegion, "instances"
+ c.Versioned = func(_ context.Context, bucket, region string) (lease.VersionedStore, error) {
+ if bucket != retBucket || region != retRegion {
+ return nil, fmt.Errorf("versioned store for %s in %s", bucket, region)
+ }
+ return rw.fake, nil
+ }
+ return b
+}
+
+func (rw *retiredWorld) putSeal(txid uint64) {
+ rw.t.Helper()
+ items := map[string]moves.Check{}
+ for _, it := range moves.OfflineSourced() {
+ items[it.ID] = moves.Check{OK: true, Reason: "fixture"}
+ }
+ seal := moves.OfflineReport{V: moves.OfflineReportVersion, Op: offID, Host: offHost, Box: "ie-a", Role: moves.OfflineRoleSealer,
+ Phase: moves.OfflinePhaseSealed, At: rw.clock(), Items: items, TombstoneOwner: moves.TombstoneOwner(offID), FinalTXID: txid}
+ if err := moves.ValidateOfflineSeal(seal, offHost, offID); err != nil {
+ rw.t.Fatal(err)
+ }
+ raw, _ := json.Marshal(seal)
+ if _, err := rw.s.Put(context.Background(), moves.OfflineSealKey(offHost, offID), raw, ""); err != nil {
+ rw.t.Fatal(err)
+ }
+}
+
+// gone is one pass of b's source step at offline-finishing over s (the
+// world's store when nil).
+func (rw *retiredWorld) gone(b *offBox, s release.Store) error {
+ if s == nil {
+ s = rw.s
+ }
+ return b.w.offlineSourceStep(context.Background(), s, rw.record())
+}
+
+func retiredPath(b *offBox) string { return moves.RetiredFile(b.dd, offHost, offID) }
+
+func (rw *retiredWorld) retired(b *offBox) (offlinecleanup.Retired, bool) {
+ rw.t.Helper()
+ raw, err := os.ReadFile(retiredPath(b))
+ if errors.Is(err, os.ErrNotExist) {
+ return offlinecleanup.Retired{}, false
+ }
+ if err != nil {
+ rw.t.Fatal(err)
+ }
+ var r offlinecleanup.Retired
+ if err := json.Unmarshal(raw, &r); err != nil {
+ rw.t.Fatalf("retired.json: %v: %s", err, raw)
+ }
+ return r, true
+}
+
+func (rw *retiredWorld) goneReported(box string) bool {
+ rw.t.Helper()
+ r, ok := rw.reportIf(moves.OfflineRoleSource, box)
+ return ok && r.Phase == moves.OfflinePhaseGone
+}
+
+func (rw *retiredWorld) noRetired(b *offBox) {
+ rw.t.Helper()
+ if _, err := os.Lstat(retiredPath(b)); !errors.Is(err, os.ErrNotExist) {
+ rw.t.Fatalf("%s has a retired.json (%v)", b.name, err)
+ }
+}
+
+// key returns the inventory entry for key, or fails.
+func inventoryKey(t *testing.T, r offlinecleanup.Retired, key string) offlinecleanup.InventoryKey {
+ t.Helper()
+ for _, k := range r.BucketInventory {
+ if k.Key == key {
+ return k
+ }
+ }
+ t.Fatalf("%s is not inventoried: %+v", key, r.BucketInventory)
+ return offlinecleanup.InventoryKey{}
+}
+
+func versionIDs(k offlinecleanup.InventoryKey) []string {
+ var out []string
+ for _, v := range k.Versions {
+ out = append(out, v.VersionID)
+ }
+ return out
+}
+
+func TestRetiredJSONWrittenOnceAtGoneBySealer(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ ie := rw.ie
+ db := retPrefix + "db/generations/0001.ltx"
+ v1 := rw.fake.Put(retBucket, db, []byte("one"))
+ v2 := rw.fake.Put(retBucket, db, []byte("two!"))
+ v3 := rw.fake.Put(retBucket, db, []byte("three"))
+ files := retPrefix + "files.tar"
+ f1 := rw.fake.Put(retBucket, files, []byte("tar"))
+ marker := rw.fake.BareDelete(retBucket, files)
+ rw.fake.Put(retBucket, retPrefix+"lock.json", []byte(`{}`))
+ rw.fake.Put(retBucket, retPrefix+"build-state.json", []byte(`{}`))
+ listing, _ := rw.fake.ListVersions(context.Background(), retBucket, retPrefix)
+ etag := map[string]string{}
+ for _, v := range listing {
+ etag[v.VersionID] = v.ETag
+ }
+
+ ieB := rw.sourceBox("ie-b") // a source box that is not the sealer
+ if err := rw.gone(ieB, nil); err != nil {
+ t.Fatal(err)
+ }
+ rw.noRetired(ieB)
+ if !rw.goneReported("ie-b") {
+ t.Fatal("the non-sealer did not report gone")
+ }
+
+ if err := rw.gone(ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ r, ok := rw.retired(ie)
+ if !ok {
+ t.Fatal("no retired.json on the sealer")
+ }
+ if err := offlinecleanup.ValidateRetired(r); err != nil {
+ t.Fatal(err)
+ }
+ rec := rw.record()
+ want := offlinecleanup.Retired{V: 1, ID: offlinecleanup.CleanupID("ie-a", offHost, offID), MoveID: offID, Host: offHost, Account: rw.acct, App: "app",
+ Box: "ie-a", Incarnation: 1, From: "", To: "uk", FinalTXID: 77, TombstoneOwner: moves.TombstoneOwner(offID), Bucket: retBucket,
+ BucketRegion: retRegion, Prefix: retPrefix, State: offlinecleanup.StateOpen, Rev: 1}
+ got := r
+ got.BucketInventory, got.LocalInventory, got.LocalLeft, got.OpenAt, got.WrittenAt = nil, nil, nil, time.Time{}, time.Time{}
+ if fmt.Sprintf("%+v", got) != fmt.Sprintf("%+v", want) {
+ t.Fatalf("retired.json\n got %+v\n want %+v", got, want)
+ }
+ if rec.Move.OpenedAt == nil || !r.OpenAt.Equal(*rec.Move.OpenedAt) || !r.WrittenAt.Equal(rw.clock()) {
+ t.Fatalf("open_at %s (record %v), written_at %s", r.OpenAt, rec.Move.OpenedAt, r.WrittenAt)
+ }
+ if len(r.BucketInventory) != 2 {
+ t.Fatalf("inventory %+v", r.BucketInventory)
+ }
+ k := inventoryKey(t, r, db)
+ if strings.Join(versionIDs(k), ",") != strings.Join([]string{v1, v2, v3}, ",") || k.Latest != v3 || !k.Versions[2].Latest || k.Versions[0].Latest || k.Versions[1].Latest {
+ t.Fatalf("the three versions: %+v", k)
+ }
+ for _, v := range k.Versions {
+ if v.ETag == "" || v.ETag != etag[v.VersionID] || v.DeleteMarker || v.Size == 0 {
+ t.Fatalf("version %+v, listed etag %q", v, etag[v.VersionID])
+ }
+ }
+ k = inventoryKey(t, r, files)
+ if strings.Join(versionIDs(k), ",") != f1+","+marker || k.Latest != marker || !k.Versions[1].DeleteMarker || k.Versions[1].ETag != "" || k.Versions[0].ETag != etag[f1] {
+ t.Fatalf("the key whose current entry is a marker: %+v", k)
+ }
+ if !rw.goneReported("ie-a") {
+ t.Fatal("no gone report after retired.json")
+ }
+
+ // A second gone pass does not rewrite, whatever the bucket holds now.
+ before, _ := os.ReadFile(retiredPath(ie))
+ fi1, _ := os.Lstat(retiredPath(ie))
+ rw.fake.Put(retBucket, db, []byte("four"))
+ rw.advance(time.Minute)
+ if err := rw.gone(ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ after, _ := os.ReadFile(retiredPath(ie))
+ fi2, _ := os.Lstat(retiredPath(ie))
+ if string(before) != string(after) || !os.SameFile(fi1, fi2) {
+ t.Fatal("a second gone pass rewrote retired.json")
+ }
+ if l := rw.fake.Listings(); len(l) != 2 { // the fixture's own listing and the first gone's
+ t.Fatalf("listings %q: a second pass listed the bucket again", l)
+ }
+
+ // An identity error writes nothing and no gone report: the live link's
+ // directory is a regular file, so its Lstat fails with ENOTDIR.
+ rw2 := newRetiredWorld(t, true)
+ writeFile(t, filepath.Join(rw2.ie.opt, "live"), "not a directory")
+ err := rw2.gone(rw2.ie, nil)
+ if !errors.Is(err, syscall.ENOTDIR) {
+ t.Fatalf("an identity error answered %v", err)
+ }
+ rw2.noRetired(rw2.ie)
+ if rw2.goneReported("ie-a") {
+ t.Fatal("a gone report after an identity error")
+ }
+}
+
+// failingSlot fails every read of offHost's incarnation slot while on.
+type failingSlot struct {
+ release.Store
+ on bool
+}
+
+func (s *failingSlot) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if s.on && key == provision.HostIncarnationKey(offHost) {
+ return nil, "", errors.New("slot read: injected")
+ }
+ return s.Store.Get(ctx, key)
+}
+
+func TestUntrackedLifetimeWritesNoRetiredAndReportsGone(t *testing.T) {
+ ctx := context.Background()
+ t.Run("incarnation 0 with no slot", func(t *testing.T) {
+ rw := newRetiredWorld(t, false)
+ rw.fake.Put(retBucket, retPrefix+"db", []byte("x"))
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ rw.noRetired(rw.ie)
+ if !rw.goneReported("ie-a") {
+ t.Fatal("no gone report")
+ }
+ if len(rw.fake.Listings()) != 0 {
+ t.Fatal("an untracked lifetime's bucket was listed")
+ }
+ })
+ t.Run("a migrating record", func(t *testing.T) {
+ rw := newRetiredWorld(t, false)
+ rec, etag, err := provision.GetRecordETag(ctx, rw.s, rw.acct, "app", offHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec.IncarnationMigrating = "lazy-1"
+ if err := provision.UpdateRecord(ctx, rw.s, *rec, etag); err != nil {
+ t.Fatal(err)
+ }
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ rw.noRetired(rw.ie)
+ if !rw.goneReported("ie-a") {
+ t.Fatal("no gone report")
+ }
+ })
+ t.Run("a stale verdict", func(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ _, etag, err := provision.ReadHostIncarnation(ctx, rw.s, offHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ next := provision.HostIncarnation{Incarnation: 2, State: provision.SlotLive, Account: rw.acct, App: "app"}
+ if _, err := provision.WriteHostIncarnation(ctx, rw.s, offHost, next, etag); err != nil {
+ t.Fatal(err)
+ }
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ rw.noRetired(rw.ie)
+ if !rw.goneReported("ie-a") {
+ t.Fatal("no gone report")
+ }
+ })
+ t.Run("an unreadable lifetime", func(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ fs := &failingSlot{Store: rw.s, on: true}
+ if err := rw.gone(rw.ie, fs); err == nil {
+ t.Fatal("an unreadable lifetime was not an error")
+ }
+ rw.noRetired(rw.ie)
+ if rw.goneReported("ie-a") {
+ t.Fatal("a gone report while the lifetime is unreadable")
+ }
+ fs.on = false
+ if err := rw.gone(rw.ie, fs); err != nil {
+ t.Fatal(err)
+ }
+ if _, ok := rw.retired(rw.ie); !ok || !rw.goneReported("ie-a") {
+ t.Fatal("the next pass wrote neither retired.json nor the gone report")
+ }
+ })
+}
+
+func TestRetiredPrefixIsNormalisedOnce(t *testing.T) {
+ for _, own := range []string{"instances", "instances/"} {
+ t.Run(own, func(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ rw.ie.w.cfg.OwnPrefix = own
+ for _, k := range []string{"db/0001.ltx", "lock.json", "build-state.json"} {
+ rw.fake.Put(retBucket, retPrefix+k, []byte(k))
+ }
+ rw.fake.Put(retBucket, retPrefix+"lock.json", []byte("again")) // excluded with all its versions
+ for _, k := range []string{"instances/shop.example.co/db", "instances/shop.example.com2/db", "instances/shop.example.com.au/db", "instances/db"} {
+ rw.fake.Put(retBucket, k, []byte("neighbour"))
+ }
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ r, ok := rw.retired(rw.ie)
+ if !ok {
+ t.Fatal("no retired.json")
+ }
+ if r.Prefix != "instances/shop.example.com/" {
+ t.Fatalf("prefix %q", r.Prefix)
+ }
+ if l := rw.fake.Listings(); len(l) != 1 || l[0] != "instances/shop.example.com/" {
+ t.Fatalf("listed %q", l)
+ }
+ if len(r.BucketInventory) != 1 || r.BucketInventory[0].Key != retPrefix+"db/0001.ltx" {
+ t.Fatalf("inventory %+v", r.BucketInventory)
+ }
+ if err := offlinecleanup.ValidateRetired(r); err != nil {
+ t.Fatal(err)
+ }
+ })
+ }
+}
+
+func TestInventoryOrderIsOldestFirstLatestLast(t *testing.T) {
+ at := time.Date(2026, 10, 20, 8, 0, 0, 0, time.UTC)
+ for _, reverse := range []bool{false, true} {
+ t.Run(fmt.Sprintf("ties reversed %v", reverse), func(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ key := retPrefix + "db/0001.ltx"
+ rw.fake.PutAt(retBucket, key, []byte("a"), at.Add(-time.Minute))
+ rw.fake.PutAt(retBucket, key, []byte("b"), at)
+ rw.fake.PutAt(retBucket, key, []byte("c"), at)
+ rw.fake.PutAt(retBucket, key, []byte("d"), at) // the latest, tied with the two before it
+ if reverse {
+ rw.fake.ListHook = func(_ string, vs []lease.ObjectVersion) []lease.ObjectVersion {
+ vs[1], vs[2] = vs[2], vs[1] // the tied noncurrent pair listed the other way round
+ return vs
+ }
+ }
+ listing, _ := rw.fake.ListVersions(context.Background(), retBucket, retPrefix)
+ seq := map[string]int{}
+ for i, v := range listing {
+ seq[v.VersionID] = i
+ }
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ r, ok := rw.retired(rw.ie)
+ if !ok {
+ t.Fatal("no retired.json")
+ }
+ k := inventoryKey(t, r, key)
+ last := k.Versions[len(k.Versions)-1]
+ if len(k.Versions) != 4 || k.Latest != last.VersionID || !last.Latest || last.VersionID != listing[0].VersionID {
+ t.Fatalf("latest: %+v", k)
+ }
+ if !k.Versions[0].LastModified.Equal(at.Add(-time.Minute)) {
+ t.Fatalf("the oldest is not first: %+v", k.Versions)
+ }
+ // The tied pair by Seq descending, Seq being its index in this
+ // listing's answer.
+ if k.Versions[1].Seq != seq[k.Versions[1].VersionID] || k.Versions[2].Seq != seq[k.Versions[2].VersionID] || k.Versions[1].Seq < k.Versions[2].Seq {
+ t.Fatalf("the tie: %+v (listing %+v)", k.Versions, listing)
+ }
+ if err := offlinecleanup.ValidateRetired(r); err != nil {
+ t.Fatal(err)
+ }
+ })
+ }
+ t.Run("two latest entries", func(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ key := retPrefix + "db/0001.ltx"
+ rw.fake.Put(retBucket, key, []byte("a"))
+ rw.fake.Put(retBucket, key, []byte("b"))
+ rw.fake.ListHook = func(_ string, vs []lease.ObjectVersion) []lease.ObjectVersion {
+ for i := range vs {
+ vs[i].IsLatest = true
+ }
+ return vs
+ }
+ if err := rw.gone(rw.ie, nil); err == nil {
+ t.Fatal("a key with two latest entries was accepted")
+ }
+ rw.noRetired(rw.ie)
+ if rw.goneReported("ie-a") {
+ t.Fatal("a gone report after a refused listing")
+ }
+ })
+}
+
+func TestNullVersionIDWritesNoRetired(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ rw.fake.Put(retBucket, retPrefix+"db/0001.ltx", []byte("real"))
+ rw.fake.PutNull(retBucket, retPrefix+"files.tar", []byte("written unversioned"))
+ if err := rw.ie.w.recordRetired(context.Background(), rw.s, rw.record()); err != nil {
+ t.Fatalf("recordRetired: %v", err)
+ }
+ rw.noRetired(rw.ie)
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ rw.noRetired(rw.ie)
+ if !rw.goneReported("ie-a") {
+ t.Fatal("no gone report")
+ }
+
+ // ValidateRetired names a null id in an otherwise valid inventory.
+ ok := newRetiredWorld(t, true)
+ ok.fake.Put(retBucket, retPrefix+"db/0001.ltx", []byte("one"))
+ ok.fake.Put(retBucket, retPrefix+"db/0001.ltx", []byte("two"))
+ if err := ok.gone(ok.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ r, _ := ok.retired(ok.ie)
+ if err := offlinecleanup.ValidateRetired(r); err != nil {
+ t.Fatal(err)
+ }
+ r.BucketInventory[0].Versions[0].VersionID = "null"
+ err := offlinecleanup.ValidateRetired(r)
+ if err == nil || !strings.Contains(err.Error(), `"null"`) || !strings.Contains(err.Error(), r.BucketInventory[0].Key) {
+ t.Fatalf("a null id: %v", err)
+ }
+}
+
+func TestRealIDsOnASuspendedBucketStillWriteRetired(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ rw.fake.Put(retBucket, retPrefix+"db/0001.ltx", []byte("before the suspension"))
+ rw.fake.SetSuspended(retBucket, true)
+ rw.fake.Put(retBucket, "instances/other.example.com/db", []byte("after, elsewhere")) // a null id outside the prefix
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ r, ok := rw.retired(rw.ie)
+ if !ok {
+ t.Fatal("no retired.json: the listing check is about ids, not the bucket's status")
+ }
+ if err := offlinecleanup.ValidateRetired(r); err != nil {
+ t.Fatal(err)
+ }
+ if !rw.goneReported("ie-a") {
+ t.Fatal("no gone report")
+ }
+}
+
+func TestWritablePathsAreLeftNotRequested(t *testing.T) {
+ setup := func(t *testing.T) *retiredWorld {
+ rw := newRetiredWorld(t, true)
+ link := release.LiveLink(rw.ie.opt, offHost)
+ if err := os.MkdirAll(filepath.Dir(link), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(filepath.Join(rw.ie.opt, "versions", "v42"), link); err != nil {
+ t.Fatal(err)
+ }
+ writeFile(t, release.ProgramsPath(rw.ie.dd, offHost), `{"schema":1,"programs":["./cmd/shop"]}`)
+ return rw
+ }
+ t.Run("a directory the test uid owns", func(t *testing.T) {
+ rw := setup(t)
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ r, _ := rw.retired(rw.ie)
+ want := []offlinecleanup.KeptPath{{Which: offlinecleanup.WhichLiveLink, Reason: "writable layout"}, {Which: offlinecleanup.WhichProgramsFile, Reason: "writable layout"}}
+ if len(r.LocalInventory) != 0 || fmt.Sprint(r.LocalLeft) != fmt.Sprint(want) {
+ t.Fatalf("inventory %+v, left %+v", r.LocalInventory, r.LocalLeft)
+ }
+ })
+ t.Run("a root-only chain, injected", func(t *testing.T) {
+ rw := setup(t)
+ var asked []string
+ rw.ie.w.cfg.rootOnlyChain = func(dir string) (bool, error) {
+ asked = append(asked, dir)
+ return true, nil
+ }
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ r, _ := rw.retired(rw.ie)
+ link, programs := release.LiveLink(rw.ie.opt, offHost), release.ProgramsPath(rw.ie.dd, offHost)
+ if fmt.Sprint(asked) != fmt.Sprint([]string{filepath.Dir(link), filepath.Dir(programs)}) {
+ t.Fatalf("chains asked %q", asked)
+ }
+ if len(r.LocalLeft) != 0 || len(r.LocalInventory) != 2 {
+ t.Fatalf("inventory %+v, left %+v", r.LocalInventory, r.LocalLeft)
+ }
+ for i, p := range []string{link, programs} {
+ id, _, _ := offlinecleanup.Identify(p)
+ if got := r.LocalInventory[i]; got.Identity != id || got.Which != []string{offlinecleanup.WhichLiveLink, offlinecleanup.WhichProgramsFile}[i] {
+ t.Fatalf("%s: %+v, want %+v", p, got, id)
+ }
+ }
+ if r.LocalInventory[0].Identity.Mode != offlinecleanup.ModeSymlink || r.LocalInventory[0].Identity.LinkTarget != filepath.Join(rw.ie.opt, "versions", "v42") {
+ t.Fatalf("the live link's identity: %+v", r.LocalInventory[0].Identity)
+ }
+ })
+ t.Run("absent paths are named nowhere", func(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ r, _ := rw.retired(rw.ie)
+ if len(r.LocalInventory) != 0 || len(r.LocalLeft) != 0 {
+ t.Fatalf("inventory %+v, left %+v", r.LocalInventory, r.LocalLeft)
+ }
+ })
+}
+
+func TestCorruptStubIsValidAndKeepsTheFirstDamage(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ path := retiredPath(rw.ie)
+ if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(path, []byte(`{"v":1,"id":`), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ damagedAt := time.Date(2026, 10, 19, 7, 30, 0, 0, time.UTC)
+ if err := os.Chtimes(path, damagedAt, damagedAt); err != nil {
+ t.Fatal(err)
+ }
+
+ // The gone step finds it and keeps it: moved aside, a stub in place,
+ // and the gone report proceeds.
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ stub := func() offlinecleanup.CorruptStub {
+ t.Helper()
+ raw, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var c offlinecleanup.CorruptStub
+ if err := json.Unmarshal(raw, &c); err != nil {
+ t.Fatal(err)
+ }
+ if err := offlinecleanup.ValidateCorruptStub(c, "ie-a", offHost, offID); err != nil {
+ t.Fatal(err)
+ }
+ return c
+ }
+ if c := stub(); !c.FoundAt.Equal(damagedAt) {
+ t.Fatalf("found_at %s, want .corrupt-1's mtime %s", c.FoundAt, damagedAt)
+ }
+ if raw, _ := os.ReadFile(path + ".corrupt-1"); string(raw) != `{"v":1,"id":` {
+ t.Fatalf(".corrupt-1 holds %q", raw)
+ }
+ if !rw.goneReported("ie-a") {
+ t.Fatal("no gone report")
+ }
+ if len(rw.fake.Listings()) != 0 {
+ t.Fatal("a corrupt obligation listed the bucket")
+ }
+
+ // A second scan changes nothing.
+ before, _ := os.ReadFile(path)
+ fi1, _ := os.Lstat(path)
+ sc, err := scanRetired(path, "ie-a", offHost, offID)
+ if err != nil || sc.Stub == nil || sc.MovedTo != "" {
+ t.Fatalf("a second scan: %+v %v", sc, err)
+ }
+ after, _ := os.ReadFile(path)
+ fi2, _ := os.Lstat(path)
+ if string(before) != string(after) || !os.SameFile(fi1, fi2) {
+ t.Fatal("a valid stub was rewritten")
+ }
+ if _, err := os.Lstat(path + ".corrupt-2"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatal("a second scan moved something aside")
+ }
+
+ // A damaged stub goes to .corrupt-2; the first damage stays, and the
+ // new stub's found_at is still .corrupt-1's.
+ if err := os.WriteFile(path, []byte(`{"corrupt":true,"id":"cu_0","host":"shop.example.com","found_at":"2026-10-24T00:00:00Z"}`), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ sc, err = scanRetired(path, "ie-a", offHost, offID)
+ if err != nil || sc.Stub == nil || sc.MovedTo != path+".corrupt-2" {
+ t.Fatalf("a damaged stub: %+v %v", sc, err)
+ }
+ if raw, _ := os.ReadFile(path + ".corrupt-1"); string(raw) != `{"v":1,"id":` {
+ t.Fatalf(".corrupt-1 was overwritten: %q", raw)
+ }
+ if raw, _ := os.ReadFile(path + ".corrupt-2"); !strings.Contains(string(raw), `"cu_0"`) {
+ t.Fatalf(".corrupt-2 holds %q", raw)
+ }
+ if c := stub(); !c.FoundAt.Equal(damagedAt) {
+ t.Fatalf("found_at %s after the second damage, want %s", c.FoundAt, damagedAt)
+ }
+
+ // A valid Retired of another obligation (another box's id) is damage
+ // here too.
+ other := newRetiredWorld(t, true)
+ if err := other.gone(other.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ raw, _ := os.ReadFile(retiredPath(other.ie))
+ if sc, err := scanRetired(retiredPath(other.ie), "ie-b", offHost, offID); err != nil || sc.MovedTo == "" || sc.Stub == nil {
+ t.Fatalf("another box's obligation: %+v %v", sc, err)
+ }
+ if moved, _ := os.ReadFile(retiredPath(other.ie) + ".corrupt-1"); string(moved) != string(raw) {
+ t.Fatal("the other obligation's file was not kept whole")
+ }
+ // And a valid Retired of this one is found as it is.
+ third := newRetiredWorld(t, true)
+ if err := third.gone(third.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ if sc, err := scanRetired(retiredPath(third.ie), "ie-a", offHost, offID); err != nil || sc.Retired == nil || sc.MovedTo != "" {
+ t.Fatalf("a valid obligation: %+v %v", sc, err)
+ }
+}
+
+// writeRetired replaces the file with Rev raised by one, and refuses an
+// obligation that does not validate without touching the file.
+func TestWriteRetiredRaisesRevAndReplaces(t *testing.T) {
+ rw := newRetiredWorld(t, true)
+ rw.fake.Put(retBucket, retPrefix+"db", []byte("x"))
+ if err := rw.gone(rw.ie, nil); err != nil {
+ t.Fatal(err)
+ }
+ path := retiredPath(rw.ie)
+ r, _ := rw.retired(rw.ie)
+ r.State = offlinecleanup.StateFinished
+ if err := writeRetired(path, &r); err != nil {
+ t.Fatal(err)
+ }
+ if got, _ := rw.retired(rw.ie); got.Rev != 2 || got.State != offlinecleanup.StateFinished || r.Rev != 2 {
+ t.Fatalf("after a write: file rev %d state %s, caller's rev %d", got.Rev, got.State, r.Rev)
+ }
+ bad := r
+ bad.State = "nowhere"
+ if err := writeRetired(path, &bad); err == nil {
+ t.Fatal("an invalid obligation was written")
+ }
+ if got, _ := rw.retired(rw.ie); got.Rev != 2 || bad.Rev != 2 {
+ t.Fatalf("a refused write changed rev: file %d, caller %d", got.Rev, bad.Rev)
+ }
+ // createRetired never replaces.
+ if err := createRetired(path, r); err == nil {
+ t.Fatal("createRetired replaced an existing file")
+ }
+ if ents, _ := os.ReadDir(filepath.Dir(path)); len(ents) != 1 {
+ t.Fatalf("temporary files left: %v", ents)
+ }
+}
diff --git a/internal/host/movesoffline.go b/internal/host/movesoffline.go
index afe3a5517..23fc04525 100644
--- a/internal/host/movesoffline.go
+++ b/internal/host/movesoffline.go
@@ -1490,6 +1490,11 @@ func (w *movesWorker) offlineSourcePark(ctx context.Context, s release.Store, re
// holds the row, so reconcile removes it; once it is gone, report gone. The
// fence was lifted at open and is lifted again, should a restart have
// re-fenced in between.
+//
+// On the move's sealer, before the gone report, the cleanup obligation's
+// evidence is recorded (recordRetired, self-serve moves plan 4): an error
+// there writes no gone report this pass; an untracked lifetime is not an
+// error, and its gone report is written as before.
func (w *movesWorker) offlineSourceGone(ctx context.Context, s release.Store, rec provision.Record) error {
m, host := rec.Move, rec.Host
rt, err := w.offlineRow(host)
@@ -1501,6 +1506,11 @@ func (w *movesWorker) offlineSourceGone(ctx context.Context, s release.Store, re
w.notice(host+"#offline-gone", "moves: %s: offline move %s: waiting for reconcile to remove the row", host, m.ID)
return nil
}
+ if _, _, sealer, _ := w.cfg.offlineRoles(m); sealer {
+ if err := w.recordRetired(ctx, s, rec); err != nil {
+ return err
+ }
+ }
r, err := w.sourceReport(host, m, nil, false, moves.OfflinePhaseGone)
if err != nil {
return err
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 6c51cded7..41310ad98 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -885,6 +885,11 @@ type moveCfg struct {
LocalPosition func(dbPath string) (uint64, error)
Timers func(host string) ([]registry.Timer, error)
InputVersions func(ctx context.Context, s release.Store, rec provision.Record) (bundle map[string]string, overlay string, err error)
+ // rootOnlyChain is offlinecleanup.RootOnlyChain, which the gone
+ // evidence asks of each local path's directory (recordRetired). A test
+ // seam: nil in production; only this package's tests set it
+ // (offlinecleanup's TestEveryE2ESeamHasOneSetter).
+ rootOnlyChain func(dir string) (bool, error)
// cutFault is a test seam: called after each side effect of the cut
// with the point's name, an error being a crash there. Nil in
// production.
diff --git a/internal/release/programhistory.go b/internal/release/programhistory.go
index 66fa5f12f..029275864 100644
--- a/internal/release/programhistory.go
+++ b/internal/release/programhistory.go
@@ -17,6 +17,11 @@ func programPath(bakeDir, host string) string {
return filepath.Join(bakeDir, "programs", host+".json")
}
+// ProgramsPath is where the program history of host lives in bakeDir,
+// <bakeDir>/programs/<host>.json: the path a finished move's cleanup
+// inventories and root removes (self-serve moves plan 4).
+func ProgramsPath(bakeDir, host string) string { return programPath(bakeDir, host) }
+
func ReadPrograms(bakeDir, host string) []string {
if strings.ContainsAny(host, `/\`) {
return nil