fix(cluster): make the advertised leader accept partition writes

core/consensus/src/impls.rs

@@ -1127,7 +1127,7 @@ pub struct ConsensusTimers {
 }
 
 /// How a restored replica joins its group.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum JoinMode {
     /// Fresh group or solo replica: plain init; the group needs its view-0
     /// primary to exist.
@@ -1151,11 +1151,72 @@ pub struct VsrRestore<'a> {
     /// View inferred from the last journaled prepare, consulted only when no
     /// 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.
+    ///
+    /// 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.
+    ///
+    /// Deliberately NOT marked superblock-durable: nothing has been written
+    /// yet, and claiming otherwise would let a restart resume a view no record
+    /// holds. The group persists on its first tick instead.
+    pub seed_view: Option<u32>,
     /// Non-zero boot incarnation; `None` keeps the default.
     pub incarnation: Option<u128>,
     pub join: JoinMode,
 }
 
+/// How a group with no consensus state in memory should come up.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct FreshGroupStart {
+    pub join: JoinMode,
+    /// Starting view for a group with no durable record; see
+    /// [`VsrRestore::seed_view`]. Always `None` when a durable record exists,
+    /// so a caller can apply the two in either order.
+    pub seed_view: Option<u32>,
+}
+
+/// Decide how a group being materialised should join, and what view it starts
+/// in when it has no record of its own.
+///
+/// One function because there are two materialisation paths and they must not
+/// drift: `build_partition_fresh` on the server, and the simulator's
+/// `init_partition`, which cannot call it (that builder does real filesystem
+/// work, and the simulator runs on in-memory storage). Both previously decided
+/// this for themselves, which is how the simulator came to exercise neither
+/// the seed nor the split it closes.
+///
+/// `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.
+#[must_use]
+pub const fn fresh_group_start(
+    restarted: bool,
+    durable_view: Option<(u32, u32)>,
+    metadata_view: Option<u32>,
+) -> FreshGroupStart {
+    let join = if restarted {
+        JoinMode::ProbeAsBackup {
+            await_state_transfer: false,
+        }
+    } else {
+        JoinMode::Init
+    };
+    // A durable record outranks the seed, and a probing backup takes neither:
+    // it must sit at or below the group's real view for the primary's
+    // `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,
+    };
+    FreshGroupStart { join, seed_view }
+}
+
 impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
     /// # Panics
     /// - If `replica >= replica_count`.
@@ -1230,6 +1291,14 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
             consensus.mark_superblock_durable(view, log_view);
         } else if let Some(view) = restore.view_fallback {
             consensus.set_view(view);
+        } else if let Some(view) = restore.seed_view {
+            tracing::info!(
+                group,
+                view,
+                "seeded a group with no history of its own into the current view"
+            );
+            consensus.set_view(view);
+            consensus.set_log_view(view);
         }
         match restore.join {
             JoinMode::Init => consensus.init(),
@@ -3988,6 +4057,51 @@ where
     }
 }
 
+#[cfg(test)]
+mod fresh_group_start_tests {
+    use super::{JoinMode, fresh_group_start};
+
+    const METADATA_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);
+        assert_eq!(start.join, JoinMode::Init);
+        assert_eq!(start.seed_view, None);
+    }
+
+    /// A durable record outranks the seed: it is what this replica actually
+    /// promised, and the seed is a guess about someone else's plane.
+    #[test]
+    fn given_a_durable_record_when_seeding_should_prefer_the_record() {
+        let start = fresh_group_start(false, Some((7, 7)), Some(METADATA_VIEW));
+        assert_eq!(start.seed_view, None);
+    }
+
+    /// A probing backup must sit at or below the group's real view for the
+    /// primary's `StartView` to move it forward. Seeded above it, the reply
+    /// 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));
+        assert!(matches!(start.join, JoinMode::ProbeAsBackup { .. }));
+        assert_eq!(start.seed_view, None);
+    }
+}
+
 #[cfg(test)]
 mod request_queue_tests {
     use super::*;

core/harness_derive/src/attrs.rs

@@ -57,6 +57,12 @@ pub struct IggyTestAttrs {
     pub seed_fn: Option<syn::Path>,
     pub cluster_nodes: ClusterNodesValue,
     pub jwks_server: Option<JwksAttrs>,
+    /// Skip the generated `start()` call so the test brings nodes up itself.
+    /// For boot-ordering tests, which need a cluster that begins serving
+    /// before every configured node has arrived; `start` waits for all of
+    /// them. Pairs with `TestHarness::start_nodes` / `start_node`. A `seed`
+    /// cannot run under it, since no node is up when it would fire.
+    pub manual_start: bool,
 }
 
 /// JWKS server attributes.
@@ -90,6 +96,7 @@ impl IggyTestAttrs {
             seed_fn: None,
             cluster_nodes: ClusterNodesValue::None,
             jwks_server: None,
+            manual_start: false,
         }
     }
 }
