krishvishal commented on code in PR #3975:
URL: https://github.com/apache/iggy/pull/3975#discussion_r3907462383
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -998,6 +1208,158 @@ 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. 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.should_increment_offset
+ {
+ return false;
+ }
+ let headroom = self
+ .durable_offset_reserved
+ .get()
+ .saturating_sub(self.mint_frontier());
+ headroom < self.offset_reservation_lease / 2
Review Comment:
Fixed. `(lease / 2).max(1)`, so the knob's documented range stays as it is.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1812,14 +2184,18 @@ where
return Err(IggyError::CannotAppendMessage);
}
- let dirty_offset = if self.should_increment_offset {
+ let next = if self.should_increment_offset {
self.dirty_offset
.load(Ordering::Relaxed)
.checked_add(1)
.ok_or(IggyError::CannotAppendMessage)?
} else {
0
};
+ // Only here: this is the only path that mints. A backup re-stamps what
+ // the primary sends (`append_received_send_messages_to_journal`) and
+ // must follow it exactly, so raising ITS counter would fork the group.
+ let dirty_offset = next;
Review Comment:
Fixed. Binding renamed back to `dirty_offset`, rebinding dropped.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -3846,98 +4319,49 @@ where
}
async fn rotate_segment(&mut self, config: &PartitionsConfig) ->
Result<(), IggyError> {
- let namespace = self.namespace();
- let old_segment_index = self.log.segments().len() - 1;
- let active_segment = self.log.active_segment_mut();
- active_segment.sealed = true;
- let start_offset = active_segment.end_offset + 1;
-
- let segment_size = self.effective_segment_size(config);
- let enforce_fsync = self.effective_enforce_fsync(config);
- let preallocate_segments = self.effective_preallocate_segments(config);
- let segment = Segment::new(start_offset, segment_size);
- // Prefer the active writer's location: a per-topic path override or a
- // config change after the initial segment was created must not scatter
- // one partition's segments across two directories. The config layout
- // only decides for a partition with no writer yet.
- let (messages_path, index_path) = self.partition_dir().map_or_else(
- || {
- (
- config.get_messages_path(
- namespace.stream_id(),
- namespace.topic_id(),
- namespace.partition_id(),
- start_offset,
- ),
- config.get_index_path(
- namespace.stream_id(),
- namespace.topic_id(),
- namespace.partition_id(),
- start_offset,
- ),
- )
- },
- |dir| {
- (
- format!("{dir}/{start_offset:0>20}.log"),
- format!("{dir}/{start_offset:0>20}.index"),
- )
- },
- );
+ let start_offset = self.log.active_segment().end_offset + 1;
+ self.rotate_segment_at(config, start_offset).await
+ }
- let storage = SegmentStorage::new(&messages_path, &index_path, 0, 0,
false)
- .await
- .map_err(|_|
IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?;
- let messages_size_bytes = storage
- .messages_writer
- .as_ref()
- .ok_or_else(||
IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?
- .size_counter();
- let messages_writer = Rc::new(
- MessagesWriter::new(
- &messages_path,
- messages_size_bytes,
- enforce_fsync,
- false,
- preallocate_segments.then_some(segment_size),
- )
- .await
- .map_err(|_|
IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?,
- );
- let index_size_bytes = storage
- .index_writer
- .as_ref()
- .ok_or_else(||
IggyError::CannotCreateSegmentIndexFile(index_path.clone()))?
- .size_counter();
- let index_writer = Rc::new(
- IggyIndexWriter::new(&index_path, index_size_bytes, enforce_fsync,
false)
- .await
- .map_err(|_|
IggyError::CannotCreateSegmentIndexFile(index_path.clone()))?,
- );
+ /// Seal the active segment and plant a fresh empty one at `start_offset`.
+ ///
+ /// Shared by the size-driven roll, which plants at `end_offset + 1`, and
the
+ /// boot re-anchor, which plants at the append point the reservation moved
the
+ /// counter to. One seal path, and one order: the new segment's files are
+ /// created BEFORE the sealed segment's writers are torn down, so a failed
+ /// create leaves the chain serviceable.
+ async fn rotate_segment_at(
Review Comment:
Fixed. `debug_assert!(start_offset > sealed_end)`, plus a debug-only
read-back asserting the anchor `covers` the sealed bounds when the gap is wider.
##########
core/partitions/src/segment_anchor.rs:
##########
@@ -0,0 +1,205 @@
+// 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.
+
+//! The record that makes a gap in the segment chain legitimate.
+//!
+//! Recovery derives each segment's bounds from its own bytes, so a gap has no
+//! author: the boot re-anchor's planted gap and a lost segment look identical.
+//! The re-anchor writes its intent down instead, and the chain guard admits a
+//! forward gap only when the far side carries an anchor naming exactly the
near
+//! side.
+//!
+//! Written and directory-fsynced BEFORE that segment is created. A crash in
the
+//! window then leaves an anchor with no segment, which the boot sweep
collects;
+//! the other order leaves a planted segment with no anchor, which the guard
+//! reads as damage on an intact chain.
+
+use compio::io::AsyncWriteAtExt;
+use consensus::state_artifact_checksum;
+use std::io;
+
+/// File extension for an anchor record, `{start_offset:020}.anchor` beside the
+/// `{start_offset:020}.log` it belongs to.
+pub const ANCHOR_EXTENSION: &str = "anchor";
+
+/// Leading bytes of an anchor record, so a file that is not one (a truncated
+/// write, an operator's copy) is refused rather than decoded.
+const ANCHOR_MAGIC: u64 = u64::from_le_bytes(*b"IGGYANCH");
+
+/// `magic`(8) + `sealed_start`(8) + `sealed_end`(8) + `checksum`(8).
+pub const ANCHOR_ENCODED_LEN: usize = 32;
+
+/// The gap one planted segment is allowed to leave behind it.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct SegmentAnchor {
+ /// Start offset of the segment that was sealed, i.e. the near side of the
+ /// gap. Names WHICH segment, so an anchor cannot be satisfied by a
+ /// different file that happens to end where this one expects.
+ pub sealed_start: u64,
+ /// End offset the sealed segment held when it was sealed. The gap runs
from
+ /// here to the planted segment's own start offset.
+ pub sealed_end: u64,
+}
+
+impl SegmentAnchor {
+ /// Encode to the fixed little-endian on-disk layout.
+ #[must_use]
+ pub fn to_bytes(&self) -> [u8; ANCHOR_ENCODED_LEN] {
+ let mut out = [0u8; ANCHOR_ENCODED_LEN];
+ out[0..8].copy_from_slice(&ANCHOR_MAGIC.to_le_bytes());
+ out[8..16].copy_from_slice(&self.sealed_start.to_le_bytes());
+ out[16..24].copy_from_slice(&self.sealed_end.to_le_bytes());
+ let checksum = state_artifact_checksum(&out[0..24]);
+ out[24..32].copy_from_slice(&checksum.to_le_bytes());
+ out
+ }
+
+ /// Decode a record, returning `None` for anything this build did not
write:
+ /// a wrong length, a wrong magic, or a checksum that does not match.
+ ///
+ /// A `None` is never treated as "no gap was intended". It means the record
+ /// proves nothing, so the gap it would have covered stays damage.
+ #[must_use]
+ pub fn from_bytes(bytes: &[u8]) -> Option<Self> {
+ if bytes.len() != ANCHOR_ENCODED_LEN {
+ return None;
+ }
+ let field = |at: usize| -> u64 {
+ let mut raw = [0u8; 8];
+ raw.copy_from_slice(&bytes[at..at + 8]);
+ u64::from_le_bytes(raw)
+ };
+ if field(0) != ANCHOR_MAGIC || field(24) !=
state_artifact_checksum(&bytes[0..24]) {
+ return None;
+ }
+ Some(Self {
+ sealed_start: field(8),
+ sealed_end: field(16),
+ })
+ }
+
+ /// Whether this anchor legitimises the gap between the segment starting at
+ /// `sealed_start` / ending at `sealed_end` and the segment it sits beside.
+ ///
+ /// BOTH bounds must match. Matching only the end offset would let an
anchor
+ /// left by an earlier incarnation of the chain cover a gap it never saw.
+ #[must_use]
+ pub const fn covers(&self, sealed_start: u64, sealed_end: u64) -> bool {
+ self.sealed_start == sealed_start && self.sealed_end == sealed_end
+ }
+}
+
+/// Path of the anchor record beside the segment starting at `start_offset`.
+#[must_use]
+pub fn anchor_path(partition_dir: &str, start_offset: u64) -> String {
+ format!("{partition_dir}/{start_offset:0>20}.{ANCHOR_EXTENSION}")
+}
+
+/// Write the anchor for a segment about to be planted at `start_offset`, then
+/// fsync the directory so the record cannot arrive after the segment it
+/// describes.
+///
+/// # Errors
+///
+/// Any I/O failure. The caller must NOT plant the segment: a planted segment
+/// whose anchor is missing reads as damage on the next boot.
+pub async fn write_anchor(
+ partition_dir: &str,
+ start_offset: u64,
+ anchor: SegmentAnchor,
+) -> io::Result<()> {
+ let path = anchor_path(partition_dir, start_offset);
+ let mut file = compio::fs::File::create(&path).await?;
Review Comment:
Fixed. Temp, fsync, rename, fsync the dir, like the superblock; the temp
uses `STAGING_SUFFIX` so every existing sweep already unlinks it.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -677,22 +692,78 @@ where
/// THIS partition's group is fenced; the rest of the node keeps serving.
#[allow(clippy::future_not_send)]
async fn write_superblock(&self, superblock: &SB, offset_frontier: u64) ->
bool {
- // ADVANCE direction: never below what this replica has already minted,
- // and never below what the record ALREADY holds. Both bounds are
- // needed and neither implies the other -- a failed install leaves the
- // counter behind the record it wrote before the swap, so maxing
against
- // the counter alone lets the fence that follows lower the durable
- // frontier. The reset direction goes through `write_superblock_inner`.
- let advanced = offset_frontier
- .max(self.offset_frontier())
- .max(self.durable_offset_frontier.get());
- self.write_superblock_inner(superblock, advanced).await
+ self.write_superblock_advancing(superblock, offset_frontier, 0)
+ .await
+ }
+
+ /// [`Self::write_superblock`] for a caller that also has a reservation to
+ /// claim. Both fields advance, neither can regress.
+ #[allow(clippy::future_not_send)]
+ async fn write_superblock_advancing(
+ &self,
+ superblock: &SB,
+ offset_frontier: u64,
+ offset_reserved: u64,
+ ) -> bool {
+ // ADVANCE direction; the reset direction goes through
+ // `write_superblock_inner`. Both bounds inside `advanced_frontier` are
+ // needed: a failed install leaves the chain behind the record it wrote
+ // before the swap, so maxing against the data alone would let the
fence
+ // that follows lower the durable frontier.
+ let advanced = self.advanced_frontier(offset_frontier);
+ // Nothing but the record witnesses a reservation, so a caller with no
+ // claim of its own (every view-change write) passes 0 and carries the
+ // recorded one forward; dropping it would let the next boot seed the
+ // counter below what an earlier append already fenced.
+ let reserved = offset_reserved.max(self.durable_offset_reserved.get());
+ self.write_superblock_inner(superblock, advanced, reserved)
+ .await
+ }
+
+ /// The advance rule for the frontier, shared by every writer that claims
+ /// one: never below what this replica holds, never below what the record
+ /// already says. Held messages, NOT [`Self::mint_frontier`], which stands
a
+ /// lease block above them after a reservation-seeded boot and names none
of
+ /// them.
+ fn advanced_frontier(&self, claim: u64) -> u64 {
+ claim
+ .max(self.held_offset_frontier())
+ .max(self.durable_offset_frontier.get())
+ }
+
+ /// Record an incoming state-transfer frontier, advancing the frontier but
+ /// SETTING the reservation to it.
Review Comment:
Fixed. Doc now says the reservation lands on the frontier this write
records, and names the clamp that puts it there.
##########
core/consensus/src/vsr_state.rs:
##########
@@ -31,20 +31,26 @@ use std::fmt;
/// Number of bytes [`VsrState::to_bytes`] produces: `cluster`(16) +
/// `replica_id`(1) + `replica_count`(1) + `view`(4) + `log_view`(4) +
/// `commit_max`(8) + `checkpoint_op`(8) + `checkpoint_checksum`(16) +
-/// `offset_frontier`(8).
-pub const ENCODED_LEN: usize = 66;
+/// `offset_frontier`(8) + `offset_reserved`(8).
+///
+/// Growing this is one-way: a record of this length is
[`VsrStateError::WrongLength`]
Review Comment:
Fixed. Moved to a `DOWNGRADE:` note beside `offset_reservation_lease` in
`config.toml` and into the PR description; the rustdoc points at it.
--
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]
