Repository: hyperledger/firefly
Pull request #1223 (2023-03-17 01:23:26)

Ensure that when we broadcast we use the org identity, not the default NS identity
Since 1.1 we've had both an org identity (required for mulit-party), and a default namespace identity (optional).
The code for definition broadcasts must use the org identity for that broadcast to be accepted, but the code was using the same priority as generic API calls - which use the NS default identity in preference.

This PR updates that so it's always the org identity, and the call will not fall-back to the default.

> Found this while discussing https://github.com/hyperledger/firefly/pull/1222 and hence in a PR chain with that change


Patch:
diff --git a/internal/definitions/sender.go b/internal/definitions/sender.go
index 2bf87a688..e41f8d8ef 100644
--- a/internal/definitions/sender.go
+++ b/internal/definitions/sender.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -105,7 +105,14 @@ func (bm *definitionSender) Name() string {
 }
 
 func (bm *definitionSender) sendDefinitionDefault(ctx context.Context, def core.Definition, tag string, waitConfirm bool) (msg *core.Message, err error) {
-	return bm.sendDefinition(ctx, def, &core.SignerRef{ /* resolve to node default */ }, tag, waitConfirm)
+	org, err := bm.identity.GetMultipartyRootOrg(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return bm.sendDefinition(ctx, def, &core.SignerRef{ /* resolve to node default */
+		Author: org.DID,
+	}, tag, waitConfirm)
 }
 
 func (bm *definitionSender) sendDefinition(ctx context.Context, def core.Definition, signingIdentity *core.SignerRef, tag string, waitConfirm bool) (msg *core.Message, err error) {
diff --git a/internal/events/event_manager.go b/internal/events/event_manager.go
index c0322f8cf..7f98b9195 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -78,7 +78,7 @@ type EventManager interface {
 	SharedStorageBlobDownloaded(ss sharedstorage.Plugin, hash fftypes.Bytes32, size int64, payloadRef string, dataID *fftypes.UUID)
 
 	// Bound token callbacks
-	TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPool) error
+	TokenPoolCreated(ctx context.Context /* allows security context to be propagated when called in-line with the send TX */, ti tokens.Plugin, pool *tokens.TokenPool) error
 	TokensTransferred(ti tokens.Plugin, transfer *tokens.TokenTransfer) error
 	TokensApproved(ti tokens.Plugin, approval *tokens.TokenApproval) error
 
diff --git a/internal/events/token_pool_created.go b/internal/events/token_pool_created.go
index 7d06b835c..559598fa4 100644
--- a/internal/events/token_pool_created.go
+++ b/internal/events/token_pool_created.go
@@ -51,7 +51,7 @@ func addPoolDetailsFromPlugin(ffPool *core.TokenPool, pluginPool *tokens.TokenPo
 }
 
 func (em *eventManager) confirmPool(ctx context.Context, pool *core.TokenPool, ev *blockchain.Event) error {
-	log.L(em.ctx).Debugf("Confirming pool ID=%s Locator='%s'", pool.ID, pool.Locator)
+	log.L(ctx).Debugf("Confirming pool ID=%s Locator='%s'", pool.ID, pool.Locator)
 	var blockchainID string
 	if ev != nil {
 		// Some pools will not include a blockchain event for creation (such as when indexing a pre-existing pool)
@@ -132,12 +132,16 @@ func (em *eventManager) loadFromOperation(ctx context.Context, pool *tokens.Toke
 // It will be at least invoked on the submitter when the pool is first created, to trigger the submitter to announce it.
 // It will be invoked on every node (including the submitter) after the pool is announced+activated, to trigger confirmation of the pool.
 // When received in any other scenario, it should be ignored.
-func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPool) (err error) {
+//
+// The context passed to this callback is dependent on what phase it is called in.
+// In the case that it is called synchronously on the submitter, in order to trigger the announcement, the original context
+// of the REST API will be propagated (so it can be used for the resolution of the org signing key).
+func (em *eventManager) TokenPoolCreated(ctx context.Context, ti tokens.Plugin, pool *tokens.TokenPool) (err error) {
 	var msgIDforRewind *fftypes.UUID
 	var announcePool *core.TokenPool
 
-	err = em.retry.Do(em.ctx, "persist token pool transaction", func(attempt int) (bool, error) {
-		err := em.database.RunAsGroup(em.ctx, func(ctx context.Context) error {
+	err = em.retry.Do(ctx, "persist token pool transaction", func(attempt int) (bool, error) {
+		err := em.database.RunAsGroup(ctx, func(ctx context.Context) error {
 			// See if this is a confirmation of an unconfirmed pool
 			existingPool, err := em.loadExisting(ctx, pool)
 			if err != nil {
@@ -145,7 +149,7 @@ func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPoo
 			}
 			if existingPool != nil {
 				if existingPool.State == core.TokenPoolStateConfirmed {
-					log.L(em.ctx).Debugf("Token pool ID=%s Locator='%s' already confirmed", existingPool.ID, pool.PoolLocator)
+					log.L(ctx).Debugf("Token pool ID=%s Locator='%s' already confirmed", existingPool.ID, pool.PoolLocator)
 					return nil // already confirmed
 				}
 				msgIDforRewind = existingPool.Message
@@ -153,7 +157,7 @@ func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPoo
 			} else if pool.TX.ID == nil {
 				// TransactionID is required if the pool doesn't exist yet
 				// (but it may be omitted when activating a pool that was received via definition broadcast)
-				log.L(em.ctx).Errorf("Invalid token pool transaction - ID is nil")
+				log.L(ctx).Errorf("Invalid token pool transaction - ID is nil")
 				return nil // move on
 			}
 
@@ -184,16 +188,16 @@ func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPoo
 		if announcePool != nil {
 			// If the pool is tied to a contract interface, resolve the methods to be used for later operations
 			if announcePool.Interface != nil && announcePool.Interface.ID != nil && announcePool.InterfaceFormat != "" {
-				log.L(em.ctx).Infof("Querying token connector interface, id=%s", announcePool.ID)
-				if err := em.assets.ResolvePoolMethods(em.ctx, announcePool); err != nil {
+				log.L(ctx).Infof("Querying token connector interface, id=%s", announcePool.ID)
+				if err := em.assets.ResolvePoolMethods(ctx, announcePool); err != nil {
 					return err
 				}
 			}
 
 			// Announce the details of the new token pool
 			// Other nodes will pass these details to their own token connector for validation/activation of the pool
-			log.L(em.ctx).Infof("Announcing token pool, id=%s", announcePool.ID)
-			err = em.defsender.DefineTokenPool(em.ctx, &core.TokenPoolAnnouncement{
+			log.L(ctx).Infof("Announcing token pool, id=%s", announcePool.ID)
+			err = em.defsender.DefineTokenPool(ctx, &core.TokenPoolAnnouncement{
 				Pool: announcePool,
 			}, false)
 		}
diff --git a/internal/tokens/fftokens/fftokens.go b/internal/tokens/fftokens/fftokens.go
index adc1ebefb..b60e4e017 100644
--- a/internal/tokens/fftokens/fftokens.go
+++ b/internal/tokens/fftokens/fftokens.go
@@ -70,7 +70,7 @@ func (cb *callbacks) OperationUpdate(ctx context.Context, nsOpID string, status
 func (cb *callbacks) TokenPoolCreated(ctx context.Context, pool *tokens.TokenPool) error {
 	// Deliver token pool creation events to every handler
 	for _, handler := range cb.handlers {
-		if err := handler.TokenPoolCreated(cb.plugin, pool); err != nil {
+		if err := handler.TokenPoolCreated(ctx, cb.plugin, pool); err != nil {
 			return err
 		}
 	}
diff --git a/mocks/eventmocks/event_manager.go b/mocks/eventmocks/event_manager.go
index f61da6151..b67e116cd 100644
--- a/mocks/eventmocks/event_manager.go
+++ b/mocks/eventmocks/event_manager.go
@@ -288,13 +288,13 @@ func (_m *EventManager) SubscriptionUpdates() chan<- *fftypes.UUID {
 	return r0
 }
 
-// TokenPoolCreated provides a mock function with given fields: ti, pool
-func (_m *EventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPool) error {
-	ret := _m.Called(ti, pool)
+// TokenPoolCreated provides a mock function with given fields: ctx, ti, pool
+func (_m *EventManager) TokenPoolCreated(ctx context.Context, ti tokens.Plugin, pool *tokens.TokenPool) error {
+	ret := _m.Called(ctx, ti, pool)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(tokens.Plugin, *tokens.TokenPool) error); ok {
-		r0 = rf(ti, pool)
+	if rf, ok := ret.Get(0).(func(context.Context, tokens.Plugin, *tokens.TokenPool) error); ok {
+		r0 = rf(ctx, ti, pool)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/mocks/tokenmocks/callbacks.go b/mocks/tokenmocks/callbacks.go
index e138da9fb..3c19a8ecd 100644
--- a/mocks/tokenmocks/callbacks.go
+++ b/mocks/tokenmocks/callbacks.go
@@ -3,6 +3,8 @@
 package tokenmocks
 
 import (
+	context "context"
+
 	tokens "github.com/hyperledger/firefly/pkg/tokens"
 	mock "github.com/stretchr/testify/mock"
 )
@@ -12,13 +14,13 @@ type Callbacks struct {
 	mock.Mock
 }
 
-// TokenPoolCreated provides a mock function with given fields: plugin, pool
-func (_m *Callbacks) TokenPoolCreated(plugin tokens.Plugin, pool *tokens.TokenPool) error {
-	ret := _m.Called(plugin, pool)
+// TokenPoolCreated provides a mock function with given fields: ctx, plugin, pool
+func (_m *Callbacks) TokenPoolCreated(ctx context.Context, plugin tokens.Plugin, pool *tokens.TokenPool) error {
+	ret := _m.Called(ctx, plugin, pool)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(tokens.Plugin, *tokens.TokenPool) error); ok {
-		r0 = rf(plugin, pool)
+	if rf, ok := ret.Get(0).(func(context.Context, tokens.Plugin, *tokens.TokenPool) error); ok {
+		r0 = rf(ctx, plugin, pool)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/pkg/tokens/plugin.go b/pkg/tokens/plugin.go
index 941a56fcd..8f581c8a9 100644
--- a/pkg/tokens/plugin.go
+++ b/pkg/tokens/plugin.go
@@ -81,7 +81,11 @@ type Callbacks interface {
 	// submitted by us, or by any other authorized party in the network.
 	//
 	// Error should only be returned in shutdown scenarios
-	TokenPoolCreated(plugin Plugin, pool *TokenPool) error
+	//
+	// Note: The context is passed on this callback (unlike most callbacks), as it might be
+	//       involved in-line with the original REST API call in the special case of the
+	//       submitter.
+	TokenPoolCreated(ctx context.Context, plugin Plugin, pool *TokenPool) error
 
 	// TokensTransferred notifies on a transfer between token accounts.
 	//


Test patch:
diff --git a/internal/definitions/sender_contracts_test.go b/internal/definitions/sender_contracts_test.go
index ce954408b..74ea7776c 100644
--- a/internal/definitions/sender_contracts_test.go
+++ b/internal/definitions/sender_contracts_test.go
@@ -62,7 +62,7 @@ func TestDefineFFIFail(t *testing.T) {
 	mcm.On("ResolveFFI", context.Background(), ffi).Return(nil)
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
-	mim.On("ResolveInputSigningIdentity", context.Background(), mock.Anything).Return(fmt.Errorf("pop"))
+	mim.On("GetMultipartyRootOrg", context.Background()).Return(nil, fmt.Errorf("pop"))
 
 	err := ds.DefineFFI(context.Background(), ffi, false)
 	assert.EqualError(t, err, "pop")
@@ -82,6 +82,11 @@ func TestDefineFFIOk(t *testing.T) {
 	mcm.On("ResolveFFI", context.Background(), ffi).Return(nil)
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
+	mim.On("GetMultipartyRootOrg", context.Background()).Return(&core.Identity{
+		IdentityBase: core.IdentityBase{
+			DID: "firefly:org1",
+		},
+	}, nil)
 	mim.On("ResolveInputSigningIdentity", context.Background(), mock.Anything).Return(nil)
 
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
@@ -137,7 +142,7 @@ func TestDefineContractAPIFail(t *testing.T) {
 	mcm.On("ResolveContractAPI", context.Background(), url, api).Return(nil)
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
-	mim.On("ResolveInputSigningIdentity", context.Background(), mock.Anything).Return(fmt.Errorf("pop"))
+	mim.On("GetMultipartyRootOrg", context.Background()).Return(nil, fmt.Errorf("pop"))
 
 	err := ds.DefineContractAPI(context.Background(), url, api, false)
 	assert.EqualError(t, err, "pop")
@@ -158,6 +163,11 @@ func TestDefineContractAPIOk(t *testing.T) {
 	mcm.On("ResolveContractAPI", context.Background(), url, api).Return(nil)
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
+	mim.On("GetMultipartyRootOrg", context.Background()).Return(&core.Identity{
+		IdentityBase: core.IdentityBase{
+			DID: "firefly:org1",
+		},
+	}, nil)
 	mim.On("ResolveInputSigningIdentity", context.Background(), mock.Anything).Return(nil)
 
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
diff --git a/internal/definitions/sender_datatype_test.go b/internal/definitions/sender_datatype_test.go
index d3d2a7f0d..86cdc2237 100644
--- a/internal/definitions/sender_datatype_test.go
+++ b/internal/definitions/sender_datatype_test.go
@@ -48,6 +48,11 @@ func TestBroadcastDatatypeBadValue(t *testing.T) {
 	mdm := ds.data.(*datamocks.Manager)
 	mdm.On("CheckDatatype", mock.Anything, mock.Anything).Return(nil)
 	mim := ds.identity.(*identitymanagermocks.Manager)
+	mim.On("GetMultipartyRootOrg", context.Background()).Return(&core.Identity{
+		IdentityBase: core.IdentityBase{
+			DID: "firefly:org1",
+		},
+	}, nil)
 	mim.On("ResolveInputSigningIdentity", mock.Anything, mock.Anything).Return(nil)
 	err := ds.DefineDatatype(context.Background(), &core.Datatype{
 		Namespace: "ns1",
@@ -91,6 +96,11 @@ func TestBroadcastOk(t *testing.T) {
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
 	mms := &syncasyncmocks.Sender{}
 
+	mim.On("GetMultipartyRootOrg", context.Background()).Return(&core.Identity{
+		IdentityBase: core.IdentityBase{
+			DID: "firefly:org1",
+		},
+	}, nil)
 	mim.On("ResolveInputSigningIdentity", mock.Anything, mock.Anything).Return(nil)
 	mdm.On("CheckDatatype", mock.Anything, mock.Anything).Return(nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
diff --git a/internal/definitions/sender_test.go b/internal/definitions/sender_test.go
index 5e86bc917..3adce3c31 100644
--- a/internal/definitions/sender_test.go
+++ b/internal/definitions/sender_test.go
@@ -73,6 +73,11 @@ func TestCreateDefinitionConfirm(t *testing.T) {
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
 	mms := &syncasyncmocks.Sender{}
 
+	mim.On("GetMultipartyRootOrg", ds.ctx).Return(&core.Identity{
+		IdentityBase: core.IdentityBase{
+			DID: "firefly:org1",
+		},
+	}, nil)
 	mim.On("ResolveInputSigningIdentity", mock.Anything, mock.Anything).Return(nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
 	mms.On("SendAndWait", mock.Anything).Return(nil)
@@ -94,6 +99,11 @@ func TestCreateDatatypeDefinitionAsNodeConfirm(t *testing.T) {
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
 	mms := &syncasyncmocks.Sender{}
 
+	mim.On("GetMultipartyRootOrg", ds.ctx).Return(&core.Identity{
+		IdentityBase: core.IdentityBase{
+			DID: "firefly:org1",
+		},
+	}, nil)
 	mim.On("ResolveInputSigningIdentity", mock.Anything, mock.Anything).Return(nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
 	mms.On("SendAndWait", mock.Anything).Return(nil)
diff --git a/internal/definitions/sender_tokenpool_test.go b/internal/definitions/sender_tokenpool_test.go
index 918916bf7..6842bcac5 100644
--- a/internal/definitions/sender_tokenpool_test.go
+++ b/internal/definitions/sender_tokenpool_test.go
@@ -127,11 +127,16 @@ func TestDefineTokenPoolOk(t *testing.T) {
 		},
 	}
 
+	mim.On("GetMultipartyRootOrg", ds.ctx).Return(&core.Identity{
+		IdentityBase: core.IdentityBase{
+			DID: "firefly:org1",
+		},
+	}, nil)
 	mim.On("ResolveInputSigningIdentity", mock.Anything, mock.Anything).Return(nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
-	mms.On("Send", context.Background()).Return(nil)
+	mms.On("Send", ds.ctx).Return(nil)
 
-	err := ds.DefineTokenPool(context.Background(), pool, false)
+	err := ds.DefineTokenPool(ds.ctx, pool, false)
 	assert.NoError(t, err)
 
 	mdm.AssertExpectations(t)
diff --git a/internal/events/token_pool_created_test.go b/internal/events/token_pool_created_test.go
index 5723d60a2..e0afa9e21 100644
--- a/internal/events/token_pool_created_test.go
+++ b/internal/events/token_pool_created_test.go
@@ -55,7 +55,7 @@ func TestTokenPoolCreatedIgnore(t *testing.T) {
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil, nil)
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 }
@@ -79,7 +79,7 @@ func TestTokenPoolCreatedIgnoreNoTX(t *testing.T) {
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 }
 
@@ -134,7 +134,7 @@ func TestTokenPoolCreatedConfirm(t *testing.T) {
 		return e.Type == core.EventTypePoolConfirmed && *e.Reference == *storedPool.ID
 	})).Return(nil).Once()
 
-	err := em.TokenPoolCreated(mti, chainPool)
+	err := em.TokenPoolCreated(em.ctx, mti, chainPool)
 	assert.NoError(t, err)
 
 	assert.Equal(t, "ERC1155", storedPool.Standard)
@@ -176,7 +176,7 @@ func TestTokenPoolCreatedAlreadyConfirmed(t *testing.T) {
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(storedPool, nil)
 
-	err := em.TokenPoolCreated(mti, chainPool)
+	err := em.TokenPoolCreated(em.ctx, mti, chainPool)
 	assert.NoError(t, err)
 
 }
@@ -220,7 +220,7 @@ func TestTokenPoolCreatedConfirmFailBadSymbol(t *testing.T) {
 		ID: opID,
 	}}, nil, nil)
 
-	err := em.TokenPoolCreated(mti, chainPool)
+	err := em.TokenPoolCreated(em.ctx, mti, chainPool)
 	assert.NoError(t, err)
 
 }
@@ -364,7 +364,7 @@ func TestTokenPoolCreatedAnnounce(t *testing.T) {
 		return pool.Pool.Namespace == "ns1" && pool.Pool.Name == "my-pool" && *pool.Pool.ID == *poolID
 	}), false).Return(nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 	mti.AssertExpectations(t)
@@ -415,7 +415,7 @@ func TestTokenPoolCreatedAnnounceBadInterface(t *testing.T) {
 		return pool.Locator == "123" && pool.Name == "my-pool"
 	})).Return(fmt.Errorf("pop"))
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.EqualError(t, err, "pop")
 
 	mti.AssertExpectations(t)
@@ -453,7 +453,7 @@ func TestTokenPoolCreatedAnnounceBadOpInputID(t *testing.T) {
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 }
@@ -492,7 +492,7 @@ func TestTokenPoolCreatedAnnounceBadOpInputNS(t *testing.T) {
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 }
@@ -536,7 +536,7 @@ func TestTokenPoolCreatedAnnounceBadSymbol(t *testing.T) {
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop")).Once()
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil).Once()
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 	mti.AssertExpectations(t)
diff --git a/internal/tokens/fftokens/fftokens_test.go b/internal/tokens/fftokens/fftokens_test.go
index 9736cfe55..3f1f41ac8 100644
--- a/internal/tokens/fftokens/fftokens_test.go
+++ b/internal/tokens/fftokens/fftokens_test.go
@@ -267,11 +267,13 @@ func TestCreateTokenPoolSynchronous(t *testing.T) {
 	mcb := &tokenmocks.Callbacks{}
 	h.SetHandler("ns1", mcb)
 
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	ctx, cancelCtx := context.WithCancel(context.Background())
+	defer cancelCtx()
+	mcb.On("TokenPoolCreated", ctx /* important this gets through */, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && *p.TX.ID == *pool.TX.ID
 	})).Return(nil)
 
-	complete, err := h.CreateTokenPool(context.Background(), nsOpID, pool)
+	complete, err := h.CreateTokenPool(ctx, nsOpID, pool)
 	assert.True(t, complete)
 	assert.NoError(t, err)
 }
@@ -425,7 +427,7 @@ func TestActivateTokenPoolSynchronous(t *testing.T) {
 
 	mcb := &tokenmocks.Callbacks{}
 	h.SetHandler("ns1", mcb)
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && p.TX.ID == nil && p.Event == nil
 	})).Return(nil)
 
@@ -473,7 +475,7 @@ func TestActivateTokenPoolSynchronousBadResponse(t *testing.T) {
 
 	mcb := &tokenmocks.Callbacks{}
 	h.SetHandler("ns1", mcb)
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && p.TX.ID == nil
 	})).Return(nil)
 
@@ -993,7 +995,7 @@ func TestPoolEvents(t *testing.T) {
 	assert.Equal(t, `{"data":{"id":"6"},"event":"ack"}`, string(msg))
 
 	// token-pool: invalid uuid (success)
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && p.TX.ID == nil && p.Event.ProtocolID == "000000000010/000020/000030"
 	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
@@ -1017,7 +1019,7 @@ func TestPoolEvents(t *testing.T) {
 	assert.Equal(t, `{"data":{"id":"7"},"event":"ack"}`, string(msg))
 
 	// token-pool: success
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && txID.Equals(p.TX.ID) && p.Event.ProtocolID == "000000000010/000020/000030"
 	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
@@ -1042,7 +1044,7 @@ func TestPoolEvents(t *testing.T) {
 	assert.Equal(t, `{"data":{"id":"8"},"event":"ack"}`, string(msg))
 
 	// token-pool: batch + callback fail
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && txID.Equals(p.TX.ID) && p.Event.ProtocolID == "000000000010/000020/000030"
 	})).Return(fmt.Errorf("pop")).Once()
 	fromServer <- fftypes.JSONObject{