Fix ShareLock deadlock (#469)

storage/encrypting_repository.go

@@ -136,6 +136,19 @@ func (er *encryptingRepository) Get(ctx context.Context, objectType types.Object
 	return obj, nil
 }
 
+func (er *encryptingRepository) GetForUpdate(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.Object, error) {
+	obj, err := er.repository.Get(ctx, objectType, criteria...)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := er.decrypt(ctx, obj); err != nil {
+		return nil, err
+	}
+
+	return obj, nil
+}
+
 func (er *encryptingRepository) List(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.ObjectList, error) {
 	objList, err := er.repository.List(ctx, objectType, criteria...)
 	if err != nil {

storage/integrity_repository.go

@@ -74,6 +74,17 @@ func (cr *integrityRepository) Get(ctx context.Context, objectType types.ObjectT
 	return obj, nil
 }
 
+func (cr *integrityRepository) GetForUpdate(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.Object, error) {
+	obj, err := cr.repository.Get(ctx, objectType, criteria...)
+	if err != nil {
+		return nil, err
+	}
+	if err := cr.validateIntegrity(obj); err != nil {
+		return nil, err
+	}
+	return obj, nil
+}
+
 func (cr *integrityRepository) List(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.ObjectList, error) {
 	objectList, err := cr.repository.List(ctx, objectType, criteria...)
 	if err != nil {

storage/interceptable_repository.go

@@ -159,6 +159,15 @@ func (ir *queryScopedInterceptableRepository) Get(ctx context.Context, objectTyp
 	return object, nil
 }
 
+func (ir *queryScopedInterceptableRepository) GetForUpdate(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.Object, error) {
+	object, err := ir.repositoryInTransaction.Get(ctx, objectType, criteria...)
+	if err != nil {
+		return nil, err
+	}
+
+	return object, nil
+}
+
 func (ir *queryScopedInterceptableRepository) List(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.ObjectList, error) {
 	objectList, err := ir.repositoryInTransaction.List(ctx, objectType, criteria...)
 	if err != nil {
@@ -251,7 +260,7 @@ func (ir *queryScopedInterceptableRepository) Update(ctx context.Context, obj ty
 	// postgres storage implementation also locks the retrieved row for update
 	objectType := obj.GetType()
 	byID := query.ByField(query.EqualsOperator, "id", obj.GetID())
-	oldObj, err := ir.Get(ctx, objectType, byID)
+	oldObj, err := ir.GetForUpdate(ctx, objectType, byID)
 	if err != nil {
 		return nil, err
 	}
@@ -442,6 +451,15 @@ func (itr *InterceptableTransactionalRepository) Get(ctx context.Context, object
 	return object, nil
 }
 
+func (itr *InterceptableTransactionalRepository) GetForUpdate(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.Object, error) {
+	object, err := itr.RawRepository.Get(ctx, objectType, criteria...)
+	if err != nil {
+		return nil, err
+	}
+
+	return object, nil
+}
+
 func (itr *InterceptableTransactionalRepository) List(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.ObjectList, error) {
 	objectList, err := itr.RawRepository.List(ctx, objectType, criteria...)
 	if err != nil {

storage/interceptable_repository_test.go

@@ -182,6 +182,13 @@ var _ = Describe("Interceptable TransactionalRepository", func() {
 			},
 		}, nil)
 
+		fakeStorage.GetForUpdateReturns(&types.ServiceBroker{
+			Base: types.Base{
+				UpdatedAt: updateTime,
+				Ready:     true,
+			},
+		}, nil)
+
 		interceptableRepository = storage.NewInterceptableTransactionalRepository(fakeStorage)
 
 		orderNone := storage.InterceptorOrder{

storage/interfaces.go

@@ -170,6 +170,9 @@ type Repository interface {
 	// Get retrieves an object using the provided id from SM DB
 	Get(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.Object, error)
 
+	// GetForUpdate retrieves an object using the provided id from SM DB while also locking the retrieved rows
+	GetForUpdate(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.Object, error)
+
 	// List retrieves all object from SM DB
 	List(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.ObjectList, error)
 

storage/postgres/query_builder.go

@@ -26,7 +26,7 @@ FROM {{.ENTITY_TABLE}}
 		ON {{.ENTITY_TABLE}}.{{.PRIMARY_KEY}} = {{.LABELS_TABLE}}.{{.REF_COLUMN}}
 	{{end}}
 {{.WHERE}}
-{{.FOR_SHARE_OF}}
+{{.FOR_UPDATE_OF}}
 {{.LIMIT}};`
 
 const SelectQueryTemplate = `
@@ -57,7 +57,7 @@ WHERE {{.ENTITY_TABLE}}.paging_sequence IN
 	(SELECT matching_resources.paging_sequence FROM matching_resources)
 {{end}}
 {{.ORDER_BY}}
-{{.FOR_SHARE_OF}};`
+{{.FOR_UPDATE_OF}};`
 
 const DeleteQueryTemplate = `
 DELETE FROM {{.ENTITY_TABLE}}
@@ -199,7 +199,7 @@ func (pq *pgQuery) resolveQueryTemplate(ctx context.Context, template string) (s
 		"REF_COLUMN":        pq.labelEntity.ReferenceColumn(),
 		"JOIN":              pq.joinSQL(),
 		"WHERE":             pq.whereSQL(),
-		"FOR_SHARE_OF":      pq.lockSQL(),
+		"FOR_UPDATE_OF":     pq.lockSQL(),
 		"ORDER_BY":          pq.orderBySQL(),
 		"ORDER_BY_SEQUENCE": pq.orderBySequenceSQL(),
 		"LIMIT":             pq.limitSQL(),
@@ -330,7 +330,7 @@ func (pq *pgQuery) lockSQL() string {
 		// Lock the rows if we are in transaction so that update operations on those rows can rely on unchanged data
 		// This allows us to handle concurrent updates on the same rows by executing them sequentially as
 		// before updating we have to anyway select the rows and can therefore lock them
-		return fmt.Sprintf("FOR SHARE OF %s", pq.entityTableName)
+		return fmt.Sprintf("FOR UPDATE OF %s", pq.entityTableName)
 	}
 	return ""
 }

storage/postgres/storage.go

@@ -212,13 +212,33 @@ func (ps *Storage) Get(ctx context.Context, objectType types.ObjectType, criteri
 	return result.ItemAt(0), nil
 }
 
+func (ps *Storage) GetForUpdate(ctx context.Context, objectType types.ObjectType, criteria ...query.Criterion) (types.Object, error) {
+	result, err := ps.list(ctx, objectType, true, criteria...)
+	if err != nil {
+		return nil, err
+	}
+	if result.Len() == 0 {
+		return nil, util.ErrNotFoundInStorage
+	}
+	return result.ItemAt(0), nil
+}
+
 func (ps *Storage) List(ctx context.Context, objType types.ObjectType, criteria ...query.Criterion) (types.ObjectList, error) {
+	return ps.list(ctx, objType, false, criteria...)
+}
+
+func (ps *Storage) list(ctx context.Context, objType types.ObjectType, forUpdate bool, criteria ...query.Criterion) (types.ObjectList, error) {
 	entity, err := ps.scheme.provide(objType)
 	if err != nil {
 		return nil, err
 	}
 
-	rows, err := ps.queryBuilder.NewQuery(entity).WithCriteria(criteria...).WithLock().List(ctx)
+	queryBuilder := ps.queryBuilder.NewQuery(entity).WithCriteria(criteria...)
+	if forUpdate {
+		queryBuilder = queryBuilder.WithLock()
+	}
+
+	rows, err := queryBuilder.List(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -242,7 +262,7 @@ func (ps *Storage) Count(ctx context.Context, objType types.ObjectType, criteria
 	if err != nil {
 		return 0, err
 	}
-	return ps.queryBuilder.NewQuery(entity).WithCriteria(criteria...).WithLock().Count(ctx)
+	return ps.queryBuilder.NewQuery(entity).WithCriteria(criteria...).Count(ctx)
 }
 
 func (ps *Storage) DeleteReturning(ctx context.Context, objType types.ObjectType, criteria ...query.Criterion) (types.ObjectList, error) {