Repository: hyperledger/firefly
Pull request #1027 (2022-08-31 17:39:01)

Use advisory lock instead of exclusive lock on events table
Advisory locks are identified with a 64-bit integer, so I've put in place a
rudimentary mapping from namespace names to a (probably) unique int64.
This should ensure consistent ordering for the writes to the events table
within a particular namespace, without blocking reads or other namespaces.


Patch:
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
index eaf809ee9..39c0b79c4 100644
--- a/docs/swagger/swagger.yaml
+++ b/docs/swagger/swagger.yaml
@@ -5249,19 +5249,12 @@ paths:
       operationId: getEvents
       parameters:
       - description: When set, the API will return the record that this item references
-          in its 'reference' field
+          in its 'ref' field
         in: query
         name: fetchreferences
         schema:
           example: "true"
           type: string
-      - description: When set, the API will return the record that this item references
-          in its 'reference' field
-        in: query
-        name: fetchreference
-        schema:
-          example: "true"
-          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -5440,13 +5433,6 @@ paths:
         required: true
         schema:
           type: string
-      - description: When set, the API will return the record that this item references
-          in its 'reference' field
-        in: query
-        name: fetchreference
-        schema:
-          example: "true"
-          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -14529,19 +14515,12 @@ paths:
           example: default
           type: string
       - description: When set, the API will return the record that this item references
-          in its 'reference' field
+          in its 'ref' field
         in: query
         name: fetchreferences
         schema:
           example: "true"
           type: string
-      - description: When set, the API will return the record that this item references
-          in its 'reference' field
-        in: query
-        name: fetchreference
-        schema:
-          example: "true"
-          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
@@ -14727,13 +14706,6 @@ paths:
         schema:
           example: default
           type: string
-      - description: When set, the API will return the record that this item references
-          in its 'reference' field
-        in: query
-        name: fetchreference
-        schema:
-          example: "true"
-          type: string
       - description: Server-side request timeout (milliseconds, or set a custom suffix
           like 10s)
         in: header
