feat: adaptive binary subdivision with remaining-based state, per-stream error resilience, and pure progress reducers

apps/engine/src/api/app.ts

@@ -133,12 +133,9 @@ const REASON_EMOJI: Record<string, string> = {
 
 const STATUS_EMOJI: Record<string, string> = {
   complete: '✅',
-  started: '🔄',
+  start: '🔄',
   running: '🔄',
-  transient_error: '⚠️',
-  system_error: '❌',
-  config_error: '❌',
-  auth_error: '🔒',
+  range_complete: '🔄',
 }
 
 function formatEof(eof: EofPayload): string {
@@ -165,6 +162,11 @@ function formatEof(eof: EofPayload): string {
     const activeStreams: { name: string; rows: number; rps: string }[] = []
 
     for (const [name, s] of Object.entries(sp)) {
+      if (s.errors?.length) {
+        errored++
+        const errMsg = s.errors[0]?.message ?? 'unknown error'
+        errorStreams.push(`❌ ${name}: ${errMsg}`)
+      }
       if (s.status === 'complete') {
         complete++
         if (s.run_record_count > 0) {
@@ -174,7 +176,7 @@ function formatEof(eof: EofPayload): string {
             rps: s.records_per_second?.toFixed(1) ?? '0',
           })
         }
-      } else if (s.status === 'started' || s.status === 'running') {
+      } else if (s.status === 'start' || s.status === 'running') {
         inProgress++
         if (s.run_record_count > 0) {
           activeStreams.push({
@@ -183,15 +185,6 @@ function formatEof(eof: EofPayload): string {
             rps: s.records_per_second?.toFixed(1) ?? '0',
           })
         }
-      } else if (
-        s.status === 'transient_error' ||
-        s.status === 'system_error' ||
-        s.status === 'config_error' ||
-        s.status === 'auth_error'
-      ) {
-        errored++
-        const errMsg = s.errors?.[0]?.message ?? s.status
-        errorStreams.push(`${STATUS_EMOJI[s.status]} ${name}: ${errMsg}`)
       } else {
         pending++
       }
@@ -475,10 +468,7 @@ export async function createApp(resolver: ConnectorResolver) {
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   function getPipeline(c: any): z.infer<typeof TypedPipelineConfig> {
     if (isJsonBody(c)) return c.req.valid('json').pipeline
-    return requireHeaderValue(
-      c.req.valid('header')['x-pipeline'],
-      'x-pipeline header is required'
-    )
+    return requireHeaderValue(c.req.valid('header')['x-pipeline'], 'x-pipeline header is required')
   }
 
   // eslint-disable-next-line @typescript-eslint/no-explicit-any

apps/engine/src/lib/backfill.test.ts

@@ -65,11 +65,14 @@ describe('pipelineSyncUntilComplete', () => {
     ])
     expect(result.attempts).toBe(2)
     expect(result.eof.reason).toBe('complete')
-    expect(onState).toHaveBeenLastCalledWith({
-      source: { streams: { customers: { cursor: 'cus_1' } }, global: {} },
-      destination: { streams: {}, global: {} },
-      engine: { streams: {}, global: {} },
-    }, 1)
+    expect(onState).toHaveBeenLastCalledWith(
+      {
+        source: { streams: { customers: { cursor: 'cus_1' } }, global: {} },
+        destination: { streams: {}, global: {} },
+        engine: { streams: {}, global: {} },
+      },
+      1
+    )
   })
 
   it('throws when pipeline_sync ends with an unexpected eof reason', async () => {
@@ -84,7 +87,9 @@ describe('pipelineSyncUntilComplete', () => {
       pipeline_teardown: vi.fn(),
       pipeline_read: vi.fn(),
       pipeline_write: vi.fn(),
-      pipeline_sync: vi.fn(() => toAsync<SyncOutput>([{ type: 'eof', eof: { reason: 'aborted' } }])),
+      pipeline_sync: vi.fn(() =>
+        toAsync<SyncOutput>([{ type: 'eof', eof: { reason: 'aborted' } }])
+      ),
     } as unknown as Engine
 
     await expect(pipelineSyncUntilComplete(engine, pipeline)).rejects.toThrow(

apps/engine/src/lib/backfill.ts

@@ -2,8 +2,7 @@ import type { EofPayload, PipelineConfig, SourceStateMessage } from '@stripe/syn
 import { emptySyncState, type SyncOutput, type SyncState } from '@stripe/sync-protocol'
 import type { Engine, SourceReadOptions } from './engine.js'
 
-export interface PipelineSyncUntilCompleteOptions
-  extends Omit<SourceReadOptions, 'state'> {
+export interface PipelineSyncUntilCompleteOptions extends Omit<SourceReadOptions, 'state'> {
   state?: SyncState
   onAttempt?: (attempt: number, state: SyncState | undefined) => void | Promise<void>
   onMessage?: (message: SyncOutput, attempt: number) => void | Promise<void>
@@ -66,7 +65,9 @@ export async function pipelineSyncUntilComplete(
     }
 
     if (eof.reason !== 'state_limit' && eof.reason !== 'time_limit') {
-      throw new Error(`pipeline_sync attempt ${attempts} ended with unexpected eof reason: ${eof.reason}`)
+      throw new Error(
+        `pipeline_sync attempt ${attempts} ended with unexpected eof reason: ${eof.reason}`
+      )
     }
   }
 }