Skip to content

feat(subnet-spiltting): deliver Splitting batch to DSM #11213

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pierugo-dfinity wants to merge 3 commits into from
Open

feat(subnet-spiltting): deliver Splitting batch to DSM #11213

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
071b50d
feat: implement batch delivery
pierugo-dfinity Aug 19, 2026
39b4f5c
fix: clippy
pierugo-dfinity Aug 26, 2026
cc7de52
refactor: simplify & clarify arguments
pierugo-dfinity Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
245 changes: 208 additions & 37 deletions rs/consensus/src/consensus/batch_delivery.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ use crate::consensus::{
use ic_consensus_chain_key::ChainKeyPayloadBuilderImpl;
use ic_consensus_dkg::get_vetkey_public_keys;
use ic_consensus_idkg::utils::get_idkg_subnet_public_keys_and_pre_signatures;
use ic_consensus_utils::{membership::Membership, pool_reader::PoolReader};
use ic_consensus_utils::{membership::Membership, pool_reader::PoolReader, subnet_splitting};
use ic_error_types::RejectCode;
use ic_https_outcalls_consensus::payload_builder::CanisterHttpPayloadBuilderImpl;
use ic_interfaces::{
Expand All @@ -24,20 +24,18 @@ use ic_protobuf::{
registry::{crypto::v1::PublicKey as PublicKeyProto, subnet::v1::InitialNiDkgTranscriptRecord},
};
use ic_types::{
Height, PrincipalId, SubnetId,
Height, NodeId, PrincipalId, SubnetId,
batch::{
Batch, BatchContent, BatchMessages, BatchSummary, BlockmakerMetrics, CanisterHttpSpent,
ChainKeyData, ConsensusResponse,
},
consensus::{
Block, BlockPayload, HasVersion,
dkg::RemoteTranscriptResult,
idkg::{self},
},
crypto::randomness_from_crypto_hashable,
crypto::threshold_sig::{
ThresholdSigPublicKey,
ni_dkg::{NiDkgId, NiDkgTag, NiDkgTranscript},
consensus::{Block, BlockPayload, HasVersion, dkg::RemoteTranscriptResult, idkg},
crypto::{
randomness_from_crypto_hashable,
threshold_sig::{
ThresholdSigPublicKey,
ni_dkg::{NiDkgId, NiDkgTag, NiDkgTranscript},
},
},
messages::{CallbackId, Payload, RejectContext},
};
Expand All @@ -46,46 +44,73 @@ use std::collections::BTreeMap;
/// Deliver all finalized blocks from
/// `message_routing.expected_batch_height` to `finalized_height` via
/// `MessageRouting` and return the last delivered batch height.
pub fn deliver_batches(
///
/// To be used exclusively by the ic-replay tool
pub fn deliver_batches_for_ic_replay(
message_routing: &dyn MessageRouting,
membership: &Membership,
pool: &PoolReader<'_>,
registry_client: &dyn RegistryClient,
subnet_id: SubnetId,
log: &ReplicaLogger,
// This argument should only be used by the ic-replay tool. If it is set to `None`, we will
// deliver all batches until the finalized height. If it is set to `Some(h)`, we will
// deliver all bathes up to the height `min(h, finalized_height)`.
subnet_id: SubnetId,
// If set to `None`, we will deliver all batches until the finalized height.
// If set to `Some(h)`, we will deliver all bathes up to the height `min(h, finalized_height)`.
max_batch_height_to_deliver: Option<Height>,
) -> Result<Height, MessageRoutingError> {
deliver_batches_with_result_processor(
deliver_batches(
message_routing,
membership,
pool,
registry_client,
subnet_id,
log,
/*maybe_node_id=*/ None,
subnet_id,
max_batch_height_to_deliver,
/*result_processor=*/ None,
/*result_processor=*/ |_, _, _| {},
)
}

/// Deliver all finalized blocks from
/// `message_routing.expected_batch_height` to `finalized_height` via
/// `MessageRouting` and return the last delivered batch height.
#[allow(clippy::type_complexity)]
pub(crate) fn deliver_batches_with_result_processor(
///
/// To be called by the finalizer.
pub(crate) fn deliver_batches_for_finalizer(
message_routing: &dyn MessageRouting,
membership: &Membership,
pool: &PoolReader<'_>,
registry_client: &dyn RegistryClient,
log: &ReplicaLogger,
node_id: NodeId,
subnet_id: SubnetId,
result_processor: impl FnMut(&Result<(), MessageRoutingError>, BlockStats, BatchStats),
) -> Result<Height, MessageRoutingError> {
deliver_batches(
message_routing,
membership,
pool,
registry_client,
log,
Some(node_id),
subnet_id,
/*max_batch_height_to_deliver=*/ None,
result_processor,
)
}

/// Deliver all finalized blocks from
/// `message_routing.expected_batch_height` to `finalized_height` via
/// `MessageRouting` and return the last delivered batch height.
fn deliver_batches(
message_routing: &dyn MessageRouting,
membership: &Membership,
pool: &PoolReader<'_>,
registry_client: &dyn RegistryClient,
log: &ReplicaLogger,
// This argument should only be used by the ic-replay tool. If it is set to `None`, we will
// deliver all batches until the finalized height. If it is set to `Some(h)`, we will
// deliver all bathes up to the height `min(h, finalized_height)`.
maybe_node_id: Option<NodeId>,
subnet_id: SubnetId,
max_batch_height_to_deliver: Option<Height>,
result_processor: Option<&dyn Fn(&Result<(), MessageRoutingError>, BlockStats, BatchStats)>,
mut result_processor: impl FnMut(&Result<(), MessageRoutingError>, BlockStats, BatchStats),
) -> Result<Height, MessageRoutingError> {
let finalized_height = pool.get_finalized_height();
// If `max_batch_height_to_deliver` is specified and smaller than
Expand Down Expand Up @@ -225,13 +250,50 @@ pub(crate) fn deliver_batches_with_result_processor(
let persist_batch = Some(height) == max_batch_height_to_deliver;
let requires_full_state_hash = block.payload.is_summary() || persist_batch;
let batch_content = match block.payload.as_ref() {
BlockPayload::Summary(_summary_payload) => BatchContent::Data {
batch_messages: BatchMessages::default(),
chain_key_data,
consensus_responses,
canister_http_spent,
requires_full_state_hash,
},
BlockPayload::Summary(_summary_payload) => {
if let Some(scheduled) = subnet_splitting::is_split_scheduled(&block) {
let node_id =
maybe_node_id.expect("Subnet splitting not yet supported in ic-replay");
Comment thread
pierugo-dfinity marked this conversation as resolved.
let subnet_splitting::PostSplitAssignment {
new_subnet_id,
other_subnet_id,
} = match subnet_splitting::get_post_split_subnet_assignment(
node_id,
&block,
registry_client,
scheduled,
) {
Ok(assignment) => assignment,
Err(err) => {
warn!(
every_n_seconds => 30,
log,
"Error getting new subnet assignment: {}",
err
);
break;
}
};

info!(
log,
"Delivering splitting block. New subnet assignment: {}", new_subnet_id
);

BatchContent::Splitting {
new_subnet_id,
other_subnet_id,
}
Comment on lines +283 to +286

@pierugo-dfinity pierugo-dfinity Aug 19, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, but according to this comment, this will be overwritten as soon as the destination subnet starts a new round after the split.
Maybe something worth looking into: both the Scheduled summary and the PostSplit summary have the same registry version, so DSM might use the cached OwnSubnetInfo by mistake. But because destination replicas are restarted before executing anything and I assume this cache is stored only in memory, this shouldn't be a problem.

cc @alin-at-dfinity for confirmation

@alin-at-dfinity alin-at-dfinity Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will work as is, as per your analysis.

We can (and probably should) make it clearer by explicitly resetting it to default for subnet B in online_split() instead of carrying it over. Not perfect, as it will result in (apparently valid) disabled features and default limits, but it will work better as documentation (we really expect this to be populated, or else).

@pierugo-dfinity pierugo-dfinity Aug 20, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Another alternative is to actually fill it with the proper values by reading the registry at the batch's registry version (which is precisely the version at which the split happened). But we'd then have two individual locations where the registry is read, not perfect either.

@alin-at-dfinity alin-at-dfinity Aug 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess you could also have BatchProcessorImpl::process_batch() already check for BatchContent::Splitting before calling read_registry() and using new_subnet_id instead of own_subnet_id as its argument. This would also transparently address any future subnet-specific registry data cached into the state.

But unless anyone absolutely requires the list of nodes in the certified state immediately after a subnet split, I would be perfectly fine with just wiping own_subnet_info for subnet B during a split.

@pierugo-dfinity pierugo-dfinity Aug 28, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So keeping it as is? (currently own_subnet_info is not wiped, it keeps the source subnet's info before getting overwritten)

@pierugo-dfinity pierugo-dfinity Aug 28, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@alin-at-dfinity

@alin-at-dfinity alin-at-dfinity Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To clarify, you can safely leave it as it is. Or you can additionally reset own_subnet_info in the split() implementation, purely for documentation purposes (with a one-line comment stating that it won't be used before it is read again in the next execution round; or something to that effect).

I'm fine either way.

} else {
BatchContent::Data {
batch_messages: BatchMessages::default(),
chain_key_data,
consensus_responses,
canister_http_spent,
requires_full_state_hash,
}
}
}
BlockPayload::Data(data_payload) => {
batch_stats.add_from_payload(&data_payload.batch);
BatchContent::Data {
Expand Down Expand Up @@ -301,9 +363,7 @@ pub(crate) fn deliver_batches_with_result_processor(
};

let result = message_routing.deliver_batch(batch);
if let Some(f) = result_processor {
f(&result, block_stats, batch_stats);
}
result_processor(&result, block_stats, batch_stats);
if let Err(err) = result {
warn!(every_n_seconds => 5, log, "Batch delivery failed: {:?}", err);
return Err(err);
Expand Down Expand Up @@ -578,16 +638,19 @@ mod tests {
//! Finalizer unit tests
use super::*;
use crate::consensus::batch_delivery::generate_responses_to_remote_dkgs;
use ic_consensus_mocks::{Dependencies, DependenciesBuilder};
use ic_crypto_test_utils_ni_dkg::dummy_transcript_for_tests;
use ic_logger::replica_logger::no_op_logger;
use ic_management_canister_types_private::{SetupInitialDKGResponse, VetKdCurve, VetKdKeyId};
use ic_test_utilities::message_routing::FakeMessageRouting;
use ic_test_utilities_registry::SubnetRecordBuilder;
use ic_test_utilities_types::ids::{subnet_test_id, test_replica_version};
use ic_types::{
PrincipalId, RegistryVersion, SubnetId,
batch::{BatchPayload, ValidationContext},
consensus::{
DataPayload, Payload as ConsensusPayload, Rank,
dkg::{DkgDataPayload, RemoteTranscriptResult},
DataPayload, HashedBlock, Payload as ConsensusPayload, Rank,
dkg::{DkgDataPayload, RemoteTranscriptResult, SplittingArgs, SubnetSplittingStatus},
},
crypto::{
CryptoHash, CryptoHashOf,
Expand All @@ -596,10 +659,16 @@ mod tests {
},
},
messages::{CallbackId, Payload},
replica_config::ReplicaConfig,
time::UNIX_EPOCH,
};
use ic_types_test_utils::ids::{NODE_1, NODE_2, NODE_3, NODE_4, SUBNET_1, SUBNET_2};
use rstest::rstest;
use std::str::FromStr;

const SOURCE_SUBNET_ID: SubnetId = SUBNET_1;
const DESTINATION_SUBNET_ID: SubnetId = SUBNET_2;

const TARGET_ID: NiDkgTargetId = NiDkgTargetId::new([8; 32]);

const EXPECTED_FRESH_SUBNET_ID_STR: &str =
Expand Down Expand Up @@ -783,4 +852,106 @@ mod tests {
SubnetId::from(PrincipalId::from_str(EXPECTED_FRESH_SUBNET_ID_STR).unwrap())
);
}

#[rstest]
#[case::node_on_source_subnet(NODE_1, SOURCE_SUBNET_ID, DESTINATION_SUBNET_ID)]
#[case::node_on_destination_subnet(NODE_4, DESTINATION_SUBNET_ID, SOURCE_SUBNET_ID)]
fn test_deliver_splitting_batch(
#[case] node_id: NodeId,
#[case] expected_new_subnet_id: SubnetId,
#[case] expected_other_subnet_id: SubnetId,
) {
ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| {
const SPLITTING_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(2);
const INTERVAL_LENGTH: u64 = 9;
let summary_height = Height::from(INTERVAL_LENGTH + 1);

let Dependencies {
mut pool,
membership,
registry,
..
} = DependenciesBuilder::multiple_subnets(
pool_config,
vec![
(
1,
SOURCE_SUBNET_ID,
SubnetRecordBuilder::from(&[NODE_1, NODE_2, NODE_3, NODE_4])
.with_dkg_interval_length(INTERVAL_LENGTH)
.build(),
),
(
SPLITTING_REGISTRY_VERSION.get(),
SOURCE_SUBNET_ID,
SubnetRecordBuilder::from(&[NODE_1, NODE_3])
.with_dkg_interval_length(INTERVAL_LENGTH)
.build(),
),
(
SPLITTING_REGISTRY_VERSION.get(),
DESTINATION_SUBNET_ID,
SubnetRecordBuilder::from(&[NODE_2, NODE_4])
.with_dkg_interval_length(INTERVAL_LENGTH)
.build(),
),
],
)
.with_replica_config(ReplicaConfig {
node_id: NODE_1,
subnet_id: SOURCE_SUBNET_ID,
replica_version: test_replica_version(),
})
.build();

pool.advance_round_normal_operation_n(INTERVAL_LENGTH);

let mut proposal = pool.make_next_block();
let block = proposal.content.as_mut();
block.context.registry_version = SPLITTING_REGISTRY_VERSION;
let mut payload = block.payload.as_ref().as_summary().clone();
payload.dkg.subnet_splitting_status = SubnetSplittingStatus::Scheduled(SplittingArgs {
source_subnet_id: SOURCE_SUBNET_ID,
destination_subnet_id: DESTINATION_SUBNET_ID,
});
block.payload = ConsensusPayload::new(
ic_types::crypto::crypto_hash,
BlockPayload::Summary(payload),
);
proposal.content = HashedBlock::new(ic_types::crypto::crypto_hash, block.clone());
pool.insert_validated(proposal.clone());
pool.notarize(&proposal);
pool.finalize(&proposal);
pool.insert_random_tape(summary_height);

let message_routing = FakeMessageRouting::new();
*message_routing.next_batch_height.write().unwrap() = summary_height;

let result = deliver_batches(
&message_routing,
&membership,
&PoolReader::new(&pool),
registry.as_ref(),
&no_op_logger(),
Some(node_id),
SOURCE_SUBNET_ID,
None,
|_, _, _| {},
);

assert_eq!(result, Ok(summary_height));
let batches = message_routing.batches.read().unwrap();
assert_eq!(batches.len(), 1);
match &batches[0].content {
BatchContent::Splitting {
new_subnet_id,
other_subnet_id,
} => {
assert_eq!(*new_subnet_id, expected_new_subnet_id);
assert_eq!(*other_subnet_id, expected_other_subnet_id);
}
other => panic!("Expected BatchContent::Splitting, got: {other:?}"),
}
})
}
}
12 changes: 6 additions & 6 deletions rs/consensus/src/consensus/finalizer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
//! into a complete finalization, at which point the block and its ancestors
//! become finalized.
use crate::consensus::{
batch_delivery::deliver_batches_with_result_processor,
batch_delivery::deliver_batches_for_finalizer,
metrics::{BatchStats, BlockStats, FinalizerMetrics},
};
use ic_consensus_utils::{
Expand Down Expand Up @@ -95,17 +95,17 @@ impl Finalizer {
}

// Try to deliver finalized batches to messaging
let _ = deliver_batches_with_result_processor(
let _ = deliver_batches_for_finalizer(
&*self.message_routing,
&self.membership,
pool,
&*self.registry_client,
self.replica_config.subnet_id,
&self.log,
None,
Some(&|result, block_stats, batch_stats| {
self.replica_config.node_id,
self.replica_config.subnet_id,
|result, block_stats, batch_stats| {
self.process_batch_delivery_result(result, block_stats, batch_stats)
}),
},
);

// Try to finalize rounds from finalized_height + 1 up to (and including)
Expand Down
6 changes: 3 additions & 3 deletions rs/replay/src/player.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ use ic_artifact_pool::{
consensus_pool::{ConsensusPoolImpl, UncachedConsensusPoolImpl},
};
use ic_config::{Config, artifact_pool::ArtifactPoolConfig, subnet_config::SubnetConfig};
use ic_consensus::consensus::batch_delivery::deliver_batches;
use ic_consensus::consensus::batch_delivery::deliver_batches_for_ic_replay;
use ic_consensus_certification::VerifierImpl;
use ic_consensus_utils::{lookup_replica_version, membership::Membership, pool_reader::PoolReader};
use ic_crypto_for_verification_only::CryptoComponentForVerificationOnly;
Expand Down Expand Up @@ -712,13 +712,13 @@ impl Player {
) -> Height {
let expected_batch_height = message_routing.expected_batch_height();
let last_batch_height = loop {
match deliver_batches(
match deliver_batches_for_ic_replay(
message_routing,
membership,
pool,
&*self.registry,
self.subnet_id,
&self.log,
self.subnet_id,
replay_target_height,
) {
Ok(h) => break h,
Expand Down
Loading