amadan

carlos / platform Public

lease: VersionedStore over region-bound S3 clients, ListVersions over the wire (claim Plan 2 Tasks 1 and 3)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude aaf519df2f0a6ab602b78a3e7536e544c03d9542 parent 24d3a71
4 files changed, +458 −0
  • internal/lease/export_test.go +9 −0
  • internal/lease/leasetest/s3server.go +218 −0
  • internal/lease/versioned.go +147 −0
  • internal/lease/versioned_test.go +84 −0
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/s3server.go b/internal/lease/leasetest/s3server.go
new file mode 100644
index 000000000..8a2cefd62
--- /dev/null
+++ b/internal/lease/leasetest/s3server.go
@@ -0,0 +1,218 @@
+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, _ := io.ReadAll(r.Body)
+ 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/versioned.go b/internal/lease/versioned.go
new file mode 100644
index 000000000..09557c517
--- /dev/null
+++ b/internal/lease/versioned.go
@@ -0,0 +1,147 @@
+package lease
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+// 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
+}
+
+var errNotYet = errors.New("lease: not implemented yet")
+
+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
+}
+func (s *versionedS3) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
+ return nil, errNotYet
+}
+func (s *versionedS3) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
+ return nil, errNotYet
+}
+func (s *versionedS3) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
+ return "", "", errNotYet
+}
+func (s *versionedS3) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
+ return errNotYet
+}
+func (s *versionedS3) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ return "", errNotYet
+}
+
+var _ VersionedStore = (*versionedS3)(nil)
diff --git a/internal/lease/versioned_test.go b/internal/lease/versioned_test.go
new file mode 100644
index 000000000..b08edd1a1
--- /dev/null
+++ b/internal/lease/versioned_test.go
@@ -0,0 +1,84 @@
+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")
+ }
+}