From db451fe9b8d37f029f3f2c970e5239fd9f17cae6 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 30 Oct 2026 00:59:05 +0900 Subject: [PATCH 1/8] docs: define S3 multipart read-set completion criteria Signed-off-by: bootjp --- ...26_partial_serializable_isolation_audit.md | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/docs/design/2026_09_26_partial_serializable_isolation_audit.md b/docs/design/2026_09_26_partial_serializable_isolation_audit.md index d7838c55f..eaa7e7baf 100644 --- a/docs/design/2026_09_26_partial_serializable_isolation_audit.md +++ b/docs/design/2026_09_26_partial_serializable_isolation_audit.md @@ -1125,6 +1125,41 @@ code at fix time and turns the `N` rows into fixes: - gRPC: none in this audit; the API gap (G8) is documented in the positioning doc and README. +#### S3 A1 completion plan (2026-10-30) + +The unpublished branch described below is historical evidence, not a merge +or acceptance receipt. Implementation now proceeds from main `c316b829e` on +`design/serializable-audit-a1-s3-validation`; the open A0 PR #1355 is separate. + +- Every completion attempt acquires one read timestamp, loads the bucket, + upload metadata, requested part descriptors and replaced manifest at that + snapshot, then commits with every consumed descriptor key in `ReadKeys`. + A retry repeats those reads and ETag checks rather than reusing a manifest + assembled from an earlier snapshot. +- UploadPart keeps its streamed blob namespace and read voucher, but checks + the upload, bucket generation and superseded descriptor again before its + descriptor commit. The read set contains the upload metadata, generation + key and part key. Bucket deletion re-puts the generation key, preserving + its maximum value, in both public and admin deletion transactions. Other + object writers do not write that key, so parallel unrelated uploads do + not acquire an accidental bucket-wide conflict. +- Support all 10,000 requested parts during staged-visibility migration as + well as normal routing. The physical read-key cap is bounded at + `2 * (10_000 + 240)` to include migration aliases and control-key headroom. + Retain every existing cap check and every alias; do not accept the older + branch's approximately 4,998-part migration limitation as completion. +- Multipart conflicts that are proven noncommits can use the existing + bounded completion retry or a 503 SlowDown response for the consumed + upload body. Unknown outcomes must not become permission to replay. +- Red/green regressions run through the production ShardedCoordinator/FSM, + including both part-read windows, deletion/recreation, superseded blob + cleanup, unrelated bucket writes, maximum part counts and alias capacity. + Test server-side data and cleanup effects, not only HTTP status codes. +- Verify all changed helper callers, full adapter/KV/store race suites, + unfiltered lint, relevant Jepsen checks, and latest-push review/CI. A1 does + not establish cross-group serializability before A2's read-lock protocol, + nor close any other audit milestone. + Status of the S3 half: landed on branch `design/serializable-audit-a1-s3` (from `main`, not pushed): `3a975625` adds `adapter/s3_multipart_occ_repro_test.go`, four cases through the real FSM From 093f07cfd76e7f96ac3bce7105cf1cb086d847b8 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 30 Oct 2026 01:02:43 +0900 Subject: [PATCH 2/8] docs: require snapshot-coherent multipart cleanup outcomes Signed-off-by: bootjp --- ...26_09_26_partial_serializable_isolation_audit.md | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/design/2026_09_26_partial_serializable_isolation_audit.md b/docs/design/2026_09_26_partial_serializable_isolation_audit.md index eaa7e7baf..4c9706f7a 100644 --- a/docs/design/2026_09_26_partial_serializable_isolation_audit.md +++ b/docs/design/2026_09_26_partial_serializable_isolation_audit.md @@ -1137,8 +1137,11 @@ or acceptance receipt. Implementation now proceeds from main `c316b829e` on A retry repeats those reads and ETag checks rather than reusing a manifest assembled from an earlier snapshot. - UploadPart keeps its streamed blob namespace and read voucher, but checks - the upload, bucket generation and superseded descriptor again before its - descriptor commit. The read set contains the upload metadata, generation + the upload, bucket generation and superseded descriptor again at exactly + the dispatched `StartTS` before its descriptor commit, using the original + pre-body read timestamp voucher and retained snapshot. Do not replace that + snapshot with a local/latest watermark or rename the already-streamed blob + and chunkref namespaces. The read set contains the upload metadata, generation key and part key. Bucket deletion re-puts the generation key, preserving its maximum value, in both public and admin deletion transactions. Other object writers do not write that key, so parallel unrelated uploads do @@ -1151,6 +1154,12 @@ or acceptance receipt. Implementation now proceeds from main `c316b829e` on - Multipart conflicts that are proven noncommits can use the existing bounded completion retry or a 503 SlowDown response for the consumed upload body. Unknown outcomes must not become permission to replay. + Track whether descriptor dispatch was attempted: clean new blobs after + pre-dispatch failure or a proven noncommit, but preserve them after an + uncertain dispatch result. Clean superseded blobs only after confirmed + success. The unknown marker takes precedence over any joined conflict, + lock or route-fence cause. This prevents a lost success reply from deleting + blobs that a committed descriptor may already reference. - Red/green regressions run through the production ShardedCoordinator/FSM, including both part-read windows, deletion/recreation, superseded blob cleanup, unrelated bucket writes, maximum part counts and alias capacity. From 9dd07b25bfdb1d606c2135347a20131e8c5525bb Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 30 Oct 2026 01:11:28 +0900 Subject: [PATCH 3/8] docs: preserve inline multipart namespaces across commit allocation Signed-off-by: bootjp --- .../2026_09_26_partial_serializable_isolation_audit.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/docs/design/2026_09_26_partial_serializable_isolation_audit.md b/docs/design/2026_09_26_partial_serializable_isolation_audit.md index 4c9706f7a..9e6ef42b3 100644 --- a/docs/design/2026_09_26_partial_serializable_isolation_audit.md +++ b/docs/design/2026_09_26_partial_serializable_isolation_audit.md @@ -1141,7 +1141,10 @@ or acceptance receipt. Implementation now proceeds from main `c316b829e` on the dispatched `StartTS` before its descriptor commit, using the original pre-body read timestamp voucher and retained snapshot. Do not replace that snapshot with a local/latest watermark or rename the already-streamed blob - and chunkref namespaces. The read set contains the upload metadata, generation + and chunkref namespaces. Allocate the descriptor commit timestamp after + chunk durability in both inline and offloaded modes; keep the reserved + inline blob version separate from the descriptor's MVCC commit timestamp. + The read set contains the upload metadata, generation key and part key. Bucket deletion re-puts the generation key, preserving its maximum value, in both public and admin deletion transactions. Other object writers do not write that key, so parallel unrelated uploads do From b047e029ee5e4d38336f3a93ffc2e10de6b48acf Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 30 Oct 2026 01:29:36 +0900 Subject: [PATCH 4/8] docs: require multipart register and final-read acceptance Signed-off-by: bootjp --- .../2026_09_26_partial_serializable_isolation_audit.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/design/2026_09_26_partial_serializable_isolation_audit.md b/docs/design/2026_09_26_partial_serializable_isolation_audit.md index 9e6ef42b3..eea985e1b 100644 --- a/docs/design/2026_09_26_partial_serializable_isolation_audit.md +++ b/docs/design/2026_09_26_partial_serializable_isolation_audit.md @@ -1167,6 +1167,13 @@ or acceptance receipt. Implementation now proceeds from main `c316b829e` on including both part-read windows, deletion/recreation, superseded blob cleanup, unrelated bucket writes, maximum part counts and alias capacity. Test server-side data and cleanup effects, not only HTTP status codes. +- Add an opt-in S3 multipart register mode to the existing Jepsen workload. + Each write races same-body part replacement with completion under one + upload ID; the expected object remains the register value regardless of + which immutable part version wins. Preserve the existing linearizability + checker and honest indeterminate outcomes. Require complete successful + final reads so lost referenced chunks cannot be hidden by failed GETs. + Keep the default PUT/GET workload unchanged and run both modes in CI. - Verify all changed helper callers, full adapter/KV/store race suites, unfiltered lint, relevant Jepsen checks, and latest-push review/CI. A1 does not establish cross-group serializability before A2's read-lock protocol, From aa2ef8d5580819715ce175fd18de19ef5cc7ccde Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 30 Oct 2026 01:45:47 +0900 Subject: [PATCH 5/8] docs: require sustained multipart acceptance and retained histories Signed-off-by: bootjp --- .../2026_09_26_partial_serializable_isolation_audit.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/docs/design/2026_09_26_partial_serializable_isolation_audit.md b/docs/design/2026_09_26_partial_serializable_isolation_audit.md index eea985e1b..ec9924433 100644 --- a/docs/design/2026_09_26_partial_serializable_isolation_audit.md +++ b/docs/design/2026_09_26_partial_serializable_isolation_audit.md @@ -1174,6 +1174,12 @@ or acceptance receipt. Implementation now proceeds from main `c316b829e` on checker and honest indeterminate outcomes. Require complete successful final reads so lost referenced chunks cannot be hidden by failed GETs. Keep the default PUT/GET workload unchanged and run both modes in CI. + Run the new multipart mode for 30 seconds at rate 10 and concurrency 10 + in each encryption job; a five-second sample produced no confirmed writes + under contention in one local run and must not be accepted as evidence. + Preserve both named S3 histories, source SHA, encryption mode and step + outcomes as unconditional CI artifacts. Do not weaken the nonvacuity or + final-read gate to obtain a passing result. - Verify all changed helper callers, full adapter/KV/store race suites, unfiltered lint, relevant Jepsen checks, and latest-push review/CI. A1 does not establish cross-group serializability before A2's read-lock protocol, From a46fba9767350874210931ad372a8a3a98d49469 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 30 Oct 2026 01:46:19 +0900 Subject: [PATCH 6/8] adapter: validate multipart snapshots and preserve uncertain writes Signed-off-by: bootjp --- .github/workflows/jepsen-test.yml | 30 + adapter/s3.go | 28 +- adapter/s3_admin.go | 10 +- adapter/s3_bucket_delete.go | 44 ++ adapter/s3_hlc_fence_test.go | 24 +- adapter/s3_multipart_benchmark_test.go | 54 ++ adapter/s3_multipart_complete.go | 104 ++- adapter/s3_multipart_occ_repro_test.go | 760 +++++++++++++++++++++ adapter/s3_multipart_read_set_test.go | 307 +++++++++ adapter/s3_mutation_outcome_test.go | 120 ++++ adapter/s3_upload_part.go | 100 ++- jepsen/src/elastickv/s3_workload.clj | 219 +++++- jepsen/test/elastickv/s3_workload_test.clj | 351 +++++++++- kv/coordinator.go | 17 +- kv/read_keys_capacity_test.go | 302 ++++++++ 15 files changed, 2334 insertions(+), 136 deletions(-) create mode 100644 adapter/s3_bucket_delete.go create mode 100644 adapter/s3_multipart_benchmark_test.go create mode 100644 adapter/s3_multipart_occ_repro_test.go create mode 100644 adapter/s3_multipart_read_set_test.go create mode 100644 adapter/s3_mutation_outcome_test.go create mode 100644 kv/read_keys_capacity_test.go diff --git a/.github/workflows/jepsen-test.yml b/.github/workflows/jepsen-test.yml index 71888e019..4181c747b 100644 --- a/.github/workflows/jepsen-test.yml +++ b/.github/workflows/jepsen-test.yml @@ -255,10 +255,40 @@ jobs: path: /tmp/jepsen-store-types.tar.gz retention-days: 7 - name: Run S3 Jepsen workload against elastickv + id: s3_register working-directory: jepsen timeout-minutes: 3 run: | timeout 120 ~/lein run -m elastickv.s3-workload --local --time-limit 5 --rate 10 --concurrency 10 --s3-ports 63901,63902,63903 --host 127.0.0.1 + - name: Run S3 multipart Jepsen workload against elastickv + id: s3_multipart + working-directory: jepsen + timeout-minutes: 3 + run: | + timeout 120 ~/lein run -m elastickv.s3-workload --local --multipart --time-limit 30 --rate 10 --concurrency 10 --s3-ports 63901,63902,63903 --host 127.0.0.1 + - name: Archive S3 acceptance histories + if: always() + env: + S3_REGISTER_OUTCOME: ${{ steps.s3_register.outcome }} + S3_MULTIPART_OUTCOME: ${{ steps.s3_multipart.outcome }} + ENCRYPTION_MODE: ${{ matrix.encryption }} + run: | + printf 'run_sha=%s\nencryption=%s\nregister=%s\nmultipart=%s\n' \ + "$GITHUB_SHA" "$ENCRYPTION_MODE" "$S3_REGISTER_OUTCOME" "$S3_MULTIPART_OUTCOME" \ + > /tmp/s3-acceptance-source.txt + bash scripts/archive-jepsen-store.sh jepsen/store/elastickv-s3-register /tmp/s3-register-history.tar.gz + bash scripts/archive-jepsen-store.sh jepsen/store/elastickv-s3-multipart-register /tmp/s3-multipart-history.tar.gz + - name: Upload S3 acceptance histories + if: always() + uses: actions/upload-artifact@v7 + with: + name: s3-acceptance-encryption-${{ matrix.encryption }} + path: | + /tmp/s3-register-history.tar.gz + /tmp/s3-multipart-history.tar.gz + /tmp/s3-acceptance-source.txt + retention-days: 14 + if-no-files-found: warn - name: Run SQS HT-FIFO Jepsen workload against elastickv working-directory: jepsen # The HT-FIFO workload runs sends and receives across a 4-partition diff --git a/adapter/s3.go b/adapter/s3.go index f6bb02b01..a6b094aa4 100644 --- a/adapter/s3.go +++ b/adapter/s3.go @@ -801,7 +801,6 @@ func (s *S3Server) deleteBucket(w http.ResponseWriter, r *http.Request, bucket s return errors.Wrap(err, "s3: allocate startTS for mutation") } readTS = readTimestamp.Timestamp() - startTS := readTS readPin := s.pinReadTS(readTS) defer readPin.Release() @@ -833,21 +832,16 @@ func (s *S3Server) deleteBucket(w http.ResponseWriter, r *http.Request, bucket s } // Phase 1: Del BucketMetaKey in a txn (OCC-protected - // against concurrent createBucket racing this delete). + // against concurrent createBucket racing this delete) and re-put + // the generation key to fence concurrent part uploads. // Phase 2 (DEL_PREFIX safety net) runs outside the txn // because the production coordinator rejects DEL_PREFIX // inside transactions and rejects mixed Del+DelPrefix // groups (kv/sharded_coordinator.go: dispatchDelPrefixBroadcast). // See AdminDeleteBucket's doc comment for the full // rationale. - dispatchCtx := readTimestamp.WithDispatchVoucher(r.Context()) - _, err = kv.DispatchWithReadTimestamp(dispatchCtx, s.coordinator, &kv.OperationGroup[kv.OP]{ - IsTxn: true, - StartTS: startTS, - Elems: []*kv.Elem[kv.OP]{{Op: kv.Del, Key: s3keys.BucketMetaKey(bucket)}}, - }) - if err != nil { - return errors.WithStack(err) + if err := s.dispatchBucketDeleteTxn(r.Context(), readTimestamp, bucket, meta.Generation); err != nil { + return err } deletedGeneration = meta.Generation return nil @@ -1390,7 +1384,7 @@ func (s *S3Server) uploadPart(w http.ResponseWriter, r *http.Request, bucket str r.Context(), r, streamBody, state, bucket, objectKey, uploadID, admissionProtocol, offload, ) if uploadErr != nil || uploadBodyErr != nil { - s.writeS3ChunkUploadError(w, uploadBodyErr, uploadErr, bucket, objectKey) + s.writeS3UploadPartError(w, uploadBodyErr, uploadErr, bucket, objectKey) return } @@ -1411,14 +1405,9 @@ func (s *S3Server) completeMultipartUpload(w http.ResponseWriter, r *http.Reques writeS3ResponseOrInternalError(w, err) return } - completion, err := s.loadS3MultipartCompletion(r.Context(), bucket, objectKey, uploadID, request) - if err != nil { - writeS3ResponseOrInternalError(w, err) - return - } - previous, err := s.commitS3MultipartCompletion(r.Context(), completion) + completion, previous, err := s.commitS3MultipartCompletion(r.Context(), bucket, objectKey, uploadID, request) if err != nil { - writeS3MutationError(w, err, bucket, objectKey) + writeS3MultipartMutationError(w, err, bucket, objectKey) return } @@ -2552,6 +2541,9 @@ func (s *S3Server) nextTxnCommitTS(ctx context.Context, startTS uint64) (uint64, } func isRetryableS3MutationErr(err error) bool { + if errors.Is(err, kv.ErrTransactionOutcomeUnknown) { + return false + } return errors.Is(err, store.ErrWriteConflict) || errors.Is(err, kv.ErrTxnLocked) || isRouteWriteFencedError(err) } diff --git a/adapter/s3_admin.go b/adapter/s3_admin.go index 36f9c781f..8ea405a76 100644 --- a/adapter/s3_admin.go +++ b/adapter/s3_admin.go @@ -444,7 +444,6 @@ func (s *S3Server) adminDeleteBucketTxnBody(ctx context.Context, name string, de return errors.Wrap(err, "s3 admin: allocate startTS for adminDeleteBucket") } readTS = readTimestamp.Timestamp() - startTS := readTS readPin := s.pinReadTS(readTS) defer readPin.Release() @@ -467,13 +466,8 @@ func (s *S3Server) adminDeleteBucketTxnBody(ctx context.Context, name string, de // AdminCreateBucket racing the delete is rejected by OCC. // retryS3Mutation handles ErrWriteConflict / ErrTxnLocked // by re-running this whole closure. - dispatchCtx := readTimestamp.WithDispatchVoucher(ctx) - if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, s.coordinator, &kv.OperationGroup[kv.OP]{ - IsTxn: true, - StartTS: startTS, - Elems: []*kv.Elem[kv.OP]{{Op: kv.Del, Key: s3keys.BucketMetaKey(name)}}, - }); err != nil { - return errors.WithStack(err) + if err := s.dispatchBucketDeleteTxn(ctx, readTimestamp, name, meta.Generation); err != nil { + return err } *deletedGeneration = meta.Generation return nil diff --git a/adapter/s3_bucket_delete.go b/adapter/s3_bucket_delete.go new file mode 100644 index 000000000..acda69b5e --- /dev/null +++ b/adapter/s3_bucket_delete.go @@ -0,0 +1,44 @@ +package adapter + +import ( + "context" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" +) + +func (s *S3Server) dispatchBucketDeleteTxn(ctx context.Context, snapshot kv.ReadTimestamp, bucket string, generation uint64) error { + elems, err := s.bucketDeleteTxnElems(ctx, bucket, generation, snapshot.Timestamp()) + if err != nil { + return err + } + _, err = kv.DispatchWithReadTimestamp(snapshot.WithDispatchVoucher(ctx), s.coordinator, &kv.OperationGroup[kv.OP]{ + IsTxn: true, StartTS: snapshot.Timestamp(), Elems: elems, + }) + return errors.WithStack(err) +} + +// Every object writer updates bucket metadata, but only create/delete update +// the generation key. Re-putting it fences streaming parts without making +// unrelated object writes abort them; preserving its maximum prevents reuse. +func (s *S3Server) bucketDeleteTxnElems(ctx context.Context, bucket string, generation, readTS uint64) ([]*kv.Elem[kv.OP], error) { + key := s3keys.BucketGenerationKey(bucket) + raw, err := s.store.GetAt(ctx, key, readTS) + switch { + case errors.Is(err, store.ErrKeyNotFound): + case err != nil: + return nil, errors.WithStack(err) + default: + stored, err := decodeS3Generation(raw) + if err != nil { + return nil, err + } + generation = max(generation, stored) + } + return []*kv.Elem[kv.OP]{ + {Op: kv.Del, Key: s3keys.BucketMetaKey(bucket)}, + {Op: kv.Put, Key: key, Value: encodeS3Generation(generation)}, + }, nil +} diff --git a/adapter/s3_hlc_fence_test.go b/adapter/s3_hlc_fence_test.go index ecdeb3c94..4a10c0cba 100644 --- a/adapter/s3_hlc_fence_test.go +++ b/adapter/s3_hlc_fence_test.go @@ -10,6 +10,7 @@ import ( "github.com/bootjp/elastickv/internal/s3keys" "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/kv/kvtest" "github.com/bootjp/elastickv/store" "github.com/stretchr/testify/require" ) @@ -256,42 +257,51 @@ func TestS3NextTxnCommitTSFailsClosedOnExpiredCeiling(t *testing.T) { "s3 nextTxnCommitTS must propagate ErrCeilingExpired from NextFenced") } -func TestS3CommitUploadPartRechecksUploadAtLatestAppliedWatermark(t *testing.T) { +func TestS3CommitUploadPartProtectsUploadRemovedAfterStartTS(t *testing.T) { st := store.NewMVCCStore() const generation = uint64(1) uploadMetaKey := s3keys.UploadMetaKey("bucket", generation, "object", "upload") + bucketMeta, err := encodeS3BucketMeta(&s3BucketMeta{BucketName: "bucket", Generation: generation}) + require.NoError(t, err) + require.NoError(t, st.PutAt(context.Background(), s3keys.BucketMetaKey("bucket"), bucketMeta, 10, 0)) require.NoError(t, st.PutAt(context.Background(), uploadMetaKey, []byte("meta"), 10, 0)) require.NoError(t, st.DeleteAt(context.Background(), uploadMetaKey, 20)) - coord := &recordingS3DispatchCoordinator{} + coord := kvtest.NewLocalCoordinator(st, kvtest.LocalOptions{OCC: kvtest.OCCAllGroups, ValidateReadKeys: true}) server := NewS3Server(nil, "", st, coord, nil) - _, err := server.commitS3UploadPart(context.Background(), &s3UploadPartState{ + _, attempted, err := server.commitS3UploadPart(context.Background(), &s3UploadPartState{ partNo: 1, readTS: 10, meta: &s3BucketMeta{Generation: generation}, uploadMetaKey: uploadMetaKey, }, s3ChunkUploadResult{}, "bucket", "object", "upload", s3UploadPartVersion{startTS: 10, commitTS: 30}) - require.ErrorContains(t, err, "upload not found") - require.Nil(t, coord.request, "an upload removed after startTS must not dispatch an orphan part") + require.True(t, attempted) + require.ErrorIs(t, err, store.ErrWriteConflict, "the read set must reject a delete after the retained snapshot") + _, err = st.GetAt(context.Background(), s3keys.UploadPartKey("bucket", generation, "object", "upload", 1), ^uint64(0)) + require.ErrorIs(t, err, store.ErrKeyNotFound, "an upload removed after startTS must not commit an orphan part") } func TestS3CommitUploadPartIncludesUploadMetaInReadSet(t *testing.T) { st := store.NewMVCCStore() const generation = uint64(1) uploadMetaKey := s3keys.UploadMetaKey("bucket", generation, "object", "upload") + bucketMeta, err := encodeS3BucketMeta(&s3BucketMeta{BucketName: "bucket", Generation: generation}) + require.NoError(t, err) + require.NoError(t, st.PutAt(context.Background(), s3keys.BucketMetaKey("bucket"), bucketMeta, 10, 0)) require.NoError(t, st.PutAt(context.Background(), uploadMetaKey, []byte("meta"), 10, 0)) coord := &recordingS3DispatchCoordinator{} server := NewS3Server(nil, "", st, coord, nil) - _, err := server.commitS3UploadPart(context.Background(), &s3UploadPartState{ + _, attempted, err := server.commitS3UploadPart(context.Background(), &s3UploadPartState{ partNo: 1, readTS: 10, meta: &s3BucketMeta{Generation: generation}, uploadMetaKey: uploadMetaKey, }, s3ChunkUploadResult{}, "bucket", "object", "upload", s3UploadPartVersion{startTS: 10, commitTS: 30}) require.NoError(t, err) + require.True(t, attempted) require.NotNil(t, coord.request) - require.Equal(t, [][]byte{uploadMetaKey}, coord.request.ReadKeys) + require.Equal(t, [][]byte{uploadMetaKey, s3keys.BucketGenerationKey("bucket"), s3keys.UploadPartKey("bucket", generation, "object", "upload", 1)}, coord.request.ReadKeys) } func seedS3ObjectForReadVoucherTest(t *testing.T, ctx context.Context, st store.MVCCStore, bucket, key string) { diff --git a/adapter/s3_multipart_benchmark_test.go b/adapter/s3_multipart_benchmark_test.go new file mode 100644 index 000000000..66aaf393a --- /dev/null +++ b/adapter/s3_multipart_benchmark_test.go @@ -0,0 +1,54 @@ +package adapter + +import ( + "context" + "fmt" + "testing" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/bootjp/elastickv/store" + json "github.com/goccy/go-json" +) + +// This measures descriptor reads and manifest assembly, not payload I/O or Raft. +func BenchmarkS3LoadMultipartDescriptors(b *testing.B) { + for _, count := range []int{1, 100, s3MaxPartsPerUpload} { + b.Run(fmt.Sprint(count), func(b *testing.B) { + ctx := context.Background() + st := store.NewMVCCStore() + server := NewS3Server(nil, "", st, nil, nil) + b.Cleanup(server.Stop) + meta, err := encodeS3BucketMeta(&s3BucketMeta{BucketName: "bench", Generation: 1}) + if err != nil { + b.Fatal(err) + } + elems := []*store.KVPairMutation{ + {Op: store.OpTypePut, Key: s3keys.BucketMetaKey("bench"), Value: meta}, + {Op: store.OpTypePut, Key: s3keys.UploadMetaKey("bench", 1, "object", "upload"), Value: []byte(`{}`)}, + } + request := s3CompleteMultipartUploadRequest{} + for i := 1; i <= count; i++ { + partNo, err := uint64FromInt(i) + if err != nil { + b.Fatal(err) + } + etag := md5Hex("part") + raw, err := json.Marshal(s3PartDescriptor{PartNo: partNo, ETag: etag, SizeBytes: s3MinPartSize, PartVersion: 1}) + if err != nil { + b.Fatal(err) + } + elems = append(elems, &store.KVPairMutation{Op: store.OpTypePut, Key: s3keys.UploadPartKey("bench", 1, "object", "upload", partNo), Value: raw}) + request.Parts = append(request.Parts, s3CompleteMultipartUploadPart{PartNumber: i, ETag: etag}) + } + if err := st.ApplyMutations(ctx, elems, nil, 0, 1); err != nil { + b.Fatal(err) + } + b.ReportAllocs() + for b.Loop() { + if _, _, err := server.loadS3MultipartCompletion(ctx, 1, "bench", "object", "upload", request); err != nil { + b.Fatal(err) + } + } + }) + } +} diff --git a/adapter/s3_multipart_complete.go b/adapter/s3_multipart_complete.go index a11bc44e2..704d5d20b 100644 --- a/adapter/s3_multipart_complete.go +++ b/adapter/s3_multipart_complete.go @@ -26,6 +26,7 @@ type s3MultipartCompletion struct { uploadMetaKey []byte uploadMeta s3UploadMeta manifestParts []s3ObjectPart + partKeys [][]byte totalSize int64 compositeETag string } @@ -64,33 +65,24 @@ func validateS3MultipartCompletionParts(parts []s3CompleteMultipartUploadPart, b return nil } -func (s *S3Server) loadS3MultipartCompletion(ctx context.Context, bucket, objectKey, uploadID string, request s3CompleteMultipartUploadRequest) (s3MultipartCompletion, error) { +func (s *S3Server) loadS3MultipartCompletion(ctx context.Context, readTS uint64, bucket, objectKey, uploadID string, request s3CompleteMultipartUploadRequest) (s3MultipartCompletion, *s3BucketMeta, error) { completion := s3MultipartCompletion{bucket: bucket, objectKey: objectKey, uploadID: uploadID} - readTS := s.readTS() - readTimestamp, err := s.beginTxnReadTimestamp(ctx, readTS, "s3 complete multipart upload preparation: begin read timestamp") - if err != nil { - return completion, errors.WithStack(err) - } - readTS = readTimestamp.Timestamp() - readPin := s.pinReadTS(readTS) - defer readPin.Release() - meta, exists, err := s.loadBucketMetaAt(ctx, bucket, readTS) if err != nil { - return completion, errors.WithStack(err) + return completion, nil, errors.WithStack(err) } if !exists || meta == nil { - return completion, newS3ResponseError(http.StatusNotFound, s3ErrNoSuchBucket, s3ErrNoSuchBucketMsg, bucket, objectKey) + return completion, nil, newS3ResponseError(http.StatusNotFound, s3ErrNoSuchBucket, s3ErrNoSuchBucketMsg, bucket, objectKey) } completion.generation = meta.Generation completion.uploadMetaKey = s3keys.UploadMetaKey(bucket, meta.Generation, objectKey, uploadID) if err := s.loadS3MultipartUploadMeta(ctx, readTS, &completion); err != nil { - return completion, err + return completion, nil, err } if err := s.loadS3MultipartParts(ctx, readTS, request.Parts, &completion); err != nil { - return completion, err + return completion, nil, err } - return completion, nil + return completion, meta, nil } func (s *S3Server) loadS3MultipartUploadMeta(ctx context.Context, readTS uint64, completion *s3MultipartCompletion) error { @@ -106,9 +98,15 @@ func (s *S3Server) loadS3MultipartUploadMeta(ctx context.Context, readTS uint64, func (s *S3Server) loadS3MultipartParts(ctx context.Context, readTS uint64, requested []s3CompleteMultipartUploadPart, completion *s3MultipartCompletion) error { completion.manifestParts = make([]s3ObjectPart, 0, len(requested)) + completion.partKeys = make([][]byte, 0, len(requested)) md5Concat := md5.New() //nolint:gosec // S3 composite ETag compatibility requires MD5. for i, requestedPart := range requested { - part, partMD5, err := s.loadS3MultipartPart(ctx, readTS, requestedPart, i == len(requested)-1, completion) + partNo, err := uint64FromInt(requestedPart.PartNumber) + if err != nil { + return err + } + partKey := s3keys.UploadPartKey(completion.bucket, completion.generation, completion.objectKey, completion.uploadID, partNo) + part, partMD5, err := s.loadS3MultipartPart(ctx, readTS, partKey, requestedPart, i == len(requested)-1, completion) if err != nil { return err } @@ -116,15 +114,14 @@ func (s *S3Server) loadS3MultipartParts(ctx context.Context, readTS uint64, requ return errors.WithStack(err) } completion.manifestParts = append(completion.manifestParts, part) + completion.partKeys = append(completion.partKeys, partKey) completion.totalSize += part.SizeBytes } completion.compositeETag = multipartCompositeETag(md5Concat, len(requested)) return nil } -func (s *S3Server) loadS3MultipartPart(ctx context.Context, readTS uint64, requested s3CompleteMultipartUploadPart, last bool, completion *s3MultipartCompletion) (s3ObjectPart, []byte, error) { - partNo := uint64(requested.PartNumber) //nolint:gosec // validated by validateS3MultipartCompletionParts before loading parts. - partKey := s3keys.UploadPartKey(completion.bucket, completion.generation, completion.objectKey, completion.uploadID, partNo) +func (s *S3Server) loadS3MultipartPart(ctx context.Context, readTS uint64, partKey []byte, requested s3CompleteMultipartUploadPart, last bool, completion *s3MultipartCompletion) (s3ObjectPart, []byte, error) { raw, err := s.store.GetAt(ctx, partKey, readTS) if err != nil { if errors.Is(err, store.ErrKeyNotFound) { @@ -157,39 +154,55 @@ func multipartCompositeETag(md5Concat hash.Hash, partCount int) string { return hex.EncodeToString(md5Concat.Sum(nil)) + fmt.Sprintf("-%d", partCount) } -func (s *S3Server) commitS3MultipartCompletion(ctx context.Context, completion s3MultipartCompletion) (*s3ObjectManifest, error) { +func writeS3MultipartMutationError(w http.ResponseWriter, err error, bucket, objectKey string) { + var responseErr *s3ResponseError + if !errors.As(err, &responseErr) && isRetryableS3MutationErr(err) { + writeS3SlowDown(w, bucket, objectKey) + return + } + writeS3MutationError(w, err, bucket, objectKey) +} + +func writeS3SlowDown(w http.ResponseWriter, bucket, objectKey string) { + writeS3Error(w, http.StatusServiceUnavailable, "SlowDown", "conflicting operation in progress; retry with backoff", bucket, objectKey) +} + +func (s *S3Server) commitS3MultipartCompletion(ctx context.Context, bucket, objectKey, uploadID string, request s3CompleteMultipartUploadRequest) (s3MultipartCompletion, *s3ObjectManifest, error) { + var completion s3MultipartCompletion var previous *s3ObjectManifest err := s.retryS3Mutation(ctx, func() error { var err error - previous, err = s.commitS3MultipartCompletionAttempt(ctx, completion) + completion, previous, err = s.commitS3MultipartCompletionAttempt(ctx, bucket, objectKey, uploadID, request) return err }) - return previous, err + return completion, previous, err } -func (s *S3Server) commitS3MultipartCompletionAttempt(ctx context.Context, completion s3MultipartCompletion) (*s3ObjectManifest, error) { +// A retry rebuilds the manifest from one retained snapshot. Descriptor keys are +// read-only dependencies: a concurrent part replacement must conflict at apply. +func (s *S3Server) commitS3MultipartCompletionAttempt(ctx context.Context, bucket, objectKey, uploadID string, request s3CompleteMultipartUploadRequest) (s3MultipartCompletion, *s3ObjectManifest, error) { readTS := s.readTS() readTimestamp, err := s.beginTxnReadTimestamp(ctx, readTS, "s3 complete multipart upload: begin read timestamp") if err != nil { - return nil, errors.Wrap(err, "s3: allocate startTS for completeMultipartUpload retry") + return s3MultipartCompletion{}, nil, errors.Wrap(err, "s3: allocate startTS for completeMultipartUpload retry") } readTS = readTimestamp.Timestamp() startTS := readTS readPin := s.pinReadTS(readTS) defer readPin.Release() - meta, err := s.validateS3MultipartCommitFences(ctx, readTS, completion) + completion, meta, err := s.loadS3MultipartCompletion(ctx, readTS, bucket, objectKey, uploadID, request) if err != nil { - return nil, err + return completion, nil, err } - commitTS, err := s.nextTxnCommitTS(ctx, startTS) + headKey := s3keys.ObjectManifestKey(bucket, meta.Generation, objectKey) + previous, _, err := s.loadObjectManifestAt(ctx, headKey, readTS) if err != nil { - return nil, errors.WithStack(err) + return completion, nil, errors.WithStack(err) } - headKey := s3keys.ObjectManifestKey(completion.bucket, meta.Generation, completion.objectKey) - previous, _, err := s.loadObjectManifestAt(ctx, headKey, readTS) + commitTS, err := s.nextTxnCommitTS(ctx, startTS) if err != nil { - return nil, errors.WithStack(err) + return completion, nil, errors.WithStack(err) } manifestBody, err := encodeS3ObjectManifest(&s3ObjectManifest{ UploadID: completion.uploadID, @@ -201,11 +214,11 @@ func (s *S3Server) commitS3MultipartCompletionAttempt(ctx context.Context, compl Parts: completion.manifestParts, }) if err != nil { - return nil, errors.WithStack(err) + return completion, nil, errors.WithStack(err) } bucketFence, err := encodeS3BucketMeta(meta) if err != nil { - return nil, errors.WithStack(err) + return completion, nil, errors.WithStack(err) } dispatchCtx := readTimestamp.WithDispatchVoucher(ctx) _, err = kv.DispatchWithReadTimestamp(dispatchCtx, s.coordinator, &kv.OperationGroup[kv.OP]{ @@ -218,29 +231,10 @@ func (s *S3Server) commitS3MultipartCompletionAttempt(ctx context.Context, compl {Op: kv.Del, Key: completion.uploadMetaKey}, {Op: kv.Del, Key: s3keys.GCUploadKey(completion.bucket, completion.generation, completion.objectKey, completion.uploadID)}, }, + ReadKeys: completion.partKeys, }) if err != nil { - return nil, errors.WithStack(err) - } - return previous, nil -} - -func (s *S3Server) validateS3MultipartCommitFences(ctx context.Context, readTS uint64, completion s3MultipartCompletion) (*s3BucketMeta, error) { - meta, exists, err := s.loadBucketMetaAt(ctx, completion.bucket, readTS) - if err != nil { - return nil, errors.WithStack(err) - } - if !exists || meta == nil { - return nil, newS3ResponseError(http.StatusNotFound, s3ErrNoSuchBucket, s3ErrNoSuchBucketMsg, completion.bucket, completion.objectKey) - } - if meta.Generation != completion.generation { - return nil, newS3ResponseError(http.StatusNotFound, "NoSuchUpload", "upload not found (bucket was recreated)", completion.bucket, completion.objectKey) - } - if _, err := s.store.GetAt(ctx, completion.uploadMetaKey, readTS); err != nil { - if errors.Is(err, store.ErrKeyNotFound) { - return nil, newS3ResponseError(http.StatusNotFound, "NoSuchUpload", "upload not found", completion.bucket, completion.objectKey) - } - return nil, errors.WithStack(err) + return completion, nil, errors.WithStack(err) } - return meta, nil + return completion, previous, nil } diff --git a/adapter/s3_multipart_occ_repro_test.go b/adapter/s3_multipart_occ_repro_test.go new file mode 100644 index 000000000..eda8efd4b --- /dev/null +++ b/adapter/s3_multipart_occ_repro_test.go @@ -0,0 +1,760 @@ +package adapter + +import ( + "bytes" + "context" + "encoding/xml" + "fmt" + "net/http" + "net/http/httptest" + "slices" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" + json "github.com/goccy/go-json" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// S3 multipart read-set reproduction harness (gap G2 of the serializable +// isolation audit). +// +// One single-node Raft group (newSingleRaftForS3Test) running the production +// kvFSM, fronted by a production kv.ShardedCoordinator with one route, so +// every S3 transaction goes through the real one-phase commit path and the +// store's OCC validation (write keys and ReadKeys against StartTS) under the +// apply lock. The S3Server reads through s3OCCReproStore, which can hold a +// handler right after a chosen GetAt returned, while the test drives a +// concurrent handler to completion. The FSM applies to the unwrapped store. +// +// The hold is at a read, not at Dispatch: the S3 handlers allocate their +// commit timestamp before they dispatch, so a handler held at Dispatch would +// commit below the concurrent writer and its late apply would be the +// apply-order gap (G0), not the read-set gap this file is about. + +const s3OCCReproWaitTimeout = 10 * time.Second + +// s3OCCReproPark holds the first GetAt whose key match accepts, after the +// read returned, until release is closed. Later matches pass straight +// through. +type s3OCCReproPark struct { + name string + match func(key []byte) bool + armed bool + parked chan struct{} + release chan struct{} + once sync.Once +} + +func (p *s3OCCReproPark) releaseNow() { + p.once.Do(func() { close(p.release) }) +} + +type s3OCCReproStore struct { + store.MVCCStore + mu sync.Mutex + parks []*s3OCCReproPark + reads []s3OCCReproRead +} + +type s3OCCReproRead struct { + key []byte + ts uint64 + requestID uint64 +} + +type s3OCCReproRequestContextKey struct{} + +func s3OCCReproRequestID(ctx context.Context) uint64 { + id, _ := ctx.Value(s3OCCReproRequestContextKey{}).(uint64) + return id +} + +func (s *s3OCCReproStore) readLog() []s3OCCReproRead { + s.mu.Lock() + defer s.mu.Unlock() + return slices.Clone(s.reads) +} + +func (s *s3OCCReproStore) addPark(name string, match func(key []byte) bool) *s3OCCReproPark { + park := &s3OCCReproPark{ + name: name, + match: match, + armed: true, + parked: make(chan struct{}), + release: make(chan struct{}), + } + s.mu.Lock() + s.parks = append(s.parks, park) + s.mu.Unlock() + return park +} + +func (s *s3OCCReproStore) releaseAll() { + s.mu.Lock() + parks := append([]*s3OCCReproPark(nil), s.parks...) + s.mu.Unlock() + for _, park := range parks { + park.releaseNow() + } +} + +func (s *s3OCCReproStore) claim(key []byte) *s3OCCReproPark { + s.mu.Lock() + defer s.mu.Unlock() + for _, park := range s.parks { + if park.armed && park.match(key) { + park.armed = false + return park + } + } + return nil +} + +func (s *s3OCCReproStore) GetAt(ctx context.Context, key []byte, ts uint64) ([]byte, error) { + val, err := s.MVCCStore.GetAt(ctx, key, ts) + s.mu.Lock() + s.reads = append(s.reads, s3OCCReproRead{key: bytes.Clone(key), ts: ts, requestID: s3OCCReproRequestID(ctx)}) + s.mu.Unlock() + if park := s.claim(key); park != nil { + close(park.parked) + select { + case <-park.release: + case <-ctx.Done(): + return nil, errors.WithStack(ctx.Err()) + } + } + return val, errors.WithStack(err) +} + +// s3OCCReproCoordinator forwards to the production ShardedCoordinator. It +// counts every Del it forwards (the async cleanups are the only S3 path that +// Dels blob keys) and, when failDelPrefix is set, fails DEL_PREFIX groups the +// way a failed bucket-delete safety net would. +type s3OCCReproCoordinator struct { + *kv.ShardedCoordinator + failDelPrefix atomic.Bool + // maxReadKeys is the largest ReadKeys set dispatched so far. + maxReadKeys atomic.Int64 + mu sync.Mutex + dels map[string]int + txns []*kv.OperationGroup[kv.OP] + requestIDs map[*kv.OperationGroup[kv.OP]]uint64 +} + +func (c *s3OCCReproCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) { + if req != nil { + if req.IsTxn { + copyReq := *req + copyReq.ReadKeys = slices.Clone(req.ReadKeys) + copyReq.Elems = slices.Clone(req.Elems) + c.mu.Lock() + c.txns = append(c.txns, ©Req) + c.requestIDs[©Req] = s3OCCReproRequestID(ctx) + c.mu.Unlock() + } + if c.failDelPrefix.Load() && s3OCCReproHasOp(req, kv.DelPrefix) { + return nil, errors.New("s3 occ repro: injected DEL_PREFIX failure") + } + if !req.IsTxn { + c.recordDels(req.Elems) + } + for n := int64(len(req.ReadKeys)); ; { + seen := c.maxReadKeys.Load() + if n <= seen || c.maxReadKeys.CompareAndSwap(seen, n) { + break + } + } + } + resp, err := c.ShardedCoordinator.Dispatch(ctx, req) + return resp, errors.WithStack(err) +} + +func s3OCCReproHasOp(req *kv.OperationGroup[kv.OP], op kv.OP) bool { + for _, elem := range req.Elems { + if elem != nil && elem.Op == op { + return true + } + } + return false +} + +func (c *s3OCCReproCoordinator) recordDels(elems []*kv.Elem[kv.OP]) { + c.mu.Lock() + defer c.mu.Unlock() + for _, elem := range elems { + if elem != nil && elem.Op == kv.Del { + c.dels[string(elem.Key)]++ + } + } +} + +func (c *s3OCCReproCoordinator) delCount(key []byte) int { + c.mu.Lock() + defer c.mu.Unlock() + return c.dels[string(key)] +} + +type s3OCCReproHarness struct { + t *testing.T + st store.MVCCStore + reads *s3OCCReproStore + coord *s3OCCReproCoordinator + server *S3Server + handlers sync.WaitGroup + requestIDs atomic.Uint64 +} + +func newS3OCCReproHarness(t *testing.T) *s3OCCReproHarness { + t.Helper() + return newS3OCCReproVisibilityHarness(t, false) +} + +func newS3OCCReproVisibilityHarness(t *testing.T, staged bool) *s3OCCReproHarness { + t.Helper() + return newS3OCCReproModeHarness(t, staged, true) +} + +func newS3OCCReproModeHarness(t *testing.T, staged, offloaded bool) *s3OCCReproHarness { + t.Helper() + + hlc := kv.NewHLC() + st := store.NewMVCCStore() + routes := distribution.NewEngine() + route := distribution.RouteDescriptor{RouteID: 1, GroupID: 1, State: distribution.RouteStateActive} + if staged { + route.StagedVisibilityActive = true + route.MigrationJobID = 9 + } + require.NoError(t, routes.ApplySnapshot(distribution.CatalogSnapshot{Version: 1, Routes: []distribution.RouteDescriptor{route}})) + fsm := kv.NewKvFSMWithHLC(st, hlc, kv.WithRouteHistory(kv.WrapDistributionEngine(routes), 1)) + engine, stop := newSingleRaftForS3Test(t, "s3-occ-repro", fsm) + t.Cleanup(stop) + + groups := map[uint64]*kv.ShardGroup{ + 1: {Engine: engine, Store: st, Txn: kv.NewLeaderProxyWithEngine(engine)}, + } + shardStore := kv.NewShardStore(routes, groups) + coord := &s3OCCReproCoordinator{ + ShardedCoordinator: kv.NewShardedCoordinator(routes, groups, 1, hlc, shardStore), + dels: map[string]int{}, + requestIDs: map[*kv.OperationGroup[kv.OP]]uint64{}, + } + reads := &s3OCCReproStore{MVCCStore: shardStore} + h := &s3OCCReproHarness{ + t: t, + st: st, + reads: reads, + coord: coord, + server: NewS3Server(nil, "", reads, coord, nil, + WithS3BlobOffloadEnabled(offloaded), + WithS3BlobCluster(newFakeS3BlobCluster()), + WithS3BlobLocalStoreResolver(&mutableS3BlobLocalStore{store: st}), + withS3BlobOffloadGCReadyForTest()), + } + t.Cleanup(func() { require.NoError(t, coord.Close()) }) + t.Cleanup(h.server.Stop) + // Release every park before the Raft engine stops so no handler + // goroutine is left blocked on a failed test. + t.Cleanup(func() { + reads.releaseAll() + h.handlers.Wait() + h.waitCleanups() + }) + return h +} + +func (h *s3OCCReproHarness) do(method, target, body string) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + req := newS3TestRequest(method, target, strings.NewReader(body)) + ctx := context.WithValue(req.Context(), s3OCCReproRequestContextKey{}, h.requestIDs.Add(1)) + h.server.handle(rec, req.WithContext(ctx)) + return rec +} + +// s3OCCReproObject is the one object every reproduction uploads; each test +// uses its own bucket. +const s3OCCReproObject = "obj" + +// s3OCCReproUpload is one in-progress multipart upload of s3OCCReproObject. +// The reproductions only race part 1. +type s3OCCReproUpload struct { + bucket string + generation uint64 + uploadID string + uploadMetaKey []byte + partKey []byte +} + +func (u s3OCCReproUpload) objectPath() string { + return "/" + u.bucket + "/" + s3OCCReproObject +} + +func (u s3OCCReproUpload) chunkKey(chunkNo uint64, part s3ObjectPart) []byte { + if part.Offloaded { + return s3keys.VersionedChunkRefKey(u.bucket, u.generation, s3OCCReproObject, u.uploadID, part.PartNo, chunkNo, part.ChunkRefVersion) + } + return s3keys.VersionedBlobKey(u.bucket, u.generation, s3OCCReproObject, u.uploadID, part.PartNo, chunkNo, part.PartVersion) +} + +// startUpload creates the bucket and a multipart upload in it. +func (h *s3OCCReproHarness) startUpload(bucket string) s3OCCReproUpload { + h.t.Helper() + rec := h.do(http.MethodPut, "/"+bucket, "") + require.Equal(h.t, http.StatusOK, rec.Code, rec.Body.String()) + meta, exists, err := h.server.loadBucketMetaAt(context.Background(), bucket, ^uint64(0)) + require.NoError(h.t, err) + require.True(h.t, exists) + + u := s3OCCReproUpload{bucket: bucket, generation: meta.Generation} + rec = h.do(http.MethodPost, u.objectPath()+"?uploads=", "") + require.Equal(h.t, http.StatusOK, rec.Code, rec.Body.String()) + var result s3InitiateMultipartUploadResult + require.NoError(h.t, xml.Unmarshal(rec.Body.Bytes(), &result)) + require.NotEmpty(h.t, result.UploadId) + u.uploadID = result.UploadId + u.uploadMetaKey = s3keys.UploadMetaKey(bucket, u.generation, s3OCCReproObject, u.uploadID) + u.partKey = s3keys.UploadPartKey(bucket, u.generation, s3OCCReproObject, u.uploadID, 1) + return u +} + +func (h *s3OCCReproHarness) uploadPart1(u s3OCCReproUpload, body string) *httptest.ResponseRecorder { + return h.do(http.MethodPut, fmt.Sprintf("%s?uploadId=%s&partNumber=1", u.objectPath(), u.uploadID), body) +} + +func (h *s3OCCReproHarness) completeWithPart1(u s3OCCReproUpload, etag string) *httptest.ResponseRecorder { + body := fmt.Sprintf(`1"%s"`, etag) + return h.do(http.MethodPost, fmt.Sprintf("%s?uploadId=%s", u.objectPath(), u.uploadID), body) +} + +// part1Descriptor returns the latest committed descriptor of part 1. +func (h *s3OCCReproHarness) part1Descriptor(u s3OCCReproUpload) (*s3PartDescriptor, bool) { + h.t.Helper() + raw, err := h.st.GetAt(context.Background(), u.partKey, ^uint64(0)) + if errors.Is(err, store.ErrKeyNotFound) { + return nil, false + } + require.NoError(h.t, err) + var descriptor s3PartDescriptor + require.NoError(h.t, json.Unmarshal(raw, &descriptor)) + return &descriptor, true +} + +// blobVersions returns the live immutable blob or chunk-reference namespaces. +func (h *s3OCCReproHarness) blobVersions(u s3OCCReproUpload) map[uint64]int { + h.t.Helper() + prefix := s3keys.ChunkRefPrefixForUpload(u.bucket, u.generation, s3OCCReproObject, u.uploadID) + if !h.server.blobOffloadEnabled { + prefix = s3keys.BlobPrefixForUpload(u.bucket, u.generation, s3OCCReproObject, u.uploadID) + } + kvs, err := h.st.ScanAt(context.Background(), prefix, prefixScanEnd(prefix), s3MetaBatchOps, ^uint64(0)) + require.NoError(h.t, err) + out := map[uint64]int{} + for _, kvp := range kvs { + _, _, _, _, _, _, version, ok := s3keys.ParseVersionedChunkRefKey(kvp.Key) + if !h.server.blobOffloadEnabled { + _, _, _, _, _, _, version, ok = s3keys.ParseBlobKey(kvp.Key) + } + require.True(h.t, ok, "unparseable blob key %q", kvp.Key) + out[version]++ + } + return out +} + +// waitCleanups returns once every asynchronous cleanup launched so far has +// finished: each holds a cleanupSem slot for its whole run, so owning every +// slot proves none is still running. +func (h *s3OCCReproHarness) waitCleanups() { + h.t.Helper() + acquired := 0 + defer func() { + for range acquired { + <-h.server.cleanupSem + } + }() + timer := time.NewTimer(s3OCCReproWaitTimeout) + defer timer.Stop() + for range cap(h.server.cleanupSem) { + select { + case h.server.cleanupSem <- struct{}{}: + acquired++ + case <-timer.C: + h.t.Fatal("asynchronous S3 cleanups did not finish") + } + } +} + +func (h *s3OCCReproHarness) waitParked(park *s3OCCReproPark) { + h.t.Helper() + select { + case <-park.parked: + case <-time.After(s3OCCReproWaitTimeout): + h.t.Fatalf("handler never reached park %q", park.name) + } +} + +func (h *s3OCCReproHarness) waitDone(done <-chan *httptest.ResponseRecorder) *httptest.ResponseRecorder { + h.t.Helper() + select { + case rec := <-done: + return rec + case <-time.After(s3OCCReproWaitTimeout): + h.t.Fatal("parked handler did not finish after release") + return nil + } +} + +func (h *s3OCCReproHarness) goHandle(fn func() *httptest.ResponseRecorder) <-chan *httptest.ResponseRecorder { + done := make(chan *httptest.ResponseRecorder, 1) + h.handlers.Add(1) + go func() { + defer h.handlers.Done() + done <- fn() + }() + return done +} + +func TestS3UploadPart_InlineInitialUploadUsesFreshDescriptorCommitTimestamp(t *testing.T) { + t.Parallel() + h := newS3OCCReproModeHarness(t, false, false) + u := h.startUpload("occ-inline-initial") + const body = "inline-payload" + rec := h.uploadPart1(u, body) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + part, found := h.part1Descriptor(u) + require.True(t, found) + require.False(t, part.Offloaded) + req := h.coord.lastTxnWriting(t, u.partKey) + require.Less(t, part.PartVersion, req.CommitTS, "immutable blob namespace must precede descriptor commit") + raw, err := h.st.GetAt(context.Background(), u.chunkKey(0, s3ObjectPart(*part)), ^uint64(0)) + require.NoError(t, err) + require.Equal(t, body, string(raw)) + rec = h.completeWithPart1(u, strings.Trim(rec.Header().Get("ETag"), `"`)) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + rec = h.do(http.MethodGet, u.objectPath(), "") + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + require.Equal(t, body, rec.Body.String()) +} + +// TestS3CompleteMultipartUpload_RacingPartReuploadKeepsReferencedChunks: +// CompleteMultipartUpload has read part 1's descriptor when a re-upload of part +// 1 with the same bytes (same ETag) commits a new part version and +// asynchronously deletes the old version's blobs. The re-upload writes only +// the part key and its chunks, none of which the completion writes, so +// nothing detects it unless the part key is in the completion's validated +// read set: the manifest then points at the superseded version whose chunks +// are gone. The re-upload lands either right after the completion read the +// descriptor or after its last read, between its read phase and its commit. +func TestS3CompleteMultipartUpload_RacingPartReuploadKeepsReferencedChunks(t *testing.T) { + t.Parallel() + cases := []struct { + name string + parkKey func(u s3OCCReproUpload) []byte + }{ + { + name: "after the part descriptor read", + parkKey: func(u s3OCCReproUpload) []byte { return u.partKey }, + }, + { + name: "between the read phase and the commit", + parkKey: func(u s3OCCReproUpload) []byte { + return s3keys.ObjectManifestKey(u.bucket, u.generation, s3OCCReproObject) + }, + }, + } + for _, offloaded := range []bool{false, true} { + for _, changedETag := range []bool{false, true} { + for i, tc := range cases { + t.Run(fmt.Sprintf("offloaded=%t/changed-etag=%t/%s", offloaded, changedETag, tc.name), func(t *testing.T) { + t.Parallel() + h := newS3OCCReproModeHarness(t, false, offloaded) + u := h.startUpload(fmt.Sprintf("occ-complete-%d", i)) + parkKey := tc.parkKey(u) + h.checkCompleteRacingPartReupload(t, u, changedETag, func(key []byte) bool { return bytes.Equal(key, parkKey) }) + }) + } + } + } +} + +func (h *s3OCCReproHarness) checkCompleteRacingPartReupload(t *testing.T, u s3OCCReproUpload, changedETag bool, parkAt func(key []byte) bool) { + t.Helper() + const body = "part-one-payload" + rec := h.uploadPart1(u, body) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + etag := strings.Trim(rec.Header().Get("ETag"), `"`) + first, ok := h.part1Descriptor(u) + require.True(t, ok) + + park := h.reads.addPark("complete: "+t.Name(), parkAt) + done := h.goHandle(func() *httptest.ResponseRecorder { return h.completeWithPart1(u, etag) }) + h.waitParked(park) + + reuploadedBody := body + if changedETag { + reuploadedBody = "changed-part-payload" + } + rec = h.uploadPart1(u, reuploadedBody) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + if changedETag { + require.NotEqual(t, etag, strings.Trim(rec.Header().Get("ETag"), `"`)) + } else { + require.Equal(t, etag, strings.Trim(rec.Header().Get("ETag"), `"`)) + } + second, ok := h.part1Descriptor(u) + require.True(t, ok) + require.NotEqual(t, first.PartVersion, second.PartVersion) + + park.releaseNow() + completeRec := h.waitDone(done) + if changedETag { + require.Equal(t, http.StatusBadRequest, completeRec.Code, completeRec.Body.String()) + require.Contains(t, completeRec.Body.String(), "InvalidPart") + _, found, err := h.server.loadObjectManifestAt(context.Background(), s3keys.ObjectManifestKey(u.bucket, u.generation, s3OCCReproObject), ^uint64(0)) + require.NoError(t, err) + require.False(t, found, "failed completion must not publish a stale manifest") + return + } + require.Equal(t, http.StatusOK, completeRec.Code, completeRec.Body.String()) + h.waitCleanups() + + manifest, found, err := h.server.loadObjectManifestAt(context.Background(), s3keys.ObjectManifestKey(u.bucket, u.generation, s3OCCReproObject), ^uint64(0)) + require.NoError(t, err) + require.True(t, found) + require.Len(t, manifest.Parts, 1) + part := manifest.Parts[0] + assert.Equalf(t, second.PartVersion, part.PartVersion, + "manifest references part version %d; the re-upload superseded it with %d (first upload was %d)", + part.PartVersion, second.PartVersion, first.PartVersion) + for chunk := range part.ChunkCount { + _, err := h.st.GetAt(context.Background(), u.chunkKey(chunk, part), ^uint64(0)) + missing := errors.Is(err, store.ErrKeyNotFound) + if !missing { + require.NoError(t, err) + } + assert.Falsef(t, missing, "manifest chunk %d of part version %d is missing", chunk, part.PartVersion) + } + + rec = h.do(http.MethodGet, u.objectPath(), "") + assert.Equal(t, http.StatusOK, rec.Code, "GET of the completed object") + assert.Equal(t, body, rec.Body.String(), "GET of the completed object") +} + +// TestS3UploadPart_RacingDeleteBucketDoesNotLandUnderDeadGeneration: +// UploadPart has read the bucket meta when DeleteBucket commits. When the +// bucket-delete safety net sweeps the upload meta, the upload-meta read key +// already stops the part; when that best-effort sweep fails (the path +// runBucketDeleteSafetyNet logs and tolerates), only the bucket meta and the +// bucket generation key record the delete, so the part and its chunks must +// not commit under the deleted generation. The delete lands either before +// the part's StartTS (right after its prepare reads) or after its last +// commit-phase read. +func TestS3UploadPart_RacingDeleteBucketDoesNotLandUnderDeadGeneration(t *testing.T) { + t.Parallel() + cases := []struct { + name string + parkKey func(u s3OCCReproUpload) []byte + // The re-read at the commit watermark sees a delete committed up to + // it; a later one is a write conflict on the generation key. + wantStatus int + wantCode string + recreate bool + }{ + { + name: "delete between the prepare reads and StartTS", + parkKey: func(u s3OCCReproUpload) []byte { return u.uploadMetaKey }, + wantStatus: http.StatusNotFound, + wantCode: "NoSuchBucket", + }, + { + name: "delete after the commit-phase reads", + parkKey: func(u s3OCCReproUpload) []byte { return u.partKey }, + wantStatus: http.StatusServiceUnavailable, + wantCode: "SlowDown", + }, + { + name: "recreate between the prepare reads and StartTS", + parkKey: func(u s3OCCReproUpload) []byte { return u.uploadMetaKey }, + wantStatus: http.StatusNotFound, + wantCode: "NoSuchUpload", + recreate: true, + }, + { + name: "recreate after the commit-phase reads", + parkKey: func(u s3OCCReproUpload) []byte { return u.partKey }, + wantStatus: http.StatusServiceUnavailable, + wantCode: "SlowDown", + recreate: true, + }, + } + for _, offloaded := range []bool{false, true} { + for i, tc := range cases { + t.Run(fmt.Sprintf("offloaded=%t/%s", offloaded, tc.name), func(t *testing.T) { + t.Parallel() + h := newS3OCCReproModeHarness(t, false, offloaded) + u := h.startUpload(fmt.Sprintf("occ-delete-bucket-%d", i)) + parkKey := tc.parkKey(u) + partRec := h.uploadPart1Racing(u, "orphan-candidate", parkKey, func() { + h.coord.failDelPrefix.Store(true) + rec := h.do(http.MethodDelete, "/"+u.bucket, "") + require.Equal(t, http.StatusNoContent, rec.Code, rec.Body.String()) + _, exists, err := h.server.loadBucketMetaAt(context.Background(), u.bucket, ^uint64(0)) + require.NoError(t, err) + require.False(t, exists) + if tc.recreate { + rec = h.do(http.MethodPut, "/"+u.bucket, "") + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + meta, exists, err := h.server.loadBucketMetaAt(context.Background(), u.bucket, ^uint64(0)) + require.NoError(t, err) + require.True(t, exists) + require.Greater(t, meta.Generation, u.generation) + } + }) + + descriptor, landed := h.part1Descriptor(u) + assert.Falsef(t, landed, + "part descriptor (version %v) committed under deleted generation %d", + descriptorVersion(descriptor), u.generation) + assert.Empty(t, h.blobVersions(u), "part chunks leaked under deleted generation") + assert.NotEqual(t, http.StatusOK, partRec.Code, "upload part must not succeed after the bucket was deleted") + assert.Equal(t, tc.wantStatus, partRec.Code, partRec.Body.String()) + assert.Contains(t, partRec.Body.String(), ""+tc.wantCode+"") + }) + } + } +} + +// TestS3UploadPart_ConcurrentObjectWriteInBucketDoesNotAbortPart: every +// object writer in a bucket re-puts the bucket meta as deleteBucket's +// emptiness fence. A part upload holds its StartTS across the whole body +// stream, so it must not fence on the bucket meta: an unrelated PutObject or +// CreateMultipartUpload in the same bucket must not abort it. +func TestS3UploadPart_ConcurrentObjectWriteInBucketDoesNotAbortPart(t *testing.T) { + t.Parallel() + cases := []struct { + name string + method string + target func(u s3OCCReproUpload) string + }{ + { + name: "PutObject of another key", + method: http.MethodPut, + target: func(u s3OCCReproUpload) string { return "/" + u.bucket + "/other" }, + }, + { + name: "CreateMultipartUpload of another key", + method: http.MethodPost, + target: func(u s3OCCReproUpload) string { return "/" + u.bucket + "/other?uploads=" }, + }, + } + for _, offloaded := range []bool{false, true} { + for i, tc := range cases { + t.Run(fmt.Sprintf("offloaded=%t/%s", offloaded, tc.name), func(t *testing.T) { + t.Parallel() + h := newS3OCCReproModeHarness(t, false, offloaded) + u := h.startUpload(fmt.Sprintf("occ-bucket-writer-%d", i)) + partRec := h.uploadPart1Racing(u, "part-body", u.partKey, func() { + rec := h.do(tc.method, tc.target(u), "other-body") + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + }) + require.Equal(t, http.StatusOK, partRec.Code, partRec.Body.String()) + _, landed := h.part1Descriptor(u) + require.True(t, landed) + }) + } + } +} + +// uploadPart1Racing uploads part 1, holds it right after its GetAt of parkKey +// returned, runs concurrent, then releases it and waits for its response and +// for every cleanup it launched. +func (h *s3OCCReproHarness) uploadPart1Racing(u s3OCCReproUpload, body string, parkKey []byte, concurrent func()) *httptest.ResponseRecorder { + h.t.Helper() + park := h.reads.addPark("upload part: "+h.t.Name(), func(key []byte) bool { + return bytes.Equal(key, parkKey) + }) + done := h.goHandle(func() *httptest.ResponseRecorder { return h.uploadPart1(u, body) }) + h.waitParked(park) + concurrent() + park.releaseNow() + rec := h.waitDone(done) + h.waitCleanups() + return rec +} + +// TestS3UploadPart_SamePartUploadsInWindowCleanExactlySupersededBlobs: upload A +// of part 1 reads the previous descriptor (v0) at its prepare snapshot, then +// upload B of part 1 commits before A's StartTS. B supersedes v0 and cleans +// it; A supersedes B, so A must clean B's blobs. Reading the previous part at +// the prepare snapshot makes A clean v0 a second time and leak B. +func TestS3UploadPart_SamePartUploadsInWindowCleanExactlySupersededBlobs(t *testing.T) { + t.Parallel() + for _, offloaded := range []bool{false, true} { + t.Run(fmt.Sprintf("offloaded=%t", offloaded), func(t *testing.T) { + t.Parallel() + h := newS3OCCReproModeHarness(t, false, offloaded) + h.checkSamePartUploadsInWindow(t) + }) + } +} + +func (h *s3OCCReproHarness) checkSamePartUploadsInWindow(t *testing.T) { + t.Helper() + u := h.startUpload("occ-same-part") + + rec := h.uploadPart1(u, "version-zero") + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + v0, ok := h.part1Descriptor(u) + require.True(t, ok) + h.waitCleanups() + + park := h.reads.addPark("upload A: prepare upload-meta read", func(key []byte) bool { + return bytes.Equal(key, u.uploadMetaKey) + }) + doneA := h.goHandle(func() *httptest.ResponseRecorder { return h.uploadPart1(u, "version-a") }) + h.waitParked(park) + + rec = h.uploadPart1(u, "version-b") + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + vB, ok := h.part1Descriptor(u) + require.True(t, ok) + + park.releaseNow() + recA := h.waitDone(doneA) + require.Equal(t, http.StatusOK, recA.Code, recA.Body.String()) + vA, ok := h.part1Descriptor(u) + require.True(t, ok) + h.waitCleanups() + + namespace := vA.PartVersion + if vA.Offloaded { + namespace = vA.ChunkRefVersion + } + assert.Equalf(t, map[uint64]int{namespace: 1}, h.blobVersions(u), + "live descriptor is version %d (v0=%d, B=%d); every other version's blobs must be cleaned", + vA.PartVersion, v0.PartVersion, vB.PartVersion) + assert.Equal(t, 1, h.coord.delCount(u.chunkKey(0, s3ObjectPart(*v0))), "v0 blobs must be cleaned exactly once") + assert.Equal(t, 1, h.coord.delCount(u.chunkKey(0, s3ObjectPart(*vB))), "B's blobs must be cleaned exactly once") + assert.Zero(t, h.coord.delCount(u.chunkKey(0, s3ObjectPart(*vA))), "the live version's blobs must not be cleaned") + + rec = h.completeWithPart1(u, strings.Trim(recA.Header().Get("ETag"), `"`)) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + rec = h.do(http.MethodGet, u.objectPath(), "") + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + require.Equal(t, "version-a", rec.Body.String()) +} + +func descriptorVersion(descriptor *s3PartDescriptor) any { + if descriptor == nil { + return nil + } + return descriptor.PartVersion +} diff --git a/adapter/s3_multipart_read_set_test.go b/adapter/s3_multipart_read_set_test.go new file mode 100644 index 000000000..3398bc74c --- /dev/null +++ b/adapter/s3_multipart_read_set_test.go @@ -0,0 +1,307 @@ +package adapter + +import ( + "bytes" + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" + json "github.com/goccy/go-json" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These contracts inspect actual HTTP transactions forwarded to the production +// coordinator. Recording does not replace dispatch, timestamp admission or OCC. +func (c *s3OCCReproCoordinator) lastTxnWriting(t *testing.T, key []byte) *kv.OperationGroup[kv.OP] { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + for i := len(c.txns) - 1; i >= 0; i-- { + for _, elem := range c.txns[i].Elems { + if bytes.Equal(elem.Key, key) { + return c.txns[i] + } + } + } + t.Fatalf("no transaction wrote %q", key) + return nil +} + +func requireS3ReadAtStartTS(t *testing.T, reads []s3OCCReproRead, key []byte, startTS uint64) { + t.Helper() + for i := len(reads) - 1; i >= 0; i-- { + if bytes.Equal(reads[i].key, key) { + assert.Equal(t, startTS, reads[i].ts, "commit-state read of %q must use exactly StartTS", key) + return + } + } + t.Errorf("no commit-state GetAt for %q at StartTS %d", key, startTS) +} + +func (h *s3OCCReproHarness) readsForTxn(req *kv.OperationGroup[kv.OP]) []s3OCCReproRead { + h.t.Helper() + h.coord.mu.Lock() + id := h.coord.requestIDs[req] + h.coord.mu.Unlock() + require.NotZero(h.t, id, "transaction must belong to a traced HTTP request") + var reads []s3OCCReproRead + for _, read := range h.reads.readLog() { + if read.requestID == id { + reads = append(reads, read) + } + } + return reads +} + +func TestS3CommitUploadPartReadSet(t *testing.T) { + t.Parallel() + for _, scenario := range []string{"first part", "previous part", "previous part between preparation and StartTS"} { + t.Run(scenario, func(t *testing.T) { + t.Parallel() + h := newS3OCCReproHarness(t) + u := h.startUpload("readset-upload") + if scenario != "first part" { + rec := h.uploadPart1(u, "previous") + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + h.waitCleanups() + } + var rec *httptest.ResponseRecorder + if scenario == "previous part between preparation and StartTS" { + rec = h.uploadPart1Racing(u, "latest", u.uploadMetaKey, func() { + concurrent := h.uploadPart1(u, "window") + require.Equal(t, http.StatusOK, concurrent.Code, concurrent.Body.String()) + }) + } else { + rec = h.uploadPart1(u, "latest") + } + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + req := h.coord.lastTxnWriting(t, u.partKey) + assert.ElementsMatch(t, [][]byte{u.uploadMetaKey, s3keys.BucketGenerationKey(u.bucket), u.partKey}, req.ReadKeys) + assert.NotContains(t, req.ReadKeys, s3keys.BucketMetaKey(u.bucket), "unrelated bucket writers must not serialize part uploads") + reads := h.readsForTxn(req) + for _, key := range [][]byte{s3keys.BucketMetaKey(u.bucket), u.uploadMetaKey, u.partKey} { + requireS3ReadAtStartTS(t, reads, key, req.StartTS) + } + }) + } +} + +func TestS3UploadPart_ReadSetUsesStartTSAfterConcurrentSamePartWrite(t *testing.T) { + t.Parallel() + for _, offloaded := range []bool{false, true} { + t.Run(fmt.Sprintf("offloaded=%t", offloaded), func(t *testing.T) { + t.Parallel() + h := newS3OCCReproModeHarness(t, false, offloaded) + u := h.startUpload("readset-after-start") + rec := h.uploadPart1(u, "original") + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + var concurrent *s3PartDescriptor + loser := h.uploadPart1Racing(u, "loser", u.partKey, func() { + rec = h.uploadPart1(u, "winner") + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + var found bool + concurrent, found = h.part1Descriptor(u) + require.True(t, found) + }) + assert.Equal(t, http.StatusServiceUnavailable, loser.Code, loser.Body.String()) + assert.Contains(t, loser.Body.String(), "SlowDown") + req := h.coord.lastTxnWriting(t, u.partKey) + assert.Less(t, req.StartTS, concurrent.PartVersion, "the winner must be newer than the loser's snapshot") + for _, key := range [][]byte{s3keys.BucketMetaKey(u.bucket), u.uploadMetaKey, u.partKey} { + requireS3ReadAtStartTS(t, h.readsForTxn(req), key, req.StartTS) + } + current, found := h.part1Descriptor(u) + require.True(t, found) + assert.Equal(t, concurrent, current, "the real FSM must reject the stale descriptor") + _, err := h.st.GetAt(context.Background(), u.chunkKey(0, s3ObjectPart(*concurrent)), ^uint64(0)) + require.NoError(t, err, "conflict cleanup must not delete the winning version") + }) + } +} + +// Descriptor-only fixtures avoid allocating 50 GiB for the 10,000-part case. +// Fixture writes and the completion both use real Raft; payload integrity is +// checked separately by the re-upload reproductions in the companion file. +func seedS3ReadSetParts(t *testing.T, h *s3OCCReproHarness, u s3OCCReproUpload, count int) uint64 { + t.Helper() + ts := h.coord.Clock().Next() + elems := make([]*kv.Elem[kv.OP], 0, count) + for i := 1; i <= count; i++ { + partNo, err := uint64FromInt(i) + require.NoError(t, err) + raw, err := json.Marshal(s3PartDescriptor{ + PartNo: partNo, ETag: md5Hex(fmt.Sprint(i)), SizeBytes: int64(s3MinPartSize), PartVersion: ts, + }) + require.NoError(t, err) + elems = append(elems, &kv.Elem[kv.OP]{ + Op: kv.Put, Key: s3keys.UploadPartKey(u.bucket, u.generation, s3OCCReproObject, u.uploadID, partNo), Value: raw, + }) + } + _, err := h.coord.Dispatch(context.Background(), &kv.OperationGroup[kv.OP]{CommitTS: ts, Elems: elems}) + require.NoError(t, err) + return ts +} + +func s3ReadSetCompletionBody(parts []int, etag func(int) string) string { + var body strings.Builder + body.WriteString("") + for _, part := range parts { + fmt.Fprintf(&body, `%d"%s"`, part, etag(part)) + } + body.WriteString("") + return body.String() +} + +func TestS3CommitMultipartCompletionReadSet(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + parts []int + etag func(int) string + wantCode string + }{ + {name: "single part", parts: []int{1}}, + {name: "every consumed part", parts: []int{1, 2, 3}}, + {name: "unconsumed uploaded part excluded", parts: []int{1, 3}}, + {name: "missing part", parts: []int{1, 4}, wantCode: "InvalidPart"}, + {name: "ETag mismatch", parts: []int{1}, etag: func(int) string { return md5Hex("changed") }, wantCode: "InvalidPart"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + h := newS3OCCReproHarness(t) + u := h.startUpload("readset-completion") + version := seedS3ReadSetParts(t, h, u, 3) + etag := tc.etag + if etag == nil { + etag = func(part int) string { return md5Hex(fmt.Sprint(part)) } + } + rec := h.do(http.MethodPost, u.objectPath()+"?uploadId="+u.uploadID, s3ReadSetCompletionBody(tc.parts, etag)) + headKey := s3keys.ObjectManifestKey(u.bucket, u.generation, s3OCCReproObject) + if tc.wantCode != "" { + require.Equal(t, http.StatusBadRequest, rec.Code, rec.Body.String()) + require.Contains(t, rec.Body.String(), ""+tc.wantCode+"") + _, err := h.st.GetAt(context.Background(), headKey, ^uint64(0)) + require.ErrorIs(t, err, store.ErrKeyNotFound) + return + } + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + req := h.coord.lastTxnWriting(t, headKey) + want := make([][]byte, 0, len(tc.parts)) + for _, part := range tc.parts { + partNo, err := uint64FromInt(part) + require.NoError(t, err) + key := s3keys.UploadPartKey(u.bucket, u.generation, s3OCCReproObject, u.uploadID, partNo) + want = append(want, key) + requireS3ReadAtStartTS(t, h.readsForTxn(req), key, req.StartTS) + for _, elem := range req.Elems { + assert.NotEqual(t, key, elem.Key, "a part descriptor is a read-only dependency") + } + } + assert.ElementsMatch(t, want, req.ReadKeys) + manifest, found, err := h.server.loadObjectManifestAt(context.Background(), headKey, ^uint64(0)) + require.NoError(t, err) + require.True(t, found) + require.Len(t, manifest.Parts, len(tc.parts)) + for i, part := range manifest.Parts { + assert.EqualValues(t, tc.parts[i], part.PartNo) + assert.Equal(t, version, part.PartVersion) + } + }) + } +} + +func TestS3DeleteBucketWritesUploadPartGenerationFence(t *testing.T) { + t.Parallel() + for _, admin := range []bool{false, true} { + for _, state := range []struct { + name string + stored uint64 + want uint64 + }{ + {name: "current", stored: 4, want: 4}, + {name: "legacy missing generation key", want: 4}, + {name: "ahead of bucket generation", stored: 7, want: 7}, + } { + t.Run(fmt.Sprintf("admin=%t/%s", admin, state.name), func(t *testing.T) { + t.Parallel() + h := newS3OCCReproHarness(t) + const bucket = "generation-fence" + metaKey := s3keys.BucketMetaKey(bucket) + generationKey := s3keys.BucketGenerationKey(bucket) + raw, err := encodeS3BucketMeta(&s3BucketMeta{BucketName: bucket, Generation: 4, CreatedAtHLC: 1, Owner: "AKIA_FULL", Acl: s3AclPrivate}) + require.NoError(t, err) + elems := []*kv.Elem[kv.OP]{{Op: kv.Put, Key: metaKey, Value: raw}} + if state.stored != 0 { + elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Put, Key: generationKey, Value: encodeS3Generation(state.stored)}) + } + _, err = h.coord.Dispatch(context.Background(), &kv.OperationGroup[kv.OP]{Elems: elems}) + require.NoError(t, err) + if admin { + require.NoError(t, h.server.AdminDeleteBucket(context.Background(), fullAdminBucketsPrincipal(), bucket)) + } else { + rec := h.do(http.MethodDelete, "/"+bucket, "") + require.Equal(t, http.StatusNoContent, rec.Code, rec.Body.String()) + } + h.waitCleanups() + req := h.coord.lastTxnWriting(t, metaKey) + assert.Len(t, req.Elems, 2) + foundGenerationFence := false + for _, elem := range req.Elems { + if bytes.Equal(elem.Key, metaKey) { + assert.Equal(t, kv.Del, elem.Op) + } + if bytes.Equal(elem.Key, generationKey) { + foundGenerationFence = true + assert.Equal(t, kv.Put, elem.Op) + assert.Equal(t, encodeS3Generation(state.want), elem.Value) + } + } + assert.True(t, foundGenerationFence, "Phase 1 must atomically invalidate in-flight UploadPart transactions") + _, err = h.st.GetAt(context.Background(), metaKey, ^uint64(0)) + require.ErrorIs(t, err, store.ErrKeyNotFound) + next, err := h.server.nextBucketGenerationAt(context.Background(), bucket, h.st.LastCommitTS()) + require.NoError(t, err) + assert.Equal(t, state.want+1, next, "recreation must never reuse the deleted generation") + }) + } + } +} + +func TestS3CompleteMultipartUploadMaxPartsFitsReadKeyCap(t *testing.T) { + t.Parallel() + for _, staged := range []bool{false, true} { + t.Run(fmt.Sprintf("staged=%t", staged), func(t *testing.T) { + t.Parallel() + h := newS3OCCReproVisibilityHarness(t, staged) + u := h.startUpload("readset-max-parts") + seedS3ReadSetParts(t, h, u, s3MaxPartsPerUpload) + parts := make([]int, s3MaxPartsPerUpload) + for i := range parts { + parts[i] = i + 1 + } + body := s3ReadSetCompletionBody(parts, func(part int) string { return md5Hex(fmt.Sprint(part)) }) + rec := h.do(http.MethodPost, u.objectPath()+"?uploadId="+u.uploadID, body) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + headKey := s3keys.ObjectManifestKey(u.bucket, u.generation, s3OCCReproObject) + req := h.coord.lastTxnWriting(t, headKey) + require.Len(t, req.ReadKeys, s3MaxPartsPerUpload, "one raw read dependency per consumed part") + for i, part := range parts { + partNo, err := uint64FromInt(part) + require.NoError(t, err) + want := s3keys.UploadPartKey(u.bucket, u.generation, s3OCCReproObject, u.uploadID, partNo) + require.Equal(t, want, req.ReadKeys[i]) + } + manifest, found, err := h.server.loadObjectManifestAt(context.Background(), headKey, ^uint64(0)) + require.NoError(t, err) + require.True(t, found) + require.Len(t, manifest.Parts, s3MaxPartsPerUpload) + }) + } +} diff --git a/adapter/s3_mutation_outcome_test.go b/adapter/s3_mutation_outcome_test.go new file mode 100644 index 000000000..a212df088 --- /dev/null +++ b/adapter/s3_mutation_outcome_test.go @@ -0,0 +1,120 @@ +package adapter + +import ( + "context" + "net/http" + "sync/atomic" + "testing" + + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" + "github.com/stretchr/testify/require" +) + +func TestS3MutationRetryPreservesUnknownOutcome(t *testing.T) { + t.Parallel() + for _, cause := range []error{store.ErrWriteConflict, kv.ErrTxnLocked, kv.ErrRouteWriteFenced} { + t.Run(cause.Error(), func(t *testing.T) { + t.Parallel() + unknown := errors.Join(kv.ErrTransactionOutcomeUnknown, cause) + attempts := 0 + err := (&S3Server{}).retryS3Mutation(context.Background(), func() error { + attempts++ + if attempts == 1 { + return unknown + } + return nil + }) + require.ErrorIs(t, err, kv.ErrTransactionOutcomeUnknown) + require.Equal(t, 1, attempts, "an uncertain write must not be replayed as a fresh mutation") + require.False(t, isRetryableS3MutationErr(unknown)) + }) + } +} + +func TestS3MutationRetryRebuildsDefinitiveConflict(t *testing.T) { + t.Parallel() + attempts := 0 + err := (&S3Server{}).retryS3Mutation(context.Background(), func() error { + attempts++ + if attempts == 1 { + return store.ErrWriteConflict + } + return nil + }) + require.NoError(t, err) + require.Equal(t, 2, attempts) +} + +type s3LostPartReplyCoordinator struct { + *s3OCCReproCoordinator + partKey []byte + err error + lost atomic.Bool +} + +func (c *s3LostPartReplyCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) { + response, err := c.s3OCCReproCoordinator.Dispatch(ctx, req) + if err != nil || req == nil || !req.IsTxn { + return response, errors.WithStack(err) + } + for _, elem := range req.Elems { + if elem.Op == kv.Put && string(elem.Key) == string(c.partKey) && c.lost.CompareAndSwap(false, true) { + return nil, errors.WithStack(c.err) + } + } + return response, nil +} + +func TestS3UploadPartLostCommitReplyPreservesReferencedChunks(t *testing.T) { + t.Parallel() + for _, offloaded := range []bool{false, true} { + name := "inline" + if offloaded { + name = "offloaded" + } + t.Run(name, func(t *testing.T) { + t.Parallel() + testS3LostPartReplyOutcomes(t, offloaded) + }) + } +} + +func testS3LostPartReplyOutcomes(t *testing.T, offloaded bool) { + t.Helper() + for _, tc := range []struct { + name string + err error + }{ + {"unknown", kv.ErrTransactionOutcomeUnknown}, + {"unknown joined conflict", errors.Join(kv.ErrTransactionOutcomeUnknown, store.ErrWriteConflict)}, + {"unknown joined lock", errors.Join(kv.ErrTransactionOutcomeUnknown, kv.ErrTxnLocked)}, + {"unknown joined fence", errors.Join(kv.ErrTransactionOutcomeUnknown, kv.ErrRouteWriteFenced)}, + {"deadline after commit", context.DeadlineExceeded}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + h := newS3OCCReproModeHarness(t, false, offloaded) + u := h.startUpload("unknown-part") + require.Equal(t, http.StatusOK, h.uploadPart1(u, "original").Code) + h.waitCleanups() + previous, exists := h.part1Descriptor(u) + require.True(t, exists) + coord := &s3LostPartReplyCoordinator{s3OCCReproCoordinator: h.coord, partKey: u.partKey, err: tc.err} + h.server.coordinator = coord + + response := h.uploadPart1(u, "replacement") + require.NotEqual(t, http.StatusOK, response.Code, "the success reply was lost") + require.True(t, coord.lost.Load(), "the failure must follow a real descriptor commit") + h.waitCleanups() + committed, exists := h.part1Descriptor(u) + require.True(t, exists) + require.NotEqual(t, previous.PartVersion, committed.PartVersion) + _, err := h.st.GetAt(t.Context(), u.chunkKey(0, s3ObjectPart(*committed)), ^uint64(0)) + require.NoError(t, err, "an uncertain response cannot authorize deleting committed chunks") + _, err = h.st.GetAt(t.Context(), u.chunkKey(0, s3ObjectPart(*previous)), ^uint64(0)) + require.NoError(t, err, "superseded chunks are retired only after confirmed success") + }) + } +} diff --git a/adapter/s3_upload_part.go b/adapter/s3_upload_part.go index 81a005a2d..26667b636 100644 --- a/adapter/s3_upload_part.go +++ b/adapter/s3_upload_part.go @@ -24,9 +24,17 @@ type s3UploadPartVersion struct { startTS uint64 readTimestamp kv.ReadTimestamp commitTS uint64 + blobVersion uint64 chunkRefVersion uint64 } +func (v s3UploadPartVersion) partVersion(offloaded bool) uint64 { + if !offloaded && v.blobVersion != 0 { + return v.blobVersion + } + return v.commitTS +} + func (s *S3Server) prepareS3UploadPart(ctx context.Context, bucket, objectKey, uploadID, partNumberRaw string) (*s3UploadPartState, error) { partNo, err := parseS3UploadPartNumber(partNumberRaw, bucket, objectKey) if err != nil { @@ -84,7 +92,7 @@ func (s *S3Server) storeS3UploadPart(ctx context.Context, request *http.Request, admissionProtocol: admissionProtocol, tooLargeMessage: "part exceeds maximum allowed size", chunkKey: func(chunkNo uint64) []byte { - return s3keys.VersionedBlobKey(bucket, state.meta.Generation, objectKey, uploadID, state.partNo, chunkNo, version.commitTS) + return s3keys.VersionedBlobKey(bucket, state.meta.Generation, objectKey, uploadID, state.partNo, chunkNo, version.blobVersion) }, chunkRefKey: func(chunkNo uint64) []byte { return s3keys.VersionedChunkRefKey(bucket, state.meta.Generation, objectKey, uploadID, state.partNo, chunkNo, version.chunkRefVersion) @@ -92,23 +100,21 @@ func (s *S3Server) storeS3UploadPart(ctx context.Context, request *http.Request, offloaded: offloaded, }) committed := false + cleanupAllowed := true defer func() { - if !committed && upload.ChunkCount > 0 { + if !committed && cleanupAllowed && upload.ChunkCount > 0 { s.cleanupPartBlobsAsync( bucket, state.meta.Generation, objectKey, uploadID, - state.partNo, upload.ChunkCount, version.commitTS, version.chunkRefVersion, upload.Offloaded, + state.partNo, upload.ChunkCount, version.partVersion(upload.Offloaded), version.chunkRefVersion, upload.Offloaded, ) } }() if err != nil || bodyErr != nil { return upload, nil, bodyErr, err } - version.commitTS, err = s.finalizeS3UploadPartCommitTS(ctx, version.startTS, version.commitTS, offloaded) - if err != nil { - return upload, nil, nil, err - } - previous, err := s.commitS3UploadPart(ctx, state, upload, bucket, objectKey, uploadID, version) + previous, attempted, err := s.commitS3UploadPart(ctx, state, upload, bucket, objectKey, uploadID, version) if err != nil { + cleanupAllowed = !attempted || isRetryableS3MutationErr(err) return upload, nil, nil, err } committed = true @@ -120,22 +126,16 @@ func (s *S3Server) allocateS3UploadPartVersionForMode(ctx context.Context, offlo if err != nil { return s3UploadPartVersion{}, err } - version := s3UploadPartVersion{startTS: startTS, readTimestamp: readTimestamp, commitTS: commitTS, chunkRefVersion: startTS} + version := s3UploadPartVersion{startTS: startTS, readTimestamp: readTimestamp, blobVersion: commitTS, chunkRefVersion: startTS} if offloaded { - // Reserve the initial commit timestamp as this attempt's immutable - // chunkref namespace before delaying the descriptor commit timestamp. version.chunkRefVersion = commitTS - // The descriptor timestamp is allocated after blob durability so the - // chunkrefs cannot commit behind a side-channel write they make reachable. - version.commitTS = 0 } + // Blob namespaces are immutable; the descriptor's commit timestamp is + // allocated only after those chunks are durable in either storage mode. return version, nil } -func (s *S3Server) finalizeS3UploadPartCommitTS(ctx context.Context, startTS, commitTS uint64, offloaded bool) (uint64, error) { - if !offloaded { - return commitTS, nil - } +func (s *S3Server) finalizeS3UploadPartCommitTS(ctx context.Context, startTS uint64) (uint64, error) { ts, err := s.nextTxnCommitTS(ctx, startTS) return ts, errors.WithStack(err) } @@ -155,20 +155,26 @@ func (s *S3Server) allocateS3UploadPartVersion(ctx context.Context) (kv.ReadTime return readTimestamp, startTS, commitTS, nil } -func (s *S3Server) commitS3UploadPart(ctx context.Context, state *s3UploadPartState, upload s3ChunkUploadResult, bucket, objectKey, uploadID string, version s3UploadPartVersion) (*s3PartDescriptor, error) { +func (s *S3Server) commitS3UploadPart(ctx context.Context, state *s3UploadPartState, upload s3ChunkUploadResult, bucket, objectKey, uploadID string, version s3UploadPartVersion) (*s3PartDescriptor, bool, error) { + partKey := s3keys.UploadPartKey(bucket, state.meta.Generation, objectKey, uploadID, state.partNo) + previous, err := s.readS3UploadPartCommitState(ctx, state, partKey, bucket, objectKey, version.startTS) + if err != nil { + return nil, false, err + } + if version.commitTS == 0 { + version.commitTS, err = s.finalizeS3UploadPartCommitTS(ctx, version.startTS) + if err != nil { + return nil, false, err + } + } descriptor := &s3PartDescriptor{ PartNo: state.partNo, ETag: upload.ETag, SizeBytes: upload.SizeBytes, - ChunkCount: upload.ChunkCount, ChunkSizes: upload.ChunkSizes, PartVersion: version.commitTS, + ChunkCount: upload.ChunkCount, ChunkSizes: upload.ChunkSizes, PartVersion: version.partVersion(upload.Offloaded), ChunkRefVersion: version.chunkRefVersion, Offloaded: upload.Offloaded, } body, err := json.Marshal(descriptor) if err != nil { - return nil, errors.WithStack(err) - } - partKey := s3keys.UploadPartKey(bucket, state.meta.Generation, objectKey, uploadID, state.partNo) - previous := s.loadPreviousS3PartDescriptor(ctx, partKey, state.readTS) - if err := s.verifyS3UploadStillExists(ctx, state.uploadMetaKey, bucket, objectKey, s.readTS()); err != nil { - return nil, err + return nil, false, errors.WithStack(err) } elems := make([]*kv.Elem[kv.OP], 0, len(upload.ChunkRefElems)+1) elems = append(elems, upload.ChunkRefElems...) @@ -179,19 +185,53 @@ func (s *S3Server) commitS3UploadPart(ctx context.Context, state *s3UploadPartSt StartTS: version.startTS, CommitTS: version.commitTS, Elems: elems, - ReadKeys: [][]byte{state.uploadMetaKey}, + ReadKeys: [][]byte{state.uploadMetaKey, s3keys.BucketGenerationKey(bucket), partKey}, }) + if err != nil { + return nil, true, errors.WithStack(err) + } + return previous, true, nil +} + +func (s *S3Server) writeS3UploadPartError(w http.ResponseWriter, bodyErr *s3PutBodyError, err error, bucket, objectKey string) { + if bodyErr == nil && isRetryableS3MutationErr(err) { + writeS3SlowDown(w, bucket, objectKey) + return + } + s.writeS3ChunkUploadError(w, bodyErr, err, bucket, objectKey) +} + +func (s *S3Server) readS3UploadPartCommitState(ctx context.Context, state *s3UploadPartState, partKey []byte, bucket, objectKey string, startTS uint64) (*s3PartDescriptor, error) { + meta, exists, err := s.loadBucketMetaAt(ctx, bucket, startTS) if err != nil { return nil, errors.WithStack(err) } - return previous, nil + if !exists || meta == nil { + return nil, newS3ResponseError(http.StatusNotFound, s3ErrNoSuchBucket, s3ErrNoSuchBucketMsg, bucket, objectKey) + } + if meta.Generation != state.meta.Generation { + return nil, newS3ResponseError(http.StatusNotFound, "NoSuchUpload", "upload not found (bucket was recreated)", bucket, objectKey) + } + if err := s.verifyS3UploadStillExists(ctx, state.uploadMetaKey, bucket, objectKey, startTS); err != nil { + return nil, err + } + return s.loadPreviousS3PartDescriptor(ctx, partKey, startTS) } -func (s *S3Server) loadPreviousS3PartDescriptor(ctx context.Context, partKey []byte, readTS uint64) *s3PartDescriptor { +func (s *S3Server) loadPreviousS3PartDescriptor(ctx context.Context, partKey []byte, readTS uint64) (*s3PartDescriptor, error) { raw, err := s.store.GetAt(ctx, partKey, readTS) if err != nil { - return nil + if errors.Is(err, store.ErrKeyNotFound) { + return nil, nil + } + return nil, errors.WithStack(err) } + return decodeS3PartDescriptorOrNil(raw), nil +} + +// Keep the legacy malformed-descriptor fallback separate from storage errors, +// which must prevent a replacement from committing after a failed read. +func decodeS3PartDescriptorOrNil(raw []byte) *s3PartDescriptor { var descriptor s3PartDescriptor if json.Unmarshal(raw, &descriptor) != nil { return nil diff --git a/jepsen/src/elastickv/s3_workload.clj b/jepsen/src/elastickv/s3_workload.clj index 317b2aa1d..8eec38131 100644 --- a/jepsen/src/elastickv/s3_workload.clj +++ b/jepsen/src/elastickv/s3_workload.clj @@ -7,6 +7,7 @@ (:gen-class) (:require [clojure.string :as str] [clj-http.client :as http] + [clj-http.conn-mgr :as conn] [elastickv.cli :as cli] [elastickv.db :as ekdb] [jepsen [checker :as checker] @@ -21,7 +22,14 @@ [knossos.model :as model] [jepsen.nemesis.combined :as combined] [jepsen.os :as os] - [jepsen.os.debian :as debian])) + [jepsen.os.debian :as debian]) + (:import [java.io StringReader] + [java.util.concurrent Callable Executors Future TimeUnit] + [javax.xml XMLConstants] + [javax.xml.parsers DocumentBuilderFactory] + [org.w3c.dom Element] + [org.xml.sax InputSource] + [org.xml.sax.helpers DefaultHandler])) ;; --------------------------------------------------------------------------- ;; Constants @@ -82,6 +90,134 @@ (throw (ex-info (str "S3 GET error: HTTP " (:status resp)) {:status (:status resp) :body (:body resp)}))))) +(defn- xml-root + "Parse S3 XML without DTDs, external resources, or user-defined entities." + [body] + (let [factory (doto (DocumentBuilderFactory/newInstance) + (.setNamespaceAware true) + (.setFeature XMLConstants/FEATURE_SECURE_PROCESSING true) + (.setFeature "http://apache.org/xml/features/disallow-doctype-decl" true) + (.setFeature "http://xml.org/sax/features/external-general-entities" false) + (.setFeature "http://xml.org/sax/features/external-parameter-entities" false) + (.setFeature "http://apache.org/xml/features/nonvalidating/load-external-dtd" false) + (.setAttribute XMLConstants/ACCESS_EXTERNAL_DTD "") + (.setAttribute XMLConstants/ACCESS_EXTERNAL_SCHEMA "") + (.setXIncludeAware false) + (.setExpandEntityReferences false)) + builder (.newDocumentBuilder factory)] + (.setErrorHandler builder (proxy [DefaultHandler] [] + (error [e] (throw e)) + (fatalError [e] (throw e)))) + (.getDocumentElement (.parse builder (InputSource. (StringReader. (or body ""))))))) + +(defn- xml-field + [^Element root field] + (let [nodes (.getElementsByTagNameNS root "*" field)] + (when-not (= 1 (.getLength nodes)) + (throw (ex-info (str "Missing or duplicate S3 XML " field) {}))) + (let [^Element node (.item nodes 0) + value (.getTextContent node)] + (when-not (and (= root (.getParentNode node)) + (zero? (.getLength (.getElementsByTagNameNS node "*" "*"))) + (not (str/blank? value))) + (throw (ex-info (str "Malformed S3 XML " field) {}))) + value))) + +(defn- success! + [response] + (when-not (<= 200 (:status response) 299) + (throw (ex-info (str "S3 multipart HTTP " (:status response)) response))) + response) + +(defn- result-root! + [response expected] + (success! response) + (let [root (xml-root (:body response))] + (when-not (and (= expected (.getLocalName root)) + (zero? (.getLength (.getElementsByTagNameNS root "*" "Error")))) + (throw (ex-info "S3 multipart error or unexpected XML result" response))) + root)) + +(defn- part-etag! + [response] + (success! response) + (let [etag (get-in response [:headers "etag"])] + (when-not (and (string? etag) (re-matches #"\"[0-9a-fA-F]{32}\"" etag)) + (throw (ex-info "Missing or malformed UploadPart ETag" response))) + etag)) + +(def ^:private multipart-race-timeout-ms 20000) + +(defn- race-requests! + "Join both bounded HTTP calls, closing their sockets before joining workers + on every exit path. No shared future executor survives this write." + [manager requests] + (let [executor (Executors/newFixedThreadPool 2) + tasks (atom []) + deadline (+ (System/nanoTime) (* 1000000 multipart-race-timeout-ms))] + (try + (doseq [request requests] + (swap! tasks conj + (.submit executor ^Callable + (bound-fn [] + (try (request) + (catch Exception e {:exception e})))))) + (mapv (fn [^Future task] + (.get task (max 1 (- deadline (System/nanoTime))) TimeUnit/NANOSECONDS)) + @tasks) + (finally + (doseq [^Future task @tasks] (.cancel task true)) + (conn/shutdown-manager manager) + (.shutdownNow executor) + (when-not (.awaitTermination executor 5 TimeUnit/SECONDS) + (throw (ex-info "Multipart HTTP workers did not terminate" {}))))))) + +(defn- response! + [response] + (if-let [e (:exception response)] (throw e) response)) + +(defn- no-such-upload? + [response] + (and (= 404 (:status response)) + (let [root (xml-root (:body response))] + (and (= "Error" (.getLocalName root)) + (= "NoSuchUpload" (xml-field root "Code")))))) + +(defn- s3-multipart-put! + [base-url k v] + (let [manager (conn/make-reusable-conn-manager {:threads 2 :default-per-route 2}) + url (object-url base-url k) + opts {:connection-manager manager :throw-exceptions false :as :string + :conn-timeout 5000 :connection-request-timeout 5000 :socket-timeout 10000 + :retry-handler (fn [_ _ _] false) :redirect-strategy :none}] + (try + (let [root (result-root! (http/post url (assoc opts :query-params {"uploads" ""})) + "InitiateMultipartUploadResult") + id (xml-field root "UploadId") + _ (when (or (> (count id) 1024) (re-find #"[\s\p{Cntrl}]" id)) + (throw (ex-info "Malformed S3 UploadId" {}))) + part-opts (assoc opts :query-params {"uploadId" id "partNumber" 1} + :body (str v) :content-type "text/plain") + etag (part-etag! (http/put url part-opts)) + ;; ETag is the server's quoted MD5, not a locally substituted hash. + body (str "1" + etag "") + [replacement completion] + (race-requests! manager + [#(http/put url part-opts) + #(http/post url (assoc opts :query-params {"uploadId" id} + :body body :content-type "application/xml"))]) + completed (result-root! (response! completion) "CompleteMultipartUploadResult") + completed-etag (xml-field completed "ETag")] + (when-not (re-matches #"\"[0-9a-fA-F]{32}-1\"" completed-etag) + (throw (ex-info "Malformed completion ETag" {}))) + ;; Upload deletion is expected only after a confirmed completion. + (let [replacement (response! replacement)] + (when-not (no-such-upload? replacement) + (when-not (= etag (part-etag! replacement)) + (throw (ex-info "Same-body replacement changed ETag" {})))))) + (finally (conn/shutdown-manager manager))))) + ;; --------------------------------------------------------------------------- ;; Jepsen client ;; --------------------------------------------------------------------------- @@ -101,16 +237,20 @@ (close! [this _test] (assoc this :url nil)) - (invoke! [_this _test op] + (invoke! [_this test op] (try - (let [[k v] (:value op)] + (let [[k v] (:value op) + ;; Both modes run on the same CI cluster; keep their registers disjoint. + object-key (if (:multipart test) (str "multipart-" k) k)] (case (:f op) :write - (do (s3-put! url k v) + (do (if (:multipart test) + (s3-multipart-put! url object-key v) + (s3-put! url k v)) (assoc op :type :ok)) :read - (let [val (s3-get url k)] + (let [val (s3-get url object-key)] (assoc op :type :ok :value (independent/tuple k val))))) (catch java.net.ConnectException _ (assoc op :type :info :error :connection-refused)) @@ -134,6 +274,41 @@ (def default-nodes ["n1" "n2" "n3" "n4" "n5"]) +(defn multipart-final-read-ops + [key-count] + (map (fn [k] {:f :read :value (independent/tuple k nil) :final true}) + (range key-count))) + +(defn multipart-final-reads-checker + "Global acceptance gate: every configured key must finish its final read; + empty histories and runs with no confirmed writes cannot pass." + [key-count] + (reify checker/Checker + (check [_ _test history _opts] + (let [expected (set (range key-count)) + writes (filter #(and (= :write (:f %)) (= :ok (:type %))) history) + reads (filter #(and (= :read (:f %)) (= :ok (:type %))) history) + finals (filter #(and (:final %) (= :read (:f %))) history) + invoked (filter #(= :invoke (:type %)) finals) + completed (remove #(= :invoke (:type %)) finals) + successful (filter #(= :ok (:type %)) completed) + key-of #(first (:value %)) + written (set (map key-of writes)) + missing (remove #(contains? (set (map key-of successful)) %) expected) + bad (filter #(or (not= :ok (:type %)) + (not (or (integer? (second (:value %))) + (and (nil? (second (:value %))) + (not (contains? written (key-of %))))))) + completed)] + {:valid? (boolean (and (seq expected) (seq writes) (seq reads) + (= expected (set (map key-of invoked))) + (= key-count (count invoked) (count completed)) + (empty? missing) (empty? bad))) + :completed-writes (count writes) + :completed-reads (count reads) + :missing-final-keys (vec (sort missing)) + :bad-final-reads (mapv #(into {} %) bad)})))) + (defn s3-register-checker "Builds the per-key S3 register checker. Knossos's default competition algorithm races its linear and WGL analyzers and returns the first verdict." @@ -169,8 +344,11 @@ (->> (gen/mix [(map (fn [v] {:f :write :value v}) (range)) (gen/repeat {:f :read})]) (gen/limit max-writes)))) - :checker (independent/checker - (s3-register-checker))})) + :checker (let [register (independent/checker (s3-register-checker))] + (if (:multipart opts) + (checker/compose {:register register + :complete-final-reads (multipart-final-reads-checker key-count)}) + register))})) (defn elastickv-s3-test "Builds a Jepsen test map that drives elastickv's S3-compatible API @@ -201,9 +379,16 @@ nemesis-gen (if nemesis-p (:generator nemesis-p) (gen/once {:type :info :f :noop})) - workload (s3-register-workload (assoc opts :node->port node->port))] + workload (s3-register-workload (assoc opts :node->port node->port)) + main-gen (->> (:generator workload) + (gen/nemesis nemesis-gen) + (gen/stagger (/ rate)) + (gen/time-limit time-limit))] (merge workload - {:name (or (:name opts) "elastickv-s3-register") + (when (:multipart opts) {:multipart true}) + {:name (or (:name opts) (if (:multipart opts) + "elastickv-s3-multipart-register" + "elastickv-s3-register")) :nodes nodes :db db :s3-host (:s3-host opts) @@ -220,10 +405,14 @@ nemesis/noop) :final-generator nil :concurrency (or (:concurrency opts) 10) - :generator (->> (:generator workload) - (gen/nemesis nemesis-gen) - (gen/stagger (/ rate)) - (gen/time-limit time-limit))})))) + :generator (if (:multipart opts) + (gen/phases main-gen + (when nemesis-p + (gen/phases + (gen/nemesis (:final-generator nemesis-p)) + (gen/sleep 10))) + (gen/clients (multipart-final-read-ops (or (:key-count opts) 10)))) + main-gen)})))) ;; --------------------------------------------------------------------------- ;; CLI @@ -231,7 +420,9 @@ (def s3-cli-opts "S3-specific CLI options, appended to common opts." - [[nil "--s3-ports PORTS" "Comma separated S3 ports (per node)." + [[nil "--multipart" "Race same-body UploadPart with CompleteMultipartUpload." + :default false] + [nil "--s3-ports PORTS" "Comma separated S3 ports (per node)." :default nil :parse-fn (fn [s] (->> (str/split s #",") diff --git a/jepsen/test/elastickv/s3_workload_test.clj b/jepsen/test/elastickv/s3_workload_test.clj index 8fed19b36..c71db74ba 100644 --- a/jepsen/test/elastickv/s3_workload_test.clj +++ b/jepsen/test/elastickv/s3_workload_test.clj @@ -1,8 +1,18 @@ (ns elastickv.s3-workload-test (:require [clojure.test :refer :all] + [clojure.string :as str] + [clojure.tools.cli :as tools.cli] + [clj-http.client :as http] + [jepsen.checker :as checker] [jepsen.client :as client] + [jepsen.generator :as gen] + [jepsen.history :as history] + [jepsen.independent :as independent] [elastickv.cli :as cli] - [elastickv.s3-workload :as workload])) + [elastickv.s3-workload :as workload]) + (:import [com.sun.net.httpserver HttpHandler HttpServer] + [java.net InetSocketAddress URLDecoder] + [java.util.concurrent CountDownLatch Executors TimeUnit])) (deftest builds-test-spec (let [test-map (workload/elastickv-s3-test {})] @@ -33,3 +43,342 @@ c (:client test-map) opened (client/open! c test-map "n1")] (is (re-find #"http://127\.0\.0\.1:9000" (:url opened))))) + +(def ^:private etag "\"70efdf2ec9b086079795c442636b55fb\"") +(def ^:private upload-id "private/a+b&c") +(def ^:private created + {:status 200 + :body (str "" + "private/a+b&c")}) +(def ^:private uploaded {:status 200 :headers {"etag" etag}}) +(def ^:private completed + {:status 200 + :body ""0123456789abcdef0123456789abcdef-1""}) +(def ^:private no-upload + {:status 404 :body "NoSuchUpload"}) + +(defn- invoke-write + "Stub the HTTP boundary, retaining clj-http request options and real workers." + ([] (invoke-write {})) + ([overrides] + (let [calls (atom []) + parts (atom 0) + started (CountDownLatch. 2) + c (workload/->S3Client {} "http://stub:9000") + request (fn [method url opts] + (let [stage (if (= :post method) + (if (contains? (:query-params opts) "uploads") :create :complete) + (if (= 1 (swap! parts inc)) :initial :replace)) + _ (swap! calls conj {:stage stage :method method :url url :opts opts}) + response (get overrides stage + (case stage :create created :complete completed uploaded))] + (when (#{:replace :complete} stage) + (.countDown started) + (when-not (.await started 2 TimeUnit/SECONDS) + (throw (ex-info "HTTP race did not start both requests" {})))) + (if (fn? response) (response) response)))] + (with-redefs [http/post #(request :post %1 %2) + http/put #(request :put %1 %2)] + {:result (client/invoke! c {:multipart true} + {:type :invoke :f :write :value (independent/tuple 3 17)}) + :calls @calls})))) + +(deftest multipart-protocol-flow + (let [{:keys [result calls]} (invoke-write) + by-stage (into {} (map (juxt :stage identity) calls))] + (is (= :ok (:type result))) + (is (= (independent/tuple 3 17) (:value result))) + (is (= 4 (count calls))) + (is (= [:create :initial] (mapv :stage (take 2 calls)))) + (is (= #{:replace :complete} (set (map :stage (drop 2 calls))))) + (is (every? #(= "http://stub:9000/jepsen-test/key-multipart-3" (:url %)) calls)) + (is (= {"uploads" ""} (get-in by-stage [:create :opts :query-params]))) + (doseq [stage [:initial :replace]] + (is (= "17" (get-in by-stage [stage :opts :body]))) + (is (= {"uploadId" upload-id "partNumber" 1} + (get-in by-stage [stage :opts :query-params])))) + (is (= {"uploadId" upload-id} (get-in by-stage [:complete :opts :query-params]))) + (is (= (str "1" + etag "") + (get-in by-stage [:complete :opts :body]))) + (doseq [{:keys [opts]} calls] + (is (pos? (:conn-timeout opts))) + (is (pos? (:connection-request-timeout opts))) + (is (pos? (:socket-timeout opts))) + (is (= :none (:redirect-strategy opts))) + (is (false? ((:retry-handler opts) nil 1 nil))))) + (let [c (workload/->S3Client {} "http://stub:9000")] + (with-redefs [http/get (fn [url _] + (is (str/ends-with? url "/key-multipart-3")) + {:status 200 :body " 17\n"})] + (is (= (independent/tuple 3 17) + (:value (client/invoke! c {:multipart true} + {:f :read :value (independent/tuple 3 nil)}))))))) + +(deftest replacement-no-such-upload-needs-confirmed-completion + (is (= :ok (get-in (invoke-write {:replace no-upload}) [:result :type]))) + (doseq [completion [no-upload + {:status 503 :body "SlowDown"} + {:status 200 :body "InternalError"} + (fn [] (throw (java.net.SocketTimeoutException. "lost reply")))]] + (let [{:keys [result calls]} (invoke-write {:replace no-upload :complete completion})] + (is (= :info (:type result))) + (is (= 1 (count (filter #(= :complete (:stage %)) calls)))))) + (doseq [replacement [{:status 404 :body "NoSuchKey"} + {:status 500 :body "failure"} + {:status 200 :headers {}} + {:status 200 :headers {"etag" "bad"}} + {:status 200 :headers {"etag" "\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\""}} + {:status 404 :body "NoSuchUploadNoSuchKey"}]] + (is (= :info (get-in (invoke-write {:replace replacement}) [:result :type]))))) + +(deftest malformed-initiation-never-starts-parts-or-completion + (doseq [body ["" "" + "" + "" + " " + "bad id" + "nested" + "ab" + "SlowDown" + "a" + "]>&id;" + "]>&id;" + "a"]] + (let [{:keys [result calls]} (invoke-write {:create {:status 200 :body body}})] + (is (= :info (:type result)) body) + (is (= [:create] (mapv :stage calls)) body))) + (doseq [status [301 400 500 503]] + (is (= :info (get-in (invoke-write {:create (assoc created :status status)}) [:result :type]))))) + +(deftest first-part-must-return-a-valid-server-etag + (doseq [response [{:status 200} + {:status 200 :headers {"etag" ""}} + {:status 200 :headers {"etag" "not-an-etag"}} + {:status 200 :headers {"etag" ""}} + {:status 200 :headers {"etag" "70efdf2ec9b086079795c442636b55fb"}} + no-upload + {:status 503 :body "SlowDown"}]] + (let [{:keys [result calls]} (invoke-write {:initial response})] + (is (= :info (:type result))) + (is (= [:create :initial] (mapv :stage calls)))))) + +(deftest completion-validates-http-and-xml-without-replaying + (doseq [response [{:status 301 :body (:body completed)} + {:status 503 :body (:body completed)} + {:status 200 :body ""} + {:status 200 :body ""} + {:status 200 :body ""} + {:status 200 :body "bad"} + {:status 200 :body "InternalError"} + {:status 200 :body "okInternalError"} + (fn [] (throw (java.net.SocketTimeoutException. "unknown completion"))) + (fn [] (throw (org.apache.http.NoHttpResponseException. "unknown completion")))]] + (let [{:keys [result calls]} (invoke-write {:complete response})] + (is (= :info (:type result))) + (is (= 1 (count (filter #(= :complete (:stage %)) calls))))))) + +(deftest race-timeout-cancels-and-joins-workers + (let [active (atom 0) + stopped (CountDownLatch. 1)] + (with-redefs-fn {#'workload/multipart-race-timeout-ms 100} + #(let [{:keys [result]} (invoke-write + {:replace (fn [] + (swap! active inc) + (try (Thread/sleep 10000) + uploaded + (finally + (swap! active dec) + (.countDown stopped))))})] + (is (= :info (:type result))) + (is (zero? @active)) + (is (zero? (.getCount stopped))))))) + +(deftest uncertain-completion-still-joins-replacement + (let [finished (atom false)] + (is (= :info (get-in (invoke-write + {:complete #(throw (java.net.SocketTimeoutException. "unknown")) + :replace #(do (Thread/sleep 20) (reset! finished true) uploaded)}) + [:result :type]))) + (is @finished))) + +(defn- with-http-stub + [respond f] + (let [server (HttpServer/create (InetSocketAddress. "127.0.0.1" 0) 0) + executor (Executors/newFixedThreadPool 4) + calls (atom [])] + (.setExecutor server executor) + (.createContext server "/" + (reify HttpHandler + (handle [_ exchange] + (try + (let [uri (.getRequestURI exchange) + query (into {} (for [pair (str/split (or (.getRawQuery uri) "") #"&") + :when (seq pair) + :let [[k v] (str/split pair #"=" 2)]] + [(URLDecoder/decode k "UTF-8") + (URLDecoder/decode (or v "") "UTF-8")])) + request {:method (.getRequestMethod exchange) + :path (.getPath uri) :query query + :body (slurp (.getRequestBody exchange))} + _ (swap! calls conj request) + response (respond request)] + (when-not (:drop response) + (doseq [[k v] (:headers response)] + (.set (.getResponseHeaders exchange) k v)) + (let [bytes (.getBytes ^String (or (:body response) "") "UTF-8")] + (.sendResponseHeaders exchange (:status response) + (if (seq bytes) (alength bytes) -1)) + (when (seq bytes) + (with-open [out (.getResponseBody exchange)] + (.write out bytes)))))) + (finally (.close exchange)))))) + (try + (.start server) + (f (str "http://127.0.0.1:" (.getPort (.getAddress server))) calls) + (finally + (.stop server 0) + (.shutdownNow executor) + (is (.awaitTermination executor 5 TimeUnit/SECONDS)))))) + +(deftest multipart-over-real-http-does-not-replay-an-uncertain-completion + (doseq [outcome [:success :no-upload :lost-reply :embedded-error]] + (let [parts (atom 0) + raced (CountDownLatch. 2)] + (with-http-stub + (fn [{:keys [method query]}] + (cond + (contains? query "uploads") created + (and (= "PUT" method) (= 1 (swap! parts inc))) uploaded + :else + (do + (.countDown raced) + (when-not (.await raced 2 TimeUnit/SECONDS) + (throw (ex-info "HTTP calls did not overlap" {}))) + (if (= "PUT" method) + (if (= :success outcome) uploaded no-upload) + (case outcome + :lost-reply {:drop true} + :embedded-error {:status 200 :body "InternalError"} + completed))))) + (fn [url calls] + (let [c (workload/->S3Client {} url) + result (client/invoke! c {:multipart true} + {:f :write :value (independent/tuple 3 17)}) + requests @calls + completions (filter #(and (= "POST" (:method %)) + (contains? (:query %) "uploadId")) requests)] + (is (= (if (#{:success :no-upload} outcome) :ok :info) (:type result)) + (str outcome ": " (:error result))) + (is (= 4 (count requests))) + (is (= 1 (count completions))) + (is (= upload-id (get-in (first completions) [:query "uploadId"]))) + (is (every? #(= "/jepsen-test/key-multipart-3" (:path %)) requests)) + (is (= ["17" "17"] (mapv :body (filter #(= "PUT" (:method %)) requests)))))))))) + +(deftest real-http-timeout-closes-sockets-and-joins-workers + (let [parts (atom 0) + raced (CountDownLatch. 2)] + (with-redefs-fn {#'workload/multipart-race-timeout-ms 500} + #(with-http-stub + (fn [{:keys [method query]}] + (cond + (contains? query "uploads") created + (and (= "PUT" method) (= 1 (swap! parts inc))) uploaded + :else + (do + (.countDown raced) + (.await raced 2 TimeUnit/SECONDS) + (if (= "PUT" method) + (do (.await (CountDownLatch. 1) 10 TimeUnit/SECONDS) uploaded) + completed)))) + (fn [url calls] + (let [c (workload/->S3Client {} url) + result (client/invoke! c {:multipart true} + {:f :write :value (independent/tuple 3 17)})] + (is (= :info (:type result))) + ;; A timed-out worker is cancelled, its socket closed, and its + ;; executor joined before invoke! can return this outcome. + (is (not= "Multipart HTTP workers did not terminate" (:error result))) + (is (= 4 (count @calls))))))))) + +(defn- events + [f k v & [attrs]] + (let [op (merge {:process 0 :f f :value (independent/tuple k v)} attrs)] + [(assoc op :type :invoke) (assoc op :type :ok)])) + +(defn- final-check + [key-count ops] + (checker/check (workload/multipart-final-reads-checker key-count) + {} (history/history ops) {})) + +(deftest final-read-gate-requires-every-key-and-nonvacuous-writes + (let [write (events :write 0 17) + read0 (events :read 0 17 {:final true}) + read1 (events :read 1 nil {:final true}) + valid (vec (concat write read0 read1))] + (is (true? (:valid? (final-check 2 valid)))) + (is (= 1 (:completed-writes (final-check 2 valid)))) + (is (= 2 (:completed-reads (final-check 2 valid)))) + (doseq [ops [[] write read0 (concat read0 read1) + (concat write read0) + (concat write read0 (take 1 read1)) + (concat write read0 (events :read 2 nil {:final true})) + (concat write read0 read0 read1) + (concat write (events :read 0 nil {:final true}) read1) + (concat write (events :read 0 :bad {:final true}) read1)]] + (is (false? (:valid? (final-check 2 ops))))) + (is (false? (:valid? (final-check 0 [])))) + (doseq [type [:info :fail]] + (let [ops (concat write read0 [(first read1) (assoc (second read1) :type type :error "HTTP 500")])] + (is (false? (:valid? (final-check 2 ops)))))))) + +(deftest failed-final-get-cannot-pass-the-gate + (let [op (assoc (first (events :read 0 nil)) :final true) + c (workload/->S3Client {} "http://stub:9000")] + (doseq [response [{:status 500 :body "missing chunks"} + {:status 200 :body "not a long"} + {:status 404 :body ""}]] + (with-redefs [http/get (fn [_ _] response)] + (let [result (client/invoke! c {:multipart true} op)] + (is (false? (:valid? (final-check 1 (concat (events :write 0 17) [op result])))))))))) + +(deftest multipart-cli-and-checker-are-opt-in + (let [default (tools.cli/parse-opts [] workload/s3-cli-opts) + enabled (tools.cli/parse-opts ["--multipart"] workload/s3-cli-opts) + normal (workload/elastickv-s3-test {:local true}) + multipart (workload/elastickv-s3-test {:local true :multipart true})] + (is (empty? (:errors enabled))) + (is (false? (get-in default [:options :multipart]))) + (is (true? (get-in enabled [:options :multipart]))) + (is (nil? (:multipart normal))) + (is (true? (:multipart multipart))) + (is (= "elastickv-s3-register" (:name normal))) + (is (= "elastickv-s3-multipart-register" (:name multipart))) + (is (nil? (:final-generator multipart))) + (is (= #{:register :complete-final-reads} (set (keys (:checkers (:checker multipart)))))) + (is (nil? (:checkers (:checker normal)))))) + +(deftest multipart-test-generator-ends-with-all-final-reads + (let [test (workload/elastickv-s3-test {:local true :multipart true :time-limit 0 :key-count 3}) + ctx (gen/context test) + ops (loop [g (:generator test), ops []] + (let [[op g'] (gen/op g test ctx)] + (if (or (nil? op) (= :pending op)) + ops + (recur g' (conj ops op))))) + finals (filter :final ops)] + (is (= [0 1 2] (sort (map (comp first :value) finals)))) + (is (every? #(= :read (:f %)) finals)))) + +(deftest default-mode-still-uses-put-and-get + (let [calls (atom []) + c (workload/->S3Client {} "http://stub:9000")] + (with-redefs [http/put (fn [url opts] (swap! calls conj [:put url (:body opts)]) {:status 200}) + http/post (fn [& _] (throw (ex-info "Unexpected multipart" {}))) + http/get (fn [url _] (swap! calls conj [:get url]) {:status 200 :body "17"})] + (is (= :ok (:type (client/invoke! c {} {:f :write :value (independent/tuple 3 17)})))) + (is (= (independent/tuple 3 17) + (:value (client/invoke! c {} {:f :read :value (independent/tuple 3 nil)})))) + (is (= [[:put "http://stub:9000/jepsen-test/key-3" "17"] + [:get "http://stub:9000/jepsen-test/key-3"]] @calls))))) diff --git a/kv/coordinator.go b/kv/coordinator.go index 4c4567020..5f82e2c55 100644 --- a/kv/coordinator.go +++ b/kv/coordinator.go @@ -1538,9 +1538,20 @@ func (c *Coordinate) toRawRequest(req *Elem[OP]) *pb.Request { var ErrInvalidRequest = errors.New("invalid request") -// maxReadKeys caps the number of keys that may appear in a transaction's read -// set. Exceeding this limit is rejected to prevent unbounded memory growth. -const maxReadKeys = 10_000 +// MaxReadKeys bounds the physical read set in a transaction request, including +// staged-visibility aliases. S3 completion must support all 10,000 parts even +// during migration: reserve 240 logical keys for metadata/fences and write +// aliases, then allow two physical keys per logical read. Keep the post-alias +// checks as well as the input checks to bound accepted physical key counts +// without dropping conflict keys or reducing the supported part count. This +// count is not a byte budget or a bound on allocations before alias checks. +const ( + maxDataReadKeys = 10_000 + readKeyHeadroom = 240 + MaxReadKeys = 2 * (maxDataReadKeys + readKeyHeadroom) +) + +const maxReadKeys = MaxReadKeys var ErrLeaderNotFound = errors.New("leader not found") diff --git a/kv/read_keys_capacity_test.go b/kv/read_keys_capacity_test.go new file mode 100644 index 000000000..e12771219 --- /dev/null +++ b/kv/read_keys_capacity_test.go @@ -0,0 +1,302 @@ +package kv + +import ( + "bytes" + "context" + "fmt" + "sync" + "testing" + + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/bootjp/elastickv/internal/s3keys" + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +const readKeysCapacityJobID = 9 + +// Observe the real proposal boundary without replacing Raft or the FSM. +type readKeysCapacityEngine struct { + raftengine.Engine + mu sync.Mutex + payloads [][]byte +} + +func (e *readKeysCapacityEngine) record(data []byte) { + e.mu.Lock() + defer e.mu.Unlock() + e.payloads = append(e.payloads, bytes.Clone(data)) +} + +func (e *readKeysCapacityEngine) Propose(ctx context.Context, data []byte) (*raftengine.ProposalResult, error) { + e.record(data) + return e.Engine.Propose(ctx, data) +} + +func (e *readKeysCapacityEngine) ProposeAdmin(ctx context.Context, data []byte) (*raftengine.ProposalResult, error) { + e.record(data) + return e.Engine.ProposeAdmin(ctx, data) +} + +func (e *readKeysCapacityEngine) proposals() [][]byte { + e.mu.Lock() + defer e.mu.Unlock() + return append([][]byte(nil), e.payloads...) +} + +func (e *readKeysCapacityEngine) requests(t *testing.T) []*pb.Request { + t.Helper() + var requests []*pb.Request + for _, payload := range e.proposals() { + if len(payload) == 0 || payload[0] != raftEncodeSingle { + continue + } + request := &pb.Request{} + require.NoError(t, proto.Unmarshal(payload[1:], request)) + requests = append(requests, request) + } + return requests +} + +type readKeysCapacityHarness struct { + coord *ShardedCoordinator + engines []*readKeysCapacityEngine + stores []store.MVCCStore +} + +func newReadKeysCapacityHarness(t *testing.T, staged, multiShard bool) readKeysCapacityHarness { + t.Helper() + routes := distribution.NewEngine() + descriptors := []distribution.RouteDescriptor{{ + RouteID: 1, GroupID: 1, State: distribution.RouteStateActive, + StagedVisibilityActive: staged, + }} + if staged { + descriptors[0].MigrationJobID = readKeysCapacityJobID + } + if multiShard { + descriptors[0].End = []byte("m") + descriptors = append(descriptors, distribution.RouteDescriptor{ + RouteID: 2, GroupID: 2, Start: []byte("m"), State: distribution.RouteStateActive, + }) + } + require.NoError(t, routes.ApplySnapshot(distribution.CatalogSnapshot{Version: 1, Routes: descriptors})) + clock := NewHLC() + groups := make(map[uint64]*ShardGroup) + h := readKeysCapacityHarness{} + for _, descriptor := range descriptors { + st := store.NewMVCCStore() + fsm := NewKvFSMWithHLC(st, clock, WithRouteHistory(WrapDistributionEngine(routes), descriptor.GroupID)) + engine, stop := newSingleRaft(t, fmt.Sprintf("read-cap-%d", descriptor.GroupID), fsm) + t.Cleanup(stop) + recorder := &readKeysCapacityEngine{Engine: engine} + groups[descriptor.GroupID] = &ShardGroup{Engine: recorder, Store: st, Txn: NewLeaderProxyWithEngine(recorder)} + h.engines = append(h.engines, recorder) + h.stores = append(h.stores, st) + } + h.coord = NewShardedCoordinator(routes, groups, 1, clock, NewShardStore(routes, groups)) + t.Cleanup(func() { require.NoError(t, h.coord.Close()) }) + return h +} + +func readKeysCapacityParts(count int) [][]byte { + keys := make([][]byte, count) + for i := range keys { + keys[i] = s3keys.UploadPartKey("capacity", 1, "object", "upload", uint64(i+1)) + } + return keys +} + +func readKeysCapacityRequest(count int) *OperationGroup[OP] { + return &OperationGroup[OP]{ + IsTxn: true, StartTS: 10, CommitTS: 100, + ReadKeys: readKeysCapacityParts(count), + Elems: []*Elem[OP]{ + {Op: Put, Key: s3keys.BucketMetaKey("capacity"), Value: []byte("bucket")}, + {Op: Put, Key: s3keys.ObjectManifestKey("capacity", 1, "object"), Value: []byte("manifest")}, + {Op: Del, Key: s3keys.UploadMetaKey("capacity", 1, "object", "upload")}, + {Op: Del, Key: s3keys.GCUploadKey("capacity", 1, "object", "upload")}, + }, + } +} + +func dispatchReadKeysCapacity(ctx context.Context, coord Coordinator, request *OperationGroup[OP], leaderStamped bool) (*CoordinateResponse, error) { + if leaderStamped { + request.CommitTS = 0 + return DispatchWithLeaderIssuedCommitTS(ctx, coord, request) + } + return coord.Dispatch(ctx, request) +} + +func TestReadKeysCapacityInputBoundary(t *testing.T) { + t.Parallel() + require.Equal(t, 20_480, MaxReadKeys) + for _, path := range []string{"coordinate", "sharded-explicit", "sharded-leader-stamped"} { + t.Run(path, func(t *testing.T) { + t.Parallel() + h := newReadKeysCapacityHarness(t, false, false) + var coord Coordinator = h.coord + if path == "coordinate" { + plain := NewCoordinatorWithEngine(NewLeaderProxyWithEngine(h.engines[0]), h.engines[0]) + t.Cleanup(func() { require.NoError(t, plain.Close()) }) + coord = plain + } + ctx := context.Background() + response, err := dispatchReadKeysCapacity(ctx, coord, readKeysCapacityRequest(MaxReadKeys), path == "sharded-leader-stamped") + require.NoError(t, err) + require.Positive(t, response.CommitIndex) + requests := h.engines[0].requests(t) + require.Len(t, requests, 1) + require.Len(t, requests[0].ReadKeys, MaxReadKeys) + before := len(h.engines[0].proposals()) + _, err = dispatchReadKeysCapacity(ctx, coord, readKeysCapacityRequest(MaxReadKeys+1), path == "sharded-leader-stamped") + require.ErrorIs(t, err, ErrInvalidRequest) + require.Len(t, h.engines[0].proposals(), before, "input overflow must fail before any proposal") + }) + } +} + +func TestReadKeysCapacityRedirectBoundary(t *testing.T) { + t.Parallel() + coord := &Coordinate{} + requests, err := coord.buildRedirectRequests(readKeysCapacityRequest(MaxReadKeys)) + require.NoError(t, err) + require.Len(t, requests, 1) + require.Len(t, requests[0].ReadKeys, MaxReadKeys) + requests, err = coord.buildRedirectRequests(readKeysCapacityRequest(MaxReadKeys + 1)) + require.ErrorIs(t, err, ErrInvalidRequest) + require.Nil(t, requests, "overflow must not produce a forwardable request") +} + +func assertReadKeysCapacityAliases(t *testing.T, request *OperationGroup[OP], physical [][]byte) { + t.Helper() + want := make(map[string]int) + for _, key := range request.ReadKeys { + want[string(key)]++ + want[string(distribution.MigrationStagedDataKey(readKeysCapacityJobID, key))]++ + } + for _, elem := range request.Elems { + if bytes.Equal(elem.Key, []byte("z")) { + continue // The second shard has no staged visibility. + } + want[string(distribution.MigrationStagedDataKey(readKeysCapacityJobID, elem.Key))]++ + } + got := make(map[string]int) + for _, key := range physical { + got[string(key)]++ + } + require.Equal(t, want, got, "every read, fence and write alias must reach Raft, including duplicates") +} + +func TestReadKeysCapacityStagedVisibility(t *testing.T) { + t.Parallel() + for _, leaderStamped := range []bool{false, true} { + t.Run(fmt.Sprintf("leader-stamped=%t", leaderStamped), func(t *testing.T) { + t.Parallel() + h := newReadKeysCapacityHarness(t, true, false) + ctx := context.Background() + request := readKeysCapacityRequest(10_000) + request.ReadKeys = append(request.ReadKeys, + s3keys.BucketMetaKey("capacity"), s3keys.BucketGenerationKey("capacity"), + s3keys.UploadMetaKey("capacity", 1, "object", "upload"), s3keys.ObjectManifestKey("capacity", 1, "object")) + // Build the read set from actual staged-only values, not synthetic aliases. + for _, key := range request.ReadKeys { + alias := distribution.MigrationStagedDataKey(readKeysCapacityJobID, key) + require.NoError(t, h.stores[0].PutAt(ctx, alias, []byte("staged"), 1, 0)) + value, err := h.coord.store.GetAt(ctx, key, request.StartTS) + require.NoError(t, err) + require.Equal(t, []byte("staged"), value) + } + response, err := dispatchReadKeysCapacity(ctx, h.coord, request, leaderStamped) + require.NoError(t, err) + require.Greater(t, response.CommitTS, request.StartTS) + requests := h.engines[0].requests(t) + require.Len(t, requests, 1) + require.Len(t, requests[0].ReadKeys, 2*(10_000+4)+4) + assertReadKeysCapacityAliases(t, request, requests[0].ReadKeys) + value, err := h.stores[0].GetAt(ctx, request.Elems[1].Key, response.CommitTS) + require.NoError(t, err) + require.Equal(t, []byte("manifest"), value) + }) + } +} + +func TestReadKeysCapacityStagedConflicts(t *testing.T) { + t.Parallel() + for _, leaderStamped := range []bool{false, true} { + t.Run(fmt.Sprintf("leader-stamped=%t", leaderStamped), func(t *testing.T) { + t.Parallel() + for _, target := range []string{"last-part", "fence", "bucket-write", "manifest-write", "upload-delete", "gc-delete"} { + t.Run(target, func(t *testing.T) { + t.Parallel() + h := newReadKeysCapacityHarness(t, true, false) + request := readKeysCapacityRequest(10_000) + fence := s3keys.BucketGenerationKey("capacity") + conflicts := map[string][]byte{ + "last-part": request.ReadKeys[len(request.ReadKeys)-1], + "fence": fence, + "bucket-write": request.Elems[0].Key, + "manifest-write": request.Elems[1].Key, + "upload-delete": request.Elems[2].Key, + "gc-delete": request.Elems[3].Key, + } + request.ReadKeys = append(request.ReadKeys, fence) + alias := distribution.MigrationStagedDataKey(readKeysCapacityJobID, conflicts[target]) + ctx := context.Background() + require.NoError(t, h.stores[0].PutAt(ctx, alias, []byte("concurrent"), request.StartTS+1, 0)) + _, err := dispatchReadKeysCapacity(ctx, h.coord, request, leaderStamped) + require.ErrorIs(t, err, store.ErrWriteConflict, "expanded capacity must not skip any OCC aliases") + requests := h.engines[0].requests(t) + require.Len(t, requests, 1, "the FSM, not the capacity guard, must reject the conflict") + assertReadKeysCapacityAliases(t, request, requests[0].ReadKeys) + _, err = h.stores[0].GetAt(ctx, request.Elems[1].Key, ^uint64(0)) + require.ErrorIs(t, err, store.ErrKeyNotFound, "a rejected completion must not publish the manifest") + }) + } + }) + } +} + +func TestReadKeysCapacityStagedBoundary(t *testing.T) { + t.Parallel() + for _, path := range []string{"explicit", "leader-stamped", "multi-shard"} { + t.Run(path, func(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + reads int + fits bool + }{ + {name: "physical-at-cap", reads: (MaxReadKeys - 4) / 2, fits: true}, + {name: "physical-over-cap", reads: (MaxReadKeys-4)/2 + 1}, + {name: "input-at-cap-expands-over", reads: MaxReadKeys}, + {name: "input-over-cap", reads: MaxReadKeys + 1}, + } { + t.Run(tc.name, func(t *testing.T) { + h := newReadKeysCapacityHarness(t, true, path == "multi-shard") + request := readKeysCapacityRequest(tc.reads) + if path == "multi-shard" { + request.Elems = append(request.Elems, &Elem[OP]{Op: Put, Key: []byte("z"), Value: []byte("second shard")}) + } + response, err := dispatchReadKeysCapacity(context.Background(), h.coord, request, path == "leader-stamped") + if !tc.fits { + require.ErrorIs(t, err, ErrInvalidRequest) + for _, engine := range h.engines { + require.Empty(t, engine.proposals(), "overflow must fail before any participant proposes") + } + return + } + require.NoError(t, err) + require.Positive(t, response.CommitIndex) + requests := h.engines[0].requests(t) + require.NotEmpty(t, requests) + require.Len(t, requests[0].ReadKeys, MaxReadKeys) + assertReadKeysCapacityAliases(t, request, requests[0].ReadKeys) + }) + } + }) + } +} From 1e2bd5e5544dde088e35dd643581310ef0436a52 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 30 Oct 2026 01:58:58 +0900 Subject: [PATCH 7/8] docs: define multipart read-key byte admission Signed-off-by: bootjp --- ...6_09_26_partial_serializable_isolation_audit.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/docs/design/2026_09_26_partial_serializable_isolation_audit.md b/docs/design/2026_09_26_partial_serializable_isolation_audit.md index ec9924433..ebc9a09e8 100644 --- a/docs/design/2026_09_26_partial_serializable_isolation_audit.md +++ b/docs/design/2026_09_26_partial_serializable_isolation_audit.md @@ -1180,6 +1180,20 @@ or acceptance receipt. Implementation now proceeds from main `c316b829e` on Preserve both named S3 histories, source SHA, encryption mode and step outcomes as unconditional CI artifacts. Do not weaken the nonvacuity or final-read gate to obtain a passing result. +- Security acceptance: reject public S3 object keys exceeding 1,024 decoded + UTF-8 bytes before proxying, body consumption or mutation. Apply the same + creation limit to admin PUT; admin GET/delete remain available for recovery + of previously stored nonconforming keys. Before allocating completion + descriptor/read-key arrays or reading descriptors, enforce a 24 MiB logical + aggregate read-key budget, including escaped segment lengths and fixed key + encoding overhead. Use division/subtraction rather than unchecked byte-count + multiplication. This supports 10,000 maximum-length conforming keys, including + zero-byte escaping; migration alias expansion remains separately bounded by + the existing physical key-count cap. The budget bounds this dependency set, + not total request/protobuf/Raft heap or concurrency, and does not change RawKV + value limits. Existing overlong public S3 keys require the admin recovery API. + Add runtime-red HTTP/admin and pre-allocation/no-descriptor-read regressions + before the fix, then repeat latest-push code/security review and CI. - Verify all changed helper callers, full adapter/KV/store race suites, unfiltered lint, relevant Jepsen checks, and latest-push review/CI. A1 does not establish cross-group serializability before A2's read-lock protocol, From 51409d937b9a822e8334f0352acc0983736fedfc Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 30 Oct 2026 02:05:51 +0900 Subject: [PATCH 8/8] adapter: bound S3 object keys and multipart read-key bytes Signed-off-by: bootjp --- adapter/s3.go | 10 +- adapter/s3_admin_objects.go | 2 +- adapter/s3_multipart_complete.go | 3 + adapter/s3_multipart_limits.go | 55 ++++++++++ adapter/s3_multipart_limits_test.go | 165 ++++++++++++++++++++++++++++ 5 files changed, 230 insertions(+), 5 deletions(-) create mode 100644 adapter/s3_multipart_limits.go create mode 100644 adapter/s3_multipart_limits_test.go diff --git a/adapter/s3.go b/adapter/s3.go index a6b094aa4..5352ddf34 100644 --- a/adapter/s3.go +++ b/adapter/s3.go @@ -493,15 +493,17 @@ func (s *S3Server) handle(w http.ResponseWriter, r *http.Request) { if s.serveS3LeaderHealthz(w, r) { return } - if s.maybeProxyToLeader(w, r) { - return - } - bucket, objectKey, hasObject, err := parseS3Path(r) if err != nil { writeS3Error(w, http.StatusBadRequest, "InvalidURI", err.Error(), bucket, objectKey) return } + if rejectOversizedS3ObjectKey(w, objectKey, hasObject) { + return + } + if s.maybeProxyToLeader(w, r) { + return + } if authErr := s.resolveAuth(r, bucket); authErr != nil { writeS3Error(w, authErr.Status, authErr.Code, authErr.Message, bucket, objectKey) diff --git a/adapter/s3_admin_objects.go b/adapter/s3_admin_objects.go index 2dcc7a213..afaea1a9a 100644 --- a/adapter/s3_admin_objects.go +++ b/adapter/s3_admin_objects.go @@ -204,7 +204,7 @@ func (s *S3Server) AdminPutObject(ctx context.Context, principal AdminPrincipal, return err } defer s.exitAdminBlobUse() - if key == "" { + if key == "" || len(key) > s3MaxObjectKeyBytes { return ErrAdminInvalidObjectKey } if err := validateS3BucketName(bucket); err != nil { diff --git a/adapter/s3_multipart_complete.go b/adapter/s3_multipart_complete.go index 704d5d20b..7dda275bf 100644 --- a/adapter/s3_multipart_complete.go +++ b/adapter/s3_multipart_complete.go @@ -97,6 +97,9 @@ func (s *S3Server) loadS3MultipartUploadMeta(ctx context.Context, readTS uint64, } func (s *S3Server) loadS3MultipartParts(ctx context.Context, readTS uint64, requested []s3CompleteMultipartUploadPart, completion *s3MultipartCompletion) error { + if err := validateS3MultipartReadKeyBudget(completion, len(requested)); err != nil { + return err + } completion.manifestParts = make([]s3ObjectPart, 0, len(requested)) completion.partKeys = make([][]byte, 0, len(requested)) md5Concat := md5.New() //nolint:gosec // S3 composite ETag compatibility requires MD5. diff --git a/adapter/s3_multipart_limits.go b/adapter/s3_multipart_limits.go new file mode 100644 index 000000000..72a235c3a --- /dev/null +++ b/adapter/s3_multipart_limits.go @@ -0,0 +1,55 @@ +package adapter + +import ( + "net/http" + "strings" + + "github.com/bootjp/elastickv/internal/s3keys" +) + +const ( + s3MaxObjectKeyBytes = 1024 + s3MultipartReadKeyByteBudget = 24 << 20 +) + +func rejectOversizedS3ObjectKey(w http.ResponseWriter, key string, hasObject bool) bool { + if !hasObject || len(key) <= s3MaxObjectKeyBytes { + return false + } + writeS3Error(w, http.StatusBadRequest, "KeyTooLongError", "object key exceeds 1024 bytes", "", "") + return true +} + +func validateS3MultipartReadKeyBudget(completion *s3MultipartCompletion, count int) error { + if len(completion.objectKey) > s3MaxObjectKeyBytes { + return newS3ResponseError(http.StatusBadRequest, "KeyTooLongError", "object key exceeds 1024 bytes", "", "") + } + if count <= 0 || count > s3MaxPartsPerUpload { + return s3MultipartReadKeyBudgetError() + } + // All part numbers and the generation have fixed-width encodings. Only + // zero escaping increases segment lengths; account for it without copying + // a caller-controlled segment or multiplying potentially large lengths. + remaining := s3MultipartReadKeyByteBudget / count + overhead := len(s3keys.UploadPartKey("", 0, "", "", 0)) + if overhead > remaining { + return s3MultipartReadKeyBudgetError() + } + remaining -= overhead + for _, segment := range []string{completion.bucket, completion.objectKey, completion.uploadID} { + if len(segment) > remaining { + return s3MultipartReadKeyBudgetError() + } + remaining -= len(segment) + escapes := strings.Count(segment, "\x00") + if escapes > remaining { + return s3MultipartReadKeyBudgetError() + } + remaining -= escapes + } + return nil +} + +func s3MultipartReadKeyBudgetError() error { + return newS3ResponseError(http.StatusBadRequest, "InvalidArgument", "multipart read keys exceed the byte budget", "", "") +} diff --git a/adapter/s3_multipart_limits_test.go b/adapter/s3_multipart_limits_test.go new file mode 100644 index 000000000..8fd0e6751 --- /dev/null +++ b/adapter/s3_multipart_limits_test.go @@ -0,0 +1,165 @@ +package adapter + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" + "github.com/stretchr/testify/require" +) + +func TestS3ObjectKeyByteLimit(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + key string + status int + }{ + {"ascii boundary", strings.Repeat("a", 1024), http.StatusOK}, + {"ascii overflow", strings.Repeat("a", 1025), http.StatusBadRequest}, + {"utf8 boundary", strings.Repeat("\xc3\xa9", 512), http.StatusOK}, + {"utf8 overflow", strings.Repeat("\xc3\xa9", 513), http.StatusBadRequest}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + h := newS3OCCReproHarness(t) + h.startUpload("key-limit") + response := h.do(http.MethodPost, "/key-limit/"+url.PathEscape(tc.key)+"?uploads", "") + require.Equal(t, tc.status, response.Code, response.Body.String()) + }) + } +} + +func TestS3AdminPutRejectsOverlongObjectKey(t *testing.T) { + t.Parallel() + h := newS3OCCReproHarness(t) + h.startUpload("admin-key-limit") + err := h.server.AdminPutObject(t.Context(), fullAdminBucketsPrincipal(), "admin-key-limit", strings.Repeat("a", 1025), strings.NewReader("value"), "text/plain") + require.ErrorIs(t, err, ErrAdminInvalidObjectKey) +} + +type s3DescriptorReadCounter struct { + store.MVCCStore + reads int +} + +func (s *s3DescriptorReadCounter) GetAt(context.Context, []byte, uint64) ([]byte, error) { + s.reads++ + return nil, errors.WithStack(store.ErrKeyNotFound) +} + +func TestS3MultipartReadKeyBudgetPrecedesAllocationAndReads(t *testing.T) { + t.Parallel() + const budget = 24 << 20 + for _, escaped := range []bool{false, true} { + t.Run(map[bool]string{false: "plain", true: "escaped"}[escaped], func(t *testing.T) { + t.Parallel() + segment := "a" + if escaped { + segment = "\x00" + } + completion := s3MultipartCompletion{bucket: strings.Repeat(segment, budget/10000), objectKey: "object", uploadID: "upload"} + requests := make([]s3CompleteMultipartUploadPart, 10000) + for i := range requests { + requests[i].PartNumber = i + 1 + } + st := &s3DescriptorReadCounter{} + err := (&S3Server{store: st}).loadS3MultipartParts(t.Context(), 1, requests, &completion) + var responseErr *s3ResponseError + require.ErrorAs(t, err, &responseErr) + require.Equal(t, "InvalidArgument", responseErr.Code) + require.Zero(t, st.reads, "reject before loading even the first descriptor") + require.Zero(t, cap(completion.partKeys), "reject before retaining the key array") + require.Zero(t, cap(completion.manifestParts), "reject before retaining manifest storage") + }) + } +} + +func TestS3MaximumConformingMultipartKeysFitByteBudget(t *testing.T) { + t.Parallel() + key := s3keys.UploadPartKey(strings.Repeat("b", 63), ^uint64(0), strings.Repeat("\x00", 1024), strings.Repeat("u", 64), 10000) + require.LessOrEqual(t, len(key)*10000, 24<<20) + require.NoError(t, validateS3MultipartReadKeyBudget(&s3MultipartCompletion{ + bucket: strings.Repeat("b", 63), objectKey: strings.Repeat("\x00", 1024), uploadID: strings.Repeat("u", 64), + }, 10000)) +} + +func TestS3MultipartReadKeyBudgetBoundary(t *testing.T) { + t.Parallel() + const count = 10000 + overhead := len(s3keys.UploadPartKey("", 0, "object", "upload", 0)) + completion := s3MultipartCompletion{bucket: strings.Repeat("a", s3MultipartReadKeyByteBudget/count-overhead), objectKey: "object", uploadID: "upload"} + require.NoError(t, validateS3MultipartReadKeyBudget(&completion, count)) + completion.bucket += "a" + require.Error(t, validateS3MultipartReadKeyBudget(&completion, count)) + for _, invalid := range []int{-1, 0, count + 1} { + require.Error(t, validateS3MultipartReadKeyBudget(&completion, invalid)) + } +} + +type s3NeverReadBody struct{ reads int } + +func (b *s3NeverReadBody) Read([]byte) (int, error) { + b.reads++ + return 0, io.EOF +} + +type s3ProxyRoutingCounter struct { + kv.Coordinator + calls int +} + +func (c *s3ProxyRoutingCounter) IsLeaderForKey([]byte) bool { + c.calls++ + return false +} + +func (*s3ProxyRoutingCounter) RaftLeaderForKey([]byte) string { return "" } + +func TestS3OverlongObjectKeyRejectedBeforeProxyAndBody(t *testing.T) { + t.Parallel() + h := newS3OCCReproHarness(t) + h.startUpload("bucket") + coordinator := &s3ProxyRoutingCounter{Coordinator: h.coord} + h.server.coordinator = coordinator + beforeReads := len(h.reads.readLog()) + body := &s3NeverReadBody{} + request := httptest.NewRequest(http.MethodPut, "/bucket/"+strings.Repeat("a", 1025), body) + response := httptest.NewRecorder() + h.server.handle(response, request) + require.Equal(t, http.StatusBadRequest, response.Code) + require.Zero(t, body.reads) + require.Zero(t, coordinator.calls) + require.Len(t, h.reads.readLog(), beforeReads) + // Positive control: entering the real proxy path does read and route. + require.True(t, h.server.maybeProxyToLeader(httptest.NewRecorder(), request)) + require.Positive(t, coordinator.calls) + require.Greater(t, len(h.reads.readLog()), beforeReads) +} + +func TestS3AdminLegacyOverlongObjectRecovery(t *testing.T) { + t.Parallel() + h := newS3OCCReproHarness(t) + h.startUpload("legacy-key") + key := strings.Repeat("a", 1025) + // Seed a pre-limit object through the existing internal fixture pipeline. + require.NoError(t, h.server.adminPutObject(t.Context(), &s3PutObjectInput{ + body: io.NopCloser(strings.NewReader("legacy")), contentLength: -1, + streamBody: &s3StreamingBody{}, contentType: "text/plain", + }, "legacy-key", key)) + body, _, err := h.server.AdminGetObject(t.Context(), fullAdminBucketsPrincipal(), "legacy-key", key) + require.NoError(t, err) + value, err := io.ReadAll(body) + require.NoError(t, err) + require.NoError(t, body.Close()) + require.Equal(t, "legacy", string(value)) + require.NoError(t, h.server.AdminDeleteObject(t.Context(), fullAdminBucketsPrincipal(), "legacy-key", key)) +}