feat: publisher plugin hardening

.github/workflows/release.yml

@@ -0,0 +1,150 @@
+name: Release
+
+on:
+  release:
+    types: [published]
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}
+
+jobs:
+  build-and-publish:
+    name: Build and Publish Release
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          ref: ${{ github.event.release.tag_name }}
+
+      - name: Set up QEMU for multi-arch builds
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Log in to GitHub Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata for Docker
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          tags: |
+            type=semver,pattern={{version}},value=${{ github.event.release.tag_name }}
+            type=semver,pattern={{major}}.{{minor}},value=${{ github.event.release.tag_name }}
+            type=semver,pattern={{major}},value=${{ github.event.release.tag_name }}
+            type=raw,value=latest,enable=${{ !github.event.release.prerelease }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          platforms: linux/amd64,linux/arm64
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Install cross-compilation tools
+        run: |
+          cargo install cross --git https://github.com/cross-rs/cross
+
+      - name: Install dependencies (protoc)
+        run: |
+          sudo apt-get update -y
+          sudo apt-get install -y protobuf-compiler
+
+      - name: Build Linux x86_64 binary
+        run: cross build --release --target x86_64-unknown-linux-gnu --locked
+        working-directory: app/rock-node
+
+      - name: Build Linux ARM64 binary
+        run: cross build --release --target aarch64-unknown-linux-gnu --locked
+        working-directory: app/rock-node
+
+      - name: Package binaries
+        run: |
+          mkdir -p release-artifacts
+
+          # Package x86_64
+          cd target/x86_64-unknown-linux-gnu/release
+          tar -czf ../../../release-artifacts/rock-node-${{ github.event.release.tag_name }}-linux-x86_64.tar.gz rock-node
+          cd -
+
+          # Package ARM64
+          cd target/aarch64-unknown-linux-gnu/release
+          tar -czf ../../../release-artifacts/rock-node-${{ github.event.release.tag_name }}-linux-arm64.tar.gz rock-node
+          cd -
+
+          # Generate checksums
+          cd release-artifacts
+          sha256sum * > checksums.txt
+          cd -
+
+      - name: Upload Release Assets
+        uses: softprops/action-gh-release@v2
+        with:
+          tag_name: ${{ github.event.release.tag_name }}
+          files: |
+            release-artifacts/rock-node-${{ github.event.release.tag_name }}-linux-x86_64.tar.gz
+            release-artifacts/rock-node-${{ github.event.release.tag_name }}-linux-arm64.tar.gz
+            release-artifacts/checksums.txt
+          append_body: true
+          body: |
+
+            ---
+
+            ## 🐳 Docker Images
+
+            Multi-architecture images published to GitHub Container Registry:
+
+            ```bash
+            # Pull specific version
+            docker pull ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.event.release.tag_name }}
+
+            # Pull latest
+            docker pull ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest
+            ```
+
+            **Platforms**: `linux/amd64`, `linux/arm64`
+
+            ## 📦 Standalone Binaries
+
+            Download and verify checksums:
+
+            ```bash
+            # Set version
+            VERSION=${{ github.event.release.tag_name }}
+
+            # Download for your platform
+            wget https://github.com/${{ github.repository }}/releases/download/${VERSION}/rock-node-${VERSION}-linux-x86_64.tar.gz
+
+            # Download checksums
+            wget https://github.com/${{ github.repository }}/releases/download/${VERSION}/checksums.txt
+
+            # Verify integrity
+            sha256sum -c checksums.txt --ignore-missing
+
+            # Extract and run
+            tar -xzf rock-node-${VERSION}-linux-x86_64.tar.gz
+            ./rock-node --config-path config/config.toml
+            ```
+
+            **Available Binaries**:
+            - `rock-node-${{ github.event.release.tag_name }}-linux-x86_64.tar.gz` (Intel/AMD 64-bit)
+            - `rock-node-${{ github.event.release.tag_name }}-linux-arm64.tar.gz` (ARM 64-bit)

config/config.e2e.toml

@@ -34,6 +34,8 @@ persistence_ack_timeout_seconds = 60
 stale_winner_timeout_seconds = 60
 winner_cleanup_interval_seconds = 300
 winner_cleanup_threshold_blocks = 1000
+max_message_size_bytes = 33554432  # 32MB
+max_items_per_set = 10000
 
 [plugins.subscriber_service]
 enabled = true

config/config.toml

@@ -5,7 +5,7 @@
 log_level = "INFO"
 database_path = "data/database"
 start_block_number = 0
-grpc_address = "127.0.0.1"
+grpc_address = "0.0.0.0"
 grpc_port = 8090
 
 # =========================================
@@ -15,7 +15,7 @@ grpc_port = 8090
 
     [plugins.observability]
     enabled = true
-    listen_address = "127.0.0.1:9600"
+    listen_address = "0.0.0.0:9600"
 
     [plugins.publish_service]
     enabled = true
@@ -32,7 +32,7 @@ grpc_port = 8090
     archive_batch_size = 100
 
     [plugins.state_management_service]
-    enabled = true
+    enabled = false
 
     [plugins.subscriber_service]
     enabled = true
@@ -42,7 +42,7 @@ grpc_port = 8090
     session_timeout_seconds = 60
 
     [plugins.verification_service]
-    enabled = false
+    enabled = true
 
     [plugins.block_access_service]
     enabled = true
@@ -51,10 +51,10 @@ grpc_port = 8090
     enabled = true
 
     [plugins.query_service]
-    enabled = true
+    enabled = false
 
     [plugins.backfill]
-    enabled = true
+    enabled = false
     mode = "GapFill"
     peers = ["http://127.0.0.1:8090"]
     check_interval_seconds = 5

crates/rock-node-backfill-plugin/src/worker.rs

@@ -259,7 +259,7 @@ impl BackfillWorker {
         }
     }
 
