Add StateSource enum and thread trie_node_writer through ChainSync

Introduce StateSource enum to properly signal state import mode:
- TrieNodes::AlreadyImported: trie nodes written incrementally, skip import
- TrieNodes::Pending: trie nodes accumulated in memory, write to DB
- KeyValues: legacy key-value import path

This eliminates 30GiB+ memory usage during state sync by:
1. Using StateSyncMode enum with Incremental and Accumulated variants
2. Skipping key-value accumulation in Incremental mode
3. Signaling to client that trie nodes are already imported

Thread trie_node_writer through ChainSync:
- Add trie_node_writer field to ChainSync struct
- Add parameter to ChainSync::new() constructor
- Pass trie_node_writer to StateSync in chain sync state initiation
- Update PolkadotSyncingStrategy to thread parameter through
- Update all test cases with None parameter

Author: liuchengxu

substrate/client/api/src/backend.rs

@@ -655,6 +655,10 @@ pub trait Backend<Block: BlockT>: AuxStore + Send + Sync {
 		expected_state_root: Block::Hash,
 	) -> sp_blockchain::Result<()>;
 
+	/// Verify that the state root node exists in the database.
+	/// Used after incremental state sync to confirm state is complete.
+	fn verify_state_root_exists(&self, root: Block::Hash) -> sp_blockchain::Result<()>;
+
 	/// Attempts to revert the chain by `n` blocks. If `revert_finalized` is set it will attempt to
 	/// revert past any finalized block, this is unsafe and can potentially leave the node in an
 	/// inconsistent state. All blocks higher than the best block are also reverted and not counting

substrate/client/api/src/in_mem.rs

@@ -758,6 +758,11 @@ impl<Block: BlockT> backend::Backend<Block> for Backend<Block> {
 		unimplemented!("Not needed for in-mem backend")
 	}
 
+	fn verify_state_root_exists(&self, _root: Block::Hash) -> sp_blockchain::Result<()> {
+		// In-memory backend doesn't support incremental state sync
+		unimplemented!("Not needed for in-mem backend")
+	}
+
 	fn revert(
 		&self,
 		_n: NumberFor<Block>,

substrate/client/consensus/common/src/block_import.rs

@@ -126,24 +126,50 @@ pub enum StorageChanges<Block: BlockT> {
 	Import(ImportedState<Block>),
 }
 
-/// Imported state data. A vector of key-value pairs that should form a trie.
+/// State of trie nodes for import.
+#[derive(Clone, Debug)]
+pub enum TrieNodeStates {
+	/// Trie nodes pending import - need to be written to database.
+	Pending(Vec<(Vec<u8>, Vec<u8>)>),
+	/// Trie nodes were already imported to database during state sync.
+	AlreadyImported {
+		/// Number of trie nodes written during state sync.
+		nodes_count: u64,
+	},
+}
+
+/// Source of state data for import.
+#[derive(Clone)]
+pub enum StateSource {
+	/// Import state from key-value pairs (legacy path).
+	/// Trie nodes will be computed via delta_trie_root.
+	KeyValues(sp_state_machine::KeyValueStates),
+	/// Import state from trie nodes directly.
+	TrieNodes(TrieNodeStates),
+}
+
+impl std::fmt::Debug for StateSource {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		match self {
+			Self::KeyValues(state) =>
+				f.debug_struct("KeyValues").field("count", &state.0.len()).finish(),
+			Self::TrieNodes(state) => f.debug_tuple("TrieNodes").field(state).finish(),
+		}
+	}
+}
+
+/// Imported state data.
 #[derive(Clone)]
 pub struct ImportedState<B: BlockT> {
 	/// Target block hash.
 	pub block: B::Hash,
-	/// State keys and values.
-	pub state: sp_state_machine::KeyValueStates,
-	/// Optional trie nodes from compact proofs.
-	/// When present, these are written directly to the STATE column,
-	/// avoiding the need to recompute trie nodes via delta_trie_root.
-	/// Contains raw (prefixed_key, value) pairs extracted from MemoryDBs.
-	pub trie_nodes: Option<Vec<(Vec<u8>, Vec<u8>)>>,
+	/// Source of state data.
+	pub source: StateSource,
 }
 
 impl<B: BlockT> PartialEq for ImportedState<B> {
 	fn eq(&self, other: &Self) -> bool {
-		// Compare block and state only; trie_nodes are derived from state
-		self.block == other.block && self.state == other.state
+		self.block == other.block
 	}
 }
 
@@ -153,7 +179,7 @@ impl<B: BlockT> std::fmt::Debug for ImportedState<B> {
 	fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
 		fmt.debug_struct("ImportedState")
 			.field("block", &self.block)
-			.field("has_trie_nodes", &self.trie_nodes.is_some())
+			.field("source", &self.source)
 			.finish()
 	}
 }

substrate/client/consensus/common/src/import_queue.rs

@@ -320,7 +320,7 @@ pub(crate) async fn verify_single_block_metered<B: BlockT, V: Verifier<B>>(
 		} else {
 			debug!(target: LOG_TARGET, "Header {} was not provided ", block.hash);
 		}
