amadan

carlos / platform Public

moves: the return lane, its seal and its baseline (§7.8 returning, J22)

ReturnLane is a staging prefix of the rollback id in the source's bucket.
ReturnSeal takes many replica objects (the as-given case) and states
FromDB and SrcCovered; its bound is the staging lease's. The return
baseline, journalled beside the seal, is what the source's install
restores to (Task 3's installSource.Baseline), encoded and read strictly.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 6cb56fb89f5064f4d4b0b3ecec615c50d28ebc54 parent efef676
2 files changed, +236 −0
  • internal/moves/returnlane.go +96 −0
  • internal/moves/returnlane_test.go +140 −0
diff --git a/internal/moves/returnlane.go b/internal/moves/returnlane.go
new file mode 100644
index 000000000..d98580d12
--- /dev/null
+++ b/internal/moves/returnlane.go
@@ -0,0 +1,96 @@
+package moves
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The return lane (moves spec §7.8 returning, Plan 4 Task 4): a dirty
+// rollback's target sends its stopped state back to the source through a
+// staging lane named by the rollback id, moves/<host>/<rb id>/, in the
+// SOURCE's bucket (M17: the target can put into the source's moves/*). It is
+// a staging lane like the cut's (round-2 finding 6): every object goes in
+// through the guarded staging writer under the staging-work lease for
+// (source region, host), and the lease's bound is recorded in the seal, so
+// the lane's park (Plan 5) acknowledges only after the last request a paused
+// writer signed has expired.
+//
+// The return seal is journalled in the claim under JournalReturnSeal, the
+// baseline the return restores to under JournalReturnBaseline. The source's
+// install (Task 3's installSource) reads the first as its Seal and the
+// second as its Baseline.
+
+// The claim journal keys of a dirty return, written in one claim write
+// before the record says returned.
+const (
+ JournalReturnSeal = "return_seal"
+ JournalReturnBaseline = "return_baseline"
+)
+
+// ReturnLane is the return lane's prefix: a staging prefix of the rollback
+// id, in the source's bucket.
+func ReturnLane(host, rbID string) string { return StagingPrefix(host, rbID) }
+
+// ReturnSeal is the seal of a return lane. replicas are the lane's replica
+// objects: one snapshot [1, n] of the stopped database for a dirty return,
+// or every level file of the staged seal copied as given (J22). The lane is
+// built from the target's stopped state and covers it exactly, so FromDB
+// and SrcCovered are both true. leaseUntil is the staging lease's bound
+// when the last lane object was written. The caller's slices are copied.
+func ReturnSeal(n uint64, replicas []SealedObject, files, timers []SealedObject, now time.Time, leaseUntil time.Time) Seal {
+ return Seal{V: SealVersion, N: n, Replica: sortObjects(replicas), Files: sortObjects(files), Timers: sortObjects(timers),
+ SrcCovered: true, FromDB: true, SealedAt: now.UTC(), LeaseUntil: leaseUntil.UTC()}
+}
+
+// errBadBaseline is a return baseline that does not name what a return
+// restores to.
+var errBadBaseline = errors.New("moves: the return baseline is not usable")
+
+// validBaseline requires what the source's install compares against: the
+// final TXID, both files digests, the timers digest and the evidence it was
+// taken under.
+func validBaseline(b provision.MoveSource) error {
+ if b.FinalTXID == 0 || b.Files == nil || b.Files.Plain == "" || b.Files.Sealed == "" || b.TimersSHA256 == "" || b.Evidence == "" {
+ return fmt.Errorf("%w: %+v", errBadBaseline, b)
+ }
+ return nil
+}
+
+// EncodeReturnBaseline is the return's baseline as the claim journal
+// carries it (Task 3's installSource.Baseline: the final TXID N', the files
+// and timers digests, the carried replication and next_due, and the target's
+// stop evidence).
+func EncodeReturnBaseline(b provision.MoveSource) (string, error) {
+ if err := validBaseline(b); err != nil {
+ return "", err
+ }
+ raw, err := json.Marshal(b)
+ if err != nil {
+ return "", err
+ }
+ return string(raw), nil
+}
+
+// DecodeReturnBaseline reads a journalled baseline strictly (no unknown
+// field, no trailing data) and validates it.
+func DecodeReturnBaseline(raw string) (provision.MoveSource, error) {
+ dec := json.NewDecoder(bytes.NewReader([]byte(raw)))
+ dec.DisallowUnknownFields()
+ var b provision.MoveSource
+ if err := dec.Decode(&b); err != nil {
+ return provision.MoveSource{}, fmt.Errorf("%w: %v", errBadBaseline, err)
+ }
+ if _, err := dec.Token(); err != io.EOF {
+ return provision.MoveSource{}, fmt.Errorf("%w: trailing data", errBadBaseline)
+ }
+ if err := validBaseline(b); err != nil {
+ return provision.MoveSource{}, err
+ }
+ return b, nil
+}
diff --git a/internal/moves/returnlane_test.go b/internal/moves/returnlane_test.go
new file mode 100644
index 000000000..eed5778b0
--- /dev/null
+++ b/internal/moves/returnlane_test.go
@@ -0,0 +1,140 @@
+package moves
+
+import (
+ "errors"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/superfly/ltx"
+)
+
+const (
+ rlHost = "a.example.com"
+ rlRB = "rb_abcdefghijklmnopqrstuvwxyz"
+)
+
+// The return lane is a staging lane (§7.3's rules apply to it, round-2
+// finding 6): a staging prefix named by the rollback id, never the move's.
+func TestReturnLaneIsTheRollbacksStagingPrefix(t *testing.T) {
+ if got := ReturnLane(rlHost, rlRB); got != StagingPrefix(rlHost, rlRB) || got != "moves/"+rlHost+"/"+rlRB+"/" {
+ t.Fatalf("ReturnLane = %q", got)
+ }
+ if ScopeLocation(rlHost, rlRB)+"/" != ReturnLane(rlHost, rlRB) {
+ t.Fatal("the lane's park scope is not the lane")
+ }
+}
+
+func returnFiles() ([]SealedObject, []SealedObject) {
+ lane := ReturnLane(rlHost, rlRB)
+ return []SealedObject{{Key: lane + "files/files-ab.sealed", VersionID: "v2", Size: 100}, {Key: lane + "files/files-ab.json", VersionID: "v3", Size: 50}},
+ []SealedObject{{Key: lane + "timers/timers.json", VersionID: "v4", Size: 40}}
+}
+
+// The return seal states the two facts a dirty return makes true (§7.8
+// returning): it is built from the stopped target's state (FromDB) and the
+// lane covers that state exactly (SrcCovered). Its bound is the staging
+// lease's, and it validates and round-trips like any seal.
+func TestReturnSealOfTheStoppedSnapshot(t *testing.T) {
+ at := time.Date(2026, 10, 8, 9, 0, 0, 0, time.UTC)
+ until := at.Add(40 * time.Minute)
+ snap := SealedObject{Key: ReturnLane(rlHost, rlRB) + "replica/0009/" + ltx.FormatFilename(1, 9), VersionID: "v1", Size: ltx.HeaderSize + 4096}
+ files, timers := returnFiles()
+ s := ReturnSeal(9, []SealedObject{snap}, files, timers, at, until)
+ if s.V != SealVersion || s.N != 9 || !s.FromDB || !s.SrcCovered || !s.SealedAt.Equal(at) || !s.Bound().Equal(until) {
+ t.Fatalf("seal %+v", s)
+ }
+ if err := s.Validate(); err != nil {
+ t.Fatal(err)
+ }
+ if !s.ReachesN() || !s.HasFiles() {
+ t.Fatalf("the snapshot does not reach N or the archive is missing: %+v", s)
+ }
+ got, err := DecodeSeal(s.Encode())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Encode() != s.Encode() || len(got.Replica) != 1 || got.Replica[0] != snap {
+ t.Fatalf("round trip: %+v", got)
+ }
+}
+
+// The "as given" return (J22): the staged seal's many level files, each in
+// the lane at the version the copy answered.
+func TestReturnSealAsGivenTakesManyReplicaObjects(t *testing.T) {
+ at := time.Date(2026, 10, 8, 9, 0, 0, 0, time.UTC)
+ lane := ReturnLane(rlHost, rlRB)
+ replica := []SealedObject{
+ {Key: lane + "replica/0000/" + ltx.FormatFilename(3, 3), VersionID: "v6", Size: ltx.HeaderSize},
+ {Key: lane + "replica/0009/" + ltx.FormatFilename(1, 2), VersionID: "v5", Size: ltx.HeaderSize + 10},
+ }
+ files, timers := returnFiles()
+ s := ReturnSeal(3, replica, files, timers, at, at.Add(time.Hour))
+ if err := s.Validate(); err != nil {
+ t.Fatal(err)
+ }
+ if len(s.Replica) != 2 || !s.ReachesN() {
+ t.Fatalf("seal %+v", s)
+ }
+ // The caller's slices are not reordered under it.
+ if replica[0].VersionID != "v6" {
+ t.Fatalf("ReturnSeal sorted the caller's slice: %+v", replica)
+ }
+ // A bound before the seal pins nothing: Validate refuses it.
+ if err := ReturnSeal(3, replica, files, timers, at, at.Add(-time.Second)).Validate(); !errors.Is(err, ErrBadSeal) {
+ t.Fatalf("a bound before the seal: %v", err)
+ }
+}
+
+func baselineForTest() provision.MoveSource {
+ return provision.MoveSource{Replication: "standard", NextDue: 1234, FinalTXID: 9, DBSHA256: "sha256:db",
+ Files: &provision.MoveFilesDigest{Plain: "sha256:plain", Sealed: "sha256:sealed", Bytes: 5, Entries: 2},
+ TimersSHA256: "sha256:timers", Evidence: "moves/bab/app/" + rlHost + "/mv_x/stopped-target.json"}
+}
+
+// The return's baseline (what the dirty return restores to, Task 3's
+// installSource.Baseline) is journalled beside the return seal and read back
+// strictly.
+func TestReturnBaselineRoundTrip(t *testing.T) {
+ b := baselineForTest()
+ raw, err := EncodeReturnBaseline(b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := DecodeReturnBaseline(raw)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Files == nil || *got.Files != *b.Files || got.FinalTXID != 9 || got.TimersSHA256 != b.TimersSHA256 || got.Evidence != b.Evidence ||
+ got.Replication != "standard" || got.NextDue != 1234 || got.DBSHA256 != "sha256:db" {
+ t.Fatalf("round trip: %+v", got)
+ }
+ for name, edit := range map[string]func(*provision.MoveSource){
+ "no final txid": func(b *provision.MoveSource) { b.FinalTXID = 0 },
+ "no files": func(b *provision.MoveSource) { b.Files = nil },
+ "no plain digest": func(b *provision.MoveSource) { b.Files.Plain = "" },
+ "no sealed digest": func(b *provision.MoveSource) { b.Files.Sealed = "" },
+ "no timers": func(b *provision.MoveSource) { b.TimersSHA256 = "" },
+ "no evidence": func(b *provision.MoveSource) { b.Evidence = "" },
+ } {
+ bad := baselineForTest()
+ files := *bad.Files
+ bad.Files = &files
+ edit(&bad)
+ if _, err := EncodeReturnBaseline(bad); err == nil {
+ t.Errorf("%s: encoded", name)
+ }
+ }
+ for name, raw := range map[string]string{
+ "empty": "",
+ "not json": "{",
+ "unknown field": strings.Replace(raw, `{`, `{"extra":1,`, 1),
+ "trailing data": raw + "{}",
+ "incomplete": `{"final_txid":9}`,
+ } {
+ if _, err := DecodeReturnBaseline(raw); err == nil {
+ t.Errorf("%s: decoded", name)
+ }
+ }
+}