From 88d6b47be876c02d8c39efe2bf284a08e7c8a60a Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 6 Jul 2026 21:43:26 +0800 Subject: [PATCH] Keep parsed storage sync non-blocking --- .../sources/[sourceId]/chunks/route.test.ts | 2 + src/domains/chunks/index.ts | 4 +- src/domains/chunks/read.test.ts | 11 +- src/domains/chunks/read.ts | 6 +- .../parsed-sync-route-workflow.test.ts | 86 +++----- .../sources/parsed-sync-route-workflow.ts | 67 +++--- .../source-reconcile-route-workflow.test.ts | 152 +++----------- .../source-reconcile-route-workflow.ts | 195 +++++++----------- src/proxy.test.ts | 10 + src/proxy.ts | 1 + 10 files changed, 185 insertions(+), 349 deletions(-) diff --git a/src/app/api/sources/[sourceId]/chunks/route.test.ts b/src/app/api/sources/[sourceId]/chunks/route.test.ts index 2f0851c..85fb1bd 100644 --- a/src/app/api/sources/[sourceId]/chunks/route.test.ts +++ b/src/app/api/sources/[sourceId]/chunks/route.test.ts @@ -498,6 +498,7 @@ describe("GET /api/sources/[sourceId]/chunks", () => { revisionKey: "job_1", page: 1, pageSize: 1, + assetUrlPolicy: "durable", }) }) @@ -659,6 +660,7 @@ describe("GET /api/sources/[sourceId]/chunks", () => { revisionKey: "job_result_1", page: 1, pageSize: 1, + assetUrlPolicy: "durable", }) }) }) diff --git a/src/domains/chunks/index.ts b/src/domains/chunks/index.ts index 56c2644..eca1830 100644 --- a/src/domains/chunks/index.ts +++ b/src/domains/chunks/index.ts @@ -181,8 +181,8 @@ export function toParsedChunkView( /** * Map an SDK `KnowledgeReadChunk` (from `knowledge.readChunks`) to the view - * model. Display reads do not request durable asset URL hardening; chat - * hardens the specific assets it returns separately. + * model. Display reads request durable asset URLs from the SDK; chat still + * hardens only the specific assets it returns separately. */ export function toParsedChunkViewFromReadChunk( chunk: KnowledgeReadChunk, diff --git a/src/domains/chunks/read.test.ts b/src/domains/chunks/read.test.ts index b6a3979..e3c1000 100644 --- a/src/domains/chunks/read.test.ts +++ b/src/domains/chunks/read.test.ts @@ -20,7 +20,7 @@ function makeReadChunk(overrides: Record = {}) { } describe("readSourceChunkPage", () => { - it("reads a page without durable asset hardening and maps chunks to the view model", async () => { + it("reads a page with durable asset URLs and maps chunks to the view model", async () => { const readChunks = vi.fn(async () => ({ document: { localDocumentId: "doc_1" }, chunks: [ @@ -48,6 +48,7 @@ describe("readSourceChunkPage", () => { revisionKey: "rev_1", page: 2, pageSize: 50, + assetUrlPolicy: "durable", }) expect(result.pagination).toEqual({ page: 2, @@ -84,6 +85,7 @@ describe("readSourceChunkPage", () => { documentId: "doc_1", page: 1, pageSize: 50, + assetUrlPolicy: "durable", }) }) }) @@ -116,6 +118,13 @@ describe("readAllSourceChunks", () => { }) expect(readChunks).toHaveBeenCalledTimes(2) + expect(readChunks).toHaveBeenNthCalledWith(1, { + documentId: "doc_1", + revisionKey: "rev_1", + page: 1, + pageSize: 200, + assetUrlPolicy: "durable", + }) expect(chunks.map((chunk) => chunk.parserChunkId)).toEqual(["c1", "c2"]) }) }) diff --git a/src/domains/chunks/read.ts b/src/domains/chunks/read.ts index 320d559..1bf4790 100644 --- a/src/domains/chunks/read.ts +++ b/src/domains/chunks/read.ts @@ -16,8 +16,8 @@ type ReadableSource = { /** * Read a single display page of parsed chunks through the SDK. The SDK serves * from configured Blob storage when fresh and falls back to Knowhere remote - * otherwise. Display reads intentionally do not request durable asset URLs: - * chat hardens only the specific assets it sends back to the user. + * otherwise. Display reads request durable asset URLs so media chunks remain + * usable whether the page came from Blob or the remote Knowhere fallback. */ export async function readSourceChunkPage(input: { readonly knowledge: Knowledge @@ -29,6 +29,7 @@ export async function readSourceChunkPage(input: { ...(input.source.revisionKey ? { revisionKey: input.source.revisionKey } : {}), page: input.params.page, pageSize: input.params.pageSize, + assetUrlPolicy: "durable", }) const chunks = response.chunks.map((chunk) => @@ -68,6 +69,7 @@ export async function readAllSourceChunks(input: { : {}), page, pageSize: loadAllPageSize, + assetUrlPolicy: "durable", }) for (const chunk of response.chunks) { chunks.push( diff --git a/src/domains/sources/parsed-sync-route-workflow.test.ts b/src/domains/sources/parsed-sync-route-workflow.test.ts index 91095f5..670ee67 100644 --- a/src/domains/sources/parsed-sync-route-workflow.test.ts +++ b/src/domains/sources/parsed-sync-route-workflow.test.ts @@ -5,10 +5,6 @@ const mocks = vi.hoisted(() => ({ makeKnowhereClientWithParsedStorage: vi.fn(), releaseSyncCapacity: vi.fn(), updateSyncStatus: vi.fn(), - findInWorkspace: vi.fn(), - markFailed: vi.fn(), - markReady: vi.fn(), - markSourceReadyAfterReconciliation: vi.fn(), loggerInfo: vi.fn(), loggerError: vi.fn(), })) @@ -20,16 +16,9 @@ vi.mock("@/integrations/knowhere", () => ({ vi.mock("./workflow-runtime", () => ({ sourceWorkflowRuntime: { updateSyncStatus: mocks.updateSyncStatus, - findInWorkspace: mocks.findInWorkspace, - markFailed: mocks.markFailed, - markReady: mocks.markReady, }, })) -vi.mock("./source-reconcile-workflow", () => ({ - markSourceReadyAfterReconciliation: mocks.markSourceReadyAfterReconciliation, -})) - vi.mock("./parsed-document-sync-capacity", () => ({ parsedDocumentSyncCapacityGuard: { acquire: mocks.acquireSyncCapacity, @@ -73,16 +62,13 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { }, }) mocks.releaseSyncCapacity.mockResolvedValue(undefined) - mocks.markSourceReadyAfterReconciliation.mockResolvedValue({ - status: "ready", - }) }) afterEach(() => { vi.clearAllMocks() }) - it("marks the source ready when sync completes in one segment", async () => { + it("records completion when sync completes in one segment", async () => { const syncParsedDocument = vi.fn(async () => ({ documentId: "doc_1", revisionKey: "rev_1", @@ -105,18 +91,13 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { "source_1", { revisionKey: "rev_1", syncStatus: "completed" }, ) - expect(mocks.markSourceReadyAfterReconciliation).toHaveBeenCalledWith({ - workspaceId: "workspace_1", - sourceId: "source_1", - documentId: "doc_1", - }) expect(mocks.releaseSyncCapacity).toHaveBeenCalledWith({ leaseToken: "lease_1", releaseReason: "completed", }) }) - it("marks ready and triggers a continuation when sync is incomplete", async () => { + it("triggers a continuation when sync is incomplete", async () => { const syncParsedDocument = vi.fn(async () => ({ documentId: "doc_1", revisionKey: "rev_1", @@ -145,11 +126,6 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { restore() } - expect(mocks.markSourceReadyAfterReconciliation).toHaveBeenCalledWith({ - workspaceId: "workspace_1", - sourceId: "source_1", - documentId: "doc_1", - }) expect(triggered).toHaveLength(1) expect(triggered[0]?.segmentIndex).toBe(1) expect(triggered[0]?.workflowRunId).toBe("doc_1-sync-rev_1-1") @@ -249,6 +225,37 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => { ]) expect(mocks.releaseSyncCapacity).not.toHaveBeenCalled() }) + + it("does not release capacity when Upstash aborts during a planned step", async () => { + const workflowAbort = new Error("planned workflow step") + workflowAbort.name = "WorkflowAbort" + const syncParsedDocument = vi.fn(async () => ({ + documentId: "doc_1", + revisionKey: "rev_1", + completed: true, + })) + mocks.makeKnowhereClientWithParsedStorage.mockReturnValue({ + client: {}, + knowledge: { syncParsedDocument }, + }) + const run: RunStep = async (id, task) => { + if (id === "sync-0-0") throw workflowAbort + return task() + } + + await expect( + parsedSyncRouteWorkflow.runParsedSyncWorkflow({ + context: { + run, + url: "https://notebook.example/api/sources/parsed-sync", + }, + payload: basePayload, + }), + ).rejects.toThrow("planned workflow step") + + expect(mocks.releaseSyncCapacity).not.toHaveBeenCalled() + expect(mocks.updateSyncStatus).not.toHaveBeenCalled() + }) }) describe("parsedSyncRouteWorkflow.markSyncFailedAfterWorkflowFailure", () => { @@ -256,9 +263,7 @@ describe("parsedSyncRouteWorkflow.markSyncFailedAfterWorkflowFailure", () => { vi.clearAllMocks() }) - it("fails a parsing source with failure_stage storage_sync", async () => { - mocks.findInWorkspace.mockResolvedValue({ id: "source_1", status: "parsing" }) - + it("records failed sync metadata without failing the source", async () => { await parsedSyncRouteWorkflow.markSyncFailedAfterWorkflowFailure( { ...basePayload, revisionKey: "rev_1" }, "boom", @@ -269,28 +274,5 @@ describe("parsedSyncRouteWorkflow.markSyncFailedAfterWorkflowFailure", () => { "source_1", { revisionKey: "rev_1", syncStatus: "failed", syncError: "boom" }, ) - expect(mocks.markFailed).toHaveBeenCalledWith( - "workspace_1", - "source_1", - expect.stringContaining("storage sync failed"), - "parsing", - "storage_sync", - ) - }) - - it("does not fail an already-ready source, only records sync_status", async () => { - mocks.findInWorkspace.mockResolvedValue({ id: "source_1", status: "ready" }) - - await parsedSyncRouteWorkflow.markSyncFailedAfterWorkflowFailure( - basePayload, - "boom", - ) - - expect(mocks.updateSyncStatus).toHaveBeenCalledWith( - "workspace_1", - "source_1", - { revisionKey: undefined, syncStatus: "failed", syncError: "boom" }, - ) - expect(mocks.markFailed).not.toHaveBeenCalled() }) }) diff --git a/src/domains/sources/parsed-sync-route-workflow.ts b/src/domains/sources/parsed-sync-route-workflow.ts index 6be3b23..e832b01 100644 --- a/src/domains/sources/parsed-sync-route-workflow.ts +++ b/src/domains/sources/parsed-sync-route-workflow.ts @@ -1,6 +1,6 @@ import "server-only" -import { Client, type WorkflowContext } from "@upstash/workflow" +import { Client, WorkflowAbort, type WorkflowContext } from "@upstash/workflow" import type { KnowledgeSyncParsedDocumentResponse } from "@ontos-ai/knowhere-sdk" import { makeKnowhereClientWithParsedStorage } from "@/integrations/knowhere" @@ -10,7 +10,6 @@ import { type ParsedSyncPayload, } from "./parsed-document-sync-scheduler" import { parsedDocumentSyncCapacityGuard } from "./parsed-document-sync-capacity" -import { markSourceReadyAfterReconciliation } from "./source-reconcile-workflow" import { sourceWorkflowRuntime } from "./workflow-runtime" type ParsedSyncWorkflowContext = Pick< @@ -72,20 +71,10 @@ async function runParsedSyncWorkflow(input: { workspaceId, }) - const preSyncReady = await context.run( - `source-ready-before-sync-${payload.segmentIndex}`, - async () => - markSourceReadyAfterReconciliation({ - workspaceId, - sourceId, - documentId, - }), - ) - if (preSyncReady.status === "gone") return - let revisionKey = payload.revisionKey let completed = false let releaseReason: SyncLeaseReleaseReason = "incomplete" + let shouldReleaseLease = true const capacity = await context.run( `acquire-sync-capacity-${payload.segmentIndex}`, @@ -196,24 +185,29 @@ async function runParsedSyncWorkflow(input: { return } } catch (error) { + if (isWorkflowControlAbort(error)) { + shouldReleaseLease = false + throw error + } releaseReason = "failed" throw error } finally { - await context.run(`release-sync-capacity-${payload.segmentIndex}`, async () => - releaseCapacityLease({ - leaseToken: capacity.leaseToken, - releaseReason, - sourceId, - documentId, - }), - ) + if (shouldReleaseLease) { + await context.run(`release-sync-capacity-${payload.segmentIndex}`, async () => + releaseCapacityLease({ + leaseToken: capacity.leaseToken, + releaseReason, + sourceId, + documentId, + }), + ) + } } logger.info("parsed-sync: parsed document sync finished", { sourceId, documentId, revisionKey, - status: preSyncReady.status, }) } @@ -261,10 +255,9 @@ async function markSyncFailedAfterWorkflowFailure( const normalized = normalizeParsedSyncPayload(payload) const reason = getSafeFailureReason(failResponse) - // Record the storage-sync failure. A source still `parsing` is failed with - // failure_stage=storage_sync so a retry resumes sync without reparsing; an - // already-ready source is left ready (it still serves via remote fallback), - // only its sync_status is marked failed for observability. + // Parsed storage is a cache/read model. Exhausted sync failure is recorded for + // observability, but the source remains ready and SDK reads can fall back to + // Knowhere remote. await sourceWorkflowRuntime.updateSyncStatus( normalized.workspaceId, normalized.sourceId, @@ -275,26 +268,11 @@ async function markSyncFailedAfterWorkflowFailure( }, ) - const source = await sourceWorkflowRuntime.findInWorkspace( - normalized.workspaceId, - normalized.sourceId, - ) - if (source?.status === "parsing") { - await sourceWorkflowRuntime.markFailed( - normalized.workspaceId, - normalized.sourceId, - `Parsed document storage sync failed: ${reason}`, - "parsing", - "storage_sync", - ) - } - logger.error("parsed-sync: marked sync failed after workflow failure", { workspaceId: normalized.workspaceId, sourceId: normalized.sourceId, documentId: normalized.documentId, segmentIndex: normalized.segmentIndex, - sourceStatus: source?.status, }) } @@ -314,6 +292,13 @@ function setContinuationTriggerForTesting( } } +function isWorkflowControlAbort(error: unknown): boolean { + return ( + (error instanceof WorkflowAbort && error.constructor === WorkflowAbort) || + (error instanceof Error && error.name === "WorkflowAbort") + ) +} + export const parsedSyncRouteWorkflow = { markSyncFailedAfterWorkflowFailure, normalizeParsedSyncPayload, diff --git a/src/domains/sources/source-reconcile-route-workflow.test.ts b/src/domains/sources/source-reconcile-route-workflow.test.ts index ee415ff..730fcf0 100644 --- a/src/domains/sources/source-reconcile-route-workflow.test.ts +++ b/src/domains/sources/source-reconcile-route-workflow.test.ts @@ -5,6 +5,7 @@ const mocks = vi.hoisted(() => ({ enqueueParsedDocumentSync: vi.fn(), releaseSyncCapacity: vi.fn(), updateSyncStatus: vi.fn(), + updateRevisionKey: vi.fn(), markFailed: vi.fn(), loggerError: vi.fn(), loggerInfo: vi.fn(), @@ -23,6 +24,7 @@ vi.mock("@/domains/sources/workflow-runtime", () => ({ sourceWorkflowRuntime: { markFailed: mocks.markFailed, updateSyncStatus: mocks.updateSyncStatus, + updateRevisionKey: mocks.updateRevisionKey, }, })) @@ -90,9 +92,11 @@ describe("sourceReconcileRouteWorkflow", () => { activeCounts, }) mocks.releaseSyncCapacity.mockResolvedValue(undefined) + mocks.enqueueParsedDocumentSync.mockResolvedValue(undefined) mocks.markSourceReadyAfterReconciliation.mockResolvedValue({ status: "ready", }) + mocks.updateRevisionKey.mockResolvedValue({ id: "source_1" }) }) afterEach(() => { @@ -116,7 +120,7 @@ describe("sourceReconcileRouteWorkflow", () => { }) }) - it("marks the source ready then syncs the parsed document", async () => { + it("marks the source ready, records pending sync, and enqueues parsed-sync", async () => { const context = createWorkflowContext() const continuations: ContinuationTriggerInput[] = [] const restore = @@ -149,46 +153,44 @@ describe("sourceReconcileRouteWorkflow", () => { restore() } - expect(wired.knowledge.syncParsedDocument).toHaveBeenCalledWith({ + expect(wired.knowledge.syncParsedDocument).not.toHaveBeenCalled() + expect(mocks.markSourceReadyAfterReconciliation).toHaveBeenCalledWith({ + workspaceId: "workspace_1", + sourceId: "source_1", documentId: "doc_1", - revisionKey: "rev_1", }) + expect(mocks.updateRevisionKey).toHaveBeenCalledWith( + "workspace_1", + "source_1", + "rev_1", + ) expect(mocks.updateSyncStatus).toHaveBeenCalledWith( "workspace_1", "source_1", - { revisionKey: "rev_1", syncStatus: "completed" }, + { revisionKey: "rev_1", syncStatus: "pending", syncError: null }, ) - expect(mocks.markSourceReadyAfterReconciliation).toHaveBeenCalledWith({ - workspaceId: "workspace_1", - sourceId: "source_1", - documentId: "doc_1", - }) - expect(mocks.acquireSyncCapacity).toHaveBeenCalledWith({ + expect(mocks.enqueueParsedDocumentSync).toHaveBeenCalledWith({ workspaceId: "workspace_1", sourceId: "source_1", documentId: "doc_1", + apiKey: "jwt_1", revisionKey: "rev_1", }) - expect(mocks.releaseSyncCapacity).toHaveBeenCalledWith({ - leaseToken: "lease_1", - releaseReason: "completed", - }) - expect(mocks.enqueueParsedDocumentSync).not.toHaveBeenCalled() + expect(mocks.acquireSyncCapacity).not.toHaveBeenCalled() + expect(mocks.releaseSyncCapacity).not.toHaveBeenCalled() expect(continuations).toEqual([]) }) - it("hands off to parsed-sync when sync is incomplete", async () => { + it("keeps the source ready when parsed-sync enqueue fails", async () => { const context = createWorkflowContext() - const syncParsedDocument = vi.fn(async () => ({ - documentId: "doc_1", - revisionKey: "rev_1", - completed: false, - })) - const wired = createClient({ syncParsedDocument }) + const wired = createClient({}) mocks.makeKnowhereClientWithParsedStorage.mockReturnValue({ client: wired.client, knowledge: wired.knowledge, }) + mocks.enqueueParsedDocumentSync.mockRejectedValue( + new Error("qstash unavailable"), + ) mocks.pollSourceReconciliation.mockResolvedValue({ kind: "ready-to-prepare", jobId: "job_1", @@ -204,120 +206,18 @@ describe("sourceReconcileRouteWorkflow", () => { }), }) + expect(wired.knowledge.syncParsedDocument).not.toHaveBeenCalled() expect(mocks.markSourceReadyAfterReconciliation).toHaveBeenCalledWith({ workspaceId: "workspace_1", sourceId: "source_1", documentId: "doc_1", }) - expect(mocks.enqueueParsedDocumentSync).toHaveBeenCalledWith({ - workspaceId: "workspace_1", - sourceId: "source_1", - documentId: "doc_1", - apiKey: "jwt_1", - revisionKey: "rev_1", - }) - expect(mocks.releaseSyncCapacity).toHaveBeenCalledWith({ - leaseToken: "lease_1", - releaseReason: "incomplete", - }) - }) - - it("records sync failure when Blob sync throws", async () => { - const context = createWorkflowContext() - const syncParsedDocument = vi.fn(async () => { - throw new Error("blob write failed") - }) - const wired = createClient({ syncParsedDocument }) - mocks.makeKnowhereClientWithParsedStorage.mockReturnValue({ - client: wired.client, - knowledge: wired.knowledge, - }) - mocks.pollSourceReconciliation.mockResolvedValue({ - kind: "ready-to-prepare", - jobId: "job_1", - documentId: "doc_1", - }) - - await expect( - sourceReconcileRouteWorkflow.runPollAndMirrorWorkflow({ - context, - payload: sourceReconcileRouteWorkflow.normalizeReconcilePayload({ - workspaceId: "workspace_1", - sourceId: "source_1", - apiKey: "jwt_1", - }), - }), - ).rejects.toThrow("blob write failed") - expect(mocks.updateSyncStatus).toHaveBeenCalledWith( "workspace_1", "source_1", - { - revisionKey: "rev_1", - syncStatus: "failed", - syncError: "blob write failed", - }, + { revisionKey: "rev_1", syncStatus: "pending", syncError: null }, ) expect(mocks.markFailed).not.toHaveBeenCalled() - expect(mocks.markSourceReadyAfterReconciliation).toHaveBeenCalledWith({ - workspaceId: "workspace_1", - sourceId: "source_1", - documentId: "doc_1", - }) - expect(mocks.releaseSyncCapacity).toHaveBeenCalledWith( - { - leaseToken: "lease_1", - releaseReason: "failed", - }, - ) - }) - - it("marks the source ready and schedules delayed sync when capacity is full", async () => { - const context = createWorkflowContext() - const wired = createClient({}) - mocks.makeKnowhereClientWithParsedStorage.mockReturnValue({ - client: wired.client, - knowledge: wired.knowledge, - }) - mocks.acquireSyncCapacity.mockResolvedValue({ - kind: "capacity-full", - reason: "document", - waitSeconds: 60, - activeCounts: { - globalActive: 10, - workspaceActive: 1, - documentActive: 1, - }, - }) - mocks.pollSourceReconciliation.mockResolvedValue({ - kind: "ready-to-prepare", - jobId: "job_1", - documentId: "doc_1", - }) - - await sourceReconcileRouteWorkflow.runPollAndMirrorWorkflow({ - context, - payload: sourceReconcileRouteWorkflow.normalizeReconcilePayload({ - workspaceId: "workspace_1", - sourceId: "source_1", - apiKey: "jwt_1", - }), - }) - - expect(mocks.markSourceReadyAfterReconciliation).toHaveBeenCalledWith({ - workspaceId: "workspace_1", - sourceId: "source_1", - documentId: "doc_1", - }) - expect(wired.knowledge.syncParsedDocument).not.toHaveBeenCalled() - expect(mocks.enqueueParsedDocumentSync).toHaveBeenCalledWith({ - workspaceId: "workspace_1", - sourceId: "source_1", - documentId: "doc_1", - apiKey: "jwt_1", - revisionKey: "rev_1", - delaySeconds: 60, - }) expect(mocks.releaseSyncCapacity).not.toHaveBeenCalled() }) diff --git a/src/domains/sources/source-reconcile-route-workflow.ts b/src/domains/sources/source-reconcile-route-workflow.ts index 1e45127..5d2cb2c 100644 --- a/src/domains/sources/source-reconcile-route-workflow.ts +++ b/src/domains/sources/source-reconcile-route-workflow.ts @@ -9,7 +9,6 @@ import { import { makeKnowhereClientWithParsedStorage } from "@/integrations/knowhere" import { logger } from "@/lib/logger" import { enqueueParsedDocumentSync } from "./parsed-document-sync-scheduler" -import { parsedDocumentSyncCapacityGuard } from "./parsed-document-sync-capacity" import { sourceWorkflowRuntime } from "./workflow-runtime" type ReconcilePayload = { @@ -41,12 +40,25 @@ type ContinuationTriggerInput = { readonly workflowRunId: string } -type SyncLeaseReleaseReason = "completed" | "incomplete" | "failed" +type RevisionKeyClient = { + readonly documents: { + readonly listChunks: ( + documentId: string, + params: { + readonly page: number + readonly pageSize: number + readonly includeAssetUrls: boolean + }, + ) => Promise<{ + readonly jobResultId?: string | null + readonly jobId?: string | null + }> + } +} const maxPollAttempts = 25 const initialDelaySeconds = 3 const maxDelaySeconds = 30 -const maxSyncStepsPerReconcile = 4 let triggerContinuation: typeof triggerReconcileContinuation = triggerReconcileContinuation @@ -57,7 +69,7 @@ async function runPollAndMirrorWorkflow(input: { }): Promise { const { context, payload } = input const { workspaceId, sourceId, apiKey } = payload - const { client, knowledge } = makeKnowhereClientWithParsedStorage(apiKey, { + const { client } = makeKnowhereClientWithParsedStorage(apiKey, { workspaceId, }) let delay = initialDelaySeconds @@ -125,21 +137,6 @@ async function runPollAndMirrorWorkflow(input: { return } - const revisionKey = await context.run("resolve-revision-key", async () => { - const firstPage = await client.documents.listChunks( - jobToPrepare.documentId, - { page: 1, pageSize: 1, includeAssetUrls: false }, - ) - return firstPage.jobResultId ?? firstPage.jobId ?? jobToPrepare.jobId - }) - await context.run("record-sync-pending", async () => - sourceWorkflowRuntime.updateSyncStatus(workspaceId, sourceId, { - revisionKey, - syncStatus: "pending", - syncError: null, - }), - ) - const ready = await context.run("source-ready", async () => markSourceReadyAfterReconciliation({ workspaceId, @@ -149,139 +146,87 @@ async function runPollAndMirrorWorkflow(input: { ) if (ready.status === "gone") return - const capacity = await context.run("acquire-sync-capacity", async () => - parsedDocumentSyncCapacityGuard.acquire({ - workspaceId, + const revisionKey = await context.run("resolve-revision-key", async () => + resolveParsedRevisionKey({ + client, sourceId, documentId: jobToPrepare.documentId, + fallbackRevisionKey: jobToPrepare.jobId, + }), + ) + + await context.run("record-source-revision-key", async () => + sourceWorkflowRuntime.updateRevisionKey(workspaceId, sourceId, revisionKey), + ) + await context.run("record-sync-pending", async () => + sourceWorkflowRuntime.updateSyncStatus(workspaceId, sourceId, { revisionKey, + syncStatus: "pending", + syncError: null, }), ) - if (capacity.kind === "source-missing") return - if (capacity.kind === "capacity-full") { - await context.run("enqueue-capacity-retry", async () => - enqueueParsedDocumentSync({ - workspaceId, - sourceId, - documentId: jobToPrepare.documentId, - apiKey, - revisionKey, - delaySeconds: capacity.waitSeconds, - }), - ) - logger.info("workflow: parsed storage sync delayed by capacity guard", { + const enqueueResult = await context.run("enqueue-parsed-sync", async () => + enqueueParsedSyncBestEffort({ + workspaceId, sourceId, documentId: jobToPrepare.documentId, + apiKey, revisionKey, - reason: capacity.reason, - waitSeconds: capacity.waitSeconds, - activeCounts: capacity.activeCounts, - }) - return - } - - let syncCompleted = false - let releaseReason: SyncLeaseReleaseReason = "incomplete" - try { - await context.run("record-sync-running", async () => - sourceWorkflowRuntime.updateSyncStatus(workspaceId, sourceId, { - revisionKey, - syncStatus: "running", - syncError: null, - }), - ) - - for (let step = 0; step < maxSyncStepsPerReconcile; step++) { - const result = await context.run(`parsed-sync-${step}`, async () => { - try { - return await knowledge.syncParsedDocument({ - documentId: jobToPrepare.documentId, - revisionKey, - }) - } catch (error) { - await sourceWorkflowRuntime.updateSyncStatus(workspaceId, sourceId, { - revisionKey, - syncStatus: "failed", - syncError: getErrorMessage(error), - }) - throw error - } - }) - if (result.completed) { - syncCompleted = true - releaseReason = "completed" - break - } - } - - if (!syncCompleted) { - await context.run("record-sync-progress", async () => - sourceWorkflowRuntime.updateSyncStatus(workspaceId, sourceId, { - revisionKey, - syncStatus: "running", - }), - ) - await context.run("enqueue-parsed-sync-continuation", async () => - enqueueParsedDocumentSync({ - workspaceId, - sourceId, - documentId: jobToPrepare.documentId, - apiKey, - revisionKey, - }), - ) - logger.info("workflow: parsed storage sync handed off to parsed-sync", { - sourceId, - documentId: jobToPrepare.documentId, - revisionKey, - }) - return - } - } catch (error) { - releaseReason = "failed" - throw error - } finally { - await context.run("release-sync-capacity", async () => - releaseCapacityLease({ - leaseToken: capacity.leaseToken, - releaseReason, - sourceId, - documentId: jobToPrepare.documentId, - }), - ) - } - - await context.run("record-sync-completed", async () => - sourceWorkflowRuntime.updateSyncStatus(workspaceId, sourceId, { - revisionKey, - syncStatus: "completed", }), ) + logger.info("workflow: source parse reconciliation finished", { sourceId, jobId: jobToPrepare.jobId, revisionKey, status: ready.status, + parsedSyncEnqueued: enqueueResult.enqueued, }) } -async function releaseCapacityLease(input: { - readonly leaseToken: string - readonly releaseReason: SyncLeaseReleaseReason +async function resolveParsedRevisionKey(input: { + readonly client: RevisionKeyClient readonly sourceId: string readonly documentId: string -}): Promise { + readonly fallbackRevisionKey: string +}): Promise { try { - await parsedDocumentSyncCapacityGuard.release({ - leaseToken: input.leaseToken, - releaseReason: input.releaseReason, + const firstPage = await input.client.documents.listChunks(input.documentId, { + page: 1, + pageSize: 1, + includeAssetUrls: false, + }) + return firstPage.jobResultId ?? firstPage.jobId ?? input.fallbackRevisionKey + } catch (error) { + logger.warn("workflow: failed to resolve parsed revision key", { + sourceId: input.sourceId, + documentId: input.documentId, + fallbackRevisionKey: input.fallbackRevisionKey, + error: getErrorMessage(error), }) + return input.fallbackRevisionKey + } +} + +async function enqueueParsedSyncBestEffort(input: { + readonly workspaceId: string + readonly sourceId: string + readonly documentId: string + readonly apiKey: string + readonly revisionKey: string +}): Promise<{ readonly enqueued: boolean }> { + try { + await enqueueParsedDocumentSync(input) + return { enqueued: true } } catch (error) { - logger.error("workflow: failed to release sync capacity lease", { + logger.error("workflow: failed to enqueue parsed storage sync", { + workspaceId: input.workspaceId, sourceId: input.sourceId, documentId: input.documentId, + revisionKey: input.revisionKey, error: getErrorMessage(error), }) + return { enqueued: false } } } diff --git a/src/proxy.test.ts b/src/proxy.test.ts index b965699..0d80c19 100644 --- a/src/proxy.test.ts +++ b/src/proxy.test.ts @@ -63,6 +63,16 @@ describe("proxy", () => { ); }); + it("allows anonymous parsed-sync workflow callbacks", () => { + const response = proxy( + new NextRequest("http://localhost:3001/api/sources/parsed-sync", { + method: "POST", + }), + ); + + expect(response.headers.get("x-middleware-next")).toBe("1"); + }); + it("allows protected app routes without a session when KNOWHERE_API_KEY is configured", () => { process.env.KNOWHERE_API_KEY = "sk_dev_key"; diff --git a/src/proxy.ts b/src/proxy.ts index 6177d3b..7527048 100644 --- a/src/proxy.ts +++ b/src/proxy.ts @@ -27,6 +27,7 @@ const PUBLIC_PATHS: readonly string[] = [ "/favicon.ico", "/api/internal/health", "/api/sources/reconcile", + "/api/sources/parsed-sync", ] const STATIC_EXTENSIONS = /\.(?:svg|png|jpe?g|gif|webp|ico|woff2?|ttf|eot|css|js|map|txt|xml|webmanifest|json|pdf)$/i