-    async fn run_continuous_stream_cycle(self: &Arc<Self>, shuffled_peers: &mut Vec<String>) {
+    async fn run_continuous_stream_cycle(self: &Arc<Self>, shuffled_peers: &mut [String]) {
         shuffled_peers.shuffle(&mut rand::rng());
 
         for peer_addr in shuffled_peers.iter() {

crates/rock-node-block-access-plugin/src/service.rs

@@ -24,7 +24,7 @@ fn code_to_string(code: block_response::Code) -> &'static str {
         block_response::Code::NotFound => "NotFound",
         block_response::Code::NotAvailable => "NotAvailable",
         block_response::Code::Unknown => "Unknown",
-        block_response::Code::Error => todo!(),
+        block_response::Code::Error => "Error",
     }
 }
 
@@ -207,6 +207,7 @@ impl BlockAccessServiceImpl {
         }
     }
 
+    #[allow(clippy::result_large_err)]
     fn record_metrics(
         &self,
         response: BlockResponse,
@@ -572,6 +573,6 @@ mod tests {
             "NotAvailable"
         );
         assert_eq!(code_to_string(block_response::Code::Unknown), "Unknown");
-        // Note: Error case will panic in debug mode, so we don't test it here
+        assert_eq!(code_to_string(block_response::Code::Error), "Error");
     }
 }

crates/rock-node-core/src/config.rs

@@ -105,6 +105,10 @@ pub struct PublishServiceConfig {
     pub stale_winner_timeout_seconds: u64,
     pub winner_cleanup_interval_seconds: u64,
     pub winner_cleanup_threshold_blocks: u64,
+    /// Maximum gRPC message size in bytes (default 32MB)
+    pub max_message_size_bytes: usize,
+    /// Maximum items per BlockItemSet
+    pub max_items_per_set: usize,
 }
 
 impl Default for PublishServiceConfig {
@@ -116,6 +120,8 @@ impl Default for PublishServiceConfig {
             stale_winner_timeout_seconds: 10,
             winner_cleanup_interval_seconds: 150,
             winner_cleanup_threshold_blocks: 1000,
+            max_message_size_bytes: 32 * 1024 * 1024, // 32MB
+            max_items_per_set: 10000,
         }
     }
 }

crates/rock-node-core/src/metrics.rs

