feat(sdk): StreamReadOps trait and test-util fixture constructors - #659
feat(sdk): StreamReadOps trait and test-util fixture constructors#659devin-ai-integration[bot] wants to merge 4 commits into
Conversation
Co-Authored-By: Shikhar Bhushan <shikhar@s2.dev>
Co-Authored-By: Shikhar Bhushan <shikhar@s2.dev>
🤖 Devin AI EngineerI'll be helping with this pull request! Here's what you should know: ✅ I will automatically:
Note: I can only respond to comments from users who have write access to this repository. ⚙️ Control Options:
|
Greptile SummaryThis PR adds two things the feldera connector needed to test against the SDK without shadow-copying SDK types: a
Confidence Score: 4/5Safe to merge; the changes are additive, well-tested, and do not touch any existing logic paths. The trait and constructors are purely additive. The sdk/src/ops.rs — the Important Files Changed
Class Diagram%%{init: {'theme': 'neutral'}}%%
classDiagram
class ReadStreamOps {
<<trait>>
+check_tail() Result~StreamPosition, S2Error~
+read_session(input ReadInput) Result~Streaming~ReadBatch~, S2Error~
}
class S2Stream {
-client BasinClient
-name StreamName
-encryption Option~EncryptionKey~
+check_tail() Result~StreamPosition, S2Error~
+read(input ReadInput) Result~ReadBatch, S2Error~
+read_session(input ReadInput) Result~ReadSession, S2Error~
+append(input AppendInput) Result~AppendAck, S2Error~
}
class MockReader {
<<example>>
+check_tail() Result~StreamPosition, S2Error~
+read_session(input ReadInput) Result~Streaming~ReadBatch~, S2Error~
}
class StreamPosition {
+seq_num u64
+timestamp u64
+new(seq_num, timestamp)$ StreamPosition
}
class ReadBatch {
+records Vec~SequencedRecord~
+tail Option~StreamPosition~
+new(records, tail)$ ReadBatch
}
class SequencedRecord {
+seq_num u64
+timestamp u64
+body Bytes
+headers Vec~Header~
+from_parts(...)$ SequencedRecord
}
S2Stream ..|> ReadStreamOps : implements
MockReader ..|> ReadStreamOps : implements
ReadBatch "1" --> "0..*" SequencedRecord
ReadBatch --> StreamPosition : tail
ReadStreamOps ..> ReadBatch : yields
ReadStreamOps ..> StreamPosition : returns
Prompt To Fix All With AIFix the following 2 code review issues. Work through them one at a time, proposing concise fixes.
---
### Issue 1 of 2
sdk/src/ops.rs:22-29
**Missing `Send + Sync` supertraits on `ReadStreamOps`**
The trait is designed for async usage, but `Box<dyn ReadStreamOps>` is `!Send` by default. Callers that store this in a struct used across `await` points or pass it to `tokio::spawn` must write `Box<dyn ReadStreamOps + Send + Sync>` everywhere. Adding `: Send + Sync` as supertraits would make `Box<dyn ReadStreamOps>` directly usable in multi-threaded tokio contexts — every existing implementor (`S2Stream`, the test `MockReader`) is already `Send + Sync`.
### Issue 2 of 2
sdk/src/ops.rs:22-29
**`read` omitted from `ReadStreamOps`**
`S2Stream` has a unary `read` method (single `ReadBatch`, not a streaming session) that is absent from `ReadStreamOps`. The trait name implies all read-side operations are covered, but a consumer needing to mock `read` alongside `check_tail`/`read_session` cannot do so through this interface. Is the omission intentional (scope limited to what feldera's connector uses), or should `read` be included for completeness?
Reviews (1): Last reviewed commit: "docs(sdk): document test fixture constru..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| #[async_trait] | ||
| pub trait ReadStreamOps { | ||
| /// Check the current tail position. | ||
| async fn check_tail(&self) -> Result<StreamPosition, S2Error>; | ||
|
|
||
| /// Open a read session stream. | ||
| async fn read_session(&self, input: ReadInput) -> Result<Streaming<ReadBatch>, S2Error>; | ||
| } |
There was a problem hiding this comment.
Send + Sync supertraits on ReadStreamOps
The trait is designed for async usage, but Box<dyn ReadStreamOps> is !Send by default. Callers that store this in a struct used across await points or pass it to tokio::spawn must write Box<dyn ReadStreamOps + Send + Sync> everywhere. Adding : Send + Sync as supertraits would make Box<dyn ReadStreamOps> directly usable in multi-threaded tokio contexts — every existing implementor (S2Stream, the test MockReader) is already Send + Sync.
Prompt To Fix With AI
This is a comment left during a code review.
Path: sdk/src/ops.rs
Line: 22-29
Comment:
**Missing `Send + Sync` supertraits on `ReadStreamOps`**
The trait is designed for async usage, but `Box<dyn ReadStreamOps>` is `!Send` by default. Callers that store this in a struct used across `await` points or pass it to `tokio::spawn` must write `Box<dyn ReadStreamOps + Send + Sync>` everywhere. Adding `: Send + Sync` as supertraits would make `Box<dyn ReadStreamOps>` directly usable in multi-threaded tokio contexts — every existing implementor (`S2Stream`, the test `MockReader`) is already `Send + Sync`.
How can I resolve this? If you propose a fix, please make it concise.Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
There was a problem hiding this comment.
Addressed: StreamReadOps now has Send + Sync supertraits.
| #[async_trait] | ||
| pub trait ReadStreamOps { | ||
| /// Check the current tail position. | ||
| async fn check_tail(&self) -> Result<StreamPosition, S2Error>; | ||
|
|
||
| /// Open a read session stream. | ||
| async fn read_session(&self, input: ReadInput) -> Result<Streaming<ReadBatch>, S2Error>; | ||
| } |
There was a problem hiding this comment.
read omitted from ReadStreamOps
S2Stream has a unary read method (single ReadBatch, not a streaming session) that is absent from ReadStreamOps. The trait name implies all read-side operations are covered, but a consumer needing to mock read alongside check_tail/read_session cannot do so through this interface. Is the omission intentional (scope limited to what feldera's connector uses), or should read be included for completeness?
Was omitting read intentional (the connector only needs check_tail/read_session), or should read be included in ReadStreamOps for completeness?
Prompt To Fix With AI
This is a comment left during a code review.
Path: sdk/src/ops.rs
Line: 22-29
Comment:
**`read` omitted from `ReadStreamOps`**
`S2Stream` has a unary `read` method (single `ReadBatch`, not a streaming session) that is absent from `ReadStreamOps`. The trait name implies all read-side operations are covered, but a consumer needing to mock `read` alongside `check_tail`/`read_session` cannot do so through this interface. Is the omission intentional (scope limited to what feldera's connector uses), or should `read` be included for completeness?
Was omitting `read` intentional (the connector only needs `check_tail`/`read_session`), or should `read` be included in `ReadStreamOps` for completeness?
How can I resolve this? If you propose a fix, please make it concise.Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
There was a problem hiding this comment.
Addressed: added the unary read to the trait for completeness (delegating impl on S2Stream, mock test extended).
|
Hmm, maybe we end up with |
|
Renaming to |
Co-Authored-By: Shikhar Bhushan <shikhar@s2.dev>
devin-ai-integration
Bot
changed the title
|
Devin, attend to Greptile's feedback |
Co-Authored-By: Shikhar Bhushan <shikhar@s2.dev>
1 participant
Summary
To unit-test its read/tailing logic, the connector in feldera/feldera#5728 had to define its own
S2StreamClienttrait and shadow copies of SDK response types (S2Batch/S2Record/S2Position) — becauseS2Streamhas no trait to mock behind, andReadBatch/SequencedRecord/StreamPositionaren't constructible in downstream tests. This ships both halves (a mockable trait is useless without constructible return types):StreamReadOps— public, dyn-safe (#[async_trait]) trait covering the stream read ops (read,read_session,check_tail), implemented forS2Stream(named to leave room for a futureStreamAppendOps):read_sessionreturnsStreaming<ReadBatch>rather than the concreteReadSessionso mocks can return any boxed stream; theS2Streamimpl just boxes itsReadSession(whoseStreamimpl already yieldsResult<ReadBatch, S2Error>).test-utilfeature — a documented public feature (distinct from the internal_hidden) gating plain-data fixture constructors:StreamPosition::new(seq_num, timestamp),SequencedRecord::from_parts(..)(gate widened from_hidden), andReadBatch::new(records, tail).sdk/tests/test_util.rsdemonstrates the downstream pattern end-to-end: aBox<dyn StreamReadOps>mock yielding constructed batches, with no shadow types.Testing
just fmt,just clippy(-D warnings),just test(679 passed; includes the feature-gated test since the repo test command uses--all-features), pluscargo clippy/test -p s2-sdk --features test-util.Part of a set of SDK improvements from reviewing feldera/feldera#5728 (see #653, #654, #655, #658).
Link to Devin session: https://app.devin.ai/sessions/6a1c8bd8dcd144128571f1cab5af0dcb
Requested by: @shikhar