feat: add Kubernetes integration test suite for cluster mode

Adds a full k3d-based integration test that deploys 3 replicas with
a real PostgreSQL coordinator to verify leader election, WS mesh,
and event distribution across pods. Also adds dataSourceOverride
passthrough to FlowcoreDataPumpCluster for test injection.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jbiskur

.github/workflows/validate.yml

@@ -43,3 +43,38 @@ jobs:
         run: deno test -A
       - name: Validate NPM build
         run: deno run -A bin/build-npm.ts
+
+  integration-test:
+    runs-on: blacksmith-4vcpu-ubuntu-2204
+    needs: build
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          token: ${{ secrets.FLOWCORE_MACHINE_GITHUB_TOKEN }}
+          submodules: true
+      - name: Setup Deno2
+        uses: denoland/setup-deno@v2
+        with:
+          deno-version: v2.x
+      - name: Install Docker
+        run: |
+          if ! command -v docker &>/dev/null; then
+            curl -fsSL https://get.docker.com | sh
+          fi
+          docker version
+      - name: Install k3d
+        run: curl -s https://raw.githubusercontent.com/k3d-io/k3d/main/install.sh | bash
+      - name: Install kubectl
+        run: |
+          curl -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl"
+          chmod +x kubectl && sudo mv kubectl /usr/local/bin/
+      - name: Run integration tests
+        run: bash integration/run-integration.sh
+      - name: Collect logs on failure
+        if: failure()
+        run: |
+          kubectl logs -n data-pump-integration-test -l app=data-pump-test --tail=200 || true
+          kubectl logs -n data-pump-integration-test -l app=postgres --tail=100 || true
+      - name: Teardown
+        if: always()
+        run: bash integration/scripts/teardown.sh || true

deno.json

@@ -32,7 +32,8 @@
       "node_modules",
       "npm",
       ".github",
-      "CHANGELOG.md"
+      "CHANGELOG.md",
+      "integration"
     ],
     "lineWidth": 120,
     "indentWidth": 2,
@@ -42,7 +43,8 @@
   "lint": {
     "exclude": [
       "node_modules",
-      "npm"
+      "npm",
+      "integration"
     ]
   }
 }

integration/app/Dockerfile

@@ -0,0 +1,8 @@
+FROM denoland/deno:2.0.0
+WORKDIR /app
+COPY deno.json deno.lock ./
+COPY src/ ./src/
+COPY integration/app/ ./integration/app/
+RUN deno install
+EXPOSE 8080
+CMD ["deno", "run", "-A", "integration/app/main.ts"]

integration/app/deps.ts

@@ -0,0 +1,3 @@
+export { default as pg } from "npm:pg@^8.13.0"
+export const { Client } = await import("npm:pg@^8.13.0")
+export type { Client as PgClient } from "npm:pg@^8.13.0"

integration/app/fake-data-source.ts

