Fixes Issues w/ Stream Crashing Over Large Replay

networks/movement/movement-full-node/src/da/stream_blocks/mod.rs

@@ -27,7 +27,7 @@ impl StreamBlocks {
 			.await
 			.context("Failed to stream blocks from DA")?;
 
-		println!("Streaming blocks from DA");
+		tracing::info!("Streaming blocks from DA");
 
 		while let Some(block_res) = blocks_from_da.next().await {
 			let response = block_res.context("Failed to get block")?;
@@ -43,13 +43,17 @@ impl StreamBlocks {
 				blob_response::BlobType::PassedThroughBlob(blob) => {
 					(blob.data, blob.timestamp, blob.blob_id, blob.height)
 				}
+				blob_response::BlobType::HeartbeatBlob(_) => {
+					tracing::info!("Receive heartbeat blob");
+					continue;
+				}
 				_ => {
 					return Err(anyhow::anyhow!("Unknown blob type"));
 				}
 			};
 
 			// pretty print (with labels) the block_id, block_timestamp, and da_height
-			println!(
+			tracing::info!(
 				"Block ID: {}, Block Timestamp: {}, DA Height: {}",
 				hex::encode(block_id),
 				// unix date string from the block timestamp which is in microseconds

networks/movement/movement-full-node/src/node/tasks/execute_settle.rs

@@ -116,6 +116,10 @@ where
 			blob_response::BlobType::PassedThroughBlob(blob) => {
 				(blob.data, blob.timestamp, blob.blob_id, blob.height)
 			}
+			blob_response::BlobType::HeartbeatBlob(_) => {
+				// Do nothing.
+				return Ok(());
+			}
 			_ => anyhow::bail!("Invalid blob type"),
 		};
 

process-compose/movement-full-node/process-compose.test-followers.yml

@@ -3,6 +3,15 @@ version: "3"
 processes:
 
   setup:
+    environment:
+      - "ETH_RPC_CONNECTION_PROTOCOL=http"
+      - "ETH_RPC_CONNECTION_HOSTNAME=0.0.0.0"
+      - "ETH_RPC_CONNECTION_PORT=8090"
+      - "ETH_WS_CONNECTION_PROTOCOL=ws"
+      - "ETH_WS_CONNECTION_HOSTNAME=0.0.0.0"
+      - "ETH_WS_CONNECTION_PORT=8090"
+      - "MAYBE_RUN_LOCAL=true"
+      - "MOVEMENT_DA_LIGHT_NODE_HTTP1=true"
     command: |
       export AWS_REGION=us-west-2
       export MOVEMENT_SYNC="leader::follower-test-$MOVEMENT_SHARED_RANDOM_1<=>{default_signer_address_whitelist,maptos,maptos-storage,movement-da-db}/**"

proto/movementlabs/protocol_units/da/light_node/v1beta1.proto

@@ -17,6 +17,7 @@ message BlobResponse {
       Blob passed_through_blob = 1;
       Blob sequenced_blob_intent = 2;
       Blob sequenced_blob_block = 3;
+      Blob heartbeat_blob = 4;
     }
 }
 

protocol-units/da/movement/protocol/da/src/lib.rs

@@ -130,6 +130,22 @@ pub trait DaOperations: Send + Sync {
 
 							last_height = height;
 						}
+						// Already executed Height are use to send Heartbeat.
+						Ok(Certificate::Height(height)) => {
+							//old certificate, use to send Heartbeat block.
+							let blob_stream = self
+								.stream_da_blobs_between_heights(height, height)
+								.await?;
+							tokio::pin!(blob_stream);
+
+							while let Some(blob_res) = blob_stream.next().await {
+								let (_, blob) = blob_res?;
+								// Ack use heigth zero to identify heart beat block.
+								// Should be changed to a type.
+								let heart_blob = (DaHeight(0u64), blob);
+								yield heart_blob;
+							}
+						}
 						Ok(Certificate::Nolo) => {
 							// Ignore Nolo
 						}
@@ -142,9 +158,9 @@ pub trait DaOperations: Send + Sync {
 							yield Err(e)?;
 						}
 						// If height is less than last height, ignore
-						_ => {
-							warn!("ignoring certificate");
-						}
+						// _ => {
+						// 	warn!("ignoring certificate");
+						// }
 					}
 				}
 			};

protocol-units/da/movement/protocol/da/src/mock/mod.rs

@@ -210,6 +210,7 @@ pub mod test {
 			vec![
 				Ok(1),                                          // First certificate
 				Err("internal error: fatal error".to_string()), // Fatal error
+				Ok(2),
 			]
 		);
 

protocol-units/da/movement/protocol/light-node/src/passthrough.rs

@@ -133,8 +133,14 @@ where
 
 			while let Some(blob) = blob_stream.next().await {
 				let (height, da_blob) = blob.map_err(|e| tonic::Status::internal(e.to_string()))?;
-				let verifed_blob = verifier.verify(da_blob, height.as_u64()).await.map_err(|e| tonic::Status::internal(e.to_string()))?;
-				let blob = verifed_blob.into_inner().to_blob_passed_through_read_response(height.as_u64()).map_err(|e| tonic::Status::internal(e.to_string()))?;
+				let blob = if height.as_u64() == 0 {
+					//Heart beat blob
+					// No need to verify the data are removed.
+					da_blob.to_blob_heartbeat_response()
+				} else {
+					let verifed_blob = verifier.verify(da_blob, height.as_u64()).await.map_err(|e| tonic::Status::internal(e.to_string()))?;
+					verifed_blob.into_inner().to_blob_passed_through_read_response(height.as_u64()).map_err(|e| tonic::Status::internal(e.to_string()))?
+				};
 				let response = StreamReadFromHeightResponse {
 					blob: Some(blob)
 				};

protocol-units/da/movement/protocol/light-node/src/sequencer.rs

@@ -279,6 +279,7 @@ where
 		let sequenced_block = match blob_type {
 			BlobType::PassedThroughBlob(blob) => BlobType::SequencedBlobBlock(blob),
 			BlobType::SequencedBlobBlock(blob) => BlobType::SequencedBlobBlock(blob),
+			BlobType::HeartbeatBlob(blob) => BlobType::HeartbeatBlob(blob),
 			_ => {
 				anyhow::bail!("Invalid blob type")
 			}

protocol-units/da/movement/protocol/util/src/blob/ir/blob.rs

@@ -141,6 +141,20 @@ impl DaBlob {
 		Ok(BlobResponse { blob_type: Some(blob_response::BlobType::SequencedBlobBlock(blob)) })
 	}
 
+	/// Converts a [DaBlob] into a [BlobResponse] with the blob passed through.
+	pub fn to_blob_heartbeat_response(self) -> BlobResponse {
+		//for heartbeat blob the data are removed.
+		let blob = Blob {
+			data: vec![],
+			signature: self.signature().to_vec(),
+			timestamp: self.timestamp(),
+			signer: self.signer().to_vec(),
+			blob_id: self.id().to_vec(),
+			height: 0,
+		};
+		BlobResponse { blob_type: Some(blob_response::BlobType::HeartbeatBlob(blob)) }
+	}
+
 	/// Converts a [DaBlob] into a [BlobResponse] with the blob passed through.
 	pub fn to_blob_passed_through_read_response(
 		self,