lock pre-existing rows

services/horizon/internal/db2/history/account_loader.go

@@ -204,6 +204,7 @@ func bulkGetOrCreate(ctx context.Context, q *Q, table string, fields []columnVal
 	// will not see the effects of the inserted_rows CTE because of the snapshot
 	// isolation semantics of postgres CTEs (see
 	// https://www.postgresql.org/docs/12/queries-with.html ).
+	preExistingRowsSuffix := "ORDER BY id ASC FOR KEY SHARE"
 	sql := `
 	WITH rows AS
 		(SELECT ` + strings.Join(unnestPart, ",") + `),
@@ -213,11 +214,14 @@ func bulkGetOrCreate(ctx context.Context, q *Q, table string, fields []columnVal
 		SELECT * FROM rows
 		ON CONFLICT (` + columns + `) DO NOTHING
 		RETURNING *
+	),
+	pre_existing_rows AS (
+		SELECT * FROM ` + table + ` WHERE (` + columns + `) IN
+		(SELECT * FROM rows) ` + preExistingRowsSuffix + `
 	)
 	SELECT *, true as inserted FROM inserted_rows
 	UNION ALL
-	SELECT *, false as inserted FROM ` + table + ` WHERE (` + columns + `) IN 
-	(SELECT * FROM rows)`
+	SELECT *, false as inserted FROM pre_existing_rows`
 
 	return q.SelectRaw(
 		ctx,