perf(fabricx/finality): batch finality via shared poller (#1849)

Signed-off-by: Evan <evanyan@sign.global>

Author: EvanYan1024

docs/configuration.md

@@ -74,12 +74,23 @@ token:
     # Only applicable for fabricx networks
     # notification: The manager is notified about finality events via a notification service (e.g. for FabricX).
     #   When a new notification arrives, an event is added to a queue for asynchronous processing.
-    #   When a client subscribes to the manager for a specific transaction, we perform an immediate query to check its status.
+    #   When a client subscribes to the manager for a specific transaction, the transaction joins a pending set
+    #   that the shared poller sweeps with batched status queries (see poller below).
     notification:
       # workers is the number of goroutines that process events in parallel. Defaults to 10.
       workers: 10
       # queueSize is the size of the event buffer. Defaults to 1000.
       queueSize: 1000
+    # Only applicable for fabricx networks
+    # poller: resolves the status of pending transactions with periodic batched committer queries.
+    poller:
+      # interval is how often the poller sweeps the pending set. Defaults to 1s.
+      interval: 1s
+      # batchSize is the maximum number of txIDs in one committer status query. Defaults to 2000.
+      batchSize: 2000
+      # pendingTTL is how long a tx stays pending before its slot is reclaimed.
+      # It should exceed the longest caller finality timeout. Defaults to 10m.
+      pendingTTL: 10m
 
   # fabricx configuration for FabricX-specific settings
   fabricx:
@@ -426,6 +437,9 @@ Default values:
 - delivery.listenerTimeout: 10s
 - notification.workers: 10
 - notification.queueSize: 1000
+- poller.interval: 1s
+- poller.batchSize: 2000
+- poller.pendingTTL: 10m
 
 ---
 

token/services/network/fabricx/driver.go

@@ -99,6 +99,7 @@ func NewDriver(
 			queryServiceProvider,
 			finalityProvider,
 			q,
+			finality2.NewConfig(configService),
 		),
 		llmProvider: lookup2.NewCronNSListenerManagerProvider(
 			queryServiceProvider,

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

@@ -0,0 +1,86 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package finality
+
+import "time"
+
+const (
+	// PollInterval is the configuration key for how often the shared poller sweeps the pending set
+	PollInterval = "token.finality.poller.interval"
+	// PollBatchSize is the configuration key for how many txIDs go into one committer status query
+	PollBatchSize = "token.finality.poller.batchSize"
+	// PendingTTL is the configuration key for how long a tx stays pending before its slot is
+	// reclaimed; it should exceed the longest caller finality timeout
+	PendingTTL = "token.finality.poller.pendingTTL"
+
+	// DefaultPollInterval is the default sweep interval
+	DefaultPollInterval = 1 * time.Second
+	// DefaultPollBatchSize is the default status query batch size
+	DefaultPollBatchSize = 2000
+	// DefaultPendingTTL is the default pending slot TTL
+	DefaultPendingTTL = 10 * time.Minute
+)
+
+// ConfigGetter models the configuration getter for the finality poller
+type ConfigGetter interface {
+	// PollInterval returns how often the shared poller sweeps the pending set
+	PollInterval() time.Duration
+	// PollBatchSize returns how many txIDs go into one committer status query
+	PollBatchSize() int
+	// PendingTTL returns how long a tx stays pending before its slot is reclaimed
+	PendingTTL() time.Duration
+}
+
+// Configuration models the configuration for the finality poller.
+//
+//go:generate counterfeiter -o mock/configuration.go -fake-name Configuration . Configuration
+type Configuration interface {
+	// GetDuration returns the duration for the given key.
+	GetDuration(key string) time.Duration
+	// GetInt returns the int for the given key.
+	GetInt(key string) int
+}
+
+// NewConfig creates a new ConfigGetter that uses the provided Configuration
+// interface to retrieve finality poller settings.
+func NewConfig(configuration Configuration) *serviceConfig {
+	return &serviceConfig{configuration: configuration}
+}
+
+type serviceConfig struct {
+	configuration Configuration
+}
+
+// PollInterval returns the sweep interval from the configuration.
+// If the configured value is not greater than 0, it returns DefaultPollInterval.
+func (c *serviceConfig) PollInterval() time.Duration {
+	if v := c.configuration.GetDuration(PollInterval); v > 0 {
+		return v
+	}
+
+	return DefaultPollInterval
+}
+
+// PollBatchSize returns the status query batch size from the configuration.
+// If the configured value is not greater than 0, it returns DefaultPollBatchSize.
+func (c *serviceConfig) PollBatchSize() int {
+	if v := c.configuration.GetInt(PollBatchSize); v > 0 {
+		return v
+	}
+
+	return DefaultPollBatchSize
+}
+
+// PendingTTL returns the pending slot TTL from the configuration.
+// If the configured value is not greater than 0, it returns DefaultPendingTTL.
+func (c *serviceConfig) PendingTTL() time.Duration {
+	if v := c.configuration.GetDuration(PendingTTL); v > 0 {
+		return v
+	}
+
+	return DefaultPendingTTL
+}

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

@@ -0,0 +1,70 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package finality_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/LFDT-Panurus/panurus/token/services/network/fabricx/finality"
+	"github.com/LFDT-Panurus/panurus/token/services/network/fabricx/finality/mock"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestServiceConfig_PollInterval(t *testing.T) {
+	t.Run("returns value from configuration when > 0", func(t *testing.T) {
+		m := &mock.Configuration{}
+		m.GetDurationReturns(2 * time.Second)
+		cfg := finality.NewConfig(m)
+
+		assert.Equal(t, 2*time.Second, cfg.PollInterval())
+		assert.Equal(t, finality.PollInterval, m.GetDurationArgsForCall(0))
+	})
+
+	t.Run("returns default when not set", func(t *testing.T) {
+		m := &mock.Configuration{}
+		cfg := finality.NewConfig(m)
+
+		assert.Equal(t, finality.DefaultPollInterval, cfg.PollInterval())
+	})
+}
+
+func TestServiceConfig_PollBatchSize(t *testing.T) {
+	t.Run("returns value from configuration when > 0", func(t *testing.T) {
+		m := &mock.Configuration{}
+		m.GetIntReturns(500)
+		cfg := finality.NewConfig(m)
+
+		assert.Equal(t, 500, cfg.PollBatchSize())
+		assert.Equal(t, finality.PollBatchSize, m.GetIntArgsForCall(0))
+	})
+
+	t.Run("returns default when not set", func(t *testing.T) {
+		m := &mock.Configuration{}
+		cfg := finality.NewConfig(m)
+
+		assert.Equal(t, finality.DefaultPollBatchSize, cfg.PollBatchSize())
+	})
+}
+
+func TestServiceConfig_PendingTTL(t *testing.T) {
+	t.Run("returns value from configuration when > 0", func(t *testing.T) {
+		m := &mock.Configuration{}
+		m.GetDurationReturns(time.Hour)
+		cfg := finality.NewConfig(m)
+
+		assert.Equal(t, time.Hour, cfg.PendingTTL())
+		assert.Equal(t, finality.PendingTTL, m.GetDurationArgsForCall(0))
+	})
+
+	t.Run("returns default when not set", func(t *testing.T) {
+		m := &mock.Configuration{}
+		cfg := finality.NewConfig(m)
+
+		assert.Equal(t, finality.DefaultPendingTTL, cfg.PendingTTL())
+	})
+}