Port Fluvio adapter to wingfoil-next - #598
Open
0-jake-0 wants to merge 2 commits into
Open
Conversation
Ports the classic `wingfoil::adapters::fluvio` module onto the Op model, behind a new `fluvio` feature: - `fluvio_sub` — a streaming topic-partition consume source over `produce_async`, emitting `Burst<FluvioEvent>` from a caller-chosen start offset. - `FluvioSinkOps::fluvio_pub` — a topic-produce sink over `consume_async_bursts`, sending every record in a burst then issuing a single flush per burst (classic's batching semantics). All classic capabilities are preserved: offset-selected partition consumption, keyed and keyless records, per-burst flush batching, and the single-record convenience sink. Deviations (documented in the module's `# Deviations from classic` block and the deviation register): the graph owns the tokio runtime, the sink connects and creates its producer lazily on the first burst, the sink is an extension trait rather than a free fn + operator trait pair, `fluvio_sub` rejects `RunMode::HistoricalFrom` at wiring (a live unbounded tail with no historical timeline), and a negative `start_offset` is rejected at wiring instead of inside the producer future. Tests: no-service wiring tests in `tests/fluvio_adapter.rs`; a parity port of the classic integration suite in `tests/fluvio_integration.rs` (testcontainers `infinyon/fluvio:0.18.1`, host networking, SC + SPU registration) behind `fluvio-integration-test`, wired into CI via `fluvio-next-integration.yml`. The classic round-trip example is ported to `examples/fluvio/`. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FKhf2tiM8nwtSBzC1St3ga
The main `Build, Test, & Lint` job runs `cargo test -p wingfoil-next --all-features --lib --tests`, which enables `fluvio-integration-test` and so runs this suite *without* `--test-threads=1`. Five of seven tests failed with `SPU register failed: spu 'spu-5001(5001)' already defined`. The cause is a real defect in the suite, not just a CI-config quirk: the cluster uses host networking on fixed ports (the SPU public endpoint the SC hands to clients has to be reachable from the host as `127.0.0.1:9010`), so only one cluster can exist per machine. Every test nonetheless tried to stand up its own container — the second one silently attached to the first cluster's SC and then failed registering an SPU that was already defined. Make the constraint explicit instead of accidental: - `fluvio_cluster()` hands out one shared cluster via `Mutex<Weak<..>>`. Each test holds an `Arc` for its duration, so the container is stopped by `Drop` once the last test finishes — no leaked container, and the suite is correct at any thread count. - `start_fluvio` attaches to a cluster that is already listening rather than racing it for the fixed ports. - SPU registration and topic creation treat "already defined" / "already exists" as success, since the cluster now outlives a single test. Docker is unavailable in the dev sandbox, so this was verified by compilation and clippy only; the suite itself runs in CI. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FKhf2tiM8nwtSBzC1St3ga
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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.
Ports the classic
wingfoil::adapters::fluviomodule onto the Op model, behind a newfluviofeature. Phase 4 item 8 (aeron, iceoryx2, fluvio) — fluvio lands first of the three because, unlike the other two, it is an ordinary async network client needing no native toolchain.Surface
fluvio_sub(&g, run_mode, conn, topic, partition, start_offset)overproduce_async: streams one topic partition from a caller-chosen offset, emitting each record as aFluvioEventin aStream<Burst<FluvioEvent>>.FluvioSinkOps::fluvio_pub(conn, topic, buffer_size)overconsume_async_bursts: sends everyFluvioRecordin a burst, then issues a singleflush()per burst (classic's batching semantics). Implemented forStream<Burst<FluvioRecord>>and, for convenience,Stream<FluvioRecord>.All classic capabilities are preserved: offset-selected partition consumption, keyed and keyless records (
RecordKey::NULL), per-burst flush batching, and the single-record convenience sink.Deviations from classic
Canonical list lives in the adapter's
# Deviations from classicmodule-doc block; summarised:&Handleparam (register A5).fluvio_subrejectsRunMode::HistoricalFromat wiring — a live, unbounded consumer that tails forever after the retained records, with no historical timeline to replay; the historical channel path would block-collect it and deadlock atstart(register B2, ruling (a)). Classic technically permitted a wall-clock historical run.FluvioSinkOpsreplaces the classic freefluvio_pub+FluvioPubOperatorspair, per the sink-as-trait convention.start_offsetis rejected at wiring rather than deferred into the producer future (the validation is pure).The sink also takes a
buffer_sizefor theconsume_async_burstsbound.Tests
tests/fluvio_adapter.rs(fluvio) — no-service wiring tests: historical rejection, negative-offset rejection, connection-refused aborting the run with adapter-named context for both source and sink, and the single-record sink form. 5 tests, green.tests/fluvio_integration.rs(fluvio-integration-test) — parity port of the classic integration suite:connection_refused,sub_from_beginning,pub_round_trip,sub_live_stream,pub_keyless_record,pub_keyed_record,sub_from_absolute_offset. Uses testcontainersinfinyon/fluvio:0.18.1with host networking and the classic SC-then-register-SPU-then-exec-SPU startup dance, carried over verbatim.Docker is unavailable in the dev sandbox, so the container-backed suite was not executed locally; it runs in CI.
Other changes
examples/fluvio/{main.rs,README.md}— the classic round-trip example ported (seed → consume → uppercase → write to a second topic)..github/workflows/fluvio-next-integration.yml+ registration inintegration-tests.yml.next/docs/port-plan.mdPhase 4 item 8 andnext/docs/deviation-register.md(B2 scope) updated.Checks
cargo fmt --all -- --check,cargo lint,cargo clippy -p wingfoil-next --all-features --all-targets -- -D warnings, andcargo test -p wingfoil-next --features fluvioall pass. The scoped clippy was substituted for the workspacecargo lint-allper the skill's sandbox caveat (the classic aeron adapter's C library does not build here); CI runs the full pass.🤖 Generated with Claude Code
https://claude.ai/code/session_01FKhf2tiM8nwtSBzC1St3ga
Generated by Claude Code