fix(links): fail health check when Kafka delivery is unavailable

.github/workflows/health-check.yml

@@ -739,6 +739,7 @@ jobs:
             -e REDIS_URL=redis://localhost:6379 \
             -e BULLMQ_REDIS_URL=redis://localhost:6379/4 \
             -e REDPANDA_BROKER=localhost:9092 \
+            -e REDPANDA_SSL=false \
             -e DASHBOARD_URL=http://localhost:3000 \
             -e CLICKHOUSE_URL=http://default:@localhost:8123/databuddy_analytics \
             links:test
@@ -757,7 +758,7 @@ jobs:
           for i in {1..30}; do
             STATUS_BODY=$(curl -sS http://localhost:2500/health/status)
             echo "Links /health/status: $STATUS_BODY"
-            if echo "$STATUS_BODY" | jq -e '.status == "ok" or .status == "degraded"' > /dev/null; then
+            if echo "$STATUS_BODY" | jq -e '.status == "ok"' > /dev/null; then
               echo "Links dependency health is valid"
               break
             fi

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,

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<void>((_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<void>((_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<void>((_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();
+	});
+});

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<boolean> | 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<boolean> {
 	if (shuttingDown) {
 		return Promise.resolve(false);
@@ -81,6 +84,10 @@ function connect(reportFailure = true): Promise<boolean> {
 		return Promise.resolve(false);
 	}
 
+	if (reportFailure) {
+		shouldReportFailure = true;
+	}
+
 	const now = Date.now();
 	if (now < nextReconnectAt) {
 		setAttributes({ kafka_reconnect_suppressed: true });
@@ -91,6 +98,12 @@ function connect(reportFailure = true): Promise<boolean> {
 		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 +123,7 @@ function connect(reportFailure = true): Promise<boolean> {
 				...(username && password
 					? { sasl: { mechanism: "scram-sha-256", username, password } }
 					: {}),
-				ssl: true,
+				ssl: useSsl,
 			});
 
 			candidate = kafka.producer({
@@ -131,10 +144,11 @@ function connect(reportFailure = true): Promise<boolean> {
 			}
 			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 +163,11 @@ function connect(reportFailure = true): Promise<boolean> {
 			producer = null;
 			nextReconnectAt = Date.now() + reconnectCooldownMs;
 			setAttributes({ kafka_connected: false });
+			kafkaConnectFailed = true;
 			return false;
 		} finally {
 			connectPromise = null;
+			shouldReportFailure = false;
 		}
 	})();
 
@@ -181,12 +197,19 @@ export function getProducerHealthState(): ProducerHealthState {
 	return "idle";
 }
 
+export function didKafkaConnectFail(): boolean {
+	return kafkaConnectFailed;
+}
+
 export async function warmProducerConnection(): Promise<void> {
 	await connect();
 }
 
 export async function refreshProducerConnection(): Promise<void> {
-	await connect(false);
+	const connected = await connect(false);
+	if (!connected) {
+		throw new Error("Kafka health probe failed");
+	}
 }
 
 async function persistLinkVisitDirectly(
@@ -249,7 +272,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 +323,11 @@ export async function sendLinkVisit(
 export async function disconnectProducer(): Promise<void> {
 	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;