Add ledger arena for warp sync

Cargo.toml

@@ -242,6 +242,7 @@ sp-runtime-interface = { default-features = false, git = "https://github.com/par
 sc-chain-spec = { default-features = false, git = "https://github.com/paritytech/polkadot-sdk.git", tag = "polkadot-stable2603" }
 sc-consensus = { default-features = false, git = "https://github.com/paritytech/polkadot-sdk.git", tag = "polkadot-stable2603" }
 sc-network = { default-features = false, git = "https://github.com/paritytech/polkadot-sdk.git", tag = "polkadot-stable2603" }
+sc-network-sync = { default-features = false, git = "https://github.com/paritytech/polkadot-sdk.git", tag = "polkadot-stable2603" }
 sc-executor = { default-features = false, git = "https://github.com/paritytech/polkadot-sdk.git", tag = "polkadot-stable2603" }
 sc-consensus-grandpa = { default-features = false, git = "https://github.com/paritytech/polkadot-sdk.git", tag = "polkadot-stable2603" }
 sc-consensus-aura = { default-features = false, git = "https://github.com/paritytech/polkadot-sdk.git", tag = "polkadot-stable2603" }

changes/added/ledger-warp-sync.md

@@ -0,0 +1,6 @@
+# Add Ledger Sync process for Warp Sync
+
+Adds support for syncing ledger state, while warp syncing. Should enable Substrate warp sync.
+
+PR: https://github.com/midnightntwrk/midnight-node/pull/1650
+Issue: https://github.com/midnightntwrk/midnight-node/issues/1648

ledger/src/lib.rs

@@ -154,6 +154,164 @@ pub fn drop_all_default_storage() {
 	ledger_9::storage::drop_default_storage_if_exists();
 }
 
+/// Parse the `vNN` from a `ledger-state[vNN]` tag embedded in a tagged blob (a `StateKey` or a
+/// genesis_state). Used to dispatch warp serialize/import (and genesis-init) to the ledger module
+/// whose `LedgerState` serialization matches: **v5 → `ledger_7`, v13 → `ledger_8`,
+/// v16/v17 → `ledger_9`**.
+/// A warp-syncing node can target a chain governed by an *older* ledger version than this build's
+/// latest (e.g. a real devnet whose arena is still v13), so the version is read from the data, not
+/// assumed to be the tip's.
+#[cfg(feature = "std")]
+pub fn ledger_state_tag_version(tagged: &[u8]) -> Option<u32> {
+	const NEEDLE: &[u8] = b"ledger-state[v";
+	let start = tagged.windows(NEEDLE.len()).position(|w| w == NEEDLE)? + NEEDLE.len();
+	let rest = &tagged[start..];
+	let end = rest.iter().position(|&b| b == b']')?;
+	core::str::from_utf8(&rest[..end]).ok()?.parse().ok()
+}
+
+/// Expand to the `(DbSeparate, DbUnified)`-parameterized call of a `Bridge` arena method on the given
+/// ledger version module (`ledger_7`/`ledger_8`/`ledger_9`), picking the DB instantiation by `unified`.
+#[cfg(feature = "std")]
+macro_rules! bridge_arena_call {
+	($ver:ident, $unified:expr, $method:ident ( $($arg:expr),* )) => {{
+		type DbSeparate = $ver::ledger_storage_local::db::ParityDb;
+		type DbUnified = $ver::ledger_storage_local::db::ParityDb<
+			sha2::Sha256,
+			$ver::ledger_storage_local::db::paritydb::OwnedDb,
+			{ midnight_primitives_ledger::LedgerStorageExt::COLUMN_OFFSET },
+		>;
+		if $unified {
+			$ver::Bridge::<$ver::TransactionSignature, DbUnified>::$method( $($arg),* )
+		} else {
+			$ver::Bridge::<$ver::TransactionSignature, DbSeparate>::$method( $($arg),* )
+		}
+	}};
+}
+
+/// Serialize the ledger arena snapshot at `state_key` into the canonical, `Ledger`-rooted warp
+/// transfer blob (trustless warp ledger-sync, server side). `unified` selects the ParityDb
+/// instantiation (separate = column offset 0; unified = offset `NUM_COLUMNS_POLKADOT`); the blob is
+/// identical across modes.
+///
+/// Dispatches to the ledger module matching the `StateKey`'s `ledger-state[vNN]` tag (see
+/// [`ledger_state_tag_version`]) — so a warp node can serve an arena governed by an older ledger
+/// version than this build's latest. Error rendered to `String` (the underlying `LedgerApiError` is
+/// version-specific).
+#[cfg(feature = "std")]
+pub fn serialize_ledger_snapshot(unified: bool, state_key: &[u8]) -> Result<Vec<u8>, String> {
+	match ledger_state_tag_version(state_key) {
+		Some(16 | 17) => {
+			bridge_arena_call!(ledger_9, unified, serialize_ledger_snapshot(state_key))
+				.map_err(|e| format!("{e:?}"))
+		},
+		Some(13) => bridge_arena_call!(ledger_8, unified, serialize_ledger_snapshot(state_key))
+			.map_err(|e| format!("{e:?}")),
+		Some(5) => bridge_arena_call!(ledger_7, unified, serialize_ledger_snapshot(state_key))
+			.map_err(|e| format!("{e:?}")),
+		other => Err(format!("unsupported ledger-state version {other:?} in StateKey")),
+	}
+}
+
+/// Whether the local ledger arena holds the ledger state `state_key` points to (the `Ledger` root
+/// node is readable). Cheap — a single arena root lookup, no DAG traversal.
+///
+/// Used by warp ledger-sync's recovery monitor to decide whether arena recovery is needed at all:
+/// a node restarted *after* a completed recovery (or a normally full-synced node) already has the
+/// state and must not re-fetch or re-gate; a node restarted *mid*-recovery does not, and must.
+/// Returns `false` for an unsupported/undecodable `StateKey` (recovery will then verify against it
+/// and fail loudly rather than silently skipping).
+#[cfg(feature = "std")]
+pub fn has_ledger_state(unified: bool, state_key: &[u8]) -> bool {
+	match ledger_state_tag_version(state_key) {
+		Some(16 | 17) => {
+			bridge_arena_call!(ledger_9, unified, get_ledger_state_root(state_key)).is_ok()
+		},
+		Some(13) => bridge_arena_call!(ledger_8, unified, get_ledger_state_root(state_key)).is_ok(),
+		Some(5) => bridge_arena_call!(ledger_7, unified, get_ledger_state_root(state_key)).is_ok(),
+		_ => false,
+	}
+}
+
+/// Failure modes of [`import_verified_ledger_snapshot`]. All are non-fatal to the chain: the caller
+/// discards the data, reports the peer, and retries from another.
+#[cfg(feature = "std")]
+#[derive(Debug)]
+pub enum SnapshotImportError {
+	/// The on-chain `StateKey` bytes failed to decode to a `TypedArenaKey<Ledger>` (the inner
+	/// `LedgerApiError` is version-specific, so it is rendered to a string here).
+	StateKeyDecode(String),
+	/// The transferred blob failed the arena's native (multi-pass, untrusted-safe) deserialization
+	/// — malformed, truncated, or internally inconsistent node graph.
+	Deserialize(std::io::Error),
+	/// The blob deserialized cleanly but its recomputed root key does **not** equal the on-chain
+	/// `StateKey`: the peer served a different (or tampered) ledger. **Never persisted.**
+	RootMismatch,
+}
+
+#[cfg(feature = "std")]
+impl core::fmt::Display for SnapshotImportError {
+	fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+		match self {
+			SnapshotImportError::StateKeyDecode(e) => {
+				write!(f, "failed to decode on-chain StateKey: {e}")
+			},
+			SnapshotImportError::Deserialize(e) => {
+				write!(f, "failed to deserialize ledger snapshot: {e}")
+			},
+			SnapshotImportError::RootMismatch => {
+				write!(f, "ledger snapshot root key does not match on-chain StateKey")
+			},
+		}
+	}
+}
+
+#[cfg(feature = "std")]
+impl std::error::Error for SnapshotImportError {}
+
+#[cfg(feature = "std")]
+/// Verify a `Ledger`-rooted warp snapshot `blob` against the on-chain `expected_state_key` and, on
+/// success, persist it into the already-open arena backend so `get_lazy(StateKey)` resolves (warp
+/// ledger-sync verification + import). `unified` selects the DB instantiation, and dispatch on the
+/// `StateKey`'s `ledger-state[vNN]` tag picks the ledger module, as in
+/// [`serialize_ledger_snapshot`].
+///
+/// The caller must hold the authoring/import gate (the arena is single-writer).
+pub fn import_verified_ledger_snapshot(
+	unified: bool,
+	blob: &[u8],
+	expected_state_key: &[u8],
+) -> Result<(), SnapshotImportError> {
+	// Dispatch on the `StateKey`'s ledger-state version (the underlying method returns the shared
+	// `SnapshotImportError` for every version, so no error mapping is needed).
+	match ledger_state_tag_version(expected_state_key) {
+		Some(16 | 17) => {
+			bridge_arena_call!(
+				ledger_9,
+				unified,
+				import_verified_ledger_snapshot(blob, expected_state_key)
+			)
+		},
+		Some(13) => {
+			bridge_arena_call!(
+				ledger_8,
+				unified,
+				import_verified_ledger_snapshot(blob, expected_state_key)
+			)
+		},
+		Some(5) => {
+			bridge_arena_call!(
+				ledger_7,
+				unified,
+				import_verified_ledger_snapshot(blob, expected_state_key)
+			)
+		},
+		other => Err(SnapshotImportError::StateKeyDecode(format!(
+			"unsupported ledger-state version {other:?} in StateKey"
+		))),
+	}
+}
+
 mod common;
 
 pub mod types {

ledger/src/versions/common/mod.rs

@@ -745,6 +745,82 @@ where
 		api.serialize(&ledger_state.as_typed_key())
 	}
 
