hubcio commented on code in PR #3399:
URL: https://github.com/apache/iggy/pull/3399#discussion_r3333076520
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -1087,6 +1088,105 @@ where
}
}
+ /// Submit a replicated client request from in-process and await the
+ /// committed reply.
+ ///
+ /// A peer (home) shard relays a client's replicated request here (shard
+ /// 0 owns the metadata consensus group) and awaits the full committed
+ /// reply over the pipeline subscriber. The home shard then writes the
+ /// reply to the originating socket -- it holds the connection and the
+ /// `vsr -> transport` mapping that this side cannot reconstruct.
+ ///
+ /// Mirrors [`Self::submit_register_in_process`] but: (1) uses
+ /// `request_preflight` (dedup / session check) instead of the register
+ /// gate, (2) returns the committed `Message<ReplyHeader>` (body = state
+ /// machine output) rather than just the commit op.
+ ///
+ /// # Errors
+ /// `NotPrimary` / `NotCaughtUp` when this node cannot accept the
+ /// prepare, `InProgress` / `PipelineFull` on pipeline pressure,
+ /// `Canceled` when preflight absorbed the request (dedup / eviction /
+ /// gap) or the pending prepare was canceled before commit.
+ ///
+ /// # Panics
+ /// On a shard without consensus (only shard 0 owns the metadata
+ /// consensus group); callers must route here only on shard 0.
+ #[allow(clippy::future_not_send)]
+ pub async fn submit_request_in_process(
+ &self,
+ message: Message<RequestHeader>,
+ ) -> Result<Message<ReplyHeader>, RegisterSubmitError> {
+ let request_header = *message.header();
+ let client_id = request_header.client;
+ let session = request_header.session;
+ let request = request_header.request;
+
+ let consensus = self
+ .consensus
+ .as_ref()
+ .expect("submit_request_in_process: consensus only exists on shard
0");
+
+ if !is_caught_up_primary(consensus) {
+ return Err(
+ if consensus.is_primary() && consensus.is_normal() &&
!consensus.is_syncing() {
+ RegisterSubmitError::NotCaughtUp
+ } else {
+ RegisterSubmitError::NotPrimary
+ },
+ );
+ }
+
+ // Dedup / session / eviction. `false` = absorbed (duplicate cached
+ // reply already resent, or evicted, or gap). Surface as Canceled so
+ // the home shard stays silent and the SDK replays.
+ if !request_preflight(consensus, &self.client_table, client_id,
session, request).await {
Review Comment:
heads up - this `request_preflight` path makes a pre-existing reply-routing
bug newly reachable for replicated client requests.
on a `Duplicate` (cached reply) or eviction (`NoSession` / `SessionTooLow`),
preflight resends via `send_to_client(client_id, ...)` in `metadata_helpers.rs`
(~lines 83-90, 95, 103). but `client_id` here is the vsr consensus id (the
random uuid the sdk mints), and `send_to_client` routes by
`client_id_owning_shard` = the top 16 bits, which only carry home-shard bits
for transport ids. a vsr uuid's top bits are random, so the resend misroutes to
a garbage shard -> no registry slot -> dropped. bites single-shard too (top16
of a random uuid is ~never 0).
effect: client loses the first reply, the sdk transparently retries the same
`(client, request)`, the server sees `Duplicate`, resends -> misroutes again ->
client hangs/livelocks until the session times out. eviction is the same - the
client never learns it was evicted.
the fresh-commit path already does this right (home shard writes the
committed reply via `send_to_client(transport_client_id, ...)`). the
dedup/eviction resends need the same treatment - return the
`Duplicate`/`Evicted` outcome to the home shard and let it resend the cached
body by transport id, since shard 0 can't reconstruct the vsr->transport
mapping.
not exercised by the tier-1 tests this pr enables (single-shard tcp happy
path, `data_integrity` gated off), so it doesn't break the suite - but it ships
in the server path. minimum bar: a TODO at the resend sites + bound the
reply-wait so a retry fails fast instead of leaving an orphaned pump parked on
`rx.recv()`. full fix before vsr client traffic goes past tier-1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
