Repository: hyperledger/firefly
Pull request #1028 (2022-08-31 17:53:38)

Add status information for subscriptions & contract listeners
 If the `fetchstatus` query param is supplied when querying `subscriptions` or `listeners` by ID, firefly will return additional status information, if it is available.

### Contract listeners:
FF environments connected to evm connectors will return listener checkpoint formation under the `status` key.

Example query: `http://localhost:5000/api/v1/namespaces/default/contracts/listeners/92e3bc0c-3c3f-455b-81d3-4124a0270b1d?fetchstatus`

Response:
```
{
    "id": "92e3bc0c-3c3f-455b-81d3-4124a0270b1d",
    "interface": {
        "id": "30775d45-61af-4f76-9d35-69b5d8ffbf42"
    },
    "namespace": "default",
    "name": "0182f410-a686-79ed-23a5-ac8c62acbf51",
    "backendId": "0182f410-a686-79ed-23a5-ac8c62acbf51",
    "location": {
        "address": "0x62d365e68fc55845ba904acef688feba7a598c9c"
    },
    "created": "2022-08-31T13:21:48.1687995Z",
    "event": {
        "name": "Changed",
        "description": "",
        "params": [
            {
                "name": "from",
                "schema": {
                    "type": "string",
                    "details": {
                        "type": "address",
                        "internalType": "address",
                        "indexed": true
                    }
                }
            },
            {
                "name": "value",
                "schema": {
                    "type": "integer",
                    "details": {
                        "type": "uint256",
                        "internalType": "uint256"
                    }
                }
            }
        ]
    },
    "signature": "Changed(address,uint256)",
    "topic": "bsc",
    "options": {
        "firstEvent": "oldest"
    },
    "status": {
        "checkpoint": {
            "block": 0,
            "transactionIndex": -1,
            "logIndex": -1
        }
    }
}
```

### Subscriptions
Subscriptions will provide offset information if available

Example Query: `http://localhost:5000/api/v1/namespaces/default/subscriptions/32cddc74-cc1a-4493-b376-45781e2d8d35?fetchstatus`

Response:
```
{
    "id": "32cddc74-cc1a-4493-b376-45781e2d8d35",
    "namespace": "default",
    "name": "simple-storage",
    "transport": "websockets",
    "filter": {
        "message": {},
        "transaction": {},
        "blockchainevent": {}
    },
    "options": {
        "firstEvent": "-1",
        "withData": false
    },
    "created": "2022-08-31T17:13:04.295134Z",
    "updated": null,
    "status": {
        "currentOffset": 31
    }
}
```


