fabricx: notification service integration (#1313)

Signed-off-by: Angelo De Caro <adc@zurich.ibm.com>

Author: adecaro

docs/core-token.md

@@ -42,20 +42,11 @@ token:
     #   b) The transaction will reach finality shortly, so we append a listener and wait for a timeout. If the listener reaches timeout, we proceed to step c.
     #   c) The transaction reached finality long ago, so we query the whole ledger for this specific transaction. If the query returns no result, we proceed to step d.
     #   d) The transaction will reach finality at some point beyond the timeout or never, so we return Unknown. Then it is up to the client to either append another listener or accept that the transaction will never reach finality.  
-    # committer: The manager subscribes to the commit pipeline and receives all finalized transactions.
-    #   If we subscribe for a transaction that hasn't been finalized yet, we will get notified once it reaches the commit pipeline.
-    #   For listeners that haven't been invoked yet, either the transaction hasn't been finalized or it was finalized before we subscribed.
-    #   For this reason, there is a periodic polling period (1s) that queries all txIDs for which there is a pending listener and invokes the listeners for the ones found.
-    #   Once we get notified about finality, we try (repeatedly) to fetch the additional tx information needed from the vault (e.g., transactionRequest).
-    #   If we work without replicas, there is no need to try more than once to fetch the additional tx information, because the finality notification and vault update happen in sync.
-    #   It is only needed in case we have replicas, where another replica may have updated the vault shortly before.
+    # 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.
+    # The field can also be left empty. In that case, the default option will be used depending on the network type each TMS refers to.
     type: delivery
-    # Only applicable when type = 'committer'
-    committer:
-      # maxRetries is the number of times we try to fetch the additional tx information from the vault.
-      maxRetries: 3
-      # retryWaitDuration is the duration to wait before retrying.
-      retryWaitDuration: 5s
     # Only applicable when type = 'delivery'
     delivery:
       # mapperParallelism is the number of goroutines that process incoming transactions in parallel. Defaults to 1.
@@ -76,6 +67,12 @@ token:
       # If the timeout is not set, then the listener will never be evicted and we will never proceed to step c.
       # We will wait forever for the transaction to return (as is done for the 'committer' type).
       listenerTimeout: 10s
+    # Only applicable when type = 'notification'
+    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
   tms:
     mytms: # unique name of this token management system
       network: default # the name of the network this TMS refers to (Fabric, etc.)

docs/services/network.md

@@ -1,4 +1,4 @@
-## The Network Service
+# The Network Service
 
 The `network` service, located under `token/services/network`, provides other services with a consistent and predictable interface to the backend (e.g., Fabric).
 Internally, the network service mirrors the structure of the Token API, consisting of a `Provider` of network instances and the `Network` instances themselves.
@@ -7,6 +7,8 @@ The network service architecture is depicted below:
 
 ![network_service.png](../imgs/network_service.png)
 
+## Fabric 
+
 The Fabric-based network implementation utilizes the Fabric Smart Client for configuration and operations, including chaincode queries and transaction broadcasting.
 
 During bootstrap, the Token SDK processes the TMS defined in the configuration.

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/gin-gonic/gin v1.11.0
 	github.com/google/pprof v0.0.0-20260202012954-cb029daf43ef
 	github.com/hashicorp/go-uuid v1.0.3
-	github.com/hyperledger-labs/fabric-smart-client v0.8.1
+	github.com/hyperledger-labs/fabric-smart-client v0.8.2-0.20260213141142-cafdc428edab
 	github.com/hyperledger/fabric-chaincode-go/v2 v2.3.0
 	github.com/hyperledger/fabric-lib-go v1.1.3-0.20240523144151-25edd1eaf5f5
 	github.com/hyperledger/fabric-protos-go-apiv2 v0.3.7

go.sum

@@ -1023,8 +1023,8 @@ github.com/huin/goupnp v1.3.0 h1:UvLUlWDNpoUdYzb2TCn+MuTWtcjXKSza2n6CBdQ0xXc=
 github.com/huin/goupnp v1.3.0/go.mod h1:gnGPsThkYa7bFi/KWmEysQRf48l2dvR5bxr2OFckNX8=
 github.com/hyperledger-labs/SmartBFT v0.0.0-20250503203013-eb005eef8866 h1:Mu/6NJsfl9g3wM15Ue7hqPq4LtgYDoABh8MO4u8aW4g=
 github.com/hyperledger-labs/SmartBFT v0.0.0-20250503203013-eb005eef8866/go.mod h1:9aNHNXsCVy/leGz2gpTC1eOL5QecxbSAGjqsLh4T1LM=
-github.com/hyperledger-labs/fabric-smart-client v0.8.1 h1:w5wE564sydDWoz4725r0GjVtb3fb8vMVHre7cEcyYQg=
-github.com/hyperledger-labs/fabric-smart-client v0.8.1/go.mod h1:bRA/GQB0XOhjbnckKjaJ0Yhj5Gf1rzwUN1VfcHyrLIk=
+github.com/hyperledger-labs/fabric-smart-client v0.8.2-0.20260213141142-cafdc428edab h1:9K0uUke+X779dsbbAbQy8ZxpVH9uYWKv/TZx/2DEqDE=
+github.com/hyperledger-labs/fabric-smart-client v0.8.2-0.20260213141142-cafdc428edab/go.mod h1:yKdQlHiq7Q5B8ngBCjwrwCOez7nLMh8R3zSdpSCtmeM=
 github.com/hyperledger/aries-bbs-go v0.0.0-20240528084656-761671ea73bc h1:3Ykk6MtyfnlzMOQry9zkxsoLWpCWZwDPqehO/BJwArM=
 github.com/hyperledger/aries-bbs-go v0.0.0-20240528084656-761671ea73bc/go.mod h1:Kofn6A6WWea1ZM8Rys5aBW9dszwJ7Ywa0kyyYL0TPYw=
 github.com/hyperledger/fabric-amcl v0.0.0-20230602173724-9e02669dceb2 h1:B1Nt8hKb//KvgGRprk0h1t4lCnwhE9/ryb1WqfZbV+M=

integration/nwo/token/template.go

@@ -13,7 +13,8 @@ token:
   selector:
     driver: {{ TokenSelector }}
   finality:
-    type: {{ FinalityType }}
+    # we leave type empty so that the default is peaked per type of network
+    type:
     delivery:
       mapperParallelism: 10
       lruSize: 100

integration/token/fungible/views/transfer.go

@@ -477,10 +477,12 @@ func (t *BroadcastPreparedTransferView) Call(context view.Context) (interface{},
 	assert.NoError(err, "failed unmarshalling transaction")
 
 	// broadcast the transaction to the ordering service
+	logger.Debugf("broadcasting transaction [%s]", tx.ID())
 	_, err = context.RunView(ttx.NewOrderingView(tx))
 	assert.NoError(err, "failed asking ordering")
 
 	if t.Finality {
+		logger.Debugf("transaction [%s], wait for finality", tx.ID())
 		// wait for finality
 		_, err = context.RunView(ttx.NewFinalityView(tx))
 		assert.NoError(err, "failed asking ordering")

token/services/auditor/auditor.go

@@ -120,6 +120,8 @@ func (a *Service) Append(ctx context.Context, tx Transaction) error {
 	logger.DebugfContext(ctx, "register tx status listener for tx [%s] at network [%s]", tx.ID(), tx.Network())
 	var r driver.FinalityListener = finality.NewListener(
 		logger,
+		net,
+		tx.Namespace(),
 		a.tmsProvider,
 		a.tmsID,
 		a.auditDB,

token/services/auditor/manager.go

@@ -122,7 +122,20 @@ func (cm *ServiceManager) RestoreTMS(tmsID token.TMSID) error {
 	return iterators.ForEach(it, func(record *storage.TokenRequestRecord) error {
 		logger.Debugf("restore transaction [%s] with status [%s]", record.TxID, TxStatusMessage[record.Status])
 
-		return net.AddFinalityListener(tmsID.Namespace, record.TxID, finality.NewListener(logger, cm.tmsProvider, tmsID, auditor.auditDB, tokenDB, auditor.finalityTracer))
+		return net.AddFinalityListener(
+			tmsID.Namespace,
+			record.TxID,
+			finality.NewListener(
+				logger,
+				net,
+				tmsID.Namespace,
+				cm.tmsProvider,
+				tmsID,
+				auditor.auditDB,
+				tokenDB,
+				auditor.finalityTracer,
+			),
+		)
 	})
 }
 

token/services/network/driver/network.go

@@ -78,9 +78,6 @@ type Network interface {
 	// In this case, the listener is not removed
 	AddFinalityListener(namespace string, txID string, listener FinalityListener) error
 
-	// RemoveFinalityListener unregisters the passed listener.
-	RemoveFinalityListener(id string, listener FinalityListener) error
-
 	// LookupTransferMetadataKey searches for a transfer metadata key containing the passed sub-key starting from the passed transaction id in the given namespace.
 	// The operation gets canceled if the passed timeout elapses or, if stopOnLastTx is true, when the last transaction in the vault is reached.
 	LookupTransferMetadataKey(namespace string, key string, timeout time.Duration) ([]byte, error)
@@ -96,7 +93,4 @@ type FinalityListenerManager interface {
 	// If the transaction id is empty, the listener will be called on status changes of any transaction.
 	// In this case, the listener is not removed
 	AddFinalityListener(namespace string, txID string, listener FinalityListener) error
-
-	// RemoveFinalityListener unregisters the passed listener.
-	RemoveFinalityListener(id string, listener FinalityListener) error
 }

token/services/network/fabric/config/config.go

@@ -24,10 +24,28 @@ type ListenerManagerConfig interface {
 	DeliveryLRUBuffer() int
 }
 
+const (
+	Type                              = "token.finality.type"
+	CommitterMaxRetries               = "token.finality.committer.maxRetries"
+	CommitterRetryWaitDuration        = "token.finality.committer.retryWaitDuration"
+	DeliveryMapperParallelism         = "token.finality.delivery.mapperParallelism"
+	DeliveryBlockProcessParallelism   = "token.finality.delivery.blockProcessParallelism"
+	DeliveryLRUSize                   = "token.finality.delivery.lruSize"
+	DeliveryLRUBuffer                 = "token.finality.delivery.lruBuffer"
+	DeliveryListenerTimeout           = "token.finality.delivery.listenerTimeout"
+	DefaultCommitterMaxRetries        = 3
+	DefaultCommitterRetryWaitDuration = 5 * time.Second
+	DefaultDeliveryMapperParallelism  = 10
+	DefaultDeliveryLRUSize            = 30
+	DefaultDeliveryLRUBuffer          = 15
+	DefaultDeliveryListenerTimeout    = 10 * time.Second
+)
+
 type ManagerType string
 
 const (
-	Delivery ManagerType = "delivery"
+	Delivery     ManagerType = "delivery"
+	Notification ManagerType = "notification"
 )
 
 func NewListenerManagerConfig(configService driver.ConfigService) *serviceListenerManagerConfig {
@@ -39,63 +57,63 @@ type serviceListenerManagerConfig struct {
 }
 
 func (c *serviceListenerManagerConfig) Type() ManagerType {
-	if v := ManagerType(c.c.GetString("token.finality.type")); len(v) > 0 {
+	if v := ManagerType(c.c.GetString(Type)); len(v) > 0 {
 		return v
 	}
 
 	return Delivery
 }
 
 func (c *serviceListenerManagerConfig) CommitterMaxRetries() int {
-	if v := c.c.GetInt("token.finality.committer.maxRetries"); v >= 0 {
+	if v := c.c.GetInt(CommitterMaxRetries); v >= 0 {
 		return v
 	}
 
-	return 3
+	return DefaultCommitterMaxRetries
 }
 
 func (c *serviceListenerManagerConfig) CommitterRetryWaitDuration() time.Duration {
-	if v := c.c.GetDuration("token.finality.committer.retryWaitDuration"); v >= 0 {
+	if v := c.c.GetDuration(CommitterRetryWaitDuration); v >= 0 {
 		return v
 	}
 
-	return 5 * time.Second
+	return DefaultCommitterRetryWaitDuration
 }
 
 func (c *serviceListenerManagerConfig) DeliveryMapperParallelism() int {
-	if v := c.c.GetInt("token.finality.delivery.mapperParallelism"); v > 0 {
+	if v := c.c.GetInt(DeliveryMapperParallelism); v > 0 {
 		return v
 	}
 
-	return 10
+	return DefaultDeliveryMapperParallelism
 }
 
 func (c *serviceListenerManagerConfig) DeliveryBlockProcessParallelism() int {
-	return c.c.GetInt("token.finality.delivery.blockProcessParallelism")
+	return c.c.GetInt(DeliveryBlockProcessParallelism)
 }
 
 func (c *serviceListenerManagerConfig) DeliveryLRUSize() int {
-	if v := c.c.GetInt("token.finality.delivery.lruSize"); v >= 0 {
+	if v := c.c.GetInt(DeliveryLRUSize); v >= 0 {
 		return v
 	}
 
-	return 30
+	return DefaultDeliveryLRUSize
 }
 
 func (c *serviceListenerManagerConfig) DeliveryLRUBuffer() int {
-	if v := c.c.GetInt("token.finality.delivery.lruBuffer"); v >= 0 {
+	if v := c.c.GetInt(DeliveryLRUBuffer); v >= 0 {
 		return v
 	}
 
-	return 15
+	return DefaultDeliveryLRUBuffer
 }
 
 func (c *serviceListenerManagerConfig) DeliveryListenerTimeout() time.Duration {
-	if v := c.c.GetDuration("token.finality.delivery.listenerTimeout"); v >= 0 {
+	if v := c.c.GetDuration(DeliveryListenerTimeout); v >= 0 {
 		return v
 	}
 
-	return 10 * time.Second
+	return DefaultDeliveryListenerTimeout
 }
 
 func (c *serviceListenerManagerConfig) String() string {