Class SnapshotService

apps/sim/lib/logs/execution/snapshot/service.ts:18–161

const logger = createLogger('SnapshotService')

export class SnapshotService implements ISnapshotService {
  async createSnapshot(
    workflowId: string,
    state: WorkflowState
  ): Promise<WorkflowExecutionSnapshot> {
    const result = await this.createSnapshotWithDeduplication(workflowId, state)
    return result.snapshot
  }

  async createSnapshotWithDeduplication(
    workflowId: string,
    state: WorkflowState
  ): Promise<SnapshotCreationResult> {
    const stateHash = this.computeStateHash(state)

    const snapshotData: WorkflowExecutionSnapshotInsert = {
      id: generateId(),
      workflowId,
      stateHash,
      stateData: state,
    }

    /**
     * Insert the snapshot, or — when an identical (workflowId, stateHash) row
     * already exists — return it without rewriting the large stateData jsonb.
     *
     * The hash is a sha256 of the normalized state, so an existing row's stateData
     * is byte-identical; there is nothing to update. The previous implementation
     * SET state_data on conflict, which rewrote the full (tens-of-KB) jsonb every
     * run. We keep a single atomic upsert — so RETURNING always yields the row and
     * there is no race with snapshot cleanup (unlike DO NOTHING + a follow-up
     * select) — but SET only the small state_hash column to itself. Under Postgres
     * MVCC the unchanged, TOASTed stateData is not rewritten: its existing
     * out-of-line storage is reused, so the per-execution write drops from the
     * full blob to a tiny heap tuple.
     */
    const [upsertedSnapshot] = await db
      .insert(workflowExecutionSnapshots)
      .values(snapshotData)
      .onConflictDoUpdate({
        target: [workflowExecutionSnapshots.workflowId, workflowExecutionSnapshots.stateHash],
        set: {
          stateHash: sql`excluded.state_hash`,
        },
      })
      .returning()

    const isNew = upsertedSnapshot.id === snapshotData.id

    logger.info(
      isNew
        ? `Created new snapshot for workflow ${workflowId} (hash: ${stateHash.slice(0, 12)}..., blocks: ${Object.keys(state.blocks || {}).length})`
        : `Reusing existing snapshot for workflow ${workflowId} (hash: ${stateHash.slice(0, 12)}...)`
    )

    return {
      snapshot: {
        ...upsertedSnapshot,