@@ -35,6 +35,7 @@ pub struct MetricsRegistry {
     pub publish_header_to_proof_duration_seconds: HistogramVec,
     pub publish_average_header_to_proof_time_seconds: GaugeVec,
     pub publish_responses_sent_total: CounterVec,
+    pub publish_session_duration_seconds: HistogramVec,
 
     // --- Persistence Plugin Metrics ---
     pub persistence_writes_total: CounterVec,
@@ -206,6 +207,16 @@ impl MetricsRegistry {
         )?;
         registry.register(Box::new(publish_responses_sent_total.clone()))?;
 
+        let publish_session_duration_seconds = HistogramVec::new(
+            Opts::new(
+                "rocknode_publish_session_duration_seconds",
+                "Duration of publisher sessions from connection to disconnection.",
+            )
+            .into(),
+            &[],
+        )?;
+        registry.register(Box::new(publish_session_duration_seconds.clone()))?;
+
         let persistence_writes_total = CounterVec::new(
             Opts::new(
                 "rocknode_persistence_writes_total",
@@ -385,6 +396,7 @@ impl MetricsRegistry {
             publish_header_to_proof_duration_seconds,
             publish_average_header_to_proof_time_seconds,
             publish_responses_sent_total,
+            publish_session_duration_seconds,
             persistence_writes_total,
             persistence_write_duration_seconds,
             persistence_event_duration_seconds,

crates/rock-node-publish-plugin/Cargo.toml

@@ -17,6 +17,7 @@ tonic = "0.13.1"
 
 # For easy error handling
 anyhow = "1"
+thiserror = "1"
 
 # For logging
 tracing = "0.1"

crates/rock-node-publish-plugin/src/error.rs

@@ -0,0 +1,43 @@
+use thiserror::Error;
+
+/// Errors that can occur during publish operations
+#[derive(Error, Debug)]
+pub enum PublishError {
+    #[error("Block item set is empty")]
+    EmptyBlockItems,
+
+    #[error("Missing block header for new block {block_number}")]
+    MissingBlockHeader { block_number: i64 },
+
+    #[error("Too many items in set: {count} (max: {max})")]
+    TooManyItems { count: usize, max: usize },
+
+    #[allow(dead_code)]
+    #[error("Block number mismatch: expected {expected}, got {received}")]
+    BlockNumberMismatch { expected: i64, received: i64 },
+
+    #[allow(dead_code)]
+    #[error("Invalid block number: {block_number}")]
+    InvalidBlockNumber { block_number: i64 },
+
+    #[allow(dead_code)]
+    #[error("Session {session_id} received items without active block")]
+    NoActiveBlock { session_id: uuid::Uuid },
+
+    #[error("Failed to encode block: {0}")]
+    Encoding(#[from] prost::EncodeError),
+
+    #[allow(dead_code)]
+    #[error("Channel send error")]
+    ChannelSend,
+
+    #[allow(dead_code)]
+    #[error("Verification timeout for block {block_number}")]
+    VerificationTimeout { block_number: u64 },
+
+    #[allow(dead_code)]
+    #[error("Verification failed for block {block_number}: {reason}")]
+    VerificationFailed { block_number: u64, reason: String },
+}
+
+pub type Result<T> = std::result::Result<T, PublishError>;

crates/rock-node-publish-plugin/src/lib.rs

@@ -23,9 +23,11 @@ use std::{
 use tonic::service::RoutesBuilder;
 use tracing::{debug, info, warn};
 
+mod error;
 mod service;
 mod session_manager;
 mod state;
+mod validation;
 
 #[derive(Debug, Default)]
 pub struct PublishPlugin {
@@ -134,9 +136,14 @@ impl Plugin for PublishPlugin {
             context: context.clone(),
             shared_state: self.shared_state.as_ref().unwrap().clone(),
         };
-        const MAX_MESSAGE_SIZE: usize = 1024 * 1024 * 32;
+        let max_message_size = context
+            .config
+            .plugins
+            .publish_service
+            .max_message_size_bytes;
         let server = BlockStreamPublishServiceServer::new(service)
-            .max_decoding_message_size(MAX_MESSAGE_SIZE);
+            .max_decoding_message_size(max_message_size)
+            .max_encoding_message_size(max_message_size);
 
         builder.add_service(server);
         Ok(true)

crates/rock-node-publish-plugin/src/service.rs

@@ -53,6 +53,8 @@ impl BlockStreamPublishService for PublishServiceImpl {
                     }
                 }
             }
+            // Log session statistics before cleanup
+            session_manager.log_session_stats();
             info!(%session_id, "Session handler task finished.");
 
             // When the session finishes, it may have been the recorded winner for