[LeeroyHannigan]

Summary

Since #419, GetRecords called validate_shard before get_stream_records so an iterator could not pair an allowed stream ARN with another stream's shard. That added two catalog queries to every poll (#438), and stream consumers poll idle shards in a loop.The stream not-found sentences are set per operation; see "Stream not-found sentences" below.

Storage statements per GetRecords call, counted from each backend's statement log (SQLite sqlx debug log, PostgreSQL log_statement = 'all', MongoDB profiler level 2, Cassandra full query log), idle shard and shard with records:

Backend	main	this branch
SQLite	4	1 (one stream_shards JOIN tables LEFT JOIN stream_records)
PostgreSQL	5 (2 catalog, 3 data)	2 (1 catalog, 1 data; the two can be different databases)
MongoDB	5	2 (1 catalog find, 1 data aggregate)
Cassandra	4 sequential	3 statements in 2 round trips (catalog tables, then the shard row and the records issued together)

Cassandra idle GetRecords over loopback, 200 calls alternating binaries in the same minute: p50 2.11 to 2.14 ms on main, 1.55 ms on this branch; p99 3.74 to 5.59 against 2.33 to 2.52. SQLite and MongoDB p50 fell 0.15 ms and 1 ms with p99 not worse.

The ownership predicate

Account and table name from the iterator's stream ARN select the table (SQLite: in the join; PostgreSQL: SELECT table_id, stream_label FROM tables WHERE account_id = $1 AND table_name = $2; MongoDB: find_one on _id; Cassandra: a primary-key read on (account_id), table_name, where main filtered on the label with ALLOW FILTERING). The data read keys the shard on table_id. Whether the shard belongs to the stream the ARN's label names is decided in one Rust function, extenddb_storage::util::shard_in_stream(table_id, shard_id, table_label, arn_label), today table_label == Some(arn_label); a table with no stream owns no shard. A shard of another table, another account, or an earlier label on the same table answers the stream not-found sentence validate_shard gave. Zero rows from the data read means the shard is not that table's; one row without a record is the idle shard; the refusal no longer depends on the row count (the shard id is selected alongside the records).

The predicate is pinned at storage level on every backend, each test asserting both directions so a lookup that dropped the name or the account cannot pass by returning whichever table row comes first: SQLite (stream_tests.rs, 7), PostgreSQL (tests/stream_shard_ownership.rs, 3), Cassandra (tests/direct/stream_ownership.rs, 3), MongoDB (stream_records_live_tests.rs, 4). Dropping the account, the table name or the label from the predicate fails the named test on each backend.

Limit bounds

GetRecords now refuses Limit under 1 with 1 validation error detected: Value at 'limit' failed to satisfy constraint: Member must have value greater than or equal to 1 and over 1000 with One or more parameter values were invalid: Limit should be less than or equal to 1000, both measured on the service with SDK parameter validation off (main let Limit: 0 reach storage, where it answered a 500 on MongoDB, and clamped 1001 to 1000). The check runs in the engine before any storage call; the constraint(member, rule) helper is written in the same shape and place as #515's so the two merge.

MongoDB page size

An earlier form of this change on MongoDB put the page in a $lookup array on the shard document, which made a whole page one BSON document under the 16 MB cap: 1000 records of about 16 KB (an 8 KB item under NEW_AND_OLD_IMAGES, the default Limit) answered 500 on every call and the consumer could not advance. The aggregation now follows the $lookup with $unwind and $project, which MongoDB coalesces so the array is never materialized and each record is its own document again, bounded by Limit. Pinned by a test that reads 40 records of about 450 KB from one shard (inserted in descending order so the $sort is what the assertion sees); without the $unwind it fails with BSONObjectTooLarge. The MongoDB storage tests run in the mongodb-rust CI job against the job's container (devtools/run-mongodb-tests).

Streams after a disable and re-enable

Measured on the service (GetShardIterator on S1, disable, re-enable, write, GetRecords with the S1 iterator): the re-enable mints a new stream ARN; the old stream is DISABLED with every shard closed; an iterator issued before the disable reads its own records to the end (NextShardIterator null) and none of the writes after the re-enable; GetShardIterator on the old ARN still answers 200. Today no backend clears stream_label on disable and the re-enable keeps it, so an old iterator keeps reading the one stream and LatestStreamArn does not change; PR #515 changes that lifecycle.The re-enable test runs one state; see "Re-enable test" below.

Tests

