numinnex commented on code in PR #3399:
URL: https://github.com/apache/iggy/pull/3399#discussion_r3335947144
##########
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:
Added 3 TODOs, will address some of those in the next PR that suppose to do
some cleanup on the most recent work.
--
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]
