fix(audit): cap rollup writes process-wide and name the run on credential access entries (#9407)

* fix(audit): cap rollup writes process-wide and name the run on credential access entries

* fix(audit): record Vertex service-account credential access

Author: waleedlatif1

apps/docs/content/docs/platform/enterprise/audit-logs.mdx

@@ -105,7 +105,7 @@ Events use identifiers such as `resource.action`. These categories are represent
 
 Authentication failures and password-reset requests use an anonymous actor and record the target account separately. Impersonation and administrative session revocations identify the operator and target organization. Table-row mutations record summaries and counts without copying row contents.
 
-When a workflow run makes the same kind of change to a table many times in quick succession, such as inserting rows in a loop, the repeats are merged into one `table.updated` entry. The merged entry keeps every row ID in `metadata.rowIds`, sums the counts, and records in `metadata.rollup` how many changes it covers and when the first and last happened. The actor of any workflow-made entry names the workflow (`metadata.actor.workflowId`) and the run (`metadata.actor.executionId`). Credential use by a workflow run is merged the same way: one `credential.accessed` entry per credential, counting the repeats.
+When a workflow run makes the same kind of change to a table many times in quick succession, such as inserting rows in a loop, the repeats are merged into one `table.updated` entry. The merged entry keeps every row ID in `metadata.rowIds`, sums the counts, and records in `metadata.rollup` how many changes it covers and when the first and last happened. The actor of any workflow-made entry names the workflow (`metadata.actor.workflowId`) and the run (`metadata.actor.executionId`). Credential use by a workflow run is merged the same way: one `credential.accessed` entry per credential, counting the repeats, that names the workflow (`metadata.workflowId`) and the run (`metadata.executionId`).
 
 Most audit writes are best effort: an outage or process loss can prevent persistence after an action succeeds. Deployment activation uses an awaited, idempotent insert before its outbox checkpoint. The feed is not a complete tool-call stream or a guaranteed transactional ledger for every operation.
 

apps/sim/executor/handlers/agent/handler.ts

@@ -2988,6 +2988,7 @@ export class AgentBlockHandler implements BlockHandler {
           actingUserId: ctx.userId,
           workspaceId: ctx.workspaceId,
           workflowId: ctx.workflowId,
+          executionId: ctx.executionId,
           callerLabel: 'vertex-agent',
         })
       }

apps/sim/executor/handlers/evaluator/handler.ts

@@ -185,6 +185,7 @@ export class EvaluatorBlockHandler implements BlockHandler {
         actingUserId: ctx.userId,
         workspaceId: ctx.workspaceId,
         workflowId: ctx.workflowId,
+        executionId: ctx.executionId,
         callerLabel: 'vertex-evaluator',
       })
     }

apps/sim/executor/handlers/router/handler.ts

@@ -120,6 +120,7 @@ export class RouterBlockHandler implements BlockHandler {
           actingUserId: ctx.userId,
           workspaceId: ctx.workspaceId,
           workflowId: ctx.workflowId,
+          executionId: ctx.executionId,
           callerLabel: 'vertex-router',
         })
       }
@@ -289,6 +290,7 @@ export class RouterBlockHandler implements BlockHandler {
           actingUserId: ctx.userId,
           workspaceId: ctx.workspaceId,
           workflowId: ctx.workflowId,
+          executionId: ctx.executionId,
           callerLabel: 'vertex-router',
         })
       }

apps/sim/executor/utils/credential-token.ts

