gorums: add stream deduplication for symmetric peers - #334
Open
meling wants to merge 8 commits into
Open
Conversation
Contributor
 |
|
Overall Grade |
Security Reliability Complexity Hygiene |
Code Review Summary
| Analyzer | Status | Updated (UTC) | Details |
|---|---|---|---|
| Go |  |
Aug 12, 2026 12:38p.m. | Review ↗ |
| Shell | |
Aug 12, 2026 12:38p.m. | Review ↗ |
Important
AI Review is run only on demand for your team. We're only showing results of static analysis review right now. To trigger AI Review, comment @deepsourcebot review on this thread.
meling
changed the title
Contributor
There was a problem hiding this comment.
Pull request overview
Adds stream deduplication support for symmetric Gorums peers so each peer-pair shares a single bidirectional stream (owned by the lower-ID peer), including a new Server.WaitForAll helper and broad test/doc coverage to validate topology, message-ID safety, ordering, and reconnect behavior.
Changes:
- Introduce
WithStreamDedupandServer.WaitForAll, plus connection-manager support for “born-shared” outbound nodes that borrow inbound peer transports. - Update stream transport/message-ID behavior to prevent ID-space collisions when sharing routers/streams across both call directions.
- Add/extend tests and documentation covering dedup topology, fail-fast semantics (
ErrStreamDown), ordering, metadata propagation, and reconnect healing.
Reviewed changes
Copilot reviewed 26 out of 26 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| stream_dedup_test.go | New end-to-end tests for stream dedup topology, fail-fast, retained configs, multicast/quorum calls, ordering, and reconnect healing |
| server.go | Add WithStreamDedup and Server.WaitForAll; clarify ConnectedPeers/PeerConfig behavior under dedup |
| server_test.go | Clarify existing peer-config test covers non-dedup (“Dual”) topology |
| internal/tests/ordering/order_test.go | Add ordering regression test for dedup shared stream (“back-channel”) behavior |
| internal/tests/metadata/metadata_test.go | Run per-message metadata test across dual vs dedup topologies and wait for peers when dedup is enabled |
| internal/stream/transport.go | Add shared-transport support, shared flag, and ErrStreamDown behavior for shared transports without a channel |
| internal/stream/testhelpers.go | Add test-only helper to create shared transports with overridden message-ID generator |
| internal/stream/teardown_deadlock_test.go | Update test comment wording to reflect stream-dedup deadlock scenario |
| internal/stream/router_test.go | Adjust expected metadata label string for server-initiated/dedup path |
| internal/stream/channel.go | Clarify eager reconnect rationale for dual + dedup stream topologies |
| internal/stream/channel_test.go | Update commentary to reflect dedup implications for stream replacement and eager reconnect |
| internal/impl/call_context.go | Ensure shared dedup nodes get per-node messages with server-initiated IDs to avoid collisions on shared streams |
| internal/impl/call_context_test.go | Extend tests to validate per-node ID behavior for shared dedup node vs regular nodes |
| internal/conn/outbound_manager.go | Create born-shared outbound nodes for lower-ID peers under dedup; validate peer address before borrowing |
| internal/conn/node.go | Add shared node creation + Node.IsShared, and align comments with shared-channel semantics |
| internal/conn/node_trysend_test.go | Update test commentary to stream-dedup terminology |
| internal/conn/node_test.go | Add test ensuring shared transports fail fast with ErrStreamDown when disconnected; assert IsShared safety on missing transport |
| internal/conn/inbound_manager.go | Add knownPeer accessor for dedup borrowing from pre-created known nodes |
| internal/conn/doc.go | Document dedup in internal conn package overview |
| internal/conn/dial_opts.go | Add StreamDedup dial option and document interaction with inbound manager |
| internal/conn/config.go | Add helpers to detect dedup-required waiting and validate dedup configuration |
| inbound_manager_test.go | Add tests ensuring dedup Extend borrows known peers and validates address/peer membership |
| gorumstest/gorumstest.go | Clarify LocalServers applies options like WithStreamDedup to every server |
| errors.go | Expand ErrStreamDown doc to describe dedup semantics; include WaitForAll in ErrStopped doc |
| doc/user-guide.md | Add user-facing “Stream Deduplication for Symmetric Peers” guidance and WaitForAll usage notes |
| doc/dev-guide.md | Document dedup internals and message-ID space rationale for shared streams |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+51
to
+58
| select { | ||
| case <-done: | ||
| return | ||
| default: | ||
| for _, node := range srv.PeerConfig().Nodes() { | ||
| _ = node.IsShared() | ||
| } | ||
| } |
Comment on lines
+96
to
+99
| // A lower-ID peer dials this node, so rather than dial back, this node | ||
| // reuses that peer's inbound connection. It works once the peer connects | ||
| // and fails with ErrStreamDown until then; [Server.WaitForAll] waits for | ||
| // the peer. |
meling
force-pushed
the
gorums/stream-dedup
branch
from
August 12, 2026 11:42
4d7d2b5 to
d88e431
Compare
A transport bundles the node ID, channel reference, router, and message-ID generator a call needs to reach one node. A shared transport is derived from a peer's transport and references the same channel, router, and generator instead of owning them, so one bidirectional stream can carry calls in both directions. Borrowing changes two behaviors. A borrower cannot dial, so a request enqueued while the owner's stream is down fails fast with ErrStreamDown rather than being silently dropped, which is what an owned transport does when it will re-establish the stream on the next send. And closing a shared transport is a no-op, since the channel and router belong to the peer. The message-ID generator comes from the peer's server-initiated space, so a borrowed call cannot collide with the remote peer's own client-initiated IDs on the same stream.
With stream deduplication on, the lower-ID peer of each pair is the only dialer. The higher-ID peer's outbound node is therefore born shared: when the configuration is built, it borrows the matching inbound peer's transport rather than dialing one of its own, and it does so before that peer has connected. Because the borrowed channel reference is the peer's own, a later reconnect is picked up without rebuilding the node. Borrowing is validated at construction. A shared node routes its calls onto the peer's channel, so the borrowed peer must be the same process this node addresses: an ID that maps to no configured peer, or to a peer at a different address, would silently carry calls elsewhere. WithPeers derives both sets from one node source, but Config.Extend can add outbound nodes from another, so the check belongs here rather than at the option. Node.IsShared reports the topology, so a deployment can tell a borrowed node from one that owns its connection.
A quorum call marshals its request once and sends the same message, carrying one client-initiated ID, to every node. That is safe while each node owns its stream, but a shared node sends on a stream that also carries the remote peer's own client-initiated IDs, where that ID could collide. A shared node now gets its own message, with an ID drawn from the peer's server-initiated space. Every other node keeps the single shared message, so the common case still marshals and constructs once.
WithStreamDedup makes each pair of peers share one connection for calls in both directions, instead of one connection per direction. It applies to the peer configuration built by WithPeers, where both sides are known and symmetric. The asymmetry it introduces is deliberate and needs to be visible to callers. Only the lower-ID peer of each pair dials, so calls to a lower-ID peer report ErrStreamDown until that peer has connected, and again if the shared stream later drops. ErrStreamDown documents both cases. WaitForAll blocks until every peer is connected and returns the peer configuration, which is the startup synchronization dedup needs and dual mode does not: without dedup, connections are established on demand, so it returns immediately. It rejects a setup it can never satisfy — a server with no node ID, or with an ID absent from its own peer list — rather than waiting for a condition that cannot occur. The tests cover the topology directly: nodes are born shared before any peer connects, borrowing is rejected for an unconfigured or misaddressed peer, node identity survives WaitForAll, and per-message metadata and request ordering hold over a shared stream as they do over a dedicated one.
The user guide gains the section a caller needs to enable it safely: which option turns it on and what it requires, that WaitForAll is the startup synchronization, and that calls to a lower-ID peer report ErrStreamDown until that peer connects. The topology is fixed when a configuration is created, so configurations retained beforehand stay valid. The developer guide records how it works underneath: born-shared outbound nodes, why a borrower cannot dial, why the owner reconnects eagerly, and why shared calls draw from the server-initiated message ID space.
Package conn has no Server type, so [Server.WaitForAll] resolved to nothing and rendered as a dead link. Name the method in plain prose instead; the server that owns it lives in the parent package.
The reader goroutines spun on a select default with no yield, one per server. The largest case starts 50 of them and holds every core flat out for the whole of WaitForAll, which is the operation the test is timing.
The new "Stream Deduplication for Symmetric Peers" section took over the "Send Queue Capacity and Backpressure" heading instead of adding its own, leaving the send-queue text as an unheaded tail of the dedup section and dropping its entry from the table of contents.
meling
force-pushed
the
gorums/stream-dedup
branch
from
August 12, 2026 12:38
d88e431 to
b8315fe
Compare
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
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.
Adds stream deduplication: each pair of symmetric peers shares one bidirectional stream for calls in both directions, instead of one connection per direction.
Enable it with
WithStreamDedupon a server configured withWithPeers. The lower-ID peer of each pair is the only dialer; the higher-ID peer's outbound node is born shared, borrowing the matching inbound peer's transport when the configuration is built, before that peer has connected. Because the borrowed channel reference is the peer's own, a reconnect is picked up without rebuilding the node.The asymmetry this introduces is visible to callers by design. Calls to a lower-ID peer report
ErrStreamDownuntil that peer has connected, and again if the shared stream later drops.Server.WaitForAllis the startup synchronization: it blocks until every peer is connected and returns the peer configuration, and it rejects a setup it can never satisfy — a server with no node ID, or with an ID absent from its own peer list — rather than waiting for a condition that cannot occur. Without dedup it returns immediately, since connections are established on demand.Three mechanisms make it safe:
ErrStreamDownrather than being silently dropped.The four commits layer it: the shared transport, the born-shared node construction that uses it, the server-space message ID, then the public option and its tests.
Verification:
go test ./... -count=2,go test -C examples ./...,go vet -tags=integration ./...,gofmt -l,make goplscheck.Last of the six core PRs. The three that follow add the benchmarking tooling.