GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — [dbsp] Reject exchange connections from other runtimes by blp · Pull Request #7814 · feldera/feldera

Conversation

@blp

@blp blp commented Nov 3, 2026 •
edited
Loading

Uh oh!

There was an error while loading. Please reload this page.

Copy link
Copy Markdown
Member

Fixes #7800.

Problem

sharded_accumulator::tests::take_received failed with this panic:

failed to deliver 0 from 0 to 0..1: UnknownExchange(0)

The receiving host has only worker 0 (0..1), but the message says that it comes from worker 0. Thus, the message came from another runtime, not from a peer. Two problems together made this possible:

Problem Effect
An exchange connection did not identify its runtime. The server accepted a connection from a client of any runtime.
A client task connected to its peer forever, also after its runtime stopped. After the peer stopped, the client could connect to a new runtime that the kernel gave the same port.

If the exchange ID had existed, the message would have been delivered as data without an error.

This is likely to affect only the tests, which run many multihost runtimes in one process. In production, each host runs in a process of its own, which a restart replaces, so the old clients stop when their process exits.

Changes

Commit Change
Stop an exchange client task after its runtime is gone. After its ExchangeClient drops, the task sends only the messages that the peer has not acknowledged, then stops. It does not stop at once, because the peer can still wait for these messages.
Reject exchange connections from other runtimes. CircuitConfig has a new exchange_uuid, which every host of a run shares. A client sends it after the exchange magic. The server sends the magic back only for its own UUID, so a client of another runtime fails to connect and cannot deliver or take over the slot of a real peer. A pipeline uses its deployment ID. The test hosts use a new random UUID.
Let a stopped exchange client connect again to finish sending. After the stop, the client still connects again when a connection fails, until the peer acknowledges all of its messages. It gives up after 60 s (10 s in the tests) and logs a warning with the number of messages. The UUID check makes this safe. Without this, a peer could hang when RuntimeHandle::join dropped the clients of a host while other hosts still needed their last messages.

The exchange magic is now feldera-xchg-v3, so all the hosts of a pipeline must run a binary with this change.

Tests