+	/// Serialize the full ledger arena snapshot at `state_key` into the canonical, `Ledger`-rooted
+	/// transfer blob used by trustless warp ledger-sync: `derived_tag_prefix ‖
+	/// TopoSortedNodes(Ledger DAG)`.
+	///
+	/// Mirrors the single-pass technique of the toolkit's `serialize_ledger_state_fast`, but roots
+	/// at `Ledger` (the `Sp` from `get_ledger` is an `Sp<Ledger>`) rather than `LedgerState`.
+	/// Because the blob is rooted at `Ledger`, its recomputed content-address root key equals the
+	/// on-chain `pallet_midnight::StateKey`, which is exactly what the client verifies against. The
+	/// tag prefix is **derived** (`GLOBAL_TAG ‖ <Ledger as Tagged>::tag()`), never hardcoded, so it
+	/// stays in lockstep with the ledger serialization format.
+	pub fn serialize_ledger_snapshot(state_key: &[u8]) -> Result<Vec<u8>, LedgerApiError> {
+		use ledger_storage_local::arena::TopoSortedNodes;
+		use midnight_serialize_local::{GLOBAL_TAG, Serializable};
+		use types::SerializationError;
+
+		let api = api::new();
+		let ledger = Self::get_ledger(&api, state_key)?;
+
+		// One `serialize_to_node_list()` pass (the derived `Serializable` impl would do two — once
+		// for `serialized_size`, once for `serialize` — each a full topo-sort of a multi-million
+		// node DAG), written directly. Byte-identical to the default impl's output.
+		let nodes: TopoSortedNodes = ledger.serialize_to_node_list();
+		let tag_prefix = format!("{}{}:", GLOBAL_TAG, <Ledger<D> as Tagged>::tag());
+		let mut bytes = Vec::with_capacity(tag_prefix.len() + nodes.serialized_size());
+		bytes.extend_from_slice(tag_prefix.as_bytes());
+		nodes.serialize(&mut bytes).map_err(|e| {
+			log::error!(target: LOG_TARGET, "Failed to serialize ledger snapshot: {e:?}");
+			LedgerApiError::Serialization(SerializationError::LedgerState)
+		})?;
+		Ok(bytes)
+	}
+
+	/// Import a verified, `Ledger`-rooted warp snapshot `blob` into the already-open arena backend,
+	/// binding it to the trie anchor `expected_state_key` (the on-chain `pallet_midnight::StateKey`
+	/// the warp-recovered trie already holds).
+	///
+	/// Reconstruction uses the arena's **native multi-pass deserializer**
+	/// (`Arena::deserialize_sp`, designed for untrusted wire input — it re-hashes every node), then
+	/// asserts the reconstructed root key equals `expected_state_key` before persisting. So a
+	/// malicious or faulty peer can at worst cause a rejected import (→ peer report + retry by the
+	/// caller), never state corruption.
+	///
+	/// Persists + flushes into the live `default_storage` so `get_lazy(StateKey)` resolves —
+	/// in-process, no restart, via the same `alloc`/`persist`/`flush` path live block execution
+	/// uses. The caller (warp client driver) MUST hold the authoring/import gate so no block
+	/// executes against the arena concurrently — the arena is single-writer.
+	pub fn import_verified_ledger_snapshot(
+		blob: &[u8],
+		expected_state_key: &[u8],
+	) -> Result<(), crate::SnapshotImportError> {
+		use crate::SnapshotImportError;
+
+		let api = api::new();
+		let expected: TypedArenaKey<Ledger<D>, D::Hasher> = api
+			.tagged_deserialize(expected_state_key)
+			.map_err(|e| SnapshotImportError::StateKeyDecode(format!("{e:?}")))?;
+
+		// Native verifying (untrusted-safe) deserialize of the `Ledger`-rooted blob into the live
+		// arena; re-allocating the loaded value yields the persistable `Sp`.
+		let ledger: Ledger<D> =
+			helpers_local::deserialize(blob).map_err(SnapshotImportError::Deserialize)?;
+		let mut sp = default_storage::<D>().arena.alloc(ledger);
+
+		// Cryptographic bind to the trie anchor: the reconstructed root must equal the on-chain
+		// `StateKey`. This is the whole security argument — reject anything else.
+		let computed: TypedArenaKey<Ledger<D>, D::Hasher> = sp.as_typed_key();
+		if computed != expected {
+			return Err(SnapshotImportError::RootMismatch);
+		}
+
+		sp.persist();
+		default_storage::<D>().with_backend(|backend| backend.flush_all_changes_to_db());
+		log::info!(target: LOG_TARGET, "Imported verified ledger snapshot ({} bytes)", blob.len());
+		Ok(())
+	}
+
 	pub fn get_unclaimed_amount(
 		state_key: &[u8],
 		beneficiary: &[u8],

node/Cargo.toml

@@ -25,6 +25,8 @@ serde_json = { workspace = true, features = ["preserve_order"] }
 serde.workspace = true
 async-trait.workspace = true
 futures.workspace = true
+# Must match the version sc-network's `request_response_config` inbound queue expects (1.9).
+async-channel = "1.9"
 
 sc-cli.workspace = true
 sc-chain-spec.workspace = true
@@ -48,6 +50,8 @@ sc-consensus-slots.workspace = true
 sc-offchain.workspace = true
 sc-client-api.workspace = true
 sc-network.workspace = true
+sc-network-sync.workspace = true
+sp-consensus.workspace = true
 sc-utils.workspace = true
 sp-consensus-aura.workspace = true
 sp-consensus-beefy.workspace = true

node/src/backend/custom_parity_db.rs

@@ -107,20 +107,67 @@ pub fn open<H: Clone + AsRef<[u8]>>(
 
 	let db = Arc::new(parity_db::Db::open_or_create(&config)?);
 
+	// Dispatch genesis-arena-init on the genesis_state's `ledger-state[vNN]` tag, not this build's
+	// latest version. A network genesis'd with an older ledger version (e.g. a real devnet whose
+	// genesis arena is still v13/ledger_8) must be initialised with the matching deserializer, or the
+	// init panics with a tag-version mismatch. (This is also what lets a fresh node warp-sync onto such
+	// a network: the genesis arena is set up under the right version, then warp recovery overwrites it
+	// with the verified target arena.)
+	let genesis_state = &storage_config.genesis_state;
+	let genesis_version = midnight_node_ledger::ledger_state_tag_version(genesis_state);
 	match storage_config.separation {
 		StorageSeparation::Separate => {
-			midnight_node_ledger::ledger_9::storage::init_storage_paritydb_separate(
-				&storage_config.db_path,
-				&storage_config.genesis_state,
-				storage_config.cache_size,
-			);
+			let dir = &storage_config.db_path;
+			let cache = storage_config.cache_size;
+			match genesis_version {
+				Some(16 | 17) => {
+					midnight_node_ledger::ledger_9::storage::init_storage_paritydb_separate(
+						dir,
+						genesis_state,
+						cache,
+					);
+				},
+				Some(13) => {
+					midnight_node_ledger::ledger_8::storage::init_storage_paritydb_separate(
+						dir,
+						genesis_state,
+						cache,
+					);
+				},
+				Some(5) => {
+					midnight_node_ledger::ledger_7::storage::init_storage_paritydb_separate(
+						dir,
+						genesis_state,
+						cache,
+					);
+				},
+				other => panic!("unsupported genesis ledger-state version {other:?}"),
+			}
 			Ok((OwnedDb(db), LedgerStorageDb::SeparateDb(storage_config.db_path.clone())))
 		},
 		StorageSeparation::Unified => {
-			midnight_node_ledger::ledger_9::storage::init_storage_paritydb_unified::<
-				_,
-				NUM_COLUMNS_POLKADOT,
-			>(OwnedDb(db.clone()), &storage_config.genesis_state, storage_config.cache_size);
+			let cache = storage_config.cache_size;
+			match genesis_version {
+				Some(16 | 17) => {
+					midnight_node_ledger::ledger_9::storage::init_storage_paritydb_unified::<
+						_,
+						NUM_COLUMNS_POLKADOT,
+					>(OwnedDb(db.clone()), genesis_state, cache);
+				},
+				Some(13) => {
+					midnight_node_ledger::ledger_8::storage::init_storage_paritydb_unified::<
+						_,
+						NUM_COLUMNS_POLKADOT,
+					>(OwnedDb(db.clone()), genesis_state, cache);
+				},
+				Some(5) => {
+					midnight_node_ledger::ledger_7::storage::init_storage_paritydb_unified::<
+						_,
+						NUM_COLUMNS_POLKADOT,
+					>(OwnedDb(db.clone()), genesis_state, cache);
+				},
+				other => panic!("unsupported genesis ledger-state version {other:?}"),
+			}
 			Ok((OwnedDb(db.clone()), LedgerStorageDb::UnifiedDb(db.clone())))
 		},
 	}

node/src/lib.rs

@@ -38,3 +38,4 @@ pub mod service;
 pub mod sidechain_params_cmd;
 pub mod subscription_bounds;
 mod util;
+pub mod warp_ledger_sync;