amadan

carlos / platform Public

moves: the seal by version; lease: streamed create-only put; provision: the cut's revert reasons (§7.3, §7.4, §8.3)

moves.Seal (N, replica/files/timers versions, src_covered, from_db,
sealed_at, lease_until), canonical Encode, strict DecodeSeal, Bound,
ReachesN. lease.FilePutIfAbsent streams the sealed archive into the
source's move-snapshots/ without holding it in memory. MoveRevert accepts
three more recorded reasons (MV-N163..165).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 0ecdeb53887867a79859476afa0317c62377ec10 parent 1686431
8 files changed, +365 −0
  • approved_moves_copy_test.go +3 −0
  • docs/superpowers/plans/2026-10-06-moves-phase1-copy.md +3 −0
  • internal/lease/versioned.go +38 −0
  • internal/lease/versioned_test.go +35 −0
  • internal/moves/seal.go +150 −0
  • internal/moves/seal_test.go +114 −0
  • internal/provision/move_test.go +8 −0
  • internal/provision/provision.go +14 −0
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 0e480faf4..7d56b56be 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -66,6 +66,7 @@ var movesCopyIDs = []string{
"MV-N145", "MV-N146", "MV-N147", "MV-N148", "MV-N149", "MV-N150",
"MV-N151", "MV-N152", "MV-N153", "MV-N154", "MV-N155", "MV-N156",
"MV-N157", "MV-N158", "MV-N159", "MV-N160", "MV-N161", "MV-N162",
+ "MV-N163", "MV-N164", "MV-N165",
}
// copyRow is one row of the record. The File cell is the file the sentence is
@@ -573,6 +574,7 @@ var movesScanFiles = []string{
"internal/lease/lockstate.go",
"internal/lease/mirror.go",
"internal/lease/staging.go",
+ "internal/lease/versioned.go",
"internal/moves/archive/archive.go",
"internal/moves/archive/extract.go",
"internal/moves/archive/limits.go",
@@ -593,6 +595,7 @@ var movesScanFiles = []string{
"internal/moves/lease.go",
"internal/moves/ltxbuild.go",
"internal/moves/readiness.go",
+ "internal/moves/seal.go",
"internal/moves/sealedclient.go",
"internal/moves/source.go",
"internal/moves/staging.go",
diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
index 58de9a907..fe62d556b 100644
--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
@@ -202,6 +202,9 @@ Columns: id, file, text, status.
| MV-N160 | internal/stopevidence/reconstruct_linux.go | pid <pid> in the agent's cgroup is not the agent | pending |
| MV-N161 | internal/stopevidence/reconstruct_linux.go | <count> process(es) hold the database open | pending |
| MV-N162 | internal/stopevidence/reconstruct_linux.go | lock.json still held | pending |
+| MV-N163 | internal/provision/provision.go | the cut ran past its deadline | pending |
+| MV-N164 | internal/provision/provision.go | the final copy kept finding changes at the source | pending |
+| MV-N165 | internal/provision/provision.go | the stopped files could not be archived | pending |
## Plan 2
diff --git a/internal/lease/versioned.go b/internal/lease/versioned.go
index c07bd037e..82dbb82fe 100644
--- a/internal/lease/versioned.go
+++ b/internal/lease/versioned.go
@@ -8,6 +8,7 @@ import (
"io"
"net/http"
"net/url"
+ "os"
"sort"
"strings"
"time"
@@ -290,6 +291,43 @@ func (s *versionedS3) PutIfAbsent(ctx context.Context, bucket, key string, body
return aws.ToString(out.VersionId), nil
}
+// FilePutIfAbsent is PutIfAbsent streaming a file from disk, for objects
+// too large to hold in memory: a move's sealed archive, written create-only
+// into the source's own move-snapshots/ (moves spec §8.3). The S3 store
+// implements it; an existing current object is ErrExists.
+type FilePutIfAbsent interface {
+ PutFileIfAbsent(ctx context.Context, bucket, key, path string) (versionID string, err error)
+}
+
+func (s *versionedS3) PutFileIfAbsent(ctx context.Context, bucket, key, path string) (string, error) {
+ c, err := s.client(bucket)
+ if err != nil {
+ return "", err
+ }
+ f, err := os.Open(path)
+ if err != nil {
+ return "", err
+ }
+ defer f.Close()
+ st, err := f.Stat()
+ if err != nil {
+ return "", err
+ }
+ // An *os.File is seekable, so the SDK hashes the payload by reading it
+ // and seeking back, never by buffering it.
+ out, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key),
+ Body: f, ContentLength: aws.Int64(st.Size()), IfNoneMatch: aws.String("*")})
+ if err != nil {
+ if code, status := apiCode(err); code == "PreconditionFailed" || status == http.StatusPreconditionFailed {
+ return "", ErrExists
+ }
+ return "", fmt.Errorf("lease: put-if-absent %s/%s: %w", bucket, key, err)
+ }
+ return aws.ToString(out.VersionId), nil
+}
+
var _ VersionedStore = (*versionedS3)(nil)
+var _ FilePutIfAbsent = (*versionedS3)(nil)
+
var _ VersionRangeOpener = (*versionedS3)(nil)
diff --git a/internal/lease/versioned_test.go b/internal/lease/versioned_test.go
index 2d7ad8f1f..b63da53f2 100644
--- a/internal/lease/versioned_test.go
+++ b/internal/lease/versioned_test.go
@@ -2,9 +2,12 @@ package lease_test
import (
"context"
+ "errors"
"io"
"net/http"
"net/http/httptest"
+ "os"
+ "path/filepath"
"testing"
"github.com/aws/aws-sdk-go-v2/aws"
@@ -146,3 +149,35 @@ func TestVersionedS3RangeWire(t *testing.T) {
})
}
}
+
+// PutFileIfAbsent streams a file from disk create-only: the object lands
+// with the file's bytes and a version id, and a second put is ErrExists
+// with the first object untouched.
+func TestVersionedS3PutFileIfAbsent(t *testing.T) {
+ ctx := context.Background()
+ st, m, _ := wireStore(t)
+ fp, ok := st.(lease.FilePutIfAbsent)
+ if !ok {
+ t.Fatal("the S3 store cannot put a file create-only")
+ }
+ path := filepath.Join(t.TempDir(), "sealed")
+ if err := os.WriteFile(path, []byte("sealed bytes"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ v, err := fp.PutFileIfAbsent(ctx, "data", "move-snapshots/h/op/files-ab.sealed", path)
+ if err != nil || v == "" {
+ t.Fatalf("first put = %q, %v", v, err)
+ }
+ if cur, ok := m.Current("data", "move-snapshots/h/op/files-ab.sealed"); !ok || string(cur) != "sealed bytes" {
+ t.Fatalf("stored %q", cur)
+ }
+ if err := os.WriteFile(path, []byte("other bytes"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fp.PutFileIfAbsent(ctx, "data", "move-snapshots/h/op/files-ab.sealed", path); !errors.Is(err, lease.ErrExists) {
+ t.Fatalf("second put = %v, want ErrExists", err)
+ }
+ if cur, _ := m.Current("data", "move-snapshots/h/op/files-ab.sealed"); string(cur) != "sealed bytes" {
+ t.Fatalf("the second put overwrote: %q", cur)
+ }
+}
diff --git a/internal/moves/seal.go b/internal/moves/seal.go
new file mode 100644
index 000000000..6e4d1abc4
--- /dev/null
+++ b/internal/moves/seal.go
@@ -0,0 +1,150 @@
+package moves
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "sort"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/superfly/ltx"
+)
+
+// The seal (§7.3 "seal by version", §7.4 staged): the exact versions of the
+// staging lane's replica/, files/ and timers/ objects the cut wrote, the
+// required final TXID N, and the two facts a recovery reads, in the claim's
+// op journal under hostclaim.JournalSeal. Landing copies these versions and
+// nothing else, so a stale round's later write (a newer version, or a
+// delete marker) is never installed.
+//
+// It is written twice under one op: at the end of `stopped` with the replica
+// and the timers (the archive does not exist yet), and at the end of
+// `staged` with the archive's two objects added. Files is empty until then,
+// and the cut never flips on a seal without them (HasFiles).
+type Seal struct {
+ V int `json:"v"`
+ N uint64 `json:"n"`
+ Replica []SealedObject `json:"replica"`
+ Files []SealedObject `json:"files,omitempty"`
+ Timers []SealedObject `json:"timers,omitempty"`
+ // SrcCovered is whether the source's own replica of holder.generation
+ // reached the stopped database's position with no pending write (§7.4,
+ // rev 7.11): it decides a rollback's clean return (§7.8).
+ SrcCovered bool `json:"src_covered"`
+ // FromDB is whether the seal was built from the stopped database itself
+ // (§5.1 step 9): false for a source that had no local database.
+ FromDB bool `json:"from_db"`
+ SealedAt time.Time `json:"sealed_at"`
+ // LeaseUntil is the latest moment a request signed by a staging writer
+ // of this lane can still land: the last guarded signature's time plus
+ // SignatureLife plus the skew (ruling G7). Bound returns it.
+ LeaseUntil time.Time `json:"lease_until"`
+}
+
+// SealVersion is the only seal format this code reads or writes.
+const SealVersion = 1
+
+// ErrBadSeal is a seal that pins nothing usable: refused on write and on
+// read, never repaired.
+var ErrBadSeal = errors.New("moves: the seal is not usable")
+
+// Bound is the time after which no staging writer that held the lease can
+// still write: the staging park's evidence_at (Plan 5).
+func (s Seal) Bound() time.Time { return s.LeaseUntil }
+
+// HasFiles reports whether the seal names the archive: `archived` and the
+// flip require it.
+func (s Seal) HasFiles() bool { return len(s.Files) > 0 }
+
+// ReachesN reports whether the sealed replica ends exactly at N: its newest
+// level file's max TXID is N. A set that stops short cannot restore to N,
+// and one that runs past it is not this stopped database's.
+func (s Seal) ReachesN() bool {
+ var newest ltx.TXID
+ for _, o := range s.Replica {
+ pos, err := sealedPosition(o.Key)
+ if err != nil {
+ return false
+ }
+ newest = max(newest, pos.max)
+ }
+ return s.N > 0 && uint64(newest) == s.N
+}
+
+// Validate refuses a seal that pins nothing: another version, no position,
+// no replica, any object without a key or a usable version id (G27: "" and
+// "null" pin nothing), a key named twice, a negative size, a replica object
+// that is not a level file or is below an LTX header (Task 8's
+// NewSealedClient refuses those), no seal time, and no bound or a bound
+// before the seal.
+func (s Seal) Validate() error {
+ if s.V != SealVersion {
+ return fmt.Errorf("%w: version %d", ErrBadSeal, s.V)
+ }
+ if s.N == 0 || len(s.Replica) == 0 {
+ return fmt.Errorf("%w: no position or no replica", ErrBadSeal)
+ }
+ if s.SealedAt.IsZero() || s.LeaseUntil.IsZero() || s.LeaseUntil.Before(s.SealedAt) {
+ return fmt.Errorf("%w: no seal time or no writers' bound", ErrBadSeal)
+ }
+ seen := map[string]bool{}
+ for _, class := range [][]SealedObject{s.Replica, s.Files, s.Timers} {
+ for _, o := range class {
+ if o.Key == "" || !lease.UsableVersionID(o.VersionID) || o.Size < 0 || seen[o.Key] {
+ return fmt.Errorf("%w: object %q version %q size %d", ErrBadSeal, o.Key, o.VersionID, o.Size)
+ }
+ seen[o.Key] = true
+ }
+ }
+ for _, o := range s.Replica {
+ if _, err := sealedPosition(o.Key); err != nil {
+ return fmt.Errorf("%w: %v", ErrBadSeal, err)
+ }
+ if o.Size < ltx.HeaderSize {
+ return fmt.Errorf("%w: replica object %s is below an LTX header", ErrBadSeal, o.Key)
+ }
+ }
+ return nil
+}
+
+func sortObjects(objs []SealedObject) []SealedObject {
+ out := append([]SealedObject(nil), objs...)
+ sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
+ return out
+}
+
+// Encode is the seal as the claim journal carries it: JSON with every
+// object list sorted by key, so one seal always encodes to one string.
+func (s Seal) Encode() string {
+ c := s
+ c.Replica, c.Files, c.Timers = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers)
+ c.SealedAt, c.LeaseUntil = s.SealedAt.UTC(), s.LeaseUntil.UTC()
+ b, err := json.Marshal(c)
+ if err != nil {
+ // Every field is a plain value: Marshal cannot fail.
+ panic(err)
+ }
+ return string(b)
+}
+
+// DecodeSeal reads a journalled seal strictly (no unknown field, no
+// trailing data) and validates it. Anything else is ErrBadSeal.
+func DecodeSeal(raw string) (Seal, error) {
+ dec := json.NewDecoder(bytes.NewReader([]byte(raw)))
+ dec.DisallowUnknownFields()
+ var s Seal
+ if err := dec.Decode(&s); err != nil {
+ return Seal{}, fmt.Errorf("%w: %v", ErrBadSeal, err)
+ }
+ if _, err := dec.Token(); err != io.EOF {
+ return Seal{}, fmt.Errorf("%w: trailing data", ErrBadSeal)
+ }
+ if err := s.Validate(); err != nil {
+ return Seal{}, err
+ }
+ s.Replica, s.Files, s.Timers = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers)
+ return s, nil
+}
diff --git a/internal/moves/seal_test.go b/internal/moves/seal_test.go
new file mode 100644
index 000000000..459cd1e6e
--- /dev/null
+++ b/internal/moves/seal_test.go
@@ -0,0 +1,114 @@
+package moves
+
+import (
+ "errors"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/superfly/ltx"
+)
+
+func sealForTest() Seal {
+ at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)
+ return Seal{V: 1, N: 7,
+ Replica: []SealedObject{
+ {Key: "moves/h/mv/replica/0001/0000000000000001-0000000000000007.ltx", VersionID: "v2", Size: ltx.HeaderSize + 10},
+ {Key: "moves/h/mv/replica/0000/0000000000000007-0000000000000007.ltx", VersionID: "v1", Size: ltx.HeaderSize},
+ },
+ Timers: []SealedObject{{Key: "moves/h/mv/timers/timers.json", VersionID: "v3", Size: 40}},
+ Files: []SealedObject{{Key: "moves/h/mv/files/files-ab.sealed", VersionID: "v4", Size: 100}, {Key: "moves/h/mv/files/files-ab.json", VersionID: "v5", Size: 50}},
+ FromDB: true,
+ SealedAt: at, LeaseUntil: at.Add(16 * time.Minute)}
+}
+
+// The seal round-trips through the claim journal's string, in one canonical
+// form: objects sorted by key, so two encodings of one seal are equal.
+func TestSealEncodeDecodeRoundTrip(t *testing.T) {
+ s := sealForTest()
+ enc := s.Encode()
+ got, err := DecodeSeal(enc)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.N != 7 || len(got.Replica) != 2 || len(got.Files) != 2 || len(got.Timers) != 1 || !got.FromDB || got.SrcCovered {
+ t.Fatalf("decoded %+v", got)
+ }
+ if got.Replica[0].Key > got.Replica[1].Key {
+ t.Fatalf("replica not sorted: %+v", got.Replica)
+ }
+ if again := got.Encode(); again != enc {
+ t.Fatalf("not canonical:\n%s\n%s", enc, again)
+ }
+ if !got.Bound().Equal(s.LeaseUntil) {
+ t.Fatalf("Bound %v, want LeaseUntil %v", got.Bound(), s.LeaseUntil)
+ }
+ if !got.HasFiles() {
+ t.Fatal("HasFiles false on a seal that names the archive")
+ }
+ got.Files = nil
+ if got.HasFiles() {
+ t.Fatal("HasFiles true with no files")
+ }
+}
+
+// A seal that pins nothing is refused, on encode-side validation and on
+// decode: a missing or "null" version, an empty key, a key twice, a replica
+// object below an LTX header, no position, no bound, another version.
+func TestSealRefusesWhatPinsNothing(t *testing.T) {
+ for name, edit := range map[string]func(*Seal){
+ "version 2": func(s *Seal) { s.V = 2 },
+ "no position": func(s *Seal) { s.N = 0 },
+ "no replica": func(s *Seal) { s.Replica = nil },
+ "null version": func(s *Seal) { s.Replica[0].VersionID = "null" },
+ "empty version": func(s *Seal) { s.Timers[0].VersionID = "" },
+ "empty key": func(s *Seal) { s.Files[0].Key = "" },
+ "duplicate key": func(s *Seal) { s.Files[1].Key = s.Files[0].Key },
+ "short replica": func(s *Seal) { s.Replica[1].Size = ltx.HeaderSize - 1 },
+ "negative size": func(s *Seal) { s.Files[0].Size = -1 },
+ "no bound": func(s *Seal) { s.LeaseUntil = time.Time{} },
+ "not sealed": func(s *Seal) { s.SealedAt = time.Time{} },
+ "replica not an ltx": func(s *Seal) { s.Replica[0].Key = "moves/h/mv/replica/lock.json" },
+ "bound before sealed": func(s *Seal) { s.LeaseUntil = s.SealedAt.Add(-time.Second) },
+ } {
+ t.Run(name, func(t *testing.T) {
+ s := sealForTest()
+ edit(&s)
+ if err := s.Validate(); !errors.Is(err, ErrBadSeal) {
+ t.Fatalf("Validate = %v, want ErrBadSeal", err)
+ }
+ if _, err := DecodeSeal(s.Encode()); !errors.Is(err, ErrBadSeal) {
+ t.Fatalf("DecodeSeal = %v, want ErrBadSeal", err)
+ }
+ })
+ }
+ for name, raw := range map[string]string{
+ "empty": "",
+ "not json": "{",
+ "unknown field": strings.Replace(sealForTest().Encode(), `"n":7`, `"n":7,"extra":1`, 1),
+ "trailing": sealForTest().Encode() + "{}",
+ } {
+ t.Run(name, func(t *testing.T) {
+ if _, err := DecodeSeal(raw); !errors.Is(err, ErrBadSeal) {
+ t.Fatalf("DecodeSeal(%q) = %v, want ErrBadSeal", raw, err)
+ }
+ })
+ }
+}
+
+// The replica reaches N: some sealed level file ends at N, and none ends
+// after it.
+func TestSealReachesN(t *testing.T) {
+ s := sealForTest()
+ if !s.ReachesN() {
+ t.Fatal("a seal whose newest file ends at N does not reach N")
+ }
+ s.N = 8
+ if s.ReachesN() {
+ t.Fatal("a seal ending at 7 reaches N = 8")
+ }
+ s.N = 6
+ if s.ReachesN() {
+ t.Fatal("a seal holding a file past N reaches N = 6")
+ }
+}
diff --git a/internal/provision/move_test.go b/internal/provision/move_test.go
index 9610093ce..22afd4eda 100644
--- a/internal/provision/move_test.go
+++ b/internal/provision/move_test.go
@@ -406,6 +406,14 @@ func TestBackToReadyAndRevert(t *testing.T) {
if err := MoveRevert(g, "some error text")(rec4); err == nil || errors.Is(err, ErrMoveConflict) || rec4.Move.State != MoveCutting || rec4.Move.Reason != "" {
t.Fatalf("an unrecorded reason: %v %+v", err, rec4.Move)
}
+ // Every recorded reason is accepted and written as itself (Task 5's
+ // reverts: the deadline, the deferred final rounds, the archive).
+ for _, reason := range []string{RevertStopEvidenceBudget, RevertCutDeadline, RevertDeferredRounds, RevertArchiveRefused} {
+ rec5 := &Record{Region: "ie", Move: &MoveState{ID: g.ID, Incarnation: g.Incarnation, State: MoveCutting, CutPhase: CutStaged}}
+ if err := MoveRevert(g, reason)(rec5); err != nil || rec5.Move.State != MoveReverting || rec5.Move.Reason != reason {
+ t.Fatalf("revert for %q: %v %+v", reason, err, rec5.Move)
+ }
+ }
// Revert is a cut's recovery only.
rec3 := &Record{Region: "uk", Move: &MoveState{ID: g.ID, Incarnation: g.Incarnation, State: MoveLanding, CutPhase: CutFlipped}}
if err := MoveRevert(g, "x")(rec3); !errors.Is(err, ErrMoveConflict) {
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index e3283abe2..f7f72e0c7 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -1459,6 +1459,14 @@ func MoveLandingPhase(g MoveGuard, from, to string, stamp func(*MoveState)) func
// reason adds a constant, a case below and a row in the copy record.
const (
RevertStopEvidenceBudget = "stop evidence budget"
+ // RevertCutDeadline: a phase after stopping found cut_deadline passed.
+ RevertCutDeadline = "the cut ran past its deadline"
+ // RevertDeferredRounds: three final mirror rounds found the source
+ // changing under them, or moved more than the priced bar.
+ RevertDeferredRounds = "the final copy kept finding changes at the source"
+ // RevertArchiveRefused: the stopped files directory cannot be archived
+ // (a special file, over the cap, too many entries, a path too long).
+ RevertArchiveRefused = "the stopped files could not be archived"
)
// MoveRevert starts the pre-flip recovery (§7.4 reverting): from a cut
@@ -1481,6 +1489,12 @@ func MoveRevert(g MoveGuard, reason string) func(*Record) error {
switch reason {
case RevertStopEvidenceBudget:
rec.Move.Reason = RevertStopEvidenceBudget
+ case RevertCutDeadline:
+ rec.Move.Reason = RevertCutDeadline
+ case RevertDeferredRounds:
+ rec.Move.Reason = RevertDeferredRounds
+ case RevertArchiveRefused:
+ rec.Move.Reason = RevertArchiveRefused
default:
return fmt.Errorf("provision: revert reason %q has no recorded sentence", reason)
}