diff --git a/token/sdk/dig/sdk.go b/token/sdk/dig/sdk.go index c9164e5728..d7f8978de5 100644 --- a/token/sdk/dig/sdk.go +++ b/token/sdk/dig/sdk.go @@ -26,7 +26,6 @@ import ( identity2 "github.com/LFDT-Panurus/panurus/token/services/identity" "github.com/LFDT-Panurus/panurus/token/services/logging" "github.com/LFDT-Panurus/panurus/token/services/network" - "github.com/LFDT-Panurus/panurus/token/services/network/common" driver3 "github.com/LFDT-Panurus/panurus/token/services/network/driver" "github.com/LFDT-Panurus/panurus/token/services/nfttx/uniqueness" "github.com/LFDT-Panurus/panurus/token/services/selector/config" @@ -107,8 +106,6 @@ func (p *SDK) Install() error { logger.Infof("Token platform enabled, installing...") err := errors2.Join( - p.Container().Provide(common.NewAcceptTxInDBFilterProvider), - // config service p.Container().Provide( digutils.Identity[*fscconfig.Provider](), diff --git a/token/services/auditor/mock/audit_transaction_store.go b/token/services/auditor/mock/audit_transaction_store.go index 6499ccc80a..847b9e072b 100644 --- a/token/services/auditor/mock/audit_transaction_store.go +++ b/token/services/auditor/mock/audit_transaction_store.go @@ -66,6 +66,20 @@ type AuditTransactionStore struct { result2 string result3 error } + GetStatusesStub func(context.Context, []string) (map[string]driver.TxStatus, error) + getStatusesMutex sync.RWMutex + getStatusesArgsForCall []struct { + arg1 context.Context + arg2 []string + } + getStatusesReturns struct { + result1 map[string]driver.TxStatus + result2 error + } + getStatusesReturnsOnCall map[int]struct { + result1 map[string]driver.TxStatus + result2 error + } GetTokenRequestStub func(context.Context, string) ([]byte, error) getTokenRequestMutex sync.RWMutex getTokenRequestArgsForCall []struct { @@ -160,20 +174,6 @@ type AuditTransactionStore struct { result1 *drivera.PageIterator[*driver.TransactionRecord] result2 error } - QueryValidationsStub func(context.Context, driver.QueryValidationRecordsParams) (driver.ValidationRecordsIterator, error) - queryValidationsMutex sync.RWMutex - queryValidationsArgsForCall []struct { - arg1 context.Context - arg2 driver.QueryValidationRecordsParams - } - queryValidationsReturns struct { - result1 driver.ValidationRecordsIterator - result2 error - } - queryValidationsReturnsOnCall map[int]struct { - result1 driver.ValidationRecordsIterator - result2 error - } ReleaseRecoveryClaimStub func(context.Context, string, string, string) error releaseRecoveryClaimMutex sync.RWMutex releaseRecoveryClaimArgsForCall []struct { @@ -460,6 +460,76 @@ func (fake *AuditTransactionStore) GetStatusReturnsOnCall(i int, result1 driver. }{result1, result2, result3} } +func (fake *AuditTransactionStore) GetStatuses(arg1 context.Context, arg2 []string) (map[string]driver.TxStatus, error) { + var arg2Copy []string + if arg2 != nil { + arg2Copy = make([]string, len(arg2)) + copy(arg2Copy, arg2) + } + fake.getStatusesMutex.Lock() + ret, specificReturn := fake.getStatusesReturnsOnCall[len(fake.getStatusesArgsForCall)] + fake.getStatusesArgsForCall = append(fake.getStatusesArgsForCall, struct { + arg1 context.Context + arg2 []string + }{arg1, arg2Copy}) + stub := fake.GetStatusesStub + fakeReturns := fake.getStatusesReturns + fake.recordInvocation("GetStatuses", []interface{}{arg1, arg2Copy}) + fake.getStatusesMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *AuditTransactionStore) GetStatusesCallCount() int { + fake.getStatusesMutex.RLock() + defer fake.getStatusesMutex.RUnlock() + return len(fake.getStatusesArgsForCall) +} + +func (fake *AuditTransactionStore) GetStatusesCalls(stub func(context.Context, []string) (map[string]driver.TxStatus, error)) { + fake.getStatusesMutex.Lock() + defer fake.getStatusesMutex.Unlock() + fake.GetStatusesStub = stub +} + +func (fake *AuditTransactionStore) GetStatusesArgsForCall(i int) (context.Context, []string) { + fake.getStatusesMutex.RLock() + defer fake.getStatusesMutex.RUnlock() + argsForCall := fake.getStatusesArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *AuditTransactionStore) GetStatusesReturns(result1 map[string]driver.TxStatus, result2 error) { + fake.getStatusesMutex.Lock() + defer fake.getStatusesMutex.Unlock() + fake.GetStatusesStub = nil + fake.getStatusesReturns = struct { + result1 map[string]driver.TxStatus + result2 error + }{result1, result2} +} + +func (fake *AuditTransactionStore) GetStatusesReturnsOnCall(i int, result1 map[string]driver.TxStatus, result2 error) { + fake.getStatusesMutex.Lock() + defer fake.getStatusesMutex.Unlock() + fake.GetStatusesStub = nil + if fake.getStatusesReturnsOnCall == nil { + fake.getStatusesReturnsOnCall = make(map[int]struct { + result1 map[string]driver.TxStatus + result2 error + }) + } + fake.getStatusesReturnsOnCall[i] = struct { + result1 map[string]driver.TxStatus + result2 error + }{result1, result2} +} + func (fake *AuditTransactionStore) GetTokenRequest(arg1 context.Context, arg2 string) ([]byte, error) { fake.getTokenRequestMutex.Lock() ret, specificReturn := fake.getTokenRequestReturnsOnCall[len(fake.getTokenRequestArgsForCall)] @@ -908,71 +978,6 @@ func (fake *AuditTransactionStore) QueryTransactionsReturnsOnCall(i int, result1 }{result1, result2} } -func (fake *AuditTransactionStore) QueryValidations(arg1 context.Context, arg2 driver.QueryValidationRecordsParams) (driver.ValidationRecordsIterator, error) { - fake.queryValidationsMutex.Lock() - ret, specificReturn := fake.queryValidationsReturnsOnCall[len(fake.queryValidationsArgsForCall)] - fake.queryValidationsArgsForCall = append(fake.queryValidationsArgsForCall, struct { - arg1 context.Context - arg2 driver.QueryValidationRecordsParams - }{arg1, arg2}) - stub := fake.QueryValidationsStub - fakeReturns := fake.queryValidationsReturns - fake.recordInvocation("QueryValidations", []interface{}{arg1, arg2}) - fake.queryValidationsMutex.Unlock() - if stub != nil { - return stub(arg1, arg2) - } - if specificReturn { - return ret.result1, ret.result2 - } - return fakeReturns.result1, fakeReturns.result2 -} - -func (fake *AuditTransactionStore) QueryValidationsCallCount() int { - fake.queryValidationsMutex.RLock() - defer fake.queryValidationsMutex.RUnlock() - return len(fake.queryValidationsArgsForCall) -} - -func (fake *AuditTransactionStore) QueryValidationsCalls(stub func(context.Context, driver.QueryValidationRecordsParams) (driver.ValidationRecordsIterator, error)) { - fake.queryValidationsMutex.Lock() - defer fake.queryValidationsMutex.Unlock() - fake.QueryValidationsStub = stub -} - -func (fake *AuditTransactionStore) QueryValidationsArgsForCall(i int) (context.Context, driver.QueryValidationRecordsParams) { - fake.queryValidationsMutex.RLock() - defer fake.queryValidationsMutex.RUnlock() - argsForCall := fake.queryValidationsArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2 -} - -func (fake *AuditTransactionStore) QueryValidationsReturns(result1 driver.ValidationRecordsIterator, result2 error) { - fake.queryValidationsMutex.Lock() - defer fake.queryValidationsMutex.Unlock() - fake.QueryValidationsStub = nil - fake.queryValidationsReturns = struct { - result1 driver.ValidationRecordsIterator - result2 error - }{result1, result2} -} - -func (fake *AuditTransactionStore) QueryValidationsReturnsOnCall(i int, result1 driver.ValidationRecordsIterator, result2 error) { - fake.queryValidationsMutex.Lock() - defer fake.queryValidationsMutex.Unlock() - fake.QueryValidationsStub = nil - if fake.queryValidationsReturnsOnCall == nil { - fake.queryValidationsReturnsOnCall = make(map[int]struct { - result1 driver.ValidationRecordsIterator - result2 error - }) - } - fake.queryValidationsReturnsOnCall[i] = struct { - result1 driver.ValidationRecordsIterator - result2 error - }{result1, result2} -} - func (fake *AuditTransactionStore) ReleaseRecoveryClaim(arg1 context.Context, arg2 string, arg3 string, arg4 string) error { fake.releaseRecoveryClaimMutex.Lock() ret, specificReturn := fake.releaseRecoveryClaimReturnsOnCall[len(fake.releaseRecoveryClaimArgsForCall)] diff --git a/token/services/network/common/filter.go b/token/services/network/common/filter.go deleted file mode 100644 index ff78b40760..0000000000 --- a/token/services/network/common/filter.go +++ /dev/null @@ -1,71 +0,0 @@ -/* -Copyright IBM Corp. All Rights Reserved. - -SPDX-License-Identifier: Apache-2.0 -*/ - -package common - -import ( - "context" - - token3 "github.com/LFDT-Panurus/panurus/token" - "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb" - "github.com/LFDT-Panurus/panurus/token/services/storage/ttxdb" - "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors" - driver2 "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver" -) - -type TransactionFilterProvider[F driver2.TransactionFilter] interface { - New(tmsID token3.TMSID) (F, error) -} - -// AcceptTxInDBFilterProvider provides instances of AcceptTxInDBsFilter based on the transaction db and audit db -// for a given TMS -type AcceptTxInDBFilterProvider struct { - ttxStoreServiceManager ttxdb.StoreServiceManager - auditStoreServiceManager auditdb.StoreServiceManager -} - -func NewAcceptTxInDBFilterProvider(ttxStoreServiceManager ttxdb.StoreServiceManager, auditStoreServiceManager auditdb.StoreServiceManager) *AcceptTxInDBFilterProvider { - return &AcceptTxInDBFilterProvider{ttxStoreServiceManager: ttxStoreServiceManager, auditStoreServiceManager: auditStoreServiceManager} -} - -func (p *AcceptTxInDBFilterProvider) New(tmsID token3.TMSID) (*AcceptTxInDBsFilter, error) { - ttxDB, err := p.ttxStoreServiceManager.StoreServiceByTMSId(tmsID) - if err != nil { - return nil, errors.WithMessagef(err, "failed to get transaction db for [%s]", tmsID) - } - auditDB, err := p.auditStoreServiceManager.StoreServiceByTMSId(tmsID) - if err != nil { - return nil, errors.WithMessagef(err, "failed to get audit db for [%s]", tmsID) - } - - return &AcceptTxInDBsFilter{ - ttxDB: ttxDB, - auditDB: auditDB, - }, nil -} - -// AcceptTxInDBsFilter uses the transaction db and the audit db to decide if a given transaction needs -// to be further processed by Panurus upon a network event about its finality -type AcceptTxInDBsFilter struct { - ttxDB *ttxdb.StoreService - auditDB *auditdb.StoreService -} - -func (t *AcceptTxInDBsFilter) Accept(txID string, env []byte) (bool, error) { - status, _, err := t.ttxDB.GetStatus(context.Background(), txID) - if err != nil { - return false, err - } - if status != ttxdb.Unknown { - return true, nil - } - status, _, err = t.auditDB.GetStatus(context.Background(), txID) - if err != nil { - return false, err - } - - return status != auditdb.Unknown, nil -} diff --git a/token/services/network/fabric/driver.go b/token/services/network/fabric/driver.go index 03af4411dd..cf34708fcc 100644 --- a/token/services/network/fabric/driver.go +++ b/token/services/network/fabric/driver.go @@ -12,7 +12,6 @@ import ( "github.com/LFDT-Panurus/panurus/token" "github.com/LFDT-Panurus/panurus/token/core/common/metrics" "github.com/LFDT-Panurus/panurus/token/services/config" - "github.com/LFDT-Panurus/panurus/token/services/network/common" "github.com/LFDT-Panurus/panurus/token/services/network/common/rws/keys" "github.com/LFDT-Panurus/panurus/token/services/network/common/rws/translator" "github.com/LFDT-Panurus/panurus/token/services/network/driver" @@ -44,7 +43,6 @@ type Driver struct { tokensManager *tokens.ServiceManager configService *config.Service viewManager *view.Manager - filterProvider *common.AcceptTxInDBFilterProvider tmsProvider *token.ManagementServiceProvider identityProvider view.IdentityProvider tracerProvider trace.TracerProvider @@ -69,7 +67,6 @@ func NewGenericDriver( configProvider *config.Service, viewManager *view.Manager, viewRegistry *view.Registry, - filterProvider *common.AcceptTxInDBFilterProvider, tmsProvider *token.ManagementServiceProvider, tracerProvider trace.TracerProvider, identityProvider view.IdentityProvider, @@ -87,7 +84,6 @@ func NewGenericDriver( tokensManager, configProvider, viewManager, - filterProvider, tmsProvider, tracerProvider, identityProvider, @@ -121,7 +117,6 @@ func NewDriver( tokensManager *tokens.ServiceManager, configService *config.Service, viewManager *view.Manager, - filterProvider *common.AcceptTxInDBFilterProvider, tmsProvider *token.ManagementServiceProvider, tracerProvider trace.TracerProvider, identityProvider view.IdentityProvider, @@ -144,7 +139,6 @@ func NewDriver( tokensManager: tokensManager, configService: configService, viewManager: viewManager, - filterProvider: filterProvider, tmsProvider: tmsProvider, identityProvider: identityProvider, tracerProvider: tracerProvider, @@ -198,7 +192,6 @@ func (d *Driver) New(network, channel string) (driver.Network, error) { fns, ch, d.configService, - d.filterProvider, d.tokensManager, d.viewManager, d.tmsProvider, diff --git a/token/services/network/fabric/network.go b/token/services/network/fabric/network.go index a06d76b865..03f5c16365 100644 --- a/token/services/network/fabric/network.go +++ b/token/services/network/fabric/network.go @@ -201,7 +201,6 @@ type Network struct { viewManager ViewManager ledger driver.Ledger configuration ncommon.Configuration - filterProvider ncommon.TransactionFilterProvider[*ncommon.AcceptTxInDBsFilter] tokensProvider *tokens.ServiceManager finalityTracer trace.Tracer localMembership *lm @@ -227,7 +226,6 @@ func NewNetwork( n *fabric.NetworkService, ch *fabric.Channel, configuration ncommon.Configuration, - filterProvider ncommon.TransactionFilterProvider[*ncommon.AcceptTxInDBsFilter], tokensProvider *tokens.ServiceManager, viewManager ViewManager, tmsProvider *token2.ManagementServiceProvider, @@ -253,7 +251,6 @@ func NewNetwork( viewManager: viewManager, ledger: ledger, configuration: configuration, - filterProvider: filterProvider, tokensProvider: tokensProvider, flm: flm, llm: llm, diff --git a/token/services/network/fabricx/driver.go b/token/services/network/fabricx/driver.go index 69242e6365..e137bb16b3 100644 --- a/token/services/network/fabricx/driver.go +++ b/token/services/network/fabricx/driver.go @@ -12,7 +12,6 @@ import ( "github.com/LFDT-Panurus/panurus/token" "github.com/LFDT-Panurus/panurus/token/core/common/metrics" "github.com/LFDT-Panurus/panurus/token/services/config" - "github.com/LFDT-Panurus/panurus/token/services/network/common" "github.com/LFDT-Panurus/panurus/token/services/network/common/rws/keys" "github.com/LFDT-Panurus/panurus/token/services/network/common/rws/translator" "github.com/LFDT-Panurus/panurus/token/services/network/driver" @@ -49,7 +48,6 @@ func NewDriver( tokensManager *tokens.ServiceManager, configs *config.Service, viewManager *view.Manager, - filterProvider *common.AcceptTxInDBFilterProvider, tmsProvider *token.ManagementServiceProvider, tracerProvider trace.TracerProvider, identityProvider view.IdentityProvider, @@ -88,7 +86,6 @@ func NewDriver( tokensManager: tokensManager, configService: configs, viewManager: viewManager, - filterProvider: filterProvider, tmsProvider: tmsProvider, tracerProvider: tracerProvider, identityProvider: identityProvider, @@ -137,7 +134,6 @@ type Driver struct { tokensManager *tokens.ServiceManager configService *config.Service viewManager *view.Manager - filterProvider *common.AcceptTxInDBFilterProvider tmsProvider *token.ManagementServiceProvider tracerProvider trace.TracerProvider identityProvider view.IdentityProvider @@ -199,7 +195,6 @@ func (d *Driver) New(network, channel string) (driver.Network, error) { fns, ch, d.configService, - d.filterProvider, d.tokensManager, d.viewManager, d.tmsProvider, diff --git a/token/services/network/fabricx/network.go b/token/services/network/fabricx/network.go index ce87fd505e..275e1bc889 100644 --- a/token/services/network/fabricx/network.go +++ b/token/services/network/fabricx/network.go @@ -35,7 +35,6 @@ func NewNetwork( n *ffabric.NetworkService, ch *ffabric.Channel, configuration common.Configuration, - filterProvider common.TransactionFilterProvider[*common.AcceptTxInDBsFilter], tokensProvider *tokens.ServiceManager, viewManager fabric.ViewManager, tmsProvider *token.ManagementServiceProvider, @@ -56,7 +55,6 @@ func NewNetwork( n, ch, configuration, - filterProvider, tokensProvider, viewManager, tmsProvider, diff --git a/token/services/storage/auditdb/store.go b/token/services/storage/auditdb/store.go index 55cebfb02e..aa7d6bd541 100644 --- a/token/services/storage/auditdb/store.go +++ b/token/services/storage/auditdb/store.go @@ -313,6 +313,13 @@ func (d *StoreService) GetStatus(ctx context.Context, txID string) (TxStatus, st return status, message, nil } +// GetStatuses returns the status of the given transaction ids in a single +// query. See driver.AuditTransactionStore.GetStatuses for details about +// missing-key semantics. +func (d *StoreService) GetStatuses(ctx context.Context, txIDs []string) (map[string]TxStatus, error) { + return d.db.GetStatuses(ctx, txIDs) +} + // GetTokenRequest returns the token request bound to the passed transaction id, if available. func (d *StoreService) GetTokenRequest(ctx context.Context, txID string) ([]byte, error) { return d.db.GetTokenRequest(ctx, txID) diff --git a/token/services/storage/db/dbtest/transactions.go b/token/services/storage/db/dbtest/transactions.go index 52f23cfbe4..84610bd895 100644 --- a/token/services/storage/db/dbtest/transactions.go +++ b/token/services/storage/db/dbtest/transactions.go @@ -143,6 +143,22 @@ func TStatus(t *testing.T, db driver3.TokenTransactionStore) { require.NoError(t, err, "error getting movements") assert.Len(t, mvs, 1) assert.Equal(t, driver3.Confirmed, mvs[0].Status, "movement status should be confirmed") + + // Batch GetStatuses: present and missing tx ids. + statuses, err := db.GetStatuses(ctx, []string{"tx1", "missing"}) + require.NoError(t, err) + assert.Len(t, statuses, 1, "missing tx ids must not occupy a map entry") + assert.Equal(t, driver3.Confirmed, statuses["tx1"]) + _, ok := statuses["missing"] + assert.False(t, ok) + + empty, err := db.GetStatuses(ctx, nil) + require.NoError(t, err) + assert.Empty(t, empty, "nil input must return an empty map without querying") + + empty, err = db.GetStatuses(ctx, []string{}) + require.NoError(t, err) + assert.Empty(t, empty, "empty-slice input must return an empty map without querying") } func TStoresTimestamp(t *testing.T, db driver3.TokenTransactionStore) { diff --git a/token/services/storage/db/driver/audit.go b/token/services/storage/db/driver/audit.go index 8a58951ce4..37f2be5303 100644 --- a/token/services/storage/db/driver/audit.go +++ b/token/services/storage/db/driver/audit.go @@ -28,6 +28,13 @@ type AuditTransactionStore interface { // It returns an error if the transaction is not found GetStatus(ctx context.Context, txID string) (TxStatus, string, error) + // GetStatuses returns the status of the given transaction ids, in a + // single query. The returned map contains an entry only for tx ids that + // were present in storage — callers should treat a missing key + // identically to GetStatus returning Unknown. An empty or nil txIDs + // slice returns an empty map without touching the database. + GetStatuses(ctx context.Context, txIDs []string) (map[string]TxStatus, error) + // QueryTransactions returns a list of transactions that match the passed params QueryTransactions(ctx context.Context, params QueryTransactionsParams, pagination driver.Pagination) (*driver.PageIterator[*TransactionRecord], error) diff --git a/token/services/storage/db/driver/ttx.go b/token/services/storage/db/driver/ttx.go index f7e944280a..f0729c0985 100644 --- a/token/services/storage/db/driver/ttx.go +++ b/token/services/storage/db/driver/ttx.go @@ -61,6 +61,13 @@ type TransactionStore interface { // It returns an error if the transaction is not found GetStatus(ctx context.Context, txID string) (TxStatus, string, error) + // GetStatuses returns the status of the given transaction ids, in a + // single query. The returned map contains an entry only for tx ids that + // were present in storage — callers should treat a missing key + // identically to GetStatus returning Unknown. An empty or nil txIDs + // slice returns an empty map without touching the database. + GetStatuses(ctx context.Context, txIDs []string) (map[string]TxStatus, error) + // QueryTransactions returns a list of transactions that match the given criteria QueryTransactions(ctx context.Context, params QueryTransactionsParams, pagination driver2.Pagination) (*driver2.PageIterator[*TransactionRecord], error) diff --git a/token/services/storage/db/sql/common/transactions.go b/token/services/storage/db/sql/common/transactions.go index a02de2e534..7747b9ea48 100644 --- a/token/services/storage/db/sql/common/transactions.go +++ b/token/services/storage/db/sql/common/transactions.go @@ -269,6 +269,40 @@ func (db *TransactionStore) GetStatus(ctx context.Context, txID string) (dbdrive return status, statusMessage, nil } +// GetStatuses fetches the status for the given tx ids in a single SELECT. +// Missing tx ids are absent from the returned map — callers should treat a +// missing key identically to GetStatus returning Unknown. Empty input +// returns an empty map without querying. +func (db *TransactionStore) GetStatuses(ctx context.Context, txIDs []string) (map[string]dbdriver.TxStatus, error) { + if len(txIDs) == 0 { + return map[string]dbdriver.TxStatus{}, nil + } + query, args := q.Select(). + FieldsByName("tx_id", "status"). + From(q.Table(db.table.Requests)). + Where(cond.In("tx_id", txIDs...)). + Format(db.ci) + + logging.Debug(logger, query, args) + rows, err := db.readDB.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer Close(rows) + + result := make(map[string]dbdriver.TxStatus, len(txIDs)) + for rows.Next() { + var txID string + var status dbdriver.TxStatus + if err := rows.Scan(&txID, &status); err != nil { + return nil, err + } + result[txID] = status + } + + return result, rows.Err() +} + func (db *TransactionStore) Notifier() (dbdriver.TransactionNotifier, error) { if db.notifier == nil { return nil, storage.ErrNotSupported diff --git a/token/services/storage/ttxdb/store.go b/token/services/storage/ttxdb/store.go index 3ff2b81488..5bc2ab695a 100644 --- a/token/services/storage/ttxdb/store.go +++ b/token/services/storage/ttxdb/store.go @@ -273,6 +273,13 @@ func (d *StoreService) GetStatus(ctx context.Context, txID string) (TxStatus, st return status, message, nil } +// GetStatuses returns the status of the given transaction ids in a single +// query. See driver.TransactionStore.GetStatuses for details about +// missing-key semantics. +func (d *StoreService) GetStatuses(ctx context.Context, txIDs []string) (map[string]TxStatus, error) { + return d.db.GetStatuses(ctx, txIDs) +} + // GetTokenRequest returns the token request bound to the passed transaction id, if available. func (d *StoreService) GetTokenRequest(ctx context.Context, txID string) ([]byte, error) { return d.db.GetTokenRequest(ctx, txID) diff --git a/token/services/ttx/dep/wrapper/dbs.go b/token/services/ttx/dep/wrapper/dbs.go index fdde7d5257..ba8e583ff2 100644 --- a/token/services/ttx/dep/wrapper/dbs.go +++ b/token/services/ttx/dep/wrapper/dbs.go @@ -7,32 +7,92 @@ SPDX-License-Identifier: Apache-2.0 package wrapper import ( + "context" + "github.com/LFDT-Panurus/panurus/token" + "github.com/LFDT-Panurus/panurus/token/services" "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb" "github.com/LFDT-Panurus/panurus/token/services/storage/ttxdb" "github.com/LFDT-Panurus/panurus/token/services/ttx/dep" + "github.com/LFDT-Panurus/panurus/token/services/ttx/dep/db" + "github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/lazy" ) +// statusStore is the subset of a transaction/audit store service needed by +// the batching decorator: the batched status lookup plus the status +// listener registry, which is passed through untouched. +type statusStore interface { + statusFetcher + AddStatusListener(txID string, ch chan db.TransactionStatusEvent) + DeleteStatusListener(txID string, ch chan db.TransactionStatusEvent) +} + +// batchingStatusDB wraps a transaction/audit store service so that +// single-tx GetStatus lookups issued by concurrent finality views are +// coalesced into one batched GetStatuses query per batch window. It +// implements both dep.TransactionDB and dep.AuditDB. The status message is +// not part of the batched query and is returned empty; the interfaces' only +// consumer (the ttx finality view) discards it. +type batchingStatusDB struct { + store statusStore + batcher *statusBatcher +} + +func newBatchingStatusDB(store statusStore) *batchingStatusDB { + return &batchingStatusDB{store: store, batcher: newStatusBatcher(store)} +} + +func (b *batchingStatusDB) GetStatus(ctx context.Context, txID string) (token.TxStatus, string, error) { + status, err := b.batcher.Get(ctx, txID) + + return status, "", err +} + +func (b *batchingStatusDB) AddStatusListener(txID string, ch chan db.TransactionStatusEvent) { + b.store.AddStatusListener(txID, ch) +} + +func (b *batchingStatusDB) DeleteStatusListener(txID string, ch chan db.TransactionStatusEvent) { + b.store.DeleteStatusListener(txID, ch) +} + type TransactionDBProvider struct { - storeServiceManager ttxdb.StoreServiceManager + dbs lazy.Provider[token.TMSID, *batchingStatusDB] } func NewTransactionDBProvider(storeServiceManager ttxdb.StoreServiceManager) *TransactionDBProvider { - return &TransactionDBProvider{storeServiceManager: storeServiceManager} + // one decorator (hence one batcher) per TMS: all finality views for a + // TMS share the same batcher, which is what makes coalescing effective + return &TransactionDBProvider{dbs: lazy.NewProviderWithKeyMapper(services.Key, func(tmsID token.TMSID) (*batchingStatusDB, error) { + store, err := storeServiceManager.StoreServiceByTMSId(tmsID) + if err != nil { + return nil, err + } + + return newBatchingStatusDB(store), nil + })} } func (t *TransactionDBProvider) TransactionDB(tmsID token.TMSID) (dep.TransactionDB, error) { - return t.storeServiceManager.StoreServiceByTMSId(tmsID) + return t.dbs.Get(tmsID) } type AuditDBProvider struct { - storeServiceManager auditdb.StoreServiceManager + dbs lazy.Provider[token.TMSID, *batchingStatusDB] } func NewAuditDBProvider(storeServiceManager auditdb.StoreServiceManager) *AuditDBProvider { - return &AuditDBProvider{storeServiceManager: storeServiceManager} + // see NewTransactionDBProvider for why decorators are cached per TMS + return &AuditDBProvider{dbs: lazy.NewProviderWithKeyMapper(services.Key, func(tmsID token.TMSID) (*batchingStatusDB, error) { + store, err := storeServiceManager.StoreServiceByTMSId(tmsID) + if err != nil { + return nil, err + } + + return newBatchingStatusDB(store), nil + })} } func (t *AuditDBProvider) AuditDB(tmsID token.TMSID) (dep.AuditDB, error) { - return t.storeServiceManager.StoreServiceByTMSId(tmsID) + return t.dbs.Get(tmsID) } diff --git a/token/services/ttx/dep/wrapper/dbs_test.go b/token/services/ttx/dep/wrapper/dbs_test.go new file mode 100644 index 0000000000..b36ba1e12e --- /dev/null +++ b/token/services/ttx/dep/wrapper/dbs_test.go @@ -0,0 +1,119 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package wrapper + +import ( + "sync" + "testing" + + "github.com/LFDT-Panurus/panurus/token" + auditmock "github.com/LFDT-Panurus/panurus/token/services/storage/auditdb/mock" + dbdriver "github.com/LFDT-Panurus/panurus/token/services/storage/db/driver" + ttxmock "github.com/LFDT-Panurus/panurus/token/services/storage/ttxdb/mock" + "github.com/LFDT-Panurus/panurus/token/services/ttx/dep/db" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeStatusStore struct { + fakeStatusFetcher + added []string + deleted []string +} + +func (f *fakeStatusStore) AddStatusListener(txID string, _ chan db.TransactionStatusEvent) { + f.mu.Lock() + defer f.mu.Unlock() + f.added = append(f.added, txID) +} + +func (f *fakeStatusStore) DeleteStatusListener(txID string, _ chan db.TransactionStatusEvent) { + f.mu.Lock() + defer f.mu.Unlock() + f.deleted = append(f.deleted, txID) +} + +func TestBatchingStatusDB_CoalescesConcurrentGetStatus(t *testing.T) { + store := &fakeStatusStore{fakeStatusFetcher: fakeStatusFetcher{responses: map[string]dbdriver.TxStatus{ + "tx1": dbdriver.Confirmed, + "tx2": dbdriver.Pending, + }}} + d := newBatchingStatusDB(store) + + ids := []string{"tx1", "tx2"} + results := make([]dbdriver.TxStatus, len(ids)) + var wg sync.WaitGroup + for i, id := range ids { + wg.Add(1) + go func(i int, id string) { + defer wg.Done() + s, _, err := d.GetStatus(t.Context(), id) + assert.NoError(t, err) + results[i] = s + }(i, id) + } + wg.Wait() + + assert.Equal(t, []dbdriver.TxStatus{dbdriver.Confirmed, dbdriver.Pending}, results) + assert.Equal(t, 1, store.callCount(), "concurrent GetStatus calls should be coalesced into a single batched query") +} + +func TestBatchingStatusDB_MissingTxIsUnknown(t *testing.T) { + store := &fakeStatusStore{fakeStatusFetcher: fakeStatusFetcher{responses: map[string]dbdriver.TxStatus{}}} + d := newBatchingStatusDB(store) + + status, msg, err := d.GetStatus(t.Context(), "missing") + require.NoError(t, err) + assert.Equal(t, dbdriver.Unknown, status) + assert.Empty(t, msg) +} + +func TestBatchingStatusDB_ListenersPassThrough(t *testing.T) { + store := &fakeStatusStore{} + d := newBatchingStatusDB(store) + + ch := make(chan db.TransactionStatusEvent, 1) + d.AddStatusListener("tx1", ch) + d.DeleteStatusListener("tx1", ch) + + assert.Equal(t, []string{"tx1"}, store.added) + assert.Equal(t, []string{"tx1"}, store.deleted) +} + +func TestTransactionDBProvider_CachesPerTMS(t *testing.T) { + manager := &ttxmock.TTXStoreServiceManager{} + provider := NewTransactionDBProvider(manager) + + tms1 := token.TMSID{Network: "n1", Channel: "c1", Namespace: "ns1"} + tms2 := token.TMSID{Network: "n2", Channel: "c2", Namespace: "ns2"} + + db1a, err := provider.TransactionDB(tms1) + require.NoError(t, err) + db1b, err := provider.TransactionDB(tms1) + require.NoError(t, err) + db2, err := provider.TransactionDB(tms2) + require.NoError(t, err) + + assert.Same(t, db1a, db1b, "same TMS ID must share one decorator (and one batcher)") + assert.NotSame(t, db1a, db2, "different TMS IDs must not share a decorator") + assert.Equal(t, 2, manager.StoreServiceByTMSIdCallCount(), "store must be resolved once per TMS ID") +} + +func TestAuditDBProvider_CachesPerTMS(t *testing.T) { + manager := &auditmock.AuditStoreServiceManager{} + provider := NewAuditDBProvider(manager) + + tms := token.TMSID{Network: "n1", Channel: "c1", Namespace: "ns1"} + + db1a, err := provider.AuditDB(tms) + require.NoError(t, err) + db1b, err := provider.AuditDB(tms) + require.NoError(t, err) + + assert.Same(t, db1a, db1b, "same TMS ID must share one decorator (and one batcher)") + assert.Equal(t, 1, manager.StoreServiceByTMSIdCallCount()) +} diff --git a/token/services/ttx/dep/wrapper/status_batcher.go b/token/services/ttx/dep/wrapper/status_batcher.go new file mode 100644 index 0000000000..698aefc683 --- /dev/null +++ b/token/services/ttx/dep/wrapper/status_batcher.go @@ -0,0 +1,103 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package wrapper + +import ( + "context" + "sync" + "time" + + dbdriver "github.com/LFDT-Panurus/panurus/token/services/storage/db/driver" +) + +// statusBatchWindow is how long a statusBatcher waits after its first +// pending lookup before flushing, to give concurrent callers a chance to +// join the same batched query. Every call pays this as a fixed latency tax +// even when nothing joins it, so it's kept small relative to a typical DB +// round trip rather than tuned for maximum coalescing. +const statusBatchWindow = 300 * time.Microsecond + +// statusFetcher resolves the status of many transaction ids in one round trip. +type statusFetcher interface { + GetStatuses(ctx context.Context, txIDs []string) (map[string]dbdriver.TxStatus, error) +} + +// statusBatcher coalesces concurrent single-tx status lookups that land +// within statusBatchWindow into a single batched GetStatuses call. +type statusBatcher struct { + fetch statusFetcher + window time.Duration + + mu sync.Mutex + pending *statusBatch +} + +type statusBatch struct { + ids []string + waiters map[string][]chan statusResult +} + +type statusResult struct { + status dbdriver.TxStatus + err error +} + +func newStatusBatcher(fetch statusFetcher) *statusBatcher { + return &statusBatcher{fetch: fetch, window: statusBatchWindow} +} + +// Get returns the status of txID, coalescing this call with any others that +// arrive within the batch window into a single GetStatuses query. If ctx is +// cancelled while waiting, Get returns ctx.Err(); the batched query still +// completes on behalf of the other waiters. +func (b *statusBatcher) Get(ctx context.Context, txID string) (dbdriver.TxStatus, error) { + ch := make(chan statusResult, 1) + + b.mu.Lock() + if b.pending == nil { + b.pending = &statusBatch{waiters: map[string][]chan statusResult{}} + batch := b.pending + time.AfterFunc(b.window, func() { b.flush(batch) }) + } + batch := b.pending + batch.ids = append(batch.ids, txID) + batch.waiters[txID] = append(batch.waiters[txID], ch) + b.mu.Unlock() + + select { + case res := <-ch: + return res.status, res.err + case <-ctx.Done(): + return dbdriver.Unknown, ctx.Err() + } +} + +// flush resolves a batch's transaction ids in one call and hands each +// waiter its result. It detaches the batch from b.pending first so that +// callers arriving after this point start a new batch instead of joining +// one that's already being flushed. The lookup deliberately runs on +// context.Background(): the batch serves multiple callers, so it must not +// be tied to any single caller's context. +func (b *statusBatcher) flush(batch *statusBatch) { + b.mu.Lock() + if b.pending == batch { + b.pending = nil + } + b.mu.Unlock() + + statuses, err := b.fetch.GetStatuses(context.Background(), batch.ids) + for txID, waiters := range batch.waiters { + res := statusResult{err: err} + if err == nil { + // zero value is dbdriver.Unknown, matching GetStatus's not-found semantics + res.status = statuses[txID] + } + for _, ch := range waiters { + ch <- res + } + } +} diff --git a/token/services/ttx/dep/wrapper/status_batcher_test.go b/token/services/ttx/dep/wrapper/status_batcher_test.go new file mode 100644 index 0000000000..644d1c5259 --- /dev/null +++ b/token/services/ttx/dep/wrapper/status_batcher_test.go @@ -0,0 +1,170 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package wrapper + +import ( + "context" + "errors" + "sync" + "testing" + + dbdriver "github.com/LFDT-Panurus/panurus/token/services/storage/db/driver" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeStatusFetcher struct { + mu sync.Mutex + calls [][]string + responses map[string]dbdriver.TxStatus + err error + block chan struct{} +} + +func (f *fakeStatusFetcher) GetStatuses(_ context.Context, txIDs []string) (map[string]dbdriver.TxStatus, error) { + f.mu.Lock() + f.calls = append(f.calls, append([]string(nil), txIDs...)) + block := f.block + f.mu.Unlock() + + if block != nil { + <-block + } + + if f.err != nil { + return nil, f.err + } + + result := make(map[string]dbdriver.TxStatus, len(txIDs)) + for _, id := range txIDs { + if s, ok := f.responses[id]; ok { + result[id] = s + } + } + + return result, nil +} + +func (f *fakeStatusFetcher) callCount() int { + f.mu.Lock() + defer f.mu.Unlock() + + return len(f.calls) +} + +func TestStatusBatcher_SingleLookup(t *testing.T) { + fetch := &fakeStatusFetcher{responses: map[string]dbdriver.TxStatus{"tx1": dbdriver.Confirmed}} + b := newStatusBatcher(fetch) + + status, err := b.Get(t.Context(), "tx1") + require.NoError(t, err) + assert.Equal(t, dbdriver.Confirmed, status) + assert.Equal(t, 1, fetch.callCount()) +} + +func TestStatusBatcher_MissingTxIsUnknown(t *testing.T) { + fetch := &fakeStatusFetcher{responses: map[string]dbdriver.TxStatus{}} + b := newStatusBatcher(fetch) + + status, err := b.Get(t.Context(), "missing") + require.NoError(t, err) + assert.Equal(t, dbdriver.Unknown, status) +} + +func TestStatusBatcher_CoalescesConcurrentLookups(t *testing.T) { + fetch := &fakeStatusFetcher{responses: map[string]dbdriver.TxStatus{ + "tx1": dbdriver.Confirmed, + "tx2": dbdriver.Pending, + "tx3": dbdriver.Deleted, + }} + b := newStatusBatcher(fetch) + + ids := []string{"tx1", "tx2", "tx3"} + results := make([]dbdriver.TxStatus, len(ids)) + var wg sync.WaitGroup + for i, id := range ids { + wg.Add(1) + go func(i int, id string) { + defer wg.Done() + s, err := b.Get(t.Context(), id) + assert.NoError(t, err) + results[i] = s + }(i, id) + } + wg.Wait() + + assert.Equal(t, []dbdriver.TxStatus{dbdriver.Confirmed, dbdriver.Pending, dbdriver.Deleted}, results) + assert.Equal(t, 1, fetch.callCount(), "concurrent lookups should be coalesced into a single batch") +} + +func TestStatusBatcher_DuplicateTxIDInSameBatch(t *testing.T) { + fetch := &fakeStatusFetcher{responses: map[string]dbdriver.TxStatus{"tx1": dbdriver.Confirmed}} + b := newStatusBatcher(fetch) + + results := make([]dbdriver.TxStatus, 2) + var wg sync.WaitGroup + for i := range results { + wg.Add(1) + go func(i int) { + defer wg.Done() + s, err := b.Get(t.Context(), "tx1") + assert.NoError(t, err) + results[i] = s + }(i) + } + wg.Wait() + + assert.Equal(t, []dbdriver.TxStatus{dbdriver.Confirmed, dbdriver.Confirmed}, results) + require.Equal(t, 1, fetch.callCount()) + assert.Equal(t, []string{"tx1", "tx1"}, fetch.calls[0], "both waiters for the same tx id should be included in the batch request") +} + +func TestStatusBatcher_SeparateBatchesAfterFlush(t *testing.T) { + fetch := &fakeStatusFetcher{responses: map[string]dbdriver.TxStatus{"tx1": dbdriver.Confirmed, "tx2": dbdriver.Confirmed}} + b := newStatusBatcher(fetch) + + // Get blocks until its own batch flushes, so these two calls can never + // land in the same window. + _, err := b.Get(t.Context(), "tx1") + require.NoError(t, err) + _, err = b.Get(t.Context(), "tx2") + require.NoError(t, err) + + assert.Equal(t, 2, fetch.callCount(), "lookups that don't overlap in time should not be coalesced") +} + +func TestStatusBatcher_PropagatesError(t *testing.T) { + wantErr := errors.New("boom") + fetch := &fakeStatusFetcher{err: wantErr} + b := newStatusBatcher(fetch) + + _, err := b.Get(t.Context(), "tx1") + require.ErrorIs(t, err, wantErr) +} + +func TestStatusBatcher_HonorsContextCancellation(t *testing.T) { + block := make(chan struct{}) + fetch := &fakeStatusFetcher{ + responses: map[string]dbdriver.TxStatus{"tx1": dbdriver.Confirmed}, + block: block, + } + b := newStatusBatcher(fetch) + + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { + _, err := b.Get(ctx, "tx1") + done <- err + }() + + cancel() + err := <-done + require.ErrorIs(t, err, context.Canceled) + + // unblock the in-flight fetch so its goroutine can finish + close(block) +}