• tests/python/test_stream_shard_ownership.py (3): an iterator naming an allowed stream with another stream's shard reads nothing; the disable and re-enable case above; Limit 0 and 1001 refused with the sentences above (raw client). On this branch's release binaries: 3 passed on SQLite (dev mode), PostgreSQL, MongoDB and Cassandra, with test_streams.py 15, tests/test_authz_resource_arn.py 43 and tests/test_stream_table_reuse.py 4 (65 per backend).
• Unit: engine 180 (get_records_limit_bounds_carry_the_measured_sentences), storage 128 (3 on shard_in_stream), SQLite 148 (7 in stream_tests), MongoDB 122 plus 4 live, Cassandra direct stream_ownership 3 and streams:: 14.
• cargo fmt, clippy 1.97.0 with -D warnings across the workspace, 1.88.0 build: clean. No dependency change.rs1024 to 1030 andcrates/storage-cassandra/tests/direct/streams.

Merge notes

• fix(streams): disabling a stream closes its shards; enabling makes a new one #515 (fix/482-stream-disable-closes-shards): eight files, eighteen hunks. crates/engine/src/streams.rs: two hunks beside the shared constraint helper (which merges), take fix(streams): disabling a stream closes its shards; enabling makes a new one #515's constants. The SQLite, PostgreSQL and MongoDB stream engines: the use line (keep both imports) and the describe_stream and validate_shard catalog reads (take fix(streams): disabling a stream closes its shards; enabling makes a new one #515's bodies); get_stream_records merges on all three. Cassandra stream_engine.rs: the use line, the get_stream_records body (merged form: this side's stream_table_row, fix(streams): disabling a stream closes its shards; enabling makes a new one #515's shard_range extended to return the row's table_id as an Option, the compare, then fix(streams): disabling a stream closes its shards; enabling makes a new one #515's bounded read), and the two validate_shard hunks (fix(streams): disabling a stream closes its shards; enabling makes a new one #515's stream_lineage::validate_shard survives). storage-sqlite/src/lib.rs and storage-cassandra/tests/direct_integration.rs: adjacent mod lines, keep both. .github/workflows/integration.yml: one --test list, keep both. After the merge shard_in_stream's body becomes fix(streams): disabling a stream closes its shards; enabling makes a new one #515's parse_shard_id rule (Labelled { label } against the ARN label, Legacy against the table's current label); that function and its three unit tests are the one place to change.
• fix(storage-postgres): U+0000 in catalog attribute names; table name check #531 (fix/367-pg-nul-key-schema-names): clean.

Not in this PR

• No backend clears stream_label when UpdateTable disables the stream, and the re-enable keeps it when shards exist (SQLite update_table.rs, PostgreSQL update_table.rs, MongoDB table_engine.rs), so LatestStreamArn does not change across a disable and re-enable where the service mints a new stream. That is Disabling a stream does not close its shards or change its status #482, fixed by fix(streams): disabling a stream closes its shards; enabling makes a new one #515.
  GetShardIterator on a shard that is not the named stream's keeps the text main had, Requested resource not found: Stream: <arn> not found.; the service answers Shard does not exist (measured 2026-10-24), and that difference is outside this PR.
• In a dev-mode build a token whose ARN does not parse answers Invalid stream ARN format: <arn> where a non-dev build answers Invalid ShardIterator from the authorization layer; unchanged by this branch.
• extenddb version reports the commit of the first build in a target directory (crates/bin/build.rs is not rerun on a worktree HEAD change).

Closes #438

PostgreSQL read bound

The PostgreSQL statement that reads a page of records joined the shard row to its records and applied ORDER BY r.sequence_number LIMIT n after the join. The outer side of the join is one row, but the planner does not use that to treat the join output as ordered by the record's sequence number. It read every record of the shard after the sequence bound, sorted them, and applied the LIMIT to the sorted rows. A call cost time in proportion to the unread backlog of the shard instead of the page size, and held a connection of the data pool for that time.

The records now come from a LEFT JOIN LATERAL subquery that carries its own ORDER BY sequence_number LIMIT n. The planner reads the primary key (shard_id, sequence_number) in order and stops after n rows. The outer query keeps the result shape: an idle shard of the named table returns one row with a NULL record, and a shard of another table returns no row. The statement text is in crates/storage-postgres/src/stream_records_sql.rs, which the engine and the plan test both use.

The measurements ran on PostgreSQL 16.15 in a container, with the shipped schema and one shard of 100,000 records of about 1 KB in a table of 299,000 stream records. The statements were run through psql with the text copied from the code, not through the server binary. The plan rows are EXPLAIN (ANALYZE) runs, each repeated once, and the second run is quoted. The drain row is three runs of a loop inside one database session, so it leaves out network and pool acquisition time.

