Repository: hyperledger/firefly
Pull request #1234 (2023-03-22 22:44:02)

Validate and restore listeners on startup, and allow deletion if not found in connector
Had a situation where we wanted to restore fresh listeners for custom contracts on a FireFly core service, and had assumed FireFly core would re-create the listeners. So deleted them on the EVMConnect runtime directly and restarted FF.

Found related issues, that meant we had to go into the FF DB and remove the contract listeners directly (and have the app recreate them):
1. FireFly did not recreate the listener. This seems a little inconsistent with both the `BatchPin` listener, and the listeners of the token connectors...
   - Proposing in this PR that we automatically recreate the listeners where possible (only Ethereum in this PR, due to existing limitations around the Fabric connector integration where it does not have existing query support to check existence)
2. The API could not be used to delete the listener, because when EVMConnect returned a `404` FireFly failed the `DELETE` API.
    - This seemed like a straight bug to me. Due to the return of status being `interface{}` and the complexity of a `nil` check there, we're adding an extra `bool` to whether it's found or not (Fabric always returns true)
3. Found in testing this change, that the check on existence of a contract listener with the same parameters was not working, passing in a `[]byte` for the `location` rather than `string`


Patch:
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index 5e8ead9c0..aa79a01a4 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -780,7 +780,7 @@ func (e *Ethereum) encodeContractLocation(ctx context.Context, location *Locatio
 	return result, err
 }
 
