mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-21 13:00:04 +08:00
fix(forks): stop copying connector-managed knowledge base documents (#6818)
* 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Vikhyath Mondreti
co-authored by
Claude Opus 5
Claude Opus 5
parent
d1e3eeea9e
commit
3a03774e42
@@ -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.
|
||||
|
||||
<Callout type="warn">
|
||||
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.
|
||||
</Callout>
|
||||
|
||||
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.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -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,106 @@ 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('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 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(*).
|
||||
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])
|
||||
@@ -1051,6 +1153,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 +1480,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<Array<Record<string, unknown>>> = []
|
||||
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 +1500,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 +1581,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 +1662,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 +1680,19 @@ describe('planForkMappedKbDocumentCopies', () => {
|
||||
}> = []
|
||||
) {
|
||||
const inserted: Array<Record<string, unknown>> = []
|
||||
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<Record<string, unknown>>) => {
|
||||
@@ -1539,7 +1701,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 +1746,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({
|
||||
|
||||
@@ -368,6 +368,11 @@ type SkillSkeletonInsert = Omit<typeof skill.$inferInsert, 'content'> & { 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)
|
||||
)
|
||||
@@ -1016,6 +1031,114 @@ 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<void> => {
|
||||
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),
|
||||
})
|
||||
}
|
||||
}
|
||||
/**
|
||||
* 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<string[]> => {
|
||||
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,
|
||||
* 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<void> => {
|
||||
try {
|
||||
const [row] = await db
|
||||
.select({ total: sql<number>`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 {
|
||||
@@ -1124,13 +1247,29 @@ export async function copyForkResourceContent(params: {
|
||||
|
||||
for (const kb of contentPlan.knowledgeBases) {
|
||||
try {
|
||||
await logSkippedConnectorDocuments(kb)
|
||||
for (const childDocumentId of await reconcileStalePlannedDocuments(kb)) {
|
||||
failedResources += 1
|
||||
failures.push({ kind: 'knowledge-document', childId: childDocumentId })
|
||||
}
|
||||
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 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),
|
||||
isNull(document.deletedAt),
|
||||
isNull(document.archivedAt)
|
||||
)
|
||||
@@ -1273,6 +1412,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,
|
||||
@@ -1284,21 +1432,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`, {
|
||||
|
||||
Reference in New Issue
Block a user