amadan

carlos / platform Public

offline-move: the sealer

The sealer of the offline operator move (plan Task 4): own report held,
fenced, quiesced, no process, no journal, no sidecar; checkpoint (lease,
sync, release); Tombstone(moved:<id>, 7 days); only then the final
position, with every source box at or below it; CarryRequest and a
byte-final mirror round into the target's moves/<host>/; the sealed files
archive, timers snapshot and env report under moves/@offline/<host>/<id>/;
env and key digests, env pin, release, replication, next due; seal.json
written once through putOfflineCreateOnly; CAS OfflineSealedBy with
evidence <seal key>@<sha256>.

After the seal every pass reads LockState first: a tombstone that is
absent, expired, lifted or someone else's is a lost stop (report failed,
page once, never taken again); only this move's is re-asserted. Finishing
lifts it once every source box reports gone; the revert lifts only this
move's, deletes the two staging prefixes and unfences.

archive.BuildFiltered and layout.NestedPlatformNames, and a ListObjectsV2
(prefix, delimiter, max-keys, continuation token) in leasetest's S3
server so real litestream replicas run against it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude f7af0242050d4b6183882371aa715da72446b3b0 parent 2639808
9 files changed, +2724 −20
  • internal/host/movesoffline_seal.go +1067 −14
  • internal/host/movesoffline_seal_test.go +1181 −0
  • internal/layout/plan.go +35 −0
  • internal/layout/plan_test.go +38 −0
  • internal/lease/leasetest/s3server.go +126 −1
  • internal/lease/leasetest/s3server_test.go +132 −0
  • internal/moves/archive/archive.go +32 −5
  • internal/moves/archive/archive_test.go +92 −0
  • internal/moves/archive/archive_unix_test.go +21 −0
