From fba392e0f6158aa7a0f021f800b1c071abb9cfef Mon Sep 17 00:00:00 2001 From: Witness Patrol Date: Mon, 12 Oct 2026 23:30:15 -0500 Subject: [PATCH 1/6] perf(dynamodb): merge writes into the cached scan order instead of re-sorting The per-table scan-order cache was invalidated wholesale on every write, so a Scan after any write cloned and re-sorted the whole table (sortScanResultsSized ~4.7% of the PGO profile). Puts, updates and deletes from every write path now append to a capped log, and the next Scan merges it into the cached order by binary-searching each touched key; an overflowing log or a bulk change (restore, rebuild, rollback) falls back to a full sort. Ordering, pagination and segments are unchanged; tests check every write type against a model, pagination across writes, segmented and GSI scans, and concurrent scans and writes. A write followed by a Scan on a 10k-item table drops from ~3.5ms to ~60us. Co-Authored-By: Claude Opus 5.5 (1M context) --- .beads/issues.jsonl | 5 + .../testdata/snapshot_inventory.json | 5 + services/dynamodb/item_ops_batch.go | 6 +- services/dynamodb/item_ops_crud.go | 6 +- services/dynamodb/item_ops_scan.go | 205 ++++++- services/dynamodb/scan_order_delta_test.go | 525 ++++++++++++++++++ .../scan_write_interleaved_bench_test.go | 54 ++ services/dynamodb/store.go | 8 + 8 files changed, 794 insertions(+), 20 deletions(-) create mode 100644 services/dynamodb/scan_order_delta_test.go create mode 100644 services/dynamodb/scan_write_interleaved_bench_test.go diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index 3997563ab1..0173a3c9a2 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":"open","priority":1,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T03:49:22Z","created_by":"Witness Patrol","updated_at":"2026-10-13T03:49:22Z","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":"open","priority":2,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T04:18:57Z","created_by":"Witness Patrol","updated_at":"2026-10-13T04:18:57Z","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,7 @@ {"_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-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":"open","priority":3,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T04:25:01Z","created_by":"Witness Patrol","updated_at":"2026-10-13T04:25:01Z","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/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_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_scan.go b/services/dynamodb/item_ops_scan.go index 86b9eb1a7d..34bf2357b6 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 } -// storeScanOrder sorts the snapshot into base-table key order (with sizes alongside) +// 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 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, 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_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/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 From e2d0c85e9e9fa5d8bfbe003b0d5078f53a9b875c Mon Sep 17 00:00:00 2001 From: Witness Patrol Date: Mon, 12 Oct 2026 23:30:20 -0500 Subject: [PATCH 2/6] test(sns,s3): publish benchmark and signature/ETag guards BenchmarkE2E_SNS measures Publish through the router. Profiling showed SNS already signs once per publish and S3 already hashes parts in a single streaming pass, so these guard that behaviour: every channel's signature verifies against the signing certificate, and UploadPart ETags and Content-MD5 validation stay correct. Co-Authored-By: Claude Opus 5.5 (1M context) --- e2e_bench_sns_test.go | 92 +++++++++++++++ services/s3/upload_part_digest_test.go | 64 ++++++++++ services/sns/signing_channels_test.go | 154 +++++++++++++++++++++++++ 3 files changed, 310 insertions(+) create mode 100644 e2e_bench_sns_test.go create mode 100644 services/s3/upload_part_digest_test.go create mode 100644 services/sns/signing_channels_test.go 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/services/s3/upload_part_digest_test.go b/services/s3/upload_part_digest_test.go new file mode 100644 index 0000000000..925e2aeb59 --- /dev/null +++ b/services/s3/upload_part_digest_test.go @@ -0,0 +1,64 @@ +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")) + }) + } +} 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") + } + } + }) + } +} From e51fdc1572b2e041ea8220120c759e98a998f823 Mon Sep 17 00:00:00 2001 From: Witness Patrol Date: Mon, 12 Oct 2026 23:30:41 -0500 Subject: [PATCH 3/6] chore(bd): sync issue state Co-Authored-By: Claude Opus 5.5 (1M context) --- .beads/issues.jsonl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index 0173a3c9a2..f0813b4e92 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -2,7 +2,7 @@ {"_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":"open","priority":1,"issue_type":"task","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T03:49:22Z","created_by":"Witness Patrol","updated_at":"2026-10-13T03:49:22Z","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} From 87e120976c466727084d1a14316325a687eab30c Mon Sep 17 00:00:00 2001 From: Witness Patrol Date: Tue, 13 Oct 2026 16:16:14 -0500 Subject: [PATCH 4/6] fix(dynamodb): ExclusiveStartKey is a position, not an item reference Scan and Query with an ExclusiveStartKey whose item had been deleted restarted from the beginning. Paging now resumes at the first item after the key in scan/query order whether or not the item still exists, on the cached-order and linear paths (GSI/LSI, TTL tables, segmented scans) and for base-table and index Queries in both directions. Tests that encoded the restart behaviour now assert the resume position. Co-Authored-By: Claude Opus 5.5 (1M context) --- .beads/issues.jsonl | 3 +- services/dynamodb/item_ops.go | 63 +++- services/dynamodb/item_ops_query.go | 5 +- services/dynamodb/item_ops_scan.go | 78 ++--- services/dynamodb/query_start_key_test.go | 9 +- .../dynamodb/scan_paging_internal_test.go | 32 -- .../start_key_position_internal_test.go | 306 ++++++++++++++++++ 7 files changed, 403 insertions(+), 93 deletions(-) create mode 100644 services/dynamodb/start_key_position_internal_test.go diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index f0813b4e92..27a475e310 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -317,7 +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":"open","priority":2,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T04:18:57Z","created_by":"Witness Patrol","updated_at":"2026-10-13T04:18:57Z","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} @@ -1543,6 +1543,7 @@ {"_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":"open","priority":3,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T04:25:01Z","created_by":"Witness Patrol","updated_at":"2026-10-13T04:25:01Z","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} 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_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 34bf2357b6..3f3d49a6ad 100644 --- a/services/dynamodb/item_ops_scan.go +++ b/services/dynamodb/item_ops_scan.go @@ -540,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:] } @@ -567,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. @@ -871,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_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/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) + } + }) + } +} From 54525f13219c5a5c6b4a4f944af0bed74839fb88 Mon Sep 17 00:00:00 2001 From: Witness Patrol Date: Tue, 13 Oct 2026 16:16:20 -0500 Subject: [PATCH 5/6] fix(s3): malformed Content-MD5 is InvalidDigest; validate bucket-level bodies A Content-MD5 that is not valid base64 or does not decode to 16 bytes now returns 400 InvalidDigest; a well-formed digest that does not match the body stays BadDigest. Bucket-level PUT/POST requests (DeleteObjects, PutBucket* configuration) previously ignored Content-MD5 entirely and now validate it against the body. Validation is shared across PutObject, UploadPart and the bucket operations. Co-Authored-By: Claude Opus 5.5 (1M context) --- services/s3/bucket_ops.go | 8 +++ services/s3/errors.go | 6 +++ services/s3/multipart.go | 16 +++--- services/s3/multipart_ops.go | 7 ++- services/s3/object_ops_put.go | 25 ++++++---- services/s3/objects.go | 15 +++--- services/s3/upload_part_digest_test.go | 69 ++++++++++++++++++++++++++ services/s3/utils.go | 63 +++++++++++++++++++++++ 8 files changed, 178 insertions(+), 31 deletions(-) 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 index 925e2aeb59..42db9fcfa8 100644 --- a/services/s3/upload_part_digest_test.go +++ b/services/s3/upload_part_digest_test.go @@ -62,3 +62,72 @@ func TestUploadPart_ContentMD5AndETag(t *testing.T) { }) } } + +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 +} From 00c0d18c8e088f1a538faecd3e33b3000c59cfa1 Mon Sep 17 00:00:00 2001 From: Witness Patrol Date: Tue, 13 Oct 2026 16:16:52 -0500 Subject: [PATCH 6/6] chore(bd): close gopherstack-o8hov Co-Authored-By: Claude Opus 5.5 (1M context) --- .beads/issues.jsonl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index 27a475e310..eb724b096c 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -1544,7 +1544,7 @@ {"_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":"open","priority":3,"issue_type":"bug","owner":"blackbird7181@gmail.com","created_at":"2026-10-13T04:25:01Z","created_by":"Witness Patrol","updated_at":"2026-10-13T04:25:01Z","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}