-func (e *Ethereum) AddContractListener(ctx context.Context, listener *core.ContractListenerInput) (err error) {
+func (e *Ethereum) AddContractListener(ctx context.Context, listener *core.ContractListener) (err error) {
 	var location *Location
 	if listener.Location != nil {
 		location, err = e.parseContractLocation(ctx, listener.Location)
@@ -806,14 +806,14 @@ func (e *Ethereum) AddContractListener(ctx context.Context, listener *core.Contr
 	return nil
 }
 
-func (e *Ethereum) DeleteContractListener(ctx context.Context, subscription *core.ContractListener) error {
-	return e.streams.deleteSubscription(ctx, subscription.BackendID)
+func (e *Ethereum) DeleteContractListener(ctx context.Context, subscription *core.ContractListener, okNotFound bool) error {
+	return e.streams.deleteSubscription(ctx, subscription.BackendID, okNotFound)
 }
 
-func (e *Ethereum) GetContractListenerStatus(ctx context.Context, subID string) (status interface{}, err error) {
-	sub, err := e.streams.getSubscription(ctx, subID)
-	if err != nil {
-		return nil, err
+func (e *Ethereum) GetContractListenerStatus(ctx context.Context, subID string, okNotFound bool) (found bool, status interface{}, err error) {
+	sub, err := e.streams.getSubscription(ctx, subID, okNotFound)
+	if err != nil || sub == nil {
+		return false, nil, err
 	}
 
 	checkpoint := &ListenerStatus{
@@ -825,7 +825,7 @@ func (e *Ethereum) GetContractListenerStatus(ctx context.Context, subID string)
 		},
 	}
 
-	return checkpoint, nil
+	return true, checkpoint, nil
 }
 
 func (e *Ethereum) GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error) {
diff --git a/internal/blockchain/ethereum/eventstream.go b/internal/blockchain/ethereum/eventstream.go
index 139f6c621..4ec519530 100644
--- a/internal/blockchain/ethereum/eventstream.go
+++ b/internal/blockchain/ethereum/eventstream.go
@@ -151,12 +151,15 @@ func (s *streamManager) getSubscriptions(ctx context.Context) (subs []*subscript
 	return subs, nil
 }
 
-func (s *streamManager) getSubscription(ctx context.Context, subID string) (sub *subscription, err error) {
+func (s *streamManager) getSubscription(ctx context.Context, subID string, okNotFound bool) (sub *subscription, err error) {
 	res, err := s.client.R().
 		SetContext(ctx).
 		SetResult(&sub).
 		Get(fmt.Sprintf("/subscriptions/%s", subID))
 	if err != nil || !res.IsSuccess() {
+		if okNotFound && res.StatusCode() == 404 {
+			return nil, nil
+		}
 		return nil, ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgEthConnectorRESTErr)
 	}
 	return sub, nil
@@ -167,7 +170,7 @@ func (s *streamManager) getSubscriptionName(ctx context.Context, subID string) (
 		return cachedValue, nil
 	}
 
-	sub, err := s.getSubscription(ctx, subID)
+	sub, err := s.getSubscription(ctx, subID, false)
 	if err != nil {
 		return "", err
 	}
@@ -205,11 +208,14 @@ func (s *streamManager) createSubscription(ctx context.Context, location *Locati
 	return &sub, nil
 }
 
-func (s *streamManager) deleteSubscription(ctx context.Context, subID string) error {
+func (s *streamManager) deleteSubscription(ctx context.Context, subID string, okNotFound bool) error {
 	res, err := s.client.R().
 		SetContext(ctx).
 		Delete("/subscriptions/" + subID)
 	if err != nil || !res.IsSuccess() {
+		if okNotFound && res.StatusCode() == 404 {
+			return nil
+		}
 		return ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgEthConnectorRESTErr)
 	}
 	return nil
diff --git a/internal/blockchain/fabric/eventstream.go b/internal/blockchain/fabric/eventstream.go
index 53f002e64..b055d1484 100644
--- a/internal/blockchain/fabric/eventstream.go
+++ b/internal/blockchain/fabric/eventstream.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -185,11 +185,14 @@ func (s *streamManager) createSubscription(ctx context.Context, location *Locati
 	return &sub, nil
 }
 
-func (s *streamManager) deleteSubscription(ctx context.Context, subID string) error {
+func (s *streamManager) deleteSubscription(ctx context.Context, subID string, okNotFound bool) error {
 	res, err := s.client.R().
 		SetContext(ctx).
 		Delete("/subscriptions/" + subID)
 	if err != nil || !res.IsSuccess() {
+		if okNotFound && res.StatusCode() == 404 {
+			return nil
+		}
 		return ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgFabconnectRESTErr)
 	}
 	return nil
diff --git a/internal/blockchain/fabric/fabric.go b/internal/blockchain/fabric/fabric.go
index aaf9c0b5f..1a35405ff 100644
--- a/internal/blockchain/fabric/fabric.go
+++ b/internal/blockchain/fabric/fabric.go
@@ -815,7 +815,7 @@ func encodeContractLocation(ctx context.Context, location *Location) (result *ff
 	return result, err
 }
 
-func (f *Fabric) AddContractListener(ctx context.Context, listener *core.ContractListenerInput) error {
+func (f *Fabric) AddContractListener(ctx context.Context, listener *core.ContractListener) error {
 	location, err := parseContractLocation(ctx, listener.Location)
 	if err != nil {
 		return err
@@ -830,13 +830,13 @@ func (f *Fabric) AddContractListener(ctx context.Context, listener *core.Contrac
 	return nil
 }
 
-func (f *Fabric) DeleteContractListener(ctx context.Context, subscription *core.ContractListener) error {
-	return f.streams.deleteSubscription(ctx, subscription.BackendID)
+func (f *Fabric) DeleteContractListener(ctx context.Context, subscription *core.ContractListener, okNotFound bool) error {
+	return f.streams.deleteSubscription(ctx, subscription.BackendID, okNotFound)
 }
 
-func (f *Fabric) GetContractListenerStatus(ctx context.Context, subID string) (interface{}, error) {
-	// Fabconnect does not currently provide any additional status info for listener subscriptions
-	return nil, nil
+func (f *Fabric) GetContractListenerStatus(ctx context.Context, subID string, okNotFound bool) (bool, interface{}, error) {
+	// Fabconnect does not currently provide any additional status info for listener subscriptions.
+	return true, nil, nil
 }
 
 func (f *Fabric) GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error) {
diff --git a/internal/contracts/manager.go b/internal/contracts/manager.go
index 75e93f979..551dfb79e 100644
--- a/internal/contracts/manager.go
+++ b/internal/contracts/manager.go
@@ -24,6 +24,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/ffapi"
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
+	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/identity"
 	"github.com/hyperledger/firefly/internal/operations"
@@ -106,7 +107,12 @@ func NewContractManager(ctx context.Context, ns string, di database.Plugin, bi b
 		core.OpTypeBlockchainContractDeploy,
 	})
 
-	return cm, nil
+	// Validate all our listeners exist on startup - consistent with the multi-party manager.
+	// This means if EVMConnect is cleared, simply a restart of the namespace on core will
+	// cause recreation of all the listeners (noting that listeners that were specified to start
+	// from latest, will start from the new latest rather than replaying from the block they
+	// started from before they were deleted).
+	return cm, cm.verifyListeners(ctx)
 }
 
 func (cm *contractManager) Name() string {
@@ -196,6 +202,32 @@ func (cm *contractManager) GetFFIs(ctx context.Context, filter ffapi.AndFilter)
 	return cm.database.GetFFIs(ctx, cm.namespace, filter)
 }
 
+func (cm *contractManager) verifyListeners(ctx context.Context) error {
+
+	var page uint64
+	var pageSize uint64 = 50
+	verifyCount := 0
+	for {
+		f := database.ContractListenerQueryFactory.NewFilterLimit(ctx, pageSize).And().Skip(page * pageSize)
+		listeners, _, err := cm.database.GetContractListeners(ctx, cm.namespace, f)
+		if err != nil {
+			return err
+		}
+		if len(listeners) == 0 {
+			log.L(ctx).Infof("Listener restore complete. Verified=%d", verifyCount)
+			return nil
+		}
+		for _, l := range listeners {
+			if err := cm.checkContractListenerExists(ctx, l); err != nil {
+				return err
+			}
+			verifyCount++
+		}
+		page++
+	}
+
+}
+
 func (cm *contractManager) writeInvokeTransaction(ctx context.Context, req *core.ContractCallRequest) (*core.Operation, error) {
 	txid, err := cm.txHelper.SubmitNewTransaction(ctx, core.TransactionTypeContractInvoke, req.IdempotencyKey)
 	if err != nil {
@@ -563,6 +595,23 @@ func (cm *contractManager) resolveEvent(ctx context.Context, ffi *fftypes.FFIRef
 	return &core.FFISerializedEvent{FFIEventDefinition: event.FFIEventDefinition}, nil
 }
 
+func (cm *contractManager) checkContractListenerExists(ctx context.Context, listener *core.ContractListener) error {
+	found, _, err := cm.blockchain.GetContractListenerStatus(ctx, listener.BackendID, true)
+	if err != nil {
+		log.L(ctx).Errorf("Validating listener %s:%s (BackendID=%s) failed: %s", listener.Signature, listener.ID, listener.BackendID, err)
+		return err
+	}
+	if found {
+		log.L(ctx).Debugf("Validated listener %s:%s (BackendID=%s)", listener.Signature, listener.ID, listener.BackendID)
+		return nil
+	}
+	if err = cm.blockchain.AddContractListener(ctx, listener); err != nil {
+		return err
+	}
+	return cm.database.UpdateContractListener(ctx, cm.namespace, listener.ID,
+		database.ContractListenerQueryFactory.NewUpdate(ctx).Set("backendid", listener.BackendID))
+}
+
 func (cm *contractManager) AddContractListener(ctx context.Context, listener *core.ContractListenerInput) (output *core.ContractListener, err error) {
 	listener.ID = fftypes.NewUUID()
 	listener.Namespace = cm.namespace
@@ -615,7 +664,7 @@ func (cm *contractManager) AddContractListener(ctx context.Context, listener *co
 		fb := database.ContractListenerQueryFactory.NewFilter(ctx)
 		if existing, _, err := cm.database.GetContractListeners(ctx, cm.namespace, fb.And(
 			fb.Eq("topic", listener.Topic),
-			fb.Eq("location", listener.Location.Bytes()),
+			fb.Eq("location", listener.Location.String()),
 			fb.Eq("signature", listener.Signature),
 		)); err != nil {
 			return err
@@ -631,7 +680,7 @@ func (cm *contractManager) AddContractListener(ctx context.Context, listener *co
 	if err := cm.validateFFIEvent(ctx, &listener.Event.FFIEventDefinition); err != nil {
 		return nil, err
 	}
-	if err = cm.blockchain.AddContractListener(ctx, listener); err != nil {
+	if err = cm.blockchain.AddContractListener(ctx, &listener.ContractListener); err != nil {
 		return nil, err
 	}
 	if listener.Name == "" {
@@ -685,7 +734,7 @@ func (cm *contractManager) GetContractListenerByNameOrIDWithStatus(ctx context.C
 	if err != nil {
 		return nil, err
 	}
-	status, err := cm.blockchain.GetContractListenerStatus(ctx, listener.BackendID)
+	_, status, err := cm.blockchain.GetContractListenerStatus(ctx, listener.BackendID, false)
 	if err != nil {
 		status = core.ListenerStatusError{
 			StatusError: err.Error(),
@@ -733,7 +782,7 @@ func (cm *contractManager) DeleteContractListenerByNameOrID(ctx context.Context,
 		if err != nil {
 			return err
 		}
-		if err = cm.blockchain.DeleteContractListener(ctx, listener); err != nil {
+		if err = cm.blockchain.DeleteContractListener(ctx, listener, true /* ok if not found */); err != nil {
 			return err
 		}
 		return cm.database.DeleteContractListenerByID(ctx, cm.namespace, listener.ID)
diff --git a/internal/database/sqlcommon/contractlisteners_sql.go b/internal/database/sqlcommon/contractlisteners_sql.go
index 5fb248733..a4e081dac 100644
--- a/internal/database/sqlcommon/contractlisteners_sql.go
+++ b/internal/database/sqlcommon/contractlisteners_sql.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -177,6 +177,36 @@ func (s *SQLCommon) GetContractListeners(ctx context.Context, namespace string,
 	return subs, s.QueryRes(ctx, contractlistenersTable, tx, fop, fi), err
 }
 
+func (s *SQLCommon) UpdateContractListener(ctx context.Context, ns string, id *fftypes.UUID, update ffapi.Update) (err error) {
+
+	ctx, tx, autoCommit, err := s.BeginOrUseTx(ctx)
+	if err != nil {
+		return err
+	}
+	defer s.RollbackTx(ctx, tx, autoCommit)
+
+	query, err := s.BuildUpdate(sq.Update(contractlistenersTable), update, contractListenerFilterFieldMap)
+	if err != nil {
+		return err
+	}
+	query = query.Where(sq.And{
+		sq.Eq{"id": id},
+		sq.Eq{"namespace": ns},
+	})
+
+	ra, err := s.UpdateTx(ctx, contractlistenersTable, tx, query, func() {
+		s.callbacks.UUIDCollectionNSEvent(database.CollectionContractListeners, core.ChangeEventTypeUpdated, ns, id)
+	})
+	if err != nil {
+		return err
+	}
+	if ra < 1 {
+		return i18n.NewError(ctx, coremsgs.Msg404NoResult)
+	}
+
+	return s.CommitTx(ctx, tx, autoCommit)
+}
+
 func (s *SQLCommon) DeleteContractListenerByID(ctx context.Context, namespace string, id *fftypes.UUID) (err error) {
 	ctx, tx, autoCommit, err := s.BeginOrUseTx(ctx)
 	if err != nil {
diff --git a/mocks/blockchainmocks/plugin.go b/mocks/blockchainmocks/plugin.go
index ceb1d1a4f..312389da9 100644
--- a/mocks/blockchainmocks/plugin.go
+++ b/mocks/blockchainmocks/plugin.go
@@ -25,11 +25,11 @@ type Plugin struct {
 }
 
 // AddContractListener provides a mock function with given fields: ctx, subscription
-func (_m *Plugin) AddContractListener(ctx context.Context, subscription *core.ContractListenerInput) error {
+func (_m *Plugin) AddContractListener(ctx context.Context, subscription *core.ContractListener) error {
 	ret := _m.Called(ctx, subscription)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *core.ContractListenerInput) error); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, *core.ContractListener) error); ok {
 		r0 = rf(ctx, subscription)
 	} else {
 		r0 = ret.Error(0)
@@ -78,13 +78,13 @@ func (_m *Plugin) Capabilities() *blockchain.Capabilities {
 	return r0
 }
 
-// DeleteContractListener provides a mock function with given fields: ctx, subscription
-func (_m *Plugin) DeleteContractListener(ctx context.Context, subscription *core.ContractListener) error {
-	ret := _m.Called(ctx, subscription)
+// DeleteContractListener provides a mock function with given fields: ctx, subscription, okNotFound
+func (_m *Plugin) DeleteContractListener(ctx context.Context, subscription *core.ContractListener, okNotFound bool) error {
+	ret := _m.Called(ctx, subscription, okNotFound)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *core.ContractListener) error); ok {
-		r0 = rf(ctx, subscription)
+	if rf, ok := ret.Get(0).(func(context.Context, *core.ContractListener, bool) error); ok {
+		r0 = rf(ctx, subscription, okNotFound)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -193,30 +193,37 @@ func (_m *Plugin) GetAndConvertDeprecatedContractConfig(ctx context.Context) (*f
 	return r0, r1, r2
 }
 
-// GetContractListenerStatus provides a mock function with given fields: ctx, subID
-func (_m *Plugin) GetContractListenerStatus(ctx context.Context, subID string) (interface{}, error) {
-	ret := _m.Called(ctx, subID)
+// GetContractListenerStatus provides a mock function with given fields: ctx, subID, okNotFound
+func (_m *Plugin) GetContractListenerStatus(ctx context.Context, subID string, okNotFound bool) (bool, interface{}, error) {
+	ret := _m.Called(ctx, subID, okNotFound)
 
-	var r0 interface{}
-	var r1 error
-	if rf, ok := ret.Get(0).(func(context.Context, string) (interface{}, error)); ok {
-		return rf(ctx, subID)
+	var r0 bool
+	var r1 interface{}
+	var r2 error
+	if rf, ok := ret.Get(0).(func(context.Context, string, bool) (bool, interface{}, error)); ok {
+		return rf(ctx, subID, okNotFound)
 	}
-	if rf, ok := ret.Get(0).(func(context.Context, string) interface{}); ok {
-		r0 = rf(ctx, subID)
+	if rf, ok := ret.Get(0).(func(context.Context, string, bool) bool); ok {
+		r0 = rf(ctx, subID, okNotFound)
 	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(interface{})
+		r0 = ret.Get(0).(bool)
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, string, bool) interface{}); ok {
+		r1 = rf(ctx, subID, okNotFound)
+	} else {
+		if ret.Get(1) != nil {
+			r1 = ret.Get(1).(interface{})
 		}
 	}
 
-	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
-		r1 = rf(ctx, subID)
+	if rf, ok := ret.Get(2).(func(context.Context, string, bool) error); ok {
+		r2 = rf(ctx, subID, okNotFound)
 	} else {
-		r1 = ret.Error(1)
+		r2 = ret.Error(2)
 	}
 
-	return r0, r1
+	return r0, r1, r2
 }
 
 // GetFFIParamValidator provides a mock function with given fields: ctx
diff --git a/mocks/databasemocks/plugin.go b/mocks/databasemocks/plugin.go
index aa6aae75b..43b42d1b8 100644
--- a/mocks/databasemocks/plugin.go
+++ b/mocks/databasemocks/plugin.go
@@ -2626,6 +2626,20 @@ func (_m *Plugin) UpdateBatch(ctx context.Context, namespace string, id *fftypes
 	return r0
 }
 
+// UpdateContractListener provides a mock function with given fields: ctx, namespace, id, update
+func (_m *Plugin) UpdateContractListener(ctx context.Context, namespace string, id *fftypes.UUID, update ffapi.Update) error {
+	ret := _m.Called(ctx, namespace, id, update)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, string, *fftypes.UUID, ffapi.Update) error); ok {
+		r0 = rf(ctx, namespace, id, update)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // UpdateData provides a mock function with given fields: ctx, namespace, id, update
 func (_m *Plugin) UpdateData(ctx context.Context, namespace string, id *fftypes.UUID, update ffapi.Update) error {
 	ret := _m.Called(ctx, namespace, id, update)
diff --git a/pkg/blockchain/plugin.go b/pkg/blockchain/plugin.go
index c3c9f293d..abe76bc77 100644
--- a/pkg/blockchain/plugin.go
+++ b/pkg/blockchain/plugin.go
@@ -86,12 +86,13 @@ type Plugin interface {
 	QueryContract(ctx context.Context, location *fftypes.JSONAny, method *fftypes.FFIMethod, input map[string]interface{}, errors []*fftypes.FFIError, options map[string]interface{}) (interface{}, error)
 
 	// AddContractListener adds a new subscription to a user-specified contract and event
-	AddContractListener(ctx context.Context, subscription *core.ContractListenerInput) error
+	AddContractListener(ctx context.Context, subscription *core.ContractListener) error
 
 	// DeleteContractListener deletes a previously-created subscription
-	DeleteContractListener(ctx context.Context, subscription *core.ContractListener) error
+	DeleteContractListener(ctx context.Context, subscription *core.ContractListener, okNotFound bool) error
 
-	GetContractListenerStatus(ctx context.Context, subID string) (interface{}, error)
+	// GetContractListenerStatus gets the status of a contract listener from the backend connector. Returns false if not found
+	GetContractListenerStatus(ctx context.Context, subID string, okNotFound bool) (bool, interface{}, error)
 
 	// GetFFIParamValidator returns a blockchain-plugin-specific validator for FFIParams and their JSON Schema
 	GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error)
diff --git a/pkg/database/plugin.go b/pkg/database/plugin.go
index ad7115f79..1a1cecbbe 100644
--- a/pkg/database/plugin.go
+++ b/pkg/database/plugin.go
@@ -474,6 +474,9 @@ type iContractListenerCollection interface {
 	// InsertContractListener - upsert a listener to an external smart contract
 	InsertContractListener(ctx context.Context, sub *core.ContractListener) (err error)
 
+	// UpdateContractListener - update contract listener by id
+	UpdateContractListener(ctx context.Context, namespace string, id *fftypes.UUID, update ffapi.Update) (err error)
+
 	// GetContractListener - get contract listener by name
 	GetContractListener(ctx context.Context, namespace, name string) (sub *core.ContractListener, err error)
 


Test patch:
diff --git a/internal/apiserver/server_test.go b/internal/apiserver/server_test.go
index 9df25251a..c8e80d956 100644
--- a/internal/apiserver/server_test.go
+++ b/internal/apiserver/server_test.go
@@ -86,6 +86,7 @@ func TestStartStopServer(t *testing.T) {
 	InitConfig()
 	apiConfig.Set(httpserver.HTTPConfPort, 0)
 	spiConfig.Set(httpserver.HTTPConfPort, 0)
+	metricsConfig.Set(httpserver.HTTPConfPort, 0)
 	config.Set(coreconfig.UIPath, "test")
 	config.Set(coreconfig.SPIEnabled, true)
 	ctx, cancel := context.WithCancel(context.Background())
@@ -104,6 +105,7 @@ func TestStartLegacyAdminConfig(t *testing.T) {
 	InitConfig()
 	apiConfig.Set(httpserver.HTTPConfPort, 0)
 	spiConfig.Set(httpserver.HTTPConfPort, 0)
+	metricsConfig.Set(httpserver.HTTPConfPort, 0)
 	config.Set(coreconfig.UIPath, "test")
 	config.Set(coreconfig.LegacyAdminEnabled, true)
 	ctx, cancel := context.WithCancel(context.Background())
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index bb1e20902..84950aa0b 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -1684,25 +1684,23 @@ func TestAddSubscription(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
-				"address": "0x123",
-			}.String()),
-			Event: &core.FFISerializedEvent{
-				FFIEventDefinition: fftypes.FFIEventDefinition{
-					Name: "Changed",
-					Params: fftypes.FFIParams{
-						{
-							Name:   "value",
-							Schema: fftypes.JSONAnyPtr(`{"type": "string", "details": {"type": "string"}}`),
-						},
+	sub := &core.ContractListener{
+		Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
+			"address": "0x123",
+		}.String()),
+		Event: &core.FFISerializedEvent{
+			FFIEventDefinition: fftypes.FFIEventDefinition{
+				Name: "Changed",
+				Params: fftypes.FFIParams{
+					{
+						Name:   "value",
+						Schema: fftypes.JSONAnyPtr(`{"type": "string", "details": {"type": "string"}}`),
 					},
 				},
 			},
-			Options: &core.ContractListenerOptions{
-				FirstEvent: string(core.SubOptsFirstEventOldest),
-			},
+		},
+		Options: &core.ContractListenerOptions{
+			FirstEvent: string(core.SubOptsFirstEventOldest),
 		},
 	}
 
@@ -1724,22 +1722,20 @@ func TestAddSubscriptionWithoutLocation(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Event: &core.FFISerializedEvent{
-				FFIEventDefinition: fftypes.FFIEventDefinition{
-					Name: "Changed",
-					Params: fftypes.FFIParams{
-						{
-							Name:   "value",
-							Schema: fftypes.JSONAnyPtr(`{"type": "string", "details": {"type": "string"}}`),
-						},
+	sub := &core.ContractListener{
+		Event: &core.FFISerializedEvent{
+			FFIEventDefinition: fftypes.FFIEventDefinition{
+				Name: "Changed",
+				Params: fftypes.FFIParams{
+					{
+						Name:   "value",
+						Schema: fftypes.JSONAnyPtr(`{"type": "string", "details": {"type": "string"}}`),
 					},
 				},
 			},
-			Options: &core.ContractListenerOptions{
-				FirstEvent: string(core.SubOptsFirstEventOldest),
-			},
+		},
+		Options: &core.ContractListenerOptions{
+			FirstEvent: string(core.SubOptsFirstEventOldest),
 		},
 	}
 
@@ -1761,19 +1757,17 @@ func TestAddSubscriptionBadParamDetails(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
-				"address": "0x123",
-			}.String()),
-			Event: &core.FFISerializedEvent{
-				FFIEventDefinition: fftypes.FFIEventDefinition{
-					Name: "Changed",
-					Params: fftypes.FFIParams{
-						{
-							Name:   "value",
-							Schema: fftypes.JSONAnyPtr(`{"type": "string", "details": {"type": ""}}`),
-						},
+	sub := &core.ContractListener{
+		Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
+			"address": "0x123",
+		}.String()),
+		Event: &core.FFISerializedEvent{
+			FFIEventDefinition: fftypes.FFIEventDefinition{
+				Name: "Changed",
+				Params: fftypes.FFIParams{
+					{
+						Name:   "value",
+						Schema: fftypes.JSONAnyPtr(`{"type": "string", "details": {"type": ""}}`),
 					},
 				},
 			},
@@ -1799,11 +1793,9 @@ func TestAddSubscriptionBadLocation(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Location: fftypes.JSONAnyPtr(""),
-			Event:    &core.FFISerializedEvent{},
-		},
+	sub := &core.ContractListener{
+		Location: fftypes.JSONAnyPtr(""),
+		Event:    &core.FFISerializedEvent{},
 	}
 
 	err := e.AddContractListener(context.Background(), sub)
@@ -1822,15 +1814,13 @@ func TestAddSubscriptionFail(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
-				"address": "0x123",
-			}.String()),
-			Event: &core.FFISerializedEvent{},
-			Options: &core.ContractListenerOptions{
-				FirstEvent: string(core.SubOptsFirstEventNewest),
-			},
+	sub := &core.ContractListener{
+		Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
+			"address": "0x123",
+		}.String()),
+		Event: &core.FFISerializedEvent{},
+		Options: &core.ContractListenerOptions{
+			FirstEvent: string(core.SubOptsFirstEventNewest),
 		},
 	}
 
@@ -1861,7 +1851,7 @@ func TestDeleteSubscription(t *testing.T) {
 	httpmock.RegisterResponder("DELETE", `http://localhost:12345/subscriptions/sb-1`,
 		httpmock.NewStringResponder(204, ""))
 
-	err := e.DeleteContractListener(context.Background(), sub)
+	err := e.DeleteContractListener(context.Background(), sub, true)
 
 	assert.NoError(t, err)
 }
@@ -1884,11 +1874,34 @@ func TestDeleteSubscriptionFail(t *testing.T) {
 	httpmock.RegisterResponder("DELETE", `http://localhost:12345/subscriptions/sb-1`,
 		httpmock.NewStringResponder(500, ""))
 
-	err := e.DeleteContractListener(context.Background(), sub)
+	err := e.DeleteContractListener(context.Background(), sub, true)
 
 	assert.Regexp(t, "FF10111", err)
 }
 
+func TestDeleteSubscriptionNotFound(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+	httpmock.ActivateNonDefault(e.client.GetClient())
+	defer httpmock.DeactivateAndReset()
+
+	e.streamID = "es-1"
+	e.streams = &streamManager{
+		client: e.client,
+	}
+
+	sub := &core.ContractListener{
+		BackendID: "sb-1",
+	}
+
+	httpmock.RegisterResponder("DELETE", `http://localhost:12345/subscriptions/sb-1`,
+		httpmock.NewStringResponder(404, ""))
+
+	err := e.DeleteContractListener(context.Background(), sub, true)
+
+	assert.NoError(t, err)
+}
+
 func TestHandleMessageContractEventOldSubscription(t *testing.T) {
 	data := fftypes.JSONAnyPtr(`
 [
@@ -3619,9 +3632,10 @@ func TestGetContractListenerStatus(t *testing.T) {
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
 	assert.NoError(t, err)
 
-	status, err := e.GetContractListenerStatus(context.Background(), "sub1")
+	found, status, err := e.GetContractListenerStatus(context.Background(), "sub1", true)
 	assert.NotNil(t, status)
 	assert.NoError(t, err)
+	assert.True(t, found)
 }
 
 func TestGetContractListenerStatusGetSubFail(t *testing.T) {
@@ -3651,9 +3665,43 @@ func TestGetContractListenerStatusGetSubFail(t *testing.T) {
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
 	assert.NoError(t, err)
 
-	status, err := e.GetContractListenerStatus(context.Background(), "sub1")
+	found, status, err := e.GetContractListenerStatus(context.Background(), "sub1", true)
 	assert.Nil(t, status)
 	assert.Regexp(t, "FF10111", err)
+	assert.False(t, found)
+}
+
+func TestGetContractListenerStatusGetSubNotFound(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+	resetConf(e)
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{}))
+	httpmock.RegisterResponder("POST", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, eventStream{ID: "es12345"}))
+	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
+		httpmock.NewJsonResponderOrPanic(200, []subscription{}))
+	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions/sub1",
+		httpmock.NewJsonResponderOrPanic(404, `not found`))
+
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
+	assert.NoError(t, err)
+
+	found, status, err := e.GetContractListenerStatus(context.Background(), "sub1", true)
+	assert.Nil(t, status)
+	assert.Nil(t, err)
+	assert.False(t, found)
 }
 
 func TestGetTransactionStatusSuccess(t *testing.T) {
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index 75cebb049..e56a53fe8 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -1425,16 +1425,14 @@ func TestAddSubscription(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
-				"channel":   "firefly",
-				"chaincode": "mycode",
-			}.String()),
-			Event: &core.FFISerializedEvent{},
-			Options: &core.ContractListenerOptions{
-				FirstEvent: string(core.SubOptsFirstEventOldest),
-			},
+	sub := &core.ContractListener{
+		Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
+			"channel":   "firefly",
+			"chaincode": "mycode",
+		}.String()),
+		Event: &core.FFISerializedEvent{},
+		Options: &core.ContractListenerOptions{
+			FirstEvent: string(core.SubOptsFirstEventOldest),
 		},
 	}
 
@@ -1462,15 +1460,13 @@ func TestAddSubscriptionNoChannel(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
-				"chaincode": "mycode",
-			}.String()),
-			Event: &core.FFISerializedEvent{},
-			Options: &core.ContractListenerOptions{
-				FirstEvent: string(core.SubOptsFirstEventOldest),
-			},
+	sub := &core.ContractListener{
+		Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
+			"chaincode": "mycode",
+		}.String()),
+		Event: &core.FFISerializedEvent{},
+		Options: &core.ContractListenerOptions{
+			FirstEvent: string(core.SubOptsFirstEventOldest),
 		},
 	}
 
@@ -1498,12 +1494,10 @@ func TestAddSubscriptionNoLocation(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Event: &core.FFISerializedEvent{},
-			Options: &core.ContractListenerOptions{
-				FirstEvent: string(core.SubOptsFirstEventOldest),
-			},
+	sub := &core.ContractListener{
+		Event: &core.FFISerializedEvent{},
+		Options: &core.ContractListenerOptions{
+			FirstEvent: string(core.SubOptsFirstEventOldest),
 		},
 	}
 
@@ -1523,11 +1517,9 @@ func TestAddSubscriptionBadLocation(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Location: fftypes.JSONAnyPtr(""),
-			Event:    &core.FFISerializedEvent{},
-		},
+	sub := &core.ContractListener{
+		Location: fftypes.JSONAnyPtr(""),
+		Event:    &core.FFISerializedEvent{},
 	}
 
 	err := e.AddContractListener(context.Background(), sub)
@@ -1546,16 +1538,14 @@ func TestAddSubscriptionFail(t *testing.T) {
 		client: e.client,
 	}
 
-	sub := &core.ContractListenerInput{
-		ContractListener: core.ContractListener{
-			Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
-				"channel":   "firefly",
-				"chaincode": "mycode",
-			}.String()),
-			Event: &core.FFISerializedEvent{},
-			Options: &core.ContractListenerOptions{
-				FirstEvent: string(core.SubOptsFirstEventNewest),
-			},
+	sub := &core.ContractListener{
+		Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
+			"channel":   "firefly",
+			"chaincode": "mycode",
+		}.String()),
+		Event: &core.FFISerializedEvent{},
+		Options: &core.ContractListenerOptions{
+			FirstEvent: string(core.SubOptsFirstEventNewest),
 		},
 	}
 
@@ -1585,7 +1575,7 @@ func TestDeleteSubscription(t *testing.T) {
 	httpmock.RegisterResponder("DELETE", `http://localhost:12345/subscriptions/sb-1`,
 		httpmock.NewStringResponder(204, ""))
 
-	err := e.DeleteContractListener(context.Background(), sub)
+	err := e.DeleteContractListener(context.Background(), sub, true)
 
 	assert.NoError(t, err)
 }
@@ -1608,11 +1598,34 @@ func TestDeleteSubscriptionFail(t *testing.T) {
 	httpmock.RegisterResponder("DELETE", `http://localhost:12345/subscriptions/sb-1`,
 		httpmock.NewStringResponder(500, "pop"))
 
-	err := e.DeleteContractListener(context.Background(), sub)
+	err := e.DeleteContractListener(context.Background(), sub, true)
 
 	assert.Regexp(t, "FF10284.*pop", err)
 }
 
+func TestDeleteSubscriptionNotFound(t *testing.T) {
+	e, cancel := newTestFabric()
+	defer cancel()
+	httpmock.ActivateNonDefault(e.client.GetClient())
+	defer httpmock.DeactivateAndReset()
+
+	e.streamID = "es-1"
+	e.streams = &streamManager{
+		client: e.client,
+	}
+
+	sub := &core.ContractListener{
+		BackendID: "sb-1",
+	}
+
+	httpmock.RegisterResponder("DELETE", `http://localhost:12345/subscriptions/sb-1`,
+		httpmock.NewStringResponder(404, "pop"))
+
+	err := e.DeleteContractListener(context.Background(), sub, true)
+
+	assert.NoError(t, err)
+}
+
 func TestHandleMessageContractEventOldSubscription(t *testing.T) {
 	data := []byte(`
 [
@@ -2721,7 +2734,7 @@ func TestGetContractListenerStatus(t *testing.T) {
 	httpmock.ActivateNonDefault(e.client.GetClient())
 	defer httpmock.DeactivateAndReset()
 
-	status, err := e.GetContractListenerStatus(context.Background(), "id")
+	_, status, err := e.GetContractListenerStatus(context.Background(), "id", true)
 	assert.Nil(t, status)
 	assert.NoError(t, err)
 }
diff --git a/internal/contracts/manager_test.go b/internal/contracts/manager_test.go
index fcf5d80ff..dd79ac909 100644
--- a/internal/contracts/manager_test.go
+++ b/internal/contracts/manager_test.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -61,6 +62,7 @@ func newTestContractManager() *contractManager {
 
 	mbi.On("Name").Return("mockblockchain").Maybe()
 
+	mdi.On("GetContractListeners", mock.Anything, "ns1", mock.Anything).Return(nil, nil, nil).Once()
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything).Maybe()
 	rag.RunFn = func(a mock.Arguments) {
 		rag.ReturnArguments = mock.Arguments{
@@ -109,6 +111,7 @@ func TestNewContractManagerFFISchemaLoader(t *testing.T) {
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(ctx, 100, 5*time.Minute), nil)
 	txHelper, _ := txcommon.NewTransactionHelper(ctx, "ns1", mdi, mdm, cmi)
 	msa := &syncasyncmocks.Bridge{}
+	mdi.On("GetContractListeners", mock.Anything, "ns1", mock.Anything).Return(nil, nil, nil)
 	mbi.On("GetFFIParamValidator", mock.Anything).Return(&ffi2abi.ParamValidator{}, nil)
 	mom.On("RegisterHandler", mock.Anything, mock.Anything, mock.Anything)
 	_, err := NewContractManager(context.Background(), "ns1", mdi, mbi, mim, mom, txHelper, msa)
@@ -700,7 +703,7 @@ func TestAddContractListenerInline(t *testing.T) {
 	mbi.On("NormalizeContractLocation", context.Background(), sub.Location).Return(sub.Location, nil)
 	mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed")
 	mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
-	mbi.On("AddContractListener", context.Background(), sub).Return(nil)
+	mbi.On("AddContractListener", context.Background(), &sub.ContractListener).Return(nil)
 	mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
 
 	result, err := cm.AddContractListener(context.Background(), sub)
@@ -737,7 +740,7 @@ func TestAddContractListenerNoLocationOK(t *testing.T) {
 
 	mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed")
 	mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
-	mbi.On("AddContractListener", context.Background(), sub).Return(nil)
+	mbi.On("AddContractListener", context.Background(), &sub.ContractListener).Return(nil)
 	mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
 
 	result, err := cm.AddContractListener(context.Background(), sub)
@@ -786,7 +789,7 @@ func TestAddContractListenerByEventPath(t *testing.T) {
 	mbi.On("NormalizeContractLocation", context.Background(), sub.Location).Return(sub.Location, nil)
 	mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed")
 	mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
-	mbi.On("AddContractListener", context.Background(), sub).Return(nil)
+	mbi.On("AddContractListener", context.Background(), &sub.ContractListener).Return(nil)
 	mdi.On("GetFFIByID", context.Background(), "ns1", interfaceID).Return(&fftypes.FFI{}, nil)
 	mdi.On("GetFFIEvent", context.Background(), "ns1", interfaceID, sub.EventPath).Return(event, nil)
 	mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
@@ -938,6 +941,148 @@ func TestAddContractListenerMissingEventOrID(t *testing.T) {
 	mbi.AssertExpectations(t)
 }
 
+func TestAddContractListenerVerifyOk(t *testing.T) {
+	cm := newTestContractManager()
+
+	ctx := context.Background()
+
+	mdi := cm.database.(*databasemocks.Plugin)
+	mdi.On("GetContractListeners", mock.Anything, "ns1", mock.MatchedBy(func(f ffapi.Filter) bool {
+		fi, _ := f.Finalize()
+		return fi.Skip == 0 && fi.Limit == 50
+	})).Return([]*core.ContractListener{
+		{ID: fftypes.NewUUID(), BackendID: "12345"},
+		{ID: fftypes.NewUUID(), BackendID: "23456"},
+	}, nil, nil).Once()
+	mdi.On("GetContractListeners", mock.Anything, "ns1", mock.MatchedBy(func(f ffapi.Filter) bool {
+		fi, _ := f.Finalize()
+		return fi.Skip == 50 && fi.Limit == 50
+	})).Return([]*core.ContractListener{}, nil, nil).Once()
+
+	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+	mbi.On("GetContractListenerStatus", ctx, "12345", true).Return(true, struct{}{}, nil)
+	mbi.On("GetContractListenerStatus", ctx, "23456", true).Return(false, nil, nil)
+	mbi.On("AddContractListener", ctx, mock.MatchedBy(func(l *core.ContractListener) bool {
+		prevBackendID := l.BackendID
+		l.BackendID = "34567"
+		return prevBackendID == "23456"
+	})).Return(nil)
+
+	mdi.On("UpdateContractListener", ctx, "ns1", mock.Anything, mock.MatchedBy(func(u ffapi.Update) bool {
+		uu, _ := u.Finalize()
+		return strings.Contains(uu.String(), "34567")
+	})).Return(nil).Once()
+
+	err := cm.verifyListeners(ctx)
+	assert.NoError(t, err)
+
+	mdi.AssertExpectations(t)
+	mbi.AssertExpectations(t)
+}
+
+func TestAddContractListenerVerifyUpdateFail(t *testing.T) {
+	cm := newTestContractManager()
+
+	ctx := context.Background()
+
+	mdi := cm.database.(*databasemocks.Plugin)
+	mdi.On("GetContractListeners", mock.Anything, "ns1", mock.MatchedBy(func(f ffapi.Filter) bool {
+		fi, _ := f.Finalize()
+		return fi.Skip == 0 && fi.Limit == 50
+	})).Return([]*core.ContractListener{
+		{ID: fftypes.NewUUID(), BackendID: "12345"},
+		{ID: fftypes.NewUUID(), BackendID: "23456"},
+	}, nil, nil).Once()
+
+	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+	mbi.On("GetContractListenerStatus", ctx, "12345", true).Return(true, struct{}{}, nil)
+	mbi.On("GetContractListenerStatus", ctx, "23456", true).Return(false, nil, nil)
+	mbi.On("AddContractListener", ctx, mock.MatchedBy(func(l *core.ContractListener) bool {
+		prevBackendID := l.BackendID
+		l.BackendID = "34567"
+		return prevBackendID == "23456"
+	})).Return(nil)
+
+	mdi.On("UpdateContractListener", ctx, "ns1", mock.Anything, mock.MatchedBy(func(u ffapi.Update) bool {
+		uu, _ := u.Finalize()
+		return strings.Contains(uu.String(), "34567")
+	})).Return(fmt.Errorf("pop")).Once()
+
+	err := cm.verifyListeners(ctx)
+	assert.Regexp(t, "pop", err)
+
+	mdi.AssertExpectations(t)
+	mbi.AssertExpectations(t)
+}
+
+func TestAddContractListenerVerifyAddFail(t *testing.T) {
+	cm := newTestContractManager()
+
+	ctx := context.Background()
+
+	mdi := cm.database.(*databasemocks.Plugin)
+	mdi.On("GetContractListeners", mock.Anything, "ns1", mock.MatchedBy(func(f ffapi.Filter) bool {
+		fi, _ := f.Finalize()
+		return fi.Skip == 0 && fi.Limit == 50
+	})).Return([]*core.ContractListener{
+		{ID: fftypes.NewUUID(), BackendID: "12345"},
+		{ID: fftypes.NewUUID(), BackendID: "23456"},
+	}, nil, nil).Once()
+
+	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+	mbi.On("GetContractListenerStatus", ctx, "12345", true).Return(true, struct{}{}, nil)
+	mbi.On("GetContractListenerStatus", ctx, "23456", true).Return(false, nil, nil)
+	mbi.On("AddContractListener", ctx, mock.MatchedBy(func(l *core.ContractListener) bool {
+		prevBackendID := l.BackendID
+		l.BackendID = "34567"
+		return prevBackendID == "23456"
+	})).Return(fmt.Errorf("pop"))
+
+	err := cm.verifyListeners(ctx)
+	assert.Regexp(t, "pop", err)
+
+	mdi.AssertExpectations(t)
+	mbi.AssertExpectations(t)
+}
+
+func TestAddContractListenerVerifyGetFail(t *testing.T) {
+	cm := newTestContractManager()
+
+	ctx := context.Background()
+
+	mdi := cm.database.(*databasemocks.Plugin)
+	mdi.On("GetContractListeners", mock.Anything, "ns1", mock.MatchedBy(func(f ffapi.Filter) bool {
+		fi, _ := f.Finalize()
+		return fi.Skip == 0 && fi.Limit == 50
+	})).Return([]*core.ContractListener{
+		{ID: fftypes.NewUUID(), BackendID: "12345"},
+		{ID: fftypes.NewUUID(), BackendID: "23456"},
+	}, nil, nil).Once()
+
+	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+	mbi.On("GetContractListenerStatus", ctx, "12345", true).Return(false, nil, fmt.Errorf("pop"))
+
+	err := cm.verifyListeners(ctx)
+	assert.Regexp(t, "pop", err)
+
+	mdi.AssertExpectations(t)
+	mbi.AssertExpectations(t)
+}
+
+func TestAddContractListenerVerifyGetListFail(t *testing.T) {
+	cm := newTestContractManager()
+
+	ctx := context.Background()
+
+	mdi := cm.database.(*databasemocks.Plugin)
+	mdi.On("GetContractListeners", mock.Anything, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop")).Once()
+
+	err := cm.verifyListeners(ctx)
+	assert.Regexp(t, "pop", err)
+
+	mdi.AssertExpectations(t)
+}
+
 func TestAddContractListenerBadName(t *testing.T) {
 	cm := newTestContractManager()
 	sub := &core.ContractListenerInput{
@@ -1128,7 +1273,7 @@ func TestAddContractListenerBlockchainFail(t *testing.T) {
 	mbi.On("NormalizeContractLocation", context.Background(), sub.Location).Return(sub.Location, nil)
 	mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed")
 	mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
-	mbi.On("AddContractListener", context.Background(), sub).Return(fmt.Errorf("pop"))
+	mbi.On("AddContractListener", context.Background(), &sub.ContractListener).Return(fmt.Errorf("pop"))
 
 	_, err := cm.AddContractListener(context.Background(), sub)
 	assert.EqualError(t, err, "pop")
@@ -1165,7 +1310,7 @@ func TestAddContractListenerUpsertSubFail(t *testing.T) {
 	mbi.On("NormalizeContractLocation", context.Background(), sub.Location).Return(sub.Location, nil)
 	mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed")
 	mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
-	mbi.On("AddContractListener", context.Background(), sub).Return(nil)
+	mbi.On("AddContractListener", context.Background(), &sub.ContractListener).Return(nil)
 	mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(fmt.Errorf("pop"))
 
 	_, err := cm.AddContractListener(context.Background(), sub)
@@ -1204,8 +1349,8 @@ func TestAddContractAPIListener(t *testing.T) {
 	mdi.On("GetFFIEvent", context.Background(), "ns1", interfaceID, "changed").Return(event, nil)
 	mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed")
 	mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
-	mbi.On("AddContractListener", context.Background(), mock.MatchedBy(func(l *core.ContractListenerInput) bool {
-		return *l.Interface.ID == *interfaceID && l.EventPath == "changed" && l.Topic == "test-topic"
+	mbi.On("AddContractListener", context.Background(), mock.MatchedBy(func(l *core.ContractListener) bool {
+		return *l.Interface.ID == *interfaceID && l.Topic == "test-topic"
 	})).Return(nil)
 	mdi.On("InsertContractListener", context.Background(), mock.MatchedBy(func(l *core.ContractListener) bool {
 		return *l.Interface.ID == *interfaceID && l.Event.Name == "changed" && l.Topic == "test-topic"
@@ -1897,7 +2042,7 @@ func TestGetContractListenerByNameOrIDWithStatus(t *testing.T) {
 	id := fftypes.NewUUID()
 	backendID := "testID"
 	mdi.On("GetContractListenerByID", context.Background(), "ns1", id).Return(&core.ContractListener{BackendID: backendID}, nil)
-	mbi.On("GetContractListenerStatus", context.Background(), backendID).Return(fftypes.JSONAnyPtr(fftypes.JSONObject{}.String()), nil)
+	mbi.On("GetContractListenerStatus", context.Background(), backendID, false).Return(true, fftypes.JSONAnyPtr(fftypes.JSONObject{}.String()), nil)
 
 	_, err := cm.GetContractListenerByNameOrIDWithStatus(context.Background(), id.String())
 	assert.NoError(t, err)
@@ -1922,7 +2067,7 @@ func TestGetContractListenerByNameOrIDWithStatusPluginFail(t *testing.T) {
 	id := fftypes.NewUUID()
 	backendID := "testID"
 	mdi.On("GetContractListenerByID", context.Background(), "ns1", id).Return(&core.ContractListener{BackendID: backendID}, nil)
-	mbi.On("GetContractListenerStatus", context.Background(), backendID).Return(nil, fmt.Errorf("pop"))
+	mbi.On("GetContractListenerStatus", context.Background(), backendID, false).Return(false, nil, fmt.Errorf("pop"))
 
 	listener, err := cm.GetContractListenerByNameOrIDWithStatus(context.Background(), id.String())
 
@@ -2088,7 +2233,7 @@ func TestDeleteContractListener(t *testing.T) {
 	}
 
 	mdi.On("GetContractListener", context.Background(), "ns1", "sub1").Return(sub, nil)
-	mbi.On("DeleteContractListener", context.Background(), sub).Return(nil)
+	mbi.On("DeleteContractListener", context.Background(), sub, true).Return(nil)
 	mdi.On("DeleteContractListenerByID", context.Background(), "ns1", sub.ID).Return(nil)
 
 	err := cm.DeleteContractListenerByNameOrID(context.Background(), "sub1")
@@ -2105,7 +2250,7 @@ func TestDeleteContractListenerBlockchainFail(t *testing.T) {
 	}
 
 	mdi.On("GetContractListener", context.Background(), "ns1", "sub1").Return(sub, nil)
-	mbi.On("DeleteContractListener", context.Background(), sub).Return(fmt.Errorf("pop"))
+	mbi.On("DeleteContractListener", context.Background(), sub, true).Return(fmt.Errorf("pop"))
 	mdi.On("DeleteContractListenerByID", context.Background(), "ns1", sub.ID).Return(nil)
 
 	err := cm.DeleteContractListenerByNameOrID(context.Background(), "sub1")
diff --git a/internal/database/sqlcommon/contractlisteners_sql_test.go b/internal/database/sqlcommon/contractlisteners_sql_test.go
index 3d29184df..b321a685a 100644
--- a/internal/database/sqlcommon/contractlisteners_sql_test.go
+++ b/internal/database/sqlcommon/contractlisteners_sql_test.go
@@ -18,6 +18,7 @@ package sqlcommon
 
 import (
 	"context"
+	"database/sql/driver"
 	"encoding/json"
 	"fmt"
 	"testing"
@@ -78,9 +79,15 @@ func TestContractListenerE2EWithDB(t *testing.T) {
 	subReadJson, _ := json.Marshal(subs[0])
 	assert.Equal(t, string(subJson), string(subReadJson))
 
+	// Update by backend ID
+	err = s.UpdateContractListener(ctx, "ns", sub.ID, database.ContractListenerQueryFactory.NewUpdate(ctx).Set("backendid", "sb-234"))
+	assert.NoError(t, err)
+
 	// Query back the listener (by name)
 	subRead, err := s.GetContractListener(ctx, "ns", "sub1")
 	assert.NoError(t, err)
+	sub.BackendID = "sb-234"
+	subJson, _ = json.Marshal(&sub)
 	subReadJson, _ = json.Marshal(subRead)
 	assert.Equal(t, string(subJson), string(subReadJson))
 
@@ -236,3 +243,43 @@ func TestContractListenerOptions(t *testing.T) {
 
 	assert.Equal(t, l.Options, li.Options)
 }
+
+func TestUpdateContractListenerFailFilter(t *testing.T) {
+	s, mock := newMockProvider().init()
+	mock.ExpectBegin()
+	err := s.UpdateContractListener(context.Background(), "ns1", fftypes.NewUUID(),
+		database.ContractListenerQueryFactory.NewUpdate(context.Background()).Set("wrong", "sb-234"))
+	assert.Regexp(t, "FF00142", err)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestUpdateContractListenerFailBegin(t *testing.T) {
+	s, mock := newMockProvider().init()
+	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
+	err := s.UpdateContractListener(context.Background(), "ns1", fftypes.NewUUID(),
+		database.ContractListenerQueryFactory.NewUpdate(context.Background()).Set("backendid", "sb-234"))
+	assert.Regexp(t, "FF00175", err)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestUpdateContractListenerFailUpdate(t *testing.T) {
+	s, mock := newMockProvider().init()
+	mock.ExpectBegin()
+	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
+	mock.ExpectRollback()
+	err := s.UpdateContractListener(context.Background(), "ns1", fftypes.NewUUID(),
+		database.ContractListenerQueryFactory.NewUpdate(context.Background()).Set("backendid", "sb-234"))
+	assert.Regexp(t, "FF00178", err)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestUpdateContractListenerNotFount(t *testing.T) {
+	s, mock := newMockProvider().init()
+	mock.ExpectBegin()
+	mock.ExpectExec("UPDATE .*").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectRollback()
+	err := s.UpdateContractListener(context.Background(), "ns1", fftypes.NewUUID(),
+		database.ContractListenerQueryFactory.NewUpdate(context.Background()).Set("backendid", "sb-234"))
+	assert.Regexp(t, "FF10143", err)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}