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/10/26 21:30:32 UTC
(iggy) 07/15: perf(server): evaluate only changed targets of a parked readiness wait
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/deferred-polling-pr
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit e90fc6b18a7fa3d79aec6e1105dc8e708133afbd
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Oct 26 19:32:28 2026 +0100
perf(server): evaluate only changed targets of a parked readiness wait
A commit woke each parked readiness wait on its partition, and the
probe ran every check on every target of the wait. A wait on 1000
partitions cost 1000 evaluations per commit, and maintenance probed
16 waits per tick whatever their size.
A probe now evaluates the targets whose partitions changed, plus a
window that maintenance assigns by counting targets, and checks the
caller once per wait. A reply can leave out a next partition whose
stored offset moved back without a commit. The client waits on that
partition again, and admission evaluates every target.
---
core/common/src/traits/message_client.rs | 3 +-
core/server/src/dispatch/partition.rs | 33 +++--
core/server/src/dispatch/reads.rs | 9 +-
core/server/src/http/handlers.rs | 2 +-
core/shard/src/lib.rs | 7 +-
core/shard/src/poll/deferred.rs | 212 ++++++++++++++++++++++++-------
core/shard/src/poll/deferred_tests.rs | 149 ++++++++++++++++++++--
7 files changed, 328 insertions(+), 87 deletions(-)
diff --git a/core/common/src/traits/message_client.rs b/core/common/src/traits/message_client.rs
index 80fe3e617..ac773b5b6 100644
--- a/core/common/src/traits/message_client.rs
+++ b/core/common/src/traits/message_client.rs
@@ -188,7 +188,8 @@ pub trait MessageClient {
/// Each start point is [`PollingStrategy::offset`] or [`PollingStrategy::next`] of
/// `consumer`, and each partition appears once. A reported partition returns at least one
/// message to a poll with the same consumer and strategy, unless a purge or retention
- /// removed the messages, or another commit moved `next` past them, in the meantime.
+ /// removed the messages, or another commit moved `next` past them, in the meantime. A
+ /// reply can leave out a `next` partition whose stored offset moved back during the wait.
///
/// The wait runs on the primary of each partition. If the partitions have more than one
/// primary, the call waits on each of them and returns the first reply that lists a
diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs
index 48a3f6d05..de2d5e6f7 100644
--- a/core/server/src/dispatch/partition.rs
+++ b/core/server/src/dispatch/partition.rs
@@ -727,9 +727,9 @@ where
let request = DeferredPollRequest::new(
consumer,
args,
+ metadata,
DeferredPollContext {
session: Some(session),
- metadata,
user_id,
user_created_at: None,
quota: PollQuotaIdentity::Session(consumer_client_id),
@@ -742,12 +742,12 @@ where
/// Resolve and authorize every partition of a readiness wait with the fence
/// and metadata identity of a poll. The owner checks them again at each
-/// evaluation.
+/// evaluation. Returns the caller that every target shares, and the targets.
pub(super) fn resolve_readiness_wait<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
wire: WaitForMessagesRequest,
(consumer_client_id, session, user_id): (u128, SessionAttachment, u32),
-) -> Result<Vec<ReadinessTarget>, IggyError>
+) -> Result<(DeferredPollContext, Vec<ReadinessTarget>), IggyError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -775,7 +775,8 @@ where
count: 1,
auto_commit: false,
};
- wire.partitions
+ let targets = wire
+ .partitions
.into_iter()
.map(|target| {
poll.partition_id = Some(target.partition_id);
@@ -794,22 +795,18 @@ where
value: args.value,
context: None,
},
- context: DeferredPollContext {
- session: Some(session.clone()),
- metadata: consumer_poll_metadata(
- shard,
- namespace,
- consumer,
- consumer_client_id,
- )?,
- user_id,
- user_created_at: None,
- quota: PollQuotaIdentity::Session(consumer_client_id),
- primary: true,
- },
+ metadata: consumer_poll_metadata(shard, namespace, consumer, consumer_client_id)?,
})
})
- .collect()
+ .collect::<Result<_, _>>()?;
+ let caller = DeferredPollContext {
+ session: Some(session),
+ user_id,
+ user_created_at: None,
+ quota: PollQuotaIdentity::Session(consumer_client_id),
+ primary: true,
+ };
+ Ok((caller, targets))
}
fn consumer_poll_metadata<B, MJ, S, SB>(
diff --git a/core/server/src/dispatch/reads.rs b/core/server/src/dispatch/reads.rs
index e1a4b2f2a..4f9173cd3 100644
--- a/core/server/src/dispatch/reads.rs
+++ b/core/server/src/dispatch/reads.rs
@@ -717,7 +717,7 @@ async fn handle_deferred_request<B, MJ, S, SB>(
}
Either::Right(wait) => {
let duration = Duration::from_micros(shard.deferred_wait_us(wait.wait_us));
- let targets = resolve_readiness_wait(
+ let (caller, targets) = resolve_readiness_wait(
shard,
wait,
(consumer_client_id, attachment.clone(), user_id),
@@ -727,14 +727,11 @@ async fn handle_deferred_request<B, MJ, S, SB>(
.map(|target| {
(
target.namespace,
- target
- .context
- .metadata
- .context(current_metadata_commit(shard)),
+ target.metadata.context(current_metadata_commit(shard)),
)
})
.collect();
- let partitions = shard.wait_for_messages(targets, duration).await?;
+ let partitions = shard.wait_for_messages(caller, targets, duration).await?;
for (namespace, context) in contexts {
authorize_served_partition_read(
shard,
diff --git a/core/server/src/http/handlers.rs b/core/server/src/http/handlers.rs
index 39e352b02..b274ec2b3 100644
--- a/core/server/src/http/handlers.rs
+++ b/core/server/src/http/handlers.rs
@@ -1355,9 +1355,9 @@ async fn poll_messages_impl(
PartitionRead::DeferredPoll(Box::new(DeferredPollRequest::new(
consumer,
args.clone(),
+ metadata,
DeferredPollContext {
session: None,
- metadata,
user_id: identity.user_id,
user_created_at: Some(identity.user_created_at.as_micros()),
quota: PollQuotaIdentity::User(identity.user_id),
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 0b8429f34..31dc5b8ac 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -2462,6 +2462,7 @@ where
#[allow(clippy::future_not_send)]
pub async fn wait_for_messages(
&self,
+ context: DeferredPollContext,
targets: Vec<ReadinessTarget>,
wait: std::time::Duration,
) -> Result<Vec<u32>, IggyError> {
@@ -2486,7 +2487,11 @@ where
let mut admitting = BTreeSet::new();
for (shard, targets) in owners {
let frame = ShardFrame::lifecycle(LifecycleFrame::ReadinessWait {
- wait: ReadinessWait { targets, deadline },
+ wait: ReadinessWait {
+ context: context.clone(),
+ targets,
+ deadline,
+ },
reply: reply.clone(),
});
self.senders
diff --git a/core/shard/src/poll/deferred.rs b/core/shard/src/poll/deferred.rs
index 1b8ad0ff3..3fac34179 100644
--- a/core/shard/src/poll/deferred.rs
+++ b/core/shard/src/poll/deferred.rs
@@ -59,10 +59,11 @@ pub enum PollQuotaIdentity {
User(u32),
}
-#[derive(Debug)]
+/// The caller of a deferred request. A readiness wait shares one context
+/// across its targets.
+#[derive(Debug, Clone)]
pub struct DeferredPollContext {
pub session: Option<SessionAttachment>,
- pub metadata: PollMetadata,
pub user_id: u32,
pub user_created_at: Option<u64>,
pub quota: PollQuotaIdentity,
@@ -78,6 +79,7 @@ pub struct DeferredPollContext {
pub struct DeferredPollRequest {
pub consumer: PollingConsumer,
pub args: PollingArgs,
+ pub metadata: PollMetadata,
pub context: DeferredPollContext,
pub options: DeferredPollOptions,
pub(crate) deadline: u64,
@@ -89,12 +91,14 @@ impl DeferredPollRequest {
pub const fn new(
consumer: PollingConsumer,
args: PollingArgs,
+ metadata: PollMetadata,
context: DeferredPollContext,
options: DeferredPollOptions,
) -> Self {
Self {
consumer,
args,
+ metadata,
context,
options,
deadline: 0,
@@ -111,22 +115,31 @@ pub struct ReadinessTarget {
pub consumer: PollingConsumer,
/// `Offset` or `Next`, the start point of a poll with the same strategy.
pub strategy: PollingStrategy,
- pub context: DeferredPollContext,
+ pub metadata: PollMetadata,
}
-/// The targets of one readiness wait that a single owner shard holds.
+/// The targets of one readiness wait that a single owner shard holds. They
+/// name one stream and topic, so one check of `context` covers all of them.
#[derive(Debug)]
pub struct ReadinessWait {
+ pub context: DeferredPollContext,
pub targets: Vec<ReadinessTarget>,
pub deadline: u64,
}
-/// A readiness waiter answers once. `admitted` holds the history and view of
-/// each target at admission, and any change rejects the waiter.
+/// A readiness waiter answers once. Its targets are sorted by namespace.
+/// `admitted` holds the history and view of each target at admission, and a
+/// change that an evaluation finds rejects the waiter.
struct ReadinessWaiter {
wait: ReadinessWait,
admitted: Vec<(PollHistoryId, u32)>,
- quota: PollQuotaIdentity,
+ /// Namespaces that changed since the last probe. A commit makes only its
+ /// own target ready, so a probe evaluates these targets.
+ dirty: Vec<IggyNamespace>,
+ /// Maintenance evaluates every target in turn: `swept` targets of its
+ /// current pass are done, and the next probe evaluates `sweep` more.
+ swept: usize,
+ sweep: usize,
reply: Sender<PartitionReadReply>,
}
@@ -353,17 +366,39 @@ impl DeferredPolls {
}
}
+ /// A tick probes polls and readiness targets up to `MAINTENANCE_BUDGET`,
+ /// so a large wait spreads over several ticks. The cursor stays before a
+ /// wait that the budget covers in part, and the next tick resumes it.
pub(crate) fn maintenance(&self) {
let mut state = self.state.borrow_mut();
let mut ids = state.ids_after(state.maintenance_cursor);
if ids.is_empty() {
ids = state.ids_after(0);
}
- if let Some(last) = ids.last() {
- state.maintenance_cursor = *last;
- }
+ let mut budget = MAINTENANCE_BUDGET;
for id in ids {
+ if budget == 0 {
+ break;
+ }
+ let partial = if let Some(waiter) = state.readiness.get_mut(&id) {
+ let remaining = waiter
+ .wait
+ .targets
+ .len()
+ .saturating_sub(waiter.swept + waiter.sweep);
+ let sweep = remaining.min(budget);
+ waiter.sweep += sweep;
+ budget -= sweep;
+ sweep < remaining
+ } else {
+ budget -= 1;
+ false
+ };
state.enqueue(id);
+ if partial {
+ break;
+ }
+ state.maintenance_cursor = id;
}
}
@@ -539,9 +574,24 @@ where
fn validate_deferred_context(
&self,
namespace: IggyNamespace,
- context: &DeferredPollContext,
- consumer: PollingConsumer,
+ request: &DeferredPollRequest,
) -> Result<(PollHistoryId, PollVisibility, u32), IggyError> {
+ self.validate_caller(namespace, &request.context)?;
+ self.validate_partition(
+ namespace,
+ request.context.primary,
+ &request.metadata,
+ request.consumer,
+ )
+ }
+
+ /// The checks of the caller, the same for every partition of one stream
+ /// and topic.
+ fn validate_caller(
+ &self,
+ namespace: IggyNamespace,
+ context: &DeferredPollContext,
+ ) -> Result<(), IggyError> {
if context
.session
.as_ref()
@@ -568,7 +618,18 @@ where
metadata.users().authorize(|permissioner| {
permissioner.poll_messages(context.user_id, namespace.stream_id(), namespace.topic_id())
})?;
- if !context.metadata.is_valid(metadata.streams(), namespace) {
+ Ok(())
+ }
+
+ /// The checks of one partition, after those of its caller.
+ fn validate_partition(
+ &self,
+ namespace: IggyNamespace,
+ primary: bool,
+ metadata: &PollMetadata,
+ consumer: PollingConsumer,
+ ) -> Result<(PollHistoryId, PollVisibility, u32), IggyError> {
+ if !metadata.is_valid(self.plane.metadata().mux_stm.streams(), namespace) {
return Err(IggyError::TransientNotAccepted);
}
self.plane
@@ -578,7 +639,7 @@ where
if partition.history_deleted() {
return Err(IggyError::HistoryUnavailable);
}
- if context.primary && consensus.is_normal_backup() {
+ if primary && consensus.is_normal_backup() {
return Err(IggyError::NotPrimary);
}
if let PollingConsumer::ConsumerGroup(group, _) = consumer
@@ -586,21 +647,19 @@ where
namespace,
group as u64,
partition.consumer_group_owner(group as u64),
- &context.metadata,
+ metadata,
)
{
return Err(error);
}
if partition.fatal().is_some()
|| partition.read_history_is_changing()
- || (context.primary && !consensus.applied_view_start())
+ || (primary && !consensus.applied_view_start())
|| partition.requires_state_transfer()
|| !consensus.is_normal()
|| consensus.is_transferring()
- || (context.primary && !consensus.is_primary())
- || !context
- .metadata
- .matches_partition(self.shards_table.epoch_for(namespace))
+ || (primary && !consensus.is_primary())
+ || !metadata.matches_partition(self.shards_table.epoch_for(namespace))
{
return Err(IggyError::TransientNotAccepted);
}
@@ -615,11 +674,8 @@ where
/// admission: another session can store an offset while it waits, and a
/// commit of this poll would then pass over a rewind.
fn validate_waiter(&self, waiter: &Waiter) -> Result<PollVisibility, IggyError> {
- let (history, visibility, view) = self.validate_deferred_context(
- waiter.namespace,
- &waiter.request.context,
- waiter.request.consumer,
- )?;
+ let (history, visibility, view) =
+ self.validate_deferred_context(waiter.namespace, &waiter.request)?;
if history != waiter.history
|| view != waiter.view
|| (waiter.next
@@ -663,8 +719,7 @@ where
}
let admission = (|| {
request.options.validate(request.args.count)?;
- let (history, _, view) =
- self.validate_deferred_context(namespace, &request.context, request.consumer)?;
+ let (history, _, view) = self.validate_deferred_context(namespace, &request)?;
let config = *self.deferred_polls.config.borrow();
let mut state = self.deferred_polls.state.borrow_mut();
let id = state.admit(request.context.quota, &config)?;
@@ -758,6 +813,11 @@ where
{
continue;
}
+ if let Some(waiter) = state.readiness.get_mut(&id)
+ && !waiter.dirty.contains(&namespace)
+ {
+ waiter.dirty.push(namespace);
+ }
state.enqueue(id);
}
}
@@ -1051,19 +1111,27 @@ where
fn park_readiness_wait(
&self,
- wait: ReadinessWait,
+ mut wait: ReadinessWait,
reply: &Sender<PartitionReadReply>,
) -> Result<PartitionReadReply, IggyError> {
- let quota = wait
+ let first = wait
.targets
.first()
.ok_or(IggyError::InvalidCommand)?
- .context
- .quota;
+ .namespace;
+ if wait.targets.iter().any(|target| {
+ target.namespace.stream_id() != first.stream_id()
+ || target.namespace.topic_id() != first.topic_id()
+ }) {
+ return Err(IggyError::InvalidCommand);
+ }
+ self.validate_caller(first, &wait.context)?;
+ // A probe finds the target of a changed namespace by binary search.
+ wait.targets.sort_unstable_by_key(|target| target.namespace);
let mut admitted = Vec::with_capacity(wait.targets.len());
let mut partitions = Vec::new();
for target in &wait.targets {
- let (history, view, ready) = self.evaluate_readiness(target)?;
+ let (history, view, ready) = self.evaluate_readiness(wait.context.primary, target)?;
admitted.push((history, view));
if ready {
partitions.push(target.partition_id);
@@ -1084,6 +1152,7 @@ where
}
let config = *self.deferred_polls.config.borrow();
let mut state = self.deferred_polls.state.borrow_mut();
+ let quota = wait.context.quota;
let id = state.admit(quota, &config)?;
self.register_waiter(
&mut state,
@@ -1096,7 +1165,9 @@ where
ReadinessWaiter {
wait,
admitted,
- quota,
+ dirty: Vec::new(),
+ swept: 0,
+ sweep: 0,
reply: reply.clone(),
},
);
@@ -1109,13 +1180,14 @@ where
/// Returns the history and view of the target, and whether a poll with its
/// strategy returns a message now. This is the empty-plan test of
- /// `build_poll_plan`.
+ /// `build_poll_plan`. The caller of the wait is checked before.
fn evaluate_readiness(
&self,
+ primary: bool,
target: &ReadinessTarget,
) -> Result<(PollHistoryId, u32, bool), IggyError> {
let (history, visibility, view) =
- self.validate_deferred_context(target.namespace, &target.context, target.consumer)?;
+ self.validate_partition(target.namespace, primary, &target.metadata, target.consumer)?;
let start = match target.strategy.kind {
PollingKind::Offset => target.strategy.value,
PollingKind::Next => self.next_offset(target.namespace, target.consumer)?,
@@ -1128,7 +1200,7 @@ where
))
}
- fn probe_readiness(&self, id: u64, waiter: ReadinessWaiter) {
+ fn probe_readiness(&self, id: u64, mut waiter: ReadinessWaiter) {
if waiter.reply.is_disconnected() {
self.metrics.deferred_polls.cancelled.inc();
self.remove_readiness_wait(id, &waiter);
@@ -1140,21 +1212,63 @@ where
self.remove_readiness_wait(id, &waiter);
return;
}
- let partitions = waiter
- .wait
- .targets
- .iter()
- .zip(&waiter.admitted)
- .filter_map(|(target, admitted)| match self.evaluate_readiness(target) {
- Ok((history, view, ready)) if (history, view) == *admitted => {
- ready.then_some(Ok(target.partition_id))
- }
- Ok(_) => Some(Err(IggyError::TransientNotAccepted)),
- Err(error) => Some(Err(error)),
+ let targets = &waiter.wait.targets;
+ // A commit marks only its own namespace. The probe evaluates the
+ // targets of marked namespaces, also of those that the pump has not
+ // taken yet, and the targets that maintenance assigned. Maintenance
+ // reaches every other target in turn. A lookup per target keeps the
+ // cost of the marks that the shard still holds within the wait size.
+ let mut evaluated: Vec<usize> = {
+ let signals = self.deferred_polls.signals.borrow();
+ waiter
+ .dirty
+ .iter()
+ .filter_map(|namespace| {
+ targets
+ .binary_search_by_key(namespace, |target| target.namespace)
+ .ok()
+ })
+ .chain(
+ targets
+ .iter()
+ .enumerate()
+ .filter(|(_, target)| signals.queued.contains(&target.namespace))
+ .map(|(index, _)| index),
+ )
+ .chain(waiter.swept..waiter.swept + waiter.sweep)
+ .collect()
+ };
+ evaluated.sort_unstable();
+ evaluated.dedup();
+ let context = &waiter.wait.context;
+ let partitions = targets
+ .first()
+ .map_or(Ok(()), |first| {
+ self.validate_caller(first.namespace, context)
})
- .collect::<Result<Vec<_>, _>>();
+ .and_then(|()| {
+ evaluated
+ .iter()
+ .filter_map(|index| targets.get(*index).zip(waiter.admitted.get(*index)))
+ .filter_map(|(target, admitted)| {
+ match self.evaluate_readiness(context.primary, target) {
+ Ok((history, view, ready)) if (history, view) == *admitted => {
+ ready.then_some(Ok(target.partition_id))
+ }
+ Ok(_) => Some(Err(IggyError::TransientNotAccepted)),
+ Err(error) => Some(Err(error)),
+ }
+ })
+ .collect::<Result<Vec<_>, _>>()
+ });
let answer = match partitions {
Ok(partitions) if partitions.is_empty() => {
+ waiter.dirty.clear();
+ waiter.swept += waiter.sweep;
+ waiter.sweep = 0;
+ if waiter.swept >= waiter.wait.targets.len() {
+ waiter.swept = 0;
+ }
self.deferred_polls
.state
.borrow_mut()
@@ -1183,7 +1297,7 @@ where
self.unwatch_partitions(self.deferred_polls.remove(
id,
&[waiter.wait.deadline],
- waiter.quota,
+ waiter.wait.context.quota,
waiter.wait.targets.iter().map(|target| target.namespace),
));
}
diff --git a/core/shard/src/poll/deferred_tests.rs b/core/shard/src/poll/deferred_tests.rs
index 98ecf367c..f3ddb9507 100644
--- a/core/shard/src/poll/deferred_tests.rs
+++ b/core/shard/src/poll/deferred_tests.rs
@@ -20,8 +20,9 @@
use std::time::Duration;
use super::*;
-use consensus::PartitionsHandle;
+use consensus::{ClientTable, PartitionsHandle, build_reply_message_with};
use iggy_binary_protocol::requests::partitions::InstallConsumerGroupOwnerRequest;
+use iggy_binary_protocol::{Operation, PrepareHeader};
use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyByteSize, IggyTimestamp};
use metadata::IggyMetadata;
use metadata::stm::consumer_group::{ConsumerGroup, ConsumerGroupMember};
@@ -148,9 +149,9 @@ fn request(owner: &TestShard, count: u32, group: bool, auto_commit: bool) -> Def
count,
auto_commit,
},
+ metadata,
DeferredPollContext {
session: None,
- metadata,
user_id: 0,
user_created_at: None,
quota: PollQuotaIdentity::Session(CLIENT),
@@ -991,14 +992,17 @@ fn readiness_target(
partition_id: u32::try_from(partition).unwrap(),
consumer: PollingConsumer::Consumer(0, partition),
strategy,
- context: DeferredPollContext {
- session: None,
- metadata,
- user_id: 0,
- user_created_at: None,
- quota: PollQuotaIdentity::Session(CLIENT),
- primary: true,
- },
+ metadata,
+ }
+}
+
+fn readiness_caller() -> DeferredPollContext {
+ DeferredPollContext {
+ session: None,
+ user_id: 0,
+ user_created_at: None,
+ quota: PollQuotaIdentity::Session(CLIENT),
+ primary: true,
}
}
@@ -1006,6 +1010,7 @@ fn wait(owner: &TestShard, targets: Vec<ReadinessTarget>) -> Receiver<PartitionR
let (reply, replies) = channel(2);
owner.admit_readiness_wait(
ReadinessWait {
+ context: readiness_caller(),
targets,
deadline: DEADLINE,
},
@@ -1178,6 +1183,125 @@ async fn replacement_history_rejects_parked_readiness() {
assert_released(&owner);
}
+/// An owner whose partitions each hold one message, and a parked wait for
+/// the next message of every partition.
+async fn parked_wait_on(partitions: usize) -> (TestShard, Receiver<PartitionReadReply>) {
+ let owner = owner_of(&vec![&["first"][..]; partitions]).await;
+ let targets = (0..partitions)
+ .map(|partition| readiness_target(&owner, partition, PollingStrategy::offset(1)))
+ .collect();
+ let replies = wait(&owner, targets);
+ assert!(answer(&replies).0.is_some());
+ (owner, replies)
+}
+
+/// The partition refuses reads but marks nothing, so only an evaluation of
+/// its target finds it.
+fn begin_state_transfer(owner: &TestShard, partition: usize) {
+ owner
+ .plane
+ .partitions()
+ .with_partition(&partition_namespace(partition), |partition| {
+ partition.consensus().begin_state_transfer_await();
+ })
+ .unwrap();
+}
+
+#[compio::test]
+async fn commit_evaluates_only_the_readiness_target_of_its_partition() {
+ let (owner, replies) = parked_wait_on(MAINTENANCE_BUDGET + 1).await;
+ begin_state_transfer(&owner, MAINTENANCE_BUDGET);
+ commit(&owner, &["second"], 2).await;
+ owner.service_deferred_polls().await;
+ assert_eq!(answer(&replies), (None, vec![0]));
+ assert_released(&owner);
+}
+
+#[compio::test]
+async fn maintenance_evaluates_a_large_readiness_wait_one_budget_per_tick() {
+ let (owner, replies) = parked_wait_on(MAINTENANCE_BUDGET + 1).await;
+ let tick = async || {
+ owner.deferred_polls.maintenance();
+ owner.service_deferred_polls().await;
+ };
+ // The first pass takes two ticks. The second starts again at the first
+ // target, so it reaches the last one on the fourth tick.
+ tick().await;
+ tick().await;
+ begin_state_transfer(&owner, MAINTENANCE_BUDGET);
+ tick().await;
+ pending(&replies);
+ tick().await;
+ assert!(matches!(
+ replies.try_recv(),
+ Ok(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted
+ ))
+ ));
+ assert_released(&owner);
+}
+
+#[compio::test]
+async fn stale_session_rejects_parked_readiness_on_a_commit() {
+ let owner = owner(&["first"]).await;
+ let mut table = ClientTable::new(1);
+ let registration = PrepareHeader {
+ client: CLIENT,
+ operation: Operation::Register,
+ op: 1,
+ ..Default::default()
+ };
+ table
+ .commit_register(
+ CLIENT,
+ 0,
+ [0x5a; 32],
+ build_reply_message_with(®istration, 0, |_| {}),
+ )
+ .unwrap();
+ let session = table.attach_session(CLIENT, 1, 0).unwrap();
+ let (reply, replies) = channel(2);
+ owner.admit_readiness_wait(
+ ReadinessWait {
+ context: DeferredPollContext {
+ session: Some(session),
+ ..readiness_caller()
+ },
+ targets: vec![readiness_target(&owner, 0, PollingStrategy::offset(1))],
+ deadline: DEADLINE,
+ },
+ &reply,
+ );
+ assert!(answer(&replies).0.is_some());
+ drop(table);
+ commit(&owner, &["second"], 2).await;
+ owner.service_deferred_polls().await;
+ assert!(matches!(
+ replies.try_recv(),
+ Ok(PartitionReadReply::Rejected(IggyError::StaleClient))
+ ));
+ assert_released(&owner);
+}
+
+#[compio::test]
+async fn readiness_refuses_targets_of_more_than_one_topic() {
+ let owner = owner(&["first"]).await;
+ let mut other_topic = readiness_target(&owner, 0, PollingStrategy::offset(1));
+ other_topic.namespace = IggyNamespace::new(0, 1, 0);
+ let replies = wait(
+ &owner,
+ vec![
+ readiness_target(&owner, 0, PollingStrategy::offset(1)),
+ other_topic,
+ ],
+ );
+ assert!(matches!(
+ replies.try_recv(),
+ Ok(PartitionReadReply::Rejected(IggyError::InvalidCommand))
+ ));
+ assert_released(&owner);
+}
+
#[compio::test]
async fn readiness_quota_counts_parked_waiters_only() {
let owner = owner(&["first"]).await;
@@ -1259,7 +1383,10 @@ async fn fan_out_merges_admissions_and_cancels_what_stays_parked() {
.unwrap();
}
};
- let (ready, ()) = futures::join!(owner.wait_for_messages(targets, duration), owners);
+ let (ready, ()) = futures::join!(
+ owner.wait_for_messages(readiness_caller(), targets, duration),
+ owners
+ );
if later {
assert_eq!(ready.unwrap(), [1]);
let Ok(ShardFrame::Lifecycle(LifecycleFrame::ReadinessCancel { waiter })) =