@@ -0,0 +1,93 @@
+import type { EventListOutput, FlowcoreEvent } from "@flowcore/sdk"
+import { TimeUuid } from "@flowcore/time-uuid"
+import { utc } from "@date-fns/utc"
+import { format, startOfHour } from "date-fns"
+import { FlowcoreDataSource } from "../../src/data-pump/data-source.ts"
+import type { FlowcoreDataPumpState } from "../../src/data-pump/types.ts"
+
+export class FakeDataSource extends FlowcoreDataSource {
+  private readonly events: FlowcoreEvent[]
+  private readonly timeBucket: string
+  private delivered = false
+
+  constructor(totalEvents: number) {
+    super({
+      auth: { getBearerToken: () => Promise.resolve("fake") },
+      dataSource: {
+        tenant: "integration-test",
+        dataCore: "test-data-core",
+        flowType: "test-flow-type",
+        eventTypes: ["test-event"],
+      },
+      noTranslation: true,
+    })
+
+    this.timeBucket = format(startOfHour(utc(new Date())), "yyyyMMddHH0000")
+    this.events = []
+
+    for (let i = 0; i < totalEvents; i++) {
+      const timeUuid = TimeUuid.now()
+      this.events.push({
+        eventId: timeUuid.toString(),
+        eventType: "test-event",
+        aggregator: `agg-${i}`,
+        payload: { index: i, data: `test-payload-${i}` },
+        metadata: {},
+        timeBucket: this.timeBucket,
+        validTime: new Date().toISOString(),
+      })
+    }
+
+    // pre-set cached IDs so the parent class never calls Flowcore API
+    this.tenantId = "integration-test"
+    this.dataCoreId = "test-data-core"
+    this.flowTypeId = "test-flow-type"
+    this.eventTypeIds = ["test-event"]
+    this.timeBuckets = [this.timeBucket]
+  }
+
+  public override async getEvents(
+    _from: FlowcoreDataPumpState,
+    amount: number,
+    _toEventId?: string,
+    _cursor?: string,
+    _includeSensitiveData?: boolean,
+  ): Promise<EventListOutput> {
+    if (this.delivered) {
+      return { events: [], nextCursor: undefined }
+    }
+    this.delivered = true
+    return {
+      events: this.events.slice(0, amount),
+      nextCursor: undefined,
+    }
+  }
+
+  public override async getTimeBuckets(_force?: boolean): Promise<string[]> {
+    return [this.timeBucket]
+  }
+
+  public override async getNextTimeBucket(_timeBucket: string): Promise<string | null> {
+    return null
+  }
+
+  public override async getClosestTimeBucket(_timeBucket: string, _getBefore?: boolean): Promise<string | null> {
+    return this.timeBucket
+  }
+
+  public override async getTenantId(): Promise<string> {
+    return "integration-test"
+  }
+
+  public override async getDataCoreId(): Promise<string> {
+    return "test-data-core"
+  }
+
+  public override async getFlowTypeId(): Promise<string> {
+    return "test-flow-type"
+  }
+
+  public override async getEventTypeIds(): Promise<string[]> {
+    return ["test-event"]
+  }
+}

integration/app/main.ts

