From 6d3451a1f010178de8efe8a2f36f0965acfdd547 Mon Sep 17 00:00:00 2001 From: jbiskur Date: Wed, 30 Sep 2026 18:58:38 +0100 Subject: [PATCH] feat(pathways): resilient cluster start that never stalls the data pump A failed startPump() or leader bootstrap left a leader that holds the lease with no running pump, so no other instance could take over. The skill now ships a production start runtime: retry the whole start with stopPump() + stopCluster() teardown, exit after the last attempt, exit on "Failed to bootstrap leader runtime after becoming leader", and a leader watchdog. Verified against @flowcore/pathways 2.10.4 and @flowcore/data-pump 0.23.3 source. Co-Authored-By: Claude Opus 5.5 (1M context) --- .claude-plugin/marketplace.json | 2 +- .claude-plugin/plugin.json | 2 +- CHANGELOG.md | 14 + plugin.json | 2 +- .../references/troubleshooting.md | 1 + skills/flowcore-pathways/SKILL.md | 44 ++- .../references/cluster-and-pump.md | 24 +- .../references/evaluation-scenarios.md | 22 ++ .../references/resilient-startup.md | 258 ++++++++++++++++++ .../references/troubleshooting.md | 6 + 10 files changed, 351 insertions(+), 24 deletions(-) create mode 100644 skills/flowcore-pathways/references/resilient-startup.md diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 1d7f746..3662ef0 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -7,7 +7,7 @@ "name": "flowcore", "source": "./", "description": "Operate Flowcore data cores, IAM and Data Pathways, and build apps with @flowcore/pathways.", - "version": "0.1.0", + "version": "0.2.0", "author": { "name": "Flowcore" } } ] diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index ed77f24..43fb7e6 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "flowcore", - "version": "0.1.0", + "version": "0.2.0", "description": "Flowcore MCP server plus skills for the platform model, IAM, Data Pathways and the @flowcore/pathways SDK.", "author": { "name": "Flowcore", "email": "support@flowcore.io", "url": "https://flowcore.io" }, "homepage": "https://flowcore.io", diff --git a/CHANGELOG.md b/CHANGELOG.md index 89d6175..982108d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,19 @@ # Changelog +## [0.2.0] - 2026-09-30 + +### Added +- `flowcore-pathways`: `references/resilient-startup.md`, a production start runtime for virtual pathways. It retries the whole start with `stopPump()` + `stopCluster()` between attempts, exits when the retries run out, exits when a leader holds the lease without a running pump, detects `Failed to bootstrap leader runtime after becoming leader` in the logger adapter, and separates liveness from readiness. It also adds handler timeouts sized to `deliveryTimeoutMs`. +- Evaluation scenario for a resilient cluster start. + +### Changed +- `flowcore-pathways`: new rule 11. A start error must never leave a silent stall. The minimal example now makes one start attempt (`startOnce`) and uses the resilient runtime. +- `flowcore-pathways` troubleshooting: rows for a leader without a pump, `Cluster already started` on retry, `{}` error logs, hanging handlers, dropped events after retries, and Bun `NODE_ENV` inlining. +- `flowcore-data-pathways` troubleshooting: row for a virtual pathway without pulses while pods are healthy. + +### Fixed +- `flowcore-pathways` cluster failure modes: a PostgreSQL outage does not make the leader step down. Lease renewal throws, the error is logged as `Lease loop error`, and the leader keeps its role (verified against `@flowcore/pathways` 2.10.4). + ## [0.1.0] - 2026-09-30 ### Added diff --git a/plugin.json b/plugin.json index 2c62a22..4f0b7b7 100644 --- a/plugin.json +++ b/plugin.json @@ -1,7 +1,7 @@ { "$schema": "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json", "name": "flowcore", - "version": "0.1.0", + "version": "0.2.0", "description": "Connects coding agents to the Flowcore event platform and teaches them to operate it: the resource hierarchy, IAM policies that actually match, virtual and managed Data Pathways, and building apps with the @flowcore/pathways SDK.", "author": { "name": "Flowcore", diff --git a/skills/flowcore-data-pathways/references/troubleshooting.md b/skills/flowcore-data-pathways/references/troubleshooting.md index 4c99178..930a06c 100644 --- a/skills/flowcore-data-pathways/references/troubleshooting.md +++ b/skills/flowcore-data-pathways/references/troubleshooting.md @@ -10,6 +10,7 @@ Start every diagnosis read-only: `get_data_pathway`, then `show_pathway_dashboar | Right event types | Compare `virtualConfig.flowTypes` (virtual) or `config.sources[].eventTypes` (managed) with the event types that actually receive events (`get_time_buckets`) | | Events exist | `get_time_buckets` + `get_events` for the event type | | Virtual: service running | The consumer's pump runs only on the cluster leader. Ask the user for the service logs and leader state. | +| Virtual: no pulses, pods healthy | The leader holds the lease without a running pump (a caught start error, or `Failed to bootstrap leader runtime after becoming leader`). Delete the leader pod to recover. Then fix the service start with the `flowcore-pathways` skill (`references/resilient-startup.md`). | | Managed: endpoint reachable | The endpoint must be public HTTPS and accept the configured auth headers | | Permissions | The pathway or service API key needs `fetch` on the data core (see `flowcore-iam`) | diff --git a/skills/flowcore-pathways/SKILL.md b/skills/flowcore-pathways/SKILL.md index 3eb77bb..b3444c5 100644 --- a/skills/flowcore-pathways/SKILL.md +++ b/skills/flowcore-pathways/SKILL.md @@ -62,7 +62,14 @@ Each rule has a reason. Keep the reason when you explain a decision to the user. (no `fireAndForget`) returns after the handler has processed the event, so you can return the real resource. Use `fireAndForget: true` only when the user asks, or for write-only and remote-consumer events. A write timeout means the event is stored but not yet processed: do not retry the write. -11. **Never invent APIs.** If an option or method is not in the installed library, say +11. **A start error must never leave a silent stall.** Retry the whole start with teardown + (`stopPump()` then `stopCluster()`) between attempts, exit the process when the retries run out, + and exit when a leader holds the lease without a running pump. Serve liveness before the start. + Reason: the lease and the pump are separate. A pod that caught a `startPump()` error, or a new + leader whose bootstrap failed (`Failed to bootstrap leader runtime after becoming leader`), keeps + the lease with no pump, and no other pod can take over. Use the runtime in + [resilient-startup](references/resilient-startup.md) for every production virtual service. +12. **Never invent APIs.** If an option or method is not in the installed library, say "I cannot verify that API" and offer a verified alternative. Known non-existent examples: `enableEncryption()`, field-level encryption, automatic key rotation, `getStatus()`. @@ -88,14 +95,15 @@ virtual pathway registration. Any other value gives a single local pump and no r ```typescript import { z } from "zod" // zod 3 (peer dependency ^3.25.63) import { + type ClusterManager, createPostgresPathwayChunkStore, createPostgresPathwayCoordinator, createPostgresPathwayDeliveryStore, createPostgresPathwayState, createPostgresPumpStateManagerFactory, - ConsoleLogger, PathwaysBuilder, } from "@flowcore/pathways" +import { pathwaysLogger } from "./pathways-logger" // see references/resilient-startup.md const connectionString = process.env.DATABASE_URL! const statePrefix = "orders_service" // one prefix per deployable sharing a database @@ -114,7 +122,7 @@ export const pathways = new PathwaysBuilder({ pathwayName: process.env.FLOWCORE_PATHWAY_NAME!, // distinct per environment pathwayLabels: { name: "Orders service", description: "Consumes order events" }, autoProvision: { pathway: true }, // register the virtual pathway (skipped in development) - logger: new ConsoleLogger(), // default is NoopLogger, which hides every log line below + logger: pathwaysLogger, // from references/resilient-startup.md; the default NoopLogger hides everything }) .withPathwayState(createPostgresPathwayState({ connectionString, statePrefix })) .withPathwayChunkStore(createPostgresPathwayChunkStore({ connectionString, statePrefix })) @@ -138,10 +146,14 @@ export const pathways = new PathwaysBuilder({ await ordersReadModel.upsert(event.payload) }) -export async function startPathways() { +// ONE start attempt. Never call it directly with a bare `.catch()`: run it through +// startPathways() in references/resilient-startup.md, which retries with teardown, exits when the +// retries run out, and watches the leader. Serve HTTP liveness first, then `void startPathways()`. +export async function startOnce(): Promise { + let cluster: ClusterManager | null = null if (isProduction) { const coordinator = await createPostgresPathwayCoordinator({ connectionString }, { statePrefix }) - await pathways.startCluster({ + cluster = await pathways.startCluster({ coordinator, advertisedAddress: process.env.POD_IP!, // host only; the library builds ws://host:port port: 9090, @@ -151,11 +163,7 @@ export async function startPathways() { stateManagerFactory: await createPostgresPumpStateManagerFactory({ connectionString, statePrefix }), notifier: { type: "websocket" }, }) -} - -export async function stopPathways() { - await pathways.stopPump() - await pathways.stopCluster() + return cluster } // In a request handler (same app handles the event): @@ -173,8 +181,10 @@ export async function recordExport(orderId: string, target: string) { } ``` -`ordersReadModel` stands for your own persistence code. In real code, pass an adapter to your own -logger that implements the library `Logger` interface (`debug`, `info`, `warn`, `error`). +`ordersReadModel` stands for your own persistence code. Give the handler a timeout (see +[resilient-startup](references/resilient-startup.md#handler-timeout)); a handler that never settles +blocks its pump group. The logger adapter must log `message`, `name` and `stack` of errors and must +keep the leader-bootstrap fatal detection. ## Verification checklist @@ -193,7 +203,12 @@ Run these before you call the work done. - [ ] Events flow: an awaited `pathways.write()` returns without timeout and the read model changes. To inspect stored events use MCP `get_time_buckets`, then `get_events` with the event type id. - [ ] Generated ORM migrations contain no `pathway_` tables (grep the SQL). -- [ ] Graceful shutdown calls `stopPump()` then `stopCluster()`. +- [ ] Graceful shutdown calls `stopPump()` then `stopCluster()`, once. +- [ ] Production start goes through the resilient runtime: no bare `startPathways().catch(...)`, retry + with teardown, exit after the last attempt, leader watchdog, fatal detection in the logger. +- [ ] Liveness does not depend on pathways; readiness reports the pathways status. +- [ ] Failure test: stop the database during boot, and delete the leader pod. The start either + recovers or the pod exits. No pod stays Ready as leader without a running pump. ## API keys and IAM @@ -211,6 +226,9 @@ every other id is a full UUID. Use the `flowcore-iam` skill for policies, roles pump, `statePrefix`, notifiers, concurrency and pump groups, pause and resume, failure modes. - [references/schema-and-migrations.md](references/schema-and-migrations.md): library-owned tables, Drizzle `tablesFilter`, read-only mirrors, deploy-time migrations. +- [references/resilient-startup.md](references/resilient-startup.md): the production start runtime + (retry with teardown, exit, leader watchdog, health endpoints, handler timeouts) and why the naive + start stalls. - [references/troubleshooting.md](references/troubleshooting.md): symptoms, causes and fixes. - [references/evaluation-scenarios.md](references/evaluation-scenarios.md): scenarios to test answers produced with this skill. diff --git a/skills/flowcore-pathways/references/cluster-and-pump.md b/skills/flowcore-pathways/references/cluster-and-pump.md index f85722c..6139fe4 100644 --- a/skills/flowcore-pathways/references/cluster-and-pump.md +++ b/skills/flowcore-pathways/references/cluster-and-pump.md @@ -29,14 +29,18 @@ await pathways.startPump({ stateManagerFactory, notifier: { type: "websocket" } In production + virtual, `startPump()` without an active cluster throws `Cluster mode must be started before production virtual pump startup`. -Shutdown: +This is one attempt. In production run it through the runtime in +[resilient-startup](resilient-startup.md): retry with `stopPump()` + `stopCluster()` between attempts, +exit when the retries run out, and exit when a leader has no running pump. Two library facts make +this necessary: -```typescript -process.on("SIGTERM", async () => { - await pathways.stopPump() - await pathways.stopCluster() -}) -``` +- A failed `startCluster()` leaves the cluster manager in place. A second call throws + `Cluster already started` until `stopCluster()` runs. +- The lease is not tied to the pump. A leader whose pump start or bootstrap failed keeps renewing the + lease, and no other instance can take over. + +Shutdown, once: `await pathways.stopPump()` then `await pathways.stopCluster()`. `stopCluster()` +releases the lease when this instance leads. ## `startCluster(options)` (`PathwayClusterOptions`) @@ -147,5 +151,9 @@ ignored (all pumps are reset, with a warning). | Leader crashes | Lease expires after `leaseTtlMs`; another instance takes over from the saved cursor. | | Follower crashes | Dropped after `staleThresholdMs`; events go to the remaining instances. | | No followers | Leader processes events itself. | -| PostgreSQL unavailable | Lease renewal fails, the leader steps down and stops its pump. No instance leads until the database is back. | +| PostgreSQL unavailable | Lease renewal throws. The error is logged as `Lease loop error` and the leader keeps its role. No other instance can acquire the lease until the database is back. | +| `startPump()` throws on the leader, and the app continues | The leader keeps the lease with no pump. Delivery stops for the whole cluster. Prevent it with [resilient-startup](resilient-startup.md). | +| A new leader fails its bootstrap | Logged as `Failed to bootstrap leader runtime after becoming leader`. No retry and no lease release. Same stall. The runtime in [resilient-startup](resilient-startup.md) exits on it. | +| A data pump group fails after a successful start | The library restarts that group with backoff (1 s doubling to 30 s, without limit). The other groups continue. | +| Handler never settles | Its pump group stops delivery. Give handlers a timeout. | | Network partition | Two leaders are possible for up to `leaseTtlMs`. Keep handlers idempotent. | diff --git a/skills/flowcore-pathways/references/evaluation-scenarios.md b/skills/flowcore-pathways/references/evaluation-scenarios.md index d449bc6..9bf02c1 100644 --- a/skills/flowcore-pathways/references/evaluation-scenarios.md +++ b/skills/flowcore-pathways/references/evaluation-scenarios.md @@ -36,6 +36,28 @@ Must not include: - managed delivery as an automatic fallback - cluster mode inside Next.js or another serverless runtime +## Resilient cluster start + +Input: "Our virtual pathway stopped delivering after a deploy, but all pods are healthy." or +"Make the production start of this pathways service robust." + +Must include: +- the cause: a leader holds the lease without a running pump, after a caught `startPump()` error or + a failed leader bootstrap (`Failed to bootstrap leader runtime after becoming leader`) +- immediate recovery: delete the leader pod, then check pulses with `get_data_pathway` / + `show_pathway_dashboard` +- a start that retries with `stopPump()` + `stopCluster()` between attempts and exits after the last + attempt +- a leader watchdog and fatal detection in the logger adapter +- liveness independent of pathways, readiness that reports the pathways status +- error logs with `message`, `name` and `stack`, and a handler timeout + +Must not include: +- `startPathways().catch(console.error)` or any start error that is only logged +- a retry that calls `startCluster()` again without `stopCluster()` +- `pathways.pump?.isRunning` as the only health signal +- disabling or switching the pathway to managed to make delivery work + ## Serverless production target Input: "Deploy the consumer to Vercel." diff --git a/skills/flowcore-pathways/references/resilient-startup.md b/skills/flowcore-pathways/references/resilient-startup.md new file mode 100644 index 0000000..90f3120 --- /dev/null +++ b/skills/flowcore-pathways/references/resilient-startup.md @@ -0,0 +1,258 @@ +# Resilient cluster start + +Verified against `@flowcore/pathways` 2.10.4 (`src/pathways/builder.ts`, +`src/pathways/cluster/cluster-manager.ts`, `src/pathways/pump/pathway-pump.ts`) and +`@flowcore/data-pump` 0.23.3. + +Use this in every production virtual service. The naive `await startCluster(); await startPump()` +works on a good day. On a bad day it leaves a pod that looks healthy while no event is delivered. + +## Why the naive start stalls + +The cluster lease and the pump are separate. The lease loop renews the lease for as long as the +process runs and the database answers. Nothing in the library ties the lease to a working pump. + +| Failure | What the library does | Result without the pattern below | +|---|---|---| +| `startPump()` throws on the leader (transient 5xx or 401 from Flowcore, IAM 403, database error, missing event type) | Stops the pump and rethrows. The cluster keeps running and keeps the lease. | The app logs the error and continues. The pod leads the cluster with no pump. Nothing is delivered and no other pod can take over. | +| `startPump()` throws on a follower | Rethrows before the pump object exists. | If this pod becomes leader later, the leadership handler sees no pump and does nothing. Same stall. | +| A follower becomes leader and the pump bootstrap fails (delivery state read, data pump start, pathway registration) | Logs `Failed to bootstrap leader runtime after becoming leader` and nothing else. No retry, no lease release. | The new leader keeps the lease with no pump. This is the most common silent stall. | +| A data pump group fails at start | `PathwayPump` already set `isRunning` to `true`, and the groups after the failed one never start. | `pathways.pump?.isRunning` reports `true` while some flow types are dead. Do not use it as the only health signal. | +| A retry calls `startCluster()` again | Throws `Cluster already started`. A failed start leaves the cluster manager in place. | The retry loop fails on every attempt and never reaches `startPump()`. | +| `logger.error(msg, { error })` in the app | `Error` fields are not enumerable. | Logs show `{}` and hide the cause. | + +The library already restarts a data pump group with backoff when it fails *after* a successful start +(1 s doubling to 30 s, without limit). The gap is the start itself and the leader bootstrap. + +## The rules + +1. **Never swallow a start error.** `startPathways().catch(console.error)` is the root cause of most + stalls. A start error must end in a retry or in a process exit. +2. **Retry the whole start, and tear down before each retry.** Call `stopPump()` then `stopCluster()` + after every failed attempt. `stopCluster()` releases the lease if this pod holds it, so another pod + can lead at once. Then start the cluster and the pump again from scratch. +3. **Bound the retries, then exit.** After the last attempt, tear down and exit with a non-zero code, + so the orchestrator restarts the pod and the crash loop is visible. A pod that stays alive with a + failed start must not hold the lease. +4. **Treat a failed leader bootstrap as fatal.** Detect `Failed to bootstrap leader runtime after + becoming leader` in the logger adapter, then tear down and exit. +5. **Watch the leader.** A leader without a running pump for longer than a grace period tears down and + exits. This catches every path that rule 4 does not. +6. **Serve HTTP before the pathways start.** Liveness must not depend on pathways. Readiness reports + the pathways status. Otherwise a slow start fails the liveness probe and hides the real error. +7. **Log `message`, `name` and `stack` of every error explicitly.** +8. **Give every handler a timeout.** A handler that never settles blocks its pump group for good. A + handler that throws is retried (`maxRetries`, default 3), then redelivered by the data pump + (`maxRedeliveryCount`, default 3), then dropped with the log line `Failed N events`. It does not + stall the pump, but the event is lost, so record it with `onAnyError`. + +## Reference implementation + +Two files. They replace `startPathways` and `stopPathways` of the minimal example in `SKILL.md`. +Node and Bun. Under Deno replace `process.exit(1)` with `Deno.exit(1)`. + +### `pathways-logger.ts` + +Pass this logger to `new PathwaysBuilder({ logger: pathwaysLogger, ... })`. Route the lines to your +own logger in real code. Keep the fatal detection. + +```typescript +import type { Logger, LoggerMeta } from "@flowcore/pathways" + +// The library logs this line and does nothing else: the leader keeps its lease with no pump. +const LEADER_BOOTSTRAP_FAILED = "Failed to bootstrap leader runtime after becoming leader" + +let onFatal: (reason: string, error?: Error) => void = () => {} + +/** Called by the pathways runtime. Kept separate so the builder module has no import cycle. */ +export function setPathwaysFatalHandler(handler: (reason: string, error?: Error) => void) { + onFatal = handler +} + +export function describeError(error: unknown) { + return error instanceof Error + ? { name: error.name, message: error.message, stack: error.stack } + : { message: String(error) } +} + +export const pathwaysLogger: Logger = { + debug: (message: string, context?: LoggerMeta) => console.debug(message, context ?? {}), + info: (message: string, context?: LoggerMeta) => console.info(message, context ?? {}), + warn: (message: string, context?: LoggerMeta) => console.warn(message, context ?? {}), + error(messageOrError: string | Error, errorOrContext?: Error | LoggerMeta, context?: LoggerMeta) { + const message = typeof messageOrError === "string" ? messageOrError : messageOrError.message + const error = messageOrError instanceof Error + ? messageOrError + : errorOrContext instanceof Error + ? errorOrContext + : undefined + const meta = errorOrContext instanceof Error ? context : errorOrContext + console.error(message, { ...meta, ...(error ? { error: describeError(error) } : {}) }) + if (message === LEADER_BOOTSTRAP_FAILED) onFatal(message, error) + }, +} +``` + +### `pathways-runtime.ts` + +```typescript +import type { ClusterManager } from "@flowcore/pathways" +import { pathways, startOnce } from "./pathways" // the module from SKILL.md, built with pathwaysLogger +import { describeError, pathwaysLogger as log, setPathwaysFatalHandler } from "./pathways-logger" + +const MAX_START_ATTEMPTS = 6 // about 30 seconds of backoff, then exit +const LEADER_CHECK_INTERVAL_MS = 10_000 +const LEADER_GRACE_MS = 60_000 // leader bootstrap includes provisioning and registration + +type PathwaysStatus = "starting" | "up" | "down" +let status: PathwaysStatus = "starting" +let lastError: string | undefined +let cluster: ClusterManager | null = null +let stopping = false +let watchdog: ReturnType | undefined + +/** For the readiness endpoint. Never use it for liveness. */ +export function pathwaysHealth() { + return { status, lastError, role: cluster?.currentRole ?? "standalone" } +} + +async function teardown() { + // stopPump first, then stopCluster: stopCluster releases the lease when this pod leads. + await pathways.stopPump().catch((err) => log.warn("stopPump failed", { error: describeError(err) })) + await pathways.stopCluster().catch((err) => log.warn("stopCluster failed", { error: describeError(err) })) + cluster = null +} + +async function fail(reason: string, error?: unknown) { + if (stopping) return + stopping = true + status = "down" + if (watchdog) clearInterval(watchdog) + log.error(`${reason}. Exiting so the orchestrator restarts this instance`, { + error: error === undefined ? undefined : describeError(error), + }) + await teardown() // release the lease so another instance can lead now + process.exit(1) +} + +function startLeaderWatchdog() { + if (!cluster) return // standalone pump: rule 4 and the start retry cover it + let unhealthySince: number | null = null + watchdog = setInterval(() => { + const leaderWithoutPump = cluster?.isLeader === true && pathways.pump?.isRunning !== true + if (!leaderWithoutPump) { + unhealthySince = null + return + } + unhealthySince ??= Date.now() + if (Date.now() - unhealthySince >= LEADER_GRACE_MS) { + void fail("Cluster leader holds the lease but runs no pump") + } + }, LEADER_CHECK_INTERVAL_MS) +} + +/** Never rejects. Start it after the HTTP server listens: `void startPathways()`. */ +export async function startPathways() { + setPathwaysFatalHandler((reason, error) => void fail(reason, error)) + + for (let attempt = 1; attempt <= MAX_START_ATTEMPTS; attempt++) { + if (stopping) return + try { + cluster = await startOnce() // startCluster (production) + startPump, one attempt + status = "up" + lastError = undefined + startLeaderWatchdog() + log.info("Pathways started", { attempt, role: cluster?.currentRole ?? "standalone" }) + return + } catch (err) { + lastError = describeError(err).message + log.error("Pathways start attempt failed", { attempt, error: describeError(err) }) + await teardown() // never keep a half-started cluster that holds the lease + if (attempt < MAX_START_ATTEMPTS) { + await new Promise((resolve) => setTimeout(resolve, Math.min(1_000 * 2 ** (attempt - 1), 30_000))) + } + } + } + await fail(`Pathways did not start after ${MAX_START_ATTEMPTS} attempts`) +} + +export async function stopPathways() { + if (stopping) return + stopping = true + if (watchdog) clearInterval(watchdog) + await teardown() +} +``` + +### Boot + +```typescript +const server = startHttpServer({ + "/health/live": () => ({ ok: true }), // no dependency on pathways or the database + "/health/ready": () => { + const health = pathwaysHealth() + return { ok: health.status === "up", pathways: health } + }, +}) +void startPathways() + +let shuttingDown = false +process.on("SIGTERM", async () => { + if (shuttingDown) return + shuttingDown = true + await server.stop() + await stopPathways() + await closeDatabasePool() + process.exit(0) +}) +``` + +`startHttpServer` and `closeDatabasePool` stand for your own code. A service whose HTTP routes await +`pathways.write()` can also return 503 from those routes while the status is not `up`. + +## Handler timeout + +```typescript +function withTimeout(work: Promise, ms: number, label: string): Promise { + let timer: ReturnType | undefined + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${label} timed out after ${ms}ms`)), ms) + }) + return Promise.race([work, timeout]).finally(() => clearTimeout(timer)) +} + +pathways.handle("order.0/order.placed.0", (event) => + withTimeout(ordersReadModel.upsert(event.payload), 5_000, `order.placed ${event.eventId}`) +) + +pathways.onAnyError((error, event, pathway) => { + log.error("Handler failed", { pathway, eventId: event.eventId, error: describeError(error) }) +}) +``` + +The timeout does not cancel the work. Keep handlers idempotent, and give database and HTTP clients +their own timeouts too. + +In cluster mode one delivery to a follower includes the in-process retries. Keep +`(maxRetries + 1) × handler timeout + retry delays` below `deliveryTimeoutMs` (default 30 000). +With the defaults (`maxRetries` 3, `retryDelayMs` 500, delays 500 + 1 000 + 1 500 ms), a 5 000 ms +timeout gives 23 000 ms. A longer total makes the leader time out the delivery and process the +event again itself. + +## Kubernetes + +- Liveness probe: `/health/live`. Readiness probe: `/health/ready`. +- `terminationGracePeriodSeconds` longer than the slowest handler plus shutdown. +- `POD_IP` from the downward API (`status.podIP`) and the cluster port open between pods. +- Alert on restarts and on pathway pulses. An exit caused by this pattern is a crash loop you can + see, which is the point. + +## Verify + +- Kill the database connection during boot: every attempt logs `Pathways start attempt failed` with + a real message, and the pod exits after the last attempt. +- Scale to two pods and delete the leader: the other pod logs `Acquired leader lease`, then + `Pump started`, within `leaseTtlMs` plus the bootstrap time. +- In a test environment only: revoke the `fetch` action of the API key and delete the leader pod: the new leader fails its + bootstrap, exits, and releases the lease. Restore the key and the cluster recovers without help. +- `get_data_pathway` and `show_pathway_dashboard` show fresh pulses after each test. diff --git a/skills/flowcore-pathways/references/troubleshooting.md b/skills/flowcore-pathways/references/troubleshooting.md index 3110a44..8968e81 100644 --- a/skills/flowcore-pathways/references/troubleshooting.md +++ b/skills/flowcore-pathways/references/troubleshooting.md @@ -5,6 +5,12 @@ hides every message quoted here. | Symptom | Likely cause | Fix | |---|---|---| +| Pods are Ready and `pathways.write()` succeeds, but nothing is processed; the pathway shows stale or no pulses | The leader holds the lease without a running pump: the app caught a `startPump()` error, or a new leader logged `Failed to bootstrap leader runtime after becoming leader` | Find the leader (`Acquired leader lease` in the logs) and delete that pod to recover now. Then adopt [resilient-startup](resilient-startup.md) so a failed start retries with teardown and ends in an exit. | +| `Cluster already started` on a start retry | The retry calls `startCluster()` again after a failed start | Call `stopPump()` and `stopCluster()` before every retry, or use [resilient-startup](resilient-startup.md). | +| Error logs show `{}` or `error: {}` | The app logs `{ error }`; `Error` fields are not enumerable | Log `message`, `name` and `stack` explicitly (see `describeError` in [resilient-startup](resilient-startup.md)). | +| One pump group stops while the others continue, and no error repeats | A handler never settles | Add a handler timeout and client timeouts. See [resilient-startup](resilient-startup.md#handler-timeout). | +| Events are skipped after handler errors, with `Failed N events` in the logs | The handler threw on every retry and every redelivery (`maxRetries` and `maxRedeliveryCount`, default 3 each), so the data pump dropped the event | Fix the handler, record failures with `onAnyError`, and replay with the control plane if needed. | +| Production behaves like development under Bun | Bun inlined `NODE_ENV` at build time | Read the runtime mode from a variable Bun does not inline, or set `runtimeEnv` explicitly. | | `Cluster mode must be started before production virtual pump startup` | Production + virtual without `startCluster()` | Call `startCluster()` before `startPump()`. On serverless, stop and ask the user (long-running service or managed). | | Production never runs a local pump and logs `Not starting local pump — production managed pathways rely on control-plane delivery` | `pathwayMode` not set; production defaults to `managed` | Set `pathwayMode: "virtual"` unless the user chose managed. | | `managedConfig.endpointUrl is required when provisioning a managed pathway` | Managed registration without an endpoint | Add `managedConfig`, or set `pathwayMode: "virtual"`. |