diff --git a/apps/links/src/index.ts b/apps/links/src/index.ts index f5afeec5f..13f40f226 100644 --- a/apps/links/src/index.ts +++ b/apps/links/src/index.ts @@ -12,6 +12,7 @@ import { evlog } from "evlog/elysia"; import { drain, enrich, flushDrain } from "./lib/logging"; import { calculateLinkReadiness } from "./lib/health"; import { + didKafkaConnectFail, disconnectProducer, getProducerHealthState, refreshProducerConnection, @@ -200,6 +201,9 @@ const app = new Elysia() redis: cache.status, redpanda: redpanda.status, }); + if (didKafkaConnectFail()) { + return Response.json({ status: "unavailable", services }, { status: 503 }); + } return Response.json( { status: readiness.status, diff --git a/apps/links/src/lib/producer.test.ts b/apps/links/src/lib/producer.test.ts index 511915442..926833462 100644 --- a/apps/links/src/lib/producer.test.ts +++ b/apps/links/src/lib/producer.test.ts @@ -308,3 +308,187 @@ describe("producer health state", () => { } }); }); + +describe("health probe failures (issue #719)", () => { + test("health refresh rejects loudly when Kafka is unreachable", async () => { + process.env.REDPANDA_BROKER = "redpanda.test:9092"; + nextProducer = makeProducer({ + connect: () => Promise.reject(new Error("tls handshake failed")), + }); + const { getProducerHealthState, refreshProducerConnection } = + await loadProducer(); + + await expect(refreshProducerConnection()).rejects.toThrow( + "Kafka health probe failed" + ); + expect(setAttributes).toHaveBeenCalledWith({ + kafka_health_connect_failed: true, + }); + expect(getProducerHealthState()).toBe("cooldown"); + }); + + test("health refresh resolves once Kafka is reachable", async () => { + process.env.REDPANDA_BROKER = "redpanda.test:9092"; + nextProducer = makeProducer(); + const { + disconnectProducer, + getProducerHealthState, + refreshProducerConnection, + } = await loadProducer(); + + await refreshProducerConnection(); + + expect(getProducerHealthState()).toBe("connected"); + await disconnectProducer(); + }); + + test("production sends still fall back to ClickHouse when Kafka is down", async () => { + process.env.REDPANDA_BROKER = "redpanda.test:9092"; + nextProducer = makeProducer({ + connect: () => Promise.reject(new Error("broker unavailable")), + }); + const { sendLinkVisit } = await loadProducer(); + + const result = await sendLinkVisit(event, event.link_id); + + expect(result).toBe(true); + expect(clickHouseInsert).toHaveBeenCalledTimes(1); + }); + + test("flags Kafka outages for the health endpoint until recovery", async () => { + process.env.REDPANDA_BROKER = "redpanda.test:9092"; + const realNow = Date.now; + let now = 1000; + Date.now = () => now; + try { + nextProducer = makeProducer({ + connect: () => Promise.reject(new Error("broker unavailable")), + }); + const { + didKafkaConnectFail, + disconnectProducer, + refreshProducerConnection, + warmProducerConnection, + } = await loadProducer(); + + expect(didKafkaConnectFail()).toBe(false); + await warmProducerConnection(); + expect(didKafkaConnectFail()).toBe(true); + + now += 60_001; + nextProducer = makeProducer(); + await refreshProducerConnection(); + + expect(didKafkaConnectFail()).toBe(false); + await disconnectProducer(); + } finally { + Date.now = realNow; + } + }); + + test("suppressed health probes still report a known outage", async () => { + process.env.REDPANDA_BROKER = "redpanda.test:9092"; + nextProducer = makeProducer({ + connect: () => Promise.reject(new Error("tls handshake failed")), + }); + const { didKafkaConnectFail, refreshProducerConnection } = + await loadProducer(); + + await expect(refreshProducerConnection()).rejects.toThrow( + "Kafka health probe failed" + ); + expect(didKafkaConnectFail()).toBe(true); + + await expect(refreshProducerConnection()).rejects.toThrow( + "Kafka health probe failed" + ); + expect(didKafkaConnectFail()).toBe(true); + }); + + test("a failing health-owned attempt does not break concurrent sends", async () => { + process.env.REDPANDA_BROKER = "redpanda.test:9092"; + let releaseConnect!: (error: Error) => void; + nextProducer = makeProducer({ + connect: () => + new Promise((_resolve, reject) => { + releaseConnect = reject; + }), + }); + const { refreshProducerConnection, sendLinkVisit } = + await loadProducer(); + + const health = refreshProducerConnection(); + await Bun.sleep(0); + const send = sendLinkVisit(event, event.link_id); + await Bun.sleep(0); + releaseConnect(new Error("tls handshake failed")); + + await expect(health).rejects.toThrow("Kafka health probe failed"); + await expect(send).resolves.toBe(true); + expect(clickHouseInsert).toHaveBeenCalledTimes(1); + }); + + test("disables SSL when REDPANDA_SSL is false", async () => { + process.env.REDPANDA_BROKER = "redpanda.test:9092"; + process.env.REDPANDA_SSL = "false"; + nextProducer = makeProducer(); + const { disconnectProducer, sendLinkVisit } = await loadProducer(); + + await sendLinkVisit(event, event.link_id); + + expect(kafkaConfigs).toEqual([ + expect.objectContaining({ + brokers: ["redpanda.test:9092"], + ssl: false, + }), + ]); + await disconnectProducer(); + }); + + test("reports connection dependency error when sendLinkVisit joins health-owned attempt", async () => { + process.env.REDPANDA_BROKER = "redpanda.test:9092"; + let releaseConnect!: (error: Error) => void; + const connectErr = new Error("connection failed"); + nextProducer = makeProducer({ + connect: () => + new Promise((_resolve, reject) => { + releaseConnect = reject; + }), + }); + const { refreshProducerConnection, sendLinkVisit } = + await loadProducer(); + + const health = refreshProducerConnection(); + await Bun.sleep(0); + const send = sendLinkVisit(event, event.link_id); + await Bun.sleep(0); + releaseConnect(connectErr); + + await expect(health).rejects.toThrow("Kafka health probe failed"); + await expect(send).resolves.toBe(true); + expect(captureError).toHaveBeenCalledWith(connectErr, { + operation: "kafka_connect", + }); + }); + + test("does not fail disconnectProducer if a pending connection fails during shutdown", async () => { + process.env.REDPANDA_BROKER = "redpanda.test:9092"; + let releaseConnect!: (error: Error) => void; + nextProducer = makeProducer({ + connect: () => + new Promise((_resolve, reject) => { + releaseConnect = reject; + }), + }); + const { disconnectProducer, warmProducerConnection } = + await loadProducer(); + + const warmup = warmProducerConnection(); + await Bun.sleep(0); + const shutdown = disconnectProducer(); + releaseConnect(new Error("connection rejected during shutdown")); + + await warmup; + await expect(shutdown).resolves.toBeUndefined(); + }); +}); diff --git a/apps/links/src/lib/producer.ts b/apps/links/src/lib/producer.ts index ae5b1d99b..183399ea7 100644 --- a/apps/links/src/lib/producer.ts +++ b/apps/links/src/lib/producer.ts @@ -16,6 +16,7 @@ const ASYNC_INSERT_BUSY_TIMEOUT_MS = 50; let producer: Producer | null = null; let connectPromise: Promise | null = null; let nextReconnectAt = 0; +let kafkaConnectFailed = false; let lastConnectErrorLogAt = 0; let lastFallbackErrorLogAt = 0; let shuttingDown = false; @@ -69,6 +70,8 @@ function captureDependencyError( captureError(error, context); } +let shouldReportFailure = false; + function connect(reportFailure = true): Promise { if (shuttingDown) { return Promise.resolve(false); @@ -87,10 +90,23 @@ function connect(reportFailure = true): Promise { return Promise.resolve(false); } + // Set only after the cooldown early return: that return path skips the + // finally that resets the flag, so setting it earlier would leak a stale + // reportFailure=true into the next real connect attempt. + if (reportFailure) { + shouldReportFailure = true; + } + if (connectPromise) { return connectPromise; } + const useSsl = + process.env.REDPANDA_SSL === "false" || + process.env.REDPANDA_SSL_ENABLED === "false" + ? false + : true; + connectPromise = (async () => { let candidate: Producer | null = null; @@ -110,7 +126,7 @@ function connect(reportFailure = true): Promise { ...(username && password ? { sasl: { mechanism: "scram-sha-256", username, password } } : {}), - ssl: true, + ssl: useSsl, }); candidate = kafka.producer({ @@ -131,10 +147,11 @@ function connect(reportFailure = true): Promise { } producer = candidate; nextReconnectAt = 0; + kafkaConnectFailed = false; setAttributes({ kafka_connected: true }); return true; } catch (error) { - if (reportFailure) { + if (shouldReportFailure) { captureDependencyError( error, { operation: "kafka_connect" }, @@ -149,9 +166,11 @@ function connect(reportFailure = true): Promise { producer = null; nextReconnectAt = Date.now() + reconnectCooldownMs; setAttributes({ kafka_connected: false }); + kafkaConnectFailed = true; return false; } finally { connectPromise = null; + shouldReportFailure = false; } })(); @@ -181,12 +200,19 @@ export function getProducerHealthState(): ProducerHealthState { return "idle"; } +export function didKafkaConnectFail(): boolean { + return kafkaConnectFailed; +} + export async function warmProducerConnection(): Promise { await connect(); } export async function refreshProducerConnection(): Promise { - await connect(false); + const connected = await connect(false); + if (!connected) { + throw new Error("Kafka health probe failed"); + } } async function persistLinkVisitDirectly( @@ -249,7 +275,13 @@ export async function sendLinkVisit( kafka_message_key: eventKey ?? "unknown", }); - const kafkaReady = await connect(); + let kafkaReady = false; + try { + kafkaReady = await connect(); + } catch (error) { + kafkaReady = false; + } + const activeProducer = producer; if (!(kafkaReady && activeProducer)) { setAttributes({ @@ -294,7 +326,11 @@ export async function sendLinkVisit( export async function disconnectProducer(): Promise { shuttingDown = true; if (connectPromise) { - await connectPromise; + try { + await connectPromise; + } catch { + // Expected health/connect probe rejection during shutdown should not fail disconnectProducer + } } const activeProducer = producer; producer = null;