[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 ownership check now runs inside the records read on every backend, the authorization decision is unchanged (crates/server is untouched), and the wire answers are byte-identical to main across the forged-iterator cases from #419 and a six-case scenario probe, in a build with authorization active.

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 wire test test_iterator_reads_only_while_its_arn_is_the_tables_stream asserts the measured behaviour in both states: while the ARN is unchanged, old iterators read what fresh ones read and the eight new keys come from the new stream; once the ARN changes, no old shard id is among the new stream's and every old iterator answers 200 with none of the new keys.

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. Every file added or grown is under 600 lines; crates/engine/src/streams.rs 463 to 486, crates/storage/src/lib.rs 1024 to 1030 and crates/storage-cassandra/tests/direct/streams.rs 637 to 655 (the last two pre-existing over 600) grew; the PostgreSQL and MongoDB stream engines shrank by one line each.

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 with a shard of another stream of the same account: the service answers Requested resource not found: Shard does not exist; main and this branch answer Requested resource not found: Stream: <arn> not found through validate_shard, a path this branch leaves as it was; fix(streams): disabling a stream closes its shards; enabling makes a new one #515 encodes the service sentence on that path (Disabling a stream does not close its shards or change its status #482).
• 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