Patch:
diff --git a/db/migrations/postgres/000097_alter_tokentransfer_allow_null_signer.down.sql b/db/migrations/postgres/000097_alter_tokentransfer_allow_null_signer.down.sql
deleted file mode 100644
index 4eed6959e..000000000
--- a/db/migrations/postgres/000097_alter_tokentransfer_allow_null_signer.down.sql
+++ /dev/null
@@ -1,1 +0,0 @@
--- No action in down
\ No newline at end of file
diff --git a/db/migrations/postgres/000097_alter_tokentransfer_allow_null_signer.up.sql b/db/migrations/postgres/000097_alter_tokentransfer_allow_null_signer.up.sql
deleted file mode 100644
index 2b3601713..000000000
--- a/db/migrations/postgres/000097_alter_tokentransfer_allow_null_signer.up.sql
+++ /dev/null
@@ -1,5 +0,0 @@
-BEGIN;
-
-ALTER TABLE tokentransfer ALTER COLUMN key DROP NOT NULL;
-
-COMMIT;
diff --git a/db/migrations/sqlite/000031_create_tokentransfer_table.up.sql b/db/migrations/sqlite/000031_create_tokentransfer_table.up.sql
index 20e821f42..ef575cb71 100644
--- a/db/migrations/sqlite/000031_create_tokentransfer_table.up.sql
+++ b/db/migrations/sqlite/000031_create_tokentransfer_table.up.sql
@@ -4,7 +4,7 @@ CREATE TABLE tokentransfer (
   type             VARCHAR(64)     NOT NULL,
   pool_protocol_id VARCHAR(1024)   NOT NULL,
   token_index      VARCHAR(1024),
-  key              VARCHAR(1024),
+  key              VARCHAR(1024)   NOT NULL,
   from_key         VARCHAR(1024),
   to_key           VARCHAR(1024),
   amount           VARCHAR(65),
diff --git a/db/migrations/sqlite/000097_alter_tokentransfer_allow_null_signer.down.sql b/db/migrations/sqlite/000097_alter_tokentransfer_allow_null_signer.down.sql
deleted file mode 100644
index f786bd66f..000000000
--- a/db/migrations/sqlite/000097_alter_tokentransfer_allow_null_signer.down.sql
+++ /dev/null
@@ -1,1 +0,0 @@
--- This is not implemented for SQLite
\ No newline at end of file
diff --git a/db/migrations/sqlite/000097_alter_tokentransfer_allow_null_signer.up.sql b/db/migrations/sqlite/000097_alter_tokentransfer_allow_null_signer.up.sql
deleted file mode 100644
index 121f077ff..000000000
--- a/db/migrations/sqlite/000097_alter_tokentransfer_allow_null_signer.up.sql
+++ /dev/null
@@ -1,1 +0,0 @@
--- SQLite does not allow removal of a NOT NULL constraint, so the CREATE TABLE has been altered for new databases only.
\ No newline at end of file
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
index eaf809ee9..df14e778b 100644
--- a/docs/swagger/swagger.yaml
+++ b/docs/swagger/swagger.yaml
@@ -3846,6 +3846,13 @@ paths:
         required: true
         schema:
           type: string
+      - description: When set, the API will return additional status information if
+          available
+        in: query
+        name: fetchstatus
+        schema:
+          example: "true"
+          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -13056,6 +13063,13 @@ paths:
         schema:
           example: default
           type: string
+      - description: When set, the API will return additional status information if
+          available
+        in: query
+        name: fetchstatus
+        schema:
+          example: "true"
+          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -22090,6 +22104,12 @@ paths:
         schema:
           example: default
           type: string
+      - description: When set, the API will return additional status information if
+          available
+        in: query
+        name: fetchstatus
+        schema:
+          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -22400,7 +22420,7 @@ paths:
       operationId: getTokenAccountPoolsNamespace
       parameters:
       - description: The key for the token account. The exact format may vary based
-          on the token connector use.
+          on the token connector use
         in: path
         name: key
         required: true
@@ -29661,6 +29681,12 @@ paths:
         required: true
         schema:
           type: string
+      - description: When set, the API will return additional status information if
+          available
+        in: query
+        name: fetchstatus
+        schema:
+          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -29964,7 +29990,7 @@ paths:
       operationId: getTokenAccountPools
       parameters:
       - description: The key for the token account. The exact format may vary based
-          on the token connector use.
+          on the token connector use
         in: path
         name: key
         required: true
diff --git a/internal/apiserver/route_get_contract_listener_by_name_or_id.go b/internal/apiserver/route_get_contract_listener_by_name_or_id.go
index 7424515f0..4408fe7d3 100644
--- a/internal/apiserver/route_get_contract_listener_by_name_or_id.go
+++ b/internal/apiserver/route_get_contract_listener_by_name_or_id.go
@@ -18,6 +18,7 @@ package apiserver
 
 import (
 	"net/http"
+	"strings"
 
 	"github.com/hyperledger/firefly-common/pkg/ffapi"
 	"github.com/hyperledger/firefly/internal/coremsgs"
@@ -32,7 +33,9 @@ var getContractListenerByNameOrID = &ffapi.Route{
 	PathParams: []*ffapi.PathParam{
 		{Name: "nameOrId", Description: coremsgs.APIParamsContractListenerNameOrID},
 	},
-	QueryParams:     nil,
+	QueryParams: []*ffapi.QueryParam{
+		{Name: "fetchstatus", Example: "true", Description: coremsgs.APIParamsFetchStatus, IsBool: true},
+	},
 	Description:     coremsgs.APIEndpointsGetContractListenerByNameOrID,
 	JSONInputValue:  nil,
 	JSONOutputValue: func() interface{} { return &core.ContractListener{} },
@@ -42,6 +45,9 @@ var getContractListenerByNameOrID = &ffapi.Route{
 			return or.Contracts() != nil
 		},
 		CoreJSONHandler: func(r *ffapi.APIRequest, cr *coreRequest) (output interface{}, err error) {
+			if strings.EqualFold(r.QP["fetchstatus"], "true") {
+				return cr.or.Contracts().GetContractListenerByNameOrIDWithStatus(cr.ctx, r.PP["nameOrId"])
+			}
 			return cr.or.Contracts().GetContractListenerByNameOrID(cr.ctx, r.PP["nameOrId"])
 		},
 	},
diff --git a/internal/apiserver/route_get_subscription_by_id.go b/internal/apiserver/route_get_subscription_by_id.go
index e099ce7b4..3b29b7960 100644
--- a/internal/apiserver/route_get_subscription_by_id.go
+++ b/internal/apiserver/route_get_subscription_by_id.go
@@ -18,6 +18,7 @@ package apiserver
 
 import (
 	"net/http"
+	"strings"
 
 	"github.com/hyperledger/firefly-common/pkg/ffapi"
 	"github.com/hyperledger/firefly/internal/coremsgs"
@@ -31,15 +32,19 @@ var getSubscriptionByID = &ffapi.Route{
 	PathParams: []*ffapi.PathParam{
 		{Name: "subid", Description: coremsgs.APIParamsSubscriptionID},
 	},
-	QueryParams:     nil,
+	QueryParams: []*ffapi.QueryParam{
+		{Name: "fetchstatus", Description: coremsgs.APIParamsFetchStatus, IsBool: true},
+	},
 	Description:     coremsgs.APIEndpointsGetSubscriptionByID,
 	JSONInputValue:  nil,
 	JSONOutputValue: func() interface{} { return &core.Subscription{} },
 	JSONOutputCodes: []int{http.StatusOK},
 	Extensions: &coreExtensions{
 		CoreJSONHandler: func(r *ffapi.APIRequest, cr *coreRequest) (output interface{}, err error) {
-			output, err = cr.or.GetSubscriptionByID(cr.ctx, r.PP["subid"])
-			return output, err
+			if strings.EqualFold(r.QP["fetchstatus"], "true") {
+				return cr.or.GetSubscriptionByIDWithStatus(cr.ctx, r.PP["subid"])
+			}
+			return cr.or.GetSubscriptionByID(cr.ctx, r.PP["subid"])
 		},
 	},
 }
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index 6e2e15bf7..bc7cf73fa 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -91,6 +91,17 @@ type Location struct {
 	Address string `json:"address"`
 }
 
+type ListenerCheckpoint struct {
+	Block            int64 `json:"block"`
+	TransactionIndex int64 `json:"transactionIndex"`
+	LogIndex         int64 `json:"logIndex"`
+}
+
+type ListenerStatus struct {
+	Checkpoint ListenerCheckpoint `json:"checkpoint"`
+	Catchup    bool               `json:"catchup"`
+}
+
 type EthconnectMessageRequest struct {
 	Headers EthconnectMessageHeaders `json:"headers,omitempty"`
 	To      string                   `json:"to"`
@@ -728,6 +739,24 @@ func (e *Ethereum) DeleteContractListener(ctx context.Context, subscription *cor
 	return e.streams.deleteSubscription(ctx, subscription.BackendID)
 }
 
+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
+	}
+
+	checkpoint := &ListenerStatus{
+		Catchup: sub.Catchup,
+		Checkpoint: ListenerCheckpoint{
+			Block:            sub.Checkpoint.Block,
+			TransactionIndex: sub.Checkpoint.TransactionIndex,
+			LogIndex:         sub.Checkpoint.LogIndex,
+		},
+	}
+
+	return checkpoint, nil
+}
+
 func (e *Ethereum) GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error) {
 	return &ffi2abi.ParamValidator{}, nil
 }
diff --git a/internal/blockchain/ethereum/eventstream.go b/internal/blockchain/ethereum/eventstream.go
index 5979f3eee..7adfc70a9 100644
--- a/internal/blockchain/ethereum/eventstream.go
+++ b/internal/blockchain/ethereum/eventstream.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/go-resty/resty/v2"
 	"github.com/hyperledger/firefly-common/pkg/ffresty"
+	"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-signer/pkg/abi"
@@ -51,12 +52,19 @@ type eventStream struct {
 }
 
 type subscription struct {
-	ID        string     `json:"id"`
-	Name      string     `json:"name,omitempty"`
-	Stream    string     `json:"stream"`
-	FromBlock string     `json:"fromBlock"`
-	Address   string     `json:"address"`
-	Event     *abi.Entry `json:"event"`
+	ID               string            `json:"id"`
+	Name             string            `json:"name,omitempty"`
+	Stream           string            `json:"stream"`
+	FromBlock        string            `json:"fromBlock"`
+	EthCompatAddress string            `json:"address,omitempty"`
+	EthCompatEvent   *abi.Entry        `json:"event,omitempty"`
+	Filters          []fftypes.JSONAny `json:"filters"`
+	subscriptionCheckpoint
+}
+
+type subscriptionCheckpoint struct {
+	Checkpoint ListenerCheckpoint `json:"checkpoint,omitempty"`
+	Catchup    bool               `json:"catchup,omitempty"`
 }
 
 func newStreamManager(client *resty.Client, cache *ccache.Cache, cacheTTL time.Duration) *streamManager {
@@ -181,11 +189,11 @@ func (s *streamManager) createSubscription(ctx context.Context, location *Locati
 		fromBlock = "latest"
 	}
 	sub := subscription{
-		Name:      subName,
-		Stream:    stream,
-		FromBlock: fromBlock,
-		Address:   location.Address,
-		Event:     abi,
+		Name:             subName,
+		Stream:           stream,
+		FromBlock:        fromBlock,
+		EthCompatAddress: location.Address,
+		EthCompatEvent:   abi,
 	}
 	res, err := s.client.R().
 		SetContext(ctx).
diff --git a/internal/blockchain/fabric/fabric.go b/internal/blockchain/fabric/fabric.go
index bd3d54442..fba79312f 100644
--- a/internal/blockchain/fabric/fabric.go
+++ b/internal/blockchain/fabric/fabric.go
@@ -833,6 +833,11 @@ func (f *Fabric) DeleteContractListener(ctx context.Context, subscription *core.
 	return f.streams.deleteSubscription(ctx, subscription.BackendID)
 }
 
+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) GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error) {
 	// Fabconnect does not require any additional validation beyond "JSON Schema correctness" at this time
 	return nil, nil
diff --git a/internal/contracts/manager.go b/internal/contracts/manager.go
index e0fe6ddd6..3d2388c52 100644
--- a/internal/contracts/manager.go
+++ b/internal/contracts/manager.go
@@ -54,6 +54,7 @@ type Manager interface {
 	AddContractListener(ctx context.Context, listener *core.ContractListenerInput) (output *core.ContractListener, err error)
 	AddContractAPIListener(ctx context.Context, apiName, eventPath string, listener *core.ContractListener) (output *core.ContractListener, err error)
 	GetContractListenerByNameOrID(ctx context.Context, nameOrID string) (*core.ContractListener, error)
+	GetContractListenerByNameOrIDWithStatus(ctx context.Context, nameOrID string) (*core.ContractListenerWithStatus, error)
 	GetContractListeners(ctx context.Context, filter database.AndFilter) ([]*core.ContractListener, *database.FilterResult, error)
 	GetContractAPIListeners(ctx context.Context, apiName, eventPath string, filter database.AndFilter) ([]*core.ContractListener, *database.FilterResult, error)
 	DeleteContractListenerByNameOrID(ctx context.Context, nameOrID string) error
@@ -572,6 +573,24 @@ func (cm *contractManager) GetContractListenerByNameOrID(ctx context.Context, na
 	return listener, nil
 }
 
+func (cm *contractManager) GetContractListenerByNameOrIDWithStatus(ctx context.Context, nameOrID string) (enrichedListener *core.ContractListenerWithStatus, err error) {
+	listener, err := cm.GetContractListenerByNameOrID(ctx, nameOrID)
+	if err != nil {
+		return nil, err
+	}
+	status, err := cm.blockchain.GetContractListenerStatus(ctx, listener.BackendID)
+	if err != nil {
+		status = core.ListenerStatusError{
+			StatusError: err.Error(),
+		}
+	}
+	enrichedListener = &core.ContractListenerWithStatus{
+		ContractListener: *listener,
+		Status:           status,
+	}
+	return enrichedListener, nil
+}
+
 func (cm *contractManager) GetContractListeners(ctx context.Context, filter database.AndFilter) ([]*core.ContractListener, *database.FilterResult, error) {
 	return cm.database.GetContractListeners(ctx, cm.namespace, filter)
 }
diff --git a/internal/coremsgs/en_api_translations.go b/internal/coremsgs/en_api_translations.go
index 7a81972b7..5548926a9 100644
--- a/internal/coremsgs/en_api_translations.go
+++ b/internal/coremsgs/en_api_translations.go
@@ -50,7 +50,7 @@ var (
 	APIParamsDID                            = ffm("api.params.DID", "The identity DID")
 	APIParamsNodeNameOrID                   = ffm("api.params.nodeNameOrID", "The name or ID of the node")
 	APIParamsOrgNameOrID                    = ffm("api.params.orgNameOrID", "The name or ID of the org")
-	APIParamsTokenAccountKey                = ffm("api.params.tokenAccountKey", "The key for the token account. The exact format may vary based on the token connector use.")
+	APIParamsTokenAccountKey                = ffm("api.params.tokenAccountKey", "The key for the token account. The exact format may vary based on the token connector use")
 	APIParamsTokenPoolNameOrID              = ffm("api.params.tokenPoolNameOrID", "The token pool name or ID")
 	APIParamsTokenTransferFromOrTo          = ffm("api.params.tokenTransferFromOrTo", "The sending or receiving token account for a token transfer")
 	APIParamsTokenTransferID                = ffm("api.params.tokenTransferID", "The token transfer ID")
@@ -63,6 +63,7 @@ var (
 	APIParamsMetadata                       = ffm("api.params.metadata", "Metadata associated with this data item")
 	APIParamsAutometa                       = ffm("api.params.autometa", "When set, FireFly will automatically generate JSON metadata with the upload details")
 	APIParamsContractAPIID                  = ffm("api.params.contractAPIID", "The ID of the contract API")
+	APIParamsFetchStatus                    = ffm("api.params.fetchStatus", "When set, the API will return additional status information if available")
 
 	APIEndpointsAdminGetNamespaceByName = ffm("api.endpoints.adminGetNamespaceByName", "Gets a namespace by name")
 	APIEndpointsAdminGetNamespaces      = ffm("api.endpoints.adminGetNamespaces", "List namespaces")
diff --git a/internal/coremsgs/en_error_messages.go b/internal/coremsgs/en_error_messages.go
index 1f9188c93..912d245cb 100644
--- a/internal/coremsgs/en_error_messages.go
+++ b/internal/coremsgs/en_error_messages.go
@@ -262,4 +262,5 @@ var (
 	MsgInvalidConnectorName               = ffe("FF10420", "Could not find name %s for %s connector")
 	MsgCannotInitLegacyNS                 = ffe("FF10421", "could not initialize legacy '%s' namespace - found conflicting V1 multi-party config in %s and %s")
 	MsgInvalidGroupMember                 = ffe("FF10422", "invalid group member - node '%s' is not owned by '%s' or any of its ancestors")
+	MsgContractListenerStatusInvalid      = ffe("FF10423", "Failed to validate contract listener status: %v", 400)
 )
diff --git a/internal/database/postgres/postgres.go b/internal/database/postgres/postgres.go
index b309be23f..9af8c78e1 100644
--- a/internal/database/postgres/postgres.go
+++ b/internal/database/postgres/postgres.go
@@ -19,7 +19,6 @@ package postgres
 import (
 	"context"
 	"fmt"
-	"math/big"
 
 	"database/sql"
 
@@ -55,21 +54,12 @@ func (psql *Postgres) MigrationsDir() string {
 	return psql.Name()
 }
 
-// Attempt to create a unique 64-bit int from the given name, by selecting 4 bytes from the
-// beginning and end of the string.
-func lockIndex(lockName string) int64 {
-	if len(lockName) >= 4 {
-		lockName = lockName[0:4] + lockName[len(lockName)-4:]
-	}
-	return big.NewInt(0).SetBytes([]byte(lockName)).Int64()
-}
-
 func (psql *Postgres) Features() sqlcommon.SQLFeatures {
 	features := sqlcommon.DefaultSQLProviderFeatures()
 	features.PlaceholderFormat = sq.Dollar
 	features.UseILIKE = false // slower than lower()
-	features.AcquireLock = func(lockName string) string {
-		return fmt.Sprintf(`SELECT pg_advisory_xact_lock(%d);`, lockIndex(lockName))
+	features.ExclusiveTableLockSQL = func(table string) string {
+		return fmt.Sprintf(`LOCK TABLE "%s" IN EXCLUSIVE MODE;`, table)
 	}
 	features.MultiRowInsert = true
 	return features
diff --git a/internal/database/sqlcommon/event_sql.go b/internal/database/sqlcommon/event_sql.go
index 9828b4463..911cffec5 100644
--- a/internal/database/sqlcommon/event_sql.go
+++ b/internal/database/sqlcommon/event_sql.go
@@ -93,17 +93,11 @@ func (s *SQLCommon) eventInserted(ctx context.Context, event *core.Event) {
 
 func (s *SQLCommon) insertEventsPreCommit(ctx context.Context, tx *txWrapper, events []*core.Event) (err error) {
 
-	namespaces := make(map[string]bool)
-	for _, event := range events {
-		namespaces[event.Namespace] = true
-	}
-	for namespace := range namespaces {
-		// We take the cost of a lock - scoped to the namespace(s) being updated.
-		// This allows us to rely on the sequence to always be increasing, even when writing events
-		// concurrently (it does not guarantee we won't get a gap in the sequences).
-		if err = s.acquireLockTx(ctx, namespace, tx); err != nil {
-			return err
-		}
+	// We take the cost of a full table lock on the events table.
+	// This allows us to rely on the sequence to always be increasing, even when writing events
+	// concurrently (it does not guarantee we won't get a gap in the sequences).
+	if err = s.lockTableExclusiveTx(ctx, eventsTable, tx); err != nil {
+		return err
 	}
 
 	if s.features.MultiRowInsert {
diff --git a/internal/database/sqlcommon/provider.go b/internal/database/sqlcommon/provider.go
index 1bdfeaa27..e4aa38138 100644
--- a/internal/database/sqlcommon/provider.go
+++ b/internal/database/sqlcommon/provider.go
@@ -28,10 +28,10 @@ const (
 )
 
 type SQLFeatures struct {
-	UseILIKE          bool
-	MultiRowInsert    bool
-	PlaceholderFormat sq.PlaceholderFormat
-	AcquireLock       func(lockName string) string
+	UseILIKE              bool
+	MultiRowInsert        bool
+	PlaceholderFormat     sq.PlaceholderFormat
+	ExclusiveTableLockSQL func(table string) string
 }
 
 func DefaultSQLProviderFeatures() SQLFeatures {
diff --git a/internal/database/sqlcommon/sqlcommon.go b/internal/database/sqlcommon/sqlcommon.go
index 318ca1b64..ddfc6bedd 100644
--- a/internal/database/sqlcommon/sqlcommon.go
+++ b/internal/database/sqlcommon/sqlcommon.go
@@ -90,6 +90,7 @@ type txWrapper struct {
 	sqlTX           *sql.Tx
 	preCommitEvents []*core.Event
 	postCommit      []func()
+	tableLocks      []string
 }
 
 func (s *SQLCommon) Init(ctx context.Context, provider Provider, config config.Section, capabilities *database.Capabilities) (err error) {
@@ -412,18 +413,28 @@ func (s *SQLCommon) addPreCommitEvent(tx *txWrapper, event *core.Event) {
 	tx.preCommitEvents = append(tx.preCommitEvents, event)
 }
 
-func (s *SQLCommon) acquireLockTx(ctx context.Context, lockName string, tx *txWrapper) error {
+func (tx *txWrapper) tableIsLocked(table string) bool {
+	for _, t := range tx.tableLocks {
+		if t == table {
+			return true
+		}
+	}
+	return false
+}
+
+func (s *SQLCommon) lockTableExclusiveTx(ctx context.Context, table string, tx *txWrapper) error {
 	l := log.L(ctx)
-	if s.features.AcquireLock != nil {
-		sqlQuery := s.features.AcquireLock(lockName)
+	if s.features.ExclusiveTableLockSQL != nil && !tx.tableIsLocked(table) {
+		sqlQuery := s.features.ExclusiveTableLockSQL(table)
 
-		l.Debugf(`SQL-> lock %s`, lockName)
+		l.Debugf(`SQL-> lock %s`, table)
 		_, err := tx.sqlTX.ExecContext(ctx, sqlQuery)
 		if err != nil {
 			l.Errorf(`SQL lock failed: %s sql=[ %s ]`, err, sqlQuery)
 			return i18n.WrapError(ctx, err, coremsgs.MsgDBLockFailed)
 		}
-		l.Debugf(`SQL<- lock %s`, lockName)
+		tx.tableLocks = append(tx.tableLocks, table)
+		l.Debugf(`SQL<- lock %s`, table)
 	}
 	return nil
 }
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 9e537add8..92f4533eb 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -74,6 +74,7 @@ type Orchestrator interface {
 	// Subscription management
 	GetSubscriptions(ctx context.Context, filter database.AndFilter) ([]*core.Subscription, *database.FilterResult, error)
 	GetSubscriptionByID(ctx context.Context, id string) (*core.Subscription, error)
+	GetSubscriptionByIDWithStatus(ctx context.Context, id string) (*core.SubscriptionWithStatus, error)
 	CreateSubscription(ctx context.Context, subDef *core.Subscription) (*core.Subscription, error)
 	CreateUpdateSubscription(ctx context.Context, subDef *core.Subscription) (*core.Subscription, error)
 	DeleteSubscription(ctx context.Context, id string) error
diff --git a/internal/orchestrator/subscriptions.go b/internal/orchestrator/subscriptions.go
index 742de7a80..af12327d3 100644
--- a/internal/orchestrator/subscriptions.go
+++ b/internal/orchestrator/subscriptions.go
@@ -76,3 +76,30 @@ func (or *orchestrator) GetSubscriptionByID(ctx context.Context, id string) (*co
 	}
 	return or.database().GetSubscriptionByID(ctx, or.namespace.Name, u)
 }
+
+func (or *orchestrator) GetSubscriptionByIDWithStatus(ctx context.Context, id string) (*core.SubscriptionWithStatus, error) {
+	sub, err := or.GetSubscriptionByID(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	if sub == nil {
+		return nil, nil
+	}
+
+	offset, err := or.database().GetOffset(ctx, core.OffsetTypeSubscription, sub.ID.String())
+	if err != nil {
+		return nil, err
+	}
+
+	subWithStatus := &core.SubscriptionWithStatus{
+		Subscription: *sub,
+	}
+
+	if offset != nil {
+		subWithStatus.Status = core.SubscriptionStatus{
+			CurrentOffset: offset.Current,
+		}
+	}
+
+	return subWithStatus, nil
+}
diff --git a/internal/tokens/fftokens/fftokens.go b/internal/tokens/fftokens/fftokens.go
index 5725f098b..93b5c36e5 100644
--- a/internal/tokens/fftokens/fftokens.go
+++ b/internal/tokens/fftokens/fftokens.go
@@ -365,6 +365,7 @@ func (ft *FFTokens) handleTokenTransfer(ctx context.Context, t core.TokenTransfe
 
 	if protocolID == "" ||
 		poolLocator == "" ||
+		signerAddress == "" ||
 		value == "" ||
 		(t != core.TokenTransferTypeMint && fromAddress == "") ||
 		(t != core.TokenTransferTypeBurn && toAddress == "") ||
@@ -437,6 +438,7 @@ func (ft *FFTokens) handleTokenApproval(ctx context.Context, data fftypes.JSONOb
 	if protocolID == "" ||
 		subject == "" ||
 		poolLocator == "" ||
+		signerAddress == "" ||
 		operatorAddress == "" ||
 		blockchainEvent == nil {
 		log.L(ctx).Errorf("Approval event is not valid - missing data: %+v", data)
diff --git a/mocks/blockchainmocks/plugin.go b/mocks/blockchainmocks/plugin.go
index 50dfe3797..2564792a6 100644
--- a/mocks/blockchainmocks/plugin.go
+++ b/mocks/blockchainmocks/plugin.go
@@ -154,6 +154,29 @@ 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)
+
+	var r0 interface{}
+	if rf, ok := ret.Get(0).(func(context.Context, string) interface{}); ok {
+		r0 = rf(ctx, subID)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(interface{})
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, subID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetFFIParamValidator provides a mock function with given fields: ctx
 func (_m *Plugin) GetFFIParamValidator(ctx context.Context) (fftypes.FFIParamValidator, error) {
 	ret := _m.Called(ctx)
diff --git a/mocks/contractmocks/manager.go b/mocks/contractmocks/manager.go
index 1924f395f..0b0125bbb 100644
--- a/mocks/contractmocks/manager.go
+++ b/mocks/contractmocks/manager.go
@@ -235,6 +235,29 @@ func (_m *Manager) GetContractListenerByNameOrID(ctx context.Context, nameOrID s
 	return r0, r1
 }
 
+// GetContractListenerByNameOrIDWithStatus provides a mock function with given fields: ctx, nameOrID
+func (_m *Manager) GetContractListenerByNameOrIDWithStatus(ctx context.Context, nameOrID string) (*core.ContractListenerWithStatus, error) {
+	ret := _m.Called(ctx, nameOrID)
+
+	var r0 *core.ContractListenerWithStatus
+	if rf, ok := ret.Get(0).(func(context.Context, string) *core.ContractListenerWithStatus); ok {
+		r0 = rf(ctx, nameOrID)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*core.ContractListenerWithStatus)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, nameOrID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetContractListeners provides a mock function with given fields: ctx, filter
 func (_m *Manager) GetContractListeners(ctx context.Context, filter database.AndFilter) ([]*core.ContractListener, *database.FilterResult, error) {
 	ret := _m.Called(ctx, filter)
diff --git a/mocks/orchestratormocks/orchestrator.go b/mocks/orchestratormocks/orchestrator.go
index e2e092127..d3ae3ef3a 100644
--- a/mocks/orchestratormocks/orchestrator.go
+++ b/mocks/orchestratormocks/orchestrator.go
@@ -971,6 +971,29 @@ func (_m *Orchestrator) GetSubscriptionByID(ctx context.Context, id string) (*co
 	return r0, r1
 }
 
+// GetSubscriptionByIDWithStatus provides a mock function with given fields: ctx, id
+func (_m *Orchestrator) GetSubscriptionByIDWithStatus(ctx context.Context, id string) (*core.SubscriptionWithStatus, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 *core.SubscriptionWithStatus
+	if rf, ok := ret.Get(0).(func(context.Context, string) *core.SubscriptionWithStatus); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*core.SubscriptionWithStatus)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetSubscriptions provides a mock function with given fields: ctx, filter
 func (_m *Orchestrator) GetSubscriptions(ctx context.Context, filter database.AndFilter) ([]*core.Subscription, *database.FilterResult, error) {
 	ret := _m.Called(ctx, filter)
diff --git a/pkg/blockchain/plugin.go b/pkg/blockchain/plugin.go
index 0af67bea0..8af1c5547 100644
--- a/pkg/blockchain/plugin.go
+++ b/pkg/blockchain/plugin.go
@@ -75,6 +75,8 @@ type Plugin interface {
 	// DeleteContractListener deletes a previously-created subscription
 	DeleteContractListener(ctx context.Context, subscription *core.ContractListener) error
 
+	GetContractListenerStatus(ctx context.Context, subID string) (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/core/contract_listener.go b/pkg/core/contract_listener.go
index d68e65b25..fb3fb1241 100644
--- a/pkg/core/contract_listener.go
+++ b/pkg/core/contract_listener.go
@@ -39,10 +39,18 @@ type ContractListener struct {
 	Options   *ContractListenerOptions `ffstruct:"ContractListener" json:"options,omitempty"`
 }
 
+type ContractListenerWithStatus struct {
+	ContractListener
+	Status interface{} `ffstruct:"ContractListenerWithStatus" json:"status,omitempty" ffexcludeinput:"true"`
+}
 type ContractListenerOptions struct {
 	FirstEvent string `ffstruct:"ContractListenerOptions" json:"firstEvent,omitempty"`
 }
 
+type ListenerStatusError struct {
+	StatusError string `ffstruct:"ListenerStatusError" json:"error,omitempty"`
+}
+
 type ContractListenerInput struct {
 	ContractListener
 	EventPath string `ffstruct:"ContractListener" json:"eventPath,omitempty"`
diff --git a/pkg/core/subscription.go b/pkg/core/subscription.go
index 95acb1278..97e80e27f 100644
--- a/pkg/core/subscription.go
+++ b/pkg/core/subscription.go
@@ -122,6 +122,15 @@ type Subscription struct {
 	Updated   *fftypes.FFTime     `ffstruct:"Subscription" json:"updated" ffexcludeinput:"true"`
 }
 
+type SubscriptionWithStatus struct {
+	Subscription
+	Status SubscriptionStatus `ffstruct:"SubscriptionWithStatus" json:"status,omitempty" ffexcludeinput:"true"`
+}
+
+type SubscriptionStatus struct {
+	CurrentOffset int64 `ffstruct:"SubscriptionStatus" json:"currentOffset,omitempty" ffexcludeinout:"true"`
+}
+
 func (so *SubscriptionOptions) UnmarshalJSON(b []byte) error {
 	so.additionalOptions = fftypes.JSONObject{}
 	err := json.Unmarshal(b, &so.additionalOptions)


Test patch:
diff --git a/internal/apiserver/route_get_contract_listener_by_name_or_id_test.go b/internal/apiserver/route_get_contract_listener_by_name_or_id_test.go
index f7ba336db..b5ec429de 100644
--- a/internal/apiserver/route_get_contract_listener_by_name_or_id_test.go
+++ b/internal/apiserver/route_get_contract_listener_by_name_or_id_test.go
@@ -43,3 +43,20 @@ func TestGetContractListenerByNameOrID(t *testing.T) {
 
 	assert.Equal(t, 200, res.Result().StatusCode)
 }
+
+func TestGetContractListenerByNameOrIDWithStatus(t *testing.T) {
+	o, r := newTestAPIServer()
+	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
+	mcm := &contractmocks.Manager{}
+	o.On("Contracts").Return(mcm)
+	id := fftypes.NewUUID()
+	req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/contracts/listeners/"+id.String()+"?fetchstatus", nil)
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+	res := httptest.NewRecorder()
+
+	mcm.On("GetContractListenerByNameOrIDWithStatus", mock.Anything, id.String()).
+		Return(&core.ContractListenerWithStatus{}, nil)
+	r.ServeHTTP(res, req)
+
+	assert.Equal(t, 200, res.Result().StatusCode)
+}
diff --git a/internal/apiserver/route_get_subscription_by_id_test.go b/internal/apiserver/route_get_subscription_by_id_test.go
index d1a8d3337..6c34a92a7 100644
--- a/internal/apiserver/route_get_subscription_by_id_test.go
+++ b/internal/apiserver/route_get_subscription_by_id_test.go
@@ -38,3 +38,19 @@ func TestGetSubscriptionByID(t *testing.T) {
 
 	assert.Equal(t, 200, res.Result().StatusCode)
 }
+
+func TestGetSubscriptionByIDWithStatus(t *testing.T) {
+	o, r := newTestAPIServer()
+	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
+	req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/subscriptions/abcd12345?fetchstatus", nil)
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+	res := httptest.NewRecorder()
+
+	o.On("GetSubscriptionByID", mock.Anything, "abcd12345").
+		Return(&core.Subscription{}, nil)
+	o.On("GetSubscriptionByIDWithStatus", mock.Anything, "abcd12345").
+		Return(&core.SubscriptionWithStatus{}, nil)
+	r.ServeHTTP(res, req)
+
+	assert.Equal(t, 200, res.Result().StatusCode)
+}
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index bdb80e10e..e3fd5deb6 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -2997,7 +2997,6 @@ func TestAddAndRemoveFireflySubscription(t *testing.T) {
 		}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3039,7 +3038,6 @@ func TestAddFireflySubscriptionV1(t *testing.T) {
 		}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 1))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3076,7 +3074,6 @@ func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
 		httpmock.NewJsonResponderOrPanic(200, subscription{}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3113,7 +3110,6 @@ func TestAddFireflySubscriptionCreateError(t *testing.T) {
 	httpmock.RegisterResponder("POST", "http://localhost:12345/",
 		httpmock.NewJsonResponderOrPanic(500, `pop`))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3149,7 +3145,6 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 		httpmock.NewJsonResponderOrPanic(500, `pop`))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
 
-	resetConf(e)
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
@@ -3165,3 +3160,74 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "FF10111", err)
 }
+
+func TestGetContractListenerStatus(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+	resetConf(e)
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	checkpoint := ListenerCheckpoint{
+		Block:            0,
+		TransactionIndex: -1,
+		LogIndex:         -1,
+	}
+
+	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(200, subscription{
+			ID: "sub1", Stream: "es12345", Name: "ff-sub-1132312312312", subscriptionCheckpoint: subscriptionCheckpoint{
+				Catchup:    false,
+				Checkpoint: checkpoint,
+			},
+		}))
+
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics)
+	assert.NoError(t, err)
+
+	status, err := e.GetContractListenerStatus(context.Background(), "sub1")
+	assert.NotNil(t, status)
+	assert.NoError(t, err)
+}
+
+func TestGetContractListenerStatusGetSubFail(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(500, `pop`))
+
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics)
+	assert.NoError(t, err)
+
+	status, err := e.GetContractListenerStatus(context.Background(), "sub1")
+	assert.Nil(t, status)
+	assert.Regexp(t, "FF10111", err)
+}
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index d9cf65dee..453cd452e 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -2519,3 +2519,14 @@ func TestSubmitNetworkActionVersionError(t *testing.T) {
 	err := e.SubmitNetworkAction(context.Background(), "", signer, core.NetworkActionTerminate, location)
 	assert.Regexp(t, "FF10284", err)
 }
+
+func TestGetContractListenerStatus(t *testing.T) {
+	e, cancel := newTestFabric()
+	defer cancel()
+	httpmock.ActivateNonDefault(e.client.GetClient())
+	defer httpmock.DeactivateAndReset()
+
+	status, err := e.GetContractListenerStatus(context.Background(), "id")
+	assert.Nil(t, status)
+	assert.NoError(t, err)
+}
diff --git a/internal/contracts/manager_test.go b/internal/contracts/manager_test.go
index 81ecfd11f..884d76d0d 100644
--- a/internal/contracts/manager_test.go
+++ b/internal/contracts/manager_test.go
@@ -1579,6 +1579,51 @@ func TestGetContractListenerByNameOrID(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestGetContractListenerByNameOrIDWithStatus(t *testing.T) {
+	cm := newTestContractManager()
+	mdi := cm.database.(*databasemocks.Plugin)
+	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+
+	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)
+
+	_, err := cm.GetContractListenerByNameOrIDWithStatus(context.Background(), id.String())
+	assert.NoError(t, err)
+}
+
+func TestGetContractListenerByNameOrIDWithStatusListenerFail(t *testing.T) {
+	cm := newTestContractManager()
+	mdi := cm.database.(*databasemocks.Plugin)
+
+	id := fftypes.NewUUID()
+	mdi.On("GetContractListenerByID", context.Background(), "ns1", id).Return(nil, fmt.Errorf("pop"))
+
+	_, err := cm.GetContractListenerByNameOrIDWithStatus(context.Background(), id.String())
+	assert.EqualError(t, err, "pop")
+}
+
+func TestGetContractListenerByNameOrIDWithStatusPluginFail(t *testing.T) {
+	cm := newTestContractManager()
+	mdi := cm.database.(*databasemocks.Plugin)
+	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+
+	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"))
+
+	listener, err := cm.GetContractListenerByNameOrIDWithStatus(context.Background(), id.String())
+
+	testError := core.ListenerStatusError{
+		StatusError: "pop",
+	}
+
+	assert.Equal(t, listener.Status, testError)
+	assert.NoError(t, err)
+}
+
 func TestGetContractListenerByNameOrIDFail(t *testing.T) {
 	cm := newTestContractManager()
 	mdi := cm.database.(*databasemocks.Plugin)
diff --git a/internal/database/postgres/postgres_test.go b/internal/database/postgres/postgres_test.go
index d27284732..60a4f7de7 100644
--- a/internal/database/postgres/postgres_test.go
+++ b/internal/database/postgres/postgres_test.go
@@ -40,8 +40,7 @@ func TestPostgresProvider(t *testing.T) {
 
 	assert.Equal(t, "postgres", psql.Name())
 	assert.Equal(t, sq.Dollar, psql.Features().PlaceholderFormat)
-	assert.Equal(t, `SELECT pg_advisory_xact_lock(8387236824920056683);`, psql.Features().AcquireLock("test-lock"))
-	assert.Equal(t, `SELECT pg_advisory_xact_lock(116);`, psql.Features().AcquireLock("t"))
+	assert.Equal(t, `LOCK TABLE "events" IN EXCLUSIVE MODE;`, psql.Features().ExclusiveTableLockSQL("events"))
 
 	insert := sq.Insert("test").Columns("col1").Values("val1")
 	insert, query := psql.ApplyInsertQueryCustomizations(insert, true)
diff --git a/internal/database/sqlcommon/event_sql_test.go b/internal/database/sqlcommon/event_sql_test.go
index ab1c57993..e6be3a0fa 100644
--- a/internal/database/sqlcommon/event_sql_test.go
+++ b/internal/database/sqlcommon/event_sql_test.go
@@ -100,10 +100,10 @@ func TestInsertEventFailBegin(t *testing.T) {
 func TestInsertEventFailLock(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
-	mock.ExpectExec("<acquire lock ns1>").WillReturnError(fmt.Errorf("pop"))
+	mock.ExpectExec("LOCK .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	eventID := fftypes.NewUUID()
-	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID, Namespace: "ns1"})
+	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID})
 	assert.Regexp(t, "FF10345", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
@@ -111,11 +111,11 @@ func TestInsertEventFailLock(t *testing.T) {
 func TestInsertEventFailInsert(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
-	mock.ExpectExec("<acquire lock ns1>").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("LOCK .*").WillReturnResult(driver.ResultNoRows)
 	mock.ExpectExec("INSERT .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	eventID := fftypes.NewUUID()
-	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID, Namespace: "ns1"})
+	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID})
 	assert.Regexp(t, "FF10116", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
@@ -124,10 +124,10 @@ func TestInsertEventFailCommit(t *testing.T) {
 	s, mock := newMockProvider().init()
 	eventID := fftypes.NewUUID()
 	mock.ExpectBegin()
-	mock.ExpectExec("<acquire lock ns1>").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("LOCK .*").WillReturnResult(driver.ResultNoRows)
 	mock.ExpectExec("INSERT .*").WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectCommit().WillReturnError(fmt.Errorf("pop"))
-	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID, Namespace: "ns1"})
+	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID})
 	assert.Regexp(t, "FF10119", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
@@ -143,7 +143,7 @@ func TestInsertEventsPreCommitMultiRowOK(t *testing.T) {
 	s.callbacks.On("OrderedUUIDCollectionNSEvent", database.CollectionEvents, core.ChangeEventTypeCreated, "ns1", ev2.ID, int64(1002))
 
 	mock.ExpectBegin()
-	mock.ExpectExec("<acquire lock ns1>").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("LOCK .*").WillReturnResult(driver.ResultNoRows)
 	mock.ExpectQuery("INSERT.*").WillReturnRows(sqlmock.NewRows([]string{sequenceColumn}).
 		AddRow(int64(1001)).
 		AddRow(int64(1002)),
@@ -164,7 +164,7 @@ func TestInsertEventsPreCommitMultiRowFail(t *testing.T) {
 	s.fakePSQLInsert = true
 	ev1 := &core.Event{ID: fftypes.NewUUID(), Namespace: "ns1"}
 	mock.ExpectBegin()
-	mock.ExpectExec("<acquire lock ns1>").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("LOCK .*").WillReturnResult(driver.ResultNoRows)
 	mock.ExpectQuery("INSERT.*").WillReturnError(fmt.Errorf("pop"))
 	ctx, tx, autoCommit, err := s.beginOrUseTx(context.Background())
 	tx.preCommitEvents = []*core.Event{ev1}
@@ -179,7 +179,7 @@ func TestInsertEventsPreCommitSingleRowFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	ev1 := &core.Event{ID: fftypes.NewUUID(), Namespace: "ns1"}
 	mock.ExpectBegin()
-	mock.ExpectExec("<acquire lock ns1>").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("LOCK .*").WillReturnResult(driver.ResultNoRows)
 	mock.ExpectExec("INSERT.*").WillReturnError(fmt.Errorf("pop"))
 	ctx, tx, autoCommit, err := s.beginOrUseTx(context.Background())
 	tx.preCommitEvents = []*core.Event{ev1}
diff --git a/internal/database/sqlcommon/provider_mock_test.go b/internal/database/sqlcommon/provider_mock_test.go
index b507af5db..2331c6415 100644
--- a/internal/database/sqlcommon/provider_mock_test.go
+++ b/internal/database/sqlcommon/provider_mock_test.go
@@ -79,8 +79,8 @@ func (mp *mockProvider) MigrationsDir() string {
 func (psql *mockProvider) Features() SQLFeatures {
 	features := DefaultSQLProviderFeatures()
 	features.UseILIKE = true
-	features.AcquireLock = func(lockName string) string {
-		return fmt.Sprintf(`<acquire lock %s>`, lockName)
+	features.ExclusiveTableLockSQL = func(table string) string {
+		return fmt.Sprintf(`LOCK TABLE "%s" IN EXCLUSIVE MODE;`, table)
 	}
 	return features
 }
diff --git a/internal/database/sqlcommon/sqlcommon_test.go b/internal/database/sqlcommon/sqlcommon_test.go
index 9fe69d94e..3c23e50b1 100644
--- a/internal/database/sqlcommon/sqlcommon_test.go
+++ b/internal/database/sqlcommon/sqlcommon_test.go
@@ -328,6 +328,19 @@ func TestQueryResSwallowError(t *testing.T) {
 	assert.Equal(t, int64(-1), *res.TotalCount)
 }
 
+func TestDoubleLock(t *testing.T) {
+	s, mdb := newMockProvider().init()
+	mdb.ExpectBegin()
+	mdb.ExpectExec("LOCK .*").WillReturnResult(driver.ResultNoRows)
+	ctx, tx, _, err := s.beginOrUseTx(context.Background())
+	assert.NoError(t, err)
+	err = s.lockTableExclusiveTx(ctx, "table1", tx)
+	assert.NoError(t, err)
+	err = s.lockTableExclusiveTx(ctx, "table1", tx)
+	assert.NoError(t, err)
+	assert.NoError(t, mdb.ExpectationsWereMet())
+}
+
 func TestInsertTxRowsBadConfig(t *testing.T) {
 	s, mdb := newMockProvider().init()
 	mdb.ExpectBegin()
diff --git a/internal/orchestrator/subscriptions_test.go b/internal/orchestrator/subscriptions_test.go
index fd6afb422..f10e41829 100644
--- a/internal/orchestrator/subscriptions_test.go
+++ b/internal/orchestrator/subscriptions_test.go
@@ -157,3 +157,63 @@ func TestGetSubscriptionDefsByIDBadID(t *testing.T) {
 	_, err := or.GetSubscriptionByID(context.Background(), "")
 	assert.Regexp(t, "FF00138", err)
 }
+
+func TestGetSGetSubscriptionsByIDWithStatus(t *testing.T) {
+	or := newTestOrchestrator()
+	defer or.cleanup(t)
+
+	u := fftypes.NewUUID()
+	sub := &core.Subscription{
+		SubscriptionRef: core.SubscriptionRef{
+			ID:        u,
+			Name:      "sub1",
+			Namespace: "ns1",
+		},
+	}
+	or.mdi.On("GetSubscriptionByID", context.Background(), "ns", u).Return(sub, nil)
+	or.mdi.On("GetOffset", context.Background(), core.OffsetTypeSubscription, u.String()).Return(&core.Offset{Current: 100}, nil)
+	subWithStatus, err := or.GetSubscriptionByIDWithStatus(context.Background(), u.String())
+	assert.NoError(t, err)
+	assert.NotNil(t, subWithStatus)
+}
+
+func TestGetSGetSubscriptionsByIDWithStatusQuerySubFail(t *testing.T) {
+	or := newTestOrchestrator()
+	defer or.cleanup(t)
+
+	u := fftypes.NewUUID()
+	or.mdi.On("GetSubscriptionByID", context.Background(), "ns", u).Return(nil, fmt.Errorf("pop"))
+	subWithStatus, err := or.GetSubscriptionByIDWithStatus(context.Background(), u.String())
+	assert.EqualError(t, err, "pop")
+	assert.Nil(t, subWithStatus)
+}
+
+func TestGetSGetSubscriptionsByIDWithStatusOffsetQueryError(t *testing.T) {
+	or := newTestOrchestrator()
+	defer or.cleanup(t)
+
+	u := fftypes.NewUUID()
+	sub := &core.Subscription{
+		SubscriptionRef: core.SubscriptionRef{
+			ID:        u,
+			Name:      "sub1",
+			Namespace: "ns1",
+		},
+	}
+	or.mdi.On("GetSubscriptionByID", context.Background(), "ns", u).Return(sub, nil)
+	or.mdi.On("GetOffset", context.Background(), core.OffsetTypeSubscription, u.String()).Return(nil, fmt.Errorf("pop"))
+	subWithStatus, err := or.GetSubscriptionByIDWithStatus(context.Background(), u.String())
+	assert.EqualError(t, err, "pop")
+	assert.Nil(t, subWithStatus)
+}
+
+func TestGetSGetSubscriptionsByIDWithStatusUnknownSub(t *testing.T) {
+	or := newTestOrchestrator()
+	defer or.cleanup(t)
+
+	u := fftypes.NewUUID()
+	or.mdi.On("GetSubscriptionByID", context.Background(), "ns", u).Return(nil, nil)
+	subWithStatus, err := or.GetSubscriptionByIDWithStatus(context.Background(), u.String())
+	assert.NoError(t, err)
+	assert.Nil(t, subWithStatus)
+}