adapter: validate S3 multipart snapshots and retain uncertain writes

.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

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)
 }
 

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

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 {

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
+}

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) {

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)
+				}
+			}
+		})
+	}
+}