numinnex commented on code in PR #4025:
URL: https://github.com/apache/iggy/pull/4025#discussion_r3905999085
##########
core/server/src/responses.rs:
##########
@@ -1652,8 +1662,148 @@ where
.map_or(0, VsrConsensus::commit_max)
}
+/// Body head of a `PolledMessages` reply:
+/// `[partition_id:4][current_offset:8][count:4]`, before the batch records.
+const POLLED_HEAD_LEN: usize = 16;
+
+/// Build the `PolledMessages` reply for the wire as a vectored frame: one
+/// buffer holding the reply header and the body head, then the poll
+/// fragments as they are. The record bytes are never copied or gathered;
+/// their reply encoding IS the storage encoding (see
+/// [`build_polled_messages_body`]), so `count` comes from walking the batch
+/// headers in place. At-rest decryption is the one case that must rewrite
+/// records, and it takes the flattening builder instead.
+pub fn build_polled_messages_reply(
+ request_header: &RoutedRequestHeader,
+ commit: u64,
+ partition_id: u32,
+ current_offset: u64,
+ fragments: PollFragments,
+ encryptor: Option<&EncryptorKind>,
+) -> Result<BusMessage, IggyError> {
+ let client_id = request_header.client;
+ let session = request_header.session;
+ if encryptor.is_some() {
+ let body = build_polled_messages_body(partition_id, current_offset,
fragments, encryptor)?;
+ let reply = build_reply_from_bytes(request_header, client_id, session,
commit, &body);
+ return Ok(reply.into_generic().into_frozen().into());
+ }
+
+ let mut frames = ResponseFragments::with_capacity(fragments.len() + 1);
+ frames.extend(fragments.into_iter().map(Fragment::into_frozen));
+ let count = polled_message_count(&frames)?;
+ let records_len: usize = frames.iter().map(Frozen::len).sum();
Review Comment:
Each fragment keeps a refcount on its **whole** source allocation, not just
the sliced window: `Frozen::slice` narrows ptr/len while the `ControlBlock`
still owns the base buffer. On the disk tier that source is a `DISK_POLL_CHUNK`
read buffer (1 MiB, `poll_plan.rs:493`, grown 4^k at `:607`), and
`walk_disk_chunk` stops as soon as `matched >= count`.
On master `build_polled_messages_body` took `PollFragments` by value and
released the chunk before the reply was enqueued. Here the pin lasts as long as
the frame sits in the per-connection mailbox (`peer_queue_capacity = 256`).
So a `count=1` poll off a cold partition ships ~600 B and retains 1 MiB. 100
stalled consumers x 256 queued is ~25.6 GiB held against a few MB of actual
reply bytes.
What makes this different from the pre-existing `256 * MAX_MESSAGE_SIZE`
ceiling: that bound is client-requested and therefore self-limiting, while this
one is server read granularity with no relation to what was asked for. And
`peer_queue_capacity` counts frames, so nothing in the system observes the
retained bytes -- there is no gauge, no byte knob, and every other
retained-payload path in the server has an explicit byte budget
(`transfer_served_cache_bytes_max` / `PARTITION_ARTIFACT_LEN_DEFAULT`).
Suggest bounding the pin: copy the fragment out when `fragment.len()` is a
small fraction of its source allocation, or read the exact batch extent on the
disk tier instead of a 1 MiB chunk.
--
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]
