From ecb2d53c73569cf79f5b17c2d2cb098bc171180e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Thu, 6 Aug 2026 08:45:12 +0000 Subject: [PATCH 01/15] tentative SQL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../stable/V073__acs_snapshot_table_name.sql | 51 +++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql index 5e0aaad57e..4813f8eec2 100644 --- a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql +++ b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql @@ -12,3 +12,54 @@ alter table acs_snapshot ((first_row_id is null and last_row_id is null and data_table_name is not null) or -- legacy table (first_row_id is not null and last_row_id is not null and data_table_name is null)); + +-- Same as acs_incremental_snapshot_data_next, but includes the create_arguments. +create table acs_incremental_snapshot_data_nextv2 +( + -- In production, we have a separate table for each scan instance so this + -- column will be the same for all rows. However, in tests we can have multiple + -- scan instances writing to the same table, so we need to include it. + snapshot_id bigint not null references acs_incremental_snapshot (snapshot_id), + + contract_id text not null, + + -- Contract data included to avoid joining with UpdateHistory tables + -- during the expensive operation of saving incremental ACS snapshots. + create_arguments jsonb not null, + created_at bigint not null, + unlocked_amulet_balance numeric not null, -- zero for non-amulet contracts + locked_amulet_balance numeric not null, -- zero for non-amulet contracts + template_id text not null, + stakeholders text[] not null +); + +-- Template table for acs_snapshot_creates_. +-- This allows the code to just CREATE TABLE LIKE acs_snapshot_creates_template or acs_snapshot_stakeholders_template. +-- Design decision: we don't have a single table per (contract_id, stakeholder) in order to avoid duplicating the create_arguments. +create table acs_snapshot_creates_template +( + contract_id text primary key, + create_arguments jsonb not null, + -- We might be able to derive signatories and observers from stakeholders, but this is easier + signatories text[] not null, + observers text[] not null, + -- plus the Amulet-specific balance columns currently computed in the working table + unlocked_amulet_balance numeric, + locked_amulet_balance numeric +); + +create table acs_snapshot_stakeholders_template +( + -- Important: insertion order should happen by (created_at, contract_id) for: + -- 1) backwards compatibility + -- 2) deterministic ordering across SVs + row_id bigint generated by default as identity primary key, + stakeholder text not null, + template_id text not null, + contract_id text not null +); + +-- Necessary indexes: +-- 1) (stakeholder, template_id, row_id) for where stakeholder=? and template_id=? order by row_id +-- 2) (stakeholder, row_id) for where stakeholder=? order by row_id +-- In both cases `include (contract_id)` allows index-only scans to then merge with acs_snapshot_creates_template From e9f0e0b6c49cd787545248d7b059ff3f6b27946e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Mon, 10 Aug 2026 16:14:52 +0000 Subject: [PATCH 02/15] support ACS snapshots per table MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../stable/V073__acs_snapshot_table_name.sql | 57 +- .../scan/automation/AcsSnapshotTrigger.scala | 6 +- .../scan/config/ScanStorageConfig.scala | 2 + .../splice/scan/store/AcsSnapshotStore.scala | 509 ++++++++++++++---- .../automation/AcsSnapshotTriggerTest.scala | 1 + .../scan/config/ScanStorageConfigTest.scala | 37 +- ...shotBulkStorageCommitFromStagingTest.scala | 1 + ...sSnapshotBulkStorageWriterFromDbTest.scala | 1 + .../bulk/UpdateHistoryBulkStorageTest.scala | 1 + 9 files changed, 456 insertions(+), 159 deletions(-) diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql index 4813f8eec2..54d57c3764 100644 --- a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql +++ b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql @@ -13,41 +13,62 @@ alter table acs_snapshot -- legacy table (first_row_id is not null and last_row_id is not null and data_table_name is null)); --- Same as acs_incremental_snapshot_data_next, but includes the create_arguments. -create table acs_incremental_snapshot_data_nextv2 +-- TODO: template ids can be interned already + +-- Same as acs_incremental_snapshot_data_next, +-- but includes ALL update_history_creates data necessary to build a CreatedEvent. +create table acs_incremental_snapshot_data_next_v2 ( -- In production, we have a separate table for each scan instance so this -- column will be the same for all rows. However, in tests we can have multiple -- scan instances writing to the same table, so we need to include it. snapshot_id bigint not null references acs_incremental_snapshot (snapshot_id), - contract_id text not null, - -- Contract data included to avoid joining with UpdateHistory tables - -- during the expensive operation of saving incremental ACS snapshots. - create_arguments jsonb not null, - created_at bigint not null, - unlocked_amulet_balance numeric not null, -- zero for non-amulet contracts - locked_amulet_balance numeric not null, -- zero for non-amulet contracts - template_id text not null, - stakeholders text[] not null + -- All the data necessary to reconstruct a created event + create_arguments jsonb not null, + event_id text not null, + record_time bigint not null, + template_id_package_id text not null, -- the package_name is already included as part of the template_id + contract_key text null, + created_at bigint not null, + signatories text[] not null, + observers text[] not null, + -- plus the Amulet-specific balance columns currently computed in the working table + unlocked_amulet_balance numeric, + locked_amulet_balance numeric ); +-- Needed for fast insert/remove by contract id +alter table acs_incremental_snapshot_data_next_v2 + add constraint acs_incremental_snapshot_data_next_v2_pk + primary key (snapshot_id, contract_id); + +-- Needed because ACS snapshots are ordered by creation time +create index acs_incremental_snapshot_data_next_v2_ca_ci + on acs_incremental_snapshot_data_next_v2 (snapshot_id, created_at, contract_id); + -- Template table for acs_snapshot_creates_. -- This allows the code to just CREATE TABLE LIKE acs_snapshot_creates_template or acs_snapshot_stakeholders_template. -- Design decision: we don't have a single table per (contract_id, stakeholder) in order to avoid duplicating the create_arguments. create table acs_snapshot_creates_template ( - contract_id text primary key, - create_arguments jsonb not null, - -- We might be able to derive signatories and observers from stakeholders, but this is easier - signatories text[] not null, - observers text[] not null, + contract_id text primary key, + -- All the data necessary to reconstruct a created event + create_arguments jsonb not null, + event_id text not null, + record_time bigint not null, + template_id_package_id text not null, -- the package_name is already included as part of the template_id + contract_key text null, + created_at bigint not null, + signatories text[] not null, + observers text[] not null, -- plus the Amulet-specific balance columns currently computed in the working table unlocked_amulet_balance numeric, locked_amulet_balance numeric ); +-- TODO: revisit this name because this does more than stakehodlering create table acs_snapshot_stakeholders_template ( -- Important: insertion order should happen by (created_at, contract_id) for: @@ -60,6 +81,6 @@ create table acs_snapshot_stakeholders_template ); -- Necessary indexes: --- 1) (stakeholder, template_id, row_id) for where stakeholder=? and template_id=? order by row_id --- 2) (stakeholder, row_id) for where stakeholder=? order by row_id +-- 1) (stakeholder, template_id, row_id) for where stakeholder=? and template_id=? (and row_id > $after) order by row_id +-- 2) (stakeholder, row_id) for where stakeholder=? (and row_id > $after) order by row_id -- In both cases `include (contract_id)` allows index-only scans to then merge with acs_snapshot_creates_template diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTrigger.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTrigger.scala index 31ed151b0f..124046d3d9 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTrigger.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTrigger.scala @@ -35,7 +35,11 @@ class AcsSnapshotTrigger( ) extends AcsSnapshotTriggerBase(store, updateHistory, context) { override val snapshotTable: IncrementalAcsSnapshotTable = - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + if (storageConfig.perAcsSnapshotTablesEnabled) { + AcsSnapshotStore.IncrementalAcsSnapshotTable.NextV2 + } else { + AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + } override val snapshotMetrics: AcsSnapshotsMetrics = new HistoryMetrics(context.metricsFactory)( MetricsContext.Empty diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfig.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfig.scala index 9eaa7a0e10..bef4b32d73 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfig.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfig.scala @@ -20,6 +20,7 @@ import java.time.temporal.{ChronoField, ChronoUnit} case class ScanStorageConfig( dbAcsSnapshotPeriodHours: Int, // Period between two consecutive acs snapshots to be computed and stored in the DB + perAcsSnapshotTablesEnabled: Boolean, // Whether each ACS snapshot should be stored in its own table bulkAcsSnapshotPeriodHours: Int, // Period between two consecutive acs snapshots to be dumped to bulk storage (currently must be <=24 hr, and a multiple of dbAcsSnapshotPeriodHours) bulkDbReadChunkSize: Int, // Chunk size to read from the DB for copying to bulk storage bulkZstdFrameSize: Long, // Size of each zstd frame. In prod, must be >= 5 MB as each frame is written as a part in multi-part upload, which are enforced by most s3 implementations to be >= 5MB each @@ -139,6 +140,7 @@ case class ScanStorageConfig( object ScanStorageConfigs { val scanStorageConfigV1 = ScanStorageConfig( dbAcsSnapshotPeriodHours = 3, + perAcsSnapshotTablesEnabled = false, bulkAcsSnapshotPeriodHours = 24, bulkDbReadChunkSize = 1000, bulkZstdFrameSize = 12L * 1024 * 1024, diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index 4a967cd727..6f8eabbd7d 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -4,40 +4,38 @@ package org.lfdecentralizedtrust.splice.scan.store import cats.data.NonEmptyVector -import com.daml.ledger.javaapi.data.CreatedEvent -import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{Amulet, LockedAmulet} -import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.{ - AcsSnapshot, - FailedToAcquireLockException, - IncrementalAcsSnapshot, - IncrementalAcsSnapshotTable, - LegacyAcsSnapshot, - PerTableAcsSnapshot, - QueryAcsSnapshotResult, - amuletQualifiedName, - lockedAmuletQualifiedName, -} -import org.lfdecentralizedtrust.splice.store.UpdateHistory.SelectFromCreateEvents -import org.lfdecentralizedtrust.splice.store.{HardLimit, Limit, LimitHelpers, UpdateHistory} -import org.lfdecentralizedtrust.splice.store.db.{AcsJdbcTypes, AcsQueries, AdvisoryLockIds} -import org.lfdecentralizedtrust.splice.util.{Contract, HoldingsSummary, PackageQualifiedName} +import com.daml.ledger.javaapi.data.{CreatedEvent, Identifier} import com.digitalasset.canton.data.CantonTimestamp import com.digitalasset.canton.lifecycle.{CloseContext, FutureUnlessShutdown} import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting} import com.digitalasset.canton.logging.{NamedLoggerFactory, NamedLogging} import com.digitalasset.canton.resource.DbStorage import com.digitalasset.canton.resource.DbStorage.Implicits.BuilderChain.toSQLActionBuilderChain +import com.digitalasset.canton.resource.DbStorage.SQLActionBuilderChain import com.digitalasset.canton.topology.PartyId import com.digitalasset.canton.tracing.TraceContext -import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.QueryParts.* +import com.google.protobuf.ByteString +import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{Amulet, LockedAmulet} +import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.* +import org.lfdecentralizedtrust.splice.store.UpdateHistory.SelectFromCreateEvents +import org.lfdecentralizedtrust.splice.store.db.{AcsJdbcTypes, AcsQueries, AdvisoryLockIds} import org.lfdecentralizedtrust.splice.store.events.SpliceCreatedEvent -import slick.dbio.{DBIOAction, Effect, NoStream} +import org.lfdecentralizedtrust.splice.store.{HardLimit, Limit, LimitHelpers, UpdateHistory} +import org.lfdecentralizedtrust.splice.util.{ + Contract, + EventId, + HoldingsSummary, + PackageQualifiedName, + ValueJsonCodecProtobuf as ProtobufCodec, +} +import slick.dbio.{DBIO, DBIOAction, Effect, NoStream} import slick.jdbc.canton.ActionBasedSQLInterpolation.Implicits.actionBasedSQLInterpolationCanton -import slick.jdbc.canton.SQLActionBuilder import slick.jdbc.{GetResult, JdbcProfile} import java.util.concurrent.Semaphore import scala.concurrent.{ExecutionContext, Future} +import scala.jdk.CollectionConverters.* +import scala.jdk.OptionConverters.* class AcsSnapshotStore( storage: DbStorage, @@ -125,7 +123,9 @@ class AcsSnapshotStore( case Some(LegacyAcsSnapshot(_, _, _, firstRowId, lastRowId, _, _)) => sql"where snapshot.row_id >= $firstRowId and snapshot.row_id <= $lastRowId" case Some(_: PerTableAcsSnapshot) => - throw io.grpc.Status.UNIMPLEMENTED.withDescription("TODO #6264").asRuntimeException() + throw io.grpc.Status.UNIMPLEMENTED + .withDescription("This should not be called if we have PerTableAcsSnapshot enabled.") + .asRuntimeException() case None => sql"where false" } @@ -252,8 +252,16 @@ class AcsSnapshotStore( sqlu"""delete from acs_snapshot where snapshot_record_time = ${snapshot.snapshotRecordTime}""", sqlu"""delete from acs_snapshot_data where row_id between ${snapshot.firstRowId} and ${snapshot.lastRowId}""", ) - case _: PerTableAcsSnapshot => - throw io.grpc.Status.UNIMPLEMENTED.withDescription("TODO #6263").asRuntimeException() + case table: PerTableAcsSnapshot => + DBIOAction.seq( + sqlu"""delete from acs_snapshot where snapshot_record_time = ${snapshot.snapshotRecordTime}""", + sqlu"""drop table #${AcsTableDDL.acsSnapshotCreatesTableName( + table.snapshotRecordTime + )};""", + sqlu"""drop table #${AcsTableDDL.acsSnapshotStakeholdersTableName( + table.snapshotRecordTime + )};""", + ) } storage.update(statement.transactionally, "deleteSnapshot") } @@ -267,7 +275,7 @@ class AcsSnapshotStore( templates: Seq[PackageQualifiedName], )(implicit tc: TraceContext): Future[QueryAcsSnapshotResult] = { for { - snapshotTrait <- storage + snapshot <- storage .querySingle( sql"""select snapshot_record_time, migration_id, history_id, first_row_id, last_row_id, unlocked_amulet_balance, locked_amulet_balance, data_table_name from acs_snapshot @@ -286,13 +294,127 @@ class AcsSnapshotStore( .asRuntimeException() ) ) - snapshot <- (snapshotTrait match { - case snapshot: LegacyAcsSnapshot => Future.successful(snapshot) - case _: PerTableAcsSnapshot => - Future.failed( - io.grpc.Status.UNIMPLEMENTED.withDescription("TODO #6264").asRuntimeException() + events <- snapshot match { + case snapshot: LegacyAcsSnapshot => + queryLegacyTable(snapshot, after, limit, partyIds, templates) + case ownTable: PerTableAcsSnapshot => + querySnapshotInOwnTable(ownTable, after, limit, partyIds, templates) + } + } yield { + val eventsInPage = + applyLimitOrFail("queryAcsSnapshot", limit, events.map(_._2)) + val afterToken = if (eventsInPage.size == limit.limit) events.lastOption.map(_._1) else None + QueryAcsSnapshotResult( + migrationId = snapshot.migrationId, + snapshotRecordTime = snapshot.snapshotRecordTime, + createdEventsInPage = eventsInPage, + afterToken = afterToken, + ) + } + } + + private def querySnapshotInOwnTable( + snapshot: PerTableAcsSnapshot, + after: Option[Long], + limit: Limit, + partyIds: Seq[PartyId], + templates: Seq[PackageQualifiedName], + )(implicit tc: TraceContext): Future[Vector[(Long, SpliceCreatedEvent)]] = { + val createsTableName = + AcsTableDDL.acsSnapshotCreatesTableName(snapshot.snapshotRecordTime) + val stakeholdersTableName = + AcsTableDDL.acsSnapshotStakeholdersTableName(snapshot.snapshotRecordTime) + val afterFilter = after.fold(sql"")(after => sql" and s.row_id > $after") + storage + .query( + (sql""" + select + s.row_id, + event_id, + record_time, + template_id_package_id, + template_id, + contract_id, + create_arguments, + contract_key, + signatories, + observers, + created_at + from #$stakeholdersTableName s + join #$createsTableName c on s.contract_id = c.contract_id + where """ ++ stakeholdersFilter(partyIds) ++ + templatesFilter(templates) ++ + afterFilter ++ sql""" + order by s.row_id + limit ${sqlLimit(limit)} + """).toActionBuilder.as[ + ( + Long, + String, + CantonTimestamp, + String, + PackageQualifiedName, + String, + String, + Option[String], + Seq[String], + Seq[String], + CantonTimestamp, + ) + ], + "querySnapshotInOwnTable", + ) + .map(_.map { + case ( + rowId, + eventId, + recordTime, + packageId, + templateIdPackageQualifiedName, + contractId, + createArguments, + contractKey, + signatories, + observers, + createdAt, + ) => + rowId -> SpliceCreatedEvent( + eventId = eventId, + recordTime = recordTime, + new CreatedEvent( + /*witnessParties = */ java.util.Collections.emptyList(), + /*offset = */ 0, // not populated + /*nodeId = */ EventId.nodeIdFromEventId(eventId), + /*templateId = */ new Identifier( + packageId, + templateIdPackageQualifiedName.qualifiedName.moduleName, + templateIdPackageQualifiedName.qualifiedName.entityName, + ), + /* packageName = */ templateIdPackageQualifiedName.packageName, + /*contractId = */ contractId, + /*arguments = */ ProtobufCodec.deserializeValue(createArguments).asRecord().get(), + /*createdEventBlob = */ ByteString.EMPTY, + /*interfaceViews = */ java.util.Collections.emptyMap(), + /*failedInterfaceViews = */ java.util.Collections.emptyMap(), + /*contractKey = */ contractKey.map(ProtobufCodec.deserializeValue).toJava, + /*signatories = */ signatories.asJava, + /*observers = */ observers.asJava, + /*createdAt = */ createdAt.toInstant, + /*acsDelta = */ false, + /*representativePackageId = */ packageId, + ), ) }) + } + + private def queryLegacyTable( + snapshot: LegacyAcsSnapshot, + after: Option[Long], + limit: Limit, + partyIds: Seq[PartyId], + templates: Seq[PackageQualifiedName], + )(implicit tc: TraceContext): Future[Vector[(Long, SpliceCreatedEvent)]] = { + for { begin <- after match { case Some(value) if value < snapshot.firstRowId || value > snapshot.lastRowId => Future.failed( @@ -306,26 +428,6 @@ class AcsSnapshotStore( case None => Future.successful(snapshot.firstRowId) } end = snapshot.lastRowId - partyIdsFilter = partyIds match { - case Nil => - // This expression is always true (scan only processes data where the DSO is stakeholder). - // It is included to make sure the query plan uses the right index (acs_snapshot_data_all_filters) - sql"and stakeholder = ${dsoParty}" - case partyIds => - (sql" and " ++ inClause("stakeholder", partyIds)).toActionBuilder - } - templatesFilter = templates match { - case Nil => sql"" - case _ => - (sql" and " ++ inClause( - "template_id", - templates.map(t => - lengthLimited( - s"${t.packageName}:${t.qualifiedName.moduleName}:${t.qualifiedName.entityName}" - ) - ), - )).toActionBuilder - } events <- storage .query( (sql""" @@ -333,7 +435,7 @@ class AcsSnapshotStore( select create_id, max(row_id) as row_id from acs_snapshot_data where row_id between $begin and $end - """ ++ partyIdsFilter ++ templatesFilter ++ sql""" + """ ++ stakeholdersFilter(partyIds) ++ templatesFilter(templates) ++ sql""" group by create_id order by row_id asc -- this CTE already will contain all snapshot rows (filtered by party id and template, if necessary). @@ -364,17 +466,29 @@ class AcsSnapshotStore( .as[(Long, SelectFromCreateEvents)], "queryAcsSnapshot.getCreatedEvents", ) - } yield { - val eventsInPage = - applyLimitOrFail("queryAcsSnapshot", limit, events.map(_._2.toCreatedEvent)) - val afterToken = if (eventsInPage.size == limit.limit) events.lastOption.map(_._1) else None - QueryAcsSnapshotResult( - migrationId = migrationId, - snapshotRecordTime = snapshot.snapshotRecordTime, - createdEventsInPage = eventsInPage, - afterToken = afterToken, - ) - } + } yield events.map { case (rowId, select) => rowId -> select.toCreatedEvent } + } + + private def stakeholdersFilter(partyIds: Seq[PartyId]) = partyIds match { + case Nil => + // This expression is always true (scan only processes data where the DSO is stakeholder). + // It is included to make sure the query plan uses the right index (acs_snapshot_data_all_filters) + sql" stakeholder = $dsoParty" + case partyIds => + inClause("stakeholder", partyIds) + } + + private def templatesFilter(templates: Seq[PackageQualifiedName]) = templates match { + case Nil => sql"" + case _ => + (sql" and " ++ inClause( + "template_id", + templates.map(t => + lengthLimited( + s"${t.packageName}:${t.qualifiedName.moduleName}:${t.qualifiedName.entityName}" + ) + ), + )).toActionBuilder } def getHoldingsState( @@ -521,7 +635,13 @@ class AcsSnapshotStore( val initializeFrom = initializeFromT match { case legacy: LegacyAcsSnapshot => legacy case _: PerTableAcsSnapshot => - throw io.grpc.Status.UNIMPLEMENTED.withDescription("TODO #6263").asRuntimeException() + // If we enable PerTableAcsSnapshots, we will necessarily initialize from a LegacyAcsSnapshot, + // never from a PerTableAcsSnapshot. Then initializeIncrementalSnapshot will never be called again. + throw io.grpc.Status.FAILED_PRECONDITION + .withDescription( + "BUG: This shouldn't be called: we shouldn't be initializing from a PerTableAcsSnapshot." + ) + .asRuntimeException() } assert(targetRecordTime.isAfter(initializeFrom.snapshotRecordTime)) val statement = for { @@ -544,11 +664,11 @@ class AcsSnapshotStore( """.as[Long].head insertedRows <- (sql""" insert into #${table.tableName} ( - """ ++ copyFromUpdateHistoryTargetColumns ++ sql""", + """ ++ table.copyFromUpdateHistoryTargetColumns ++ sql""", snapshot_id ) select - """ ++ copyFromUpdateHistorySourceColumns ++ sql""", + """ ++ table.copyFromUpdateHistorySourceColumns ++ sql""", $snapshotId from acs_snapshot_data d join update_history_creates c on d.create_id=c.row_id @@ -612,11 +732,11 @@ class AcsSnapshotStore( insertedRows <- (sql""" insert into #${table.tableName} ( - """ ++ copyFromUpdateHistoryTargetColumns ++ sql""", + """ ++ table.copyFromUpdateHistoryTargetColumns ++ sql""", snapshot_id ) select - """ ++ copyFromUpdateHistorySourceColumns ++ sql""", + """ ++ table.copyFromUpdateHistorySourceColumns ++ sql""", $snapshotId from update_history_creates c where history_id = $historyId @@ -656,7 +776,110 @@ class AcsSnapshotStore( assert(snapshot.tableName == table.tableName) assert(snapshot.historyId == historyId) assert(snapshot.recordTime == snapshot.targetRecordTime) - val statement = for { + // TODO: split into insert into legacy table and insert into new table + create indexes + val statement: DBIO[Int] = table match { + case IncrementalAcsSnapshotTable.NextV2 => + saveV2IncrementalSnapshot(table, snapshot, nextSnapshotTargetRecordTime)(tc) + case IncrementalAcsSnapshotTable.Next | IncrementalAcsSnapshotTable.Backfill => + saveLegacyIncrementalSnapshotStatement(table, snapshot, nextSnapshotTargetRecordTime) + } + storage.queryAndUpdate( + withExclusiveSnapshotDataLock( + withIncrementalSnapshotIdempotencyCheck( + table, + statement, + Some(snapshot), + ) + ), + "saveIncrementalSnapshot", + ) + } + + private def saveV2IncrementalSnapshot( + table: IncrementalAcsSnapshotTable, + snapshot: IncrementalAcsSnapshot, + nextSnapshotTargetRecordTime: CantonTimestamp, + )(implicit tc: TraceContext) = { + val createsTableName = + AcsTableDDL.acsSnapshotCreatesTableName(snapshot.targetRecordTime) + val stakeholdersTableName = + AcsTableDDL.acsSnapshotStakeholdersTableName(snapshot.targetRecordTime) + + for { + _ <- sqlu"create table #$createsTableName like acs_snapshot_creates_template" + _ <- sqlu"create table #$stakeholdersTableName like acs_snapshot_stakeholders_template" + copiedCreateRows <- (sql""" + insert into #$createsTableName (contract_id, create_arguments, event_id, record_time, template_id_package_id, contract_key, created_at, signatories, observers, unlocked_amulet_balance, locked_amulet_balance) + select s.contract_id, s.create_arguments, s.event-id, s.record_time, s.template_id_package_id, s.contract_key, s.creataed_at, s.signatories, s.observers, """ ++ IncrementalAcsSnapshotTable.QueryParts + .unlockedAmuletBalance() ++ sql", " ++ IncrementalAcsSnapshotTable.QueryParts + .lockedAmuletBalance() ++ sql""" + from #${table.tableName} s + where s.snapshot_id = ${snapshot.snapshotId} + -- ensure consistent ordering across SVs + order by created_at, contract_id + """).toActionBuilder.asUpdate + copiedStakeholderRows <- sqlu""" + insert into #${stakeholdersTableName} (stakeholder, template_id, contract_id) + select stakeholder, s.template_id, contract_id + from #${table.tableName} s + cross join unnest(array_cat(s.signatories, s.observers)) as stakeholder + where s.snapshot_id = ${snapshot.snapshotId} + order by created_at, contract_id + """ + + (unlocked_amulet_balance, locked_amulet_balance) <- sql""" + select + sum(s.unlocked_amulet_balance) AS unlocked_amulet_balance, + sum(s.locked_amulet_balance) AS locked_amulet_balance + from #${table.tableName} s + where snapshot_id = ${snapshot.snapshotId} + """.as[(BigDecimal, BigDecimal)].head + + _ <- sqlu""" + insert into acs_snapshot ( + snapshot_record_time, + migration_id, + history_id, + first_row_id, + last_row_id, + unlocked_amulet_balance, + locked_amulet_balance, + table_name + ) + values ( + ${snapshot.recordTime}, + ${snapshot.migrationId}, + ${snapshot.historyId}, + null, + null, + ${unlocked_amulet_balance}, + ${locked_amulet_balance}, + ${createsTableName} + ) + """ + + _ <- sqlu""" + update acs_incremental_snapshot + set + target_record_time = ${nextSnapshotTargetRecordTime} + where snapshot_id = ${snapshot.snapshotId} + """ + } yield { + logger.debug( + s"Saved incremental snapshot ${snapshot.snapshotId} at ${snapshot.recordTime} with $copiedCreateRows create rows and $copiedStakeholderRows stakeholder rows." + + s" Next snapshot target record time: $nextSnapshotTargetRecordTime" + ) + // This doesn't make much sense anymore + copiedCreateRows + } + } + + private def saveLegacyIncrementalSnapshotStatement( + table: IncrementalAcsSnapshotTable, + snapshot: IncrementalAcsSnapshot, + nextSnapshotTargetRecordTime: CantonTimestamp, + )(implicit tc: TraceContext): DBIOAction[Int, NoStream, Effect.Read & Effect.Write] = { + for { // Note: Only one client can write to acs_snapshot_data at a time, enforced via advisory locks. // We therefore don't need to worry about concurrent writes between getting max_row_id_before and using it. max_row_id_before <- sql""" @@ -724,16 +947,6 @@ class AcsSnapshotStore( ) copied_rows } - storage.queryAndUpdate( - withExclusiveSnapshotDataLock( - withIncrementalSnapshotIdempotencyCheck( - table, - statement, - Some(snapshot), - ) - ), - "saveIncrementalSnapshot", - ) } /** Updates an incremental snapshot to a new record time. @@ -760,11 +973,11 @@ class AcsSnapshotStore( insertedRows <- (sql""" insert into #${table.tableName} ( - """ ++ copyFromUpdateHistoryTargetColumns ++ sql""", + """ ++ table.copyFromUpdateHistoryTargetColumns ++ sql""", snapshot_id ) select - """ ++ copyFromUpdateHistorySourceColumns ++ sql""", + """ ++ table.copyFromUpdateHistorySourceColumns ++ sql""", ${snapshot.snapshotId} from update_history_creates c where history_id = $historyId @@ -825,13 +1038,105 @@ object AcsSnapshotStore { "Failed to acquire advisory lock for writing to the acs snapshot table." ) - sealed trait IncrementalAcsSnapshotTable { def tableName: String } + sealed trait IncrementalAcsSnapshotTable { + def tableName: String + protected def storeCreatedArguments: Boolean + def copyFromUpdateHistoryTargetColumns = + IncrementalAcsSnapshotTable.QueryParts.copyFromUpdateHistoryTargetColumns( + storeCreatedArguments + ) + def copyFromUpdateHistorySourceColumns = + IncrementalAcsSnapshotTable.QueryParts.copyFromUpdateHistorySourceColumns( + storeCreatedArguments + ) + } object IncrementalAcsSnapshotTable { case object Next extends IncrementalAcsSnapshotTable { val tableName: String = "acs_incremental_snapshot_data_next" + override protected def storeCreatedArguments: Boolean = false + } + case object NextV2 extends IncrementalAcsSnapshotTable { + val tableName: String = "acs_incremental_snapshot_data_next_v2" + override protected def storeCreatedArguments: Boolean = true } case object Backfill extends IncrementalAcsSnapshotTable { val tableName: String = "acs_incremental_snapshot_data_backfill" + override protected def storeCreatedArguments: Boolean = false + } + + object QueryParts { + + private[IncrementalAcsSnapshotTable] def copyFromUpdateHistoryTargetColumns( + copyCreatedEventData: Boolean + ): SQLActionBuilderChain = { + (if (copyCreatedEventData) + sql""" + create_arguments, + event_id, + record_time, + template_id_package_id, + contract_key, + signatories, + observers, + """ + else sql"") ++ (if (copyCreatedEventData) sql"" else sql"create_id,") ++ sql""" + contract_id, + created_at, + unlocked_amulet_balance, + locked_amulet_balance, + template_id, + stakeholders + """ + } + + private[IncrementalAcsSnapshotTable] def copyFromUpdateHistorySourceColumns( + copyCreatedEventData: Boolean + ): SQLActionBuilderChain = { + (if (copyCreatedEventData) + sql""" + c.create_arguments, + c.event_id, + c.record_time, + c.template_id_package_id, + c.contract_key, + c.signatories, + c.observers, + """ + else sql"") ++ (if (copyCreatedEventData) sql"" + else sql"c.row_id,") ++ sql""" + c.contract_id, + c.created_at, + """ ++ unlockedAmuletBalance() ++ sql"," ++ + lockedAmuletBalance() ++ sql""", + concat(c.package_name, ':', c.template_id_module_name, ':', c.template_id_entity_name) as template_id, + array_cat(c.signatories, c.observers) + """ + } + + def unlockedAmuletBalance() = { + sql""" + case + when package_name = ${Amulet.COMPANION.PACKAGE_NAME} + and template_id_module_name = ${Amulet.COMPANION.TEMPLATE_ID.getModuleName} + and template_id_entity_name = ${Amulet.COMPANION.TEMPLATE_ID.getEntityName} + then (c.create_arguments->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric + else 0 + end + """ + } + + def lockedAmuletBalance() = { + sql""" + case + when package_name = ${LockedAmulet.COMPANION.PACKAGE_NAME} + and template_id_module_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getModuleName} + and template_id_entity_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getEntityName} + then (c.create_arguments->'record'->'fields'->0->'value'->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric + else 0 + end, + """ + } + } } @@ -872,42 +1177,6 @@ object AcsSnapshotStore { ) } - object QueryParts { - - val copyFromUpdateHistoryTargetColumns: SQLActionBuilder = - sql""" - create_id, - contract_id, - created_at, - unlocked_amulet_balance, - locked_amulet_balance, - template_id, - stakeholders - """ - val copyFromUpdateHistorySourceColumns: SQLActionBuilder = - sql""" - c.row_id, - c.contract_id, - c.created_at, - case - when package_name = ${Amulet.COMPANION.PACKAGE_NAME} - and template_id_module_name = ${Amulet.COMPANION.TEMPLATE_ID.getModuleName} - and template_id_entity_name = ${Amulet.COMPANION.TEMPLATE_ID.getEntityName} - then (c.create_arguments->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric - else 0 - end, - case - when package_name = ${LockedAmulet.COMPANION.PACKAGE_NAME} - and template_id_module_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getModuleName} - and template_id_entity_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getEntityName} - then (c.create_arguments->'record'->'fields'->0->'value'->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric - else 0 - end, - concat(c.package_name, ':', c.template_id_module_name, ':', c.template_id_entity_name) as template_id, - array_cat(c.signatories, c.observers) - """ - } - sealed trait AcsSnapshot extends PrettyPrinting { val snapshotRecordTime: CantonTimestamp val migrationId: Long @@ -1055,6 +1324,14 @@ object AcsSnapshotStore { }) } + object AcsTableDDL { + def acsSnapshotCreatesTableName(snapshotRecordTime: CantonTimestamp) = + s"acs_snapshot_creates_${snapshotRecordTime.toEpochMilli}" + + def acsSnapshotStakeholdersTableName(snapshotRecordTime: CantonTimestamp) = + s"acs_snapshot_stakeholders_${snapshotRecordTime.toEpochMilli}" + } + def apply( storage: DbStorage, updateHistory: UpdateHistory, diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala index e256b7a9b9..1531a53ec7 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala @@ -692,6 +692,7 @@ class AcsSnapshotTriggerTest private def storageConfig = ScanStorageConfig( dbAcsSnapshotPeriodHours = 1, + perAcsSnapshotTablesEnabled = false, // TODO: test this bulkAcsSnapshotPeriodHours = 1, // ignored in this test bulkDbReadChunkSize = 1, // ignored in this test bulkZstdFrameSize = 0L, // ignored in this test diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfigTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfigTest.scala index 408c014bd3..0e2ffb1fd7 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfigTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfigTest.scala @@ -12,41 +12,30 @@ class ScanStorageConfigTest "ScanStorageConfig" should { "computeSnapshotTimeAfter" should { + def mkConfig(periodHours: Int) = ScanStorageConfig( + dbAcsSnapshotPeriodHours = periodHours, + perAcsSnapshotTablesEnabled = false, + bulkAcsSnapshotPeriodHours = 4, + bulkDbReadChunkSize = 1, + bulkZstdFrameSize = 0L, + bulkMaxFileSize = 0L, + zstdCompressionLevel = 0, + ) + "return correct time if the previous one is not a valid snapshot time" in { - val config = ScanStorageConfig( - dbAcsSnapshotPeriodHours = 2, - bulkAcsSnapshotPeriodHours = 4, - bulkDbReadChunkSize = 1, - bulkZstdFrameSize = 0L, - bulkMaxFileSize = 0L, - zstdCompressionLevel = 0, - ) + val config = mkConfig(periodHours = 2) val prev = cantonTimestamp("2007-12-03T11:30:00.00Z") val next = cantonTimestamp("2007-12-03T12:00:00.00Z") config.computeDbSnapshotTimeAfter(prev) shouldBe next } "return correct time if the previous one is a valid snapshot time" in { - val config = ScanStorageConfig( - dbAcsSnapshotPeriodHours = 2, - bulkAcsSnapshotPeriodHours = 4, - bulkDbReadChunkSize = 1, - bulkZstdFrameSize = 0L, - bulkMaxFileSize = 0L, - zstdCompressionLevel = 0, - ) + val config = mkConfig(periodHours = 2) val prev = cantonTimestamp("2007-12-03T12:00:00.00Z") val next = cantonTimestamp("2007-12-03T14:00:00.00Z") config.computeDbSnapshotTimeAfter(prev) shouldBe next } "return correct time if the next one is on the day after" in { - val config = ScanStorageConfig( - dbAcsSnapshotPeriodHours = 4, - bulkAcsSnapshotPeriodHours = 8, - bulkDbReadChunkSize = 1, - bulkZstdFrameSize = 0L, - bulkMaxFileSize = 0L, - zstdCompressionLevel = 0, - ) + val config = mkConfig(periodHours = 4) val prev = cantonTimestamp("2007-12-03T21:00:00.00Z") val next = cantonTimestamp("2007-12-04T00:00:00.00Z") config.computeDbSnapshotTimeAfter(prev) shouldBe next diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala index 44adcd701e..0933b0a5fc 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala @@ -45,6 +45,7 @@ class AcsSnapshotBulkStorageCommitFromStagingTest val bulkStorageTestConfig = ScanStorageConfig( dbAcsSnapshotPeriodHours = 3, + perAcsSnapshotTablesEnabled = false, bulkAcsSnapshotPeriodHours = 24, bulkDbReadChunkSize = 1000, bulkZstdFrameSize = 10000L, diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageWriterFromDbTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageWriterFromDbTest.scala index 639122a087..7809d990ec 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageWriterFromDbTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageWriterFromDbTest.scala @@ -65,6 +65,7 @@ class AcsSnapshotBulkStorageWriterFromDbTest val acsSnapshotSize = 48500 val bulkStorageTestConfig = ScanStorageConfig( dbAcsSnapshotPeriodHours = 3, + perAcsSnapshotTablesEnabled = false, bulkAcsSnapshotPeriodHours = 24, bulkDbReadChunkSize = 1000, bulkZstdFrameSize = 10000L, diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala index 62ddcd750e..2900a237d2 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala @@ -50,6 +50,7 @@ class UpdateHistoryBulkStorageTest val maxFileSize = 25000L val bulkStorageTestConfig = ScanStorageConfig( dbAcsSnapshotPeriodHours = 1, + perAcsSnapshotTablesEnabled = false, bulkAcsSnapshotPeriodHours = 2, bulkDbReadChunkSize = 500, bulkZstdFrameSize = 10000L, From 97cf5a278f958652699b15cc00d1cf0f63eeed5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Mon, 10 Aug 2026 16:27:22 +0000 Subject: [PATCH 03/15] fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../splice/scan/store/AcsSnapshotStore.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index 6f8eabbd7d..c62a711303 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -434,7 +434,7 @@ class AcsSnapshotStore( with snapshot as ( select create_id, max(row_id) as row_id from acs_snapshot_data - where row_id between $begin and $end + where row_id between $begin and $end and """ ++ stakeholdersFilter(partyIds) ++ templatesFilter(templates) ++ sql""" group by create_id order by row_id asc @@ -1133,7 +1133,7 @@ object AcsSnapshotStore { and template_id_entity_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getEntityName} then (c.create_arguments->'record'->'fields'->0->'value'->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric else 0 - end, + end """ } From 344def0e3570d2cefa372a9592e46c8ecc1ac398 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Mon, 10 Aug 2026 17:11:44 +0000 Subject: [PATCH 04/15] add v2 tests and get it to work [ci] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../stable/V073__acs_snapshot_table_name.sql | 9 ++- .../splice/util/QualifiedName.scala | 8 +++ .../splice/scan/store/AcsSnapshotStore.scala | 29 ++++++--- .../store/db/AcsSnapshotStoreTest.scala | 61 +++++++++++-------- 4 files changed, 70 insertions(+), 37 deletions(-) diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql index 54d57c3764..5e961821ad 100644 --- a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql +++ b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql @@ -26,10 +26,17 @@ create table acs_incremental_snapshot_data_next_v2 contract_id text not null, -- All the data necessary to reconstruct a created event + -- TODO: also unecessary + template_id text not null, + -- TODO: this should be unnecessary + stakeholders text[] not null, create_arguments jsonb not null, event_id text not null, record_time bigint not null, - template_id_package_id text not null, -- the package_name is already included as part of the template_id + template_id_package_id text not null, + package_name text not null, + template_id_module_name text not null, + template_id_entity_name text not null, contract_key text null, created_at bigint not null, signatories text[] not null, diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/QualifiedName.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/QualifiedName.scala index 962bf45ce4..fc7622f247 100644 --- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/QualifiedName.scala +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/QualifiedName.scala @@ -45,6 +45,14 @@ object PackageQualifiedName { QualifiedName(companion.TEMPLATE_ID.getModuleName, companion.TEMPLATE_ID.getEntityName), ) } + + def assertFromString(s: String): PackageQualifiedName = { + val segments = s.split(":") + if (segments.length != 3) { + throw new IllegalArgumentException(s"Expect qualified name with two identifiers but got $s") + } + PackageQualifiedName(segments(0), QualifiedName(segments(1), segments(2))) + } } object QualifiedName { diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index c62a711303..0eaf654801 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -334,7 +334,7 @@ class AcsSnapshotStore( record_time, template_id_package_id, template_id, - contract_id, + s.contract_id, create_arguments, contract_key, signatories, @@ -353,7 +353,7 @@ class AcsSnapshotStore( String, CantonTimestamp, String, - PackageQualifiedName, + String, String, String, Option[String], @@ -370,7 +370,7 @@ class AcsSnapshotStore( eventId, recordTime, packageId, - templateIdPackageQualifiedName, + rawTemplateIdPackageQualifiedName, contractId, createArguments, contractKey, @@ -378,6 +378,8 @@ class AcsSnapshotStore( observers, createdAt, ) => + val templateIdPackageQualifiedName = + PackageQualifiedName.assertFromString(rawTemplateIdPackageQualifiedName) rowId -> SpliceCreatedEvent( eventId = eventId, recordTime = recordTime, @@ -806,11 +808,12 @@ class AcsSnapshotStore( AcsTableDDL.acsSnapshotStakeholdersTableName(snapshot.targetRecordTime) for { - _ <- sqlu"create table #$createsTableName like acs_snapshot_creates_template" - _ <- sqlu"create table #$stakeholdersTableName like acs_snapshot_stakeholders_template" + _ <- sqlu"create table #$createsTableName (like acs_snapshot_creates_template including all)" + _ <- + sqlu"create table #$stakeholdersTableName (like acs_snapshot_stakeholders_template including all)" copiedCreateRows <- (sql""" insert into #$createsTableName (contract_id, create_arguments, event_id, record_time, template_id_package_id, contract_key, created_at, signatories, observers, unlocked_amulet_balance, locked_amulet_balance) - select s.contract_id, s.create_arguments, s.event-id, s.record_time, s.template_id_package_id, s.contract_key, s.creataed_at, s.signatories, s.observers, """ ++ IncrementalAcsSnapshotTable.QueryParts + select s.contract_id, s.create_arguments, s.event_id, s.record_time, s.template_id_package_id, s.contract_key, s.created_at, s.signatories, s.observers, """ ++ IncrementalAcsSnapshotTable.QueryParts .unlockedAmuletBalance() ++ sql", " ++ IncrementalAcsSnapshotTable.QueryParts .lockedAmuletBalance() ++ sql""" from #${table.tableName} s @@ -822,7 +825,7 @@ class AcsSnapshotStore( insert into #${stakeholdersTableName} (stakeholder, template_id, contract_id) select stakeholder, s.template_id, contract_id from #${table.tableName} s - cross join unnest(array_cat(s.signatories, s.observers)) as stakeholder + cross join unnest(s.stakeholders) as stakeholder where s.snapshot_id = ${snapshot.snapshotId} order by created_at, contract_id """ @@ -844,7 +847,7 @@ class AcsSnapshotStore( last_row_id, unlocked_amulet_balance, locked_amulet_balance, - table_name + data_table_name ) values ( ${snapshot.recordTime}, @@ -1075,6 +1078,9 @@ object AcsSnapshotStore { event_id, record_time, template_id_package_id, + package_name, + template_id_module_name, + template_id_entity_name, contract_key, signatories, observers, @@ -1098,6 +1104,9 @@ object AcsSnapshotStore { c.event_id, c.record_time, c.template_id_package_id, + c.package_name, + c.template_id_module_name, + c.template_id_entity_name, c.contract_key, c.signatories, c.observers, @@ -1119,7 +1128,7 @@ object AcsSnapshotStore { when package_name = ${Amulet.COMPANION.PACKAGE_NAME} and template_id_module_name = ${Amulet.COMPANION.TEMPLATE_ID.getModuleName} and template_id_entity_name = ${Amulet.COMPANION.TEMPLATE_ID.getEntityName} - then (c.create_arguments->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric + then (create_arguments->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric else 0 end """ @@ -1131,7 +1140,7 @@ object AcsSnapshotStore { when package_name = ${LockedAmulet.COMPANION.PACKAGE_NAME} and template_id_module_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getModuleName} and template_id_entity_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getEntityName} - then (c.create_arguments->'record'->'fields'->0->'value'->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric + then (create_arguments->'record'->'fields'->0->'value'->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric else 0 end """ diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala index 4c27a6c133..6af6df3e9d 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala @@ -1,7 +1,7 @@ package org.lfdecentralizedtrust.splice.store.db import cats.data.NonEmptyVector -import com.daml.ledger.javaapi.data.{Unit as damlUnit} +import com.daml.ledger.javaapi.data.Unit as damlUnit import com.daml.ledger.javaapi.data.codegen.ContractId import com.daml.metrics.api.noop.NoOpMetricsFactory import org.lfdecentralizedtrust.splice.environment.DarResources @@ -32,8 +32,9 @@ import java.time.Instant import scala.concurrent.Future import scala.util.{Failure, Success} import StoreTestBase.* +import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.IncrementalAcsSnapshotTable -class AcsSnapshotStoreTest +trait AcsSnapshotStoreTest extends StoreTestBase with HasExecutionContext with StoreErrors @@ -42,6 +43,8 @@ class AcsSnapshotStoreTest with AcsJdbcTypes with AcsTables { + val nextTable: IncrementalAcsSnapshotTable + private val DefaultMigrationId = 0L private val timestamp1 = CantonTimestamp.Epoch.plusSeconds(3600) private val timestamp2 = CantonTimestamp.Epoch.plusSeconds(3600 * 2) @@ -1077,9 +1080,7 @@ class AcsSnapshotStoreTest for { updateHistory <- mkUpdateHistory() store = mkStore(updateHistory) - incrementalSnapshotN <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next - ) + incrementalSnapshotN <- store.getIncrementalSnapshot(nextTable) incrementalSnapshotB <- store.getIncrementalSnapshot( AcsSnapshotStore.IncrementalAcsSnapshotTable.Backfill ) @@ -1122,13 +1123,13 @@ class AcsSnapshotStoreTest } yield snapshot1) _ <- store.initializeIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot1.value, timestamp2, ) incrementalSnapshotN <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) incrementalSnapshotB <- store.getIncrementalSnapshot( AcsSnapshotStore.IncrementalAcsSnapshotTable.Backfill @@ -1136,10 +1137,10 @@ class AcsSnapshotStoreTest _ <- clueF(s"Update snapshot")(for { snapshot <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- store.updateIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, timestamp2, ) @@ -1147,10 +1148,10 @@ class AcsSnapshotStoreTest _ <- clueF(s"Save snapshot")(for { snapshot <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- store.saveIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, nextSnapshotTargetRecordTime = timestamp3, ) @@ -1193,13 +1194,13 @@ class AcsSnapshotStoreTest snapshotRecordTime = timestamp1.minusSeconds(1L) _ <- store.initializeIncrementalSnapshotFromImportUpdates( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshotRecordTime, timestamp2, DefaultMigrationId, ) incrementalSnapshotN <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) incrementalSnapshotB <- store.getIncrementalSnapshot( AcsSnapshotStore.IncrementalAcsSnapshotTable.Backfill @@ -1253,7 +1254,7 @@ class AcsSnapshotStoreTest _ <- clueF(s"Snapshot A: start from import updates at T0")( storeM1.initializeIncrementalSnapshotFromImportUpdates( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, recordTime = timestamp1, targetRecordTime = timestamp1.plusSeconds(9L), 1L, @@ -1266,10 +1267,10 @@ class AcsSnapshotStoreTest for { (_, cid1) <- ingestHistory(updateHistoryM1, 0L) snapshot <- storeM1.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- storeM1.updateIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, // Note: there is no event at exactly T9 timestamp1.plusSeconds(9L), @@ -1279,11 +1280,11 @@ class AcsSnapshotStoreTest _ <- clueF(s"Snapshot A: finalize snapshot at T9")(for { snapshot <- storeM1.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ = snapshot.value.recordTime shouldBe timestamp1.plusSeconds(9L) _ <- storeM1.saveIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, nextSnapshotTargetRecordTime = timestamp1.plusSeconds(20L), ) @@ -1311,10 +1312,10 @@ class AcsSnapshotStoreTest )(for { (_, cid2) <- ingestHistory(updateHistoryM1, 12L) snapshot <- storeM1.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- storeM1.updateIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, // Note: there is an event at exactly T20 timestamp1.plusSeconds(20L), @@ -1323,11 +1324,11 @@ class AcsSnapshotStoreTest _ <- clueF(s"Snapshot B: finalize snapshot at T20")(for { snapshot <- storeM1.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ = snapshot.value.recordTime shouldBe timestamp1.plusSeconds(20L) _ <- storeM1.saveIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, nextSnapshotTargetRecordTime = timestamp1.plusSeconds(30L), ) @@ -1364,23 +1365,23 @@ class AcsSnapshotStoreTest _ <- ingestCreate(updateHistory, amuletRules(), CantonTimestamp.MinValue) _ <- store.initializeIncrementalSnapshotFromImportUpdates( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, timestamp2, timestamp3, DefaultMigrationId, ) incrementalSnapshotBefore <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- store.deleteIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, incrementalSnapshotBefore.value, ) incrementalSnapshotAfter <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) } yield { incrementalSnapshotAfter shouldBe None @@ -1531,3 +1532,11 @@ class AcsSnapshotStoreTest } yield () } + +class LegacyAcsSnapshotStoreTest extends AcsSnapshotStoreTest { + override val nextTable: IncrementalAcsSnapshotTable = IncrementalAcsSnapshotTable.Next +} + +class TablePerAcsSnapshotStoreTest extends AcsSnapshotStoreTest { + override val nextTable: IncrementalAcsSnapshotTable = IncrementalAcsSnapshotTable.NextV2 +} From 421f68c6c9c112e289d02979a5473295bec3c99d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Tue, 11 Aug 2026 09:40:56 +0000 Subject: [PATCH 05/15] [ci] refactor and remove TODOs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../stable/V073__acs_snapshot_table_name.sql | 8 +- .../splice/scan/store/AcsSnapshotStore.scala | 111 ++++++++++-------- 2 files changed, 67 insertions(+), 52 deletions(-) diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql index 5e961821ad..ec739bd217 100644 --- a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql +++ b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql @@ -26,19 +26,17 @@ create table acs_incremental_snapshot_data_next_v2 contract_id text not null, -- All the data necessary to reconstruct a created event - -- TODO: also unecessary - template_id text not null, - -- TODO: this should be unnecessary - stakeholders text[] not null, create_arguments jsonb not null, event_id text not null, record_time bigint not null, template_id_package_id text not null, + -- template_id = package_name:template_id_module_name:template_id_entity_name package_name text not null, template_id_module_name text not null, template_id_entity_name text not null, contract_key text null, created_at bigint not null, + -- stakeholders = array_cat(signatories, observers) signatories text[] not null, observers text[] not null, -- plus the Amulet-specific balance columns currently computed in the working table @@ -75,12 +73,12 @@ create table acs_snapshot_creates_template locked_amulet_balance numeric ); --- TODO: revisit this name because this does more than stakehodlering create table acs_snapshot_stakeholders_template ( -- Important: insertion order should happen by (created_at, contract_id) for: -- 1) backwards compatibility -- 2) deterministic ordering across SVs + -- then we can sort by row_id to preserve that order and support pagination with after: Long row_id bigint generated by default as identity primary key, stakeholder text not null, template_id text not null, diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index 0eaf654801..ef60551944 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -778,7 +778,6 @@ class AcsSnapshotStore( assert(snapshot.tableName == table.tableName) assert(snapshot.historyId == historyId) assert(snapshot.recordTime == snapshot.targetRecordTime) - // TODO: split into insert into legacy table and insert into new table + create indexes val statement: DBIO[Int] = table match { case IncrementalAcsSnapshotTable.NextV2 => saveV2IncrementalSnapshot(table, snapshot, nextSnapshotTargetRecordTime)(tc) @@ -823,9 +822,9 @@ class AcsSnapshotStore( """).toActionBuilder.asUpdate copiedStakeholderRows <- sqlu""" insert into #${stakeholdersTableName} (stakeholder, template_id, contract_id) - select stakeholder, s.template_id, contract_id + select stakeholder, concat(s.package_name, ':', s.template_id_module_name, ':', s.template_id_entity_name) as template_id, contract_id from #${table.tableName} s - cross join unnest(s.stakeholders) as stakeholder + cross join unnest(array_cat(s.observers, s.signatories)) as stakeholder where s.snapshot_id = ${snapshot.snapshotId} order by created_at, contract_id """ @@ -1043,37 +1042,43 @@ object AcsSnapshotStore { sealed trait IncrementalAcsSnapshotTable { def tableName: String - protected def storeCreatedArguments: Boolean - def copyFromUpdateHistoryTargetColumns = - IncrementalAcsSnapshotTable.QueryParts.copyFromUpdateHistoryTargetColumns( - storeCreatedArguments - ) - def copyFromUpdateHistorySourceColumns = - IncrementalAcsSnapshotTable.QueryParts.copyFromUpdateHistorySourceColumns( - storeCreatedArguments - ) + def copyFromUpdateHistoryTargetColumns: SQLActionBuilderChain + def copyFromUpdateHistorySourceColumns: SQLActionBuilderChain } object IncrementalAcsSnapshotTable { - case object Next extends IncrementalAcsSnapshotTable { - val tableName: String = "acs_incremental_snapshot_data_next" - override protected def storeCreatedArguments: Boolean = false - } case object NextV2 extends IncrementalAcsSnapshotTable { val tableName: String = "acs_incremental_snapshot_data_next_v2" - override protected def storeCreatedArguments: Boolean = true + + override def copyFromUpdateHistoryTargetColumns: SQLActionBuilderChain = + QueryParts.v2CopyFromUpdateHistoryTargetColumns + + override def copyFromUpdateHistorySourceColumns: SQLActionBuilderChain = + QueryParts.v2CopyFromUpdateHistorySourceColumns + } + case object Next extends IncrementalAcsSnapshotTable { + val tableName: String = "acs_incremental_snapshot_data_next" + + override def copyFromUpdateHistoryTargetColumns: SQLActionBuilderChain = + QueryParts.legacyCopyFromUpdateHistoryTargetColumns + + override def copyFromUpdateHistorySourceColumns: SQLActionBuilderChain = + QueryParts.legacyCopyFromUpdateHistorySourceColumns } case object Backfill extends IncrementalAcsSnapshotTable { val tableName: String = "acs_incremental_snapshot_data_backfill" - override protected def storeCreatedArguments: Boolean = false + + override def copyFromUpdateHistoryTargetColumns: SQLActionBuilderChain = + QueryParts.legacyCopyFromUpdateHistoryTargetColumns + + override def copyFromUpdateHistorySourceColumns: SQLActionBuilderChain = + QueryParts.legacyCopyFromUpdateHistorySourceColumns } object QueryParts { - private[IncrementalAcsSnapshotTable] def copyFromUpdateHistoryTargetColumns( - copyCreatedEventData: Boolean - ): SQLActionBuilderChain = { - (if (copyCreatedEventData) - sql""" + private[IncrementalAcsSnapshotTable] val v2CopyFromUpdateHistoryTargetColumns + : SQLActionBuilderChain = { + sql""" create_arguments, event_id, record_time, @@ -1084,22 +1089,16 @@ object AcsSnapshotStore { contract_key, signatories, observers, - """ - else sql"") ++ (if (copyCreatedEventData) sql"" else sql"create_id,") ++ sql""" - contract_id, - created_at, - unlocked_amulet_balance, - locked_amulet_balance, - template_id, - stakeholders - """ + contract_id, + created_at, + unlocked_amulet_balance, + locked_amulet_balance + """ } - private[IncrementalAcsSnapshotTable] def copyFromUpdateHistorySourceColumns( - copyCreatedEventData: Boolean - ): SQLActionBuilderChain = { - (if (copyCreatedEventData) - sql""" + private[IncrementalAcsSnapshotTable] val v2CopyFromUpdateHistorySourceColumns + : SQLActionBuilderChain = { + sql""" c.create_arguments, c.event_id, c.record_time, @@ -1110,16 +1109,34 @@ object AcsSnapshotStore { c.contract_key, c.signatories, c.observers, - """ - else sql"") ++ (if (copyCreatedEventData) sql"" - else sql"c.row_id,") ++ sql""" - c.contract_id, - c.created_at, - """ ++ unlockedAmuletBalance() ++ sql"," ++ - lockedAmuletBalance() ++ sql""", - concat(c.package_name, ':', c.template_id_module_name, ':', c.template_id_entity_name) as template_id, - array_cat(c.signatories, c.observers) - """ + c.contract_id, + c.created_at,""" ++ + unlockedAmuletBalance() ++ sql"," ++ + lockedAmuletBalance() + } + + private[IncrementalAcsSnapshotTable] val legacyCopyFromUpdateHistoryTargetColumns + : SQLActionBuilderChain = { + sql""" + create_id, + template_id, + stakeholders, + contract_id, + created_at, + unlocked_amulet_balance, + locked_amulet_balance""" + } + + private[IncrementalAcsSnapshotTable] val legacyCopyFromUpdateHistorySourceColumns + : SQLActionBuilderChain = { + sql""" + c.row_id, + concat(c.package_name, ':', c.template_id_module_name, ':', c.template_id_entity_name) as template_id, + array_cat(c.signatories, c.observers) as stakeholder, + c.contract_id, + c.created_at, + """ ++ unlockedAmuletBalance() ++ sql"," ++ + lockedAmuletBalance() } def unlockedAmuletBalance() = { From cc61bf0dbd990b831abc6b3c6be7d0f53d7bee7e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Tue, 11 Aug 2026 09:58:59 +0000 Subject: [PATCH 06/15] fix/add todos [ci] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../splice/scan/store/AcsSnapshotStore.scala | 1 + .../automation/AcsSnapshotTriggerTest.scala | 39 +++++++++++++------ 2 files changed, 29 insertions(+), 11 deletions(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index ef60551944..ca2d59643b 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -828,6 +828,7 @@ class AcsSnapshotStore( where s.snapshot_id = ${snapshot.snapshotId} order by created_at, contract_id """ + // TODO: we should create the necessary indexes (unlocked_amulet_balance, locked_amulet_balance) <- sql""" select diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala index 1531a53ec7..a5ce4b4619 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala @@ -14,12 +14,14 @@ import org.scalatest.wordspec.AnyWordSpec import scala.concurrent.Future -class AcsSnapshotTriggerTest +trait AcsSnapshotTriggerTest extends AnyWordSpec with BaseTest with HasExecutionContext with HasActorSystem { + protected def storageConfig: ScanStorageConfig + "AcsSnapshotTrigger" should { "initialize from an existing snapshot" in { @@ -690,16 +692,6 @@ class AcsSnapshotTriggerTest } } - private def storageConfig = ScanStorageConfig( - dbAcsSnapshotPeriodHours = 1, - perAcsSnapshotTablesEnabled = false, // TODO: test this - bulkAcsSnapshotPeriodHours = 1, // ignored in this test - bulkDbReadChunkSize = 1, // ignored in this test - bulkZstdFrameSize = 0L, // ignored in this test - bulkMaxFileSize = 0L, // ignored in this test - zstdCompressionLevel = 0, // ignored in this test - ) - private def unused0[T]: () => Future[T] = () => fail("This argument should not be used") private def unused1[T]: Long => Future[T] = _ => fail("This argument should not be used") @@ -749,3 +741,28 @@ class AcsSnapshotTriggerTest private def migration3 = 3L private def migration4 = 4L } + +class LegacyAcsSnapshotTriggerTest extends AcsSnapshotTriggerTest{ + protected def storageConfig = ScanStorageConfig( + dbAcsSnapshotPeriodHours = 1, + perAcsSnapshotTablesEnabled = false, + bulkAcsSnapshotPeriodHours = 1, // ignored in this test + bulkDbReadChunkSize = 1, // ignored in this test + bulkZstdFrameSize = 0L, // ignored in this test + bulkMaxFileSize = 0L, // ignored in this test + zstdCompressionLevel = 0, // ignored in this test + ) +} + + +class V2AcsSnapshotTriggerTest extends AcsSnapshotTriggerTest{ + protected def storageConfig = ScanStorageConfig( + dbAcsSnapshotPeriodHours = 1, + perAcsSnapshotTablesEnabled = true, + bulkAcsSnapshotPeriodHours = 1, // ignored in this test + bulkDbReadChunkSize = 1, // ignored in this test + bulkZstdFrameSize = 0L, // ignored in this test + bulkMaxFileSize = 0L, // ignored in this test + zstdCompressionLevel = 0, // ignored in this test + ) +} From d59c0827f25d20147a7b00b3ba48b91a92a18eab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Tue, 11 Aug 2026 09:59:59 +0000 Subject: [PATCH 07/15] [ci] scalafmt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../splice/scan/automation/AcsSnapshotTriggerTest.scala | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala index a5ce4b4619..d27668ac7c 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala @@ -742,7 +742,7 @@ trait AcsSnapshotTriggerTest private def migration4 = 4L } -class LegacyAcsSnapshotTriggerTest extends AcsSnapshotTriggerTest{ +class LegacyAcsSnapshotTriggerTest extends AcsSnapshotTriggerTest { protected def storageConfig = ScanStorageConfig( dbAcsSnapshotPeriodHours = 1, perAcsSnapshotTablesEnabled = false, @@ -754,8 +754,7 @@ class LegacyAcsSnapshotTriggerTest extends AcsSnapshotTriggerTest{ ) } - -class V2AcsSnapshotTriggerTest extends AcsSnapshotTriggerTest{ +class V2AcsSnapshotTriggerTest extends AcsSnapshotTriggerTest { protected def storageConfig = ScanStorageConfig( dbAcsSnapshotPeriodHours = 1, perAcsSnapshotTablesEnabled = true, From 5f4e43fa078e0e326ed025f90d20357f6c315939 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Tue, 11 Aug 2026 10:02:51 +0000 Subject: [PATCH 08/15] [ci] prevent tablename collisions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../stable/V073__acs_snapshot_table_name.sql | 2 +- .../splice/scan/store/AcsSnapshotStore.scala | 22 ++++++++++--------- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql index ec739bd217..200f31d44e 100644 --- a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql +++ b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql @@ -53,7 +53,7 @@ alter table acs_incremental_snapshot_data_next_v2 create index acs_incremental_snapshot_data_next_v2_ca_ci on acs_incremental_snapshot_data_next_v2 (snapshot_id, created_at, contract_id); --- Template table for acs_snapshot_creates_. +-- Template table for acs_snapshot_creates__. -- This allows the code to just CREATE TABLE LIKE acs_snapshot_creates_template or acs_snapshot_stakeholders_template. -- Design decision: we don't have a single table per (contract_id, stakeholder) in order to avoid duplicating the create_arguments. create table acs_snapshot_creates_template diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index ca2d59643b..b349c69807 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -256,10 +256,12 @@ class AcsSnapshotStore( DBIOAction.seq( sqlu"""delete from acs_snapshot where snapshot_record_time = ${snapshot.snapshotRecordTime}""", sqlu"""drop table #${AcsTableDDL.acsSnapshotCreatesTableName( - table.snapshotRecordTime + historyId, + table.snapshotRecordTime, )};""", sqlu"""drop table #${AcsTableDDL.acsSnapshotStakeholdersTableName( - table.snapshotRecordTime + historyId, + table.snapshotRecordTime, )};""", ) } @@ -321,9 +323,9 @@ class AcsSnapshotStore( templates: Seq[PackageQualifiedName], )(implicit tc: TraceContext): Future[Vector[(Long, SpliceCreatedEvent)]] = { val createsTableName = - AcsTableDDL.acsSnapshotCreatesTableName(snapshot.snapshotRecordTime) + AcsTableDDL.acsSnapshotCreatesTableName(historyId, snapshot.snapshotRecordTime) val stakeholdersTableName = - AcsTableDDL.acsSnapshotStakeholdersTableName(snapshot.snapshotRecordTime) + AcsTableDDL.acsSnapshotStakeholdersTableName(historyId, snapshot.snapshotRecordTime) val afterFilter = after.fold(sql"")(after => sql" and s.row_id > $after") storage .query( @@ -802,9 +804,9 @@ class AcsSnapshotStore( nextSnapshotTargetRecordTime: CantonTimestamp, )(implicit tc: TraceContext) = { val createsTableName = - AcsTableDDL.acsSnapshotCreatesTableName(snapshot.targetRecordTime) + AcsTableDDL.acsSnapshotCreatesTableName(historyId, snapshot.targetRecordTime) val stakeholdersTableName = - AcsTableDDL.acsSnapshotStakeholdersTableName(snapshot.targetRecordTime) + AcsTableDDL.acsSnapshotStakeholdersTableName(historyId, snapshot.targetRecordTime) for { _ <- sqlu"create table #$createsTableName (like acs_snapshot_creates_template including all)" @@ -1352,11 +1354,11 @@ object AcsSnapshotStore { } object AcsTableDDL { - def acsSnapshotCreatesTableName(snapshotRecordTime: CantonTimestamp) = - s"acs_snapshot_creates_${snapshotRecordTime.toEpochMilli}" + def acsSnapshotCreatesTableName(historyId: Long, snapshotRecordTime: CantonTimestamp) = + s"acs_snapshot_creates_${historyId}_${snapshotRecordTime.toEpochMilli}" - def acsSnapshotStakeholdersTableName(snapshotRecordTime: CantonTimestamp) = - s"acs_snapshot_stakeholders_${snapshotRecordTime.toEpochMilli}" + def acsSnapshotStakeholdersTableName(historyId: Long, snapshotRecordTime: CantonTimestamp) = + s"acs_snapshot_stakeholders_${historyId}_${snapshotRecordTime.toEpochMilli}" } def apply( From d1a433d7cdd16c0ff4ae212b9647654682d05c57 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Tue, 18 Aug 2026 16:14:45 +0000 Subject: [PATCH 09/15] fix imports MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../splice/scan/store/AcsSnapshotStore.scala | 18 ++---------------- 1 file changed, 2 insertions(+), 16 deletions(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index e7a88cda9d..08260a98d7 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -18,12 +18,7 @@ import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.{ } import org.lfdecentralizedtrust.splice.store.UpdateHistory.SelectFromCreateEvents import org.lfdecentralizedtrust.splice.store.{HardLimit, Limit, LimitHelpers, UpdateHistory} -import org.lfdecentralizedtrust.splice.store.db.{ - AcsJdbcTypes, - AcsQueries, - AdvisoryLocks, - AdvisoryLockIds, -} +import org.lfdecentralizedtrust.splice.store.db.{AdvisoryLocks} import org.lfdecentralizedtrust.splice.util.{Contract, HoldingsSummary, PackageQualifiedName} import com.digitalasset.canton.data.CantonTimestamp import com.digitalasset.canton.lifecycle.{CloseContext, FutureUnlessShutdown} @@ -35,19 +30,10 @@ import com.digitalasset.canton.resource.DbStorage.SQLActionBuilderChain import com.digitalasset.canton.topology.PartyId import com.digitalasset.canton.tracing.TraceContext import com.google.protobuf.ByteString -import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{Amulet, LockedAmulet} import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.* -import org.lfdecentralizedtrust.splice.store.UpdateHistory.SelectFromCreateEvents import org.lfdecentralizedtrust.splice.store.db.{AcsJdbcTypes, AcsQueries, AdvisoryLockIds} import org.lfdecentralizedtrust.splice.store.events.SpliceCreatedEvent -import org.lfdecentralizedtrust.splice.store.{HardLimit, Limit, LimitHelpers, UpdateHistory} -import org.lfdecentralizedtrust.splice.util.{ - Contract, - EventId, - HoldingsSummary, - PackageQualifiedName, - ValueJsonCodecProtobuf as ProtobufCodec, -} +import org.lfdecentralizedtrust.splice.util.{EventId, ValueJsonCodecProtobuf as ProtobufCodec} import slick.dbio.{DBIO, DBIOAction, Effect, NoStream} import slick.jdbc.canton.ActionBasedSQLInterpolation.Implicits.actionBasedSQLInterpolationCanton import slick.jdbc.{GetResult, JdbcProfile} From 01572e7083d518e6fe7205cc536ea4776848edaf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Wed, 19 Aug 2026 11:08:57 +0000 Subject: [PATCH 10/15] ddl lock on drop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../splice/scan/store/AcsSnapshotStore.scala | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index 08260a98d7..b87cecd599 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -246,14 +246,16 @@ class AcsSnapshotStore( case table: PerTableAcsSnapshot => DBIOAction.seq( sqlu"""delete from acs_snapshot where snapshot_record_time = ${snapshot.snapshotRecordTime}""", - sqlu"""drop table #${AcsTableDDL.acsSnapshotCreatesTableName( + AdvisoryLocks.withDdlLock(sqlu"""drop table #${AcsTableDDL.acsSnapshotCreatesTableName( historyId, table.snapshotRecordTime, - )};""", - sqlu"""drop table #${AcsTableDDL.acsSnapshotStakeholdersTableName( - historyId, - table.snapshotRecordTime, - )};""", + )};"""), + AdvisoryLocks.withDdlLock( + sqlu"""drop table #${AcsTableDDL.acsSnapshotStakeholdersTableName( + historyId, + table.snapshotRecordTime, + )};""" + ), ) } storage.update(statement.transactionally, "deleteSnapshot") From 0d6406f46866eb361caca091dbed1e99ce2f337f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Wed, 19 Aug 2026 11:10:38 +0000 Subject: [PATCH 11/15] distinct on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../splice/scan/store/AcsSnapshotStore.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index b87cecd599..76b528bc8c 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -323,7 +323,7 @@ class AcsSnapshotStore( storage .query( (sql""" - select + select distinct on(c.contract_id) s.row_id, event_id, record_time, From 73301afd789d0999c758dbbd1388c0eefc66aadb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Wed, 19 Aug 2026 11:18:10 +0000 Subject: [PATCH 12/15] Move migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- ..._snapshot_table_name.sql => V074__acs_snapshot_table_name.sql} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename apps/common/src/main/resources/db/migration/canton-network/postgres/stable/{V073__acs_snapshot_table_name.sql => V074__acs_snapshot_table_name.sql} (100%) diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V074__acs_snapshot_table_name.sql similarity index 100% rename from apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql rename to apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V074__acs_snapshot_table_name.sql From fa958f9810aee2aa50f29f4f63ba89719bb273a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Wed, 19 Aug 2026 11:19:48 +0000 Subject: [PATCH 13/15] fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../splice/scan/store/AcsSnapshotStore.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index 76b528bc8c..7b2df849a2 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -258,7 +258,7 @@ class AcsSnapshotStore( ), ) } - storage.update(statement.transactionally, "deleteSnapshot") + storage.queryAndUpdate(statement.transactionally, "deleteSnapshot") } def queryAcsSnapshot( @@ -323,7 +323,7 @@ class AcsSnapshotStore( storage .query( (sql""" - select distinct on(c.contract_id) + select --distinct on(c.contract_id) s.row_id, event_id, record_time, From 8f9fb170c4acc9ec17a7d86200d05cc6a57d3164 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Wed, 19 Aug 2026 11:35:39 +0000 Subject: [PATCH 14/15] properly fix the query which probably breaks indexes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../splice/scan/store/AcsSnapshotStore.scala | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index 7b2df849a2..a16262f651 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -323,7 +323,15 @@ class AcsSnapshotStore( storage .query( (sql""" - select --distinct on(c.contract_id) + with contracts as ( + select distinct on (contract_id) contract_id, row_id, template_id + from #$stakeholdersTableName + where """ ++ stakeholdersFilter(partyIds) ++ + templatesFilter(templates) ++ + afterFilter ++ sql""" + order by contract_id + ) + select s.row_id, event_id, record_time, @@ -335,11 +343,8 @@ class AcsSnapshotStore( signatories, observers, created_at - from #$stakeholdersTableName s + from contracts s join #$createsTableName c on s.contract_id = c.contract_id - where """ ++ stakeholdersFilter(partyIds) ++ - templatesFilter(templates) ++ - afterFilter ++ sql""" order by s.row_id limit ${sqlLimit(limit)} """).toActionBuilder.as[ From d5181852e7fecf9bf94e4b3e7269a5ac0f090f2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Oriol=20Mu=C3=B1oz?= Date: Wed, 19 Aug 2026 14:43:32 +0000 Subject: [PATCH 15/15] whatever MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Oriol Muñoz --- .../splice/scan/store/AcsSnapshotStore.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index a16262f651..086643e617 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -330,6 +330,7 @@ class AcsSnapshotStore( templatesFilter(templates) ++ afterFilter ++ sql""" order by contract_id + limit ${sqlLimit(limit)} ) select s.row_id, @@ -345,8 +346,8 @@ class AcsSnapshotStore( created_at from contracts s join #$createsTableName c on s.contract_id = c.contract_id + -- This will only sort over LIMIT rows, which is acceptable order by s.row_id - limit ${sqlLimit(limit)} """).toActionBuilder.as[ ( Long,