krishvishal commented on code in PR #3975:
URL: https://github.com/apache/iggy/pull/3975#discussion_r3911643726
##########
core/configs/src/server_config/partition.rs:
##########
@@ -177,6 +191,16 @@ impl Validatable<ConfigurationError> for PartitionConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ if self.offset_reservation_lease == 0
Review Comment:
You are right, and my reply was wrong: the file was written but never made
it into ccf5adcde. Re-added, three cases, mirroring the other two knobs. Not
pushed yet, so it will show up in the next push rather than on this head.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -998,6 +1262,169 @@ where
self.write_superblock(superblock.as_ref(), frontier).await
}
+ /// Whether the offset reservation is close enough to being consumed that
it
+ /// should be extended NOW, off the append path.
+ ///
+ /// The append fence is correct but badly placed: it writes the superblock
+ /// inline in the shard's frame pump, where the consensus tick is a sibling
+ /// arm, so its two fsyncs delay heartbeats for every group on the core.
The
+ /// fix is to make the fence's fast path
+ /// (`durable_offset_reserved > end_offset`) the only path it ever takes
under
+ /// load, by extending from the tick instead.
+ ///
+ /// HALF a block of headroom, which is a wide margin on purpose:
over-claiming
+ /// costs nothing but offset space, while arriving late puts the write
back on
+ /// the append path. Floored at 1, since validation admits a lease of 1 and
+ /// `1 / 2` would never trigger, leaving every append to pay the inline
claim.
+ /// A partition that has never minted is skipped -- its first append
+ /// legitimately pays for the first claim, and extending every idle
partition
+ /// at boot would write a superblock per partition for nothing.
+ #[must_use]
+ pub fn needs_offset_reservation_extension(&self) -> bool {
+ if self.consensus.replica_count() > 1
+ || self.superblock.is_none()
+ || !self.offset_space.append_live
+ {
+ return false;
+ }
+ let headroom = self
+ .durable_offset_reserved
+ .get()
+ .saturating_sub(self.mint_frontier());
+ headroom < (self.offset_reservation_lease / 2).max(1)
+ }
+
+ /// Extend the reservation a full block past the current append point.
+ ///
+ /// Pairs with [`Self::needs_offset_reservation_extension`]; the caller is
the
+ /// shard tick, so this write is off the append path. A failure needs no
+ /// handling beyond the writer's own logging and backoff: the fence at the
+ /// mint is still there, and it is what refuses the append if the ceiling
+ /// never caught up.
+ #[allow(clippy::future_not_send)]
+ pub async fn extend_offset_reservation(&self) -> bool {
+ let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else {
+ return true;
+ };
+ if self.superblock_write_is_backed_off() {
+ return false;
+ }
+ let _superblock_guard = self.superblock_lock.acquire().await;
+ self.write_offset_claim(superblock.as_ref(), self.mint_frontier())
+ .await
+ }
+
+ /// Upper bound on the offsets a pending `SendMessages` request will mint,
for
+ /// fencing it BEFORE it enters the pipeline.
+ ///
+ /// `project` assigns an op, not a base offset, so the exact range is
unknown
+ /// until the mint runs under `write_lock`. This is deliberately loose: a
+ /// request pipelined behind others can land above it, and the fence at the
+ /// mint stays as the exact check. It does not need to be tight -- the
claim
+ /// runs a whole lease block past whatever it is handed, so one of these
+ /// covers every batch in flight unless a run of them crosses a block
+ /// boundary.
+ ///
+ /// `None` above one replica, where nothing is reserved, and when the body
is
+ /// not one canonical batch, which `convert_request_message` has already
+ /// rejected by the time this runs.
+ ///
+ /// Header decode, NOT `decode_batch_slice`: the verifying decode fails on
+ /// every ordinary send, because `convert_request_message` runs at
+ /// [`ChecksumMode::Skip`] and leaves `batch_checksum` zeroed, which would
+ /// silently drop the fence back to the mint. It also re-hashes every body
+ /// `admit_wire_request` already hashed.
+ fn request_mint_ceiling(&self, message: &Message<RoutedRequestHeader>) ->
Option<u64> {
+ if self.consensus.replica_count() > 1 {
+ return None;
+ }
+ let body = message
+ .as_slice()
+
.get(std::mem::size_of::<RoutedRequestHeader>()..message.header().size as
usize)?;
+ let count = BatchHeader::decode(body).ok()?.message_count;
+ Some(self.mint_frontier().saturating_add(u64::from(count)))
+ }
+
+ /// The append fence: make sure the durable record already permits every
+ /// offset up to and including `end_offset` before the caller lets them
+ /// exist.
+ ///
+ /// `SendMessagesResponse` hands clients concrete base offsets and the poll
+ /// path serves committed messages out of the resident journal, so an
offset
+ /// is client-visible long before the threshold-gated flush names it in a
+ /// segment, and a crash in between hands a second message an offset a
client
+ /// already holds. Fencing here rather than at commit puts it upstream of
+ /// every way a NEWLY minted offset escapes -- the reply, the poll tier,
the
+ /// peers a prepare reaches -- on the one path both a primary's mint and a
+ /// backup's re-stamp take. Journal repair
+ /// (`append_repaired_send_messages`) is the exception and needs none: it
+ /// re-journals offsets a peer already minted and fenced, so there is
nothing
+ /// new to claim. Claiming through `end_offset + 1 + lease` rather than
from
+ /// the live counter needs no special case for an oversized batch.
+ ///
+ /// SOLO ONLY, like everything the reservation feeds:
`restore_offset_frontier`
+ /// seeds no counter from it above one replica, and the boot re-anchor that
+ /// shapes the chain around it never runs there either. A replicated group
+ /// paying a superblock write per block would buy nothing -- and an ack
there
+ /// already means a quorum journaled the batch, so re-minting needs a
+ /// FULL-cluster crash.
+ ///
+ /// `false` when the write was attempted and failed. Fail-closed: the send
is
+ /// rejected with nothing externalised, exactly as a failed view persist
+ /// withholds its sends.
+ ///
+ /// BYPASSES the retry backoff, like `record_frontier_before_quarantine`
and
Review Comment:
Fixed, and good catch: the bypass turned the wedge count into a function of
producer retry rate. `write_superblock_inner` now samples the backoff window
before the write and, on a failure inside it, neither counts nor extends. So
the count is one per backoff step again, which is what `superblock_wedged`
reads as elapsed time, and a bypassing writer can no longer both starve the
gated writers and pull the exit forward. Lands in the next push.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1564,7 +2001,7 @@ where
// commit). Also used below as the poll's high-water bound: this
function
// is fully synchronous, so the single load cannot drift mid-plan.
let commit_offset = self.offsets().commit_offset;
- if !self.should_increment_offset || args.count == 0 {
+ if !self.offset_space.append_live || args.count == 0 {
Review Comment:
Fixed. Gated on `committed_seeded`, which is the bit the comment above it
describes. Lands in the next push.
--
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]
