You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/06/19 15:18:08 UTC
(iggy) branch data-integrity-vsr updated (fa2c7f2f7 -> b46634972)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch data-integrity-vsr
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit fa2c7f2f7 fix(partitions): persist data on vsr backups, add data integrity test
new 9d27f7074 fix(partitions): persist data on vsr backups, add data integrity test
new b46634972 fix(consensus): prevent shard panics under in-process Logout storm
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (fa2c7f2f7)
\
N -- N -- N refs/heads/data-integrity-vsr (b46634972)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/consensus/src/impls.rs | 26 ++--
core/consensus/src/plane_helpers.rs | 46 +++++-
.../verify_cluster_replica_data_identical.rs | 155 +++++++++++++++++++--
core/journal/src/prepare_journal.rs | 9 +-
core/metadata/src/impls/metadata.rs | 15 +-
core/partitions/src/iggy_partition.rs | 128 +++++++++++------
core/partitions/src/journal.rs | 153 ++++++++++++++++++++
core/shard/src/lib.rs | 6 +-
8 files changed, 457 insertions(+), 81 deletions(-)
(iggy) 01/02: fix(partitions): persist data on vsr backups, add data integrity test
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch data-integrity-vsr
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 9d27f70742724b9a9471544dbe664fd90563971f
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Jun 19 11:40:28 2026 +0200
fix(partitions): persist data on vsr backups, add data integrity test
VSR's hash chain and checksum-keyed recovery require a committed
op's on-disk bytes to match on every replica. Backups persisted
0-byte segments: the partition commit path drained the in-memory
pipeline, which only the primary fills. And the append path
stamped base_timestamp from a local now(), diverging bytes per
node even once persisted.
Backups now source committable ops from the journal when the
pipeline is empty, and commit_messages flushes only the committed
prefix (op <= commit_max), keeping the uncommitted tail resident.
A backup thus never writes uncommitted bytes to its segment and
never drops the headers a later commit needs - which would
otherwise wedge commit_min below commit_max. base_timestamp
reuses the prepare's monotonic timestamp, stamped once by the
primary and replicated. A 3-node data-integrity test gates the
cross-replica byte-identity.
---
core/integration/tests/data_integrity/mod.rs | 11 +
.../verify_cluster_replica_data_identical.rs | 376 +++++++++++++++++++++
core/integration/tests/mod.rs | 15 +-
core/journal/src/prepare_journal.rs | 6 +-
core/partitions/src/iggy_partition.rs | 139 ++++++--
core/partitions/src/journal.rs | 153 +++++++++
6 files changed, 659 insertions(+), 41 deletions(-)
diff --git a/core/integration/tests/data_integrity/mod.rs b/core/integration/tests/data_integrity/mod.rs
index c095a56bc..136a807cd 100644
--- a/core/integration/tests/data_integrity/mod.rs
+++ b/core/integration/tests/data_integrity/mod.rs
@@ -15,7 +15,18 @@
// specific language governing permissions and limitations
// under the License.
+// The legacy verifiers run only without `vsr`: they rely on Http/WebSocket/Quic
+// transports, the bench producer, or consumer groups, none of which the VSR
+// server serves yet.
+#[cfg(not(feature = "vsr"))]
mod verify_after_server_restart;
+#[cfg(not(feature = "vsr"))]
mod verify_consumer_group_partition_assignment;
+#[cfg(not(feature = "vsr"))]
mod verify_no_plaintext_credentials_on_disk;
+#[cfg(not(feature = "vsr"))]
mod verify_user_login_after_restart;
+
+// Cross-replica on-disk data identity is VSR-only.
+#[cfg(feature = "vsr")]
+mod verify_cluster_replica_data_identical;
diff --git a/core/integration/tests/data_integrity/verify_cluster_replica_data_identical.rs b/core/integration/tests/data_integrity/verify_cluster_replica_data_identical.rs
new file mode 100644
index 000000000..0d7f4a815
--- /dev/null
+++ b/core/integration/tests/data_integrity/verify_cluster_replica_data_identical.rs
@@ -0,0 +1,376 @@
+// 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.
+
+#![cfg(feature = "vsr")]
+
+//! Cross-replica on-disk data identity.
+//!
+//! VSR replicates each plane's log as a hash chain: every prepare header carries
+//! `parent == previous entry checksum`, and recovery / view-change locate entries
+//! by `(op, checksum)`. That makes one property load-bearing for safety: the
+//! committed on-disk bytes for a given op must be identical on every replica. If
+//! they differ, per-replica checksums differ, the chain identity breaks across
+//! nodes, and recovery's checksum-keyed lookups miss.
+//!
+//! This test produces a single batch, then a small concurrent burst, then a long
+//! sequential run to a 3-node cluster, and compares the partition segment `.log`
+//! and the metadata-plane files byte-for-byte across every node. The burst
+//! pipelines ops so backups journal ahead of their commit frontier (the window in
+//! which a backup commit must persist only the committed prefix and keep the
+//! uncommitted tail resident); the sequential run that follows drives many backup
+//! commit cycles and would expose a backup left permanently behind. The segment
+//! `.index` is excluded - see `is_comparable`.
+
+use iggy::prelude::*;
+use integration::iggy_harness;
+use std::collections::{BTreeMap, BTreeSet};
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+use tokio::time::sleep;
+
+const STREAM_NAME: &str = "di-stream";
+const TOPIC_NAME: &str = "di-topic";
+const CLUSTER_NODES: usize = 3;
+const MESSAGES: u32 = 64;
+
+// Sequential multi-batch phase: drive the backup commit path through many
+// commit cycles (each batch is its own op), so `collect_committable_from_journal`
+// runs repeatedly and `commit_messages` keeps proving the retain-the-tail /
+// persist-only-committed contract op after op. Deterministic, unlike the burst.
+const SEQUENTIAL_BATCHES: u32 = 24;
+const SEQUENTIAL_BATCH_MESSAGES: u32 = 3;
+
+// Concurrent producers that pipeline several ops at once so backups journal ops
+// ahead of their commit frontier (op > commit_max) - the window in which a
+// backup commit must persist ONLY the committed prefix and keep the uncommitted
+// tail resident. Kept small and the connections are held open until after the
+// cluster is stopped: a large fan-out plus a simultaneous disconnect storm trips
+// unrelated metadata/consensus concurrency asserts (mass in-process Logout, a
+// view-change with an over-deep in-flight range) and destabilises the cluster
+// before this test can observe its own property.
+const BURST_CLIENTS: usize = 8;
+const BURST_SENDS_PER_CLIENT: u32 = 3;
+const BURST_BATCH_MESSAGES: u32 = 2;
+
+// Let replication and per-batch persistence reach every replica before the
+// cluster is stopped and files are read at rest. Generous on purpose: backups
+// must drain their commit frontier and persist every committed batch.
+const REPLICATION_SETTLE: Duration = Duration::from_secs(10);
+
+// `messages_required_to_save = 1` forces every committed batch to persist to its
+// segment immediately on every node, so each replica materialises the segment
+// files while running (the VSR server serves no flush_unsaved_buffer, and
+// shutdown-flush would couple the test to drain behaviour). The harness applies
+// the `server(...)` config to every cluster node.
+#[iggy_harness(
+ cluster_nodes = 3,
+ server(partition.messages_required_to_save = 1)
+)]
+async fn segment_and_metadata_bytes_identical_across_replicas(harness: &mut TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
+ client.create_stream(STREAM_NAME).await.unwrap();
+ client
+ .create_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+
+ // Phase 1: a single batch. Exercises the simple one-op commit on every
+ // replica and keeps the original byte-identity baseline.
+ let mut messages: Vec<IggyMessage> = (0..MESSAGES)
+ .map(|i| {
+ IggyMessage::builder()
+ .id(u128::from(i + 1))
+ .payload(format!("replica-identity-payload-{i:04}").into())
+ .build()
+ .unwrap()
+ })
+ .collect();
+ client
+ .send_messages(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+
+ // Phase 2: a concurrent burst that pipelines ops through the single
+ // partition's consensus group, so backups journal ops above their
+ // commit_max. On the buggy path a backup over-drains its journal on the first
+ // committed op, persisting uncommitted bytes and dropping the headers of
+ // still-uncommitted ops; that wedges its commit_min permanently (the dropped
+ // headers never reappear), so it stops persisting from then on. The
+ // connections are returned and held open until after the cluster stops: a
+ // mid-test disconnect storm would trip an unrelated metadata Logout race.
+ let _burst_clients =
+ send_concurrent_burst(harness.tcp_root_clients(BURST_CLIENTS).await.unwrap()).await;
+
+ // Phase 3: a long sequential run AFTER the burst. Each batch is its own
+ // committed op, exercising the backup journal-fallback commit and the
+ // persist-only-committed / retain-the-tail path once per op. It also exposes
+ // a PERMANENT wedge from phase 2: a backup whose commit_min stalled cannot
+ // advance past the gap, so none of these later batches reach its segment and
+ // its `.log` falls behind the primary's at rest. A healthy backup persists
+ // every batch and stays byte-identical.
+ for batch_index in 0..SEQUENTIAL_BATCHES {
+ let mut messages: Vec<IggyMessage> = (0..SEQUENTIAL_BATCH_MESSAGES)
+ .map(|message_index| {
+ let unique = batch_index * SEQUENTIAL_BATCH_MESSAGES + message_index;
+ IggyMessage::builder()
+ .id(u128::from(MESSAGES + unique + 1) << 64)
+ .payload(format!("sequential-{batch_index:04}-{message_index:02}").into())
+ .build()
+ .unwrap()
+ })
+ .collect();
+ client
+ .send_messages(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+ }
+
+ sleep(REPLICATION_SETTLE).await;
+
+ let data_paths: Vec<PathBuf> = harness
+ .all_servers()
+ .iter()
+ .map(|server| server.data_path())
+ .collect();
+ assert_eq!(
+ data_paths.len(),
+ CLUSTER_NODES,
+ "expected {CLUSTER_NODES} node data dirs"
+ );
+
+ // Stop the whole cluster so every segment / metadata file is at rest. The
+ // burst connections are still held; dropping them only after stop avoids the
+ // concurrent in-process Logout race on the metadata plane.
+ harness.stop().await.unwrap();
+ drop(_burst_clients);
+
+ let per_node: Vec<BTreeMap<String, Vec<u8>>> = data_paths
+ .iter()
+ .map(|root| collect_comparable_files(root))
+ .collect();
+
+ for (idx, node) in per_node.iter().enumerate() {
+ eprintln!(
+ "node {idx}: {} comparable file(s): {:?}",
+ node.len(),
+ node.iter()
+ .map(|(rel, bytes)| format!("{rel} ({} B)", bytes.len()))
+ .collect::<Vec<_>>()
+ );
+ }
+
+ assert_replica_data_identical(&per_node);
+}
+
+/// Drive a concurrent send burst across independent connections so the primary
+/// pipelines multiple ops at once and backups journal ops ahead of their commit
+/// frontier. Each client sends its own batches in a loop; the sends race so that
+/// while one op is committing on a backup, later ops are already resident in its
+/// journal. Message ids are disjoint per client to avoid any cross-batch id
+/// collision. Every send must succeed: a wedged backup still acks (the primary
+/// commits on quorum), so the divergence shows up on disk, not as a send error.
+///
+/// The clients are returned so the caller can hold the connections open until
+/// after the cluster is stopped (a simultaneous disconnect storm trips an
+/// unrelated metadata-plane in-process Logout race).
+async fn send_concurrent_burst(clients: Vec<IggyClient>) -> Vec<IggyClient> {
+ let mut handles = Vec::with_capacity(clients.len());
+ for (client_index, client) in clients.into_iter().enumerate() {
+ // Each task owns its own connection so the sends run concurrently and
+ // the primary's pipeline fills with multiple in-flight ops; the client
+ // is handed back so the connection stays open past the burst.
+ let handle = tokio::spawn(async move {
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ let partitioning = Partitioning::partition_id(0);
+ for send_index in 0..BURST_SENDS_PER_CLIENT {
+ let base = ((client_index as u128) << 32) | (u128::from(send_index) << 8);
+ let mut messages: Vec<IggyMessage> = (0..BURST_BATCH_MESSAGES)
+ .map(|message_index| {
+ IggyMessage::builder()
+ .id(base + u128::from(message_index) + 1)
+ .payload(
+ format!(
+ "burst-{client_index:02}-{send_index:02}-{message_index:02}"
+ )
+ .into(),
+ )
+ .build()
+ .unwrap()
+ })
+ .collect();
+ client
+ .send_messages(&stream, &topic, &partitioning, &mut messages)
+ .await
+ .unwrap();
+ }
+ client
+ });
+ handles.push(handle);
+ }
+
+ let mut clients = Vec::with_capacity(handles.len());
+ for handle in handles {
+ clients.push(handle.await.unwrap());
+ }
+ clients
+}
+
+/// A file (relative to a node's data dir) whose bytes must match across replicas:
+/// the partition segment `.log` and the metadata-plane files. Per-node files
+/// (logs, runtime, config, stdout) are excluded by construction.
+///
+/// The segment `.index` is deliberately NOT compared. It is a local sparse index
+/// (one entry per persist flush), not replicated data and not part of the VSR
+/// hash chain; recovery rebuilds it from the `.log`. Its byte length tracks the
+/// number of `commit_messages` flushes, which differs by commit cadence - the
+/// primary commits one op per quorum ack (one flush per op), a backup commits a
+/// whole heartbeat's committed range in a single `commit_journal` (one flush for
+/// many ops). So the `.index` legitimately differs across replicas for a
+/// multi-batch stream even though every committed `.log` byte is identical;
+/// requiring it to match would assert an incidental local-cache layout, not the
+/// cross-replica data identity this test exists to protect.
+fn is_comparable(rel: &str) -> bool {
+ let is_segment = rel.starts_with("streams/") && rel.ends_with(".log");
+ let is_metadata = rel.starts_with("metadata/") || rel.starts_with("state/");
+ is_segment || is_metadata
+}
+
+fn collect_comparable_files(root: &Path) -> BTreeMap<String, Vec<u8>> {
+ let mut files = BTreeMap::new();
+ let mut stack = vec![root.to_path_buf()];
+ while let Some(dir) = stack.pop() {
+ let Ok(entries) = fs::read_dir(&dir) else {
+ continue;
+ };
+ for entry in entries.flatten() {
+ let path = entry.path();
+ let Ok(file_type) = entry.file_type() else {
+ continue;
+ };
+ if file_type.is_dir() {
+ stack.push(path);
+ } else if file_type.is_file()
+ && let Ok(rel) = path.strip_prefix(root)
+ {
+ let rel = rel.to_string_lossy().replace('\\', "/");
+ if is_comparable(&rel) {
+ let bytes = fs::read(&path)
+ .unwrap_or_else(|e| panic!("failed to read {}: {e}", path.display()));
+ files.insert(rel, bytes);
+ }
+ }
+ }
+ }
+ files
+}
+
+fn assert_replica_data_identical(per_node: &[BTreeMap<String, Vec<u8>>]) {
+ // Guard against a vacuous pass: node 0 must actually hold a produced segment,
+ // otherwise nothing was persisted and the comparison proves nothing.
+ let node0 = &per_node[0];
+ assert!(
+ node0
+ .keys()
+ .any(|k| k.starts_with("streams/") && k.ends_with(".log")),
+ "node 0 holds no segment .log under streams/ - no partition data was persisted, \
+ so the cross-replica comparison would be vacuous. Comparable files: {:?}",
+ node0.keys().collect::<Vec<_>>()
+ );
+
+ let all_keys: BTreeSet<&str> = per_node
+ .iter()
+ .flat_map(|node| node.keys().map(String::as_str))
+ .collect();
+
+ let mut problems = Vec::new();
+ for key in all_keys {
+ let mut reference: Option<(usize, &[u8])> = None;
+ for (idx, node) in per_node.iter().enumerate() {
+ let Some(bytes) = node.get(key) else {
+ problems.push(format!(
+ "`{key}` present on some replicas but MISSING on node {idx}"
+ ));
+ continue;
+ };
+ let bytes: &[u8] = bytes;
+ match reference {
+ None => reference = Some((idx, bytes)),
+ Some((ref_idx, ref_bytes)) => {
+ if bytes != ref_bytes {
+ problems.push(describe_mismatch(key, ref_idx, ref_bytes, idx, bytes));
+ }
+ }
+ }
+ }
+ }
+
+ assert!(
+ problems.is_empty(),
+ "cross-replica data divergence ({} issue(s)):\n{}",
+ problems.len(),
+ problems.join("\n")
+ );
+}
+
+fn describe_mismatch(key: &str, a_idx: usize, a: &[u8], b_idx: usize, b: &[u8]) -> String {
+ let window = |buf: &[u8], at: usize| {
+ let start = at.saturating_sub(8);
+ let end = (at + 8).min(buf.len());
+ buf[start..end]
+ .iter()
+ .map(|byte| format!("{byte:02x}"))
+ .collect::<Vec<_>>()
+ .join(" ")
+ };
+ match a.iter().zip(b.iter()).position(|(x, y)| x != y) {
+ Some(at) => format!(
+ "`{key}`: bytes differ between node {a_idx} ({} B) and node {b_idx} ({} B) at offset {at}. \
+ node{a_idx}=[{}] node{b_idx}=[{}]. For a segment .log this is the per-replica \
+ base_timestamp / batch_checksum stamped in `stamp_prepare_for_persistence`, or a \
+ backup that persisted past its commit frontier.",
+ a.len(),
+ b.len(),
+ window(a, at),
+ window(b, at),
+ ),
+ None => format!(
+ "`{key}`: length differs between node {a_idx} ({} B) and node {b_idx} ({} B)",
+ a.len(),
+ b.len(),
+ ),
+ }
+}
diff --git a/core/integration/tests/mod.rs b/core/integration/tests/mod.rs
index 6f59ae6e7..e2dfcaf9e 100644
--- a/core/integration/tests/mod.rs
+++ b/core/integration/tests/mod.rs
@@ -38,17 +38,10 @@ mod cluster;
mod config_provider;
#[cfg(not(feature = "vsr"))]
mod connectors;
-// TODO(vsr): enable the `data_integrity` suite under the `vsr` feature once
-// the full VSR path is done. Tier 1 (`verify_user_login_after_restart`,
-// `verify_no_plaintext_credentials_on_disk`) is server-side ready (create_user
-// reply body, get_users/get_user via `frontend()`, change_password, cross-shard
-// committed-reply routing) -- proven by `sdk::hello_world::replicated_*` under
-// vsr. Blocked on: (1) these tests use `[Tcp, Http, Quic, WebSocket]` but only
-// Tcp works under vsr today (Http does no VSR framing; WebSocket/Quic hit the
-// compio `buffer.rs:83` bug); (2) the module also carries
-// `verify_after_server_restart` + `verify_consumer_group_partition_assignment`,
-// which need send/poll messages + consumer groups (out of Tier 1 scope).
-#[cfg(not(feature = "vsr"))]
+// Compiled in both lanes; members are gated individually (see its `mod.rs`).
+// Under `vsr` only the cross-replica on-disk identity test is active. The four
+// legacy verifiers stay non-vsr: they use Http/WebSocket/Quic, the bench
+// producer, or consumer groups, which the VSR server does not serve yet.
mod data_integrity;
#[cfg(not(feature = "vsr"))]
mod mcp;
diff --git a/core/journal/src/prepare_journal.rs b/core/journal/src/prepare_journal.rs
index 809a10517..2117d1341 100644
--- a/core/journal/src/prepare_journal.rs
+++ b/core/journal/src/prepare_journal.rs
@@ -294,7 +294,11 @@ impl PrepareJournal {
// Until a producer computes them, verification here would be
// trivially-passing noise. Without it, a body bit-flip that
// leaves the header valid is replayed silently as corrupt
- // state.
+ // state. Committed bytes are meant to be byte-identical across
+ // replicas (deterministic apply, timestamp replicated not
+ // re-projected), so once the producer computes the integrity fields
+ // they should agree on every node and this check can be turned on
+ // without per-replica false positives.
// Check if the full entry fits
if pos + entry_size > file_len {
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 029fd35fd..024c7ea84 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -55,8 +55,8 @@ use server_common::{
Message, SegmentStorage,
iobuf::{Frozen, Owned},
send_messages2::{
- COMMAND_HEADER_SIZE, convert_request_message, decode_batch_slice, decode_prepare_slice,
- stamp_prepare_for_persistence,
+ COMMAND_HEADER_SIZE, SendMessages2Ref, convert_request_message, decode_batch_slice,
+ decode_prepare_slice, stamp_prepare_for_persistence,
},
sharding::IggyNamespace,
};
@@ -528,9 +528,12 @@ where
0
};
- // TODO: Replace this with monotonic broker timestamp assignment. If wall clock
- // time goes backwards, clamp to the partition/log max timestamp instead.
- let batch_timestamp = IggyTimestamp::now().as_micros();
+ // Reuse the prepare's monotonic timestamp, assigned once by the primary
+ // in `project()` (`next_monotonic_timestamp`) and replicated verbatim to
+ // every backup. Sourcing it here instead of a fresh local `now()` makes
+ // the persisted `base_timestamp` (and the `batch_checksum` derived from
+ // it) byte-identical across replicas; a local `now()` diverges per node.
+ let batch_timestamp = header.timestamp;
let (message, batch, batch_messages_count) =
stamp_prepare_for_persistence(message, dirty_offset, batch_timestamp)
.map_err(|_| IggyError::CannotAppendMessage)?;
@@ -1433,7 +1436,19 @@ where
pub async fn commit_journal(&mut self, config: &PartitionsConfig) {
self.clear_pending_consumer_offset_commits_if_view_changed();
- let drained = drain_committable_prefix(self.consensus());
+ // The primary commits inline via `on_ack` (it drains its own pipeline).
+ // Backups never populate the pipeline - they journal replicated prepares
+ // in `apply_replicated_operation` - so the pipeline drain is empty for
+ // them. Fall back to the journal so backups durably persist committed
+ // data. `commit_messages` then flushes only the committed prefix and
+ // keeps the uncommitted tail journal-resident, so a later commit of that
+ // tail still finds its headers here (no wedge). Pipeline-first keeps a
+ // freshly promoted primary (rebuilt pipeline) draining there, avoiding a
+ // double-count against `advance_commit_min`.
+ let mut drained = drain_committable_prefix(self.consensus());
+ if drained.is_empty() {
+ drained = self.collect_committable_from_journal();
+ }
if drained.is_empty() {
return;
}
@@ -1450,6 +1465,25 @@ where
}
}
+ /// Committable entries (ops `commit_min+1 ..= commit_max`) read from the
+ /// journal, for a backup whose pipeline is empty. Stops at the first missing
+ /// op: a replication gap must not be skipped, or `advance_commit_min`'s
+ /// sequential contract breaks. Like the metadata plane's `commit_journal`,
+ /// the journal keeps its committed entries until they are flushed
+ /// (`commit_messages` drains only the committed prefix), so this read finds
+ /// every committed op while the uncommitted tail stays resident.
+ fn collect_committable_from_journal(&self) -> Vec<PipelineEntry> {
+ let from_op = self.consensus.commit_min() + 1;
+ let commit_max = self.consensus.commit_max();
+ self.log
+ .journal()
+ .inner
+ .committed_headers_from(from_op, commit_max)
+ .into_iter()
+ .map(PipelineEntry::new)
+ .collect()
+ }
+
async fn apply_replicated_operation(
&mut self,
message: Message<PrepareHeader>,
@@ -1572,17 +1606,34 @@ where
return Ok(());
}
- let (frozen_batches, index_bytes, batch_count) = {
- let entries = self.log.journal().inner.entries();
+ // Persist and evict ONLY the committed prefix (op <= commit_max). A
+ // backup journals replicated prepares ahead of the commit frontier;
+ // flushing the uncommitted tail would write per-replica-timing bytes to
+ // its segment (cross-replica divergence) and drop the headers those ops
+ // need when their own commit later lands (commit_min wedge). All segment
+ // range / stats / durable-offset accounting below is computed from the
+ // committed entries, not the resident-journal snapshot above.
+ let commit_max = self.consensus.commit_max();
+ let committed_entries = self
+ .log
+ .journal()
+ .inner
+ .drain_committed_prefix(commit_max)
+ .await;
+ let retained_info = recompute_journal_info(&self.log.journal().inner.entries());
+ self.log.journal_mut().info = retained_info;
+
+ let (frozen_batches, index_bytes, batch_count, committed_info) = {
let segment = self.log.active_segment();
let mut file_position = segment.size.as_bytes_u64();
self.log.ensure_indexes();
let indexes = self.log.active_indexes_mut().expect("indexes must exist");
let mut flush_index = None;
- let mut frozen = Vec::with_capacity(entries.len());
+ let mut frozen = Vec::with_capacity(committed_entries.len());
let mut batch_count = 0u32;
+ let mut committed_info = JournalInfo::default();
- for entry in entries {
+ for entry in committed_entries {
let Ok(batch) = decode_prepare_slice(entry.as_slice()) else {
continue;
};
@@ -1602,23 +1653,22 @@ where
}
file_position += batch.header.total_size() as u64;
batch_count += message_count;
+ accumulate_committed_info(&mut committed_info, &batch, message_count);
frozen.push(entry);
}
let index_bytes =
flush_index.map(|index| crate::iggy_index::IggyIndexCache::serialize(&index));
- (frozen, index_bytes, batch_count)
+ (frozen, index_bytes, batch_count, committed_info)
};
+ // No committed SendMessages batch was resident (e.g. only uncommitted
+ // ops, or a committed consumer-offset prefix that is not persisted to a
+ // segment). The drain already evicted/retained correctly; nothing to
+ // flush.
let Some(index_bytes) = index_bytes else {
- warn!(
- target: "iggy.partitions.diag",
- plane = "partitions",
- namespace_raw = self.namespace().inner(),
- "failed to build sparse index entry from pending journal batches"
- );
- return Err(IggyError::InvalidCommand);
+ return Ok(());
};
self.persist_frozen_batches_to_disk(frozen_batches, index_bytes, batch_count)
@@ -1631,26 +1681,23 @@ where
// subsequent segment boundary off the file contents.
let segment_index = self.log.segments().len() - 1;
let segment = &mut self.log.segments_mut()[segment_index];
- if segment.start_timestamp == 0 && journal_info.first_timestamp != 0 {
- segment.start_timestamp = journal_info.first_timestamp;
+ if segment.start_timestamp == 0 && committed_info.first_timestamp != 0 {
+ segment.start_timestamp = committed_info.first_timestamp;
}
- segment.end_timestamp = journal_info.end_timestamp;
- segment.max_timestamp = segment.max_timestamp.max(journal_info.max_timestamp);
- segment.end_offset = journal_info.current_offset;
+ segment.end_timestamp = committed_info.end_timestamp;
+ segment.max_timestamp = segment.max_timestamp.max(committed_info.max_timestamp);
+ segment.end_offset = committed_info.current_offset;
if is_full {
self.rotate_segment(config).await?;
}
- let _ = self.log.journal_mut().inner.commit();
- self.log.journal_mut().info = JournalInfo::default();
-
self.stats
- .increment_size_bytes(journal_info.size.as_bytes_u64());
+ .increment_size_bytes(committed_info.size.as_bytes_u64());
self.stats
- .increment_messages_count(u64::from(journal_info.messages_count));
+ .increment_messages_count(u64::from(committed_info.messages_count));
- let durable_offset = journal_info.current_offset;
+ let durable_offset = committed_info.current_offset;
self.offset.store(durable_offset, Ordering::Release);
self.stats.set_current_offset(durable_offset);
Ok(())
@@ -2215,6 +2262,40 @@ where
}
}
+/// Rebuild `JournalInfo` from the entries left resident after a committed
+/// prefix is drained, so the next persist threshold reflects only the
+/// uncommitted tail. Mirrors the per-append accounting in `append_messages`,
+/// folded over every retained `SendMessages` batch. Non-`SendMessages` ops
+/// (consumer offsets) carry no segment accounting and are skipped.
+fn recompute_journal_info(entries: &[Frozen<4096>]) -> JournalInfo {
+ let mut info = JournalInfo::default();
+ for entry in entries {
+ let Ok(batch) = decode_prepare_slice(entry.as_slice()) else {
+ continue;
+ };
+ let message_count = batch.message_count();
+ if message_count == 0 {
+ continue;
+ }
+ accumulate_committed_info(&mut info, &batch, message_count);
+ }
+ info
+}
+
+/// Fold one decoded `SendMessages` batch into a running `JournalInfo`, matching
+/// the field updates `append_messages` applies per append. `current_offset` is
+/// the batch's last message offset; the batch carries a contiguous offset run.
+fn accumulate_committed_info(info: &mut JournalInfo, batch: &SendMessages2Ref<'_>, count: u32) {
+ info.messages_count += count;
+ info.size += IggyByteSize::from(batch.header.total_size() as u64);
+ info.current_offset = batch.header.base_offset + u64::from(count) - 1;
+ if info.first_timestamp == 0 {
+ info.first_timestamp = batch.header.base_timestamp;
+ }
+ info.end_timestamp = batch.header.base_timestamp;
+ info.max_timestamp = info.max_timestamp.max(batch.header.base_timestamp);
+}
+
/// Walk stamped `[256B SendMessages2Header][blob]` batches in one disk
/// chunk, pushing matching fragments. Returns bytes consumed: the start
/// of the first batch that did not fully fit in the chunk (the caller
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index 107a935b0..eb881f264 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -233,6 +233,62 @@ impl PartitionJournal<PartitionJournalMemStorage> {
entries
}
+ /// Drain entries with `op <= commit_max`, returning them in append order,
+ /// and retain entries with `op > commit_max` in place with their indexes
+ /// rebuilt for the compacted layout.
+ ///
+ /// A backup journals replicated prepares up to a full pipeline ahead of the
+ /// commit frontier. Only the committed prefix may be flushed to a segment
+ /// and evicted; persisting and clearing the uncommitted tail would write
+ /// per-replica-timing bytes to disk (cross-replica divergence) and drop the
+ /// headers those ops need when their own commit later lands (`commit_min`
+ /// wedge). Entries are append-ordered, which is op-ascending on a backup, so
+ /// the committed prefix is the front; the few uncommitted entries are
+ /// re-appended after the drain.
+ pub async fn drain_committed_prefix(&self, commit_max: u64) -> Vec<JournalBuffer> {
+ let header_ops: Vec<u64> = {
+ let headers = unsafe { &*self.headers.get() };
+ headers.iter().map(|header| header.op).collect()
+ };
+ let all_entries = {
+ let inner = unsafe { &*self.inner.get() };
+ inner.storage.drain()
+ };
+
+ {
+ let headers = unsafe { &mut *self.headers.get() };
+ headers.clear();
+ let op_to_storage_offset = unsafe { &mut *self.op_to_storage_offset.get() };
+ op_to_storage_offset.clear();
+ let offset_to_op = unsafe { &mut *self.offset_to_op.get() };
+ offset_to_op.clear();
+ let timestamp_to_op = unsafe { &mut *self.timestamp_to_op.get() };
+ timestamp_to_op.clear();
+ }
+
+ let mut committed = Vec::new();
+ let mut retained = Vec::new();
+ for (op, entry) in header_ops.into_iter().zip(all_entries) {
+ if op <= commit_max {
+ committed.push(entry);
+ } else {
+ retained.push(entry);
+ }
+ }
+
+ // Re-append the uncommitted tail so storage offsets and the op /
+ // offset / timestamp indexes match the compacted layout. `append`
+ // rebuilds every index from the entry header. The bytes were valid
+ // when first appended, so re-appending them cannot fail.
+ for entry in retained {
+ self.append(entry)
+ .await
+ .expect("re-appending a retained journal entry must not fail");
+ }
+
+ committed
+ }
+
pub fn is_empty(&self) -> bool {
let inner = unsafe { &*self.inner.get() };
inner.storage.is_empty()
@@ -259,6 +315,31 @@ where
headers.iter().find(|header| header.op == op).copied()
}
+ /// Headers for the contiguous op run `from_op ..= commit_max`, in op order,
+ /// stopping at the first missing op. A replication gap must not be skipped:
+ /// the caller advances `commit_min` strictly by one, so a hole would break
+ /// that contract. Single ordered pass over the resident headers (O(n)),
+ /// versus an `O(n*range)` repeated `header_by_op` scan.
+ pub fn committed_headers_from(&self, from_op: u64, commit_max: u64) -> Vec<PrepareHeader> {
+ if from_op > commit_max {
+ return Vec::new();
+ }
+ let by_op: BTreeMap<u64, PrepareHeader> = {
+ let headers = unsafe { &*self.headers.get() };
+ headers.iter().map(|header| (header.op, *header)).collect()
+ };
+ let mut result = Vec::with_capacity(usize::try_from(commit_max - from_op + 1).unwrap_or(0));
+ let mut op = from_op;
+ while op <= commit_max {
+ let Some(header) = by_op.get(&op) else {
+ break;
+ };
+ result.push(*header);
+ op += 1;
+ }
+ result
+ }
+
/// Oldest message offset still resident in the in-memory journal, if
/// any. Polls below this must fall back to the on-disk segments.
pub fn oldest_resident_offset(&self) -> Option<u64> {
@@ -621,4 +702,76 @@ mod tests {
"cloning a journal entry must yield identical bytes (refcount bump, not deep copy)"
);
}
+
+ #[compio::test]
+ async fn drain_committed_prefix_retains_uncommitted_tail() {
+ // A backup journals ops ahead of the commit frontier. Draining the
+ // committed prefix (op <= commit_max) must flush only those and keep the
+ // uncommitted tail resident + readable, with its headers intact, so a
+ // later commit of that tail still finds it (no commit_min wedge).
+ let journal = PartitionJournal::<PartitionJournalMemStorage>::default();
+ for op in 1..=4 {
+ journal
+ .append(build_prepare(op, HEADER_SIZE + 16).into_frozen())
+ .await
+ .expect("append");
+ }
+
+ let committed = journal.drain_committed_prefix(2).await;
+ assert_eq!(
+ committed.len(),
+ 2,
+ "ops 1 and 2 are the committed prefix and must be returned"
+ );
+
+ // Committed ops are evicted from the index; uncommitted ops remain.
+ assert!(journal.header_by_op(1).is_none(), "op 1 must be evicted");
+ assert!(journal.header_by_op(2).is_none(), "op 2 must be evicted");
+ let header3 = journal.header_by_op(3).expect("op 3 must be retained");
+ let header4 = journal.header_by_op(4).expect("op 4 must be retained");
+
+ // Retained entries are still byte-readable after the storage rebuild.
+ for header in [header3, header4] {
+ let entry = journal
+ .entry(&header)
+ .await
+ .expect("retained entry must read back");
+ let stored = bytemuck::checked::try_from_bytes::<PrepareHeader>(
+ &entry[..std::mem::size_of::<PrepareHeader>()],
+ )
+ .expect("retained entry must hold a valid prepare header");
+ assert_eq!(stored.op, header.op);
+ }
+
+ // A second drain advancing the frontier flushes the rest with no gap.
+ let rest = journal.drain_committed_prefix(4).await;
+ assert_eq!(rest.len(), 2, "ops 3 and 4 flush on the next drain");
+ assert!(journal.is_empty(), "journal is empty once all ops flushed");
+ }
+
+ #[compio::test]
+ async fn committed_headers_from_stops_at_gap() {
+ let journal = PartitionJournal::<PartitionJournalMemStorage>::default();
+ for op in [1u64, 2, 4] {
+ journal
+ .append(build_prepare(op, HEADER_SIZE + 16).into_frozen())
+ .await
+ .expect("append");
+ }
+
+ // Contiguous run from op 1 stops before the missing op 3 even though
+ // op 4 is resident and within commit_max.
+ let run = journal.committed_headers_from(1, 4);
+ let ops: Vec<u64> = run.iter().map(|header| header.op).collect();
+ assert_eq!(
+ ops,
+ vec![1, 2],
+ "must stop at the op 3 gap, not skip to op 4"
+ );
+
+ assert!(
+ journal.committed_headers_from(5, 4).is_empty(),
+ "from_op past commit_max yields nothing"
+ );
+ }
}
(iggy) 02/02: fix(consensus): prevent shard panics under in-process Logout storm
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch data-integrity-vsr
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b46634972902006891f6a0d24dd04d1a05f9d1d8
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Jun 19 16:47:10 2026 +0200
fix(consensus): prevent shard panics under in-process Logout storm
A disconnect storm spawns many concurrent in-process Logout submits
on shard 0, breaking two invariants that held only under serial
submission.
dispatch_prepare_and_await (metadata plane) snapshotted view and
commit_min before on_replicate().await and asserted them unchanged
after. A sibling on_ack legitimately advances commit_min while a
task is parked in the await, and a view change advances the view,
tripping the debug_assert. The snapshots fed only the assert; no
release-mode logic read them. Both conditions are handled
downstream - a view change drops the reply_sender so the receiver
resolves Canceled, and loopback acks are op-routed - so the assert
and snapshots are removed.
A new primary floors its view-change pipeline rebuild at the max
commit across the DoViewChange quorum. The DVC carried commit_min
(locally applied), which can lag commit_max (known committed) by
more than the pipeline depth: on_ack pops the committable prefix at
once then applies it across an await per entry, while concurrent
submits push op ahead. The rebuild range then exceeds
PIPELINE_PREPARE_QUEUE_MAX and the assert, or the pipeline push,
panics the shard in release builds. Carry commit_max instead: every
replica holds op - commit_max <= pipeline depth, so the rebuilt
range stays within capacity, while quorum intersection keeps the
floor at or below the winner's op. The committed-but-unapplied tail
is replayed by CommitJournal. Only the value written to the
DoViewChange commit field changes; the wire layout is untouched.
---
core/consensus/src/impls.rs | 26 +++++++++++----------
core/consensus/src/plane_helpers.rs | 46 ++++++++++++++++++++++++++++++++++++-
core/metadata/src/impls/metadata.rs | 15 ++++--------
core/shard/src/lib.rs | 6 ++---
4 files changed, 66 insertions(+), 27 deletions(-)
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index bc6fec07b..5171b4aa1 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -1069,7 +1069,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
///
/// Returns a list of actions to take based on fired timeouts.
/// Empty vec means no actions needed.
- pub fn tick(&self, plane: PlaneKind, current_op: u64, current_commit: u64) -> Vec<VsrAction> {
+ pub fn tick(&self, plane: PlaneKind, current_op: u64) -> Vec<VsrAction> {
let mut actions = Vec::new();
let mut timeouts = self.timeouts.borrow_mut();
@@ -1091,11 +1091,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
if timeouts.fired(TimeoutKind::DoViewChangeMessage) {
drop(timeouts);
- actions.extend(self.handle_do_view_change_message_timeout(
- plane,
- current_op,
- current_commit,
- ));
+ actions.extend(self.handle_do_view_change_message_timeout(plane, current_op));
timeouts = self.timeouts.borrow_mut();
}
@@ -1206,7 +1202,6 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
&self,
plane: PlaneKind,
current_op: u64,
- current_commit: u64,
) -> Vec<VsrAction> {
if self.status.get() != Status::ViewChange {
return Vec::new();
@@ -1230,7 +1225,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
target: self.primary_index(self.view.get()),
log_view: self.log_view.get(),
op: current_op,
- commit: current_commit,
+ // commit_max, not commit_min: see `handle_start_view_change`.
+ commit: self.commit_max.get(),
namespace: self.namespace,
};
emit_sim_event(
@@ -1471,8 +1467,14 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
let primary_candidate = self.primary_index(self.view.get());
let current_op = self.sequencer.current_sequence();
- // DVC uses commit_min: the replica's actual execution progress.
- let current_commit = self.commit_min.get();
+ // DVC carries commit_max (highest known-committed), not commit_min
+ // (locally applied). The new primary floors its pipeline rebuild at
+ // max(commit) across the quorum; only commit_max bounds that range
+ // to pipeline depth (every replica holds op - commit_max <= depth).
+ // commit_min can lag far behind and overflow the rebuild. The
+ // committed-but-unapplied tail (commit_min..commit_max] is replayed
+ // by the new primary's CommitJournal, not the pipeline.
+ let current_commit = self.commit_max.get();
// Start DVC timeout
self.timeouts
@@ -1605,8 +1607,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
}
let current_op = self.sequencer.current_sequence();
- // Use commit_min: the replica's actual execution progress.
- let current_commit = self.commit_min.get();
+ // commit_max, not commit_min: see `handle_start_view_change`.
+ let current_commit = self.commit_max.get();
// If we haven't sent our own DVC yet, record it
if !self.sent_own_do_view_change.get() {
diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs
index e261b79ae..c00aa063c 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -456,7 +456,8 @@ pub async fn send_prepare_ok<B, P>(
#[cfg(test)]
mod tests {
use super::*;
- use crate::{Consensus, LocalPipeline};
+ use crate::{Consensus, LocalPipeline, VsrAction};
+ use iggy_binary_protocol::StartViewChangeHeader;
use message_bus::SendError;
use server_common::{MESSAGE_ALIGN, iobuf::Frozen};
@@ -536,6 +537,49 @@ mod tests {
assert!(replicate_preflight(&consensus, current.header()).is_ok());
}
+ // A DoViewChange must carry commit_max (highest known-committed), not
+ // commit_min (locally applied). The new primary floors its pipeline rebuild
+ // at the max DVC commit across the quorum, and only commit_max keeps that
+ // range within pipeline depth (every replica holds op - commit_max <= depth).
+ // commit_min can lag commit_max by more than the depth, so flooring on it
+ // would rebuild an over-deep range and panic the shard. Regression marker.
+ #[test]
+ fn do_view_change_carries_commit_max_not_commit_min() {
+ let consensus = VsrConsensus::new(1, 1, 3, 0, NoopBus, LocalPipeline::new());
+ consensus.init();
+
+ // Diverge the frontiers: applied (commit_min=5) lags known-committed
+ // (commit_max=13) by more than PIPELINE_PREPARE_QUEUE_MAX (8).
+ consensus.advance_commit_max(13);
+ for op in 1..=5 {
+ consensus.advance_commit_min(op);
+ }
+ assert_eq!(consensus.commit_min(), 5);
+ assert_eq!(consensus.commit_max(), 13);
+
+ // An SVC for a higher view from another replica moves this node into the
+ // view change; with f = 1 SVC excluding self, it emits its own DVC.
+ let svc =
+ Message::<StartViewChangeHeader>::new(std::mem::size_of::<StartViewChangeHeader>())
+ .transmute_header(|_, new: &mut StartViewChangeHeader| {
+ new.command = Command2::StartViewChange;
+ new.view = 1;
+ new.replica = 0;
+ new.namespace = 0;
+ });
+ let actions = consensus.handle_start_view_change(PlaneKind::Metadata, svc.header());
+
+ let dvc_commit = actions.iter().find_map(|action| match action {
+ VsrAction::SendDoViewChange { commit, .. } => Some(*commit),
+ _ => None,
+ });
+ assert_eq!(
+ dvc_commit,
+ Some(13),
+ "DVC must carry commit_max (13), not commit_min (5)"
+ );
+ }
+
#[test]
fn loopback_push_and_drain() {
let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus, LocalPipeline::new());
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 1fc2d994d..bbac9d2d7 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -1267,12 +1267,6 @@ where
prepare: Message<PrepareHeader>,
) -> Result<Message<ReplyHeader>, Canceled> {
consensus.verify_pipeline();
- // Snapshot (view, commit_min) pre-subscribe. Validate it after
- // `on_replicate` returns: another task could mutate either during the
- // await and silently invalidate the catch-up gate, with release builds
- // proceeding on a stale view-state.
- let view_snapshot = consensus.view();
- let commit_min_snapshot = consensus.commit_min();
let receiver = consensus.pipeline_message_with_subscriber(PlaneKind::Metadata, &prepare);
// Re-check gate post-subscribe: `pipeline_message_with_subscriber`
// can drop the borrow. No commit-max advance flips the gate today;
@@ -1281,11 +1275,12 @@ where
is_caught_up_primary(consensus),
"dispatch_prepare_and_await: gate flipped between check and dispatch"
);
+ // `on_replicate` awaits: a sibling in-process submit may commit
+ // (commit_min advances) or a view change may land (view advances) while
+ // parked here. Both are handled downstream - a view change drops the
+ // reply_sender so `receiver` resolves `Canceled`, and loopback acks are
+ // op-routed - so no post-await view/commit invariant holds or is needed.
self.on_replicate(prepare).await;
- debug_assert!(
- consensus.view() == view_snapshot && consensus.commit_min() == commit_min_snapshot,
- "dispatch_prepare_and_await: view/commit_min advanced across on_replicate await"
- );
let mut loopback = Vec::new();
consensus.drain_loopback_into(&mut loopback);
for message in loopback {
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 8dec40089..691f22cca 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -1673,8 +1673,7 @@ where
let consensus = partition.consensus();
let current_op = consensus.sequencer().current_sequence();
- let current_commit = consensus.commit_min();
- let actions = consensus.tick(PlaneKind::Partitions, current_op, current_commit);
+ let actions = consensus.tick(PlaneKind::Partitions, current_op);
dispatch_vsr_actions::<B, _, MJ>(consensus, None, &actions).await;
dispatch_partition_journal_actions(consensus, partition, &actions).await;
}
@@ -1697,8 +1696,7 @@ where
};
let current_op = consensus.sequencer().current_sequence();
- let current_commit = consensus.commit_min();
- let actions = consensus.tick(PlaneKind::Metadata, current_op, current_commit);
+ let actions = consensus.tick(PlaneKind::Metadata, current_op);
dispatch_vsr_actions(consensus, metadata.journal.as_ref(), &actions).await;
}