amadan

carlos / platform Public

moves: per-operation staging keys; lease: copies record their destination version ids (§7.3)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude f504d92626a09b2fba7841c220c91e059e0f4b92 parent 2cfd51f
4 files changed, +306 −8
  • internal/lease/mirror.go +63 −7
  • internal/lease/mirror_test.go +142 −1
  • internal/moves/staging.go +49 −0
  • internal/moves/staging_test.go +52 −0
diff --git a/internal/lease/mirror.go b/internal/lease/mirror.go
index cb6ed72b0..9e0ab9e11 100644
--- a/internal/lease/mirror.go
+++ b/internal/lease/mirror.go
@@ -44,6 +44,25 @@ type MirrorStore interface {
Put(ctx context.Context, bucket, key string, body []byte) error
}
+// VersionedCopier is the optional half of MirrorStore a versioned bucket
+// supports: the copy's response carries the destination's version id,
+// which a move's seal records (moves spec §7.3, "seal by version"). The
+// source never lists the target's staging, so the response is the only
+// place the id can come from. Only the staging mirror (Task 6,
+// lease.NewStagingMirror, through Task 14's guardedMirror) implements it;
+// the ordinary s3Mirror and every plain MirrorStore record no versions and
+// MirrorRound.Versions stays empty, so production mirroring (NewS3Mirror,
+// the legacy worker) is unchanged (round-4 finding 7).
+type VersionedCopier interface {
+ CopyVersioned(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (etag, versionID string, err error)
+}
+
+// VersionedPutter is the same for Put: the archive, the timers snapshot
+// and move.json are sealed by the version their PUT returned.
+type VersionedPutter interface {
+ PutVersioned(ctx context.Context, bucket, key string, body []byte) (versionID string, err error)
+}
+
// ErrCopySourceChanged is a conditional Copy whose source was overwritten
// after the round listed it. The round defers that object to the next
// round rather than failing: a source that keeps changing is a busy
@@ -74,6 +93,12 @@ type MirrorRound struct {
// Deferred > 0.
Deferred int
BytesDeferred int64
+ // Versions maps a relative key to the destination version id of the
+ // copy that produced its current destination object, for every current
+ // source key whose version is known (a copy this round, or a receipt
+ // that still vouches for the destination). Non-nil only when the store
+ // is a VersionedCopier; a plain MirrorStore records none.
+ Versions map[string]string
}
// DeltaBytes is how much the round found to move: copied plus deferred.
@@ -118,12 +143,17 @@ type receiptBook struct {
}
// copyReceipt is one completed copy whose destination ETag differs from
-// its source's. Only such copies get one: where the ETags already match,
-// the plain equality test needs no help.
+// its source's, or (on a versioned store) one whose version id is known.
+// Otherwise, where the ETags already match, the plain equality test needs
+// no help.
type copyReceipt struct {
SrcETag string `json:"src_etag"`
SrcSize int64 `json:"src_size"`
DstETag string `json:"dst_etag"`
+ // DstVersionID is the destination version the copy returned. Written
+ // only by a VersionedCopier; a versioned round skips an object only
+ // when a receipt carrying one still vouches for it.
+ DstVersionID string `json:"dst_version_id,omitempty"`
}
// MirrorPrefix makes dstBucket/dstPrefix an exact mirror of
@@ -170,6 +200,12 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB
}
sort.Strings(rel)
var round MirrorRound
+ // Version collection is decided once, by the store: a VersionedCopier
+ // is in receipt-required mode, any other store keeps the plain rules.
+ vc, versioned := m.(VersionedCopier)
+ if versioned {
+ round.Versions = map[string]string{}
+ }
for _, r := range rel {
// The lease's lock.json is COORDINATION state, not data (live
// rehearsal catch, 2026-08-29): the renewer rewrites it on every
@@ -183,7 +219,7 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB
sk, dk := srcPrefix+r, dstPrefix+r
s := src[sk]
d, staged := dst[dk]
- if staged && d.ETag == s.ETag {
+ if staged && d.ETag == s.ETag && !versioned {
continue
}
if staged {
@@ -191,11 +227,24 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB
if err != nil {
return round, err
}
- if ok && rcpt.SrcETag == s.ETag && rcpt.SrcSize == s.Size && rcpt.DstETag == d.ETag && d.Size == s.Size {
+ // A versioned round must be able to name the version it would
+ // seal, so a matching receipt has to carry one; without it the
+ // object is recopied (a fresh version id; the stale version
+ // stays unreferenced).
+ if ok && rcpt.SrcETag == s.ETag && rcpt.SrcSize == s.Size && rcpt.DstETag == d.ETag && d.Size == s.Size && (!versioned || rcpt.DstVersionID != "") {
+ if versioned {
+ round.Versions[r] = rcpt.DstVersionID
+ }
continue
}
}
- etag, err := m.Copy(ctx, srcBucket, sk, s.ETag, dstBucket, dk)
+ var etag, ver string
+ var err error
+ if versioned {
+ etag, ver, err = vc.CopyVersioned(ctx, srcBucket, sk, s.ETag, dstBucket, dk)
+ } else {
+ etag, err = m.Copy(ctx, srcBucket, sk, s.ETag, dstBucket, dk)
+ }
if errors.Is(err, ErrCopySourceChanged) {
book.drop(r)
round.Changed++
@@ -209,10 +258,17 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB
}
round.Changed++
round.BytesCopied += s.Size
- if etag != "" && etag != s.ETag {
+ if ver != "" {
+ round.Versions[r] = ver
+ }
+ if ver != "" || (etag != "" && etag != s.ETag) {
+ dstETag := etag
+ if dstETag == "" {
+ dstETag = s.ETag
+ }
// A failed read here loses this receipt (one recopy next
// round), never a wrong skip.
- if err := book.set(ctx, r, copyReceipt{SrcETag: s.ETag, SrcSize: s.Size, DstETag: etag}); err != nil {
+ if err := book.set(ctx, r, copyReceipt{SrcETag: s.ETag, SrcSize: s.Size, DstETag: dstETag, DstVersionID: ver}); err != nil {
return round, err
}
} else {
diff --git a/internal/lease/mirror_test.go b/internal/lease/mirror_test.go
index 35cff26b0..3abf06856 100644
--- a/internal/lease/mirror_test.go
+++ b/internal/lease/mirror_test.go
@@ -287,7 +287,9 @@ func TestMirrorPrefixReceiptsSkipAnUnchangedMultipartObject(t *testing.T) {
}
puts := f.puts
for i := 0; i < 3; i++ {
- if r := mirrorRound(t, f, testReceipts); r != (MirrorRound{}) {
+ // MirrorRound gained a map (Versions), so it is no longer comparable
+ // with ==; this is the same "nothing happened" assertion, field by field.
+ if r := mirrorRound(t, f, testReceipts); r.Changed != 0 || r.BytesCopied != 0 || r.Deferred != 0 || r.BytesDeferred != 0 || r.Versions != nil {
t.Fatalf("unchanged round %d = %+v, want nothing copied", i, r)
}
}
@@ -480,3 +482,142 @@ func TestMirrorPrefixReceiptReadFailureFailsTheRound(t *testing.T) {
t.Fatalf("receipts after a failed read = %+v, want them untouched (%+v)", got, before)
}
}
+
+// fakeVersionedMirror is fakeMirror plus version ids on copies and puts,
+// the S3 behaviour MirrorPrefix records for the seal (moves spec §7.3).
+type fakeVersionedMirror struct {
+ *fakeMirror
+ n int
+}
+
+func (f *fakeVersionedMirror) CopyVersioned(ctx context.Context, sb, sk, se, db, dk string) (string, string, error) {
+ etag, err := f.fakeMirror.Copy(ctx, sb, sk, se, db, dk)
+ if err != nil {
+ return "", "", err
+ }
+ f.n++
+ return etag, fmt.Sprintf("v%d", f.n), nil
+}
+
+func (f *fakeVersionedMirror) PutVersioned(ctx context.Context, b, k string, body []byte) (string, error) {
+ if err := f.fakeMirror.Put(ctx, b, k, body); err != nil {
+ return "", err
+ }
+ f.n++
+ return fmt.Sprintf("v%d", f.n), nil
+}
+
+func TestMirrorPrefixRecordsDestinationVersions(t *testing.T) {
+ f := &fakeVersionedMirror{fakeMirror: newFakeMirror()}
+ f.put("src", "instances/h/0000/a.ltx", "e1")
+ f.put("src", "instances/h/0009/s.ltx", "e2")
+ rc := Receipts{Bucket: "src", Key: "move-receipts/h/mv.json"}
+ r1, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if r1.Versions["0000/a.ltx"] == "" || r1.Versions["0009/s.ltx"] == "" {
+ t.Fatalf("versions after round 1: %v", r1.Versions)
+ }
+ // Round 2 copies nothing; the versions come from the receipts.
+ r2, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if r2.BytesCopied != 0 || r2.Versions["0000/a.ltx"] != r1.Versions["0000/a.ltx"] || len(r2.Versions) != 2 {
+ t.Fatalf("round 2: %+v", r2)
+ }
+ // A crash between the copy and the receipt write: the next round sees
+ // the destination's ETag equal to the source's but no receipt (or a
+ // receipt for an older version of the key). It recopies rather than
+ // sealing a version it cannot vouch for.
+ f.putSized("src", "instances/h/0000/a.ltx", "e1-new", 64)
+ f.n++ // a copy whose receipt was never persisted
+ f.store("dst", "moves/h/mv/replica/0000/a.ltx", fakeObject{MirrorObject{ETag: "e1-new", Size: 64}, "e1-new"})
+ r4, err := MirrorPrefix(context.Background(), f, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
+ if err != nil || r4.Versions["0000/a.ltx"] == r1.Versions["0000/a.ltx"] || r4.Changed == 0 || r4.BytesCopied != 64 {
+ t.Fatalf("stale receipt sealed: %+v %v", r4, err)
+ }
+ // The plain store's second identical round copies nothing (the
+ // existing TestMirrorPrefix contract, kept): see the next block.
+ // A plain MirrorStore records no versions and keeps today's rule: an
+ // equal ETag skips the copy with no receipt needed.
+ g := newFakeMirror()
+ g.putSized("src", "instances/h/0000/a.ltx", "e1", 64)
+ r3, err := MirrorPrefix(context.Background(), g, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
+ if err != nil || r3.Versions != nil || r3.BytesCopied != 64 {
+ t.Fatalf("plain store round 1: %+v %v", r3, err)
+ }
+ r3b, err := MirrorPrefix(context.Background(), g, "src", "instances/h/", "dst", "moves/h/mv/replica/", rc)
+ if err != nil || r3b.Changed != 0 || r3b.BytesCopied != 0 {
+ t.Fatalf("plain store recopied an unchanged object: %+v %v", r3b, err)
+ }
+}
+
+// A multipart source's copy lands with a destination ETag that differs from
+// the source's, so the receipt-match branch (not the equal-ETag one) must
+// also vouch for the version.
+func TestVersionedMirrorRecordsMultipartVersions(t *testing.T) {
+ ctx := context.Background()
+ f := &fakeVersionedMirror{fakeMirror: newFakeMirror()}
+ f.putMultipart("src", "instances/h/ltx/1/a.ltx", "v1", 100<<20)
+ f.putMultipart("src", "instances/h/ltx/1/b.ltx", "w1", 50<<20)
+ rc := testReceipts
+ round := func() MirrorRound {
+ t.Helper()
+ r, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/", rc)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return r
+ }
+ r1 := round()
+ if r1.Changed != 2 || r1.Versions["ltx/1/a.ltx"] == "" || r1.Versions["ltx/1/b.ltx"] == "" || len(r1.Versions) != 2 {
+ t.Fatalf("round 1: %+v", r1)
+ }
+ if d := f.buckets["dst"]["moves/h/ltx/1/a.ltx"]; d.ETag == "v1-27" {
+ t.Fatalf("fake did not model the multipart ETag change: %v", d.ETag)
+ }
+ r2 := round()
+ if r2.Changed != 0 || r2.BytesCopied != 0 || len(r2.Versions) != 2 ||
+ r2.Versions["ltx/1/a.ltx"] != r1.Versions["ltx/1/a.ltx"] || r2.Versions["ltx/1/b.ltx"] != r1.Versions["ltx/1/b.ltx"] {
+ t.Fatalf("round 2: %+v (round 1 versions %v)", r2, r1.Versions)
+ }
+ // A receipt hand-written without dst_version_id (the plain path, or
+ // before this change) is recopied exactly once, then carries one.
+ book := readReceipts(t, f.fakeMirror)
+ rcpt := book["ltx/1/a.ltx"]
+ if rcpt.DstVersionID == "" {
+ t.Fatalf("receipt carries no version: %+v", rcpt)
+ }
+ rcpt.DstVersionID = ""
+ book["ltx/1/a.ltx"] = rcpt
+ body, err := json.Marshal(receiptBook{V: 1, SrcBucket: "src", SrcPrefix: "instances/h/", DstBucket: "dst", DstPrefix: "moves/h/", Objects: book})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := f.fakeMirror.Put(ctx, rc.Bucket, rc.Key, body); err != nil {
+ t.Fatal(err)
+ }
+ r3 := round()
+ if r3.Changed != 1 || r3.BytesCopied != 100<<20 || r3.Versions["ltx/1/a.ltx"] == "" || r3.Versions["ltx/1/a.ltx"] == r1.Versions["ltx/1/a.ltx"] {
+ t.Fatalf("round 3 (versionless receipt): %+v", r3)
+ }
+ r4 := round()
+ if r4.Changed != 0 || r4.Versions["ltx/1/a.ltx"] != r3.Versions["ltx/1/a.ltx"] || len(r4.Versions) != 2 {
+ t.Fatalf("round 4: %+v", r4)
+ }
+}
+
+// Version collection is opt-in: the production s3Mirror must never enter
+// the receipt-required mode, since an unversioned destination can never
+// supply a version receipt and every unchanged object would recopy forever
+// (round-4 finding 7).
+func TestPlainS3MirrorRecordsNoVersions(t *testing.T) {
+ if _, ok := interface{}(&s3Mirror{}).(VersionedCopier); ok {
+ t.Fatal("s3Mirror implements VersionedCopier")
+ }
+ if _, ok := interface{}(&s3Mirror{}).(VersionedPutter); ok {
+ t.Fatal("s3Mirror implements VersionedPutter")
+ }
+}
diff --git a/internal/moves/staging.go b/internal/moves/staging.go
new file mode 100644
index 000000000..d2258b254
--- /dev/null
+++ b/internal/moves/staging.go
@@ -0,0 +1,49 @@
+package moves
+
+import "fmt"
+
+// The staging layout (§7.3): one prefix per operation in the target
+// bucket, one subprefix per artifact class, so MirrorPrefix on replica/
+// can never delete an archive, a snapshot or move.json.
+func StagingPrefix(host, opID string) string { return "moves/" + host + "/" + opID + "/" }
+func ReplicaPrefix(host, opID string) string { return StagingPrefix(host, opID) + "replica/" }
+func FilesPrefix(host, opID string) string { return StagingPrefix(host, opID) + "files/" }
+func TimersPrefix(host, opID string) string { return StagingPrefix(host, opID) + "timers/" }
+func MoveJSONKey(host, opID string) string { return StagingPrefix(host, opID) + "move.json" }
+
+// ScopeLocation is the lane as a park scope location: no trailing slash
+// (claim rev 6.23 §2 grammar, INV-23).
+func ScopeLocation(host, opID string) string { return "moves/" + host + "/" + opID }
+
+// MoveJSON is written create-only before the first round (§7.3). The
+// landing requires its incarnation.
+type MoveJSON struct {
+ MoveID string `json:"move_id"`
+ Incarnation string `json:"incarnation"`
+ Host string `json:"host"`
+ From string `json:"from"`
+ To string `json:"to"`
+}
+
+// LaneKey is T10's probe object in the source bucket (§7.3).
+func LaneKey(opID string) string { return "moves/_lane/" + opID }
+
+// Source-bucket keys: receipts and sealed snapshots, per operation (§6.3).
+func ReceiptsKey(host, opID string) string { return "move-receipts/" + host + "/" + opID + ".json" }
+func SnapshotsPrefix(host, opID string) string { return "move-snapshots/" + host + "/" + opID + "/" }
+
+// LedgerKey is the per-operation ledger (the move_json_version and archive
+// seal facts the rounds record), beside the copy receipts and a different
+// object from them.
+func LedgerKey(host, opID string) string { return "move-receipts/" + host + "/" + opID + "-moves.json" }
+
+// Deployment-bucket keys (§5.3, §7.3, §13): reports, evidence and the log.
+func ReportPrefix(acct, app, host, opID string) string {
+ return "moves/" + acct + "/" + app + "/" + host + "/" + opID + "/"
+}
+func SourceReportKey(prefix string) string { return prefix + "source.json" }
+func TargetReportKey(prefix string) string { return prefix + "target.json" }
+func StoppedKey(prefix, role string) string { return prefix + "stopped-" + role + ".json" }
+func LogKey(prefix string, seq int, event string) string {
+ return fmt.Sprintf("%slog/%07d-%s.json", prefix, seq, event)
+}
diff --git a/internal/moves/staging_test.go b/internal/moves/staging_test.go
new file mode 100644
index 000000000..65fe4e17b
--- /dev/null
+++ b/internal/moves/staging_test.go
@@ -0,0 +1,52 @@
+package moves
+
+import (
+ "encoding/json"
+ "testing"
+)
+
+func TestStagingKeysAreScopedPerOperation(t *testing.T) {
+ host, op := "a.go.example.com", "mv_abcdefghijklmnopqrstuvwxyz"
+ if StagingPrefix(host, op) != "moves/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz/" {
+ t.Fatal(StagingPrefix(host, op))
+ }
+ if ReplicaPrefix(host, op) != StagingPrefix(host, op)+"replica/" || FilesPrefix(host, op) != StagingPrefix(host, op)+"files/" || TimersPrefix(host, op) != StagingPrefix(host, op)+"timers/" {
+ t.Fatal("subprefixes")
+ }
+ if MoveJSONKey(host, op) != StagingPrefix(host, op)+"move.json" {
+ t.Fatal(MoveJSONKey(host, op))
+ }
+ if LaneKey(op) != "moves/_lane/mv_abcdefghijklmnopqrstuvwxyz" || ReceiptsKey(host, op) != "move-receipts/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz.json" || SnapshotsPrefix(host, op) != "move-snapshots/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz/" {
+ t.Fatal("source-bucket keys")
+ }
+ p := ReportPrefix("bab", "app", host, op)
+ if p != "moves/bab/app/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz/" || StoppedKey(p, "source") != p+"stopped-source.json" || TargetReportKey(p) != p+"target.json" || SourceReportKey(p) != p+"source.json" || LogKey(p, 7, "flipped") != p+"log/0000007-flipped.json" {
+ t.Fatal("deployment-bucket keys")
+ }
+ // The park scope grammar has no trailing slash (claim rev 6.23 §2).
+ if ScopeLocation(host, op) != "moves/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz" {
+ t.Fatal(ScopeLocation(host, op))
+ }
+}
+
+// The per-operation ledger sits beside the receipts in the source bucket
+// (Task 14, ruling F30), and is a different key from them.
+func TestLedgerKeyIsBesideTheReceiptsAndDistinctFromThem(t *testing.T) {
+ host, op := "a.go.example.com", "mv_abcdefghijklmnopqrstuvwxyz"
+ if got := LedgerKey(host, op); got != "move-receipts/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz-moves.json" {
+ t.Fatal(got)
+ }
+ if LedgerKey(host, op) == ReceiptsKey(host, op) {
+ t.Fatal("ledger and receipts share a key")
+ }
+}
+
+func TestMoveJSONCarriesTheWireFieldNames(t *testing.T) {
+ b, err := json.Marshal(MoveJSON{MoveID: "mv_x", Incarnation: "i", Host: "h", From: "f", To: "t"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if string(b) != `{"move_id":"mv_x","incarnation":"i","host":"h","from":"f","to":"t"}` {
+ t.Fatal(string(b))
+ }
+}