| 1 | package lease |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "net/http" |
| 10 | "net/url" |
| 11 | "os" |
| 12 | "sort" |
| 13 | "strings" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/aws/aws-sdk-go-v2/aws" |
| 17 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 18 | "github.com/aws/smithy-go" |
| 19 | "github.com/aws/smithy-go/encoding/httpbinding" |
| 20 | smithyhttp "github.com/aws/smithy-go/transport/http" |
| 21 | ) |
| 22 | |
| 23 | // ObjectVersion is one entry of a versioned listing: an object version, or |
| 24 | // a delete marker (DeleteMarker true, no ETag or Size). |
| 25 | type ObjectVersion struct { |
| 26 | Key string |
| 27 | VersionID string |
| 28 | ETag string // unquoted |
| 29 | Size int64 |
| 30 | IsLatest bool |
| 31 | DeleteMarker bool |
| 32 | LastModified time.Time |
| 33 | } |
| 34 | |
| 35 | var ( |
| 36 | ErrExists = errors.New("lease: object already exists") |
| 37 | ErrDestinationExists = errors.New("lease: copy destination already exists") |
| 38 | ErrIsDeleteMarker = errors.New("lease: version is a delete marker") |
| 39 | ErrNoVersionID = errors.New("lease: a version id is required") |
| 40 | ) |
| 41 | |
| 42 | // VersionedStore is the data bucket addressed by exact object version (host |
| 43 | // lifecycle claim spec C3, §6.2, §6.5). No method deletes or overwrites by |
| 44 | // bare key. |
| 45 | type VersionedStore interface { |
| 46 | // ListVersions returns every version and delete marker under prefix: |
| 47 | // key order; within a key the IsLatest entry first, the rest newest |
| 48 | // first with ties unordered. Any page error fails the whole call. |
| 49 | ListVersions(ctx context.Context, bucket, prefix string) ([]ObjectVersion, error) |
| 50 | // OpenVersion streams one exact version. Missing: ErrMirrorNotFound. |
| 51 | // A delete marker: ErrIsDeleteMarker. |
| 52 | OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) |
| 53 | // GetVersion reads one exact version whole (OpenVersion's errors). |
| 54 | GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) |
| 55 | // CopyVersion copies one exact source version to dstKey. createOnly |
| 56 | // adds If-None-Match: *; an existing current destination object gives |
| 57 | // ErrDestinationExists. It returns the destination's new version id |
| 58 | // and ETag. |
| 59 | CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (dstVersionID, dstETag string, err error) |
| 60 | // DeleteVersion removes one exact version; gone already is success. |
| 61 | // An empty versionID is ErrNoVersionID, without any request. |
| 62 | DeleteVersion(ctx context.Context, bucket, key, versionID string) error |
| 63 | // PutIfAbsent writes body only if key holds no current object version. |
| 64 | PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (versionID string, err error) |
| 65 | } |
| 66 | |
| 67 | // VersionRangeOpener optionally streams a range of one exact object version. |
| 68 | // A negative length reads to the end; a positive length bounds the read. |
| 69 | // Zero length and negative offsets are invalid. Errors match OpenVersion. |
| 70 | type VersionRangeOpener interface { |
| 71 | OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error) |
| 72 | } |
| 73 | |
| 74 | // versionedS3 binds every bucket to its own regional client explicitly. |
| 75 | // An unbound bucket is an error, never a guess. |
| 76 | type versionedS3 struct { |
| 77 | clients map[string]*awss3.Client // bucket -> client for its region |
| 78 | } |
| 79 | |
| 80 | // NewVersionedS3 builds one client per region from the default credential |
| 81 | // chain (with s3compat options, as newLeaserClient does) and binds each |
| 82 | // bucket to its region's client. |
| 83 | func NewVersionedS3(ctx context.Context, regionByBucket map[string]string) (VersionedStore, error) { |
| 84 | byRegion := map[string]*awss3.Client{} |
| 85 | out := &versionedS3{clients: map[string]*awss3.Client{}} |
| 86 | for bucket, region := range regionByBucket { |
| 87 | c, ok := byRegion[region] |
| 88 | if !ok { |
| 89 | var err error |
| 90 | if c, _, err = newLeaserClient(ctx, region); err != nil { |
| 91 | return nil, fmt.Errorf("lease: versioned client for %s: %w", region, err) |
| 92 | } |
| 93 | byRegion[region] = c |
| 94 | } |
| 95 | out.clients[bucket] = c |
| 96 | } |
| 97 | return out, nil |
| 98 | } |
| 99 | |
| 100 | func (s *versionedS3) client(bucket string) (*awss3.Client, error) { |
| 101 | c, ok := s.clients[bucket] |
| 102 | if !ok { |
| 103 | return nil, fmt.Errorf("lease: bucket %q is not bound to a region", bucket) |
| 104 | } |
| 105 | return c, nil |
| 106 | } |
| 107 | |
| 108 | func (s *versionedS3) ListVersions(ctx context.Context, bucket, prefix string) ([]ObjectVersion, error) { |
| 109 | c, err := s.client(bucket) |
| 110 | if err != nil { |
| 111 | return nil, err |
| 112 | } |
| 113 | p := awss3.NewListObjectVersionsPaginator(c, &awss3.ListObjectVersionsInput{ |
| 114 | Bucket: aws.String(bucket), Prefix: aws.String(prefix), |
| 115 | }) // StopOnDuplicateToken stays off: a key legitimately repeats across version pages |
| 116 | var out []ObjectVersion |
| 117 | for p.HasMorePages() { |
| 118 | page, err := p.NextPage(ctx) |
| 119 | if err != nil { |
| 120 | return nil, fmt.Errorf("lease: list versions %s/%s: %w", bucket, prefix, err) |
| 121 | } |
| 122 | for _, v := range page.Versions { |
| 123 | out = append(out, ObjectVersion{Key: aws.ToString(v.Key), VersionID: aws.ToString(v.VersionId), |
| 124 | ETag: strings.Trim(aws.ToString(v.ETag), `"`), Size: aws.ToInt64(v.Size), |
| 125 | IsLatest: aws.ToBool(v.IsLatest), LastModified: aws.ToTime(v.LastModified)}) |
| 126 | } |
| 127 | for _, d := range page.DeleteMarkers { |
| 128 | out = append(out, ObjectVersion{Key: aws.ToString(d.Key), VersionID: aws.ToString(d.VersionId), |
| 129 | IsLatest: aws.ToBool(d.IsLatest), DeleteMarker: true, LastModified: aws.ToTime(d.LastModified)}) |
| 130 | } |
| 131 | } |
| 132 | sort.SliceStable(out, func(i, j int) bool { |
| 133 | if out[i].Key != out[j].Key { |
| 134 | return out[i].Key < out[j].Key |
| 135 | } |
| 136 | if out[i].IsLatest != out[j].IsLatest { |
| 137 | return out[i].IsLatest |
| 138 | } |
| 139 | return out[i].LastModified.After(out[j].LastModified) // ties: unspecified (Global Constraints) |
| 140 | }) |
| 141 | return out, nil |
| 142 | } |
| 143 | |
| 144 | // apiCode returns the smithy error code and HTTP status of err. |
| 145 | func apiCode(err error) (string, int) { |
| 146 | var ae smithy.APIError |
| 147 | code := "" |
| 148 | if errors.As(err, &ae) { |
| 149 | code = ae.ErrorCode() |
| 150 | } |
| 151 | var re *smithyhttp.ResponseError |
| 152 | status := 0 |
| 153 | if errors.As(err, &re) { |
| 154 | status = re.HTTPStatusCode() |
| 155 | } |
| 156 | return code, status |
| 157 | } |
| 158 | |
| 159 | // mapReadErr normalizes only object or version absence. A missing bucket |
| 160 | // (NoSuchBucket, also a 404) stays an error: it is never "already gone". |
| 161 | func mapReadErr(err error) error { |
| 162 | code, status := apiCode(err) |
| 163 | switch { |
| 164 | case code == "NoSuchBucket": |
| 165 | return err |
| 166 | case code == "NoSuchKey" || code == "NoSuchVersion" || code == "NotFound" || status == http.StatusNotFound: |
| 167 | return ErrMirrorNotFound |
| 168 | case status == http.StatusMethodNotAllowed: |
| 169 | return ErrIsDeleteMarker // S3 answers 405 for a delete marker's version id |
| 170 | } |
| 171 | return err |
| 172 | } |
| 173 | |
| 174 | func (s *versionedS3) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) { |
| 175 | return s.openVersion(ctx, bucket, key, versionID, nil) |
| 176 | } |
| 177 | |
| 178 | func (s *versionedS3) OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error) { |
| 179 | if versionID == "" { |
| 180 | return nil, ErrNoVersionID |
| 181 | } |
| 182 | if offset < 0 || length == 0 || (length > 0 && offset > (1<<63-1)-(length-1)) { |
| 183 | return nil, errors.New("lease: invalid version range") |
| 184 | } |
| 185 | rangeHeader := fmt.Sprintf("bytes=%d-", offset) |
| 186 | if length > 0 { |
| 187 | rangeHeader += fmt.Sprint(offset + length - 1) |
| 188 | } |
| 189 | return s.openVersion(ctx, bucket, key, versionID, &rangeHeader) |
| 190 | } |
| 191 | |
| 192 | func (s *versionedS3) openVersion(ctx context.Context, bucket, key, versionID string, byteRange *string) (io.ReadCloser, error) { |
| 193 | if versionID == "" { |
| 194 | return nil, ErrNoVersionID |
| 195 | } |
| 196 | c, err := s.client(bucket) |
| 197 | if err != nil { |
| 198 | return nil, err |
| 199 | } |
| 200 | out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID), Range: byteRange}) |
| 201 | if err != nil { |
| 202 | if m := mapReadErr(err); m != err { |
| 203 | return nil, m |
| 204 | } |
| 205 | return nil, fmt.Errorf("lease: get %s/%s@%s: %w", bucket, key, versionID, err) |
| 206 | } |
| 207 | return out.Body, nil |
| 208 | } |
| 209 | |
| 210 | func (s *versionedS3) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) { |
| 211 | rc, err := s.OpenVersion(ctx, bucket, key, versionID) |
| 212 | if err != nil { |
| 213 | return nil, err |
| 214 | } |
| 215 | defer rc.Close() |
| 216 | return io.ReadAll(rc) |
| 217 | } |
| 218 | |
| 219 | func (s *versionedS3) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) { |
| 220 | if srcVersionID == "" { |
| 221 | return "", "", ErrNoVersionID |
| 222 | } |
| 223 | if _, err := s.client(srcBucket); err != nil { |
| 224 | return "", "", err |
| 225 | } |
| 226 | c, err := s.client(dstBucket) // a copy executes in the destination's region |
| 227 | if err != nil { |
| 228 | return "", "", err |
| 229 | } |
| 230 | in := &awss3.CopyObjectInput{ |
| 231 | Bucket: aws.String(dstBucket), |
| 232 | Key: aws.String(dstKey), |
| 233 | CopySource: aws.String(httpbinding.EscapePath(srcBucket+"/"+srcKey, false) + "?versionId=" + url.QueryEscape(srcVersionID)), |
| 234 | } |
| 235 | if createOnly { |
| 236 | in.IfNoneMatch = aws.String("*") |
| 237 | } |
| 238 | out, err := c.CopyObject(ctx, in) |
| 239 | if err != nil { |
| 240 | code, status := apiCode(err) |
| 241 | var ae smithy.APIError |
| 242 | switch { |
| 243 | case createOnly && (code == "PreconditionFailed" || status == http.StatusPreconditionFailed): |
| 244 | return "", "", ErrDestinationExists |
| 245 | case code == "InvalidRequest" && errors.As(err, &ae) && strings.Contains(strings.ToLower(ae.ErrorMessage()), "delete marker"): |
| 246 | return "", "", ErrIsDeleteMarker |
| 247 | default: |
| 248 | if m := mapReadErr(err); m != err { |
| 249 | return "", "", m |
| 250 | } |
| 251 | } |
| 252 | return "", "", fmt.Errorf("lease: copy %s/%s@%s -> %s/%s: %w", srcBucket, srcKey, srcVersionID, dstBucket, dstKey, err) |
| 253 | } |
| 254 | etag := "" |
| 255 | if out.CopyObjectResult != nil { |
| 256 | etag = strings.Trim(aws.ToString(out.CopyObjectResult.ETag), `"`) |
| 257 | } |
| 258 | return aws.ToString(out.VersionId), etag, nil |
| 259 | } |
| 260 | |
| 261 | func (s *versionedS3) DeleteVersion(ctx context.Context, bucket, key, versionID string) error { |
| 262 | if versionID == "" { |
| 263 | return ErrNoVersionID // never a bare-key delete (spec C3), and no request is sent |
| 264 | } |
| 265 | c, err := s.client(bucket) |
| 266 | if err != nil { |
| 267 | return err |
| 268 | } |
| 269 | if _, err := c.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID)}); err != nil { |
| 270 | if mapReadErr(err) == ErrMirrorNotFound { |
| 271 | return nil // already gone |
| 272 | } |
| 273 | return fmt.Errorf("lease: delete %s/%s@%s: %w", bucket, key, versionID, err) |
| 274 | } |
| 275 | return nil |
| 276 | } |
| 277 | |
| 278 | func (s *versionedS3) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) { |
| 279 | c, err := s.client(bucket) |
| 280 | if err != nil { |
| 281 | return "", err |
| 282 | } |
| 283 | out, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), |
| 284 | Body: bytes.NewReader(body), IfNoneMatch: aws.String("*")}) |
| 285 | if err != nil { |
| 286 | if code, status := apiCode(err); code == "PreconditionFailed" || status == http.StatusPreconditionFailed { |
| 287 | return "", ErrExists |
| 288 | } |
| 289 | return "", fmt.Errorf("lease: put-if-absent %s/%s: %w", bucket, key, err) |
| 290 | } |
| 291 | return aws.ToString(out.VersionId), nil |
| 292 | } |
| 293 | |
| 294 | // FilePutIfAbsent is PutIfAbsent streaming a file from disk, for objects |
| 295 | // too large to hold in memory: a move's sealed archive, written create-only |
| 296 | // into the source's own move-snapshots/ (moves spec §8.3). The S3 store |
| 297 | // implements it; an existing current object is ErrExists. |
| 298 | type FilePutIfAbsent interface { |
| 299 | PutFileIfAbsent(ctx context.Context, bucket, key, path string) (versionID string, err error) |
| 300 | } |
| 301 | |
| 302 | func (s *versionedS3) PutFileIfAbsent(ctx context.Context, bucket, key, path string) (string, error) { |
| 303 | c, err := s.client(bucket) |
| 304 | if err != nil { |
| 305 | return "", err |
| 306 | } |
| 307 | f, err := os.Open(path) |
| 308 | if err != nil { |
| 309 | return "", err |
| 310 | } |
| 311 | defer f.Close() |
| 312 | st, err := f.Stat() |
| 313 | if err != nil { |
| 314 | return "", err |
| 315 | } |
| 316 | // An *os.File is seekable, so the SDK hashes the payload by reading it |
| 317 | // and seeking back, never by buffering it. |
| 318 | out, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), |
| 319 | Body: f, ContentLength: aws.Int64(st.Size()), IfNoneMatch: aws.String("*")}) |
| 320 | if err != nil { |
| 321 | if code, status := apiCode(err); code == "PreconditionFailed" || status == http.StatusPreconditionFailed { |
| 322 | return "", ErrExists |
| 323 | } |
| 324 | return "", fmt.Errorf("lease: put-if-absent %s/%s: %w", bucket, key, err) |
| 325 | } |
| 326 | return aws.ToString(out.VersionId), nil |
| 327 | } |
| 328 | |
| 329 | var _ VersionedStore = (*versionedS3)(nil) |
| 330 | |
| 331 | var _ FilePutIfAbsent = (*versionedS3)(nil) |
| 332 | |
| 333 | var _ VersionRangeOpener = (*versionedS3)(nil) |
| 334 | |
amadan