diff --git a/internal/apiserver/route_get_event_by_id.go b/internal/apiserver/route_get_event_by_id.go
index c045ea810..ce6d154a8 100644
--- a/internal/apiserver/route_get_event_by_id.go
+++ b/internal/apiserver/route_get_event_by_id.go
@@ -18,7 +18,6 @@ package apiserver
 
 import (
 	"net/http"
-	"strings"
 
 	"github.com/hyperledger/firefly-common/pkg/ffapi"
 	"github.com/hyperledger/firefly/internal/coremsgs"
@@ -32,19 +31,15 @@ var getEventByID = &ffapi.Route{
 	PathParams: []*ffapi.PathParam{
 		{Name: "eid", Description: coremsgs.APIParamsEventID},
 	},
-	QueryParams: []*ffapi.QueryParam{
-		{Name: "fetchreference", Example: "true", Description: coremsgs.APIParamsFetchReference, IsBool: true},
-	},
+	QueryParams:     nil,
 	Description:     coremsgs.APIEndpointsGetEventByID,
 	JSONInputValue:  nil,
 	JSONOutputValue: func() interface{} { return &core.Event{} },
 	JSONOutputCodes: []int{http.StatusOK},
 	Extensions: &coreExtensions{
 		CoreJSONHandler: func(r *ffapi.APIRequest, cr *coreRequest) (output interface{}, err error) {
-			if strings.EqualFold(r.QP["fetchreference"], "true") {
-				return cr.or.GetEventByIDWithReference(cr.ctx, r.PP["eid"])
-			}
-			return cr.or.GetEventByID(cr.ctx, r.PP["eid"])
+			output, err = cr.or.GetEventByID(cr.ctx, r.PP["eid"])
+			return output, err
 		},
 	},
 }
diff --git a/internal/apiserver/route_get_events.go b/internal/apiserver/route_get_events.go
index 7d10b7a12..8167b8de8 100644
--- a/internal/apiserver/route_get_events.go
+++ b/internal/apiserver/route_get_events.go
@@ -33,7 +33,6 @@ var getEvents = &ffapi.Route{
 	PathParams: nil,
 	QueryParams: []*ffapi.QueryParam{
 		{Name: "fetchreferences", Example: "true", Description: coremsgs.APIParamsFetchReferences, IsBool: true},
-		{Name: "fetchreference", Example: "true", Description: coremsgs.APIParamsFetchReference, IsBool: true},
 	},
 	Description:     coremsgs.APIEndpointsGetEvents,
 	JSONInputValue:  nil,
@@ -42,7 +41,7 @@ var getEvents = &ffapi.Route{
 	Extensions: &coreExtensions{
 		FilterFactory: database.EventQueryFactory,
 		CoreJSONHandler: func(r *ffapi.APIRequest, cr *coreRequest) (output interface{}, err error) {
-			if strings.EqualFold(r.QP["fetchreferences"], "true") || strings.EqualFold(r.QP["fetchreference"], "true") {
+			if strings.EqualFold(r.QP["fetchreferences"], "true") {
 				return filterResult(cr.or.GetEventsWithReferences(cr.ctx, cr.filter))
 			}
 			return filterResult(cr.or.GetEvents(cr.ctx, cr.filter))
diff --git a/internal/coremsgs/en_api_translations.go b/internal/coremsgs/en_api_translations.go
index 7a81972b7..0895c3a23 100644
--- a/internal/coremsgs/en_api_translations.go
+++ b/internal/coremsgs/en_api_translations.go
@@ -41,8 +41,7 @@ var (
 	APIParamsDatatypeName                   = ffm("api.params.datatypeName", "The name of the datatype")
 	APIParamsDatatypeVersion                = ffm("api.params.datatypeVersion", "The version of the datatype")
 	APIParamsEventID                        = ffm("api.params.eventID", "The event ID")
-	APIParamsFetchReferences                = ffm("api.params.fetchReferences", "When set, the API will return the record that this item references in its 'reference' field")
-	APIParamsFetchReference                 = ffm("api.params.fetchReference", "When set, the API will return the record that this item references in its 'reference' field")
+	APIParamsFetchReferences                = ffm("api.params.fetchReferences", "When set, the API will return the record that this item references in its 'ref' field")
 	APIParamsGroupHash                      = ffm("api.params.groupID", "The hash of the group")
 	APIParamsFetchVerifiers                 = ffm("api.params.fetchVerifiers", "When set, the API will return the verifier for this identity")
 	APIParamsIdentityID                     = ffm("api.params.identityID", "The identity ID, which is a UUID generated by FireFly")
diff --git a/internal/database/postgres/postgres.go b/internal/database/postgres/postgres.go
index 9af8c78e1..b309be23f 100644
--- a/internal/database/postgres/postgres.go
+++ b/internal/database/postgres/postgres.go
@@ -19,6 +19,7 @@ package postgres
 import (
 	"context"
 	"fmt"
+	"math/big"
 
 	"database/sql"
 
@@ -54,12 +55,21 @@ 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.ExclusiveTableLockSQL = func(table string) string {
-		return fmt.Sprintf(`LOCK TABLE "%s" IN EXCLUSIVE MODE;`, table)
+	features.AcquireLock = func(lockName string) string {
+		return fmt.Sprintf(`SELECT pg_advisory_xact_lock(%d);`, lockIndex(lockName))
 	}
 	features.MultiRowInsert = true
 	return features
diff --git a/internal/database/sqlcommon/event_sql.go b/internal/database/sqlcommon/event_sql.go
index 911cffec5..9828b4463 100644
--- a/internal/database/sqlcommon/event_sql.go
+++ b/internal/database/sqlcommon/event_sql.go
@@ -93,11 +93,17 @@ func (s *SQLCommon) eventInserted(ctx context.Context, event *core.Event) {
 
 func (s *SQLCommon) insertEventsPreCommit(ctx context.Context, tx *txWrapper, events []*core.Event) (err error) {
 
-	// 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
+	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
+		}
 	}
 
 	if s.features.MultiRowInsert {
diff --git a/internal/database/sqlcommon/provider.go b/internal/database/sqlcommon/provider.go
index e4aa38138..1bdfeaa27 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
-	ExclusiveTableLockSQL func(table string) string
+	UseILIKE          bool
+	MultiRowInsert    bool
+	PlaceholderFormat sq.PlaceholderFormat
+	AcquireLock       func(lockName string) string
 }
 
 func DefaultSQLProviderFeatures() SQLFeatures {
diff --git a/internal/database/sqlcommon/sqlcommon.go b/internal/database/sqlcommon/sqlcommon.go
index ddfc6bedd..318ca1b64 100644
--- a/internal/database/sqlcommon/sqlcommon.go
+++ b/internal/database/sqlcommon/sqlcommon.go
@@ -90,7 +90,6 @@ 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) {
@@ -413,28 +412,18 @@ func (s *SQLCommon) addPreCommitEvent(tx *txWrapper, event *core.Event) {
 	tx.preCommitEvents = append(tx.preCommitEvents, event)
 }
 
-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 {
+func (s *SQLCommon) acquireLockTx(ctx context.Context, lockName string, tx *txWrapper) error {
 	l := log.L(ctx)
-	if s.features.ExclusiveTableLockSQL != nil && !tx.tableIsLocked(table) {
-		sqlQuery := s.features.ExclusiveTableLockSQL(table)
+	if s.features.AcquireLock != nil {
+		sqlQuery := s.features.AcquireLock(lockName)
 
-		l.Debugf(`SQL-> lock %s`, table)
+		l.Debugf(`SQL-> lock %s`, lockName)
 		_, 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)
 		}
-		tx.tableLocks = append(tx.tableLocks, table)
-		l.Debugf(`SQL<- lock %s`, table)
+		l.Debugf(`SQL<- lock %s`, lockName)
 	}
 	return nil
 }
diff --git a/internal/orchestrator/data_query.go b/internal/orchestrator/data_query.go
index dbaea570f..ea67f42e5 100644
--- a/internal/orchestrator/data_query.go
+++ b/internal/orchestrator/data_query.go
@@ -135,18 +135,6 @@ func (or *orchestrator) GetEventByID(ctx context.Context, id string) (*core.Even
 	return or.database().GetEventByID(ctx, or.namespace.Name, u)
 }
 
-func (or *orchestrator) GetEventByIDWithReference(ctx context.Context, id string) (*core.EnrichedEvent, error) {
-	u, err := fftypes.ParseUUID(ctx, id)
-	if err != nil {
-		return nil, err
-	}
-	event, err := or.database().GetEventByID(ctx, or.namespace.Name, u)
-	if err != nil {
-		return nil, err
-	}
-	return or.events.EnrichEvent(ctx, event)
-}
-
 func (or *orchestrator) GetTransactions(ctx context.Context, filter database.AndFilter) ([]*core.Transaction, *database.FilterResult, error) {
 	return or.database().GetTransactions(ctx, or.namespace.Name, filter)
 }
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 9e537add8..5d6ba3424 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -103,7 +103,6 @@ type Orchestrator interface {
 	GetOperationByID(ctx context.Context, id string) (*core.Operation, error)
 	GetOperations(ctx context.Context, filter database.AndFilter) ([]*core.Operation, *database.FilterResult, error)
 	GetEventByID(ctx context.Context, id string) (*core.Event, error)
-	GetEventByIDWithReference(ctx context.Context, id string) (*core.EnrichedEvent, error)
 	GetEvents(ctx context.Context, filter database.AndFilter) ([]*core.Event, *database.FilterResult, error)
 	GetEventsWithReferences(ctx context.Context, filter database.AndFilter) ([]*core.EnrichedEvent, *database.FilterResult, error)
 	GetBlockchainEventByID(ctx context.Context, id string) (*core.BlockchainEvent, error)
diff --git a/mocks/orchestratormocks/orchestrator.go b/mocks/orchestratormocks/orchestrator.go
index e2e092127..35e9c8cca 100644
--- a/mocks/orchestratormocks/orchestrator.go
+++ b/mocks/orchestratormocks/orchestrator.go
@@ -515,29 +515,6 @@ func (_m *Orchestrator) GetEventByID(ctx context.Context, id string) (*core.Even
 	return r0, r1
 }
 
-// GetEventByIDWithReference provides a mock function with given fields: ctx, id
-func (_m *Orchestrator) GetEventByIDWithReference(ctx context.Context, id string) (*core.EnrichedEvent, error) {
-	ret := _m.Called(ctx, id)
-
-	var r0 *core.EnrichedEvent
-	if rf, ok := ret.Get(0).(func(context.Context, string) *core.EnrichedEvent); ok {
-		r0 = rf(ctx, id)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*core.EnrichedEvent)
-		}
-	}
-
-	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
-}
-
 // GetEvents provides a mock function with given fields: ctx, filter
 func (_m *Orchestrator) GetEvents(ctx context.Context, filter database.AndFilter) ([]*core.Event, *database.FilterResult, error) {
 	ret := _m.Called(ctx, filter)


Test patch:
diff --git a/internal/apiserver/route_get_event_by_id_test.go b/internal/apiserver/route_get_event_by_id_test.go
index 72ef9c4d1..4a189bd7f 100644
--- a/internal/apiserver/route_get_event_by_id_test.go
+++ b/internal/apiserver/route_get_event_by_id_test.go
@@ -38,17 +38,3 @@ func TestGetEventByID(t *testing.T) {
 
 	assert.Equal(t, 200, res.Result().StatusCode)
 }
-
-func TestGetEventByIDWithReference(t *testing.T) {
-	o, r := newTestAPIServer()
-	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
-	req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/events/abcd12345?fetchreference", nil)
-	req.Header.Set("Content-Type", "application/json; charset=utf-8")
-	res := httptest.NewRecorder()
-
-	o.On("GetEventByIDWithReference", mock.Anything, "abcd12345").
-		Return(&core.EnrichedEvent{}, nil)
-	r.ServeHTTP(res, req)
-
-	assert.Equal(t, 200, res.Result().StatusCode)
-}
diff --git a/internal/apiserver/route_get_events_test.go b/internal/apiserver/route_get_events_test.go
index 400692d7d..fb605cc27 100644
--- a/internal/apiserver/route_get_events_test.go
+++ b/internal/apiserver/route_get_events_test.go
@@ -63,26 +63,3 @@ func TestGetEventsWithReferences(t *testing.T) {
 	assert.Equal(t, int64(0), resWithCount.Count)
 	assert.Equal(t, int64(10), resWithCount.Total)
 }
-
-func TestGetEventsWithFetchReference(t *testing.T) {
-	o, r := newTestAPIServer()
-	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
-	req := httptest.NewRequest("GET", "/api/v1/namespaces/mynamespace/events?fetchreference", nil)
-	req.Header.Set("Content-Type", "application/json; charset=utf-8")
-	res := httptest.NewRecorder()
-
-	var ten int64 = 10
-	o.On("GetEventsWithReferences", mock.Anything, mock.Anything).
-		Return([]*core.EnrichedEvent{}, &database.FilterResult{
-			TotalCount: &ten,
-		}, nil)
-	r.ServeHTTP(res, req)
-
-	assert.Equal(t, 200, res.Result().StatusCode)
-	var resWithCount filterResultsWithCount
-	err := json.NewDecoder(res.Body).Decode(&resWithCount)
-	assert.NoError(t, err)
-	assert.NotNil(t, resWithCount.Items)
-	assert.Equal(t, int64(0), resWithCount.Count)
-	assert.Equal(t, int64(10), resWithCount.Total)
-}
diff --git a/internal/database/postgres/postgres_test.go b/internal/database/postgres/postgres_test.go
index 60a4f7de7..d27284732 100644
--- a/internal/database/postgres/postgres_test.go
+++ b/internal/database/postgres/postgres_test.go
@@ -40,7 +40,8 @@ func TestPostgresProvider(t *testing.T) {
 
 	assert.Equal(t, "postgres", psql.Name())
 	assert.Equal(t, sq.Dollar, psql.Features().PlaceholderFormat)
-	assert.Equal(t, `LOCK TABLE "events" IN EXCLUSIVE MODE;`, psql.Features().ExclusiveTableLockSQL("events"))
+	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"))
 
 	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 e6be3a0fa..ab1c57993 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("LOCK .*").WillReturnError(fmt.Errorf("pop"))
+	mock.ExpectExec("<acquire lock ns1>").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	eventID := fftypes.NewUUID()
-	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID})
+	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID, Namespace: "ns1"})
 	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("LOCK .*").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("<acquire lock ns1>").WillReturnResult(driver.ResultNoRows)
 	mock.ExpectExec("INSERT .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	eventID := fftypes.NewUUID()
-	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID})
+	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID, Namespace: "ns1"})
 	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("LOCK .*").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("<acquire lock ns1>").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})
+	err := s.InsertEvent(context.Background(), &core.Event{ID: eventID, Namespace: "ns1"})
 	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("LOCK .*").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("<acquire lock ns1>").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("LOCK .*").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("<acquire lock ns1>").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("LOCK .*").WillReturnResult(driver.ResultNoRows)
+	mock.ExpectExec("<acquire lock ns1>").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 2331c6415..b507af5db 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.ExclusiveTableLockSQL = func(table string) string {
-		return fmt.Sprintf(`LOCK TABLE "%s" IN EXCLUSIVE MODE;`, table)
+	features.AcquireLock = func(lockName string) string {
+		return fmt.Sprintf(`<acquire lock %s>`, lockName)
 	}
 	return features
 }
diff --git a/internal/database/sqlcommon/sqlcommon_test.go b/internal/database/sqlcommon/sqlcommon_test.go
index 3c23e50b1..9fe69d94e 100644
--- a/internal/database/sqlcommon/sqlcommon_test.go
+++ b/internal/database/sqlcommon/sqlcommon_test.go
@@ -328,19 +328,6 @@ 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/data_query_test.go b/internal/orchestrator/data_query_test.go
index b30d807e3..a696f678b 100644
--- a/internal/orchestrator/data_query_test.go
+++ b/internal/orchestrator/data_query_test.go
@@ -691,49 +691,6 @@ func TestGetEventsWithReferencesEnrichFail(t *testing.T) {
 	assert.EqualError(t, err, "pop")
 }
 
-func TestGetEventWithReferenceQueryFail(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	id := fftypes.NewUUID()
-	or.mdi.On("GetEventByID", mock.Anything, "ns", mock.Anything).Return(nil, fmt.Errorf("pop"))
-	_, err := or.GetEventByIDWithReference(context.Background(), id.String())
-	assert.EqualError(t, err, "pop")
-}
-
-func TestGetEventWithReferenceParseUUIDFail(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	_, err := or.GetEventByIDWithReference(context.Background(), "1")
-	assert.Regexp(t, "FF00138", err)
-}
-
-func TestGetEventWithReferenceOK(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-
-	// Setup the IDs
-	ref1 := fftypes.NewUUID()
-	ev1 := fftypes.NewUUID()
-
-	blockchainEvent := &core.Event{
-		ID:        ev1,
-		Sequence:  10000001,
-		Reference: ref1,
-		Type:      core.EventTypeBlockchainEventReceived,
-	}
-
-	or.mem.On("EnrichEvent", mock.Anything, blockchainEvent).Return(&core.EnrichedEvent{
-		Event: *blockchainEvent,
-		BlockchainEvent: &core.BlockchainEvent{
-			ID: ref1,
-		},
-	}, nil)
-
-	or.mdi.On("GetEventByID", mock.Anything, "ns", mock.Anything).Return(blockchainEvent, nil)
-	_, err := or.GetEventByIDWithReference(context.Background(), ev1.String())
-	assert.NoError(t, err)
-}
-
 func TestGetBlockchainEventByID(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)