fix(knowledge): expose safe recovery diagnostics

Author: icecrasher321

apps/sim/app/api/webhooks/outbox/process/route.ts

@@ -10,10 +10,12 @@ import { enterpriseIssuanceOutboxHandlers } from '@/lib/billing/enterprise-provi
 import { membershipBillingOutboxHandlers } from '@/lib/billing/organizations/membership-reconciliation'
 import { billingOutboxHandlers } from '@/lib/billing/webhooks/outbox-handlers'
 import { processOutboxEvents } from '@/lib/core/outbox/service'
+import { DeadlineExceededError } from '@/lib/core/utils/deadline'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { directGrantOutboxHandlers } from '@/lib/invitations/direct-grant'
 import { slackSearchOutboxHandlers } from '@/lib/knowledge/application/slack-search/outbox'
+import { getConnectorFailureDiagnostic } from '@/lib/knowledge/connectors/connector-error'
 import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler'
 import { recoverKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-recovery'
 import { organizationResourceCleanupOutboxHandlers } from '@/lib/organizations/resource-cleanup'
@@ -66,8 +68,17 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       if (Date.now() - startedAt < 770_000) {
         recoveredDocuments = await recoverKnowledgeDocumentProcessing()
       }
-    } catch {
-      logger.error('Stored document recovery failed', { requestId })
+    } catch (error) {
+      logger.error('Stored document recovery failed', {
+        requestId,
+        error: getConnectorFailureDiagnostic(error) ?? {
+          category: error instanceof DeadlineExceededError ? 'deadline' : 'internal',
+          message:
+            error instanceof DeadlineExceededError
+              ? error.message
+              : 'Unexpected stored-document recovery failure',
+        },
+      })
     }
 
     // Reap fork background-work rows stuck `processing` past their TTL (worker crash /

apps/sim/lib/core/utils/deadline.test.ts

@@ -1,6 +1,6 @@
 /** @vitest-environment node */
 import { afterEach, describe, expect, it, vi } from 'vitest'
-import { withinDeadline } from '@/lib/core/utils/deadline'
+import { DeadlineExceededError, withinDeadline } from '@/lib/core/utils/deadline'
 
 afterEach(() => vi.useRealTimers())
 
@@ -17,7 +17,7 @@ describe('bounded asynchronous operations', () => {
       signal.throwIfAborted()
       mutate()
     }, Date.now() + 100)
-    const rejection = expect(operation).rejects.toThrow('Operation deadline expired')
+    const rejection = expect(operation).rejects.toBeInstanceOf(DeadlineExceededError)
     await vi.advanceTimersByTimeAsync(100)
     await rejection
     resolve()
@@ -26,6 +26,14 @@ describe('bounded asynchronous operations', () => {
     expect(vi.getTimerCount()).toBe(0)
   })
 
+  it('rejects an expired deadline before starting dependency work', async () => {
+    const dependency = vi.fn()
+    await expect(withinDeadline(dependency, Date.now() - 1)).rejects.toBeInstanceOf(
+      DeadlineExceededError
+    )
+    expect(dependency).not.toHaveBeenCalled()
+  })
+
   it('clears its timer after successful completion and propagates caller cancellation', async () => {
     vi.useFakeTimers()
     await expect(withinDeadline(async () => 7, Date.now() + 100)).resolves.toBe(7)

apps/sim/lib/core/utils/deadline.ts

@@ -1,3 +1,11 @@
+/** Identifies deadline expiry without exposing arbitrary dependency error messages. */
+export class DeadlineExceededError extends Error {
+  constructor() {
+    super('Operation deadline expired')
+    this.name = 'DeadlineExceededError'
+  }
+}
+
 /** Cancels the caller's wait even when a storage client's connection or command queue stalls. */
 export async function withinDeadline<T>(
   operation: (signal: AbortSignal) => Promise<T>,
@@ -6,7 +14,7 @@ export async function withinDeadline<T>(
 ): Promise<T> {
   signal?.throwIfAborted()
   const remainingMs = deadlineAt - Date.now()
-  if (remainingMs <= 0) throw new Error('Operation deadline expired')
+  if (remainingMs <= 0) throw new DeadlineExceededError()
   const controller = new AbortController()
   const abort = () => controller.abort(signal?.reason)
   signal?.addEventListener('abort', abort, { once: true })
@@ -17,7 +25,7 @@ export async function withinDeadline<T>(
   const rejectAborted = () => rejectWait(controller.signal.reason)
   controller.signal.addEventListener('abort', rejectAborted, { once: true })
   const timer = setTimeout(() => {
-    controller.abort(new Error('Operation deadline expired'))
+    controller.abort(new DeadlineExceededError())
   }, remainingMs)
   try {
     return await Promise.race([operation(controller.signal), aborted])

apps/sim/lib/knowledge/connectors/listing-checkpoint.ts

@@ -74,9 +74,11 @@ export function beginListingCheckpoint(input: {
 }
 
 /**
- * Persists one provider page only after its documents and observations land.
- * A crash replays at most that page; EOF is durable so reconciliation can resume
- * independently. Runtime caches and access tokens never enter the checkpoint.
+ * Pins an optional current-page replay cursor before processing, without advancing
+ * the listed count or marking completion. Advances the cursor and persists EOF only
+ * after the page's documents and observations land. A crash replays at most that page;
+ * durable EOF lets reconciliation resume independently. Runtime caches and access
+ * tokens never enter the checkpoint.
  */
 export async function runResumableListing(input: {
   connectorConfig: Pick<ConnectorConfig, 'listDocuments' | 'isListingCursorInvalidError'>