amadan

carlos / platform Public

lease: read, stream, copy and delete by exact version, create-only put (claim Plan 2 Tasks 4 and 5)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 075349296f51f0244d43c866dbf184f19b0fd5a8 parent aaf519d
2 files changed, +141 −7
  • internal/lease/versioned.go +128 −7
  • internal/lease/versioned_test.go +13 −0
diff --git a/internal/lease/versioned.go b/internal/lease/versioned.go
index 09557c517..54377d71b 100644
--- a/internal/lease/versioned.go
+++ b/internal/lease/versioned.go
@@ -1,16 +1,22 @@
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
@@ -91,8 +97,6 @@ func (s *versionedS3) client(bucket string) (*awss3.Client, error) {
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 {
@@ -128,20 +132,137 @@ func (s *versionedS3) ListVersions(ctx context.Context, bucket, prefix string) (
})
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) {
- return nil, errNotYet
+ 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) {
- return nil, errNotYet
+ 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) {
- return "", "", errNotYet
+ 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 {
- return errNotYet
+ 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) {
- return "", errNotYet
+ 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
index b08edd1a1..a7d415af3 100644
--- a/internal/lease/versioned_test.go
+++ b/internal/lease/versioned_test.go
@@ -82,3 +82,16 @@ func TestVersionedS3RefusesUnboundBucket(t *testing.T) {
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)
+ }
+}