Fix: Stop silently dropping failed finality events (#1419)

Signed-off-by: atharrva01 <atharvaborade568@gmail.com>

Author: atharrva01

integration/token/common/views/finality.go

@@ -90,3 +90,7 @@ func (l *finalityListener) OnStatus(ctx context.Context, txID string, status int
 	fmt.Printf("Received finality from network for TX [%s][%d]", txID, status)
 	l.success()
 }
+
+func (l *finalityListener) OnError(ctx context.Context, txID string, err error) {
+	fmt.Printf("Finality error for TX [%s]: %v", txID, err)
+}

token/services/network/driver/network.go

@@ -21,6 +21,8 @@ import (
 type FinalityListener interface {
 	// OnStatus is called when the status of a transaction changes
 	OnStatus(ctx context.Context, txID string, status int, message string, tokenRequestHash []byte)
+	// OnError is called when the finality event cannot be delivered after all retries are exhausted
+	OnError(ctx context.Context, txID string, err error)
 }
 
 type TransientMap = map[string][]byte

token/services/network/fabricx/finality/finality.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	cdriver "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver"
@@ -27,6 +28,13 @@ import (
 
 var logger = logging.MustGetLogger()
 
+const (
+	// defaultMaxRetries is the number of times a ListenerEvent will retry on transient errors.
+	defaultMaxRetries = 3
+	// defaultRetryInterval is the initial backoff delay; it doubles after each attempt.
+	defaultRetryInterval = time.Second
+)
+
 // ConfigService models the configuration service needed by the NSListenerManager
 //
 //go:generate counterfeiter -o mock/cs.go -fake-name ConfigService . ConfigService
@@ -101,17 +109,58 @@ type ListenerEvent struct {
 	StatusMessage string
 	// Namespace is the namespace of the transaction
 	Namespace string
+
+	// MaxRetries is the number of retry attempts on transient errors (0 uses defaultMaxRetries).
+	MaxRetries int
+	// RetryInterval is the initial backoff delay between retries, doubling each attempt (0 uses defaultRetryInterval).
+	RetryInterval time.Duration
 }
 
-// Process handles a finality event notification.
+// Process handles a finality event notification with exponential-backoff retries.
 // If the status is Unknown or Busy, it triggers a manual transaction check.
 // If the status is Valid, it retrieves the token request hash from the ledger.
-// Finally, it notifies the wrapped listener with the transaction's status and hash.
+// It notifies the wrapped listener on success, or calls OnError if all retries are exhausted.
 func (l *ListenerEvent) Process(ctx context.Context) error {
+	maxRetries := l.MaxRetries
+	if maxRetries <= 0 {
+		maxRetries = defaultMaxRetries
+	}
+	retryInterval := l.RetryInterval
+	if retryInterval <= 0 {
+		retryInterval = defaultRetryInterval
+	}
+
+	delay := retryInterval
+	for attempt := 0; attempt <= maxRetries; attempt++ {
+		err := l.process(ctx)
+		if err == nil {
+			return nil
+		}
+		if attempt == maxRetries {
+			logger.Errorf("[ListenerEvent] tx [%s] failed after %d attempts: %v — notifying listener", l.TxID, maxRetries+1, err)
+			l.Listener.OnError(ctx, l.TxID, err)
+
+			return nil
+		}
+		logger.Warnf("[ListenerEvent] tx [%s] attempt %d/%d failed: %v, retrying in %v", l.TxID, attempt+1, maxRetries+1, err, delay)
+		select {
+		case <-time.After(delay):
+			delay *= 2
+		case <-ctx.Done():
+			logger.Warnf("[ListenerEvent] tx [%s] context canceled during retry backoff", l.TxID)
+
+			return nil
+		}
+	}
+
+	return nil
+}
+
+// process executes a single attempt at handling the finality event.
+func (l *ListenerEvent) process(ctx context.Context) error {
 	logger.Debugf("[ListenerEvent] get notification for [%s], status [%d]", l.TxID, l.Status)
 
 	if l.Status == fdriver.Unknown || l.Status == fdriver.Busy {
-		// perform a query
 		txCheck := TxCheck{
 			QueryService:  l.QueryService,
 			KeyTranslator: l.KeyTranslator,
@@ -120,14 +169,12 @@ func (l *ListenerEvent) Process(ctx context.Context) error {
 			Namespace:     l.Namespace,
 		}
 		if err := txCheck.Process(ctx); err == nil {
-			// this means that the query has notified the event
 			return nil
 		}
 	}
 
 	var tokenRequestHash []byte
 	if l.Status == fdriver.Valid {
-		// fetch token request hash key
 		key, err := l.KeyTranslator.CreateTokenRequestKey(l.TxID)
 		if err != nil {
 			return errors.Wrapf(err, "can't create for token request [%s]", l.TxID)
@@ -345,6 +392,13 @@ func (o *OnlyOnceListener) OnStatus(ctx context.Context, txID string, status int
 	})
 }
 
+// OnError forwards the error to the wrapped listener only if it hasn't been notified before.
+func (o *OnlyOnceListener) OnError(ctx context.Context, txID string, err error) {
+	o.once.Do(func() {
+		o.listener.OnError(ctx, txID, err)
+	})
+}
+
 // fabricXFSCStatus maps Fabric-X transaction status codes to FSC validation codes.
 func fabricXFSCStatus(c int32) fdriver.ValidationCode {
 	switch protoblocktx.Status(c) {

token/services/network/fabricx/finality/finality_test.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	cdriver "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver"
@@ -206,6 +207,83 @@ func TestListenerEvent_Process_Busy_TxCheckSucceeds(t *testing.T) {
 	assert.Equal(t, 1, mockListener.OnStatusCallCount())
 }
 
+func TestListenerEvent_Process_RetriesOnTransientError(t *testing.T) {
+	ctx := t.Context()
+	txID := "tx123"
+	namespace := "token-namespace"
+	tokenRequestHash := []byte("hash123")
+	key := "token-request-key"
+
+	mockQS := &mock.QueryService{}
+	mockKT := &mock.KeyTranslator{}
+	mockListener := &mock.Listener{}
+
+	var attempts int
+	// Fail twice, succeed on third attempt
+	mockQS.GetStateStub = func(_ string, _ string) (*cdriver.VaultValue, error) {
+		attempts++
+		if attempts < 3 {
+			return nil, errors.New("transient peer error")
+		}
+
+		return &cdriver.VaultValue{Raw: tokenRequestHash}, nil
+	}
+	mockKT.CreateTokenRequestKeyReturns(key, nil)
+
+	event := &finality.ListenerEvent{
+		QueryService:  mockQS,
+		KeyTranslator: mockKT,
+		Listener:      mockListener,
+		TxID:          txID,
+		Status:        fdriver.Valid,
+		Namespace:     namespace,
+		MaxRetries:    3,
+		RetryInterval: 10 * time.Millisecond,
+	}
+
+	err := event.Process(ctx)
+	require.NoError(t, err)
+
+	// Should have retried and eventually called OnStatus
+	assert.Equal(t, 3, attempts)
+	assert.Equal(t, 1, mockListener.OnStatusCallCount())
+	assert.Equal(t, 0, mockListener.OnErrorCallCount())
+}
+
+func TestListenerEvent_Process_CallsOnErrorAfterAllRetriesExhausted(t *testing.T) {
+	ctx := t.Context()
+	txID := "tx123"
+	key := "token-request-key"
+
+	mockQS := &mock.QueryService{}
+	mockKT := &mock.KeyTranslator{}
+	mockListener := &mock.Listener{}
+
+	mockKT.CreateTokenRequestKeyReturns(key, nil)
+	mockQS.GetStateReturns(nil, errors.New("persistent peer error"))
+
+	event := &finality.ListenerEvent{
+		QueryService:  mockQS,
+		KeyTranslator: mockKT,
+		Listener:      mockListener,
+		TxID:          txID,
+		Status:        fdriver.Valid,
+		Namespace:     "token-namespace",
+		MaxRetries:    2,
+		RetryInterval: 10 * time.Millisecond,
+	}
+
+	err := event.Process(ctx)
+	require.NoError(t, err)
+
+	// All retries exhausted — OnStatus must NOT be called, OnError must be called once
+	assert.Equal(t, 0, mockListener.OnStatusCallCount())
+	assert.Equal(t, 1, mockListener.OnErrorCallCount())
+	_, callTxID, callErr := mockListener.OnErrorArgsForCall(0)
+	assert.Equal(t, txID, callTxID)
+	assert.Contains(t, callErr.Error(), "persistent peer error")
+}
+
 func TestListenerEvent_Process_CreateTokenRequestKeyError(t *testing.T) {
 	ctx := t.Context()
 	txID := "tx123"
@@ -224,12 +302,18 @@ func TestListenerEvent_Process_CreateTokenRequestKeyError(t *testing.T) {
 		Status:        fdriver.Valid,
 		StatusMessage: "",
 		Namespace:     "token-namespace",
+		MaxRetries:    2,
+		RetryInterval: 10 * time.Millisecond,
 	}
 
 	err := event.Process(ctx)
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "can't create for token request")
-	assert.Contains(t, err.Error(), "key creation failed")
+	require.NoError(t, err)
+	// All retries exhausted — listener must be notified via OnError
+	assert.Equal(t, 0, mockListener.OnStatusCallCount())
+	assert.Equal(t, 1, mockListener.OnErrorCallCount())
+	_, callTxID, callErr := mockListener.OnErrorArgsForCall(0)
+	assert.Equal(t, txID, callTxID)
+	assert.Contains(t, callErr.Error(), "key creation failed")
 }
 
 func TestListenerEvent_Process_GetStateError(t *testing.T) {
@@ -252,12 +336,18 @@ func TestListenerEvent_Process_GetStateError(t *testing.T) {
 		Status:        fdriver.Valid,
 		StatusMessage: "",
 		Namespace:     "token-namespace",
+		MaxRetries:    2,
+		RetryInterval: 10 * time.Millisecond,
 	}
 
 	err := event.Process(ctx)
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "can't get state for token request")
-	assert.Contains(t, err.Error(), "state retrieval failed")
+	require.NoError(t, err)
+	// All retries exhausted — listener must be notified via OnError
+	assert.Equal(t, 0, mockListener.OnStatusCallCount())
+	assert.Equal(t, 1, mockListener.OnErrorCallCount())
+	_, callTxID, callErr := mockListener.OnErrorArgsForCall(0)
+	assert.Equal(t, txID, callTxID)
+	assert.Contains(t, callErr.Error(), "state retrieval failed")
 }
 
 func TestListenerEvent_String(t *testing.T) {

token/services/network/fabricx/finality/mock/fl.go

@@ -98,39 +98,45 @@ func (eq *EventQueue) start() {
 	}
 }
 
-// worker represents a single goroutine that pulls events from the queue
-// and processes them until the channel is closed or the context is canceled.
-// It includes panic recovery to prevent worker crashes from affecting the pool.
+// worker is the top-level goroutine for a worker. It delegates to runWorker
+// and restarts on panic so the pool does not degrade over time.
 func (eq *EventQueue) worker(id int) {
 	defer eq.wg.Done()
+	for {
+		if stopped := eq.runWorker(id); stopped {
+			return
+		}
+		// runWorker returned false after recovering from a panic — restart the loop.
+	}
+}
+
+// runWorker processes events until the channel is closed or context is canceled.
+// It returns true for a normal exit and false when recovered from a panic.
+func (eq *EventQueue) runWorker(id int) (stopped bool) {
 	defer func() {
 		if r := recover(); r != nil {
-			logger.Errorf("Worker %d recovered from panic: %v", id, r)
-			// Don't restart worker to prevent unbounded goroutine creation
-			// The pool will continue with remaining workers
+			logger.Errorf("Worker %d recovered from panic: %v, restarting", id, r)
+			stopped = false
 		}
 	}()
 
 	for {
 		select {
 		case event, ok := <-eq.events:
 			if !ok {
-				// Channel closed, worker exits
 				logger.Debugf("Worker %d shutting down", id)
 
-				return
+				return true
 			}
 
-			// Process the event with context
 			if err := event.Process(eq.ctx); err != nil {
-				logger.Debugf("Worker %d: error processing event: %v", id, err)
+				logger.Errorf("Worker %d: error processing event [%v]: %v", id, event, err)
 			}
 
 		case <-eq.ctx.Done():
-			// Context canceled, exit gracefully
 			logger.Debugf("Worker %d received shutdown signal", id)
 
-			return
+			return true
 		}
 	}
 }

token/services/network/fabricx/finality/queue/queue.go

@@ -307,7 +307,7 @@ func TestEventProcessing_Success(t *testing.T) {
 	}
 }
 
-// TestEventProcessing_WithError tests event processing that returns errors
+// TestEventProcessing_WithError tests that a failing event is processed once and the error is logged
 func TestEventProcessing_WithError(t *testing.T) {
 	cfg := queue.Config{Workers: 2, QueueSize: 10}
 	eq, err := queue.NewEventQueue(cfg)
@@ -326,7 +326,8 @@ func TestEventProcessing_WithError(t *testing.T) {
 
 	// Wait for processing
 	time.Sleep(100 * time.Millisecond)
-	assert.True(t, event.wasProcessed())
+	// Queue does not retry — event is processed exactly once
+	assert.Equal(t, int32(1), atomic.LoadInt32(&event.processed))
 }
 
 // TestEventProcessing_WithPanic tests worker recovery from panic