@@ -0,0 +1,133 @@
+import { Client } from "npm:pg@^8.13.0"
+import { FlowcoreDataPumpCluster } from "../../src/data-pump/data-pump-cluster.ts"
+import { PostgresCoordinator } from "./postgres-coordinator.ts"
+import { PostgresStateManager } from "./postgres-state-manager.ts"
+import { FakeDataSource } from "./fake-data-source.ts"
+
+const DATABASE_URL = Deno.env.get("DATABASE_URL") ?? "postgres://postgres:postgres@localhost:5432/datapump_test"
+const POD_NAME = Deno.env.get("POD_NAME") ?? "local-pod"
+const POD_IP = Deno.env.get("POD_IP") ?? "127.0.0.1"
+const TOTAL_EVENTS = parseInt(Deno.env.get("TOTAL_EVENTS") ?? "100", 10)
+const WS_PORT = parseInt(Deno.env.get("WS_PORT") ?? "8080", 10)
+
+const log = {
+  debug: (msg: string, meta?: Record<string, unknown>) => console.log(`[DEBUG] [${POD_NAME}] ${msg}`, meta ?? ""),
+  info: (msg: string, meta?: Record<string, unknown>) => console.log(`[INFO]  [${POD_NAME}] ${msg}`, meta ?? ""),
+  warn: (msg: string, meta?: Record<string, unknown>) => console.warn(`[WARN]  [${POD_NAME}] ${msg}`, meta ?? ""),
+  error: (msg: string | Error, meta?: Record<string, unknown>) =>
+    console.error(`[ERROR] [${POD_NAME}] ${msg}`, meta ?? ""),
+}
+
+// connect to PG
+const db = new Client({ connectionString: DATABASE_URL })
+await db.connect()
+log.info("Connected to PostgreSQL")
+
+// create tables
+await db.query(`
+  CREATE TABLE IF NOT EXISTS flowcore_pump_leases (
+    key TEXT PRIMARY KEY,
+    holder TEXT NOT NULL,
+    expires_at TIMESTAMPTZ NOT NULL
+  );
+
+  CREATE TABLE IF NOT EXISTS flowcore_pump_instances (
+    instance_id TEXT PRIMARY KEY,
+    address TEXT NOT NULL,
+    last_heartbeat TIMESTAMPTZ NOT NULL DEFAULT NOW()
+  );
+
+  CREATE TABLE IF NOT EXISTS pump_state (
+    id TEXT PRIMARY KEY,
+    time_bucket TEXT NOT NULL,
+    event_id TEXT,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+  );
+
+  CREATE TABLE IF NOT EXISTS processed_events (
+    id SERIAL PRIMARY KEY,
+    pod_name TEXT NOT NULL,
+    event_id TEXT NOT NULL,
+    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+  );
+`)
+log.info("Database tables ready")
+
+// create components
+const coordinator = new PostgresCoordinator(db)
+const stateManager = new PostgresStateManager(db)
+const fakeDataSource = new FakeDataSource(TOTAL_EVENTS)
+
+// create cluster
+const cluster = new FlowcoreDataPumpCluster({
+  auth: { getBearerToken: () => Promise.resolve("fake") },
+  dataSource: {
+    tenant: "integration-test",
+    dataCore: "test-data-core",
+    flowType: "test-flow-type",
+    eventTypes: ["test-event"],
+  },
+  stateManager,
+  coordinator,
+  dataSourceOverride: fakeDataSource,
+  advertisedAddress: `ws://${POD_IP}:${WS_PORT}`,
+  noTranslation: true,
+  processor: {
+    concurrency: 5,
+    handler: async (events) => {
+      for (const event of events) {
+        await db.query(`INSERT INTO processed_events (pod_name, event_id) VALUES ($1, $2)`, [
+          POD_NAME,
+          event.eventId,
+        ])
+      }
+      log.info(`Processed ${events.length} events`)
+    },
+  },
+  notifier: { type: "poller", intervalMs: 2000 },
+  leaseTtlMs: 15000,
+  leaseRenewIntervalMs: 5000,
+  heartbeatIntervalMs: 3000,
+  workerConcurrency: 5,
+  logger: log,
+})
+
+// start HTTP + WS server
+Deno.serve({ port: WS_PORT }, (req) => {
+  const url = new URL(req.url)
+
+  if (url.pathname === "/health") {
+    return new Response(
+      JSON.stringify({
+        instanceId: cluster.id,
+        isLeader: cluster.isLeaderInstance,
+        workerCount: cluster.activeWorkerCount,
+        isRunning: cluster.isRunning,
+        podName: POD_NAME,
+      }),
+      { headers: { "content-type": "application/json" } },
+    )
+  }
+
+  if (req.headers.get("upgrade")?.toLowerCase() === "websocket") {
+    const { socket, response } = Deno.upgradeWebSocket(req)
+    cluster.handleConnection(socket)
+    return response
+  }
+
+  return new Response("Not found", { status: 404 })
+})
+
+log.info(`HTTP/WS server listening on :${WS_PORT}`)
+
+// start cluster
+await cluster.start()
+log.info("Cluster started")
+
+// graceful shutdown
+Deno.addSignalListener("SIGTERM", async () => {
+  log.info("SIGTERM received, shutting down...")
+  await cluster.stop()
+  await db.end()
+  Deno.exit(0)
+})

integration/app/postgres-coordinator.ts

