amadan
moves: the staging-work lease (45 min TTL) and the SafeUntil guard; lease: AcquireTTL and the single-signature staging mirror (§7.3)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
13 files changed,
+1085
−27
internal/lease/checkpoint_test.go+5 −5internal/lease/endpoint.go+1 −1internal/lease/health.go+7 −2internal/lease/lease.go+11 −8internal/lease/lease_test.go+5 −5internal/lease/leasetest/bucket.go+25 −0internal/lease/leasetest/s3server.go+40 −6internal/lease/leasetest/s3server_test.go+83 −0internal/lease/staging.go+346 −0internal/lease/staging_test.go+311 −0internal/lease/staging_url_test.go+20 −0internal/moves/lease.go+70 −0internal/moves/lease_test.go+161 −0
diff --git a/internal/lease/checkpoint_test.go b/internal/lease/checkpoint_test.go| index 2727cfda4..cca33dbe7 100644 |
| --- a/internal/lease/checkpoint_test.go |
| +++ b/internal/lease/checkpoint_test.go |
| @@ -123,7 +123,7 @@ func localHeld(t *testing.T, backend *lockStore, ttl time.Duration) *Held { |
| t.Cleanup(srv.Close) |
| fakeCreds(t) |
| t.Setenv("CARLOS_S3_ENDPOINT", srv.URL) |
| - h, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl) |
| + h, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl) |
| if e != nil { |
| t.Fatal(e) |
| } |
| @@ -207,7 +207,7 @@ func TestHeldCheckpointVacuumRestoreAndTransfer(t *testing.T) { |
| t.Fatal("checkpoint released lock") |
| } |
| // The actual leaser sees the same still-owned lock after both checkpoints. |
| - if peer, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", 10*time.Second); e == nil { |
| + if peer, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", 10*time.Second); e == nil { |
| peer.Release(context.Background()) |
| t.Fatal("acquisition gap") |
| } |
| @@ -362,7 +362,7 @@ func TestBlockedRenewalDeadlinePrecedesTakeover(t *testing.T) { |
| if !time.Now().Before(original.ExpiresAt) { |
| t.Fatal("termination did not precede lease expiry") |
| } |
| - if peer, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl); e == nil { |
| + if peer, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl); e == nil { |
| peer.Release(context.Background()) |
| t.Fatal("peer took live lock") |
| } |
| @@ -370,7 +370,7 @@ func TestBlockedRenewalDeadlinePrecedesTakeover(t *testing.T) { |
| // A separate real leaser can take over the expired CAS object. Its acquisition |
| // uses If-Match, so disable the synthetic blocked-renewal hook first. |
| backend.beforePut = nil |
| - peer, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl) |
| + peer, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", ttl) |
| if e != nil { |
| t.Fatal(e) |
| } |
| @@ -454,7 +454,7 @@ func TestAcquireLateSuccessGrantsNoHandle(t *testing.T) { |
| defer srv.Close() |
| fakeCreds(t) |
| t.Setenv("CARLOS_S3_ENDPOINT", srv.URL) |
| - h, e := acquireTTL(context.Background(), "test-bucket", "one.example", "test-region", 300*time.Millisecond) |
| + h, e := AcquireTTL(context.Background(), "test-bucket", "one.example", "test-region", 300*time.Millisecond) |
| if e == nil || h != nil { |
| if h != nil { |
| h.Release(context.Background()) |
diff --git a/internal/lease/endpoint.go b/internal/lease/endpoint.go| index d01061b8c..c2e3dfe78 100644 |
| --- a/internal/lease/endpoint.go |
| +++ b/internal/lease/endpoint.go |
| @@ -93,7 +93,7 @@ func regionFor(explicit, chainResolved string) string { |
| // Ceph RGW. Passing that client to lss3.Leaser.SetClient is the ONLY way the |
| // fix reaches litestream, which has no provider handling in its leaser at all. |
| // |
| -// It returns the region it resolved alongside the client, so acquireTTL can |
| +// It returns the region it resolved alongside the client, so AcquireTTL can |
| // carry that exact value onto the Held for Replicate's replica client rather |
| // than letting litestream substitute a different default of its own. |
| func newLeaserClient(ctx context.Context, region string) (*awss3.Client, string, error) { |
diff --git a/internal/lease/health.go b/internal/lease/health.go| index 69814f15c..4b3b9193a 100644 |
| --- a/internal/lease/health.go |
| +++ b/internal/lease/health.go |
| @@ -18,8 +18,13 @@ var ( |
| const terminationMargin = 5 * time.Second |
| -// Production uses the fixed margin. Short TTLs exist only through acquireTTL, |
| -// the package's private test seam. |
| +// RenewMargin is the margin Held.SafeUntil subtracts from the lease's |
| +// expiry at production TTLs: SafeUntil is at most TTL-RenewMargin ahead of |
| +// a fresh acquire. The moves staging guard reasons from it. |
| +const RenewMargin = terminationMargin |
| + |
| +// Production uses the fixed margin. Short TTLs exist only through AcquireTTL |
| +// (tests; production callers pass LeaseTTL or the moves staging term). |
| func marginForTTL(ttl time.Duration) time.Duration { |
| if ttl < 3*terminationMargin { |
| return ttl / 10 |
diff --git a/internal/lease/lease.go b/internal/lease/lease.go| index effcd0cd9..d4ae8b158 100644 |
| --- a/internal/lease/lease.go |
| +++ b/internal/lease/lease.go |
| @@ -118,16 +118,19 @@ type Held struct { |
| // (env/profile) resolves it, or — under a CARLOS_S3_ENDPOINT override, where |
| // there is no chain to resolve from — endpoint.go's non-AWS default. |
| func Acquire(ctx context.Context, bucket, prefix, region string) (*Held, error) { |
| - return acquireTTL(ctx, bucket, prefix, region, LeaseTTL) |
| + return AcquireTTL(ctx, bucket, prefix, region, LeaseTTL) |
| } |
| -// acquireTTL is Acquire with the lease TTL as a seam. Production always |
| -// passes LeaseTTL; tests pass a short TTL so an expiry-and-takeover sequence |
| -// (the failover race) fits inside a test run instead of 15 minutes. The TTL |
| -// governs renewal spacing too — startRenewer ticks at leaser.TTL/3 — so a |
| -// short-TTL lease is renewed proportionally often and lapses proportionally |
| -// fast, which is what makes "the holder went dark" observable. |
| -func acquireTTL(ctx context.Context, bucket, prefix, region string, ttl time.Duration) (*Held, error) { |
| +// AcquireTTL is Acquire with the lease TTL as a parameter. Production |
| +// Acquire passes LeaseTTL; the moves staging-work lease passes its longer |
| +// term (moves.StagingLeaseTTL, whose renewals have to leave a SigV4 |
| +// signature's whole life in front of SafeUntil); tests pass a short TTL so |
| +// an expiry-and-takeover sequence (the failover race) fits inside a test |
| +// run instead of 15 minutes. The TTL governs renewal spacing too: |
| +// startRenewer ticks at leaser.TTL/3, so a short-TTL lease is renewed |
| +// proportionally often and lapses proportionally fast, which is what |
| +// makes "the holder went dark" observable. |
| +func AcquireTTL(ctx context.Context, bucket, prefix, region string, ttl time.Duration) (*Held, error) { |
| // The leaser wants a ready-made S3 client (unlike the replica client, |
| // which builds its own). Default AWS credential chain, and the |
| // CARLOS_S3_ENDPOINT override applied — see endpoint.go for why the lock |
diff --git a/internal/lease/lease_test.go b/internal/lease/lease_test.go| index ca31c0733..2584b3e2b 100644 |
| --- a/internal/lease/lease_test.go |
| +++ b/internal/lease/lease_test.go |
| @@ -254,14 +254,14 @@ func TestReplicateWithLease_MutualExclusion(t *testing.T) { |
| // instance from the regional bucket — is the existing activator wake path and |
| // is covered by the activator's own tests; nothing about it is new here. |
| // |
| -// Runs on a 15s TTL through the acquireTTL seam; at the production LeaseTTL |
| +// Runs on a 15s TTL through the AcquireTTL seam; at the production LeaseTTL |
| // the same sequence would take 15 minutes. |
| func TestDarkHostCannotReclaimAfterFailover(t *testing.T) { |
| bucket, prefix, region := liveConfig(t) |
| ctx := context.Background() |
| const shortTTL = 15 * time.Second |
| - a, err := acquireTTL(ctx, bucket, prefix, region, shortTTL) |
| + a, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL) |
| if err != nil { |
| t.Fatalf("A initial acquire: %v", err) |
| } |
| @@ -281,7 +281,7 @@ func TestDarkHostCannotReclaimAfterFailover(t *testing.T) { |
| // (Both "hosts" run in this one process and therefore share the leaser's |
| // owner string, so the expiry — not the name — is what identifies the |
| // holder here.) |
| - peer, err := acquireTTL(ctx, bucket, prefix, region, shortTTL) |
| + peer, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL) |
| if err == nil { |
| elapsed := time.Since(acquiredAt) |
| _ = peer.Release(ctx) |
| @@ -301,7 +301,7 @@ func TestDarkHostCannotReclaimAfterFailover(t *testing.T) { |
| time.Sleep(shortTTL + 5*time.Second) // let A's abandoned lease lapse |
| - b, err := acquireTTL(ctx, bucket, prefix, region, shortTTL) |
| + b, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL) |
| if err != nil { |
| t.Fatalf("B acquire after A's lease expired (the failover wake): %v", err) |
| } |
| @@ -309,7 +309,7 @@ func TestDarkHostCannotReclaimAfterFailover(t *testing.T) { |
| defer b.Release(ctx) |
| // A returns from the dead and tries to resume. |
| - if revenant, err := acquireTTL(ctx, bucket, prefix, region, shortTTL); err == nil { |
| + if revenant, err := AcquireTTL(ctx, bucket, prefix, region, shortTTL); err == nil { |
| elapsed := time.Since(bHeldFrom) |
| _ = revenant.Release(ctx) |
| if elapsed >= shortTTL { |
diff --git a/internal/lease/leasetest/bucket.go b/internal/lease/leasetest/bucket.go| index 94450f5e0..cc1f4a824 100644 |
| --- a/internal/lease/leasetest/bucket.go |
| +++ b/internal/lease/leasetest/bucket.go |
| @@ -356,6 +356,31 @@ func (f *VersionedBucket) putIfAbsentLocked(bucket, key string, body []byte) (st |
| return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)}), nil |
| } |
| +// putLocked is a plain PUT: it stores a new version beside the existing |
| +// ones, as S3 with versioning on does. putIfAbsentLocked is the create-only |
| +// variant (If-None-Match: *). |
| +func (f *VersionedBucket) putLocked(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 |
| + } |
| + return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)}), nil |
| +} |
| + |
| +// bareDeleteLocked is a DELETE naming no version: S3 with versioning on |
| +// adds a delete marker and removes nothing. |
| +func (f *VersionedBucket) bareDeleteLocked(bucket, key string) (string, error) { |
| + if err := f.known(bucket); err != nil { |
| + return "", err |
| + } |
| + if err := f.fail("delete", key); err != nil { |
| + return "", err |
| + } |
| + return f.push(bucket, key, version{deleteMarker: true}), nil |
| +} |
| + |
| func (f *VersionedBucket) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) { |
| f.mu.Lock() |
| defer f.mu.Unlock() |
diff --git a/internal/lease/leasetest/s3server.go b/internal/lease/leasetest/s3server.go| index 92cba0b9d..931926ea8 100644 |
| --- a/internal/lease/leasetest/s3server.go |
| +++ b/internal/lease/leasetest/s3server.go |
| @@ -107,7 +107,16 @@ func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int |
| 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")) |
| + versionID := q.Get("versionId") |
| + if versionID == "" { // no version named: the current one |
| + cur := f.buckets[bucket][key] |
| + if len(cur) == 0 || cur[0].deleteMarker { |
| + s3err(w, http.StatusNotFound, "NoSuchKey") |
| + return |
| + } |
| + versionID = cur[0].id |
| + } |
| + body, err := f.read(bucket, key, versionID) |
| if err != nil { |
| writeErr(w, err) |
| return |
| @@ -144,11 +153,12 @@ func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int |
| 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 |
| + var id string |
| + if r.Header.Get("If-None-Match") == "*" { |
| + id, err = f.putIfAbsentLocked(bucket, key, body) |
| + } else { |
| + id, err = f.putLocked(bucket, key, body) // a new version, as S3 with versioning on |
| } |
| - id, err := f.putIfAbsentLocked(bucket, key, body) |
| if err != nil { |
| writeErr(w, err) |
| return |
| @@ -161,6 +171,15 @@ func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int |
| return |
| } |
| w.WriteHeader(http.StatusNoContent) |
| + case r.Method == http.MethodDelete && key != "": |
| + id, err := f.bareDeleteLocked(bucket, key) // versioning on: a delete marker |
| + if err != nil { |
| + writeErr(w, err) |
| + return |
| + } |
| + w.Header().Set("x-amz-delete-marker", "true") |
| + w.Header().Set("x-amz-version-id", id) |
| + w.WriteHeader(http.StatusNoContent) |
| default: |
| s3err(w, http.StatusMethodNotAllowed, "MethodNotAllowed") |
| } |
| @@ -232,7 +251,22 @@ func (f *VersionedBucket) serveCopy(w http.ResponseWriter, r *http.Request, buck |
| return |
| } |
| sb, sk, _ := strings.Cut(unesc, "/") |
| - id, etag, err := f.copyLocked(sb, sk, sq.Get("versionId"), bucket, key, r.Header.Get("If-None-Match") == "*") |
| + srcVersion := sq.Get("versionId") |
| + if srcVersion == "" { // no version named: the source's current one |
| + cur := f.buckets[sb][sk] |
| + if len(cur) == 0 || cur[0].deleteMarker { |
| + s3err(w, http.StatusNotFound, "NoSuchKey") |
| + return |
| + } |
| + srcVersion = cur[0].id |
| + } |
| + if im := r.Header.Get("x-amz-copy-source-if-match"); im != "" { |
| + if v, ok := f.find(sb, sk, srcVersion); ok && strings.Trim(im, `"`) != v.etag { |
| + s3err(w, http.StatusPreconditionFailed, "PreconditionFailed") |
| + return |
| + } |
| + } |
| + id, etag, err := f.copyLocked(sb, sk, srcVersion, 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 |
diff --git a/internal/lease/leasetest/s3server_test.go b/internal/lease/leasetest/s3server_test.go| index 7ea3b6785..75663ed4c 100644 |
| --- a/internal/lease/leasetest/s3server_test.go |
| +++ b/internal/lease/leasetest/s3server_test.go |
| @@ -1,6 +1,7 @@ |
| package leasetest |
| import ( |
| + "context" |
| "errors" |
| "io" |
| "net/http" |
| @@ -36,3 +37,85 @@ func TestS3ServerFailedUploadStoresNothing(t *testing.T) { |
| t.Fatalf("failed upload changed the bucket:\n%s", m.Snapshot()) |
| } |
| } |
| + |
| +func do(t *testing.T, h http.Handler, method, path string, body string, hdr map[string]string) *httptest.ResponseRecorder { |
| + t.Helper() |
| + req := httptest.NewRequest(method, path, strings.NewReader(body)) |
| + for k, v := range hdr { |
| + req.Header.Set(k, v) |
| + } |
| + rec := httptest.NewRecorder() |
| + h.ServeHTTP(rec, req) |
| + return rec |
| +} |
| + |
| +// A plain PUT (no If-None-Match) stores a new version beside the existing |
| +// one, as S3 with versioning on does; the create-only PUT keeps refusing |
| +// an existing current object. A plain DELETE adds a delete marker. |
| +func TestS3ServerPlainPutCreatesAVersion(t *testing.T) { |
| + m := NewVersionedBucket() |
| + m.AddBucket("data") |
| + h, _ := NewS3Server(m) |
| + first := do(t, h, http.MethodPut, "/data/k", "one", nil) |
| + second := do(t, h, http.MethodPut, "/data/k", "two", nil) |
| + if first.Code != 200 || second.Code != 200 { |
| + t.Fatalf("plain puts: %d %d", first.Code, second.Code) |
| + } |
| + v1, v2 := first.Header().Get("x-amz-version-id"), second.Header().Get("x-amz-version-id") |
| + if v1 == "" || v2 == "" || v1 == v2 { |
| + t.Fatalf("version ids %q %q", v1, v2) |
| + } |
| + if got, _ := m.Current("data", "k"); string(got) != "two" { |
| + t.Fatalf("current %q", got) |
| + } |
| + if old, err := m.GetVersion(context.Background(), "data", "k", v1); err != nil || string(old) != "one" { |
| + t.Fatalf("the first version is gone: %q %v", old, err) |
| + } |
| + // a GET naming no version reads the current one |
| + if rec := do(t, h, http.MethodGet, "/data/k", "", nil); rec.Code != 200 || rec.Body.String() != "two" { |
| + t.Fatalf("plain get: %d %q", rec.Code, rec.Body.String()) |
| + } |
| + if rec := do(t, h, http.MethodGet, "/data/absent", "", nil); rec.Code != http.StatusNotFound { |
| + t.Fatalf("plain get of an absent key: %d", rec.Code) |
| + } |
| + // the create-only path is unchanged |
| + if rec := do(t, h, http.MethodPut, "/data/k", "three", map[string]string{"If-None-Match": "*"}); rec.Code != http.StatusPreconditionFailed { |
| + t.Fatalf("create-only over a current object: %d", rec.Code) |
| + } |
| + if rec := do(t, h, http.MethodPut, "/data/fresh", "x", map[string]string{"If-None-Match": "*"}); rec.Code != 200 || rec.Header().Get("x-amz-version-id") == "" { |
| + t.Fatalf("create-only on a fresh key: %d", rec.Code) |
| + } |
| + // a bare DELETE leaves a delete marker, the versions stay |
| + if rec := do(t, h, http.MethodDelete, "/data/k", "", nil); rec.Code != http.StatusNoContent { |
| + t.Fatalf("bare delete: %d", rec.Code) |
| + } |
| + if _, ok := m.Current("data", "k"); ok { |
| + t.Fatal("the key still reads after a bare delete") |
| + } |
| + if old, err := m.GetVersion(context.Background(), "data", "k", v2); err != nil || string(old) != "two" { |
| + t.Fatalf("a bare delete removed a version: %q %v", old, err) |
| + } |
| +} |
| + |
| +// A copy that names no source version copies the source's current one, and |
| +// honors x-amz-copy-source-if-match against its ETag (S3's behavior; the |
| +// staging mirror copies this way). |
| +func TestS3ServerCopyOfCurrentVersion(t *testing.T) { |
| + m := NewVersionedBucket() |
| + m.AddBucket("src") |
| + m.AddBucket("dst") |
| + m.Put("src", "a/b c+d", []byte("body")) |
| + h, _ := NewS3Server(m) |
| + src := "/src/a/b%20c%2Bd" |
| + rec := do(t, h, http.MethodPut, "/dst/out", "", map[string]string{"x-amz-copy-source": src, "x-amz-copy-source-if-match": `"` + md5hex([]byte("body")) + `"`}) |
| + if rec.Code != 200 || rec.Header().Get("x-amz-version-id") == "" || !strings.Contains(rec.Body.String(), md5hex([]byte("body"))) { |
| + t.Fatalf("copy: %d %q %v", rec.Code, rec.Body.String(), rec.Header()) |
| + } |
| + rec = do(t, h, http.MethodPut, "/dst/out2", "", map[string]string{"x-amz-copy-source": src, "x-amz-copy-source-if-match": `"deadbeef"`}) |
| + if rec.Code != http.StatusPreconditionFailed || !strings.Contains(rec.Body.String(), "<Code>PreconditionFailed</Code>") { |
| + t.Fatalf("stale if-match: %d %q", rec.Code, rec.Body.String()) |
| + } |
| + if _, ok := m.Current("dst", "out2"); ok { |
| + t.Fatal("a failed precondition copied") |
| + } |
| +} |
diff --git a/internal/lease/staging.go b/internal/lease/staging.go| new file mode 100644 |
| index 000000000..5e0a691cb |
| --- /dev/null |
| +++ b/internal/lease/staging.go |
| @@ -0,0 +1,346 @@ |
| +package lease |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/xml" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "net/http" |
| + "net/url" |
| + "strings" |
| + "time" |
| + |
| + "github.com/aws/aws-sdk-go-v2/aws" |
| + v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4" |
| + awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| +) |
| + |
| +// The staging mirror (moves spec §7.3, model 14): the writer of a move's |
| +// staging area, whose every mutation is signed at a timestamp the caller |
| +// passes and sent exactly once. |
| +// |
| +// Why it signs by hand. A SigV4 signature stays valid for 15 minutes, and |
| +// the SDK signs when its middleware runs, after any retry, backoff or |
| +// pause the caller cannot see. The moves guard (moves.Guarded) proves the |
| +// staging lease is safe for a signature's whole life at one instant and |
| +// hands that instant to the mutation as signAt. This type signs AT signAt |
| +// and reads no clock of its own, so a pause of any length between the |
| +// check and the send (a frozen VM, a stalled goroutine) yields a request |
| +// dated at the check, which S3 refuses once that date is 15 minutes old: |
| +// exactly the life the guard required of the lease. There is no retry and |
| +// no re-signing; a caller that wants another attempt goes back through |
| +// the guard and gets a new timestamp. The one gap left is between the |
| +// signing and the socket write, bounded by the request timeout and the |
| +// guard's skew allowance (§7.3). |
| +// |
| +// *StagingMirror deliberately satisfies none of MirrorStore's mutations, |
| +// VersionedCopier or VersionedPutter: each of those lacks a signAt, and |
| +// only the guard hands one out, so no code path can sign outside a |
| +// check. Callers hold it as a StagingWriter behind a guard adapter. |
| + |
| +// StagingWriter is what the move lane needs of its staging store: the |
| +// four guarded mutations and the two reads. *StagingMirror is the |
| +// production implementation; the lane's tests supply a fake. |
| +type StagingWriter interface { |
| + // CopyVersionedAt is a server-side copy into the staging bucket, |
| + // conditional on srcETag when it is set (ErrCopySourceChanged when the |
| + // source no longer carries it). It returns the destination's ETag |
| + // (unquoted) and version id. |
| + CopyVersionedAt(ctx context.Context, signAt time.Time, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (etag, versionID string, err error) |
| + // PutVersionedAt writes body over key and returns the new version id. |
| + PutVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (versionID string, err error) |
| + // PutIfAbsentVersionedAt is PutVersionedAt with If-None-Match: *; an |
| + // existing object gives ErrExists. |
| + PutIfAbsentVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (versionID string, err error) |
| + // DeleteAt removes key (on a versioned bucket, adds a delete marker). |
| + DeleteAt(ctx context.Context, signAt time.Time, bucket, key string) error |
| + List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) |
| + Get(ctx context.Context, bucket, key string) ([]byte, error) |
| +} |
| + |
| +var _ StagingWriter = (*StagingMirror)(nil) |
| + |
| +const ( |
| + stagingRequestTimeout = 120 * time.Second |
| + unsignedPayload = "UNSIGNED-PAYLOAD" |
| + // stagingMaxResponse caps what is read of any response: the bodies |
| + // that matter (a CopyObjectResult, an error document) are tiny. |
| + stagingMaxResponse = 1 << 20 |
| +) |
| + |
| +// StagingMirror signs and sends staging mutations itself. See the package |
| +// comment above. Build it with NewStagingMirror (production) or |
| +// NewStagingMirrorAt (tests, an endpoint override). |
| +type StagingMirror struct { |
| + reads MirrorStore // List and Get: ordinary reads, no guard needed |
| + endpoint string // base URL override (path-style); "" is AWS |
| + region string // the destination bucket's region: where requests go and what they are signed for |
| + creds aws.CredentialsProvider |
| + signer *v4.Signer |
| + client *http.Client |
| +} |
| + |
| +func newStagingSigner() *v4.Signer { |
| + // S3 canonicalizes the path as sent, never double-escaped. |
| + return v4.NewSigner(func(o *v4.SignerOptions) { o.DisableURIPathEscaping = true }) |
| +} |
| + |
| +// NewStagingMirror builds the production mirror: reads through the same |
| +// two-region pair NewS3Mirror uses, mutations signed with the default |
| +// credential chain for dstRegion (the staging bucket's region). |
| +// CARLOS_S3_ENDPOINT, when set, replaces the AWS endpoint, path-style. |
| +func NewStagingMirror(ctx context.Context, srcRegion, dstRegion string) (*StagingMirror, error) { |
| + reads, err := NewS3Mirror(ctx, srcRegion, dstRegion) |
| + if err != nil { |
| + return nil, err |
| + } |
| + var opts []func(*awsconfig.LoadOptions) error |
| + if dstRegion != "" { |
| + opts = append(opts, awsconfig.WithRegion(dstRegion)) |
| + } |
| + cfg, err := awsconfig.LoadDefaultConfig(ctx, opts...) |
| + if err != nil { |
| + return nil, fmt.Errorf("load aws config: %w", err) |
| + } |
| + if cfg.Credentials == nil { |
| + return nil, errors.New("staging mirror: no AWS credentials to sign with") |
| + } |
| + return &StagingMirror{ |
| + reads: reads, |
| + endpoint: s3Endpoint(), |
| + region: regionFor(dstRegion, cfg.Region), |
| + creds: cfg.Credentials, |
| + signer: newStagingSigner(), |
| + client: &http.Client{Timeout: stagingRequestTimeout}, |
| + }, nil |
| +} |
| + |
| +// NewStagingMirrorAt is the constructor with the endpoint and credentials |
| +// given, path-style, for tests against a local S3 (leasetest.NewS3Server). |
| +// Reads go to the same endpoint. |
| +func NewStagingMirrorAt(endpoint, region string, creds aws.CredentialsProvider) *StagingMirror { |
| + if !strings.Contains(endpoint, "://") { |
| + endpoint = "https://" + endpoint |
| + } |
| + c := awss3.New(awss3.Options{ |
| + Region: region, |
| + Credentials: creds, |
| + BaseEndpoint: &endpoint, |
| + UsePathStyle: true, |
| + RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired, |
| + ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired, |
| + }) |
| + return &StagingMirror{ |
| + reads: &s3Mirror{src: c, dst: c, byBucket: map[string]*awss3.Client{}}, |
| + endpoint: endpoint, |
| + region: region, |
| + creds: creds, |
| + signer: newStagingSigner(), |
| + client: &http.Client{Timeout: stagingRequestTimeout}, |
| + } |
| +} |
| + |
| +func (m *StagingMirror) List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) { |
| + return m.reads.List(ctx, bucket, prefix) |
| +} |
| + |
| +func (m *StagingMirror) Get(ctx context.Context, bucket, key string) ([]byte, error) { |
| + return m.reads.Get(ctx, bucket, key) |
| +} |
| + |
| +// s3Escape percent-encodes everything but the RFC 3986 unreserved set and |
| +// "/", which is what S3 canonicalizes object paths (and the copy-source |
| +// header) to. |
| +func s3Escape(s string) string { |
| + const hexdigits = "0123456789ABCDEF" |
| + var b strings.Builder |
| + for i := 0; i < len(s); i++ { |
| + c := s[i] |
| + switch { |
| + case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9', |
| + c == '-', c == '_', c == '.', c == '~', c == '/': |
| + b.WriteByte(c) |
| + default: |
| + b.WriteByte('%') |
| + b.WriteByte(hexdigits[c>>4]) |
| + b.WriteByte(hexdigits[c&15]) |
| + } |
| + } |
| + return b.String() |
| +} |
| + |
| +// objectURL addresses bucket/key: the override path-style, else AWS |
| +// virtual-hosted (path-style for a dotted bucket name, whose name the |
| +// wildcard certificate does not cover). |
| +func (m *StagingMirror) objectURL(bucket, key string) (*url.URL, error) { |
| + var u *url.URL |
| + var path string |
| + switch { |
| + case m.endpoint != "": |
| + base, err := url.Parse(m.endpoint) |
| + if err != nil { |
| + return nil, fmt.Errorf("staging mirror: endpoint %q: %w", m.endpoint, err) |
| + } |
| + u, path = base, strings.TrimSuffix(base.Path, "/")+"/"+bucket+"/"+key |
| + case strings.Contains(bucket, "."): |
| + u, path = &url.URL{Scheme: "https", Host: "s3." + m.region + ".amazonaws.com"}, "/"+bucket+"/"+key |
| + default: |
| + u, path = &url.URL{Scheme: "https", Host: bucket + ".s3." + m.region + ".amazonaws.com"}, "/"+key |
| + } |
| + raw := s3Escape(path) |
| + dec, err := url.PathUnescape(raw) |
| + if err != nil { |
| + return nil, err |
| + } |
| + out := *u |
| + out.Path, out.RawPath, out.RawQuery = dec, raw, "" |
| + return &out, nil |
| +} |
| + |
| +// send signs one request at signAt and sends it once. Nothing here reads a |
| +// clock, retries, or re-signs. It returns the status, headers and (capped) |
| +// body of whatever S3 answered. |
| +func (m *StagingMirror) send(ctx context.Context, signAt time.Time, method, bucket, key string, hdr http.Header, body []byte) (int, http.Header, []byte, error) { |
| + if signAt.IsZero() { |
| + return 0, nil, nil, errors.New("staging mirror: no signing time (a mutation signs at a Guarded.Before check)") |
| + } |
| + u, err := m.objectURL(bucket, key) |
| + if err != nil { |
| + return 0, nil, nil, err |
| + } |
| + var rd io.Reader |
| + payloadHash := unsignedPayload |
| + if body != nil { |
| + rd = bytes.NewReader(body) |
| + sum := sha256.Sum256(body) |
| + payloadHash = hex.EncodeToString(sum[:]) |
| + } |
| + req, err := http.NewRequestWithContext(ctx, method, u.String(), rd) |
| + if err != nil { |
| + return 0, nil, nil, err |
| + } |
| + for k, vs := range hdr { |
| + req.Header[k] = vs |
| + } |
| + req.Header.Set("X-Amz-Content-Sha256", payloadHash) |
| + creds, err := m.creds.Retrieve(ctx) |
| + if err != nil { |
| + return 0, nil, nil, fmt.Errorf("staging mirror: credentials: %w", err) |
| + } |
| + if err := m.signer.SignHTTP(ctx, creds, req, payloadHash, "s3", m.region, signAt); err != nil { |
| + return 0, nil, nil, fmt.Errorf("staging mirror: sign: %w", err) |
| + } |
| + resp, err := m.client.Do(req) |
| + if err != nil { |
| + return 0, nil, nil, fmt.Errorf("staging mirror: %s %s/%s: %w", method, bucket, key, err) |
| + } |
| + defer resp.Body.Close() |
| + rb, err := io.ReadAll(io.LimitReader(resp.Body, stagingMaxResponse)) |
| + if err != nil { |
| + return resp.StatusCode, resp.Header, nil, fmt.Errorf("staging mirror: %s %s/%s: read response: %w", method, bucket, key, err) |
| + } |
| + return resp.StatusCode, resp.Header, rb, nil |
| +} |
| + |
| +// s3ErrorBody is the document S3 puts in an error response, and in the |
| +// body of a 200 copy that failed after the headers went out. |
| +type s3ErrorBody struct { |
| + XMLName xml.Name `xml:"Error"` |
| + Code string `xml:"Code"` |
| + Message string `xml:"Message"` |
| +} |
| + |
| +// s3Code is the error document's Code, or "" if the body is not one. |
| +func s3Code(body []byte) string { |
| + var e s3ErrorBody |
| + if xml.Unmarshal(body, &e) != nil { |
| + return "" |
| + } |
| + return e.Code |
| +} |
| + |
| +func stagingFailure(op string, status int, body []byte) error { |
| + code := s3Code(body) |
| + if code == "" { |
| + code = http.StatusText(status) |
| + } |
| + return fmt.Errorf("staging mirror: %s: %s (HTTP %d)", op, code, status) |
| +} |
| + |
| +func (m *StagingMirror) CopyVersionedAt(ctx context.Context, signAt time.Time, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, string, error) { |
| + hdr := http.Header{} |
| + hdr.Set("X-Amz-Copy-Source", "/"+srcBucket+"/"+s3Escape(srcKey)) |
| + if srcETag != "" { |
| + hdr.Set("X-Amz-Copy-Source-If-Match", `"`+normalizeETag(srcETag)+`"`) |
| + } |
| + op := "copy " + srcBucket + "/" + srcKey |
| + status, h, body, err := m.send(ctx, signAt, http.MethodPut, dstBucket, dstKey, hdr, nil) |
| + if err != nil { |
| + return "", "", err |
| + } |
| + if status == http.StatusPreconditionFailed && srcETag != "" && s3Code(body) == "PreconditionFailed" { |
| + return "", "", ErrCopySourceChanged |
| + } |
| + if status/100 != 2 { |
| + return "", "", stagingFailure(op, status, body) |
| + } |
| + // S3 may answer 200 and then fail the copy, putting an <Error> document |
| + // where the result belongs. |
| + if code := s3Code(body); code != "" { |
| + return "", "", fmt.Errorf("staging mirror: %s: %s (HTTP %d, late failure)", op, code, status) |
| + } |
| + var res struct { |
| + XMLName xml.Name `xml:"CopyObjectResult"` |
| + ETag string `xml:"ETag"` |
| + } |
| + if err := xml.Unmarshal(body, &res); err != nil || res.ETag == "" { |
| + return "", "", fmt.Errorf("staging mirror: %s: no CopyObjectResult in a 200 response", op) |
| + } |
| + return normalizeETag(res.ETag), h.Get("X-Amz-Version-Id"), nil |
| +} |
| + |
| +func (m *StagingMirror) put(ctx context.Context, signAt time.Time, bucket, key string, body []byte, createOnly bool) (string, error) { |
| + if body == nil { |
| + body = []byte{} |
| + } |
| + hdr := http.Header{} |
| + hdr.Set("Content-Type", "application/json") |
| + if createOnly { |
| + hdr.Set("If-None-Match", "*") |
| + } |
| + status, h, rb, err := m.send(ctx, signAt, http.MethodPut, bucket, key, hdr, body) |
| + if err != nil { |
| + return "", err |
| + } |
| + if createOnly && status == http.StatusPreconditionFailed { |
| + return "", ErrExists |
| + } |
| + if status/100 != 2 { |
| + return "", stagingFailure("put "+bucket+"/"+key, status, rb) |
| + } |
| + return h.Get("X-Amz-Version-Id"), nil |
| +} |
| + |
| +func (m *StagingMirror) PutVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (string, error) { |
| + return m.put(ctx, signAt, bucket, key, body, false) |
| +} |
| + |
| +func (m *StagingMirror) PutIfAbsentVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (string, error) { |
| + return m.put(ctx, signAt, bucket, key, body, true) |
| +} |
| + |
| +func (m *StagingMirror) DeleteAt(ctx context.Context, signAt time.Time, bucket, key string) error { |
| + status, _, body, err := m.send(ctx, signAt, http.MethodDelete, bucket, key, nil, nil) |
| + if err != nil { |
| + return err |
| + } |
| + if status/100 != 2 { |
| + return stagingFailure("delete "+bucket+"/"+key, status, body) |
| + } |
| + return nil |
| +} |
diff --git a/internal/lease/staging_test.go b/internal/lease/staging_test.go| new file mode 100644 |
| index 000000000..4a100b89e |
| --- /dev/null |
| +++ b/internal/lease/staging_test.go |
| @@ -0,0 +1,311 @@ |
| +package lease_test |
| + |
| +import ( |
| + "context" |
| + "crypto/hmac" |
| + "crypto/md5" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "net/http" |
| + "net/http/httptest" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/aws/aws-sdk-go-v2/aws" |
| + |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/lease/leasetest" |
| +) |
| + |
| +const ( |
| + stagingAK = "AKIATEST" |
| + stagingSecret = "test-secret" |
| + stagingRegion = "eu-west-2" |
| +) |
| + |
| +var stagingCreds = aws.CredentialsProviderFunc(func(context.Context) (aws.Credentials, error) { |
| + return aws.Credentials{AccessKeyID: stagingAK, SecretAccessKey: stagingSecret}, nil |
| +}) |
| + |
| +func hmacSHA(key []byte, s string) []byte { |
| + h := hmac.New(sha256.New, key) |
| + h.Write([]byte(s)) |
| + return h.Sum(nil) |
| +} |
| + |
| +// verifySigV4 recomputes the signature of a received request from first |
| +// principles (the AWS spec's canonical request and key derivation, not the |
| +// SDK's code), so the hand-built request is checked against an independent |
| +// reading of what S3 would check. It returns the X-Amz-Date it saw. |
| +func verifySigV4(r *http.Request) error { |
| + auth := r.Header.Get("Authorization") |
| + rest, ok := strings.CutPrefix(auth, "AWS4-HMAC-SHA256 ") |
| + if !ok { |
| + return fmt.Errorf("authorization %q", auth) |
| + } |
| + f := map[string]string{} |
| + for _, part := range strings.Split(rest, ", ") { |
| + k, v, _ := strings.Cut(part, "=") |
| + f[k] = v |
| + } |
| + date := r.Header.Get("X-Amz-Date") |
| + scope := date[:8] + "/" + stagingRegion + "/s3/aws4_request" |
| + if f["Credential"] != stagingAK+"/"+scope { |
| + return fmt.Errorf("credential scope %q, want %q", f["Credential"], scope) |
| + } |
| + var canon strings.Builder |
| + canon.WriteString(r.Method + "\n") |
| + canon.WriteString(strings.SplitN(r.RequestURI, "?", 2)[0] + "\n") // the path as sent: S3 canonicalizes it once-encoded |
| + canon.WriteString(r.URL.RawQuery + "\n") |
| + for _, h := range strings.Split(f["SignedHeaders"], ";") { |
| + v := r.Header.Get(h) |
| + if h == "host" { |
| + v = r.Host |
| + } |
| + if h == "content-length" { |
| + v = fmt.Sprint(r.ContentLength) |
| + } |
| + canon.WriteString(h + ":" + strings.TrimSpace(v) + "\n") |
| + } |
| + canon.WriteString("\n" + f["SignedHeaders"] + "\n" + r.Header.Get("X-Amz-Content-Sha256")) |
| + sum := sha256.Sum256([]byte(canon.String())) |
| + sts := "AWS4-HMAC-SHA256\n" + date + "\n" + scope + "\n" + hex.EncodeToString(sum[:]) |
| + k := hmacSHA([]byte("AWS4"+stagingSecret), date[:8]) |
| + k = hmacSHA(k, stagingRegion) |
| + k = hmacSHA(k, "s3") |
| + k = hmacSHA(k, "aws4_request") |
| + if got := hex.EncodeToString(hmacSHA(k, sts)); got != f["Signature"] { |
| + return fmt.Errorf("signature mismatch\ncanonical request:\n%s", canon.String()) |
| + } |
| + return nil |
| +} |
| + |
| +type recorded struct { |
| + method, uri, date, payload, contentType, copySource, ifMatch, ifNoneMatch, body string |
| + sigErr error |
| +} |
| + |
| +// stagingFixture is a staging mirror over leasetest's local S3, with every |
| +// request recorded and its signature verified on arrival. Optionally an |
| +// override answers instead of the model (a failing S3). |
| +func stagingFixture(t *testing.T, override func(w http.ResponseWriter, r *http.Request) bool) (*lease.StagingMirror, *leasetest.VersionedBucket, func() []recorded) { |
| + t.Helper() |
| + bucket := leasetest.NewVersionedBucket() |
| + bucket.AddBucket("data-uk") |
| + bucket.AddBucket("data-ie") |
| + handler, _ := leasetest.NewS3Server(bucket) |
| + var mu sync.Mutex |
| + var reqs []recorded |
| + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + body, _ := io.ReadAll(r.Body) |
| + mu.Lock() |
| + reqs = append(reqs, recorded{r.Method, r.RequestURI, r.Header.Get("X-Amz-Date"), r.Header.Get("X-Amz-Content-Sha256"), |
| + r.Header.Get("Content-Type"), r.Header.Get("X-Amz-Copy-Source"), r.Header.Get("X-Amz-Copy-Source-If-Match"), |
| + r.Header.Get("If-None-Match"), string(body), verifySigV4(r)}) |
| + mu.Unlock() |
| + r.Body = io.NopCloser(strings.NewReader(string(body))) |
| + if override != nil && override(w, r) { |
| + return |
| + } |
| + handler.ServeHTTP(w, r) |
| + })) |
| + t.Cleanup(srv.Close) |
| + return lease.NewStagingMirrorAt(srv.URL, stagingRegion, stagingCreds), bucket, func() []recorded { |
| + mu.Lock() |
| + defer mu.Unlock() |
| + return append([]recorded(nil), reqs...) |
| + } |
| +} |
| + |
| +var signAt = time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC) |
| + |
| +func TestStagingMirrorCopyVersioned(t *testing.T) { |
| + m, bucket, reqs := stagingFixture(t, nil) |
| + bucket.Put("data-ie", "instances/h/generations/0 1+a/lvl.ltx", []byte("body")) |
| + etag := fmt.Sprintf("%x", md5sum("body")) |
| + ctx := context.Background() |
| + gotETag, ver, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "instances/h/generations/0 1+a/lvl.ltx", etag, "data-uk", "moves/h/op/replica/x y.ltx") |
| + if err != nil || gotETag != etag || ver == "" { |
| + t.Fatalf("copy: etag %q ver %q err %v", gotETag, ver, err) |
| + } |
| + if got, _ := bucket.Current("data-uk", "moves/h/op/replica/x y.ltx"); string(got) != "body" { |
| + t.Fatalf("destination %q", got) |
| + } |
| + r := reqs()[0] |
| + if r.sigErr != nil { |
| + t.Fatal(r.sigErr) |
| + } |
| + if r.date != "20261006T090000Z" || r.payload != "UNSIGNED-PAYLOAD" || r.method != "PUT" { |
| + t.Fatalf("request %+v", r) |
| + } |
| + if r.copySource != "/data-ie/instances/h/generations/0%201%2Ba/lvl.ltx" || r.ifMatch != `"`+etag+`"` { |
| + t.Fatalf("copy headers: source %q if-match %q", r.copySource, r.ifMatch) |
| + } |
| + if r.uri != "/data-uk/moves/h/op/replica/x%20y.ltx" { |
| + t.Fatalf("destination path %q", r.uri) |
| + } |
| + // no source ETag: no precondition header |
| + if _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "instances/h/generations/0 1+a/lvl.ltx", "", "data-uk", "moves/h/op/replica/z"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rs := reqs(); rs[1].ifMatch != "" || rs[1].sigErr != nil { |
| + t.Fatalf("unconditional copy: %+v", rs[1]) |
| + } |
| +} |
| + |
| +func md5sum(s string) [16]byte { return md5.Sum([]byte(s)) } |
| + |
| +func TestStagingMirrorCopyFailures(t *testing.T) { |
| + m, bucket, reqs := stagingFixture(t, func(w http.ResponseWriter, r *http.Request) bool { |
| + switch { |
| + case strings.Contains(r.URL.Path, "late"): |
| + w.WriteHeader(200) |
| + io.WriteString(w, "<Error><Code>InternalError</Code><Message>x</Message></Error>") |
| + return true |
| + case strings.Contains(r.URL.Path, "boom"): |
| + w.WriteHeader(503) |
| + io.WriteString(w, "<Error><Code>SlowDown</Code><Message>x</Message></Error>") |
| + return true |
| + case strings.Contains(r.URL.Path, "empty"): |
| + w.WriteHeader(200) |
| + return true |
| + } |
| + return false |
| + }) |
| + bucket.Put("data-ie", "k", []byte("body")) |
| + ctx := context.Background() |
| + // the source changed since it was listed |
| + if _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "0000", "data-uk", "d"); !errors.Is(err, lease.ErrCopySourceChanged) { |
| + t.Fatalf("stale source: %v", err) |
| + } |
| + if _, ok := bucket.Current("data-uk", "d"); ok { |
| + t.Fatal("a stale-source copy landed") |
| + } |
| + // a 200 carrying an <Error> body is a failure |
| + if _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "late"); err == nil || !strings.Contains(err.Error(), "InternalError") { |
| + t.Fatalf("late failure: %v", err) |
| + } |
| + if _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "empty"); err == nil { |
| + t.Fatal("a 200 with no result is not a success") |
| + } |
| + // any other status names the code, and is sent once (no retry) |
| + before := len(reqs()) |
| + _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "boom") |
| + if err == nil || !strings.Contains(err.Error(), "SlowDown") || errors.Is(err, lease.ErrCopySourceChanged) { |
| + t.Fatalf("503: %v", err) |
| + } |
| + if n := len(reqs()) - before; n != 1 { |
| + t.Fatalf("a failing request was sent %d times", n) |
| + } |
| +} |
| + |
| +func TestStagingMirrorPutAndDelete(t *testing.T) { |
| + m, bucket, reqs := stagingFixture(t, nil) |
| + ctx := context.Background() |
| + v1, err := m.PutVersionedAt(ctx, signAt, "data-uk", "moves/h/op/timers/t.json", []byte(`{"a":1}`)) |
| + if err != nil || v1 == "" { |
| + t.Fatalf("put: %q %v", v1, err) |
| + } |
| + v2, err := m.PutVersionedAt(ctx, signAt, "data-uk", "moves/h/op/timers/t.json", []byte(`{"a":2}`)) |
| + if err != nil || v2 == "" || v2 == v1 { |
| + t.Fatalf("overwrite: %q %v", v2, err) |
| + } |
| + sum := sha256.Sum256([]byte(`{"a":1}`)) |
| + r := reqs()[0] |
| + if r.sigErr != nil || r.payload != hex.EncodeToString(sum[:]) || r.contentType != "application/json" || r.ifNoneMatch != "" || r.body != `{"a":1}` { |
| + t.Fatalf("put request %+v", r) |
| + } |
| + if _, err := m.PutIfAbsentVersionedAt(ctx, signAt, "data-uk", "moves/h/op/move.json", []byte("{}")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rs := reqs(); rs[2].ifNoneMatch != "*" || rs[2].sigErr != nil { |
| + t.Fatalf("create-only request %+v", rs[2]) |
| + } |
| + if _, err := m.PutIfAbsentVersionedAt(ctx, signAt, "data-uk", "moves/h/op/move.json", []byte("{}")); !errors.Is(err, lease.ErrExists) { |
| + t.Fatalf("second create-only: %v", err) |
| + } |
| + // an empty body still signs and lands |
| + if _, err := m.PutVersionedAt(ctx, signAt, "data-uk", "empty", nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := m.DeleteAt(ctx, signAt, "data-uk", "moves/h/op/timers/t.json"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, ok := bucket.Current("data-uk", "moves/h/op/timers/t.json"); ok { |
| + t.Fatal("still current after DeleteAt") |
| + } |
| + for i, r := range reqs() { |
| + if r.sigErr != nil { |
| + t.Fatalf("request %d: %v", i, r.sigErr) |
| + } |
| + } |
| + // a missing bucket is an error naming the code |
| + if _, err := m.PutVersionedAt(ctx, signAt, "nope", "k", []byte("x")); err == nil || !strings.Contains(err.Error(), "NoSuchBucket") { |
| + t.Fatalf("missing bucket: %v", err) |
| + } |
| +} |
| + |
| +// The signature carries the timestamp it was given and nothing else: the |
| +// same inputs sign identically however much later they are sent, and a |
| +// signAt in the far past is used as is (the mirror reads no clock). |
| +func TestStagingMirrorSignsOnlyAtSignAt(t *testing.T) { |
| + var auths []string |
| + m, _, reqs := stagingFixture(t, func(w http.ResponseWriter, r *http.Request) bool { |
| + auths = append(auths, r.Header.Get("Authorization")) |
| + return false |
| + }) |
| + ctx := context.Background() |
| + for i := 0; i < 2; i++ { |
| + if _, err := m.PutVersionedAt(ctx, signAt, "data-uk", "k", []byte("x")); err != nil { |
| + t.Fatal(err) |
| + } |
| + time.Sleep(1100 * time.Millisecond) // a second boundary passes between sends |
| + } |
| + if len(auths) != 2 || auths[0] != auths[1] { |
| + t.Fatalf("same signAt, different signatures: %q", auths) |
| + } |
| + old := time.Date(2001, 1, 2, 3, 4, 5, 0, time.UTC) |
| + if _, err := m.PutVersionedAt(ctx, old, "data-uk", "k", []byte("x")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rs := reqs(); rs[2].date != "20010102T030405Z" { |
| + t.Fatalf("date %q", rs[2].date) |
| + } |
| +} |
| + |
| +// A mutation without a signing time (a caller that bypassed the guard) |
| +// sends nothing. |
| +func TestStagingMirrorRefusesAZeroSignTime(t *testing.T) { |
| + m, _, reqs := stagingFixture(t, nil) |
| + ctx := context.Background() |
| + if _, err := m.PutVersionedAt(ctx, time.Time{}, "data-uk", "k", []byte("x")); err == nil { |
| + t.Fatal("put signed at the zero time") |
| + } |
| + if _, _, err := m.CopyVersionedAt(ctx, time.Time{}, "data-ie", "k", "", "data-uk", "d"); err == nil { |
| + t.Fatal("copy signed at the zero time") |
| + } |
| + if err := m.DeleteAt(ctx, time.Time{}, "data-uk", "k"); err == nil { |
| + t.Fatal("delete signed at the zero time") |
| + } |
| + if n := len(reqs()); n != 0 { |
| + t.Fatalf("%d requests sent", n) |
| + } |
| +} |
| + |
| +// Get is the ordinary read, delegated. |
| +func TestStagingMirrorGetDelegates(t *testing.T) { |
| + m, bucket, _ := stagingFixture(t, nil) |
| + bucket.Put("data-uk", "r", []byte("receipts")) |
| + got, err := m.Get(context.Background(), "data-uk", "r") |
| + if err != nil || string(got) != "receipts" { |
| + t.Fatalf("get: %q %v", got, err) |
| + } |
| + if _, err := m.Get(context.Background(), "data-uk", "absent"); !errors.Is(err, lease.ErrMirrorNotFound) { |
| + t.Fatalf("get of an absent key: %v", err) |
| + } |
| +} |
diff --git a/internal/lease/staging_url_test.go b/internal/lease/staging_url_test.go| new file mode 100644 |
| index 000000000..b844e23b7 |
| --- /dev/null |
| +++ b/internal/lease/staging_url_test.go |
| @@ -0,0 +1,20 @@ |
| +package lease |
| + |
| +import "testing" |
| + |
| +func TestStagingObjectURL(t *testing.T) { |
| + cases := []struct { |
| + name, endpoint, bucket, key, want string |
| + }{ |
| + {"virtual-hosted", "", "data-uk", "moves/h/op/a b+c", "https://data-uk.s3.eu-west-2.amazonaws.com/moves/h/op/a%20b%2Bc"}, |
| + {"dotted bucket goes path-style", "", "data.uk", "k", "https://s3.eu-west-2.amazonaws.com/data.uk/k"}, |
| + {"endpoint override is path-style", "https://store.example:9000", "data-uk", "a/é", "https://store.example:9000/data-uk/a/%C3%A9"}, |
| + } |
| + for _, c := range cases { |
| + m := &StagingMirror{endpoint: c.endpoint, region: "eu-west-2"} |
| + u, err := m.objectURL(c.bucket, c.key) |
| + if err != nil || u.String() != c.want { |
| + t.Errorf("%s: got %v %v, want %s", c.name, u, err, c.want) |
| + } |
| + } |
| +} |
diff --git a/internal/moves/lease.go b/internal/moves/lease.go| new file mode 100644 |
| index 000000000..bc3211fd4 |
| --- /dev/null |
| +++ b/internal/moves/lease.go |
| @@ -0,0 +1,70 @@ |
| +package moves |
| + |
| +import ( |
| + "errors" |
| + "fmt" |
| + "time" |
| +) |
| + |
| +// SignatureLife is how long S3 accepts a SigV4 signature (15 minutes, |
| +// verified §15 step 8). A staging writer may sign only while its lease is |
| +// safe for that whole life plus the clock skew (§7.3, model 14, 18). |
| +const SignatureLife = 15 * time.Minute |
| + |
| +// StagingLeaseTTL is the staging-work lease's term (D3). The ordinary |
| +// lease.LeaseTTL is 15 minutes, which can never leave SignatureLife plus |
| +// the skew in front of SafeUntil; 45 minutes gives every renewal cycle |
| +// room, and the renewer keeps the bound ahead while the lane works. |
| +const StagingLeaseTTL = 45 * time.Minute |
| + |
| +// DefaultSkew is the clock-skew allowance the guard subtracts. |
| +const DefaultSkew = time.Minute |
| + |
| +// ErrLeaseUnsafe is the guard's refusal: the staging lease is too close to |
| +// its end (or already lost) for another request to be signed. |
| +var ErrLeaseUnsafe = errors.New("moves: staging lease too close to its end to sign another request") |
| + |
| +// StagingLeasePrefix is the claim's shared staging-work lease (claim C3, |
| +// §7.3 D3): in the deployment bucket, keyed by target region and host, so |
| +// a source that cannot read the target's lock.json still excludes other |
| +// writers of the same lane. |
| +func StagingLeasePrefix(targetRegion, host string) string { |
| + return "control/staging-leases/" + targetRegion + "/" + host |
| +} |
| + |
| +// SafeUntilGuard is the one method of *lease.Held the guard reads. |
| +type SafeUntilGuard interface{ SafeUntil() (time.Time, error) } |
| + |
| +// Guarded wraps every staging request: the check runs immediately before |
| +// fn signs, and fn signs AT THE CHECK'S TIMESTAMP, which Before hands it. |
| +// A request that cannot complete within the signature's life before the |
| +// lease ends is dropped; a request signed at the check and sent after any |
| +// pause carries a date S3 stops accepting exactly when the guard's bound |
| +// said it must. fn must issue exactly one request through a client that |
| +// signs once, at signAt, and never retries (lease.StagingMirror's At |
| +// methods); a caller that wants a retry calls Before again. |
| +type Guarded struct { |
| + Lease SafeUntilGuard |
| + Now func() time.Time |
| + Skew time.Duration // zero means DefaultSkew |
| +} |
| + |
| +// Before refuses with ErrLeaseUnsafe unless the lease stays safe for the |
| +// signature's whole life plus the skew from now, where now is read once |
| +// and handed to fn as the signing time. lease.Held.SafeUntil already |
| +// subtracts the renewal margin; Before subtracts the signature's life. |
| +func (g Guarded) Before(fn func(signAt time.Time) error) error { |
| + safe, err := g.Lease.SafeUntil() |
| + if err != nil { |
| + return fmt.Errorf("%w: %v", ErrLeaseUnsafe, err) |
| + } |
| + skew := g.Skew |
| + if skew == 0 { |
| + skew = DefaultSkew |
| + } |
| + now := g.Now() |
| + if now.Add(SignatureLife + skew).After(safe) { |
| + return ErrLeaseUnsafe |
| + } |
| + return fn(now) |
| +} |
diff --git a/internal/moves/lease_test.go b/internal/moves/lease_test.go| new file mode 100644 |
| index 000000000..179eaf6cc |
| --- /dev/null |
| +++ b/internal/moves/lease_test.go |
| @@ -0,0 +1,161 @@ |
| +package moves |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "net/http" |
| + "net/http/httptest" |
| + "testing" |
| + "time" |
| + |
| + "github.com/aws/aws-sdk-go-v2/aws" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/lease/leasetest" |
| +) |
| + |
| +type fixedLease struct{ until time.Time } |
| + |
| +func (f fixedLease) SafeUntil() (time.Time, error) { return f.until, nil } |
| + |
| +type failedLease struct{ err error } |
| + |
| +func (f failedLease) SafeUntil() (time.Time, error) { return time.Time{}, f.err } |
| + |
| +// §7.3 "sign only before SafeUntil, and sign at the check": the guard |
| +// leaves room for the signature's whole valid life and the clock skew. |
| +func TestGuardedRefusesWithinTheSignatureLife(t *testing.T) { |
| + t0 := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC) |
| + g := Guarded{Lease: fixedLease{t0.Add(30 * time.Minute)}, Now: func() time.Time { return t0 }, Skew: time.Minute} |
| + ran := 0 |
| + var got time.Time |
| + if err := g.Before(func(at time.Time) error { ran++; got = at; return nil }); err != nil || ran != 1 || !got.Equal(t0) { |
| + t.Fatalf("29 minutes of room: err %v ran %d signAt %v", err, ran, got) |
| + } |
| + g.Now = func() time.Time { return t0.Add(14*time.Minute + 1*time.Second) } |
| + if err := g.Before(func(time.Time) error { ran++; return nil }); !errors.Is(err, ErrLeaseUnsafe) || ran != 1 { |
| + t.Fatalf("inside the 15 min + skew window: err %v ran %d", err, ran) |
| + } |
| + if StagingLeasePrefix("uk", "a.go.example.com") != "control/staging-leases/uk/a.go.example.com" { |
| + t.Fatal(StagingLeasePrefix("uk", "a.go.example.com")) |
| + } |
| + // The real settings: a fresh staging lease (45 min TTL) is usable for |
| + // its first 29 minutes minus the renewal margin; the ordinary 15 min |
| + // lease would never pass the guard, which is why StagingLeaseTTL exists. |
| + if StagingLeaseTTL <= SignatureLife+DefaultSkew+15*time.Minute { |
| + t.Fatalf("StagingLeaseTTL %v leaves no room for the signature life", StagingLeaseTTL) |
| + } |
| +} |
| + |
| +// A lease that cannot answer SafeUntil (lost, spent) is unsafe, and fn |
| +// never runs. |
| +func TestGuardedRefusesALostLease(t *testing.T) { |
| + g := Guarded{Lease: failedLease{lease.ErrLost}, Now: time.Now} |
| + err := g.Before(func(time.Time) error { t.Fatal("ran on a lost lease"); return nil }) |
| + if !errors.Is(err, ErrLeaseUnsafe) { |
| + t.Fatalf("lost lease: %v", err) |
| + } |
| +} |
| + |
| +// The boundary is exact: now + life + skew == safe is allowed, one |
| +// nanosecond more is not. A zero Skew means DefaultSkew. |
| +func TestGuardedBoundary(t *testing.T) { |
| + t0 := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC) |
| + safe := t0.Add(SignatureLife + DefaultSkew) |
| + ok := Guarded{Lease: fixedLease{safe}, Now: func() time.Time { return t0 }} |
| + if err := ok.Before(func(time.Time) error { return nil }); err != nil { |
| + t.Fatalf("exactly at the bound: %v", err) |
| + } |
| + late := Guarded{Lease: fixedLease{safe}, Now: func() time.Time { return t0.Add(1) }} |
| + if err := late.Before(func(time.Time) error { return nil }); !errors.Is(err, ErrLeaseUnsafe) { |
| + t.Fatalf("one ns past the bound: %v", err) |
| + } |
| + // fn's own error comes back untouched. |
| + boom := errors.New("boom") |
| + if err := ok.Before(func(time.Time) error { return boom }); err != boom { |
| + t.Fatalf("fn error: %v", err) |
| + } |
| +} |
| + |
| +// The real lease settings, by arithmetic: lease.Held's SafeUntil is |
| +// Until minus the renewal margin, so a fresh lease's SafeUntil is at most |
| +// TTL ahead. A fresh StagingLeaseTTL lease passes the guard; a fresh |
| +// lease.LeaseTTL lease never can, which is why StagingLeaseTTL exists. |
| +// lease.AcquireTTL itself is covered where lease's own Acquire tests are |
| +// (lease_test.go, S3-gated): this package adds no second S3 fixture. |
| +func TestGuardAgainstRealLeaseSettings(t *testing.T) { |
| + t0 := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC) |
| + fresh := func(ttl time.Duration) SafeUntilGuard { |
| + return fixedLease{t0.Add(ttl - lease.RenewMargin)} // what Held.SafeUntil answers right after Acquire |
| + } |
| + if err := (Guarded{Lease: fresh(StagingLeaseTTL), Now: func() time.Time { return t0 }}).Before(func(time.Time) error { return nil }); err != nil { |
| + t.Fatalf("a fresh staging lease is refused: %v", err) |
| + } |
| + if err := (Guarded{Lease: fresh(lease.LeaseTTL), Now: func() time.Time { return t0 }}).Before(func(time.Time) error { return nil }); !errors.Is(err, ErrLeaseUnsafe) { |
| + t.Fatalf("a fresh 15 minute lease passed the guard: %v", err) |
| + } |
| +} |
| + |
| +// The staging mirror signs at the check and never again: a request |
| +// carries the X-Amz-Date of the timestamp Guarded.Before captured, and a |
| +// second attempt is a second Before. The fixture is a recording handler |
| +// (the test asserts the signed header, not AWS's acceptance of it) in |
| +// front of internal/lease/leasetest's local S3, which stores a plain PUT |
| +// as a new version beside its create-only PUT. |
| +func TestStagingMirrorSignsAtTheCheck(t *testing.T) { |
| + bucket := leasetest.NewVersionedBucket() |
| + bucket.AddBucket("data-uk") |
| + handler, _ := leasetest.NewS3Server(bucket, "data-uk") |
| + var dates []string |
| + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + dates = append(dates, r.Header.Get("X-Amz-Date")) |
| + handler.ServeHTTP(w, r) |
| + })) |
| + defer srv.Close() |
| + creds := aws.CredentialsProviderFunc(func(context.Context) (aws.Credentials, error) { |
| + return aws.Credentials{AccessKeyID: "AKIATEST", SecretAccessKey: "secret"}, nil |
| + }) |
| + m := lease.NewStagingMirrorAt(srv.URL, "eu-west-2", creds) |
| + t0 := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC) |
| + now := t0 |
| + g := Guarded{Lease: fixedLease{t0.Add(30 * time.Minute)}, Now: func() time.Time { return now }} |
| + // First write: checked and signed at t0. |
| + if err := g.Before(func(at time.Time) error { |
| + _, err := m.PutVersionedAt(context.Background(), at, "data-uk", "moves/h/mv/timers/t.json", []byte("{}")) |
| + return err |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + // Second write: the check happens at t0+3m; then the goroutine pauses |
| + // past every bound before the request is signed and sent. The |
| + // signature must carry the CHECK's time, so no pause can mint a fresh |
| + // signature after the lease is gone (round-4 finding 3). |
| + now = t0.Add(3 * time.Minute) |
| + if err := g.Before(func(at time.Time) error { |
| + now = t0.Add(40 * time.Minute) // the pause: the wall clock is past the lease |
| + _, err := m.PutVersionedAt(context.Background(), at, "data-uk", "moves/h/mv/timers/t.json", []byte("{}")) |
| + return err |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(dates) != 2 || dates[0] != "20261006T090000Z" || dates[1] != "20261006T090300Z" { |
| + t.Fatalf("X-Amz-Date per request must be the check's time: %v", dates) |
| + } |
| + // The lease is now unsafe: a third Before refuses and sends nothing. |
| + if err := g.Before(func(time.Time) error { t.Fatal("signed after the bound"); return nil }); !errors.Is(err, ErrLeaseUnsafe) || len(dates) != 2 { |
| + t.Fatalf("after the bound: %v, %d requests", err, len(dates)) |
| + } |
| + // create-only through the same signed path |
| + now = t0.Add(4 * time.Minute) |
| + put := func() error { |
| + return g.Before(func(at time.Time) error { |
| + _, err := m.PutIfAbsentVersionedAt(context.Background(), at, "data-uk", "moves/h/mv/move.json", []byte("{}")) |
| + return err |
| + }) |
| + } |
| + if err := put(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := put(); !errors.Is(err, lease.ErrExists) { |
| + t.Fatalf("second create-only put: %v", err) |
| + } |
| +} |