manually clear BigQuery with datatransfer instead of relying on table expiry

cluster/pulumi/common-sv/src/bigQuery.ts

@@ -355,7 +355,11 @@ function installBigqueryStagingDataset(scanBigQuery: ScanBigQueryConfig): gcp.bi
     friendlyName: `${scanBigQuery.dataset} Staging Dataset`,
     location: cloudsdkComputeRegion(),
     deleteContentsOnDestroy: true,
-    defaultTableExpirationMs: THREE_DAYS_MS,
+    // ISSUE#6814: Do not rely on ingestion timestamps for retention in staging. 
+    // GCP calculates expiration from the table creation date, which will delete 
+    // staging tables at the 3-day mark even if production sync is incomplete.
+
+    //defaultTableExpirationMs: THREE_DAYS_MS,
     labels: {
       cluster: CLUSTER_BASENAME,
       datastream_id: 'stag_prod',
@@ -374,6 +378,21 @@ function installBigqueryProdDataset(scanBigQuery: ScanBigQueryConfig): gcp.bigqu
     },
   });
 }
+// ============================================================================
+// IAM PERMISSIONS for SCHEDULED QUERIES
+// ============================================================================
+function installBqTransferServiceAgentPermission(): gcp.projects.IAMMember {
+  const currentProject = gcp.organizations.getProjectOutput({});
+  const projectId = currentProject.apply(p => p.projectId!);
+
+  return new gcp.projects.IAMMember('bq-transfer-token-creator', {
+    project: projectId,
+    role: 'roles/iam.serviceAccountTokenCreator',
+    member: currentProject.apply(
+      p => `serviceAccount:service-${p.number}@gcp-sa-bigquerydatatransfer.iam.gserviceaccount.com`
+    ),
+  });
+}
 
 // ============================================================================
 // HOURLY DEDUPLICATION & SCHEDULED QUERIES
