Prevent concurrent changes to same entity from different submissions

lib/model/query/entities.js

@@ -15,7 +15,7 @@ const { map, mergeRight, pickAll } = require('ramda');
 const { blankStringToNull, construct } = require('../../util/util');
 const { QueryOptions } = require('../../util/db');
 const { odataFilter, odataOrderBy } = require('../../data/odata-filter');
-const { odataToColumnMap, parseSubmissionXml, getDiffProp, ConflictType } = require('../../data/entity');
+const { odataToColumnMap, parseSubmissionXml, getDiffProp, ConflictType, normalizeUuid } = require('../../data/entity');
 const { isTrue } = require('../../util/http');
 const Problem = require('../../util/problem');
 const { getOrReject, runSequentially } = require('../../util/promise');
@@ -382,7 +382,7 @@ const _getFormDefActions = (oneFirst, datasetId, formDefId) => oneFirst(sql`
 
 // Main submission event processing function, which runs within a transaction
 // so any errors can be rolled back and logged as an entity processing error.
-const _processSubmissionEvent = (event, parentEvent) => async ({ Audits, Datasets, Entities, Submissions, Forms, oneFirst }) => {
+const _processSubmissionEvent = (event, parentEvent) => async ({ Audits, Datasets, Entities, Submissions, Forms, oneFirst, run }) => {
   const { submissionId, submissionDefId } = event.details;
   const forceOutOfOrderProcessing = parentEvent?.details?.force === true;
 
@@ -441,6 +441,9 @@ const _processSubmissionEvent = (event, parentEvent) => async ({ Audits, Dataset
       throw Problem.user.entityActionNotPermitted({ action, permitted: permittedActions });
   }
 
+  // Prevent concurrent changes to the entity.
+  await _lockEntity(run, normalizeUuid(entityData.system.id));
+
   let maybeEntity = null;
   // Try update before create (if both are specified)
   if (entityData.system.update === '1' || entityData.system.update === 'true')
@@ -613,7 +616,12 @@ const _get = (includeSource) => {
 // We can't use `FOR UPDATE` clause because of "Read Committed Isolation Level",
 // i.e. blocked transaction gets the row version that was at the start of the command,
 // (after lock is released by the first transaction), even if transaction with lock has updated that row.
-const _lockEntity = (exec, uuid) => exec(sql`SELECT pg_advisory_xact_lock(id) FROM entities WHERE uuid = ${uuid};`);
+const _lockEntity = (exec, uuid) => {
+  // pg_advisory_xact_lock() takes a bigint. A 16-digit hex number could exceed
+  // the bigint max, so we only use the first 15 digits of the UUID.
+  const lockId = Number.parseInt(uuid.replaceAll('-', '').slice(0, 15), 16);
+  return exec(sql`SELECT pg_advisory_xact_lock(${lockId})`);
+};
 
 const assignCurrentVersionCreator = (entity) => {
   const currentVersion = new Entity.Def(entity.aux.currentVersion, { creator: entity.aux.currentVersionCreator });