address more comments

Author: icecrasher321

apps/sim/app/api/workspaces/[id]/fork/diff/route.ts

@@ -4,6 +4,7 @@ import { getForkDiffContract } from '@/lib/api/contracts/workspace-fork'
 import { parseRequest } from '@/lib/api/server'
 import { getSession } from '@/lib/auth'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { loadSourceDeployedStates } from '@/lib/workspaces/fork/copy/deploy-bridge'
 import { assertCanPromote } from '@/lib/workspaces/fork/lineage/authz'
 import { computeForkPromotePlan } from '@/lib/workspaces/fork/promote/promote-plan'
 
@@ -21,12 +22,17 @@ export const GET = withRouteHandler(
 
     const auth = await assertCanPromote(id, otherWorkspaceId, direction, session.user.id)
 
+    const { deployedWorkflows, sourceStates } = await loadSourceDeployedStates(
+      auth.sourceWorkspaceId
+    )
     const plan = await computeForkPromotePlan({
       executor: db,
       edge: auth.edge,
       sourceWorkspaceId: auth.sourceWorkspaceId,
       targetWorkspaceId: auth.targetWorkspaceId,
       direction,
+      deployedSourceWorkflows: deployedWorkflows,
+      sourceStates,
     })
 
     const toRef = (reference: (typeof plan.unmappedRequired)[number]) => ({

apps/sim/lib/workspaces/fork/copy/deploy-bridge.ts

@@ -57,6 +57,31 @@ export async function getActiveDeploymentVersionNumber(
   return row?.version ?? null
 }
 
+/**
+ * Read a source workspace's deployed workflows and each one's active deployed state
+ * on the global pool. Fork/promote callers MUST run this BEFORE opening their
+ * transaction: doing these heavy per-workflow reads inside the tx checks out a
+ * SECOND pooled connection while the tx holds the first, which can deadlock the
+ * pool at saturation (primary pool max is 15). The source is read-only for the
+ * operation, so a pre-transaction snapshot is the value that gets force-pushed.
+ *
+ * Holds every source state in memory at once (bounded by the workspace's deployed
+ * workflow count) - the apply step needs each state to write its target inside the
+ * single atomic transaction, so it cannot stream them one at a time.
+ */
+export async function loadSourceDeployedStates(sourceWorkspaceId: string): Promise<{
+  deployedWorkflows: DeployedWorkflowSummary[]
+  sourceStates: Map<string, WorkflowState>
+}> {
+  const deployedWorkflows = await listDeployedWorkflows(db, sourceWorkspaceId)
+  const sourceStates = new Map<string, WorkflowState>()
+  for (const wf of deployedWorkflows) {
+    const state = await readDeployedState(wf.id, sourceWorkspaceId)
+    if (state) sourceStates.set(wf.id, state)
+  }
+  return { deployedWorkflows, sourceStates }
+}
+
 /**
  * Read a workflow's active deployed state as a `WorkflowState`. Returns null ONLY
  * when the workflow genuinely has no active deployment (a legitimate skip); real

apps/sim/lib/workspaces/fork/create-fork.ts

@@ -17,7 +17,7 @@ import {
   copyWorkflowStateIntoTarget,
   resolveForkFolderMapping,
 } from '@/lib/workspaces/fork/copy/copy-workflows'
-import { listDeployedWorkflows, readDeployedState } from '@/lib/workspaces/fork/copy/deploy-bridge'
+import { loadSourceDeployedStates } from '@/lib/workspaces/fork/copy/deploy-bridge'
 import {
   type ForkMappingUpsert,
   type ForkResourceType,
@@ -28,7 +28,6 @@ import type { ForkRemapKind } from '@/lib/workspaces/fork/remap/remap-references
 import type { WorkspaceWithOwner } from '@/lib/workspaces/permissions/utils'
 import type { WorkspaceCreationPolicy } from '@/lib/workspaces/policy'
 import { WORKSPACE_MODE } from '@/lib/workspaces/policy'
-import type { WorkflowState } from '@/stores/workflows/workflow/types'
 
 const logger = createLogger('WorkspaceForkCreate')
 
@@ -88,16 +87,10 @@ export async function createFork(params: CreateForkParams): Promise<CreateForkRe
   const selection = params.selection ?? EMPTY_SELECTION
   const childName = params.name?.trim() || `${source.name} (fork)`
 
-  // Read the source's deployed workflows + states BEFORE the transaction. These are
-  // global-pool reads of the (unchanged) source; issuing them inside the fork tx
-  // would trip the cross-connection tripwire (and check out a second pooled
-  // connection that can deadlock at saturation).
-  const deployedWorkflows = await listDeployedWorkflows(db, source.id)
-  const sourceStates = new Map<string, WorkflowState>()
-  for (const wf of deployedWorkflows) {
-    const state = await readDeployedState(wf.id, source.id)
-    if (state) sourceStates.set(wf.id, state)
-  }
+  // Read the source's deployed workflows + states BEFORE the transaction so these
+  // global-pool reads don't check out a second pooled connection from inside the
+  // fork tx (which can deadlock the pool at saturation).
+  const { deployedWorkflows, sourceStates } = await loadSourceDeployedStates(source.id)
 
   const { result, blobTasks, contentPlan } = await db.transaction(async (tx) => {
     const now = new Date()

apps/sim/lib/workspaces/fork/lineage/lineage.ts

@@ -87,10 +87,26 @@ export async function resolveForkEdge(
 
 /**
  * Serialize concurrent promote/rollback on a fork edge with a transaction-scoped
- * advisory lock keyed by the edge (the child workspace id). `hashtext` is 32-bit,
- * so distinct edges can share a lock slot - at worst unnecessary serialization,
- * never a correctness issue.
+ * advisory lock keyed by the edge (the child workspace id). `hashtextextended`
+ * (64-bit, matching every other advisory lock in the repo) makes a collision
+ * between distinct keys astronomically unlikely; a collision would only cause
+ * unnecessary serialization, never a correctness issue.
  */
 export async function acquireForkEdgeLock(tx: DbOrTx, childWorkspaceId: string): Promise<void> {
-  await tx.execute(sql`select pg_advisory_xact_lock(hashtext(${`fork-edge:${childWorkspaceId}`}))`)
+  await tx.execute(
+    sql`select pg_advisory_xact_lock(hashtextextended(${`fork-edge:${childWorkspaceId}`}, 0))`
+  )
+}
+
+/**
+ * Serialize every promote/rollback whose TARGET is this workspace. Sibling forks
+ * promote into the same parent on different edge locks, so the edge lock alone does
+ * not serialize them; this lock does, keeping concurrent syncs into one target from
+ * interleaving and keeping rollback's "newest sync" check race-free. Always acquire
+ * this BEFORE {@link acquireForkEdgeLock} so the two are taken in a consistent order.
+ */
+export async function acquireForkTargetLock(tx: DbOrTx, targetWorkspaceId: string): Promise<void> {
+  await tx.execute(
+    sql`select pg_advisory_xact_lock(hashtextextended(${`fork-target:${targetWorkspaceId}`}, 0))`
+  )
 }

apps/sim/lib/workspaces/fork/promote/promote-plan.ts

@@ -2,7 +2,7 @@ import { workflow } from '@sim/db/schema'
 import { generateId } from '@sim/utils/id'
 import { and, eq, isNull } from 'drizzle-orm'
 import type { DbOrTx } from '@/lib/db/types'
-import { listDeployedWorkflows, readDeployedState } from '@/lib/workspaces/fork/copy/deploy-bridge'
+import type { DeployedWorkflowSummary } from '@/lib/workspaces/fork/copy/deploy-bridge'
 import type { ForkEdge } from '@/lib/workspaces/fork/lineage/lineage'
 import { detectForkCascadeReferences } from '@/lib/workspaces/fork/mapping/cascade'
 import { buildForkResolver, getEdgeMappingRows } from '@/lib/workspaces/fork/mapping/mapping-store'
@@ -13,6 +13,7 @@ import {
   type ForkReferenceResolver,
   scanWorkflowReferences,
 } from '@/lib/workspaces/fork/remap/remap-references'
+import type { WorkflowState } from '@/stores/workflows/workflow/types'
 
 export interface ForkPromotePlanItem {
   sourceWorkflowId: string
@@ -100,8 +101,23 @@ export async function computeForkPromotePlan(params: {
   sourceWorkspaceId: string
   targetWorkspaceId: string
   direction: 'push' | 'pull'
+  /**
+   * Source deployed workflows + their states, read by the caller BEFORE its
+   * transaction (see `loadSourceDeployedStates`) so the plan never checks out a
+   * second pooled connection from inside a tx.
+   */
+  deployedSourceWorkflows: DeployedWorkflowSummary[]
+  sourceStates: Map<string, WorkflowState>
 }): Promise<ForkPromotePlan> {
-  const { executor, edge, sourceWorkspaceId, targetWorkspaceId, direction } = params
+  const {
+    executor,
+    edge,
+    sourceWorkspaceId,
+    targetWorkspaceId,
+    direction,
+    deployedSourceWorkflows,
+    sourceStates,
+  } = params
 
   const mappingRows = await getEdgeMappingRows(executor, edge.childWorkspaceId)
   const [targetEnvKeys, sourceEnvKeys] = await Promise.all([
@@ -118,8 +134,7 @@ export async function computeForkPromotePlan(params: {
     else identityMap.set(row.childResourceId, row.parentResourceId)
   }
 
-  const [deployedSourceWorkflows, targetWorkflows, sourceWorkflowRows] = await Promise.all([
-    listDeployedWorkflows(executor, sourceWorkspaceId),
+  const [targetWorkflows, sourceWorkflowRows] = await Promise.all([
     executor
       .select({ id: workflow.id, name: workflow.name, updatedAt: workflow.updatedAt })
       .from(workflow)
@@ -138,13 +153,12 @@ export async function computeForkPromotePlan(params: {
   // parent's originals; undeployed sources are simply skipped (target left as-is).
   const existingSourceIds = new Set(sourceWorkflowRows.map((w) => w.id))
 
-  // Build the items and scan references in one pass, loading each source's deployed
-  // state, scanning it, then discarding it - peak memory stays at one workflow state
-  // (the deployed-state cache, bounded globally, retains originals for the copy loop).
+  // Build the items and scan references in one pass from the pre-read source states
+  // (loaded before the caller's transaction; see loadSourceDeployedStates).
   const items: ForkPromotePlanItem[] = []
   const referenceByKey = new Map<string, ForkReference>()
   for (const source of deployedSourceWorkflows) {
-    const sourceState = await readDeployedState(source.id, sourceWorkspaceId)
+    const sourceState = sourceStates.get(source.id)
     if (!sourceState) continue
 
     const mappedTargetId = identityMap.get(source.id)
@@ -180,12 +194,14 @@ export async function computeForkPromotePlan(params: {
     items,
   })
 
+  const writtenTargetIds = new Set(items.map((item) => item.targetWorkflowId))
   const archivedTargetIds: string[] = []
   for (const row of mappingRows) {
     if (row.resourceType !== 'workflow' || row.childResourceId == null) continue
     const mappedSourceId = sourceIsParent ? row.parentResourceId : row.childResourceId
     const mappedTargetId = sourceIsParent ? row.childResourceId : row.parentResourceId
     if (existingSourceIds.has(mappedSourceId)) continue
+    if (writtenTargetIds.has(mappedTargetId)) continue
     if (targetActiveIds.has(mappedTargetId)) archivedTargetIds.push(mappedTargetId)
   }
   const archivedTargets = archivedTargetIds.map((id) => ({

apps/sim/lib/workspaces/fork/promote/promote-run-store.ts

@@ -100,49 +100,29 @@ export async function upsertPromoteRun(
   return id
 }
 
-/** Remove one direction's undo point (keyed by target) after a successful rollback. */
-export async function deletePromoteRun(
+/**
+ * Remove EVERY undo point targeting this workspace. Called after a rollback so the
+ * undo is single-level: only the latest sync into a target is ever undoable, and
+ * once it is undone there is no stack of older syncs to walk back into.
+ */
+export async function deleteAllPromoteRunsForTarget(
   tx: DbOrTx,
-  childWorkspaceId: string,
   targetWorkspaceId: string
 ): Promise<void> {
   await tx
     .delete(workspaceForkPromoteRun)
-    .where(
-      and(
-        eq(workspaceForkPromoteRun.childWorkspaceId, childWorkspaceId),
-        eq(workspaceForkPromoteRun.targetWorkspaceId, targetWorkspaceId)
-      )
-    )
+    .where(eq(workspaceForkPromoteRun.targetWorkspaceId, targetWorkspaceId))
 }
 
 /**
- * The undo point targeting this workspace, with the edge counterpart needed to
- * call rollback. `sourceWorkspaceId` is the "other" workspace the promote came
- * from (rollback resolves the edge from target + other).
+ * The newest undo point targeting this workspace. A workspace can be the target of
+ * several edges (pushes from its children, a pull from its parent), so order by
+ * recency: this is the ONLY undoable sync - older ones are stale the moment a newer
+ * sync lands, and rollback refuses them.
  */
-export async function getUndoableRunForTarget(
+export async function getLatestPromoteRunForTarget(
   executor: DbOrTx,
   targetWorkspaceId: string
-): Promise<{ sourceWorkspaceId: string; direction: 'push' | 'pull' } | null> {
-  const [row] = await executor
-    .select({
-      sourceWorkspaceId: workspaceForkPromoteRun.sourceWorkspaceId,
-      direction: workspaceForkPromoteRun.direction,
-    })
-    .from(workspaceForkPromoteRun)
-    .where(eq(workspaceForkPromoteRun.targetWorkspaceId, targetWorkspaceId))
-    // A workspace can be the target of several edges; surface the most recent.
-    .orderBy(desc(workspaceForkPromoteRun.createdAt))
-    .limit(1)
-  return row ?? null
-}
-
-/** The undo point whose target is this workspace and whose edge counterpart is `otherWorkspaceId`. */
-export async function getPromoteRunForRollback(
-  executor: DbOrTx,
-  targetWorkspaceId: string,
-  childWorkspaceId: string
 ): Promise<PromoteRunRow | null> {
   const [row] = await executor
     .select({
@@ -155,13 +135,22 @@ export async function getPromoteRunForRollback(
       createdAt: workspaceForkPromoteRun.createdAt,
     })
     .from(workspaceForkPromoteRun)
-    .where(
-      and(
-        eq(workspaceForkPromoteRun.childWorkspaceId, childWorkspaceId),
-        eq(workspaceForkPromoteRun.targetWorkspaceId, targetWorkspaceId)
-      )
-    )
+    .where(eq(workspaceForkPromoteRun.targetWorkspaceId, targetWorkspaceId))
+    .orderBy(desc(workspaceForkPromoteRun.createdAt))
     .limit(1)
   if (!row) return null
   return { ...row, snapshot: row.snapshot as PromoteRunSnapshot }
 }
+
+/**
+ * The "other" workspace and direction of the latest sync into this target, for the
+ * UI's undo affordance. `sourceWorkspaceId` is the workspace the sync came from
+ * (rollback resolves the edge from target + other).
+ */
+export async function getUndoableRunForTarget(
+  executor: DbOrTx,
+  targetWorkspaceId: string
+): Promise<{ sourceWorkspaceId: string; direction: 'push' | 'pull' } | null> {
+  const run = await getLatestPromoteRunForTarget(executor, targetWorkspaceId)
+  return run ? { sourceWorkspaceId: run.sourceWorkspaceId, direction: run.direction } : null
+}

apps/sim/lib/workspaces/fork/promote/promote.ts

@@ -12,9 +12,13 @@ import {
 } from '@/lib/workspaces/fork/copy/copy-workflows'
 import {
   getActiveDeploymentVersionNumber,
-  readDeployedState,
+  loadSourceDeployedStates,
 } from '@/lib/workspaces/fork/copy/deploy-bridge'
-import { acquireForkEdgeLock, type ForkEdge } from '@/lib/workspaces/fork/lineage/lineage'
+import {
+  acquireForkEdgeLock,
+  acquireForkTargetLock,
+  type ForkEdge,
+} from '@/lib/workspaces/fork/lineage/lineage'
 import {
   type ForkMappingUpsert,
   upsertEdgeMappings,
@@ -101,7 +105,17 @@ export async function promoteFork(params: PromoteForkParams): Promise<PromoteFor
 
   const targetMembers = (await getUsersWithPermissions(targetWorkspaceId)).map((m) => m.userId)
 
+  // Read the source's deployed workflows + states BEFORE the transaction so these
+  // heavy per-workflow reads never check out a second pooled connection from inside
+  // the promote tx (which can deadlock the pool at saturation). The source is
+  // read-only here, so this pre-tx snapshot is exactly what gets force-pushed.
+  const { deployedWorkflows, sourceStates } = await loadSourceDeployedStates(sourceWorkspaceId)
+
   const txResult: PromoteTxBlocked | PromoteTxApplied = await db.transaction(async (tx) => {
+    // Target lock before edge lock (consistent ordering): the target lock serializes
+    // every sync into this target so sibling forks can't interleave writes, and so
+    // rollback's "newest sync" check stays race-free against a concurrent promote.
+    await acquireForkTargetLock(tx, targetWorkspaceId)
     await acquireForkEdgeLock(tx, edge.childWorkspaceId)
 
     const plan = await computeForkPromotePlan({
@@ -110,6 +124,8 @@ export async function promoteFork(params: PromoteForkParams): Promise<PromoteFor
       sourceWorkspaceId,
       targetWorkspaceId,
       direction,
+      deployedSourceWorkflows: deployedWorkflows,
+      sourceStates,
     })
 
     if (plan.unmappedRequired.length > 0) {
@@ -143,12 +159,10 @@ export async function promoteFork(params: PromoteForkParams): Promise<PromoteFor
     const createdTargetIds: string[] = []
     const writtenItems: typeof plan.items = []
     for (const item of plan.items) {
-      // Re-read the source's deployed state one workflow at a time so peak memory
-      // stays at a single workflow state. Only items actually written below feed
-      // the snapshot, identity rows, and deploy list - a source that lost its
-      // active deployment between plan and copy is skipped cleanly (no phantom
-      // mapping/deploy of a never-created target).
-      const sourceState = await readDeployedState(item.sourceWorkflowId, sourceWorkspaceId)
+      // Use the pre-read source state (loaded above, before the tx). An item only
+      // exists when its state was present at read time, so this lookup hits; the
+      // guard stays as defense so the written counts below never over-report.
+      const sourceState = sourceStates.get(item.sourceWorkflowId)
       if (!sourceState) continue
       if (item.mode === 'replace') {
         const priorVersion = await getActiveDeploymentVersionNumber(tx, item.targetWorkflowId)