@@ -384,20 +403,13 @@ const rawSqlTemplate = fs.readFileSync(path.join(__dirname, 'hourly_append.sql')
 function installHourlyScheduledQueries(
   namespace: ExactNamespace,
   stagingDataset: gcp.bigquery.Dataset,
-  prodDataset: gcp.bigquery.Dataset
+  prodDataset: gcp.bigquery.Dataset,
+  transferServiceAgentPermission: gcp.projects.IAMMember
 ) {
   const currentProject = gcp.organizations.getProjectOutput({});
   const projectId = currentProject.apply(p => p.projectId!);
   const schemaName = scanAppDatabaseName(namespace);
 
-  const transferServiceAgentPermission = new gcp.projects.IAMMember('bq-transfer-token-creator', {
-    project: projectId,
-    role: 'roles/iam.serviceAccountTokenCreator',
-    member: currentProject.apply(
-      p => `serviceAccount:service-${p.number}@gcp-sa-bigquerydatatransfer.iam.gserviceaccount.com`
-    ),
-  });
-
   Object.entries(replicatedTables).forEach(([tableName, tableConfig]) => {
     const primaryKeyExpr = tableConfig.primaryKey;
     const colName = tableConfig.datePartitionColumn;
@@ -463,7 +475,68 @@ function installHourlyScheduledQueries(
     );
   });
 }
+// ============================================================================
+// Purging data older than 7 days from the staging table
+// ============================================================================
+function installDailyPurgeScheduledQueries(
+  namespace: ExactNamespace,
+  stagingDataset: gcp.bigquery.Dataset,
+  transferServiceAgentPermission: gcp.projects.IAMMember,
+  retentionPeriodSeconds: number = 604800,  
+) {
+  const currentProject = gcp.organizations.getProjectOutput({});
+  const projectId = currentProject.apply(p => p.projectId!);
+  const schemaName = scanAppDatabaseName(namespace);
+  const retentionDays = retentionPeriodSeconds / 86400;
+
+  Object.entries(replicatedTables).forEach(([tableName, tableConfig]) => {
+    const timeExpression =
+      tableConfig.timeType === 'datastream_metadata'
+        ? `TIMESTAMP_MILLIS(datastream_metadata.${tableConfig.datePartitionColumn})`
+        : `TIMESTAMP_MICROS(${tableConfig.datePartitionColumn})`;
+
+    const procedureBody = pulumi
+      .all([projectId, stagingDataset.datasetId])
+      .apply(([proj, stagingDs]) => {
+        const stagingTable = `\`${proj}.${stagingDs}.${schemaName}_${tableName}\``;
+
+        return `
+          DELETE FROM ${stagingTable}
+          WHERE ${timeExpression} < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL ${retentionPeriodSeconds} SECOND);
+        `;
+      });
+
+    const routineId = `sp_purge_old_records_${tableName}`;
+
+    // Create the cleanup Stored Procedure
+    const purgeRoutine = new gcp.bigquery.Routine(`${tableName}-purge-routine`, {
+      datasetId: stagingDataset.datasetId,
+      routineId: routineId,
+      routineType: 'PROCEDURE',
+      language: 'SQL',
+      definitionBody: procedureBody,
+    });
+
+    // Schedule the Stored Procedure to run daily
+    new gcp.bigquery.DataTransferConfig(
+      `${CLUSTER_BASENAME}_${tableName}-daily-purge`,
+      {
+        displayName: `${CLUSTER_BASENAME}_${tableName} Daily Retention Purge`,
+        location: cloudsdkComputeRegion(),
+        serviceAccountName: pulumi.interpolate`bigquery@${projectId}.iam.gserviceaccount.com`,
+        dataSourceId: 'scheduled_query',
+        schedule: 'every day 05:21', // Runs daily at 05:21 AM 
 
+        params: {
+          query: pulumi.interpolate`CALL \`${projectId}.${stagingDataset.datasetId}.${routineId}\`();`,
+        },
+      },
+      {
+        dependsOn: [transferServiceAgentPermission, purgeRoutine],
+      }
+    );
+  });
+}
 // ============================================================================
 // CONNECTION PROFILES & NETWORKING
 // ============================================================================
@@ -775,6 +848,7 @@ export async function configureScanBigQuery({
     enableStagProdDatastream,
     legacyDesiredState,
     stagProdDesiredState,
+    retentionPeriodSeconds,
   } = bigQueryConfig;
 
   if (!enableLegacyDatastream && !enableStagProdDatastream) {
@@ -856,8 +930,9 @@ export async function configureScanBigQuery({
       slots.slot2,
       stagProdDesiredState
     );
-
-    installHourlyScheduledQueries(namespace, stagingDataset, prodDataset);
+    const transferServiceAgentPermission = installBqTransferServiceAgentPermission();
+    installHourlyScheduledQueries(namespace, stagingDataset, prodDataset, transferServiceAgentPermission);
+    installDailyPurgeScheduledQueries(namespace, stagingDataset, transferServiceAgentPermission, retentionPeriodSeconds);
   }
   // TODO (DACH-NY/canton-network-internal#6451) not sure if this function needs to return anything,
   // but we need to return something to satisfy the ScanBigQuery type.

cluster/pulumi/common-sv/src/singleSvConfig.ts

@@ -107,6 +107,12 @@ export const ScanBigQueryConfigSchema = z
     enableStagProdDatastream: z.boolean().default(false),
     legacyDesiredState: z.enum(['RUNNING', 'PAUSED']).default('RUNNING'),
     stagProdDesiredState: z.enum(['RUNNING', 'PAUSED']).default('RUNNING'),
+    retentionPeriodSeconds: z
+    .number()
+    .int()
+    .min(259200) // 3 days in seconds
+    .refine((v) => v % 86400 === 0) // enforces 24 hour day cut out
+    .default(604800), // 604800 = 7 days
   })
   .strict(); // Keeps strict mode safe now that all known fields are explicitly defined