You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/07/28 11:16:44 UTC
(iggy) 02/04: temp
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch state_transfer_metadata_clients_table
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 71aca79067319a909adc68a76bfee5604256a526
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Tue Jul 28 10:42:06 2026 +0200
temp
---
core/common/src/error/iggy_error.rs | 3 +
core/configs/src/server_config/sharding.rs | 28 +++++
core/consensus/src/plane_helpers.rs | 33 ++++++
core/integration/tests/server/cg_vsr.rs | 85 +++++++++++++++
.../tests/server/concurrent_addition.rs | 12 +-
core/message_bus/src/transports/quic.rs | 59 ++++++++++
core/metadata/src/impls/metadata.rs | 86 +++++++++++++++
core/sdk/src/quic/quic_client.rs | 72 ++++++++++++
core/sdk/src/tcp/tcp_client.rs | 121 +++++++++++++++++++++
core/sdk/src/vsr.rs | 7 ++
core/sdk/src/websocket/websocket_client.rs | 4 +
core/server-ng/src/auth.rs | 21 ++++
core/server-ng/src/responses.rs | 40 +++++++
core/shard/src/lib.rs | 7 ++
vsr-server-test-status.md | 96 ++++++++++++++++
15 files changed, 672 insertions(+), 2 deletions(-)
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index 4d9d8c0bf..f2b515c94 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -122,8 +122,11 @@ pub enum IggyError {
InvalidPersonalAccessTokenExpiry = 56,
#[error("Request transiently not committed; retry")]
TransientNotCommitted = 57,
+<<<<<<< Updated upstream
#[error("Request transiently not accepted; retry, on any replica")]
TransientNotAccepted = 58,
+=======
+>>>>>>> Stashed changes
#[error("Not connected")]
NotConnected = 61,
#[error("Client shutdown")]
diff --git a/core/configs/src/server_config/sharding.rs b/core/configs/src/server_config/sharding.rs
index 87a411e1f..edef0f6a3 100644
--- a/core/configs/src/server_config/sharding.rs
+++ b/core/configs/src/server_config/sharding.rs
@@ -20,11 +20,39 @@ use serde::{Deserialize, Serialize};
use super::defaults::SERVER_CONFIG;
use configs::ConfigEnv;
+<<<<<<< Updated upstream
// `CpuAllocation`/`NumaConfig` are pure config types and live in their own
// leaf crate so both `configs` and `shard_allocator` can share them without
// pulling each other's heavier dependency trees. Re-exported here to keep the
// `configs::sharding::*` path stable for existing callers.
pub use cpu_allocation::{CpuAllocation, NumaConfig};
+=======
+/// Default capacity of the per-shard inter-shard inbox channel. Sized
+/// comfortably above the consensus working set, which is roughly
+/// `PIPELINE_PREPARE_QUEUE_MAX (= 32) * replica_count * directions`
+/// frames in flight per shard, without allowing a runaway producer to
+/// eat unbounded memory. Tunable via `[system.sharding] inbox_capacity`
+/// in TOML.
+///
+/// The capacity must also absorb the worst-case cross-shard client
+/// Reply burst. Unlike consensus frames, client Replies have no VSR
+/// retransmit path: a Reply lost on full inbox is gone and the client
+/// times out. A reasonable lower bound is
+/// `max_inflight_client_requests / num_shards` (assuming requests are
+/// distributed evenly across owning shards) plus the consensus
+/// headroom above.
+///
+/// Consensus frames and client-reply forwards share this one channel,
+/// so the two headrooms are not independent: a consensus burst or
+/// retransmit storm can fill the inbox with consensus frames exactly
+/// when a client Reply needs the space. A single `inbox_capacity` knob
+/// cannot isolate the two frame classes - size it for the sum of both
+/// worst cases occurring together. Watch the drop-site `tracing` logs
+/// (and, once a per-shard exporter lands, the `frame_drops_total`
+/// `{variant="forward_client_send"}` counter) to detect when the bound
+/// is too low in production.
+pub const DEFAULT_INBOX_CAPACITY: usize = 1024;
+>>>>>>> Stashed changes
/// Sharding config for the legacy `core/server`. That server consumes only
/// `cpu_allocation` and `pin_cores`; the bus / shutdown / reconcile knobs are
diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs
index c0694bcc4..f261e8a9c 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -359,6 +359,7 @@ where
.expect("reply buffer must contain a valid reply message")
}
+<<<<<<< Updated upstream
/// Builds a `Reply` carrying only a single-entry result section, no payload:
/// the generic `[count=1][index=0][result=code]` rejection frame.
///
@@ -374,12 +375,31 @@ where
/// `commit` is the primary's current commit position and is informational
/// here: a rejection reply is sent, never cached in the `ClientTable`, so it
/// is not subject to the `commit_reply` regression order.
+=======
+/// Builds a `Reply` carrying only a metadata result section, no payload.
+///
+/// For a request that could not be committed *right now* (not-caught-up /
+/// in-flight / pipeline-full / view-change cancel): the body is
+/// `[count=1][index=0][result=code]`. The SDK decodes the nonzero `result` to a
+/// typed retryable error (`IggyError::TransientNotCommitted`) and replays the
+/// same `request_id` immediately, instead of waiting out its response
+/// read-timeout.
+///
+/// Stamped from the request header (no prepare exists for an uncommitted op).
+/// `commit` is the primary's current commit position and is informational here:
+/// a transient reply is sent, never cached in the `ClientTable`, so it is not
+/// subject to the `commit_reply` regression order.
+>>>>>>> Stashed changes
///
/// # Panics
/// If the constructed message buffer is not a valid reply.
#[must_use]
#[allow(clippy::cast_possible_truncation)]
+<<<<<<< Updated upstream
pub fn build_result_rejection_reply(
+=======
+pub fn build_transient_reply(
+>>>>>>> Stashed changes
request_header: &RequestHeader,
commit: u64,
code: u32,
@@ -391,7 +411,13 @@ pub fn build_result_rejection_reply(
let total_size = header_size + RESULT_BODY_LEN;
let mut buffer = bytes::BytesMut::zeroed(total_size);
+<<<<<<< Updated upstream
let header = ReplyHeader {
+=======
+ let header = bytemuck::checked::try_from_bytes_mut::<ReplyHeader>(&mut buffer[..header_size])
+ .expect("zeroed bytes are valid");
+ *header = ReplyHeader {
+>>>>>>> Stashed changes
cluster: request_header.cluster,
size: total_size as u32,
view: request_header.view,
@@ -400,10 +426,14 @@ pub fn build_result_rejection_reply(
replica: request_header.replica,
request_checksum: request_header.request_checksum,
client: request_header.client,
+<<<<<<< Updated upstream
// Position-typed like the sibling builders (`build_reply_from_request`
// stamps `op: commit` too); inert on this path -- rejections are never
// cached -- but keeps the wire field convention for frame inspection.
op: commit,
+=======
+ op: request_header.session,
+>>>>>>> Stashed changes
commit,
timestamp: request_header.timestamp,
request: request_header.request,
@@ -411,7 +441,10 @@ pub fn build_result_rejection_reply(
namespace: request_header.namespace,
..Default::default()
};
+<<<<<<< Updated upstream
buffer[..header_size].copy_from_slice(bytemuck::bytes_of(&header));
+=======
+>>>>>>> Stashed changes
let body = &mut buffer[header_size..];
body[0..4].copy_from_slice(&1u32.to_le_bytes());
diff --git a/core/integration/tests/server/cg_vsr.rs b/core/integration/tests/server/cg_vsr.rs
new file mode 100644
index 000000000..ecf595a89
--- /dev/null
+++ b/core/integration/tests/server/cg_vsr.rs
@@ -0,0 +1,85 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Consumer-group scenarios that run against server-ng (vsr): round-robin
+//! membership (join) plus group polling (single- and multi-client), where the
+//! server picks each member's partition by advancing its round-robin cursor.
+
+use crate::server::scenarios::{
+ consumer_group_auto_commit_reconnection_scenario,
+ consumer_group_duplicate_name_create_scenario, consumer_group_join_scenario,
+ consumer_group_new_messages_after_restart_scenario, consumer_group_offset_cleanup_scenario,
+ consumer_group_with_multiple_clients_polling_messages_scenario,
+ consumer_group_with_single_client_polling_messages_scenario,
+};
+use integration::iggy_harness;
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn duplicate_name_create_preserves_live_group(harness: &TestHarness) {
+ consumer_group_duplicate_name_create_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn join(harness: &TestHarness) {
+ consumer_group_join_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn single_client(harness: &TestHarness) {
+ consumer_group_with_single_client_polling_messages_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn multiple_clients(harness: &TestHarness) {
+ consumer_group_with_multiple_clients_polling_messages_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn offset_cleanup(harness: &TestHarness) {
+ consumer_group_offset_cleanup_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn auto_commit_reconnection(harness: &TestHarness) {
+ consumer_group_auto_commit_reconnection_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn new_messages_after_restart(harness: &TestHarness) {
+ consumer_group_new_messages_after_restart_scenario::run(harness).await;
+}
diff --git a/core/integration/tests/server/concurrent_addition.rs b/core/integration/tests/server/concurrent_addition.rs
index ff1a4347e..e590078f1 100644
--- a/core/integration/tests/server/concurrent_addition.rs
+++ b/core/integration/tests/server/concurrent_addition.rs
@@ -37,12 +37,19 @@ use test_case::test_matrix;
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
))]
-#[test_matrix(
+// vsr excludes HTTP only (server-ng exposes no HTTP listener).
+#[cfg_attr(not(feature = "vsr"), test_matrix(
[tcp(), http(), quic(), websocket()],
[user(), stream(), topic(), partition(), consumer_group()],
[hot(), cold()],
[barrier_on(), barrier_off()]
-)]
+))]
+#[cfg_attr(feature = "vsr", test_matrix(
+ [tcp(), quic(), websocket()],
+ [user(), stream(), topic(), partition(), consumer_group()],
+ [hot(), cold()],
+ [barrier_on(), barrier_off()]
+))]
async fn matrix(
harness: &TestHarness,
transport: TransportProtocol,
@@ -57,6 +64,7 @@ fn tcp() -> TransportProtocol {
TransportProtocol::Tcp
}
+#[cfg(not(feature = "vsr"))]
fn http() -> TransportProtocol {
TransportProtocol::Http
}
diff --git a/core/message_bus/src/transports/quic.rs b/core/message_bus/src/transports/quic.rs
index 8fdf49e83..1feed7924 100644
--- a/core/message_bus/src/transports/quic.rs
+++ b/core/message_bus/src/transports/quic.rs
@@ -33,6 +33,7 @@
//! 2. read exactly one inbound frame from the `RecvStream`, note its
//! `request` id, and forward it to [`ActorContext::in_tx`];
//! 3. wait on [`ActorContext::rx`] for the reply whose `request` id matches,
+<<<<<<< Updated upstream
//! discarding any stale/duplicate reply left by an earlier attempt's bidi
//! (the SDK replays an explicit transient with the SAME request id); write
//! the matched reply, then `finish()` to deliver pending data + FIN;
@@ -55,6 +56,26 @@
//! pending reply with it -- nothing can strand or cross request ids. A
//! future pipelining upgrade would additionally run per-bidi handlers
//! concurrently (and would then require a real per-op correlation id).
+=======
+//! discarding any stale/duplicate reply left by an abandoned bidi; on
+//! [`REPLY_WAIT_TIMEOUT`] abandon this bidi (the dispatch stayed silent on
+//! a transient submit, expecting a replay) and accept the next; write the
+//! matched reply, then `finish()` to deliver pending data + FIN;
+//! 4. drop the bidi pair and loop to the next `accept_bi`.
+//!
+//! The `accept_bi` loop is still serial (the iggy SDK serialises per
+//! connection via an internal `RwLock`, so one bidi at a time is a semantic
+//! match), but the reply wait is now `request_id`-KEYED rather than "next
+//! frame on `rx`". This matters because the server stays deliberately silent
+//! on a transient submit (it expects the SDK to replay): an unkeyed,
+//! untimed wait would block this bidi - and, serial, the whole connection -
+//! forever, so the replay's fresh bidi could never be accepted (acute at
+//! `max_concurrent_bidi_streams=1`). Keying + the timeout let a transient
+//! bidi be abandoned safely: the SDK replays with the SAME request id, so a
+//! late reply for the abandoned request matches the replay's bidi and is
+//! consumed there, never mis-delivered to a later request. A future
+//! pipelining upgrade would additionally run per-bidi handlers concurrently.
+>>>>>>> Stashed changes
//!
//! The handshake itself is driven by [`accept_handshake`] (no bidi
//! captured). To avoid a single slow / hostile peer wedging the entire
@@ -291,6 +312,7 @@ pub(crate) const DEFAULT_CLOSE_GRACE: Duration = Duration::from_secs(2);
/// A replicated submit the server cannot yet commit is answered with silence
/// (the server expects the SDK to replay), so without a bound the per-bidi
/// wait - and, since `accept_bi` is serial, the whole connection - would wedge:
+<<<<<<< Updated upstream
/// Backstop for a bidi whose dispatch never produced a reply (a true drop:
/// both partition queues full, or a pathologically slow commit). Longer than
/// the SDK's whole-request deadline (30s), so the client always times out
@@ -300,6 +322,14 @@ pub(crate) const DEFAULT_CLOSE_GRACE: Duration = Duration::from_secs(2);
/// the next data-plane op and shift every subsequent reply off by one.
/// Closing the connection drops the mailbox and any late reply with it.
const REPLY_WAIT_BACKSTOP: Duration = Duration::from_mins(1);
+=======
+/// the SDK's replay opens a fresh bidi that, at `max_concurrent_bidi_streams=1`,
+/// can only be accepted once this one is released. Kept below the SDK's
+/// per-attempt retry interval so the slot frees before the client replays; a
+/// late reply for the abandoned request carries the same request id and is
+/// matched (and consumed) by the replay's bidi, so abandoning early is safe.
+const REPLY_WAIT_TIMEOUT: Duration = Duration::from_secs(2);
+>>>>>>> Stashed changes
/// `Command2` byte offset shared by every consensus header (after the
/// fixed checksum/cluster/size/view/release prefix).
@@ -430,7 +460,11 @@ impl TransportConn for QuicTransportConn {
};
// Request id this bidi is waiting on. Used to match its reply and
// to discard stale/duplicate replies left by an abandoned bidi.
+<<<<<<< Updated upstream
// `None` (unparsable) degrades to accepting the first reply.
+=======
+ // `None` (unparseable) degrades to accepting the first reply.
+>>>>>>> Stashed changes
let request_id = req
.as_slice()
.get(..HEADER_SIZE)
@@ -444,6 +478,7 @@ impl TransportConn for QuicTransportConn {
// Wait for THIS bidi's reply, keyed by request id. The bus `rx` is
// a single unkeyed per-connection channel, so filter by id and
+<<<<<<< Updated upstream
// discard replies belonging to an earlier attempt's bidi (an
// explicit-transient replay reuses the SAME request id, so a late
// reply matches and is consumed by the replay's bidi - never
@@ -453,11 +488,24 @@ impl TransportConn for QuicTransportConn {
// pathologically slow. `REPLY_WAIT_BACKSTOP` (see its doc) closes
// the whole connection instead of abandoning the bidi, so a late
// reply can never strand and shift onto the next data-plane op.
+=======
+ // discard replies belonging to an abandoned earlier bidi (the SDK
+ // replays with the SAME request id, so a late reply matches and is
+ // consumed by the replay's bidi - never mis-delivered to a later
+ // request). On timeout the dispatch produced no reply (a transient
+ // submit: the server stays silent expecting a replay); abandon this
+ // bidi and accept the next so the replay's fresh stream is not
+ // blocked behind us (acute at `max_concurrent_bidi_streams=1`).
+>>>>>>> Stashed changes
let mut matched: Option<BusMessage> = None;
let mut connection_gone = false;
{
let mut reply_deadline =
+<<<<<<< Updated upstream
std::pin::pin!(compio::time::sleep(REPLY_WAIT_BACKSTOP).fuse());
+=======
+ std::pin::pin!(compio::time::sleep(REPLY_WAIT_TIMEOUT).fuse());
+>>>>>>> Stashed changes
loop {
futures::select! {
() = shutdown_fut.as_mut() => {
@@ -466,8 +514,12 @@ impl TransportConn for QuicTransportConn {
break;
}
() = reply_deadline.as_mut() => {
+<<<<<<< Updated upstream
warn!(%label, %peer, ?request_id, "quic: no reply within backstop; closing connection so no reply can strand");
connection_gone = true;
+=======
+ debug!(%label, %peer, ?request_id, "quic: no reply within timeout; abandoning bidi for client replay");
+>>>>>>> Stashed changes
break;
}
res = rx.recv().fuse() => if let Ok(m) = res {
@@ -496,10 +548,17 @@ impl TransportConn for QuicTransportConn {
break;
}
let Some(first) = matched else {
+<<<<<<< Updated upstream
// Unreachable in practice: every no-reply path above sets
// `connection_gone`. Close the send half defensively.
if let Err(e) = send.finish() {
debug!(%label, %peer, error = ?e, "quic: send.finish() failed with no reply");
+=======
+ // Timeout: close our send half so the SDK's read ends, then
+ // accept the next bidi (the client's replay).
+ if let Err(e) = send.finish() {
+ debug!(%label, %peer, error = ?e, "quic: send.finish() failed after reply timeout");
+>>>>>>> Stashed changes
}
continue;
};
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 7fa76a4d1..159d4be6d 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -29,7 +29,11 @@ use consensus::{
PipelineEntry, Plane, PlaneIdentity, PlaneKind, PreflightOutcome, Project, ReplicaLogContext,
RequestLogEvent, Sequencer, SimEventKind, VsrConsensus, ack_preflight, ack_quorum_reached,
apply_preflight_consensus_plane, build_eviction_message, build_reply_message,
+<<<<<<< Updated upstream
build_reply_message_with, build_result_rejection_reply, emit_sim_event,
+=======
+ build_reply_message_with, build_transient_reply, drain_committable_prefix, emit_sim_event,
+>>>>>>> Stashed changes
fence_old_prepare_by_commit, is_caught_up_primary,
panic_if_hash_chain_would_break_in_same_view, peek_committable_head, pipeline_prepare_common,
register_preflight, replicate_preflight, replicate_to_next_in_chain, request_preflight,
@@ -785,6 +789,7 @@ where
// Durable here but not yet committed, and the primary is retransmitting
// it: our original PrepareOk was lost (e.g. the primary's inbox
// overflowed under a client burst). Re-forward the tail down the chain
+<<<<<<< Updated upstream
// so a downstream replica that missed it recovers, then re-ack ONLY
// the retransmitted op. The primary's retransmit cycle walks every
// un-acked op in the window (`retransmit_targets`), so a lost ack for
@@ -793,6 +798,16 @@ where
// backups, which can overflow the primary's inbox -- the very failure
// this path recovers from. Both downstream and primary are idempotent
// on a duplicate (replica, op).
+=======
+ // so a downstream replica that missed it recovers, then re-ack every
+ // op in our uncommitted prepared suffix - not just this one. Acks are
+ // tracked per-op (see `ack_quorum_reached`), and the primary's commit
+ // frontier only advances over a contiguous quorum-acked prefix, so a
+ // lost ack for a lower op keeps that prefix pinned even after we re-ack
+ // the tail. Re-acking the whole suffix the backup holds restores
+ // quorum on the gap. Bounded by the pipeline depth; both downstream
+ // and primary are idempotent on a duplicate (replica, op).
+>>>>>>> Stashed changes
#[allow(clippy::cast_possible_truncation)]
if journal.handle().header(header.op as usize).is_some() {
warn!(
@@ -803,10 +818,23 @@ where
op = header.op,
commit = consensus.commit_max(),
operation = ?header.operation,
+<<<<<<< Updated upstream
"journal already holds prepare, re-forwarding + re-acking it"
);
self.replicate(&message).await;
self.send_prepare_ok(&header).await;
+=======
+ "journal already holds prepare, re-forwarding + re-acking suffix"
+ );
+ self.replicate(&message).await;
+ for op in (consensus.commit_max() + 1)..=header.op {
+ if let Some(suffix_header) =
+ journal.handle().header(op as usize).map(|header| *header)
+ {
+ self.send_prepare_ok(&suffix_header).await;
+ }
+ }
+>>>>>>> Stashed changes
return;
}
@@ -1603,6 +1631,7 @@ where
.as_ref()
.expect("submit_request_in_process: consensus only exists on shard 0");
+<<<<<<< Updated upstream
// Not-primary is transient: the same request replayed against the
// current primary commits fine. Reply with the explicit transient
// frame (relayed to the socket by the home shard) so the client
@@ -1622,6 +1651,19 @@ where
&request_header,
consensus.commit_max(),
IggyError::TransientNotAccepted.as_code(),
+=======
+ // Not-primary / not-caught-up is transient: the same request replayed
+ // once a primary is caught up commits fine. Reply with the explicit
+ // `TransientNotCommitted` frame (relayed to the socket by the home shard)
+ // so the client replays immediately rather than waiting out its
+ // read-timeout. The frame keeps the lockstep TCP/WS stream in framing
+ // sync, so the client resends on the same connection (session intact).
+ if !is_caught_up_primary(consensus) {
+ return Ok(build_transient_reply(
+ &request_header,
+ consensus.commit_max(),
+ IggyError::TransientNotCommitted.as_code(),
+>>>>>>> Stashed changes
)
.into_generic());
}
@@ -1631,6 +1673,7 @@ where
// routing), so a Replay/Evict/NotReady is returned to the home shard as
// the reply -- `handle_client_request` writes it to the originating
// socket by transport id, exactly like a fresh commit. Drop (client-bug
+<<<<<<< Updated upstream
// already-applied / future-epoch) surfaces as Canceled so the home
// shard stays silent.
match request_preflight(
@@ -1641,6 +1684,10 @@ where
request,
request_checksum,
) {
+=======
+ // stale/gap) surfaces as Canceled so the home shard stays silent.
+ match request_preflight(consensus, &self.client_table, client_id, session, request) {
+>>>>>>> Stashed changes
PreflightOutcome::Dispatch => {}
PreflightOutcome::Replay(reply) => {
return server_common::Message::<GenericHeader>::try_from(
@@ -1657,7 +1704,11 @@ where
// In-flight prepare from this client: replaying the same request_id
// is absorbed until the original commits, then served from cache.
PreflightOutcome::NotReady => {
+<<<<<<< Updated upstream
return Ok(build_result_rejection_reply(
+=======
+ return Ok(build_transient_reply(
+>>>>>>> Stashed changes
&request_header,
consensus.commit_max(),
IggyError::TransientNotCommitted.as_code(),
@@ -1667,6 +1718,7 @@ where
PreflightOutcome::Drop => return Err(MetadataSubmitError::Canceled),
}
+<<<<<<< Updated upstream
// Prepare queue full: backpressure, not failure. Absorb into the
// request queue with this caller's reply subscriber; the commit
// path promotes it as slots free up and the await below resolves
@@ -1699,6 +1751,16 @@ where
)
.into_generic()),
};
+=======
+ // Pipeline full: backpressure, not failure. Tell the client to replay.
+ if consensus.pipeline().borrow().is_full() {
+ return Ok(build_transient_reply(
+ &request_header,
+ consensus.commit_max(),
+ IggyError::TransientNotCommitted.as_code(),
+ )
+ .into_generic());
+>>>>>>> Stashed changes
}
// The acting-user RBAC stamp lives in the shared `prepare_request`
@@ -1719,7 +1781,11 @@ where
// so reply with the transient frame rather than staying silent.
match self.dispatch_prepare_and_await(consensus, prepare).await {
Ok(reply) => Ok(reply.into_generic()),
+<<<<<<< Updated upstream
Err(Canceled) => Ok(build_result_rejection_reply(
+=======
+ Err(Canceled) => Ok(build_transient_reply(
+>>>>>>> Stashed changes
&request_header,
consensus.commit_max(),
IggyError::TransientNotCommitted.as_code(),
@@ -1809,7 +1875,11 @@ where
let Some(consensus) = self.consensus.as_ref() else {
return;
};
+<<<<<<< Updated upstream
if !consensus.is_primary() || !consensus.is_normal() || consensus.is_transferring() {
+=======
+ if !consensus.is_primary() || !consensus.is_normal() || consensus.is_syncing() {
+>>>>>>> Stashed changes
return;
}
let Some(journal) = self.journal.as_ref() else {
@@ -1842,6 +1912,7 @@ where
return;
}
+<<<<<<< Updated upstream
// Interleave push + drain per header instead of push-all-then-drain-once:
// each `on_ack` below can promote a full window of buffered requests
// (`drain_request_queue_into_prepares`), and every promoted prepare
@@ -1874,6 +1945,14 @@ where
return false;
}
for message in loopback.drain(..) {
+=======
+ for header in &headers {
+ self.send_prepare_ok(header).await;
+ }
+ let mut loopback = Vec::new();
+ consensus.drain_loopback_into(&mut loopback);
+ for message in loopback {
+>>>>>>> Stashed changes
match message.header().command {
Command2::PrepareOk => match message.try_into_typed::<PrepareOkHeader>() {
Ok(prepare_ok) => self.on_ack(prepare_ok).await,
@@ -1888,6 +1967,7 @@ where
),
}
}
+<<<<<<< Updated upstream
true
}
@@ -2140,6 +2220,12 @@ where
/// Promote buffered requests into free prepare slots after a commit
/// batch drains.
+=======
+ }
+
+ /// Promote up to `slots_freed` buffered requests into prepares after
+ /// `on_ack` commits a prefix.
+>>>>>>> Stashed changes
///
/// # Safety
/// Re-preflight per iteration: `commit_journal` may have advanced the
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index 9a731f608..6086e5234 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -63,6 +63,17 @@ const NAME: &str = "Iggy";
/// under lock per send, every later request on the client.
const RESPONSE_READ_TIMEOUT: Duration = Duration::from_secs(30);
+<<<<<<< Updated upstream
+=======
+/// Per-attempt response read timeout for the same-connection transient resend
+/// loop (see `send_raw`). A replicated submit the server can't yet commit is
+/// answered with silence; rather than wait the full `RESPONSE_READ_TIMEOUT`
+/// once, abandon the unanswered bidi after this shorter interval and resend the
+/// same request on a fresh stream. Bounded overall by `RESPONSE_READ_TIMEOUT`.
+#[cfg(feature = "vsr")]
+const TRANSIENT_RETRY_INTERVAL: Duration = Duration::from_secs(3);
+
+>>>>>>> Stashed changes
/// Backoff before replaying a request the server answered with an explicit
/// `TransientNotCommitted` frame (not-caught-up / in-flight / pipeline-full /
/// view-change cancel). Unlike a silent timeout, this reply arrives promptly, so
@@ -177,6 +188,13 @@ impl BinaryTransport for QuicClient {
#[cfg(feature = "vsr")]
if skip_auto_login {
*self.skip_auto_login_once.lock().await = true;
+ // The replayed login/register must mint a fresh Register: the failed
+ // attempt already consumed the one-shot register request id and may
+ // have half-bound the session.
+ *self
+ .consensus_session
+ .lock()
+ .expect("consensus session mutex poisoned") = ConsensusSession::new();
}
let server_address = self.current_server_address.lock().await.to_string();
info!(
@@ -723,6 +741,7 @@ impl QuicClient {
crate::vsr::encode_request_header(&mut consensus_session, code, &payload)?
};
trace!("Sending a QUIC VSR request of size {request_size} with code: {code}");
+<<<<<<< Updated upstream
// Same-connection transient resend, gated on the EXPLICIT
// `TransientNotCommitted` frame only. The server answers every
// transient submit with that frame (it is pre-commit by
@@ -736,6 +755,20 @@ impl QuicClient {
// back as terminal `ConsumerOffsetNotFound`). A silent deadline
// expiry surfaces `Disconnected` and takes the
// reconnect path in `send_raw_with_response`, same as TCP.
+=======
+ // Same-connection transient resend. A replicated submit the
+ // server cannot yet commit is answered with silence - it expects
+ // the client to replay. The connection and login session are
+ // intact; only this bidi went unanswered. So abandon it after a
+ // short read timeout and resend the SAME request header (same
+ // request id, so the server dedups/commits it idempotently) on a
+ // fresh bidi, bounded overall by `RESPONSE_READ_TIMEOUT`. This
+ // needs no reconnect or relogin, so it recovers transients even
+ // with auto-login disabled. Login/register is one-shot (a replay
+ // must mint a fresh session), so it is left to the reconnect path
+ // in `send_raw_with_response` and not resent here.
+ let allow_resend = !is_login_register_code(code);
+>>>>>>> Stashed changes
let header_bytes = bytemuck::bytes_of(&request_header);
let deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
loop {
@@ -762,14 +795,23 @@ impl QuicClient {
if remaining.is_zero() {
return Err(IggyError::Disconnected);
}
+<<<<<<< Updated upstream
match QuicClient::handle_response(
&mut recv,
response_buffer_size as usize,
remaining,
+=======
+ let attempt_timeout = remaining.min(TRANSIENT_RETRY_INTERVAL);
+ match QuicClient::handle_response(
+ &mut recv,
+ response_buffer_size as usize,
+ attempt_timeout,
+>>>>>>> Stashed changes
)
.await
{
Ok(reply) => return Ok(reply),
+<<<<<<< Updated upstream
// `TransientNotCommitted` = the server replied with an
// explicit retry frame because it could not commit yet
// (not-caught-up / in-flight / pipeline-full /
@@ -787,6 +829,36 @@ impl QuicClient {
let remaining =
deadline.saturating_duration_since(tokio::time::Instant::now());
tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
+=======
+ // `Disconnected` = our read timed out (server never
+ // answered). `EmptyResponse` = the server abandoned the
+ // bidi (finished it with no body). `TransientNotCommitted`
+ // = the server replied with an explicit retry frame
+ // because it could not commit yet (not-caught-up /
+ // in-flight / pipeline-full / view-change cancel). All
+ // three mean "resend the same request on a fresh bidi";
+ // the session stays intact so no reconnect/relogin needed.
+ // Login/register resends on `TransientNotCommitted` too
+ // (nothing committed -> the same Register replays
+ // idempotently, no fresh session needed); for the silent
+ // paths login still falls through to the reconnect+relogin
+ // path in `send_raw_with_response`.
+ Err(
+ error @ (IggyError::Disconnected
+ | IggyError::EmptyResponse
+ | IggyError::TransientNotCommitted),
+ ) if (allow_resend || error == IggyError::TransientNotCommitted)
+ && tokio::time::Instant::now() < deadline =>
+ {
+ // The explicit frame returns promptly (no read
+ // timeout elapsed), so pace the replay; the silent
+ // paths already waited out `attempt_timeout`.
+ if error == IggyError::TransientNotCommitted {
+ let remaining = deadline
+ .saturating_duration_since(tokio::time::Instant::now());
+ tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
+ }
+>>>>>>> Stashed changes
warn!(
"QUIC request code {code} not committed (transient); resending on a new stream"
);
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index f827b9f55..c21e2f04c 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -72,6 +72,7 @@ const RESPONSE_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_sec
#[cfg(feature = "vsr")]
const NOT_READY_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50);
+<<<<<<< Updated upstream
/// How long a request replays `TransientNotCommitted` on the SAME connection
/// before the client re-checks cluster leadership. A node that stopped being
/// primary (view change while the connection stayed up) answers transient
@@ -81,6 +82,8 @@ const NOT_READY_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_
#[cfg(feature = "vsr")]
const TRANSIENT_FAILOVER_CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2);
+=======
+>>>>>>> Stashed changes
/// TCP client for interacting with the Iggy API.
/// It requires a valid server address.
#[derive(Debug)]
@@ -726,6 +729,7 @@ impl TcpClient {
}
#[cfg(feature = "vsr")]
+<<<<<<< Updated upstream
{
// One overall deadline bounds the request across transient replays
// AND leader failovers, matching the previous single-connection
@@ -782,10 +786,123 @@ impl TcpClient {
return Err(IggyError::Disconnected);
}
other => return other,
+=======
+ let consensus_session = self.consensus_session.clone();
+ // SAFETY: we run code holding the `stream` lock in a task so we can't be cancelled while holding the lock.
+ let result = tokio::spawn(async move {
+ let mut stream = stream.lock().await;
+ if let Some(stream) = stream.as_mut() {
+ #[cfg(feature = "vsr")]
+ {
+ // Encode the request header ONCE: `next_request_id` advances
+ // here, so a transient replay must reuse the same id for the
+ // server's dedup. The connection is lockstep (one request in
+ // flight per client), so a complete reply leaves the stream at
+ // a clean frame boundary -- a `TransientNotCommitted` answer
+ // (the server could not commit yet: not-caught-up / in-flight
+ // / pipeline-full / view-change cancel) lets us resend the
+ // SAME request on the SAME connection with no reconnect and
+ // the session intact. Bounded overall by RESPONSE_READ_TIMEOUT.
+ let (request_header, request_size) = {
+ let mut consensus_session = consensus_session
+ .lock()
+ .expect("consensus session mutex poisoned");
+ crate::vsr::encode_request_header(&mut consensus_session, code, &payload)?
+ };
+ trace!("Sending a TCP VSR request of size {request_size} with code: {code}");
+ let header_bytes = bytemuck::bytes_of(&request_header);
+ // One deadline bounds the whole request including transient
+ // replays, so a slow primary cannot stretch a single call
+ // beyond RESPONSE_READ_TIMEOUT.
+ let retry_deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
+ loop {
+ stream.write(header_bytes).await?;
+ if !payload.is_empty() {
+ stream.write(&payload).await?;
+ }
+ stream.flush().await?;
+ trace!("Sent a TCP request with code: {code}, waiting for a response...");
+
+ let mut response_header = [0u8; iggy_binary_protocol::HEADER_SIZE];
+ // `stream.read` delegates to `read_exact`; on success it
+ // always returns the requested length, so no short-read
+ // guard is needed here.
+ //
+ // Deadline guards against server-side reply loss (e.g. a
+ // stalled replication quorum that never commits the op):
+ // the connection is lockstep, so an unanswered read would
+ // hold the stream lock forever and wedge every later
+ // request on this client. On expiry drop the stream --
+ // a late reply would desync framing for the next request.
+ //
+ // One deadline spans BOTH the header and body reads: a
+ // reply that delivers a header then stalls must not get a
+ // fresh full timeout for the body.
+ let header_read = tokio::time::timeout_at(
+ retry_deadline,
+ stream.read(&mut response_header),
+ )
+ .await;
+ let Ok(header_read) = header_read else {
+ error!(
+ "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting for VSR response header for TCP request with code: {code}",
+ );
+ return Err(IggyError::Disconnected);
+ };
+ header_read.map_err(|error| {
+ error!(
+ "Failed to read VSR response header for TCP request with code: {code}: {error}",
+ );
+ IggyError::Disconnected
+ })?;
+
+ let response_size = crate::vsr::response_size(&response_header)?;
+ let body_size = response_size - iggy_binary_protocol::HEADER_SIZE;
+ let body = if body_size > 0 {
+ let mut body = BytesMut::with_capacity(body_size);
+ let body_read = tokio::time::timeout_at(
+ retry_deadline,
+ stream.read_buf(&mut body, body_size),
+ )
+ .await;
+ let Ok(body_read) = body_read else {
+ error!(
+ "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting for VSR response body for TCP request with code: {code}",
+ );
+ return Err(IggyError::Disconnected);
+ };
+ body_read.map_err(|error| {
+ error!(
+ "Failed to read VSR response body for TCP request with code: {code}: {error}",
+ );
+ IggyError::Disconnected
+ })?;
+ body.freeze()
+ } else {
+ Bytes::new()
+ };
+
+ match crate::vsr::decode_response_split(&response_header, body) {
+ // Server could not commit yet but answered with a
+ // complete frame; the lockstep stream is in sync, so
+ // replay the same request id on this connection after
+ // a short pause (no reconnect, session intact).
+ Err(IggyError::TransientNotCommitted)
+ if tokio::time::Instant::now() < retry_deadline =>
+ {
+ let remaining = retry_deadline
+ .saturating_duration_since(tokio::time::Instant::now());
+ tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
+ }
+ other => return other,
+ }
+ }
+>>>>>>> Stashed changes
}
}
}
+<<<<<<< Updated upstream
#[cfg(not(feature = "vsr"))]
{
let stream = self.stream.clone();
@@ -793,6 +910,10 @@ impl TcpClient {
let result = tokio::spawn(async move {
let mut stream = stream.lock().await;
if let Some(stream) = stream.as_mut() {
+=======
+ #[cfg(not(feature = "vsr"))]
+ {
+>>>>>>> Stashed changes
let payload_length = payload.len() + REQUEST_INITIAL_BYTES_LENGTH;
trace!("Sending a TCP request of size {payload_length} with code: {code}");
stream.write(&(payload_length as u32).to_le_bytes()).await?;
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 671fbd5e9..9451b9c09 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -274,12 +274,19 @@ fn split_metadata_result(operation: Operation, body: Bytes) -> Result<Bytes, Igg
// login decodes to `TransientNotCommitted` and the SDK replays it. The one
// exception is a terminal failure, which ships an empty body (no result
// section) and is passed through to fail the typed `LoginRegisterResponse`
+<<<<<<< Updated upstream
// decode. `Operation::is_result_framed` is the shared source of truth with
// the server-side encode sites; the Register empty-body-is-terminal nuance
// is the one SDK-side addition. Other reads, data-plane ops, and Logout
// carry no result section and pass through untouched.
let result_framed =
operation.is_result_framed() || (operation == Operation::Register && !body.is_empty());
+=======
+ // decode. Reads, the partition data plane, and Logout carry no result
+ // section and pass through untouched.
+ let result_framed =
+ operation.is_metadata() || (operation == Operation::Register && !body.is_empty());
+>>>>>>> Stashed changes
if !result_framed {
return Ok(body);
}
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index 537d68de2..5a8912c6f 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -774,7 +774,11 @@ impl WebSocketClient {
// Server could not commit yet but answered with a complete
// frame; the lockstep stream is in sync, so replay the same
// request id on this connection after a short pause.
+<<<<<<< Updated upstream
Err(IggyError::TransientNotCommitted | IggyError::TransientNotAccepted)
+=======
+ Err(IggyError::TransientNotCommitted)
+>>>>>>> Stashed changes
if tokio::time::Instant::now() < retry_deadline =>
{
let remaining =
diff --git a/core/server-ng/src/auth.rs b/core/server-ng/src/auth.rs
index 56f4cbea8..027e1b4e2 100644
--- a/core/server-ng/src/auth.rs
+++ b/core/server-ng/src/auth.rs
@@ -25,6 +25,7 @@ use crate::bootstrap::{ShellBus, ShellShard};
use crate::dispatch::submit_register_on_owner;
use crate::login_register::LoginRegisterError;
use crate::responses::{build_empty_reply, build_login_register_reply, current_metadata_commit};
+<<<<<<< Updated upstream
use crate::session_manager::{ClientSdkInfo, SessionManager};
use consensus::{MetadataHandle, build_result_rejection_reply};
use iggy_binary_protocol::PrepareHeader;
@@ -34,6 +35,13 @@ use iggy_common::defaults::{
};
use iggy_common::{IggyError, IggyTimestamp, PersonalAccessToken, UserStatus};
use journal::{Journal, JournalHandle};
+=======
+use crate::session_manager::SessionManager;
+use consensus::{MetadataHandle, build_transient_reply};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::{IggyError, IggyTimestamp, PersonalAccessToken, UserStatus};
+use message_bus::MessageBus;
+>>>>>>> Stashed changes
use metadata::impls::metadata::StreamsFrontend;
use server_common::Message;
use server_common::crypto;
@@ -306,6 +314,7 @@ pub(crate) async fn surface_login_failure<B, MJ, S>(
/// same login on the same connection. Only call for transient errors -- see
/// [`surface_login_failure`].
#[allow(clippy::future_not_send)]
+<<<<<<< Updated upstream
async fn send_login_transient_reply<B, MJ, S>(
shard: &Rc<ShellShard<B, MJ, S>>,
transport_client_id: u128,
@@ -324,6 +333,18 @@ async fn send_login_transient_reply<B, MJ, S>(
request_header,
commit,
IggyError::TransientNotAccepted.as_code(),
+=======
+async fn send_login_transient_reply(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request_header: &RequestHeader,
+) {
+ let commit = current_metadata_commit(shard);
+ let reply = build_transient_reply(
+ request_header,
+ commit,
+ IggyError::TransientNotCommitted.as_code(),
+>>>>>>> Stashed changes
);
if let Err(error) = shard
.bus
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index 9cf3378ed..e11d6af54 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -1275,6 +1275,7 @@ pub(crate) fn build_login_register_reply(
user_id: u32,
) -> Message<ReplyHeader> {
// Result-framed like every metadata reply: a zero result-count (success)
+<<<<<<< Updated upstream
// followed by the `LoginRegisterResponse` payload. A transient Register
// instead ships a `[count=1][index=0][TransientNotCommitted]` frame
// (`build_transient_reply`), which the SDK decodes and replays. The matching
@@ -1290,11 +1291,28 @@ pub(crate) fn build_login_register_reply(
body.extend_from_slice(&[0u8; RESULT_COUNT_LEN]);
body.extend_from_slice(&payload);
build_reply_from_bytes(
+=======
+ // followed by the `[user_id][session]` payload. A transient Register instead
+ // ships a `[count=1][index=0][TransientNotCommitted]` frame
+ // (`build_transient_reply`), which the SDK decodes and replays. The matching
+ // strip is in the SDK `split_metadata_result` (Register is result-framed).
+ build_reply_with_body(
+>>>>>>> Stashed changes
request_header,
client_id,
session,
commit,
+<<<<<<< Updated upstream
&Bytes::from(body),
+=======
+ RESULT_COUNT_LEN + 12,
+ |body| {
+ body[..RESULT_COUNT_LEN].copy_from_slice(&0u32.to_le_bytes());
+ body[RESULT_COUNT_LEN..RESULT_COUNT_LEN + 4].copy_from_slice(&user_id.to_le_bytes());
+ body[RESULT_COUNT_LEN + 4..RESULT_COUNT_LEN + 12]
+ .copy_from_slice(&session.to_le_bytes());
+ },
+>>>>>>> Stashed changes
)
}
@@ -1340,6 +1358,14 @@ pub(crate) fn build_raw_pat_reply(
return Ok(committed);
}
let header_len = std::mem::size_of::<ReplyHeader>();
+ // A `Reply` whose result section is nonzero is not a successful commit but a
+ // committed business rejection or a `TransientNotCommitted` retry frame, both
+ // with no payload and no token to ship. Pass it through so the client decodes
+ // the typed result (and, for a transient, replays) instead of having a raw
+ // token grafted onto a rejection body.
+ if result_code(&committed.as_slice()[header_len..]) != Some(0) {
+ return Ok(committed);
+ }
let committed_header =
bytemuck::checked::try_from_bytes::<ReplyHeader>(&committed.as_slice()[..header_len])
.map_err(|_| IggyError::InvalidFormat)?;
@@ -1359,6 +1385,7 @@ pub(crate) fn build_raw_pat_reply(
return Ok(committed);
}
let token = WireName::new(raw.as_str()).map_err(|_| IggyError::InvalidFormat)?;
+<<<<<<< Updated upstream
let response = RawPersonalAccessTokenResponse { token };
// The SDK strips a leading result section from every metadata reply
// (`split_metadata_result`), so the spliced body must carry the success
@@ -1367,6 +1394,19 @@ pub(crate) fn build_raw_pat_reply(
let mut body = BytesMut::with_capacity(RESULT_COUNT_LEN + response.encoded_size());
body.put_u32_le(0);
response.encode(&mut body);
+=======
+ let token_payload = RawPersonalAccessTokenResponse { token }.to_bytes();
+ // Metadata reply bodies are framed `[result_count:u32][payload]`; a success
+ // is `count == 0` followed by the payload (see `ApplyReply::write_reply_body`
+ // / `result_code`). The committed reply carried an empty payload with this
+ // same framing, so reproduce it here - prepend the zero result-count rather
+ // than shipping a bare `RawPersonalAccessTokenResponse`, which the client
+ // would misread as the result-count and reject as a bogus error.
+ let mut framed = Vec::with_capacity(RESULT_COUNT_LEN + token_payload.len());
+ framed.extend_from_slice(&[0u8; RESULT_COUNT_LEN]);
+ framed.extend_from_slice(&token_payload);
+ let body = Bytes::from(framed);
+>>>>>>> Stashed changes
let reply = build_reply_from_bytes(
request_header,
request_header.client,
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index a9fe3232e..dd10aa793 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -3400,10 +3400,14 @@ where
Input = Message<PrepareHeader>,
Output = metadata::stm::result::ApplyReply,
Error = iggy_common::IggyError,
+<<<<<<< Updated upstream
> + StreamsFrontend
+ metadata::stm::snapshot::RestoreSnapshotInPlace<
metadata::stm::snapshot::MetadataSnapshot,
>,
+=======
+ > + StreamsFrontend,
+>>>>>>> Stashed changes
{
let metadata = self.plane.metadata();
let Some(ref consensus) = metadata.consensus else {
@@ -3420,6 +3424,7 @@ where
// forever (commit_min stuck below commit_max). See
// `IggyMetadata::repair_primary_self_acks`.
metadata.repair_primary_self_acks().await;
+<<<<<<< Updated upstream
// Backstop for commit work stranded by a canceled `on_ack` driver
// (a future dropped at its journal-read or wire-reply await): no
@@ -3496,6 +3501,8 @@ where
.await;
}
}
+=======
+>>>>>>> Stashed changes
}
}
diff --git a/vsr-server-test-status.md b/vsr-server-test-status.md
new file mode 100644
index 000000000..82a497789
--- /dev/null
+++ b/vsr-server-test-status.md
@@ -0,0 +1,96 @@
+# Iggy integration tests under the cluster `vsr` feature
+
+Status of the `integration` test suite (`core/integration/tests/`) run against
+the next-gen clustered server (`--features vsr`, 3-node VSR cluster), release,
+`--test-threads=1` (io_uring OOM otherwise).
+
+## Scope: what even runs under vsr
+
+The integration binary has **878 tests** without `vsr` (legacy server). With
+`--features vsr` only **302** compile/run — the rest are `cfg(not vsr)`-gated or
+HTTP-only (server-ng exposes no HTTP listener) and **cannot run under vsr**.
+
+| Top module | non-vsr | vsr-runnable | excluded from vsr |
+| --- | ---: | ---: | --- |
+| `server` | 285 | **183** | 102 (HTTP transport variants + `cfg(not vsr)` fns) |
+| `data_integrity` | 108 | **96** | 12 (HTTP variants) |
+| `sdk` | 14 | **12** | 2 (HTTP variants) |
+| `storage` | 11 | **11** | 0 |
+| `cli` | 174 | 0 | whole module `cfg(not vsr)` (no HTTP/CLI path) |
+| `cluster` | 96 | 0 | whole module `cfg(not vsr)` |
+| `connectors` | 135 | 0 | whole module `cfg(not vsr)` (connectors runtime) |
+| `mcp` | 41 | 0 | whole module `cfg(not vsr)` |
+| `config_provider` | 9 | 0 | whole module `cfg(not vsr)` |
+| `state` | 5 | 0 | whole module `cfg(not vsr)` |
+| **Total** | **878** | **302** | **576** |
+
+Not covered here: `cargo test -p server` / `-p server-ng` crate unit tests — a
+separate surface with no vsr cluster mode.
+
+## Result of the vsr-runnable sweep: 302 / 302 pass
+
+| Module | Pass | Fail |
+| --- | ---: | ---: |
+| `server` | 183 | 0 |
+| `data_integrity` | 96 | 0 |
+| `sdk` | 12 | 0 |
+| `storage` | 11 | 0 |
+| **Total** | **302** | **0** |
+
+`server` submodule detail: specific 2/2, scenarios 3/3, purge_delete 6/6,
+message_cleanup 7/7, general 12/12 (incl. `authentication`), cg_vsr 21/21
+(incl. `duplicate_name_create`), message_retrieval 72/72, concurrent_addition 60/60.
+
+Run as a single test-binary invocation per module (proper per-test cluster
+lifecycle). The earlier 5 failures were burst transients; rapid back-to-back
+isolated runs can still surface a separate harness flake — "Timed out waiting
+for VSR replica mesh to form" — which is cluster-startup contention, not a code
+defect.
+
+## Previously failing (5) — fixed by the transient-reply refactor
+
+All five failed with **`Disconnected`** under concurrent / burst load on
+**tcp / websocket** (never QUIC):
+
+| Test | Old failure |
+| --- | --- |
+| `server::concurrent_addition::matrix::tcp_user_cold_barrier_off_expects` | assertion: expected AlreadyExists, got Disconnected |
+| `server::concurrent_addition::matrix::websocket_topic_hot_barrier_off_expects` | assertion: 19/20 ok, `Err(Disconnected)` |
+| `server::concurrent_addition::matrix::websocket_user_cold_barrier_off_expects` | assertion: got Disconnected |
+| `server::concurrent_addition::matrix::websocket_user_hot_barrier_off_expects` | client setup: "login failed: Disconnected" |
+| `data_integrity::verify_consumer_group_partition_assignment::should_not_reshuffle_partitions_when_new_member_joins` (websocket) | `unwrap()`: "login failed: Disconnected" |
+
+### Old root cause
+
+Under burst load the metadata primary transiently can't commit a request (or
+accept a login); the server stayed **silent** on the transient and relied on the
+SDK read-timeout to replay. The **tcp/websocket** lockstep stream can't safely
+resend after a silent timeout (a late reply would desync framing) → it must
+reconnect → reconnect loses the VSR session → can't idempotent-replay → gated off
+→ surfaces `Disconnected`. QUIC alone survived (per-request streams let it resend
+on the same connection).
+
+### Fix: explicit `TransientNotCommitted` reply frame (replaces silence)
+
+The server now replies with an explicit `TransientNotCommitted` (IggyError code
+57) result-code frame instead of staying silent. The complete frame keeps the
+lockstep stream in framing sync, so the SDK replays the **same request id on the
+same connection** (no reconnect, session intact) — idempotent via `ClientTable`
+dedup. Applied to **both** the metadata request path and the login/register path.
+
+- common: `IggyError::TransientNotCommitted = 57`.
+- consensus `metadata_helpers.rs`: `PreflightOutcome::Drop` split into `NotReady`
+ (in-flight / not-caught-up — transient) vs `Drop` (stale/gap/newer-session —
+ client bug, still silent). `plane_helpers.rs`: `build_transient_reply`.
+- metadata `submit_request_in_process`: transients (not-caught-up, pipeline-full,
+ in-flight, view-change cancel) reply with the transient frame instead of `Err`.
+- server-ng `responses.rs`: login success reply is now result-framed
+ (`[count=0][user_id][session]`); `auth.rs` `surface_login_failure` sends the
+ transient frame on a non-terminal (transient) login.
+- sdk `tcp`/`websocket`/`quic` `send_raw`: encode the header once, replay the same
+ request on `TransientNotCommitted` (incl. login), bounded by
+ `RESPONSE_READ_TIMEOUT` (30s), paced by `NOT_READY_RETRY_INTERVAL` (50ms).
+
+Verified: build + clippy `-D warnings` + crate unit tests clean; full
+vsr-runnable sweep 302/302, 0 regressions (`authentication` and `sdk` login
+paths green, so the login-reply result-framing decodes correctly).