feat: add grpc socket and flattn tx batches to allow for lower allocations (#3297)

* add grpc socket and flattn tx batches to allow for lower allocations

* redo proto

* docs: update changelog for grpc execution transport

* remove extra txs

Author: tac0turtle

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changes
 
 - Optimization of mutex usage in cache for reaper [#3286](https://github.com/evstack/ev-node/pull/3286)
+- Add Unix domain socket support for gRPC execution endpoints via `unix:///path/to/socket` [#3297](https://github.com/evstack/ev-node/pull/3297)
+- **BREAKING:** Replace legacy gRPC execution `txs` payload fields with `tx_batch` so clients and servers use contiguous transaction buffers [#3297](https://github.com/evstack/ev-node/pull/3297)
 - Optimize metadata writes by making it async in cache store [#3298](https://github.com/evstack/ev-node/pull/3298)
 - Reduce tx cache retention to avoid OOM under (really) heavy tx load [#3299](https://github.com/evstack/ev-node/pull/3299)
 

apps/grpc/README.md

@@ -1,13 +1,13 @@
 # gRPC Single Sequencer App
 
-This application runs a Evolve node with a single sequencer that connects to a remote execution client via gRPC. It allows you to use any execution layer that implements the Evolve execution gRPC interface.
+This application runs an Evolve node with a single sequencer that connects to an execution client via gRPC. It allows you to use any execution layer that implements the Evolve execution gRPC interface.
 
 ## Overview
 
 The gRPC single sequencer app provides:
 
-- A Evolve consensus node with single sequencer
-- Connection to remote execution clients via gRPC
+- An Evolve consensus node with single sequencer
+- Connection to execution clients via TCP or Unix domain socket gRPC
 - Full data availability layer integration
 - P2P networking capabilities
 
@@ -58,11 +58,20 @@ Start the Evolve node with:
   --da.auth-token your-da-token
 ```
 
+For a same-machine executor, use a Unix domain socket endpoint:
+
+```bash
+./evgrpc start \
+  --root-dir ~/.evgrpc \
+  --grpc-executor-url unix:///tmp/evolve-executor.sock \
+  --da.address http://localhost:7980
+```
+
 ## Command-Line Flags
 
 ### gRPC-specific Flags
 
-- `--grpc-executor-url`: URL of the gRPC execution service (default: `http://localhost:50051`)
+- `--grpc-executor-url`: URL of the gRPC execution service, either `http://host:port` or `unix:///path/to/socket` (default: `http://localhost:50051`)
 
 ### Common Evolve Flags
 

apps/grpc/cmd/run.go

@@ -28,7 +28,7 @@ import (
 
 const (
 	grpcDbName = "grpc-single"
-	// FlagGrpcExecutorURL is the flag for the gRPC executor endpoint
+	// FlagGrpcExecutorURL is the flag for the gRPC executor endpoint.
 	FlagGrpcExecutorURL = "grpc-executor-url"
 )
 
@@ -163,11 +163,10 @@ func createGRPCExecutionClient(cmd *cobra.Command) (execution.Executor, error) {
 		return nil, fmt.Errorf("%s flag is required", FlagGrpcExecutorURL)
 	}
 
-	// Create and return the gRPC client
-	return executiongrpc.NewClient(executorURL), nil
+	return executiongrpc.NewClient(executorURL)
 }
 
 // addGRPCFlags adds flags specific to the gRPC execution client
 func addGRPCFlags(cmd *cobra.Command) {
-	cmd.Flags().String(FlagGrpcExecutorURL, "http://localhost:50051", "URL of the gRPC execution service")
+	cmd.Flags().String(FlagGrpcExecutorURL, "http://localhost:50051", "URL of the gRPC execution service, or unix:///path/to/executor.sock")
 }

buf.yaml

@@ -14,3 +14,6 @@ lint:
 breaking:
   use:
     - FILE
+  ignore_only:
+    FIELD_NO_DELETE:
+      - proto/evnode/v1/execution.proto

client/crates/types/src/proto/evnode.v1.messages.rs

@@ -76,6 +76,19 @@ pub struct SignedHeader {
     #[prost(message, optional, tag = "3")]
     pub signer: ::core::option::Option<Signer>,
 }
+/// DAHeaderEnvelope is a wrapper around SignedHeader for DA submission.
+/// It is binary compatible with SignedHeader (fields 1-3) but adds an envelope signature.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct DaHeaderEnvelope {
+    #[prost(message, optional, tag = "1")]
+    pub header: ::core::option::Option<Header>,
+    #[prost(bytes = "vec", tag = "2")]
+    pub signature: ::prost::alloc::vec::Vec<u8>,
+    #[prost(message, optional, tag = "3")]
+    pub signer: ::core::option::Option<Signer>,
+    #[prost(bytes = "vec", tag = "4")]
+    pub envelope_signature: ::prost::alloc::vec::Vec<u8>,
+}
 /// Signer is a signer of a block in the blockchain.
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct Signer {
@@ -139,6 +152,28 @@ pub struct Vote {
     #[prost(bytes = "vec", tag = "5")]
     pub validator_address: ::prost::alloc::vec::Vec<u8>,
 }
+/// P2PSignedHeader
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct P2pSignedHeader {
+    #[prost(message, optional, tag = "1")]
+    pub header: ::core::option::Option<Header>,
+    #[prost(bytes = "vec", tag = "2")]
+    pub signature: ::prost::alloc::vec::Vec<u8>,
+    #[prost(message, optional, tag = "3")]
+    pub signer: ::core::option::Option<Signer>,
+    #[prost(uint64, optional, tag = "4")]
+    pub da_height_hint: ::core::option::Option<u64>,
+}
+/// P2PData
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct P2pData {
+    #[prost(message, optional, tag = "1")]
+    pub metadata: ::core::option::Option<Metadata>,
+    #[prost(bytes = "vec", repeated, tag = "2")]
+    pub txs: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
+    #[prost(uint64, optional, tag = "3")]
+    pub da_height_hint: ::core::option::Option<u64>,
+}
 /// State is the state of the blockchain.
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct State {
@@ -159,6 +194,24 @@ pub struct State {
     #[prost(bytes = "vec", tag = "9")]
     pub last_header_hash: ::prost::alloc::vec::Vec<u8>,
 }
+/// RaftBlockState represents a replicated block state
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct RaftBlockState {
+    #[prost(uint64, tag = "1")]
+    pub height: u64,
+    #[prost(uint64, tag = "2")]
+    pub last_submitted_da_header_height: u64,
+    #[prost(uint64, tag = "3")]
+    pub last_submitted_da_data_height: u64,
+    #[prost(bytes = "vec", tag = "4")]
+    pub hash: ::prost::alloc::vec::Vec<u8>,
+    #[prost(uint64, tag = "5")]
+    pub timestamp: u64,
+    #[prost(bytes = "vec", tag = "6")]
+    pub header: ::prost::alloc::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "7")]
+    pub data: ::prost::alloc::vec::Vec<u8>,
+}
 /// SequencerDACheckpoint tracks the position in the DA where transactions were last processed
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct SequencerDaCheckpoint {
@@ -212,6 +265,17 @@ pub struct Batch {
     #[prost(bytes = "vec", repeated, tag = "1")]
     pub txs: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
 }
+/// BlockData contains data retrieved from a single DA height.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct BlockData {
+    #[prost(uint64, tag = "1")]
+    pub height: u64,
+    /// Unix timestamp in nanoseconds
+    #[prost(int64, tag = "2")]
+    pub timestamp: i64,
+    #[prost(bytes = "vec", repeated, tag = "3")]
+    pub blobs: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
+}
 /// InitChainRequest contains the genesis parameters for chain initialization
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct InitChainRequest {
@@ -231,28 +295,32 @@ pub struct InitChainResponse {
     /// Hash representing initial state
     #[prost(bytes = "vec", tag = "1")]
     pub state_root: ::prost::alloc::vec::Vec<u8>,
-    /// Maximum allowed bytes for transactions in a block
-    #[prost(uint64, tag = "2")]
-    pub max_bytes: u64,
 }
 /// GetTxsRequest is the request for fetching transactions
 ///
 /// Empty for now, may include filtering criteria in the future
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct GetTxsRequest {}
+/// TxBatch stores ordered transactions in one contiguous bytes buffer.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct TxBatch {
+    /// Concatenated transaction bytes.
+    #[prost(bytes = "vec", tag = "1")]
+    pub data: ::prost::alloc::vec::Vec<u8>,
+    /// Byte length for each transaction in order.
+    #[prost(uint32, repeated, tag = "2")]
+    pub tx_sizes: ::prost::alloc::vec::Vec<u32>,
+}
 /// GetTxsResponse contains the available transactions
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct GetTxsResponse {
-    /// Slice of valid transactions from mempool
-    #[prost(bytes = "vec", repeated, tag = "1")]
-    pub txs: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
+    /// Valid transactions from mempool in contiguous batch form.
+    #[prost(message, optional, tag = "2")]
+    pub tx_batch: ::core::option::Option<TxBatch>,
 }
 /// ExecuteTxsRequest contains transactions and block context for execution
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ExecuteTxsRequest {
-    /// Ordered list of transactions to execute
-    #[prost(bytes = "vec", repeated, tag = "1")]
-    pub txs: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
     /// Height of block being created (must be > 0)
     #[prost(uint64, tag = "2")]
     pub block_height: u64,
@@ -262,6 +330,9 @@ pub struct ExecuteTxsRequest {
     /// Previous block's state root hash
     #[prost(bytes = "vec", tag = "4")]
     pub prev_state_root: ::prost::alloc::vec::Vec<u8>,
+    /// Ordered transactions to execute in contiguous batch form.
+    #[prost(message, optional, tag = "5")]
+    pub tx_batch: ::core::option::Option<TxBatch>,
 }
 /// ExecuteTxsResponse contains the result of transaction execution
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
@@ -285,6 +356,73 @@ pub struct SetFinalRequest {
 /// Empty response, errors are returned via gRPC status
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct SetFinalResponse {}
+/// GetExecutionInfoRequest requests execution layer parameters
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetExecutionInfoRequest {}
+/// GetExecutionInfoResponse contains execution layer parameters
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetExecutionInfoResponse {
+    /// Maximum gas allowed for transactions in a block
+    /// For non-gas-based execution layers, this should be 0
+    #[prost(uint64, tag = "1")]
+    pub max_gas: u64,
+}
+/// FilterTxsRequest contains transactions to validate and filter
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct FilterTxsRequest {
+    /// Maximum cumulative size allowed (0 means no size limit)
+    #[prost(uint64, tag = "2")]
+    pub max_bytes: u64,
+    /// Maximum cumulative gas allowed (0 means no gas limit)
+    #[prost(uint64, tag = "3")]
+    pub max_gas: u64,
+    /// Whether force-included transactions are present
+    #[prost(bool, tag = "4")]
+    pub has_force_included_transaction: bool,
+    /// All transactions (force-included + mempool) in contiguous batch form.
+    #[prost(message, optional, tag = "5")]
+    pub tx_batch: ::core::option::Option<TxBatch>,
+}
+/// FilterTxsResponse contains the filter status for each transaction
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct FilterTxsResponse {
+    /// Filter status for each transaction (same length as txs in request)
+    #[prost(enumeration = "FilterStatus", repeated, tag = "1")]
+    pub statuses: ::prost::alloc::vec::Vec<i32>,
+}
+/// FilterStatus represents the result of filtering a transaction
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum FilterStatus {
+    /// Transaction will make it to the next batch
+    FilterOk = 0,
+    /// Transaction will be filtered out because invalid (too big, malformed, etc.)
+    FilterRemove = 1,
+    /// Transaction is valid but postponed for later processing due to size/gas constraint
+    FilterPostpone = 2,
+}
+impl FilterStatus {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            Self::FilterOk => "FILTER_OK",
+            Self::FilterRemove => "FILTER_REMOVE",
+            Self::FilterPostpone => "FILTER_POSTPONE",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "FILTER_OK" => Some(Self::FilterOk),
+            "FILTER_REMOVE" => Some(Self::FilterRemove),
+            "FILTER_POSTPONE" => Some(Self::FilterPostpone),
+            _ => None,
+        }
+    }
+}
 /// Block contains all the components of a complete block
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct Block {