Case	Join form in the first commit of this PR	LATERAL form
First page of 1000	102.7 ms, 22,580 buffers, 100,000 records fetched	0.56 ms, 158 buffers, 1000 records fetched
Page of 1000 after sequence 50000	53.0 ms, 50,000 records fetched	0.85 ms
Generic plan, first page	81.0 ms	1.49 ms
Generic plan, page after sequence 50000	51.5 ms	1.10 ms
Drain of the shard in 101 calls of 1000 records	5,184 to 5,230 ms	142.6 to 143.6 ms

The read on main, which looked the shard up in separate statements before reading its records, drained the same shard in 108.5 to 111.3 ms, measured the same way (three runs). The join form and the LATERAL form returned the same pages in the same order for six sequence bounds; an md5 of the page bytes matched for each.

Four PostgreSQL storage tests are added, three in crates/storage-postgres/tests/stream_shard_ownership.rs and one beside it:

• five_pages_read_every_record_once_in_sequence_order reads five pages in turn and expects every record once, in sequence order.
• an_idle_shard_reads_an_empty_page_with_and_without_a_bound expects an empty page, not the not-found error, from a shard with no record to read.
• a_page_is_in_sequence_order_when_the_records_were_written_in_reverse (its own file, stream_records_page_order.rs) writes ten records with the highest sequence number first and expects pages of 1, 2, 3 and 3, 4, 5; with the subquery's ORDER BY removed the first page is 8, 9, 10.
• a_page_visits_at_most_the_page_size_of_the_shards_records runs EXPLAIN (ANALYZE) on both statements, in the plan chosen for the parameters and in the generic plan, from the start of a shard of five pages and after its second page, and expects the scans of stream_records to visit no more rows than the page holds.

Stream not-found sentences

The sentence for a stream that does not exist is set per operation. On main, every stream not-found sentence on every backend was Requested resource not found: Stream: <arn> not found., with a trailing period. With this PR the sentences are:

Operation and case	Answer	Basis
DescribeStream, stream ARN of an existing table with a label the table never had	Requested resource not found: Stream: <arn> not found (no period)	Measured on DynamoDB in us-east-1 on 2026-10-24, two runs with the same answer
GetShardIterator, unknown stream ARN, existing shard id	Requested resource not found: Stream does not exist	Measured on DynamoDB in us-east-1 on 2026-10-24, two runs with the same answer
GetRecords, iterator that names an unknown stream, or a shard that is not the named stream's	Requested resource not found: Stream does not exist	Not measured. The service's iterator tokens are opaque, so a token of that kind cannot be built against it. GetRecords uses the GetShardIterator sentence
GetShardIterator, shard that is not the named stream's	Requested resource not found: Stream: <arn> not found. (trailing period)	Same text as on main. The service answers Requested resource not found: Shard does not exist (measured 2026-10-24); that difference is outside this PR

Three operations change their text compared with main: DescribeStream loses the trailing period, and GetShardIterator and GetRecords answer Stream does not exist for an unknown stream. tests/python/test_streams.py (TestStreamNotFoundSentences) sends DescribeStream, GetShardIterator and GetRecords to an unknown stream ARN of an existing table and compares the code and the message of each. test_streams.py and test_stream_shard_ownership.py pass on SQLite, PostgreSQL and MongoDB.

File sizes

crates/engine/src/streams.rs goes from 463 to 597 lines. The PostgreSQL stream engine goes from 617 to 609 lines and the MongoDB stream engine from 810 to 814. Both engines were over 600 lines before this PR and are not split here. Three other files this PR grows were over 600 lines already: crates/storage/src/lib.rs (1,024 to 1,030), crates/storage-postgres/src/lib.rs (877 to 879) and crates/storage-cassandra/tests/direct/streams.rs (637 to 655). Every file this PR adds is under 600 lines. The largest is crates/storage-postgres/tests/stream_shard_ownership.rs at 587.

Re-enable test

test_iterator_reads_only_while_its_arn_is_the_tables_stream runs one state. After a disable and an enable, every backend keeps the table's stream ARN, so the state with a new ARN cannot be reached. The test asserts that precondition, so a backend that starts issuing a new ARN fails with that message, and a comment names #515 as the change that makes the other state reachable.

Other changes in this update

