diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index 3997563ab1..eb724b096c 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -1,6 +1,9 @@ {"_type":"issue","id":"gopherstack-3js4","title":"dynamodb: DeleteTable leaves fisReplicationPaused keyed by a name-deterministic ARN, so a recreated table starts replication-paused","status":"closed","priority":0,"issue_type":"bug","created_at":"2026-09-04T07:54:20Z","updated_at":"2026-09-04T08:00:46Z","closed_at":"2026-09-04T08:00:46Z","close_reason":"fixed and regression-tested on chore/parity-sweep-2026-09-03","labels":["parity-campaign"],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-wnmd","title":"lambda: DeleteFunction leaves ~17 side maps including resource-policy permissions and aliases; a recreated function inherits them","status":"closed","priority":0,"issue_type":"bug","created_at":"2026-09-04T07:54:19Z","updated_at":"2026-09-04T08:00:45Z","started_at":"2026-09-04T07:54:30Z","closed_at":"2026-09-04T08:00:45Z","close_reason":"fixed: DeleteFunction now reuses deleteFunctionMapsLocked; verified via the real HTTP handler that a recreated function no longer inherits the resource policy, aliases or concurrency","labels":["parity-campaign"],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-gmc5","title":"HANDOFF 2026-08-06: uncommitted in-flight work on chore/parity-upgrade","description":"Session ended mid-flight. 18 commits pushed on chore/parity-upgrade (PR #2414). The working tree has UNCOMMITTED work that survives on disk — do not discard it.\n\nUNCOMMITTED, VERIFIED COMPLETE (safe to commit after re-running gates):\n- services/grafana (12 files) — B to A. New test/integration/grafana_test.go, real cross-service validation via a new cross_service.go (captures ctx.Config at Provider.Init, resolves sibling handlers lazily on first request, no cli.go edits — REUSE THIS PATTERN for outposts/mgn/resiliencehub), chaos-driven FAILED/DEGRADED transitions, ListVersions moved to structural_gaps. Unit gates verified green by the main thread.\n- services/networkmanager (15 files) — gap to A. New test/integration/networkmanager_test.go, cross-service validation against EC2/DirectConnect, a real single-hop TGW route-analysis walk replacing a hardcoded NOT_CONNECTED, and a real core-network policy diff engine. Unit gates verified green.\n\nUNCOMMITTED, MID-EDIT (an agent was still working when the session ended — REVIEW BEFORE TRUSTING):\n- services/bedrockagent, services/cleanrooms, pkgs/httputils, cli.go, test/integration/tag_routing_test.go\n\nTHE BLOCKER — read this before committing anything above.\nTestIntegration_Grafana_WorkspaceLifecycle/Tags FAILS (grafana_test.go:521, TagResource should succeed). Root cause is a router bug class, NOT grafana:\nSeveral services' RouteMatcher do an unguarded strings.HasPrefix(path, \"/tags/\") with no SigV4 service-scope guard, so they swallow other services' tag requests. Confirmed in services/bedrockagent/handler.go:229-234 and services/cleanrooms/handler.go:312-323. A previous pass had masked this by escalating networkManagerMatchPriority to 88; that escalation was reverted (correctly) which un-masked cleanrooms. Do NOT re-escalate priority — cleanrooms beats grafana regardless of networkmanager's priority.\nCorrect fix, in progress: guard each prefix fallback so it does not match when ExtractServiceFromRequest names a different known service; sweep EVERY service RouteMatcher for the same pattern (check /resourcepolicy, /flows, /agents, /prompts too); extend test/integration/tag_routing_test.go to cover every service serving /tags/ in ONE binary run. A shared helper (service.PrefixMatcherWithScopeGuard) was proposed so the convention cannot be skipped. Tracked as gopherstack-sokq.\nThis class is invisible when services test alone — every one passes in isolation.\n\nNEXT SERVICES for the all-services-A program (2 herdr tabs max, sonnet, see the herdr-delegate skill): outposts (gopherstack-b9mg), mgn (gopherstack-xd34), resiliencehub (gopherstack-lxs2). directconnect already landed in 198990e82.\n\nPROCESS NOTE: the directconnect agent committed despite an explicit instruction not to. Main thread commits; re-state that in every brief and verify with git log.","status":"closed","priority":0,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-08-06T19:27:57Z","created_by":"Witness Patrol","updated_at":"2026-08-07T05:29:00Z","closed_at":"2026-08-07T05:29:00Z","close_reason":"Handoff complete. All work it described has landed and pushed: grafana and networkmanager reached A, the router prefix-collision fix (ef896bcf1), and the follow-on services. Nothing uncommitted remains.","dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"gopherstack-cfam2","title":"perf(s3): stream MD5 for UploadPart instead of hashing a buffered copy","description":"PGO 2026-10-12: md5 in UploadPart = ~4.5-6% CPU. Hash while reading the body (single pass), avoid extra copies; ETag and checksum behaviour unchanged.","notes":"Investigated 2026-10-13: UploadPart and PutObject already hash in a single streaming MultiWriter pass; md5 cost is the ETag floor. Added Content-MD5/ETag guard test.","status":"closed","priority":1,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T03:49:23Z","created_by":"Witness Patrol","updated_at":"2026-10-13T04:24:59Z","closed_at":"2026-10-13T04:24:59Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"gopherstack-6dyao","title":"perf(dynamodb): cache sorted scan order instead of re-sorting each Scan","description":"PGO 2026-10-12: sortScanResultsSized via storeScanOrder = ~4.7% CPU. Cache sorted order per table/index, invalidate on writes; identical results/pagination.","status":"closed","priority":1,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T03:49:22Z","created_by":"Witness Patrol","updated_at":"2026-10-13T04:30:24Z","closed_at":"2026-10-13T04:30:24Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"gopherstack-xd3vh","title":"perf(sns): avoid RSA-signing every notification on Publish","description":"PGO 2026-10-12: sns notificationSigner.signSHA1 (RSA PKCS1v15) = ~4.5% total CPU. Sign once per message per signature version and reuse across subscribers, and only when a delivery needs the signature; keep wire output identical.","notes":"Investigated 2026-10-13: Publish already signs once per distinct body and skips signing when no consumer needs it; remaining cost is one RSA-2048 sign per publish (AWS-faithful key size). Only lever is a 1024-bit key (rejected: fidelity/CodeQL). Added BenchmarkE2E_SNS + signature-verification guard test.","status":"closed","priority":1,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T03:49:18Z","created_by":"Witness Patrol","updated_at":"2026-10-13T04:24:57Z","closed_at":"2026-10-13T04:24:57Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-8r9ey","title":"Athena integration test DDL/DML now FAILED under SQL engine","description":"CI integration-tests(0): TestIntegration_Athena_DDLAndDMLQueryLifecycle athena_test.go:175 expected SUCCEEDED got FAILED after the SQLite engine (f67e74448/7401e0e93). Find which statement fails (StateChangeReason), decide if engine should accept it (AWS-valid) or test is unrealistic; fix with regression test.","status":"closed","priority":1,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-12T05:16:45Z","created_by":"Witness Patrol","updated_at":"2026-10-12T06:27:13Z","closed_at":"2026-10-12T06:27:13Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-gb98x","title":"Flaky: iotdataplane TestGetConnection_ReportsLiveBrokerSession/clean_with_socket","description":"Failed in CI unit-tests(0) on PR #2482 (wire_sdk_roundtrip_test.go:69 Not equal); passes 20x locally under -race. Likely races broker session registration vs GetConnection under load. Make deterministic (wait on session-registered signal, no sleeps) and prove with -count=50 -race.","status":"closed","priority":1,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-11T22:56:45Z","created_by":"Witness Patrol","updated_at":"2026-10-12T00:00:28Z","closed_at":"2026-10-12T00:00:28Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-cefu7.10","title":"Per-service feature-depth comparison for existing AWS services","description":"For every service we both cover, diff documented operations/behaviours against the reference set and file per-service gaps. Track privately; repo docs must not mention the comparison.","status":"open","priority":1,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-10-11T14:33:30Z","created_by":"Witness Patrol","updated_at":"2026-10-11T14:33:30Z","dependencies":[{"issue_id":"gopherstack-cefu7.10","depends_on_id":"gopherstack-cefu7","type":"parent-child","created_at":"2026-10-11T09:33:30Z","created_by":"Witness Patrol","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} @@ -314,6 +317,7 @@ {"_type":"issue","id":"gopherstack-rnd","title":"EventBridge Pipes AWS-accuracy audit (GH#1818)","status":"closed","priority":1,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-05-29T21:27:34Z","created_by":"mayor","updated_at":"2026-07-26T14:59:24Z","closed_at":"2026-07-26T14:59:24Z","close_reason":"Superseded: pipes audited 2026-07-24 (5d5b2188), overall A. Execution gaps closed; only remaining gap is a proven impossibility (no in-repo Kafka/AMQP broker).","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-o2j","title":"OpenSearch AWS-accuracy audit (GH#1817)","status":"closed","priority":1,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-05-29T21:27:29Z","created_by":"mayor","updated_at":"2026-07-26T14:59:23Z","closed_at":"2026-07-26T14:59:23Z","close_reason":"Superseded: opensearch audited 2026-07-25 (acb2e23f9, parity-4), overall A-, gaps documented. Issue predates the parity campaigns.","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-g3y","title":"EC2 batch-4 audit: VPC endpoints, TGW, NACL, Route Tables, NAT Gateway. Real stateful emulation, 2k+ lines.","status":"closed","priority":1,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-05-29T10:49:37Z","created_by":"mayor","updated_at":"2026-07-26T14:59:24Z","closed_at":"2026-07-26T14:59:24Z","close_reason":"Superseded: all five families exist and ec2 audited 2026-07-25 (parity-4) at overall A. Remaining EC2 field-diff work tracked in gopherstack-8pce.","dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"gopherstack-nwr62","title":"DynamoDB Scan: ExclusiveStartKey of a deleted item restarts from the beginning","description":"Real DynamoDB resumes after the key's position. A test (TestScanPaging_StartKeyOfDeletedItemRestartsLikeLinearPath) currently locks in the wrong behaviour. Fix with binary-search insertion point; update test; regression test.","status":"closed","priority":2,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T04:18:57Z","created_by":"Witness Patrol","updated_at":"2026-10-13T20:59:57Z","closed_at":"2026-10-13T20:59:57Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-cefu7.51","title":"ID shape audit: uuid.NewString()[:N] IDs vs SDK patterns","description":"Transfer IDs violated SDK patterns. Same construction remains in glue(10), efs(5), opensearch(4), codeartifact(4), mq/ecs/cloudfront(2), ssm, rds, kafka, eks, kinesis. Audit each against the pinned SDK pattern; add a guard where patterns are machine-readable.","status":"open","priority":2,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-10-12T19:56:20Z","created_by":"Witness Patrol","updated_at":"2026-10-12T19:56:20Z","dependencies":[{"issue_id":"gopherstack-cefu7.51","depends_on_id":"gopherstack-cefu7","type":"parent-child","created_at":"2026-10-12T14:56:19Z","created_by":"Witness Patrol","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-cefu7.50","title":"Organizations: async CreateAccount (IN_PROGRESS -\u003e SUCCEEDED)","description":"CreateAccount completes synchronously; model the async status with an opt-in delay.","status":"open","priority":2,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-10-12T19:56:17Z","created_by":"Witness Patrol","updated_at":"2026-10-12T19:56:17Z","dependencies":[{"issue_id":"gopherstack-cefu7.50","depends_on_id":"gopherstack-cefu7","type":"parent-child","created_at":"2026-10-12T14:56:17Z","created_by":"Witness Patrol","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-cefu7.49","title":"IoT: MQTT broker port configurable; jobs MQTT topics; policy doc validation","description":"Broker hardcoded to 1883 (collides with a running instance); $aws/things/+/jobs/* topics not served; policy document structure not validated.","status":"open","priority":2,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-10-12T19:56:14Z","created_by":"Witness Patrol","updated_at":"2026-10-12T19:56:14Z","dependencies":[{"issue_id":"gopherstack-cefu7.49","depends_on_id":"gopherstack-cefu7","type":"parent-child","created_at":"2026-10-12T14:56:14Z","created_by":"Witness Patrol","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} @@ -1539,6 +1543,8 @@ {"_type":"issue","id":"go-hwb.106","title":"S3 Control: ~48 missing ops (access points/grants/batch jobs/MRAP)","description":"## S3 Control — Service Deep Dive\n\nAudit of [services/s3control/](services/s3control/) and UI in [ui/src/routes/s3control/](ui/src/routes/s3control/).\n\n### 1. Missing SDK Operations\n~48 missing ([sdk_completeness_test.go#L21](services/s3control/sdk_completeness_test.go#L21)): `DeleteAccessGrant`, `DeleteBucket`, `GetAccessPoint`, `ListAccessPoints`, `PutAccessPointPolicy`, Access Grants, Access Points, Batch Jobs, MRAP, Storage Lens Group. Only 13 supported (public access block + partial).\n\n### 2. Missing UI / Dashboard Features\nPublic access block display only. Missing: access points mgmt, access grants, batch job UI, MRAP, storage lens groups, Object Lambda.\n\n### 3. Goroutine / Resource / Lock Leaks\nClean. Map cloning on snapshot ([persistence.go#L44](services/s3control/persistence.go#L44)).\n\n### 4. Performance Optimizations\n1. 10+ separate maps — consolidate with typed keys to reduce Reset cost.\n2. Atomic counter for IDs ([backend.go#L178](services/s3control/backend.go#L178)) good.\n\n### Suggested Order\n1. Access Points (Create/Get/List/Put policy)\n2. Access Grants (Create/Delete/List)\n3. Batch Jobs + MRAP + Storage Lens Group\n4. Consolidate map structure\n\n\n---\n**Source:** https://github.com/BlackbirdWorks/gopherstack/issues/1223\n","status":"closed","priority":2,"issue_type":"task","owner":"andrew.bishop9625@gmail.com","created_at":"2026-05-02T18:28:35Z","created_by":"mayor","updated_at":"2026-07-30T16:59:49Z","closed_at":"2026-07-30T16:59:49Z","close_reason":"STALE (parity-5 verification sweep): verified against code, not PARITY.md prose. The authoritative reflective TestSDKCompleteness test passes for this service with an empty notImplemented list, i.e. zero unaccounted SDK operations - refuting the ticket's missing-ops claim. UI route page exists and is substantial. Ticket was auto-generated 2026-05-02, before the parity-3/4/5 campaigns did this work.","external_ref":"gh-1223","labels":["ai-queue"],"dependencies":[{"issue_id":"go-hwb.106","depends_on_id":"go-hwb","type":"parent-child","created_at":"2026-05-02T13:28:35Z","created_by":"Witness Patrol","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"go-hwb.102","title":"Timestream Write: SDK complete; per-table WriteRecords locks","description":"## Timestream Write — Service Deep Dive\n\nAudit of [services/timestreamwrite/](services/timestreamwrite/) and shared UI in [ui/src/routes/timestream/](ui/src/routes/timestream/).\n\n### 1. Missing SDK Operations\n**0 missing.** 20 ops implemented including `CreateDatabase`, `CreateTable`, `WriteRecords`, `CreateBatchLoadTask`, `ResumeBatchLoadTask`, tags.\n\n### 2. Missing UI / Dashboard Features\nShared UI covers DBs + tables + scheduled queries. Full CRUD. Batch load UI could be enhanced.\n\n### 3. Goroutine / Resource / Lock Leaks\nClean. 4 nested maps under single `lockmetrics.RWMutex` ([backend.go#L159](services/timestreamwrite/backend.go#L159)).\n\n### 4. Performance Optimizations\n1. **Single mutex serializes WriteRecords across tables** — partition by table-ARN for ~10x throughput.\n2. Dispatch pre-built ([handler.go#L62](services/timestreamwrite/handler.go#L62)).\n\n### Suggested Order\n1. Per-table-ARN partition locks for `WriteRecords`\n2. Batch load UI polish\n\n\n---\n**Source:** https://github.com/BlackbirdWorks/gopherstack/issues/1227\n","status":"closed","priority":2,"issue_type":"task","owner":"andrew.bishop9625@gmail.com","created_at":"2026-05-02T18:28:34Z","created_by":"mayor","updated_at":"2026-07-30T16:59:48Z","closed_at":"2026-07-30T16:59:48Z","close_reason":"STALE (parity-5 verification sweep): verified against code, not PARITY.md prose. The authoritative reflective TestSDKCompleteness test passes for this service with an empty notImplemented list, i.e. zero unaccounted SDK operations - refuting the ticket's missing-ops claim. UI route page exists and is substantial. Ticket was auto-generated 2026-05-02, before the parity-3/4/5 campaigns did this work.","external_ref":"gh-1227","labels":["ai-queue"],"dependencies":[{"issue_id":"go-hwb.102","depends_on_id":"go-hwb","type":"parent-child","created_at":"2026-05-02T13:28:34Z","created_by":"Witness Patrol","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"go-hwb","title":"Epic: ai-queue from BlackbirdWorks/gopherstack","description":"Autonomous grinding of GitHub issues labeled 'ai-queue' from BlackbirdWorks/gopherstack. Each child bead corresponds to one GitHub issue (external-ref gh-N). Launched via gt mountain for wave-based dispatch with Witness failure tracking and merge-on-CI-pass via Refinery.","status":"open","priority":2,"issue_type":"epic","owner":"andrew.bishop9625@gmail.com","created_at":"2026-05-02T18:27:46Z","created_by":"mayor","updated_at":"2026-05-02T18:27:46Z","labels":["ai-queue"],"dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"gopherstack-6wwqt","title":"DynamoDB GSI/LSI: stable order for items sharing an index key","description":"Index scans/queries sort by index key with an unstable sort and no table-key tiebreak, so tied items have no stable order and a deleted ExclusiveStartKey on a tie skips all ties. Add the table primary key as tiebreak (matches DynamoDB's deterministic paging); regression test.","status":"open","priority":3,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T20:59:53Z","created_by":"Witness Patrol","updated_at":"2026-10-13T20:59:53Z","dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"gopherstack-o8hov","title":"S3 UploadPart: malformed Content-MD5 returns BadDigest; real S3 returns InvalidDigest","description":"Bad base64 in Content-MD5 should be InvalidDigest (400); a valid-but-mismatched digest stays BadDigest. Regression test required.","status":"closed","priority":3,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T04:25:01Z","created_by":"Witness Patrol","updated_at":"2026-10-13T21:16:23Z","closed_at":"2026-10-13T21:16:23Z","close_reason":"Closed","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-cefu7.17","title":"Oracle Cloud (OCI) local emulator","description":"New OCI emulation (Object Storage, Streaming, Vault, Functions, IAM) with SDK-verified shapes.","status":"open","priority":3,"issue_type":"feature","owner":"blackbird7181@gmail.com","created_at":"2026-10-11T14:33:59Z","created_by":"Witness Patrol","updated_at":"2026-10-11T14:33:59Z","dependencies":[{"issue_id":"gopherstack-cefu7.17","depends_on_id":"gopherstack-cefu7","type":"parent-child","created_at":"2026-10-11T09:33:59Z","created_by":"Witness Patrol","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-cefu7.14","title":"Web console covering all supported clouds","description":"Extend the dashboard to browse/manage resources for AWS plus Azure/GCP/OCI once those exist.","status":"open","priority":3,"issue_type":"feature","owner":"blackbird7181@gmail.com","created_at":"2026-10-11T14:33:44Z","created_by":"Witness Patrol","updated_at":"2026-10-11T14:33:44Z","dependencies":[{"issue_id":"gopherstack-cefu7.14","depends_on_id":"gopherstack-cefu7","type":"parent-child","created_at":"2026-10-11T09:33:44Z","created_by":"Witness Patrol","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"gopherstack-cefu7.13","title":"Standalone CLI to manage local cloud emulators","description":"CLI for start/stop/status/reset/snapshot of local emulators across clouds.","status":"open","priority":3,"issue_type":"feature","owner":"blackbird7181@gmail.com","created_at":"2026-10-11T14:33:41Z","created_by":"Witness Patrol","updated_at":"2026-10-11T14:33:41Z","dependencies":[{"issue_id":"gopherstack-cefu7.13","depends_on_id":"gopherstack-cefu7","type":"parent-child","created_at":"2026-10-11T09:33:40Z","created_by":"Witness Patrol","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} diff --git a/e2e_bench_sns_test.go b/e2e_bench_sns_test.go new file mode 100644 index 0000000000..d0204056ad --- /dev/null +++ b/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) + } + }) + } +} diff --git a/pkgs/persistence/testdata/snapshot_inventory.json b/pkgs/persistence/testdata/snapshot_inventory.json index cf851b6a67..f39e6e994c 100644 --- a/pkgs/persistence/testdata/snapshot_inventory.json +++ b/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", diff --git a/services/dynamodb/item_ops.go b/services/dynamodb/item_ops.go index 13b60a5242..2470648f89 100644 --- a/services/dynamodb/item_ops.go +++ b/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 diff --git a/services/dynamodb/item_ops_batch.go b/services/dynamodb/item_ops_batch.go index 63a61ad9ab..bb9c8a6d22 100644 --- a/services/dynamodb/item_ops_batch.go +++ b/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. diff --git a/services/dynamodb/item_ops_crud.go b/services/dynamodb/item_ops_crud.go index ab9eb73b20..50d49a4338 100644 --- a/services/dynamodb/item_ops_crud.go +++ b/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] diff --git a/services/dynamodb/item_ops_query.go b/services/dynamodb/item_ops_query.go index 99437ee70d..0a5cd85034 100644 --- a/services/dynamodb/item_ops_query.go +++ b/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, diff --git a/services/dynamodb/item_ops_scan.go b/services/dynamodb/item_ops_scan.go index 86b9eb1a7d..3f3d49a6ad 100644 --- a/services/dynamodb/item_ops_scan.go +++ b/services/dynamodb/item_ops_scan.go @@ -81,7 +81,7 @@ func (db *InMemoryDB) scanCore( // A shallow slice copy is safe: writes always replace items[i] with a new map; // they never mutate an existing map in place, so our pointers remain valid. presorted := aws.ToString(input.IndexName) == "" - snap := snapshotTableForScan(table, presorted) + snap := snapshotTableForScan(table, presorted, false) // Get key schema definitions (reconstruct the table temporarily for getScanKeySchema) snapshotTable := &Table{ @@ -138,13 +138,14 @@ type scanView struct { gsiList []models.GlobalSecondaryIndex lsiList []models.LocalSecondaryIndex attrDefs []models.AttributeDefinition + delta []scanOp version uint64 sorted bool } // snapshotTableForScan captures scan state under one read lock. A valid sorted cache is // reused as-is, so a paged Scan copies nothing. -func snapshotTableForScan(table *Table, wantSorted bool) scanView { +func snapshotTableForScan(table *Table, wantSorted, forceFull bool) scanView { table.mu.RLock("Scan") defer table.mu.RUnlock() @@ -158,16 +159,12 @@ func snapshotTableForScan(table *Table, wantSorted bool) scanView { version: table.itemsVersion, } - if wantSorted { - if c := table.scanOrder.Load(); c != nil && c.version == table.itemsVersion { - snap.items, snap.sizes, snap.sorted = c.items, c.sizes, true - - return snap - } + if wantSorted && !forceFull && table.useCachedScanOrder(&snap) { + return snap + } - if len(table.itemSizes) == len(table.Items) { - snap.sizes = slices.Clone(table.itemSizes) - } + if wantSorted && len(table.itemSizes) == len(table.Items) { + snap.sizes = slices.Clone(table.itemSizes) } snap.items = slices.Clone(table.Items) @@ -175,6 +172,30 @@ func snapshotTableForScan(table *Table, wantSorted bool) scanView { return snap } +// useCachedScanOrder fills snap from the cached order, either as-is or with the +// pending write log to merge. Call under table.mu. +func (t *Table) useCachedScanOrder(snap *scanView) bool { + c := t.scanOrder.Load() + if c == nil { + return false + } + + if c.version == t.itemsVersion { + snap.items, snap.sizes, snap.sorted = c.items, c.sizes, true + + return true + } + + ops, ok := t.scanOpsSince(c) + if !ok { + return false + } + + snap.items, snap.sizes, snap.delta = c.items, c.sizes, slices.Clone(ops) + + return true +} + // scanOrderCache is the key-sorted item order of a table at one itemsVersion. type scanOrderCache struct { items []map[string]any @@ -182,25 +203,181 @@ type scanOrderCache struct { version uint64 } +// scanOp is one item-store write since the cached scan order: item is the +// stored item (or the removed one when deleted), size its byte size. +type scanOp struct { + item map[string]any + size int + deleted bool +} + +// scanLogCap bounds the write log kept for incremental scan-order rebuilds; +// past it the next Scan re-sorts from scratch. +const scanLogCap = 1024 + // itemsChanged invalidates the cached scan order; call it under table.mu.Lock -// before any mutation of t.Items. +// before any mutation of t.Items that is not logged via itemStored/itemRemoved. func (t *Table) itemsChanged() { t.itemsVersion++ + t.scanLog = nil + t.scanLogBase = t.itemsVersion +} + +// itemStored records a put/update of item for incremental scan-order upkeep. +func (t *Table) itemStored(item map[string]any, size int) { + t.logScanOp(scanOp{item: item, size: size}) +} + +// itemRemoved records a delete of item for incremental scan-order upkeep. +func (t *Table) itemRemoved(item map[string]any) { + t.logScanOp(scanOp{item: item, deleted: true}) +} + +func (t *Table) logScanOp(op scanOp) { + if t.scanOrder.Load() == nil || len(t.scanLog) >= scanLogCap { + t.itemsChanged() + + return + } + + t.itemsVersion++ + t.scanLog = append(t.scanLog, op) +} + +// scanOpsSince returns the writes applied after c was built, if still logged. +// Call under table.mu (read or write). +func (t *Table) scanOpsSince(c *scanOrderCache) ([]scanOp, bool) { + if c.version < t.scanLogBase || c.version > t.itemsVersion || + uint64(len(t.scanLog)) != t.itemsVersion-t.scanLogBase { + return nil, false + } + + ops := t.scanLog[c.version-t.scanLogBase:] + if len(ops) > len(c.items) { + return nil, false + } + + return ops, true } -// storeScanOrder sorts the snapshot into base-table key order (with sizes alongside) +// storeScanOrder returns the snapshot in base-table key order (with sizes alongside) // and caches it at the snapshot's version. The result is shared and read-only. func (t *Table) storeScanOrder( snap scanView, pkDef, skDef models.KeySchemaElement, ) ([]map[string]any, []int) { + table := &Table{AttributeDefinitions: snap.attrDefs} + + if snap.delta != nil { + if items, sizes, ok := applyScanOps(snap.items, snap.sizes, snap.delta, pkDef, skDef, table); ok { + t.scanOrder.Store(&scanOrderCache{items: items, sizes: sizes, version: snap.version}) + + return items, sizes + } + + snap = snapshotTableForScan(t, true, true) + } + items, sizes := snap.items, snap.sizes - sortScanResultsSized(items, sizes, pkDef, skDef, &Table{AttributeDefinitions: snap.attrDefs}) + sortScanResultsSized(items, sizes, pkDef, skDef, table) t.scanOrder.Store(&scanOrderCache{items: items, sizes: sizes, version: snap.version}) return items, sizes } +type scanKey struct { + pk, sk string + pkN, skN float64 +} + +type scanTouch struct { + op scanOp + entry scanSortEntry + pos int + found bool +} + +// applyScanOps merges logged writes into an already key-sorted order without +// re-sorting it: each touched key is binary-searched, then one pass copies the +// untouched runs. Returns ok=false when the order cannot be reconciled. +func applyScanOps( + items []map[string]any, + sizes []int, + ops []scanOp, + pkDef, skDef models.KeySchemaElement, + table *Table, +) ([]map[string]any, []int, bool) { + pkType := getAttributeType(table.AttributeDefinitions, pkDef.AttributeName, "S") + var skType string + if skDef.AttributeName != "" { + skType = getAttributeType(table.AttributeDefinitions, skDef.AttributeName, "S") + } + + hasSK := skDef.AttributeName != "" + haveSizes := sizes != nil && len(sizes) == len(items) + + touched := make(map[scanKey]*scanTouch, len(ops)) + + for _, op := range ops { + e := populateScanSortEntry(op.item, pkDef, skDef, pkType, skType) + k := scanKey{pk: e.pkStr, sk: e.skStr, pkN: e.pkNum, skN: e.skNum} + touched[k] = &scanTouch{op: op, entry: e} + } + + list := make([]*scanTouch, 0, len(touched)) + for _, tc := range touched { + tc.pos, tc.found = slices.BinarySearchFunc(items, tc.entry, func(it map[string]any, target scanSortEntry) int { + return compareScanSortEntries( + populateScanSortEntry(it, pkDef, skDef, pkType, skType), target, pkType, skType, hasSK, + ) + }) + list = append(list, tc) + } + + slices.SortFunc(list, func(a, b *scanTouch) int { + return compareScanSortEntries(a.entry, b.entry, pkType, skType, hasSK) + }) + + out := make([]map[string]any, 0, len(items)+len(list)) + + var outSizes []int + if haveSizes { + outSizes = make([]int, 0, len(items)+len(list)) + } + + cur := 0 + + for _, tc := range list { + if tc.pos < cur { + return nil, nil, false + } + + out = append(out, items[cur:tc.pos]...) + if haveSizes { + outSizes = append(outSizes, sizes[cur:tc.pos]...) + } + + cur = tc.pos + if tc.found { + cur++ + } + + if !tc.op.deleted { + out = append(out, tc.op.item) + if haveSizes { + outSizes = append(outSizes, tc.op.size) + } + } + } + + out = append(out, items[cur:]...) + if haveSizes { + outSizes = append(outSizes, sizes[cur:]...) + } + + return out, outSizes, true +} + // buildScanOutput enforces read throughput and assembles the scan page. func (db *InMemoryDB) buildScanOutput( ctx context.Context, @@ -363,7 +540,7 @@ func (db *InMemoryDB) scanCandidates( presorted bool, ) ([]map[string]any, []int) { if presorted && snap.ttlAttr == "" && aws.ToInt32(input.TotalSegments) <= 1 { - start := locateAfterStartKey(snap, input.ExclusiveStartKey, pkDef, skDef) + start := locateAfterStartKey(snap.items, input.ExclusiveStartKey, pkDef, skDef, snap.keySchema, snap.attrDefs) if snap.sizes != nil { return snap.items[start:], snap.sizes[start:] } @@ -390,63 +567,55 @@ func (db *InMemoryDB) scanCandidates( // Apply parallel-scan segment filter (Segment / TotalSegments). candidate = applySegmentFilter(candidate, input, pkDef) - // Apply ExclusiveStartKey: skip items up to and including the start-key item. - candidate = applyExclusiveStartKey( - candidate, - input.ExclusiveStartKey, - pkDef, - skDef, - snap.keySchema, - ) + candidate = candidate[locateAfterStartKey( + candidate, input.ExclusiveStartKey, pkDef, skDef, snap.keySchema, snap.attrDefs, + ):] return candidate, nil } -// locateAfterStartKey returns the index just past the ExclusiveStartKey item in the sorted -// snapshot (0 when absent or unmatched), using binary search with a linear fallback. +// locateAfterStartKey returns the index of the first item after ExclusiveStartKey in the +// key-sorted items. The key is a position, so a deleted key's item resumes at the next item. +// On ties (GSI/LSI) it skips past the matching item, or past all ties when none matches. func locateAfterStartKey( - snap scanView, + items []map[string]any, startKey map[string]types.AttributeValue, pkDef, skDef models.KeySchemaElement, + tableKeySchema []models.KeySchemaElement, + attrDefs []models.AttributeDefinition, ) int { if len(startKey) == 0 { return 0 } wireKey := models.FromSDKItem(startKey) - tablePKDef, tableSKDef := getPKAndSK(snap.keySchema) - pkType := getAttributeType(snap.attrDefs, pkDef.AttributeName, "S") + tablePKDef, tableSKDef := getPKAndSK(tableKeySchema) + pkType := getAttributeType(attrDefs, pkDef.AttributeName, "S") var skType string if skDef.AttributeName != "" { - skType = getAttributeType(snap.attrDefs, skDef.AttributeName, "S") + skType = getAttributeType(attrDefs, skDef.AttributeName, "S") } hasSK := skDef.AttributeName != "" target := populateScanSortEntry(wireKey, pkDef, skDef, pkType, skType) - - idx, _ := slices.BinarySearchFunc(snap.items, target, func(item map[string]any, t scanSortEntry) int { + compare := func(item map[string]any) int { return compareScanSortEntries( - populateScanSortEntry(item, pkDef, skDef, pkType, skType), - t, - pkType, - skType, - hasSK, + populateScanSortEntry(item, pkDef, skDef, pkType, skType), target, pkType, skType, hasSK, ) - }) - - matches := func(i int) bool { - return i >= 0 && i < len(snap.items) && - itemMatchesStartKeyMap(snap.items[i], wireKey, pkDef, skDef, tablePKDef, tableSKDef) } - if matches(idx) && !matches(idx-1) { - return idx + 1 - } + idx, _ := slices.BinarySearchFunc(items, target, func(item map[string]any, _ scanSortEntry) int { + return compare(item) + }) - rest := applyExclusiveStartKey(snap.items, startKey, pkDef, skDef, snap.keySchema) + for ; idx < len(items) && compare(items[idx]) == 0; idx++ { + if itemMatchesStartKeyMap(items[idx], wireKey, pkDef, skDef, tablePKDef, tableSKDef) { + return idx + 1 + } + } - return len(snap.items) - len(rest) + return idx } // scanPage iterates candidate items up to 1MB or limit, applying filter and projection. @@ -694,28 +863,6 @@ func isItemInIndex( return true } -func applyExclusiveStartKey( - candidate []map[string]any, - exclusiveStartKey map[string]types.AttributeValue, - pkDef, skDef models.KeySchemaElement, - tableKeySchema []models.KeySchemaElement, -) []map[string]any { - if len(exclusiveStartKey) == 0 { - return candidate - } - - startKey := models.FromSDKItem(exclusiveStartKey) - tablePKDef, tableSKDef := getPKAndSK(tableKeySchema) - - for i, item := range candidate { - if itemMatchesStartKeyMap(item, startKey, pkDef, skDef, tablePKDef, tableSKDef) { - return candidate[i+1:] - } - } - - return candidate -} - // maxParallelScanSegments is the upper bound on TotalSegments for parallel Scan. const maxParallelScanSegments = 1_000_000 diff --git a/services/dynamodb/query_start_key_test.go b/services/dynamodb/query_start_key_test.go index 8b525c1c07..42eb868c7b 100644 --- a/services/dynamodb/query_start_key_test.go +++ b/services/dynamodb/query_start_key_test.go @@ -40,10 +40,15 @@ func TestQuery_ExclusiveStartKeyVariants(t *testing.T) { want: []string{"o2", "o3", "o4"}, }, { - name: "unknown key restarts", + name: "absent key past end", startKey: map[string]any{"pk": map[string]any{"S": "c"}, "sk": map[string]any{"S": "zz"}}, limit: 2, - want: []string{"o0", "o1"}, + want: []string{}, + }, + { + name: "absent key in middle", + startKey: map[string]any{"pk": map[string]any{"S": "c"}, "sk": map[string]any{"S": "o15"}}, + want: []string{"o2", "o3", "o4"}, }, } diff --git a/services/dynamodb/scan_order_delta_test.go b/services/dynamodb/scan_order_delta_test.go new file mode 100644 index 0000000000..3aec9976cb --- /dev/null +++ b/services/dynamodb/scan_order_delta_test.go @@ -0,0 +1,525 @@ +package dynamodb_test + +import ( + "math/rand/v2" + "slices" + "strconv" + "sync" + "testing" + + "github.com/blackbirdworks/gopherstack/services/dynamodb" + + "github.com/aws/aws-sdk-go-v2/aws" + sdk "github.com/aws/aws-sdk-go-v2/service/dynamodb" + "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type deltaSchema struct { + name string + keyType types.ScalarAttributeType + rangeKey bool +} + +const deltaTable = "T" + +type deltaKey struct{ pk, sk int } + +func (s deltaSchema) val(n int) types.AttributeValue { + if s.keyType == types.ScalarAttributeTypeN { + return &types.AttributeValueMemberN{Value: strconv.Itoa(n - 20)} + } + + return &types.AttributeValueMemberS{Value: "k" + strconv.Itoa(1000+n)} +} + +func (s deltaSchema) key(k deltaKey) map[string]types.AttributeValue { + out := map[string]types.AttributeValue{"pk": s.val(k.pk)} + if s.rangeKey { + out["sk"] = s.val(k.sk) + } + + return out +} + +func (s deltaSchema) item(k deltaKey, v string) map[string]types.AttributeValue { + out := s.key(k) + out["v"] = &types.AttributeValueMemberS{Value: v} + + return out +} + +func (s deltaSchema) create(t *testing.T, db *dynamodb.InMemoryDB, gsi bool) { + t.Helper() + + ks := []types.KeySchemaElement{{AttributeName: aws.String("pk"), KeyType: types.KeyTypeHash}} + ad := []types.AttributeDefinition{{AttributeName: aws.String("pk"), AttributeType: s.keyType}} + + if s.rangeKey { + ks = append(ks, types.KeySchemaElement{AttributeName: aws.String("sk"), KeyType: types.KeyTypeRange}) + ad = append(ad, types.AttributeDefinition{AttributeName: aws.String("sk"), AttributeType: s.keyType}) + } + + in := &sdk.CreateTableInput{ + TableName: aws.String(deltaTable), KeySchema: ks, AttributeDefinitions: ad, + BillingMode: types.BillingModePayPerRequest, + } + + if gsi { + in.AttributeDefinitions = append(in.AttributeDefinitions, + types.AttributeDefinition{AttributeName: aws.String("v"), AttributeType: types.ScalarAttributeTypeS}) + in.GlobalSecondaryIndexes = []types.GlobalSecondaryIndex{{ + IndexName: aws.String("byv"), + KeySchema: []types.KeySchemaElement{{AttributeName: aws.String("v"), KeyType: types.KeyTypeHash}}, + Projection: &types.Projection{ProjectionType: types.ProjectionTypeAll}, + }} + } + + _, err := db.CreateTable(t.Context(), in) + require.NoError(t, err) +} + +func (s deltaSchema) less(a, b deltaKey) int { + if a.pk != b.pk { + return a.pk - b.pk + } + + if s.rangeKey { + return a.sk - b.sk + } + + return 0 +} + +func (s deltaSchema) write(t *testing.T, db *dynamodb.InMemoryDB, op string, k deltaKey, v string) { + t.Helper() + + var err error + + ctx := t.Context() + + switch op { + case "put": + _, err = db.PutItem(ctx, &sdk.PutItemInput{TableName: aws.String(deltaTable), Item: s.item(k, v)}) + case "delete": + _, err = db.DeleteItem(ctx, &sdk.DeleteItemInput{TableName: aws.String(deltaTable), Key: s.key(k)}) + case "update": + _, err = db.UpdateItem(ctx, &sdk.UpdateItemInput{ + TableName: aws.String(deltaTable), Key: s.key(k), UpdateExpression: aws.String("SET v = :v"), + ExpressionAttributeValues: map[string]types.AttributeValue{":v": &types.AttributeValueMemberS{Value: v}}, + }) + case "batchput": + _, err = db.BatchWriteItem(ctx, &sdk.BatchWriteItemInput{RequestItems: map[string][]types.WriteRequest{ + deltaTable: {{PutRequest: &types.PutRequest{Item: s.item(k, v)}}}, + }}) + case "batchdelete": + _, err = db.BatchWriteItem(ctx, &sdk.BatchWriteItemInput{RequestItems: map[string][]types.WriteRequest{ + deltaTable: {{DeleteRequest: &types.DeleteRequest{Key: s.key(k)}}}, + }}) + case "txput": + _, err = db.TransactWriteItems(ctx, &sdk.TransactWriteItemsInput{TransactItems: []types.TransactWriteItem{ + {Put: &types.Put{TableName: aws.String(deltaTable), Item: s.item(k, v)}}, + }}) + case "txdelete": + _, err = db.TransactWriteItems(ctx, &sdk.TransactWriteItemsInput{TransactItems: []types.TransactWriteItem{ + {Delete: &types.Delete{TableName: aws.String(deltaTable), Key: s.key(k)}}, + }}) + } + + require.NoError(t, err) +} + +type deltaModel map[deltaKey]string + +func (m deltaModel) apply(op string, k deltaKey, v string) { + switch op { + case "delete", "batchdelete", "txdelete": + delete(m, k) + default: + m[k] = v + } +} + +func (s deltaSchema) want(m deltaModel) []string { + keys := make([]deltaKey, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + + slices.SortFunc(keys, s.less) + + out := make([]string, 0, len(keys)) + for _, k := range keys { + out = append(out, strconv.Itoa(k.pk)+"/"+strconv.Itoa(k.sk)+"="+m[k]) + } + + return out +} + +func (s deltaSchema) decode(item map[string]types.AttributeValue) string { + num := func(av types.AttributeValue) int { + if n, ok := av.(*types.AttributeValueMemberN); ok { + i, _ := strconv.Atoi(n.Value) + + return i + 20 + } + + i, _ := strconv.Atoi(av.(*types.AttributeValueMemberS).Value[1:]) + + return i - 1000 + } + + sk := 0 + if s.rangeKey { + sk = num(item["sk"]) + } + + return strconv.Itoa(num(item["pk"])) + "/" + strconv.Itoa(sk) + "=" + item["v"].(*types.AttributeValueMemberS).Value +} + +func (s deltaSchema) scanPaged(t *testing.T, db *dynamodb.InMemoryDB, limit int32) []string { + t.Helper() + + got := []string{} + + var start map[string]types.AttributeValue + + for range 5000 { + out, err := db.Scan(t.Context(), &sdk.ScanInput{ + TableName: aws.String(deltaTable), Limit: aws.Int32(limit), ExclusiveStartKey: start, + }) + require.NoError(t, err) + + for _, it := range out.Items { + got = append(got, s.decode(it)) + } + + if len(out.LastEvaluatedKey) == 0 { + return got + } + + start = out.LastEvaluatedKey + } + + t.Fatal("scan did not terminate") + + return nil +} + +func deltaSchemas() []deltaSchema { + return []deltaSchema{ + {name: "hash_s", keyType: types.ScalarAttributeTypeS}, + {name: "hash_n", keyType: types.ScalarAttributeTypeN}, + {name: "range_s", keyType: types.ScalarAttributeTypeS, rangeKey: true}, + {name: "range_n", keyType: types.ScalarAttributeTypeN, rangeKey: true}, + } +} + +func deltaOps() []string { + return []string{"put", "delete", "update", "batchput", "batchdelete", "txput", "txdelete"} +} + +func (s deltaSchema) randomKey(rng *rand.Rand) deltaKey { + k := deltaKey{pk: rng.IntN(12)} + if s.rangeKey { + k.sk = rng.IntN(6) + } + + return k +} + +func TestScanOrderDelta_WritesBetweenScans(t *testing.T) { + t.Parallel() + + for _, s := range deltaSchemas() { + t.Run(s.name, func(t *testing.T) { + t.Parallel() + + db := dynamodb.NewInMemoryDB() + s.create(t, db, false) + + rng := rand.New(rand.NewPCG(7, 9)) + model := deltaModel{} + + for round := range 60 { + batch := 1 + rng.IntN(8) + for range batch { + op := deltaOps()[rng.IntN(len(deltaOps()))] + k := s.randomKey(rng) + v := strconv.Itoa(rng.IntN(1000)) + + if op == "update" { + if _, ok := model[k]; !ok { + op = "put" + } + } + + s.write(t, db, op, k, v) + model.apply(op, k, v) + } + + want := s.want(model) + for _, limit := range []int32{1, 4, 1000} { + require.Equal(t, want, s.scanPaged(t, db, limit), "round %d limit %d", round, limit) + } + } + }) + } +} + +func TestScanOrderDelta_LogOverflow(t *testing.T) { + t.Parallel() + + for _, s := range deltaSchemas() { + t.Run(s.name, func(t *testing.T) { + t.Parallel() + + db := dynamodb.NewInMemoryDB() + s.create(t, db, false) + + model := deltaModel{} + s.write(t, db, "put", deltaKey{}, "seed") + model.apply("put", deltaKey{}, "seed") + require.Equal(t, s.want(model), s.scanPaged(t, db, 100)) + + for i := range 1500 { + k := deltaKey{pk: i % 53} + if s.rangeKey { + k.sk = i % 7 + } + + v := strconv.Itoa(i) + s.write(t, db, "put", k, v) + model.apply("put", k, v) + } + + assert.Equal(t, s.want(model), s.scanPaged(t, db, 100)) + }) + } +} + +func TestScanOrderDelta_PaginationAcrossWrites(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + op string + rest []string + key int + }{ + {name: "insert_after_cursor", op: "put", key: 5, rest: []string{"5/0=y", "7/0=x", "9/0=x"}}, + {name: "insert_before_cursor", op: "put", key: 0, rest: []string{"7/0=x", "9/0=x"}}, + {name: "delete_after_cursor", op: "delete", key: 7, rest: []string{"9/0=x"}}, + {name: "delete_before_cursor", op: "delete", key: 1, rest: []string{"7/0=x", "9/0=x"}}, + {name: "overwrite_after_cursor", op: "put", key: 7, rest: []string{"7/0=y", "9/0=x"}}, + } + + s := deltaSchema{name: "hash_n", keyType: types.ScalarAttributeTypeN} + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + db := dynamodb.NewInMemoryDB() + s.create(t, db, false) + + for _, n := range []int{1, 2, 3, 7, 9} { + s.write(t, db, "put", deltaKey{pk: n}, "x") + } + + first, err := db.Scan(t.Context(), &sdk.ScanInput{TableName: aws.String(deltaTable), Limit: aws.Int32(3)}) + require.NoError(t, err) + require.Len(t, first.Items, 3) + require.NotEmpty(t, first.LastEvaluatedKey) + + s.write(t, db, tt.op, deltaKey{pk: tt.key}, "y") + + rest, err := db.Scan(t.Context(), &sdk.ScanInput{ + TableName: aws.String(deltaTable), ExclusiveStartKey: first.LastEvaluatedKey, + }) + require.NoError(t, err) + + got := make([]string, 0, len(rest.Items)) + for _, it := range rest.Items { + got = append(got, s.decode(it)) + } + + assert.Equal(t, tt.rest, got) + }) + } +} + +func TestScanOrderDelta_GSIScanAfterWrites(t *testing.T) { + t.Parallel() + + s := deltaSchema{name: "hash_s", keyType: types.ScalarAttributeTypeS} + db := dynamodb.NewInMemoryDB() + s.create(t, db, true) + + scanIdx := func() []string { + out, err := db.Scan(t.Context(), &sdk.ScanInput{ + TableName: aws.String(deltaTable), IndexName: aws.String("byv"), + }) + require.NoError(t, err) + + got := make([]string, 0, len(out.Items)) + for _, it := range out.Items { + got = append(got, s.decode(it)) + } + + slices.Sort(got) + + return got + } + + model := deltaModel{} + steps := []struct { + op string + v string + k int + }{ + {"put", "a", 3}, {"put", "b", 1}, {"put", "c", 2}, {"update", "z", 1}, {"delete", "", 3}, + {"batchput", "d", 4}, {"txput", "e", 5}, {"batchdelete", "", 2}, {"txdelete", "", 5}, {"put", "q", 4}, + } + + for i, st := range steps { + k := deltaKey{pk: st.k} + s.write(t, db, st.op, k, st.v) + model.apply(st.op, k, st.v) + + want := s.want(model) + slices.Sort(want) + assert.Equal(t, want, scanIdx(), "step %d", i) + assert.Equal(t, s.want(model), s.scanPaged(t, db, 2), "base step %d", i) + } +} + +func TestScanOrderDelta_SegmentedScansAfterWrites(t *testing.T) { + t.Parallel() + + for _, s := range deltaSchemas() { + t.Run(s.name, func(t *testing.T) { + t.Parallel() + + db := dynamodb.NewInMemoryDB() + s.create(t, db, false) + + rng := rand.New(rand.NewPCG(11, 13)) + model := deltaModel{} + + for round := range 25 { + for range 1 + rng.IntN(6) { + op := deltaOps()[rng.IntN(len(deltaOps()))] + k := s.randomKey(rng) + v := strconv.Itoa(rng.IntN(1000)) + + if op == "update" { + if _, ok := model[k]; !ok { + op = "put" + } + } + + s.write(t, db, op, k, v) + model.apply(op, k, v) + } + + var union []string + + for seg := range int32(3) { + out, err := db.Scan(t.Context(), &sdk.ScanInput{ + TableName: aws.String(deltaTable), Segment: aws.Int32(seg), TotalSegments: aws.Int32(3), + }) + require.NoError(t, err) + + for _, it := range out.Items { + union = append(union, s.decode(it)) + } + } + + want := s.want(model) + slices.Sort(want) + slices.Sort(union) + require.Equal(t, want, union, "round %d", round) + } + }) + } +} + +func TestScanOrderDelta_ConcurrentScansAndWrites(t *testing.T) { + t.Parallel() + + for _, s := range deltaSchemas() { + t.Run(s.name, func(t *testing.T) { + t.Parallel() + + db := dynamodb.NewInMemoryDB() + s.create(t, db, false) + + var wg sync.WaitGroup + + for w := range 4 { + wg.Go(func() { + for i := range 150 { + k := deltaKey{pk: w*100 + i%50} + if s.rangeKey { + k.sk = i % 3 + } + + op := deltaOps()[(i+w)%len(deltaOps())] + if op == "update" { + op = "put" + } + + s.write(t, db, op, k, strconv.Itoa(i)) + } + }) + } + + for range 3 { + wg.Go(func() { + for range 100 { + out, err := db.Scan(t.Context(), &sdk.ScanInput{TableName: aws.String(deltaTable)}) + if err != nil { + t.Errorf("scan: %v", err) + + return + } + + got := make([]string, 0, len(out.Items)) + for _, it := range out.Items { + got = append(got, s.decode(it)) + } + + assert.True(t, slices.IsSortedFunc(got, compareDecoded)) + } + }) + } + + wg.Wait() + + final := s.scanPaged(t, db, 1000) + assert.True(t, slices.IsSortedFunc(final, compareDecoded)) + }) + } +} + +func compareDecoded(a, b string) int { + ak, bk := decodedKey(a), decodedKey(b) + if ak[0] != bk[0] { + return ak[0] - bk[0] + } + + return ak[1] - bk[1] +} + +func decodedKey(s string) [2]int { + var pk, sk int + + slash := slices.Index([]byte(s), '/') + eq := slices.Index([]byte(s), '=') + pk, _ = strconv.Atoi(s[:slash]) + sk, _ = strconv.Atoi(s[slash+1 : eq]) + + return [2]int{pk, sk} +} diff --git a/services/dynamodb/scan_paging_internal_test.go b/services/dynamodb/scan_paging_internal_test.go index b9b4b4133a..b1209f15f9 100644 --- a/services/dynamodb/scan_paging_internal_test.go +++ b/services/dynamodb/scan_paging_internal_test.go @@ -167,38 +167,6 @@ func TestScanPaging_FastPathMatchesLinearPath(t *testing.T) { } } -func TestScanPaging_StartKeyOfDeletedItemRestartsLikeLinearPath(t *testing.T) { - t.Parallel() - - schema := pagingSchema{name: "s", keyType: types.ScalarAttributeTypeS} - db := NewInMemoryDB() - db.SetDefaultRegion("us-east-1") - newPagingTable(t, db, "plain", schema, false) - newPagingTable(t, db, "twin", schema, true) - - for _, tbl := range []string{"plain", "twin"} { - for i := range 6 { - _, err := db.PutItem(t.Context(), &sdk.PutItemInput{ - TableName: aws.String(tbl), - Item: map[string]types.AttributeValue{ - "pk": &types.AttributeValueMemberS{Value: "k" + strconv.Itoa(i)}, - }, - }) - require.NoError(t, err) - } - } - - ghost := map[string]types.AttributeValue{"pk": &types.AttributeValueMemberS{Value: "k25"}} - want, err := db.Scan(t.Context(), &sdk.ScanInput{TableName: aws.String("twin"), ExclusiveStartKey: ghost}) - require.NoError(t, err) - - got, err := db.Scan(t.Context(), &sdk.ScanInput{TableName: aws.String("plain"), ExclusiveStartKey: ghost}) - require.NoError(t, err) - - assert.Equal(t, want.Items, got.Items) - assert.Equal(t, want.Count, got.Count) -} - func TestScanOrderCache_SizesMatchItems(t *testing.T) { t.Parallel() diff --git a/services/dynamodb/scan_write_interleaved_bench_test.go b/services/dynamodb/scan_write_interleaved_bench_test.go new file mode 100644 index 0000000000..3b2a61f947 --- /dev/null +++ b/services/dynamodb/scan_write_interleaved_bench_test.go @@ -0,0 +1,54 @@ +package dynamodb_test + +import ( + "strconv" + "testing" + + "github.com/blackbirdworks/gopherstack/services/dynamodb" + + "github.com/aws/aws-sdk-go-v2/aws" + sdk "github.com/aws/aws-sdk-go-v2/service/dynamodb" + "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "github.com/stretchr/testify/require" +) + +func BenchmarkScanAfterWrite(b *testing.B) { + for _, size := range []int{1000, 10000} { + b.Run(strconv.Itoa(size), func(b *testing.B) { + db := dynamodb.NewInMemoryDB() + ctx := b.Context() + _, err := db.CreateTable(ctx, &sdk.CreateTableInput{ + TableName: aws.String("T"), + KeySchema: []types.KeySchemaElement{ + {AttributeName: aws.String("pk"), KeyType: types.KeyTypeHash}, + }, + AttributeDefinitions: []types.AttributeDefinition{ + {AttributeName: aws.String("pk"), AttributeType: types.ScalarAttributeTypeS}, + }, + }) + require.NoError(b, err) + + for i := range size { + _, err = db.PutItem(ctx, &sdk.PutItemInput{ + TableName: aws.String("T"), + Item: scanCacheItem("k"+strconv.Itoa(i*7919%size), "v"), + }) + require.NoError(b, err) + } + + b.ReportAllocs() + b.ResetTimer() + + for i := 0; b.Loop(); i++ { + _, err = db.PutItem(ctx, &sdk.PutItemInput{ + TableName: aws.String("T"), + Item: scanCacheItem("k"+strconv.Itoa(i%size), strconv.Itoa(i)), + }) + require.NoError(b, err) + + _, err = db.Scan(ctx, &sdk.ScanInput{TableName: aws.String("T"), Limit: aws.Int32(10)}) + require.NoError(b, err) + } + }) + } +} diff --git a/services/dynamodb/start_key_position_internal_test.go b/services/dynamodb/start_key_position_internal_test.go new file mode 100644 index 0000000000..557536d0a3 --- /dev/null +++ b/services/dynamodb/start_key_position_internal_test.go @@ -0,0 +1,306 @@ +package dynamodb + +import ( + "fmt" + "slices" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + sdk "github.com/aws/aws-sdk-go-v2/service/dynamodb" + "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func skey(v string) types.AttributeValue { return &types.AttributeValueMemberS{Value: v} } + +func bk(pk, sk string) map[string]types.AttributeValue { + return map[string]types.AttributeValue{"pk": skey(pk), "sk": skey(sk)} +} + +func gk(gs, sk string) map[string]types.AttributeValue { + return map[string]types.AttributeValue{"gk": skey("ga"), "gs": skey(gs), "pk": skey("a"), "sk": skey(sk)} +} + +func startKeyTable(t *testing.T, ttl bool) *InMemoryDB { + t.Helper() + + db := NewInMemoryDB() + db.SetDefaultRegion("us-east-1") + + str := func(n string) types.AttributeDefinition { + return types.AttributeDefinition{AttributeName: aws.String(n), AttributeType: types.ScalarAttributeTypeS} + } + + _, err := db.CreateTable(t.Context(), &sdk.CreateTableInput{ + TableName: aws.String("t"), + KeySchema: []types.KeySchemaElement{ + {AttributeName: aws.String("pk"), KeyType: types.KeyTypeHash}, + {AttributeName: aws.String("sk"), KeyType: types.KeyTypeRange}, + }, + AttributeDefinitions: []types.AttributeDefinition{str("pk"), str("sk"), str("gk"), str("gs")}, + BillingMode: types.BillingModePayPerRequest, + GlobalSecondaryIndexes: []types.GlobalSecondaryIndex{{ + IndexName: aws.String("gsi"), + KeySchema: []types.KeySchemaElement{ + {AttributeName: aws.String("gk"), KeyType: types.KeyTypeHash}, + {AttributeName: aws.String("gs"), KeyType: types.KeyTypeRange}, + }, + Projection: &types.Projection{ProjectionType: types.ProjectionTypeAll}, + }}, + }) + require.NoError(t, err) + + if ttl { + _, err = db.UpdateTimeToLive(t.Context(), &sdk.UpdateTimeToLiveInput{ + TableName: aws.String("t"), + TimeToLiveSpecification: &types.TimeToLiveSpecification{ + AttributeName: aws.String("ttl"), Enabled: aws.Bool(true), + }, + }) + require.NoError(t, err) + } + + for _, id := range []string{"a1", "a3", "a5", "b1", "b3"} { + _, err = db.PutItem(t.Context(), &sdk.PutItemInput{ + TableName: aws.String("t"), + Item: map[string]types.AttributeValue{ + "pk": skey(id[:1]), "sk": skey("s" + id[1:]), "gk": skey("g" + id[:1]), "gs": skey("x" + id[1:]), + }, + }) + require.NoError(t, err) + } + + return db +} + +func ids(items []map[string]types.AttributeValue) []string { + out := make([]string, 0, len(items)) + for _, it := range items { + out = append(out, it["pk"].(*types.AttributeValueMemberS).Value+it["sk"].(*types.AttributeValueMemberS).Value) + } + + return out +} + +func TestExclusiveStartKey_IsPositionNotItemReference(t *testing.T) { + t.Parallel() + + all := []string{"as1", "as3", "as5", "bs1", "bs3"} + tests := []struct { + start map[string]types.AttributeValue + name string + want []string + }{ + {bk("a", "s2"), "deleted middle", all[1:]}, + {bk("a", "s9"), "deleted end of partition", all[3:]}, + {bk("c", "s1"), "deleted at end", nil}, + {bk("0", "s0"), "deleted before first", all}, + {bk("a", "s3"), "existing", all[2:]}, + } + + for _, ttl := range []bool{false, true} { + for _, tt := range tests { + t.Run(fmt.Sprintf("scan ttl=%v %s", ttl, tt.name), func(t *testing.T) { + t.Parallel() + + db := startKeyTable(t, ttl) + out, err := db.Scan(t.Context(), &sdk.ScanInput{ + TableName: aws.String("t"), ExclusiveStartKey: tt.start, + }) + require.NoError(t, err) + assert.Len(t, out.Items, len(tt.want)) + if len(tt.want) > 0 { + assert.Equal(t, tt.want, ids(out.Items)) + } + }) + } + } +} + +func TestExclusiveStartKey_HashOnlyScan(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + start string + want []string + }{ + {"deleted middle", "k4", []string{"k5", "k7"}}, + {"deleted at end", "k9", nil}, + {"deleted before first", "k0", []string{"k1", "k3", "k5", "k7"}}, + {"existing", "k3", []string{"k5", "k7"}}, + } + + for _, ttl := range []bool{false, true} { + for _, tt := range tests { + t.Run(fmt.Sprintf("ttl=%v %s", ttl, tt.name), func(t *testing.T) { + t.Parallel() + + db := NewInMemoryDB() + db.SetDefaultRegion("us-east-1") + newPagingTable(t, db, "h", pagingSchema{name: "h", keyType: types.ScalarAttributeTypeS}, ttl) + + for _, k := range []string{"k1", "k3", "k5", "k7"} { + _, err := db.PutItem(t.Context(), &sdk.PutItemInput{ + TableName: aws.String("h"), Item: map[string]types.AttributeValue{"pk": skey(k)}, + }) + require.NoError(t, err) + } + + out, err := db.Scan(t.Context(), &sdk.ScanInput{ + TableName: aws.String("h"), + ExclusiveStartKey: map[string]types.AttributeValue{"pk": skey(tt.start)}, + }) + require.NoError(t, err) + + got := make([]string, 0, len(out.Items)) + for _, it := range out.Items { + got = append(got, it["pk"].(*types.AttributeValueMemberS).Value) + } + + assert.Equal(t, tt.want, append([]string(nil), got...)) + assert.Len(t, got, len(tt.want)) + }) + } + } +} + +func TestExclusiveStartKey_SegmentedScan(t *testing.T) { + t.Parallel() + + const segments = 3 + + db := startKeyTable(t, false) + start := bk("a", "s2") + + got := make([]string, 0, 5) + want := make([]string, 0, 5) + + for seg := range int32(segments) { + full, err := db.Scan(t.Context(), &sdk.ScanInput{ + TableName: aws.String("t"), Segment: aws.Int32(seg), TotalSegments: aws.Int32(segments), + }) + require.NoError(t, err) + + out, err := db.Scan(t.Context(), &sdk.ScanInput{ + TableName: aws.String("t"), Segment: aws.Int32(seg), TotalSegments: aws.Int32(segments), + ExclusiveStartKey: start, + }) + require.NoError(t, err) + + for _, id := range ids(full.Items) { + if id > "as2" { + want = append(want, id) + } + } + + got = append(got, ids(out.Items)...) + } + + slices.Sort(got) + slices.Sort(want) + assert.Equal(t, []string{"as3", "as5", "bs1", "bs3"}, got) + assert.Equal(t, want, got) +} + +func TestExclusiveStartKey_GSIScanAndQuery(t *testing.T) { + t.Parallel() + + all := []string{"as1", "as3", "as5", "bs1", "bs3"} + pastEnd := map[string]types.AttributeValue{"gk": skey("gz"), "gs": skey("x9"), "pk": skey("z"), "sk": skey("s9")} + tests := []struct { + start map[string]types.AttributeValue + name string + want []string + query bool + reverse bool + }{ + {gk("x2", "s2"), "scan deleted middle", all[1:], false, false}, + {pastEnd, "scan deleted end", nil, false, false}, + {gk("x0", "s0"), "scan deleted before first", all, false, false}, + {gk("x2", "s2"), "query deleted middle", []string{"as3", "as5"}, true, false}, + {gk("x9", "s9"), "query deleted end", nil, true, false}, + {gk("x0", "s0"), "query deleted before first", all[:3], true, false}, + {gk("x4", "s4"), "reverse deleted middle", []string{"as3", "as1"}, true, true}, + {gk("x0", "s0"), "reverse deleted end", nil, true, true}, + {gk("x9", "s9"), "reverse before first", []string{"as5", "as3", "as1"}, true, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + db := startKeyTable(t, false) + + var items []map[string]types.AttributeValue + + if tt.query { + out, err := db.Query(t.Context(), &sdk.QueryInput{ + TableName: aws.String("t"), IndexName: aws.String("gsi"), + KeyConditionExpression: aws.String("gk = :g"), + ExpressionAttributeValues: map[string]types.AttributeValue{":g": skey("ga")}, + ScanIndexForward: aws.Bool(!tt.reverse), + ExclusiveStartKey: tt.start, + }) + require.NoError(t, err) + + items = out.Items + } else { + out, err := db.Scan(t.Context(), &sdk.ScanInput{ + TableName: aws.String("t"), IndexName: aws.String("gsi"), ExclusiveStartKey: tt.start, + }) + require.NoError(t, err) + + items = out.Items + } + + got := ids(items) + assert.Len(t, got, len(tt.want)) + if len(tt.want) > 0 { + assert.Equal(t, tt.want, got) + } + }) + } +} + +func TestExclusiveStartKey_BaseTableQuery(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + sk string + want []string + reverse bool + }{ + {"deleted middle", "s2", []string{"as3", "as5"}, false}, + {"deleted end", "s9", nil, false}, + {"deleted before first", "s0", []string{"as1", "as3", "as5"}, false}, + {"reverse deleted middle", "s4", []string{"as3", "as1"}, true}, + {"reverse deleted end", "s0", nil, true}, + {"reverse before first", "s9", []string{"as5", "as3", "as1"}, true}, + {"existing", "s3", []string{"as5"}, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + db := startKeyTable(t, false) + out, err := db.Query(t.Context(), &sdk.QueryInput{ + TableName: aws.String("t"), + KeyConditionExpression: aws.String("pk = :p"), + ExpressionAttributeValues: map[string]types.AttributeValue{":p": skey("a")}, + ScanIndexForward: aws.Bool(!tt.reverse), + ExclusiveStartKey: map[string]types.AttributeValue{"pk": skey("a"), "sk": skey(tt.sk)}, + }) + require.NoError(t, err) + + got := ids(out.Items) + assert.Len(t, got, len(tt.want)) + if len(tt.want) > 0 { + assert.Equal(t, tt.want, got) + } + }) + } +} diff --git a/services/dynamodb/store.go b/services/dynamodb/store.go index 8ae38f3a12..71cd2ed8cf 100644 --- a/services/dynamodb/store.go +++ b/services/dynamodb/store.go @@ -312,6 +312,7 @@ type Table struct { mu *lockmetrics.RWMutex activateTimer *time.Timer scanOrder atomic.Pointer[scanOrderCache] + scanLog []scanOp Tags *tags.Tags `json:"Tags,omitempty"` AutoScaling *autoScalingSettings `json:"AutoScaling,omitempty"` // ReplicaAutoScaling holds per-replica read-capacity autoscaling settings, @@ -358,6 +359,7 @@ type Table struct { StreamShards []StreamShard `json:"StreamShards,omitempty"` ProvisionedThroughput models.ProvisionedThroughputDescription `json:"ProvisionedThroughput"` itemsVersion uint64 + scanLogBase uint64 totalItemSizeBytes int64 streamSeq int64 StreamHead int `json:"StreamHead,omitempty"` @@ -701,6 +703,12 @@ func (t *Table) initializeIndexes() { // rebuildIndexes rebuilds all indexes from existing items (used after table creation or batch updates). func (t *Table) rebuildIndexes() { t.itemsChanged() + t.rebuildDerivedState() +} + +// rebuildDerivedState recomputes key indexes and item sizes without touching the +// scan-order log; callers must have logged or invalidated their Items changes. +func (t *Table) rebuildDerivedState() { t.initializeIndexes() // Rebuild the item-size accounting alongside the key indexes. itemSizes has diff --git a/services/s3/bucket_ops.go b/services/s3/bucket_ops.go index 46de21148c..63cc57588d 100644 --- a/services/s3/bucket_ops.go +++ b/services/s3/bucket_ops.go @@ -45,6 +45,14 @@ func (h *S3Handler) handleBucketOperation( r *http.Request, bucket string, ) { + if r.Method == http.MethodPut || r.Method == http.MethodPost { + if err := verifyBodyContentMD5(r); err != nil { + WriteError(ctx, w, r, err) + + return + } + } + switch r.Method { case http.MethodPut: h.routeBucketPut(ctx, w, r, bucket) diff --git a/services/s3/errors.go b/services/s3/errors.go index 5f0933a22f..b05b9b8bd6 100644 --- a/services/s3/errors.go +++ b/services/s3/errors.go @@ -78,6 +78,7 @@ var ( ErrNoMetadataTableConfig = errors.New(errNoSuchConfig) ErrNoSuchTagSet = errors.New("NoSuchTagSet") ErrBadChecksum = errors.New("BadDigest") + ErrInvalidDigest = errors.New("InvalidDigest") ErrDeleteMarker = errors.New("DeleteMarker") ErrLatestDeleteMarker = errors.New("LatestDeleteMarker") ErrTooManyTags = errors.New("BadRequest") @@ -411,6 +412,11 @@ func coreErrorTableObject() []s3ErrorEntry { func configErrorTable() []s3ErrorEntry { return []s3ErrorEntry{ + {ErrInvalidDigest, s3ErrorInfo{ + "InvalidDigest", + "The Content-MD5 you specified was invalid.", + http.StatusBadRequest, + }}, {ErrNoBucketPolicy, s3ErrorInfo{ "NoSuchBucketPolicy", "The bucket policy does not exist", diff --git a/services/s3/multipart.go b/services/s3/multipart.go index 1db69cc0a1..1228b41a20 100644 --- a/services/s3/multipart.go +++ b/services/s3/multipart.go @@ -7,7 +7,6 @@ import ( "crypto/sha1" //nolint:gosec // SHA1 required for S3 checksum compatibility "crypto/sha256" "crypto/sha512" - "encoding/base64" "encoding/hex" "encoding/xml" "fmt" @@ -196,16 +195,13 @@ func (b *InMemoryBackend) UploadPart( etag := hex.EncodeToString(md5Hasher.Sum(nil)) // 2. Validate Content-MD5 from context if present. - if md5Header, ok := ctx.Value(md5Key).(string); ok && md5Header != "" { - decoded, dErr := base64.StdEncoding.DecodeString(md5Header) - if dErr != nil || len(decoded) != md5.Size { - return nil, ErrBadChecksum - } + decoded, dErr := contentMD5FromContext(ctx) + if dErr != nil { + return nil, dErr + } - computed := md5Hasher.Sum(nil) - if !bytes.Equal(computed, decoded) { - return nil, ErrBadChecksum - } + if decoded != nil && !bytes.Equal(md5Hasher.Sum(nil), decoded) { + return nil, ErrBadChecksum } if vErr := b.verifyChecksum(input, s3Hasher, algo); vErr != nil { diff --git a/services/s3/multipart_ops.go b/services/s3/multipart_ops.go index 097b0d61ab..a7b6598efb 100644 --- a/services/s3/multipart_ops.go +++ b/services/s3/multipart_ops.go @@ -156,8 +156,11 @@ func (h *S3Handler) uploadPart( // the real bytes, not the chunk-signature envelope. r = maybeDecodeChunkedBody(r) - if md5Header := r.Header.Get("Content-MD5"); md5Header != "" { - ctx = context.WithValue(ctx, md5Key, md5Header) + ctx, md5Err := withContentMD5(ctx, r) + if md5Err != nil { + WriteError(ctx, w, r, md5Err) + + return } algo, crc32p, crc32cp, sha1p, sha256p := extractAlgoAndChecksums(r) diff --git a/services/s3/object_ops_put.go b/services/s3/object_ops_put.go index b2cbd79cf3..9df12f3213 100644 --- a/services/s3/object_ops_put.go +++ b/services/s3/object_ops_put.go @@ -38,6 +38,15 @@ func (h *S3Handler) setPutObjectResponseHeaders(w http.ResponseWriter, ver *s3.P } } +// validatePutObjectHeaders rejects invalid tag sets (>10 tags, over-long key/value) and oversized metadata. +func validatePutObjectHeaders(r *http.Request) error { + if err := validateTaggingHeader(r.Header.Get("X-Amz-Tagging")); err != nil { + return err + } + + return validateUserMetadataSize(r.Header) +} + func (h *S3Handler) putObject( ctx context.Context, w http.ResponseWriter, @@ -62,14 +71,7 @@ func (h *S3Handler) putObject( // ETag) is the real object bytes, not the chunk-signature envelope. r = maybeDecodeChunkedBody(r) - // Reject invalid tag sets (>10 tags, over-long key/value) before writing. - if err := validateTaggingHeader(r.Header.Get("X-Amz-Tagging")); err != nil { - WriteError(ctx, w, r, err) - - return - } - - if err := validateUserMetadataSize(r.Header); err != nil { + if err := validatePutObjectHeaders(r); err != nil { WriteError(ctx, w, r, err) return @@ -97,8 +99,11 @@ func (h *S3Handler) putObject( logger.Load(ctx).DebugContext(ctx, "S3 putObject input", "bucket", bucketName, "key", key, "contentType", r.Header.Get("Content-Type")) - if md5Header := r.Header.Get("Content-MD5"); md5Header != "" { - ctx = context.WithValue(ctx, md5Key, md5Header) + ctx, md5Err := withContentMD5(ctx, r) + if md5Err != nil { + WriteError(ctx, w, r, md5Err) + + return } algo, crc32p, crc32cp, sha1p, sha256p := extractAlgoAndChecksums(r) diff --git a/services/s3/objects.go b/services/s3/objects.go index 984c34bf89..9ce8a78373 100644 --- a/services/s3/objects.go +++ b/services/s3/objects.go @@ -3,7 +3,6 @@ package s3 import ( "bytes" "context" - "crypto/md5" //nolint:gosec // MD5 required for S3 ETag compatibility "encoding/base64" "encoding/binary" "encoding/hex" @@ -1292,15 +1291,13 @@ func (b *InMemoryBackend) computeObjectHashes( // validateContentMD5 validates the Content-MD5 header from context against the computed etag. func (b *InMemoryBackend) validateContentMD5(ctx context.Context, _ []byte, etag string) error { - if md5Header, ok := ctx.Value(md5Key).(string); ok && md5Header != "" { - decoded, dErr := base64.StdEncoding.DecodeString(md5Header) - if dErr != nil || len(decoded) != md5.Size { - return ErrBadChecksum - } + decoded, dErr := contentMD5FromContext(ctx) + if dErr != nil { + return dErr + } - if hex.EncodeToString(decoded) != etag { - return ErrBadChecksum - } + if decoded != nil && hex.EncodeToString(decoded) != etag { + return ErrBadChecksum } return nil diff --git a/services/s3/upload_part_digest_test.go b/services/s3/upload_part_digest_test.go new file mode 100644 index 0000000000..42db9fcfa8 --- /dev/null +++ b/services/s3/upload_part_digest_test.go @@ -0,0 +1,133 @@ +package s3_test + +import ( + "bytes" + "crypto/md5" + "encoding/base64" + "encoding/hex" + "fmt" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestUploadPart_ContentMD5AndETag(t *testing.T) { + t.Parallel() + + data := bytes.Repeat([]byte("part-data"), 1000) + sum := md5.Sum(data) + + tests := []struct { + name string + contentMD5 string + wantBody string + wantCode int + }{ + {name: "none", wantCode: http.StatusOK}, + {name: "valid", contentMD5: base64.StdEncoding.EncodeToString(sum[:]), wantCode: http.StatusOK}, + { + name: "mismatch", + contentMD5: base64.StdEncoding.EncodeToString(make([]byte, md5.Size)), + wantCode: http.StatusBadRequest, wantBody: "BadDigest", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + handler, backend := newTestHandler(t) + uploadID := doMultipartUpload(t, handler, backend, "bkt") + + req := httptest.NewRequest(http.MethodPut, + fmt.Sprintf("/bkt/obj?partNumber=1&uploadId=%s", uploadID), bytes.NewReader(data)) + if tt.contentMD5 != "" { + req.Header.Set("Content-MD5", tt.contentMD5) + } + + rec := httptest.NewRecorder() + serveS3Handler(handler, rec, req) + + assert.Equal(t, tt.wantCode, rec.Code, rec.Body.String()) + + if tt.wantBody != "" { + assert.Contains(t, rec.Body.String(), tt.wantBody) + + return + } + + assert.Equal(t, `"`+hex.EncodeToString(sum[:])+`"`, rec.Header().Get("ETag")) + }) + } +} + +func TestContentMD5_ErrorCodes(t *testing.T) { + t.Parallel() + + body := []byte("k1") + sum := md5.Sum(body) + good := base64.StdEncoding.EncodeToString(sum[:]) + wrongSum := base64.StdEncoding.EncodeToString(make([]byte, md5.Size)) + badB64 := "!!!not-base64!!!" + shortLen := base64.StdEncoding.EncodeToString([]byte("short")) + + tests := []struct { + name string + op string + contentMD5 string + wantCode string + wantStatus int + }{ + {name: "put_bad_b64", op: "put", contentMD5: badB64, wantCode: "InvalidDigest", wantStatus: 400}, + {name: "put_short", op: "put", contentMD5: shortLen, wantCode: "InvalidDigest", wantStatus: 400}, + {name: "put_mismatch", op: "put", contentMD5: wrongSum, wantCode: "BadDigest", wantStatus: 400}, + {name: "put_good", op: "put", contentMD5: good, wantStatus: 200}, + {name: "part_bad_b64", op: "part", contentMD5: badB64, wantCode: "InvalidDigest", wantStatus: 400}, + {name: "part_short", op: "part", contentMD5: shortLen, wantCode: "InvalidDigest", wantStatus: 400}, + {name: "part_mismatch", op: "part", contentMD5: wrongSum, wantCode: "BadDigest", wantStatus: 400}, + {name: "delete_bad_b64", op: "delete", contentMD5: badB64, wantCode: "InvalidDigest", wantStatus: 400}, + {name: "delete_short", op: "delete", contentMD5: shortLen, wantCode: "InvalidDigest", wantStatus: 400}, + {name: "delete_mismatch", op: "delete", contentMD5: wrongSum, wantCode: "BadDigest", wantStatus: 400}, + {name: "delete_good", op: "delete", contentMD5: good, wantStatus: 200}, + {name: "cfg_bad_b64", op: "cfg", contentMD5: badB64, wantCode: "InvalidDigest", wantStatus: 400}, + {name: "cfg_mismatch", op: "cfg", contentMD5: wrongSum, wantCode: "BadDigest", wantStatus: 400}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + handler, backend := newTestHandler(t) + uploadID := doMultipartUpload(t, handler, backend, "bkt") + mustPutObject(t, backend, "bkt", "k1", []byte("x")) + + var req *http.Request + + switch tt.op { + case "put": + req = httptest.NewRequest(http.MethodPut, "/bkt/new", bytes.NewReader(body)) + case "part": + req = httptest.NewRequest(http.MethodPut, + fmt.Sprintf("/bkt/obj?partNumber=1&uploadId=%s", uploadID), bytes.NewReader(body)) + case "delete": + req = httptest.NewRequest(http.MethodPost, "/bkt?delete", bytes.NewReader(body)) + case "cfg": + req = httptest.NewRequest(http.MethodPut, "/bkt?tagging", + bytes.NewReader([]byte(``))) + } + + req.Header.Set("Content-MD5", tt.contentMD5) + + rec := httptest.NewRecorder() + serveS3Handler(handler, rec, req) + + assert.Equal(t, tt.wantStatus, rec.Code, rec.Body.String()) + + if tt.wantCode != "" { + assert.Contains(t, rec.Body.String(), ""+tt.wantCode+"") + } + }) + } +} diff --git a/services/s3/utils.go b/services/s3/utils.go index 37532043b8..8f427b8719 100644 --- a/services/s3/utils.go +++ b/services/s3/utils.go @@ -1,6 +1,8 @@ package s3 import ( + "bytes" + "context" "crypto/md5" //nolint:gosec // S3 checksum algorithm, not a security use of MD5 "crypto/sha1" //nolint:gosec // SHA1 required for S3 checksum compatibility "crypto/sha256" @@ -8,6 +10,7 @@ import ( "encoding/base64" "encoding/binary" "hash/crc32" + "io" "net/http" "net/url" "strings" @@ -132,3 +135,63 @@ func queryParam(r *http.Request, name string) string { return r.URL.Query().Get(name) } + +// decodeContentMD5 returns ErrInvalidDigest unless header is base64 of a 16-byte MD5. +func decodeContentMD5(header string) ([]byte, error) { + decoded, err := base64.StdEncoding.DecodeString(header) + if err != nil || len(decoded) != md5.Size { + return nil, ErrInvalidDigest + } + + return decoded, nil +} + +// contentMD5FromContext returns the decoded Content-MD5 stored by the handler, if any. +func contentMD5FromContext(ctx context.Context) ([]byte, error) { + header, ok := ctx.Value(md5Key).(string) + if !ok || header == "" { + return nil, nil + } + + return decodeContentMD5(header) +} + +// withContentMD5 validates the Content-MD5 header format and stashes it on ctx for the backend. +func withContentMD5(ctx context.Context, r *http.Request) (context.Context, error) { + header := r.Header.Get("Content-MD5") + if header == "" { + return ctx, nil + } + + if _, err := decodeContentMD5(header); err != nil { + return ctx, err + } + + return context.WithValue(ctx, md5Key, header), nil +} + +// verifyBodyContentMD5 checks Content-MD5 against the request body and restores the body. +func verifyBodyContentMD5(r *http.Request) error { + header := r.Header.Get("Content-MD5") + if header == "" { + return nil + } + + want, err := decodeContentMD5(header) + if err != nil { + return err + } + + body, err := io.ReadAll(r.Body) + if err != nil { + return err + } + + r.Body = io.NopCloser(bytes.NewReader(body)) + + if sum := md5.Sum(body); !bytes.Equal(sum[:], want) { //nolint:gosec // Content-MD5 is MD5 by spec + return ErrBadChecksum + } + + return nil +} diff --git a/services/sns/signing_channels_test.go b/services/sns/signing_channels_test.go new file mode 100644 index 0000000000..89b691e275 --- /dev/null +++ b/services/sns/signing_channels_test.go @@ -0,0 +1,154 @@ +package sns_test + +import ( + "context" + "crypto" + "crypto/rsa" + "crypto/sha1" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "io" + "net/http" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/blackbirdworks/gopherstack/pkgs/events" + "github.com/blackbirdworks/gopherstack/services/sns" +) + +func verifyNotificationSig( + t *testing.T, pub *rsa.PublicKey, version, msgID, topic, message, ts, sig string, +) { + t.Helper() + + canonical := sns.CanonicalNotificationStringForTest(msgID, topic, "subj", message, ts) + + raw, err := base64.StdEncoding.DecodeString(sig) + require.NoError(t, err) + + if version == "2" { + h := sha256.Sum256([]byte(canonical)) + assert.NoError(t, rsa.VerifyPKCS1v15(pub, crypto.SHA256, h[:], raw)) + + return + } + + h := sha1.Sum([]byte(canonical)) + assert.NoError(t, rsa.VerifyPKCS1v15(pub, crypto.SHA1, h[:], raw)) +} + +func TestPublishSignaturesVerifyOnEveryChannel(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + version string + }{ + {name: "sha1", version: "1"}, + {name: "sha256", version: "2"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + const publishes = 6 + + var mu sync.Mutex + + httpEnvs := make(chan snsNotificationEnvelope, publishes*2) + ts := newNotificationServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + + var env snsNotificationEnvelope + if json.Unmarshal(body, &env) == nil { + httpEnvs <- env + } + + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + b := newA1679Backend(t) + + var sqsEvents []*events.SNSPublishedEvent + + emitter := events.NewInMemoryEmitter[*events.SNSPublishedEvent]() + emitter.Subscribe(func(_ context.Context, ev *events.SNSPublishedEvent) error { + mu.Lock() + defer mu.Unlock() + + sqsEvents = append(sqsEvents, ev) + + return nil + }) + b.SetPublishEmitter(emitter) + + tp, err := b.CreateTopic("sig-channels", map[string]string{"SignatureVersion": tt.version}) + require.NoError(t, err) + + _, err = b.Subscribe(tp.TopicArn, "sqs", "arn:aws:sqs:us-east-1:000000000000:q", "") + require.NoError(t, err) + + for range 2 { + _, err = b.Subscribe(tp.TopicArn, "http", ts.URL, "") + require.NoError(t, err) + } + + var wg sync.WaitGroup + + msg := `{"default":"def-body","http":"http-body"}` + + for range publishes { + wg.Go(func() { + _, pErr := b.Publish(tp.TopicArn, msg, "subj", "json", nil) + assert.NoError(t, pErr) + }) + } + + wg.Wait() + + pub := parseSigningCertPublicKey(t, b) + + mu.Lock() + defer mu.Unlock() + + require.Len(t, sqsEvents, publishes) + + for _, ev := range sqsEvents { + assert.Equal(t, tt.version, ev.SignatureVersion) + verifyNotificationSig( + t, pub, tt.version, ev.MessageID, ev.TopicARN, ev.Message, ev.Timestamp, ev.Signature, + ) + + for _, sub := range ev.Subscriptions { + if sub.Protocol != "sqs" { + continue + } + + assert.Equal(t, "def-body", sub.Body(ev.Message)) + verifyNotificationSig(t, pub, tt.version, ev.MessageID, ev.TopicARN, + sub.Body(ev.Message), ev.Timestamp, sub.SignatureFor(ev.Signature)) + } + } + + for range publishes * 2 { + select { + case env := <-httpEnvs: + assert.Equal(t, "http-body", env.Message) + assert.Equal(t, tt.version, env.SignatureVersion) + verifyNotificationSig( + t, pub, tt.version, env.MessageID, env.TopicArn, env.Message, env.Timestamp, env.Signature, + ) + case <-time.After(5 * time.Second): + t.Fatal("HTTP delivery did not arrive") + } + } + }) + } +}