-		return Err(BlockImportError::IncompleteHeader(peer))
+		return Err(BlockImportError::IncompleteHeader(peer));
 	};
 
 	trace!(target: LOG_TARGET, "Header {} has {:?} logs", block.hash, header.digest().logs().len());
@@ -348,7 +348,7 @@ pub(crate) async fn verify_single_block_metered<B: BlockT, V: Verifier<B>>(
 		BlockImportStatus::ImportedUnknown { .. } => (),
 		r => {
 			// Any other successful result means that the block is already imported.
-			return Ok(SingleBlockVerificationOutcome::Imported(r))
+			return Ok(SingleBlockVerificationOutcome::Imported(r));
 		},
 	}
 

substrate/client/consensus/common/src/import_queue/basic_queue.rs

@@ -127,7 +127,7 @@ impl<B: BlockT> BasicQueueHandle<B> {
 impl<B: BlockT> ImportQueueService<B> for BasicQueueHandle<B> {
 	fn import_blocks(&mut self, origin: BlockOrigin, blocks: Vec<IncomingBlock<B>>) {
 		if blocks.is_empty() {
-			return
+			return;
 		}
 
 		trace!(target: LOG_TARGET, "Scheduling {} blocks for import", blocks.len());
@@ -195,7 +195,7 @@ impl<B: BlockT> ImportQueue<B> for BasicQueue<B> {
 		loop {
 			if let Err(_) = self.result_port.next_action(link).await {
 				log::error!(target: "sync", "poll_actions: Background import task is no longer alive");
-				return
+				return;
 			}
 		}
 	}
@@ -237,7 +237,7 @@ async fn block_import_process<B: BlockT>(
 					target: LOG_TARGET,
 					"Stopping block import because the import channel was closed!",
 				);
-				return
+				return;
 			},
 		};
 
@@ -304,7 +304,7 @@ impl<B: BlockT> BlockImportWorker<B> {
 						target: LOG_TARGET,
 						"Stopping block import because result channel was closed!",
 					);
-					return
+					return;
 				}
 
 				// Make sure to first process all justifications
@@ -317,13 +317,13 @@ impl<B: BlockT> BlockImportWorker<B> {
 								target: LOG_TARGET,
 								"Stopping block import because justification channel was closed!",
 							);
-							return
+							return;
 						},
 					}
 				}
 
 				if let Poll::Ready(()) = futures::poll!(&mut block_import_process) {
-					return
+					return;
 				}
 
 				// All futures that we polled are now pending.
@@ -423,7 +423,7 @@ async fn import_many_blocks<B: BlockT, V: Verifier<B>>(
 			Some(b) => b,
 			None => {
 				// No block left to import, success!
-				return ImportManyBlocksResult { block_count: count, imported, results }
+				return ImportManyBlocksResult { block_count: count, imported, results };
 			},
 		};
 

substrate/client/consensus/common/src/import_queue/buffered_link.rs

@@ -160,7 +160,7 @@ impl<B: BlockT> BufferedLinkReceiver<B> {
 	pub async fn next_action(&mut self, link: &dyn Link<B>) -> Result<(), ()> {
 		if let Some(msg) = self.rx.next().await {
 			self.send_actions(msg, link);
-			return Ok(())
+			return Ok(());
 		}
 		Err(())
 	}

substrate/client/consensus/common/src/lib.rs

@@ -25,7 +25,7 @@ pub mod metrics;
 pub use block_import::{
 	BlockCheckParams, BlockImport, BlockImportParams, ForkChoiceStrategy, ImportResult,
 	ImportedAux, ImportedState, JustificationImport, JustificationSyncLink, StateAction,
-	StorageChanges,
+	StateSource, StorageChanges, TrieNodeStates,
 };
 pub use import_queue::{
 	import_single_block, BasicQueue, BlockImportError, BlockImportStatus, BoxBlockImport,

substrate/client/consensus/common/src/longest_chain.rs

@@ -99,7 +99,7 @@ where
 					"Requested a finality target using max number {} below the base number {}",
 					max_number, base_number
 				);
-				return Err(Application(msg.into()))
+				return Err(Application(msg.into()));
 			}
 
 			while current_head.number() > &max_number {
@@ -116,7 +116,7 @@ where
 					"Requested a finality target using a base {:?} not in the best chain {:?}",
 					base_hash, best_hash,
 				);
-				return Err(Application(msg.into()))
+				return Err(Application(msg.into()));
 			}
 			let current_hash = *current_head.parent_hash();
 			current_head = blockchain

substrate/client/db/src/lib.rs

@@ -2376,6 +2376,19 @@ impl<Block: BlockT> sc_client_api::backend::Backend<Block> for Backend<Block> {
 		Ok(())
 	}
 
+	fn verify_state_root_exists(&self, root: Block::Hash) -> ClientResult<()> {
+		if self.storage.db.get(columns::STATE, root.as_ref()).is_none() {
+			return Err(sp_blockchain::Error::Backend(format!(
+				"State root {root:?} not found in database after incremental state sync"
+			)));
+		}
+		log::info!(
+			target: "state",
+			"Verified state root {root:?} exists in database"
+		);
+		Ok(())
+	}
+
 	fn revert(
 		&self,
 		n: NumberFor<Block>,