• MongoDB: a record document without record_data answers Internal (Missing record_data), as it does on main. The new aggregation had dropped such a document as an idle page. The aggregation now marks every joined record, and two live tests pin both halves: the document without record_data, and the idle shard.
• A shard iterator token with fewer than four fields answers Invalid ShardIterator in the engine, as the authorization layer does. A build without the layer (developer mode) answered such a token as expired.
• The doc comment on StreamEngine::get_stream_records in crates/storage/src/lib.rs and the comment in handle_get_records in crates/engine/src/streams.rs now say that only SQLite resolves the shard in the same round trip as the records, and that PostgreSQL, MongoDB and Cassandra read the table first. The comments in crates/server/src/request_helpers.rs and tests/test_authz_resource_arn.py now say that the storage read resolves the shard against the stream, not that the engine checks it before reading.
• docs/dynamodb-limits.md lists the GetRecords limit as enforced and gives its two sentences.
• The SQLite stream tests and the Python shard ownership test carry the copyright line.

[robinnsc]

Approving: nothing here blocks. The ownership check holds up. Forged iterators on the SQLite, PostgreSQL and MongoDB release binaries (another table's shard, another account, an older or near-miss label, a deleted and recreated table, a table with no stream, swapped shards on every iterator type) gave the same answers as main apart from the intended sentence change. Dropping the account, the table name, the shard's table id or the label from the predicate failed a test on each of the four backends. The statement counts also matched what we logged: PostgreSQL 5 to 2 and MongoDB 5 to 2, plus SQLite 4 to 1 from reading the code. A few things in the description:

• The #515 merge notes look out of date. git merge-tree --write-tree against #515's head (5f22dc4) gives 9 conflicted files and 19 hunks. The extra file is crates/storage-postgres/src/lib.rs, from the LATERAL commit (pub mod stream_records_sql next to #515's mod stream_lifecycle). One of the four Cassandra hunks is in describe_stream, not validate_shard. After resolving the textual conflicts it still won't compile: both sides add a stream_table_row method on CassandraEngine with different return types (stream_engine.rs:68 here, stream_lifecycle.rs:387 there), so one needs renaming. The merged tree would also hold two copies of each not-found sentence (util::* and stream_lineage::*).
• On main's MongoDB, Limit: 0 didn't answer a 500. limit(0) means no limit, so the base binary returned the whole shard (3,753 records in our run). The fix is the same, but main's behaviour was worse than described. Similarly, in dev mode main read a token with no stream ARN with only the account check, which this PR now refuses; that might be worth a line.
• Main's records statement is still a bit faster per full page in database time than the LATERAL form (see the inline comment), which the plan table doesn't show because it compares against the first commit's join form. Over the network, the saving is the three removed round trips.
• A few counts have moved on: engine 182, storage 134, SQLite 151, MongoDB 125 plus 6 live, and test_streams.py 16. There are also a couple of pasted fragments ("No dependency change.rs` 1024 to 1030 ...", "loop.The stream", "lifecycle.The re-enable") and a bullet that lost its marker in "Not in this PR".
• Refusing Limit over 1000 matches the service, but the botocore model has no maximum, so a client written against ExtendDB alone that sends 1001 or more now gets a ValidationException where main clamped. The Kinesis adapter clamps to 1000 itself. Probably worth a release note.

Separately from this PR: on PostgreSQL a consumer can skip records written by concurrent writers, because the sequence comes from nextval('stream_seq') inside the write transaction (data/tx_helpers.rs:340-344), so a later sequence can commit first and the reader moves past the earlier one. With 8 BatchWriteItem threads and a LATEST reader per shard, 0 to 11 of 8,000 records were skipped on the base binary as well as this branch. The records are in stream_records, so a fresh TRIM_HORIZON read finds them. It's outside #438, so it probably wants its own issue.

[robinnsc]

This outer ORDER BY adds a Sort over the page the LATERAL subquery already returns in key order. On a 100k-record shard, a page of 1000 took 2.2 ms in EXPLAIN against 1.2 ms without it and about 1.1 ms for main's records statement, or roughly 10.3 ms against 8.7 ms client side. It's invisible behind the ~85 ms the server spends per page, and an idle or Limit-1 poll sorts one row, so this is optional. The ordering is needed for a guarantee, since NextShardIterator comes from the last record, but nothing pins it: removing it from both statements still passes every PostgreSQL test, because the nested loop happens to emit in order. If you want the database time back, one option would be to drop it and sort records by dynamodb.sequence_number in Rust (the values are fixed width), with a test that would fail on an unordered page. Keeping it and adding such a test would also be fine.

[robinnsc]

Small test gap: replacing this call with main's unwrap_or(1000).min(1000) still passes the engine tests, because get_records_limit_bounds_carry_the_measured_sentences calls the function directly. Only tests/python/test_stream_shard_ownership.py covers the wiring. A unit test through handle_get_records with Limit: 0 would pin it on the Rust side.