@@ -22,6 +22,7 @@ export interface ResolveExecutorCredentialTokenParams {
   credentialId: string
   userId?: string
   workflowId?: string
+  executionId?: string
   /** Tool consuming the token; required by the managed-OAuth scope policy. */
   toolId?: string
   /** Display label for the thrown failure ("Failed to obtain credential for X: ..."). */
@@ -112,6 +113,7 @@ export async function resolveExecutorCredentialToken(
     requestId,
     credentialId,
     workflowId,
+    executionId: params.executionId,
     toolId,
     scopes: params.scopes,
     impersonateEmail: params.impersonateEmail,

apps/sim/executor/utils/vertex-credential.ts

@@ -1,6 +1,7 @@
 import { createLogger } from '@sim/logger'
 import { canUseCredential, getCredentialActorContext } from '@/lib/credentials/access'
 import { getServiceAccountToken } from '@/lib/oauth/credential-service'
+import { recordCredentialAccess } from '@/lib/oauth/token-resolution'
 import { resolveExecutorCredentialToken } from '@/executor/utils/credential-token'
 
 const logger = createLogger('VertexCredential')
@@ -12,6 +13,7 @@ export interface ResolveVertexCredentialParams {
   workspaceId: string | null | undefined
   /** Pins the token request to this workflow's workspace. */
   workflowId?: string
+  executionId?: string
   callerLabel?: string
 }
 
@@ -27,6 +29,7 @@ export async function resolveVertexCredential({
   actingUserId,
   workspaceId,
   workflowId,
+  executionId,
   callerLabel = 'vertex',
 }: ResolveVertexCredentialParams): Promise<string> {
   const requestId = `${callerLabel}-${Date.now()}`
@@ -57,6 +60,15 @@ export async function resolveVertexCredential({
     const accessToken = await getServiceAccountToken(cred.id, [
       'https://www.googleapis.com/auth/cloud-platform',
     ])
+    recordCredentialAccess({
+      actorId: actingUserId,
+      workspaceId: cred.workspaceId,
+      resourceId: cred.id,
+      providerId: cred.providerId,
+      credentialType: 'service_account',
+      workflowId,
+      executionId,
+    })
     logger.info(`[${requestId}] Successfully resolved Vertex AI service account credential`)
     return accessToken
   }
@@ -70,6 +82,7 @@ export async function resolveVertexCredential({
     credentialId,
     userId: actingUserId,
     workflowId,
+    executionId,
     toolLabel: 'Vertex AI',
   })
 

apps/sim/lib/oauth/token-resolution.ts

@@ -58,6 +58,8 @@ export interface ResolveCredentialTokenInput {
   requestId: string
   credentialId?: string
   workflowId?: string
+  /** Workflow run the token is for; recorded on the credential-access audit entry. */
+  executionId?: string
   /** Canonical provider scopes, used only by service-account token minting. */
   scopes?: string[]
   /** Google domain-wide-delegation subject for service-account credentials. */
@@ -112,9 +114,12 @@ export function recordCredentialAccess(params: {
   resourceId: string
   providerId: string | null | undefined
   credentialType: 'oauth' | 'service_account'
+  workflowId?: string
+  executionId?: string
   auditRequest?: CredentialAuditRequest
 }): void {
-  const { actorId, workspaceId, resourceId, providerId, credentialType } = params
+  const { actorId, workspaceId, resourceId, providerId, credentialType, workflowId, executionId } =
+    params
   const credentialLabel = `${credentialType === 'oauth' ? 'OAuth' : 'service account'} credential for provider ${providerId ?? 'unknown'}`
   const audit = {
     workspaceId,
@@ -126,6 +131,8 @@ export function recordCredentialAccess(params: {
     metadata: {
       provider: providerId,
       credentialType,
+      ...(workflowId ? { workflowId } : {}),
+      ...(executionId ? { executionId } : {}),
     },
   }
   if (params.auditRequest) {
@@ -211,9 +218,20 @@ export async function completeOAuthCredentialToken(params: {
   resolvedCredentialId: string
   actorId?: string
   workspaceId: string | null
+  workflowId?: string
+  executionId?: string
   auditRequest?: CredentialAuditRequest
 }): Promise<ResolveCredentialTokenResult> {
-  const { requestId, credential, resolvedCredentialId, actorId, workspaceId, auditRequest } = params
+  const {
+    requestId,
+    credential,
+    resolvedCredentialId,
+    actorId,
+    workspaceId,
+    workflowId,
+    executionId,
+    auditRequest,
+  } = params
   const contextValidation = validateOAuthCredentialContext(credential)
   if (!contextValidation.ok) {
     return { ok: false, status: 401, error: contextValidation.error }
@@ -229,6 +247,8 @@ export async function completeOAuthCredentialToken(params: {
         resourceId: resolvedCredentialId,
         providerId: credential.providerId,
         credentialType: 'oauth',
+        workflowId,
+        executionId,
         auditRequest,
       })
     }
@@ -301,6 +321,7 @@ export async function resolveCredentialToken(
     requestId,
     credentialId,
     workflowId,
+    executionId,
     scopes,
     impersonateEmail,
     callerUserId,
@@ -355,6 +376,8 @@ export async function resolveCredentialToken(
             resourceId: resolved.credentialId,
             providerId: resolved.providerId,
             credentialType: 'service_account',
+            workflowId,
+            executionId,
             auditRequest,
           })
         }
@@ -433,6 +456,8 @@ export async function resolveCredentialToken(
       resolvedCredentialId,
       actorId: authz.requesterUserId,
       workspaceId: authz.workspaceId ?? null,
+      workflowId,
+      executionId,
       auditRequest,
     })
   } catch (error) {
@@ -477,6 +502,7 @@ export async function resolveCredentialAccessToken(
       requestId,
       credentialId,
       workflowId: input.workflowId,
+      executionId: input.executionId,
       scopes: input.scopes,
       /**
        * In-process callers forward raw subblock state, where an untouched

apps/sim/tools/index.ts

@@ -1931,6 +1931,7 @@ async function executeToolImplementation(
                 credentialId,
                 userId,
                 workflowId,
+                executionId: scope.executionId,
                 toolId,
                 toolLabel,
                 scopes: providerScopes,

packages/audit/src/rollup.ts

@@ -13,7 +13,10 @@ const ROLLUP_MAX_ENTRIES = 100
 /** Pending rollups held by the process; reaching it writes every one of them. */
 const ROLLUP_MAX_PENDING = 1_000
 
-/** Rows inserted at once by one write, so a large flush never floods the pool. */
+/**
+ * Rollup rows the process inserts at once, across every window timer, size
+ * flush, and scope end, so a burst of distinct rollups never floods the pool.
+ */
 const ROLLUP_WRITE_CONCURRENCY = 10
 
 export interface AuditRollupSummary {
@@ -67,6 +70,8 @@ interface PendingRollup {
 }
 
 const pending = new Map<string, PendingRollup>()
+const queuedWrites: Array<{ rollup: PendingRollup; written: () => void }> = []
+let activeWrites = 0
 const scopeStorage = new AsyncLocalStorage<RollupScope>()
 let nextScopeId = 0
 
@@ -196,11 +201,31 @@ function toAuditParams(rollup: PendingRollup): AuditLogParams {
 
 /**
  * One insert per row, so a row that cannot be written never takes others with
- * it, at most {@link ROLLUP_WRITE_CONCURRENCY} at a time.
+ * it. Rows wait in one process-wide queue, at most
+ * {@link ROLLUP_WRITE_CONCURRENCY} in flight.
  */
-async function writePending(rollups: PendingRollup[]): Promise<void> {
-  for (let index = 0; index < rollups.length; index += ROLLUP_WRITE_CONCURRENCY) {
-    await Promise.all(rollups.slice(index, index + ROLLUP_WRITE_CONCURRENCY).map(writeRollup))
+function writePending(rollups: PendingRollup[]): Promise<void> {
+  return Promise.all(
+    rollups.map(
+      (rollup) =>
+        new Promise<void>((written) => {
+          queuedWrites.push({ rollup, written })
+          drainWrites()
+        })
+    )
+  ).then(() => undefined)
+}
+
+function drainWrites(): void {
+  while (activeWrites < ROLLUP_WRITE_CONCURRENCY) {
+    const next = queuedWrites.shift()
+    if (!next) return
+    activeWrites += 1
+    void writeRollup(next.rollup).finally(() => {
+      activeWrites -= 1
+      next.written()
+      drainWrites()
+    })
   }
 }