Add streaming pipeline: ProtocolStream, TransactionalStream, CdcStream #17
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Adds streamProtocol method to ArrowFlight service that wraps streamQuery with protocol-level message handling: - Data: New records with block ranges - Reorg: Chain reorganization detected with invalidation ranges - Watermark: Confirmation that block ranges are complete Includes validation logic ported from Rust implementation: - validatePrevHash: Genesis vs non-genesis block rules - validateNetworks: Network consistency across batches - validateConsecutiveness: Hash chain and gap detection Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Implements a full transactional layer on top of the protocol stream, providing exactly-once semantics for data processing: - Transaction IDs: Monotonically increasing IDs for each event - StateStore: Pluggable persistence via Context.Tag (InMemoryStateStore included) - Watermark Buffer: Tracks watermarks for reorg recovery point calculation - CommitHandle: Explicit commit control with idempotent semantics - Rewind Detection: Detects and invalidates uncommitted transactions on restart - Retention Window: Prunes old watermarks outside configurable block window Also adds protocol-stream reorg tests ported from the Rust implementation. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Refactors the protocol stream functionality from ArrowFlight into its own ProtocolStream Context.Tag service: - Add protocol-stream/service.ts with ProtocolStream service definition - Remove streamProtocol, detectReorgs, and related code from arrow-flight.ts - Update TransactionalStream to depend on ProtocolStream instead of ArrowFlight - Update exports in protocol-stream/index.ts This improves separation of concerns and allows the protocol stream logic to be used independently or replaced with alternative implementations. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Replace subdirectory index.ts barrel files with root-level .ts files that serve as the public API for each module: - protocol-stream/index.ts → protocol-stream.ts - transactional-stream/index.ts → transactional-stream.ts The package.json exports field (./* → ./src/*.ts) already supports this pattern, so imports like "@edgeandnode/amp/protocol-stream" resolve to the new root-level files. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Replace vitest imports with @effect/vitest and convert all tests to it.effect with per-test Effect.provide for isolation. Remove runWithStore, runWithState, and runWithActor helpers. Use Effect.exit instead of Effect.runPromiseExit for failure tests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Effect is already lazy, so wrapping service methods in `() =>` thunks is redundant. Convert all zero-arg thunk methods to plain Effect values across StateStore, StateActor, CommitHandle, and AdminApi interfaces. Also remove unused imports/variables in reorg.test.ts. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…ubdirectory Move all domain models from src/models.ts to src/core/domain.ts with a src/core.ts barrel, and split the monolithic src/arrow-flight.ts into errors.ts, types.ts, transport.ts, and service.ts submodules — matching the existing protocol-stream/transactional-stream directory pattern. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Port Rust CDC stream to TypeScript/Effect. Wraps TransactionalStream with a BatchStore to persist batch content, enabling Delete events with original data during reorgs and rewinds. - CdcEvent union: Insert (data forwarding) and Delete (lazy batch iterator) - BatchStore service: pluggable persistence (append, seek, load, prune) - InMemoryBatchStore: Ref-backed reference implementation - CdcStream service: Data→Insert, Undo→Delete, Watermark auto-commit+prune - 17 tests covering batch store conformance and stream integration Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
IMax153
changed the title
ProtocolStream and TransactionalStream
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
Adds the core streaming pipeline for the Amp TypeScript SDK, porting the Rust streaming architecture to TypeScript/Effect:
Data,Watermark, andReorgmessages.Data→Insert,Undo→Delete(with lazy batch iterator), and auto-commits watermarks internally.Also includes project structure improvements:
models.tsintocore/domain.tswithcore.tsbarrelarrow-flight.tsinto subdirectory module (errors.ts,types.ts,transport.ts,service.ts)manifest-buildermoduleService Dependency Chain
Test plan
pnpm test)pnpm check:recursive)dprint check)🤖 Generated with Claude Code