Test Fails without
rejects_a_client_of_another_runtime the UUID check (with the same UnknownExchange panic as #7800)
client_connect_fails_with_another_exchange_uuid the UUID check
dropped_client_does_not_reconnect the client stop
dropped_client_finishes_sending the client stop
dropped_client_reconnects_to_finish_sending the reconnection after the stop

I made sure that each test fails when its fix is disabled.

The exchange, sharded accumulator, and multihost tests ran 25 times with no failures or hangs. On main, the #7800 failure showed up in 3 of 38 runs.

The dbsp lib tests pass, except trace::test::storage_calls_are_charged_to_the_worker_for_their_time_off_the_cpu. That test also fails on main on my machine, so it is not related.

Written with Claude.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions
mythical-fred-oss[bot]
mythical-fred-oss Bot reviewed Nov 3, 2026
View reviewed changes
Comment thread crates/dbsp/src/operator/communication/exchange.rs Outdated
{
warn!("connection to {description} dropped ({error}), waiting to retry")
Ok(()) => return,
Err(_) if stop.is_cancelled() => return,

@mythical-fred-oss mythical-fred-oss Bot Nov 3, 2026

Copy link
Copy Markdown
Contributor

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pitfall (medium): after stop, both this arm and the connect arm at L714 drop any unacknowledged messages without logging anything. If a peer is still waiting for those final messages (which is the reason the drain exists), it hangs and the logs give no clue why. In the unit tests, inject_fault fails about 1% of sends, so this can happen there too. Could you log at warn! when channel.has_unacknowledged() is true, e.g. "giving up on N unacknowledged messages to {description} after runtime stopped: {error}"?

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

@blp blp Nov 3, 2026

Copy link
Copy Markdown
Member Author

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in eab311c. Both paths now log at warn! when messages are unacknowledged, for example:

giving up on 1 unacknowledged messages to 127.0.0.1:37235 for exchange with workers 0..1 after the runtime stopped: unexpected end of file

The same commit adds the two tests from the summary:

Test Fails without
dropped_client_gives_up_after_a_failure the "give up after the stop" arm (the client connects again)
layout_without_runtime_id_gets_nil #[serde(default)] on runtime_id

(Claude)

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions
@mythical-fred-oss

mythical-fred-oss Bot commented Nov 3, 2026

Copy link
Copy Markdown
Contributor

LGTM. The fix for #7800 is sound: the client stop and the runtime-ID check each close one half of the race, and every new test fails when its fix is turned off.

What I ran (debug builds)

Check Result
cargo test -p dbsp --lib communication::exchange 25/25 pass
cargo test -p dbsp --lib sharded_accumulator 26/26 pass
new tests + take_received, looped 15× 15/15 green, no flakes
cargo test -p feldera-types --lib pass
cargo check -p dbsp_connectors pass
pre-commit run --files <changed> pass

Gaps in test coverage (medium)

  • No test covers the drain path where the client's runtime stops and then the connect or send fails (Err(_) if stop.is_cancelled()). Today the remaining messages are dropped without a log line; see the inline comment.
  • No test for Layout serde backward compatibility, i.e. deserializing a Multihost layout that has no runtime_id and getting the nil UUID through #[serde(default)]. It is cheap to add if a Layout is ever persisted or sent between components.

Gates: tests ✅, flakiness ✅ (the negative 500 ms wait cannot fail falsely), no new unsafe, no new dependencies (uuid is already used), commit messages ✅. The v2 exchange magic is an internal wire change, and a pipeline's hosts always run the same binary, so no user-facing breaking change or docs update is needed.

All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

@blp blp self-assigned this Nov 3, 2026
@blp blp added bug Something isn't working DBSP core Related to the core DBSP library ft Fault tolerant, distributed, and scale-out implementation CI/CD rust Pull requests that update Rust code enterprise Issue related to Feldera Enterprise features. multihost Related to multihost or distributed pipelines labels Nov 3, 2026
@blp
blp requested a review from ryzhyk November 3, 2026 21:55
mihaibudiu
mihaibudiu approved these changes Nov 3, 2026
View reviewed changes
Comment thread crates/dbsp/src/operator/communication/exchange.rs Outdated
/// acknowledges all of them, or at the first failure to connect or send.
/// Without the stop, this task would connect forever, and after the peer
/// stops, it could connect to another runtime that reuses the peer's
/// address.

@mihaibudiu mihaibudiu Nov 3, 2026

Copy link
Copy Markdown
Contributor

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this seems to be counter to the recovery-oriented computing matra that says you should not have a clean shutdown path

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

@blp blp Nov 3, 2026

Copy link
Copy Markdown
Member Author

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't understand this comment.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

@mihaibudiu mihaibudiu Nov 3, 2026

Copy link
Copy Markdown
Contributor

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this the setup or teardown path?

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

@blp blp Nov 3, 2026

Copy link
Copy Markdown
Member Author

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The change in this commit affects what happens when a multihost DBSP runtime is dropped. Before this commit, this leaked the send task: it remained running and trying to connect forever until the process died. In an ordinary pipeline process, this was not a big deal, because the process died soon after the runtime drops, since the whole point of the pipeline is to host the runtime. However, when we run the unit tests, we create and drop many runtimes, and leaking the send task was both wasteful in CPU and memory, and harmful, because if another multihost runtime started up and ended up on on the same TCP port as one that the send task is trying to connect to, then it could send bogus data to it. This commit fixes the leak, and the following commit "Reject exchange connections from other runtimes" fixes the problem of two different runtimes talking to each other through this kind of accident. These are both primarily problems that arise in CI; they are less likely to surface in production.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

@mihaibudiu mihaibudiu Nov 3, 2026

Copy link
Copy Markdown
Contributor

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it; it's worth cleaning up in testing, because it makes tests faster.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

@blp blp Nov 3, 2026

Copy link
Copy Markdown
Member Author

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, or in hypothetical other users of the dbsp crate that happen to use multiple runtimes.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions
Comment thread crates/feldera-types/src/circuit.rs Outdated
///
/// The nil UUID if not set (see [Layout::with_runtime_id]).
#[serde(default)]
runtime_id: Uuid,

@mihaibudiu mihaibudiu Nov 3, 2026

Copy link
Copy Markdown
Contributor

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I find this name can be confused with the "runtime version"

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

@blp blp Nov 3, 2026

Copy link
Copy Markdown
Member Author

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll rename it to exchange_uuid.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions
blp added 3 commits November 4, 2026 08:18
@blp
[dbsp] Stop an exchange client task after its runtime is gone.
5af4481
An exchange client task connected to its peer forever, even after its
runtime and its peer stopped.  In the tests, it tried again about once
per millisecond, and it could connect to a new runtime that listened on
the old address of its peer (see issue #7800).

Now, after its `ExchangeClient` drops, the task sends only the messages
that the peer has not acknowledged, because the peer can still wait for
them, and then stops.  It also stops at the first failure to connect or
send.  It does not stop at once, because the connection uses zero
linger, so closing it would discard data that is not sent yet.

This problem is likely to affect only the tests, which run many
multihost runtimes in one process.  In production, each host runs in a
process of its own, which a restart replaces, so the old clients stop
when their process exits.

Signed-off-by: Ben Pfaff <[email protected]>
@blp
[dbsp] Reject exchange connections from other runtimes.
eb768dd
An exchange connection did not identify the runtime that it belongs to.
Thus, a client of another runtime could connect to the address of this
runtime and deliver its messages here.  This made
`sharded_accumulator::tests::take_received` fail with
`UnknownExchange(0)` (issue #7800): a client of an earlier test
connected to a port that the kernel gave to a host of a new test.  If
the exchange ID had existed, the message would have been delivered as
data without an error.

This problem is likely to affect only the tests.  It needs a client to
outlive its runtime in the same process, and a new runtime to listen on
the old address.  The tests run many multihost runtimes in one process,
on ports that the kernel selects and can reuse.  In production, each
host runs in a process of its own, which a restart replaces, so the old
clients stop when their process exits.  The previous commit makes the
first condition less likely, because a client now stops after its
runtime is gone, but a client can still connect while it sends its
last messages.

To fix the problem, `CircuitConfig` now has `exchange_uuid`, which every
host of a run shares.  A client sends it after the exchange magic.  The
server sends the magic back only if the UUID is its own, so a client of
another runtime fails to connect and waits for the backoff before it
tries again, and it cannot deliver or take over the slot of a real
peer.  A pipeline uses its deployment ID.  The test hosts use a new
random UUID.  The magic is now version 3.

Fixes #7800.

Signed-off-by: Ben Pfaff <[email protected]>
@blp
[dbsp] Let a stopped exchange client connect again to finish sending.
cf2e93b
After its runtime was gone, an exchange client sent the messages that
its peer had not acknowledged, but it gave up at the first failure to
connect or send, and dropped the messages without a log line.  A peer
can still need them: `RuntimeHandle::join` drops the clients of a host
when the workers of that host finish, even while the workers of the
other hosts run.  In the multihost tests, which inject a failure into
about 1% of the sends, this probably made a peer hang in about 1 run
in 13.

Now, after the stop, the client still connects again when a connection
fails, until the peer acknowledges all of its messages.  It gives up
after 60 seconds, or 10 seconds in the tests, and then logs a warning
with the number of messages.  The check of the exchange UUID makes the
reconnection safe: if another runtime listens on the old address, the
client cannot deliver there.

This commit also makes the tests accept a ping before the client closes
the connection.

Signed-off-by: Ben Pfaff <[email protected]>
@blp
blp force-pushed the multihost-runtime-id branch from eab311c to cf2e93b Compare November 4, 2026 19:14
@blp
blp enabled auto-merge November 4, 2026 19:14
@blp
blp disabled auto-merge November 4, 2026 19:15
@blp
blp enabled auto-merge November 4, 2026 19:15
@blp
blp disabled auto-merge November 4, 2026 20:29
@blp
blp added this pull request to the merge queue Nov 4, 2026
@blp blp mentioned this pull request Nov 4, 2026
[dbsp] End an exchange client's task after the client is dropped #7842
Closed
1 of 12 tasks
Merged via the queue into main with commit 64e5cfe Nov 5, 2026
1 check passed

Uh oh!

There was an error while loading. Please reload this page.

@blp
blp deleted the multihost-runtime-id branch November 5, 2026 03:28
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@mihaibudiu mihaibudiu mihaibudiu approved these changes
@mythical-fred-oss mythical-fred-oss[bot] mythical-fred-oss[bot] left review comments
@ryzhyk ryzhyk Awaiting requested review from ryzhyk

Uh oh!

There was an error while loading. Please reload this page.

Assignees

@blp blp

Labels

bug Something isn't working CI/CD DBSP core Related to the core DBSP library enterprise Issue related to Feldera Enterprise features. ft Fault tolerant, distributed, and scale-out implementation multihost Related to multihost or distributed pipelines rust Pull requests that update Rust code

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

[dbsp] Flaky: sharded_accumulator::tests::take_received panics with UnknownExchange on multihost harness

Uh oh!

There was an error while loading. Please reload this page.

2 participants

@blp @mihaibudiu
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.
You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.