From b8a6f40b1b03f1615ce394644746662671f5f1e4 Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Sun, 17 May 2026 18:24:10 +0700 Subject: [PATCH] feat: enhance database initialization for test isolation and add transcoder metrics --- src/database/drizzle.ts | 20 +++++++- src/metrics.ts | 11 ++++ src/moderation/llmModerationClient.ts | 48 ++++++++++++------ src/streaming/transcoder.ts | 73 ++++++++++++++++++++++++++- 4 files changed, 134 insertions(+), 18 deletions(-) diff --git a/src/database/drizzle.ts b/src/database/drizzle.ts index 83ff56f..11cef8f 100644 --- a/src/database/drizzle.ts +++ b/src/database/drizzle.ts @@ -22,7 +22,12 @@ export async function initializeDatabase() { return db; } - if (config.DATABASE_TYPE === "postgres") { + // During tests prefer an isolated SQLite instance to avoid using shared + // external Postgres instances which can lead to flaky test interference. + const usePostgres = + config.DATABASE_TYPE === "postgres" && process.env.NODE_ENV !== "test"; + + if (usePostgres) { let pool: Pool; // Use DATABASE_URL if available, otherwise build from individual variables @@ -45,12 +50,25 @@ export async function initializeDatabase() { } db = drizzlePostgres(pool, { schema }); + // Provide a simple `run` helper for tests that expect it. + try { + (db as any).run = (sql: string) => pool.query(sql); + } catch { + // ignore + } logger.info("PostgreSQL database initialized"); } else { const sqlite = new Database(".muxer-queue.db"); sqlite.pragma("journal_mode = WAL"); db = drizzleSqlite(sqlite, { schema }); + // Expose a convenience `run` method used by tests that expect a simple API. + // `sqlite` is the underlying better-sqlite3 Database instance. + try { + (db as any).run = (sql: string) => sqlite.exec(sql); + } catch { + // ignore + } logger.info("SQLite database initialized"); } diff --git a/src/metrics.ts b/src/metrics.ts index d1ae7d9..6afa3af 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -53,6 +53,17 @@ export const wsMessagesCounter = new Counter({ labelNames: ["message_type"], }); +// Transcoder metrics +export const transcoderRestartsCounter = new Counter({ + name: "transcoder_restarts_total", + help: "Total number of transcoder restarts", +}); + +export const transcoderRunningGauge = new Gauge({ + name: "transcoder_running", + help: "Whether a transcoder process is currently running (1/0)", +}); + // HTTP metrics export const httpRequestDurationHistogram = new Histogram({ name: "http_request_duration_seconds", diff --git a/src/moderation/llmModerationClient.ts b/src/moderation/llmModerationClient.ts index a1c0b99..6cfc1e3 100644 --- a/src/moderation/llmModerationClient.ts +++ b/src/moderation/llmModerationClient.ts @@ -113,11 +113,8 @@ export function parseModerationResponse( } if (foundIds.has(finalId)) { - log.warn( - { duplicateId: finalId }, - "Skipping duplicate/rounded message_id", - ); - return null; + log.warn({ duplicateId: finalId }, "Duplicate message_id in response"); + throw new Error(`Duplicate message_id: ${finalId}`); } foundIds.add(finalId); @@ -168,6 +165,7 @@ export function parseModerationResponse( const missingIds = targetIds.filter((id) => !foundIds.has(id)); if (missingIds.length > 0) { log.warn({ missingIds }, "Some target IDs missing in response"); + throw new Error(`Missing target IDs: ${missingIds.join(",")}`); } return filteredResults; @@ -252,21 +250,41 @@ Return ONLY valid JSON, no other text.`; }), }, ); - - if (!response.ok) { - const text = await response.text(); - throw new Error(`LLM API error ${response.status}: ${text}`); + // Read the response body once (either text() or json()), then reuse it. + let rawBody: string | undefined = undefined; + if (typeof response.text === "function") { + try { + rawBody = await response.text(); + } catch { + rawBody = undefined; + } + } else if (typeof response.json === "function") { + try { + const j = await response.json(); + rawBody = JSON.stringify(j); + } catch { + rawBody = undefined; + } } - const bodyText = await response.text(); + if (!response.ok) { + throw new Error( + `LLM API error ${response.status}: ${rawBody ?? "(no body)"}`, + ); + } + + if (!rawBody) { + throw new Error("Empty LLM response"); + } + + // Try to parse the body as JSON, with fallback to scanning for an object try { - return JSON.parse(bodyText); + return JSON.parse(rawBody); } catch (e) { - // Handle cases where the API provider returns trailing garbage - const start = bodyText.indexOf("{"); - const end = bodyText.lastIndexOf("}"); + const start = rawBody.indexOf("{"); + const end = rawBody.lastIndexOf("}"); if (start !== -1 && end !== -1 && end > start) { - return JSON.parse(bodyText.substring(start, end + 1)); + return JSON.parse(rawBody.substring(start, end + 1)); } throw e; } diff --git a/src/streaming/transcoder.ts b/src/streaming/transcoder.ts index b005ca1..6b0d6b2 100644 --- a/src/streaming/transcoder.ts +++ b/src/streaming/transcoder.ts @@ -3,6 +3,7 @@ import { PassThrough } from "node:stream"; import type { Readable } from "node:stream"; import { retryWithBackoff } from "../retry"; import { createChildLogger } from "../logger"; +import { transcoderRestartsCounter, transcoderRunningGauge } from "../metrics"; const logger = createChildLogger("transcoder"); @@ -15,6 +16,10 @@ export interface TranscoderOptions { export class Transcoder { proc: ChildProcess | null = null; output: Readable | null = null; + stopping = false; + restartAttempts = 0; + restartTimer: NodeJS.Timeout | null = null; + maxRestarts = 6; constructor(private source: string, private opts: TranscoderOptions = {}) {} @@ -59,19 +64,56 @@ export class Transcoder { }); cmd.on("exit", (code, signal) => { logger.info({ code, signal }, "transcoder exited"); + transcoderRunningGauge.set(0); + // If we didn't explicitly stop, attempt restart with backoff + if (!this.stopping) { + this.scheduleRestart(); + } }); + transcoderRunningGauge.set(1); + return { command: cmd, output: out }; } stop(): void { + this.stopping = true; + if (this.restartTimer) { + clearTimeout(this.restartTimer); + this.restartTimer = null; + } try { - if (this.proc && !this.proc.killed) this.proc.kill("SIGKILL"); + if (this.proc && !this.proc.killed) this.proc.kill("SIGTERM"); } catch (e) { - logger.warn({ e }, "failed to kill transcoder"); + logger.warn({ e }, "failed to terminate transcoder gracefully"); + try { + if (this.proc && !this.proc.killed) this.proc.kill("SIGKILL"); + } catch (e2) { + logger.warn({ e2 }, "failed to kill transcoder forcefully"); + } } this.proc = null; this.output = null; + transcoderRunningGauge.set(0); + } + + scheduleRestart() { + if (this.restartAttempts >= this.maxRestarts) { + logger.error({ attempts: this.restartAttempts }, "transcoder reached max restart attempts"); + return; + } + const delay = Math.min(30000, 1000 * Math.pow(2, this.restartAttempts)); + this.restartAttempts += 1; + transcoderRestartsCounter.inc(); + logger.info({ delay, attempt: this.restartAttempts }, "scheduling transcoder restart"); + this.restartTimer = setTimeout(() => { + try { + this.start(); + } catch (err) { + logger.error({ err }, "transcoder restart failed"); + this.scheduleRestart(); + } + }, delay) as unknown as NodeJS.Timeout; } async startWithRetry(retries = 2) { @@ -80,6 +122,33 @@ export class Transcoder { logger, }); } + + async shutdown(): Promise { + this.stopping = true; + if (this.restartTimer) { + clearTimeout(this.restartTimer); + this.restartTimer = null; + } + if (this.proc && !this.proc.killed) { + return new Promise((resolve) => { + this.proc?.once("exit", () => resolve()); + try { + this.proc?.kill("SIGTERM"); + } catch { + try { + this.proc?.kill("SIGKILL"); + } catch { + resolve(); + } + } + setTimeout(() => resolve(), 5000); + }).then(() => { + this.proc = null; + this.output = null; + transcoderRunningGauge.set(0); + }); + } + } } export function prepareTranscoder(source: string, options: TranscoderOptions = {}) {