@@ -0,0 +1,64 @@
+import type { Client as PgClient } from "npm:pg@^8.13.0"
+import type { FlowcoreDataPumpCoordinator } from "../../src/data-pump/types.ts"
+
+export class PostgresCoordinator implements FlowcoreDataPumpCoordinator {
+  constructor(private readonly db: PgClient) {}
+
+  async acquireLease(instanceId: string, key: string, ttlMs: number): Promise<boolean> {
+    const expiresAt = new Date(Date.now() + ttlMs)
+    const result = await this.db.query(
+      `INSERT INTO flowcore_pump_leases (key, holder, expires_at)
+       VALUES ($1, $2, $3)
+       ON CONFLICT (key) DO UPDATE
+         SET holder = $2, expires_at = $3
+         WHERE flowcore_pump_leases.expires_at < NOW() OR flowcore_pump_leases.holder = $2
+       RETURNING holder`,
+      [key, instanceId, expiresAt],
+    )
+    return result.rowCount > 0 && result.rows[0].holder === instanceId
+  }
+
+  async renewLease(instanceId: string, key: string, ttlMs: number): Promise<boolean> {
+    const expiresAt = new Date(Date.now() + ttlMs)
+    const result = await this.db.query(
+      `UPDATE flowcore_pump_leases SET expires_at = $3 WHERE key = $1 AND holder = $2`,
+      [key, instanceId, expiresAt],
+    )
+    return (result.rowCount ?? 0) > 0
+  }
+
+  async releaseLease(instanceId: string, key: string): Promise<void> {
+    await this.db.query(`DELETE FROM flowcore_pump_leases WHERE key = $1 AND holder = $2`, [key, instanceId])
+  }
+
+  async register(instanceId: string, address: string): Promise<void> {
+    await this.db.query(
+      `INSERT INTO flowcore_pump_instances (instance_id, address, last_heartbeat)
+       VALUES ($1, $2, NOW())
+       ON CONFLICT (instance_id) DO UPDATE SET address = $2, last_heartbeat = NOW()`,
+      [instanceId, address],
+    )
+  }
+
+  async heartbeat(instanceId: string): Promise<void> {
+    await this.db.query(`UPDATE flowcore_pump_instances SET last_heartbeat = NOW() WHERE instance_id = $1`, [
+      instanceId,
+    ])
+  }
+
+  async unregister(instanceId: string): Promise<void> {
+    await this.db.query(`DELETE FROM flowcore_pump_instances WHERE instance_id = $1`, [instanceId])
+  }
+
+  async getInstances(staleThresholdMs: number): Promise<Array<{ instanceId: string; address: string }>> {
+    const threshold = new Date(Date.now() - staleThresholdMs)
+    const result = await this.db.query(
+      `SELECT instance_id, address FROM flowcore_pump_instances WHERE last_heartbeat > $1`,
+      [threshold],
+    )
+    return result.rows.map((row: { instance_id: string; address: string }) => ({
+      instanceId: row.instance_id,
+      address: row.address,
+    }))
+  }
+}

integration/app/postgres-state-manager.ts

@@ -0,0 +1,26 @@
+import type { Client as PgClient } from "npm:pg@^8.13.0"
+import type { FlowcoreDataPumpState, FlowcoreDataPumpStateManager } from "../../src/data-pump/types.ts"
+
+export class PostgresStateManager implements FlowcoreDataPumpStateManager {
+  constructor(private readonly db: PgClient) {}
+
+  async getState(): Promise<FlowcoreDataPumpState | null> {
+    const result = await this.db.query(
+      `SELECT time_bucket, event_id FROM pump_state ORDER BY updated_at DESC LIMIT 1`,
+    )
+    if (result.rows.length === 0) return null
+    return {
+      timeBucket: result.rows[0].time_bucket,
+      eventId: result.rows[0].event_id,
+    }
+  }
+
+  async setState(state: FlowcoreDataPumpState): Promise<void> {
+    await this.db.query(
+      `INSERT INTO pump_state (id, time_bucket, event_id, updated_at)
+       VALUES ('default', $1, $2, NOW())
+       ON CONFLICT (id) DO UPDATE SET time_bucket = $1, event_id = $2, updated_at = NOW()`,
+      [state.timeBucket, state.eventId],
+    )
+  }
+}