Merge branch 'julien/speedup-submitter' into julien/fiber

Author: julienrbrt

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"
 )
 
@@ -165,11 +165,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")
 }

block/internal/cache/manager.go

@@ -77,10 +77,14 @@ type PendingManager interface {
 	GetPendingData(ctx context.Context) ([]*types.SignedData, [][]byte, error)
 	SetLastSubmittedHeaderHeight(ctx context.Context, height uint64)
 	GetLastSubmittedHeaderHeight() uint64
+	ResetInFlightHeaderRange(start, end uint64)
 	SetLastSubmittedDataHeight(ctx context.Context, height uint64)
 	GetLastSubmittedDataHeight() uint64
+	ResetInFlightDataRange(start, end uint64)
 	NumPendingHeaders() uint64
 	NumPendingData() uint64
+	NumPendingHeadersTotal() uint64
+	NumPendingDataTotal() uint64
 }
 
 // Manager combines CacheManager and PendingManager.
@@ -311,6 +315,14 @@ func (m *implementation) SetLastSubmittedHeaderHeight(ctx context.Context, heigh
 	m.pendingHeaders.SetLastSubmittedHeaderHeight(ctx, height)
 }
 
+func (m *implementation) ResetInFlightHeaderHeight() {
+	m.pendingHeaders.ResetInFlightHeaderRange(0, 0)
+}
+
+func (m *implementation) ResetInFlightHeaderRange(start, end uint64) {
+	m.pendingHeaders.ResetInFlightHeaderRange(start, end)
+}
+
 func (m *implementation) GetLastSubmittedDataHeight() uint64 {
 	return m.pendingData.GetLastSubmittedDataHeight()
 }
@@ -319,6 +331,10 @@ func (m *implementation) SetLastSubmittedDataHeight(ctx context.Context, height
 	m.pendingData.SetLastSubmittedDataHeight(ctx, height)
 }
 
+func (m *implementation) ResetInFlightDataRange(start, end uint64) {
+	m.pendingData.ResetInFlightDataRange(start, end)
+}
+
 func (m *implementation) NumPendingHeaders() uint64 {
 	return m.pendingHeaders.NumPendingHeaders()
 }
@@ -327,6 +343,14 @@ func (m *implementation) NumPendingData() uint64 {
 	return m.pendingData.NumPendingData()
 }
 
+func (m *implementation) NumPendingHeadersTotal() uint64 {
+	return m.pendingHeaders.NumPendingHeadersTotal()
+}
+
+func (m *implementation) NumPendingDataTotal() uint64 {
+	return m.pendingData.NumPendingDataTotal()
+}
+
 // SetPendingEvent sets the event at the specified height.
 func (m *implementation) SetPendingEvent(height uint64, event *common.DAHeightEvent) {
 	m.pendingMu.Lock()

block/internal/cache/manager_test.go

@@ -221,6 +221,12 @@ func TestPendingHeadersAndData_Flow(t *testing.T) {
 	// update last submitted heights and re-check
 	cm.SetLastSubmittedHeaderHeight(ctx, 1)
 	cm.SetLastSubmittedDataHeight(ctx, 2)
+	cm.ResetInFlightHeaderRange(1, 3)
+	cm.ResetInFlightDataRange(2, 3)
+
+	// numPending views (before getPending claims items)
+	assert.Equal(t, uint64(2), cm.NumPendingHeaders())
+	assert.Equal(t, uint64(1), cm.NumPendingData())
 
 	headers, _, err = cm.GetPendingHeaders(ctx)
 	require.NoError(t, err)
@@ -231,10 +237,6 @@ func TestPendingHeadersAndData_Flow(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, signedData, 1)
 	assert.Equal(t, uint64(3), signedData[0].Height())
-
-	// numPending views
-	assert.Equal(t, uint64(2), cm.NumPendingHeaders())
-	assert.Equal(t, uint64(1), cm.NumPendingData())
 }
 
 func TestManager_TxOperations(t *testing.T) {