[verifier] Fix timeout bug (#596)

#### Type of change

- Bug fix
- Improvement (improvement to code, performance, etc)
 
#### Description

- Fix timeout behaviour to start from the last batch submitted (instead
of the last received item)
- Batch buffer capacity (`outputCh`) increased to 1 to prevent blocking
- Flattened `Config` structure by removing nested `ExecutorConfig`
- Updated all references across tests, configuration files, and
templates

#### Related issues

- resolves #554

Signed-off-by: Liran Funaro <liran.funaro@gmail.com>

Author: liran-funaro

cmd/config/app_config_test.go

@@ -247,12 +247,10 @@ func TestReadConfigVerifier(t *testing.T) {
 		configFilePath:       emptyConfig(t),
 		expectedServerConfig: newServeConfig(verifier.DefaultServerPort, verifier.DefaultMonitoringPort),
 		expectedServiceConfig: &verifier.Config{
-			ParallelExecutor: verifier.ExecutorConfig{
-				Parallelism:       verifier.DefaultParallelism,
-				BatchSizeCutoff:   verifier.DefaultBatchSizeCutoff,
-				BatchTimeCutoff:   verifier.DefaultBatchTimeCutoff,
-				ChannelBufferSize: verifier.DefaultChannelBufferSize,
-			},
+			Parallelism:       verifier.DefaultParallelism,
+			BatchSizeCutoff:   verifier.DefaultBatchSizeCutoff,
+			BatchTimeCutoff:   verifier.DefaultBatchTimeCutoff,
+			ChannelBufferSize: verifier.DefaultChannelBufferSize,
 		},
 	}, {
 		name:           "sample",
@@ -261,12 +259,10 @@ func TestReadConfigVerifier(t *testing.T) {
 			"verifier", verifier.DefaultServerPort, verifier.DefaultMonitoringPort,
 		),
 		expectedServiceConfig: &verifier.Config{
-			ParallelExecutor: verifier.ExecutorConfig{
-				BatchSizeCutoff:   verifier.DefaultBatchSizeCutoff,
-				BatchTimeCutoff:   10 * time.Millisecond,
-				ChannelBufferSize: verifier.DefaultChannelBufferSize,
-				Parallelism:       40,
-			},
+			BatchSizeCutoff:   verifier.DefaultBatchSizeCutoff,
+			BatchTimeCutoff:   10 * time.Millisecond,
+			ChannelBufferSize: verifier.DefaultChannelBufferSize,
+			Parallelism:       40,
 		},
 	}}
 

cmd/config/samples/verifier.yaml

@@ -39,28 +39,27 @@ monitoring:
     #   - Must be less than or equal to requests-per-second
     burst: 0
 
-parallel-executor:
-  # Minimum number of verification responses to collect before emitting a batch result.
-  # Larger batches improve verification efficiency through parallel processing
-  # across multiple goroutines. Batches are processed when this size is reached
-  # or batch-time-cutoff expires, whichever comes first.
-  # Default: 50
-  batch-size-cutoff: 50
-  # Maximum time to wait for a batch to reach batch-size-cutoff before processing
-  # it anyway. Ensures low latency even when signature arrival rate is low.
-  # Balances latency vs. batching efficiency.
-  # Default: 500ms
-  batch-time-cutoff: 10ms
-  # Buffer size for internal channels used in the signature verification pipeline.
-  # Larger buffers reduce channel contention and improve throughput but increase
-  # memory usage.
-  # Default: 50
-  channel-buffer-size: 50
-  # Number of goroutines used to verify signatures in parallel within each batch.
-  # Should be tuned based on CPU core count and signature verification workload.
-  # Higher values improve throughput on multi-core systems but increase CPU usage.
-  # Default: 4
-  parallelism: 40
+# Minimum number of verification responses to collect before emitting a batch result.
+# Larger batches improve verification efficiency through parallel processing
+# across multiple goroutines. Batches are processed when this size is reached
+# or batch-time-cutoff expires, whichever comes first.
+# Default: 50
+batch-size-cutoff: 50
+# Maximum time to wait for a batch to reach batch-size-cutoff before processing
+# it anyway. Ensures low latency even when signature arrival rate is low.
+# Balances latency vs. batching efficiency.
+# Default: 500ms
+batch-time-cutoff: 10ms
+# Buffer size for internal channels used in the signature verification pipeline.
+# Larger buffers reduce channel contention and improve throughput but increase
+# memory usage.
+# Default: 50
+channel-buffer-size: 50
+# Number of goroutines used to verify signatures in parallel within each batch.
+# Should be tuned based on CPU core count and signature verification workload.
+# Higher values improve throughput on multi-core systems but increase CPU usage.
+# Default: 4
+parallelism: 40
 
 # Logging configuration
 logging:

cmd/config/templates/verifier.yaml.tmpl

@@ -4,8 +4,7 @@
 #
 {{ include "service" . | indent 0 }}
 
-parallel-executor:
-  parallelism: 40
-  batch-time-cutoff: {{ .VerifierBatchTimeCutoff | default "10ms" }}
-  batch-size-cutoff: {{ .VerifierBatchSizeCutoff | default 50 }}
-  channel-buffer-size: 50
+parallelism: 40
+batch-time-cutoff: {{ .VerifierBatchTimeCutoff | default "10ms" }}
+batch-size-cutoff: {{ .VerifierBatchSizeCutoff | default 50 }}
+channel-buffer-size: 50

cmd/config/viper.go

@@ -56,10 +56,10 @@ func NewViperWithSidecarDefaults() *viper.Viper {
 // NewViperWithVerifierDefaults returns a viper instance with the verifier default values.
 func NewViperWithVerifierDefaults() *viper.Viper {
 	v := newViperWithServiceDefault("verifier", verifier.DefaultServerPort, verifier.DefaultMonitoringPort)
-	v.SetDefault("parallel-executor.parallelism", verifier.DefaultParallelism)
-	v.SetDefault("parallel-executor.batch-time-cutoff", verifier.DefaultBatchTimeCutoff)
-	v.SetDefault("parallel-executor.batch-size-cutoff", verifier.DefaultBatchSizeCutoff)
-	v.SetDefault("parallel-executor.channel-buffer-size", verifier.DefaultChannelBufferSize)
+	v.SetDefault("parallelism", verifier.DefaultParallelism)
+	v.SetDefault("batch-time-cutoff", verifier.DefaultBatchTimeCutoff)
+	v.SetDefault("batch-size-cutoff", verifier.DefaultBatchSizeCutoff)
+	v.SetDefault("channel-buffer-size", verifier.DefaultChannelBufferSize)
 	return v
 }
 

loadgen/client_test.go

@@ -136,12 +136,10 @@ func startVerifiers(t *testing.T, serverTLS, clientTLS connection.TLSConfig) *co
 	endpoints := make([]*connection.Endpoint, 2)
 	for i := range endpoints {
 		service := verifier.New(&verifier.Config{
-			ParallelExecutor: verifier.ExecutorConfig{
-				BatchSizeCutoff:   50,
-				BatchTimeCutoff:   10 * time.Millisecond,
-				ChannelBufferSize: 50,
-				Parallelism:       40,
-			},
+			BatchSizeCutoff:   50,
+			BatchTimeCutoff:   10 * time.Millisecond,
+			ChannelBufferSize: 50,
+			Parallelism:       40,
 		})
 
 		serverConfig := test.NewLocalHostServiceConfig(serverTLS)

service/verifier/config.go

@@ -13,11 +13,6 @@ import (
 type (
 	// Config describes the signature verifier parameters.
 	Config struct {
-		ParallelExecutor ExecutorConfig `mapstructure:"parallel-executor"`
-	}
-
-	// ExecutorConfig describes the execution parameters.
-	ExecutorConfig struct {
 		// Parallelism How many parallel go routines will be launched
 		Parallelism int `mapstructure:"parallelism" validate:"required,gt=0"`
 		// BatchSizeCutoff The minimum amount of responses we need to collect before emitting a response

service/verifier/parallel_executor.go

@@ -23,7 +23,7 @@ type (
 		outputSingleCh chan *verificationOutput
 		outputCh       chan []*committerpb.TxStatus
 		verifier       *verifier
-		config         *ExecutorConfig
+		config         *Config
 	}
 
 	verificationOutput struct {
@@ -32,12 +32,12 @@ type (
 	}
 )
 
-func newParallelExecutor(config *ExecutorConfig) *parallelExecutor {
+func newParallelExecutor(config *Config) *parallelExecutor {
 	channelCapacity := config.ChannelBufferSize * config.Parallelism
 	return &parallelExecutor{
 		config:         config,
 		inputCh:        make(chan *servicepb.TxWithRef, channelCapacity),
-		outputCh:       make(chan []*committerpb.TxStatus),
+		outputCh:       make(chan []*committerpb.TxStatus, 1),
 		outputSingleCh: make(chan *verificationOutput, channelCapacity),
 		verifier:       newVerifier(),
 	}
@@ -61,21 +61,30 @@ func (e *parallelExecutor) handleChannelInput(ctx context.Context) {
 func (e *parallelExecutor) handleCutoff(ctx context.Context) {
 	var outputBuffer []*committerpb.TxStatus
 	chOut := channel.NewWriter(ctx, e.outputCh)
+
+	var cutTimeout <-chan time.Time
 	cutBatch := func(size int) {
 		for len(outputBuffer) >= size {
 			batchSize := min(e.config.BatchSizeCutoff, len(outputBuffer))
 			logger.Debugf("Cuts batch with %d/%d of the outputs.", batchSize, len(outputBuffer))
 			chOut.Write(outputBuffer[:batchSize])
 			outputBuffer = outputBuffer[batchSize:]
+			// Reset the timer if we submitted a batch.
+			cutTimeout = nil
 		}
 	}
 	for {
+		if cutTimeout == nil {
+			cutTimeout = time.After(e.config.BatchTimeCutoff)
+		}
 		select {
 		case <-ctx.Done():
 			return
-		case <-time.After(e.config.BatchTimeCutoff):
-			logger.Debugf("Attempts to cut a batch (timout). (buffer size: %d)", len(outputBuffer))
+		case <-cutTimeout:
+			logger.Debugf("Attempts to cut a batch (timeout). (buffer size: %d)", len(outputBuffer))
 			cutBatch(1)
+			// Reset the timer since it ended.
+			cutTimeout = nil
 		case output := <-e.outputSingleCh:
 			logger.Debugf("Attempts to emit a batch (response). (buffer size: %d)", len(outputBuffer)+1)
 			outputBuffer = append(outputBuffer, output.status)

service/verifier/parallel_executor_test.go

@@ -60,7 +60,7 @@ func TestRegularTransactionBatching(t *testing.T) {
 
 func startExecutor(ctx context.Context, t *testing.T, batchSizeCutoff int) *parallelExecutor {
 	t.Helper()
-	config := &ExecutorConfig{
+	config := &Config{
 		BatchSizeCutoff:   batchSizeCutoff,
 		BatchTimeCutoff:   1 * time.Hour,
 		Parallelism:       1,

service/verifier/verifier_server.go

@@ -74,7 +74,7 @@ func (s *Server) StartStream(stream servicepb.Verifier_StartStreamServer) error
 	defer s.metrics.ActiveStreams.Dec()
 
 	// We create a new executor for each stream to avoid answering to the wrong stream.
-	executor := newParallelExecutor(&s.config.ParallelExecutor)
+	executor := newParallelExecutor(s.config)
 	g, gCtx := errgroup.WithContext(stream.Context())
 	g.Go(func() error {
 		return s.handleInputs(gCtx, stream, executor)

service/verifier/verifier_server_test.go

@@ -564,19 +564,17 @@ func defaultCryptoParameters(t *testing.T) cryptoParameters {
 
 func defaultConfigWithTLS(tlsConfig connection.TLSConfig) (*Config, *serve.Config) {
 	config := &Config{
-		ParallelExecutor: ExecutorConfig{
-			BatchSizeCutoff:   3,
-			BatchTimeCutoff:   1 * time.Hour,
-			Parallelism:       3,
-			ChannelBufferSize: 1,
-		},
+		BatchSizeCutoff:   3,
+		BatchTimeCutoff:   1 * time.Hour,
+		Parallelism:       3,
+		ChannelBufferSize: 1,
 	}
 	return config, test.NewLocalHostServiceConfig(tlsConfig)
 }
 
 func defaultConfigQuickCutoff() (*Config, *serve.Config) {
 	config, serverConfig := defaultConfigWithTLS(test.InsecureTLSConfig)
-	config.ParallelExecutor.BatchSizeCutoff = 1
+	config.BatchSizeCutoff = 1
 	return config, serverConfig
 }