Lock acs_snapshot_data while writing

[ci]

Signed-off-by: Robert Autenrieth <robert.autenrieth@digitalasset.com>

Author: rautenrieth-da

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala

@@ -25,7 +25,7 @@ import com.digitalasset.canton.resource.DbStorage.Implicits.BuilderChain.toSQLAc
 import com.digitalasset.canton.topology.PartyId
 import com.digitalasset.canton.tracing.TraceContext
 import org.lfdecentralizedtrust.splice.store.events.SpliceCreatedEvent
-import slick.dbio.DBIOAction
+import slick.dbio.{DBIOAction, Effect, NoStream}
 import slick.jdbc.canton.ActionBasedSQLInterpolation.Implicits.actionBasedSQLInterpolationCanton
 import slick.jdbc.{GetResult, JdbcProfile}
 
@@ -158,12 +158,32 @@ class AcsSnapshotStore(
         join creates_to_insert on inserted_rows.create_id = creates_to_insert.row_id
         having min(inserted_rows.row_id) is not null;
              """).toActionBuilder.asUpdate
-      storage.update(statement, "insertNewSnapshot")
+      storage.update(withExclusiveSnapshotDataLock(statement), "insertNewSnapshot")
     }.andThen { _ =>
       AcsSnapshotStore.PreventConcurrentSnapshotsSemaphore.release()
     }
   }
 
+  /** Wraps the given action in a transaction that holds an exclusive lock on the acs_snapshot_data table.
+    *
+    *  Note: The acs_snapshot_data table must not have interleaved rows from two different acs snapshots.
+    *  In rare cases, it can happen that the application crashes while writing a snapshot, then
+    *  restarts and starts writing a different snapshot while the previous statement is still running.
+    *  The exclusive lock prevents this.
+    *  Once obtained, a lock is held for the remainder of the current transaction.
+    *  In case the application crashes while holding the lock, the server should close the connection
+    *  and abort the transaction as soon as it detects a disconnect.
+    *  See [[com.digitalasset.canton.platform.store.backend.postgresql.PostgresDataSourceConfig]] for our connection keepalive settings.
+    *  With default settings, the server should detect a dead connection within ~15sec.
+    */
+  private def withExclusiveSnapshotDataLock[T, E <: Effect](
+      action: DBIOAction[T, NoStream, E]
+  ): DBIOAction[T, NoStream, E & Effect.Write & Effect.Transactional] =
+    (for {
+      _ <- sqlu"LOCK TABLE acs_snapshot_data IN EXCLUSIVE MODE"
+      result <- action
+    } yield result).transactionally
+
   def deleteSnapshot(
       snapshot: AcsSnapshot
   )(implicit