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

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);
@@ -87,10 +90,23 @@ function connect(reportFailure = true): Promise<boolean> {
 		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<boolean> {
 				...(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<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 +166,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 +200,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 +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<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;