amadan

carlos / platform Public

Merge remote-tracking branch 'origin/feat/hostclaim-genselect' into feat/hostclaim-box

claude bf3b5e81690634b13a4a04789715533db9619f34 parent eea8d9bparent d7621a2
3 files changed, +1354 −0
  • internal/genselect/compatibility_test.go +66 −0
  • internal/genselect/select.go +509 −0
  • internal/genselect/select_test.go +779 −0
diff --git a/internal/genselect/compatibility_test.go b/internal/genselect/compatibility_test.go
new file mode 100644
index 000000000..dfa60d848
--- /dev/null
+++ b/internal/genselect/compatibility_test.go
@@ -0,0 +1,66 @@
+package genselect
+
+import (
+ "io"
+ "log/slog"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ filereplica "github.com/benbjohnson/litestream/file"
+ "github.com/superfly/ltx"
+)
+
+func TestCalcRestorePlanCompatibility(t *testing.T) {
+ type coords struct {
+ level int
+ min, max uint64
+ }
+ tests := []struct {
+ name string
+ files []coords
+ }{
+ {"compacted_overlap", []coords{{9, 1, 11}, {1, 10, 13}}},
+ {"genuine_gap", []coords{{9, 1, 11}, {0, 13, 13}}},
+ {"successor_snapshot", []coords{{9, 1, 11}, {9, 1, 20}, {0, 13, 13}, {0, 21, 21}}},
+ {"origin", []coords{{0, 1, 1}, {1, 2, 5}}},
+ {"tie_breaks", []coords{{9, 1, 11}, {8, 12, 19}, {8, 12, 20}, {1, 10, 20}, {2, 10, 20}}},
+ {"anchor_is_not_continuation", []coords{{9, 1, 11}, {0, 11, 11}, {0, 12, 12}}},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFixture(t)
+ f.origin()
+ client := filereplica.NewReplicaClient(t.TempDir())
+ for _, c := range tt.files {
+ f.file("", c.level, c.min, c.max)
+ path := client.LTXFilePath(c.level, ltx.TXID(c.min), ltx.TXID(c.max))
+ if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(path, header(t, c.min, c.max), 0600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ want, wantErr := litestream.CalcRestorePlan(t.Context(), client, 0, time.Time{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
+ got, err := Select(t.Context(), f.b, f.req)
+ if (err != nil) != (wantErr != nil) {
+ t.Fatalf("selector %v; litestream %v", err, wantErr)
+ }
+ if err != nil {
+ return
+ }
+ if len(got.Files) != len(want) {
+ t.Fatalf("length %d != %d", len(got.Files), len(want))
+ }
+ for i, v := range want {
+ x := got.Files[i]
+ if x.Level != v.Level || x.MinTXID != uint64(v.MinTXID) || x.MaxTXID != uint64(v.MaxTXID) {
+ t.Fatalf("file %d: %+v != %+v", i, x, v)
+ }
+ }
+ })
+ }
+}
diff --git a/internal/genselect/select.go b/internal/genselect/select.go
new file mode 100644
index 000000000..70e6ad3c8
--- /dev/null
+++ b/internal/genselect/select.go
@@ -0,0 +1,509 @@
+package genselect
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/superfly/ltx"
+)
+
+var (
+ ErrListing = errors.New("genselect: incomplete or invalid listing")
+ ErrDisagreement = errors.New("genselect: marker and holder disagree")
+ ErrIncarnation = errors.New("genselect: marker incarnation mismatch")
+ ErrMarker = errors.New("genselect: malformed marker")
+ ErrVersionID = errors.New("genselect: immutable version id required")
+ ErrDeleteMarker = errors.New("genselect: selected version is a delete marker")
+ ErrPinnedMissing = errors.New("genselect: pinned snapshot version missing")
+ ErrPinnedMismatch = errors.New("genselect: pinned snapshot mismatch")
+ ErrGap = errors.New("genselect: non-contiguous ltx files")
+ ErrEmpty = errors.New("genselect: no restorable data")
+ ErrRegion = errors.New("genselect: baseline belongs to another region")
+ ErrCovered = errors.New("genselect: local baseline is covered")
+ ErrInput = errors.New("genselect: invalid input")
+ ErrRoot = errors.New("genselect: invalid root")
+)
+
+type Reader interface {
+ ListVersions(context.Context, string, string) ([]lease.ObjectVersion, error)
+ GetVersion(context.Context, string, string, string) ([]byte, error)
+ OpenVersion(context.Context, string, string, string) (io.ReadCloser, error)
+}
+
+var _ Reader = (lease.VersionedStore)(nil)
+
+type Pin struct {
+ SnapshotKey string `json:"snapshot_key"`
+ SnapshotVersion string `json:"snapshot_version"`
+ SnapshotTXID uint64 `json:"snapshot_txid"`
+ SnapshotCRC64 uint64 `json:"snapshot_crc64"`
+}
+
+type Marker struct {
+ Op string `json:"op"`
+ Incarnation string `json:"incarnation"`
+ Host string `json:"host"`
+ Box string `json:"box"`
+ InstalledAt time.Time `json:"installed_at"`
+ ManifestSHA256 string `json:"manifest_sha256"`
+ Pin
+}
+
+// PreparedPin is supplied only from an authorized, completed build journal.
+// The caller binds this exact generation and pin to the authorized holder,
+// operation and destination region before calling SelectPrepared.
+type PreparedPin struct {
+ Generation string
+ SnapshotKey string
+ SnapshotVersion string
+ SnapshotTXID uint64
+ SnapshotCRC64 uint64
+}
+
+type Request struct {
+ Bucket, Region, Host string
+ Holder hostclaim.Holder
+}
+
+type File struct {
+ Key string `json:"key"`
+ VersionID string `json:"version_id"`
+ Level int `json:"level"`
+ MinTXID uint64 `json:"min_txid"`
+ MaxTXID uint64 `json:"max_txid"`
+ Size int64 `json:"size"`
+}
+
+type Plan struct {
+ Generation string
+ Files []File
+ Root *File
+ Marker *Marker
+ MarkerVersion string
+ Pin *Pin
+ MaxTXID uint64
+ MaxLevel0TXID uint64
+}
+
+type candidate struct {
+ generation, region string
+ covered bool
+}
+
+func origin(g string) string {
+ if g == "" {
+ return hostclaim.RootGeneration
+ }
+ return g
+}
+
+func component(s string) bool {
+ return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
+}
+
+func generation(s string) bool { return component(s) && s != hostclaim.RootGeneration }
+
+func versionID(s string) bool { return strings.TrimSpace(s) != "" && s != "null" }
+
+func prefix(location, gen string) string {
+ if gen == hostclaim.RootGeneration {
+ return location
+ }
+ return location + "g-" + gen + "/"
+}
+
+func coordinates(key, base string) (File, bool) {
+ if !strings.HasPrefix(key, base) {
+ return File{}, false
+ }
+ parts := strings.Split(strings.TrimPrefix(key, base), "/")
+ if len(parts) != 2 || len(parts[0]) != 4 {
+ return File{}, false
+ }
+ level, err := strconv.ParseUint(parts[0], 16, 16)
+ if err != nil || level > 9 || fmt.Sprintf("%04x", level) != parts[0] {
+ return File{}, false
+ }
+ min, max, err := ltx.ParseFilename(parts[1])
+ if err != nil || min == 0 || min > max || ltx.FormatFilename(min, max) != parts[1] {
+ return File{}, false
+ }
+ return File{Key: key, Level: int(level), MinTXID: uint64(min), MaxTXID: uint64(max)}, true
+}
+
+func exactFile(v lease.ObjectVersion, base string) (File, error) {
+ f, ok := coordinates(v.Key, base)
+ if !ok || v.Size < 0 {
+ return File{}, ErrListing
+ }
+ if !versionID(v.VersionID) {
+ return File{}, ErrVersionID
+ }
+ if v.DeleteMarker {
+ return File{}, ErrDeleteMarker
+ }
+ f.VersionID, f.Size = v.VersionID, v.Size
+ return f, nil
+}
+
+func requestLocation(s Reader, req Request) (string, error) {
+ h := req.Holder
+ location := strings.TrimSuffix(h.Location, "/")
+ if s == nil || req.Bucket == "" || req.Region == "" || !component(req.Host) || !component(h.Incarnation) ||
+ (location != "instances/"+req.Host && location != "instances/"+req.Host+"/"+h.Incarnation) ||
+ (h.Generation != "" && !generation(h.Generation)) {
+ return "", ErrInput
+ }
+ return location + "/", nil
+}
+
+func listVersions(ctx context.Context, s Reader, bucket, base string) ([]lease.ObjectVersion, map[string]lease.ObjectVersion, error) {
+ // VersionedStore drains all pages before returning. Never use partial data.
+ versions, err := s.ListVersions(ctx, bucket, base)
+ if err != nil {
+ return nil, nil, fmt.Errorf("%w: %w", ErrListing, err)
+ }
+ latest := make(map[string]lease.ObjectVersion)
+ keys := make(map[string]bool)
+ for _, v := range versions {
+ if !strings.HasPrefix(v.Key, base) {
+ return nil, nil, ErrListing
+ }
+ keys[v.Key] = true
+ if v.IsLatest {
+ if _, exists := latest[v.Key]; exists {
+ return nil, nil, ErrListing
+ }
+ latest[v.Key] = v
+ }
+ }
+ for k := range keys {
+ if _, ok := latest[k]; !ok {
+ return nil, nil, ErrListing
+ }
+ }
+ return versions, latest, nil
+}
+
+// Select resolves the ordinary baseline using the marker and unfinished run.
+// It returns identities only. The consumer must verify every complete file.
+func Select(ctx context.Context, s Reader, req Request) (*Plan, error) {
+ location, err := requestLocation(s, req)
+ if err != nil {
+ return nil, err
+ }
+ versions, latest, err := listVersions(ctx, s, req.Bucket, location)
+ if err != nil {
+ return nil, err
+ }
+ h := req.Holder
+ p := &Plan{Files: []File{}}
+ markerGen := hostclaim.RootGeneration
+ if mv, ok := latest[location+"generation.json"]; ok {
+ if !versionID(mv.VersionID) {
+ return nil, ErrVersionID
+ }
+ if mv.DeleteMarker {
+ return nil, ErrDeleteMarker
+ }
+ data, err := s.GetVersion(ctx, req.Bucket, mv.Key, mv.VersionID)
+ if err != nil {
+ return nil, fmt.Errorf("%w: %w", ErrMarker, err)
+ }
+ var fields map[string]json.RawMessage
+ if err := json.Unmarshal(data, &fields); err != nil {
+ return nil, ErrMarker
+ }
+ if raw, ok := fields["snapshot_txid"]; !ok || string(raw) == "null" {
+ return nil, ErrMarker
+ }
+ var m Marker
+ dec := json.NewDecoder(bytes.NewReader(data))
+ // Unknown additive fields are allowed. Malformed/trailing JSON is not.
+ if err := dec.Decode(&m); err != nil {
+ return nil, fmt.Errorf("%w: %w", ErrMarker, err)
+ }
+ if err := dec.Decode(new(any)); err != io.EOF {
+ return nil, ErrMarker
+ }
+ if m.Incarnation != h.Incarnation {
+ return nil, ErrIncarnation
+ }
+ if m.Host != req.Host || !generation(m.Op) || m.Box == "" || m.InstalledAt.IsZero() {
+ return nil, ErrMarker
+ }
+ markerGen, p.Marker, p.MarkerVersion = m.Op, &m, mv.VersionID
+ }
+ run, err := unfinished(h, markerGen, req.Region)
+ if err != nil {
+ return nil, err
+ }
+ selected := candidate{generation: markerGen, region: req.Region}
+ for _, c := range run {
+ if c.region != "" && c.region != req.Region {
+ continue
+ }
+ if c.covered {
+ return nil, ErrCovered
+ }
+ if c.generation == markerGen || hasObject(latest, location, c.generation) {
+ selected = c
+ break
+ }
+ }
+ p.Generation = selected.generation
+ base := prefix(location, p.Generation)
+ var pin *Pin
+ if p.Marker != nil && p.Generation == markerGen {
+ cp := p.Marker.Pin
+ pin = &cp
+ }
+ return selectFiles(ctx, s, req.Bucket, base, versions, latest, p, pin)
+}
+
+// SelectPrepared restores only the explicitly authorized prepared generation.
+// It never reads the marker or walks admissions, and never chooses a baseline.
+func SelectPrepared(ctx context.Context, s Reader, req Request, prepared PreparedPin) (*Plan, error) {
+ location, err := requestLocation(s, req)
+ if err != nil {
+ return nil, err
+ }
+ if !generation(prepared.Generation) {
+ return nil, ErrInput
+ }
+ // Empty prepared roots require a separate bootstrap protocol, not fallback.
+ if prepared.SnapshotTXID == 0 {
+ return nil, ErrPinnedMismatch
+ }
+ base := prefix(location, prepared.Generation)
+ versions, latest, err := listVersions(ctx, s, req.Bucket, base)
+ if err != nil {
+ return nil, err
+ }
+ pin := &Pin{SnapshotKey: prepared.SnapshotKey, SnapshotVersion: prepared.SnapshotVersion,
+ SnapshotTXID: prepared.SnapshotTXID, SnapshotCRC64: prepared.SnapshotCRC64}
+ p := &Plan{Generation: prepared.Generation, Files: []File{}}
+ return selectFiles(ctx, s, req.Bucket, base, versions, latest, p, pin)
+}
+
+// Both entry points share exact-version root selection and continuation.
+func selectFiles(ctx context.Context, s Reader, bucket, base string, versions []lease.ObjectVersion,
+ latest map[string]lease.ObjectVersion, p *Plan, pin *Pin) (*Plan, error) {
+ if pin != nil {
+ if pin.SnapshotTXID == 0 {
+ if pin.SnapshotKey != "" || pin.SnapshotVersion != "" || pin.SnapshotCRC64 != 0 {
+ return nil, ErrPinnedMismatch
+ }
+ } else {
+ f, ok := coordinates(pin.SnapshotKey, base)
+ if !ok || f.Level != 9 || f.MinTXID != 1 || f.MaxTXID != pin.SnapshotTXID {
+ return nil, ErrPinnedMismatch
+ }
+ if !versionID(pin.SnapshotVersion) {
+ return nil, ErrVersionID
+ }
+ }
+ p.Pin = pin
+ }
+ var current []File
+ for _, v := range latest {
+ f, ok := coordinates(v.Key, base)
+ if !ok && strings.HasPrefix(v.Key, base) {
+ rel := strings.TrimPrefix(v.Key, base)
+ parts := strings.Split(rel, "/")
+ if len(parts) == 2 && len(parts[0]) == 4 && strings.HasSuffix(parts[1], ".ltx") {
+ if level, e := strconv.ParseUint(parts[0], 16, 16); e == nil && level <= 9 {
+ return nil, ErrListing
+ }
+ }
+ }
+ if !ok || v.DeleteMarker {
+ continue
+ } // Retention tombstones are not current files.
+ f, err := exactFile(v, base)
+ if err != nil {
+ return nil, err
+ }
+ if f.Level == 9 && f.MinTXID != 1 {
+ return nil, ErrRoot
+ }
+ current = append(current, f)
+ if f.Level == 0 && f.MaxTXID > p.MaxLevel0TXID {
+ p.MaxLevel0TXID = f.MaxTXID
+ }
+ }
+ floor := uint64(0)
+ if pin != nil {
+ floor = pin.SnapshotTXID
+ }
+ for _, f := range current {
+ // A late duplicate at the pinned key never replaces its acknowledged version.
+ if pin != nil && floor > 0 && f.Key == pin.SnapshotKey {
+ continue
+ }
+ if f.Level == 9 && f.MinTXID == 1 && f.MaxTXID >= floor && (p.Root == nil || better(f, *p.Root)) {
+ cp := f
+ p.Root = &cp
+ }
+ }
+ pinned := false
+ if p.Root == nil && floor > 0 {
+ var pv *lease.ObjectVersion
+ for _, v := range versions {
+ if v.Key == pin.SnapshotKey && v.VersionID == pin.SnapshotVersion {
+ cp := v
+ pv = &cp
+ break
+ }
+ }
+ if pv == nil {
+ return nil, ErrPinnedMissing
+ }
+ f, err := exactFile(*pv, base)
+ if err != nil {
+ return nil, err
+ }
+ p.Root, pinned = &f, true
+ }
+ if p.Root != nil {
+ if err := checkRoot(ctx, s, bucket, *p.Root, pinned); err != nil {
+ return nil, err
+ }
+ p.Files = append(p.Files, *p.Root)
+ p.MaxTXID = p.Root.MaxTXID
+ }
+ for {
+ var best *File
+ for _, f := range current {
+ if f.Level == 9 || p.MaxTXID == ^uint64(0) || !ltx.IsContiguous(ltx.TXID(p.MaxTXID), ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID)) {
+ continue
+ }
+ if best == nil || better(f, *best) {
+ cp := f
+ best = &cp
+ }
+ }
+ if best == nil {
+ break
+ }
+ if p.Root == nil {
+ if err := checkRoot(ctx, s, bucket, *best, false); err != nil {
+ return nil, err
+ }
+ cp := *best
+ p.Root = &cp
+ }
+ p.Files = append(p.Files, *best)
+ p.MaxTXID = best.MaxTXID
+ }
+ for _, f := range current {
+ if f.Level < 9 && f.MaxTXID > p.MaxTXID {
+ return nil, ErrGap
+ }
+ }
+ if len(p.Files) == 0 {
+ return nil, ErrEmpty
+ }
+ return p, nil
+}
+
+func unfinished(h hostclaim.Holder, marker, region string) ([]candidate, error) {
+ gen := origin(h.Generation)
+ if gen == marker {
+ return []candidate{{generation: marker, region: region}}, nil
+ }
+ var out []candidate
+ covered := false
+ seen := map[string]bool{}
+ for i := len(h.Admissions) - 1; i >= 0; i-- {
+ a := h.Admissions[i]
+ if origin(a.Generation) != gen || a.Region != region {
+ return nil, ErrDisagreement
+ }
+ if a.Mode == "warm" {
+ if origin(a.Predecessor) != gen || a.PredecessorRegion != region {
+ return nil, ErrDisagreement
+ }
+ continue
+ }
+ if a.Mode != "cold" || !generation(a.Generation) || seen[gen] {
+ return nil, ErrDisagreement
+ }
+ seen[gen] = true
+ out = append(out, candidate{generation: gen, region: region, covered: covered})
+ gen, region, covered = origin(a.Predecessor), a.PredecessorRegion, a.Covered
+ if gen == marker {
+ return append(out, candidate{generation: gen, region: region, covered: covered}), nil
+ }
+ if !generation(gen) || region == "" {
+ return nil, ErrDisagreement
+ }
+ }
+ return nil, ErrDisagreement
+}
+
+func hasObject(latest map[string]lease.ObjectVersion, location, gen string) bool {
+ base := prefix(location, gen)
+ for k := range latest {
+ if !strings.HasPrefix(k, base) {
+ continue
+ }
+ if gen == hostclaim.RootGeneration {
+ if _, ok := coordinates(k, base); !ok {
+ continue
+ }
+ }
+ return true
+ }
+ return false
+}
+
+func better(a, b File) bool {
+ if a.MaxTXID != b.MaxTXID {
+ return a.MaxTXID > b.MaxTXID
+ }
+ if a.MinTXID != b.MinTXID {
+ return a.MinTXID < b.MinTXID
+ }
+ return a.Level > b.Level
+}
+
+func checkRoot(ctx context.Context, s Reader, bucket string, f File, pinned bool) error {
+ category := ErrRoot
+ if pinned {
+ category = ErrPinnedMismatch
+ }
+ rc, err := s.OpenVersion(ctx, bucket, f.Key, f.VersionID)
+ if err != nil {
+ if errors.Is(err, lease.ErrIsDeleteMarker) {
+ return fmt.Errorf("%w: %w", ErrDeleteMarker, err)
+ }
+ if pinned && errors.Is(err, lease.ErrMirrorNotFound) {
+ return fmt.Errorf("%w: %w", ErrPinnedMissing, err)
+ }
+ return fmt.Errorf("%w: %w", category, err)
+ }
+ dec := ltx.NewDecoder(rc)
+ err = dec.DecodeHeader()
+ closeErr := rc.Close()
+ if err != nil {
+ return fmt.Errorf("%w: %w", category, err)
+ }
+ if closeErr != nil {
+ return fmt.Errorf("%w: %w", category, closeErr)
+ }
+ h := dec.Header()
+ if uint64(h.MinTXID) != f.MinTXID || uint64(h.MaxTXID) != f.MaxTXID {
+ return category
+ }
+ return nil
+}
diff --git a/internal/genselect/select_test.go b/internal/genselect/select_test.go
new file mode 100644
index 000000000..64e7a82ca
--- /dev/null
+++ b/internal/genselect/select_test.go
@@ -0,0 +1,779 @@
+package genselect
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/superfly/ltx"
+)
+
+type fixture struct {
+ t *testing.T
+ b *leasetest.VersionedBucket
+ req Request
+ marker Marker
+ markerVersion string
+ pin File
+ store Reader
+}
+
+func newFixture(t *testing.T) *fixture {
+ b := leasetest.NewVersionedBucket()
+ b.AddBucket("data")
+ b.PageSize = 1
+ f := &fixture{t: t, b: b, store: b, req: Request{Bucket: "data", Region: "ie", Host: "h",
+ Holder: hostclaim.Holder{Incarnation: "inc", Location: "instances/h/inc", Generation: "g0"}}}
+ f.pin = f.file("g0", 9, 1, 11)
+ f.file("g0", 0, 11, 11)
+ f.marker = Marker{Op: "g0", Incarnation: "inc", Host: "h", Box: "box", InstalledAt: time.Unix(1, 0).UTC(),
+ ManifestSHA256: strings.Repeat("a", 64), Pin: Pin{SnapshotKey: f.pin.Key, SnapshotVersion: f.pin.VersionID, SnapshotTXID: 11}}
+ f.publish()
+ return f
+}
+
+func header(t *testing.T, min, max uint64) []byte {
+ t.Helper()
+ h := ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096,
+ MinTXID: ltx.TXID(min), MaxTXID: ltx.TXID(max)}
+ b, err := h.MarshalBinary()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+}
+
+func (f *fixture) file(gen string, level int, min, max uint64) File {
+ f.t.Helper()
+ key := fmt.Sprintf("%s%04x/%016x-%016x.ltx", prefix(strings.TrimSuffix(f.req.Holder.Location, "/")+"/", origin(gen)), level, min, max)
+ b := header(f.t, min, max)
+ id := f.b.Put("data", key, b)
+ return File{Key: key, VersionID: id, Level: level, MinTXID: min, MaxTXID: max, Size: int64(len(b))}
+}
+
+func (f *fixture) publish() {
+ f.t.Helper()
+ b, err := json.Marshal(f.marker)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ f.markerVersion = f.b.Put("data", f.req.Holder.Location+"/generation.json", b)
+}
+
+func (f *fixture) chain(base string) {
+ f.req.Holder.Generation = "g2"
+ f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
+ {Generation: "g1", Predecessor: base, Region: "ie", PredecessorRegion: "ie", Mode: "cold"},
+ {Generation: "g2", Predecessor: "g1", Region: "ie", PredecessorRegion: "ie", Mode: "cold"},
+ }
+}
+
+func (f *fixture) origin() {
+ f.req.Holder.Generation = ""
+ if err := f.b.DeleteVersion(f.t.Context(), "data", f.req.Holder.Location+"/generation.json", f.markerVersion); err != nil {
+ f.t.Fatal(err)
+ }
+ f.markerVersion = ""
+}
+
+// These are pure-selection ports, not simulations of wake, park or replication.
+// Paths refer to platform-moves-design/internal/sim/hostlife at rev 7.13.
+func TestModelWitnesses(t *testing.T) {
+ tests := []struct {
+ name, source string
+ setup func(*fixture) []File
+ gen string
+ wantErr error
+ }{
+ {"wRestoreOverlap", "witness_test.go:1204", func(f *fixture) []File {
+ return []File{f.pin, f.file("g0", 1, 10, 13)}
+ }, "g0", nil},
+ {"wRestoreGap", "witness_test.go:1233", func(f *fixture) []File {
+ f.file("g0", 0, 13, 13)
+ return nil
+ }, "", ErrGap},
+ {"wRootAdvance", "witness_test.go:1295", func(f *fixture) []File {
+ f.file("g0", 0, 13, 13)
+ root := f.file("g0", 9, 1, 20)
+ f.b.BareDelete("data", f.pin.Key)
+ return []File{root, f.file("g0", 0, 21, 21)}
+ }, "g0", nil},
+ {"wOriginSwept/origin", "witness_test.go:2373", func(f *fixture) []File {
+ f.origin()
+ return []File{f.file("", 0, 1, 1), f.file("", 1, 2, 5)}
+ }, hostclaim.RootGeneration, nil},
+ {"wOriginSwept/unfinished_chain", "witness_test.go:2373; protocol_test.go:1188", func(f *fixture) []File {
+ f.origin()
+ f.chain(hostclaim.RootGeneration)
+ return []File{f.file("", 0, 1, 3)}
+ }, hostclaim.RootGeneration, nil},
+ {"wBaselineParked/all_empty", "witness_test.go:2168; protocol_test.go:1154", func(f *fixture) []File {
+ f.chain("g0")
+ return []File{f.pin}
+ }, "g0", nil},
+ {"wBaselineParked/newest_nonempty", "witness_test.go:2168; protocol_test.go:1172", func(f *fixture) []File {
+ f.chain("g0")
+ return []File{f.file("g1", 0, 1, 1)}
+ }, "g1", nil},
+ {"wBaselineParked/holder_nonempty", "protocol_test.go:1161", func(f *fixture) []File {
+ f.chain("g0")
+ f.file("g1", 0, 1, 9)
+ return []File{f.file("g2", 0, 1, 1)}
+ }, "g2", nil},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ t.Log("port of", tt.source)
+ f := newFixture(t)
+ want := tt.setup(f)
+ before := f.b.Snapshot()
+ f.b.FailOn = func(op, key string) error {
+ if op != "list" && op != "get" {
+ t.Fatalf("mutation %s %s", op, key)
+ }
+ return nil
+ }
+ got, err := Select(t.Context(), f.store, f.req)
+ if !errors.Is(err, tt.wantErr) {
+ t.Fatalf("got %v, want %v", err, tt.wantErr)
+ }
+ if f.b.Snapshot() != before {
+ t.Fatal("selection mutated storage")
+ }
+ if tt.wantErr != nil {
+ if got != nil {
+ t.Fatal("partial plan")
+ }
+ return
+ }
+ if got.Generation != tt.gen || !reflect.DeepEqual(got.Files, want) {
+ t.Fatalf("got %+v, want %s %+v", got, tt.gen, want)
+ }
+ if got.MarkerVersion != f.markerVersion || got.Root == nil || *got.Root != want[0] || got.MaxTXID != want[len(want)-1].MaxTXID {
+ t.Fatalf("root, marker or final TXID: %+v", got)
+ }
+ })
+ }
+}
+
+type altered struct {
+ Reader
+ list func([]lease.ObjectVersion) ([]lease.ObjectVersion, error)
+ open func(context.Context, string, string, string) (io.ReadCloser, error)
+}
+
+func (s altered) ListVersions(ctx context.Context, bucket, base string) ([]lease.ObjectVersion, error) {
+ vs, err := s.Reader.ListVersions(ctx, bucket, base)
+ if err != nil {
+ return nil, err
+ }
+ if s.list != nil {
+ return s.list(vs)
+ }
+ return vs, nil
+}
+func (s altered) OpenVersion(ctx context.Context, bucket, key, id string) (io.ReadCloser, error) {
+ if s.open != nil {
+ return s.open(ctx, bucket, key, id)
+ }
+ return s.Reader.OpenVersion(ctx, bucket, key, id)
+}
+
+func TestR13_1UnfinishedRegionalWalk(t *testing.T) {
+ tests := []struct {
+ name string
+ setup func(*fixture) []File
+ gen string
+ want error
+ }{
+ {"interrupted_clean_return", func(f *fixture) []File {
+ return []File{f.pin}
+ }, "g0", nil},
+ {"another_interrupted_admission", func(f *fixture) []File {
+ f.req.Holder.Generation = "R2"
+ f.req.Holder.Admissions = append(f.req.Holder.Admissions,
+ hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold"})
+ return []File{f.pin}
+ }, "g0", nil},
+ {"covered_foreign_link_deeper_in_run", func(f *fixture) []File {
+ f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
+ {Generation: "T", Predecessor: "g0", Region: "us", PredecessorRegion: "ie", Mode: "cold"},
+ {Generation: "U", Predecessor: "T", Region: "uk", PredecessorRegion: "us", Mode: "cold", Covered: true},
+ {Generation: "R", Predecessor: "U", Region: "ie", PredecessorRegion: "uk", Mode: "cold"},
+ }
+ return []File{f.pin}
+ }, "g0", nil},
+ {"uncovered_foreign_link", func(f *fixture) []File {
+ f.req.Holder.Admissions[1].Covered = false
+ return []File{f.pin}
+ }, "g0", nil},
+ {"foreign_objects_never_selected", func(f *fixture) []File {
+ f.req.Holder.Admissions[1].Covered = false
+ f.file("T", 9, 1, 99)
+ return []File{f.pin}
+ }, "g0", nil},
+ {"holder_objects_with_covered_foreign_predecessor", func(f *fixture) []File {
+ return []File{f.file("R", 0, 1, 1)}
+ }, "R", nil},
+ {"newest_local_objects", func(f *fixture) []File {
+ f.req.Holder.Generation = "R2"
+ f.req.Holder.Admissions = append(f.req.Holder.Admissions,
+ hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold"})
+ return []File{f.file("R", 0, 1, 1)}
+ }, "R", nil},
+ {"unsafe_covered_local_baseline", func(f *fixture) []File {
+ f.req.Holder.Generation = "R2"
+ f.req.Holder.Admissions = append(f.req.Holder.Admissions,
+ hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold", Covered: true})
+ f.file("R", 0, 1, 1)
+ return nil
+ }, "", ErrCovered},
+ {"unsafe_covered_empty_local_baseline", func(f *fixture) []File {
+ f.req.Holder.Generation = "R2"
+ f.req.Holder.Admissions = append(f.req.Holder.Admissions,
+ hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold", Covered: true})
+ return nil
+ }, "", ErrCovered},
+ {"unsafe_covered_local_marker", func(f *fixture) []File {
+ f.req.Holder.Admissions[0].Covered = true
+ return nil
+ }, "", ErrCovered},
+ {"older_covered_local_not_needed", func(f *fixture) []File {
+ f.req.Holder.Admissions[0].Covered = true
+ return []File{f.file("R", 0, 1, 1)}
+ }, "R", nil},
+ {"covered_warm_session_in_run", func(f *fixture) []File {
+ f.req.Holder.Admissions = append(f.req.Holder.Admissions,
+ hostclaim.AdmissionRecord{Generation: "R", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "warm", Covered: true})
+ return []File{f.pin}
+ }, "g0", nil},
+ {"absent_marker_origin", func(f *fixture) []File {
+ f.origin()
+ f.req.Holder.Generation = "R"
+ f.req.Holder.Admissions[0].Predecessor = hostclaim.RootGeneration
+ return []File{f.file("", 0, 1, 1)}
+ }, hostclaim.RootGeneration, nil},
+ {"foreign_predecessor_region_marker_fallback", func(f *fixture) []File {
+ f.chain("g0")
+ f.req.Holder.Admissions[0].PredecessorRegion = "us"
+ return []File{f.pin}
+ }, "g0", nil},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFixture(t)
+ f.req.Holder.Generation = "R"
+ f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
+ {Generation: "T", Predecessor: "g0", Region: "us", PredecessorRegion: "ie", Mode: "cold"},
+ {Generation: "R", Predecessor: "T", Region: "ie", PredecessorRegion: "us", Mode: "cold", Coverage: "clean-return", Covered: true},
+ }
+ want := tt.setup(f)
+ before := f.b.Snapshot()
+ holderBefore, _ := json.Marshal(f.req.Holder)
+ got, err := Select(t.Context(), f.store, f.req)
+ holderAfter, _ := json.Marshal(f.req.Holder)
+ if f.b.Snapshot() != before || string(holderAfter) != string(holderBefore) {
+ t.Fatal("selection mutated inputs or storage")
+ }
+ if !errors.Is(err, tt.want) {
+ t.Fatalf("got %+v / %v, want %v", got, err, tt.want)
+ }
+ if tt.want != nil {
+ if got != nil {
+ t.Fatal("partial plan")
+ }
+ return
+ }
+ if got == nil || got.Generation != tt.gen || !reflect.DeepEqual(got.Files, want) ||
+ got.Root == nil || *got.Root != want[0] || got.MaxTXID != want[len(want)-1].MaxTXID ||
+ got.MarkerVersion != f.markerVersion {
+ t.Fatalf("got %+v, want %s %+v", got, tt.gen, want)
+ }
+ if tt.gen == f.marker.Op && (got.Pin == nil || *got.Pin != f.marker.Pin) {
+ t.Fatalf("marker pin not preserved: %+v", got.Pin)
+ }
+ })
+ }
+}
+
+func TestRefusals(t *testing.T) {
+ tests := []struct {
+ name string
+ setup func(*fixture)
+ want error
+ }{
+ {"marker_holder_disagreement", func(f *fixture) { f.req.Holder.Generation = "unknown" }, ErrDisagreement},
+ {"nonempty_unexplained_disagreement", func(f *fixture) { f.req.Holder.Generation = "unknown"; f.file("unknown", 0, 1, 1) }, ErrDisagreement},
+ {"covered_local_marker", func(f *fixture) { f.chain("g0"); f.req.Holder.Admissions[0].Covered = true }, ErrCovered},
+ {"broken_chain", func(f *fixture) { f.chain("g0"); f.req.Holder.Admissions[1].Predecessor = "missing" }, ErrDisagreement},
+ {"cyclic_chain", func(f *fixture) { f.chain("g0"); f.req.Holder.Admissions[0].Predecessor = "g2" }, ErrDisagreement},
+ {"incomplete_listing", func(f *fixture) { f.b.FailListAfterPages = 1 }, ErrListing},
+ {"partial_data_with_error", func(f *fixture) {
+ f.store = altered{Reader: f.b, list: func(vs []lease.ObjectVersion) ([]lease.ObjectVersion, error) { return vs, io.ErrUnexpectedEOF }}
+ }, ErrListing},
+ {"incomplete_unfinished_listing", func(f *fixture) { f.chain("g0"); f.b.FailListAfterPages = 1 }, ErrListing},
+ {"pinned_version_missing", func(f *fixture) {
+ f.b.Put("data", f.pin.Key, header(f.t, 1, 11))
+ if err := f.b.DeleteVersion(f.t.Context(), "data", f.pin.Key, f.pin.VersionID); err != nil {
+ f.t.Fatal(err)
+ }
+ }, ErrPinnedMissing},
+ {"pinned_version_vanished_after_listing", func(f *fixture) {
+ f.store = altered{Reader: f.b, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
+ return nil, lease.ErrMirrorNotFound
+ }}
+ }, ErrPinnedMissing},
+ {"pinned_header_mismatch", func(f *fixture) {
+ f.marker.SnapshotVersion = f.b.Put("data", f.pin.Key, header(f.t, 1, 12))
+ f.publish()
+ }, ErrPinnedMismatch},
+ {"pinned_key_mismatch", func(f *fixture) { f.marker.SnapshotTXID = 12; f.publish() }, ErrPinnedMismatch},
+ {"pinned_delete_marker", func(f *fixture) { f.marker.SnapshotVersion = f.b.BareDelete("data", f.pin.Key); f.publish() }, ErrDeleteMarker},
+ {"selected_delete_marker_on_open", func(f *fixture) {
+ f.store = altered{Reader: f.b, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
+ return nil, lease.ErrIsDeleteMarker
+ }}
+ }, ErrDeleteMarker},
+ {"marker_delete_marker", func(f *fixture) { f.b.BareDelete("data", f.req.Holder.Location+"/generation.json") }, ErrDeleteMarker},
+ {"marker_read_error", func(f *fixture) {
+ f.b.FailOn = func(op, key string) error {
+ if op == "get" {
+ return io.ErrUnexpectedEOF
+ }
+ return nil
+ }
+ }, ErrMarker},
+ {"marker_disappears", func(f *fixture) {
+ f.b.FailOn = func(op, key string) error {
+ if op == "get" {
+ return lease.ErrMirrorNotFound
+ }
+ return nil
+ }
+ }, ErrMarker},
+ {"marker_incarnation", func(f *fixture) { f.marker.Incarnation = "previous"; f.publish() }, ErrIncarnation},
+ {"marker_host", func(f *fixture) { f.marker.Host = "other"; f.publish() }, ErrMarker},
+ {"malformed_marker", func(f *fixture) { f.b.Put("data", f.req.Holder.Location+"/generation.json", []byte("{")) }, ErrMarker},
+ {"trailing_marker_json", func(f *fixture) {
+ b, _ := json.Marshal(f.marker)
+ f.b.Put("data", f.req.Holder.Location+"/generation.json", append(b, []byte(" {}")...))
+ }, ErrMarker},
+ {"empty_pin_version", func(f *fixture) { f.marker.SnapshotVersion = ""; f.publish() }, ErrVersionID},
+ {"null_pin_version", func(f *fixture) { f.marker.SnapshotVersion = "null"; f.publish() }, ErrVersionID},
+ {"empty_current_version", func(f *fixture) { alterVersion(f, "") }, ErrVersionID},
+ {"null_current_version", func(f *fixture) { alterVersion(f, "null") }, ErrVersionID},
+ {"tombstoned_unfinished_is_not_empty", func(f *fixture) {
+ f.chain("g0")
+ dead := f.file("g2", 0, 1, 1)
+ f.b.BareDelete("data", dead.Key)
+ }, ErrEmpty},
+ {"metadata_only_unfinished", func(f *fixture) {
+ f.chain("g0")
+ f.b.Put("data", f.req.Holder.Location+"/g-g2/build-intent.json", []byte("{}"))
+ }, ErrEmpty},
+ {"empty_published_generation", func(f *fixture) {
+ f.marker.Op = "empty"
+ f.marker.Pin = Pin{}
+ f.req.Holder.Generation = "empty"
+ f.publish()
+ }, ErrEmpty},
+ {"bad_successor_no_fallback", func(f *fixture) { r := f.file("g0", 9, 1, 20); f.b.Put("data", r.Key, header(f.t, 1, 19)) }, ErrRoot},
+ {"missing_snapshot_txid", func(f *fixture) {
+ b, _ := json.Marshal(f.marker)
+ var fields map[string]any
+ json.Unmarshal(b, &fields)
+ delete(fields, "snapshot_txid")
+ b, _ = json.Marshal(fields)
+ f.b.Put("data", f.req.Holder.Location+"/generation.json", b)
+ }, ErrMarker},
+ {"malformed_current_ltx", func(f *fixture) { f.b.Put("data", f.req.Holder.Location+"/g-g0/0000/not-a-range.ltx", []byte("bad")) }, ErrListing},
+ {"invalid_snapshot_min", func(f *fixture) { f.file("g0", 9, 2, 20) }, ErrRoot},
+ {"gap_without_root", func(f *fixture) { f.origin(); f.file("", 0, 2, 2) }, ErrGap},
+ {"location_escape", func(f *fixture) { f.req.Holder.Location = "instances/h/../other" }, ErrInput},
+ {"holder_root_sentinel", func(f *fixture) { f.req.Holder.Generation = hostclaim.RootGeneration }, ErrInput},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFixture(t)
+ tt.setup(f)
+ before := f.b.Snapshot()
+ got, err := Select(t.Context(), f.store, f.req)
+ if got != nil || !errors.Is(err, tt.want) {
+ t.Fatalf("got %+v / %v, want %v", got, err, tt.want)
+ }
+ if f.b.Snapshot() != before {
+ t.Fatal("mutated storage")
+ }
+ })
+ }
+}
+
+func TestContinuationRootRefusals(t *testing.T) {
+ for _, gen := range []string{"", "fresh"} {
+ name := gen
+ if name == "" {
+ name = "origin"
+ }
+ for level := 0; level < 9; level++ {
+ for _, corruption := range []string{"malformed_header", "truncated_file", "header_range_mismatch"} {
+ t.Run(fmt.Sprintf("%s/level_%d/%s", name, level, corruption), func(t *testing.T) {
+ f := newFixture(t)
+ if gen == "" {
+ f.origin()
+ } else {
+ f.marker.Op = gen
+ f.marker.Pin = Pin{}
+ f.req.Holder.Generation = gen
+ f.publish()
+ }
+ root := f.file(gen, level, 1, 1)
+ body := header(t, 1, 1)
+ switch corruption {
+ case "malformed_header":
+ body[0] ^= 0xff
+ case "truncated_file":
+ body = body[:len(body)-1]
+ case "header_range_mismatch":
+ body = header(t, 1, 2)
+ }
+ f.b.Put("data", root.Key, body)
+ before := f.b.Snapshot()
+ got, err := Select(t.Context(), f.store, f.req)
+ if got != nil || !errors.Is(err, ErrRoot) {
+ t.Fatalf("got %+v / %v, want nil / %v", got, err, ErrRoot)
+ }
+ if f.b.Snapshot() != before {
+ t.Fatal("selection mutated storage")
+ }
+ })
+ }
+ }
+ }
+}
+
+func TestMaxLevel0TXID(t *testing.T) {
+ tests := []struct {
+ name string
+ setup func(*fixture)
+ wantLevel0 uint64
+ wantRestored uint64
+ }{
+ {"current_anchor", func(f *fixture) {}, 11, 11},
+ {"current_continuation", func(f *fixture) {
+ f.file("g0", 0, 12, 12)
+ f.file("g0", 0, 13, 13)
+ }, 13, 13},
+ {"noncurrent_data", func(f *fixture) {
+ old := f.file("g0", 0, 30, 30)
+ f.b.BareDelete("data", old.Key)
+ }, 11, 11},
+ {"delete_marker_only", func(f *fixture) {
+ old := f.file("g0", 0, 40, 40)
+ f.b.BareDelete("data", old.Key)
+ if err := f.b.DeleteVersion(f.t.Context(), "data", old.Key, old.VersionID); err != nil {
+ f.t.Fatal(err)
+ }
+ }, 11, 11},
+ {"newer_restored_snapshot", func(f *fixture) {
+ f.file("g0", 9, 1, 30)
+ }, 11, 30},
+ {"no_current_level0", func(f *fixture) {
+ anchor := f.file("g0", 0, 11, 11)
+ f.b.BareDelete("data", anchor.Key)
+ }, 0, 11},
+ }
+ for _, entry := range []string{"ordinary", "prepared"} {
+ for _, tt := range tests {
+ t.Run(entry+"/"+tt.name, func(t *testing.T) {
+ f := newFixture(t)
+ tt.setup(f)
+ var got *Plan
+ var err error
+ if entry == "ordinary" {
+ got, err = Select(t.Context(), f.store, f.req)
+ } else {
+ p := f.marker.Pin
+ got, err = SelectPrepared(t.Context(), f.store, f.req, PreparedPin{
+ Generation: "g0", SnapshotKey: p.SnapshotKey, SnapshotVersion: p.SnapshotVersion,
+ SnapshotTXID: p.SnapshotTXID, SnapshotCRC64: p.SnapshotCRC64,
+ })
+ }
+ if err != nil || got == nil {
+ t.Fatalf("got %+v / %v, want a plan", got, err)
+ }
+ if got.MaxLevel0TXID != tt.wantLevel0 || got.MaxTXID != tt.wantRestored {
+ t.Fatalf("MaxLevel0TXID=%d, MaxTXID=%d; want %d, %d",
+ got.MaxLevel0TXID, got.MaxTXID, tt.wantLevel0, tt.wantRestored)
+ }
+ })
+ }
+ }
+}
+
+func alterVersion(f *fixture, id string) {
+ f.file("g0", 0, 12, 12)
+ f.store = altered{Reader: f.b, list: func(vs []lease.ObjectVersion) ([]lease.ObjectVersion, error) {
+ for i := range vs {
+ if strings.Contains(vs[i].Key, "0000/000000000000000c-") {
+ vs[i].VersionID = id
+ }
+ }
+ return vs, nil
+ }}
+}
+
+func TestSelectionDetails(t *testing.T) {
+ tests := []struct {
+ name string
+ setup func(*fixture) []File
+ }{
+ {"pinned_noncurrent_duplicate", func(f *fixture) []File {
+ f.b.Put("data", f.pin.Key, header(f.t, 1, 11))
+ return []File{f.pin}
+ }},
+ {"pinned_under_retention_tombstone", func(f *fixture) []File {
+ f.b.BareDelete("data", f.pin.Key)
+ return []File{f.pin}
+ }},
+ {"retention_tombstones_ignored", func(f *fixture) []File {
+ f.b.BareDelete("data", f.file("g0", 0, 13, 13).Key)
+ return []File{f.pin}
+ }},
+ {"max_then_min_then_level", func(f *fixture) []File {
+ f.file("g0", 8, 12, 19)
+ f.file("g0", 8, 12, 20)
+ f.file("g0", 1, 10, 20)
+ return []File{f.pin, f.file("g0", 2, 10, 20)}
+ }},
+ {"origin_snapshot_and_isolation", func(f *fixture) []File {
+ f.origin()
+ f.file("other", 9, 1, 999)
+ f.b.Put("data", f.req.Holder.Location+"/child/0009/0000000000000001-000000000000ffff.ltx", header(f.t, 1, 65535))
+ return []File{f.file("", 9, 1, 20), f.file("", 0, 21, 21)}
+ }},
+ {"legacy_location", func(f *fixture) []File {
+ f.req.Holder.Location = "instances/h"
+ f.req.Holder.Generation = ""
+ return []File{f.file("", 0, 1, 1)}
+ }},
+ {"fresh_zero_pin_marker", func(f *fixture) []File {
+ f.marker.Op = "fresh"
+ f.marker.Pin = Pin{}
+ f.req.Holder.Generation = "fresh"
+ f.publish()
+ return []File{f.file("fresh", 0, 1, 1)}
+ }},
+ {"warm_record_in_unfinished_run", func(f *fixture) []File {
+ f.chain("g0")
+ f.req.Holder.Admissions = append(f.req.Holder.Admissions, hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g2", Region: "ie", PredecessorRegion: "ie", Mode: "warm"})
+ return []File{f.pin}
+ }},
+ {"root_advance_past_missing_old_pin", func(f *fixture) []File {
+ if err := f.b.DeleteVersion(f.t.Context(), "data", f.pin.Key, f.pin.VersionID); err != nil {
+ f.t.Fatal(err)
+ }
+ return []File{f.file("g0", 9, 1, 30)}
+ }},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFixture(t)
+ want := tt.setup(f)
+ got, err := Select(t.Context(), f.store, f.req)
+ if err != nil || !reflect.DeepEqual(got.Files, want) {
+ t.Fatalf("got %+v / %v, want %+v", got, err, want)
+ }
+ })
+ }
+}
+
+// Port of resumeOpen's prepared branch (protocol_test.go:658) and
+// restorePinned/restoreWith (:183/:187), moves §5.1 / §7.14.
+func TestPreparedRestore(t *testing.T) {
+ tests := []struct {
+ name string
+ setup func(*fixture, *PreparedPin, File) []File
+ want error
+ ordinaryRefuses bool
+ }{
+ {"absent_target_marker_covered_source", func(f *fixture, p *PreparedPin, root File) []File {
+ f.origin()
+ f.req.Holder.Generation = "g2"
+ return []File{root, f.file("g2", 1, 29, 33)}
+ }, nil, true},
+ {"absent_target_marker_trimmed_source", func(f *fixture, p *PreparedPin, root File) []File {
+ f.origin()
+ f.req.Holder.Generation = "g2"
+ f.req.Holder.Admissions = f.req.Holder.Admissions[1:]
+ return []File{root}
+ }, nil, true},
+ {"unpublished_historical_pin", func(f *fixture, p *PreparedPin, root File) []File {
+ f.b.Put("data", root.Key, header(f.t, 1, 30))
+ return []File{root}
+ }, nil, false},
+ {"marker_pin_disagreement_ignored", func(f *fixture, p *PreparedPin, root File) []File {
+ f.marker.Op = "g2" // Its old g0 pin cannot override the authorized pin.
+ f.publish()
+ return []File{root}
+ }, nil, false},
+ {"malformed_marker_ignored", func(f *fixture, p *PreparedPin, root File) []File {
+ f.b.Put("data", f.req.Holder.Location+"/generation.json", []byte("{"))
+ return []File{root}
+ }, nil, false},
+ {"marker_tombstone_ignored", func(f *fixture, p *PreparedPin, root File) []File {
+ f.b.BareDelete("data", f.req.Holder.Location+"/generation.json")
+ return []File{root}
+ }, nil, false},
+ {"missing_prepared_root_no_current_key_fallback", func(f *fixture, p *PreparedPin, root File) []File {
+ f.b.Put("data", root.Key, header(f.t, 1, 30))
+ deletePreparedRoot(f, root)
+ return nil
+ }, ErrPinnedMissing, false},
+ {"empty_prepared_generation_never_older_baseline", func(f *fixture, p *PreparedPin, root File) []File {
+ f.chain("g0")
+ deletePreparedRoot(f, root)
+ versions, err := f.b.ListVersions(f.t.Context(), "data", prefix(f.req.Holder.Location+"/", "g2"))
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ for _, v := range versions {
+ if err := f.b.DeleteVersion(f.t.Context(), "data", v.Key, v.VersionID); err != nil {
+ f.t.Fatal(err)
+ }
+ }
+ baseline, err := Select(f.t.Context(), f.b, f.req)
+ if err != nil || baseline.Generation != "g0" {
+ f.t.Fatalf("ordinary baseline: %+v / %v", baseline, err)
+ }
+ return nil
+ }, ErrPinnedMissing, false},
+ {"zero_txid_prepared_pin_refuses", func(f *fixture, p *PreparedPin, root File) []File {
+ *p = PreparedPin{Generation: "empty"}
+ return nil
+ }, ErrPinnedMismatch, false},
+ {"prepared_pin_key_mismatch", func(f *fixture, p *PreparedPin, root File) []File {
+ p.SnapshotTXID++
+ return nil
+ }, ErrPinnedMismatch, false},
+ {"prepared_pin_other_generation", func(f *fixture, p *PreparedPin, root File) []File {
+ p.SnapshotKey = f.pin.Key
+ p.SnapshotVersion = f.pin.VersionID
+ p.SnapshotTXID = f.pin.MaxTXID
+ return nil
+ }, ErrPinnedMismatch, false},
+ {"prepared_pin_header_mismatch", func(f *fixture, p *PreparedPin, root File) []File {
+ p.SnapshotVersion = f.b.Put("data", root.Key, header(f.t, 1, 29))
+ return nil
+ }, ErrPinnedMismatch, false},
+ {"prepared_pin_empty_version", func(f *fixture, p *PreparedPin, root File) []File {
+ p.SnapshotVersion = ""
+ return nil
+ }, ErrVersionID, false},
+ {"prepared_pin_deleted_version", func(f *fixture, p *PreparedPin, root File) []File {
+ p.SnapshotVersion = f.b.BareDelete("data", root.Key)
+ return nil
+ }, ErrDeleteMarker, false},
+ {"prepared_root_advance_past_missing_pin", func(f *fixture, p *PreparedPin, root File) []File {
+ deletePreparedRoot(f, root)
+ f.file("other", 9, 1, 999)
+ return []File{f.file("g2", 9, 1, 40), f.file("g2", 0, 41, 41)}
+ }, nil, false},
+ {"prepared_bad_successor_no_fallback", func(f *fixture, p *PreparedPin, root File) []File {
+ r := f.file("g2", 9, 1, 40)
+ f.b.Put("data", r.Key, header(f.t, 1, 39))
+ return nil
+ }, ErrRoot, false},
+ {"prepared_incomplete_listing", func(f *fixture, p *PreparedPin, root File) []File {
+ f.b.FailListAfterPages = 1
+ return nil
+ }, ErrListing, false},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFixture(t)
+ f.req.Holder.Generation = "g2"
+ f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
+ {Generation: "source", Predecessor: "older", Region: "us", PredecessorRegion: "us", Mode: "cold", Covered: true},
+ {Generation: "g2", Predecessor: "source", Region: "ie", PredecessorRegion: "us", Mode: "cold"},
+ }
+ root := f.file("g2", 9, 1, 30)
+ f.file("g2", 0, 30, 30)
+ prepared := PreparedPin{Generation: "g2", SnapshotKey: root.Key, SnapshotVersion: root.VersionID, SnapshotTXID: 30, SnapshotCRC64: 123}
+ want := tt.setup(f, &prepared, root)
+ if tt.ordinaryRefuses {
+ got, err := Select(t.Context(), f.store, f.req)
+ if got != nil || !errors.Is(err, ErrDisagreement) {
+ t.Fatalf("ordinary selection: %+v / %v", got, err)
+ }
+ }
+ before := f.b.Snapshot()
+ holderBefore, _ := json.Marshal(f.req.Holder)
+ f.b.FailOn = func(op, key string) error {
+ if op != "list" && op != "get" {
+ t.Fatalf("mutation %s %s", op, key)
+ }
+ return nil
+ }
+ reader := preparedReader{Reader: f.store, t: t, base: prefix(f.req.Holder.Location+"/", prepared.Generation)}
+ got, err := SelectPrepared(t.Context(), reader, f.req, prepared)
+ holderAfter, _ := json.Marshal(f.req.Holder)
+ if f.b.Snapshot() != before || string(holderBefore) != string(holderAfter) {
+ t.Fatal("selection mutated inputs or storage")
+ }
+ if !errors.Is(err, tt.want) {
+ t.Fatalf("got %+v / %v, want %v", got, err, tt.want)
+ }
+ if tt.want != nil {
+ if got != nil {
+ t.Fatal("partial plan")
+ }
+ return
+ }
+ wantPin := Pin{SnapshotKey: prepared.SnapshotKey, SnapshotVersion: prepared.SnapshotVersion, SnapshotTXID: prepared.SnapshotTXID, SnapshotCRC64: prepared.SnapshotCRC64}
+ if got == nil || got.Generation != prepared.Generation || !reflect.DeepEqual(got.Files, want) ||
+ got.Root == nil || *got.Root != want[0] || got.MaxTXID != want[len(want)-1].MaxTXID ||
+ got.Marker != nil || got.MarkerVersion != "" || got.Pin == nil || *got.Pin != wantPin {
+ t.Fatalf("got %+v, want prepared %s %+v", got, prepared.Generation, want)
+ }
+ })
+ }
+}
+
+func deletePreparedRoot(f *fixture, root File) {
+ f.t.Helper()
+ if err := f.b.DeleteVersion(f.t.Context(), "data", root.Key, root.VersionID); err != nil {
+ f.t.Fatal(err)
+ }
+}
+
+type preparedReader struct {
+ Reader
+ t *testing.T
+ base string
+}
+
+func (s preparedReader) ListVersions(ctx context.Context, bucket, base string) ([]lease.ObjectVersion, error) {
+ if base != s.base {
+ s.t.Fatalf("prepared listing escaped generation: %s", base)
+ }
+ return s.Reader.ListVersions(ctx, bucket, base)
+}
+func (s preparedReader) GetVersion(context.Context, string, string, string) ([]byte, error) {
+ s.t.Fatal("prepared restore must not read a marker")
+ return nil, ErrMarker
+}
+func (s preparedReader) OpenVersion(ctx context.Context, bucket, key, id string) (io.ReadCloser, error) {
+ if !strings.HasPrefix(key, s.base) {
+ s.t.Fatalf("prepared open escaped generation: %s", key)
+ }
+ return s.Reader.OpenVersion(ctx, bucket, key, id)
+}