Repository: hyperledger/firefly
Pull request #210 (2021-09-27 18:20:39)

Move TokenPoolCreated handler into AssetManager

codecov-commenter: # [Codecov](https://codecov.io/gh/hyperledger/firefly/pull/210?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) Report
> Merging [#210](https://codecov.io/gh/hyperledger/firefly/pull/210?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) (61f12cc) into [main](https://codecov.io/gh/hyperledger/firefly/commit/193227301a823618721e9ab791ad7f6eebf89929?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) (1932273) will **not change** coverage.
> The diff coverage is `100.00%`.

> :exclamation: Current head 61f12cc differs from pull request most recent head 80eece5. Consider uploading reports for the commit 80eece5 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/hyperledger/firefly/pull/210/graphs/tree.svg?width=650&height=150&src=pr&token=ViopDrTWPu&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger)](https://codecov.io/gh/hyperledger/firefly/pull/210?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger)

```diff
@@            Coverage Diff            @@
##              main      #210   +/-   ##
=========================================
  Coverage   100.00%   100.00%           
=========================================
  Files          203       203           
  Lines        11026     11033    +7     
=========================================
+ Hits         11026     11033    +7     
```


| [Impacted Files](https://codecov.io/gh/hyperledger/firefly/pull/210?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) | Coverage Δ | |
|---|---|---|
| [internal/events/event\_manager.go](https://codecov.io/gh/hyperledger/firefly/pull/210/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZXZlbnRzL2V2ZW50X21hbmFnZXIuZ28=) | `100.00% <ø> (ø)` | |
| [internal/assets/manager.go](https://codecov.io/gh/hyperledger/firefly/pull/210/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvYXNzZXRzL21hbmFnZXIuZ28=) | `100.00% <100.00%> (ø)` | |
| [internal/assets/token\_pool\_created.go](https://codecov.io/gh/hyperledger/firefly/pull/210/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvYXNzZXRzL3Rva2VuX3Bvb2xfY3JlYXRlZC5nbw==) | `100.00% <100.00%> (ø)` | |
| [internal/orchestrator/bound\_callbacks.go](https://codecov.io/gh/hyperledger/firefly/pull/210/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvb3JjaGVzdHJhdG9yL2JvdW5kX2NhbGxiYWNrcy5nbw==) | `100.00% <100.00%> (ø)` | |
| [internal/orchestrator/orchestrator.go](https://codecov.io/gh/hyperledger/firefly/pull/210/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvb3JjaGVzdHJhdG9yL29yY2hlc3RyYXRvci5nbw==) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/hyperledger/firefly/pull/210?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/hyperledger/firefly/pull/210?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger). Last update [1932273...80eece5](https://codecov.io/gh/hyperledger/firefly/pull/210?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger).


Patch:
diff --git a/internal/assets/manager.go b/internal/assets/manager.go
index f1b69efeb..d37e9e172 100644
--- a/internal/assets/manager.go
+++ b/internal/assets/manager.go
@@ -22,7 +22,9 @@ import (
 	"github.com/hyperledger/firefly/internal/config"
 	"github.com/hyperledger/firefly/internal/data"
 	"github.com/hyperledger/firefly/internal/i18n"
+	"github.com/hyperledger/firefly/internal/retry"
 	"github.com/hyperledger/firefly/internal/syncasync"
+	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/fftypes"
 	"github.com/hyperledger/firefly/pkg/identity"
@@ -35,6 +37,10 @@ type Manager interface {
 	GetTokenPools(ctx context.Context, ns, typeName string, filter database.AndFilter) ([]*fftypes.TokenPool, *database.FilterResult, error)
 	GetTokenPool(ctx context.Context, ns, typeName, name string) (*fftypes.TokenPool, error)
 	GetTokenAccounts(ctx context.Context, ns, typeName, name string, filter database.AndFilter) ([]*fftypes.TokenAccount, *database.FilterResult, error)
+
+	// Bound token callbacks
+	TokenPoolCreated(tk tokens.Plugin, pool *fftypes.TokenPool, signingIdentity string, protocolTxID string, additionalInfo fftypes.JSONObject) error
+
 	Start() error
 	WaitStop()
 }
@@ -46,6 +52,8 @@ type assetManager struct {
 	data      data.Manager
 	syncasync syncasync.Bridge
 	tokens    map[string]tokens.Plugin
+	retry     retry.Retry
+	txhelper  txcommon.Helper
 }
 
 func NewAssetManager(ctx context.Context, di database.Plugin, ii identity.Plugin, dm data.Manager, sa syncasync.Bridge, ti map[string]tokens.Plugin) (Manager, error) {
@@ -59,6 +67,12 @@ func NewAssetManager(ctx context.Context, di database.Plugin, ii identity.Plugin
 		data:      dm,
 		syncasync: sa,
 		tokens:    ti,
+		retry: retry.Retry{
+			InitialDelay: config.GetDuration(config.AssetManagerRetryInitialDelay),
+			MaximumDelay: config.GetDuration(config.AssetManagerRetryMaxDelay),
+			Factor:       config.GetFloat64(config.AssetManagerRetryFactor),
+		},
+		txhelper: txcommon.NewTransactionHelper(di),
 	}
 	return am, nil
 }
diff --git a/internal/events/token_pool_created.go b/internal/assets/token_pool_created.go
similarity index 81%
rename from internal/events/token_pool_created.go
rename to internal/assets/token_pool_created.go
index 74ab7fb5c..bcdb042dc 100644
--- a/internal/events/token_pool_created.go
+++ b/internal/assets/token_pool_created.go
@@ -14,7 +14,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package events
+package assets
 
 import (
 	"context"
@@ -25,12 +25,12 @@ import (
 	"github.com/hyperledger/firefly/pkg/tokens"
 )
 
-func (em *eventManager) persistTokenPoolTransaction(ctx context.Context, pool *fftypes.TokenPool, signingIdentity string, protocolTxID string, additionalInfo fftypes.JSONObject) (valid bool, err error) {
+func (am *assetManager) persistTokenPoolTransaction(ctx context.Context, pool *fftypes.TokenPool, signingIdentity string, protocolTxID string, additionalInfo fftypes.JSONObject) (valid bool, err error) {
 	if pool.ID == nil || pool.TX.ID == nil {
 		log.L(ctx).Errorf("Invalid token pool '%s'. Missing ID (%v) or transaction ID (%v)", pool.ID, pool.ID, pool.TX.ID)
 		return false, nil // this is not retryable
 	}
-	return em.txhelper.PersistTransaction(ctx, &fftypes.Transaction{
+	return am.txhelper.PersistTransaction(ctx, &fftypes.Transaction{
 		ID: pool.TX.ID,
 		Subject: fftypes.TransactionSubject{
 			Namespace: pool.Namespace,
@@ -43,13 +43,13 @@ func (em *eventManager) persistTokenPoolTransaction(ctx context.Context, pool *f
 	})
 }
 
-func (em *eventManager) persistTokenPool(ctx context.Context, pool *fftypes.TokenPool) (valid bool, err error) {
+func (am *assetManager) persistTokenPool(ctx context.Context, pool *fftypes.TokenPool) (valid bool, err error) {
 	l := log.L(ctx)
 	if err := fftypes.ValidateFFNameField(ctx, pool.Name, "name"); err != nil {
 		l.Errorf("Invalid token pool '%s' - invalid name '%s': %a", pool.ID, pool.Name, err)
 		return false, nil // This is not retryable
 	}
-	err = em.database.UpsertTokenPool(ctx, pool)
+	err = am.database.UpsertTokenPool(ctx, pool)
 	if err != nil {
 		if err == database.IDMismatch {
 			log.L(ctx).Errorf("Invalid token pool '%s'. ID mismatch with existing record", pool.ID)
@@ -61,12 +61,12 @@ func (em *eventManager) persistTokenPool(ctx context.Context, pool *fftypes.Toke
 	return true, nil
 }
 
-func (em *eventManager) TokenPoolCreated(tk tokens.Plugin, pool *fftypes.TokenPool, signingIdentity string, protocolTxID string, additionalInfo fftypes.JSONObject) error {
-	return em.retry.Do(em.ctx, "persist token pool", func(attempt int) (bool, error) {
-		err := em.database.RunAsGroup(em.ctx, func(ctx context.Context) error {
-			valid, err := em.persistTokenPoolTransaction(ctx, pool, signingIdentity, protocolTxID, additionalInfo)
+func (am *assetManager) TokenPoolCreated(tk tokens.Plugin, pool *fftypes.TokenPool, signingIdentity string, protocolTxID string, additionalInfo fftypes.JSONObject) error {
+	return am.retry.Do(am.ctx, "persist token pool", func(attempt int) (bool, error) {
+		err := am.database.RunAsGroup(am.ctx, func(ctx context.Context) error {
+			valid, err := am.persistTokenPoolTransaction(ctx, pool, signingIdentity, protocolTxID, additionalInfo)
 			if valid && err == nil {
-				valid, err = em.persistTokenPool(ctx, pool)
+				valid, err = am.persistTokenPool(ctx, pool)
 			}
 			if err != nil {
 				return err
@@ -74,11 +74,11 @@ func (em *eventManager) TokenPoolCreated(tk tokens.Plugin, pool *fftypes.TokenPo
 			if !valid {
 				log.L(ctx).Warnf("Token pool rejected id=%s author=%s", pool.ID, signingIdentity)
 				event := fftypes.NewEvent(fftypes.EventTypePoolRejected, pool.Namespace, pool.ID)
-				return em.database.InsertEvent(ctx, event)
+				return am.database.InsertEvent(ctx, event)
 			}
 			log.L(ctx).Infof("Token pool created id=%s author=%s", pool.ID, signingIdentity)
 			event := fftypes.NewEvent(fftypes.EventTypePoolConfirmed, pool.Namespace, pool.ID)
-			return em.database.InsertEvent(ctx, event)
+			return am.database.InsertEvent(ctx, event)
 		})
 		return err != nil, err // retry indefinitely (until context closes)
 	})
diff --git a/internal/config/config.go b/internal/config/config.go
index 114389ef8..f8a7dd208 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -200,7 +200,13 @@ var (
 	// SubscriptionsRetryMaxDelay is the initial retry delay
 	SubscriptionsRetryMaxDelay = rootKey("subscription.retry.maxDelay")
 	// SubscriptionsRetryFactor the backoff factor to use for retry of database operations
-	SubscriptionsRetryFactor = rootKey("event.dispatcher.retry.factor")
+	SubscriptionsRetryFactor = rootKey("subscription.retry.factor")
+	// AssetManagerRetryInitialDelay is the initial retry delay
+	AssetManagerRetryInitialDelay = rootKey("asset.manager.retry.initDelay")
+	// AssetManagerRetryMaxDelay is the initial retry delay
+	AssetManagerRetryMaxDelay = rootKey("asset.manager.retry.maxDelay")
+	// AssetManagerRetryFactor the backoff factor to use for retry of database operations
+	AssetManagerRetryFactor = rootKey("asset.manager.retry.factor")
 	// UIEnabled set to false to disable the UI (default is true, so UI will be enabled if ui.path is valid)
 	UIEnabled = rootKey("ui.enabled")
 	// UIPath the path on which to serve the UI
@@ -322,6 +328,9 @@ func Reset() {
 	viper.SetDefault(string(SubscriptionsRetryInitialDelay), "250ms")
 	viper.SetDefault(string(SubscriptionsRetryMaxDelay), "30s")
 	viper.SetDefault(string(SubscriptionsRetryFactor), 2.0)
+	viper.SetDefault(string(AssetManagerRetryInitialDelay), "250ms")
+	viper.SetDefault(string(AssetManagerRetryMaxDelay), "30s")
+	viper.SetDefault(string(AssetManagerRetryFactor), 2.0)
 	viper.SetDefault(string(UIEnabled), true)
 	viper.SetDefault(string(ValidatorCacheSize), "1Mb")
 	viper.SetDefault(string(ValidatorCacheTTL), "1h")
diff --git a/internal/events/event_manager.go b/internal/events/event_manager.go
index 7dd86a4a5..aaccca358 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -38,7 +38,6 @@ import (
 	"github.com/hyperledger/firefly/pkg/fftypes"
 	"github.com/hyperledger/firefly/pkg/identity"
 	"github.com/hyperledger/firefly/pkg/publicstorage"
-	"github.com/hyperledger/firefly/pkg/tokens"
 )
 
 type EventManager interface {
@@ -62,9 +61,6 @@ type EventManager interface {
 	BLOBReceived(dx dataexchange.Plugin, peerID string, hash fftypes.Bytes32, payloadRef string) error
 	MessageReceived(dx dataexchange.Plugin, peerID string, data []byte) error
 
-	// Bound token callbacks
-	TokenPoolCreated(tk tokens.Plugin, pool *fftypes.TokenPool, signingIdentity string, protocolTxID string, additionalInfo fftypes.JSONObject) error
-
 	// Internal events
 	sysmessaging.SystemEvents
 }
diff --git a/internal/orchestrator/bound_callbacks.go b/internal/orchestrator/bound_callbacks.go
index 9ca7b515e..35d8d04d9 100644
--- a/internal/orchestrator/bound_callbacks.go
+++ b/internal/orchestrator/bound_callbacks.go
@@ -17,6 +17,7 @@
 package orchestrator
 
 import (
+	"github.com/hyperledger/firefly/internal/assets"
 	"github.com/hyperledger/firefly/internal/events"
 	"github.com/hyperledger/firefly/pkg/blockchain"
 	"github.com/hyperledger/firefly/pkg/dataexchange"
@@ -28,6 +29,7 @@ type boundCallbacks struct {
 	bi blockchain.Plugin
 	dx dataexchange.Plugin
 	ei events.EventManager
+	am assets.Manager
 }
 
 func (bc *boundCallbacks) BlockchainOpUpdate(operationID *fftypes.UUID, txState blockchain.TransactionStatus, errorMessage string, opOutput fftypes.JSONObject) error {
@@ -55,5 +57,5 @@ func (bc *boundCallbacks) MessageReceived(peerID string, data []byte) error {
 }
 
 func (bc *boundCallbacks) TokenPoolCreated(plugin tokens.Plugin, pool *fftypes.TokenPool, signingIdentity string, protocolTxID string, additionalInfo fftypes.JSONObject) error {
-	return bc.ei.TokenPoolCreated(plugin, pool, signingIdentity, protocolTxID, additionalInfo)
+	return bc.am.TokenPoolCreated(plugin, pool, signingIdentity, protocolTxID, additionalInfo)
 }
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 8e6450286..ee0e4ab66 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -171,6 +171,7 @@ func (or *orchestrator) Init(ctx context.Context, cancelCtx context.CancelFunc)
 	or.bc.bi = or.blockchain
 	or.bc.ei = or.events
 	or.bc.dx = or.dataexchange
+	or.bc.am = or.assets
 	return err
 }
 
diff --git a/mocks/assetmocks/manager.go b/mocks/assetmocks/manager.go
index c91910ca1..b0f67b6b2 100644
--- a/mocks/assetmocks/manager.go
+++ b/mocks/assetmocks/manager.go
@@ -9,6 +9,8 @@ import (
 	fftypes "github.com/hyperledger/firefly/pkg/fftypes"
 
 	mock "github.com/stretchr/testify/mock"
+
+	tokens "github.com/hyperledger/firefly/pkg/tokens"
 )
 
 // Manager is an autogenerated mock type for the Manager type
@@ -163,6 +165,20 @@ func (_m *Manager) Start() error {
 	return r0
 }
 
+// TokenPoolCreated provides a mock function with given fields: tk, pool, signingIdentity, protocolTxID, additionalInfo
+func (_m *Manager) TokenPoolCreated(tk tokens.Plugin, pool *fftypes.TokenPool, signingIdentity string, protocolTxID string, additionalInfo fftypes.JSONObject) error {
+	ret := _m.Called(tk, pool, signingIdentity, protocolTxID, additionalInfo)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(tokens.Plugin, *fftypes.TokenPool, string, string, fftypes.JSONObject) error); ok {
+		r0 = rf(tk, pool, signingIdentity, protocolTxID, additionalInfo)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // WaitStop provides a mock function with given fields:
 func (_m *Manager) WaitStop() {
 	_m.Called()
diff --git a/mocks/eventmocks/event_manager.go b/mocks/eventmocks/event_manager.go
index bcb4331e9..a6e774ea0 100644
--- a/mocks/eventmocks/event_manager.go
+++ b/mocks/eventmocks/event_manager.go
@@ -14,8 +14,6 @@ import (
 	mock "github.com/stretchr/testify/mock"
 
 	system "github.com/hyperledger/firefly/internal/events/system"
-
-	tokens "github.com/hyperledger/firefly/pkg/tokens"
 )
 
 // EventManager is an autogenerated mock type for the EventManager type
@@ -231,20 +229,6 @@ func (_m *EventManager) SubscriptionUpdates() chan<- *fftypes.UUID {
 	return r0
 }
 
-// TokenPoolCreated provides a mock function with given fields: tk, pool, signingIdentity, protocolTxID, additionalInfo
-func (_m *EventManager) TokenPoolCreated(tk tokens.Plugin, pool *fftypes.TokenPool, signingIdentity string, protocolTxID string, additionalInfo fftypes.JSONObject) error {
-	ret := _m.Called(tk, pool, signingIdentity, protocolTxID, additionalInfo)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(tokens.Plugin, *fftypes.TokenPool, string, string, fftypes.JSONObject) error); ok {
-		r0 = rf(tk, pool, signingIdentity, protocolTxID, additionalInfo)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
 // TransferResult provides a mock function with given fields: dx, trackingID, status, info, opOutput
 func (_m *EventManager) TransferResult(dx dataexchange.Plugin, trackingID string, status fftypes.OpStatus, info string, opOutput fftypes.JSONObject) error {
 	ret := _m.Called(dx, trackingID, status, info, opOutput)


Test patch:
diff --git a/internal/assets/manager_test.go b/internal/assets/manager_test.go
index 3fc6b2cc5..53935d641 100644
--- a/internal/assets/manager_test.go
+++ b/internal/assets/manager_test.go
@@ -46,6 +46,10 @@ func newTestAssets(t *testing.T) (*assetManager, func()) {
 	mii.On("Resolve", mock.Anything, "UTNodeID").Return(defaultIdentity, nil).Maybe()
 	ctx, cancel := context.WithCancel(context.Background())
 	a, err := NewAssetManager(ctx, mdi, mii, mdm, msa, map[string]tokens.Plugin{"magic-tokens": mti})
+	rag := mdi.On("RunAsGroup", ctx, mock.Anything).Maybe()
+	rag.RunFn = func(a mock.Arguments) {
+		rag.ReturnArguments = mock.Arguments{a[1].(func(context.Context) error)(a[0].(context.Context))}
+	}
 	assert.NoError(t, err)
 	return a.(*assetManager), cancel
 }
diff --git a/internal/events/token_pool_created_test.go b/internal/assets/token_pool_created_test.go
similarity index 82%
rename from internal/events/token_pool_created_test.go
rename to internal/assets/token_pool_created_test.go
index 6c9638108..484f93102 100644
--- a/internal/events/token_pool_created_test.go
+++ b/internal/assets/token_pool_created_test.go
@@ -14,7 +14,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package events
+package assets
 
 import (
 	"fmt"
@@ -29,9 +29,9 @@ import (
 )
 
 func TestTokenPoolCreatedSuccess(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	am, cancel := newTestAssets(t)
 	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
+	mdi := am.database.(*databasemocks.Plugin)
 	mti := &tokenmocks.Plugin{}
 
 	pool := &fftypes.TokenPool{
@@ -45,42 +45,42 @@ func TestTokenPoolCreatedSuccess(t *testing.T) {
 	}
 
 	mdi.On("GetTransactionByID", mock.Anything, pool.TX.ID).Return(nil, nil)
-	mdi.On("UpsertTransaction", em.ctx, mock.MatchedBy(func(tx *fftypes.Transaction) bool {
+	mdi.On("UpsertTransaction", am.ctx, mock.MatchedBy(func(tx *fftypes.Transaction) bool {
 		return tx.Subject.Type == fftypes.TransactionTypeTokenPool
 	}), false).Return(nil)
-	mdi.On("UpsertTokenPool", em.ctx, pool).Return(nil)
-	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *fftypes.Event) bool {
+	mdi.On("UpsertTokenPool", am.ctx, pool).Return(nil)
+	mdi.On("InsertEvent", am.ctx, mock.MatchedBy(func(ev *fftypes.Event) bool {
 		return ev.Type == fftypes.EventTypePoolConfirmed && ev.Reference == pool.ID && ev.Namespace == pool.Namespace
 	})).Return(nil)
 
 	info := fftypes.JSONObject{"some": "info"}
-	err := em.TokenPoolCreated(mti, pool, "0x12345", "tx1", info)
+	err := am.TokenPoolCreated(mti, pool, "0x12345", "tx1", info)
 	assert.NoError(t, err)
 	mdi.AssertExpectations(t)
 }
 
 func TestTokenPoolMissingID(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	am, cancel := newTestAssets(t)
 	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
+	mdi := am.database.(*databasemocks.Plugin)
 	mti := &tokenmocks.Plugin{}
 
 	pool := &fftypes.TokenPool{}
 
-	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *fftypes.Event) bool {
+	mdi.On("InsertEvent", am.ctx, mock.MatchedBy(func(ev *fftypes.Event) bool {
 		return ev.Type == fftypes.EventTypePoolRejected && ev.Reference == pool.ID && ev.Namespace == pool.Namespace
 	})).Return(nil)
 
 	info := fftypes.JSONObject{"some": "info"}
-	err := em.TokenPoolCreated(mti, pool, "0x12345", "tx1", info)
+	err := am.TokenPoolCreated(mti, pool, "0x12345", "tx1", info)
 	assert.NoError(t, err)
 	mdi.AssertExpectations(t)
 }
 
 func TestTokenPoolBadNamespace(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	am, cancel := newTestAssets(t)
 	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
+	mdi := am.database.(*databasemocks.Plugin)
 	mti := &tokenmocks.Plugin{}
 
 	pool := &fftypes.TokenPool{
@@ -91,20 +91,20 @@ func TestTokenPoolBadNamespace(t *testing.T) {
 		},
 	}
 
-	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *fftypes.Event) bool {
+	mdi.On("InsertEvent", am.ctx, mock.MatchedBy(func(ev *fftypes.Event) bool {
 		return ev.Type == fftypes.EventTypePoolRejected && ev.Reference == pool.ID && ev.Namespace == pool.Namespace
 	})).Return(nil)
 
 	info := fftypes.JSONObject{"some": "info"}
-	err := em.TokenPoolCreated(mti, pool, "0x12345", "tx1", info)
+	err := am.TokenPoolCreated(mti, pool, "0x12345", "tx1", info)
 	assert.NoError(t, err)
 	mdi.AssertExpectations(t)
 }
 
 func TestTokenPoolBadName(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	am, cancel := newTestAssets(t)
 	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
+	mdi := am.database.(*databasemocks.Plugin)
 	mti := &tokenmocks.Plugin{}
 
 	pool := &fftypes.TokenPool{
@@ -117,23 +117,23 @@ func TestTokenPoolBadName(t *testing.T) {
 	}
 
 	mdi.On("GetTransactionByID", mock.Anything, pool.TX.ID).Return(nil, nil)
-	mdi.On("UpsertTransaction", em.ctx, mock.MatchedBy(func(tx *fftypes.Transaction) bool {
+	mdi.On("UpsertTransaction", am.ctx, mock.MatchedBy(func(tx *fftypes.Transaction) bool {
 		return tx.Subject.Type == fftypes.TransactionTypeTokenPool
 	}), false).Return(nil)
-	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *fftypes.Event) bool {
+	mdi.On("InsertEvent", am.ctx, mock.MatchedBy(func(ev *fftypes.Event) bool {
 		return ev.Type == fftypes.EventTypePoolRejected && ev.Reference == pool.ID && ev.Namespace == pool.Namespace
 	})).Return(nil)
 
 	info := fftypes.JSONObject{"some": "info"}
-	err := em.TokenPoolCreated(mti, pool, "0x12345", "tx1", info)
+	err := am.TokenPoolCreated(mti, pool, "0x12345", "tx1", info)
 	assert.NoError(t, err)
 	mdi.AssertExpectations(t)
 }
 
 func TestTokenPoolGetTransactionFail(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	am, cancel := newTestAssets(t)
 	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
+	mdi := am.database.(*databasemocks.Plugin)
 
 	pool := &fftypes.TokenPool{
 		ID: fftypes.NewUUID(),
@@ -148,16 +148,16 @@ func TestTokenPoolGetTransactionFail(t *testing.T) {
 	mdi.On("GetTransactionByID", mock.Anything, mock.Anything).Return(nil, fmt.Errorf("pop"))
 
 	info := fftypes.JSONObject{"some": "info"}
-	valid, err := em.persistTokenPoolTransaction(em.ctx, pool, "0x12345", "tx1", info)
+	valid, err := am.persistTokenPoolTransaction(am.ctx, pool, "0x12345", "tx1", info)
 	assert.EqualError(t, err, "pop")
 	assert.False(t, valid)
 	mdi.AssertExpectations(t)
 }
 
 func TestTokenPoolGetTransactionInvalidMatch(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	am, cancel := newTestAssets(t)
 	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
+	mdi := am.database.(*databasemocks.Plugin)
 
 	pool := &fftypes.TokenPool{
 		ID: fftypes.NewUUID(),
@@ -174,16 +174,16 @@ func TestTokenPoolGetTransactionInvalidMatch(t *testing.T) {
 	}, nil)
 
 	info := fftypes.JSONObject{"some": "info"}
-	valid, err := em.persistTokenPoolTransaction(em.ctx, pool, "0x12345", "tx1", info)
+	valid, err := am.persistTokenPoolTransaction(am.ctx, pool, "0x12345", "tx1", info)
 	assert.NoError(t, err)
 	assert.False(t, valid)
 	mdi.AssertExpectations(t)
 }
 
 func TestTokenPoolNewTXUpsertFail(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	am, cancel := newTestAssets(t)
 	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
+	mdi := am.database.(*databasemocks.Plugin)
 
 	pool := &fftypes.TokenPool{
 		ID: fftypes.NewUUID(),
@@ -199,16 +199,16 @@ func TestTokenPoolNewTXUpsertFail(t *testing.T) {
 	mdi.On("UpsertTransaction", mock.Anything, mock.Anything, false).Return(fmt.Errorf("pop"))
 
 	info := fftypes.JSONObject{"some": "info"}
-	valid, err := em.persistTokenPoolTransaction(em.ctx, pool, "0x12345", "tx1", info)
+	valid, err := am.persistTokenPoolTransaction(am.ctx, pool, "0x12345", "tx1", info)
 	assert.EqualError(t, err, "pop")
 	assert.False(t, valid)
 	mdi.AssertExpectations(t)
 }
 
 func TestTokenPoolExistingTXHashMismatch(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	am, cancel := newTestAssets(t)
 	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
+	mdi := am.database.(*databasemocks.Plugin)
 
 	pool := &fftypes.TokenPool{
 		ID: fftypes.NewUUID(),
@@ -231,14 +231,14 @@ func TestTokenPoolExistingTXHashMismatch(t *testing.T) {
 	mdi.On("UpsertTransaction", mock.Anything, mock.Anything, false).Return(database.HashMismatch)
 
 	info := fftypes.JSONObject{"some": "info"}
-	valid, err := em.persistTokenPoolTransaction(em.ctx, pool, "0x12345", "tx1", info)
+	valid, err := am.persistTokenPoolTransaction(am.ctx, pool, "0x12345", "tx1", info)
 	assert.NoError(t, err)
 	assert.False(t, valid)
 	mdi.AssertExpectations(t)
 }
 
 func TestTokenPoolIDMismatch(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	em, cancel := newTestAssets(t)
 	defer cancel()
 	mdi := em.database.(*databasemocks.Plugin)
 	mti := &tokenmocks.Plugin{}
@@ -269,7 +269,7 @@ func TestTokenPoolIDMismatch(t *testing.T) {
 }
 
 func TestTokenPoolUpsertFailAndRetry(t *testing.T) {
-	em, cancel := newTestEventManager(t)
+	em, cancel := newTestAssets(t)
 	defer cancel()
 	mdi := em.database.(*databasemocks.Plugin)
 	mti := &tokenmocks.Plugin{}
diff --git a/internal/orchestrator/bound_callbacks_test.go b/internal/orchestrator/bound_callbacks_test.go
index 8be411d0d..893a5f3e5 100644
--- a/internal/orchestrator/bound_callbacks_test.go
+++ b/internal/orchestrator/bound_callbacks_test.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/hyperledger/firefly/mocks/assetmocks"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
 	"github.com/hyperledger/firefly/mocks/dataexchangemocks"
 	"github.com/hyperledger/firefly/mocks/eventmocks"
@@ -34,7 +35,8 @@ func TestBoundCallbacks(t *testing.T) {
 	mbi := &blockchainmocks.Plugin{}
 	mdx := &dataexchangemocks.Plugin{}
 	mti := &tokenmocks.Plugin{}
-	bc := boundCallbacks{bi: mbi, dx: mdx, ei: mei}
+	mam := &assetmocks.Manager{}
+	bc := boundCallbacks{bi: mbi, dx: mdx, ei: mei, am: mam}
 
 	info := fftypes.JSONObject{"hello": "world"}
 	batch := &blockchain.BatchPin{TransactionID: fftypes.NewUUID()}
@@ -66,7 +68,7 @@ func TestBoundCallbacks(t *testing.T) {
 	err = bc.MessageReceived("peer1", []byte{})
 	assert.EqualError(t, err, "pop")
 
-	mei.On("TokenPoolCreated", mti, pool, "0x12345", "tx12345", info).Return(fmt.Errorf("pop"))
+	mam.On("TokenPoolCreated", mti, pool, "0x12345", "tx12345", info).Return(fmt.Errorf("pop"))
 	err = bc.TokenPoolCreated(mti, pool, "0x12345", "tx12345", info)
 	assert.EqualError(t, err, "pop")
 }