From a61805346808d6aa9b886280c33da98873c3027e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 18:36:33 -0700 Subject: [PATCH 1/2] fix(project-files): requeue rendered-artifact search after commit and trim Project read transactions - Project artifact reads no longer requeue search inside the read transaction, which held the Project and source row shared and then needed them exclusively; the requeue runs after commit with its owner lock bounded by the short artifact-ready lock timeout and is best effort - Artifact, inline, and CSV preview reads authorize from their own snapshots instead of a separate preflight; artifact inputs are snapshotted only when referenced and download with bounded concurrency after a declared-size budget check; the request signal reaches render and compile - Project inline delivery streams and caches content-addressed key requests like workspace inline - Project search admits, withholds previews, and reports provenance after its snapshot commits, classifies provenance only when a gate or delivery observer needs it, and lists folders directly - A Project page embedding another owner's image is a missing input instead of a 500 - Page limits are shared from page-document.server; knip and the CI isolate build are tightened --- .github/workflows/checks.yml | 5 +- .../api/projects/[id]/files/inline/route.ts | 11 +- .../content-rendered.integration.ts | 132 ++++++ .../projects/files/application/artifacts.ts | 416 +++++++++++------- .../projects/files/application/previews.ts | 132 +++--- .../lib/projects/files/application/search.ts | 108 +++-- apps/sim/lib/projects/membership.ts | 25 +- apps/sim/lib/workspace-files/page-compile.ts | 10 +- .../workspace-files/page-document.server.ts | 6 +- .../workspace-files/search/artifact-ready.ts | 8 +- .../lib/workspace-files/search/delivery.ts | 8 +- .../workspace-files/search/owner-policy.ts | 8 +- knip.jsonc | 6 +- 13 files changed, 565 insertions(+), 310 deletions(-) diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index c57afab1d3b..6bd415230ae 100644 --- a/.github/workflows/checks.yml +++ b/.github/workflows/checks.yml @@ -243,11 +243,14 @@ jobs: working-directory: packages/db run: bun run db:migrate + # JOBS=4 caps node-gyp parallelism as docker/app.Dockerfile does (laverdet/isolated-vm#428). - name: Build native isolate addon for Project file rendering if: matrix.group == 'project-files' working-directory: apps/sim + env: + JOBS: 4 run: | - addon_dir="$(node -p 'require("node:path").dirname(require.resolve("isolated-vm/package.json", { paths: ["./apps/sim"] }))')" + addon_dir="$(node -p 'require("node:path").dirname(require.resolve("isolated-vm/package.json", { paths: ["."] }))')" cd "$addon_dir" bun run install diff --git a/apps/sim/app/api/projects/[id]/files/inline/route.ts b/apps/sim/app/api/projects/[id]/files/inline/route.ts index 8caab953b32..84efd1af8dd 100644 --- a/apps/sim/app/api/projects/[id]/files/inline/route.ts +++ b/apps/sim/app/api/projects/[id]/files/inline/route.ts @@ -5,8 +5,8 @@ import { internalRateLimits, internalSessionAuth, } from '@/lib/api/server/routes' -import { presentProjectFileContent } from '@/lib/projects/files/api' import { readProjectInlineFile } from '@/lib/projects/files/application' +import { presentFileDelivery, workspaceFileCacheControl } from '@/lib/uploads/server/delivery' export const GET = defineInternalBinaryRoute({ contract: getInlineProjectFileContract, @@ -21,5 +21,12 @@ export const GET = defineInternalBinaryRoute({ referenceFileId: query.fileId, }), useCase: readProjectInlineFile, - present: presentProjectFileContent, + present: ({ file, stream, contentLength, contentAddressed }) => + presentFileDelivery({ + body: stream, + filename: file.name, + contentType: file.type, + contentLength, + cacheControl: workspaceFileCacheControl(contentAddressed), + }), }) diff --git a/apps/sim/lib/projects/files/__integration__/content-rendered.integration.ts b/apps/sim/lib/projects/files/__integration__/content-rendered.integration.ts index e98a1e4b4b5..89839ae0871 100644 --- a/apps/sim/lib/projects/files/__integration__/content-rendered.integration.ts +++ b/apps/sim/lib/projects/files/__integration__/content-rendered.integration.ts @@ -5,6 +5,7 @@ import { dirname, join } from 'node:path' import type { ResourceDelegatedPrincipal } from '@sim/auth/principal' import { db } from '@sim/db' import { + fileSearchDispatchQueue, folder, idempotencyKey, member, @@ -14,6 +15,7 @@ import { subscription, user, workspace, + workspaceFileSearchRevision, workspaceFiles, } from '@sim/db/schema' import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' @@ -44,11 +46,13 @@ import { } from '@/lib/projects/files/application/previews' import { searchProjectFileContent } from '@/lib/projects/files/application/search' import { revertProjectFileVersion } from '@/lib/projects/files/application/versions' +import { lockProjectShared } from '@/lib/projects/membership' import * as storageCleanup from '@/lib/uploads/contexts/workspace/file-storage-cleanup-outbox' import * as storage from '@/lib/uploads/core/storage-service' import { compileFileDocument } from '@/lib/uploads/documents/compile' import * as heic from '@/lib/uploads/server/heic' import { observeWorkspaceFileDelivery } from '@/lib/workspace-files/application/file-delivery-observer' +import { FILE_SEARCH_ARTIFACT_READY_LOCK_TIMEOUT_MS } from '@/lib/workspace-files/search/constants' import * as searchDelivery from '@/lib/workspace-files/search/delivery' import * as searchRepository from '@/lib/workspace-files/search/repository' @@ -306,6 +310,7 @@ describe('Project delivery admission', () => { } vi.spyOn(searchRepository, 'searchFileIndex').mockResolvedValue(prepared) vi.spyOn(searchDelivery, 'loadFileSearchDelivery').mockResolvedValue({ + current: new Set(), provenance: new Map(), outOfScope: new Set(['moved']), inputs: [], @@ -671,6 +676,130 @@ describe('Project rendered artifacts against PostgreSQL and private storage', () } ) + check( + 'a generated artifact read beside a concurrent Project reader completes and requeues search', + async () => { + const f = await fixture() + const source = await createProjectFile.execute({ + principal: f.principal, + input: { + ...createInput(f.projectId, 'const title = "concurrent"'), + name: 'deck.pptx', + contentType: 'text/x-pptxgenjs', + }, + }) + const artifact = Buffer.from('PK\u0003\u0004document') + vi.spyOn(sandboxTask, 'runSandboxTask').mockResolvedValue(artifact) + let markHeld = () => {} + const held = new Promise((resolve) => { + markHeld = resolve + }) + let release = () => {} + const released = new Promise((resolve) => { + release = resolve + }) + const concurrentReader = db.transaction(async (tx) => { + await lockProjectShared(tx, f.projectId) + markHeld() + await released + }) + await held + try { + const rendered = await observeWorkspaceFileDelivery( + async () => release(), + () => + readProjectFileArtifact.execute({ + principal: f.principal, + input: { projectId: f.projectId, fileId: source.file.id, maxBytes: 1_000_000 }, + }) + ) + expect(rendered.buffer).toEqual(artifact) + } finally { + release() + await concurrentReader + } + const [revision] = await db + .select() + .from(workspaceFileSearchRevision) + .where(eq(workspaceFileSearchRevision.fileId, source.file.id)) + expect(revision?.status).toBe('pending') + expect(revision?.sourceContentUpdatedAt).toEqual(source.file.contentUpdatedAt) + } + ) + + check( + 'a contended search requeue gives up within its short bound without failing the read', + async () => { + const f = await fixture() + const source = await createProjectFile.execute({ + principal: f.principal, + input: { + ...createInput(f.projectId, 'const title = "contended"'), + name: 'deck.pptx', + contentType: 'text/x-pptxgenjs', + }, + }) + const artifact = Buffer.from('PK\u0003\u0004document') + vi.spyOn(sandboxTask, 'runSandboxTask').mockResolvedValue(artifact) + let markHeld = () => {} + const held = new Promise((resolve) => { + markHeld = resolve + }) + let release = () => {} + const released = new Promise((resolve) => { + release = resolve + }) + const concurrentReader = db.transaction(async (tx) => { + await lockProjectShared(tx, f.projectId) + markHeld() + await released + }) + await held + try { + const started = performance.now() + const rendered = await readProjectFileArtifact.execute({ + principal: f.principal, + input: { projectId: f.projectId, fileId: source.file.id, maxBytes: 1_000_000 }, + }) + expect(rendered.buffer).toEqual(artifact) + expect(performance.now() - started).toBeLessThan( + FILE_SEARCH_ARTIFACT_READY_LOCK_TIMEOUT_MS + 2_000 + ) + } finally { + release() + await concurrentReader + } + const revisions = await db + .select() + .from(workspaceFileSearchRevision) + .where(eq(workspaceFileSearchRevision.fileId, source.file.id)) + expect(revisions).toEqual([]) + } + ) + + check("a Project page embedding another owner's image is a missing input", async () => { + const f = await fixture() + for (const owner of [`workspace=${f.workspaceId}`, `project=${generateId()}`]) { + const page = await createProjectFile.execute({ + principal: f.principal, + input: { + ...createInput( + f.projectId, + `---\ntitle: Design\n---\n\n![Diagram](sim:file/${generateId()}?${owner})` + ), + name: `${generateId()}.html`, + contentType: 'text/x-sim-page', + }, + }) + await expect( + readProjectFileArtifact.execute({ + principal: f.principal, + input: { projectId: f.projectId, fileId: page.file.id, maxBytes: 1_000_000 }, + }) + ).rejects.toMatchObject({ code: 'not_found' }) + } + }) + check( 'artifact source cannot import a foreign Project asset through a valid source-file capability', async () => { @@ -846,6 +975,9 @@ afterAll(async () => { .where( sql`${outboxEvent.payload}::jsonb ->> 'key' LIKE ${`project/${f.projectId}/%`} OR ${outboxEvent.payload}::jsonb -> 'owner' ->> 'entityId' = ${f.projectId}` ) + await db + .delete(fileSearchDispatchQueue) + .where(eq(fileSearchDispatchQueue.entityId, f.projectId)) await db.delete(workspaceFiles).where(eq(workspaceFiles.projectId, f.projectId)) await db.delete(folder).where(eq(folder.projectId, f.projectId)) await deleteWorkspaceFixture(db, eq(workspace.id, f.workspaceId)) diff --git a/apps/sim/lib/projects/files/application/artifacts.ts b/apps/sim/lib/projects/files/application/artifacts.ts index fd3e4853aee..63cc36d03d9 100644 --- a/apps/sim/lib/projects/files/application/artifacts.ts +++ b/apps/sim/lib/projects/files/application/artifacts.ts @@ -1,9 +1,14 @@ +import type { Principal } from '@sim/auth/principal' import type { WorkspaceFileRow } from '@sim/db/schema' import { workspaceFiles } from '@sim/db/schema' +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' import { and, asc, inArray, isNull } from 'drizzle-orm' import { requireAdmittedContent } from '@/lib/core/application/content-delivery' import { isDocSandboxEnabled } from '@/lib/core/config/env-flags' +import { runWithOutboundOrganization } from '@/lib/core/network/context.server' import { OrchestrationError } from '@/lib/core/orchestration/types' +import { MATERIALIZE_CONCURRENCY, mapWithConcurrency } from '@/lib/core/utils/concurrency' import { assertKnownSizeWithinLimit } from '@/lib/core/utils/stream-limits' import type { DbTransaction } from '@/lib/db/types' import type { @@ -35,17 +40,34 @@ import { } from '@/lib/uploads/documents' import { fileDocumentInputIdentity } from '@/lib/uploads/documents/input-identity' import { resolveServableImageBytes } from '@/lib/uploads/server/image-derivative' -import { MAX_BUFFERED_TRANSFER_BYTES } from '@/lib/uploads/shared/types' +import { getWorkspaceFileSize, MAX_BUFFERED_TRANSFER_BYTES } from '@/lib/uploads/shared/types' import { reportWorkspaceFileDelivery } from '@/lib/workspace-files/application/file-delivery-observer' import { fileOwnerCondition } from '@/lib/workspace-files/ownership-query' -import { isSimPageSource, SIM_PAGE_CONTENT_TYPE } from '@/lib/workspace-files/page-compile' +import { + isSimPageSource, + ProjectPageForeignEmbedError, + SIM_PAGE_CONTENT_TYPE, +} from '@/lib/workspace-files/page-compile' import { renderSimPageDocument } from '@/lib/workspace-files/page-document' import { collectSimPageFileReferences, inlineSimPageImages, + MAX_INLINE_IMAGE_BYTES, + MAX_INLINE_IMAGE_REFERENCES, + MAX_INLINE_TOTAL_BYTES, } from '@/lib/workspace-files/page-document.server' import { createFileReadReceipt, type FileReadReceipt } from '@/lib/workspace-files/read-receipt' -import { markFileSearchArtifactReadyInTx } from '@/lib/workspace-files/search/artifact-ready' +import { + type FileSearchArtifactCompletion, + requeueFileSearchForRenderedArtifact, +} from '@/lib/workspace-files/search/artifact-ready' + +const logger = createLogger('ProjectFileArtifacts') + +/** Generated-document input limits; `compileFileDocument` enforces the same values on the bytes. */ +const MAX_DOCUMENT_INPUTS = 500 +const MAX_DOCUMENT_INPUT_BYTES = 25 * 1024 * 1024 +const MAX_DOCUMENT_INPUT_TOTAL_BYTES = 50 * 1024 * 1024 type ProjectOwner = { entityType: 'project'; entityId: string } interface ArtifactInput extends ProjectFileTarget { @@ -56,18 +78,21 @@ interface ArtifactInput extends ProjectFileTarget { } interface SourceSnapshot { file: WorkspaceFileRow - content: Buffer + owner: ProjectOwner + organizationId: string | null + secretProvenance: WorkspaceFileSecretProvenance | undefined } -interface ArtifactManifest { - source: WorkspaceFileRow - dependencies: WorkspaceFileRow[] +interface ArtifactPlan { + /** The decoded source, only for a page or generated document. */ + source?: string pageHtml?: string generatedDocument: boolean - secretProvenance: WorkspaceFileSecretProvenance + dependencyIds: string[] } interface PreparedArtifact { artifactKey?: string - manifest: ArtifactManifest + source: WorkspaceFileRow + dependencies: WorkspaceFileRow[] buffer: Buffer contentType: string dependsOnReferencedFiles: boolean @@ -81,6 +106,12 @@ interface ArtifactResult { receipt: FileReadReceipt } +/** + * Search requeues carried from the commit to the post-commit phase. The requeue holds the Project + * exclusively, so it never runs inside the read's transaction, which holds the Project shared. + */ +const renderedSearchArtifacts = new WeakMap() + function requireSameRevision(current: WorkspaceFileRow | undefined, expected: WorkspaceFileRow) { if ( !current || @@ -112,10 +143,96 @@ async function loadDependencies(tx: DbTransaction, owner: ProjectOwner, ids: str return rows } +/** Classifies the source bytes and names the same-Project inputs its render reads. */ +async function planArtifact( + file: WorkspaceFileRow, + content: Buffer, + owner: ProjectOwner +): Promise { + const pageCandidate = + file.contentType === SIM_PAGE_CONTENT_TYPE || file.originalName.toLowerCase().endsWith('.html') + const format = await getE2BDocFormat(file.originalName) + if (!pageCandidate && format === null) return { generatedDocument: false, dependencyIds: [] } + const source = content.toString('utf-8') + const page = pageCandidate && isSimPageSource(source) + let pageHtml: string | undefined + if (page) { + try { + pageHtml = renderSimPageDocument(source, { projectId: owner.entityId }) + } catch (error) { + if (error instanceof ProjectPageForeignEmbedError) + throw new OrchestrationError('not_found', 'Document input not found') + throw error + } + } + const generatedDocument = + !page && format !== null && !isCompiledDocumentBuffer(file.originalName, content) + const references = pageHtml ? collectSimPageFileReferences(pageHtml) : [] + if (references.some((reference) => reference.projectId && reference.projectId !== owner.entityId)) + throw new OrchestrationError('not_found', 'Document input not found') + const dependencyIds = [ + ...new Set( + pageHtml + ? references.map((reference) => reference.fileId) + : generatedDocument && format + ? collectReferencedFileIds( + source, + isDocSandboxEnabled + ? getDocumentSourceLanguage(source, format, file.contentType) + : 'javascript' + ) + : [] + ), + ] + if (dependencyIds.length > (pageHtml ? MAX_INLINE_IMAGE_REFERENCES : MAX_DOCUMENT_INPUTS)) + throw new OrchestrationError('payload_too_large', 'Document exceeds the referenced input limit') + return { source, pageHtml, generatedDocument, dependencyIds } +} + +/** + * Reads inputs with bounded fan-out. Declared sizes are checked before any read, and each read is + * capped at its declared size plus the unclaimed budget, so accurate rows never exceed the total. + */ +async function downloadDependencies( + dependencies: readonly WorkspaceFileRow[], + limits: { perFile: number; total: number }, + signal?: AbortSignal +) { + const declared = dependencies.map((dependency) => getWorkspaceFileSize(dependency)) + for (const size of declared) assertKnownSizeWithinLimit(size, limits.perFile, 'document input') + const declaredTotal = declared.reduce((sum, size) => sum + size, 0) + assertKnownSizeWithinLimit(declaredTotal, limits.total, 'document inputs') + const unclaimed = limits.total - declaredTotal + const failed = new AbortController() + const abort = signal ? AbortSignal.any([signal, failed.signal]) : failed.signal + let received = 0 + try { + return await mapWithConcurrency( + dependencies, + MATERIALIZE_CONCURRENCY, + async (dependency, index) => { + abort.throwIfAborted() + const content = await downloadFile({ + key: dependency.key, + context: 'project', + maxBytes: Math.min(limits.perFile, declared[index] + unclaimed), + signal: abort, + }) + received += content.length + assertKnownSizeWithinLimit(received, limits.total, 'document inputs') + return { fileId: dependency.id, contentType: dependency.contentType, content } + } + ) + } catch (error) { + failed.abort(error) + throw error + } +} + const snapshotSource = defineAuthorizedProjectFileUseCase< typeof projectFileOperations.readArtifact, ArtifactInput, - { file: WorkspaceFileRow; secretProvenance: WorkspaceFileSecretProvenance | undefined } + SourceSnapshot >({ operation: projectFileOperations.readArtifact, async execute({ @@ -129,7 +246,12 @@ const snapshotSource = defineAuthorizedProjectFileUseCase< const file = context.file if (!file) throw new OrchestrationError('not_found', 'File not found') const evidence = await getBoundWorkspaceFileSecretProvenanceByMetadata(tx, [file]) - return { file, secretProvenance: evidence.get(file.id) } + return { + file, + owner: context.owner, + organizationId: context.organizationId, + secretProvenance: evidence.get(file.id), + } }, afterSuccess: ({ principal, result }) => requireAdmittedContent(principal, { @@ -140,76 +262,131 @@ const snapshotSource = defineAuthorizedProjectFileUseCase< }), }) -const snapshotArtifact = defineAuthorizedProjectFileUseCase< +/** Locks the named inputs against the unchanged source and admits their merged provenance before any input read. */ +const snapshotDependencies = defineAuthorizedProjectFileUseCase< typeof projectFileOperations.readArtifact, - ArtifactInput & { source: SourceSnapshot }, - ArtifactManifest + ArtifactInput & { source: WorkspaceFileRow; dependencyIds: string[] }, + { dependencies: WorkspaceFileRow[]; secretProvenance: WorkspaceFileSecretProvenance } >({ operation: projectFileOperations.readArtifact, async execute({ input, context, tx }) { - requireSameRevision(context.file, input.source.file) - const source = input.source.content.toString('utf-8') - const page = - (input.source.file.contentType === SIM_PAGE_CONTENT_TYPE || - input.source.file.originalName.toLowerCase().endsWith('.html')) && - isSimPageSource(source) - const pageHtml = page - ? renderSimPageDocument(source, { projectId: context.projectId }) - : undefined - const format = await getE2BDocFormat(input.source.file.originalName) - const generatedDocument = - !page && - format !== null && - !isCompiledDocumentBuffer(input.source.file.originalName, input.source.content) - const references = pageHtml ? collectSimPageFileReferences(pageHtml) : [] - if ( - references.some( - (reference) => reference.projectId && reference.projectId !== context.projectId - ) - ) { - throw new OrchestrationError('not_found', 'Document input not found') - } - const ids = [ - ...new Set( - pageHtml - ? references.map((reference) => reference.fileId) - : generatedDocument && format - ? collectReferencedFileIds( - source, - isDocSandboxEnabled - ? getDocumentSourceLanguage(source, format, input.source.file.contentType) - : 'javascript' - ) - : [] - ), - ] - if (ids.length > (pageHtml ? 256 : 500)) - throw new OrchestrationError( - 'payload_too_large', - 'Document exceeds the referenced input limit' - ) - const dependencies = await loadDependencies(tx, context.owner, ids) + requireSameRevision(context.file, input.source) + const dependencies = await loadDependencies(tx, context.owner, input.dependencyIds) const evidence = await getBoundWorkspaceFileSecretProvenanceByMetadata(tx, [ - input.source.file, + input.source, ...dependencies, ]) return { - source: input.source.file, dependencies, - pageHtml, - generatedDocument, secretProvenance: mergeWorkspaceFileSecretProvenance(...evidence.values()), } }, - afterSuccess: ({ principal, result }) => + afterSuccess: ({ principal, input, result }) => requireAdmittedContent(principal, { kind: 'workspace-file', - id: result.source.id, + id: input.source.id, provenance: result.secretProvenance, view: 'opaque', }), }) +async function renderArtifact( + principal: Principal, + input: ArtifactInput, + snapshot: SourceSnapshot, + signal?: AbortSignal +): Promise { + const { file, owner } = snapshot + const artifact = await resolveDocumentRender( + file.originalName, + { maxBytes: input.maxBytes, signal }, + async (): Promise => { + const content = await downloadFile({ + key: file.key, + context: 'project', + maxBytes: input.maxBytes, + signal, + }) + const plan = await planArtifact(file, content, owner) + const dependencies = + plan.dependencyIds.length === 0 + ? [] + : ( + await snapshotDependencies.execute({ + principal, + input: { ...input, source: file, dependencyIds: plan.dependencyIds }, + }) + ).dependencies + const inputs = await downloadDependencies( + dependencies, + plan.pageHtml + ? { perFile: MAX_INLINE_IMAGE_BYTES, total: MAX_INLINE_TOTAL_BYTES } + : { perFile: MAX_DOCUMENT_INPUT_BYTES, total: MAX_DOCUMENT_INPUT_TOTAL_BYTES }, + signal + ) + if (plan.pageHtml) { + const byId = new Map(inputs.map((input) => [input.fileId, input])) + const page = await inlineSimPageImages( + plan.pageHtml, + async ({ fileId }) => { + const asset = byId.get(fileId) + if (!asset) throw new OrchestrationError('not_found', 'Document input not found') + return { bytes: asset.content, contentType: asset.contentType, identity: fileId } + }, + { strict: true } + ) + const buffer = Buffer.from(page.html, 'utf-8') + assertKnownSizeWithinLimit(buffer.length, input.maxBytes, 'rendered page') + return { + source: file, + dependencies, + buffer, + contentType: 'text/html', + dependsOnReferencedFiles: inputs.length > 0, + } + } + if (plan.generatedDocument && plan.source !== undefined) { + const compiled = await compileFileDocument({ + owner, + sourceMime: file.contentType, + source: plan.source, + fileName: file.originalName, + inputs, + inputIdentity: fileDocumentInputIdentity(owner, dependencies), + maxBytes: input.maxBytes, + signal, + }) + return { source: file, dependencies, ...compiled } + } + return { + source: file, + dependencies, + buffer: content, + contentType: file.contentType, + dependsOnReferencedFiles: false, + } + } + ) + const derivative = input.preview + ? await resolveServableImageBytes(artifact.buffer, file.key, { owner }) + : null + if (derivative) + assertKnownSizeWithinLimit(derivative.buffer.length, input.maxBytes, 'image preview') + return { ...artifact, ...derivative } +} + +/** Indexing is best effort: a contended or failed requeue never fails a delivered read. */ +async function requeueRenderedSearchArtifact(completion: FileSearchArtifactCompletion) { + try { + await requeueFileSearchForRenderedArtifact(completion) + } catch (error) { + logger.warn('Could not requeue file search for a rendered artifact', { + fileId: completion.file.fileId, + error: getErrorMessage(error), + }) + } +} + /** Renders only source-owned dependencies, then rechecks the source, every input, and current authority. */ export const readProjectFileArtifact = defineAuthorizedProjectFileUseCase< typeof projectFileOperations.readArtifact, @@ -218,7 +395,7 @@ export const readProjectFileArtifact = defineAuthorizedProjectFileUseCase< PreparedArtifact >({ operation: projectFileOperations.readArtifact, - async prepare({ principal, input, context }) { + async prepareFromSnapshot({ principal, input, request }) { if ( !Number.isSafeInteger(input.maxBytes) || input.maxBytes <= 0 || @@ -226,109 +403,24 @@ export const readProjectFileArtifact = defineAuthorizedProjectFileUseCase< ) { throw new OrchestrationError('validation', 'Invalid document byte limit') } - const { file } = await snapshotSource.execute({ principal, input }) - const artifact = await resolveDocumentRender( - file.originalName, - { maxBytes: input.maxBytes }, - async () => { - const content = await downloadFile({ - key: file.key, - context: 'project', - maxBytes: input.maxBytes, - }) - const manifest = await snapshotArtifact.execute({ - principal, - input: { ...input, source: { file, content } }, - }) - let remaining = manifest.pageHtml ? 32 * 1024 * 1024 : 50 * 1024 * 1024 - const inputs = [] - for (const dependency of manifest.dependencies) { - const bytes = await downloadFile({ - key: dependency.key, - context: 'project', - maxBytes: Math.min(remaining, manifest.pageHtml ? 8 * 1024 * 1024 : 25 * 1024 * 1024), - }) - remaining -= bytes.length - inputs.push({ - fileId: dependency.id, - contentType: dependency.contentType, - content: bytes, - }) - } - if (manifest.pageHtml) { - const byId = new Map(inputs.map((input) => [input.fileId, input])) - const page = await inlineSimPageImages( - manifest.pageHtml, - async ({ fileId }) => { - const asset = byId.get(fileId) - if (!asset) throw new OrchestrationError('not_found', 'Document input not found') - return { bytes: asset.content, contentType: asset.contentType, identity: fileId } - }, - { strict: true } - ) - const buffer = Buffer.from(page.html, 'utf-8') - assertKnownSizeWithinLimit(buffer.length, input.maxBytes, 'rendered page') - return { - manifest, - buffer, - contentType: 'text/html', - dependsOnReferencedFiles: inputs.length > 0, - } - } - if (manifest.generatedDocument) { - const inputIdentity = fileDocumentInputIdentity(context.owner, manifest.dependencies) - const compiled = await compileFileDocument({ - owner: context.owner, - sourceMime: file.contentType, - source: content.toString('utf-8'), - fileName: file.originalName, - inputs, - inputIdentity, - maxBytes: input.maxBytes, - }) - return { manifest, ...compiled } - } - return { - manifest, - buffer: content, - contentType: file.contentType, - dependsOnReferencedFiles: false, - } - } + const snapshot = await snapshotSource.execute({ principal, input }) + return runWithOutboundOrganization(snapshot.organizationId, () => + renderArtifact(principal, input, snapshot, request?.signal) ) - const derivative = input.preview - ? await resolveServableImageBytes(artifact.buffer, artifact.manifest.source.key, { - owner: context.owner, - }) - : null - if (derivative) - assertKnownSizeWithinLimit(derivative.buffer.length, input.maxBytes, 'image preview') - return { ...artifact, ...derivative } }, async execute({ input, context, tx, prepared }) { if (!prepared) throw new Error('Prepared document is unavailable') - requireSameRevision(context.file, prepared.manifest.source) + requireSameRevision(context.file, prepared.source) const dependencies = await loadDependencies( tx, context.owner, - prepared.manifest.dependencies.map((file) => file.id) + prepared.dependencies.map((file) => file.id) ) const currentById = new Map(dependencies.map((file) => [file.id, file])) - for (const expected of prepared.manifest.dependencies) + for (const expected of prepared.dependencies) requireSameRevision(currentById.get(expected.id), expected) const source = context.file if (!source) throw new OrchestrationError('not_found', 'File not found') - if (prepared.artifactKey) - await markFileSearchArtifactReadyInTx(tx, { - owner: context.owner, - file: { fileId: source.id, key: source.key, contentUpdatedAt: source.contentUpdatedAt }, - dependencies: dependencies.map((file) => ({ - fileId: file.id, - key: file.key, - sourceContentUpdatedAt: file.contentUpdatedAt, - })), - artifactKey: prepared.artifactKey, - }) const evidence = await getBoundWorkspaceFileSecretProvenanceByMetadata(tx, [ source, ...dependencies, @@ -343,7 +435,7 @@ export const readProjectFileArtifact = defineAuthorizedProjectFileUseCase< const folders = source.folderId ? await listFileFolders(context.owner, { scope: 'all' }, tx) : [] - return { + const result = { file: mapFileRecord(source, context.owner, buildWorkspaceFileFolderPathMap(folders)), buffer: prepared.buffer, contentType: prepared.contentType, @@ -351,6 +443,18 @@ export const readProjectFileArtifact = defineAuthorizedProjectFileUseCase< secretProvenance, receipt: createFileReadReceipt(context.owner, [source, ...dependencies]), } + if (prepared.artifactKey) + renderedSearchArtifacts.set(result, { + owner: context.owner, + file: { fileId: source.id, key: source.key, contentUpdatedAt: source.contentUpdatedAt }, + dependencies: dependencies.map((file) => ({ + fileId: file.id, + key: file.key, + sourceContentUpdatedAt: file.contentUpdatedAt, + })), + artifactKey: prepared.artifactKey, + }) + return result }, async afterSuccess({ principal, result }) { await requireAdmittedContent(principal, { @@ -360,5 +464,7 @@ export const readProjectFileArtifact = defineAuthorizedProjectFileUseCase< view: 'opaque', }) await reportWorkspaceFileDelivery(result.secretProvenance, result.file.id) + const completion = renderedSearchArtifacts.get(result) + if (completion) await requeueRenderedSearchArtifact(completion) }, }) diff --git a/apps/sim/lib/projects/files/application/previews.ts b/apps/sim/lib/projects/files/application/previews.ts index 9b51da02bc8..d7432c65523 100644 --- a/apps/sim/lib/projects/files/application/previews.ts +++ b/apps/sim/lib/projects/files/application/previews.ts @@ -2,7 +2,9 @@ import type { Principal } from '@sim/auth/principal' import { type WorkspaceFileRow, workspaceFiles } from '@sim/db/schema' import { and, eq, isNull } from 'drizzle-orm' import { requireAdmittedContent } from '@/lib/core/application/content-delivery' +import { runWithOutboundOrganization } from '@/lib/core/network/context.server' import { OrchestrationError } from '@/lib/core/orchestration/types' +import { nodeReadableToWebStream } from '@/lib/core/utils/node-stream' import type { DbTransaction } from '@/lib/db/types' import { getCsvPreviewSlice } from '@/lib/file-parsers/csv-preview-slice' import type { @@ -11,17 +13,12 @@ import type { } from '@/lib/projects/files/application/authorization' import { defineAuthorizedProjectFileUseCase } from '@/lib/projects/files/application/authorized-use-case' import { projectFileOperations } from '@/lib/projects/files/application/operations' -import { - buildWorkspaceFileFolderPathMap, - listFileFolders, - mapFileRecord, -} from '@/lib/uploads/contexts/workspace' import { getBoundWorkspaceFileSecretProvenanceByMetadata, type WorkspaceFileSecretProvenance, } from '@/lib/uploads/contexts/workspace/file-secret-provenance' -import { downloadFile } from '@/lib/uploads/core/storage-service' -import { MAX_BUFFERED_TRANSFER_BYTES } from '@/lib/uploads/shared/types' +import { downloadFileStream } from '@/lib/uploads/core/storage-service' +import { getWorkspaceFileSize } from '@/lib/uploads/shared/types' import { reportWorkspaceFileDelivery } from '@/lib/workspace-files/application/file-delivery-observer' import { fileOwnerCondition } from '@/lib/workspace-files/ownership-query' @@ -35,7 +32,11 @@ interface InlineInput { const resolveInlineRecord = defineAuthorizedProjectFileUseCase< typeof projectFileOperations.readInline, InlineInput, - { file: WorkspaceFileRow; secretProvenance: WorkspaceFileSecretProvenance | undefined } + { + file: WorkspaceFileRow + organizationId: string | null + secretProvenance: WorkspaceFileSecretProvenance | undefined + } >({ operation: projectFileOperations.readInline, async execute({ @@ -65,7 +66,11 @@ const resolveInlineRecord = defineAuthorizedProjectFileUseCase< .limit(1) if (!file) throw new OrchestrationError('not_found', 'File not found') const evidence = await getBoundWorkspaceFileSecretProvenanceByMetadata(tx, [file]) - return { file, secretProvenance: evidence.get(file.id) } + return { + file, + organizationId: context.organizationId, + secretProvenance: evidence.get(file.id), + } }, afterSuccess: ({ principal, input, result }) => requireAdmittedContent(principal, { @@ -76,63 +81,48 @@ const resolveInlineRecord = defineAuthorizedProjectFileUseCase< }), }) -/** Resolves a private object under its Project before reading, then fences its current head again. */ -const readInlineFile = defineAuthorizedProjectFileUseCase({ - operation: projectFileOperations.readInline, - async prepare({ principal, input }: { principal: Principal; input: InlineInput }) { - const { file } = await resolveInlineRecord.execute({ principal, input }) - const content = await downloadFile({ - key: file.key, - context: 'project', - maxBytes: MAX_BUFFERED_TRANSFER_BYTES, - }) - return { file, content } +type ResolveInlineArgs = Parameters[0] + +/** + * Streams a private object resolved under its Project, then fences its current head after storage + * opens. A key request names the exact immutable object it streamed, so it is content-addressed. + */ +export const readProjectInlineFile = { + operation: resolveInlineRecord.operation, + delegationAudience: resolveInlineRecord.delegationAudience, + /** HEAD resolves the canonical image without fetching its bytes. */ + async authorize(args: ResolveInlineArgs) { + await resolveInlineRecord.execute(args) }, - async execute({ context, tx, prepared }) { - if (!prepared) throw new Error('Inline file bytes are unavailable') - const [file] = await tx - .select() - .from(workspaceFiles) - .where( - and( - fileOwnerCondition(context.owner), - eq(workspaceFiles.id, prepared.file.id), - isNull(workspaceFiles.deletedAt) - ) + async execute(args: ResolveInlineArgs) { + const snapshot = await resolveInlineRecord.execute(args) + const stream = nodeReadableToWebStream( + await runWithOutboundOrganization(snapshot.organizationId, () => + downloadFileStream({ key: snapshot.file.key, context: 'project' }) ) - .for('share') - .limit(1) - if ( - !file || - file.key !== prepared.file.key || - file.contentUpdatedAt.getTime() !== prepared.file.contentUpdatedAt.getTime() ) - throw new OrchestrationError('conflict', 'File changed while preparing its preview') - const evidence = await getBoundWorkspaceFileSecretProvenanceByMetadata(tx, [file]) - const folders = file.folderId ? await listFileFolders(context.owner, { scope: 'all' }, tx) : [] - return { - file: mapFileRecord(file, context.owner, buildWorkspaceFileFolderPathMap(folders)), - content: prepared.content, - secretProvenance: evidence.get(file.id), + try { + const { file, secretProvenance } = await resolveInlineRecord.execute({ + ...args, + input: { projectId: args.input.projectId, referenceFileId: snapshot.file.id }, + }) + if ( + file.key !== snapshot.file.key || + file.contentUpdatedAt.getTime() !== snapshot.file.contentUpdatedAt.getTime() + ) + throw new OrchestrationError('conflict', 'File changed while preparing its preview') + await reportWorkspaceFileDelivery(secretProvenance, file.id) + return { + file: { id: file.id, name: file.originalName, type: file.contentType }, + stream, + contentLength: getWorkspaceFileSize(file), + contentAddressed: args.input.key !== undefined, + } + } catch (error) { + await stream.cancel(error).catch(() => undefined) + throw error } }, - async afterSuccess({ principal, result }) { - await requireAdmittedContent(principal, { - kind: 'workspace-file', - id: result.file.id, - provenance: result.secretProvenance ?? { status: 'unknown' }, - view: 'opaque', - }) - await reportWorkspaceFileDelivery(result.secretProvenance, result.file.id) - }, -}) - -/** HEAD resolves the canonical image without fetching its bytes. */ -export const readProjectInlineFile = { - ...readInlineFile, - async authorize(args: Parameters[0]) { - await resolveInlineRecord.execute(args) - }, } interface CsvPreviewInput extends ProjectFileTarget { @@ -144,29 +134,23 @@ interface CsvPreviewInput extends ProjectFileTarget { /** Reads only the bounded CSV slice and rejects head changes before exposing its rows. */ export const readProjectFileCsvPreview = defineAuthorizedProjectFileUseCase({ operation: projectFileOperations.readContent, - async prepare({ + async prepareFromSnapshot({ input, - context, principal, }: { principal: Principal input: CsvPreviewInput - context: ProjectFileAuthorizationContext }) { - const file = context.file - if (!file || file.key !== input.key) throw new OrchestrationError('not_found', 'File not found') - const { file: admitted } = await resolveInlineRecord.execute({ + const { file, organizationId } = await resolveInlineRecord.execute({ principal, - input: { projectId: input.projectId, referenceFileId: file.id, view: 'excerpt' }, + input: { projectId: input.projectId, referenceFileId: input.fileId, view: 'excerpt' }, }) - if ( - admitted.key !== file.key || - admitted.contentUpdatedAt.getTime() !== file.contentUpdatedAt.getTime() - ) - throw new OrchestrationError('conflict', 'File changed while preparing its preview') + if (file.key !== input.key) throw new OrchestrationError('not_found', 'File not found') return { file, - slice: await getCsvPreviewSlice({ key: file.key, context: 'project', signal: input.signal }), + slice: await runWithOutboundOrganization(organizationId, () => + getCsvPreviewSlice({ key: file.key, context: 'project', signal: input.signal }) + ), } }, async execute({ input, context, tx, prepared }) { diff --git a/apps/sim/lib/projects/files/application/search.ts b/apps/sim/lib/projects/files/application/search.ts index bc793df2d8d..341d63a9171 100644 --- a/apps/sim/lib/projects/files/application/search.ts +++ b/apps/sim/lib/projects/files/application/search.ts @@ -1,15 +1,14 @@ import { getPostgresErrorCode } from '@sim/utils/errors' -import { admitEach } from '@/lib/core/application/content-delivery' +import { admitEach, mustClassifyContent } from '@/lib/core/application/content-delivery' import { OrchestrationError } from '@/lib/core/orchestration/types' import { defineAuthorizedProjectFileUseCase } from '@/lib/projects/files/application/authorized-use-case' -import { listProjectFileFolders } from '@/lib/projects/files/application/folders' import { projectFileOperations } from '@/lib/projects/files/application/operations' import { listFileFolders } from '@/lib/uploads/contexts/workspace' +import { mergeWorkspaceFileSecretProvenance } from '@/lib/uploads/contexts/workspace/file-secret-provenance' import { - mergeWorkspaceFileSecretProvenance, - type WorkspaceFileSecretProvenance, -} from '@/lib/uploads/contexts/workspace/file-secret-provenance' -import { reportWorkspaceFileDelivery } from '@/lib/workspace-files/application/file-delivery-observer' + hasWorkspaceFileDeliveryObserver, + reportWorkspaceFileDelivery, +} from '@/lib/workspace-files/application/file-delivery-observer' import type { SearchWorkspaceFileContentInput } from '@/lib/workspace-files/application/search-workspace-file-content' import { loadFileSearchDelivery, @@ -22,31 +21,24 @@ import { } from '@/lib/workspace-files/search/pattern' import { type FileSearchResult, searchFileIndex } from '@/lib/workspace-files/search/repository' -export interface SearchProjectFileContentInput +interface SearchProjectFileContentInput extends Omit { projectId: string } -export const searchProjectFileContent = defineAuthorizedProjectFileUseCase< +/** Searches the index, then validates the snapshot against current rows under Project authority. */ +const snapshotSearch = defineAuthorizedProjectFileUseCase< typeof projectFileOperations.searchContent, SearchProjectFileContentInput, - FileSearchResult & { secretProvenance: WorkspaceFileSecretProvenance }, + { prepared: FileSearchResult; delivery: Awaited> }, FileSearchResult >({ operation: projectFileOperations.searchContent, isolationLevel: 'repeatable read', - async prepare({ input, principal, context, request }) { + async prepare({ input, context, request }) { const signal = input.signal ?? request?.signal signal?.throwIfAborted() - const folders = - input.folderPaths === undefined - ? [] - : ( - await listProjectFileFolders.execute({ - principal, - input: { projectId: context.projectId }, - }) - ).folders + const folders = input.folderPaths === undefined ? [] : await listFileFolders(context.owner) try { return await searchFileIndex({ owner: context.owner, @@ -70,45 +62,12 @@ export const searchProjectFileContent = defineAuthorizedProjectFileUseCase< input.folderPaths === undefined ? [] : await listFileFolders(context.owner, {}, tx) const delivery = await loadFileSearchDelivery(tx, { owner: context.owner, - classify: true, + classify: mustClassifyContent(principal) || hasWorkspaceFileDeliveryObserver(), prepared, scope: resolveFileSearchFolderScope(folders, input), signal: input.signal ?? request?.signal, }) - let secretProvenance: WorkspaceFileSecretProvenance = { status: 'exact', entries: [] } - const current = prepared.sources.filter( - (source) => - !delivery.outOfScope.has(source.identity.fileId) && - delivery.provenance.has(source.identity.fileId) - ) - const admitted = await admitEach(principal, current, async (sources) => - sources.map((source) => ({ - kind: 'workspace-file' as const, - id: source.identity.fileId, - provenance: delivery.provenance.get(source.identity.fileId) ?? { - status: 'unknown' as const, - }, - view: 'excerpt' as const, - })) - ) - const admittedIds = new Set(admitted.map((source) => source.identity.fileId)) - const results = prepared.results - .filter((hit) => !delivery.outOfScope.has(hit.fileId)) - .map((hit) => - admittedIds.has(hit.fileId) - ? hit - : { - ...hit, - text: '[Preview withheld: read or grep this file to inspect protected text.]', - } - ) - for (const source of admitted) { - secretProvenance = mergeWorkspaceFileSecretProvenance( - secretProvenance, - delivery.provenance.get(source.identity.fileId) ?? { status: 'unknown' } - ) - } - return { ...prepared, results, count: results.length, sources: admitted, secretProvenance } + return { prepared, delivery } } catch (error) { if (['55P03', '25P04'].includes(getPostgresErrorCode(error) ?? '')) { throw new OrchestrationError( @@ -119,5 +78,44 @@ export const searchProjectFileContent = defineAuthorizedProjectFileUseCase< throw error } }, - afterSuccess: ({ result }) => reportWorkspaceFileDelivery(result.secretProvenance), }) + +/** Admission runs after the snapshot commits, as on every Project read; row and Project locks would fence nothing. */ +export const searchProjectFileContent = { + operation: snapshotSearch.operation, + delegationAudience: snapshotSearch.delegationAudience, + authorize: snapshotSearch.authorize, + async execute(args: Parameters[0]): Promise { + const { prepared, delivery } = await snapshotSearch.execute(args) + const provenanceOf = (fileId: string) => + delivery.provenance.get(fileId) ?? { status: 'unknown' as const } + const current = prepared.sources.filter((source) => + delivery.current.has(source.identity.fileId) + ) + const admitted = await admitEach(args.principal, current, async (sources) => + sources.map((source) => ({ + kind: 'workspace-file' as const, + id: source.identity.fileId, + provenance: provenanceOf(source.identity.fileId), + view: 'excerpt' as const, + })) + ) + const admittedIds = new Set(admitted.map((source) => source.identity.fileId)) + const results = prepared.results + .filter((hit) => !delivery.outOfScope.has(hit.fileId)) + .map((hit) => + admittedIds.has(hit.fileId) + ? hit + : { + ...hit, + text: '[Preview withheld: read or grep this file to inspect protected text.]', + } + ) + await reportWorkspaceFileDelivery( + mergeWorkspaceFileSecretProvenance( + ...admitted.map((source) => provenanceOf(source.identity.fileId)) + ) + ) + return { ...prepared, results, count: results.length, sources: admitted } + }, +} diff --git a/apps/sim/lib/projects/membership.ts b/apps/sim/lib/projects/membership.ts index 54aa3b522a3..840296c86e2 100644 --- a/apps/sim/lib/projects/membership.ts +++ b/apps/sim/lib/projects/membership.ts @@ -22,17 +22,19 @@ import { ProjectConflictError } from '@/lib/projects/errors' const PROJECT_LOCK_TIMEOUT_MS = 5_000 /** - * Waits in `acquire` are bounded by {@link PROJECT_LOCK_TIMEOUT_MS} and a timeout or - * deadlock becomes a retryable Project conflict. The caller's own `lock_timeout` is - * restored afterwards, so the bound never leaks into the work done under the locks. + * Waits in `acquire` are bounded by `timeoutMs` ({@link PROJECT_LOCK_TIMEOUT_MS} unless a caller + * needs a shorter wait) and a timeout or deadlock becomes a retryable Project conflict. The + * caller's own `lock_timeout` is restored afterwards, so the bound never leaks into the work done + * under the locks. */ async function withProjectLockTimeout( tx: DbTransaction, message: string, - acquire: () => Promise + acquire: () => Promise, + timeoutMs = PROJECT_LOCK_TIMEOUT_MS ): Promise { try { - return await withLockTimeout(tx, PROJECT_LOCK_TIMEOUT_MS, acquire) + return await withLockTimeout(tx, timeoutMs, acquire) } catch (error) { const code = getPostgresErrorCode(error) if (code === '55P03' || code === '40P01') throw new ProjectConflictError(message) @@ -71,9 +73,16 @@ export async function lockProjectBackfillWrites( } /** Canonical Project mutex; membership and lifecycle writers hold it until commit. */ -export async function lockProject(tx: DbTransaction, projectId: string): Promise { - await withProjectLockTimeout(tx, PROJECT_CHANGING, () => - acquireAdvisoryXactLock(tx, 'project', projectLockKey(projectId)) +export async function lockProject( + tx: DbTransaction, + projectId: string, + options: { timeoutMs?: number } = {} +): Promise { + await withProjectLockTimeout( + tx, + PROJECT_CHANGING, + () => acquireAdvisoryXactLock(tx, 'project', projectLockKey(projectId)), + options.timeoutMs ) } diff --git a/apps/sim/lib/workspace-files/page-compile.ts b/apps/sim/lib/workspace-files/page-compile.ts index 9c8804f9605..6c41cd3da03 100644 --- a/apps/sim/lib/workspace-files/page-compile.ts +++ b/apps/sim/lib/workspace-files/page-compile.ts @@ -246,6 +246,14 @@ const FILE_RESOURCE_ROUTES: Record< : `/projects/${entityId}/files/${fileId}`, } +/** A Project page embeds a file another owner holds; servers report it as a missing input. */ +export class ProjectPageForeignEmbedError extends Error { + constructor() { + super('Project pages can only embed files owned by the same Project') + this.name = 'ProjectPageForeignEmbedError' + } +} + function resolveSimFileLinks(html: string, owner: EditableFileOwner | undefined, baseUrl: string) { return html.replace( /\b(src|href)="sim:file\/([^"]+)"/g, @@ -257,7 +265,7 @@ function resolveSimFileLinks(html: string, owner: EditableFileOwner | undefined, owner?.entityType === 'project' && (parsed.owner?.entityType !== 'project' || parsed.owner.entityId !== owner.entityId) ) { - throw new Error('Project pages can only embed files owned by the same Project') + throw new ProjectPageForeignEmbedError() } const path = parsed.owner ? FILE_RESOURCE_ROUTES[parsed.owner.entityType]( diff --git a/apps/sim/lib/workspace-files/page-document.server.ts b/apps/sim/lib/workspace-files/page-document.server.ts index 118db37cf6f..a915cfff2cb 100644 --- a/apps/sim/lib/workspace-files/page-document.server.ts +++ b/apps/sim/lib/workspace-files/page-document.server.ts @@ -4,17 +4,17 @@ import { getFileMetadataById } from '@/lib/uploads/server/metadata' import { renderSimPageDocument } from '@/lib/workspace-files/page-document' /** Images past this size stay as URL references rather than bloating the document. */ -const MAX_INLINE_IMAGE_BYTES = 8 * 1024 * 1024 +export const MAX_INLINE_IMAGE_BYTES = 8 * 1024 * 1024 /** * Ceiling on everything a single document inlines. A per-image limit does not bound * the page on its own — N images each just under it still cost N times it. Images * that do not fit what is left keep their URL reference, exactly like an oversized one. */ -const MAX_INLINE_TOTAL_BYTES = 32 * 1024 * 1024 +export const MAX_INLINE_TOTAL_BYTES = 32 * 1024 * 1024 /** Bounds metadata reads even when the page references many missing or empty images. */ -const MAX_INLINE_IMAGE_REFERENCES = 256 +export const MAX_INLINE_IMAGE_REFERENCES = 256 const IMAGE_SRC = /src="[^"]*\/api\/(?:files\/view\/([^"]+)|projects\/([^/"]+)\/files\/([^/"]+)\/content)"/g diff --git a/apps/sim/lib/workspace-files/search/artifact-ready.ts b/apps/sim/lib/workspace-files/search/artifact-ready.ts index 64de6fabd3a..96e1b83fae9 100644 --- a/apps/sim/lib/workspace-files/search/artifact-ready.ts +++ b/apps/sim/lib/workspace-files/search/artifact-ready.ts @@ -9,6 +9,7 @@ import type { DbTransaction } from '@/lib/db/types' import type { EditableFileOwner } from '@/lib/workspace-files/ownership' import { FILE_SEARCH_ARTIFACT_READY_LOCK_TIMEOUT_MS } from '@/lib/workspace-files/search/constants' import { lockFileSearchInputs } from '@/lib/workspace-files/search/index-state' +import { lockFileSearchOwners } from '@/lib/workspace-files/search/owner-policy' import { fileSearchOwnerCondition, fileSearchOwnerFields, @@ -82,7 +83,8 @@ async function isFileSearchArtifactCurrent( /** * Requeues search for a freshly rendered artifact unless its revision already uses it. The common - * already-current case takes no locks; the locked requeue is bounded by a short lock timeout. + * already-current case takes no locks; the locked requeue, owner lock included, is bounded by a + * short lock timeout so a contended owner fails fast instead of stalling the read. */ export async function requeueFileSearchForRenderedArtifact( input: FileSearchArtifactCompletion @@ -93,6 +95,10 @@ export async function requeueFileSearchForRenderedArtifact( await configureFileSearchTransaction(tx, { lockTimeout: FILE_SEARCH_ARTIFACT_READY_LOCK_TIMEOUT_MS, }) + // Held here under the short bound; the reentrant owner lock in the mark then returns at once. + await lockFileSearchOwners(tx, [owner], { + lockTimeoutMs: FILE_SEARCH_ARTIFACT_READY_LOCK_TIMEOUT_MS, + }) await markFileSearchArtifactReadyInTx(tx, input) }) } diff --git a/apps/sim/lib/workspace-files/search/delivery.ts b/apps/sim/lib/workspace-files/search/delivery.ts index 306b3e577ee..51a87967f4b 100644 --- a/apps/sim/lib/workspace-files/search/delivery.ts +++ b/apps/sim/lib/workspace-files/search/delivery.ts @@ -47,9 +47,9 @@ export function resolveFileSearchFolderScope( /** * Validates a private index snapshot against current rows after its caller authorizes the owner. * Each source is judged on its own: one that left the folder scope is listed in `outOfScope` for - * the caller to drop, and one whose file, revision, or inputs changed since the search gets - * `unknown` provenance so its stale preview is withheld. Only inputs of current sources are - * returned for access checks. + * the caller to drop, and only one whose file, revision, and inputs are unchanged since the search + * is listed in `current`; any other has a stale preview to withhold. `classify` adds the merged + * provenance of each current source. Only inputs of current sources are returned for access checks. */ export async function loadFileSearchDelivery( tx: DbTransaction, @@ -67,6 +67,7 @@ export async function loadFileSearchDelivery( classify: boolean } ): Promise<{ + current: Set provenance: Map outOfScope: Set inputs: { fileId: string; fileContext: string }[] @@ -190,6 +191,7 @@ export async function loadFileSearchDelivery( current.flatMap((source) => source.dependencies.map((input) => input.fileId)) ) return { + current: new Set(current.map((source) => source.identity.fileId)), provenance: sources, outOfScope, inputs: rows diff --git a/apps/sim/lib/workspace-files/search/owner-policy.ts b/apps/sim/lib/workspace-files/search/owner-policy.ts index f30bcd41892..218e7af48ac 100644 --- a/apps/sim/lib/workspace-files/search/owner-policy.ts +++ b/apps/sim/lib/workspace-files/search/owner-policy.ts @@ -9,7 +9,7 @@ import { import type { EditableFileOwner } from '@/lib/workspace-files/ownership' const policies: FileOwnerAdapters<{ - lock(tx: DbTransaction, entityId: string): Promise + lock(tx: DbTransaction, entityId: string, timeoutMs?: number): Promise tryLock(tx: DbTransaction, entityId: string): Promise }> = { workspace: { @@ -19,7 +19,7 @@ const policies: FileOwnerAdapters<{ }, }, project: { - lock: lockProject, + lock: (tx, id, timeoutMs) => lockProject(tx, id, { timeoutMs }), tryLock: (tx, id) => tryAcquireAdvisoryXactLock(tx, 'file_search_dispatch', `project:${id}`), }, } @@ -28,7 +28,7 @@ const policies: FileOwnerAdapters<{ export async function lockFileSearchOwners( tx: DbTransaction, owners: readonly EditableFileOwner[], - options: { skipBusy?: boolean } = {} + options: { skipBusy?: boolean; lockTimeoutMs?: number } = {} ): Promise { const ordered = [ ...new Map( @@ -44,7 +44,7 @@ export async function lockFileSearchOwners( if (options.skipBusy) { if (!(await policy.tryLock(tx, owner.entityId))) continue } else { - await policy.lock(tx, owner.entityId) + await policy.lock(tx, owner.entityId, options.lockTimeoutMs) } acquired.push(owner) } diff --git a/knip.jsonc b/knip.jsonc index a5142e18537..ce97194abb7 100644 --- a/knip.jsonc +++ b/knip.jsonc @@ -48,10 +48,10 @@ // Generated contracts mirror their source of truth; regenerating them must not // trip the unused-export ratchet, and hand edits would be overwritten. "lib/mothership/generated/**": ["exports", "types", "duplicates"], - // Consumed only by the stacked Project file PRs: the rendered-content contracts by #9017, - // the client aliases by #9034. Remove each entry once its consumer merges. + // Client aliases consumed only by the stacked Project file PR #9034. Remove each entry once + // its consumer merges. "lib/api/contracts/projects.ts": ["types"], - "lib/api/contracts/project-files.ts": ["exports", "types"], + "lib/api/contracts/project-files.ts": ["types"], "lib/api/contracts/project-file-folders.ts": ["types"], "lib/api/contracts/project-file-lifecycle.ts": ["types"], "sandboxes/index.ts": ["files"], From 2383eb9ceae29ceca7fdf7229b4cf3f57aba2c56 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 18:43:10 -0700 Subject: [PATCH 2/2] test(project-files): start requeue checks from no search revision --- .../content-rendered.integration.ts | 26 ++++++++++++------- .../projects/files/application/artifacts.ts | 1 + 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/apps/sim/lib/projects/files/__integration__/content-rendered.integration.ts b/apps/sim/lib/projects/files/__integration__/content-rendered.integration.ts index 89839ae0871..10298483fdf 100644 --- a/apps/sim/lib/projects/files/__integration__/content-rendered.integration.ts +++ b/apps/sim/lib/projects/files/__integration__/content-rendered.integration.ts @@ -143,6 +143,19 @@ async function fixture() { return { ...ids, principal: createSessionPrincipal({ userId: editorId }) } } +/** File creation marks a revision pending, so a requeue check starts from no revision at all. */ +async function clearSearchRevision(fileId: string) { + await db.delete(workspaceFileSearchRevision).where(eq(workspaceFileSearchRevision.fileId, fileId)) +} + +async function searchRevision(fileId: string) { + const [revision] = await db + .select() + .from(workspaceFileSearchRevision) + .where(eq(workspaceFileSearchRevision.fileId, fileId)) + return revision +} + function createInput(projectId: string, content = 'Project architecture') { return { projectId, @@ -704,6 +717,7 @@ describe('Project rendered artifacts against PostgreSQL and private storage', () await released }) await held + await clearSearchRevision(source.file.id) try { const rendered = await observeWorkspaceFileDelivery( async () => release(), @@ -718,10 +732,7 @@ describe('Project rendered artifacts against PostgreSQL and private storage', () release() await concurrentReader } - const [revision] = await db - .select() - .from(workspaceFileSearchRevision) - .where(eq(workspaceFileSearchRevision.fileId, source.file.id)) + const revision = await searchRevision(source.file.id) expect(revision?.status).toBe('pending') expect(revision?.sourceContentUpdatedAt).toEqual(source.file.contentUpdatedAt) } @@ -755,6 +766,7 @@ describe('Project rendered artifacts against PostgreSQL and private storage', () await released }) await held + await clearSearchRevision(source.file.id) try { const started = performance.now() const rendered = await readProjectFileArtifact.execute({ @@ -769,11 +781,7 @@ describe('Project rendered artifacts against PostgreSQL and private storage', () release() await concurrentReader } - const revisions = await db - .select() - .from(workspaceFileSearchRevision) - .where(eq(workspaceFileSearchRevision.fileId, source.file.id)) - expect(revisions).toEqual([]) + expect(await searchRevision(source.file.id)).toBeUndefined() } ) diff --git a/apps/sim/lib/projects/files/application/artifacts.ts b/apps/sim/lib/projects/files/application/artifacts.ts index 63cc36d03d9..b356ebda15e 100644 --- a/apps/sim/lib/projects/files/application/artifacts.ts +++ b/apps/sim/lib/projects/files/application/artifacts.ts @@ -465,6 +465,7 @@ export const readProjectFileArtifact = defineAuthorizedProjectFileUseCase< }) await reportWorkspaceFileDelivery(result.secretProvenance, result.file.id) const completion = renderedSearchArtifacts.get(result) + renderedSearchArtifacts.delete(result) if (completion) await requeueRenderedSearchArtifact(completion) }, })