Repository navigation
Conversation
yesyayen
left a comment
yesyayen
left a comment
There was a problem hiding this comment.
Thank you for your first contribution to ExtendDB, @forhappy! Could you please update the PR description using the PR template?
My understanding of the issue
Today, an empty open shard and a fully read closed shard look the same to the engine's handler, because both return no records. An open shard should keep returning NextShardIterator, because new records may still arrive. A closed, fully read shard should omit it, so the consumer knows it can stop polling. Your StreamContinuation change lets storage tell the handler which case it is: More keeps the iterator, and End drops it.
| /// Return `End` only after observing a closed shard and exhausting its | ||
| /// records. An empty open shard returns `More(None)`. A final nonempty | ||
| /// page may return `End`; its records must still reach the caller. | ||
| /// |
There was a problem hiding this comment.
Please document the guarantees behind End, shard closure must account for in-flight writes and prevent further writes. Re-enabling a stream must create new shards.
| let continuation = if closed && !has_more { | ||
| StreamContinuation::End | ||
| } else { |
There was a problem hiding this comment.
As is, this change has no effect in ExtendDB: none of our backends closes a shard, so End never fires. Could you add shard closing? Disabling a stream should close its shards (set ending_sequence_number and report DISABLED in DescribeStream), and re-enabling should create a new stream with new shards.
| let continuation = if closed && !has_more { | ||
| StreamContinuation::End | ||
| } else { | ||
| StreamContinuation::More(last_seq) | ||
| }; | ||
| Ok((records, continuation)) |
There was a problem hiding this comment.
Please add tests for the new continuation path on PostgreSQL and MongoDB, not only SQLite. Please also add an end-to-end test that disables a stream, reads each shard to the end, and checks that every record arrives once and NextShardIterator is omitted.
LeeroyHannigan
left a comment
LeeroyHannigan
left a comment
There was a problem hiding this comment.
I read the diff, built the head, and measured the closed shard case against DynamoDB in us-east-1. Gates pass on your branch: fmt, clippy with -D warnings, 1224 workspace tests, and the stream wire tests on SQLite and PostgreSQL. The service drops NextShardIterator on the same response that carries a closed shard's final records, so your continuation type and its contract match it. Nothing in the tree writes stream_shards.ending_sequence_number, so End never fires: on a build of this branch every GetRecords after UpdateTable StreamEnabled=false still returns an iterator. The branch also needs a rebase onto #419 and #338, and the new Limit check and the limits doc row need measuring against the service. Once the inline items are addressed I will approve.
| return Err(StorageError::Validation("Invalid ShardIterator".to_owned())); | ||
| } | ||
| }; | ||
| let closed = ending_sequence.is_some(); |
There was a problem hiding this comment.
Blocking: closed is false in every server run, so End never reaches a client, because no code writes stream_shards.ending_sequence_number: the two inserts are this file at line 57 and crates/storage-postgres/src/stream_engine.rs:134, neither supplies it, and the tests below set it with their own UPDATE. I built this branch and ran the sequence end to end: after UpdateTable StreamEnabled=false, DescribeStream still reports ENABLED with no EndingSequenceNumber on any shard, and every GetRecords returns NextShardIterator, on SQLite and on PostgreSQL. Closing a stream's shards when it is disabled would make this change observable, and yesyayen asked for that on 3 October.
| return None; | ||
| }; | ||
| let sequence = last_sequence.as_deref().unwrap_or(previous_sequence); | ||
| let token = format!("{shard_id}|AFTER_SEQUENCE_NUMBER|{sequence}|{created_at}"); |
There was a problem hiding this comment.
Blocking on rebase: main builds this token through encode_shard_iterator, which carries the stream ARN as a fifth field (#419, merged after you opened this). A four-field token is classified Expired and the server answers ExpiredIteratorException before the handler runs (crates/server/src/request_helpers.rs:210-218), so a consumer would fail on its second call. Call encode_shard_iterator and pass it.stream_arn through, and assert the result with decode_shard_iterator in the unit test rather than against the four-field string.
| } | ||
|
|
||
| /// Stream records and explicit continuation, including closed-shard exhaustion. | ||
| pub type StreamRecordsResult = Result<(Vec<StreamRecord>, StreamContinuation), StorageError>; |
There was a problem hiding this comment.
Blocking on rebase: #338 added the Cassandra backend, whose get_stream_records still returns (Vec<StreamRecord>, Option<String>), so this type change does not compile against it (E0053 at crates/storage-cassandra/src/stream_engine.rs:131), and two of its tests read the second element as an Option (crates/storage-cassandra/tests/direct/streams.rs:521 and :563). I resolved the three textual conflicts locally and cargo check --workspace --all-targets passes once that backend is ported, so the work is mechanical. Please say in the description whether Cassandra gets the lookahead or stays on More for now.
| .ok() | ||
| .filter(|limit| (1..=1000).contains(limit)) | ||
| .ok_or_else(|| { | ||
| StorageError::Validation("Limit must be between 1 and 1000".to_owned()) |
There was a problem hiding this comment.
Major: this range check is new behaviour outside the stated fix, and the message is not the one the service sends. Measured in us-east-1: Limit=0 and a negative Limit are rejected with 1 validation error detected: Value at 'limit' failed to satisfy constraint: Member must have value greater than or equal to 1, while Limit=1001 is rejected with One or more parameter values were invalid: Limit should be less than or equal to 1000. Either drop the check from this PR and file it separately, or carry those two messages; note that the AWS SDKs reject a Limit below 1 client-side, so only a raw HTTP client reaches this line.
| | Simultaneous shard readers | 2 (1 for global tables) | Not enforced | No concurrent reader tracking | | ||
| | Max write capacity with streams (provisioned) | 40,000 WCU | Enforced | Same as table WCU limit | | ||
| | GetRecords: max records per call | 1,000 | Not enforced | No per-call record count limit | | ||
| | GetRecords: max records per call | 1,000 | Enforced | Handler caps pages; storage rejects out-of-range limits | |
There was a problem hiding this comment.
Major: this row now says the 1,000 cap is enforced, and on a build of this branch GetRecords with Limit=1001 returns 200 with an iterator, on both SQLite and PostgreSQL. The handler clamps with .min(1000) (crates/engine/src/streams.rs:221) before storage sees the value, so the new range check cannot reject it, and the service sends 400 ValidationException for the same request. Make the handler reject an over-limit request, or leave this row and the technical-debt entry as they were.
| } else { | ||
| StreamContinuation::More(last_seq) | ||
| }; | ||
| Ok((records, continuation)) |
There was a problem hiding this comment.
Major: the continuation path has tests on SQLite only, so the closed && !has_more branch in this file and in the MongoDB producer is unexercised, which is the point yesyayen raised on 3 October. A test that disables a stream, drains each shard and asserts that each record arrives once and that the final response omits NextShardIterator would also pin the contract the trait doc now promises.
| /// | ||
| /// Return `End` only after observing a closed shard and exhausting its | ||
| /// records. An empty open shard returns `More(None)`. A final nonempty | ||
| /// page may return `End`; its records must still reach the caller. |
There was a problem hiding this comment.
Major: this says when to return End but not what closing a shard guarantees, which is the other half of yesyayen's first comment. Please state that a closed shard takes no further writes, that records written before closure are still served, and that re-enabling a stream produces a new stream with new shards. The service does that last part: re-enabling gave a new ARN with a new millisecond label and four new shard ids, none shared with the old stream, which stayed readable as DISABLED.
| let opts = FindOptions::builder() | ||
| .sort(doc! { "sequence_number": 1 }) | ||
| .limit(limit) | ||
| .limit((limit + 1) as i64) |
There was a problem hiding this comment.
Minor: the lookahead row is fetched on every call but read only when the shard is closed, so until shards close this is one extra row per GetRecords on each of the three backends. Fetching it only for a closed shard would leave the open path as it is.
Summary
StreamEngine::get_stream_recordsan explicit continuation state so an empty open shard remains pollable and an exhausted closed shard ends its iterator.This replaces draft PR #371 with the same commit on the crabbuild-owned fork branch. BeyondDB pins this commit from the crabbuild fork so its release cadence is independent of upstream merge timing.
Verification
PostgreSQL and MongoDB runtime integration tests remain to be run in their database environments. This changes the storage trait consumed by external backends, so contract review is especially useful.