From b25816fde9861bba599947a7d95831091570c71d Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Tue, 18 Aug 2026 13:48:08 -0700 Subject: [PATCH 1/4] fix(forks): stop copying connector-managed knowledge base documents A fork copies a KB's documents but never its connectors, so a connector-sourced document arrives with `connector_id` nulled and its `external_id` intact. The sync engine keys every existing/tombstone/ exclusion lookup off `connector_id`, so that copy is invisible to it - never updated, reconciled, or purged - and `doc_connector_external_id_idx` does not constrain it either, since its `connector_id` is NULL. Attaching a connector in the child then re-ingests every page as a NEW row on top of the snapshot. Each fork hop re-copies the previous hop's orphans and adds one more generation, so a prod -> UAT -> staging chain leaves three rows per page and a knowledge search returns the same page three times, one of them serving content frozen at the fork date. Exclude connector-managed documents from all four doors a document can enter a fork through: the whole-KB content copy, the in-transaction placeholder pre-creation, the sync-only copy into an already-mapped KB, and the content fill (guarded for payloads planned by a pre-change worker mid-rollout). The placeholder path matters as much as the copy loop - filtering only the content phase would leave a permanently archived row behind a persisted `knowledge_document` mapping. Skipped on both sides, the reference clears like any other uncopied document's. A document whose connector was deleted already has a null `connector_id` (the FK is ON DELETE SET NULL) and is static in the source too, so it still copies. One count(*) per copied KB logs what was left behind, since a fully connector-synced KB now forks to zero documents. Co-Authored-By: Claude Opus 5 (1M context) --- .../docs/en/platform/enterprise/forks.mdx | 23 +++- .../lib/copy/copy-resources.test.ts | 113 +++++++++++++++++- .../lib/copy/copy-resources.ts | 64 +++++++++- 3 files changed, 191 insertions(+), 9 deletions(-) diff --git a/apps/docs/content/docs/en/platform/enterprise/forks.mdx b/apps/docs/content/docs/en/platform/enterprise/forks.mdx index 7185eb39627..ebaa22fb6ea 100644 --- a/apps/docs/content/docs/en/platform/enterprise/forks.mdx +++ b/apps/docs/content/docs/en/platform/enterprise/forks.mdx @@ -174,7 +174,7 @@ How each resource behaves at **fork** time vs **sync** time. Use this when you a | [Excluded workflows](#excluded-workflows) | Never | Never — not sent, not overwritten, not archived | | Files | Optional copy (default on) | Map or copy | | Tables | Optional copy (default on) | Map or copy | -| Knowledge bases (+ documents) | Optional copy; referenced docs come with the KB | Map or copy; documents follow the KB | +| Knowledge bases (+ documents) | Optional copy; uploaded documents come with the KB, [connector-synced ones do not](#connector-synced-documents-are-not-copied) | Map or copy; documents follow the KB | | Custom tools | Optional copy (default on) | Map or copy | | Skills | Optional copy (default on) | Map or copy | | External MCP servers | Optional copy (config only; sign-in cleared) | Map or copy (config only; sign-in cleared) | @@ -227,10 +227,27 @@ Only **deployed** workflows move. Deploy is the commit; sync is the force push/p | | Behavior | |---|----------| -| **Fork** | Optional copy (default on). Tag definitions come with the knowledge base. Documents that the forked workflows actually reference are included. Deselect → knowledge base / document fields clear. | +| **Fork** | Optional copy (default on). Tag definitions come with the knowledge base, along with every **uploaded** document in it. Deselect → knowledge base / document fields clear. | | **Sync** | Map or copy the knowledge base. Documents are not mapped by themselves — they follow the knowledge base (copied with it, or re-picked when you map to an existing one). | -**Example:** An agent searches knowledge base “Product docs.” Fork with that knowledge base selected → the child gets the base, tags, and the documents the agent used. On sync, mapping to the child’s existing “Product docs” means re-picking which document the tool should use. +**Example:** An agent searches knowledge base “Product docs.” Fork with that knowledge base selected → the child gets the base, tags, and the uploaded documents. On sync, mapping to the child’s existing “Product docs” means re-picking which document the tool should use. + +#### Connector-synced documents are not copied + +Connectors themselves never cross a fork edge — the child gets no Confluence, Notion, Google Drive, or other sync running against it. Documents that a **connector** put in the knowledge base are therefore not copied either. Only documents you **uploaded** come across. + + + Fork a knowledge base whose content is entirely connector-synced and the child gets the base, its tags, and its settings — but **no documents**. Add the connector in the child to fill it. + + +This is deliberate. A copied connector document would arrive detached from any connector, so nothing would ever update, re-sync, or remove it — and when you added the connector in the child it would ingest every page again *alongside* the stale copy. Chain a few forks (prod → UAT → staging) and each hop leaves another dead generation behind, so one page comes back several times in a single knowledge search. Skipping them keeps the child’s own connector the single owner of that content. + +| To get connector content into the child | Do this | +|---|---| +| Keep it live | Add the same connector in the child and let it sync. It re-ingests everything, so nothing is lost. | +| Keep a frozen snapshot | Download the documents from the source and upload them to the child’s knowledge base — uploaded documents copy on every later fork. | + +A document whose connector was **deleted** in the source is no longer connector-managed, so it copies like any other uploaded document. --- diff --git a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts index b965aa0654e..5d1f73e39dc 100644 --- a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts +++ b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts @@ -6,7 +6,9 @@ import { folder as folderTable } from '@sim/db/schema' import { sha256Hex } from '@sim/security/hash' import { dbChainMockFns, + flattenMockConditions, resetDbChainMock, + schemaMock, storageServiceMock, storageServiceMockFns, } from '@sim/testing' @@ -311,6 +313,28 @@ describe('copyForkResourceContent', () => { expect(mockPersistCopiedResourceMappings).not.toHaveBeenCalled() }) + it('never copies a connector-managed document out of the source knowledge base', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([]) + + const result = await copyForkResourceContent({ + contentPlan: basePlan({ + knowledgeBases: [{ sourceId: 'src-kb', childId: 'child-kb', documentIdMap: {} }], + }), + requestId: 'test', + }) + + expect(result).toEqual({ copied: 1, failed: 0, failures: [] }) + // The row queue returns whatever is enqueued regardless of the predicate, so the exclusion + // is only observable in the condition tree. Pinned to the column so the assertion keeps its + // meaning if another nullable filter joins the same clause. + const pageWhere = dbChainMockFns.where.mock.calls.at(-1)?.[0] + expect( + flattenMockConditions(pageWhere).some( + (node) => node.type === 'isNull' && node.column === schemaMock.document.connectorId + ) + ).toBe(true) + }) + it('uses the blob content digest so a retry cannot adopt an older failed snapshot', async () => { dbChainMockFns.limit .mockResolvedValueOnce([sourceDoc]) @@ -1051,6 +1075,25 @@ describe('copyForkResourceContent', () => { }) }) + it('U-docs: refuses a connector-managed source planned before the exclusion existed', async () => { + // A payload queued by a pre-change worker during a rolling deploy: the planner would no + // longer emit this entry, so the fill must drop the placeholder rather than detach a copy + // of a connector-managed document into the existing target KB. + dbChainMockFns.limit + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([{ ...sourceDoc, connectorId: 'connector-1' }]) + + const result = await copyForkResourceContent({ + contentPlan: mappedDocumentPlan(), + requestId: 'test', + }) + + expect(result.copied).toBe(0) + expect(result.failures).toEqual([{ kind: 'knowledge-document', childId: 'child-doc-1' }]) + expect(storageServiceMockFns.mockDownloadFile).not.toHaveBeenCalled() + expect(mockIncrementStorageUsageInTx).not.toHaveBeenCalled() + }) + it('U-docs: refuses to charge when the target knowledge base moved workspaces', async () => { queueMappedDocumentCopy() dbChainMockFns.for.mockResolvedValueOnce([{ workspaceId: 'other-workspace' }]) @@ -1359,10 +1402,12 @@ describe('copyForkResourceContainers knowledge-base tag definitions', () => { // would make every source folder look already-present and suppress the mirroring. let folderCall = 0 const inserts: Array>> = [] + const wheres: Array<{ table: unknown; condition: unknown }> = [] const tx = { select: () => ({ from: (table: unknown) => ({ - where: () => { + where: (condition: unknown) => { + wheres.push({ table, condition }) if (table === folderTable) { return Promise.resolve(folderCall++ === 0 ? sourceFolders : []) } @@ -1377,7 +1422,7 @@ describe('copyForkResourceContainers knowledge-base tag definitions', () => { }, }), } - return { tx: tx as unknown as DbOrTx, inserts } + return { tx: tx as unknown as DbOrTx, inserts, wheres } } const kbSelection = { @@ -1458,6 +1503,35 @@ describe('copyForkResourceContainers knowledge-base tag definitions', () => { expect(inserts).toHaveLength(1) }) + it('does not pre-create a placeholder for a referenced connector-managed document', async () => { + const { tx, wheres } = makeKbTx([[sourceBase], [], []]) + + const result = await copyForkResourceContainers({ + tx, + sourceWorkspaceId: 'src-ws', + childWorkspaceId: 'child-ws', + userId: 'user-1', + now: new Date(), + selection: kbSelection, + workflowIdMap: new Map(), + referencedDocumentIds: ['doc-1'], + documentMappingContext: { edgeChildWorkspaceId: 'child-ws', sourceIsParent: true }, + }) + + // Must agree with the content phase's exclusion: a placeholder with no content copy behind + // it would stay archived forever while its persisted mapping pointed at it. + const placeholderWhere = wheres.find(({ table }) => table === schemaMock.document)?.condition + expect( + flattenMockConditions(placeholderWhere).some( + (node) => node.type === 'isNull' && node.column === schemaMock.document.connectorId + ) + ).toBe(true) + expect(result.mappingEntries.some((entry) => entry.resourceType === 'knowledge_document')).toBe( + false + ) + expect(result.contentPlan.knowledgeBases[0].documentIdMap).toEqual({}) + }) + it('mirrors the source knowledge-base folder and copies the KB into it, not the target root', async () => { const foldered = { ...sourceBase, folderId: 'kb-folder' } const { tx, inserts } = makeKbTx( @@ -1510,7 +1584,9 @@ describe('planForkMappedKbDocumentCopies', () => { fileSize: 123, filename: `${id}.pdf`, mimeType: 'application/pdf', - connectorId: 'connector-1', + // Hand-uploaded: connector-managed documents are filtered out by the candidate query and + // can never reach the placeholder insert. + connectorId: null, deletedAt: null, archivedAt: null, }) @@ -1526,11 +1602,19 @@ describe('planForkMappedKbDocumentCopies', () => { }> = [] ) { const inserted: Array> = [] + const wheres: unknown[] = [] let selectCalls = 0 const tx = { select: () => { const rows = selectCalls++ === 0 ? docs : existingTargets - return { from: () => ({ where: () => Promise.resolve(rows) }) } + return { + from: () => ({ + where: (condition: unknown) => { + wheres.push(condition) + return Promise.resolve(rows) + }, + }), + } }, insert: () => ({ values: (rows: Array>) => { @@ -1539,7 +1623,7 @@ describe('planForkMappedKbDocumentCopies', () => { }, }), } - return { tx: tx as unknown as DbOrTx, inserted, selectCalls: () => selectCalls } + return { tx: tx as unknown as DbOrTx, inserted, wheres, selectCalls: () => selectCalls } } const mappedKbResolver: ForkReferenceResolver = (kind, id) => @@ -1584,6 +1668,25 @@ describe('planForkMappedKbDocumentCopies', () => { ]) }) + it('never considers a connector-managed doc as a candidate for the mapped target KB', async () => { + const { tx, wheres } = makeTx([]) + await planForkMappedKbDocumentCopies({ + tx, + resolver: mappedKbResolver, + referencedDocumentIds: ['doc-1'], + alreadyCopiedSourceDocIds: new Set(), + now, + }) + + // The tx mock returns its rows regardless of the predicate, so the exclusion is only + // observable in the condition tree. + expect( + flattenMockConditions(wheres[0]).some( + (node) => node.type === 'isNull' && node.column === schemaMock.document.connectorId + ) + ).toBe(true) + }) + it('skips a referenced doc whose parent KB is not mapped (reference is left to be cleared)', async () => { const { tx, inserted } = makeTx([sourceRow('doc-1', 'unmapped-kb')]) const result = await planForkMappedKbDocumentCopies({ diff --git a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts index dc3c9dbcbdd..76f7d64e67b 100644 --- a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts +++ b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts @@ -368,6 +368,11 @@ type SkillSkeletonInsert = Omit & { conten * {@link copyForkResourceContent} to copy best-effort after commit. Secrets are * never copied: MCP OAuth tokens are omitted (re-auth required) and KB connectors * are not copied (the child is a content snapshot without live sync). + * + * Because the child gets no connector, connector-MANAGED documents are not copied + * either - only hand-uploaded ones. A detached copy is unreachable by the sync engine + * (which keys off `connector_id`), so re-attaching a connector in the child would layer + * a fresh generation on top of it instead of updating it. See {@link copyForkResourceContent}. */ export async function copyForkResourceContainers( params: CopyResourcesParams @@ -784,6 +789,12 @@ export async function copyForkResourceContainers( * Each deterministic placeholder is archived with no storage key and zero bytes, so it is * non-billable until {@link copyForkResourceContent} activates it atomically with accounting. * Documents whose parent KB is not copied are skipped, leaving their references to be cleared. + * + * Connector-managed documents are skipped for the same reason {@link copyForkResourceContent} + * excludes them from the bulk copy - a detached snapshot the child's connector would duplicate. + * Skipping them HERE too is what keeps the two sides consistent: a placeholder with no content + * phase behind it would stay archived forever while its persisted `knowledge_document` mapping + * pointed at it. Their references clear like any other uncopied document's. */ async function createForkDocumentPlaceholders(params: { tx: DbOrTx @@ -803,6 +814,7 @@ async function createForkDocumentPlaceholders(params: { and( inArray(document.id, referencedDocumentIds), inArray(document.knowledgeBaseId, Array.from(kbIdMap.keys())), + isNull(document.connectorId), isNull(document.deletedAt), isNull(document.archivedAt) ) @@ -845,7 +857,9 @@ async function createForkDocumentPlaceholders(params: { * Documents whose parent KB is being copied THIS sync are handled by * {@link createForkDocumentPlaceholders} under that copied KB and are excluded here via * `alreadyCopiedSourceDocIds`. A referenced document whose parent KB is not mapped at all is left - * untouched, so its reference is cleared as before. + * untouched, so its reference is cleared as before. Connector-managed documents are excluded for + * the reason given on {@link copyForkResourceContent} - and the exclusion matters MORE here, since + * the target KB is an existing one that may already run its own connector over the same source. */ export async function planForkMappedKbDocumentCopies(params: { tx: DbOrTx @@ -877,6 +891,7 @@ export async function planForkMappedKbDocumentCopies(params: { .where( and( inArray(document.id, candidateIds), + isNull(document.connectorId), isNull(document.deletedAt), isNull(document.archivedAt) ) @@ -1124,13 +1139,51 @@ export async function copyForkResourceContent(params: { for (const kb of contentPlan.knowledgeBases) { try { + // Connector-managed documents are excluded from the copy (see the predicate below), and a + // KB can be entirely connector-sourced - so report what was left behind rather than letting + // the child silently land with fewer documents than the source. + const [{ skipped: connectorManaged = 0 } = {}] = await db + .select({ skipped: sql`count(*)` }) + .from(document) + .where( + and( + eq(document.knowledgeBaseId, kb.sourceId), + isNotNull(document.connectorId), + isNull(document.deletedAt), + isNull(document.archivedAt) + ) + ) + if (Number(connectorManaged) > 0) { + logger.info( + `[${requestId}] Skipped connector-managed documents in a copied knowledge base`, + { + sourceKnowledgeBaseId: kb.sourceId, + childKnowledgeBaseId: kb.childId, + skipped: Number(connectorManaged), + } + ) + } let afterDocId: string | null = null for (;;) { // Only copy LIVE documents - exclude soft-deleted and archived rows, matching // how the rest of the KB system treats them as gone (chunks/tags/search filter // both). A fork must not resurrect documents removed from the source base. + // + // Connector-managed documents (`connector_id IS NOT NULL`) are excluded too. A copy can + // only be a DETACHED snapshot - the child gets no connector (see + // `copyForkResourceContainers`), and the sync engine keys every existing/tombstone/ + // exclusion lookup off `connector_id`, so a detached copy is invisible to it: it can + // never be updated, reconciled, or purged, and `doc_connector_external_id_idx` + // (UNIQUE on `(connector_id, external_id)`) does not constrain it because its + // `connector_id` is NULL. Attaching a connector to the child then re-ingests every + // page as a NEW row on top of the snapshot, stacking one duplicate generation per fork + // hop and returning the same page several times from one retrieval. Skipping them + // instead leaves the child's connector as the single owner of that content. A source + // document whose connector was DELETED already has a null `connector_id` (the FK is + // ON DELETE SET NULL) and is static content in the source too, so it still copies. const liveDocs = and( eq(document.knowledgeBaseId, kb.sourceId), + isNull(document.connectorId), isNull(document.deletedAt), isNull(document.archivedAt) ) @@ -1273,6 +1326,15 @@ export async function copyForkResourceContent(params: { if (!source) { throw new Error(`Source document ${docEntry.sourceDocId} is missing`) } + if (source.connectorId) { + // Only reachable from a payload planned before connector-managed documents were excluded + // (a rolling deploy). Fail the entry instead of filling it: the per-document cleanup + // below drops the archived placeholder and clears its references, which is the outcome + // the planner would now produce anyway. + throw new Error( + `Source document ${docEntry.sourceDocId} is connector-managed and is not copied across a fork edge` + ) + } const resolvedBillingContext = await getBillingContext() await copyKbDocument({ source, From acfd00ab6fa3e64a5d52a89627efafcb08d1edeb Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Tue, 18 Aug 2026 13:52:15 -0700 Subject: [PATCH 2/4] fix(forks): keep the skipped-document count from failing a copied KB The connector-managed count feeds a log line, but it sat inside the KB's try block, so a transient failure on a COUNT(*) would roll back a copy that had otherwise succeeded and clear every reference to it. Move it into a helper that swallows its own error. Counting is not copying: only the copy itself may fail a resource. Test proven red by removing the catch - the mutation reports a knowledge-base failure. Co-Authored-By: Claude Opus 5 (1M context) --- .../lib/copy/copy-resources.test.ts | 18 +++++ .../lib/copy/copy-resources.ts | 79 ++++++++++--------- 2 files changed, 61 insertions(+), 36 deletions(-) diff --git a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts index 5d1f73e39dc..1f7484922b6 100644 --- a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts +++ b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts @@ -335,6 +335,24 @@ describe('copyForkResourceContent', () => { ).toBe(true) }) + it('keeps a copied KB alive when the skipped-document count fails', async () => { + // The count only feeds a log line. Letting it throw into the KB's catch would roll back a + // perfectly good copy and clear every reference to it over a failed COUNT(*). + dbChainMockFns.where.mockImplementationOnce(() => { + throw new Error('count failed') + }) + dbChainMockFns.limit.mockResolvedValueOnce([]) + + const result = await copyForkResourceContent({ + contentPlan: basePlan({ + knowledgeBases: [{ sourceId: 'src-kb', childId: 'child-kb', documentIdMap: {} }], + }), + requestId: 'test', + }) + + expect(result).toEqual({ copied: 1, failed: 0, failures: [] }) + }) + it('uses the blob content digest so a retry cannot adopt an older failed snapshot', async () => { dbChainMockFns.limit .mockResolvedValueOnce([sourceDoc]) diff --git a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts index 76f7d64e67b..ff06cef8f88 100644 --- a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts +++ b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts @@ -1031,6 +1031,39 @@ export async function copyForkResourceContent(params: { billingContext ??= await resolveStorageBillingContext(childWorkspaceId) return billingContext } + /** + * Report the connector-managed documents a copied KB leaves behind, since a fully + * connector-synced base lands in the child with no documents at all. Strictly observability, + * so it swallows its own failure: counting is not copying, and a transient error here must not + * take down the KB the way a failed document does. + */ + const logSkippedConnectorDocuments = async (kb: ForkContentKbEntry): Promise => { + try { + const [row] = await db + .select({ total: sql`count(*)` }) + .from(document) + .where( + and( + eq(document.knowledgeBaseId, kb.sourceId), + isNotNull(document.connectorId), + isNull(document.deletedAt), + isNull(document.archivedAt) + ) + ) + const skipped = Number(row?.total ?? 0) + if (skipped === 0) return + logger.info(`[${requestId}] Skipped connector-managed documents in a copied knowledge base`, { + sourceKnowledgeBaseId: kb.sourceId, + childKnowledgeBaseId: kb.childId, + skipped, + }) + } catch (error) { + logger.warn(`[${requestId}] Failed to count the documents a copied knowledge base skipped`, { + sourceKnowledgeBaseId: kb.sourceId, + error: getErrorMessage(error), + }) + } + } for (const table of contentPlan.tables) { try { @@ -1139,48 +1172,22 @@ export async function copyForkResourceContent(params: { for (const kb of contentPlan.knowledgeBases) { try { - // Connector-managed documents are excluded from the copy (see the predicate below), and a - // KB can be entirely connector-sourced - so report what was left behind rather than letting - // the child silently land with fewer documents than the source. - const [{ skipped: connectorManaged = 0 } = {}] = await db - .select({ skipped: sql`count(*)` }) - .from(document) - .where( - and( - eq(document.knowledgeBaseId, kb.sourceId), - isNotNull(document.connectorId), - isNull(document.deletedAt), - isNull(document.archivedAt) - ) - ) - if (Number(connectorManaged) > 0) { - logger.info( - `[${requestId}] Skipped connector-managed documents in a copied knowledge base`, - { - sourceKnowledgeBaseId: kb.sourceId, - childKnowledgeBaseId: kb.childId, - skipped: Number(connectorManaged), - } - ) - } + await logSkippedConnectorDocuments(kb) let afterDocId: string | null = null for (;;) { // Only copy LIVE documents - exclude soft-deleted and archived rows, matching // how the rest of the KB system treats them as gone (chunks/tags/search filter // both). A fork must not resurrect documents removed from the source base. // - // Connector-managed documents (`connector_id IS NOT NULL`) are excluded too. A copy can - // only be a DETACHED snapshot - the child gets no connector (see - // `copyForkResourceContainers`), and the sync engine keys every existing/tombstone/ - // exclusion lookup off `connector_id`, so a detached copy is invisible to it: it can - // never be updated, reconciled, or purged, and `doc_connector_external_id_idx` - // (UNIQUE on `(connector_id, external_id)`) does not constrain it because its - // `connector_id` is NULL. Attaching a connector to the child then re-ingests every - // page as a NEW row on top of the snapshot, stacking one duplicate generation per fork - // hop and returning the same page several times from one retrieval. Skipping them - // instead leaves the child's connector as the single owner of that content. A source - // document whose connector was DELETED already has a null `connector_id` (the FK is - // ON DELETE SET NULL) and is static content in the source too, so it still copies. + // Connector-managed documents are excluded too, because a copy could only ever be a + // DETACHED snapshot: the child gets no connector (see `copyForkResourceContainers`), and + // the sync engine keys every existing/tombstone/exclusion lookup off `connector_id`, so + // the copy is invisible to it - never updated, reconciled, or purged. Attaching a + // connector in the child then re-ingests every page as a NEW row on top of the snapshot, + // stacking one dead generation per fork hop. Skipping them leaves the child's own + // connector as the single owner of that content. A source document whose connector was + // DELETED already has a null `connector_id` (the FK is ON DELETE SET NULL) and is static + // content in the source too, so it still copies. const liveDocs = and( eq(document.knowledgeBaseId, kb.sourceId), isNull(document.connectorId), From 5d7749b197d78390b2ea83f1971bec5e2db41ab2 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Tue, 18 Aug 2026 13:57:13 -0700 Subject: [PATCH 3/4] fix(forks): clean up full-KB placeholders planned before the exclusion The mapped-KB fill guarded a pre-change plan, but the full-KB path did not: a placeholder planned by an old worker for a connector-managed document is simply no longer returned by the page query, so nothing fills it and it stays archived behind a live mapping that a remapped document-selector still resolves to. Report those child ids as failed documents so the shared cleanup clears their references and drops the rows, and delete their persisted identity so a later sync does not resolve to a row cleanup removes. Keyed on the SOURCE being connector-managed, which can never become copyable, so it cannot race a concurrent attempt mid-fill the way a "source is gone" check could. The mapping drop is now one helper shared with the mapped-KB catch. Test proven red by removing the reconciliation block. Co-Authored-By: Claude Opus 5 (1M context) --- .../lib/copy/copy-resources.test.ts | 36 ++++++++++ .../lib/copy/copy-resources.ts | 68 +++++++++++++++---- 2 files changed, 89 insertions(+), 15 deletions(-) diff --git a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts index 1f7484922b6..b60d49f5cd0 100644 --- a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts +++ b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts @@ -335,6 +335,42 @@ describe('copyForkResourceContent', () => { ).toBe(true) }) + it('drops a full-KB placeholder a pre-change worker planned for a connector-managed doc', async () => { + // Rolling deploy: the fork tx ran on the old code and planned a placeholder for a + // connector-managed document, which this worker's page query no longer returns. Nothing + // would ever fill it, so it must be reported for cleanup rather than left archived behind a + // live mapping that a remapped document-selector still resolves to. + dbChainMockFns.where.mockImplementationOnce(() => ({ + // The skipped-document count. + then: (resolve: (rows: unknown[]) => unknown) => resolve([{ total: 1 }]), + })) + dbChainMockFns.where.mockImplementationOnce(() => ({ + // The stale-plan probe: the planned source is connector-managed. + then: (resolve: (rows: unknown[]) => unknown) => resolve([{ id: 'doc-1' }]), + })) + dbChainMockFns.limit.mockResolvedValueOnce([]) + + const result = await copyForkResourceContent({ + contentPlan: basePlan({ + knowledgeBases: [ + { sourceId: 'src-kb', childId: 'child-kb', documentIdMap: { 'doc-1': 'child-doc-1' } }, + ], + documentMappingContext: { edgeChildWorkspaceId: 'edge-child-ws', sourceIsParent: false }, + }), + requestId: 'test', + }) + + expect(result.failed).toBe(1) + expect(result.failures).toEqual([{ kind: 'knowledge-document', childId: 'child-doc-1' }]) + // The persisted identity goes too, or a later sync resolves to the row cleanup deletes. + expect(mockDeleteCopiedResourceMappingsByTargets).toHaveBeenCalledWith({ + executor: expect.anything(), + edgeChildWorkspaceId: 'edge-child-ws', + sourceIsParent: false, + targets: [{ resourceType: 'knowledge_document', resourceId: 'child-doc-1' }], + }) + }) + it('keeps a copied KB alive when the skipped-document count fails', async () => { // The count only feeds a log line. Letting it throw into the KB's catch would roll back a // perfectly good copy and clear every reference to it over a failed COUNT(*). diff --git a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts index ff06cef8f88..9d85638cfb4 100644 --- a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts +++ b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts @@ -1031,6 +1031,29 @@ export async function copyForkResourceContent(params: { billingContext ??= await resolveStorageBillingContext(childWorkspaceId) return billingContext } + /** + * Drop the persisted `knowledge_document` identity for a copied document that will not exist, + * so a later sync resolves the reference afresh instead of to a row the cleanup removes. + * Isolated like the rest of the post-commit phase: a mapping-cleanup failure is logged, never + * rethrown, since the caller is already reporting the document as failed. + */ + const dropCopiedDocumentMapping = async (childDocumentId: string): Promise => { + const mappingContext = contentPlan.documentMappingContext + if (!mappingContext) return + try { + await deleteCopiedResourceMappingsByTargets({ + executor: db, + edgeChildWorkspaceId: mappingContext.edgeChildWorkspaceId, + sourceIsParent: mappingContext.sourceIsParent, + targets: [{ resourceType: 'knowledge_document', resourceId: childDocumentId }], + }) + } catch (mappingCleanupError) { + logger.error(`[${requestId}] Failed to clean mapping for a failed copied document`, { + childDocumentId, + error: getErrorMessage(mappingCleanupError), + }) + } + } /** * Report the connector-managed documents a copied KB leaves behind, since a fully * connector-synced base lands in the child with no documents at all. Strictly observability, @@ -1173,6 +1196,35 @@ export async function copyForkResourceContent(params: { for (const kb of contentPlan.knowledgeBases) { try { await logSkippedConnectorDocuments(kb) + // A worker from before this exclusion (a rolling deploy) could have planned a placeholder + // for a connector-managed document. The page query below no longer returns its source, so + // nothing would ever fill it - leaving an archived empty row that a persisted mapping and + // a remapped `document-selector` still resolve to. Report those child ids as failed + // documents instead, so the shared cleanup clears the references and drops the rows. + // + // Keyed on the SOURCE being connector-managed, which is deterministic: such a document can + // never become copyable, so this can never race a concurrent attempt mid-fill (unlike a + // "source is gone" check, which could). + const plannedSourceIds = Object.keys(kb.documentIdMap) + if (plannedSourceIds.length > 0) { + const stalePlanned = await db + .select({ id: document.id }) + .from(document) + .where(and(inArray(document.id, plannedSourceIds), isNotNull(document.connectorId))) + for (const { id } of stalePlanned) { + const childDocumentId = kb.documentIdMap[id] + if (!childDocumentId) continue + // Left in `documentIdMap` deliberately: if the KB itself later fails, its failure + // lists the same child id again, and the cleanup keys failed ids by kind in a Set. + await dropCopiedDocumentMapping(childDocumentId) + failedResources += 1 + failures.push({ kind: 'knowledge-document', childId: childDocumentId }) + logger.warn( + `[${requestId}] Dropping a fork placeholder planned for a connector-managed document`, + { sourceDocumentId: id, childDocumentId, childKnowledgeBaseId: kb.childId } + ) + } + } let afterDocId: string | null = null for (;;) { // Only copy LIVE documents - exclude soft-deleted and archived rows, matching @@ -1353,21 +1405,7 @@ export async function copyForkResourceContent(params: { }) copiedResources += 1 } catch (error) { - if (contentPlan.documentMappingContext) { - try { - await deleteCopiedResourceMappingsByTargets({ - executor: db, - edgeChildWorkspaceId: contentPlan.documentMappingContext.edgeChildWorkspaceId, - sourceIsParent: contentPlan.documentMappingContext.sourceIsParent, - targets: [{ resourceType: 'knowledge_document', resourceId: docEntry.childDocId }], - }) - } catch (mappingCleanupError) { - logger.error(`[${requestId}] Failed to clean mapping for a failed copied document`, { - childDocumentId: docEntry.childDocId, - error: getErrorMessage(mappingCleanupError), - }) - } - } + await dropCopiedDocumentMapping(docEntry.childDocId) failedResources += 1 failures.push({ kind: 'knowledge-document', childId: docEntry.childDocId }) logger.warn(`[${requestId}] Failed to copy document into mapped KB during sync`, { From ae315cc4c81a53d0127676ad434963d9168b8943 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Tue, 18 Aug 2026 14:09:29 -0700 Subject: [PATCH 4/4] fix(forks): make the stale-plan probe best-effort The probe ran inside the KB try, so a transient SELECT would reach the catch, roll back a complete copy, delete the child base, and clear every reference to it. Weighing it as "load-bearing, so fail closed" was wrong: the probe runs on EVERY copied KB that has referenced documents, while the state it repairs exists only inside a rollout window. Failing closed traded a common-path outage against a rare-squared one. It now swallows its own failure with a loud error log, leaving that pre-existing state in place rather than destroying a good copy. Test proven red by removing the catch - the mutation reports the KB failure. Co-Authored-By: Claude Opus 5 (1M context) --- .../lib/copy/copy-resources.test.ts | 24 ++++++ .../lib/copy/copy-resources.ts | 83 ++++++++++++------- 2 files changed, 79 insertions(+), 28 deletions(-) diff --git a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts index b60d49f5cd0..274605ad669 100644 --- a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts +++ b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.test.ts @@ -371,6 +371,30 @@ describe('copyForkResourceContent', () => { }) }) + it('keeps a copied KB alive when the stale-plan probe fails', async () => { + // The probe runs on every KB with referenced documents, but the state it repairs only exists + // inside a rollout window. Letting it reach the KB catch would delete a complete copy and + // clear every reference to it over a transient SELECT. + dbChainMockFns.where.mockImplementationOnce(() => ({ + then: (resolve: (rows: unknown[]) => unknown) => resolve([{ total: 0 }]), + })) + dbChainMockFns.where.mockImplementationOnce(() => { + throw new Error('stale-plan probe failed') + }) + dbChainMockFns.limit.mockResolvedValueOnce([]) + + const result = await copyForkResourceContent({ + contentPlan: basePlan({ + knowledgeBases: [ + { sourceId: 'src-kb', childId: 'child-kb', documentIdMap: { 'doc-1': 'child-doc-1' } }, + ], + }), + requestId: 'test', + }) + + expect(result).toEqual({ copied: 1, failed: 0, failures: [] }) + }) + it('keeps a copied KB alive when the skipped-document count fails', async () => { // The count only feeds a log line. Letting it throw into the KB's catch would roll back a // perfectly good copy and clear every reference to it over a failed COUNT(*). diff --git a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts index 9d85638cfb4..a50830cebce 100644 --- a/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts +++ b/apps/sim/ee/workspace-forking/lib/copy/copy-resources.ts @@ -1054,6 +1054,58 @@ export async function copyForkResourceContent(params: { }) } } + /** + * Find the placeholders a worker from before this exclusion (a rolling deploy) planned for + * connector-managed documents, drop their persisted identities, and return the child ids to + * report as failed documents - the page query no longer returns their sources, so nothing + * would ever fill them, leaving archived empty rows that a mapping and a remapped + * `document-selector` still resolve to. + * + * Keyed on the SOURCE being connector-managed, which is deterministic: such a document can + * never become copyable, so this cannot race a concurrent attempt sitting between + * {@link ensureKbDocumentPlaceholder} and {@link finalizeKbDocument} (a "planned but unfilled" + * sweep would). + * + * Best-effort, like the count above: this probe runs on EVERY copied KB that has referenced + * documents, while the state it repairs exists only inside a rollout window. Letting a + * transient failure reach the KB's catch would delete an otherwise-complete copy and clear + * every reference to it - far worse, and far more likely, than the dangling placeholder it + * guards against. A failure is logged loudly and leaves that pre-existing state in place. + */ + const reconcileStalePlannedDocuments = async (kb: ForkContentKbEntry): Promise => { + const plannedSourceIds = Object.keys(kb.documentIdMap) + if (plannedSourceIds.length === 0) return [] + try { + const stalePlanned = await db + .select({ id: document.id }) + .from(document) + .where(and(inArray(document.id, plannedSourceIds), isNotNull(document.connectorId))) + const staleChildIds: string[] = [] + for (const { id } of stalePlanned) { + const childDocumentId = kb.documentIdMap[id] + if (!childDocumentId) continue + // Left in `documentIdMap` deliberately: if the KB itself later fails, its failure lists + // the same child id again, and the cleanup keys failed ids by kind in a Set. + await dropCopiedDocumentMapping(childDocumentId) + staleChildIds.push(childDocumentId) + logger.warn( + `[${requestId}] Dropping a fork placeholder planned for a connector-managed document`, + { sourceDocumentId: id, childDocumentId, childKnowledgeBaseId: kb.childId } + ) + } + return staleChildIds + } catch (error) { + logger.error( + `[${requestId}] Failed to reconcile fork placeholders planned for connector-managed documents`, + { + sourceKnowledgeBaseId: kb.sourceId, + childKnowledgeBaseId: kb.childId, + error: getErrorMessage(error), + } + ) + return [] + } + } /** * Report the connector-managed documents a copied KB leaves behind, since a fully * connector-synced base lands in the child with no documents at all. Strictly observability, @@ -1196,34 +1248,9 @@ export async function copyForkResourceContent(params: { for (const kb of contentPlan.knowledgeBases) { try { await logSkippedConnectorDocuments(kb) - // A worker from before this exclusion (a rolling deploy) could have planned a placeholder - // for a connector-managed document. The page query below no longer returns its source, so - // nothing would ever fill it - leaving an archived empty row that a persisted mapping and - // a remapped `document-selector` still resolve to. Report those child ids as failed - // documents instead, so the shared cleanup clears the references and drops the rows. - // - // Keyed on the SOURCE being connector-managed, which is deterministic: such a document can - // never become copyable, so this can never race a concurrent attempt mid-fill (unlike a - // "source is gone" check, which could). - const plannedSourceIds = Object.keys(kb.documentIdMap) - if (plannedSourceIds.length > 0) { - const stalePlanned = await db - .select({ id: document.id }) - .from(document) - .where(and(inArray(document.id, plannedSourceIds), isNotNull(document.connectorId))) - for (const { id } of stalePlanned) { - const childDocumentId = kb.documentIdMap[id] - if (!childDocumentId) continue - // Left in `documentIdMap` deliberately: if the KB itself later fails, its failure - // lists the same child id again, and the cleanup keys failed ids by kind in a Set. - await dropCopiedDocumentMapping(childDocumentId) - failedResources += 1 - failures.push({ kind: 'knowledge-document', childId: childDocumentId }) - logger.warn( - `[${requestId}] Dropping a fork placeholder planned for a connector-managed document`, - { sourceDocumentId: id, childDocumentId, childKnowledgeBaseId: kb.childId } - ) - } + for (const childDocumentId of await reconcileStalePlannedDocuments(kb)) { + failedResources += 1 + failures.push({ kind: 'knowledge-document', childId: childDocumentId }) } let afterDocId: string | null = null for (;;) {