amadan

carlos / platform Public

lease: versioned storage primitives for the host lifecycle claim (Plan 2)

lease.VersionedStore: exact-version list, stream, copy, delete and
create-only put over explicitly region-bound S3 clients; never a bare-key
delete. leasetest: a versioned bucket model, an S3 wire server over it,
and a shared contract suite run against both. IAM: each box role lists,
reads and deletes object versions in its own data bucket only. Nothing
calls it in production yet.

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Squashed from feat/hostclaim-storage into main.
Co-Authored-By: claude <claude~paul@keymail.dev>
codex pushed by claude 3ca43d338643a8ca17290c6896e97e5cfc251cb3 parent b58e127
10 files changed, +1262 −0
  • infra/modules/region-host/main.tf +14 −0
  • infra/modules/region-remote-host/main.tf +14 −0
  • internal/lease/export_test.go +9 −0
  • internal/lease/leasetest/bucket.go +340 −0
  • internal/lease/leasetest/bucket_test.go +15 −0
  • internal/lease/leasetest/contract.go +245 −0
  • internal/lease/leasetest/s3server.go +222 −0
  • internal/lease/leasetest/s3server_test.go +38 −0
  • internal/lease/versioned.go +268 −0
  • internal/lease/versioned_test.go +97 −0

CI passed — run details

