perf(dynamodb): merge writes into cached scan order; fix ExclusiveStartKey resume and S3 InvalidDigest

e2e_bench_sns_test.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	sdksns "github.com/aws/aws-sdk-go-v2/service/sns"
+	sdksqs "github.com/aws/aws-sdk-go-v2/service/sqs"
+	sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	e2eSNSQueues  = 3
+	e2eSNSHTTPSub = 3
+)
+
+func BenchmarkE2E_SNS(b *testing.B) {
+	e, _ := benchServer(b)
+	cfg := newBenchAWSConfig(b, benchTransport{e: e})
+	ctx := context.Background()
+
+	snsC := sdksns.NewFromConfig(cfg, func(o *sdksns.Options) { o.BaseEndpoint = aws.String(benchEndpoint) })
+	sqsC := sdksqs.NewFromConfig(cfg, func(o *sdksqs.Options) { o.BaseEndpoint = aws.String(benchEndpoint) })
+
+	sink := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	b.Cleanup(sink.Close)
+
+	newTopic := func(b *testing.B, name string, queues, httpSubs int) string {
+		b.Helper()
+
+		tp, err := snsC.CreateTopic(ctx, &sdksns.CreateTopicInput{Name: aws.String(name)})
+		require.NoError(b, err)
+
+		for i := range queues {
+			q, qerr := sqsC.CreateQueue(ctx, &sdksqs.CreateQueueInput{
+				QueueName: aws.String(name + "-q" + strconv.Itoa(i)),
+			})
+			require.NoError(b, qerr)
+
+			attrs, aerr := sqsC.GetQueueAttributes(ctx, &sdksqs.GetQueueAttributesInput{
+				QueueUrl:       q.QueueUrl,
+				AttributeNames: []sqstypes.QueueAttributeName{sqstypes.QueueAttributeNameQueueArn},
+			})
+			require.NoError(b, aerr)
+
+			_, serr := snsC.Subscribe(ctx, &sdksns.SubscribeInput{
+				TopicArn: tp.TopicArn, Protocol: aws.String("sqs"),
+				Endpoint: aws.String(attrs.Attributes[string(sqstypes.QueueAttributeNameQueueArn)]),
+			})
+			require.NoError(b, serr)
+		}
+
+		for range httpSubs {
+			_, serr := snsC.Subscribe(ctx, &sdksns.SubscribeInput{
+				TopicArn: tp.TopicArn, Protocol: aws.String("http"), Endpoint: aws.String(sink.URL),
+			})
+			require.NoError(b, serr)
+		}
+
+		return aws.ToString(tp.TopicArn)
+	}
+
+	for _, tc := range []struct {
+		name          string
+		queues, https int
+	}{
+		{"publish_no_subs", 0, 0},
+		{"publish_3sqs", e2eSNSQueues, 0},
+		{"publish_3sqs_3http", e2eSNSQueues, e2eSNSHTTPSub},
+	} {
+		b.Run(tc.name, func(b *testing.B) {
+			arn := newTopic(b, "e2e-bench-"+tc.name, tc.queues, tc.https)
+
+			b.ReportAllocs()
+
+			for b.Loop() {
+				_, err := snsC.Publish(ctx, &sdksns.PublishInput{
+					TopicArn: aws.String(arn), Message: aws.String("e2e bench notification"),
+					Subject: aws.String("bench"),
+				})
+				require.NoError(b, err)
+			}
+		})
+	}
+}

pkgs/persistence/testdata/snapshot_inventory.json

@@ -8912,6 +8912,8 @@
       "Table.mu *lockmetrics.RWMutex",
       "Table.pkIndex map[string]int",
       "Table.pkskIndex map[string]map[string]int",
+      "Table.scanLog []scanOp",
+      "Table.scanLogBase uint64",
       "Table.scanOrder atomic.Pointer[scanOrderCache]",
       "Table.streamSeq int64",
       "Table.streamTrimSeq int64",
@@ -8943,6 +8945,9 @@
       "pitrSnapshot.Taken time.Time `json:\"Taken\"`",
       "replicaAutoScalingSettings.GlobalSecondaryIndexes map[string]*autoScalingThroughput `json:\"GlobalSecondaryIndexes,omitempty\"`",
       "replicaAutoScalingSettings.Read *autoScalingThroughput `json:\"Read,omitempty\"`",
+      "scanOp.deleted bool",
+      "scanOp.item map[string]any",
+      "scanOp.size int",
       "secondaryIndex.pkOnly map[string]map[int]struct{}",
       "secondaryIndex.pksk map[string]map[string]map[int]struct{}",
       "storedExport.BilledSizeBytes int64",

services/dynamodb/item_ops.go

