You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/08/31 19:43:11 UTC
(iggy) branch master updated: fix(cluster): stop a late partition materialiser from wedging its group (#4007)
This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new e9b1c9a12 fix(cluster): stop a late partition materialiser from wedging its group (#4007)
e9b1c9a12 is described below
commit e9b1c9a12d0baaf74129d6925c2f114663536151
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Aug 31 21:43:07 2026 +0200
fix(cluster): stop a late partition materialiser from wedging its group (#4007)
A partition group with no history of its own seeds its view and
log_view from the metadata view this replica happens to publish.
Nothing bounds that view by the group's real one, so a replica that
materialises the group late, after a further metadata election,
starts Normal on an empty log above every peer. That empty log is
canonical in the next DVC merge: as a backup it collects a nack
quorum against committed ops and the group deadlocks for good, the
superblock persisting the view so a restart re-wedges; as the
primary-by-index it answers the real primary's heartbeat with a
StartView at op 0 that resets the peers' heads under their commit
floor, and every new write is fenced. Introduced with the seed in
#3985.
Record the view the admitting primary mints the create in on the
committed metadata entry and seed from that instead. Every replica
reads the same value, and the group's view never drops below it, so
an empty log at the creation view is the floor a view-0 start used
to be, never canonical over committed history. The reconciler no
longer needs the published-view sentinel, its deferral, or the
double atomic read that came with it.
The value rides the request body, not the prepare header: a
header's view is the one field a post-view-change retransmit
restamps (replicate_preflight refuses the frame otherwise), so a
header-derived value is a per-replica property of delivery history
and durably diverges the metadata STM. The body is sealed by
checksum_body and never restamped; entries journaled before the
trailing field decode it as 0, the safe floor.
Chosen over electing into the seed: no extra round trip per group,
no consensus change, and a late materialiser joins as a backup at
or below its peers instead of forcing an election it may win with
an empty log. The snapshot format moves to version 5 with the new
trailing field defaulted, so version 3 and 4 checkpoints still
decode.
---
.../create_partitions_with_assignments.rs | 44 +++++
.../topics/create_topic_with_assignments.rs | 46 +++++
core/consensus/src/impls.rs | 47 +++--
core/metadata/src/impls/metadata.rs | 7 +
core/metadata/src/stm/authz.rs | 1 +
core/metadata/src/stm/consumer_group.rs | 1 +
core/metadata/src/stm/snapshot.rs | 33 +++-
core/metadata/src/stm/stream.rs | 126 +++++++++++-
core/server/src/bootstrap.rs | 10 +-
core/server/src/dispatch.rs | 1 +
core/server/src/partition_helpers.rs | 33 ++--
core/server/src/partition_reconciler.rs | 219 +++++----------------
core/server/src/responses.rs | 6 +-
core/shard/src/lib.rs | 11 +-
core/simulator/src/lib.rs | 217 +++++++++++++++++---
core/simulator/src/replica.rs | 6 +-
16 files changed, 544 insertions(+), 264 deletions(-)
diff --git a/core/binary_protocol/src/requests/partitions/create_partitions_with_assignments.rs b/core/binary_protocol/src/requests/partitions/create_partitions_with_assignments.rs
index 1350541af..904ed6e5f 100644
--- a/core/binary_protocol/src/requests/partitions/create_partitions_with_assignments.rs
+++ b/core/binary_protocol/src/requests/partitions/create_partitions_with_assignments.rs
@@ -29,6 +29,10 @@ fn usize_to_u32(value: usize, context: &str) -> u32 {
pub struct CreatePartitionsWithAssignmentsRequest {
pub request: CreatePartitionsRequest,
pub partitions: Vec<CreatedPartitionAssignment>,
+ /// View the admitting primary minted this create in. Rides the body, which
+ /// `checksum_body` seals, so unlike the header's `view` it is never restamped
+ /// by a post-view-change retransmit: every replica decodes the same value.
+ pub created_view: u32,
}
impl WireEncode for CreatePartitionsWithAssignmentsRequest {
@@ -40,6 +44,7 @@ impl WireEncode for CreatePartitionsWithAssignmentsRequest {
.iter()
.map(WireEncode::encoded_size)
.sum::<usize>()
+ + 4
}
fn encode(&self, buf: &mut BytesMut) {
@@ -55,6 +60,7 @@ impl WireEncode for CreatePartitionsWithAssignmentsRequest {
for partition in &self.partitions {
partition.encode(buf);
}
+ buf.put_u32_le(self.created_view);
}
}
@@ -81,10 +87,22 @@ impl WireDecode for CreatePartitionsWithAssignmentsRequest {
partitions.push(partition);
}
+ // Trailing field: entries journaled before it existed end here and read
+ // 0, the view every plane starts in. 1-3 leftover bytes are corruption,
+ // not an old shape, so they still error.
+ let created_view = if offset < buf.len() {
+ let view = read_u32_le(buf, offset)?;
+ offset += 4;
+ view
+ } else {
+ 0
+ };
+
Ok((
Self {
request,
partitions,
+ created_view,
},
offset,
))
@@ -117,10 +135,36 @@ mod tests {
consensus_group_id: 12,
},
],
+ created_view: 9,
};
let bytes = request.to_bytes();
let (decoded, consumed) = CreatePartitionsWithAssignmentsRequest::decode(&bytes).unwrap();
assert_eq!(consumed, bytes.len());
assert_eq!(decoded, request);
}
+
+ #[test]
+ fn decode_without_trailing_view_reads_zero() {
+ let request = CreatePartitionsWithAssignmentsRequest {
+ request: CreatePartitionsRequest {
+ stream_id: WireIdentifier::numeric(1),
+ topic_id: WireIdentifier::numeric(2),
+ partitions_count: 1,
+ },
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 3,
+ consensus_group_id: 11,
+ }],
+ created_view: 9,
+ };
+ // Bytes journaled before the field existed: same shape, no trailing u32.
+ let bytes = request.to_bytes();
+ let old_bytes = &bytes[..bytes.len() - 4];
+ let (decoded, consumed) =
+ CreatePartitionsWithAssignmentsRequest::decode(old_bytes).unwrap();
+ assert_eq!(consumed, old_bytes.len());
+ assert_eq!(decoded.created_view, 0);
+ assert_eq!(decoded.partitions, request.partitions);
+ assert_eq!(decoded.request, request.request);
+ }
}
diff --git a/core/binary_protocol/src/requests/topics/create_topic_with_assignments.rs b/core/binary_protocol/src/requests/topics/create_topic_with_assignments.rs
index b6665b618..632286530 100644
--- a/core/binary_protocol/src/requests/topics/create_topic_with_assignments.rs
+++ b/core/binary_protocol/src/requests/topics/create_topic_with_assignments.rs
@@ -38,6 +38,10 @@ pub struct CreateTopicWithAssignmentsRequest {
pub request: CreateTopicRequest,
pub derived_options: WireOptions,
pub partitions: Vec<CreatedPartitionAssignment>,
+ /// View the admitting primary minted this create in. Rides the body, which
+ /// `checksum_body` seals, so unlike the header's `view` it is never restamped
+ /// by a post-view-change retransmit: every replica decodes the same value.
+ pub created_view: u32,
}
impl WireEncode for CreateTopicWithAssignmentsRequest {
@@ -51,6 +55,7 @@ impl WireEncode for CreateTopicWithAssignmentsRequest {
.iter()
.map(WireEncode::encoded_size)
.sum::<usize>()
+ + 4
}
fn encode(&self, buf: &mut BytesMut) {
@@ -71,6 +76,7 @@ impl WireEncode for CreateTopicWithAssignmentsRequest {
for partition in &self.partitions {
partition.encode(buf);
}
+ buf.put_u32_le(self.created_view);
}
}
@@ -116,11 +122,23 @@ impl WireDecode for CreateTopicWithAssignmentsRequest {
partitions.push(partition);
}
+ // Trailing field: entries journaled before it existed end here and read
+ // 0, the view every plane starts in. 1-3 leftover bytes are corruption,
+ // not an old shape, so they still error.
+ let created_view = if offset < buf.len() {
+ let view = read_u32_le(buf, offset)?;
+ offset += 4;
+ view
+ } else {
+ 0
+ };
+
Ok((
Self {
request,
derived_options,
partitions,
+ created_view,
},
offset,
))
@@ -165,6 +183,7 @@ mod tests {
consensus_group_id: 2,
},
],
+ created_view: 7,
};
let bytes = request.to_bytes();
let (decoded, consumed) = CreateTopicWithAssignmentsRequest::decode(&bytes).unwrap();
@@ -172,6 +191,32 @@ mod tests {
assert_eq!(decoded, request);
}
+ #[test]
+ fn decode_without_trailing_view_reads_zero() {
+ let request = CreateTopicWithAssignmentsRequest {
+ request: CreateTopicRequest {
+ stream_id: WireIdentifier::numeric(1),
+ partitions_count: 1,
+ name: WireName::new("events").unwrap(),
+ options: WireOptions::empty(),
+ },
+ derived_options: WireOptions::empty(),
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 1,
+ }],
+ created_view: 7,
+ };
+ // Bytes journaled before the field existed: same shape, no trailing u32.
+ let bytes = request.to_bytes();
+ let old_bytes = &bytes[..bytes.len() - 4];
+ let (decoded, consumed) = CreateTopicWithAssignmentsRequest::decode(old_bytes).unwrap();
+ assert_eq!(consumed, old_bytes.len());
+ assert_eq!(decoded.created_view, 0);
+ assert_eq!(decoded.partitions, request.partitions);
+ assert_eq!(decoded.request, request.request);
+ }
+
#[test]
fn roundtrip_with_empty_blocks() {
let request = CreateTopicWithAssignmentsRequest {
@@ -183,6 +228,7 @@ mod tests {
},
derived_options: WireOptions::empty(),
partitions: vec![],
+ created_view: 0,
};
let bytes = request.to_bytes();
let (decoded, consumed) = CreateTopicWithAssignmentsRequest::decode(&bytes).unwrap();
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 28865344f..edc60c086 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -1152,14 +1152,21 @@ pub struct VsrRestore<'a> {
/// durable record exists; `log_view` cannot be inferred and stays 0.
pub view_fallback: Option<u32>,
/// Starting `(view, log_view)` for a group with NO history of its own,
- /// consulted only when neither of the two above applies.
+ /// consulted only when neither of the two above applies: the view the
+ /// group was created in, as recorded by the metadata plane.
+ ///
+ /// Safe to start `Normal` in without a quorum only because every replica
+ /// seeds the same value and the group's real view never drops below it:
+ /// an empty log at the creation view is the floor a view-0 start used to
+ /// be, never canonical over committed history. Seeded above the group's
+ /// real view it would be canonical: the empty log wins the next DVC merge
+ /// and wedges the group on its committed ops.
///
/// Sets `log_view` as well as `view`, unlike `view_fallback`: the log is
/// empty, so there is no history to misattribute to the view, and a
/// primary whose `log_view` lags its `view` is treated as mid-transition
/// and answers no `RequestStartView` probe (see
- /// [`VsrConsensus::handle_request_start_view`]) - it would hold its own group's
- /// probes open forever.
+ /// [`VsrConsensus::handle_request_start_view`]).
///
/// Deliberately NOT marked superblock-durable: nothing has been written
/// yet, and claiming otherwise would let a restart resume a view no record
@@ -1192,12 +1199,14 @@ pub struct FreshGroupStart {
///
/// `restarted` is the caller's own evidence of a prior life, since the two
/// paths read it differently: a partition directory already on disk for the
-/// server, a retained in-memory log for the simulator.
+/// server, a retained in-memory log for the simulator. `created_view` is the
+/// view the metadata plane created the group in; see
+/// [`VsrRestore::seed_view`].
#[must_use]
pub const fn fresh_group_start(
restarted: bool,
durable_view: Option<(u32, u32)>,
- metadata_view: Option<u32>,
+ created_view: u32,
) -> FreshGroupStart {
let join = if restarted {
JoinMode::ProbeAsBackup {
@@ -1211,7 +1220,7 @@ pub const fn fresh_group_start(
// `StartView` to move it forward, and seeded above that the reply reads as
// stale and is dropped.
let seed_view = match (durable_view, join) {
- (None, JoinMode::Init) => metadata_view,
+ (None, JoinMode::Init) => Some(created_view),
_ => None,
};
FreshGroupStart { join, seed_view }
@@ -1295,7 +1304,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
tracing::info!(
group,
view,
- "seeded a group with no history of its own into the current view"
+ "seeded a group with no history of its own at its creation view"
);
consensus.set_view(view);
consensus.set_log_view(view);
@@ -4061,33 +4070,23 @@ where
mod fresh_group_start_tests {
use super::{JoinMode, fresh_group_start};
- const METADATA_VIEW: u32 = 4;
+ const CREATED_VIEW: u32 = 4;
/// The case the seed exists for: a group created after the metadata plane
/// elected must not start at view 0, or it names a replica the roster does
/// not advertise and no client can be routed to it.
#[test]
- fn given_a_fresh_group_when_the_metadata_view_moved_should_seed_that_view() {
- let start = fresh_group_start(false, None, Some(METADATA_VIEW));
- assert_eq!(start.join, JoinMode::Init);
- assert_eq!(start.seed_view, Some(METADATA_VIEW));
- }
-
- /// No published view is "no opinion", not "view 0". The caller defers
- /// materialising rather than seeding; this only asserts nothing is
- /// invented here.
- #[test]
- fn given_a_fresh_group_when_no_metadata_view_is_known_should_not_seed() {
- let start = fresh_group_start(false, None, None);
+ fn given_a_fresh_group_when_created_after_an_election_should_seed_the_creation_view() {
+ let start = fresh_group_start(false, None, CREATED_VIEW);
assert_eq!(start.join, JoinMode::Init);
- assert_eq!(start.seed_view, None);
+ assert_eq!(start.seed_view, Some(CREATED_VIEW));
}
/// A durable record outranks the seed: it is what this replica actually
- /// promised, and the seed is a guess about someone else's plane.
+ /// promised, and the seed only describes where the group began.
#[test]
fn given_a_durable_record_when_seeding_should_prefer_the_record() {
- let start = fresh_group_start(false, Some((7, 7)), Some(METADATA_VIEW));
+ let start = fresh_group_start(false, Some((7, 7)), CREATED_VIEW);
assert_eq!(start.seed_view, None);
}
@@ -4096,7 +4095,7 @@ mod fresh_group_start_tests {
/// reads as stale and the replica never rejoins.
#[test]
fn given_a_prior_life_when_seeding_should_probe_without_a_seed() {
- let start = fresh_group_start(true, None, Some(METADATA_VIEW));
+ let start = fresh_group_start(true, None, CREATED_VIEW);
assert!(matches!(start.join, JoinMode::ProbeAsBackup { .. }));
assert_eq!(start.seed_view, None);
}
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index b7b7f8337..6ff355e47 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -3425,6 +3425,11 @@ where
request,
derived_options,
partitions,
+ // Minted once here, on the admitting primary, and sealed by
+ // `checksum_body`: the header's `view` is restamped on a
+ // post-view-change retransmit, so a body-carried copy is the
+ // only per-op view every replica commits identically.
+ created_view: consensus.view(),
}
.to_bytes();
Ok(build_prepare_message(
@@ -3458,6 +3463,8 @@ where
let body = PersistedCreatePartitionsRequest {
request,
partitions,
+ // Same body-carried view as `PersistedCreateTopicRequest`.
+ created_view: consensus.view(),
}
.to_bytes();
Ok(build_prepare_message(
diff --git a/core/metadata/src/stm/authz.rs b/core/metadata/src/stm/authz.rs
index 667b87f02..1e7ec1ec3 100644
--- a/core/metadata/src/stm/authz.rs
+++ b/core/metadata/src/stm/authz.rs
@@ -486,6 +486,7 @@ mod tests {
fn create_topic_body(stream_id: u32, name: &str) -> bytes::Bytes {
CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: CreateTopicRequest {
stream_id: WireIdentifier::numeric(stream_id),
partitions_count: 1,
diff --git a/core/metadata/src/stm/consumer_group.rs b/core/metadata/src/stm/consumer_group.rs
index 22a66c7d4..0e70642bb 100644
--- a/core/metadata/src/stm/consumer_group.rs
+++ b/core/metadata/src/stm/consumer_group.rs
@@ -969,6 +969,7 @@ mod tests {
IggyTimestamp::now(),
);
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: CreateTopicRequest {
stream_id: WireIdentifier::numeric(0),
partitions_count: 1,
diff --git a/core/metadata/src/stm/snapshot.rs b/core/metadata/src/stm/snapshot.rs
index b59a96000..2185b439e 100644
--- a/core/metadata/src/stm/snapshot.rs
+++ b/core/metadata/src/stm/snapshot.rs
@@ -52,13 +52,17 @@ use crate::stm::user::UsersSnapshot;
/// position out of place. The bump is what turns that into
/// `UnsupportedFormatVersion` instead of an opaque deserializer error, or worse a
/// silent misread.
-pub const SNAPSHOT_FORMAT_VERSION: u32 = 4;
+///
+/// Version 5: `PartitionSnapshot` gained `created_view` as a trailing, defaulted
+/// field.
+pub const SNAPSHOT_FORMAT_VERSION: u32 = 5;
/// Oldest format version [`MetadataSnapshot::decode`] still reads.
///
-/// Version 4 added the client table's dedup fences as a trailing, defaulted
-/// field, so a version 3 checkpoint decodes under the current layout and simply
-/// carries none. Accepting it is what lets a node upgrade in place instead of
+/// Versions 4 and 5 each appended a trailing, defaulted field (the client
+/// table's dedup fences, then `PartitionSnapshot::created_view`), so a version 3
+/// or 4 checkpoint decodes under the current layout and simply carries the
+/// defaults. Accepting them is what lets a node upgrade in place instead of
/// refusing its own last checkpoint; anything older than 3 changed field
/// positions and is refused as before.
pub const MIN_READABLE_SNAPSHOT_FORMAT_VERSION: u32 = 3;
@@ -568,7 +572,7 @@ mod tests {
// operator's boot reading one field's bytes as another's. Changing either
// number is the reminder to change the other.
const FIELD_COUNT: u32 = 7;
- const PINNED_VERSION: u32 = 4;
+ const PINNED_VERSION: u32 = 5;
let encoded = MetadataSnapshot::new(0).encode().unwrap();
let mut cursor = encoded.as_slice();
@@ -597,6 +601,8 @@ mod tests {
// keeps version 3 readable, so a further append here needs the same
// treatment or a bump.
const CLIENT_TABLE_FIELD_COUNT: u32 = 2;
+ // Version 5 appended `created_view` the same way.
+ const PARTITION_FIELD_COUNT: u32 = 7;
let client_table = consensus::ClientTableSnapshot {
slots: Vec::new(),
@@ -634,6 +640,22 @@ mod tests {
"TopicSnapshot's field count changed; bump SNAPSHOT_FORMAT_VERSION with it"
);
+ let partition = PartitionSnapshot {
+ id: 0,
+ consensus_group_id: 0,
+ created_at: IggyTimestamp::default(),
+ created_revision: 0,
+ deleted_up_to_offset: 0,
+ purge_generation: 0,
+ created_view: 0,
+ };
+ let encoded = rmp_serde::to_vec(&partition).unwrap();
+ assert_eq!(
+ rmp::decode::read_array_len(&mut encoded.as_slice()).unwrap(),
+ PARTITION_FIELD_COUNT,
+ "PartitionSnapshot's field count changed; default the new field or bump the version"
+ );
+
let stream = StreamSnapshot {
id: 0,
name: String::new(),
@@ -861,6 +883,7 @@ mod tests {
created_revision: 0,
deleted_up_to_offset: 0,
purge_generation: 0,
+ created_view: 0,
}],
consumer_groups: Vec::new(),
// Nonzero and distinct from every id above so the
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index 5e6d7a4d9..119a9f476 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -89,6 +89,10 @@ pub struct PartitionSnapshot {
/// `#[serde(default)]` so pre-purge snapshots restore at 0.
#[serde(default)]
pub purge_generation: u64,
+ /// `#[serde(default)]` so snapshots predating this field restore at 0, the
+ /// view every group started in before it existed.
+ #[serde(default)]
+ pub created_view: u32,
}
#[derive(Debug, Clone)]
@@ -101,6 +105,15 @@ pub struct Partition {
/// reused the slab key, so the local partition is stale and must be torn
/// down before rebuild.
pub created_revision: u64,
+ /// View the admitting primary minted the create in, read off the request
+ /// body (never the prepare header, whose `view` a post-view-change
+ /// retransmit restamps per delivery). Every replica seeds the partition's
+ /// consensus group from it, so they all name the same primary: the metadata
+ /// primary at creation, which is what the roster advertises. Never above
+ /// the group's current view, since the group starts here and views only
+ /// climb, so a replica materialising late joins at or below its peers, the
+ /// floor a view-0 start used to be.
+ pub created_view: u32,
/// Replicated delete watermark: the reconciler on every replica removes
/// sealed segments with `end_offset` below this. Advanced monotonically by
/// `TruncatePartition` (the resolved form of a client `DeleteSegments`).
@@ -123,12 +136,14 @@ impl Partition {
consensus_group_id: u64,
created_at: IggyTimestamp,
created_revision: u64,
+ created_view: u32,
) -> Self {
Self {
id,
consensus_group_id,
created_at,
created_revision,
+ created_view,
deleted_up_to_offset: 0,
purge_generation: 0,
}
@@ -1363,6 +1378,22 @@ impl Streams {
/// nothing in the type enforces that density.
#[must_use]
pub fn created_revision_for_namespace(&self, namespace: IggyNamespace) -> Option<u64> {
+ self.with_committed_partition(namespace, |partition| partition.created_revision)
+ }
+
+ /// Committed [`Partition::created_view`] for the exact partition the
+ /// `namespace` tuple denotes; resolved like
+ /// [`Self::created_revision_for_namespace`].
+ #[must_use]
+ pub fn created_view_for_namespace(&self, namespace: IggyNamespace) -> Option<u32> {
+ self.with_committed_partition(namespace, |partition| partition.created_view)
+ }
+
+ fn with_committed_partition<T>(
+ &self,
+ namespace: IggyNamespace,
+ read: impl FnOnce(&Partition) -> T,
+ ) -> Option<T> {
self.inner.read(|inner| {
let stream = inner.items.get(namespace.stream_id())?;
let topic = stream.topics.get(namespace.topic_id())?;
@@ -1370,13 +1401,13 @@ impl Streams {
if let Some(partition) = topic.partitions.get(partition_id)
&& partition.id == partition_id
{
- return Some(partition.created_revision);
+ return Some(read(partition));
}
topic
.partitions
.iter()
.find(|partition| partition.id == partition_id)
- .map(|partition| partition.created_revision)
+ .map(read)
})
}
@@ -1408,6 +1439,7 @@ impl Streams {
stream_slab: usize,
topic_slab: usize,
target_partitions: &[CreatedPartitionAssignment],
+ created_view: u32,
) {
let stream_wire =
WireIdentifier::numeric(u32::try_from(stream_slab).expect("sim stream slab fits u32"));
@@ -1439,6 +1471,7 @@ impl Streams {
},
derived_options: WireOptions::empty(),
partitions,
+ created_view,
},
IggyTimestamp::from(1),
))
@@ -1458,12 +1491,20 @@ impl Streams {
/// the missing partition. Mirrors [`Users::ensure_root_user`](crate::stm::user::Users::ensure_root_user): a seed
/// helper that bypasses consensus, never a production runtime path.
///
+ /// `created_view` stands in for the view a real create's admitting primary
+ /// mints into the request body; see [`Partition::created_view`].
+ ///
/// # Panics
/// Panics if the apply is attempted on a reader handle rather than the
/// writer (never for the simulator's writer-backed STM), or if a slab id
/// exceeds the `u32` wire identifier space.
#[cfg(any(test, feature = "simulator"))]
- pub fn seed_namespace(&self, namespace: IggyNamespace, consensus_group_id: u64) {
+ pub fn seed_namespace(
+ &self,
+ namespace: IggyNamespace,
+ consensus_group_id: u64,
+ created_view: u32,
+ ) {
let stream_slab = namespace.stream_id();
let topic_slab = namespace.topic_id();
let partition_id =
@@ -1496,7 +1537,7 @@ impl Streams {
})
.collect();
self.seed_stream_slabs(stream_slab);
- self.seed_topic_slabs(stream_slab, topic_slab, &target_partitions);
+ self.seed_topic_slabs(stream_slab, topic_slab, &target_partitions, created_view);
if self.created_revision_for_namespace(namespace).is_some() {
return;
@@ -1529,6 +1570,7 @@ impl Streams {
.expect("sim partition count fits u32"),
},
partitions,
+ created_view,
},
IggyTimestamp::from(1),
))
@@ -1850,6 +1892,7 @@ impl StateHandler for CreateTopicWithAssignmentsRequest {
consensus_group_id: partition.consensus_group_id,
created_at: timestamp,
created_revision: new_revision,
+ created_view: self.created_view,
deleted_up_to_offset: 0,
purge_generation: 0,
};
@@ -2140,6 +2183,7 @@ impl StateHandler for CreatePartitionsWithAssignmentsRequest {
consensus_group_id: partition.consensus_group_id,
created_at: timestamp,
created_revision: new_revision,
+ created_view: self.created_view,
deleted_up_to_offset: 0,
purge_generation: 0,
});
@@ -2247,6 +2291,7 @@ impl Snapshotable for Streams {
consensus_group_id: p.consensus_group_id,
created_at: p.created_at,
created_revision: p.created_revision,
+ created_view: p.created_view,
deleted_up_to_offset: p.deleted_up_to_offset,
purge_generation: p.purge_generation,
})
@@ -2365,6 +2410,7 @@ impl StreamsInner {
consensus_group_id: p.consensus_group_id,
created_at: p.created_at,
created_revision: p.created_revision,
+ created_view: p.created_view,
deleted_up_to_offset: p.deleted_up_to_offset,
purge_generation: p.purge_generation,
})
@@ -2489,6 +2535,7 @@ mod tests {
..TopicCreateOptions::default()
};
let request = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: WireCreateTopicRequest {
stream_id: WireIdentifier::numeric(0),
partitions_count: 1,
@@ -2523,6 +2570,51 @@ mod tests {
);
}
+ /// The seed every replica materialises a group from is the body-carried
+ /// `created_view`, minted once by the admitting primary. A header-derived
+ /// view would differ per replica after a post-view-change retransmit.
+ #[test]
+ fn create_ops_record_the_body_carried_view_on_the_partition() {
+ let mut inner = StreamsInner::new();
+ create_stream(&mut inner, "s");
+
+ let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 7,
+ request: WireCreateTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ partitions_count: 1,
+ name: WireName::new("t").unwrap(),
+ options: WireOptions::empty(),
+ },
+ derived_options: WireOptions::empty(),
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 1,
+ }],
+ };
+ let reply = StateHandler::apply(&create_topic, &mut inner, IggyTimestamp::from(1));
+ assert_eq!(reply.code, 0, "create topic must succeed");
+
+ let create_partitions = CreatePartitionsWithAssignmentsRequest {
+ created_view: 9,
+ request: CreatePartitionsRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ partitions_count: 1,
+ },
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 2,
+ }],
+ };
+ let reply = StateHandler::apply(&create_partitions, &mut inner, IggyTimestamp::from(2));
+ assert_eq!(reply.code, 0, "create partitions must succeed");
+
+ let topic = inner.items.get(0).unwrap().topics.get(0).unwrap();
+ assert_eq!(topic.partitions[0].created_view, 7);
+ assert_eq!(topic.partitions[1].created_view, 9);
+ }
+
/// A client may send the literal 0 that means "resolve the default". The
/// merge lets that explicit entry win over the derived one, so the map has
/// to be rewritten from the resolved value: otherwise the persisted map
@@ -2558,6 +2650,7 @@ mod tests {
&mut sentinel,
);
let request = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: WireCreateTopicRequest {
stream_id: WireIdentifier::numeric(0),
partitions_count: 1,
@@ -2656,6 +2749,7 @@ mod tests {
for stream_id in 0..2u32 {
for topic_name in ["logs", "events"] {
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(stream_id, 2, topic_name),
derived_options: WireOptions::empty(),
partitions: vec![
@@ -2724,6 +2818,7 @@ mod tests {
let mut inner = StreamsInner::new();
create_stream(&mut inner, "stream");
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 2, "topic"),
derived_options: WireOptions::empty(),
partitions: vec![
@@ -2752,6 +2847,7 @@ mod tests {
let mut inner = StreamsInner::new();
create_stream(&mut inner, "stream");
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 2, "topic"),
derived_options: WireOptions::empty(),
partitions: vec![
@@ -2768,6 +2864,7 @@ mod tests {
let _ = StateHandler::apply(&create_topic, &mut inner, IggyTimestamp::now());
let create_partitions = CreatePartitionsWithAssignmentsRequest {
+ created_view: 0,
request: WireCreatePartitionsRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
@@ -2807,6 +2904,7 @@ mod tests {
let mut inner = StreamsInner::new();
create_stream(&mut inner, "stream");
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 2, "topic"),
derived_options: WireOptions::empty(),
partitions: vec![
@@ -2841,6 +2939,7 @@ mod tests {
let mut inner = StreamsInner::new();
create_stream(&mut inner, "stream");
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 2, "topic"),
derived_options: WireOptions::empty(),
partitions: vec![
@@ -2857,6 +2956,7 @@ mod tests {
let _ = StateHandler::apply(&create_topic, &mut inner, IggyTimestamp::now());
let create_partitions = CreatePartitionsWithAssignmentsRequest {
+ created_view: 0,
request: WireCreatePartitionsRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
@@ -2894,6 +2994,7 @@ mod tests {
create_stream(&mut inner, "stream");
// Topic missing => validation failure path
let create_partitions = CreatePartitionsWithAssignmentsRequest {
+ created_view: 0,
request: WireCreatePartitionsRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(99),
@@ -2936,6 +3037,7 @@ mod tests {
let mut inner = StreamsInner::new();
create_stream(&mut inner, "stream");
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, partitions_count, "topic"),
derived_options: WireOptions::empty(),
partitions: (0..partitions_count)
@@ -2991,6 +3093,7 @@ mod tests {
let mut inner = StreamsInner::new();
create_stream(&mut inner, "stream");
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 1, "topic"),
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
@@ -3090,6 +3193,7 @@ mod tests {
fn given_counted_partitions_when_apply_purge_stream_should_zero_every_topic() {
let mut inner = inner_with_registered_partition();
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 1, "metrics"),
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
@@ -3188,6 +3292,7 @@ mod tests {
let mut inner = StreamsInner::new();
create_stream(&mut inner, "alpha");
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 1, "logs"),
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
@@ -3302,6 +3407,7 @@ mod tests {
let mut inner = StreamsInner::new();
create_stream(&mut inner, "alpha");
let create_topic = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 1, "logs"),
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
@@ -3396,6 +3502,7 @@ mod tests {
for index in 0..MAX_TOPICS {
let request = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 0, &format!("t{index}")),
derived_options: WireOptions::empty(),
partitions: Vec::new(),
@@ -3405,6 +3512,7 @@ mod tests {
}
let overflow = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 0, "one-too-many"),
derived_options: WireOptions::empty(),
partitions: Vec::new(),
@@ -3434,6 +3542,7 @@ mod tests {
create_stream(&mut inner, "s");
let seed = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 1, "t"),
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
@@ -3448,6 +3557,7 @@ mod tests {
// Resolves to MAX_PARTITIONS - 1: the last legal id.
let last = CreatePartitionsWithAssignmentsRequest {
+ created_view: 0,
request: WireCreatePartitionsRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
@@ -3465,6 +3575,7 @@ mod tests {
);
let overflow = CreatePartitionsWithAssignmentsRequest {
+ created_view: 0,
request: WireCreatePartitionsRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
@@ -3496,6 +3607,7 @@ mod tests {
create_stream(&mut inner, "s");
let request = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 1, "t"),
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
@@ -3567,6 +3679,7 @@ mod tests {
create_stream(&mut inner, "s");
for name in ["a", "b", "c"] {
let request = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 0, name),
derived_options: WireOptions::empty(),
partitions: Vec::new(),
@@ -3625,6 +3738,7 @@ mod tests {
create_stream(&mut inner, "s");
let request = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 2, "t"),
derived_options: WireOptions::empty(),
partitions: vec![
@@ -3659,6 +3773,7 @@ mod tests {
create_stream(&mut inner, "s");
let seed = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 1, "t"),
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
@@ -3672,6 +3787,7 @@ mod tests {
);
let duplicate = CreatePartitionsWithAssignmentsRequest {
+ created_view: 0,
request: WireCreatePartitionsRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
@@ -3710,6 +3826,7 @@ mod tests {
create_stream(&mut inner, "s");
let seed = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
request: make_topic_request(0, 1, "t"),
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
@@ -3723,6 +3840,7 @@ mod tests {
);
let distinct = CreatePartitionsWithAssignmentsRequest {
+ created_view: 0,
request: WireCreatePartitionsRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 3c15d647b..9a58b8b78 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -1052,10 +1052,10 @@ async fn shard_main(
|mux_stm| {
ensure_default_root_user(mux_stm);
},
- |mux_stm, client, timestamp| {
+ |mux_stm, client, stamp| {
mux_stm
.streams()
- .remove_consumer_group_member(client, timestamp);
+ .remove_consumer_group_member(client, stamp);
},
)
.await
@@ -1297,7 +1297,6 @@ async fn shard_main(
topology.cluster_id,
topology.self_replica_id,
topology.replica_count,
- Arc::clone(&metadata_view),
));
let reconcile_periodic = config
.system
@@ -2081,10 +2080,7 @@ async fn build_shard_for_thread(
topology.cluster_id,
topology.self_replica_id,
topology.replica_count,
- // Quarantine-and-rebuild always finds a partition
- // directory already there, so this joins as a probing
- // backup and learns the live view; nothing to seed.
- None,
+ partition_metadata.created_view,
Rc::clone(&bus),
)
.await?
diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs
index 44cfd93e2..0a494f524 100644
--- a/core/server/src/dispatch.rs
+++ b/core/server/src/dispatch.rs
@@ -4330,6 +4330,7 @@ mod tests {
partition_id: 0,
consensus_group_id: 1,
}],
+ created_view: 0,
};
md.mux_stm
.update(prepare_message(
diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs
index f548f5de7..a6f709880 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -518,11 +518,11 @@ pub(crate) async fn open_partition_superblock(
/// The namespace arrives packed, so its components are in range by
/// construction. Metadata admission is what bounds them.
///
-/// `view_seed` is the view a group with no durable record of its own starts
-/// in, and it is the metadata plane's current view: see the `seed_view`
-/// comment below for why a group left at view 0 is unreachable. `None` keeps
-/// the historical view-0 start, and is also what a restart materialization
-/// gets, since it probes for the live view instead.
+/// `created_view` is the view a group with no durable record of its own starts
+/// in: the metadata plane's view when it committed the create, recorded on
+/// the committed partition so every replica seeds the same value. See the
+/// `seed_view` comment below for why a group left at view 0 is unreachable. A
+/// restart materialization ignores it and probes for the live view instead.
///
/// The returned partition's `offset` / `dirty_offset` are `0` and
/// `should_increment_offset` is `false`, mirroring a clean append starting
@@ -542,7 +542,7 @@ pub async fn build_partition_fresh(
cluster_id: u128,
self_replica_id: u8,
replica_count: u8,
- view_seed: Option<u32>,
+ created_view: u32,
bus: Rc<IggyMessageBus>,
) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerError> {
let stream_id = namespace.stream_id();
@@ -604,7 +604,8 @@ pub async fn build_partition_fresh(
.map(|state| (state.view, state.log_view));
// Shared with the simulator's `init_partition`, which cannot call this
// builder; see `fresh_group_start`.
- let FreshGroupStart { join, seed_view } = fresh_group_start(restarted, durable_view, view_seed);
+ let FreshGroupStart { join, seed_view } =
+ fresh_group_start(restarted, durable_view, created_view);
// Request queue holds 2x the prepare depth (buffered requests drain as
// prepares commit); depth is the per-partition `[partition]` knob.
let prepare_queue_depth = config.partition.prepare_queue_depth;
@@ -626,16 +627,16 @@ pub async fn build_partition_fresh(
// than the roster advertises as leader, and nothing routes a
// partition write across that gap: the client is sent to the
// metadata leader and refused there for the whole budget. Seeding
- // from the metadata view keeps the two congruent for a group born
- // after a metadata election.
+ // from the view the create was admitted in keeps the two congruent
+ // for a group born after a metadata election.
//
- // Replicas can still disagree on the seed: each publishes its own
- // metadata view on a 100ms poll, so one may read V while another
- // has yet to see the election. That resolves the way any view
- // disagreement does, the higher view winning through `StartView` -
- // but only because the seed sets `log_view` too. The caller is
- // what keeps a replica from seeding a view it has no opinion on;
- // see `ReconcilerCtx::partition_view_seed`.
+ // The seed comes off the committed partition, not this replica's
+ // live metadata view: the two differ once the metadata plane
+ // elects again, and a seed above the group's real view is an
+ // empty log that outranks committed history in the next DVC
+ // merge. The value rides the create's request body (a header view
+ // is restamped per delivery), so every replica commits the same
+ // one and a late materialiser lands at or below its peers.
seed_view,
incarnation: None,
join,
diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs
index e9114dabd..dea36c9c0 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -170,7 +170,6 @@
//! -- `PrepareHeader.reserved` has room, but it is a `#[repr(C)]` wire change.
use crate::bootstrap::ServerShard;
-use crate::cluster_meta::METADATA_VIEW_UNKNOWN;
use crate::partition_helpers::{build_partition_fresh, delete_partitions_from_disk};
use ahash::{AHashMap, AHashSet};
use configs::server::ServerConfig;
@@ -188,7 +187,6 @@ use shard::{Receiver, Sender};
use std::cell::{Cell, RefCell};
use std::rc::Rc;
use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use tracing::{debug, error, trace, warn};
@@ -227,11 +225,6 @@ pub struct ReconcilerCtx {
pub cluster_id: u128,
pub self_replica_id: u8,
pub replica_count: u8,
- /// The metadata plane's view, published by shard 0 and shared with every
- /// shard's roster. Read when materialising a partition so its consensus
- /// group starts in the same view the roster's advertised leader comes
- /// from; the unknown-view sentinel until the first publish.
- pub metadata_view: Arc<AtomicU64>,
failure_state: RefCell<AHashMap<(IggyNamespace, FailureCause), FailureRecord>>,
/// `Streams::revision` observed at the end of the last pass that fully
/// converged. Paired with `last_pass_noop` for the fast-skip in
@@ -251,7 +244,6 @@ impl ReconcilerCtx {
cluster_id: u128,
self_replica_id: u8,
replica_count: u8,
- metadata_view: Arc<AtomicU64>,
) -> Self {
Self {
shard,
@@ -260,48 +252,12 @@ impl ReconcilerCtx {
cluster_id,
self_replica_id,
replica_count,
- metadata_view,
failure_state: RefCell::new(AHashMap::new()),
last_revision: Cell::new(None),
last_pass_noop: Cell::new(false),
}
}
- /// The view a partition group materialised now should start in.
- ///
- /// `None` means this replica has no opinion yet, NOT "start at view 0":
- /// shard 0 publishes the unknown-view sentinel before its first tick and
- /// for as long as it has ceded a recovered view's primaryship. Treating
- /// that as 0 is how a single replica of a group ends up view-0 while its
- /// peers are at V, which is the split this seed exists to close. Callers
- /// on a replicated group defer materialising instead; see
- /// [`Self::partition_view_seed_ready`].
- ///
- /// # Panics
- /// If the published view does not fit a `u32`. The publisher writes
- /// `u64::from(consensus.view())` or the sentinel handled above, so a value
- /// past `u32::MAX` is memory corruption, not a state a retry can clear.
- /// Silently falling back to `None` here would restore the view-0 start
- /// this change exists to remove, on the one path nobody would look at.
- fn partition_view_seed(&self) -> Option<u32> {
- let view = self.metadata_view.load(Ordering::Relaxed);
- if view == METADATA_VIEW_UNKNOWN {
- return None;
- }
- Some(u32::try_from(view).expect("published metadata view must fit a u32"))
- }
-
- /// Whether a group may be materialised now.
- ///
- /// A solo replica is always ready: with one replica every view names it
- /// primary, so there is no peer to disagree with and no split to cause.
- /// A replicated group waits until this replica knows the metadata view,
- /// so it cannot seed a view-0 group underneath peers that already moved.
- /// The wait is one reconcile pass; shard 0 republishes every 100ms.
- fn partition_view_seed_ready(&self) -> bool {
- self.replica_count <= 1 || self.partition_view_seed().is_some()
- }
-
fn is_backed_off(&self, ns: IggyNamespace, cause: FailureCause, now: Instant) -> bool {
let state = self.failure_state.borrow();
if state.is_empty() {
@@ -461,12 +417,6 @@ struct PassCounters {
/// has not applied yet. Counted so the pass does not arm the fast-skip
/// while work is in flight; applying it bumps no revision.
already_staged: usize,
- /// Materialisations held back because this replica has no metadata view to
- /// seed from yet. Counted so the pass does not arm the fast-skip: shard 0
- /// publishing a view bumps no `Streams::revision` and does not wake the
- /// reconciler, so an armed skip would strand every deferred group until
- /// some unrelated commit happened to bump the revision.
- view_unpublished: usize,
}
impl PassCounters {
@@ -483,7 +433,6 @@ impl PassCounters {
+ self.deferred
+ self.parked_reclaimed
+ self.already_staged
- + self.view_unpublished
}
}
@@ -529,7 +478,7 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
}
let target = snapshot_target_namespaces(ctx);
- let target_set: AHashSet<IggyNamespace> = target.iter().map(|(ns, _)| *ns).collect();
+ let target_set: AHashSet<IggyNamespace> = target.iter().map(|partition| partition.ns).collect();
let mut counters = PassCounters::default();
reconcile_additions(ctx, target, &mut counters).await;
@@ -581,14 +530,19 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
#[allow(clippy::too_many_lines)]
async fn reconcile_additions(
ctx: &ReconcilerCtx,
- target: Vec<(IggyNamespace, u64)>,
+ target: Vec<TargetPartition>,
counters: &mut PassCounters,
) {
let shard_id = ctx.shard.id;
let partitions = ctx.shard.plane.partitions();
let total_shards = u32::from(ctx.total_shards);
- for (ns, epoch) in target {
+ for TargetPartition {
+ ns,
+ epoch,
+ created_view,
+ } in target
+ {
if partitions.contains(&ns) {
// Tombstoned but still in the map. Two cases, told apart by
// whether teardown's disk delete succeeded:
@@ -710,22 +664,6 @@ async fn reconcile_additions(
continue;
}
- // Materialising before this replica knows the metadata view would seed
- // the group at view 0 under peers that already elected past it. Defer
- // the whole pass rather than the namespace: every group on this shard
- // reads the same view, so none of them can be seeded correctly yet.
- if !ctx.partition_view_seed_ready() {
- debug!(
- shard = shard_id,
- "metadata view not published yet; deferring partition materialisation"
- );
- // Every remaining namespace reads the same view, so none of them
- // can be seeded either. Counted before breaking so the pass does
- // not read as converged, which would fast-skip the retry.
- counters.view_unpublished += 1;
- break;
- }
-
// Resolve the shared stats `Arc` only for namespaces actually
// built, not once per committed partition every pass. A topic that
// vanished between the target snapshot and this read defers to the
@@ -743,7 +681,7 @@ async fn reconcile_additions(
ctx.cluster_id,
ctx.self_replica_id,
ctx.replica_count,
- ctx.partition_view_seed(),
+ created_view,
Rc::clone(&ctx.shard.bus),
)
.await
@@ -1124,11 +1062,21 @@ fn snapshot_topic_live_groups(ctx: &ReconcilerCtx) -> AHashMap<(usize, usize), A
})
}
-/// Committed `(namespace, created_revision)` pairs. The epoch lets the
-/// additions pass detect a stale local incarnation after slab-key reuse
-/// without an `Arc<TopicStats>` clone per partition; stats are fetched
-/// lazily in [`fetch_partition_stats`] only for namespaces actually built.
-fn snapshot_target_namespaces(ctx: &ReconcilerCtx) -> Vec<(IggyNamespace, u64)> {
+/// One committed partition the additions pass has to account for.
+struct TargetPartition {
+ ns: IggyNamespace,
+ /// The committed `created_revision`. Lets the pass detect a stale local
+ /// incarnation after slab-key reuse without an `Arc<TopicStats>` clone per
+ /// partition; stats are fetched lazily in [`fetch_partition_stats`] only
+ /// for namespaces actually built.
+ epoch: u64,
+ /// The view a fresh materialisation seeds its consensus group with; see
+ /// `build_partition_fresh`.
+ created_view: u32,
+}
+
+/// Every committed partition, as the additions pass needs it.
+fn snapshot_target_namespaces(ctx: &ReconcilerCtx) -> Vec<TargetPartition> {
ctx.shard.plane.metadata().mux_stm.streams().read(|inner| {
// TODO(krishna): O(committed partitions) per non-skipped pass (here +
// reconcile_removals). The revision fast-skip hides this in steady
@@ -1138,8 +1086,11 @@ fn snapshot_target_namespaces(ctx: &ReconcilerCtx) -> Vec<(IggyNamespace, u64)>
for (_, stream) in &inner.items {
for (topic_id, topic) in &stream.topics {
for partition in &topic.partitions {
- let ns = IggyNamespace::new(stream.id, topic_id, partition.id);
- entries.push((ns, partition.created_revision));
+ entries.push(TargetPartition {
+ ns: IggyNamespace::new(stream.id, topic_id, partition.id),
+ epoch: partition.created_revision,
+ created_view: partition.created_view,
+ });
}
}
}
@@ -1279,8 +1230,8 @@ pub fn install_tick_handler(shard: &Rc<ServerShard>, wake_tx: WakeTx) {
#[cfg(test)]
mod tests {
use super::{
- AtomicU64, FailureCause, FailureRecord, METADATA_VIEW_UNKNOWN, ReconcilerCtx,
- build_partition_fresh, delete_partitions_from_disk, fetch_partition_stats, reconcile_once,
+ FailureCause, FailureRecord, ReconcilerCtx, build_partition_fresh,
+ delete_partitions_from_disk, fetch_partition_stats, reconcile_once,
};
use configs::server::{ServerConfig, ServerSystemConfig};
use consensus::{MetadataHandle, PartitionsHandle};
@@ -1303,6 +1254,7 @@ mod tests {
use metadata::IggyMetadata;
use metadata::MuxStateMachine;
use metadata::impls::metadata::IggySnapshot;
+ use metadata::impls::metadata::StreamsFrontend;
use metadata::stm::StateMachine;
use metadata::stm::stream::Streams;
use metadata::stm::user::Users;
@@ -1514,6 +1466,7 @@ mod tests {
},
derived_options: WireOptions::empty(),
partitions: assignments,
+ created_view: 0,
};
mux.update(build_prepare(
op,
@@ -1732,32 +1685,19 @@ mod tests {
CLUSTER_ID,
0,
1,
- Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)),
))
}
- /// [`make_ctx`] for a REPLICATED group, sharing `metadata_view` with the
- /// caller so a test can publish a view mid-run.
- ///
- /// Replica count is what decides whether materialisation waits on a
- /// published view: a solo replica is primary in every view, so there is no
- /// peer to disagree with and nothing to wait for. Every other test here
- /// runs solo and takes that short circuit.
- fn make_cluster_ctx(
- shard: Rc<TestShard>,
- total_shards: u16,
- config: Rc<ServerConfig>,
- metadata_view: Arc<AtomicU64>,
- ) -> Rc<ReconcilerCtx> {
- Rc::new(ReconcilerCtx::new(
- shard,
- total_shards,
- config,
- CLUSTER_ID,
- 0,
- 3,
- metadata_view,
- ))
+ /// The committed `created_view` of `ns`, what a reconcile pass would seed
+ /// a fresh build with.
+ fn created_view(ctx: &ReconcilerCtx, ns: IggyNamespace) -> u32 {
+ ctx.shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .created_view_for_namespace(ns)
+ .expect("committed namespace records its creation view")
}
/// Tests run reconcile + pump-side apply inline since no real pump exists.
@@ -1902,7 +1842,7 @@ mod tests {
CLUSTER_ID,
0,
1,
- None,
+ created_view(&ctx, ns),
Rc::clone(&ctx.shard.bus),
)
.await
@@ -1932,7 +1872,7 @@ mod tests {
CLUSTER_ID,
0,
1,
- None,
+ created_view(&ctx, ns),
Rc::clone(&ctx.shard.bus),
)
.await
@@ -2056,73 +1996,6 @@ mod tests {
/// `CreatePartitions` on an existing topic adds new namespaces; the
/// reconciler picks them up on the next pass without touching the
/// partitions it already materialised.
- /// A replicated group is NOT materialised while this replica has no
- /// metadata view to seed from, and IS once one is published.
- ///
- /// Seeding is a local read: shard 0 publishes the unknown sentinel before
- /// its first tick and for as long as it has ceded a recovered view. Taking
- /// that for view 0 would start the group naming replica 0 underneath peers
- /// that already elected past it, which is the split the seed exists to
- /// close, reintroduced one replica at a time. Waiting costs one reconcile
- /// pass; the publisher reposts every 100ms.
- #[compio::test]
- async fn given_no_published_metadata_view_when_reconciling_should_defer_materialisation() {
- let tmp = TempDir::new().expect("tempdir for system path");
- let config = test_config(&tmp);
- let mux = TestMux::default();
- seed_stream(&mux, 1, "stream-deferred");
- seed_topic(&mux, 2, 0, "topic-deferred", vec![assignment(0, 1)]);
-
- let shard = build_test_shard(0, &config, mux);
- let metadata_view = Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN));
- let ctx = make_cluster_ctx(
- Rc::clone(&shard),
- 1,
- Rc::new(config),
- Arc::clone(&metadata_view),
- );
-
- reconcile_pass(&ctx).await;
- assert_eq!(
- shard.plane.partitions().len(),
- 0,
- "a replicated group must not materialise before this replica knows the metadata \
- view: seeded at 0 it names replica 0 whatever the metadata plane elected"
- );
-
- // The publisher posts a real view; the deferred pass now converges.
- metadata_view.store(1, Ordering::Relaxed);
- reconcile_pass(&ctx).await;
- assert_eq!(
- shard.plane.partitions().len(),
- 1,
- "once a view is published the deferred group must materialise on the next pass"
- );
- }
-
- /// A SOLO replica never waits. It is primary in every view, so there is no
- /// peer for it to disagree with, and blocking on a publisher that a
- /// single-node deployment may never run would wedge materialisation
- /// outright.
- #[compio::test]
- async fn given_a_solo_replica_when_no_view_is_published_should_still_materialise() {
- let tmp = TempDir::new().expect("tempdir for system path");
- let config = test_config(&tmp);
- let mux = TestMux::default();
- seed_stream(&mux, 1, "stream-solo");
- seed_topic(&mux, 2, 0, "topic-solo", vec![assignment(0, 1)]);
-
- let shard = build_test_shard(0, &config, mux);
- let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
-
- reconcile_pass(&ctx).await;
- assert_eq!(
- shard.plane.partitions().len(),
- 1,
- "a solo replica must materialise without waiting on a published metadata view"
- );
- }
-
#[compio::test]
async fn reconcile_picks_up_create_partitions_increments() {
let tmp = TempDir::new().expect("tempdir for system path");
@@ -2161,6 +2034,7 @@ mod tests {
partitions_count: 2,
},
partitions: vec![assignment(0, 3), assignment(1, 4)],
+ created_view: 0,
},
))
.expect("CreatePartitions apply succeeds");
@@ -3353,6 +3227,7 @@ mod tests {
partitions_count: 1,
},
partitions: vec![assignment(0, 3)],
+ created_view: 0,
},
))
.expect("CreatePartitions apply succeeds");
diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs
index 21b8e6497..8165bdf63 100644
--- a/core/server/src/responses.rs
+++ b/core/server/src/responses.rs
@@ -1869,7 +1869,7 @@ mod tests {
use metadata::stm::stream::{Partition, StreamsInner};
let streams = StreamsInner::new();
- let partition = Partition::new(0, 1, IggyTimestamp::from(1u64), 0);
+ let partition = Partition::new(0, 1, IggyTimestamp::from(1u64), 0, 0);
// Registry miss: the owning shard has not started building.
let predicted = partition_response(&streams, 0, 0, &partition).expect("response builds");
@@ -1914,8 +1914,8 @@ mod tests {
options: iggy_common::ResourceOptions::default(),
stats: topic_stats.clone(),
partitions: vec![
- Partition::new(0, 1, created_at, 0),
- Partition::new(1, 1, created_at, 0),
+ Partition::new(0, 1, created_at, 0, 0),
+ Partition::new(1, 1, created_at, 0, 0),
],
round_robin_counter: Arc::new(AtomicUsize::new(0)),
consumer_groups: ahash::AHashMap::default(),
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 2decb3934..98674232c 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -3555,6 +3555,9 @@ where
/// materialisation and wrong for a restart: a rebuilt partition with no data
/// reports `commit_offset` 0, which reads as a regression rather than a
/// harness that discarded the log.
+ ///
+ /// `created_view` is the view the metadata plane created the namespace in;
+ /// see `fresh_group_start`.
// `feature = "simulator"` alone, unlike its neighbours: the body names items
// `partitions` gates the same way, and a `test` arm cannot turn those on.
// Under `cargo test -p shard` that arm fires from shard's own `cfg(test)`
@@ -3569,7 +3572,7 @@ where
recovered_state: Option<consensus::VsrState>,
retained: Option<partitions::RetainedPartitionState>,
restore_frontier: bool,
- metadata_view: Option<u32>,
+ created_view: u32,
) where
B: MessageBus + Clone,
{
@@ -3590,15 +3593,15 @@ where
// The SAME decision `build_partition_fresh` makes, not a copy of it.
// This path cannot call that builder (it does real filesystem work and
// this runs on in-memory storage), and while the two decided
- // separately the simulator exercised neither the metadata-view seed nor
- // the plane split it closes. `retained` is populated only by the
+ // separately the simulator exercised neither the creation-view seed
+ // nor the plane split it closes. `retained` is populated only by the
// restart path, which is this path's evidence of a prior life.
let durable_view = recovered_state
.as_ref()
.map(|state| (state.view, state.log_view));
let restarted = retained.is_some() && self.partition_consensus.replica_count > 1;
let consensus::FreshGroupStart { join, seed_view } =
- consensus::fresh_group_start(restarted, durable_view, metadata_view);
+ consensus::fresh_group_start(restarted, durable_view, created_view);
// Recorded view first, exactly as the two boot paths order it: restoring
// after `init` would advertise a view older than the recorded one.
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 4e006c573..7b722eb9b 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -170,6 +170,12 @@ pub struct Simulator {
/// a run with it on studies a system more durable than Iggy is. See
/// `IggyShard::init_partition`.
restore_partition_frontier: bool,
+ /// The view each namespace was created in, what production records on the
+ /// committed partition and every replica seeds its group from. Read from the
+ /// metadata plane once, at the first seed, and reused for every later seed
+ /// of that namespace, a restart's included, so no replica seeds a view its
+ /// peers did not.
+ partition_created_views: HashMap<IggyNamespace, u32>,
/// Replies a setup handshake pulled off the wire that were not its own.
///
/// `await_setup_reply` steps the whole simulator, so it sees every client's
@@ -472,6 +478,7 @@ impl Simulator {
entry_rng: Xoshiro256PlusPlus::seed_from_u64(SimSeeds::derive(seed).entry_shard),
evicted: Vec::new(),
restore_partition_frontier: false,
+ partition_created_views: HashMap::new(),
deferred_client_replies: Vec::new(),
seed,
shell,
@@ -494,11 +501,19 @@ impl Simulator {
// segment storage so that branch can be deleted.
#[allow(clippy::cast_possible_truncation)]
pub fn init_partition(&mut self, namespace: IggyNamespace) {
+ let Some(created_view) = self.partition_created_view(namespace) else {
+ return;
+ };
for (i, replica) in self.replicas.iter().enumerate() {
if self.crashed.contains(&(i as u8)) {
continue;
}
- materialise_partition(replica, namespace, self.restore_partition_frontier);
+ materialise_partition(
+ replica,
+ namespace,
+ self.restore_partition_frontier,
+ created_view,
+ );
}
}
@@ -513,7 +528,10 @@ impl Simulator {
/// dispatch-shell poll path.
///
#[allow(clippy::cast_possible_truncation)]
- pub fn seed_stream_topic_partition(&self, namespace: IggyNamespace) {
+ pub fn seed_stream_topic_partition(&mut self, namespace: IggyNamespace) {
+ let Some(created_view) = self.partition_created_view(namespace) else {
+ return;
+ };
for (i, replica) in self.replicas.iter().enumerate() {
if self.crashed.contains(&(i as u8)) {
continue;
@@ -525,8 +543,20 @@ impl Simulator {
.metadata()
.mux_stm
.streams()
- .seed_namespace(namespace, namespace.inner());
+ .seed_namespace(namespace, namespace.inner(), created_view);
+ }
+ }
+
+ /// The view `namespace` was created in: the metadata plane's view at its
+ /// first seed, as the prepare of a real create carries it, and the same value
+ /// for every seed after. `None` with no live metadata consensus to read.
+ fn partition_created_view(&mut self, namespace: IggyNamespace) -> Option<u32> {
+ if let Some(&created_view) = self.partition_created_views.get(&namespace) {
+ return Some(created_view);
}
+ let created_view = self.metadata_view()?;
+ self.partition_created_views.insert(namespace, created_view);
+ Some(created_view)
}
/// Log `client` in against the deterministic root user through the dispatch
@@ -1063,9 +1093,13 @@ impl Simulator {
let partition_logs = self.retain_partition_logs(idx, &partition_superblocks);
// SORTED: seed order decides slab ids and `HashMap` order is per-process, so
// an unsorted walk would stop replay being byte-identical. Also drives the
- // re-materialisation loop below, which must agree with it.
- let mut materialised: Vec<IggyNamespace> = partition_superblocks.keys().copied().collect();
- materialised.sort_unstable_by_key(IggyNamespace::inner);
+ // re-materialisation loop below, which must agree with it. Each carries the
+ // view it was created in, as the metadata a real boot replays would.
+ let mut seed_namespaces: Vec<(IggyNamespace, u32)> = partition_superblocks
+ .keys()
+ .map(|&namespace| (namespace, self.partition_created_views[&namespace]))
+ .collect();
+ seed_namespaces.sort_unstable_by_key(|(namespace, _)| namespace.inner());
// Durable VSR state from the retained superblock, before the rebuild, as
// production reads it in `restore_metadata_consensus`.
@@ -1116,7 +1150,7 @@ impl Simulator {
recovered_state,
metadata_incarnation,
(shard_idx == 0).then(|| replica_data_dir.clone()).flatten(),
- &materialised,
+ &seed_namespaces,
);
if shard_idx == 0 {
metadata_bundle =
@@ -1153,11 +1187,12 @@ impl Simulator {
// makes the carried-forward superblock load-bearing: the group recovers its
// recorded `(view, log_view)` instead of re-entering view 0. The metadata half
// of the seed already ran inside `new_shard`, ahead of the replay.
- for namespace in materialised {
+ for (namespace, created_view) in seed_namespaces {
materialise_partition(
&self.replicas[idx],
namespace,
self.restore_partition_frontier,
+ created_view,
);
}
@@ -1334,6 +1369,21 @@ impl Simulator {
})
}
+ /// The metadata plane's view, as the first live replica owning a metadata
+ /// consensus sees it.
+ fn metadata_view(&self) -> Option<u32> {
+ (0..self.replica_count)
+ .filter(|replica_idx| !self.crashed.contains(replica_idx))
+ .find_map(|replica_idx| {
+ self.replicas[usize::from(replica_idx)].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .map(consensus::VsrConsensus::view)
+ })
+ }
+
#[must_use]
pub(crate) fn primary_index(&self, namespace: IggyNamespace) -> Option<u8> {
(0..self.replica_count)
@@ -1357,14 +1407,19 @@ impl Simulator {
/// re-opens every partition directory it owns, so the sim must re-materialise too,
/// else the superblock a restart carries forward is never read back and the
/// recovered-view branch is dead code.
-fn materialise_partition(replica: &SimReplica, namespace: IggyNamespace, restore_frontier: bool) {
+fn materialise_partition(
+ replica: &SimReplica,
+ namespace: IggyNamespace,
+ restore_frontier: bool,
+ created_view: u32,
+) {
let shard_count = u32::try_from(replica.shards.len()).expect("shard count fits u32");
let owner = calculate_shard_assignment(&namespace, shard_count);
// Commit the namespace first: a partition the metadata plane never heard of is
// a shape production cannot produce, and the shard refuses client traffic whose
// routing-row epoch it cannot match against a committed `created_revision`.
let streams = replica.shards[0].plane.metadata().mux_stm.streams();
- streams.seed_namespace(namespace, namespace.inner());
+ streams.seed_namespace(namespace, namespace.inner(), created_view);
// No committed revision means the seed could not re-add the namespace, which
// happens once a metadata workload has deleted its stream or topic: the seed's
// `CreatePartitions` is then a committed REJECTION rather than an error, so it
@@ -1375,6 +1430,12 @@ fn materialise_partition(replica: &SimReplica, namespace: IggyNamespace, restore
let Some(epoch) = streams.created_revision_for_namespace(namespace) else {
return;
};
+ // Read back rather than trusted from the argument: the seed above is a no-op
+ // for a namespace a metadata op already committed, and production seeds from
+ // the committed partition, never from a live view.
+ let created_view = streams
+ .created_view_for_namespace(namespace)
+ .expect("a committed partition records its creation view");
// One store per group, minted on first materialisation and reused after, so the
// recorded view survives a replica restart.
let superblock = Rc::clone(
@@ -1392,24 +1453,13 @@ fn materialise_partition(replica: &SimReplica, namespace: IggyNamespace, restore
// a second materialisation with no restart between would otherwise resurrect a
// log the live partition has moved past.
let retained = replica.partition_logs.borrow_mut().remove(&namespace);
- // The view this replica's metadata plane is in, which is what a fresh
- // group seeds from. Production reads it off the roster value shard 0
- // publishes; here shard 0's consensus is right there. Without it the
- // simulator materialises every group at view 0 and can never produce the
- // plane split that costs production its writes.
- let metadata_view = replica.shards[0]
- .plane
- .metadata()
- .consensus
- .as_ref()
- .map(consensus::VsrConsensus::view);
replica.shards[usize::from(owner)].init_partition(
namespace,
Some(superblock),
recovered_state,
retained,
restore_frontier,
- metadata_view,
+ created_view,
);
for shard in &replica.shards {
shard.shards_table().insert(
@@ -1576,8 +1626,9 @@ mod tests {
primary back at replica 0 both planes agree and the split cannot show"
);
- // Materialise a brand-new group. Every live replica seeds from its own
- // metadata view, which is the value production reads off the roster.
+ // Materialise a brand-new group. Every live replica seeds from the view
+ // the create was committed in, which production reads off the committed
+ // partition.
let namespace = IggyNamespace::new(1, 1, 0);
sim.init_partition(namespace);
@@ -1617,6 +1668,120 @@ mod tests {
}
}
+ /// A replica that missed a group's creation materialises it after the
+ /// metadata plane elected again. It must seed the view the group was CREATED
+ /// in, not the live metadata view: seeded above the group's real view, its
+ /// empty log outranks every peer in the next DVC merge and the committed ops
+ /// collect a nack quorum, wedging the group for good.
+ #[test]
+ fn given_a_late_materialiser_when_the_metadata_view_moved_on_should_seed_the_creation_view() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ replica_count as usize,
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let settle = |sim: &mut Simulator| {
+ for _ in 0..800 {
+ sim.step();
+ }
+ };
+ let reply_within = |sim: &mut Simulator, budget: usize| {
+ (0..budget).find_map(|_| {
+ let replies = sim.step();
+ (!replies.is_empty()).then(|| replies[0].deep_copy())
+ })
+ };
+
+ // Replica 0 is down while the group is created, at a metadata view its
+ // crash moved off 0.
+ sim.replica_crash(0);
+ settle(&mut sim);
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(namespace);
+ let created_view = sim
+ .partition_consensus_state(1, namespace)
+ .expect("replica 1 materialised the group")
+ .view;
+ assert!(
+ created_view > 0,
+ "crashing replica 0 must have moved the metadata plane off view 0, or a \
+ creation-view seed is indistinguishable from a view-0 start"
+ );
+
+ // Commit a write, so the group holds history a wrong seed could outrank.
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+ let primary = sim
+ .primary_index(namespace)
+ .expect("a live replica hosts the group");
+ let request = client.send_messages(namespace, &[Bytes::from_static(b"before")]);
+ sim.submit_request(client_id, primary, request.into_generic());
+ reply_within(&mut sim, 200).expect("the write commits on the fresh group");
+
+ // Replica 0 returns without the group, then the metadata plane elects
+ // again, so its live view now exceeds the view the group was created in.
+ sim.replica_restart(0);
+ settle(&mut sim);
+ let metadata_primary = sim
+ .metadata_primary_index()
+ .expect("a live replica owns metadata consensus");
+ sim.replica_crash(metadata_primary);
+ settle(&mut sim);
+ let moved_view = sim
+ .metadata_view()
+ .expect("a live replica owns metadata consensus");
+ assert!(
+ moved_view > created_view,
+ "the second election must move the metadata view ({moved_view}) past the \
+ group's creation view ({created_view})"
+ );
+
+ // Replica 0 materialises the group now.
+ sim.init_partition(namespace);
+ let seeded = sim
+ .partition_consensus_state(0, namespace)
+ .expect("replica 0 materialised the group")
+ .view;
+ assert_eq!(
+ seeded, created_view,
+ "a late materialiser must seed the group's creation view, not the live metadata \
+ view {moved_view}: above the group's real view its empty log wins the next merge"
+ );
+
+ // With replica 0 in, the group has a quorum again: it elects past its
+ // dead primary and commits a new write.
+ let settled_primary = (0..4)
+ .find_map(|_| {
+ settle(&mut sim);
+ (0..replica_count)
+ .filter(|replica_idx| !sim.is_crashed(*replica_idx))
+ .find(|replica_idx| {
+ sim.partition_consensus_state(usize::from(*replica_idx), namespace)
+ .is_some_and(|state| state.status == Status::Normal && state.is_primary)
+ })
+ })
+ .expect("the group must elect a live primary once replica 0 joins it");
+ let request = client.send_messages(namespace, &[Bytes::from_static(b"after")]);
+ sim.submit_request(client_id, settled_primary, request.into_generic());
+ reply_within(&mut sim, 400).expect(
+ "a write must commit after the late materialiser joined; a wedged merge \
+ answers nothing",
+ );
+ }
+
/// A replica that advanced its view, persisted it through the superblock gate,
/// then crashed recovers that view from its own disk, not a fresh 0. The
/// split-brain guarantee: a replica never forgets a view it acted in. Impossible
@@ -2533,7 +2698,7 @@ mod tests {
executor.run_until_stalled(POLL_BUDGET); // borrow acquired; task parks
let grow = Rc::clone(&sim.replicas[0].shards[0]);
executor.spawn(async move {
- grow.init_partition(ns_grow, None, None, None, false, None);
+ grow.init_partition(ns_grow, None, None, None, false, 0);
});
executor.run_until_stalled(POLL_BUDGET); // grow while the borrow is live
}))
@@ -2568,7 +2733,7 @@ mod tests {
executor.run_until_stalled(POLL_BUDGET);
let grow = Rc::clone(&sim.replicas[0].shards[0]);
executor.spawn(async move {
- grow.init_partition(ns_grow, None, None, None, false, None);
+ grow.init_partition(ns_grow, None, None, None, false, 0);
});
executor.run_until_stalled(POLL_BUDGET);
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index bdbeed1b5..638a61dce 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -155,7 +155,7 @@ pub fn new_shard(
recovered_state: Option<VsrState>,
incarnation: u128,
data_dir: Option<std::path::PathBuf>,
- seed_namespaces: &[server_common::sharding::IggyNamespace],
+ seed_namespaces: &[(server_common::sharding::IggyNamespace, u32)],
) -> (Rc<Replica>, Option<SimMetadataBundle>) {
// Metadata is single-writer, mirroring the server bootstrap. Shard 0 owns
// the only writable STM; every peer shard rebuilds a reader-mode mirror from
@@ -327,8 +327,8 @@ pub fn new_shard(
// `materialise_partition`'s own seed is then a no-op.
if shard_idx == 0 {
let streams = metadata.mux_stm.streams();
- for &namespace in seed_namespaces {
- streams.seed_namespace(namespace, namespace.inner());
+ for &(namespace, created_view) in seed_namespaces {
+ streams.seed_namespace(namespace, namespace.inner(), created_view);
}
}