numinnex commented on code in PR #4169:
URL: https://github.com/apache/iggy/pull/4169#discussion_r4003024367
##########
core/partitions/src/poll_plan.rs:
##########
@@ -383,16 +387,82 @@ pub enum DiskReadOutcome {
Faulted,
}
+/// Largest first read of a disk poll, and the size every poll used to read
+/// whatever it asked for. A batch wider than this still grows past it through
+/// the re-read path below; this bounds only where a walk starts.
+const DISK_POLL_CHUNK_MAX: u64 = 1 << 20;
+
+/// Smallest first read of a disk poll. Below this the syscall and the segment
+/// walk cost more than the bytes the smaller read saves, and a poll for a
+/// handful of messages would issue a read per batch.
+const DISK_POLL_CHUNK_MIN: u64 = 64 << 10;
+
+/// How the chunk loop over one segment ended.
+enum SegmentWalk {
+ /// The segment is exhausted or the requested count is filled. The walk
+ /// may continue into the next segment.
+ Done,
+ /// Fail-closed: the segment may hold present-but-unreadable or corrupt
+ /// data, so no later segment may be served over it.
+ Faulted,
+}
+
+/// The state one disk walk carries across its segments.
+struct DiskWalk {
+ /// Byte offset into the segment being walked; reset at each boundary.
+ position: u64,
+ matched: u32,
+ fragments: PollFragments<4096>,
+ last_matching_offset: Option<u64>,
+ #[cfg(feature = "poll-diagnostics")]
+ requested_bytes: u64,
+ #[cfg(feature = "poll-diagnostics")]
+ chunk_reads: u32,
+}
+
+impl DiskWalk {
+ fn starting_at(position: u64) -> Self {
+ Self {
+ position,
+ matched: 0,
+ fragments: PollFragments::new(),
+ last_matching_offset: None,
+ #[cfg(feature = "poll-diagnostics")]
+ requested_bytes: 0,
+ #[cfg(feature = "poll-diagnostics")]
+ chunk_reads: 0,
+ }
+ }
+}
+
impl DiskReadPlan {
+ /// Bytes to read for the next `remaining` messages.
+ ///
+ /// A poll asks for a message count, and the walk reads bytes, so the two
+ /// are bridged by the partition's own mean encoded size. Reading a fixed
+ /// megabyte instead costs a poll for a thousand hundred-byte messages ten
+ /// times the bytes it returns, and the sparse-selection copy that follows
+ /// scales with the chunk rather than with the selection.
+ ///
+ /// The estimate is deliberately not a bound. Messages vary in size, the
+ /// starting offset can sit inside a batch the index resolved before it,
+ /// and an underestimate only costs another read of the next chunk.
+ fn chunk_len(&self, remaining: u32) -> u64 {
+ let Some(bytes_per_message) = self.bytes_per_message else {
+ return DISK_POLL_CHUNK_MAX;
+ };
+ u64::from(bytes_per_message)
+ .saturating_mul(u64::from(remaining))
+ .saturating_add(COMMAND_HEADER_SIZE as u64)
+ .clamp(DISK_POLL_CHUNK_MIN, DISK_POLL_CHUNK_MAX)
Review Comment:
`chunk_len(c) = mean*c + COMMAND_HEADER_SIZE` equals the producer's batch
size exactly at `c == N` (the +256 cancels the floor loss). One message below
that, the chunk lands under a whole batch, `consumed == 0` discards the entire
read, and the `*4` ladder at :636 starts.
Against the old flat 1 MiB (one read for any `c`): 4.99x the bytes at `c =
N-1`, 2.50x at `c = 500`, 2.10x at `c = 100`, with 2-3x the syscalls.
The SDK rustdoc (`consumer.rs:314`) and
`examples/rust/src/stream-builder/stream-consumer-config` both use
`.batch_length(100)`, while
`bench/src/actors/consumer/client/high_level.rs:138` sets the consumer's batch
length from the producer's, so iggy-bench only ever generates the single
neutral point `c == N`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2987,12 +2993,18 @@ where
self.check_local_poll_key(kind, consumer_id)
.map_err(|error| self.poll_capacity_error(error))?;
let consensus = self.consensus();
- if !consensus.is_primary()
- || !consensus.is_normal()
- || consensus.is_transferring()
- || self
- .durable_consumer_offsets
- .covers(kind, consumer_id, offset)
+ // A replica that cannot originate the prepare cannot record this
+ // progress anywhere a peer will ever see. `Ok(None)` would leave
+ // `complete_poll` applying the offset to local state alone, so the
+ // poll would report progress the group never agreed, and a later
+ // read on the primary would hand the same messages out again.
+ // Refusing keeps the outcome retriable on a replica that can commit.
+ if !consensus.is_primary() || !consensus.is_normal() ||
consensus.is_transferring() {
+ return Err(IggyError::TransientNotAccepted);
Review Comment:
This refusal reaches HTTP as a 503 (`http/error.rs:83`), but `GET
.../messages` is in neither `partition_write_routes` nor `forwardable_routes`,
and `forward.rs:265` short-circuits every GET, so no HTTP client can reach a
node that would serve it.
TCP/QUIC recover via the roster walk. All four HTTP surfaces (Rust, Java,
.NET, Python) treat code 58 as fatal, and the Rust/Python high-level consumers
send `auto_commit=true` at stock settings (`consumer_builder.rs:69` ->
`consumer.rs:745` -> `http/messages.rs:40`).
On any cluster with >=2 replicas, replica 0 is primary for every partition
until a view change, so 2 of 3 HTTP endpoints now fail every auto-commit poll
that previously returned messages.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