@@ -2,6 +2,7 @@ package dynamodb
 
 import (
 	"bytes"
+	"cmp"
 	"context"
 	"encoding/base64"
 	"fmt"
@@ -686,13 +687,26 @@ func getAttributeType(
 	return defaultType
 }
 
-// findExclusiveStartIndex finds the index after the ExclusiveStartKey in the candidates list.
-// Returns 0 if ExclusiveStartKey is nil or not found.
+// findExclusiveStartIndex returns the index just past the ExclusiveStartKey in candidates,
+// sorted ascending by the index sort key.
 func findExclusiveStartIndex(
 	candidates []map[string]any,
 	exclusiveStartKey map[string]any,
 	keySchema []models.KeySchemaElement,
 	tableKeySchema []models.KeySchemaElement,
+) int {
+	return queryStartIndex(candidates, exclusiveStartKey, keySchema, tableKeySchema, true)
+}
+
+// queryStartIndex returns the index of the first candidate after the ExclusiveStartKey in
+// query order. The key is a position: if its item is gone, the next candidate in sort order
+// is returned. candidates must already be sorted in the query direction.
+func queryStartIndex(
+	candidates []map[string]any,
+	exclusiveStartKey map[string]any,
+	keySchema []models.KeySchemaElement,
+	tableKeySchema []models.KeySchemaElement,
+	forward bool,
 ) int {
 	if len(exclusiveStartKey) == 0 {
 		return 0
@@ -707,7 +721,50 @@ func findExclusiveStartIndex(
 		}
 	}
 
-	return 0
+	startSK, ok := exclusiveStartKey[skDef.AttributeName]
+	if skDef.AttributeName == "" || !ok {
+		return 0
+	}
+
+	typ := sortKeyType(startSK)
+
+	for i, item := range candidates {
+		c := compareSortKeyValues(item[skDef.AttributeName], startSK, typ)
+		if !forward {
+			c = -c
+		}
+
+		if c > 0 {
+			return i
+		}
+	}
+
+	return len(candidates)
+}
+
+// sortKeyType returns the DynamoDB scalar type tag ("N", "S" or "B") of a wire attribute.
+func sortKeyType(v any) string {
+	if m, ok := v.(map[string]any); ok {
+		for _, t := range []string{"N", "B"} {
+			if _, has := m[t]; has {
+				return t
+			}
+		}
+	}
+
+	return "S"
+}
+
+// compareSortKeyValues orders two sort-key values the way sortCandidates does.
+func compareSortKeyValues(a, b any, typ string) int {
+	if typ == "N" {
+		an, _ := dynamoattr.ParseNumeric(a)
+		bn, _ := dynamoattr.ParseNumeric(b)
+
+		return cmp.Compare(an, bn)
+	}
+
+	return cmp.Compare(dynamoattr.ToString(a), dynamoattr.ToString(b))
 }
 
 // itemMatchesStartKeyMap reports whether item matches the ExclusiveStartKey for the given

services/dynamodb/item_ops_batch.go

@@ -936,7 +936,7 @@ func (db *InMemoryDB) applyBatchDeletes(table *Table, indices []int) {
 		// Capture stream record (REMOVE)
 		table.appendStreamRecord(streamEventRemove, table.Items[idx], nil, "", "")
 
-		table.itemsChanged()
+		table.itemRemoved(table.Items[idx])
 		// Delete by swapping with last and truncating. table.itemSizes must be
 		// kept in lockstep with table.Items (same swap, same truncation) so its
 		// length never drifts from Items and totalItemSizeBytes stays accurate
@@ -952,7 +952,7 @@ func (db *InMemoryDB) applyBatchDeletes(table *Table, indices []int) {
 	// rebuildIndexes recomputes the key indexes AND itemSizes/totalItemSizeBytes
 	// from the surviving Items, so the size accounting stays consistent with the
 	// deletions above.
-	table.rebuildIndexes()
+	table.rebuildDerivedState()
 }
 
 // updateBatchIndexes incrementally updates every index (primary and
@@ -1050,7 +1050,7 @@ func (db *InMemoryDB) handleBatchPutWithIndex(table *Table, item map[string]any)
 	// this, DescribeTable's TableSizeBytes silently excludes anything written
 	// via BatchWriteItem, and itemSizes/Items can drift out of length-sync.
 	itemSize, _ := CalculateItemSize(item)
-	table.itemsChanged()
+	table.itemStored(item, itemSize)
 	oldItem, matchIndex := db.findMatchForPut(table, item)
 	if matchIndex != -1 {
 		// Capture stream event (MODIFY) before overwriting in place.

services/dynamodb/item_ops_crud.go

@@ -233,7 +233,7 @@ func (db *InMemoryDB) doPut(table *Table, item map[string]any, matchIndex int) {
 }
 
 func (db *InMemoryDB) doPutWithSize(table *Table, item map[string]any, matchIndex int, itemSize int) {
-	table.itemsChanged()
+	table.itemStored(item, itemSize)
 	if matchIndex != -1 {
 		oldItem := table.Items[matchIndex]
 		table.totalItemSizeBytes += int64(itemSize) - int64(table.itemSizes[matchIndex])
@@ -1040,7 +1040,7 @@ func (db *InMemoryDB) commitUpdate(
 	matchIndex int,
 ) {
 	updatedSize, _ := CalculateItemSize(updated)
-	table.itemsChanged()
+	table.itemStored(updated, updatedSize)
 
 	if matchIndex != -1 {
 		table.totalItemSizeBytes += int64(updatedSize) - int64(table.itemSizes[matchIndex])
@@ -1207,7 +1207,7 @@ func (db *InMemoryDB) deleteItemAtIndex(table *Table, matchIndex int) {
 
 	table.updateSecondaryIndexes(item, matchIndex, nil, 0)
 
-	table.itemsChanged()
+	table.itemRemoved(item)
 	// Swap with last strategy for O(1) deletion
 	lastIdx := len(table.Items) - 1
 	deletedSize := table.itemSizes[matchIndex]

services/dynamodb/item_ops_query.go

@@ -612,7 +612,10 @@ func (db *InMemoryDB) processQueryResults(
 ) (*pageResult, error) {
 	exclusiveStartKey := models.FromSDKItem(input.ExclusiveStartKey)
 
-	startIndex := findExclusiveStartIndex(candidates, exclusiveStartKey, keySchema, tableKeySchema)
+	startIndex := queryStartIndex(
+		candidates, exclusiveStartKey, keySchema, tableKeySchema,
+		input.ScanIndexForward == nil || *input.ScanIndexForward,
+	)
 
 	items, lastEvaluatedKey, scannedCount, err := db.collectQueryPage(
 		ctx,