diff --git a/infra/modules/region-host/main.tf b/infra/modules/region-host/main.tf
index 14b7724a6..97a13e100 100644
--- a/infra/modules/region-host/main.tf
+++ b/infra/modules/region-host/main.tf
@@ -174,6 +174,20 @@ resource "aws_iam_role_policy" "host_buckets" {
Action = ["s3:ListBucket", "s3:GetBucketLocation"]
Resource = [var.deployment_bucket_arn, var.data_bucket_arn]
},
+ # Host lifecycle claim, Plan 2 (spec C3, §6.2, §6.5): exact-version
+ # list, read and delete on this box's OWN data bucket. Sibling-region
+ # moves/* grants are unchanged: the destination region retires its
+ # own staging (spec N13).
+ {
+ Effect = "Allow"
+ Action = ["s3:GetObjectVersion", "s3:DeleteObjectVersion"]
+ Resource = ["${var.data_bucket_arn}/*"]
+ },
+ {
+ Effect = "Allow"
+ Action = ["s3:ListBucketVersions"]
+ Resource = [var.data_bucket_arn]
+ },
{
Effect = "Allow"
Action = ["ssm:GetParameter"]
diff --git a/infra/modules/region-remote-host/main.tf b/infra/modules/region-remote-host/main.tf
index d18648e76..d86783977 100644
--- a/infra/modules/region-remote-host/main.tf
+++ b/infra/modules/region-remote-host/main.tf
@@ -91,6 +91,20 @@ resource "aws_iam_user_policy" "host" {
Action = ["s3:ListBucket", "s3:GetBucketLocation"]
Resource = [aws_s3_bucket.data.arn]
},
+ # Host lifecycle claim, Plan 2 (spec C3, §6.2, §6.5): exact-version
+ # list, read and delete on this box's OWN data bucket. Sibling-region
+ # moves/* grants are unchanged: the destination region retires its
+ # own staging (spec N13).
+ {
+ Effect = "Allow"
+ Action = ["s3:GetObjectVersion", "s3:DeleteObjectVersion"]
+ Resource = ["${aws_s3_bucket.data.arn}/*"]
+ },
+ {
+ Effect = "Allow"
+ Action = ["s3:ListBucketVersions"]
+ Resource = [aws_s3_bucket.data.arn]
+ },
],
[for arn in var.sibling_staging_arns : {
Effect = "Allow"
diff --git a/internal/lease/export_test.go b/internal/lease/export_test.go
new file mode 100644
index 000000000..c6c589e69
--- /dev/null
+++ b/internal/lease/export_test.go
@@ -0,0 +1,9 @@
+package lease
+
+import awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+
+// NewVersionedS3WithClients is the test-only constructor: tests bind
+// buckets to clients pointed at leasetest.NewS3Server.
+func NewVersionedS3WithClients(clients map[string]*awss3.Client) VersionedStore {
+ return &versionedS3{clients: clients}
+}
diff --git a/internal/lease/leasetest/bucket.go b/internal/lease/leasetest/bucket.go
new file mode 100644
index 000000000..fe13f4d98
--- /dev/null
+++ b/internal/lease/leasetest/bucket.go
@@ -0,0 +1,340 @@
+// Package leasetest holds test doubles for internal/lease. VersionedBucket
+// is the reference model of a versioned S3 bucket that every plan of the
+// host lifecycle claim tests against (spec §13).
+package leasetest
+
+import (
+ "bytes"
+ "context"
+ "crypto/md5"
+ "encoding/hex"
+ "fmt"
+ "io"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+type version struct {
+ id string
+ body []byte
+ etag string
+ deleteMarker bool
+ modified time.Time
+}
+
+// VersionedBucket is an in-memory, versioned, multi-bucket S3 model.
+type VersionedBucket struct {
+ mu sync.Mutex
+ buckets map[string]map[string][]version // bucket -> key -> versions, newest first
+ next int
+ clock time.Time
+
+ PageSize int // listing page size; default 1000
+ FailListAfterPages int // fail a listing once this many pages have been served; 0 never
+ FailOn func(op, key string) error // op: list|get|copy|delete|put
+}
+
+func NewVersionedBucket() *VersionedBucket {
+ return &VersionedBucket{buckets: map[string]map[string][]version{}, PageSize: 1000,
+ clock: time.Date(2026, 10, 5, 0, 0, 0, 0, time.UTC)}
+}
+
+func md5hex(b []byte) string { s := md5.Sum(b); return hex.EncodeToString(s[:]) }
+
+func (f *VersionedBucket) AddBucket(name string) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if f.buckets[name] == nil {
+ f.buckets[name] = map[string][]version{}
+ }
+}
+
+// --- locked helpers: callers hold f.mu ---
+
+func (f *VersionedBucket) fail(op, key string) error {
+ if f.FailOn != nil {
+ return f.FailOn(op, key)
+ }
+ return nil
+}
+
+func (f *VersionedBucket) known(bucket string) error {
+ if _, ok := f.buckets[bucket]; !ok {
+ return fmt.Errorf("leasetest: no bucket %q", bucket)
+ }
+ return nil
+}
+
+func (f *VersionedBucket) push(bucket, key string, v version) string {
+ f.next++
+ f.clock = f.clock.Add(time.Second)
+ v.id, v.modified = fmt.Sprintf("v%06d", f.next), f.clock
+ f.buckets[bucket][key] = append([]version{v}, f.buckets[bucket][key]...)
+ return v.id
+}
+
+func (f *VersionedBucket) hasCurrent(bucket, key string) bool {
+ vs := f.buckets[bucket][key]
+ return len(vs) > 0 && !vs[0].deleteMarker
+}
+
+func (f *VersionedBucket) find(bucket, key, id string) (version, bool) {
+ for _, v := range f.buckets[bucket][key] {
+ if v.id == id {
+ return v, true
+ }
+ }
+ return version{}, false
+}
+
+// listAll is the whole ordered listing under prefix (key order, newest
+// first within a key).
+func (f *VersionedBucket) listAll(bucket, prefix string) []lease.ObjectVersion {
+ o := f.buckets[bucket]
+ keys := make([]string, 0, len(o))
+ for k := range o {
+ if strings.HasPrefix(k, prefix) {
+ keys = append(keys, k)
+ }
+ }
+ sort.Strings(keys)
+ var all []lease.ObjectVersion
+ for _, k := range keys {
+ for i, v := range o[k] {
+ ov := lease.ObjectVersion{Key: k, VersionID: v.id, IsLatest: i == 0, DeleteMarker: v.deleteMarker, LastModified: v.modified}
+ if !v.deleteMarker {
+ ov.ETag, ov.Size = v.etag, int64(len(v.body))
+ }
+ all = append(all, ov)
+ }
+ }
+ return all
+}
+
+// pageOf returns page number page (0-based) of all, whether more follow,
+// and the failure FailListAfterPages injects.
+func (f *VersionedBucket) pageOf(all []lease.ObjectVersion, page int) ([]lease.ObjectVersion, bool, error) {
+ if f.FailListAfterPages > 0 && page >= f.FailListAfterPages {
+ return nil, false, fmt.Errorf("leasetest: injected listing failure on page %d", page+1)
+ }
+ size := f.PageSize
+ if size <= 0 {
+ size = 1000
+ }
+ start := page * size
+ if start > len(all) {
+ start = len(all)
+ }
+ end := start + size
+ if end >= len(all) {
+ return all[start:], false, nil
+ }
+ return all[start:end], true, nil
+}
+
+// --- exported helpers for tests ---
+
+func (f *VersionedBucket) Put(bucket, key string, body []byte) string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)})
+}
+
+func (f *VersionedBucket) PutMultipart(bucket, key string, body []byte) string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body) + "-2"})
+}
+
+func (f *VersionedBucket) BareDelete(bucket, key string) string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.push(bucket, key, version{deleteMarker: true})
+}
+
+func (f *VersionedBucket) LatestID(bucket, key string) string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if vs := f.buckets[bucket][key]; len(vs) > 0 {
+ return vs[0].id
+ }
+ return ""
+}
+
+func (f *VersionedBucket) Current(bucket, key string) ([]byte, bool) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ vs := f.buckets[bucket][key]
+ if len(vs) == 0 || vs[0].deleteMarker {
+ return nil, false
+ }
+ return vs[0].body, true
+}
+
+// Snapshot renders the whole state, for before/after comparison.
+func (f *VersionedBucket) Snapshot() string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ var b strings.Builder
+ names := make([]string, 0, len(f.buckets))
+ for n := range f.buckets {
+ names = append(names, n)
+ }
+ sort.Strings(names)
+ for _, n := range names {
+ for _, v := range f.listAll(n, "") {
+ fmt.Fprintf(&b, "%s/%s@%s:%v:%s\n", n, v.Key, v.VersionID, v.DeleteMarker, v.ETag)
+ }
+ }
+ return b.String()
+}
+
+// --- lease.VersionedStore ---
+
+func (f *VersionedBucket) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if err := f.known(bucket); err != nil {
+ return nil, err
+ }
+ if err := f.fail("list", prefix); err != nil {
+ return nil, err
+ }
+ all := f.listAll(bucket, prefix)
+ var out []lease.ObjectVersion
+ for page := 0; ; page++ {
+ vs, more, err := f.pageOf(all, page)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, vs...)
+ if !more {
+ return out, nil
+ }
+ }
+}
+
+func (f *VersionedBucket) read(bucket, key, versionID string) ([]byte, error) {
+ if versionID == "" {
+ return nil, lease.ErrNoVersionID
+ }
+ if err := f.known(bucket); err != nil {
+ return nil, err
+ }
+ if err := f.fail("get", key); err != nil {
+ return nil, err
+ }
+ v, ok := f.find(bucket, key, versionID)
+ if !ok {
+ return nil, lease.ErrMirrorNotFound
+ }
+ if v.deleteMarker {
+ return nil, lease.ErrIsDeleteMarker
+ }
+ return append([]byte(nil), v.body...), nil
+}
+
+func (f *VersionedBucket) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.read(bucket, key, versionID)
+}
+
+func (f *VersionedBucket) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ b, err := f.read(bucket, key, versionID)
+ if err != nil {
+ return nil, err
+ }
+ return io.NopCloser(bytes.NewReader(b)), nil
+}
+
+func (f *VersionedBucket) copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
+ if srcVersionID == "" {
+ return "", "", lease.ErrNoVersionID
+ }
+ if err := f.known(srcBucket); err != nil {
+ return "", "", err
+ }
+ if err := f.known(dstBucket); err != nil {
+ return "", "", err
+ }
+ if err := f.fail("copy", srcKey); err != nil {
+ return "", "", err
+ }
+ v, ok := f.find(srcBucket, srcKey, srcVersionID)
+ if !ok {
+ return "", "", lease.ErrMirrorNotFound
+ }
+ if v.deleteMarker {
+ return "", "", lease.ErrIsDeleteMarker
+ }
+ if createOnly && f.hasCurrent(dstBucket, dstKey) {
+ return "", "", lease.ErrDestinationExists
+ }
+ etag := md5hex(v.body) // a copy is single-part (§13 row 9)
+ return f.push(dstBucket, dstKey, version{body: append([]byte(nil), v.body...), etag: etag}), etag, nil
+}
+
+func (f *VersionedBucket) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey, createOnly)
+}
+
+func (f *VersionedBucket) deleteLocked(bucket, key, versionID string) error {
+ if versionID == "" {
+ return lease.ErrNoVersionID
+ }
+ if err := f.known(bucket); err != nil {
+ return err
+ }
+ if err := f.fail("delete", key); err != nil {
+ return err
+ }
+ vs := f.buckets[bucket][key]
+ for i, v := range vs {
+ if v.id == versionID {
+ rest := append(vs[:i:i], vs[i+1:]...)
+ if len(rest) == 0 {
+ delete(f.buckets[bucket], key)
+ } else {
+ f.buckets[bucket][key] = rest
+ }
+ return nil
+ }
+ }
+ return nil
+}
+
+func (f *VersionedBucket) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.deleteLocked(bucket, key, versionID)
+}
+
+func (f *VersionedBucket) putIfAbsentLocked(bucket, key string, body []byte) (string, error) {
+ if err := f.known(bucket); err != nil {
+ return "", err
+ }
+ if err := f.fail("put", key); err != nil {
+ return "", err
+ }
+ if f.hasCurrent(bucket, key) {
+ return "", lease.ErrExists
+ }
+ return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)}), nil
+}
+
+func (f *VersionedBucket) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.putIfAbsentLocked(bucket, key, body)
+}
+
+var _ lease.VersionedStore = (*VersionedBucket)(nil)
diff --git a/internal/lease/leasetest/bucket_test.go b/internal/lease/leasetest/bucket_test.go
new file mode 100644
index 000000000..d711a6f49
--- /dev/null
+++ b/internal/lease/leasetest/bucket_test.go
@@ -0,0 +1,15 @@
+package leasetest
+
+import (
+ "testing"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+func TestVersionedBucketContract(t *testing.T) {
+ RunVersionedStoreContract(t, func(t *testing.T) (lease.VersionedStore, *VersionedBucket, string, func() int) {
+ m := NewVersionedBucket()
+ m.AddBucket("data")
+ return m, m, "data", nil
+ })
+}
diff --git a/internal/lease/leasetest/contract.go b/internal/lease/leasetest/contract.go
new file mode 100644
index 000000000..d1a823408
--- /dev/null
+++ b/internal/lease/leasetest/contract.go
@@ -0,0 +1,245 @@
+package leasetest
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+// Maker builds a store under test, the model backing it, the bucket name,
+// and (for wire stores) a request counter.
+type Maker func(t *testing.T) (st lease.VersionedStore, m *VersionedBucket, bucket string, served func() int)
+
+// RunVersionedStoreContract is the behaviour every lease.VersionedStore
+// must have (spec §13, as covered by Plan 2).
+func RunVersionedStoreContract(t *testing.T, mk Maker) {
+ ctx := context.Background()
+
+ t.Run("listing: key order, latest first, markers included, prefix bounded", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ m.Put(b, "p/a", []byte("one"))
+ m.Put(b, "p/a", []byte("two"))
+ m.BareDelete(b, "p/a")
+ m.Put(b, "p/b", []byte("x"))
+ m.Put(b, "q/c", []byte("other prefix"))
+ vs, err := st.ListVersions(ctx, b, "p/")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(vs) != 4 || vs[0].Key != "p/a" || !vs[0].IsLatest || !vs[0].DeleteMarker || vs[3].Key != "p/b" || !vs[3].IsLatest {
+ t.Fatalf("listing = %+v", vs)
+ }
+ for _, v := range vs[1:3] {
+ if v.Key != "p/a" || v.IsLatest || v.DeleteMarker || v.VersionID == "" {
+ t.Fatalf("non-latest entries = %+v", vs[1:3])
+ }
+ }
+ })
+
+ t.Run("listing spans pages, including many versions of one key", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ m.PageSize = 3
+ for i := 0; i < 5; i++ {
+ m.Put(b, "p/same", []byte{byte(i)})
+ }
+ for i := 0; i < 4; i++ {
+ m.Put(b, fmt.Sprintf("p/k%02d", i), []byte{byte(i)})
+ }
+ vs, err := st.ListVersions(ctx, b, "p/")
+ if err != nil || len(vs) != 9 {
+ t.Fatalf("paged listing = %d, %v; want 9", len(vs), err)
+ }
+ seen := map[string]bool{}
+ for _, v := range vs {
+ if seen[v.Key+"@"+v.VersionID] {
+ t.Fatalf("duplicate %s@%s", v.Key, v.VersionID)
+ }
+ seen[v.Key+"@"+v.VersionID] = true
+ }
+ })
+
+ t.Run("listing fails whole on a page error", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ m.PageSize = 2
+ for i := 0; i < 6; i++ {
+ m.Put(b, fmt.Sprintf("p/k%d", i), []byte{byte(i)})
+ }
+ m.FailListAfterPages = 2
+ if vs, err := st.ListVersions(ctx, b, "p/"); err == nil || vs != nil {
+ t.Fatalf("got %d entries and error %v; want no entries and an error", len(vs), err)
+ }
+ })
+
+ t.Run("read and stream by exact version; markers and missing", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ k := "dir/a b+c%?d"
+ v1 := m.Put(b, k, []byte("old"))
+ m.Put(b, k, []byte("new"))
+ if body, err := st.GetVersion(ctx, b, k, v1); err != nil || string(body) != "old" {
+ t.Fatalf("GetVersion = %q, %v", body, err)
+ }
+ rc, err := st.OpenVersion(ctx, b, k, v1)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, _ := io.ReadAll(rc)
+ rc.Close()
+ if string(body) != "old" {
+ t.Fatalf("OpenVersion = %q", body)
+ }
+ mk1 := m.BareDelete(b, k)
+ if _, err := st.GetVersion(ctx, b, k, mk1); !errors.Is(err, lease.ErrIsDeleteMarker) {
+ t.Fatalf("GetVersion(marker) = %v, want ErrIsDeleteMarker", err)
+ }
+ if _, err := st.GetVersion(ctx, b, k, "v999999"); !errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Fatalf("GetVersion(missing) = %v, want ErrMirrorNotFound", err)
+ }
+ // A version hidden behind a marker is still readable by id.
+ if body, err := st.GetVersion(ctx, b, k, v1); err != nil || string(body) != "old" {
+ t.Fatalf("GetVersion behind a marker = %q, %v", body, err)
+ }
+ })
+
+ t.Run("copy by exact version; create-only; markers", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ k := "src/a b+c%?d"
+ v1 := m.Put(b, k, []byte("old"))
+ m.Put(b, k, []byte("new"))
+ dv, etag, err := st.CopyVersion(ctx, b, k, v1, b, "dst/x", true)
+ if err != nil || dv == "" || etag == "" || strings.Contains(etag, "-") {
+ t.Fatalf("CopyVersion = %q %q %v", dv, etag, err)
+ }
+ if cur, ok := m.Current(b, "dst/x"); !ok || string(cur) != "old" {
+ t.Fatalf("copied %q", cur)
+ }
+ if _, _, err := st.CopyVersion(ctx, b, k, v1, b, "dst/x", true); !errors.Is(err, lease.ErrDestinationExists) {
+ t.Fatalf("create-only copy onto existing = %v", err)
+ }
+ m.BareDelete(b, "dst/x")
+ if _, _, err := st.CopyVersion(ctx, b, k, v1, b, "dst/x", true); err != nil {
+ t.Fatalf("create-only copy over a delete marker = %v", err)
+ }
+ mid := m.BareDelete(b, k)
+ if _, _, err := st.CopyVersion(ctx, b, k, mid, b, "dst/y", false); !errors.Is(err, lease.ErrIsDeleteMarker) {
+ t.Fatalf("copy of a marker = %v, want ErrIsDeleteMarker", err)
+ }
+ mpID := m.PutMultipart(b, "src/mp", []byte("payload"))
+ if _, etag, err := st.CopyVersion(ctx, b, "src/mp", mpID, b, "dst/mp", false); err != nil || strings.Contains(etag, "-") {
+ t.Fatalf("multipart copy etag = %q, %v; want single-part", etag, err)
+ }
+ })
+
+ t.Run("delete by version: exact, idempotent, never bare", func(t *testing.T) {
+ st, m, b, served := mk(t)
+ v1 := m.Put(b, "k", []byte("one"))
+ v2 := m.Put(b, "k", []byte("two"))
+ if err := st.DeleteVersion(ctx, b, "k", v2); err != nil {
+ t.Fatal(err)
+ }
+ if cur, ok := m.Current(b, "k"); !ok || string(cur) != "one" {
+ t.Fatalf("after deleting the current version: %q %v; want the older one current", cur, ok)
+ }
+ if err := st.DeleteVersion(ctx, b, "k", v2); err != nil {
+ t.Fatalf("re-delete = %v", err)
+ }
+ if err := st.DeleteVersion(ctx, b, "never", "v000001"); err != nil {
+ t.Fatalf("delete on a missing key = %v", err)
+ }
+ before, reqs := m.Snapshot(), 0
+ if served != nil {
+ reqs = served()
+ }
+ if err := st.DeleteVersion(ctx, b, "k", ""); !errors.Is(err, lease.ErrNoVersionID) {
+ t.Fatalf("empty version id = %v, want ErrNoVersionID", err)
+ }
+ if _, err := st.GetVersion(ctx, b, "k", ""); !errors.Is(err, lease.ErrNoVersionID) {
+ t.Fatalf("GetVersion(\"\") = %v, want ErrNoVersionID", err)
+ }
+ if _, err := st.OpenVersion(ctx, b, "k", ""); !errors.Is(err, lease.ErrNoVersionID) {
+ t.Fatalf("OpenVersion(\"\") = %v, want ErrNoVersionID", err)
+ }
+ if _, _, err := st.CopyVersion(ctx, b, "k", "", b, "k2", false); !errors.Is(err, lease.ErrNoVersionID) {
+ t.Fatalf("CopyVersion(\"\") = %v, want ErrNoVersionID", err)
+ }
+ if m.Snapshot() != before {
+ t.Fatal("state changed on an empty-version delete")
+ }
+ if served != nil && served() != reqs {
+ t.Fatal("a request was sent for an empty version id")
+ }
+ _ = v1
+ })
+
+ t.Run("delete a noncurrent version keeps the current one", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ v1 := m.Put(b, "k", []byte("one"))
+ v2 := m.Put(b, "k", []byte("two"))
+ if err := st.DeleteVersion(ctx, b, "k", v1); err != nil {
+ t.Fatal(err)
+ }
+ if cur, ok := m.Current(b, "k"); !ok || string(cur) != "two" || m.LatestID(b, "k") != v2 {
+ t.Fatalf("current after a noncurrent delete = %q %v %s", cur, ok, m.LatestID(b, "k"))
+ }
+ if _, err := st.GetVersion(ctx, b, "k", v1); !errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Fatalf("deleted version still readable: %v", err)
+ }
+ })
+
+ t.Run("a missing bucket is an error, never absence", func(t *testing.T) {
+ st, _, _, _ := mk(t)
+ if err := st.DeleteVersion(ctx, "nobucket", "k", "v000001"); err == nil {
+ t.Error("delete in a missing bucket succeeded")
+ }
+ if _, err := st.GetVersion(ctx, "nobucket", "k", "v000001"); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Errorf("get in a missing bucket = %v", err)
+ }
+ if _, err := st.ListVersions(ctx, "nobucket", ""); err == nil {
+ t.Error("list of a missing bucket succeeded")
+ }
+ })
+
+ t.Run("put if absent", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ if v, err := st.PutIfAbsent(ctx, b, "m/closed.json", []byte("{}")); err != nil || v == "" {
+ t.Fatalf("first = %q %v", v, err)
+ }
+ if _, err := st.PutIfAbsent(ctx, b, "m/closed.json", []byte("{}")); !errors.Is(err, lease.ErrExists) {
+ t.Fatalf("second = %v, want ErrExists", err)
+ }
+ m.BareDelete(b, "m/closed.json")
+ if _, err := st.PutIfAbsent(ctx, b, "m/closed.json", []byte("{}")); err != nil {
+ t.Fatalf("over a marker = %v", err)
+ }
+ })
+
+ t.Run("injected failures surface as errors from every operation", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ v := m.Put(b, "k", []byte("x"))
+ boom := errors.New("boom")
+ m.FailOn = func(op, key string) error { return boom }
+ if _, err := st.ListVersions(ctx, b, ""); err == nil {
+ t.Error("list: no error")
+ }
+ if _, err := st.GetVersion(ctx, b, "k", v); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Errorf("get: %v", err)
+ }
+ if _, _, err := st.CopyVersion(ctx, b, "k", v, b, "k2", false); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Errorf("copy: %v", err)
+ }
+ if err := st.DeleteVersion(ctx, b, "k", v); err == nil {
+ t.Error("delete: no error")
+ }
+ if _, err := st.PutIfAbsent(ctx, b, "new", []byte("x")); err == nil {
+ t.Error("put: no error")
+ }
+ m.FailOn = nil
+ if _, ok := m.Current(b, "k"); !ok {
+ t.Fatal("a failed delete removed data")
+ }
+ })
+}
diff --git a/internal/lease/leasetest/s3server.go b/internal/lease/leasetest/s3server.go
new file mode 100644
index 000000000..f0877aeb5
--- /dev/null
+++ b/internal/lease/leasetest/s3server.go
@@ -0,0 +1,222 @@
+package leasetest
+
+import (
+ "encoding/xml"
+ "errors"
+ "io"
+ "net/http"
+ "net/url"
+ "strings"
+ "sync/atomic"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+type xmlVersion struct {
+ XMLName xml.Name `xml:"Version"`
+ Key string
+ VersionId string
+ IsLatest bool
+ LastModified string
+ ETag string
+ Size int64
+}
+
+type xmlMarker struct {
+ XMLName xml.Name `xml:"DeleteMarker"`
+ Key string
+ VersionId string
+ IsLatest bool
+ LastModified string
+}
+
+type listVersionsResult struct {
+ XMLName xml.Name `xml:"ListVersionsResult"`
+ Name string
+ Prefix string
+ KeyMarker string
+ VersionIdMarker string
+ NextKeyMarker string `xml:",omitempty"`
+ NextVersionIdMarker string `xml:",omitempty"`
+ MaxKeys int
+ IsTruncated bool
+ Entries []any
+}
+
+func s3err(w http.ResponseWriter, status int, code string) { s3errMsg(w, status, code, code) }
+
+func s3errMsg(w http.ResponseWriter, status int, code, msg string) {
+ w.Header().Set("Content-Type", "application/xml")
+ w.WriteHeader(status)
+ io.WriteString(w, "<Error><Code>"+code+"</Code><Message>"+msg+"</Message></Error>")
+}
+
+// copyMarkerMsg is S3's answer to a COPY naming a delete marker's version.
+const copyMarkerMsg = "The source of a copy request may not specifically refer to a delete marker by version id."
+
+// writeErr maps a model error onto the S3 response real S3 gives.
+func writeErr(w http.ResponseWriter, err error) {
+ switch {
+ case errors.Is(err, lease.ErrMirrorNotFound):
+ s3err(w, http.StatusNotFound, "NoSuchVersion")
+ case errors.Is(err, lease.ErrIsDeleteMarker):
+ w.Header().Set("x-amz-delete-marker", "true")
+ s3err(w, http.StatusMethodNotAllowed, "MethodNotAllowed")
+ case errors.Is(err, lease.ErrDestinationExists), errors.Is(err, lease.ErrExists):
+ s3err(w, http.StatusPreconditionFailed, "PreconditionFailed")
+ default:
+ s3err(w, http.StatusInternalServerError, "InternalError")
+ }
+}
+
+// NewS3Server speaks the subset of the S3 wire protocol VersionedStore uses,
+// path-style, over f, holding f.mu for each request. The second result
+// counts requests served. With serves non-empty, the endpoint answers only
+// for those request buckets (a regional endpoint); any other bucket gets
+// 301 PermanentRedirect, as S3 answers a wrong-region request. A copy's
+// source bucket may be any bucket in f.
+func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int) {
+ var n atomic.Int64
+ allowed := map[string]bool{}
+ for _, b := range serves {
+ allowed[b] = true
+ }
+ h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ n.Add(1)
+ bucket, rawKey, _ := strings.Cut(strings.TrimPrefix(r.URL.EscapedPath(), "/"), "/")
+ key, err := url.PathUnescape(rawKey)
+ if err != nil {
+ s3err(w, http.StatusBadRequest, "InvalidURI")
+ return
+ }
+ q := r.URL.Query()
+ if len(allowed) > 0 && !allowed[bucket] {
+ s3err(w, http.StatusMovedPermanently, "PermanentRedirect")
+ return
+ }
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if err := f.known(bucket); err != nil {
+ s3err(w, http.StatusNotFound, "NoSuchBucket")
+ return
+ }
+ switch {
+ case r.Method == http.MethodGet && key == "" && q.Has("versions"):
+ f.serveList(w, bucket, q)
+ case r.Method == http.MethodGet && key != "":
+ body, err := f.read(bucket, key, q.Get("versionId"))
+ if err != nil {
+ writeErr(w, err)
+ return
+ }
+ w.Write(body)
+ case r.Method == http.MethodPut && r.Header.Get("x-amz-copy-source") != "":
+ f.serveCopy(w, r, bucket, key)
+ case r.Method == http.MethodPut:
+ body, err := io.ReadAll(r.Body)
+ if err != nil {
+ s3err(w, http.StatusBadRequest, "IncompleteBody") // a failed upload stores nothing
+ return
+ }
+ if r.Header.Get("If-None-Match") != "*" {
+ s3err(w, http.StatusNotImplemented, "NotImplemented") // VersionedStore only puts create-only
+ return
+ }
+ id, err := f.putIfAbsentLocked(bucket, key, body)
+ if err != nil {
+ writeErr(w, err)
+ return
+ }
+ w.Header().Set("x-amz-version-id", id)
+ w.Header().Set("ETag", `"`+md5hex(body)+`"`)
+ case r.Method == http.MethodDelete && key != "" && q.Get("versionId") != "":
+ if err := f.deleteLocked(bucket, key, q.Get("versionId")); err != nil {
+ writeErr(w, err)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+ default:
+ s3err(w, http.StatusMethodNotAllowed, "MethodNotAllowed")
+ }
+ })
+ return h, func() int { return int(n.Load()) }
+}
+
+func (f *VersionedBucket) serveList(w http.ResponseWriter, bucket string, q url.Values) {
+ if err := f.fail("list", q.Get("prefix")); err != nil {
+ writeErr(w, err)
+ return
+ }
+ all := f.listAll(bucket, q.Get("prefix"))
+ km, vm := q.Get("key-marker"), q.Get("version-id-marker")
+ start := 0
+ if km != "" {
+ start = len(all) // a marker not found resumes at the end
+ for i, v := range all {
+ if v.Key == km && v.VersionID == vm {
+ start = i + 1
+ break
+ }
+ }
+ }
+ size := f.PageSize
+ if size <= 0 {
+ size = 1000
+ }
+ page := (start + size - 1) / size // pages already served before this one
+ vs, more, err := f.pageOf(all[start:], 0)
+ if f.FailListAfterPages > 0 && page >= f.FailListAfterPages {
+ err = errors.New("leasetest: injected listing failure")
+ }
+ if err != nil {
+ writeErr(w, err)
+ return
+ }
+ res := listVersionsResult{Name: bucket, Prefix: q.Get("prefix"), KeyMarker: km, VersionIdMarker: vm, MaxKeys: size, IsTruncated: more}
+ for _, v := range vs {
+ lm := v.LastModified.UTC().Format(time.RFC3339)
+ if v.DeleteMarker {
+ res.Entries = append(res.Entries, xmlMarker{Key: v.Key, VersionId: v.VersionID, IsLatest: v.IsLatest, LastModified: lm})
+ } else {
+ res.Entries = append(res.Entries, xmlVersion{Key: v.Key, VersionId: v.VersionID, IsLatest: v.IsLatest, LastModified: lm, ETag: `"` + v.ETag + `"`, Size: v.Size})
+ }
+ }
+ if more && len(vs) > 0 {
+ res.NextKeyMarker, res.NextVersionIdMarker = vs[len(vs)-1].Key, vs[len(vs)-1].VersionID
+ }
+ w.Header().Set("Content-Type", "application/xml")
+ xml.NewEncoder(w).Encode(res)
+}
+
+func (f *VersionedBucket) serveCopy(w http.ResponseWriter, r *http.Request, bucket, key string) {
+ src := strings.TrimPrefix(r.Header.Get("x-amz-copy-source"), "/")
+ srcPath, rawQ, _ := strings.Cut(src, "?")
+ if strings.ContainsAny(srcPath, "+? ") {
+ s3err(w, http.StatusBadRequest, "InvalidArgument") // AWS's encoder escapes these; a raw one is a client bug
+ return
+ }
+ sq, err := url.ParseQuery(rawQ)
+ if err != nil {
+ s3err(w, http.StatusBadRequest, "InvalidArgument")
+ return
+ }
+ unesc, err := url.PathUnescape(srcPath)
+ if err != nil {
+ s3err(w, http.StatusBadRequest, "InvalidArgument")
+ return
+ }
+ sb, sk, _ := strings.Cut(unesc, "/")
+ id, etag, err := f.copyLocked(sb, sk, sq.Get("versionId"), bucket, key, r.Header.Get("If-None-Match") == "*")
+ if errors.Is(err, lease.ErrIsDeleteMarker) {
+ s3errMsg(w, http.StatusBadRequest, "InvalidRequest", copyMarkerMsg) // a COPY is 400, not GET's 405
+ return
+ }
+ if err != nil {
+ writeErr(w, err)
+ return
+ }
+ w.Header().Set("x-amz-version-id", id)
+ w.Header().Set("Content-Type", "application/xml")
+ io.WriteString(w, `<CopyObjectResult><ETag>"`+etag+`"</ETag><LastModified>2026-10-05T00:00:00Z</LastModified></CopyObjectResult>`)
+}
diff --git a/internal/lease/leasetest/s3server_test.go b/internal/lease/leasetest/s3server_test.go
new file mode 100644
index 000000000..7ea3b6785
--- /dev/null
+++ b/internal/lease/leasetest/s3server_test.go
@@ -0,0 +1,38 @@
+package leasetest
+
+import (
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+)
+
+// failingReader yields some bytes and then an error, like a dropped upload.
+type failingReader struct{ r io.Reader }
+
+func (f *failingReader) Read(p []byte) (int, error) {
+ n, err := f.r.Read(p)
+ if err == io.EOF {
+ return n, errors.New("connection reset")
+ }
+ return n, err
+}
+
+func TestS3ServerFailedUploadStoresNothing(t *testing.T) {
+ m := NewVersionedBucket()
+ m.AddBucket("data")
+ h, _ := NewS3Server(m)
+ before := m.Snapshot()
+ req := httptest.NewRequest(http.MethodPut, "/data/k", &failingReader{strings.NewReader("partial")})
+ req.Header.Set("If-None-Match", "*")
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, req)
+ if rec.Code < 400 {
+ t.Fatalf("failed upload answered %d", rec.Code)
+ }
+ if m.Snapshot() != before {
+ t.Fatalf("failed upload changed the bucket:\n%s", m.Snapshot())
+ }
+}
diff --git a/internal/lease/versioned.go b/internal/lease/versioned.go
new file mode 100644
index 000000000..54377d71b
--- /dev/null
+++ b/internal/lease/versioned.go
@@ -0,0 +1,268 @@
+package lease
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ "github.com/aws/smithy-go"
+ "github.com/aws/smithy-go/encoding/httpbinding"
+ smithyhttp "github.com/aws/smithy-go/transport/http"
+)
+
+// ObjectVersion is one entry of a versioned listing: an object version, or
+// a delete marker (DeleteMarker true, no ETag or Size).
+type ObjectVersion struct {
+ Key string
+ VersionID string
+ ETag string // unquoted
+ Size int64
+ IsLatest bool
+ DeleteMarker bool
+ LastModified time.Time
+}
+
+var (
+ ErrExists = errors.New("lease: object already exists")
+ ErrDestinationExists = errors.New("lease: copy destination already exists")
+ ErrIsDeleteMarker = errors.New("lease: version is a delete marker")
+ ErrNoVersionID = errors.New("lease: a version id is required")
+)
+
+// VersionedStore is the data bucket addressed by exact object version (host
+// lifecycle claim spec C3, §6.2, §6.5). No method deletes or overwrites by
+// bare key.
+type VersionedStore interface {
+ // ListVersions returns every version and delete marker under prefix:
+ // key order; within a key the IsLatest entry first, the rest newest
+ // first with ties unordered. Any page error fails the whole call.
+ ListVersions(ctx context.Context, bucket, prefix string) ([]ObjectVersion, error)
+ // OpenVersion streams one exact version. Missing: ErrMirrorNotFound.
+ // A delete marker: ErrIsDeleteMarker.
+ OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error)
+ // GetVersion reads one exact version whole (OpenVersion's errors).
+ GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error)
+ // CopyVersion copies one exact source version to dstKey. createOnly
+ // adds If-None-Match: *; an existing current destination object gives
+ // ErrDestinationExists. It returns the destination's new version id
+ // and ETag.
+ CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (dstVersionID, dstETag string, err error)
+ // DeleteVersion removes one exact version; gone already is success.
+ // An empty versionID is ErrNoVersionID, without any request.
+ DeleteVersion(ctx context.Context, bucket, key, versionID string) error
+ // PutIfAbsent writes body only if key holds no current object version.
+ PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (versionID string, err error)
+}
+
+// versionedS3 binds every bucket to its own regional client explicitly.
+// An unbound bucket is an error, never a guess.
+type versionedS3 struct {
+ clients map[string]*awss3.Client // bucket -> client for its region
+}
+
+// NewVersionedS3 builds one client per region from the default credential
+// chain (with s3compat options, as newLeaserClient does) and binds each
+// bucket to its region's client.
+func NewVersionedS3(ctx context.Context, regionByBucket map[string]string) (VersionedStore, error) {
+ byRegion := map[string]*awss3.Client{}
+ out := &versionedS3{clients: map[string]*awss3.Client{}}
+ for bucket, region := range regionByBucket {
+ c, ok := byRegion[region]
+ if !ok {
+ var err error
+ if c, _, err = newLeaserClient(ctx, region); err != nil {
+ return nil, fmt.Errorf("lease: versioned client for %s: %w", region, err)
+ }
+ byRegion[region] = c
+ }
+ out.clients[bucket] = c
+ }
+ return out, nil
+}
+
+func (s *versionedS3) client(bucket string) (*awss3.Client, error) {
+ c, ok := s.clients[bucket]
+ if !ok {
+ return nil, fmt.Errorf("lease: bucket %q is not bound to a region", bucket)
+ }
+ return c, nil
+}
+
+func (s *versionedS3) ListVersions(ctx context.Context, bucket, prefix string) ([]ObjectVersion, error) {
+ c, err := s.client(bucket)
+ if err != nil {
+ return nil, err
+ }
+ p := awss3.NewListObjectVersionsPaginator(c, &awss3.ListObjectVersionsInput{
+ Bucket: aws.String(bucket), Prefix: aws.String(prefix),
+ }) // StopOnDuplicateToken stays off: a key legitimately repeats across version pages
+ var out []ObjectVersion
+ for p.HasMorePages() {
+ page, err := p.NextPage(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("lease: list versions %s/%s: %w", bucket, prefix, err)
+ }
+ for _, v := range page.Versions {
+ out = append(out, ObjectVersion{Key: aws.ToString(v.Key), VersionID: aws.ToString(v.VersionId),
+ ETag: strings.Trim(aws.ToString(v.ETag), `"`), Size: aws.ToInt64(v.Size),
+ IsLatest: aws.ToBool(v.IsLatest), LastModified: aws.ToTime(v.LastModified)})
+ }
+ for _, d := range page.DeleteMarkers {
+ out = append(out, ObjectVersion{Key: aws.ToString(d.Key), VersionID: aws.ToString(d.VersionId),
+ IsLatest: aws.ToBool(d.IsLatest), DeleteMarker: true, LastModified: aws.ToTime(d.LastModified)})
+ }
+ }
+ sort.SliceStable(out, func(i, j int) bool {
+ if out[i].Key != out[j].Key {
+ return out[i].Key < out[j].Key
+ }
+ if out[i].IsLatest != out[j].IsLatest {
+ return out[i].IsLatest
+ }
+ return out[i].LastModified.After(out[j].LastModified) // ties: unspecified (Global Constraints)
+ })
+ return out, nil
+}
+
+// apiCode returns the smithy error code and HTTP status of err.
+func apiCode(err error) (string, int) {
+ var ae smithy.APIError
+ code := ""
+ if errors.As(err, &ae) {
+ code = ae.ErrorCode()
+ }
+ var re *smithyhttp.ResponseError
+ status := 0
+ if errors.As(err, &re) {
+ status = re.HTTPStatusCode()
+ }
+ return code, status
+}
+
+// mapReadErr normalizes only object or version absence. A missing bucket
+// (NoSuchBucket, also a 404) stays an error: it is never "already gone".
+func mapReadErr(err error) error {
+ code, status := apiCode(err)
+ switch {
+ case code == "NoSuchBucket":
+ return err
+ case code == "NoSuchKey" || code == "NoSuchVersion" || code == "NotFound" || status == http.StatusNotFound:
+ return ErrMirrorNotFound
+ case status == http.StatusMethodNotAllowed:
+ return ErrIsDeleteMarker // S3 answers 405 for a delete marker's version id
+ }
+ return err
+}
+
+func (s *versionedS3) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
+ if versionID == "" {
+ return nil, ErrNoVersionID
+ }
+ c, err := s.client(bucket)
+ if err != nil {
+ return nil, err
+ }
+ out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID)})
+ if err != nil {
+ if m := mapReadErr(err); m != err {
+ return nil, m
+ }
+ return nil, fmt.Errorf("lease: get %s/%s@%s: %w", bucket, key, versionID, err)
+ }
+ return out.Body, nil
+}
+
+func (s *versionedS3) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
+ rc, err := s.OpenVersion(ctx, bucket, key, versionID)
+ if err != nil {
+ return nil, err
+ }
+ defer rc.Close()
+ return io.ReadAll(rc)
+}
+
+func (s *versionedS3) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
+ if srcVersionID == "" {
+ return "", "", ErrNoVersionID
+ }
+ if _, err := s.client(srcBucket); err != nil {
+ return "", "", err
+ }
+ c, err := s.client(dstBucket) // a copy executes in the destination's region
+ if err != nil {
+ return "", "", err
+ }
+ in := &awss3.CopyObjectInput{
+ Bucket: aws.String(dstBucket),
+ Key: aws.String(dstKey),
+ CopySource: aws.String(httpbinding.EscapePath(srcBucket+"/"+srcKey, false) + "?versionId=" + url.QueryEscape(srcVersionID)),
+ }
+ if createOnly {
+ in.IfNoneMatch = aws.String("*")
+ }
+ out, err := c.CopyObject(ctx, in)
+ if err != nil {
+ code, status := apiCode(err)
+ var ae smithy.APIError
+ switch {
+ case createOnly && (code == "PreconditionFailed" || status == http.StatusPreconditionFailed):
+ return "", "", ErrDestinationExists
+ case code == "InvalidRequest" && errors.As(err, &ae) && strings.Contains(strings.ToLower(ae.ErrorMessage()), "delete marker"):
+ return "", "", ErrIsDeleteMarker
+ default:
+ if m := mapReadErr(err); m != err {
+ return "", "", m
+ }
+ }
+ return "", "", fmt.Errorf("lease: copy %s/%s@%s -> %s/%s: %w", srcBucket, srcKey, srcVersionID, dstBucket, dstKey, err)
+ }
+ etag := ""
+ if out.CopyObjectResult != nil {
+ etag = strings.Trim(aws.ToString(out.CopyObjectResult.ETag), `"`)
+ }
+ return aws.ToString(out.VersionId), etag, nil
+}
+
+func (s *versionedS3) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
+ if versionID == "" {
+ return ErrNoVersionID // never a bare-key delete (spec C3), and no request is sent
+ }
+ c, err := s.client(bucket)
+ if err != nil {
+ return err
+ }
+ if _, err := c.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID)}); err != nil {
+ if mapReadErr(err) == ErrMirrorNotFound {
+ return nil // already gone
+ }
+ return fmt.Errorf("lease: delete %s/%s@%s: %w", bucket, key, versionID, err)
+ }
+ return nil
+}
+
+func (s *versionedS3) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ c, err := s.client(bucket)
+ if err != nil {
+ return "", err
+ }
+ out, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key),
+ Body: bytes.NewReader(body), 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)
diff --git a/internal/lease/versioned_test.go b/internal/lease/versioned_test.go
new file mode 100644
index 000000000..a7d415af3
--- /dev/null
+++ b/internal/lease/versioned_test.go
@@ -0,0 +1,97 @@
+package lease_test
+
+import (
+ "context"
+ "net/http/httptest"
+ "testing"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/credentials"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+)
+
+func wireClient(t *testing.T, url string) *awss3.Client {
+ t.Helper()
+ return awss3.New(awss3.Options{
+ Region: "eu-west-1", BaseEndpoint: aws.String(url), UsePathStyle: true,
+ Credentials: credentials.NewStaticCredentialsProvider("k", "s", ""),
+ RetryMaxAttempts: 1,
+ })
+}
+
+// wireStore is a versionedS3 whose buckets "data" and "other" are served
+// by two separate servers over one model, so a wrong-region routing is a
+// visible failure.
+func wireStore(t *testing.T) (lease.VersionedStore, *leasetest.VersionedBucket, func() int) {
+ t.Helper()
+ m := leasetest.NewVersionedBucket()
+ m.AddBucket("data")
+ m.AddBucket("other")
+ h1, n1 := leasetest.NewS3Server(m, "data", "nobucket")
+ h2, n2 := leasetest.NewS3Server(m, "other")
+ s1, s2 := httptest.NewServer(h1), httptest.NewServer(h2)
+ t.Cleanup(s1.Close)
+ t.Cleanup(s2.Close)
+ // "nobucket" is bound but absent from the model: the server answers
+ // NoSuchBucket, for the missing-bucket contract case.
+ st := lease.NewVersionedS3WithClients(map[string]*awss3.Client{
+ "data": wireClient(t, s1.URL), "other": wireClient(t, s2.URL),
+ "nobucket": wireClient(t, s1.URL),
+ })
+ return st, m, func() int { return n1() + n2() }
+}
+
+func TestVersionedS3Contract(t *testing.T) {
+ leasetest.RunVersionedStoreContract(t, func(t *testing.T) (lease.VersionedStore, *leasetest.VersionedBucket, string, func() int) {
+ st, m, served := wireStore(t)
+ return st, m, "data", served
+ })
+}
+
+// Each bucket reaches only its own endpoint: a client bound to the wrong
+// endpoint gets 301 and the operation fails.
+func TestVersionedS3RoutesEachBucketToItsEndpoint(t *testing.T) {
+ ctx := context.Background()
+ st, m, _ := wireStore(t)
+ for _, b := range []string{"data", "other"} {
+ v := m.Put(b, "k", []byte(b))
+ if vs, err := st.ListVersions(ctx, b, ""); err != nil || len(vs) != 1 {
+ t.Fatalf("list %s = %v, %v", b, vs, err)
+ }
+ if err := st.DeleteVersion(ctx, b, "k", v); err != nil {
+ t.Fatalf("delete %s = %v", b, err)
+ }
+ }
+ m2 := leasetest.NewVersionedBucket()
+ m2.AddBucket("data")
+ h, _ := leasetest.NewS3Server(m2, "other")
+ srv := httptest.NewServer(h)
+ t.Cleanup(srv.Close)
+ wrong := lease.NewVersionedS3WithClients(map[string]*awss3.Client{"data": wireClient(t, srv.URL)})
+ if _, err := wrong.ListVersions(ctx, "data", ""); err == nil {
+ t.Fatal("a wrong-region endpoint answered")
+ }
+}
+
+func TestVersionedS3RefusesUnboundBucket(t *testing.T) {
+ st, _, _ := wireStore(t)
+ if _, err := st.ListVersions(context.Background(), "unbound", ""); err == nil {
+ t.Fatal("listing an unbound bucket did not fail")
+ }
+}
+
+func TestVersionedS3CopyEscapesLikeAWS(t *testing.T) {
+ st, m, _ := wireStore(t)
+ key := "src/a b+c%?d"
+ v1 := m.Put("data", key, []byte("old"))
+ m.Put("data", key, []byte("new"))
+ if _, _, err := st.CopyVersion(context.Background(), "data", key, v1, "other", "dst/k", true); err != nil {
+ t.Fatal(err)
+ }
+ if cur, ok := m.Current("other", "dst/k"); !ok || string(cur) != "old" {
+ t.Fatalf("copied %q, want the pinned old version", cur)
+ }
+}