You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/14 18:00:11 UTC
(iggy) branch server_perf_cut_extra_work updated (1e4a01df7 -> c2f5633fd)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch server_perf_cut_extra_work
in repository https://gitbox.apache.org/repos/asf/iggy.git
from 1e4a01df7 fix(sdk): synchronize offset routing command tables
new e3c53a11c fix(sdk): fence stale routes and preserve raw poll replies
new c2f5633fd test(integration): raise segment rotation memory limit
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../scenarios/segment_rotation_race_scenario.rs | 2 +-
core/sdk/src/poll_routing.rs | 308 +++++++++++++++++++--
foreign/node/src/client/client.socket.test.ts | 34 +++
foreign/node/src/client/client.socket.ts | 17 +-
4 files changed, 338 insertions(+), 23 deletions(-)
(iggy) 02/02: test(integration): raise segment rotation memory limit
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch server_perf_cut_extra_work
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c2f5633fd8710a9f5f4148b7408696bed82ec853
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Sep 14 19:59:18 2026 +0200
test(integration): raise segment rotation memory limit
---
.../tests/server/scenarios/segment_rotation_race_scenario.rs | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs b/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
index fb963ad0a..ba0071d3f 100644
--- a/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
+++ b/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
@@ -50,7 +50,7 @@ const SEGMENT_SIZE: u64 = 1024 * 1024;
/// this rolls ~240 at the same server memory footprint. The old 512 B cap
/// rolled one per batch (~50k), which no legal segment size can reproduce.
const PAYLOAD_FILLER_LEN: usize = 2 * 1024;
-const MAX_ALLOWED_MEMORY_BYTES: u64 = 200 * 1024 * 1024;
+const MAX_ALLOWED_MEMORY_BYTES: u64 = 300 * 1024 * 1024;
/// Runs the segment rotation race condition test with multiple protocols.
/// Uses all available transports from the harness (TCP, HTTP, QUIC, WebSocket).
(iggy) 01/02: fix(sdk): fence stale routes and preserve raw poll replies
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch server_perf_cut_extra_work
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit e3c53a11c6a8b6cae30f276b47d5a477aeb6ee7a
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Sep 14 19:55:00 2026 +0200
fix(sdk): fence stale routes and preserve raw poll replies
Rust routing replies could restore an old group attachment after a
coordinator reconnect. Fence route publication and use by session
generation, including requests waiting on a primary connection.
Node primary routing discarded the raw-response option. Preserve the
original frame while still handling routing refusals internally.
Add deterministic reconnect and TCP/TLS raw-response regressions.
---
core/sdk/src/poll_routing.rs | 308 ++++++++++++++++++++++++--
foreign/node/src/client/client.socket.test.ts | 34 +++
foreign/node/src/client/client.socket.ts | 17 +-
3 files changed, 337 insertions(+), 22 deletions(-)
diff --git a/core/sdk/src/poll_routing.rs b/core/sdk/src/poll_routing.rs
index 042b8693d..612b16d0d 100644
--- a/core/sdk/src/poll_routing.rs
+++ b/core/sdk/src/poll_routing.rs
@@ -18,6 +18,7 @@
//! Auto-commit polls and offset writes use persistent data connections while
//! the coordinator retains group membership. Only explicit non-admission permits
//! rerouting. Replicated writes keep the data session's own deduplication identity.
+//! Routes and attachments are fenced by the coordinator's session generation.
//! Cluster routing requires servers supporting the routing and attachment commands.
use crate::leader_aware::{node_address, transport_port};
@@ -83,6 +84,7 @@ pub(crate) trait PollTransport: BinaryClient + Send + Sync + Sized {
#[derive(Debug)]
struct PollRoute {
+ generation: u64,
endpoint: String,
consumer_session: AttachConsumerSessionRequest,
}
@@ -102,6 +104,7 @@ pub(crate) struct PollRouter<T> {
pub(crate) metadata_watermark: Arc<AtomicU64>,
/// Zero means no successful topology read, not a standalone server.
pub(crate) roster_size: AtomicUsize,
+ session_generation: AtomicU64,
routes: Mutex<HashMap<RouteKey, Arc<PollRoute>>>,
connections: Mutex<HashMap<String, ConnectionSlot<T>>>,
credentials: Mutex<Option<(Credentials, u32)>>,
@@ -113,6 +116,7 @@ impl<T> Default for PollRouter<T> {
Self {
metadata_watermark: Arc::default(),
roster_size: AtomicUsize::new(0),
+ session_generation: AtomicU64::new(0),
routes: Mutex::default(),
connections: Mutex::default(),
credentials: Mutex::default(),
@@ -127,14 +131,18 @@ impl<T> PollRouter<T> {
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
- self.routes
+ let mut routes = self
+ .routes
.lock()
- .unwrap_or_else(std::sync::PoisonError::into_inner)
- .clear();
- self.connections
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let mut connections = self
+ .connections
.lock()
- .unwrap_or_else(std::sync::PoisonError::into_inner)
- .clear();
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ // Serialize invalidation with route and connection publication.
+ self.session_generation.fetch_add(1, Ordering::AcqRel);
+ routes.clear();
+ connections.clear();
}
pub(crate) fn remember_credentials(&self, credentials: Credentials, user_id: u32) {
@@ -311,6 +319,7 @@ impl<T: PollTransport> PollRouter<T> {
.connections
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
+ self.validate_route(&route)?;
if let Some(slot) = connections.get(&route.endpoint) {
Arc::clone(slot)
} else {
@@ -324,6 +333,7 @@ impl<T: PollTransport> PollRouter<T> {
}
};
let mut connection = slot.lock().await;
+ self.validate_route(&route)?;
if connection
.as_ref()
.is_some_and(|connection| !connection.usable)
@@ -338,6 +348,7 @@ impl<T: PollTransport> PollRouter<T> {
}
Err(error) => return Err(error),
};
+ self.validate_route(&route)?;
*connection = Some(PollConnection {
client,
consumer_session: None,
@@ -365,6 +376,7 @@ impl<T: PollTransport> PollRouter<T> {
.map_err(unaccepted_data_error)?;
connection.consumer_session = Some(route.consumer_session);
}
+ self.validate_route(&route)?;
let result = connection
.client
.send_poll_request(code, payload.clone())
@@ -394,14 +406,16 @@ impl<T: PollTransport> PollRouter<T> {
key: &RouteKey,
payload: &Bytes,
) -> Result<Arc<PollRoute>, IggyError> {
+ let generation = self.session_generation.load(Ordering::Acquire);
if let Some(route) = self
.routes
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get(key)
.filter(|route| {
- route.consumer_session.metadata_watermark
- >= self.metadata_watermark.load(Ordering::Acquire)
+ route.generation == generation
+ && route.consumer_session.metadata_watermark
+ >= self.metadata_watermark.load(Ordering::Acquire)
})
{
return Ok(Arc::clone(route));
@@ -429,6 +443,7 @@ impl<T: PollTransport> PollRouter<T> {
return Err(IggyError::FeatureUnavailable);
}
let route = Arc::new(PollRoute {
+ generation,
endpoint: node_address(&node, port),
consumer_session: response.consumer_session,
});
@@ -436,12 +451,23 @@ impl<T: PollTransport> PollRouter<T> {
.routes
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
+ self.validate_route(&route)?;
if routes.len() >= MAX_CACHED_ROUTES {
routes.clear();
}
routes.insert(key.clone(), Arc::clone(&route));
Ok(route)
}
+
+ fn validate_route(&self, route: &PollRoute) -> Result<(), IggyError> {
+ if route.generation != self.session_generation.load(Ordering::Acquire)
+ || route.consumer_session.metadata_watermark
+ < self.metadata_watermark.load(Ordering::Acquire)
+ {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ Ok(())
+ }
}
pub(crate) fn matches_session_user(user: &Identifier, user_id: u32, username: &str) -> bool {
@@ -498,6 +524,7 @@ mod tests {
use std::collections::VecDeque;
use std::str::FromStr;
use std::sync::atomic::AtomicUsize;
+ use tokio::sync::Notify;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Channel {
@@ -508,11 +535,42 @@ mod tests {
type Exchange = (Channel, u32, Result<Bytes, IggyError>);
+ #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+ enum PausePoint {
+ Connect,
+ Reply(u32),
+ }
+
+ #[derive(Debug)]
+ struct Pause {
+ point: PausePoint,
+ reached: Notify,
+ resume: Notify,
+ }
+
#[derive(Debug, Default)]
struct Script {
exchanges: Mutex<VecDeque<Exchange>>,
- attachments: Mutex<Vec<u64>>,
+ attachments: Mutex<Vec<AttachConsumerSessionRequest>>,
connections: AtomicUsize,
+ pause: Mutex<Option<Arc<Pause>>>,
+ }
+
+ impl Script {
+ async fn pause_at(&self, point: PausePoint) {
+ let pause = {
+ let mut pause = self.pause.lock().unwrap();
+ if pause.as_ref().is_some_and(|pause| pause.point == point) {
+ pause.take()
+ } else {
+ None
+ }
+ };
+ if let Some(pause) = pause {
+ pause.reached.notify_one();
+ pause.resume.notified().await;
+ }
+ }
}
#[derive(Debug)]
@@ -537,11 +595,11 @@ mod tests {
.expect("unexpected request");
assert_eq!((channel, code), (expected_channel, expected_code));
if code == ATTACH_CONSUMER_SESSION_CODE {
- self.script.attachments.lock().unwrap().push(
- AttachConsumerSessionRequest::decode_from(payload)
- .unwrap()
- .metadata_watermark,
- );
+ self.script
+ .attachments
+ .lock()
+ .unwrap()
+ .push(AttachConsumerSessionRequest::decode_from(payload).unwrap());
}
result
}
@@ -606,13 +664,179 @@ mod tests {
const PROTOCOL: TransportProtocol = TransportProtocol::Tcp;
async fn connect_poll_client(&self, _endpoint: &str) -> Result<Self, IggyError> {
self.script.connections.fetch_add(1, Ordering::Relaxed);
+ self.script.pause_at(PausePoint::Connect).await;
Ok(Self {
channel: Channel::Data,
script: Arc::clone(&self.script),
})
}
async fn send_poll_request(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
- self.exchange(self.channel, code, &payload)
+ let result = self.exchange(self.channel, code, &payload);
+ self.script.pause_at(PausePoint::Reply(code)).await;
+ result
+ }
+ }
+
+ #[tokio::test]
+ async fn reconnect_fences_in_flight_routes_connections_and_attachments() {
+ for code in [
+ POLL_MESSAGES_ON_PRIMARY_CODE,
+ STORE_CONSUMER_OFFSET_CODE,
+ DELETE_CONSUMER_OFFSET_CODE,
+ ] {
+ let routing_code = if code == POLL_MESSAGES_ON_PRIMARY_CODE {
+ GET_POLL_ROUTING_CODE
+ } else {
+ GET_CONSUMER_OFFSET_ROUTING_CODE
+ };
+ for point in [
+ PausePoint::Reply(routing_code),
+ PausePoint::Connect,
+ PausePoint::Reply(ATTACH_CONSUMER_SESSION_CODE),
+ ] {
+ let mut replacement = PollRoutingResponse::decode_from(&routing()).unwrap();
+ replacement.consumer_session.session = 2;
+ let mut exchanges = vec![(Channel::Coordinator, routing_code, Ok(routing()))];
+ if point == PausePoint::Reply(ATTACH_CONSUMER_SESSION_CODE) {
+ exchanges.push((
+ Channel::Data,
+ ATTACH_CONSUMER_SESSION_CODE,
+ Ok(Bytes::new()),
+ ));
+ }
+ exchanges.extend([
+ (
+ Channel::Coordinator,
+ routing_code,
+ Ok(replacement.to_bytes()),
+ ),
+ (
+ Channel::Data,
+ ATTACH_CONSUMER_SESSION_CODE,
+ Ok(Bytes::new()),
+ ),
+ (Channel::Data, code, Ok(Bytes::from_static(b"resumed"))),
+ (Channel::Data, code, Ok(Bytes::from_static(b"warm"))),
+ ]);
+ let (router, coordinator, request) = fixture(exchanges);
+ let pause = Arc::new(Pause {
+ point,
+ reached: Notify::new(),
+ resume: Notify::new(),
+ });
+ *coordinator.script.pause.lock().unwrap() = Some(Arc::clone(&pause));
+ let pending = routed_request(&router, &coordinator, &request, code);
+ tokio::pin!(pending);
+ tokio::select! {
+ result = &mut pending => panic!("request finished before reconnect: {result:?}"),
+ () = pause.reached.notified() => {}
+ }
+ router.clear_session();
+ router.clear_session();
+ router.metadata_watermark.store(11, Ordering::Release);
+ pause.resume.notify_one();
+
+ assert_eq!(pending.await.unwrap(), "resumed", "{code} at {point:?}");
+ assert_eq!(
+ routed_request(&router, &coordinator, &request, code)
+ .await
+ .unwrap(),
+ "warm"
+ );
+ let routes = router.routes.lock().unwrap();
+ assert_eq!(routes.len(), 1);
+ assert!(routes.iter().all(|((command, _), route)| {
+ *command == routing_code && route.consumer_session.session == 2
+ }));
+ replacement.consumer_session.metadata_watermark = 11;
+ assert_eq!(
+ *coordinator.script.attachments.lock().unwrap(),
+ if point == PausePoint::Reply(ATTACH_CONSUMER_SESSION_CODE) {
+ vec![
+ PollRoutingResponse::decode_from(&routing())
+ .unwrap()
+ .consumer_session,
+ replacement.consumer_session,
+ ]
+ } else {
+ vec![replacement.consumer_session]
+ }
+ );
+ assert_eq!(
+ coordinator.script.connections.load(Ordering::Relaxed),
+ if point == PausePoint::Reply(routing_code) {
+ 1
+ } else {
+ 2
+ }
+ );
+ assert!(coordinator.script.exchanges.lock().unwrap().is_empty());
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn reconnect_fences_queued_requests_without_replaying_in_flight_data() {
+ for code in [
+ POLL_MESSAGES_ON_PRIMARY_CODE,
+ STORE_CONSUMER_OFFSET_CODE,
+ DELETE_CONSUMER_OFFSET_CODE,
+ ] {
+ let routing_code = if code == POLL_MESSAGES_ON_PRIMARY_CODE {
+ GET_POLL_ROUTING_CODE
+ } else {
+ GET_CONSUMER_OFFSET_ROUTING_CODE
+ };
+ let mut replacement = PollRoutingResponse::decode_from(&routing()).unwrap();
+ replacement.consumer_session.session = 2;
+ let (router, coordinator, request) = fixture([
+ (Channel::Coordinator, routing_code, Ok(routing())),
+ (
+ Channel::Data,
+ ATTACH_CONSUMER_SESSION_CODE,
+ Ok(Bytes::new()),
+ ),
+ (Channel::Data, code, Ok(Bytes::from_static(b"in flight"))),
+ (
+ Channel::Coordinator,
+ routing_code,
+ Ok(replacement.to_bytes()),
+ ),
+ (
+ Channel::Data,
+ ATTACH_CONSUMER_SESSION_CODE,
+ Ok(Bytes::new()),
+ ),
+ (Channel::Data, code, Ok(Bytes::from_static(b"queued"))),
+ ]);
+ let pause = Arc::new(Pause {
+ point: PausePoint::Reply(code),
+ reached: Notify::new(),
+ resume: Notify::new(),
+ });
+ *coordinator.script.pause.lock().unwrap() = Some(Arc::clone(&pause));
+ let in_flight = routed_request(&router, &coordinator, &request, code);
+ tokio::pin!(in_flight);
+ tokio::select! {
+ result = &mut in_flight => panic!("data reply was not paused: {result:?}"),
+ () = pause.reached.notified() => {}
+ }
+ let queued = routed_request(&router, &coordinator, &request, code);
+ tokio::pin!(queued);
+ assert!(futures::poll!(&mut queued).is_pending());
+ router.clear_session();
+ router.clear_session();
+ pause.resume.notify_one();
+
+ let (in_flight, queued) = tokio::join!(in_flight, queued);
+ assert_eq!(in_flight.unwrap(), "in flight");
+ assert_eq!(queued.unwrap(), "queued");
+ assert_eq!(
+ coordinator.script.attachments.lock().unwrap().last(),
+ Some(&replacement.consumer_session)
+ );
+ assert_eq!(coordinator.script.connections.load(Ordering::Relaxed), 2);
+ assert!(coordinator.script.exchanges.lock().unwrap().is_empty());
}
}
@@ -768,7 +992,17 @@ mod tests {
"after metadata"
);
assert_eq!(router.poll(&coordinator, &request).await.unwrap(), "warm");
- assert_eq!(*coordinator.script.attachments.lock().unwrap(), [1, 11]);
+ assert_eq!(
+ coordinator
+ .script
+ .attachments
+ .lock()
+ .unwrap()
+ .iter()
+ .map(|session| session.metadata_watermark)
+ .collect::<Vec<_>>(),
+ [1, 11]
+ );
assert_eq!(coordinator.script.connections.load(Ordering::Relaxed), 1);
assert!(coordinator.script.exchanges.lock().unwrap().is_empty());
}
@@ -850,7 +1084,17 @@ mod tests {
"resumed"
);
assert_eq!(coordinator.script.connections.load(Ordering::Relaxed), 1);
- assert_eq!(*coordinator.script.attachments.lock().unwrap(), [1]);
+ assert_eq!(
+ coordinator
+ .script
+ .attachments
+ .lock()
+ .unwrap()
+ .iter()
+ .map(|session| session.metadata_watermark)
+ .collect::<Vec<_>>(),
+ [1]
+ );
assert!(coordinator.script.exchanges.lock().unwrap().is_empty());
}
@@ -899,6 +1143,36 @@ mod tests {
assert!(coordinator.script.exchanges.lock().unwrap().is_empty());
}
+ async fn routed_request(
+ router: &PollRouter<Transport>,
+ coordinator: &Transport,
+ request: &PollMessagesRequest,
+ code: u32,
+ ) -> Result<Bytes, IggyError> {
+ let payload = match code {
+ POLL_MESSAGES_ON_PRIMARY_CODE => return router.poll(coordinator, request).await,
+ STORE_CONSUMER_OFFSET_CODE => StoreConsumerOffsetRequest {
+ consumer: request.consumer.clone(),
+ stream_id: request.stream_id.clone(),
+ topic_id: request.topic_id.clone(),
+ partition_id: request.partition_id,
+ offset: 10,
+ ack: AckLevel::Quorum,
+ }
+ .to_bytes(),
+ DELETE_CONSUMER_OFFSET_CODE => DeleteConsumerOffsetRequest {
+ consumer: request.consumer.clone(),
+ stream_id: request.stream_id.clone(),
+ topic_id: request.topic_id.clone(),
+ partition_id: request.partition_id,
+ ack: AckLevel::Quorum,
+ }
+ .to_bytes(),
+ _ => panic!("unexpected routed command: {code}"),
+ };
+ router.write_offset(coordinator, code, payload).await
+ }
+
fn fixture(
exchanges: impl IntoIterator<Item = Exchange>,
) -> (PollRouter<Transport>, Transport, PollMessagesRequest) {
diff --git a/foreign/node/src/client/client.socket.test.ts b/foreign/node/src/client/client.socket.test.ts
index 46fb582d3..aac3dacdd 100644
--- a/foreign/node/src/client/client.socket.test.ts
+++ b/foreign/node/src/client/client.socket.test.ts
@@ -356,6 +356,40 @@ const startPollCluster = async (
};
describe('primary auto-commit polling', () => {
+ for (const transport of ['TCP', 'TLS'] as const) {
+ for (const refused of [false, true]) {
+ it(`preserves raw poll responses over ${transport}${refused ? ' after a refusal' : ''}`, async () => {
+ const expected = replyFrame(Operation.NonReplicated, Buffer.from('raw primary'));
+ let polls = 0;
+ const cluster = await startPollCluster((frame, socket) => {
+ if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.PollMessagesOnPrimary)
+ return false;
+ polls += 1;
+ socket.write(refused && polls === 1
+ ? replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58)
+ : expected);
+ return true;
+ }, undefined, transport);
+ let resets = 0;
+ cluster.client.on('sessionReset', () => { resets += 1; });
+ try {
+ for (let poll = 0; poll < 2; poll += 1) {
+ const response = await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload(),
+ { handleResponse: false });
+ assert.ok(Buffer.isBuffer(response), 'raw polls must return the complete VSR frame');
+ assert.deepEqual(response, expected);
+ }
+ assert.equal(polls, refused ? 3 : 2);
+ assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPollRouting), refused ? 2 : 1);
+ assert.equal(countCommand(cluster.primary, COMMAND_CODE.AttachConsumerSession), refused ? 2 : 1);
+ assert.equal(resets, 0);
+ } finally {
+ await cluster.close();
+ }
+ });
+ }
+ }
+
for (const transport of ['TCP', 'TLS'] as const) {
it(`retains the coordinator session and reuses authenticated primary polls over ${transport}`, async () => {
const cluster = await startPollCluster(undefined, undefined, transport);
diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts
index 2de4b1f59..5d13626d5 100644
--- a/foreign/node/src/client/client.socket.ts
+++ b/foreign/node/src/client/client.socket.ts
@@ -36,7 +36,7 @@ import { deserializeNode } from '../wire/cluster/cluster.utils.js';
import { COMMAND_CODE } from '../wire/command.code.js';
import { serializeIdentifier } from '../wire/identifier.utils.js';
import {
- Command, HEADER_SIZE, REPLY_OFFSET, peekCommand, readReplyOperation
+ Command, HEADER_SIZE, REPLY_OFFSET, peekCommand, readReplyOperation, readStatus
} from '../wire/vsr/header.js';
import { Operation, isKnownOperation } from '../wire/vsr/operation.js';
import {
@@ -336,7 +336,7 @@ export class CommandResponseStream extends EventEmitter {
if (this.clustered === undefined)
throw new Error('cannot determine the poll routing topology');
if (this.clustered)
- return await this._pollOnPrimary(payload, deadline);
+ return await this._pollOnPrimary(payload, deadline, handleResponse);
}
// The roster read is itself a queued command and the queue is
@@ -585,7 +585,8 @@ export class CommandResponseStream extends EventEmitter {
private async _pollOnPrimary(
payload: Buffer,
- deadline: number
+ deadline: number,
+ handleResponse: boolean
): Promise<CommandResponse> {
const key = payload.subarray(0, -POLL_OPTIONS_SIZE).toString('hex');
while (Date.now() < deadline && !this.connection.ending) {
@@ -673,7 +674,7 @@ export class CommandResponseStream extends EventEmitter {
throw responseError(COMMAND_CODE.PollMessages, TRANSIENT_NOT_ACCEPTED);
polling = true;
return await entry.client._queueCommand(COMMAND_CODE.PollMessagesOnPrimary,
- payload, true, true, false, deadline);
+ payload, handleResponse, true, false, deadline);
} catch (error) {
if (error instanceof ResponseError &&
error.errorCode === TRANSIENT_NOT_ACCEPTED) {
@@ -911,8 +912,14 @@ export class CommandResponseStream extends EventEmitter {
requestWritten ||= exchangeState.written;
}
this._observeMetadataReply(response);
- if (!handleResp)
+ if (!handleResp) {
+ // Routing still needs refusals when the caller decodes the frame.
+ if (command === COMMAND_CODE.PollMessagesOnPrimary &&
+ peekCommand(response) === Command.Reply &&
+ readStatus(response) === TRANSIENT_NOT_ACCEPTED)
+ throw responseError(command, TRANSIENT_NOT_ACCEPTED);
return response as unknown as CommandResponse;
+ }
try {
parsed = decodeVsrResponse(response, command);
break;