You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "diegomrsantos (via GitHub)" <gi...@apache.org> on 2026/09/10 09:48:02 UTC
[PR] fix(partitions): complete polls on the owning shard [iggy]
diegomrsantos opened a new pull request, #4119:
URL: https://github.com/apache/iggy/pull/4119
Fixes #4117.
## Problem
A disk poll can remain pending while purge replaces a partition's message history and resets its offsets to zero. The existing `PollPlan` carries shared handles that allow its detached worker to update consumer offsets and group `last_polled` state after the read finishes. A result created before purge can therefore update the new history, even when its numeric offset is valid again because new messages now occupy the same offsets.
For example, an old poll for offsets 0 through 2 can complete after purge and record offset 2. If five new messages have since been appended at offsets 0 through 4, the consumer's next `Next` poll starts at offset 3 and skips the first three new messages.
## Implementation
- `PollPlan` now contains read resources and immutable request context. Its worker returns an owned `PollReadResult` and cannot mutate consumer progress.
- Disk results return to the existing shard inbox through a restricted completion sender. The partition owner validates the captured history and completes the poll synchronously.
- A process local `PollHistoryId` changes before purge, state installation, partition replacement, and other paths that replace message history. Results from an older history are rejected with `TransientNotAccepted` before progress changes or a reply is authorized.
- Automatic commit admission, capacity reservation, progress updates, and group `last_polled` updates now occur behind the partition completion API. Replication remains asynchronous, so a successful poll reply does not wait for the durable commit.
- Each queued automatic commit carries its request, history identity, and capacity reservation together. Stale queued commits are discarded before operation assignment, and the reservation table keyed only by consumer is removed.
- Resident reads remain inline and use the same partition completion rules.
The benchmark support is a separate commit. It adds explicit polling and latency selectors, raw poll observations, run manifests, bounded deadlines, pooled latency statistics, and reusable campaign scripts.
## Correctness verification
- The baseline reproduces the delayed disk poll failure after purge, including a new history in which the stale offset is numerically valid.
- The candidate passes the affected crate suites on Linux with 791 tests and ten targeted cluster, purge, and reconnection integration tests.
- Native server, shard, and simulator verification passes 620 tests, including a successful reply becoming available before deliberately stalled replication completes.
- Benchmark verification passes 17 Rust tests, three report tests, and nine tests for the campaign scripts.
- Formatting, dependency sorting with `--no-format`, changed TOML formatting, Markdown lint, whitespace checks, and scoped newline checks pass.
- Clippy with all features, all targets, and warnings denied passes for the affected crates on Linux. On macOS, the same check reaches existing `unused_self` and `unnecessary_wraps` findings in unchanged `core/server/src/shard_allocator.rs`; allowing those two categories passes.
## Performance
The comparison uses the same low level benchmark client for the baseline at `b692160e4` and this candidate. Each case has ten paired runs, with five pairs run baseline first and five run candidate first. Three A/A pairs per case provide a small noise diagnostic. All runs use TCP with `TCP_NODELAY` and 256 byte messages. No measured empty polls, errors, timeouts, or cancellations were recorded.
The values below are geometric means of the paired candidate to baseline ratios, expressed as percentage changes. Positive throughput favors the candidate; positive p99 means higher latency. Parentheses contain the lower and upper one sided 95% bootstrap bounds, which together form a central 90% interval.
| Case | Linux throughput | Linux p99 | macOS throughput | macOS p99 |
| --- | ---: | ---: | ---: | ---: |
| resident | +2.38% (-0.36% to +5.10%) | +3.60% (-7.49% to +16.82%) | +5.05% (-0.98% to +11.60%) | -7.48% (-21.96% to +11.03%) |
| `disk-offset` | -4.79% (-6.91% to -2.54%) | +9.44% (+4.04% to +14.80%) | -6.83% (-9.42% to -3.44%) | +1.35% (-7.81% to +9.96%) |
| `disk-next` | -2.22% (-5.12% to +0.80%) | +9.55% (-0.27% to +21.37%) | -4.49% (-8.42% to -0.42%) | +6.15% (-3.39% to +16.45%) |
| `disk-batch` | +1.10% (-1.75% to +4.17%) | -0.68% (-6.61% to +6.13%) | -5.57% (-10.81% to +0.01%) | +2.40% (-11.99% to +17.98%) |
| `disk-group` | -0.92% (-4.44% to +1.82%) | +1.33% (-4.51% to +10.46%) | -0.67% (-6.66% to +5.82%) | -4.63% (-18.27% to +11.47%) |
| `disk-writes` | -2.33% (-4.45% to -0.11%) | +7.43% (-1.53% to +18.09%) | +8.70% (+0.96% to +17.04%) | -18.55% (-35.16% to +3.31%) |
| `disk-fsync` | +7.12% (-6.47% to +21.69%) | -1.54% (-9.39% to +7.73%) | +0.35% (-12.90% to +18.86%) | +0.44% (-4.67% to +6.75%) |
The most repeatable cost is the `disk-offset` case. Median throughput changes from 3,609.0 to 3,461.6 polls per second in the Linux VM and from 5,515.1 to 5,016.2 on macOS. Median p99 changes from 515 to 557 microseconds on Linux and from 293 to 315 microseconds on macOS. Estimated CPU cost per completed poll rises by 5.63% on Linux and 4.60% on macOS.
The other cases have wider uncertainty. In particular, the resident, batched, group, and fsync measurements do not establish either a consistent gain or unchanged performance.
### Environment and limits
The Linux measurements ran in a native ARM64 container inside Docker Desktop's Linux virtual machine. The container used CPUs 0 through 3 and an 8 GiB memory limit, with the server on CPU 0 and clients on CPUs 1 and 2. Data lived in a Docker volume. The server used its required `io_uring` backend through a seccomp profile that permits the three `io_uring` system calls.
The macOS measurements ran on an Apple M1 Pro with 32 GiB RAM and the native I/O backend, without CPU affinity. Both versions used the repository's Rust 1.98.0 toolchain and identical release settings within each platform.
These results describe one sequential consumer against one shard. Group cases use one group consumer. Reads through the disk path may be served from the operating system file cache. The Linux result describes the Docker Desktop ARM64 virtual machine rather than native Linux hardware. The three A/A pairs expose noise but are too few to characterize it precisely, especially in the cases with background writes and fsync.
The reusable commands and analysis method are documented in `scripts/benchmarks/README.md`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997389576
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
Review Comment:
nonblocking docs follow-up: could we spell this out in the public poll api docs? a lost reply can come after the auto-commit cursor has advanced, so blindly retrying `Next` can skip messages the caller never saw. worth documenting recovery from client-known explicit offsets for each partition.
##########
core/partitions/src/iggy_partition.rs:
##########
Review Comment:
nonblocking: with `AutoCommitCtx` gone, `durable_consumer_offsets` and both capacity trackers no longer need the outer `Rc` wrappers. could we own those fields directly and keep sharing inside the reservation tokens/counters? that would save three allocations and indirections per partition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3994067578
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
Review Comment:
I completed the remeasurement and a separate experiment with reserved completion
capacity. The reserved lane passed its scoped correctness checks on a699ff91c, but it
did not show a performance benefit, so it remains an experimental patch.
The first comparison used a699ff91c6b9126502ea112ac676d96d8d107975 against
dc2b382097ed7fb6cfeb8a31fed0fbc74af4433e, the master revision already merged
into the PR. Throughput was 4.01% lower without concurrent writes, with a 95%
interval of 1.94% to 6.26% lower, and 5.41% lower with writes, with an interval
of 2.59% to 8.41% lower.
Both campaigns predate 7cdf64cfa6096d8a1cf1fa807cd22c41469015bb, which adds the
disconnected receiver check. These results do not establish the latest PR
revision's performance.
The second comparison used a699ff91c as the baseline and added a reserved
completion lane. It keeps the existing allocation and owner history validation.
Throughput changed by +0.64% without writes, interval −0.64% to +2.40%, and
−3.42% with writes, interval −7.67% to approximately zero. With writes, p99
latency changed by +13.86%, interval +0.31% to +29.94%.
For the concurrent producer cases, medians across runs were:
| Experiment and version | Producer batches per poll | Written messages per poll | Median poll latency | p99 poll latency |
| --- | ---: | ---: | ---: | ---: |
| First comparison, master | 0.586 | 58.60 | 268.5 µs | 575.5 µs |
| First comparison, PR | 0.613 | 61.28 | 278.5 µs | 603.5 µs |
| Lane experiment, PR | 0.564 | 56.37 | 271.0 µs | 507.5 µs |
| Lane experiment, reserved lane | 0.574 | 57.40 | 274.0 µs | 574.0 µs |
The producer target is 50 MB/s in batches of 100 messages. These are logical
producer operations, not disk syscalls. The two experiments ran at different
times, so their absolute values should not be compared across experiments.
Each case has ten pairs of 33,000 explicit offset polls, with five pairs in each
order, one shard and one sequential TCP consumer. Each poll returns one 256 byte
message; automatic commits are disabled. Percentage changes are geometric means
of paired ratios, and the intervals resample whole pairs.
Each experiment also has four baseline repeatability pairs per case. All
1,848,000 clean polls in each experiment passed the accounting audit.
This is a Linux VM on an M1 Pro with disk data that may be cached. One identical
binary pair in the lane experiment differed by 8.54% in throughput. The mixed
throughput interval ends very close to zero, so I would not claim a universal
slowdown from it. It does not support calling this implementation a performance
fix. Separate diagnostics also did not show a handoff improvement.
[Reports and evidence bundles](https://github.com/diegomrsantos/iggy/releases/tag/pr-4119-benchmark-evidence-2026-09-12) contain the source patch, revisions,
build and run commands, configuration, raw observations, diagnostic results,
validation logs, analysis scripts and SHA256 hashes. The evidence is hosted
on my fork; the lane change has not been added to this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on PR #4119:
URL: https://github.com/apache/iggy/pull/4119#issuecomment-5631709486
I’d prioritize this one, since it fixes the correctness issue. #4122 contains the benchmark tooling and works against master without this fix. Neither PR depends on the other, so they can be reviewed independently, but I’d merge this one first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on PR #4119:
URL: https://github.com/apache/iggy/pull/4119#issuecomment-5634296827
Added concise docs explaining the reservation ownership boundary. Previously, `AutoCommitSubmit` carried an existing guard through the shard inbox, whose `Send` bound applied even to sends to the same shard. Now the inbox carries read results, and the owner creates and retains guards in local request or replication state. This permits `Rc` and `Cell` while preserving synchronous counter updates. The docs also explain why capacity checking and acquisition must stay in one owner operation: individual atomics would not protect admission split across owner turns. This update changes documentation only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3992156459
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -15148,6 +15804,252 @@ mod retention_tests {
}
}
+#[cfg(test)]
+mod purge_poll_tests {
+ //! A disk poll with automatic commit starts reading messages at offsets 0-2.
+ //! Before it completes, purge removes those messages and clears consumer
+ //! progress. Fresh messages are then appended starting at offset 0.
+ //! Completing the old poll must not advance progress over the fresh messages.
+
+ use super::tests::{journal_send_batch, repair_config, test_partition};
+ use super::*;
+ use crate::PollFragments;
+ use iggy_common::PollingStrategy;
+ use server_common::send_messages::decode_batch_slice;
+
+ #[compio::test]
+ async fn given_pending_disk_poll_when_purged_should_preserve_fresh_consumer_progress() {
+ Box::pin(assert_delayed_poll_preserves_fresh_progress(true, 2)).await;
+ }
+
+ #[compio::test]
+ async fn given_pending_disk_poll_when_fresh_history_covers_old_offset_should_preserve_progress()
+ {
+ // The delayed poll targets old offsets 0-2. After purge, five fresh
+ // messages occupy offsets 0-4. Recording progress 2 from the old poll
+ // would make `Next` skip fresh offsets 0-2 even though 2 is in range.
+ Box::pin(assert_delayed_poll_preserves_fresh_progress(true, 5)).await;
+ }
+
+ #[compio::test]
+ async fn given_disk_poll_without_auto_commit_when_purged_should_preserve_fresh_progress() {
+ for fresh_message_count in [2, 5] {
+ Box::pin(assert_delayed_poll_preserves_fresh_progress(
+ false,
+ fresh_message_count,
+ ))
+ .await;
+ }
+ }
+
+ async fn assert_delayed_poll_preserves_fresh_progress(
+ auto_commit: bool,
+ fresh_message_count: u32,
+ ) {
+ let config = repair_config();
+ let (_directory, mut partition) = Box::pin(disk_poll_partition(&config)).await;
+ let consumer = PollingConsumer::Consumer(7, 0);
Review Comment:
Thanks, that makes sense. With an individual consumer and `auto_commit=false`, the old implementation did not update consumer progress, so the unchanged-offset and `Next` assertions do not demonstrate that the fix prevented a stale update.
The test still checks that the old result is rejected with `TransientNotAccepted`. There is also a separate group test using a constructed result, but we are missing a group test that exercises a real disk read across purge.
I’ll add that case with automatic commits disabled and ensure the delayed read produces a nonempty result. After purge and rejection of the old completion, it should verify that both the group’s last polled and committed offsets remain unset, before any fresh group read changes them. This would catch the old behavior where a delayed read restores a stale last polled mark after purge, potentially delaying partition handoff.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3992101216
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
Review Comment:
Thanks, I agree this extra step deserves attention. In `disk-offset`, the client tracks its next offset itself and automatic commits are disabled. The old worker could therefore reply directly after reading. With this PR, it allocates a completion and waits for the owner to process it before replying, so there is additional overhead in this case.
The measured slowdown is consistent with that extra work. Since the results describe an earlier revision, I’d like to rerun the comparison on the current code and check how much time goes into allocation and waiting for the owner. This case has no concurrent producer, so testing with concurrent writes would also help us understand the queue contention you mentioned.
The completion queue options discussed in the other thread seem worth exploring. We can preserve the owner’s history validation while looking for a cheaper way to return completed reads, then use the measurements to guide the choice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3992396053
##########
core/server/src/dispatch/partition.rs:
##########
@@ -830,6 +489,19 @@ where
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
+ None => {
+ // A timeout drops the receiver without canceling the owner. Its
+ // eventual completion may still accept progress, so neither an
+ // empty success nor TransientNotAccepted describes this outcome.
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ "partition read reply unavailable; acceptance unknown"
+ );
+ Err(ReadPolledMessagesError::Rejected(
+ IggyError::ShardCommunicationError,
Review Comment:
Thanks for pointing this out. HTTP already returned `504` when the partition owner did not reply before this PR, while binary transports returned a successful empty poll. This PR changes the binary response to `ShardCommunicationError` with code `11001`, so both now report a failure, but their error identities remain inconsistent.
The numbers represent different things. `11001` is an Iggy error code, whereas `504` is an HTTP status. However, the HTTP response also uses `504` as its JSON error ID, with `partition_read_timeout` as its code. We could preserve an appropriate HTTP status while exposing the same Iggy error identity in the body.
I suggest aligning this with the distinction discussed in the other thread. When we know the request was never submitted, both transports could return `TransientNotAccepted`, which already maps to HTTP `503`. When submission succeeded but no reply arrived, we should preserve the unknown outcome. For an actual timeout, HTTP could retain `504` while including error ID `11001`, without implying that the original poll had no effect.
This would need an explicit HTTP mapping. Passing `ShardCommunicationError` through the current generic handler would otherwise produce `400 Bad Request`, which would incorrectly attribute the failure to the client.
Does that approach match the contract you have in mind, or would you prefer a different mapping?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3998911146
##########
core/partitions/src/iggy_partitions.rs:
##########
@@ -495,6 +488,46 @@ where
Some(partition.build_poll_plan(consumer, args, validate_checksum))
}
+ /// Validate and accept a poll synchronously on the owning pump.
+ /// Send the reply before driving any returned continuation through
+ /// [`Self::replicate_poll_completion`]. Success does not acknowledge a
+ /// durable offset commit.
+ ///
+ /// Acceptance is independent of reply delivery. A caller that stopped
+ /// waiting does not cancel completion: an accepted read can still advance
+ /// progress and admit an automatic commit even if its reply cannot be sent.
+ /// A history rejection accepts no progress from this read, but cannot
+ /// replace a response the transport already sent.
+ ///
+ /// # Errors
+ /// Rejects stale history, unavailable partition or admission state, invalid
+ /// consumer identifiers, and exhausted consumer, queue, or journal capacity.
+ /// Rejection does not accept progress from this read.
+ pub fn complete_poll(
+ &self,
+ namespace: &IggyNamespace,
+ result: PollReadResult,
+ ) -> Result<PollCompletion, IggyError> {
+ let partition = self
+ .get_mut_by_ns(namespace)
+ .ok_or(IggyError::TransientNotAccepted)?;
+ partition.complete_poll(result)
+ }
+
+ /// Drive an accepted poll's replication on the owning pump after its reply.
+ /// Use the continuation returned by [`Self::complete_poll`] for this namespace.
+ /// This call may suspend with a partition borrow, so it must remain on the
+ /// owning pump. A missing or tombstoned namespace drops the continuation.
+ pub async fn replicate_poll_completion(
+ &self,
+ namespace: &IggyNamespace,
+ replication: PollReplication,
+ ) {
+ if let Some(partition) = self.get_mut_by_ns(namespace) {
Review Comment:
A namespace-lookup miss silently drops the continuation, and the doc above sanctions it. By this point `admit_poll_auto_commit` has already called `request.project(consensus)`, consuming an op number, and pushed the prepare into `prepare_queue`. Only `on_replicate` journals and broadcasts it, so a dropped continuation leaves an unjournaled prepare holding that op and stalls the partition's commit chain there. The sole caller in `core/shard/src/poll.rs` has no suspension point between `complete_poll` and this call, since `reply.try_send` is synchronous, so nothing can tombstone the namespace in that window and it is unreachable today. The concern is that the split between op assignment and journaling is new in this diff and the miss is documented as acceptable rather than asserted against.
##########
core/shard/src/metrics.rs:
##########
@@ -175,6 +177,7 @@ const VARIANTS: [&str; VARIANT_COUNT] = [
frame_drop_variant::REPLICA_HANDSHAKE_ACK,
frame_drop_variant::PARTITION_AUTO_COMMIT,
Review Comment:
This entry has no producer left. Deleting `submit_auto_commit_offset` and the `LifecycleFrame::AutoCommitSubmit` arm removed the only `record_frame_drop` call for it, which was at `core/shard/src/lib.rs:2129` on the base. A repo-wide grep on this head finds only the constant at line 115 and this array slot. `frame_drops_total{variant="partition_auto_commit"}` goes permanently silent after deploy while the code still reads as a live series, so any dashboard or alert keyed on that label breaks quietly instead of being explicitly retired.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -830,6 +489,19 @@ where
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
+ None => {
Review Comment:
This arm catches every `None`, so `reply_was_none = other.is_none()` in the fallback arm below it can never be true. The fallback now only fires on an unexpected `Some(_)` variant, realistically `PartitionReadReply::NotFound`. An operator reading `reply_was_none=false` on a "partition read failed; replying empt
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997811440
##########
core/server_common/src/poll.rs:
##########
@@ -0,0 +1,202 @@
+// 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.
+
+use std::cell::Cell;
+use std::rc::Rc;
+use std::sync::Arc;
+
+use iggy_common::ConsumerKind;
+
+/// Identity of one serviceable message history. It is never serialized.
+///
+/// The allocation identifies the history even if a rebuilt partition reuses
+/// its namespace and offsets. Pending reads keep it alive, preventing address
+/// reuse while they can still complete. `Arc` also keeps completion frames
+/// `Send`, as required by the shard inbox.
+/// `Default` creates a fresh identity. Only clones compare equal.
+#[derive(Clone, Debug, Default)]
+pub struct PollHistoryId(Arc<()>);
Review Comment:
Implemented in [98a6e0c00](https://github.com/apache/iggy/commit/98a6e0c00accac2e185394f33044ba93e065416f). `PollHistoryId` now holds a copied `u64`, assigned by a checked `AtomicU64` counter shared by the process. Exhaustion cannot wrap or reuse an ID, including across partition recreation. Individual polls only copy the ID; they do not access the counter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997019493
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
Review Comment:
Following up with longer measurements of the shutdown receive reuse patch. The local patch creates and pins `stop.recv().fuse()` once before the owner loop. Shutdown stays first in the biased select, and disk completions still return to the owner for validation.
This addresses an existing cost that the PR exercises more often. In the earlier profiles of master `dc2b38209` and PR `a699ff91c`, sampled CPU time in the shutdown receive function rose from about 4.1 to 7.8 µs per poll without concurrent writes, nearly doubling. With writes it rose from 6.83 to 10.35 µs per poll. The extra owner turn can recreate the receive and repeat Crossfire's backoff for an empty channel. These are sampled CPU estimates, not an additive explanation of the entire elapsed time regression.
A fresh comparison of `7cdf64cfa` against that revision plus reuse used 12 pairs per workload. These longer results supersede the preliminary gains above:
| Load | Elapsed µs/poll, before → after | Written messages/poll, before → after | Time reduction, 95% interval |
| --- | ---: | ---: | ---: |
| No producer | 270.009 → 262.870 | 0 → 0 | 2.64% [1.55%, 3.57%] |
| Producer targeting 50 MB/s | 310.369 → 298.296 | 60.97 → 58.73 | 3.89% [2.59%, 5.33%] |
Fresh resident and consumer group checks, with 12 pairs each, also showed smaller mean elapsed times. The patch passed 244 shard and simulator tests, including shutdown wakeup, draining queued completions and sender disconnection. Five existing simulator tests remain ignored.
I then ran a fresh comparison of `7cdf64cfa` plus reuse against preserved master `dc2b38209`, with 64 pairs per workload:
| Load | Master → PR + reuse, µs/poll | Written messages/poll, master → PR + reuse | Slowdown, 95% interval |
| --- | ---: | ---: | ---: |
| No producer | 234.201 → 234.771 | 0 → 0 | +0.24% [−0.08%, +0.61%] |
| Producer targeting 50 MB/s | 258.465 → 260.416 | 50.90 → 51.34 | +0.75% [+0.34%, +1.20%] |
All 256 runs completed: 8,448,000 measured polls, with 33,000 polls after 100 warmup polls per run. Each poll returns one 256 byte message, using explicit offsets with automatic commits disabled. Version order was randomized and balanced within each workload. All runs passed the raw workload audit, and an independent verifier reproduced the primary estimates and intervals. The concurrent producer sustained 50.09–50.60 MB/s in batches of 100 messages.
Elapsed times and written messages per poll are means across runs. The mean time intervals use 20,000 resamples of whole pairs. The two comparisons ran at different times, so absolute values should be compared within each table. Writes are logical producer operations, not disk syscalls.
The same fresh comparison measured p99 poll latency:
| Load | p99 change | Adjusted interval |
| --- | ---: | ---: |
| No producer | +1.57% | −0.79% to +5.20% |
| Producer targeting 50 MB/s | −1.33% | −4.13% to +0.66% |
These are geometric means of paired ratios of p99 values from individual runs. Positive values mean higher latency. The intervals use 100,000 resamples of whole pairs and retain the preregistered 98.75% bounds per workload, accounting for both workloads and two planned checkpoints. With concurrent writes, the upper bound is below the 3% slowdown threshold. Without writes, that threshold remains unresolved. All pairs remain included.
In these ARM64 Linux VM workloads, the estimated remaining mean overhead is below 1%, and both 95% intervals end below 1.3%. The comparison with reuse on both master and PR mentioned above remains outstanding, so the benefit of applying reuse to plain master and the exact fraction of the regression it explains are still unmeasured.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997334241
##########
core/server/src/dispatch/partition.rs:
##########
@@ -830,6 +489,19 @@ where
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
+ None => {
+ // A timeout drops the receiver without canceling the owner. Its
+ // eventual completion may still accept progress, so neither an
+ // empty success nor TransientNotAccepted describes this outcome.
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ "partition read reply unavailable; acceptance unknown"
+ );
+ Err(ReadPolledMessagesError::Rejected(
+ IggyError::ShardCommunicationError,
Review Comment:
Fixed in [736c4a025](https://github.com/apache/iggy/commit/736c4a02561f8876be4fdc48dba6f4bef06f9bcb). Missing partition read replies now keep HTTP 504 and carry `ShardCommunicationError` (ID `11001`) in the JSON body. Regression tests cover this mapping and HTTP 503 with `TransientNotAccepted`. All 451 server tests and strict Clippy passed on macOS.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3993369536
##########
core/simulator/src/lib.rs:
##########
@@ -1701,6 +1706,87 @@ mod tests {
panic!("request {request_id} did not receive a reply");
}
+ #[test]
+ fn given_admitted_resident_poll_when_replica_send_stalls_should_reply_before_replication() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+ let client_id = 1u128;
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let mut sim = Simulator::new(
+ 3,
+ std::iter::once(client_id),
+ packet::PacketSimulatorOptions {
+ node_count: 3,
+ client_count: 1,
+ seed: 0xC011_EC71,
+ ..packet::PacketSimulatorOptions::default()
+ },
+ );
+ sim.init_partition(namespace);
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+ let produced = submit_and_wait_for_reply(
+ &mut sim,
+ client_id,
+ 0,
+ client.send_messages(
+ namespace,
+ &[Bytes::from_static(b"reply-before-replication")],
+ ),
+ );
+ assert_eq!(produced.header().status, 0);
+ sim.run_pumps();
+
+ let owner = Rc::clone(sim.replicas[0].partition_shard(namespace));
+ let consumer = PollingConsumer::Consumer(7, 0);
+ let args = PollingArgs::new(iggy_common::PollingStrategy::next(), 1, true);
+ let plan = owner
+ .plane
+ .partitions()
+ .build_poll_snapshot(&namespace, consumer, &args)
+ .expect("poll snapshot");
+ assert!(
+ !plan.needs_off_pump_io(),
Review Comment:
Added in [a699ff91c](https://github.com/apache/iggy/commit/a699ff91c6b9126502ea112ac676d96d8d107975). The new regression holds a group read result, replaces the partition with different messages at the same offsets, and delivers the old result through the actual completion sender, inbox, router, and owner pump. It checks `TransientNotAccepted` and both group offsets remaining unset before a fresh result succeeds through the same route. In an isolated copy, disabling the history check made the test fail because the stale result restored `last_polled` to `Some(2)`, confirming the delayed result is nonempty and the regression catches stale progress. All 99 shard tests and strict Clippy passed on both macOS and Linux.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4119:
URL: https://github.com/apache/iggy/pull/4119#issuecomment-5631622901
@diegomrsantos which one should be reviewed/merged first, this one or #4122?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3996312330
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
Review Comment:
Following up on the extra pump turn: profiles pointed to CPU work in the shutdown receive. `run_message_pump` recreates `stop.recv().fuse()` on every turn, repeating Crossfire’s initial backoff and receive registration while the channel is empty. I tested a local patch that pins this future once outside the loop and reuses it. Completions still return to the owner for validation.
Against `7cdf64cf`, I ran four pairs per load on the same ARM64 Linux VM: 16 runs of 33,000 explicit offset polls, one sequential TCP consumer, 256 bytes per message and automatic commits disabled.
| Load | Elapsed µs per poll, before → after | Throughput change | Written messages per poll, before → after |
| --- | ---: | ---: | ---: |
| No producer | 304.27 → 273.41 | +11.27% | 0 → 0 |
| Producer targeting 50 MB/s | 369.44 → 340.29 | +8.14% | 71.51 → 66.30 |
Elapsed time and write counts are means across runs; throughput changes are geometric means of paired ratios. The producer sustained 48.5–50.4 MB/s in batches of 100. All four pairs improved without writes; three improved with writes, where results were noisier. All polls passed the audit. The 244 shard and simulator tests passed, including new shutdown coverage.
The patch is still local and these results are preliminary. Since master also recreates this receive, the next comparison will test preserved master `dc2b38209` and PR `a699ff91c`, each with and without reuse, to measure the remaining gap under both loads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4119:
URL: https://github.com/apache/iggy/pull/4119#issuecomment-5617060199
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4119?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 85.58%. Comparing base ([`1290e4f`](https://app.codecov.io/gh/apache/iggy/commit/1290e4fffcfca85abdad6a213567d61e1f2f21f4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`bb71934`](https://app.codecov.io/gh/apache/iggy/commit/bb71934bce1920b99eaef605476c2d05f76ddde5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4119 +/- ##
============================================
- Coverage 86.01% 85.58% -0.43%
Complexity 1451 1451
============================================
Files 1248 1246 -2
Lines 194939 193875 -1064
Branches 160243 159384 -859
============================================
- Hits 167669 165926 -1743
- Misses 23163 23855 +692
+ Partials 4107 4094 -13
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4119/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/4119/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `86.90% <ø> (-0.02%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4119/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `67.56% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4119/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `77.06% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4119/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.31% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4119/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.65% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4119/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `89.96% <ø> (-6.30%)` | :arrow_down: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4119/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `69.40% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/4119?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/consensus/src/impls.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fconsensus%2Fsrc%2Fimpls.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25zZW5zdXMvc3JjL2ltcGxzLnJz) | `92.76% <ø> (ø)` | |
| [core/partitions/src/consumer\_offset\_capacity.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Fconsumer_offset_capacity.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9jb25zdW1lcl9vZmZzZXRfY2FwYWNpdHkucnM=) | `98.97% <ø> (-0.06%)` | :arrow_down: |
| [core/partitions/src/iggy\_partition.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Figgy_partition.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9pZ2d5X3BhcnRpdGlvbi5ycw==) | `91.95% <ø> (-0.08%)` | :arrow_down: |
| [core/partitions/src/iggy\_partitions.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Figgy_partitions.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9pZ2d5X3BhcnRpdGlvbnMucnM=) | `85.88% <ø> (+0.47%)` | :arrow_up: |
| [core/partitions/src/lib.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9saWIucnM=) | `0.00% <ø> (ø)` | |
| [core/partitions/src/poll\_plan.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Fpoll_plan.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9wb2xsX3BsYW4ucnM=) | `92.54% <ø> (-1.19%)` | :arrow_down: |
| [core/partitions/src/state\_transfer.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Fstate_transfer.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9zdGF0ZV90cmFuc2Zlci5ycw==) | `69.60% <ø> (+0.02%)` | :arrow_up: |
| [core/server/src/boot/mod.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fboot%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Jvb3QvbW9kLnJz) | `87.30% <ø> (-0.03%)` | :arrow_down: |
| [core/server/src/boot/recovery.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fboot%2Frecovery.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Jvb3QvcmVjb3ZlcnkucnM=) | `95.44% <ø> (-0.02%)` | :arrow_down: |
| [core/server/src/dispatch/mod.rs](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fdispatch%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Rpc3BhdGNoL21vZC5ycw==) | `93.05% <ø> (+0.10%)` | :arrow_up: |
| ... and [9 more](https://app.codecov.io/gh/apache/iggy/pull/4119?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [54 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4119/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3991530282
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
Review Comment:
Agreed. With `auto_commit=true`, the internal timeout can drop the receiver while the owner still accepts the result and advances the cursor. A subsequent `Next` can then skip the batch the client never received.
The previous implementation also advanced progress before attempting the reply. On timeout, however, it returned a successful empty poll, even though the detached read could still finish and advance progress. This PR returns `ShardCommunicationError` instead, so the caller is told that the read failed to produce a reply rather than being given an empty success. That improves the reported outcome, but it does not cancel the read or prevent later progress updates. The new completion queue also adds another delay window.
I’ll add a regression that delays a nonempty completion past the timeout, then checks the cursor and a subsequent `Next`. What contract should we guarantee here? Should timeout cancel any poll that has not yet been admitted, ensuring it cannot advance progress afterward? Or is eventual admission acceptable if the caller receives an error indicating an unknown outcome? Once admission has happened, a lost reply can still leave the outcome unknown.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3991621992
##########
core/server/src/dispatch/partition.rs:
##########
@@ -830,6 +489,19 @@ where
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
+ None => {
Review Comment:
Good point. `partition_read` already combined submission failures and unavailable replies into `None` before this PR. The binary poll handler previously converted all those cases into a successful empty poll. I changed that to `ShardCommunicationError` because a timeout can leave the owner running and advancing progress, making empty success misleading.
However, that mapping is too broad for the cases you identified. If routing fails, the sender is missing, or the initial inbox rejects the request, we know this poll never reached the owner.
I propose returning `TransientNotAccepted` for those known submission failures, allowing the existing SDK retry logic to handle them. Timeouts and lost replies after submission would retain `ShardCommunicationError`, since acceptance remains uncertain.
Does that distinction make sense to you, or would you prefer a different error mapping or approach?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3992324714
##########
core/server_common/src/poll.rs:
##########
@@ -0,0 +1,202 @@
+// 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.
+
+use std::cell::Cell;
+use std::rc::Rc;
+use std::sync::Arc;
+
+use iggy_common::ConsumerKind;
+
+/// Identity of one serviceable message history. It is never serialized.
+///
+/// The allocation identifies the history even if a rebuilt partition reuses
+/// its namespace and offsets. Pending reads keep it alive, preventing address
+/// reuse while they can still complete. `Arc` also keeps completion frames
+/// `Send`, as required by the shard inbox.
+/// `Default` creates a fresh identity. Only clones compare equal.
+#[derive(Clone, Debug, Default)]
+pub struct PollHistoryId(Arc<()>);
Review Comment:
Thanks, I think a numeric ID is worth trying here. The current allocation is created when a history is created or invalidated, and polls clone the existing `Arc` without allocating a new history object. Switching to a number would still remove the reference count updates from each poll.
We can keep the implementation small by using an `AtomicU64` shared by the process to assign fresh history IDs, with a checked update that prevents wrapping and reuse. The counter would run only when the history changes. Each poll would carry a plain `u64`, so it would not access the atomic counter. This also preserves uniqueness when a partition is recreated with the same namespace.
I’ll retain the tests for partition replacement and stale results, and include the change in the planned performance comparison. It removes some polling overhead, although we should measure its effect separately from the completion allocation and queue processing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3992940440
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -15148,6 +15804,252 @@ mod retention_tests {
}
}
+#[cfg(test)]
+mod purge_poll_tests {
+ //! A disk poll with automatic commit starts reading messages at offsets 0-2.
+ //! Before it completes, purge removes those messages and clears consumer
+ //! progress. Fresh messages are then appended starting at offset 0.
+ //! Completing the old poll must not advance progress over the fresh messages.
+
+ use super::tests::{journal_send_batch, repair_config, test_partition};
+ use super::*;
+ use crate::PollFragments;
+ use iggy_common::PollingStrategy;
+ use server_common::send_messages::decode_batch_slice;
+
+ #[compio::test]
+ async fn given_pending_disk_poll_when_purged_should_preserve_fresh_consumer_progress() {
+ Box::pin(assert_delayed_poll_preserves_fresh_progress(true, 2)).await;
+ }
+
+ #[compio::test]
+ async fn given_pending_disk_poll_when_fresh_history_covers_old_offset_should_preserve_progress()
+ {
+ // The delayed poll targets old offsets 0-2. After purge, five fresh
+ // messages occupy offsets 0-4. Recording progress 2 from the old poll
+ // would make `Next` skip fresh offsets 0-2 even though 2 is in range.
+ Box::pin(assert_delayed_poll_preserves_fresh_progress(true, 5)).await;
+ }
+
+ #[compio::test]
+ async fn given_disk_poll_without_auto_commit_when_purged_should_preserve_fresh_progress() {
+ for fresh_message_count in [2, 5] {
+ Box::pin(assert_delayed_poll_preserves_fresh_progress(
+ false,
+ fresh_message_count,
+ ))
+ .await;
+ }
+ }
+
+ async fn assert_delayed_poll_preserves_fresh_progress(
+ auto_commit: bool,
+ fresh_message_count: u32,
+ ) {
+ let config = repair_config();
+ let (_directory, mut partition) = Box::pin(disk_poll_partition(&config)).await;
+ let consumer = PollingConsumer::Consumer(7, 0);
Review Comment:
Done in [70268ab7b](https://github.com/apache/iggy/commit/70268ab7b98c3573b1b7b02ee277cc00fc3c1556). The new test covers a real group disk read across purge with automatic commits disabled. It checks that the delayed result is nonempty and that rejecting it leaves both `last_polled` and the committed offset unset before any fresh read.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3993500488
##########
core/server/src/dispatch/partition.rs:
##########
@@ -830,6 +489,19 @@ where
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
+ None => {
Review Comment:
Yes, split it that way. TransientNotAccepted for the three that prove the read
never started, ShardCommunicationError only once submission succeeded.
One addition. Do not just report the uncertainty, shrink it. Probe
crossfire::Sender::is_disconnected() on the reply sender in complete_poll
before admit_poll_auto_commit. A timed-out caller is already gone when the read
lands, so the completion rejects and nothing advances. ShardCommunicationError
then means a rare lost reply instead of the common case.
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,461 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+#[cfg(test)]
+mod completion_tests;
+#[cfg(test)]
+mod test_support;
+#[cfg(test)]
+mod timeout_tests;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
Review Comment:
Needs the re-measurement before merge, with a concurrent producer. This hop is
on every disk poll and the posted numbers describe an earlier revision.
Report writes per poll next to latency. The completion adds a frame and an
allocation, no I/O, so a surviving regression is queueing behind consensus and
write frames, which the reserved lane also fixes.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2840,19 +2866,282 @@ where
}
}
- /// Reject completion from a removed partition or a view whose retained
- /// prepares have not yet been accounted for by the pump.
- #[must_use]
- pub fn auto_commit_admission_ready(&self, applied: &crate::AutoCommitApplied) -> bool {
- applied.belongs_to(&self.durable_consumer_offsets)
- && self.observed_view == self.consensus.view()
+ /// Accept a read only while its message history still belongs to this owner.
+ /// Validation, admission, and progress updates must stay in one synchronous
+ /// owner turn so purge or recovery cannot interleave between them.
+ /// Nonempty group reads update `last_polled` even without automatic commits.
+ /// Rejection leaves this read's progress unapplied.
+ pub(crate) fn complete_poll(
+ &mut self,
+ result: PollReadResult,
+ ) -> Result<PollCompletion, IggyError> {
+ self.resynchronize_consumer_offset_reservations();
Review Comment:
Move this below the history guard. Every rejected stale completion pays the
reconciliation before being thrown away.
##########
core/shard/src/poll/timeout_tests.rs:
##########
@@ -0,0 +1,277 @@
+// 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.
+
+use std::cell::RefCell;
+use std::rc::Rc;
+use std::time::Duration;
+
+use consensus::PartitionsHandle;
+use futures::channel::oneshot;
+use iggy_common::PollingStrategy;
+use message_bus::{
+ BusMessage, ClientForwardFn, ConnectionLostFn, JoinHandle, MessageBus, ReplicaForwardFn,
+ SendError,
+};
+use metadata::IggyMetadata;
+use partitions::{IggyPartitions, PollFragments, PollingArgs, PollingConsumer};
+use server_common::MESSAGE_ALIGN;
+use server_common::iobuf::Frozen;
+use server_common::send_messages::decode_batch_slice;
+use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+
+use super::completion::PollCompletionSender;
+use super::test_support::{PollTestMetadata, partition_with_messages};
+use crate::shards_table::{PapayaShardsTable, ShardsTable};
+use crate::{
+ IggyShard, LifecycleFrame, PARTITION_READ_TIMEOUT, PartitionConsensusConfig, PartitionRead,
+ PartitionReadReply, ReplicaTopology, ShardFrame, ShardIdentity, shard_channel,
+};
+
+/// Characterize the current timeout behavior: losing the caller does not cancel
+/// owner acceptance. This records the unseen messages that a later Next skips;
+/// it does not prescribe the eventual cancellation contract.
+#[compio::test]
+#[allow(clippy::too_many_lines)]
+async fn given_auto_commit_poll_when_completion_arrives_after_timeout_should_skip_unseen_messages_on_next()
+ {
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let consumer = PollingConsumer::Consumer(7, 0);
+ let (expire_timeout, timeout_elapsed) = oneshot::channel();
+ let bus = PollTimeoutBus {
+ next_timeout: Rc::new(RefCell::new(Some(timeout_elapsed))),
+ };
+ let mut owner = owner_with_messages(&bus, namespace).await;
+ let (owner_sender, owner_inbox, _owner_reply_lane) = shard_channel(0, 2, 1);
+ owner.attach_senders(vec![owner_sender.clone()]);
+ let partitions = owner.plane.partitions();
+ let (stored_offset, partition_offset) = partitions
+ .consumer_offset_read(&namespace, consumer)
+ .expect("fixture partition exists");
+ assert_eq!(stored_offset, None);
+ assert_eq!(
+ partition_offset, 3,
+ "four messages are available at offsets 0 through 3"
+ );
+
+ // Start a real routed request, but let the test control when its completed
+ // read reaches the owner. The fourth message will identify the next batch.
+ let first_poll = owner.partition_read(
+ namespace,
+ PartitionRead::Poll {
+ consumer,
+ args: PollingArgs {
+ strategy: PollingStrategy::next(),
+ count: 3,
+ auto_commit: true,
+ },
+ },
+ );
+ futures::pin_mut!(first_poll);
+ assert!(futures::poll!(first_poll.as_mut()).is_pending());
+ let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead {
+ namespace: requested_namespace,
+ read:
+ PartitionRead::Poll {
+ consumer: requested_consumer,
+ args,
+ },
+ reply,
+ }) = owner_inbox
+ .try_recv()
+ .expect("poll reached the owner inbox")
+ else {
+ panic!("expected the routed poll request");
+ };
+ let read_plan = partitions
+ .build_poll_snapshot(&requested_namespace, requested_consumer, &args)
+ .expect("poll has a read snapshot");
+ // Resident bytes keep disk scheduling out of this test. Holding the result
+ // here models the delay before acceptance shared with detached disk reads.
+ assert!(!read_plan.needs_off_pump_io());
+ let delayed_result = read_plan.execute_resident();
+
+ // The reply sender is still alive, so None must come from the timeout.
+ expire_timeout
+ .send(())
+ .expect("requester is waiting on its timer");
+ assert!(
+ first_poll.await.is_none(),
+ "the caller received no successful poll reply"
+ );
+ assert!(
+ matches!(
+ reply.try_send(PartitionReadReply::Ack),
+ Err(crossfire::TrySendError::Disconnected(_))
+ ),
+ "timeout closed the reply channel before completion"
+ );
+ let (stored_offset, _) = partitions
+ .consumer_offset_read(&namespace, consumer)
+ .unwrap();
+ assert_eq!(
+ stored_offset, None,
+ "reading alone has not accepted progress"
+ );
+
+ // Deliver the held result through the completion sender and owner inbox.
+ // Current behavior admits its automatic commit even though delivery fails.
+ PollCompletionSender::new(
+ Some(owner_sender),
+ namespace,
+ reply,
+ owner.metrics().clone(),
+ )
+ .complete(delayed_result);
+ let ShardFrame::Lifecycle(LifecycleFrame::PollCompleted(completion)) = owner_inbox
+ .try_recv()
+ .expect("late completion reached the owner inbox")
+ else {
+ panic!("expected the completed poll");
+ };
+ owner.on_poll_completed(*completion).await;
+ let (stored_offset, _) = partitions
+ .consumer_offset_read(&namespace, consumer)
+ .unwrap();
+ assert_eq!(
+ stored_offset,
+ Some(2),
+ "the nonempty late result advances local progress through the three unseen messages"
+ );
+
+ // A new Next now returns only offset 3. Disabling its automatic commit
+ // keeps the final cursor attributable to the timed out poll alone.
+ let next_poll = owner.partition_read(
+ namespace,
+ PartitionRead::Poll {
+ consumer,
+ args: PollingArgs {
+ strategy: PollingStrategy::next(),
+ count: 4,
+ auto_commit: false,
+ },
+ },
+ );
+ futures::pin_mut!(next_poll);
+ assert!(futures::poll!(next_poll.as_mut()).is_pending());
+ let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead {
+ namespace,
+ read,
+ reply,
+ }) = owner_inbox
+ .try_recv()
+ .expect("subsequent Next reached the owner inbox")
+ else {
+ panic!("expected the subsequent poll request");
+ };
+ owner.on_partition_read(namespace, read, reply).await;
+ let Some(PartitionReadReply::Poll { fragments, .. }) = next_poll.await else {
+ panic!("subsequent Next must return messages successfully");
+ };
+ assert_eq!(
+ message_offsets(&fragments),
+ vec![3],
+ "Next skips offsets 0 through 2, which the caller never received"
Review Comment:
Contract: a poll that cannot deliver its reply must not advance progress. Duplicate beats skip, and that is the invariant this PR restores for purge.
`crossfire::Sender::is_disconnected()` is on the reply sender. Probe it in `complete_poll` before `admit_poll_auto_commit`. A timed-out caller is already gone when the read lands, so the completion rejects and the next poll re-reads. The race window drops from the whole disk read to a few instructions, and losing it costs a duplicate.
Then flip this assertion to re-delivery. Pinned as-is, the next reader takes the skip for the intended contract.
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
Review Comment:
Re-measure before merge, with a concurrent producer. Give writes per poll next to latency. The completion adds a frame and an allocation, no I/O, so if the regression survives it is queueing behind consensus and write frames, which the reserved lane in the other thread also fixes.
##########
core/server_common/src/poll.rs:
##########
@@ -0,0 +1,202 @@
+// 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.
+
+use std::cell::Cell;
+use std::rc::Rc;
+use std::sync::Arc;
+
+use iggy_common::ConsumerKind;
+
+/// Identity of one serviceable message history. It is never serialized.
+///
+/// The allocation identifies the history even if a rebuilt partition reuses
+/// its namespace and offsets. Pending reads keep it alive, preventing address
+/// reuse while they can still complete. `Arc` also keeps completion frames
+/// `Send`, as required by the shard inbox.
+/// `Default` creates a fresh identity. Only clones compare equal.
+#[derive(Clone, Debug, Default)]
+pub struct PollHistoryId(Arc<()>);
Review Comment:
Do it here, not in a follow-up. The point is the invariant, not the allocation. Arc::ptr_eq only works because pending reads keep the old allocation alive so the address cannot be reused. A u64 counter cannot alias at all, and it is Send, so the "keeps frames Send" note does not argue for Arc either.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -830,6 +489,19 @@ where
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
+ None => {
+ // A timeout drops the receiver without canceling the owner. Its
+ // eventual completion may still accept progress, so neither an
+ // empty success nor TransientNotAccepted describes this outcome.
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ "partition read reply unavailable; acceptance unknown"
+ );
+ Err(ReadPolledMessagesError::Rejected(
+ IggyError::ShardCommunicationError,
Review Comment:
Yes. Keep 504 as the status and carry 11001 as the body error id. Add the
explicit mapping, the generic handler would give 400 and blame the client.
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
+ Ok(PollCompletion {
+ fragments,
+ current_offset,
+ replication,
+ }) => {
+ // The owner has admitted progress, but the offset may still be
+ // queued. Release the reply before waiting for replication.
+ // A poll reply does not acknowledge a durable offset commit.
+ let _ = reply.try_send(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ });
+ if let Some(replication) = replication {
+ partitions
+ .replicate_poll_completion(&namespace, replication)
+ .await;
+ }
+ }
+ Err(error) => {
+ if matches!(error, IggyError::TooManyConsumerOffsets) {
+ self.metrics.record_consumer_offset_denied(consumer_kind);
+ }
+ let _ = reply.try_send(PartitionReadReply::Rejected(error));
+ }
+ }
+ }
+}
+
+/// Read an owned snapshot without borrowing the partition or changing progress.
+/// The completion sender returns the result to the owner for acceptance.
+#[allow(clippy::future_not_send)]
+async fn read_poll(
+ namespace: IggyNamespace,
+ plan: PollPlan,
+ completion: completion::PollCompletionSender,
+) {
+ let poll_started = std::time::Instant::now();
+ let result = plan.execute().await;
+ let elapsed = poll_started.elapsed();
+ if elapsed > std::time::Duration::from_secs(1) {
+ tracing::warn!(
+ namespace_raw = namespace.inner(),
+ elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
+ "slow partition poll; gather side may have timed out"
+ );
+ }
+ completion.complete(result);
+}
+
+mod completion {
+ use super::PollCompleted;
+ use crate::coordinator::classify_try_send_err;
+ use crate::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant};
+ use crate::{LifecycleFrame, PartitionReadReply, Sender, ShardFrame, TaggedSender};
+ use iggy_common::IggyError;
+ use partitions::PollReadResult;
+ use server_common::sharding::IggyNamespace;
+
+ /// The read task can return bytes to the owner or reject a lost completion.
+ /// The reply sender stays private so it cannot authorize successful reads.
+ pub(super) struct PollCompletionSender {
+ /// Originating owner's inbox, whose absence causes a rejection.
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ /// May be used here only to reject a result that cannot reach its owner.
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ }
+
+ impl PollCompletionSender {
+ /// Bind the result and reply route to the shard that built the snapshot.
+ pub(super) const fn new(
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ ) -> Self {
+ Self {
+ inbox,
+ namespace,
+ reply,
+ metrics,
+ }
+ }
+
+ /// Return the result to the owner without waiting for inbox capacity.
+ /// A missing, full, or disconnected inbox records a dropped completion
+ /// and attempts a `TransientNotAccepted` reply. Successful enqueueing
+ /// leaves the reply to the owner.
+ pub(super) fn complete(self, result: PollReadResult) {
+ let Some(inbox) = self.inbox else {
+ self.metrics.record_frame_drop(
+ frame_drop_variant::PARTITION_POLL_COMPLETION,
+ frame_drop_reason::UNROUTABLE,
+ );
+ let _ = self.reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ };
+ let frame =
+ ShardFrame::lifecycle(LifecycleFrame::PollCompleted(Box::new(PollCompleted {
+ namespace: self.namespace,
+ result,
+ reply: self.reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Some(std::time::Instant::now()),
+ })));
+ if let Err(error) = inbox.try_send(frame) {
Review Comment:
Do not wait on the inbox. A worker waiting for capacity keeps the batch and the read handle for as long as the pressure lasts, so memory and descriptors grow without bound.
Size the completion lane to the maximum in-flight disk polls instead, which the read budget already caps, or reserve the slot when the poll is dispatched. Then try_send cannot fail for capacity. Leave the unroutable and disconnected arms alone.
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
Review Comment:
Cancel it. A poll that cannot deliver its reply must not advance progress.
Duplicate delivery is fine, a silent skip is not, and that is the same
invariant this PR restores for purge.
Eventual admission with an error is not acceptable. The client cannot act on
"unknown outcome" and the skip is silent from its side.
crossfire::Sender::is_disconnected() is on the reply sender. Probe it in
complete_poll before admit_poll_auto_commit and reject when the caller is
gone. That covers the real case, since a timed-out caller is already gone when
the read lands. The remaining race is a few instructions wide and costs a
duplicate, not a skip.
Then flip the new test to assert re-delivery. Pinned as a skip it reads as the
intended contract.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3993967321
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
Review Comment:
Added the disconnect guard in [7cdf64cfa](https://github.com/apache/iggy/commit/7cdf64cfa6096d8a1cf1fa807cd22c41469015bb). `on_poll_completed` now discards the result when the reply receiver is already disconnected, before `complete_poll` can admit an automatic commit or change progress. The same commit updates the timeout regression to require redelivery. The race between the connection check and admission remains as discussed above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on PR #4119:
URL: https://github.com/apache/iggy/pull/4119#issuecomment-5633401290
Added a short overview of poll completion and concise docs for the new types, fields, and functions. These explain history validation, reservation lifetimes, admission checks, and why poll replies can precede durable offset commits. This update changes documentation only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3993202596
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
Review Comment:
Added the regression in [ad9fc761b](https://github.com/apache/iggy/commit/ad9fc761b03ba1cc64ed25897817a1c6cfabbb92). It holds a nonempty result past the requester timeout with `auto_commit=true`, verifies that the reply channel has closed, then delivers the result through the completion sender and owner inbox. The test records the current behavior: local progress advances through offsets 0–2, and a subsequent `Next` returns only offset 3, skipping the messages the caller never received. This documents the behavior while leaving the timeout and cancellation contract open.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3993707768
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
Review Comment:
Agreed. A completion arriving after the requester has timed out should leave progress unchanged. I’ll add the disconnect check before admission and change the regression to verify that the next poll returns the unseen messages.
There is one remaining ordering detail I’d like to account for. The task awaiting the reply and the partition owner can run on different shards. In [`partition_read`](https://github.com/apache/iggy/blob/a699ff91c6b9126502ea112ac676d96d8d107975/core/shard/src/lib.rs#L1992), the requesting shard sends the read to the shard selected by the namespace, then waits on its own reply receiver with a timeout. One thread per shard therefore still allows the requester’s timeout to run concurrently with the owner’s completion handler.
With the proposed check, this sequence remains possible:
1. The owner checks `is_disconnected()` and sees that the receiver is present.
2. The requesting shard times out and drops the receiver.
3. The owner admits the automatic commit and advances progress through offset 2.
4. The owner attempts to return messages 0–2, but the reply cannot be delivered.
5. The next `Next` poll starts at offset 3.
The check prevents the case where the requester is already gone when completion is processed. In the sequence above, however, the remaining race still skips unseen messages because progress advances before the reply attempt. The absence of an `.await` protects the owner’s state from interleaving on its own thread, but it does not prevent another shard from closing the receiver.
To make cancellation and admission mutually exclusive across shards, they need a shared decision: if cancellation wins, the owner must leave progress untouched; if acceptance wins, the requester must preserve the reply path instead of abandoning it when its timer fires. Checking whether the receiver is connected only observes its state at that moment.
I suggest keeping the disconnect check and the redelivery regression, and adding a deterministic test that places timeout between the check and admission. That would let us verify the additional coordination needed for the stated invariant. This concern is specifically about the internal timeout between the requesting shard and the owner.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #4119:
URL: https://github.com/apache/iggy/pull/4119#issuecomment-5662511820
/dst
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997019493
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
Review Comment:
Following up with longer measurements of the shutdown receive reuse patch. The patch is now included in [3d9116daa](https://github.com/apache/iggy/commit/3d9116daab434f79fbe65b2757e1df6e4a349d1a). It creates and pins `stop.recv().fuse()` once before the owner loop. Shutdown stays first in the biased select, and disk completions still return to the owner for validation.
This addresses an existing cost that the PR exercises more often. In the earlier profiles of master `dc2b38209` and PR `a699ff91c`, sampled CPU time in the shutdown receive function rose from about 4.1 to 7.8 µs per poll without concurrent writes, nearly doubling. With writes it rose from 6.83 to 10.35 µs per poll. The extra owner turn can recreate the receive and repeat Crossfire's backoff for an empty channel. These are sampled CPU estimates, not an additive explanation of the entire elapsed time regression.
A fresh comparison of `7cdf64cfa` against that revision plus reuse used 12 pairs per workload. These longer results supersede the preliminary gains above:
| Load | Elapsed µs/poll, before → after | Written messages/poll, before → after | Time reduction, 95% interval |
| --- | ---: | ---: | ---: |
| No producer | 270.009 → 262.870 | 0 → 0 | 2.64% [1.55%, 3.57%] |
| Producer targeting 50 MB/s | 310.369 → 298.296 | 60.97 → 58.73 | 3.89% [2.59%, 5.33%] |
Fresh resident and consumer group checks, with 12 pairs each, also showed smaller mean elapsed times. The patch passed 244 shard and simulator tests, including shutdown wakeup, draining queued completions and sender disconnection. Five existing simulator tests remain ignored.
I then ran a fresh comparison of `7cdf64cfa` plus reuse against preserved master `dc2b38209`, with 64 pairs per workload:
| Load | Master → PR + reuse, µs/poll | Written messages/poll, master → PR + reuse | Slowdown, 95% interval |
| --- | ---: | ---: | ---: |
| No producer | 234.201 → 234.771 | 0 → 0 | +0.24% [−0.08%, +0.61%] |
| Producer targeting 50 MB/s | 258.465 → 260.416 | 50.90 → 51.34 | +0.75% [+0.34%, +1.20%] |
All 256 runs completed: 8,448,000 measured polls, with 33,000 polls after 100 warmup polls per run. Each poll returns one 256 byte message, using explicit offsets with automatic commits disabled. Version order was randomized and balanced within each workload. All runs passed the raw workload audit, and an independent verifier reproduced the primary estimates and intervals. The concurrent producer sustained 50.09–50.60 MB/s in batches of 100 messages.
Elapsed times and written messages per poll are means across runs. The mean time intervals use 20,000 resamples of whole pairs. The two comparisons ran at different times, so absolute values should be compared within each table. Writes are logical producer operations, not disk syscalls.
The same fresh comparison measured p99 poll latency:
| Load | p99 change | Adjusted interval |
| --- | ---: | ---: |
| No producer | +1.57% | −0.79% to +5.20% |
| Producer targeting 50 MB/s | −1.33% | −4.13% to +0.66% |
These are geometric means of paired ratios of p99 values from individual runs. Positive values mean higher latency. The intervals use 100,000 resamples of whole pairs and retain the preregistered 98.75% bounds per workload, accounting for both workloads and two planned checkpoints. With concurrent writes, the upper bound is below the 3% slowdown threshold. Without writes, that threshold remains unresolved. All pairs remain included.
In these ARM64 Linux VM workloads, the estimated remaining mean overhead is below 1%, and both 95% intervals end below 1.3%. The comparison with reuse on both master and PR mentioned above remains outstanding, so the benefit of applying reuse to plain master and the exact fraction of the regression it explains are still unmeasured.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4119:
URL: https://github.com/apache/iggy/pull/4119#issuecomment-5616715708
@diegomrsantos have you joined our discord? if yes, what's your username?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4119:
URL: https://github.com/apache/iggy/pull/4119#issuecomment-5616676094
#4092 will be merged first :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997392487
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,466 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the reply connection, history, and recovery state, admits
+//! any automatic commit, and updates progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+#[cfg(test)]
+mod completion_tests;
+#[cfg(test)]
+mod test_support;
+#[cfg(test)]
+mod timeout_tests;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
Review Comment:
optional: each disk poll clones and drops 30 `Arc`-backed metric handles, even though completion metrics are only used when enqueueing fails. could we share the needed metrics once during shard setup? the latency impact is unmeasured, so this isn't a blocker. we'd need to preserve the `const` metrics accessor and `Send` bootstrap inputs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3993873223
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
Review Comment:
I looked into existing concurrency and messaging contracts to make the shared decision mentioned above more concrete.
If we want **confirmed cancellation to guarantee that the poll cannot advance progress**, the requester and owner need to compete for one explicit state transition:
```text
Pending → Cancelled
Pending → Completing → Completed(result)
```
An atomic compare-and-exchange could decide which side wins:
- **Cancellation wins:** the owner discards the result without updating consumer progress, group `last_polled`, or admitting an automatic commit.
- **Completion wins:** cancellation can no longer succeed. The owner finishes validation and admission, applies the permitted effects, and publishes the outcome. That outcome can still be a rejection if validation or admission fails.
The partition state would remain exclusively owned by its shard. Only the request’s cancellation decision needs to be shared. This follows an established executor pattern: Python’s [`Future.set_running_or_notify_cancel()`](https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Future.set_running_or_notify_cancel) arbitrates between cancellation and execution, with cancellation no longer succeeding once execution has won.
The requester’s behavior is essential to that contract. If its timer fires after the owner has claimed completion, it must preserve the receiver and wait for the outcome to maintain the internal handoff guarantee. That may extend the wait beyond the original deadline. Returning an error with an unknown outcome instead would describe the uncertainty, but would not recover the unseen messages or satisfy the no-skip requirement.
I also found that [Crossfire explicitly documents](https://docs.rs/crossfire/latest/crossfire/struct.MTx.html#method.try_send) that a buffered send can succeed while the receiver concurrently disappears, leaving nobody to receive the message. Therefore, moving `try_send()` before the effects would not establish delivery either. It could additionally expose success before admission fails. Rolling back after a failed send would require undoing all associated progress and consensus admission, rather than simply restoring one offset.
This distinguishes two atomicity questions. We can make cancellation and permission to apply effects mutually exclusive within the process. Actual receipt by the external client requires a delivery and recovery protocol: a connection can fail after the owner completes but before the client observes the reply. [gRPC documents that same ambiguity](https://grpc.io/docs/what-is-grpc/core-concepts/#rpc-termination). Our poll reply also currently acknowledges admission rather than a durable offset commit, so admission, durability, and receipt are separate boundaries.
For reliable consumption across those failures, established messaging systems retain eligibility for redelivery until the consumer acknowledges the required work. When processing must be protected, acknowledgement follows processing; failure before acknowledgement can then produce duplicates. [RabbitMQ documents this tradeoff](https://www.rabbitmq.com/docs/confirms#consumer-acknowledgement-modes-and-data-safety-considerations), including the weaker safety of automatic acknowledgement.
Kafka offers a useful comparison, but its automatic commits operate through the consumer client. Its current position and committed recovery position are separate. Obtaining at-least-once behavior with automatic commits requires processing all records returned by `poll()` before subsequent polls or closing. That differs from the server admitting the current batch’s automatic commit before replying. [Kafka consumer documentation](https://kafka.apache.org/41/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html)
Another possible recovery contract is replayable polling: retrying the same request ID recovers the same batch without advancing again. Here, the effects and the recorded request outcome need to be committed together, following the atomicity requirement described in [AWS’s idempotency guidance](https://aws.amazon.com/builders-library/making-retries-safe-with-idempotent-APIs/). For Iggy, this would also require defining batch retention, request expiry, and recovery across owner failure.
Any broader delivery contract must include consumer-group `last_polled`, which advances even without automatic commits. Disabling `auto_commit` alone therefore would not cover every path. The shared cancellation decision addresses the internal timeout; acknowledgements or replayable requests provide recovery when the external reply is lost.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3992226398
##########
core/simulator/src/lib.rs:
##########
@@ -1407,12 +1407,16 @@ impl Simulator {
"partition not found for namespace {namespace:?} on replica {replica_idx}"
)));
};
- // Partitions are driven directly, so a poll's auto-commit is never
- // replicated (the serving shard's job in the real server). Offset discarded.
- let (fragments, _commit_offset, auto_commit) = futures::executor::block_on(plan.execute())?;
- if let Some(applied) = auto_commit {
- applied.admit(|_| Ok::<(), IggyError>(()))?;
+ let partitions = shard.plane.partitions();
+ let completion =
+ partitions.complete_poll(&namespace, futures::executor::block_on(plan.execute()))?;
+ if let Some(replication) = completion.replication {
+ // SimOutbox sends enqueue immediately, so replication cannot suspend.
+ futures::executor::block_on(
+ partitions.replicate_poll_completion(&namespace, replication),
Review Comment:
Thanks for catching this. The comment is too strong. Replica sends normally finish immediately, but the new test mechanism can pause them, so the helper cannot safely assume that replication will never suspend.
Before this PR, the helper used `block_on` for the resident read but did not replicate automatic commits. This PR adds the replication wait alongside the ability to pause a replica send. If a test calls the helper while that pause is active, it can block before reaching the instruction that releases the send.
The existing test avoids this by using the asynchronous path, but the helper should handle that situation safely too. I’ll adjust it to use the simulator’s normal asynchronous execution path so the calling test can continue and release paused work, and add regression coverage for this combination.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz merged PR #4119:
URL: https://github.com/apache/iggy/pull/4119
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3994070692
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
+ Ok(PollCompletion {
+ fragments,
+ current_offset,
+ replication,
+ }) => {
+ // The owner has admitted progress, but the offset may still be
+ // queued. Release the reply before waiting for replication.
+ // A poll reply does not acknowledge a durable offset commit.
+ let _ = reply.try_send(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ });
+ if let Some(replication) = replication {
+ partitions
+ .replicate_poll_completion(&namespace, replication)
+ .await;
+ }
+ }
+ Err(error) => {
+ if matches!(error, IggyError::TooManyConsumerOffsets) {
+ self.metrics.record_consumer_offset_denied(consumer_kind);
+ }
+ let _ = reply.try_send(PartitionReadReply::Rejected(error));
+ }
+ }
+ }
+}
+
+/// Read an owned snapshot without borrowing the partition or changing progress.
+/// The completion sender returns the result to the owner for acceptance.
+#[allow(clippy::future_not_send)]
+async fn read_poll(
+ namespace: IggyNamespace,
+ plan: PollPlan,
+ completion: completion::PollCompletionSender,
+) {
+ let poll_started = std::time::Instant::now();
+ let result = plan.execute().await;
+ let elapsed = poll_started.elapsed();
+ if elapsed > std::time::Duration::from_secs(1) {
+ tracing::warn!(
+ namespace_raw = namespace.inner(),
+ elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
+ "slow partition poll; gather side may have timed out"
+ );
+ }
+ completion.complete(result);
+}
+
+mod completion {
+ use super::PollCompleted;
+ use crate::coordinator::classify_try_send_err;
+ use crate::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant};
+ use crate::{LifecycleFrame, PartitionReadReply, Sender, ShardFrame, TaggedSender};
+ use iggy_common::IggyError;
+ use partitions::PollReadResult;
+ use server_common::sharding::IggyNamespace;
+
+ /// The read task can return bytes to the owner or reject a lost completion.
+ /// The reply sender stays private so it cannot authorize successful reads.
+ pub(super) struct PollCompletionSender {
+ /// Originating owner's inbox, whose absence causes a rejection.
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ /// May be used here only to reject a result that cannot reach its owner.
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ }
+
+ impl PollCompletionSender {
+ /// Bind the result and reply route to the shard that built the snapshot.
+ pub(super) const fn new(
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ ) -> Self {
+ Self {
+ inbox,
+ namespace,
+ reply,
+ metrics,
+ }
+ }
+
+ /// Return the result to the owner without waiting for inbox capacity.
+ /// A missing, full, or disconnected inbox records a dropped completion
+ /// and attempts a `TransientNotAccepted` reply. Successful enqueueing
+ /// leaves the reply to the owner.
+ pub(super) fn complete(self, result: PollReadResult) {
+ let Some(inbox) = self.inbox else {
+ self.metrics.record_frame_drop(
+ frame_drop_variant::PARTITION_POLL_COMPLETION,
+ frame_drop_reason::UNROUTABLE,
+ );
+ let _ = self.reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ };
+ let frame =
+ ShardFrame::lifecycle(LifecycleFrame::PollCompleted(Box::new(PollCompleted {
+ namespace: self.namespace,
+ result,
+ reply: self.reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Some(std::time::Instant::now()),
+ })));
+ if let Err(error) = inbox.try_send(frame) {
Review Comment:
I tested the option that reserves capacity when the disk poll is dispatched.
Each pending read owns one slot until its result is dequeued. Completed reads
use a dedicated bounded lane, and the owner processes one completion after
ordinary work so both lanes keep making progress. A read is rejected before
disk I/O if no slot is available. Finished reads do not wait for inbox space.
The missing and disconnected route checks remain, as does owner history
validation.
The new regression fills the ordinary inbox, then returns two nonempty reads.
It verifies delivery and observes an ordinary request between their acceptances.
It fails if completion service is removed or if the entire completion queue
is drained at once. Additional tests cover reservation release and shutdown, plus the behavior
after caller cancellation at a699ff91c. In total, 103 shard tests and 143
simulator tests passed on macOS and Linux.
This experiment used a699ff91c, before 7cdf64cfa added the disconnected receiver
check. It preserves the earlier timeout characterization. Adopting the lane
would require rebasing it and updating that regression for the new policy.
The performance comparison did not show an improvement, and the point estimates
were worse with a concurrent producer. The implementation remains an experimental patch. The [performance reply](https://github.com/apache/iggy/pull/4119#discussion_r3994067578) and the [evidence bundle](https://github.com/diegomrsantos/iggy/releases/tag/pr-4119-benchmark-evidence-2026-09-12) contain the measurements and
[exact patch](https://github.com/diegomrsantos/iggy/releases/download/pr-4119-benchmark-evidence-2026-09-12/reserved-lane-candidate.patch).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3991208625
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
Review Comment:
Adds one main-lane frame, one pump turn and one heap allocation per disk poll. The completion queues behind arbitrary consensus and write frames. This is the measured `disk-offset` regression.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -15148,6 +15804,252 @@ mod retention_tests {
}
}
+#[cfg(test)]
+mod purge_poll_tests {
+ //! A disk poll with automatic commit starts reading messages at offsets 0-2.
+ //! Before it completes, purge removes those messages and clears consumer
+ //! progress. Fresh messages are then appended starting at offset 0.
+ //! Completing the old poll must not advance progress over the fresh messages.
+
+ use super::tests::{journal_send_batch, repair_config, test_partition};
+ use super::*;
+ use crate::PollFragments;
+ use iggy_common::PollingStrategy;
+ use server_common::send_messages::decode_batch_slice;
+
+ #[compio::test]
+ async fn given_pending_disk_poll_when_purged_should_preserve_fresh_consumer_progress() {
+ Box::pin(assert_delayed_poll_preserves_fresh_progress(true, 2)).await;
+ }
+
+ #[compio::test]
+ async fn given_pending_disk_poll_when_fresh_history_covers_old_offset_should_preserve_progress()
+ {
+ // The delayed poll targets old offsets 0-2. After purge, five fresh
+ // messages occupy offsets 0-4. Recording progress 2 from the old poll
+ // would make `Next` skip fresh offsets 0-2 even though 2 is in range.
+ Box::pin(assert_delayed_poll_preserves_fresh_progress(true, 5)).await;
+ }
+
+ #[compio::test]
+ async fn given_disk_poll_without_auto_commit_when_purged_should_preserve_fresh_progress() {
+ for fresh_message_count in [2, 5] {
+ Box::pin(assert_delayed_poll_preserves_fresh_progress(
+ false,
+ fresh_message_count,
+ ))
+ .await;
+ }
+ }
+
+ async fn assert_delayed_poll_preserves_fresh_progress(
+ auto_commit: bool,
+ fresh_message_count: u32,
+ ) {
+ let config = repair_config();
+ let (_directory, mut partition) = Box::pin(disk_poll_partition(&config)).await;
+ let consumer = PollingConsumer::Consumer(7, 0);
Review Comment:
No-auto-commit purge test uses `PollingConsumer::Consumer(7, 0)`. On base an individual consumer with `auto_commit=false` updates neither `consumer_offsets` nor `last_polled_offsets`, so every assertion passes without the fix.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -830,6 +489,19 @@ where
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
+ None => {
Review Comment:
Maps every `None` from `partition_read` to `ShardCommunicationError` (11001). Three of those cases prove the read never started: `core/shard/src/lib.rs:1998` unroutable namespace, `core/shard/src/lib.rs:2012` no sender, `core/shard/src/lib.rs:2013` full inbox. Base answered an empty poll here.
##########
core/simulator/src/lib.rs:
##########
@@ -1701,6 +1706,87 @@ mod tests {
panic!("request {request_id} did not receive a reply");
}
+ #[test]
+ fn given_admitted_resident_poll_when_replica_send_stalls_should_reply_before_replication() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+ let client_id = 1u128;
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let mut sim = Simulator::new(
+ 3,
+ std::iter::once(client_id),
+ packet::PacketSimulatorOptions {
+ node_count: 3,
+ client_count: 1,
+ seed: 0xC011_EC71,
+ ..packet::PacketSimulatorOptions::default()
+ },
+ );
+ sim.init_partition(namespace);
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+ let produced = submit_and_wait_for_reply(
+ &mut sim,
+ client_id,
+ 0,
+ client.send_messages(
+ namespace,
+ &[Bytes::from_static(b"reply-before-replication")],
+ ),
+ );
+ assert_eq!(produced.header().status, 0);
+ sim.run_pumps();
+
+ let owner = Rc::clone(sim.replicas[0].partition_shard(namespace));
+ let consumer = PollingConsumer::Consumer(7, 0);
+ let args = PollingArgs::new(iggy_common::PollingStrategy::next(), 1, true);
+ let plan = owner
+ .plane
+ .partitions()
+ .build_poll_snapshot(&namespace, consumer, &args)
+ .expect("poll snapshot");
+ assert!(
+ !plan.needs_off_pump_io(),
Review Comment:
The one new simulator test asserts `!plan.needs_off_pump_io()`, and no disk tier exists. The new inbox hop is never modeled.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -830,6 +489,19 @@ where
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
+ None => {
+ // A timeout drops the receiver without canceling the owner. Its
+ // eventual completion may still accept progress, so neither an
+ // empty success nor TransientNotAccepted describes this outcome.
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ "partition read reply unavailable; acceptance unknown"
+ );
+ Err(ReadPolledMessagesError::Rejected(
+ IggyError::ShardCommunicationError,
Review Comment:
Binary transports answer 11001 while HTTP answers 504 at `core/server/src/http/handlers.rs:1322`. The parity rule is stated at `core/server/src/http/error.rs:474`.
##########
core/server_common/src/poll.rs:
##########
@@ -0,0 +1,202 @@
+// 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.
+
+use std::cell::Cell;
+use std::rc::Rc;
+use std::sync::Arc;
+
+use iggy_common::ConsumerKind;
+
+/// Identity of one serviceable message history. It is never serialized.
+///
+/// The allocation identifies the history even if a rebuilt partition reuses
+/// its namespace and offsets. Pending reads keep it alive, preventing address
+/// reuse while they can still complete. `Arc` also keeps completion frames
+/// `Send`, as required by the shard inbox.
+/// `Default` creates a fresh identity. Only clones compare equal.
+#[derive(Clone, Debug, Default)]
+pub struct PollHistoryId(Arc<()>);
Review Comment:
`PollHistoryId` is an `Arc<()>` compared by address where a `u64` counter serves. Zero lines saved, one allocation removed per poll.
##########
core/simulator/src/lib.rs:
##########
@@ -1407,12 +1407,16 @@ impl Simulator {
"partition not found for namespace {namespace:?} on replica {replica_idx}"
)));
};
- // Partitions are driven directly, so a poll's auto-commit is never
- // replicated (the serving shard's job in the real server). Offset discarded.
- let (fragments, _commit_offset, auto_commit) = futures::executor::block_on(plan.execute())?;
- if let Some(applied) = auto_commit {
- applied.admit(|_| Ok::<(), IggyError>(()))?;
+ let partitions = shard.plane.partitions();
+ let completion =
+ partitions.complete_poll(&namespace, futures::executor::block_on(plan.execute()))?;
+ if let Some(replication) = completion.replication {
+ // SimOutbox sends enqueue immediately, so replication cannot suspend.
+ futures::executor::block_on(
+ partitions.replicate_poll_completion(&namespace, replication),
Review Comment:
Comment states replica sends never suspend, but `core/simulator/src/bus.rs:234` awaits under `cfg(test)`, which deadlocks the `block_on` at `core/simulator/src/lib.rs:1415`.
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
+ Ok(PollCompletion {
+ fragments,
+ current_offset,
+ replication,
+ }) => {
+ // The owner has admitted progress, but the offset may still be
+ // queued. Release the reply before waiting for replication.
+ // A poll reply does not acknowledge a durable offset commit.
+ let _ = reply.try_send(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ });
+ if let Some(replication) = replication {
+ partitions
+ .replicate_poll_completion(&namespace, replication)
+ .await;
+ }
+ }
+ Err(error) => {
+ if matches!(error, IggyError::TooManyConsumerOffsets) {
+ self.metrics.record_consumer_offset_denied(consumer_kind);
+ }
+ let _ = reply.try_send(PartitionReadReply::Rejected(error));
+ }
+ }
+ }
+}
+
+/// Read an owned snapshot without borrowing the partition or changing progress.
+/// The completion sender returns the result to the owner for acceptance.
+#[allow(clippy::future_not_send)]
+async fn read_poll(
+ namespace: IggyNamespace,
+ plan: PollPlan,
+ completion: completion::PollCompletionSender,
+) {
+ let poll_started = std::time::Instant::now();
+ let result = plan.execute().await;
+ let elapsed = poll_started.elapsed();
+ if elapsed > std::time::Duration::from_secs(1) {
+ tracing::warn!(
+ namespace_raw = namespace.inner(),
+ elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
+ "slow partition poll; gather side may have timed out"
+ );
+ }
+ completion.complete(result);
+}
+
+mod completion {
+ use super::PollCompleted;
+ use crate::coordinator::classify_try_send_err;
+ use crate::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant};
+ use crate::{LifecycleFrame, PartitionReadReply, Sender, ShardFrame, TaggedSender};
+ use iggy_common::IggyError;
+ use partitions::PollReadResult;
+ use server_common::sharding::IggyNamespace;
+
+ /// The read task can return bytes to the owner or reject a lost completion.
+ /// The reply sender stays private so it cannot authorize successful reads.
+ pub(super) struct PollCompletionSender {
+ /// Originating owner's inbox, whose absence causes a rejection.
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ /// May be used here only to reject a result that cannot reach its owner.
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ }
+
+ impl PollCompletionSender {
+ /// Bind the result and reply route to the shard that built the snapshot.
+ pub(super) const fn new(
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ ) -> Self {
+ Self {
+ inbox,
+ namespace,
+ reply,
+ metrics,
+ }
+ }
+
+ /// Return the result to the owner without waiting for inbox capacity.
+ /// A missing, full, or disconnected inbox records a dropped completion
+ /// and attempts a `TransientNotAccepted` reply. Successful enqueueing
+ /// leaves the reply to the owner.
+ pub(super) fn complete(self, result: PollReadResult) {
+ let Some(inbox) = self.inbox else {
+ self.metrics.record_frame_drop(
+ frame_drop_variant::PARTITION_POLL_COMPLETION,
+ frame_drop_reason::UNROUTABLE,
+ );
+ let _ = self.reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ };
+ let frame =
+ ShardFrame::lifecycle(LifecycleFrame::PollCompleted(Box::new(PollCompleted {
+ namespace: self.namespace,
+ result,
+ reply: self.reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Some(std::time::Instant::now()),
+ })));
+ if let Err(error) = inbox.try_send(frame) {
Review Comment:
`PollCompletionSender::complete` calls `try_send` on the main lane. A full inbox discards a finished read and answers `TransientNotAccepted`.
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
Review Comment:
Commits progress at `core/partitions/src/iggy_partition.rs:2901`, then tries the reply at `core/shard/src/poll.rs:195` and discards the result. Budget expiry at `core/shard/src/lib.rs:2021` drops the receiver while the client still waits. Needs `auto_commit=true` with `Next`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3991722812
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
+ Ok(PollCompletion {
+ fragments,
+ current_offset,
+ replication,
+ }) => {
+ // The owner has admitted progress, but the offset may still be
+ // queued. Release the reply before waiting for replication.
+ // A poll reply does not acknowledge a durable offset commit.
+ let _ = reply.try_send(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ });
+ if let Some(replication) = replication {
+ partitions
+ .replicate_poll_completion(&namespace, replication)
+ .await;
+ }
+ }
+ Err(error) => {
+ if matches!(error, IggyError::TooManyConsumerOffsets) {
+ self.metrics.record_consumer_offset_denied(consumer_kind);
+ }
+ let _ = reply.try_send(PartitionReadReply::Rejected(error));
+ }
+ }
+ }
+}
+
+/// Read an owned snapshot without borrowing the partition or changing progress.
+/// The completion sender returns the result to the owner for acceptance.
+#[allow(clippy::future_not_send)]
+async fn read_poll(
+ namespace: IggyNamespace,
+ plan: PollPlan,
+ completion: completion::PollCompletionSender,
+) {
+ let poll_started = std::time::Instant::now();
+ let result = plan.execute().await;
+ let elapsed = poll_started.elapsed();
+ if elapsed > std::time::Duration::from_secs(1) {
+ tracing::warn!(
+ namespace_raw = namespace.inner(),
+ elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
+ "slow partition poll; gather side may have timed out"
+ );
+ }
+ completion.complete(result);
+}
+
+mod completion {
+ use super::PollCompleted;
+ use crate::coordinator::classify_try_send_err;
+ use crate::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant};
+ use crate::{LifecycleFrame, PartitionReadReply, Sender, ShardFrame, TaggedSender};
+ use iggy_common::IggyError;
+ use partitions::PollReadResult;
+ use server_common::sharding::IggyNamespace;
+
+ /// The read task can return bytes to the owner or reject a lost completion.
+ /// The reply sender stays private so it cannot authorize successful reads.
+ pub(super) struct PollCompletionSender {
+ /// Originating owner's inbox, whose absence causes a rejection.
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ /// May be used here only to reject a result that cannot reach its owner.
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ }
+
+ impl PollCompletionSender {
+ /// Bind the result and reply route to the shard that built the snapshot.
+ pub(super) const fn new(
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ ) -> Self {
+ Self {
+ inbox,
+ namespace,
+ reply,
+ metrics,
+ }
+ }
+
+ /// Return the result to the owner without waiting for inbox capacity.
+ /// A missing, full, or disconnected inbox records a dropped completion
+ /// and attempts a `TransientNotAccepted` reply. Successful enqueueing
+ /// leaves the reply to the owner.
+ pub(super) fn complete(self, result: PollReadResult) {
+ let Some(inbox) = self.inbox else {
+ self.metrics.record_frame_drop(
+ frame_drop_variant::PARTITION_POLL_COMPLETION,
+ frame_drop_reason::UNROUTABLE,
+ );
+ let _ = self.reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ };
+ let frame =
+ ShardFrame::lifecycle(LifecycleFrame::PollCompleted(Box::new(PollCompleted {
+ namespace: self.namespace,
+ result,
+ reply: self.reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Some(std::time::Instant::now()),
+ })));
+ if let Err(error) = inbox.try_send(frame) {
Review Comment:
Agreed. Once the disk read finishes, the `complete` method on `PollCompletionSender` tries to enqueue the result in the owner’s main inbox. If that inbox is full, it drops the completed result and attempts a `TransientNotAccepted` reply. This is deliberate behavior, and the existing test explicitly verifies it.
The rejection preserves progress safety. The owner has not accepted this completion, so this poll has neither advanced the consumer cursor nor admitted an automatic commit. `TransientNotAccepted` is therefore appropriate even though the disk read already happened. The concern is wasted work. Under sustained pressure, retries can repeatedly read and discard the same messages, adding load.
A related failure path existed before this PR. Polls that needed to submit an automatic commit could already reject after reading their messages if the owner’s inbox refused that submission. However, polls without automatic commits, or whose commits needed no submission, could reply directly. This PR makes every disk poll depend on completion inbox capacity. Resident polls still complete inline.
We need to retain owner validation before allowing progress or a successful reply, since purge or partition replacement can happen while disk I/O is pending. I see two possible improvements.
- **Wait asynchronously for inbox capacity from the detached worker.** This retains the completed batch instead of immediately discarding it. The current sender’s ordinary `send` blocks, so using it could deadlock the shard thread that must drain the inbox. We would need an asynchronous sender, with waiting outside the pump. This also requires a bound on outstanding reads or retained bytes. A bounded inbox alone does not bound waiting workers. Completions would still compete with other main inbox traffic.
- **Reserve dedicated completion capacity before starting disk I/O.** Each admitted read would then have space to return its result. If capacity is exhausted, we could defer or reject before doing the read. This addresses wasted I/O and competition with other traffic more directly, but introduces additional admission and scheduling machinery, including fair service for completions.
Either approach needs explicit timeout and shutdown handling, because the requester’s timeout currently does not cancel detached work. History validation must still happen when the owner accepts the result.
The current policy favors immediate rejection and releasing memory. Would you prefer retaining completed reads with bounded asynchronous waiting, reserving completion capacity before reading, or another approach?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997391587
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2840,19 +2866,282 @@ where
}
}
- /// Reject completion from a removed partition or a view whose retained
- /// prepares have not yet been accounted for by the pump.
- #[must_use]
- pub fn auto_commit_admission_ready(&self, applied: &crate::AutoCommitApplied) -> bool {
- applied.belongs_to(&self.durable_consumer_offsets)
- && self.observed_view == self.consensus.view()
+ /// Accept a read only while its message history still belongs to this owner.
+ /// Validation, admission, and progress updates must stay in one synchronous
+ /// owner turn so purge or recovery cannot interleave between them.
+ /// Nonempty group reads update `last_polled` even without automatic commits.
+ /// Rejection leaves this read's progress unapplied.
+ pub(crate) fn complete_poll(
+ &mut self,
+ result: PollReadResult,
+ ) -> Result<PollCompletion, IggyError> {
+ self.resynchronize_consumer_offset_reservations();
Review Comment:
Fixed in [cbdd9c43c](https://github.com/apache/iggy/commit/cbdd9c43c315523120d75d9f8f0dead297cb72f9). `complete_poll` now checks history and partition availability before reconciling reservations. The regression fails with the previous ordering and verifies that a fresh completion still reconciles before automatic commit admission. All 985 tests across the affected crates passed on macOS, with eight existing ignored tests, and strict Clippy passed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997644345
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
+ Ok(PollCompletion {
+ fragments,
+ current_offset,
+ replication,
+ }) => {
+ // The owner has admitted progress, but the offset may still be
+ // queued. Release the reply before waiting for replication.
+ // A poll reply does not acknowledge a durable offset commit.
+ let _ = reply.try_send(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ });
+ if let Some(replication) = replication {
+ partitions
+ .replicate_poll_completion(&namespace, replication)
+ .await;
+ }
+ }
+ Err(error) => {
+ if matches!(error, IggyError::TooManyConsumerOffsets) {
+ self.metrics.record_consumer_offset_denied(consumer_kind);
+ }
+ let _ = reply.try_send(PartitionReadReply::Rejected(error));
+ }
+ }
+ }
+}
+
+/// Read an owned snapshot without borrowing the partition or changing progress.
+/// The completion sender returns the result to the owner for acceptance.
+#[allow(clippy::future_not_send)]
+async fn read_poll(
+ namespace: IggyNamespace,
+ plan: PollPlan,
+ completion: completion::PollCompletionSender,
+) {
+ let poll_started = std::time::Instant::now();
+ let result = plan.execute().await;
+ let elapsed = poll_started.elapsed();
+ if elapsed > std::time::Duration::from_secs(1) {
+ tracing::warn!(
+ namespace_raw = namespace.inner(),
+ elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
+ "slow partition poll; gather side may have timed out"
+ );
+ }
+ completion.complete(result);
+}
+
+mod completion {
+ use super::PollCompleted;
+ use crate::coordinator::classify_try_send_err;
+ use crate::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant};
+ use crate::{LifecycleFrame, PartitionReadReply, Sender, ShardFrame, TaggedSender};
+ use iggy_common::IggyError;
+ use partitions::PollReadResult;
+ use server_common::sharding::IggyNamespace;
+
+ /// The read task can return bytes to the owner or reject a lost completion.
+ /// The reply sender stays private so it cannot authorize successful reads.
+ pub(super) struct PollCompletionSender {
+ /// Originating owner's inbox, whose absence causes a rejection.
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ /// May be used here only to reject a result that cannot reach its owner.
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ }
+
+ impl PollCompletionSender {
+ /// Bind the result and reply route to the shard that built the snapshot.
+ pub(super) const fn new(
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ ) -> Self {
+ Self {
+ inbox,
+ namespace,
+ reply,
+ metrics,
+ }
+ }
+
+ /// Return the result to the owner without waiting for inbox capacity.
+ /// A missing, full, or disconnected inbox records a dropped completion
+ /// and attempts a `TransientNotAccepted` reply. Successful enqueueing
+ /// leaves the reply to the owner.
+ pub(super) fn complete(self, result: PollReadResult) {
+ let Some(inbox) = self.inbox else {
+ self.metrics.record_frame_drop(
+ frame_drop_variant::PARTITION_POLL_COMPLETION,
+ frame_drop_reason::UNROUTABLE,
+ );
+ let _ = self.reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ };
+ let frame =
+ ShardFrame::lifecycle(LifecycleFrame::PollCompleted(Box::new(PollCompleted {
+ namespace: self.namespace,
+ result,
+ reply: self.reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Some(std::time::Instant::now()),
+ })));
+ if let Err(error) = inbox.try_send(frame) {
Review Comment:
Implemented in [49e2e15d4](https://github.com/apache/iggy/commit/49e2e15d4efce576f0ff0bd867dc546cf75e7e64). Disk polls reserve dedicated completion capacity before I/O and retain it until dequeue or discard, even after requester timeout. Exhaustion rejects before reading; completed reads never wait for inbox space.
`sharding.poll_completion_capacity` is independent of `inbox_capacity`. The pump services completions alongside ordinary work, and missing or disconnected routes still reject.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997781175
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
Review Comment:
@hubcio Added in [3b9619ea2](https://github.com/apache/iggy/commit/3b9619ea2346b1909c7d24a55ad345b99f429a64). The public poll API now explains lost replies and recovery from checkpoints for each partition, with links from `Next`, HTTP polling, and the consumer docs. It covers staying with explicit offsets until caught up, retention limits, and why a successful reply does not acknowledge a durable offset commit.
##########
core/partitions/src/iggy_partition.rs:
##########
Review Comment:
Done in [3b9619ea2](https://github.com/apache/iggy/commit/3b9619ea2346b1909c7d24a55ad345b99f429a64). The partition now owns the durable offset table and both capacity trackers directly, removing three allocations per partition. Reservation tokens and counters retain their existing sharing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3992191507
##########
core/simulator/src/lib.rs:
##########
@@ -1701,6 +1706,87 @@ mod tests {
panic!("request {request_id} did not receive a reply");
}
+ #[test]
+ fn given_admitted_resident_poll_when_replica_send_stalls_should_reply_before_replication() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+ let client_id = 1u128;
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let mut sim = Simulator::new(
+ 3,
+ std::iter::once(client_id),
+ packet::PacketSimulatorOptions {
+ node_count: 3,
+ client_count: 1,
+ seed: 0xC011_EC71,
+ ..packet::PacketSimulatorOptions::default()
+ },
+ );
+ sim.init_partition(namespace);
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+ let produced = submit_and_wait_for_reply(
+ &mut sim,
+ client_id,
+ 0,
+ client.send_messages(
+ namespace,
+ &[Bytes::from_static(b"reply-before-replication")],
+ ),
+ );
+ assert_eq!(produced.header().status, 0);
+ sim.run_pumps();
+
+ let owner = Rc::clone(sim.replicas[0].partition_shard(namespace));
+ let consumer = PollingConsumer::Consumer(7, 0);
+ let args = PollingArgs::new(iggy_common::PollingStrategy::next(), 1, true);
+ let plan = owner
+ .plane
+ .partitions()
+ .build_poll_snapshot(&namespace, consumer, &args)
+ .expect("poll snapshot");
+ assert!(
+ !plan.needs_off_pump_io(),
Review Comment:
Thanks, you’re right. That test deliberately uses a resident read to verify that the client receives its reply while replication is stalled. It does not exercise the extra inbox step used by disk completions.
The existing disk purge test calls the completion handler directly, while the channel tests check enqueueing and rejection without processing the completion through the owner. We are missing coverage of those pieces working together.
I’ll add a test that holds back a nonempty result, purges or replaces the partition, and then delivers the old result through the actual completion sender and inbox. The owner should reject it without advancing progress. A fresh result should also succeed through the same route.
We can control when the result is released without building a complete disk model into the simulator, while keeping the existing disk test for actual I/O coverage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3993282720
##########
core/simulator/src/lib.rs:
##########
@@ -1407,12 +1407,16 @@ impl Simulator {
"partition not found for namespace {namespace:?} on replica {replica_idx}"
)));
};
- // Partitions are driven directly, so a poll's auto-commit is never
- // replicated (the serving shard's job in the real server). Offset discarded.
- let (fragments, _commit_offset, auto_commit) = futures::executor::block_on(plan.execute())?;
- if let Some(applied) = auto_commit {
- applied.admit(|_| Ok::<(), IggyError>(()))?;
+ let partitions = shard.plane.partitions();
+ let completion =
+ partitions.complete_poll(&namespace, futures::executor::block_on(plan.execute()))?;
+ if let Some(replication) = completion.replication {
+ // SimOutbox sends enqueue immediately, so replication cannot suspend.
+ futures::executor::block_on(
+ partitions.replicate_poll_completion(&namespace, replication),
Review Comment:
Fixed in [13afb497a](https://github.com/apache/iggy/commit/13afb497ab56f3f49e7810d893cad610f54e6d1e). `Simulator::poll_messages` now returns an owned future and routes the poll through the partition owner, removing both `block_on` calls. The regression uses the helper with a replica send paused, checks that the nonempty reply is available before release, and then verifies that the automatic commit reaches every replica after replication resumes. The simulator demo also uses the asynchronous helper. All 143 simulator tests passed on both macOS and Linux, with five existing ignored tests on each platform.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on PR #4119:
URL: https://github.com/apache/iggy/pull/4119#issuecomment-5619237457
> #4092 will be merged first :)
Thanks for letting me know. I'll rebase this one onto #4092
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3993970807
##########
core/shard/src/poll/timeout_tests.rs:
##########
@@ -0,0 +1,277 @@
+// 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.
+
+use std::cell::RefCell;
+use std::rc::Rc;
+use std::time::Duration;
+
+use consensus::PartitionsHandle;
+use futures::channel::oneshot;
+use iggy_common::PollingStrategy;
+use message_bus::{
+ BusMessage, ClientForwardFn, ConnectionLostFn, JoinHandle, MessageBus, ReplicaForwardFn,
+ SendError,
+};
+use metadata::IggyMetadata;
+use partitions::{IggyPartitions, PollFragments, PollingArgs, PollingConsumer};
+use server_common::MESSAGE_ALIGN;
+use server_common::iobuf::Frozen;
+use server_common::send_messages::decode_batch_slice;
+use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+
+use super::completion::PollCompletionSender;
+use super::test_support::{PollTestMetadata, partition_with_messages};
+use crate::shards_table::{PapayaShardsTable, ShardsTable};
+use crate::{
+ IggyShard, LifecycleFrame, PARTITION_READ_TIMEOUT, PartitionConsensusConfig, PartitionRead,
+ PartitionReadReply, ReplicaTopology, ShardFrame, ShardIdentity, shard_channel,
+};
+
+/// Characterize the current timeout behavior: losing the caller does not cancel
+/// owner acceptance. This records the unseen messages that a later Next skips;
+/// it does not prescribe the eventual cancellation contract.
+#[compio::test]
+#[allow(clippy::too_many_lines)]
+async fn given_auto_commit_poll_when_completion_arrives_after_timeout_should_skip_unseen_messages_on_next()
+ {
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let consumer = PollingConsumer::Consumer(7, 0);
+ let (expire_timeout, timeout_elapsed) = oneshot::channel();
+ let bus = PollTimeoutBus {
+ next_timeout: Rc::new(RefCell::new(Some(timeout_elapsed))),
+ };
+ let mut owner = owner_with_messages(&bus, namespace).await;
+ let (owner_sender, owner_inbox, _owner_reply_lane) = shard_channel(0, 2, 1);
+ owner.attach_senders(vec![owner_sender.clone()]);
+ let partitions = owner.plane.partitions();
+ let (stored_offset, partition_offset) = partitions
+ .consumer_offset_read(&namespace, consumer)
+ .expect("fixture partition exists");
+ assert_eq!(stored_offset, None);
+ assert_eq!(
+ partition_offset, 3,
+ "four messages are available at offsets 0 through 3"
+ );
+
+ // Start a real routed request, but let the test control when its completed
+ // read reaches the owner. The fourth message will identify the next batch.
+ let first_poll = owner.partition_read(
+ namespace,
+ PartitionRead::Poll {
+ consumer,
+ args: PollingArgs {
+ strategy: PollingStrategy::next(),
+ count: 3,
+ auto_commit: true,
+ },
+ },
+ );
+ futures::pin_mut!(first_poll);
+ assert!(futures::poll!(first_poll.as_mut()).is_pending());
+ let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead {
+ namespace: requested_namespace,
+ read:
+ PartitionRead::Poll {
+ consumer: requested_consumer,
+ args,
+ },
+ reply,
+ }) = owner_inbox
+ .try_recv()
+ .expect("poll reached the owner inbox")
+ else {
+ panic!("expected the routed poll request");
+ };
+ let read_plan = partitions
+ .build_poll_snapshot(&requested_namespace, requested_consumer, &args)
+ .expect("poll has a read snapshot");
+ // Resident bytes keep disk scheduling out of this test. Holding the result
+ // here models the delay before acceptance shared with detached disk reads.
+ assert!(!read_plan.needs_off_pump_io());
+ let delayed_result = read_plan.execute_resident();
+
+ // The reply sender is still alive, so None must come from the timeout.
+ expire_timeout
+ .send(())
+ .expect("requester is waiting on its timer");
+ assert!(
+ first_poll.await.is_none(),
+ "the caller received no successful poll reply"
+ );
+ assert!(
+ matches!(
+ reply.try_send(PartitionReadReply::Ack),
+ Err(crossfire::TrySendError::Disconnected(_))
+ ),
+ "timeout closed the reply channel before completion"
+ );
+ let (stored_offset, _) = partitions
+ .consumer_offset_read(&namespace, consumer)
+ .unwrap();
+ assert_eq!(
+ stored_offset, None,
+ "reading alone has not accepted progress"
+ );
+
+ // Deliver the held result through the completion sender and owner inbox.
+ // Current behavior admits its automatic commit even though delivery fails.
+ PollCompletionSender::new(
+ Some(owner_sender),
+ namespace,
+ reply,
+ owner.metrics().clone(),
+ )
+ .complete(delayed_result);
+ let ShardFrame::Lifecycle(LifecycleFrame::PollCompleted(completion)) = owner_inbox
+ .try_recv()
+ .expect("late completion reached the owner inbox")
+ else {
+ panic!("expected the completed poll");
+ };
+ owner.on_poll_completed(*completion).await;
+ let (stored_offset, _) = partitions
+ .consumer_offset_read(&namespace, consumer)
+ .unwrap();
+ assert_eq!(
+ stored_offset,
+ Some(2),
+ "the nonempty late result advances local progress through the three unseen messages"
+ );
+
+ // A new Next now returns only offset 3. Disabling its automatic commit
+ // keeps the final cursor attributable to the timed out poll alone.
+ let next_poll = owner.partition_read(
+ namespace,
+ PartitionRead::Poll {
+ consumer,
+ args: PollingArgs {
+ strategy: PollingStrategy::next(),
+ count: 4,
+ auto_commit: false,
+ },
+ },
+ );
+ futures::pin_mut!(next_poll);
+ assert!(futures::poll!(next_poll.as_mut()).is_pending());
+ let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead {
+ namespace,
+ read,
+ reply,
+ }) = owner_inbox
+ .try_recv()
+ .expect("subsequent Next reached the owner inbox")
+ else {
+ panic!("expected the subsequent poll request");
+ };
+ owner.on_partition_read(namespace, read, reply).await;
+ let Some(PartitionReadReply::Poll { fragments, .. }) = next_poll.await else {
+ panic!("subsequent Next must return messages successfully");
+ };
+ assert_eq!(
+ message_offsets(&fragments),
+ vec![3],
+ "Next skips offsets 0 through 2, which the caller never received"
Review Comment:
Updated in [7cdf64cfa](https://github.com/apache/iggy/commit/7cdf64cfa6096d8a1cf1fa807cd22c41469015bb). The regression now requires an unchanged cursor, no automatic commit queued or assigned, and offsets `[0, 1, 2, 3]` on the next poll after the original requester times out. It fails without the disconnect guard and passes with it. This covers timeout before owner completion; the concurrent cancellation race is discussed in the [completion thread](https://github.com/apache/iggy/pull/4119#discussion_r3993707768).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3999336841
##########
core/server/src/dispatch/partition.rs:
##########
@@ -830,6 +489,19 @@ where
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
+ None => {
Review Comment:
Changed the fallback to `Some(other)` and replaced `reply_was_none` with `reply = ?other`. The warning now identifies the unexpected reply variant.
Fixed in [c7e6fb165](https://github.com/apache/iggy/commit/c7e6fb1655bd48c18c2fb567917f88ec1bdf7fcc).
##########
core/partitions/src/iggy_partitions.rs:
##########
@@ -495,6 +488,46 @@ where
Some(partition.build_poll_plan(consumer, args, validate_checksum))
}
+ /// Validate and accept a poll synchronously on the owning pump.
+ /// Send the reply before driving any returned continuation through
+ /// [`Self::replicate_poll_completion`]. Success does not acknowledge a
+ /// durable offset commit.
+ ///
+ /// Acceptance is independent of reply delivery. A caller that stopped
+ /// waiting does not cancel completion: an accepted read can still advance
+ /// progress and admit an automatic commit even if its reply cannot be sent.
+ /// A history rejection accepts no progress from this read, but cannot
+ /// replace a response the transport already sent.
+ ///
+ /// # Errors
+ /// Rejects stale history, unavailable partition or admission state, invalid
+ /// consumer identifiers, and exhausted consumer, queue, or journal capacity.
+ /// Rejection does not accept progress from this read.
+ pub fn complete_poll(
+ &self,
+ namespace: &IggyNamespace,
+ result: PollReadResult,
+ ) -> Result<PollCompletion, IggyError> {
+ let partition = self
+ .get_mut_by_ns(namespace)
+ .ok_or(IggyError::TransientNotAccepted)?;
+ partition.complete_poll(result)
+ }
+
+ /// Drive an accepted poll's replication on the owning pump after its reply.
+ /// Use the continuation returned by [`Self::complete_poll`] for this namespace.
+ /// This call may suspend with a partition borrow, so it must remain on the
+ /// owning pump. A missing or tombstoned namespace drops the continuation.
+ pub async fn replicate_poll_completion(
+ &self,
+ namespace: &IggyNamespace,
+ replication: PollReplication,
+ ) {
+ if let Some(partition) = self.get_mut_by_ns(namespace) {
Review Comment:
Replaced the optional lookup with an invariant assertion and documented that the continuation must run on the same owner pump immediately after the synchronous reply attempt. The new regression checks a tombstoned namespace after acceptance has queued an unjournaled prepare, and was verified to fail with the old silent drop and pass with the assertion.
Fixed in [c7e6fb165](https://github.com/apache/iggy/commit/c7e6fb1655bd48c18c2fb567917f88ec1bdf7fcc).
##########
core/shard/src/metrics.rs:
##########
@@ -175,6 +177,7 @@ const VARIANTS: [&str; VARIANT_COUNT] = [
frame_drop_variant::REPLICA_HANDSHAKE_ACK,
frame_drop_variant::PARTITION_AUTO_COMMIT,
Review Comment:
Removed `PARTITION_AUTO_COMMIT` and its table entry, retiring `frame_drops_total{variant="partition_auto_commit"}` with the deleted submit path. `frame_drops_total{variant="partition_poll_completion"}` measures failures to reserve or deliver poll completions, while `partition_consumer_offsets_denied_total` measures consumer offset capacity denials. Those are different events, so the retired series has no direct replacement.
Fixed in [c7e6fb165](https://github.com/apache/iggy/commit/c7e6fb1655bd48c18c2fb567917f88ec1bdf7fcc).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997476826
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,425 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the history and recovery state, admits any automatic
+//! commit, and updates consumer progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
+ );
+ self.bus.spawn(read_poll(namespace, plan, completion));
+ return;
+ }
+ Some(plan) => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "resident",
+ "partition poll dispatch"
+ );
+ self.on_poll_completed(PollCompleted {
+ namespace,
+ result: plan.execute_resident(),
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: None,
+ })
+ .await;
+ return;
+ }
+ }
+ }
+ PartitionRead::ConsumerOffset { consumer } => partitions
+ .consumer_offset_read(&namespace, consumer)
+ .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| {
+ PartitionReadReply::ConsumerOffset {
+ stored,
+ current_offset,
+ }
+ }),
+ PartitionRead::GroupOffsetState { group_id } => partitions
+ .group_offset_state(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |(last_polled, committed)| {
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }),
+ PartitionRead::ClearGroupLastPolled { group_id } => partitions
+ .clear_group_last_polled(&namespace, group_id)
+ .map_or(PartitionReadReply::NotFound, |()| PartitionReadReply::Ack),
+ PartitionRead::ResolveSegmentDeleteOffset { count } => partitions
+ .segment_delete_resolution(&namespace, count)
+ .map_or(PartitionReadReply::NotFound, |(up_to_offset, lagging)| {
+ PartitionReadReply::SegmentDeleteOffset {
+ up_to_offset,
+ lagging,
+ }
+ }),
+ };
+ let _ = reply.try_send(result);
+ }
+
+ /// Accept a read on the owner's pump, then attempt the reply before replication.
+ /// Replication may suspend. A closed reply channel does not cancel an
+ /// automatic commit that the owner has already admitted.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_poll_completed(&self, completion: PollCompleted) {
+ let PollCompleted {
+ namespace,
+ result,
+ reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at,
+ } = completion;
+ #[cfg(feature = "poll-diagnostics")]
+ if let Some(queued_at) = queued_at {
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "completion",
+ tier = "disk",
+ queue_wait_us = u64::try_from(queued_at.elapsed().as_micros()).unwrap_or(u64::MAX),
+ "partition poll completion"
+ );
+ }
+ let partitions = self.plane.partitions();
+ let consumer_kind = result.consumer_kind();
+ match partitions.complete_poll(&namespace, result) {
+ Ok(PollCompletion {
+ fragments,
+ current_offset,
+ replication,
+ }) => {
+ // The owner has admitted progress, but the offset may still be
+ // queued. Release the reply before waiting for replication.
+ // A poll reply does not acknowledge a durable offset commit.
+ let _ = reply.try_send(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ });
+ if let Some(replication) = replication {
+ partitions
+ .replicate_poll_completion(&namespace, replication)
+ .await;
+ }
+ }
+ Err(error) => {
+ if matches!(error, IggyError::TooManyConsumerOffsets) {
+ self.metrics.record_consumer_offset_denied(consumer_kind);
+ }
+ let _ = reply.try_send(PartitionReadReply::Rejected(error));
+ }
+ }
+ }
+}
+
+/// Read an owned snapshot without borrowing the partition or changing progress.
+/// The completion sender returns the result to the owner for acceptance.
+#[allow(clippy::future_not_send)]
+async fn read_poll(
+ namespace: IggyNamespace,
+ plan: PollPlan,
+ completion: completion::PollCompletionSender,
+) {
+ let poll_started = std::time::Instant::now();
+ let result = plan.execute().await;
+ let elapsed = poll_started.elapsed();
+ if elapsed > std::time::Duration::from_secs(1) {
+ tracing::warn!(
+ namespace_raw = namespace.inner(),
+ elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
+ "slow partition poll; gather side may have timed out"
+ );
+ }
+ completion.complete(result);
+}
+
+mod completion {
+ use super::PollCompleted;
+ use crate::coordinator::classify_try_send_err;
+ use crate::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant};
+ use crate::{LifecycleFrame, PartitionReadReply, Sender, ShardFrame, TaggedSender};
+ use iggy_common::IggyError;
+ use partitions::PollReadResult;
+ use server_common::sharding::IggyNamespace;
+
+ /// The read task can return bytes to the owner or reject a lost completion.
+ /// The reply sender stays private so it cannot authorize successful reads.
+ pub(super) struct PollCompletionSender {
+ /// Originating owner's inbox, whose absence causes a rejection.
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ /// May be used here only to reject a result that cannot reach its owner.
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ }
+
+ impl PollCompletionSender {
+ /// Bind the result and reply route to the shard that built the snapshot.
+ pub(super) const fn new(
+ inbox: Option<TaggedSender>,
+ namespace: IggyNamespace,
+ reply: Sender<PartitionReadReply>,
+ metrics: ShardMetrics,
+ ) -> Self {
+ Self {
+ inbox,
+ namespace,
+ reply,
+ metrics,
+ }
+ }
+
+ /// Return the result to the owner without waiting for inbox capacity.
+ /// A missing, full, or disconnected inbox records a dropped completion
+ /// and attempts a `TransientNotAccepted` reply. Successful enqueueing
+ /// leaves the reply to the owner.
+ pub(super) fn complete(self, result: PollReadResult) {
+ let Some(inbox) = self.inbox else {
+ self.metrics.record_frame_drop(
+ frame_drop_variant::PARTITION_POLL_COMPLETION,
+ frame_drop_reason::UNROUTABLE,
+ );
+ let _ = self.reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ };
+ let frame =
+ ShardFrame::lifecycle(LifecycleFrame::PollCompleted(Box::new(PollCompleted {
+ namespace: self.namespace,
+ result,
+ reply: self.reply,
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Some(std::time::Instant::now()),
+ })));
+ if let Err(error) = inbox.try_send(frame) {
Review Comment:
Agreed that reserving completion capacity before disk I/O would bound outstanding work without making workers wait for inbox space.
Could you point me to the existing budget that caps outstanding disk polls? In the current path, `on_partition_read` builds the snapshot and spawns `read_poll` without acquiring a permit. `PARTITION_READ_TIMEOUT` bounds the requester’s wait but leaves the disk read running. The request inbox also frees its slot when the owner dequeues the request.
For a completion lane of capacity `N`, we need:
```text
running disk polls + completions awaiting the owner <= N
```
Releasing a read permit when I/O finishes would let new reads start while earlier results still occupy the completion lane. Capacity therefore needs to remain accounted for until the completion is dequeued or discarded, even after requester timeout. That can be a permit retained by the operation or a reservation transferred into queue occupancy. Unrelated frames must not consume that capacity.
If an existing budget covers this lifetime, we can reuse it. Otherwise, reserving a completion slot before dispatch establishes the guarantee directly: reject before disk I/O when no slot is available, while preserving the existing handling for missing and disconnected routes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(partitions): complete polls on the owning shard [iggy]
Posted by "diegomrsantos (via GitHub)" <gi...@apache.org>.
diegomrsantos commented on code in PR #4119:
URL: https://github.com/apache/iggy/pull/4119#discussion_r3997781814
##########
core/shard/src/poll.rs:
##########
@@ -0,0 +1,466 @@
+// 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.
+
+//! Poll reads complete under the partition owner's control.
+//!
+//! 1. The owner snapshots the history identity and read resources.
+//! 2. Resident reads complete inline. Disk reads run in a detached task and
+//! return their result through the owner's inbox.
+//! 3. The owner checks the reply connection, history, and recovery state, admits
+//! any automatic commit, and updates progress before releasing the reply.
+//!
+//! A disk read can yield while purge or state transfer replaces the history,
+//! even on the same shard thread. The detached task therefore cannot advance
+//! progress or authorize a successful reply. Completion validation and progress
+//! updates run synchronously on the owner, before any replication wait.
+
+use crate::shards_table::ShardsTable;
+use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender};
+use consensus::PartitionsHandle;
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use partitions::{PollCompletion, PollPlan, PollReadResult};
+use server_common::sharding::IggyNamespace;
+
+#[cfg(test)]
+mod completion_tests;
+#[cfg(test)]
+mod test_support;
+#[cfg(test)]
+mod timeout_tests;
+
+/// A read result awaiting acceptance by its partition owner.
+/// Disk tasks send it through the inbox. Resident reads pass it directly to
+/// the same completion handler.
+///
+/// The inbox requires `Send` even for messages addressed to this same shard.
+/// The owner acquires any capacity reservation while accepting the result.
+/// Guards stay in its local request queue or replication state, outside this
+/// channel's `Send` boundary.
+pub struct PollCompleted {
+ /// Namespace whose current partition must validate the captured history.
+ namespace: IggyNamespace,
+ /// Snapshot facts that have not yet been accepted as consumer progress.
+ result: PollReadResult,
+ /// Return path for the accepted read or a rejection.
+ reply: Sender<PartitionReadReply>,
+ /// Inbox enqueue time for disk diagnostics, or `None` for resident completion.
+ #[cfg(feature = "poll-diagnostics")]
+ queued_at: Option<std::time::Instant>,
+}
+
+impl<B, MJ, S, M, T, SB> IggyShard<B, MJ, S, M, T, SB>
+where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ SB: SuperblockStore,
+{
+ /// Execute a routed read on the partition owner's pump.
+ /// Partitions missing materialized data reject the read. Resident polls finish
+ /// inline. Disk polls return to the pump for acceptance after detached I/O.
+ #[allow(clippy::future_not_send)]
+ pub(crate) async fn on_partition_read(
+ &self,
+ namespace: IggyNamespace,
+ read: PartitionRead,
+ reply: Sender<PartitionReadReply>,
+ ) {
+ let partitions = self.plane.partitions();
+ if partitions.with_partition(
+ &namespace,
+ partitions::IggyPartition::requires_state_transfer,
+ ) == Some(true)
+ {
+ let _ = reply.try_send(PartitionReadReply::Rejected(
+ IggyError::TransientNotAccepted,
+ ));
+ return;
+ }
+ let result = match read {
+ PartitionRead::Poll { consumer, args } => {
+ match partitions.build_poll_snapshot(&namespace, consumer, &args) {
+ None => PartitionReadReply::NotFound,
+ Some(plan) if plan.needs_off_pump_io() => {
+ #[cfg(feature = "poll-diagnostics")]
+ tracing::debug!(
+ target: "iggy.shard.poll_diagnostics",
+ namespace_raw = namespace.inner(),
+ phase = "dispatch",
+ tier = "disk",
+ "partition poll dispatch"
+ );
+ let completion = completion::PollCompletionSender::new(
+ self.senders.get(usize::from(self.id)).cloned(),
+ namespace,
+ reply,
+ self.metrics.clone(),
Review Comment:
Done in [3b9619ea2](https://github.com/apache/iggy/commit/3b9619ea2346b1909c7d24a55ad345b99f429a64). The completion lane shares the frame drop family and cache through its existing state, removing the full `ShardMetrics` clone from each disk read dispatch. The `const` accessor and `Send` bootstrap inputs are preserved. The regression verifies lazy metric creation and that a late completion updates the registered counter after lane teardown. Latency remains unmeasured.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org