diff --git a/.github/workflows/jepsen-test.yml b/.github/workflows/jepsen-test.yml
index 52298327d..2706255c3 100644
--- a/.github/workflows/jepsen-test.yml
+++ b/.github/workflows/jepsen-test.yml
@@ -59,7 +59,6 @@ jobs:
- name: Pre-fetch Go modules and build binary
run: |
mkdir -p "$GOCACHE" /tmp/go-tmp
- GOPATH=$(go env GOPATH)
export GOCACHE GOTMPDIR=/tmp/go-tmp
go mod download
go build -o /tmp/elastickv-bin .
@@ -143,7 +142,7 @@ jobs:
done
echo "Waiting for redis (63791-63793), dynamo (63801-63803), s3 (63901-63903), sqs (63501-63503) listeners and FUSE mounts..."
- for i in {1..90}; do
+ for _ in {1..90}; do
if nc -z 127.0.0.1 63791 && nc -z 127.0.0.1 63792 && nc -z 127.0.0.1 63793 \
&& nc -z 127.0.0.1 63801 && nc -z 127.0.0.1 63802 && nc -z 127.0.0.1 63803 \
&& nc -z 127.0.0.1 63901 && nc -z 127.0.0.1 63902 && nc -z 127.0.0.1 63903 \
@@ -255,10 +254,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 Standard Jepsen workload against elastickv
working-directory: jepsen
timeout-minutes: 3
diff --git a/adapter/s3.go b/adapter/s3.go
index f6bb02b01..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)
@@ -801,7 +803,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 +834,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 +1386,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 +1407,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)
+ completion, previous, err := s.commitS3MultipartCompletion(r.Context(), bucket, objectKey, uploadID, request)
if err != nil {
- writeS3ResponseOrInternalError(w, err)
- return
- }
- previous, err := s.commitS3MultipartCompletion(r.Context(), completion)
- if err != nil {
- writeS3MutationError(w, err, bucket, objectKey)
+ writeS3MultipartMutationError(w, err, bucket, objectKey)
return
}
@@ -2552,6 +2543,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_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_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..7dda275bf 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 {
@@ -105,10 +97,19 @@ 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.
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 +117,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 +157,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 +217,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 +234,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_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))
+}
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/docs/design/2026_09_26_partial_serializable_isolation_audit.md b/docs/design/2026_09_26_partial_serializable_isolation_audit.md
index d7838c55f..ebc9a09e8 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,80 @@ 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 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. 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
+ 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.
+ 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.
+ 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.
+ 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.
+- 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,
+ 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
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)
+ })
+ }
+ })
+ }
+}