diff --git a/internal/host/movesoffline_seal.go b/internal/host/movesoffline_seal.go
index e13489c57..f0e7b62b6 100644
--- a/internal/host/movesoffline_seal.go
+++ b/internal/host/movesoffline_seal.go
@@ -2,27 +2,1080 @@ package host
import (
"context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+ "github.com/superfly/ltx"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/alerting"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/instancekey"
+ "github.com/carlosframework/platform/internal/layout"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/park"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
-// offlineSealerStep is the sealer's part of an offline move
-// (docs/superpowers/plans/2026-10-20-offline-operator-move.md, Task 4, which
-// owns this file). offlineRound calls it on the box whose moveCfg.Box equals
-// rec.Move.Sealer, in every offline state, after this box's source step has
-// written its report for the pass. rec carries an offline move
-// (rec.Move.Offline()) and the box is capable (moveCfg.OfflineCapable).
-// Every step must be an ensure; an error is logged by the pass, throttled
-// per host, and the next pass runs the step again.
+// The sealer of an offline operator move (docs/superpowers/plans/2026-10-20-
+// offline-operator-move.md, Task 4): the one source box that turns "every
+// source box holds, fences and reports" into a stop at the data layer, and
+// records what the target must land.
+//
+// The tombstone is the linearization point of the whole move. Every wake on
+// every box of the source region takes the host's single-writer lease
+// (instances/<host>/lock.json in the source's own bucket) before it can
+// write anything, and from the tombstone on that lease is held by
+// moved:<id>. The order is the plan's (Review Focus 2):
//
-// The immutable seal record (moves.OfflineSealKey(host, id), added on
-// Task 6's branch) is written once through putOfflineCreateOnly, never
-// rewritten: on a re-run it returns the stored body with created false, and
-// the step compares instead of overwriting. The per-pass sealer report goes
-// through writeOfflineReport, which replaces.
+// 1. this box's own report for the pass says held, fenced, quiesced, no
+// process and no layout journal, and no source box keeps a sidecar;
+// 2. the local database is checkpointed when it is ahead of the replica
+// (cfg.Checkpoint takes the lease, syncs and releases it);
+// 3. lease.Tombstone, which never replaces a live holder;
+// 4. only then the final position is read from the replica, and every
+// source box's local position must be at or below it;
+// 5. CarryRequest, a final MirrorPrefix round with nothing deferred, and
+// the staged listing;
+// 6. to 9. the files archive, the timers, the env and its pin, the release,
+// replication and next due;
+// 10. seal.json, written once and never rewritten (putOfflineCreateOnly),
+// then the record's CAS to offline-sealed.
//
-// Task 3 leaves it a no-op.
+// Every step is an ensure. Until seal.json exists the whole seal is taken
+// again on each pass (a tombstone that is already this move's is a no-op,
+// the mirror resumes from its receipts, the archive is rebuilt and its
+// objects overwritten), so a crash anywhere before it converges. Once
+// seal.json exists the sealed position is fixed, and from then on the
+// tombstone is never taken again: each pass reads lease.LockState first,
+// and a tombstone that is absent, expired or anyone else's is a lost stop.
+// The pass reports failed and pages; the operator aborts. Only a tombstone
+// held by this move is re-asserted (lease.Reassert, If-Match on the ETag it
+// reads). After it is lifted (finish, abort) it is never re-asserted.
+
+// The objects the sealer writes beside the sealed archive, under
+// moves.OfflineArchivePrefix(host, id) in the target's data bucket. The
+// landing (Task 5) reads them there.
+const (
+ // offlineArchiveSealed is the sealed files archive (archive.Seal of
+ // archive.BuildFiltered's output). Its sha256 is the seal's
+ // FilesSealed.
+ offlineArchiveSealed = "files.sealed"
+ // offlineArchiveManifest is its archive.Manifest (JSON), with the
+ // sealed data key and every entry.
+ offlineArchiveManifest = "files.json"
+ // offlineTimersObject is the moves.TimersSnapshot of the row's active
+ // timers. moves.TimersDigest of its rows is the seal's TimersDigest.
+ offlineTimersObject = "timers.json"
+ // offlineEnvReportObject is the source's moves.EnvReport (names and
+ // HMACs, never a value), so the landing can name the variables that
+ // differ (moves.CompareEnv) when the env pin does not match. Its
+ // Digest is the seal's EnvDigest.
+ offlineEnvReportObject = "env-report.json"
+)
+
+// offlineSealFresh is how old this box's own source report may be when the
+// sealer reads it: the source step writes it earlier in the same pass, so
+// anything older means that step failed this pass.
+const offlineSealFresh = 2 * time.Minute
+
+// offlinePageLostStop is the page rule of a lost stop: one page per move.
+const offlinePageLostStop = "offline-lost-stop"
+
+// offlineSealFaults is the sealer's test seam: a worker's entry is called
+// after each step with the step's name, and an error returned is a crash
+// there. Keyed by the worker, so tests that run side by side never share
+// one. Empty in production.
+var offlineSealFaults sync.Map // *movesWorker → func(point string) error
+
+func (w *movesWorker) sealFault(point string) error {
+ if f, ok := offlineSealFaults.Load(w); ok {
+ return f.(func(string) error)(point)
+ }
+ return nil
+}
+
+// offlineSealerStep is the sealer's part of an offline move. offlineRound
+// calls it on the box whose label is rec.Move.Sealer, after this box's own
+// source step has written its report for the pass.
func (w *movesWorker) offlineSealerStep(ctx context.Context, s release.Store, rec provision.Record) error {
+ m := rec.Move
+ if !m.Offline() || w.cfg.Box == "" || m.Sealer != w.cfg.Box {
+ return nil
+ }
+ switch m.State {
+ case provision.OfflineSealing:
+ return w.offlineSeal(ctx, s, rec)
+ case provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded, provision.OfflineOpen:
+ raw, err := w.offlineSealRecord(ctx, s, rec)
+ if err != nil {
+ return err
+ }
+ return w.offlineKeepStop(ctx, s, rec, raw)
+ case provision.OfflineFinishing:
+ return w.offlineFinishLift(ctx, s, rec)
+ case provision.OfflineReverting:
+ return w.offlineSealRevert(ctx, s, rec)
+ }
+ return nil
+}
+
+// tombstoneAt is where the host's single-writer lease lives in this box's
+// own bucket: the activator's lease location, as ExecBackend.Acquire takes
+// it for every wake.
+func (c moveCfg) tombstoneAt(host string) (bucket, location, region string, err error) {
+ if c.OwnBucket == "" {
+ return "", "", "", errors.New("offline move: this box has no hibernate bucket")
+ }
+ return c.OwnBucket, activator.HibernatePrefix(c.OwnPrefix, host), c.OwnBucketRegion, nil
+}
+
+// tombState is the lease as lease.LockState read it: an absent lock has no
+// owner and a zero expiry.
+type tombState struct {
+ owner string
+ expires time.Time
+}
+
+// mine reports a tombstone this move holds now. "Expired" is the lease's
+// own reading (Acquire takes a lease the moment its expiry has passed), so
+// it is judged against the wall clock the lease is written by.
+func (t tombState) mine(id string) bool {
+ return t.owner == moves.TombstoneOwner(id) && time.Now().Before(t.expires)
+}
+
+// free reports a lease nothing holds: absent, or past its expiry, whoever
+// wrote it (a lifted tombstone included: lease.IsLiftedTombstone).
+func (t tombState) free() bool {
+ return t.owner == "" || !time.Now().Before(t.expires)
+}
+
+func (t tombState) String() string {
+ switch {
+ case t.owner == "":
+ return "absent"
+ case lease.IsLiftedTombstone(t.owner, t.expires) && t.expires.Unix() == 0:
+ // LiftTombstone's expiry: the epoch.
+ return fmt.Sprintf("lifted (%s)", t.owner)
+ case t.free():
+ return fmt.Sprintf("expired (%s at %s)", t.owner, t.expires.UTC().Format(time.RFC3339))
+ }
+ return fmt.Sprintf("held by %s until %s", t.owner, t.expires.UTC().Format(time.RFC3339))
+}
+
+// offlineLock reads the host's lease.
+func (w *movesWorker) offlineLock(ctx context.Context, host string) (tombState, error) {
+ bucket, loc, region, err := w.cfg.tombstoneAt(host)
+ if err != nil {
+ return tombState{}, err
+ }
+ _, owner, expires, err := lease.LockState(ctx, bucket, loc, region)
+ if err != nil {
+ return tombState{}, fmt.Errorf("read the lease of %s: %w", host, err)
+ }
+ return tombState{owner: owner, expires: expires}, nil
+}
+
+// offlineTarget is the target's coordinates as its edges file publishes
+// them: the box at move.to (the lexically first, as every move reads the
+// fleet), matched exactly and then by placement under the home binding.
+func (w *movesWorker) offlineTarget(ctx context.Context, s release.Store, m *provision.MoveState) (edgesFile, error) {
+ fleet, err := loadMoveFleet(ctx, s)
+ if err != nil {
+ return edgesFile{}, fmt.Errorf("read the fleet's edges files: %w", err)
+ }
+ if t, ok := fleet.targets[m.To]; ok {
+ return t, nil
+ }
+ codes := make([]string, 0, len(fleet.targets))
+ for code := range fleet.targets {
+ codes = append(codes, code)
+ }
+ sort.Strings(codes)
+ for _, code := range codes {
+ if park.PlacementMatches(w.cfg.ParkHomeRegion, m.To, park.PlacementRegion(code)) {
+ return fleet.targets[code], nil
+ }
+ }
+ return edgesFile{}, fmt.Errorf("the target region %q publishes no data bucket yet", m.To)
+}
+
+// offlineMirror is the mirror pair between this box's bucket and region.
+func (w *movesWorker) offlineMirror(ctx context.Context, region string) (lease.MirrorStore, error) {
+ if w.cfg.Mirror == nil {
+ return nil, errors.New("offline move: no S3 mirror on this box")
+ }
+ return w.mirror(ctx, region)
+}
+
+// replicaPosition is the newest TXID of the replica listed in objs under
+// prefix (newestLTX), 0 when it holds no LTX file.
+func replicaPosition(objs map[string]lease.MirrorObject, prefix string) uint64 {
+ newest := newestLTX(objs, prefix, nil)
+ if newest == "" {
+ return 0
+ }
+ _, maxTX, err := ltx.ParseFilename(filepath.Base(newest))
+ if err != nil {
+ return 0
+ }
+ return uint64(maxTX)
+}
+
+// sealRefusal is a seal item that failed for a reason a later pass will
+// meet again: the pass reports it (phase failed) and the operator aborts.
+type sealRefusal struct {
+ item, reason string
+}
+
+func (r *sealRefusal) Error() string { return r.item + ": " + r.reason }
+
+func sealRefuse(item, format string, args ...any) error {
+ return &sealRefusal{item: item, reason: fmt.Sprintf(format, args...)}
+}
+
+// offlineSeal is the sealer's step in offline-sealing.
+func (w *movesWorker) offlineSeal(ctx context.Context, s release.Store, rec provision.Record) error {
+ m, host := rec.Move, rec.Host
+ sealKey := moves.OfflineSealKey(host, m.ID)
+ if sealKey == "" {
+ return fmt.Errorf("offline move %s: %s has no seal key", m.ID, host)
+ }
+ raw, _, err := s.Get(ctx, sealKey)
+ switch {
+ case err == nil:
+ // Sealed on an earlier pass, which crashed before the CAS: the
+ // position is fixed, so the stop is only kept, never taken again.
+ return w.offlineKeepStop(ctx, s, rec, raw)
+ case !errors.Is(err, os.ErrNotExist):
+ return fmt.Errorf("read %s: %w", sealKey, err)
+ }
+ items := map[string]moves.Check{}
+ seal, err := w.offlineTakeSeal(ctx, s, rec, items)
+ var ref *sealRefusal
+ if errors.As(err, &ref) {
+ items[ref.item] = moves.Check{OK: false, Reason: ref.reason}
+ r := moves.OfflineReport{Op: m.ID, Host: host, Box: w.cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseFailed,
+ At: w.cfg.now().UTC(), Items: items}
+ return errors.Join(fmt.Errorf("offline move %s: %s cannot be sealed: %w", m.ID, host, err), writeOfflineReport(ctx, s, r))
+ }
+ if err != nil {
+ return fmt.Errorf("offline move %s: sealing %s: %w", m.ID, host, err)
+ }
+ body, err := json.Marshal(seal)
+ if err != nil {
+ return err
+ }
+ stored, created, err := putOfflineCreateOnly(ctx, s, sealKey, body)
+ if err != nil {
+ return err
+ }
+ if created {
+ log.Printf("moves: %s: offline move %s: sealed at TXID %d (%d staged objects, files %s)", host, m.ID, seal.FinalTXID, len(seal.Staged), seal.FilesPlain)
+ }
+ if err := w.sealFault("seal"); err != nil {
+ return err
+ }
+ return w.offlineKeepStop(ctx, s, rec, stored)
+}
+
+// offlineTakeSeal runs steps 1 to 9 and returns the seal to write. items
+// collects each item's check as it is decided; a *sealRefusal is the item
+// that failed.
+func (w *movesWorker) offlineTakeSeal(ctx context.Context, s release.Store, rec provision.Record, items map[string]moves.Check) (moves.OfflineReport, error) {
+ cfg := w.cfg
+ m, host := rec.Move, rec.Host
+ var seal moves.OfflineReport
+
+ // 1. The stop, as the source reports say it now: this box held, fenced,
+ // quiesced, with no process and no layout journal; every box the seal
+ // named held; no sidecar anywhere.
+ if len(m.Boxes) == 0 {
+ return seal, errors.New("the record names no source boxes")
+ }
+ reports, err := readOfflineReports(ctx, s, host, m.ID, moves.OfflineRoleSource)
+ if err != nil {
+ return seal, fmt.Errorf("read the source reports: %w", err)
+ }
+ now := cfg.now().UTC()
+ own, ok := reports[cfg.Box]
+ if !ok || own.Phase != moves.OfflinePhaseStopped || !own.Held || !own.Fenced || !own.Quiesced || own.Procs != 0 || own.LayoutJournal || now.Sub(own.At) > offlineSealFresh {
+ return seal, fmt.Errorf("this box has not stopped %s this pass: report found %v, held %v, fenced %v, quiesced %v, processes %d, layout journal %v, at %s",
+ host, ok, own.Held, own.Fenced, own.Quiesced, own.Procs, own.LayoutJournal, own.At.Format(time.RFC3339))
+ }
+ boxes := append([]string(nil), m.Boxes...)
+ sort.Strings(boxes)
+ for _, b := range boxes {
+ r, ok := reports[b]
+ if !ok || !r.Held {
+ return seal, fmt.Errorf("waiting for %s to report %s held", b, host)
+ }
+ }
+ for _, b := range boxes {
+ if reports[b].Sidecar {
+ return seal, sealRefuse("sidecar", "%s keeps a sidecar file for %s", b, host)
+ }
+ }
+ items["sidecar"] = moves.Check{OK: true, Reason: "no source box keeps a sidecar file"}
+
+ rt, err := w.offlineRow(host)
+ if err != nil {
+ return seal, err
+ }
+ if rt == nil {
+ return seal, fmt.Errorf("this box has no row for %s", host)
+ }
+ tgt, err := w.offlineTarget(ctx, s, m)
+ if err != nil {
+ return seal, err
+ }
+ mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion)
+ if err != nil {
+ return seal, err
+ }
+ bucket, loc, region, err := cfg.tombstoneAt(host)
+ if err != nil {
+ return seal, err
+ }
+ srcPrefix := loc + "/"
+ dstPrefix := activator.MovesPrefix(host)
+ dbPath := rt.DBPath
+ if dbPath == "" {
+ dbPath = cfg.flatDBPath(host, rt)
+ }
+
+ // 2. The checkpoint, while no tombstone is down: the local database is
+ // synced under the lease, which the checkpoint releases again. That
+ // carries a tail the stop failed to checkpoint, whether litestream had
+ // already turned it into a local LTX file (the local position is past
+ // the replica's) or it is still in the WAL (the positions are equal),
+ // as the legacy cut re-checkpoints unconditionally. A local copy behind
+ // the replica is a stale warm cache: the replica is the instance, and
+ // syncing the stale copy into it is the one thing never done. A
+ // tombstone already this move's means an earlier pass checkpointed and
+ // took it, and nothing has woken since; the checkpoint could not take
+ // the lease anyway.
+ local, err := w.localState(host, rt)
+ if err != nil {
+ return seal, err
+ }
+ objs, err := mirror.List(ctx, bucket, srcPrefix)
+ if err != nil {
+ return seal, fmt.Errorf("list the replica: %w", err)
+ }
+ before, err := w.offlineLock(ctx, host)
+ if err != nil {
+ return seal, err
+ }
+ replica := replicaPosition(objs, srcPrefix)
+ switch {
+ case before.mine(m.ID):
+ case local.LocalTXID < replica:
+ log.Printf("moves: %s: offline move %s: the local copy (position %d) is behind the replica (%d); the replica is sealed as it is", host, m.ID, local.LocalTXID, replica)
+ case lexists(dbPath):
+ if cfg.Checkpoint == nil {
+ return seal, errors.New("no checkpoint on this box")
+ }
+ if err := cfg.Checkpoint(ctx, dbPath); err != nil {
+ return seal, fmt.Errorf("final checkpoint of %s: %w", dbPath, err)
+ }
+ log.Printf("moves: %s: offline move %s: checkpointed the local database (position %d, replica %d)", host, m.ID, local.LocalTXID, replica)
+ case local.LocalTXID > replica:
+ return seal, sealRefuse("db-tail", "this box's local position %d is past the replica's %d and there is no database at %s to checkpoint", local.LocalTXID, replica, dbPath)
+ }
+ if err := w.sealFault("checkpoint"); err != nil {
+ return seal, err
+ }
+
+ // 3. The tombstone: from here no wake on any box can take the lease.
+ owner := moves.TombstoneOwner(m.ID)
+ if err := lease.Tombstone(ctx, bucket, loc, region, owner, moves.TombstoneTTL); err != nil {
+ return seal, fmt.Errorf("tombstone %s/%s/lock.json: %w", bucket, loc, err)
+ }
+ if err := w.sealFault("tombstone"); err != nil {
+ return seal, err
+ }
+
+ // 4. The final position, read only now, and no source box past it.
+ objs, err = mirror.List(ctx, bucket, srcPrefix)
+ if err != nil {
+ return seal, fmt.Errorf("list the replica: %w", err)
+ }
+ final := replicaPosition(objs, srcPrefix)
+ if final == 0 {
+ return seal, sealRefuse("db", "the replica at %s/%s holds no database", bucket, srcPrefix)
+ }
+ mine, err := w.localState(host, rt) // after the checkpoint, not as reported before it
+ if err != nil {
+ return seal, err
+ }
+ for _, b := range boxes {
+ pos := reports[b].LocalTXID
+ if b == cfg.Box {
+ pos = mine.LocalTXID
+ }
+ if pos > final {
+ return seal, sealRefuse("db-tail", "%s's local position %d is past the replica's final position %d", b, pos, final)
+ }
+ }
+ items["db-tail"] = moves.Check{OK: true, Reason: fmt.Sprintf("every source box is at or below TXID %d", final)}
+ items["lease"] = moves.Check{OK: true, Reason: fmt.Sprintf("%s holds %s/lock.json", owner, loc)}
+ if err := w.sealFault("final"); err != nil {
+ return seal, err
+ }
+
+ // 5. The request record into the build state, then the byte-final
+ // round into the target's staging lane and its listing.
+ if cfg.CarryRequest != nil {
+ if err := cfg.CarryRequest(ctx, host, bucket, loc); err != nil {
+ return seal, fmt.Errorf("carry the request record into the build state: %w", err)
+ }
+ items["build-state"] = moves.Check{OK: true, Reason: "the request record is in the build state the replica prefix carries"}
+ } else {
+ items["build-state"] = moves.Check{OK: true, Reason: "this box keeps no request records; the build state travels as the replica prefix holds it"}
+ }
+ receipts := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)}
+ round, err := lease.MirrorPrefix(ctx, mirror, bucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts)
+ if err != nil {
+ return seal, err
+ }
+ if round.Deferred > 0 {
+ return seal, fmt.Errorf("final mirror round: %d object(s) changed at the source during the copy", round.Deferred)
+ }
+ source, err := mirror.List(ctx, bucket, srcPrefix)
+ if err != nil {
+ return seal, fmt.Errorf("list the replica: %w", err)
+ }
+ listed, err := mirror.List(ctx, tgt.DataBucket, dstPrefix)
+ if err != nil {
+ return seal, fmt.Errorf("list the staged copy: %w", err)
+ }
+ staged := map[string]moves.StagedObject{}
+ for k, o := range listed {
+ staged[strings.TrimPrefix(k, dstPrefix)] = moves.StagedObject{Size: o.Size, ETag: o.ETag}
+ }
+ if err := sameListing(source, srcPrefix, staged); err != nil {
+ return seal, fmt.Errorf("the staged copy is not the replica: %w", err)
+ }
+ if replicaPosition(listed, dstPrefix) != final {
+ return seal, fmt.Errorf("the staged copy's newest TXID is %d, the replica's %d", replicaPosition(listed, dstPrefix), final)
+ }
+ items["db"] = moves.Check{OK: true, Reason: fmt.Sprintf("%d objects staged at %s/%s, newest TXID %d", len(staged), tgt.DataBucket, dstPrefix, final)}
+ if err := w.sealFault("mirror"); err != nil {
+ return seal, err
+ }
+
+ // 6. The files and <host>.env, sealed, beside the staging lane.
+ prefix := moves.OfflineArchivePrefix(host, m.ID)
+ files, envPath, configPath, stray, err := w.offlineFilesOf(host, rt)
+ if err != nil {
+ return seal, fmt.Errorf("read what %s keeps: %w", host, err)
+ }
+ if len(stray) > 0 {
+ sort.Strings(stray)
+ return seal, sealRefuse("files", "%s keeps %s beside its files directory %s, which the move carries alone; move it into %s or remove it", host, strings.Join(stray, ", "), files, files)
+ }
+ a, err := w.offlineArchive(ctx, rec, mirror, tgt.DataBucket, prefix, files, envPath)
+ if err != nil {
+ return seal, err
+ }
+ items["files"] = moves.Check{OK: true, Reason: fmt.Sprintf("%d files, %d bytes, sealed at %s%s", a.entries, a.bytes, prefix, offlineArchiveSealed)}
+ if err := w.sealFault("archive"); err != nil {
+ return seal, err
+ }
+
+ // 7. The timers.
+ if cfg.Registry == nil {
+ return seal, errors.New("no registry on this box")
+ }
+ rows, err := cfg.Registry.TimersFor(host)
+ if err != nil {
+ return seal, fmt.Errorf("read the timers: %w", err)
+ }
+ active := moves.ActiveTimers(rows)
+ timersBody, err := moves.TimersSnapshot{V: 1, Host: host, Rows: active, TakenAt: now}.Encode()
+ if err != nil {
+ return seal, err
+ }
+ if err := mirror.Put(ctx, tgt.DataBucket, prefix+offlineTimersObject, timersBody); err != nil {
+ return seal, fmt.Errorf("put the timers: %w", err)
+ }
+ timersDigest := moves.TimersDigest(rows)
+ items["timers"] = moves.Check{OK: true, Reason: fmt.Sprintf("%d active timers at %s%s", len(active), prefix, offlineTimersObject)}
+ if err := w.sealFault("timers"); err != nil {
+ return seal, err
+ }
+
+ // 8. The env file, its effective instance key, and the env pin.
+ envSHA, keySHA, err := offlineEnvDigests(envPath, configPath, a.envSHA)
+ if err != nil {
+ return seal, err
+ }
+ envWhat := "absent on the source, so none is carried"
+ if envSHA != "" {
+ envWhat = "carried in the archive, sha256 " + envSHA
+ }
+ items["env"] = moves.Check{OK: true, Reason: envWhat}
+ if cfg.LaunchLayers == nil {
+ return seal, sealRefuse("env-pin", "this box has no exec backend to compose the launch environment")
+ }
+ if cfg.SecretsKey == "" {
+ return seal, sealRefuse("env-pin", "this box holds no CARLOS_SECRETS_KEY to key the env report")
+ }
+ env, instance, _, err := cfg.LaunchLayers(*rt)
+ if err != nil {
+ return seal, fmt.Errorf("compose the launch environment: %w", err)
+ }
+ mac, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
+ if err != nil {
+ return seal, sealRefuse("env-pin", "the env report key: %v", err)
+ }
+ envReport := moves.ReportLaunch(mac, env, instance)
+ envReportBody, err := json.Marshal(envReport)
+ if err != nil {
+ return seal, err
+ }
+ if err := mirror.Put(ctx, tgt.DataBucket, prefix+offlineEnvReportObject, envReportBody); err != nil {
+ return seal, fmt.Errorf("put the env report: %w", err)
+ }
+ items["env-pin"] = moves.Check{OK: true, Reason: "launch environment digest " + envReport.Digest}
+ if err := w.sealFault("env"); err != nil {
+ return seal, err
+ }
+
+ // 9. The build, the replication rung and the next due time.
+ if cfg.LiveVersion == nil {
+ return seal, errors.New("no version store on this box")
+ }
+ version, err := cfg.LiveVersion(host)
+ if err != nil {
+ return seal, fmt.Errorf("read the live build: %w", err)
+ }
+ if version == "" {
+ return seal, sealRefuse("release", "no build is live for %s on this box", host)
+ }
+ items["release"] = moves.Check{OK: true, Reason: "live build " + version}
+ items["replication"] = moves.Check{OK: true, Reason: fmt.Sprintf("replication %q", rt.Replication)}
+ items["next-due"] = moves.Check{OK: true, Reason: fmt.Sprintf("next due %d", rt.NextDue)}
+
+ // The tombstone must still be this move's when the seal is written:
+ // the position it records was read under it. One lost before the seal
+ // is taken again on the next pass, which seals afresh.
+ st, err := w.offlineLock(ctx, host)
+ if err != nil {
+ return seal, err
+ }
+ if !st.mine(m.ID) {
+ return seal, fmt.Errorf("the lease is %s, not this move's tombstone; the next pass takes it again", st)
+ }
+ seal = moves.OfflineReport{
+ V: moves.OfflineReportVersion, Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseSealed,
+ At: now, Items: items, TombstoneOwner: owner,
+ FinalTXID: final, Staged: staged,
+ FilesPlain: a.plain, FilesSealed: a.sealed, FilesBytes: a.bytes, FilesEntries: a.entries,
+ EnvSHA256: envSHA, KeySHA256: keySHA, EnvDigest: envReport.Digest,
+ Release: version, TimersDigest: timersDigest, Replication: rt.Replication, NextDue: rt.NextDue,
+ }
+ if err := moves.ValidateOfflineSeal(seal, host, m.ID); err != nil {
+ return seal, err
+ }
+ return seal, nil
+}
+
+// sameListing reports whether staged (keyed relative to the staging
+// prefix) holds exactly the source's objects under srcPrefix, lock.json
+// aside (the mirror never copies it), size for size. ETags may differ: a
+// multipart source's copy gets a new one (lease.Receipts).
+func sameListing(source map[string]lease.MirrorObject, srcPrefix string, staged map[string]moves.StagedObject) error {
+ want := map[string]int64{}
+ for k, o := range source {
+ rel := strings.TrimPrefix(k, srcPrefix)
+ if rel == "lock.json" {
+ continue
+ }
+ want[rel] = o.Size
+ }
+ if len(want) != len(staged) {
+ return fmt.Errorf("%d objects at the source, %d staged", len(want), len(staged))
+ }
+ for rel, size := range want {
+ got, ok := staged[rel]
+ if !ok || got.Size != size {
+ return fmt.Errorf("%s: staged %v, source size %d", rel, got, size)
+ }
+ }
return nil
}
+
+// offlineFilesOf is what the archive carries for the row's layout, and
+// what it cannot.
+//
+// The target lands the instance flat (the row is minted at flat paths), so
+// the archive's files root is the app's files directory as the flat layout
+// names it: <DataDir>/<host>, with <host>.env beside it. The layout pass
+// maps the flat data directory onto the nested instance directory name for
+// name (layout/extras.go: an app finds its files beside its database, and
+// nested the database is <instance dir>/<host>.db), so a nested row's
+// files directory is <instance dir>/<host>, read through the raw
+// directory as every agent read of it is, and its env file is
+// <instance dir>/<host>.env. Archiving the whole instance directory would
+// land the app's files one level down on the target (<DataDir>/<host>/
+// <host>/...), where the app would not find them.
+//
+// stray is every entry the app keeps beside its files directory, which
+// the flat files directory cannot carry: on a nested row every entry of
+// the instance directory that is neither the platform's
+// (layout.NestedPlatformNames) nor <host>; on a flat row every entry of
+// the data directory the layout's naming rules give this host
+// (layout.FlatLeftovers; <host>-blobs is the live example) that is
+// neither. The seal refuses on any, rather than leave it behind.
+// configPath is the rendered configuration beside the env file, which the
+// effective instance key may come from.
+func (w *movesWorker) offlineFilesOf(host string, rt *registry.Route) (files, envPath, configPath string, stray []string, err error) {
+ dd := w.cfg.DataDir
+ platform := layout.NestedPlatformNames(host)
+ if rt != nil && provision.NestedPaths(dd, *rt) {
+ raw := layout.RawInstanceDir(dd, host)
+ ents, err := os.ReadDir(raw)
+ if err != nil && !errors.Is(err, os.ErrNotExist) {
+ return "", "", "", nil, err
+ }
+ for _, e := range ents {
+ if !platform[e.Name()] && e.Name() != host {
+ stray = append(stray, filepath.Join(raw, e.Name()))
+ }
+ }
+ return filepath.Join(raw, host), filepath.Join(raw, host+".env"), filepath.Join(raw, host+".config.env"), stray, nil
+ }
+ var hosts []string
+ if w.cfg.Registry != nil {
+ rows, err := w.cfg.Registry.List()
+ if err != nil {
+ return "", "", "", nil, err
+ }
+ for _, r := range rows {
+ hosts = append(hosts, r.Host)
+ }
+ }
+ for _, p := range layout.FlatLeftovers(dd, host, hosts) {
+ if name := filepath.Base(p); !platform[name] && name != host {
+ stray = append(stray, p)
+ }
+ }
+ return filepath.Join(dd, host), filepath.Join(dd, host+".env"), filepath.Join(dd, host+".config.env"), stray, nil
+}
+
+// sealedArchive is what the seal records of the archive.
+type sealedArchive struct {
+ plain, sealed string
+ bytes int64
+ entries int
+ // envSHA is the env entry's digest as the archive recorded it
+ // ("sha256:<hex>"), "" when the archive carries no env.
+ envSHA string
+}
+
+// offlineArchive builds the archive of files and envPath into a scratch
+// directory on the data disk, seals it under the deployment's config key
+// bound to the record's account and app, and puts it and its manifest
+// under prefix in the target bucket. A file the archive refuses is a
+// refusal of the files item, never a partial archive; a file that changed
+// while it was read is retried. The files directory holds nothing of the
+// platform's but the files marker, which every build leaves out, so no
+// filter is needed (archive.Build is BuildFiltered with none).
+func (w *movesWorker) offlineArchive(ctx context.Context, rec provision.Record, mirror lease.MirrorStore, bucket, prefix, files, envPath string) (sealedArchive, error) {
+ cfg := w.cfg
+ var out sealedArchive
+ if cfg.ConfigPub == nil {
+ return out, sealRefuse("files", "this box holds no CARLOS_SECRETS_KEY to seal the archive under")
+ }
+ dir := moves.OfflineParkDir(cfg.DataDir, rec.Host, rec.Move.ID, moves.OfflineRoleSealer)
+ if dir == "" {
+ return out, fmt.Errorf("no scratch directory for %s under %q", rec.Host, cfg.DataDir)
+ }
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ return out, err
+ }
+ tmp, err := os.MkdirTemp(dir, "archive-")
+ if err != nil {
+ return out, err
+ }
+ defer os.RemoveAll(tmp)
+ built, err := os.Create(filepath.Join(tmp, "archive.tar.zst"))
+ if err != nil {
+ return out, err
+ }
+ defer built.Close()
+ plain, entries, size, err := archive.Build(files, envPath, built)
+ switch {
+ case errors.Is(err, archive.ErrChanged):
+ return out, fmt.Errorf("archive the files: %w; built again next pass", err)
+ case err != nil:
+ return out, sealRefuse("files", "%v", err)
+ }
+ if _, err := built.Seek(0, io.SeekStart); err != nil {
+ return out, err
+ }
+ sealedFile, err := os.Create(filepath.Join(tmp, "archive.sealed"))
+ if err != nil {
+ return out, err
+ }
+ defer sealedFile.Close()
+ dataKey, sealedSHA, err := archive.Seal(built, cfg.ConfigPub, appconfig.Binding{Account: rec.Account, App: rec.App}, sealedFile)
+ if err != nil {
+ return out, fmt.Errorf("seal the archive: %w", err)
+ }
+ manifest, err := json.Marshal(archive.Manifest{V: archive.ManifestVersion, Op: rec.Move.ID, Plain: plain, Sealed: sealedSHA, Bytes: size,
+ Entries: entries, SealedDataKey: dataKey, Chunk: archive.ChunkSize})
+ if err != nil {
+ return out, err
+ }
+ body, err := os.ReadFile(sealedFile.Name())
+ if err != nil {
+ return out, err
+ }
+ if err := mirror.Put(ctx, bucket, prefix+offlineArchiveSealed, body); err != nil {
+ return out, fmt.Errorf("put the sealed archive: %w", err)
+ }
+ if err := mirror.Put(ctx, bucket, prefix+offlineArchiveManifest, manifest); err != nil {
+ return out, fmt.Errorf("put the archive manifest: %w", err)
+ }
+ out = sealedArchive{plain: plain, sealed: sealedSHA, bytes: size, entries: len(entries)}
+ for _, e := range entries {
+ if e.Path == archive.EnvEntry {
+ out.envSHA = e.SHA256
+ }
+ }
+ return out, nil
+}
+
+// readOwnFile reads a regular file without following a link, as the agent
+// reads every file a nested tenant can write: exists false when there is
+// none.
+func readOwnFile(path string) (body []byte, exists bool, err error) {
+ info, err := os.Lstat(path)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, false, nil
+ }
+ if err != nil {
+ return nil, false, err
+ }
+ if !info.Mode().IsRegular() {
+ return nil, true, fmt.Errorf("%s is not a regular file", path)
+ }
+ f, err := os.Open(path)
+ if err != nil {
+ return nil, true, err
+ }
+ defer f.Close()
+ got, err := f.Stat()
+ if err != nil {
+ return nil, true, err
+ }
+ if !os.SameFile(info, got) {
+ return nil, true, fmt.Errorf("%s changed while it was read", path)
+ }
+ body, err = io.ReadAll(io.LimitReader(f, instancekey.EnvReadLimit+1))
+ if err != nil {
+ return nil, true, err
+ }
+ if len(body) > instancekey.EnvReadLimit {
+ return nil, true, fmt.Errorf("%s is larger than an env file", path)
+ }
+ return body, true, nil
+}
+
+// offlineEnvDigests is the env item's two values: the lowercase hex sha256
+// of <host>.env's bytes ("" when the source has none), and the lowercase
+// hex sha256 of the instance key a launch hands the app
+// (instancekey.Effective over the rendered configuration and the own file,
+// instancekey.Digest; "" when neither supplies one). archived is the env
+// entry's digest as the archive recorded it ("sha256:<hex>", or "" for no
+// env entry); the file read here must be the one archived.
+func offlineEnvDigests(envPath, configPath, archived string) (envSHA, keySHA string, err error) {
+ own, exists, err := readOwnFile(envPath)
+ if err != nil {
+ return "", "", fmt.Errorf("read the env file: %w", err)
+ }
+ if exists {
+ sum := sha256.Sum256(own)
+ envSHA = hex.EncodeToString(sum[:])
+ }
+ if want := strings.TrimPrefix(archived, "sha256:"); want != envSHA {
+ return "", "", fmt.Errorf("the env file changed while it was archived (archived %q, now %q); sealed again next pass", want, envSHA)
+ }
+ config, _, err := readOwnFile(configPath)
+ if err != nil {
+ return "", "", fmt.Errorf("read the rendered configuration: %w", err)
+ }
+ if key, _ := instancekey.Effective(instancekey.ParseEnv(config), instancekey.ParseEnv(own)); key != "" {
+ keySHA = instancekey.Digest(key)
+ }
+ return envSHA, keySHA, nil
+}
+
+// offlineSealRecord reads seal.json, nil when it is not there.
+func (w *movesWorker) offlineSealRecord(ctx context.Context, s release.Store, rec provision.Record) ([]byte, error) {
+ key := moves.OfflineSealKey(rec.Host, rec.Move.ID)
+ raw, _, err := s.Get(ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, fmt.Errorf("read %s: %w", key, err)
+ }
+ return raw, nil
+}
+
+// decodeSeal parses and checks the seal record; nil raw is no seal.
+func decodeSeal(raw []byte, host, id string) (*moves.OfflineReport, error) {
+ if raw == nil {
+ return nil, nil
+ }
+ var seal moves.OfflineReport
+ if err := json.Unmarshal(raw, &seal); err != nil {
+ return nil, fmt.Errorf("the seal record does not parse: %w", err)
+ }
+ if err := moves.ValidateOfflineSeal(seal, host, id); err != nil {
+ return nil, err
+ }
+ return &seal, nil
+}
+
+// offlineKeepStop is every pass after the seal, until the tombstone is
+// lifted: read the lease first; a tombstone that is not this move's any
+// more is a lost stop, never taken again; this move's is re-asserted. In
+// offline-sealing (a seal written by a pass that crashed before the CAS)
+// it also makes the CAS. Then it rewrites the living sealer report: the
+// seal's values and items, the tombstone's owner and expiry, and At.
+func (w *movesWorker) offlineKeepStop(ctx context.Context, s release.Store, rec provision.Record, raw []byte) error {
+ cfg := w.cfg
+ m, host := rec.Move, rec.Host
+ seal, err := decodeSeal(raw, host, m.ID)
+ if err != nil {
+ return fmt.Errorf("offline move %s: %s: %w; it is never rewritten, so abort the move", m.ID, host, err)
+ }
+ if seal == nil && m.State == provision.OfflineSealing {
+ return fmt.Errorf("offline move %s: %s has no seal record", m.ID, host)
+ }
+ st, err := w.offlineLock(ctx, host)
+ if err != nil {
+ return err
+ }
+ if !st.mine(m.ID) {
+ return w.offlineLostStop(ctx, s, rec, seal, st)
+ }
+ bucket, loc, region, err := cfg.tombstoneAt(host)
+ if err != nil {
+ return err
+ }
+ if err := lease.Reassert(ctx, bucket, loc, region, moves.TombstoneOwner(m.ID), moves.TombstoneTTL); err != nil {
+ var held *lease.HeldError
+ if errors.As(err, &held) {
+ return w.offlineLostStop(ctx, s, rec, seal, tombState{owner: held.Owner, expires: held.ExpiresAt})
+ }
+ return fmt.Errorf("offline move %s: re-assert the tombstone of %s: %w", m.ID, host, err)
+ }
+ if st, err = w.offlineLock(ctx, host); err != nil {
+ return err
+ }
+ if !st.mine(m.ID) {
+ return w.offlineLostStop(ctx, s, rec, seal, st)
+ }
+ if m.State == provision.OfflineSealing {
+ src := provision.MoveSource{
+ FinalTXID: seal.FinalTXID, Replication: seal.Replication, NextDue: seal.NextDue,
+ Files: &provision.MoveFilesDigest{Plain: seal.FilesPlain, Sealed: seal.FilesSealed, Bytes: seal.FilesBytes, Entries: seal.FilesEntries},
+ TimersSHA256: seal.TimersDigest, Release: seal.Release, EnvDigest: seal.EnvDigest,
+ Evidence: moves.OfflineSealEvidence(host, m.ID, raw),
+ }
+ _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, host, provision.OfflineSealedBy(m.ID, cfg.Box, src))
+ switch {
+ case errors.Is(err, provision.ErrMoveConflict):
+ log.Printf("moves: %s: offline move %s: the record moved on before the seal's CAS: %v", host, m.ID, err)
+ case err != nil:
+ return fmt.Errorf("offline move %s: record the seal of %s: %w", m.ID, host, err)
+ default:
+ nudgeSerial(ctx, s)
+ log.Printf("moves: %s: offline move %s: sealed", host, m.ID)
+ }
+ if err := w.sealFault("cas"); err != nil {
+ return err
+ }
+ }
+ living := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer}
+ if seal != nil {
+ living = *seal
+ }
+ living.Phase = moves.OfflinePhaseSealed
+ living.At = cfg.now().UTC()
+ living.TombstoneOwner = st.owner
+ living.TombstoneUntil = st.expires.UTC()
+ return writeOfflineReport(ctx, s, living)
+}
+
+// offlineLostStop reports a stop that no longer holds: the tombstone is
+// absent, expired, lifted or someone else's after the seal. It is never
+// taken again (a wake may have written the replica in between, so the
+// sealed position is no longer the source's), the living report says
+// failed with the lease item not OK, and an operator is paged once. The
+// operator aborts the move.
+func (w *movesWorker) offlineLostStop(ctx context.Context, s release.Store, rec provision.Record, seal *moves.OfflineReport, st tombState) error {
+ cfg := w.cfg
+ m, host := rec.Move, rec.Host
+ reason := fmt.Sprintf("lost stop: the tombstone at the source's lease is %s, not held by %s", st, moves.TombstoneOwner(m.ID))
+ r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer}
+ if seal != nil {
+ r = *seal
+ }
+ r.Phase = moves.OfflinePhaseFailed
+ r.At = cfg.now().UTC()
+ r.TombstoneOwner, r.TombstoneUntil = "", time.Time{}
+ items := map[string]moves.Check{}
+ for k, v := range r.Items {
+ items[k] = v
+ }
+ items["lease"] = moves.Check{OK: false, Reason: reason}
+ r.Items = items
+ werr := writeOfflineReport(ctx, s, r)
+ subject := fmt.Sprintf("offline move of %s lost its stop", host)
+ body := fmt.Sprintf("The offline move %s of %s is %s, and the tombstone that stops every wake in the source region is no longer held: %s. The move cannot go on. Read carlos move offline status --host %s and abort the move.", m.ID, host, m.State, st, host)
+ w.firePage(ctx, host, m.ID, offlinePageLostStop, alerting.LevelIncident, subject, body)
+ log.Printf("moves: %s: offline move %s: LOST STOP: %s", host, m.ID, reason)
+ return errors.Join(fmt.Errorf("offline move %s: %s: %s", m.ID, host, reason), werr)
+}
+
+// offlineFinishLift is the sealer's step in offline-finishing. Until every
+// box the seal named reports gone (it parked at open and reconcile has
+// removed its row since), the stop is kept as before. Then the tombstone
+// is lifted (only this move's, with the ETag read), the copy receipts are
+// cleared and the report says lifted. A lease that is already free (lifted
+// by an earlier pass) is left as it is; one someone else holds live is a
+// lost stop.
+func (w *movesWorker) offlineFinishLift(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ m, host := rec.Move, rec.Host
+ reports, err := readOfflineReports(ctx, s, host, m.ID, moves.OfflineRoleSource)
+ if err != nil {
+ return fmt.Errorf("read the source reports: %w", err)
+ }
+ gone := len(m.Boxes) > 0
+ for _, b := range m.Boxes {
+ if r, ok := reports[b]; !ok || r.Phase != moves.OfflinePhaseGone {
+ gone = false
+ }
+ }
+ raw, err := w.offlineSealRecord(ctx, s, rec)
+ if err != nil {
+ return err
+ }
+ if !gone {
+ return w.offlineKeepStop(ctx, s, rec, raw)
+ }
+ st, err := w.offlineLock(ctx, host)
+ if err != nil {
+ return err
+ }
+ switch {
+ case st.mine(m.ID):
+ bucket, loc, region, err := cfg.tombstoneAt(host)
+ if err != nil {
+ return err
+ }
+ if _, err := lease.LiftTombstone(ctx, bucket, loc, region, func(owner string) bool { return owner == moves.TombstoneOwner(m.ID) }); err != nil {
+ return fmt.Errorf("offline move %s: lift the tombstone of %s: %w", m.ID, host, err)
+ }
+ log.Printf("moves: %s: offline move %s: tombstone lifted", host, m.ID)
+ case st.free():
+ default:
+ seal, _ := decodeSeal(raw, host, m.ID)
+ return w.offlineLostStop(ctx, s, rec, seal, st)
+ }
+ if err := w.sealFault("lift"); err != nil {
+ return err
+ }
+ own, err := w.offlineMirror(ctx, cfg.OwnBucketRegion)
+ if err != nil {
+ return err
+ }
+ clearOfflineReceipts(ctx, own, cfg.OwnBucket, host)
+ r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseLifted, At: cfg.now().UTC()}
+ return writeOfflineReport(ctx, s, r)
+}
+
+// clearOfflineReceipts is clearMoveReceipts as an ensure: receipts that are
+// already gone are not deleted again, so a pass that repeats finish or
+// revert writes nothing (on a versioned bucket every DELETE of an absent
+// key leaves one more delete marker).
+func clearOfflineReceipts(ctx context.Context, m lease.MirrorStore, bucket, host string) {
+ rc := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)}
+ if _, err := m.Get(ctx, rc.Bucket, rc.Key); errors.Is(err, lease.ErrMirrorNotFound) {
+ return
+ }
+ clearMoveReceipts(ctx, m, rc, host)
+}
+
+// offlineSealRevert is the sealer's step in offline-reverting: lift the
+// tombstone if it is this move's (a lease anyone else holds, or none, is
+// left alone), delete the two prefixes this move wrote in the target's
+// bucket (the staged replica and the archive), clear the copy receipts,
+// lift this move's fence (the source step lifts it too) and report
+// reverted. Every part is an ensure.
+func (w *movesWorker) offlineSealRevert(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ m, host := rec.Move, rec.Host
+ bucket, loc, region, err := cfg.tombstoneAt(host)
+ if err != nil {
+ return err
+ }
+ lifted, err := lease.LiftTombstone(ctx, bucket, loc, region, func(owner string) bool { return owner == moves.TombstoneOwner(m.ID) })
+ var held *lease.HeldError
+ switch {
+ case err == nil:
+ if lifted {
+ log.Printf("moves: %s: offline move %s: tombstone lifted on abort", host, m.ID)
+ }
+ case errors.As(err, &held):
+ w.notice(host+"#offline-revert-lease", "moves: %s: offline move %s: the lease is %s's, not this move's; left as it is", host, m.ID, held.Owner)
+ default:
+ return fmt.Errorf("offline move %s: lift the tombstone of %s: %w", m.ID, host, err)
+ }
+ if err := w.sealFault("revert-lift"); err != nil {
+ return err
+ }
+ tgt, err := w.offlineTarget(ctx, s, m)
+ if err != nil {
+ return err
+ }
+ mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion)
+ if err != nil {
+ return err
+ }
+ for _, prefix := range []string{activator.MovesPrefix(host), moves.OfflineArchivePrefix(host, m.ID)} {
+ if prefix == "" || prefix == "moves/" {
+ return fmt.Errorf("offline move %s: no staging prefix for %s", m.ID, host)
+ }
+ if err := lease.DeletePrefix(ctx, mirror, tgt.DataBucket, prefix); err != nil {
+ return fmt.Errorf("offline move %s: delete %s/%s: %w", m.ID, tgt.DataBucket, prefix, err)
+ }
+ }
+ clearOfflineReceipts(ctx, mirror, bucket, host)
+ w.offlineLiftFence(host, m)
+ r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC()}
+ return writeOfflineReport(ctx, s, r)
+}
diff --git a/internal/host/movesoffline_seal_test.go b/internal/host/movesoffline_seal_test.go
new file mode 100644
index 000000000..1d6004da0
--- /dev/null
+++ b/internal/host/movesoffline_seal_test.go
@@ -0,0 +1,1181 @@
+//go:build linux
+
+package host
+
+import (
+ "bytes"
+ "context"
+ "crypto/ecdh"
+ "crypto/sha256"
+ "database/sql"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strings"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ "github.com/superfly/ltx"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/instancekey"
+ "github.com/carlosframework/platform/internal/layout"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The sealer (Task 4 of docs/superpowers/plans/2026-10-20-offline-operator-
+// move.md), tested through offlineRound, the moves pass's dispatch, over
+// the S3 wire protocol (leasetest.NewS3Server, with buckets data-ie and
+// data-uk behind one endpoint), real litestream replicas written by
+// lease.ReplicateWithLease, the real lease package, a real registry and
+// activator per box, and the production edges-file writer for the target's
+// coordinates. The deployment store is T3's DirStore world.
+
+const (
+ sealIE = "data-ie"
+ sealUK = "data-uk"
+ sealIERegion = "eu-west-1"
+ sealUKRegion = "eu-west-2"
+ sealPrefix = "instances"
+ sealVersion = "v42"
+)
+
+type sealWorld struct {
+ *offWorld
+ fake *leasetest.VersionedBucket
+ keyB64 string
+ priv *ecdh.PrivateKey
+ ie, uk *offBox
+ pages []string
+}
+
+func newSealWorld(t *testing.T) *sealWorld {
+ t.Helper()
+ fake := leasetest.NewVersionedBucket()
+ fake.AddBucket(sealIE)
+ fake.AddBucket(sealUK)
+ h, _ := leasetest.NewS3Server(fake)
+ srv := httptest.NewServer(h) // 127.0.0.1 only
+ t.Cleanup(srv.Close)
+ t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
+ t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
+ t.Setenv("AWS_SESSION_TOKEN", "")
+ t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
+ t.Setenv("AWS_CONFIG_FILE", filepath.Join(t.TempDir(), "absent"))
+ t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(t.TempDir(), "absent"))
+ t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
+ keyB64, _, err := appconfig.GenerateKey()
+ if err != nil {
+ t.Fatal(err)
+ }
+ priv, err := appconfig.ParsePrivateKey(keyB64, "test key")
+ if err != nil {
+ t.Fatal(err)
+ }
+ sw := &sealWorld{offWorld: newOffWorld(t), fake: fake, keyB64: keyB64, priv: priv}
+ sw.ie = sw.sourceBox("ie-a")
+ sw.uk = sw.box("uk-box", "uk", map[string]string{"CARLOS_HIBERNATE_BUCKET": sealUK, "CARLOS_SECRETS_KEY": keyB64})
+ var last []byte
+ if err := edgesFileWrite(context.Background(), sw.s, sw.uk.reg, edgesFile{Box: "uk-box", Placement: "uk", DataBucket: sealUK, DataBucketRegion: sealUKRegion}, "", &last, false); err != nil {
+ t.Fatal(err)
+ }
+ return sw
+}
+
+// sourceBox is a home box armed as armMovesPass arms one: armOffline, then
+// the hibernate coordinates, the S3 mirror, the production checkpoint and
+// the box's exec backend for the launch environment.
+func (sw *sealWorld) sourceBox(name string) *offBox {
+ b := sw.box(name, "", map[string]string{"CARLOS_HIBERNATE_BUCKET": sealIE, "CARLOS_SECRETS_KEY": sw.keyB64})
+ sw.arm(b)
+ return b
+}
+
+func (sw *sealWorld) arm(b *offBox) {
+ c := &b.w.cfg
+ c.OwnBucket, c.OwnBucketRegion, c.OwnPrefix = sealIE, sealIERegion, sealPrefix
+ c.Mirror = func(ctx context.Context, dst string) (lease.MirrorStore, error) {
+ return lease.NewS3Mirror(ctx, sealIERegion, dst)
+ }
+ // armMovesPass's Checkpoint, as it is wired there.
+ c.Checkpoint = func(ctx context.Context, dbPath string) error {
+ host := strings.TrimSuffix(filepath.Base(dbPath), ".db")
+ rep, err := lease.ReplicateWithLease(ctx, dbPath, sealIE, activator.HibernatePrefix(sealPrefix, host), sealIERegion)
+ if err != nil {
+ return err
+ }
+ defer rep.Close(ctx)
+ return rep.SyncAndWait(ctx)
+ }
+ c.CarryRequest = func(ctx context.Context, host, bucket, prefix string) error {
+ sw.fake.Put(bucket, prefix+"/build-state.json", []byte(`{"request":"carried"}`))
+ return nil
+ }
+ c.LaunchLayers = (&activator.ExecBackend{DataDir: b.dd}).LaunchLayers
+ c.Page = func(_ context.Context, level, subject, body string) error {
+ sw.pages = append(sw.pages, level+": "+subject)
+ return nil
+ }
+}
+
+// pass is a tick in the agent's order (reconcile, the edges file, the moves
+// pass) whose moves pass is movesPass's offline part: the keep view's
+// refresh from the record list, then offlineRound, returning the round's
+// error rather than logging it.
+func (sw *sealWorld) pass(b *offBox) error {
+ sw.t.Helper()
+ b.reconcile()
+ b.edges()
+ ctx := context.Background()
+ refreshOfflineView(ctx, sw.s, b.w.cfg.Offline)
+ return b.w.offlineRound(ctx, sw.s, sw.record())
+}
+
+func (sw *sealWorld) mustPass(b *offBox) {
+ sw.t.Helper()
+ if err := sw.pass(b); err != nil {
+ sw.t.Fatal(err)
+ }
+}
+
+func (sw *sealWorld) dbPath(b *offBox) string {
+ sw.t.Helper()
+ rt := b.row(offHost)
+ if rt == nil {
+ sw.t.Fatal("no row")
+ }
+ return rt.DBPath
+}
+
+// instance gives b a live instance of offHost, as a box that has served it
+// has: the row (minted ready by reconcile), its replication rung, next due
+// and timers, a database written through lease.ReplicateWithLease into
+// data-ie, the files directory, <host>.env with an instance key, the
+// rendered configuration and a live build.
+func (sw *sealWorld) instance(b *offBox) {
+ t := sw.t
+ t.Helper()
+ b.tick()
+ rt := b.row(offHost)
+ if rt == nil || rt.HoldOp != "" {
+ t.Fatalf("fixture: row %+v", rt)
+ }
+ if err := b.reg.SetReplication(offHost, registry.ReplicationBatched); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.reg.SetNextDue(offHost, 1_800_000_000); err != nil {
+ t.Fatal(err)
+ }
+ for _, tm := range []registry.Timer{
+ {Host: offHost, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", Path: "/jobs/nightly", NextAt: 1_800_000_100},
+ {Host: offHost, Name: "reminder", Kind: "once", Path: "/jobs/remind", NextAt: 1_800_000_200},
+ {Host: offHost, Name: "fired", Kind: "once", Path: "/jobs/fired", LastAt: 1_700_000_000, LastStatus: "200"},
+ } {
+ if err := b.reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ }
+ writeFile(t, filepath.Join(b.dd, offHost, "instance.key"), "app key file")
+ writeFile(t, filepath.Join(b.dd, offHost, "uploads", "a.png"), "png bytes")
+ writeFile(t, filepath.Join(b.dd, offHost+".env"), "CARLOS_INSTANCE_KEY=the-instance-key\nCARLOS_ADMIN_TOKEN=token\n")
+ writeFile(t, filepath.Join(b.dd, offHost+".config.env"), "APP_COLOUR=green\n")
+ sw.writeRows(b, rt.DBPath, 0, 20, true)
+ versions := filepath.Join(b.opt, "versions", sealVersion, sw.acct, "app")
+ if err := os.MkdirAll(versions, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.MkdirAll(filepath.Dir(release.LiveLink(b.opt, offHost)), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(versions, release.LiveLink(b.opt, offHost)); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// writeRows inserts rows [from, to) into the database at dbPath, then, with
+// replicate, syncs it to data-ie under the lease as a wake's stop does.
+func (sw *sealWorld) writeRows(b *offBox, dbPath string, from, to int, replicate bool) {
+ t := sw.t
+ t.Helper()
+ db, err := sql.Open("sqlite", dbPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, q := range []string{`PRAGMA journal_mode=WAL`, `CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`} {
+ if _, err := db.Exec(q); err != nil {
+ t.Fatal(err)
+ }
+ }
+ for i := from; i < to; i++ {
+ if _, err := db.Exec(`INSERT INTO t (id, body) VALUES (?, ?)`, i, fmt.Sprintf("row-%d", i)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := db.Close(); err != nil {
+ t.Fatal(err)
+ }
+ if !replicate {
+ return
+ }
+ ctx := context.Background()
+ rep, err := lease.ReplicateWithLease(ctx, dbPath, sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := rep.SyncAndWait(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if err := rep.Close(ctx); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func countRows(t *testing.T, dbPath string) int {
+ t.Helper()
+ db, err := sql.Open("sqlite", dbPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ var n int
+ if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
+ t.Fatal(err)
+ }
+ return n
+}
+
+// start moves offHost to the record's offline-stopping, ticks every source
+// box until it holds and reports, and names the sealer and the boxes as the
+// seal door does.
+func (sw *sealWorld) startAndSeal(sealer string, boxes ...*offBox) {
+ sw.t.Helper()
+ sw.start()
+ names := []string{}
+ for _, b := range boxes {
+ b.tick()
+ names = append(names, b.name)
+ }
+ sw.move(provision.OfflineSeal(offID, sealer, names))
+}
+
+func (sw *sealWorld) mirror() lease.MirrorStore {
+ sw.t.Helper()
+ m, err := lease.NewS3Mirror(context.Background(), sealIERegion, sealUKRegion)
+ if err != nil {
+ sw.t.Fatal(err)
+ }
+ return m
+}
+
+func (sw *sealWorld) list(bucket, prefix string) map[string]lease.MirrorObject {
+ sw.t.Helper()
+ objs, err := sw.mirror().List(context.Background(), bucket, prefix)
+ if err != nil {
+ sw.t.Fatal(err)
+ }
+ return objs
+}
+
+func (sw *sealWorld) lock() (owner string, expires time.Time, generation int64) {
+ sw.t.Helper()
+ body, ok := sw.fake.Current(sealIE, sealPrefix+"/"+offHost+"/lock.json")
+ if !ok {
+ return "", time.Time{}, 0
+ }
+ var l litestream.Lease
+ if err := json.Unmarshal(body, &l); err != nil {
+ sw.t.Fatal(err)
+ }
+ return l.Owner, l.ExpiresAt, l.Generation
+}
+
+func (sw *sealWorld) sealRecord() (moves.OfflineReport, []byte, bool) {
+ sw.t.Helper()
+ raw, _, err := sw.s.Get(context.Background(), moves.OfflineSealKey(offHost, offID))
+ if errors.Is(err, os.ErrNotExist) {
+ return moves.OfflineReport{}, nil, false
+ }
+ if err != nil {
+ sw.t.Fatal(err)
+ }
+ var r moves.OfflineReport
+ if err := json.Unmarshal(raw, &r); err != nil {
+ sw.t.Fatal(err)
+ }
+ return r, raw, true
+}
+
+// acquireAsAWake is what every wake does first (ExecBackend.Acquire).
+func acquireAsAWake(ctx context.Context) (*lease.Held, error) {
+ return lease.Acquire(ctx, sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion)
+}
+
+func wantHeldByMove(t *testing.T, err error, what string) {
+ t.Helper()
+ var he *lease.HeldError
+ if !errors.As(err, &he) || he.Owner != moves.TombstoneOwner(offID) {
+ t.Fatalf("%s: %v, want a HeldError naming %s", what, err, moves.TombstoneOwner(offID))
+ }
+}
+
+// openArchive fetches the sealed archive and its manifest from data-uk,
+// verifies it and extracts it into a fresh flat files directory and env
+// path, returning them.
+func (sw *sealWorld) openArchive(seal moves.OfflineReport) (files, env string) {
+ t := sw.t
+ t.Helper()
+ prefix := moves.OfflineArchivePrefix(offHost, offID)
+ sealed, ok := sw.fake.Current(sealUK, prefix+offlineArchiveSealed)
+ if !ok {
+ t.Fatal("no sealed archive")
+ }
+ rawManifest, ok := sw.fake.Current(sealUK, prefix+offlineArchiveManifest)
+ if !ok {
+ t.Fatal("no archive manifest")
+ }
+ var m archive.Manifest
+ if err := json.Unmarshal(rawManifest, &m); err != nil {
+ t.Fatal(err)
+ }
+ bind := appconfig.Binding{Account: sw.acct, App: "app"}
+ if err := archive.Verify(bytes.NewReader(sealed), m, sw.priv, bind); err != nil {
+ t.Fatalf("Verify: %v", err)
+ }
+ sum := sha256.Sum256(sealed)
+ if m.Plain != seal.FilesPlain || m.Sealed != seal.FilesSealed || "sha256:"+hex.EncodeToString(sum[:]) != seal.FilesSealed || m.Bytes != seal.FilesBytes || len(m.Entries) != seal.FilesEntries {
+ t.Fatalf("manifest %+v against the seal %+v", m, seal)
+ }
+ var plain bytes.Buffer
+ if err := archive.Open(bytes.NewReader(sealed), m.SealedDataKey, sw.priv, bind, &plain); err != nil {
+ t.Fatalf("Open: %v", err)
+ }
+ parent := t.TempDir()
+ files, env = filepath.Join(parent, offHost), filepath.Join(parent, offHost+".env")
+ // A *bytes.Reader, not the buffer: zstd decodes a *bytes.Buffer under
+ // 128 KiB from its Bytes() without draining it, and Extract's
+ // end-of-stream check then finds the whole archive still unread.
+ if err := archive.Extract(bytes.NewReader(plain.Bytes()), files, env); err != nil {
+ t.Fatalf("Extract: %v", err)
+ }
+ if got, _, _, err := archive.PlainDigest(files, env); err != nil || got != seal.FilesPlain {
+ t.Fatalf("the extraction digests to %s (%v), the seal says %s", got, err, seal.FilesPlain)
+ }
+ return files, env
+}
+
+func readString(t *testing.T, p string) string {
+ t.Helper()
+ b, err := os.ReadFile(p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return string(b)
+}
+
+// (a) The happy seal, end to end, and the passes after it.
+func TestOfflineSealHappyPath(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ // A tail the stop never checkpointed: rows in the local database only.
+ sw.writeRows(ie, sw.dbPath(ie), 20, 25, false)
+ sw.startAndSeal("ie-a", ie)
+ ctx := context.Background()
+
+ sw.mustPass(ie)
+ rec := sw.record()
+ if rec.Move.State != provision.OfflineSealed {
+ t.Fatalf("state %s, want sealed", rec.Move.State)
+ }
+ seal, raw, ok := sw.sealRecord()
+ if !ok {
+ t.Fatal("no seal record")
+ }
+ if err := moves.ValidateOfflineSeal(seal, offHost, offID); err != nil {
+ t.Fatal(err)
+ }
+ src := rec.Move.Source
+ if src == nil || src.Evidence != moves.OfflineSealEvidence(offHost, offID, raw) || src.FinalTXID != seal.FinalTXID ||
+ src.Files == nil || src.Files.Plain != seal.FilesPlain || src.Files.Sealed != seal.FilesSealed || src.TimersSHA256 != seal.TimersDigest ||
+ src.Release != sealVersion || src.EnvDigest != seal.EnvDigest || src.Replication != registry.ReplicationBatched || src.NextDue != 1_800_000_000 {
+ t.Fatalf("record source %+v, seal %+v", src, seal)
+ }
+
+ // The tombstone, and a wake meets it.
+ owner, expires, _ := sw.lock()
+ if owner != moves.TombstoneOwner(offID) || time.Until(expires) < moves.TombstoneTTL-time.Minute {
+ t.Fatalf("lock %s until %s", owner, expires)
+ }
+ if h, err := acquireAsAWake(ctx); err == nil {
+ h.Release(ctx)
+ t.Fatal("a wake took the lease after the seal")
+ } else {
+ wantHeldByMove(t, err, "a wake after the seal")
+ }
+
+ // The staged copy is the replica, object for object, and its listing is
+ // the seal's.
+ srcObjs := sw.list(sealIE, sealPrefix+"/"+offHost+"/")
+ staged := sw.list(sealUK, activator.MovesPrefix(offHost))
+ if len(staged) != len(seal.Staged) || len(staged) != len(srcObjs)-1 {
+ t.Fatalf("staged %d objects, seal %d, source %d (lock.json included)", len(staged), len(seal.Staged), len(srcObjs))
+ }
+ for k, o := range staged {
+ rel := strings.TrimPrefix(k, activator.MovesPrefix(offHost))
+ if s := seal.Staged[rel]; s.Size != o.Size || s.ETag != o.ETag {
+ t.Fatalf("%s staged %+v, sealed %+v", rel, o, s)
+ }
+ if so := srcObjs[sealPrefix+"/"+offHost+"/"+rel]; so.Size != o.Size {
+ t.Fatalf("%s staged %+v, source %+v", rel, o, so)
+ }
+ }
+ if _, ok := seal.Staged["build-state.json"]; !ok {
+ t.Fatal("the carried build state is not staged")
+ }
+ if replicaPosition(staged, activator.MovesPrefix(offHost)) != seal.FinalTXID {
+ t.Fatal("the staged newest TXID is not the seal's")
+ }
+ // The staged replica restores to every row, the tail included: the
+ // checkpoint ran before the tombstone.
+ restored := filepath.Join(t.TempDir(), "restored.db")
+ if ok, err := lease.Restore(ctx, restored, sealUK, strings.TrimSuffix(activator.MovesPrefix(offHost), "/"), sealUKRegion); err != nil || !ok {
+ t.Fatalf("restore the staged copy: %v %v", ok, err)
+ }
+ if n := countRows(t, restored); n != 25 {
+ t.Fatalf("the staged copy holds %d rows, want 25", n)
+ }
+
+ // The archive opens, verifies and extracts to the same plaintext, with
+ // the files and the env byte for byte.
+ files, env := sw.openArchive(seal)
+ for rel, want := range map[string]string{"instance.key": "app key file", "uploads/a.png": "png bytes"} {
+ if got := readString(t, filepath.Join(files, rel)); got != want {
+ t.Fatalf("%s = %q", rel, got)
+ }
+ }
+ envBody := readString(t, filepath.Join(ie.dd, offHost+".env"))
+ if readString(t, env) != envBody {
+ t.Fatal("the env file did not travel byte for byte")
+ }
+ sum := sha256.Sum256([]byte(envBody))
+ if seal.EnvSHA256 != hex.EncodeToString(sum[:]) || seal.KeySHA256 != instancekey.Digest("the-instance-key") {
+ t.Fatalf("env %s key %s", seal.EnvSHA256, seal.KeySHA256)
+ }
+
+ // The timers snapshot (active rows only) and the env report.
+ prefix := moves.OfflineArchivePrefix(offHost, offID)
+ timersBody, _ := sw.fake.Current(sealUK, prefix+offlineTimersObject)
+ snap, err := moves.DecodeTimers(timersBody)
+ if err != nil || len(snap.Rows) != 2 || moves.TimersDigest(snap.Rows) != seal.TimersDigest {
+ t.Fatalf("timers snapshot %+v %v", snap, err)
+ }
+ reportBody, _ := sw.fake.Current(sealUK, prefix+offlineEnvReportObject)
+ var envReport moves.EnvReport
+ if err := json.Unmarshal(reportBody, &envReport); err != nil || envReport.Digest != seal.EnvDigest || envReport.Vars["APP_COLOUR"] == "" || strings.Contains(string(reportBody), "green") {
+ t.Fatalf("env report %s %v", reportBody, err)
+ }
+ if seal.Release != sealVersion || seal.Replication != registry.ReplicationBatched || seal.NextDue != 1_800_000_000 {
+ t.Fatalf("seal %+v", seal)
+ }
+
+ // The living report: the tombstone and the pass's time.
+ living := sw.report(moves.OfflineRoleSealer, "ie-a")
+ if living.Phase != moves.OfflinePhaseSealed || living.TombstoneOwner != moves.TombstoneOwner(offID) || !living.TombstoneUntil.Equal(expires) || !living.At.Equal(sw.clock()) {
+ t.Fatalf("living report %+v", living)
+ }
+ if ok, bad := living.SourceItemsOK(); !ok {
+ t.Fatalf("living report items %v", bad)
+ }
+
+ // Later passes re-assert the expiry, keep the tombstone's generation,
+ // never touch the seal record, and rewrite the living report.
+ _, _, gen := sw.lock()
+ for _, state := range []func(){func() {}, sw.flip, sw.landed} {
+ state()
+ time.Sleep(5 * time.Millisecond)
+ sw.advance(time.Second)
+ if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
+ t.Fatal(err)
+ }
+ owner2, expires2, gen2 := sw.lock()
+ if owner2 != owner || !expires2.After(expires) || gen2 != gen {
+ t.Fatalf("after a later pass: %s until %s (generation %d), before %s until %s (generation %d)", owner2, expires2, gen2, owner, expires, gen)
+ }
+ expires = expires2
+ living = sw.report(moves.OfflineRoleSealer, "ie-a")
+ if !living.TombstoneUntil.Equal(expires2) || !living.At.Equal(sw.clock()) || living.Phase != moves.OfflinePhaseSealed {
+ t.Fatalf("living report after a later pass: %+v", living)
+ }
+ if _, again, _ := sw.sealRecord(); !bytes.Equal(again, raw) {
+ t.Fatal("the seal record was rewritten")
+ }
+ }
+ if len(sw.pages) != 0 {
+ t.Fatalf("paged: %v", sw.pages)
+ }
+}
+
+// (b) A live writer's lease: with a local database the checkpoint cannot
+// take it, without one the tombstone cannot; either way the state stays
+// sealing and nothing is staged or sealed. A wake that races the tombstone
+// meets it.
+func TestOfflineSealWaitsForALiveWriter(t *testing.T) {
+ ctx := context.Background()
+ for _, local := range []bool{true, false} {
+ t.Run(fmt.Sprintf("local database %v", local), func(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ if !local {
+ db := sw.dbPath(ie)
+ for _, p := range []string{db, db + "-wal", db + "-shm", activatorMetaDir(db)} {
+ if err := os.RemoveAll(p); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ sw.startAndSeal("ie-a", ie)
+ wake, err := acquireAsAWake(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ err = sw.pass(ie)
+ var he *lease.HeldError
+ if !errors.As(err, &he) || he.Owner == moves.TombstoneOwner(offID) {
+ t.Fatalf("a live writer: %v", err)
+ }
+ if !local && !strings.Contains(err.Error(), "tombstone") {
+ t.Fatalf("not refused at the tombstone: %v", err)
+ }
+ if sw.record().Move.State != provision.OfflineSealing {
+ t.Fatal("the state moved")
+ }
+ if _, _, ok := sw.sealRecord(); ok {
+ t.Fatal("sealed under a live writer")
+ }
+ if n := len(sw.list(sealUK, "moves/")); n != 0 {
+ t.Fatalf("%d objects staged under a live writer", n)
+ }
+ if err := wake.Err(); err != nil {
+ t.Fatalf("the writer lost its lease: %v", err)
+ }
+ if err := wake.Release(ctx); err != nil {
+ t.Fatal(err)
+ }
+
+ // The race: a wake that comes for the lease the moment after
+ // the tombstone is written is refused by it.
+ raced := false
+ offlineSealFaults.Store(ie.w, func(point string) error {
+ if point == "tombstone" {
+ raced = true
+ h, err := acquireAsAWake(ctx)
+ if err == nil {
+ h.Release(ctx)
+ return errors.New("a wake took the lease after the tombstone")
+ }
+ wantHeldByMove(t, err, "a wake racing the tombstone")
+ }
+ return nil
+ })
+ t.Cleanup(func() { offlineSealFaults.Delete(ie.w) })
+ sw.mustPass(ie)
+ if !raced || sw.record().Move.State != provision.OfflineSealed {
+ t.Fatalf("raced %v, state %s", raced, sw.record().Move.State)
+ }
+ })
+ }
+}
+
+// (c) A source box whose local position is past the replica's final
+// position refuses the seal: the db-tail item fails, the report says
+// failed, nothing is sealed and the state stays sealing.
+func TestOfflineSealRefusesASourceAheadOfTheFinalPosition(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ ieB := sw.sourceBox("ie-b")
+ ieB.tick()
+ final := replicaPosition(sw.list(sealIE, sealPrefix+"/"+offHost+"/"), sealPrefix+"/"+offHost+"/")
+ if final == 0 {
+ t.Fatal("fixture: no replica")
+ }
+ // ie-b woke the host once and kept a position past the replica's.
+ meta := filepath.Join(activatorMetaDir(provision.DefaultDBPath(ieB.dd, offHost)), "ltx", "0")
+ writeFile(t, filepath.Join(meta, ltxName(final+1, final+5)), "ltx")
+ sw.startAndSeal("ie-a", ie, ieB)
+ if r := sw.report(moves.OfflineRoleSource, "ie-b"); r.LocalTXID != final+5 {
+ t.Fatalf("fixture: ie-b reports %d", r.LocalTXID)
+ }
+ err := sw.pass(ie)
+ var ref *sealRefusal
+ if !errors.As(err, &ref) || ref.item != "db-tail" || !strings.Contains(ref.reason, "ie-b") {
+ t.Fatalf("pass: %v", err)
+ }
+ r := sw.report(moves.OfflineRoleSealer, "ie-a")
+ if r.Phase != moves.OfflinePhaseFailed || r.Items["db-tail"].OK || !r.Items["sidecar"].OK {
+ t.Fatalf("report %+v", r)
+ }
+ if _, _, ok := sw.sealRecord(); ok || sw.record().Move.State != provision.OfflineSealing {
+ t.Fatal("sealed past a source ahead of the replica")
+ }
+}
+
+func ltxName(min, max uint64) string {
+ return ltx.FormatFilename(ltx.TXID(min), ltx.TXID(max))
+}
+
+// (d) After the seal the stop is kept, never taken again: a tombstone
+// stolen by another move, expired, lifted or deleted is a lost stop. The
+// pass reports failed with the lease item down, pages once for the move,
+// and leaves the lease exactly as it found it.
+func TestOfflineSealLostStopIsNeverRetaken(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ sw.startAndSeal("ie-a", ie)
+ sw.mustPass(ie)
+ ctx := context.Background()
+ key := sealPrefix + "/" + offHost + "/lock.json"
+ ours, _ := sw.fake.Current(sealIE, key)
+ lockBody := func(owner string, expires time.Time) []byte {
+ b, err := json.Marshal(litestream.Lease{Generation: 1, Owner: owner, ExpiresAt: expires})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+ }
+ cases := []struct {
+ name string
+ damage func()
+ }{
+ {"stolen by another move", func() {
+ sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID2), time.Now().Add(time.Hour)))
+ }},
+ {"stolen by a wake", func() {
+ sw.fake.Put(sealIE, key, lockBody("box-x", time.Now().Add(time.Hour)))
+ }},
+ {"expired", func() {
+ sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID), time.Now().Add(-time.Second)))
+ }},
+ {"lifted", func() {
+ if _, err := lease.LiftTombstone(ctx, sealIE, sealPrefix+"/"+offHost, sealIERegion, func(o string) bool { return o == moves.TombstoneOwner(offID) }); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ {"deleted", func() { sw.fake.BareDelete(sealIE, key) }},
+ }
+ for _, c := range cases {
+ // The stop holds: a pass is clean.
+ sw.fake.Put(sealIE, key, ours)
+ if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
+ t.Fatalf("%s: before: %v", c.name, err)
+ }
+ c.damage()
+ before := sw.fake.Snapshot()
+ for i := range 2 {
+ err := ie.w.offlineSealerStep(ctx, sw.s, sw.record())
+ if err == nil || !strings.Contains(err.Error(), "lost stop") {
+ t.Fatalf("%s, pass %d: %v", c.name, i, err)
+ }
+ if sw.fake.Snapshot() != before {
+ t.Fatalf("%s, pass %d: the lease was written after the stop was lost", c.name, i)
+ }
+ r := sw.report(moves.OfflineRoleSealer, "ie-a")
+ if r.Phase != moves.OfflinePhaseFailed || r.Items["lease"].OK || !strings.Contains(r.Items["lease"].Reason, "lost stop") || r.TombstoneOwner != "" || !r.Items["files"].OK {
+ t.Fatalf("%s: report %+v", c.name, r)
+ }
+ }
+ }
+ if len(sw.pages) != 1 || !strings.HasPrefix(sw.pages[0], "incident: ") {
+ t.Fatalf("pages %v, want one incident", sw.pages)
+ }
+ if sw.record().Move.State != provision.OfflineSealed {
+ t.Fatal("the record moved")
+ }
+}
+
+// (e) A crash after each step, then the next pass: one tombstone (one
+// generation, created once), staged objects copied once and identical, the
+// archive the seal names, and a sealed record.
+func TestOfflineSealReplaysAfterACrashAtEachStep(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ sw.writeRows(ie, sw.dbPath(ie), 20, 23, false)
+ sw.startAndSeal("ie-a", ie)
+ ctx := context.Background()
+ points := []string{"checkpoint", "tombstone", "final", "mirror", "archive", "timers", "env", "seal", "cas"}
+ t.Cleanup(func() { offlineSealFaults.Delete(ie.w) })
+ var gen int64
+ var stagedAfter map[string]lease.MirrorObject
+ for i, crash := range points {
+ hit := false
+ offlineSealFaults.Store(ie.w, func(point string) error {
+ if point == crash {
+ hit = true
+ return errors.New("crash at " + point)
+ }
+ return nil
+ })
+ err := sw.pass(ie)
+ if !hit || err == nil || !strings.Contains(err.Error(), "crash at "+crash) {
+ t.Fatalf("crash at %s: hit %v, %v", crash, hit, err)
+ }
+ if i >= 1 { // the tombstone is down from the crash after it on
+ owner, _, g := sw.lock()
+ if owner != moves.TombstoneOwner(offID) || (gen != 0 && g != gen) {
+ t.Fatalf("after the crash at %s: lock %s generation %d (was %d)", crash, owner, g, gen)
+ }
+ gen = g
+ }
+ if crash == "mirror" {
+ stagedAfter = sw.list(sealUK, activator.MovesPrefix(offHost))
+ }
+ wantState := provision.OfflineSealing
+ if crash == "cas" {
+ wantState = provision.OfflineSealed
+ }
+ if s := sw.record().Move.State; s != wantState {
+ t.Fatalf("after the crash at %s: state %s", crash, s)
+ }
+ }
+ offlineSealFaults.Delete(ie.w)
+ sw.mustPass(ie)
+ if sw.record().Move.State != provision.OfflineSealed {
+ t.Fatal("not sealed after the replays")
+ }
+ if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || r.TombstoneOwner != moves.TombstoneOwner(offID) {
+ t.Fatalf("living report %+v", r)
+ }
+ // One tombstone: created once (one version with the move's owner and
+ // no expiry change before the seal), one generation throughout.
+ if _, _, g := sw.lock(); g != gen {
+ t.Fatalf("generation %d, was %d", g, gen)
+ }
+ versions, err := sw.fake.ListVersions(ctx, sealIE, sealPrefix+"/"+offHost+"/lock.json")
+ if err != nil {
+ t.Fatal(err)
+ }
+ created := 0
+ for _, v := range versions {
+ if v.DeleteMarker {
+ continue
+ }
+ body, err := sw.fake.GetVersion(ctx, sealIE, v.Key, v.VersionID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var l litestream.Lease
+ if err := json.Unmarshal(body, &l); err == nil && l.Owner == moves.TombstoneOwner(offID) && l.Generation != gen {
+ t.Fatalf("a tombstone of another generation %d", l.Generation)
+ }
+ if l.Owner == moves.TombstoneOwner(offID) {
+ created++
+ }
+ }
+ // The tombstone is written once before the seal (the replays after it
+ // find it this move's and write nothing); after the seal the re-asserts
+ // of the last two passes rewrite it, same generation.
+ if created != 3 {
+ t.Fatalf("%d versions of the tombstone", created)
+ }
+ // The staged objects: identical to those after the first mirror round,
+ // and each copied once.
+ final := sw.list(sealUK, activator.MovesPrefix(offHost))
+ if len(final) != len(stagedAfter) {
+ t.Fatalf("staged %d objects, %d after the first round", len(final), len(stagedAfter))
+ }
+ for k, o := range final {
+ if stagedAfter[k] != o {
+ t.Fatalf("%s changed: %+v, was %+v", k, o, stagedAfter[k])
+ }
+ vs, err := sw.fake.ListVersions(ctx, sealUK, k)
+ if err != nil || len(vs) != 1 {
+ t.Fatalf("%s has %d versions (%v)", k, len(vs), err)
+ }
+ }
+ seal, _, ok := sw.sealRecord()
+ if !ok {
+ t.Fatal("no seal record")
+ }
+ sw.openArchive(seal)
+ if n := len(seal.Staged); n != len(final) {
+ t.Fatalf("seal names %d staged objects, %d staged", n, len(final))
+ }
+}
+
+// (f) The revert: the sealer lifts this move's tombstone, deletes exactly
+// the staging lane and the archive prefix in the target's bucket, clears
+// the receipts and lifts the fence; a wake can take the lease again. A
+// lease that is not this move's is left alone. Every part is an ensure.
+func TestOfflineSealRevertLiftsAndUnfences(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ sw.startAndSeal("ie-a", ie)
+ sw.mustPass(ie)
+ ctx := context.Background()
+ // Neighbours the revert must not touch.
+ sw.fake.Put(sealUK, "moves/other.example.com/0000/x.ltx", []byte("x"))
+ sw.fake.Put(sealUK, "moves/@offline/other.example.com/"+offID2+"/files.sealed", []byte("x"))
+ sw.fake.Put(sealUK, sealPrefix+"/"+offHost+"/0000/y.ltx", []byte("y"))
+ // Copy receipts, as a round over a multipart source leaves them.
+ sw.fake.Put(sealIE, activator.MoveReceiptsKey(offHost), []byte(`{"v":1}`))
+ sw.flip()
+ sw.abort()
+ if sw.record().Move.State != provision.OfflineReverting {
+ t.Fatal("fixture: not reverting")
+ }
+ if !ie.svc.Fenced(offHost) {
+ t.Fatal("fixture: not fenced")
+ }
+ // The sealer's own step lifts the fence too (the source step does it
+ // as well, earlier in the same round).
+ if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
+ t.Fatal(err)
+ }
+ if ie.svc.Fenced(offHost) {
+ t.Fatal("still fenced after the revert")
+ }
+ owner, expires, _ := sw.lock()
+ if !lease.IsLiftedTombstone(owner, expires) || owner != moves.TombstoneOwner(offID) {
+ t.Fatalf("lock %s %s, want this move's lifted tombstone", owner, expires)
+ }
+ for _, prefix := range []string{activator.MovesPrefix(offHost), moves.OfflineArchivePrefix(offHost, offID)} {
+ if n := len(sw.list(sealUK, prefix)); n != 0 {
+ t.Fatalf("%s holds %d objects", prefix, n)
+ }
+ }
+ for _, k := range []string{"moves/other.example.com/0000/x.ltx", "moves/@offline/other.example.com/" + offID2 + "/files.sealed", sealPrefix + "/" + offHost + "/0000/y.ltx"} {
+ if _, ok := sw.fake.Current(sealUK, k); !ok {
+ t.Fatalf("the revert deleted %s", k)
+ }
+ }
+ if _, ok := sw.fake.Current(sealIE, activator.MoveReceiptsKey(offHost)); ok {
+ t.Fatal("the receipts are still there")
+ }
+ if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseReverted {
+ t.Fatalf("report %+v", r)
+ }
+ // Again: nothing changes, and the lifted tombstone is not taken again.
+ before := sw.fake.Snapshot()
+ sw.mustPass(ie)
+ if sw.fake.Snapshot() != before {
+ t.Fatal("a second revert pass wrote to the buckets")
+ }
+ wake, err := acquireAsAWake(ctx)
+ if err != nil {
+ t.Fatalf("a wake after the revert: %v", err)
+ }
+ // A lease that is not this move's (a live wake now) is left alone.
+ if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
+ t.Fatal(err)
+ }
+ if err := wake.Err(); err != nil {
+ t.Fatalf("the revert disturbed a wake's lease: %v", err)
+ }
+ wake.Release(ctx)
+ // Another move's tombstone is never lifted by this one.
+ other, err := json.Marshal(litestream.Lease{Generation: 9, Owner: moves.TombstoneOwner(offID2), ExpiresAt: time.Now().Add(time.Hour)})
+ if err != nil {
+ t.Fatal(err)
+ }
+ sw.fake.Put(sealIE, sealPrefix+"/"+offHost+"/lock.json", other)
+ if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
+ t.Fatal(err)
+ }
+ if owner, expires, _ := sw.lock(); owner != moves.TombstoneOwner(offID2) || !time.Now().Before(expires) {
+ t.Fatalf("another move's tombstone was lifted: %s %s", owner, expires)
+ }
+}
+
+// The revert before anything was sealed (abort while sealing failed on a
+// refused item) lifts the tombstone the failed seal took, and a revert
+// with no tombstone at all is a clean pass.
+func TestOfflineSealRevertBeforeTheSeal(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ writeFile(t, filepath.Join(ie.dd, offHost, "big"), "")
+ if err := os.Truncate(filepath.Join(ie.dd, offHost, "big"), archive.Cap+1); err != nil {
+ t.Fatal(err)
+ }
+ sw.startAndSeal("ie-a", ie)
+ if err := sw.pass(ie); err == nil {
+ t.Fatal("sealed over the cap")
+ }
+ if owner, _, _ := sw.lock(); owner != moves.TombstoneOwner(offID) {
+ t.Fatal("fixture: no tombstone")
+ }
+ sw.abort()
+ sw.mustPass(ie)
+ if owner, expires, _ := sw.lock(); !lease.IsLiftedTombstone(owner, expires) {
+ t.Fatalf("lock %s %s after the revert", owner, expires)
+ }
+ if n := len(sw.list(sealUK, "moves/")); n != 0 {
+ t.Fatalf("%d objects left in the staging prefixes", n)
+ }
+ sw.mustPass(ie)
+}
+
+// The files item: a file over the cap, a FIFO, and an entry of the host's
+// beside its files directory are failed items (the report says which),
+// never a partial archive and never a seal; once the file is gone the next
+// pass seals.
+func TestOfflineSealFailsAFileOverTheCapAndAFIFO(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ sw.startAndSeal("ie-a", ie)
+ files := filepath.Join(ie.dd, offHost)
+ big := filepath.Join(files, "big.bin")
+ writeFile(t, big, "")
+ if err := os.Truncate(big, archive.Cap+1); err != nil {
+ t.Fatal(err)
+ }
+ fifo := filepath.Join(files, "pipe")
+ blobs := filepath.Join(ie.dd, offHost+"-blobs")
+ cases := []struct {
+ name string
+ setup func()
+ want string
+ }{
+ {"over the cap", func() {}, strings.TrimPrefix(archive.ErrOverCap.Error(), "archive: ")},
+ {"a FIFO", func() {
+ if err := os.Remove(big); err != nil {
+ t.Fatal(err)
+ }
+ if err := syscall.Mkfifo(fifo, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }, strings.TrimPrefix(archive.ErrSpecialFile.Error(), "archive: ") + ": pipe"},
+ // The flat layout's rules give <host>-blobs to the host, and the
+ // archive carries the files directory alone.
+ {"an entry beside the files directory", func() {
+ if err := os.Remove(fifo); err != nil {
+ t.Fatal(err)
+ }
+ writeFile(t, filepath.Join(blobs, "b"), "blob")
+ }, blobs},
+ }
+ for _, c := range cases {
+ c.setup()
+ err := sw.pass(ie)
+ var ref *sealRefusal
+ if !errors.As(err, &ref) || ref.item != "files" || !strings.Contains(ref.reason, c.want) {
+ t.Fatalf("%s: %v", c.name, err)
+ }
+ r := sw.report(moves.OfflineRoleSealer, "ie-a")
+ if r.Phase != moves.OfflinePhaseFailed || r.Items["files"].OK || !r.Items["db"].OK {
+ t.Fatalf("%s: report %+v", c.name, r)
+ }
+ if _, _, ok := sw.sealRecord(); ok || sw.record().Move.State != provision.OfflineSealing {
+ t.Fatalf("%s: sealed", c.name)
+ }
+ if n := len(sw.list(sealUK, moves.OfflineArchivePrefix(offHost, offID))); n != 0 {
+ t.Fatalf("%s: %d archive objects written", c.name, n)
+ }
+ }
+ if err := os.RemoveAll(blobs); err != nil {
+ t.Fatal(err)
+ }
+ sw.mustPass(ie)
+ if sw.record().Move.State != provision.OfflineSealed {
+ t.Fatal("not sealed once the files were fixed")
+ }
+ if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || !r.Items["files"].OK {
+ t.Fatalf("report %+v", r)
+ }
+}
+
+// A nested instance, made nested by the real layout pass, which takes the
+// flat files directory <data>/<host> into the instance directory as
+// <instance dir>/<host>, where the app finds it beside its database. The
+// archive is read through the raw instance directory and carries that
+// files directory and the env file, never the database, its litestream
+// state, the rendered configuration or any other platform entry; it
+// extracts flat with the app's files at the flat files directory's root,
+// where the flat app looks. An entry beside the files directory that the
+// flat layout cannot carry refuses the files item until it is gone.
+func TestOfflineSealArchivesANestedInstanceFlat(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ ctx := context.Background()
+ // The bind mount at <dd>/instances, as a link in the fixture.
+ if err := os.MkdirAll(filepath.Join(ie.dd, ".instances"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(".instances", filepath.Join(ie.dd, "instances")); err != nil {
+ t.Fatal(err)
+ }
+ // The layout pass works on the row's own paths; the sockets live in a
+ // short directory of the test's, as the pass's tests keep them.
+ sd, err := os.MkdirTemp("", "ly")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(sd) })
+ rt := ie.row(offHost)
+ if err := ie.reg.SetInstancePaths(offHost, rt.DBPath, rt.Socket, rt.DBPath, filepath.Join(sd, offHost+".sock")); err != nil {
+ t.Fatal(err)
+ }
+ pass := &layout.Pass{Registry: ie.reg, DataDir: ie.dd, SocketDir: sd, SocketsGID: -1, Now: time.Now, SweptAtStart: true, Logf: t.Logf}
+ res := pass.Converge(ctx, layout.ModeNested)
+ migrated := false
+ for _, r := range res {
+ if r.Host == offHost && r.Outcome == "migrated" {
+ migrated = true
+ }
+ }
+ if !migrated {
+ t.Fatalf("the layout pass: %+v", res)
+ }
+ ie.nested = true
+ raw := layout.RawInstanceDir(ie.dd, offHost)
+ if rt := ie.row(offHost); !provision.NestedPaths(ie.dd, *rt) {
+ t.Fatalf("fixture: row %+v is not nested", rt)
+ }
+ if lexists(filepath.Join(ie.dd, offHost)) {
+ t.Fatal("fixture: the layout pass left the flat files directory")
+ }
+ for _, name := range []string{offHost + ".db", offHost + ".env", offHost + ".config.env", ".activator-" + offHost + ".db-litestream", filepath.Join(offHost, "instance.key")} {
+ if !lexists(filepath.Join(raw, name)) {
+ t.Fatalf("fixture: the layout pass did not move %s into the instance directory", name)
+ }
+ }
+ // More platform entries, which stay behind.
+ writeFile(t, filepath.Join(raw, offHost+".db.gen"), "generation")
+ writeFile(t, filepath.Join(raw, ".files-gen"), "marker")
+ writeFile(t, filepath.Join(raw, offHost+instancekey.AdoptionSuffix), "{}")
+ // An entry of the app's beside its files directory, which no flat
+ // files directory can hold.
+ stray := filepath.Join(raw, offHost+"-blobs", "b")
+ writeFile(t, stray, "blob")
+ sw.startAndSeal("ie-a", ie)
+ err = sw.pass(ie)
+ var ref *sealRefusal
+ if !errors.As(err, &ref) || ref.item != "files" || !strings.Contains(ref.reason, offHost+"-blobs") {
+ t.Fatalf("a stray entry: %v", err)
+ }
+ if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseFailed || r.Items["files"].OK {
+ t.Fatalf("report %+v", r)
+ }
+ if err := os.RemoveAll(filepath.Dir(stray)); err != nil {
+ t.Fatal(err)
+ }
+ sw.mustPass(ie)
+ if sw.record().Move.State != provision.OfflineSealed {
+ t.Fatal("not sealed")
+ }
+ seal, _, _ := sw.sealRecord()
+ files, env := sw.openArchive(seal)
+ got := map[string]bool{}
+ if err := filepath.Walk(files, func(p string, info os.FileInfo, err error) error {
+ if err != nil {
+ return err
+ }
+ if !info.IsDir() {
+ rel, _ := filepath.Rel(files, p)
+ got[rel] = true
+ }
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 2 || !got["instance.key"] || !got[filepath.Join("uploads", "a.png")] {
+ t.Fatalf("the archive carried %v, want the app's two files", got)
+ }
+ if readString(t, env) != readString(t, filepath.Join(raw, offHost+".env")) {
+ t.Fatal("the env file did not travel byte for byte")
+ }
+ if seal.KeySHA256 != instancekey.Digest("the-instance-key") {
+ t.Fatal("the key digest")
+ }
+}
+
+// Finishing lifts the tombstone only once every box the seal named reports
+// gone: until then each pass keeps the stop. After the lift a wake takes
+// the lease, the report says lifted, and a later pass changes nothing.
+func TestOfflineSealFinishingLiftsOnlyAfterEverySourceIsGone(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ ieB := sw.sourceBox("ie-b")
+ sw.startAndSeal("ie-a", ie, ieB)
+ sw.mustPass(ie)
+ ieB.tick()
+ if sw.record().Move.State != provision.OfflineSealed {
+ t.Fatal("not sealed")
+ }
+ sw.flip()
+ sw.landed()
+ sw.open()
+ sw.mustPass(ie)
+ sw.mustPass(ieB)
+ for _, b := range []string{"ie-a", "ie-b"} {
+ if r := sw.report(moves.OfflineRoleSource, b); r.Phase != moves.OfflinePhaseParked {
+ t.Fatalf("%s: %+v", b, r)
+ }
+ }
+ sw.finishing()
+ ctx := context.Background()
+ // ie-a goes; ie-b has not run a pass since.
+ for range 3 {
+ sw.mustPass(ie)
+ }
+ if r := sw.report(moves.OfflineRoleSource, "ie-a"); r.Phase != moves.OfflinePhaseGone {
+ t.Fatalf("ie-a: %+v", r)
+ }
+ owner, _, _ := sw.lock()
+ if owner != moves.TombstoneOwner(offID) {
+ t.Fatal("fixture: no tombstone")
+ }
+ if _, err := acquireAsAWake(ctx); err == nil {
+ t.Fatal("the tombstone was lifted before ie-b was gone")
+ }
+ if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || r.TombstoneOwner != moves.TombstoneOwner(offID) {
+ t.Fatalf("sealer report while waiting: %+v", r)
+ }
+ for range 3 {
+ sw.mustPass(ieB)
+ }
+ if r := sw.report(moves.OfflineRoleSource, "ie-b"); r.Phase != moves.OfflinePhaseGone {
+ t.Fatalf("ie-b: %+v", r)
+ }
+ sw.mustPass(ie)
+ if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseLifted {
+ t.Fatalf("sealer report: %+v", r)
+ }
+ owner, expires, _ := sw.lock()
+ if !lease.IsLiftedTombstone(owner, expires) || owner != moves.TombstoneOwner(offID) {
+ t.Fatalf("lock %s %s", owner, expires)
+ }
+ before := sw.fake.Snapshot()
+ sw.mustPass(ie)
+ if sw.fake.Snapshot() != before {
+ t.Fatal("a pass after the lift wrote to the buckets")
+ }
+ wake, err := acquireAsAWake(ctx)
+ if err != nil {
+ t.Fatalf("a wake after the lift: %v", err)
+ }
+ wake.Release(ctx)
+ if len(sw.pages) != 0 {
+ t.Fatalf("paged: %v", sw.pages)
+ }
+}
diff --git a/internal/layout/plan.go b/internal/layout/plan.go
index ed1b17194..d1d52d435 100644
--- a/internal/layout/plan.go
+++ b/internal/layout/plan.go
@@ -12,6 +12,7 @@ import (
"path/filepath"
"github.com/carlosframework/platform/internal/controlname"
+ "github.com/carlosframework/platform/internal/instancekey"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/release"
)
@@ -154,6 +155,40 @@ func nestedFiles(dataDir, socketDir, host string) []string {
}
}
+// filesMarkerName is the files directory's generation marker
+// (archive.MarkerName, hostfiles' marker), named here rather than imported:
+// the archive package depends on hostfiles and appconfig, which this
+// package has no other reason to.
+const filesMarkerName = ".files-gen"
+
+// NestedPlatformNames is every name in a nested host's instance directory
+// that is the platform's, not the app's: the names nestedFiles puts in the
+// raw instance directory (the database, its WAL, SHM and quarantine, the
+// own and sidecar env files, the rendered configuration, the activator's
+// litestream shadow; the socket is in the run directory and is not here),
+// the database's generation marker (<db>.gen), litestream's stock meta
+// directory (<db>-litestream), the files generation marker (.files-gen) and
+// the instance-key adoption record. Everything else in the directory is the
+// app's files. The offline move's archive leaves these out
+// (archive.BuildFiltered), so the files it carries are the app's alone and
+// extract flat.
+func NestedPlatformNames(host string) map[string]bool {
+ const dataDir, socketDir = "/data", "/run/carlos"
+ raw := rawInstanceDir(dataDir, host)
+ out := map[string]bool{}
+ for _, p := range nestedFiles(dataDir, socketDir, host) {
+ if filepath.Dir(p) == raw {
+ out[filepath.Base(p)] = true
+ }
+ }
+ db := host + ".db"
+ out[db+".gen"] = true
+ out[db+"-litestream"] = true
+ out[filesMarkerName] = true
+ out[host+instancekey.AdoptionSuffix] = true
+ return out
+}
+
// PlanFor derives the plan for host in the given direction from the host
// name alone (design §8.2 step 3). The endpoints are the row values the
// pass will compare against and swap; the artifacts are the files (and one
diff --git a/internal/layout/plan_test.go b/internal/layout/plan_test.go
index 27381c460..f1d21e34e 100644
--- a/internal/layout/plan_test.go
+++ b/internal/layout/plan_test.go
@@ -150,6 +150,44 @@ func TestRawRunDirIsTheDottedRawSocketTree(t *testing.T) {
}
}
+// NestedPlatformNames is exactly the plan's raw-directory names (every
+// nested artifact the pass moves into the instance directory, the socket
+// aside) plus the generation markers, litestream's stock meta directory and
+// the adoption record; and it covers every name a nested plan's artifacts
+// land at in the raw directory, so no platform file of the layout pass can
+// reach an app archive.
+func TestNestedPlatformNamesAreThePlansRawNames(t *testing.T) {
+ const h = "h.example.com"
+ got := NestedPlatformNames(h)
+ want := map[string]bool{
+ h + ".db": true, h + ".db-wal": true, h + ".db-shm": true, h + ".db.corrupt": true,
+ h + ".env": true, h + "-sidecar.env": true, h + ".config.env": true,
+ ".activator-" + h + ".db-litestream": true, ".files-gen": true,
+ h + ".db.gen": true, h + ".db-litestream": true, h + ".instance-key-adoption.json": true,
+ }
+ if len(got) != len(want) {
+ t.Fatalf("NestedPlatformNames = %v, want %v", got, want)
+ }
+ for n := range want {
+ if !got[n] {
+ t.Errorf("%s missing", n)
+ }
+ }
+ p, err := PlanFor("/data", "/run/carlos", h, ModeNested)
+ if err != nil {
+ t.Fatal(err)
+ }
+ raw := rawInstanceDir("/data", h)
+ for _, a := range p.Artifacts {
+ if filepath.Dir(a.Dst) == raw && !got[filepath.Base(a.Dst)] {
+ t.Errorf("artifact %s lands at %s, which NestedPlatformNames does not name", a.Name, a.Dst)
+ }
+ if a.Name == "socket" && got[filepath.Base(a.Dst)] {
+ t.Errorf("the socket is in the run directory, not the instance directory")
+ }
+ }
+}
+
func TestPlanForRefusesBadInput(t *testing.T) {
if _, err := PlanFor("/data", "/run/carlos", "h.example.com", Mode("sideways")); err == nil {
t.Error("unknown mode accepted")
diff --git a/internal/lease/leasetest/s3server.go b/internal/lease/leasetest/s3server.go
index 051b9f391..ee7819d3b 100644
--- a/internal/lease/leasetest/s3server.go
+++ b/internal/lease/leasetest/s3server.go
@@ -2,6 +2,7 @@ package leasetest
import (
"crypto/sha256"
+ "encoding/base64"
"encoding/hex"
"encoding/xml"
"errors"
@@ -9,6 +10,7 @@ import (
"io"
"net/http"
"net/url"
+ "sort"
"strconv"
"strings"
"sync/atomic"
@@ -105,7 +107,7 @@ func writeErr(w http.ResponseWriter, err error) {
}
// NewS3Server speaks the subset of the S3 wire protocol VersionedStore uses,
-// path-style, over f, holding f.mu for each request. The second result
+// and ListObjectsV2 (serveListV2), path-style, over f, holding f.mu for each request. The second result
// counts requests served. With serves non-empty, the endpoint answers only
// for those request buckets (a regional endpoint); any other bucket gets
// 301 PermanentRedirect, as S3 answers a wrong-region request. A copy's
@@ -145,6 +147,8 @@ func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int
switch {
case r.Method == http.MethodGet && key == "" && q.Has("versions"):
f.serveList(w, bucket, q)
+ case r.Method == http.MethodGet && key == "" && q.Get("list-type") == "2":
+ f.serveListV2(w, bucket, q)
case r.Method == http.MethodGet && key != "":
versionID := q.Get("versionId")
if versionID == "" { // no version named: the current one
@@ -289,6 +293,127 @@ func (f *VersionedBucket) serveList(w http.ResponseWriter, bucket string, q url.
xml.NewEncoder(w).Encode(res)
}
+type xmlObject struct {
+ Key string
+ LastModified string
+ ETag string
+ Size int64
+ StorageClass string
+}
+
+type xmlCommonPrefix struct {
+ Prefix string
+}
+
+type listBucketResult struct {
+ XMLName xml.Name `xml:"ListBucketResult"`
+ Name string
+ Prefix string
+ Delimiter string `xml:",omitempty"`
+ MaxKeys int
+ KeyCount int
+ IsTruncated bool
+ ContinuationToken string `xml:",omitempty"`
+ NextContinuationToken string `xml:",omitempty"`
+ StartAfter string `xml:",omitempty"`
+ Contents []xmlObject
+ CommonPrefixes []xmlCommonPrefix
+}
+
+// serveListV2 is ListObjectsV2, the subset litestream's replica client
+// (its restore lists with a delimiter) and lease.NewS3Mirror page through:
+// the current objects under prefix in key order (an object whose newest
+// version is a delete marker is not listed), a delimiter rolling every key
+// that holds it after the prefix into one common prefix, max-keys (at most
+// 1000, and at most PageSize when that is set lower; keys and common
+// prefixes both count) and continuation-token pagination. The token is
+// opaque to the client; here it is the last key or common prefix of the
+// previous page, and a page resumes after it (after every key under it,
+// for a common prefix). start-after is honoured on the first page.
+func (f *VersionedBucket) serveListV2(w http.ResponseWriter, bucket string, q url.Values) {
+ prefix, delim := q.Get("prefix"), q.Get("delimiter")
+ if err := f.fail("list", prefix); err != nil {
+ writeErr(w, err)
+ return
+ }
+ size := 1000
+ if f.PageSize > 0 && f.PageSize < size {
+ size = f.PageSize
+ }
+ if raw := q.Get("max-keys"); raw != "" {
+ n, err := strconv.Atoi(raw)
+ if err != nil || n < 0 {
+ s3err(w, http.StatusBadRequest, "InvalidArgument")
+ return
+ }
+ size = min(size, n)
+ }
+ after := q.Get("start-after")
+ token := q.Get("continuation-token")
+ if token != "" {
+ last, err := base64.StdEncoding.DecodeString(token)
+ if err != nil || len(last) == 0 {
+ s3errMsg(w, http.StatusBadRequest, "InvalidArgument", "The continuation token provided is incorrect")
+ return
+ }
+ after = string(last)
+ }
+ // entries are the page candidates in key order: a key, or a common
+ // prefix (isPrefix) standing for every key under it.
+ type entry struct {
+ name string
+ isPrefix bool
+ }
+ var keys []string
+ for k, vs := range f.buckets[bucket] {
+ if strings.HasPrefix(k, prefix) && len(vs) > 0 && !vs[0].deleteMarker {
+ keys = append(keys, k)
+ }
+ }
+ sort.Strings(keys)
+ var entries []entry
+ for _, k := range keys {
+ e := entry{name: k}
+ if delim != "" {
+ if i := strings.Index(k[len(prefix):], delim); i >= 0 {
+ e = entry{name: k[:len(prefix)+i+len(delim)], isPrefix: true}
+ }
+ }
+ if n := len(entries); n > 0 && entries[n-1] == e {
+ continue
+ }
+ // A common prefix that ended the last page is that page's token
+ // itself, and so is every key rolled into it.
+ if e.name <= after {
+ continue
+ }
+ entries = append(entries, e)
+ }
+ res := listBucketResult{Name: bucket, Prefix: prefix, Delimiter: delim, MaxKeys: size, ContinuationToken: token}
+ if token == "" {
+ res.StartAfter = q.Get("start-after")
+ }
+ page := entries
+ if len(page) > size {
+ page = page[:size]
+ res.IsTruncated = size > 0
+ }
+ for _, e := range page {
+ if e.isPrefix {
+ res.CommonPrefixes = append(res.CommonPrefixes, xmlCommonPrefix{Prefix: e.name})
+ continue
+ }
+ v := f.buckets[bucket][e.name][0]
+ res.Contents = append(res.Contents, xmlObject{Key: e.name, LastModified: v.modified.UTC().Format(time.RFC3339), ETag: `"` + v.etag + `"`, Size: int64(len(v.body)), StorageClass: "STANDARD"})
+ }
+ res.KeyCount = len(page)
+ if res.IsTruncated && len(page) > 0 {
+ res.NextContinuationToken = base64.StdEncoding.EncodeToString([]byte(page[len(page)-1].name))
+ }
+ w.Header().Set("Content-Type", "application/xml")
+ xml.NewEncoder(w).Encode(res)
+}
+
func (f *VersionedBucket) serveCopy(w http.ResponseWriter, r *http.Request, bucket, key string) {
src := strings.TrimPrefix(r.Header.Get("x-amz-copy-source"), "/")
srcPath, rawQ, _ := strings.Cut(src, "?")
diff --git a/internal/lease/leasetest/s3server_test.go b/internal/lease/leasetest/s3server_test.go
index 3b3581c2b..93e5efee8 100644
--- a/internal/lease/leasetest/s3server_test.go
+++ b/internal/lease/leasetest/s3server_test.go
@@ -4,12 +4,17 @@ import (
"context"
"crypto/sha256"
"encoding/hex"
+ "encoding/xml"
"errors"
+ "fmt"
"io"
"net/http"
"net/http/httptest"
+ "net/url"
"strings"
"testing"
+
+ "github.com/carlosframework/platform/internal/lease"
)
// failingReader yields some bytes and then an error, like a dropped upload.
@@ -287,3 +292,130 @@ func TestS3ServerConditionalDeleteUsesReadETag(t *testing.T) {
t.Fatal("plain delete changed", got.Code)
}
}
+
+// listV2 decodes one ListObjectsV2 page from the raw wire.
+func listV2(t *testing.T, h http.Handler, query string) (listBucketResult, int) {
+ t.Helper()
+ rec := do(t, h, http.MethodGet, "/data?list-type=2"+query, "", nil)
+ var res listBucketResult
+ if rec.Code == http.StatusOK {
+ if err := xml.Unmarshal(rec.Body.Bytes(), &res); err != nil {
+ t.Fatalf("decode %q: %v", rec.Body.String(), err)
+ }
+ }
+ return res, rec.Code
+}
+
+// ListObjectsV2 on the wire: the current objects under a prefix in key
+// order, each with its size and quoted ETag; a key whose newest version is
+// a delete marker, an older version and a sibling prefix are not listed;
+// max-keys pages, and the continuation token resumes after the page;
+// start-after skips; a bad token and a negative max-keys are 400s; a
+// delimiter rolls keys into common prefixes, which page like keys.
+func TestS3ServerListObjectsV2(t *testing.T) {
+ m := NewVersionedBucket()
+ m.AddBucket("data")
+ h, _ := NewS3Server(m)
+ for _, k := range []string{"p/c", "p/a", "p/b/x", "p/d", "q/a", "p2/a"} {
+ m.Put("data", k, []byte("body of "+k))
+ }
+ m.Put("data", "p/a", []byte("newer"))
+ m.BareDelete("data", "p/d")
+ res, code := listV2(t, h, "&prefix=p/")
+ if code != http.StatusOK {
+ t.Fatal(code)
+ }
+ var keys []string
+ for _, o := range res.Contents {
+ keys = append(keys, o.Key)
+ }
+ if strings.Join(keys, ",") != "p/a,p/b/x,p/c" || res.IsTruncated || res.KeyCount != 3 || res.NextContinuationToken != "" {
+ t.Fatalf("listing %+v", res)
+ }
+ if o := res.Contents[0]; o.Size != int64(len("newer")) || o.ETag != `"`+md5hex([]byte("newer"))+`"` {
+ t.Fatalf("p/a listed as %+v", o)
+ }
+
+ first, _ := listV2(t, h, "&prefix=p/&max-keys=2")
+ if len(first.Contents) != 2 || !first.IsTruncated || first.NextContinuationToken == "" || first.Contents[1].Key != "p/b/x" {
+ t.Fatalf("first page %+v", first)
+ }
+ second, _ := listV2(t, h, "&prefix=p/&max-keys=2&continuation-token="+url.QueryEscape(first.NextContinuationToken))
+ if len(second.Contents) != 1 || second.Contents[0].Key != "p/c" || second.IsTruncated || second.ContinuationToken != first.NextContinuationToken {
+ t.Fatalf("second page %+v", second)
+ }
+ after, _ := listV2(t, h, "&prefix=p/&start-after=p/a")
+ if len(after.Contents) != 2 || after.Contents[0].Key != "p/b/x" {
+ t.Fatalf("start-after %+v", after)
+ }
+ if _, code := listV2(t, h, "&continuation-token=%21%21"); code != http.StatusBadRequest {
+ t.Fatal("bad token", code)
+ }
+ if _, code := listV2(t, h, "&max-keys=-1"); code != http.StatusBadRequest {
+ t.Fatal("negative max-keys", code)
+ }
+ // A delimiter rolls p/b/x into the common prefix p/b/, which counts
+ // against max-keys and can end a page.
+ m.Put("data", "p/b/y", []byte("y"))
+ d1, _ := listV2(t, h, "&prefix=p/&delimiter=/&max-keys=2")
+ if len(d1.Contents) != 1 || d1.Contents[0].Key != "p/a" || len(d1.CommonPrefixes) != 1 || d1.CommonPrefixes[0].Prefix != "p/b/" ||
+ d1.KeyCount != 2 || !d1.IsTruncated || d1.Delimiter != "/" {
+ t.Fatalf("delimited first page %+v", d1)
+ }
+ d2, _ := listV2(t, h, "&prefix=p/&delimiter=/&max-keys=2&continuation-token="+url.QueryEscape(d1.NextContinuationToken))
+ if len(d2.Contents) != 1 || d2.Contents[0].Key != "p/c" || len(d2.CommonPrefixes) != 0 || d2.IsTruncated {
+ t.Fatalf("delimited second page %+v", d2)
+ }
+ top, _ := listV2(t, h, "&delimiter=/")
+ var tops []string
+ for _, p := range top.CommonPrefixes {
+ tops = append(tops, p.Prefix)
+ }
+ if len(top.Contents) != 0 || strings.Join(tops, ",") != "p/,p2/,q/" {
+ t.Fatalf("top level %+v", top)
+ }
+}
+
+// The SDK's own paginator, through lease.NewS3Mirror, reads every page:
+// 2,500 keys at the default 1,000 a page, and again at a PageSize of 7.
+func TestS3ServerListObjectsV2PaginatesTheSDK(t *testing.T) {
+ m := NewVersionedBucket()
+ m.AddBucket("data")
+ h, _ := NewS3Server(m)
+ srv := httptest.NewServer(h)
+ defer srv.Close()
+ t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
+ t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
+ t.Setenv("AWS_SESSION_TOKEN", "")
+ t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
+ t.Setenv("AWS_CONFIG_FILE", t.TempDir()+"/absent")
+ t.Setenv("AWS_SHARED_CREDENTIALS_FILE", t.TempDir()+"/absent")
+ t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
+ want := map[string]bool{}
+ for i := range 2500 {
+ k := fmt.Sprintf("big/%05d", i)
+ m.Put("data", k, []byte(k))
+ want[k] = true
+ }
+ m.Put("data", "bigger/x", []byte("x")) // a sibling prefix, never listed
+ ctx := context.Background()
+ mirror, err := lease.NewS3Mirror(ctx, "eu-west-1", "eu-west-1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, page := range []int{1000, 7} {
+ m.PageSize = page
+ got, err := mirror.List(ctx, "data", "big/")
+ if err != nil {
+ t.Fatalf("page size %d: %v", page, err)
+ }
+ if len(got) != len(want) {
+ t.Fatalf("page size %d: listed %d keys, want %d", page, len(got), len(want))
+ }
+ for k, o := range got {
+ if !want[k] || o.Size != int64(len(k)) || o.ETag != md5hex([]byte(k)) {
+ t.Fatalf("page size %d: %s listed as %+v", page, k, o)
+ }
+ }
+ }
+}
diff --git a/internal/moves/archive/archive.go b/internal/moves/archive/archive.go
index 1c3287872..c48336a52 100644
--- a/internal/moves/archive/archive.go
+++ b/internal/moves/archive/archive.go
@@ -143,7 +143,14 @@ type item struct {
// marker. The files directory is read through an os.Root, so nothing a
// tenant does inside it (a directory swapped for a symlink mid-walk) can
// make the walk or a later open leave it.
-func walk(root *os.Root, envPath string) ([]item, error) {
+//
+// skip, when non-nil, is asked each path relative to the files directory
+// (slash-separated, as the archive names it) before anything else is read
+// about it: a path it accepts is left out as the marker is, a directory
+// with everything under it, and is neither typed nor counted, so a
+// platform artifact of any kind (a socket, a FIFO) cannot fail the walk.
+// envPath is never asked.
+func walk(root *os.Root, envPath string, skip func(rel string) bool) ([]item, error) {
var items []item
add := func(it item) error {
if len(items) >= entryLimit {
@@ -160,9 +167,10 @@ func walk(root *os.Root, envPath string) ([]item, error) {
if p == "." {
return nil
}
- if p == MarkerName {
+ if p == MarkerName || (skip != nil && skip(p)) {
// The marker is never archived, nor anything under it if
- // it is (wrongly) a directory.
+ // it is (wrongly) a directory; nor is a path the caller
+ // skips.
if d.IsDir() {
return fs.SkipDir
}
@@ -323,6 +331,11 @@ func openFilesRoot(filesDir string) (*os.Root, error) {
// build writes the tar of filesDir and envPath to out and returns what
// Build and PlainDigest report.
func build(filesDir, envPath string, out io.Writer) (built, error) {
+ return buildFiltered(filesDir, envPath, nil, out)
+}
+
+// buildFiltered is build over the paths skip leaves in (walk).
+func buildFiltered(filesDir, envPath string, skip func(rel string) bool, out io.Writer) (built, error) {
root, err := openFilesRoot(filesDir)
if err != nil {
return built{}, err
@@ -330,7 +343,7 @@ func build(filesDir, envPath string, out io.Writer) (built, error) {
if root != nil {
defer root.Close()
}
- items, err := walk(root, envPath)
+ items, err := walk(root, envPath, skip)
if err != nil {
return built{}, err
}
@@ -410,11 +423,25 @@ func copyItem(tw io.Writer, it item, size int64) (Entry, error) {
// can never collide with the overlay. A missing envPath (or "") means the
// archive has no env entry; a missing filesDir is an empty one.
func Build(filesDir, envPath string, w io.Writer) (plain string, entries []Entry, bytes int64, err error) {
+ return BuildFiltered(filesDir, envPath, nil, w)
+}
+
+// BuildFiltered is Build over the files directory without the paths skip
+// accepts. skip is asked each path relative to filesDir, slash-separated
+// (a top-level entry is its bare name); a skipped directory is skipped
+// whole, and a skipped entry of any type is never read, typed or counted.
+// The env entry is never asked. A nil skip is Build. The offline move
+// archives a nested instance directory this way, leaving out the
+// platform's own files there (layout.NestedPlatformNames): the archive is
+// then the one Build would write over a flat files directory holding the
+// same app files, so it extracts flat and PlainDigest of the extraction
+// equals plain.
+func BuildFiltered(filesDir, envPath string, skip func(rel string) bool, w io.Writer) (plain string, entries []Entry, bytes int64, err error) {
zw, err := zstd.NewWriter(w, zstd.WithEncoderConcurrency(1), zstd.WithWindowSize(zstdWindow))
if err != nil {
return "", nil, 0, err
}
- b, err := build(filesDir, envPath, zw)
+ b, err := buildFiltered(filesDir, envPath, skip, zw)
if err != nil {
zw.Close()
return "", nil, 0, err
diff --git a/internal/moves/archive/archive_test.go b/internal/moves/archive/archive_test.go
index 9cf54e7d7..49ef8e1fd 100644
--- a/internal/moves/archive/archive_test.go
+++ b/internal/moves/archive/archive_test.go
@@ -104,6 +104,98 @@ func TestBuildIsDeterministicAndExcludesTheMarker(t *testing.T) {
}
}
+// BuildFiltered over a nested-shaped directory leaves out exactly what
+// skip accepts (a top-level file, a top-level directory with everything
+// under it, and a symlink, which unskipped would refuse the walk), never
+// asks about env, and writes byte for byte the archive Build writes over a
+// flat directory holding only the app's files; that archive extracts flat
+// and its extraction digests to the same plaintext. Build is BuildFiltered
+// with a nil skip.
+func TestBuildFilteredLeavesOutWhatSkipAccepts(t *testing.T) {
+ flat, env := filesFixture(t)
+ nested := filepath.Join(t.TempDir(), "nested")
+ if err := os.CopyFS(nested, os.DirFS(flat)); err != nil {
+ t.Fatal(err)
+ }
+ for _, rel := range []string{"a.txt", "sub", "sub/b.bin", "env", "sub/env", "sub/env/x", ".files-gen"} {
+ info, err := os.Lstat(filepath.Join(flat, rel))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chmod(filepath.Join(nested, rel), info.Mode().Perm()); err != nil {
+ t.Fatal(err)
+ }
+ }
+ mustWrite(t, filepath.Join(nested, "h.db"), []byte("database"), 0o600)
+ if err := os.MkdirAll(filepath.Join(nested, ".activator-h.db-litestream", "ltx", "0"), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ mustWrite(t, filepath.Join(nested, ".activator-h.db-litestream", "ltx", "0", "f.ltx"), []byte("ltx"), 0o600)
+ if err := os.Symlink("/etc/passwd", filepath.Join(nested, "h.config.env")); err != nil {
+ t.Fatal(err)
+ }
+ // A nested file that shares a skipped name below the top level is kept.
+ mustWrite(t, filepath.Join(nested, "sub", "h.db"), []byte("an app file"), 0o644)
+ mustWrite(t, filepath.Join(flat, "sub", "h.db"), []byte("an app file"), 0o644)
+ platform := map[string]bool{"h.db": true, ".activator-h.db-litestream": true, "h.config.env": true}
+ var asked []string
+ skip := func(rel string) bool {
+ asked = append(asked, rel)
+ return !strings.Contains(rel, "/") && platform[rel]
+ }
+ if _, _, _, err := Build(nested, env, io.Discard); !errors.Is(err, ErrSpecialFile) {
+ t.Fatalf("unfiltered, the symlink should refuse the walk: %v", err)
+ }
+ var got, want bytes.Buffer
+ plain, entries, n, err := BuildFiltered(nested, env, skip, &got)
+ if err != nil {
+ t.Fatal(err)
+ }
+ wplain, wentries, wn, err := Build(flat, env, &want)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if plain != wplain || n != wn || len(entries) != len(wentries) || !bytes.Equal(got.Bytes(), want.Bytes()) {
+ t.Fatalf("filtered %s %d %v, flat %s %d %v", plain, n, entries, wplain, wn, wentries)
+ }
+ for _, rel := range asked {
+ if strings.HasPrefix(rel, ".activator-h.db-litestream/") {
+ t.Fatalf("asked about %q inside a skipped directory", rel)
+ }
+ }
+ if !slicesContains(asked, "sub/h.db") || !slicesContains(asked, "h.db") {
+ t.Fatalf("asked %v", asked)
+ }
+ // Extract flat, then digest the extraction as Build would.
+ parent := t.TempDir()
+ files, outEnv := filepath.Join(parent, "h"), filepath.Join(parent, "h.env")
+ if err := Extract(bytes.NewReader(got.Bytes()), files, outEnv); err != nil {
+ t.Fatal(err)
+ }
+ if p, _, _, err := PlainDigest(files, outEnv); err != nil || p != plain {
+ t.Fatalf("extraction digests to %s (%v), want %s", p, err, plain)
+ }
+ for _, name := range []string{"h.db", ".activator-h.db-litestream", "h.config.env"} {
+ if _, err := os.Lstat(filepath.Join(files, name)); !errors.Is(err, fs.ErrNotExist) {
+ t.Fatalf("%s extracted: %v", name, err)
+ }
+ }
+ // A nil skip is Build.
+ var nilSkip bytes.Buffer
+ if p, _, _, err := BuildFiltered(flat, env, nil, &nilSkip); err != nil || p != wplain || !bytes.Equal(nilSkip.Bytes(), want.Bytes()) {
+ t.Fatalf("nil skip: %s %v", p, err)
+ }
+}
+
+func slicesContains(s []string, v string) bool {
+ for _, x := range s {
+ if x == v {
+ return true
+ }
+ }
+ return false
+}
+
func mustDecode(t *testing.T, b []byte) io.Reader {
t.Helper()
zr, err := zstd.NewReader(bytes.NewReader(b))
diff --git a/internal/moves/archive/archive_unix_test.go b/internal/moves/archive/archive_unix_test.go
index caab62dde..ef987b669 100644
--- a/internal/moves/archive/archive_unix_test.go
+++ b/internal/moves/archive/archive_unix_test.go
@@ -135,3 +135,24 @@ func TestBuildRefusalsCarryTheirPath(t *testing.T) {
})
}
}
+
+// A FIFO (a platform socket or pipe in a nested instance directory) that
+// skip accepts is never opened or typed: BuildFiltered archives around it.
+// The same FIFO unskipped is refused, naming it.
+func TestBuildFilteredNeverTypesASkippedFIFO(t *testing.T) {
+ files := t.TempDir()
+ if err := os.WriteFile(filepath.Join(files, "a"), []byte("a"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := syscall.Mkfifo(filepath.Join(files, "app.fifo"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ var pe *PathError
+ if _, _, _, err := BuildFiltered(files, "", nil, io.Discard); !errors.Is(err, ErrSpecialFile) || !errors.As(err, &pe) || pe.Path != "app.fifo" {
+ t.Fatalf("unskipped FIFO: %v", err)
+ }
+ _, entries, _, err := BuildFiltered(files, "", func(rel string) bool { return rel == "app.fifo" }, io.Discard)
+ if err != nil || len(entries) != 1 || entries[0].Path != "files/a" {
+ t.Fatalf("skipped FIFO: %v %v", entries, err)
+ }
+}