Stripe api test server & Tests & Bug Fixes (#245)

* test server

* use hono, clean up api, add tests

* add docker-backed test server harness

* add all-api test server coverage, stripe failure regression tests

* retry apis

* disruption issue tests

* fix false ready status on sync errors

* clean up e2e docker containers and bound buildkit cache

* format

* derive list capabilities from OpenAPI spec instead of hardcoding & tests

* typo

* cleanup

* exlcude new tests as well

* don't run unrelated tests

* don't test older than 2020

* Fix classifySyncErrors, remvoe backfill-pipeline-workflow.ts , cleanup, Revert unrelated dest-postgres changes

* Replace stub object generation with OpenAPI schema-based generator and optimize test server

* cleanup

* move retry to stripe-source

* don't exclude the tests

* Fix error recovery race across continueAsNew

* Emit auth_error for 401/403, keep system_error transient

* Log transient errors suppressed by permanent failures

* JSDoc, allow hyphens, add missing api_version

* separate job for test server

* start docker

* reduce object size, pin version that support v2 in v2test

* consolidate error

Author: Yostra

.github/workflows/ci.yml

@@ -446,7 +446,13 @@ jobs:
             echo "::warning::E2E tests skipped — STRIPE_API_KEY not available (fork PR?)"
             exit 0
           fi
-          pnpm --filter @stripe/sync-e2e exec vitest run --exclude 'service-docker.test.ts'
+          pnpm --filter @stripe/sync-e2e exec vitest run \
+            --exclude 'service-docker.test.ts' \
+            --exclude 'test-server-all-api.test.ts' \
+            --exclude 'test-server-sync.test.ts' \
+            --exclude 'test-sync-e2e.test.ts' \
+            --exclude 'test-sync-engine.test.ts' \
+            --exclude 'test-e2e-network.test.ts'  # ↑ run in e2e_test_server job
         env:
           STRIPE_API_KEY: ${{ secrets.STRIPE_API_KEY }}
           POSTGRES_URL: 'postgres://postgres:postgres@localhost:55432/postgres'
@@ -456,6 +462,54 @@ jobs:
         if: always()
         run: '[ -f /tmp/vitest-skip-warnings.txt ] && cat /tmp/vitest-skip-warnings.txt || true'
 
+  # ---------------------------------------------------------------------------
+  # E2E Test Server — test-server suites that need Docker Compose infrastructure
+  # ---------------------------------------------------------------------------
+  e2e_test_server:
+    name: E2E Test Server
+    runs-on: ubuntu-24.04-arm
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Install pnpm
+        uses: pnpm/action-setup@v5
+
+      - name: Set up Node
+        uses: actions/setup-node@v6
+        with:
+          node-version-file: ./.nvmrc
+          cache: pnpm
+
+      - name: Install dependencies & build
+        run: pnpm install --frozen-lockfile && pnpm build
+
+      - name: Start Docker Compose stack
+        run: |
+          docker compose -f compose.yml -f compose.dev.yml -f e2e/compose.e2e.yml \
+            up --build -d --wait temporal engine service worker
+
+      - name: Test server suites (parallel)
+        run: |
+          pnpm --filter @stripe/sync-e2e exec vitest run \
+            test-server-all-api.test.ts \
+            test-server-sync.test.ts \
+            test-sync-e2e.test.ts \
+            test-sync-engine.test.ts
+        env:
+          SKIP_SETUP: '1'
+
+      - name: Network interruption tests (pauses containers)
+        run: |
+          pnpm --filter @stripe/sync-e2e exec vitest run \
+            test-e2e-network.test.ts
+        env:
+          SKIP_SETUP: '1'
+
+      - name: Skipped test warnings
+        if: always()
+        run: '[ -f /tmp/vitest-skip-warnings.txt ] && cat /tmp/vitest-skip-warnings.txt || true'
+
   # ---------------------------------------------------------------------------
   # E2E Service — service + worker Docker containers end-to-end (every push/PR)
   # ---------------------------------------------------------------------------
@@ -755,7 +809,7 @@ jobs:
         run: pnpm install --frozen-lockfile && pnpm build
 
       - name: CDN e2e tests
-        run: pnpm --filter @stripe/sync-e2e run test -- openapi-cdn.test.ts
+        run: pnpm --filter @stripe/sync-e2e exec vitest run openapi-cdn.test.ts
         env:
           STRIPE_SPEC_CDN_BASE_URL: ${{ needs.docs.outputs.deployment_url }}/stripe-api-specs
 

apps/service/src/__tests__/workflow.test.ts

@@ -8,19 +8,23 @@ import { CONTINUE_AS_NEW_THRESHOLD } from '../lib/utils.js'
 
 type SourceInput = unknown
 
-// workflowsPath points to the compiled workflow directory.
-const workflowsPath = path.resolve(process.cwd(), 'dist/temporal/workflows')
+// Point directly at the workflow index to avoid resolving the legacy dist/temporal/workflows.js file.
+const workflowsPath = path.resolve(process.cwd(), 'dist/temporal/workflows/index.js')
 
 const emptyState = { streams: {}, global: {} }
 const noErrors: RunResult = { errors: [], state: emptyState }
+const permanentSyncError: RunResult = {
+  errors: [{ message: 'permanent sync failure', failure_type: 'auth_error', stream: 'customers' }],
+  state: emptyState,
+}
 
 // Workflows now receive only the pipelineId string
 const testPipelineId = 'test_pipe'
 
 function stubActivities(overrides: Partial<SyncActivities> = {}): SyncActivities {
-  return {
+  const activities = {
     discoverCatalog: async () => ({ streams: [] }),
-    pipelineSetup: async () => ({}),
+    pipelineSetup: async () => {},
     pipelineSync: async () => noErrors,
     readIntoQueue: async () => ({ count: 0, state: emptyState }),
     writeGoogleSheetsFromQueue: async () => ({
@@ -33,6 +37,13 @@ function stubActivities(overrides: Partial<SyncActivities> = {}): SyncActivities
     updatePipelineStatus: async () => {},
     ...overrides,
   }
+
+  return {
+    ...activities,
+    setup: activities.pipelineSetup,
+    sync: activities.pipelineSync,
+    teardown: activities.pipelineTeardown,
+  } as SyncActivities
 }
 
 /** Signal the workflow to delete. */
@@ -69,7 +80,6 @@ describe('pipelineWorkflow (unit — stubbed activities)', () => {
       activities: stubActivities({
         pipelineSetup: async () => {
           setupCalled = true
-          return {}
         },
         pipelineSync: async () => {
           runCallCount++
@@ -89,7 +99,7 @@ describe('pipelineWorkflow (unit — stubbed activities)', () => {
       await new Promise((r) => setTimeout(r, 2000))
 
       const status = await handle.query('status')
-      expect(status.iteration).toBeGreaterThan(0)
+      expect((status as { iteration: number }).iteration).toBeGreaterThan(0)
 
       await signalDelete(handle)
       await handle.result()
@@ -334,6 +344,82 @@ describe('pipelineWorkflow (unit — stubbed activities)', () => {
     })
   })
 
+  it('transitions to error instead of ready when reconcile returns permanent sync errors', async () => {
+    const statusWrites: string[] = []
+
+    const worker = await Worker.create({
+      connection: testEnv.nativeConnection,
+      taskQueue: 'test-queue-3b-error',
+      workflowsPath,
+      activities: stubActivities({
+        updatePipelineStatus: async (_id: string, status: string) => {
+          statusWrites.push(status)
+        },
+        pipelineSync: async (_pipelineId: string, opts?) => {
+          if (opts?.input) return noErrors
+          return { ...permanentSyncError, eof: { reason: 'complete' as const } }
+        },
+      }),
+    })
+
+    await worker.runUntil(async () => {
+      const handle = await testEnv.client.workflow.start('pipelineWorkflow', {
+        args: [testPipelineId],
+        workflowId: 'test-sync-3b-error',
+        taskQueue: 'test-queue-3b-error',
+      })
+
+      await new Promise((r) => setTimeout(r, 500))
+      await signalDelete(handle)
+      await handle.result()
+
+      expect(statusWrites).toContain('error')
+      expect(statusWrites).not.toContain('ready')
+    })
+  })
+
+  it('retries transient sync activity failures and still reaches ready', async () => {
+    const statusWrites: string[] = []
+    let reconcileCalls = 0
+
+    const worker = await Worker.create({
+      connection: testEnv.nativeConnection,
+      taskQueue: 'test-queue-3b-retry',
+      workflowsPath,
+      activities: stubActivities({
+        updatePipelineStatus: async (_id: string, status: string) => {
+          statusWrites.push(status)
+        },
+        pipelineSync: async (_pipelineId: string, opts?) => {
+          if (opts?.input) return noErrors
+
+          reconcileCalls++
+          if (reconcileCalls === 1) {
+            throw new Error('transient sync failure')
+          }
+
+          return { ...noErrors, eof: { reason: 'complete' as const } }
+        },
+      }),
+    })
+
+    await worker.runUntil(async () => {
+      const handle = await testEnv.client.workflow.start('pipelineWorkflow', {
+        args: [testPipelineId],
+        workflowId: 'test-sync-3b-retry',
+        taskQueue: 'test-queue-3b-retry',
+      })
+
+      await new Promise((r) => setTimeout(r, 2500))
+      await signalDelete(handle)
+      await handle.result()
+
+      expect(reconcileCalls).toBeGreaterThanOrEqual(2)
+      expect(statusWrites).toContain('ready')
+      expect(statusWrites).not.toContain('error')
+    })
+  })
+
   it('queues live events while paused and drains them after resume', async () => {
     const syncCalls: { input?: SourceInput[] }[] = []
 
@@ -464,7 +550,6 @@ describe('pipelineWorkflow (unit — stubbed activities)', () => {
       activities: stubActivities({
         pipelineSetup: async () => {
           setupCalls++
-          return {}
         },
         pipelineSync: async () => {
           syncCallCount++

apps/service/src/temporal/activities/_shared.ts

@@ -4,6 +4,7 @@ import { createRemoteEngine } from '@stripe/sync-engine'
 import { Kafka } from 'kafkajs'
 import type { Producer } from 'kafkajs'
 import type { PipelineStore } from '../../lib/stores.js'
+import type { SyncRunError } from '../sync-errors.js'
 
 export interface ActivitiesContext {
   /** Remote engine client — satisfies the {@link Engine} interface over HTTP. Drop-in replacement for a local engine. */
@@ -104,7 +105,7 @@ export function createActivitiesContext(opts: {
 }
 
 export interface RunResult {
-  errors: Array<{ message: string; failure_type?: string; stream?: string }>
+  errors: SyncRunError[]
   state: SourceState
 }
 

apps/service/src/temporal/activities/pipeline-sync.ts

@@ -1,6 +1,8 @@
+import { ApplicationFailure } from '@temporalio/activity'
 import type { SourceInputMessage, SourceReadOptions } from '@stripe/sync-engine'
 import type { ActivitiesContext } from './_shared.js'
 import { asIterable, drainMessages, type RunResult } from './_shared.js'
+import { classifySyncErrors, summarizeSyncErrors } from '../sync-errors.js'
 
 export function createPipelineSyncActivity(context: ActivitiesContext) {
   return async function pipelineSync(
@@ -28,6 +30,18 @@ export function createPipelineSyncActivity(context: ActivitiesContext) {
         destination: { type, [type]: destConfig },
       })
     }
+    const { transient, permanent } = classifySyncErrors(errors)
+    if (permanent.length > 0) {
+      if (transient.length > 0) {
+        console.warn(
+          `Transient errors suppressed by permanent failures: ${summarizeSyncErrors(transient)}`
+        )
+      }
+      return { errors, state, eof }
+    }
+    if (transient.length > 0) {
+      throw ApplicationFailure.retryable(summarizeSyncErrors(transient), 'TransientSyncError')
+    }
     return { errors, state, eof }
   }
 }

apps/service/src/temporal/sync-errors.ts

@@ -0,0 +1,37 @@
+export type SyncRunError = {
+  message: string
+  failure_type?: string
+  stream?: string
+}
+
+export type ClassifiedSyncErrors = {
+  transient: SyncRunError[]
+  permanent: SyncRunError[]
+}
+
+const PERMANENT_FAILURE_TYPES = new Set(['config_error', 'auth_error'])
+
+export function classifySyncErrors(errors: SyncRunError[]): ClassifiedSyncErrors {
+  const transient: SyncRunError[] = []
+  const permanent: SyncRunError[] = []
+
+  for (const error of errors) {
+    if (PERMANENT_FAILURE_TYPES.has(error.failure_type ?? '')) {
+      permanent.push(error)
+    } else {
+      transient.push(error)
+    }
+  }
+
+  return { transient, permanent }
+}
+
+export function summarizeSyncErrors(errors: SyncRunError[]): string {
+  return errors
+    .map((error) => {
+      const failureType = error.failure_type ?? 'unknown_error'
+      const stream = error.stream ? `/${error.stream}` : ''
+      return `[${failureType}${stream}] ${error.message}`
+    })
+    .join('; ')
+}