@@ -265,6 +272,9 @@ impl Parse for IggyTestAttrs {
                 AttrItem::JwksServer(jwks) => {
                     attrs.jwks_server = Some(jwks);
                 }
+                AttrItem::ManualStart => {
+                    attrs.manual_start = true;
+                }
             }
         }
 
@@ -282,6 +292,7 @@ enum AttrItem {
     Seed(syn::Path),
     ClusterNodes(ClusterNodesValue),
     JwksServer(JwksAttrs),
+    ManualStart,
 }
 
 impl Parse for AttrItem {
@@ -317,6 +328,7 @@ impl Parse for AttrItem {
                 let jwks = parse_jwks_attrs(&content)?;
                 Ok(AttrItem::JwksServer(jwks))
             }
+            "manual_start" => Ok(AttrItem::ManualStart),
             _ => Err(syn::Error::new(
                 ident.span(),
                 format!("unknown attribute: {ident_str}"),

core/harness_derive/src/codegen.rs

@@ -557,6 +557,11 @@ fn generate_harness_setup(
 /// after server but before MCP and connectors runtime (which may depend on seed data).
 /// Fixture seeds are combined with the global seed.
 fn generate_start_and_seed(attrs: &IggyTestAttrs, fixture_seed: TokenStream) -> TokenStream {
+    // `manual_start` hands the bring-up to the test body, so there is nothing
+    // to emit and no client for a seed to run against.
+    if attrs.manual_start {
+        return TokenStream::new();
+    }
     let has_fixture_seed = !fixture_seed.is_empty();
     match (&attrs.seed_fn, has_fixture_seed) {
         (Some(seed_fn), true) => {
@@ -837,6 +842,7 @@ mod tests {
             },
             seed_fn: None,
             cluster_nodes: crate::attrs::ClusterNodesValue::None,
+            manual_start: false,
             jwks_server: None,
         };
         let variants = generate_variants(&attrs);

core/integration/src/harness/disk.rs

@@ -448,14 +448,29 @@ pub fn find_partition_superblock_dir(root: &Path) -> Option<PathBuf> {
 /// Index of the node the metadata roster marks as leader, resolved by matching
 /// the roster's TCP port against each node's bound address.
 ///
+/// Reads through node 0. Use [`leader_node_index_via`] where node 0 may be
+/// down or where the roster must be read from a named node.
+///
 /// # Panics
 /// If no root client connects, the roster query fails, no node is marked
 /// leader, or the leader's port matches no harness node.
 pub async fn leader_node_index(harness: &TestHarness) -> usize {
+    leader_node_index_via(harness, 0).await
+}
+
+/// [`leader_node_index`] reading the roster through node `via`.
+///
+/// Named separately because which node answers matters once node 0 may be
+/// down: the roster read is auth-gated, so it needs a node that can complete a
+/// login, and a client built against a dead node cannot.
+///
+/// # Panics
+/// As [`leader_node_index`].
+pub async fn leader_node_index_via(harness: &TestHarness, via: usize) -> usize {
     let client = harness
-        .root_client_for_node(0)
+        .root_client_for_node(via)
         .await
-        .expect("a root client (redirecting to the leader if node 0 is not it)");
+        .expect("a root client (redirecting to the leader if the dialed node is not it)");
     let metadata = client
         .get_cluster_metadata()
         .await