diff --git a/api/db/connect.js b/api/db/connect.js
index 3534884b57..a63d3301b6 100644
--- a/api/db/connect.js
+++ b/api/db/connect.js
@@ -1,10 +1,12 @@
require('dotenv').config();
-const { isEnabled } = require('@librechat/api');
+const { isEnabled, instrumentMongooseQueryMetrics } = require('@librechat/api');
const { logger } = require('@librechat/data-schemas');
const mongoose = require('mongoose');
const MONGO_URI = process.env.MONGO_URI;
+instrumentMongooseQueryMetrics(mongoose);
+
if (!MONGO_URI) {
throw new Error('Please define the MONGO_URI environment variable');
}
diff --git a/api/server/index.metrics.spec.js b/api/server/index.metrics.spec.js
index 3f29347292..c907aca3b6 100644
--- a/api/server/index.metrics.spec.js
+++ b/api/server/index.metrics.spec.js
@@ -72,16 +72,18 @@ describe('Server metrics route', () => {
mongoServer = await MongoMemoryServer.create();
process.env.MONGO_URI = mongoServer.getUri();
process.env.PORT = '0';
+ process.env.METRICS_SECRET = 'test-secret';
app = require('~/server');
await healthCheckPoll(app);
});
afterEach(() => {
- delete process.env.METRICS_SECRET;
+ process.env.METRICS_SECRET = 'test-secret';
});
afterAll(async () => {
+ delete process.env.METRICS_SECRET;
await mongoServer.stop();
await mongoose.disconnect();
});
diff --git a/packages/api/src/app/metrics.spec.ts b/packages/api/src/app/metrics.spec.ts
index 6bb66ed278..80984dcb8b 100644
--- a/packages/api/src/app/metrics.spec.ts
+++ b/packages/api/src/app/metrics.spec.ts
@@ -1,5 +1,19 @@
///
-import { normalizePath } from './metrics';
+import { EventEmitter } from 'events';
+import express from 'express';
+import type { Request, Response } from 'express';
+import {
+ createMetrics,
+ instrumentMongooseQueryMetrics,
+ normalizePath,
+ recordGenerationJob,
+ recordGenerationStreamResumePendingEvents,
+ recordGenerationStreamSubscription,
+ recordOpenIDUserLookup,
+ setGenerationJobsInFlight,
+} from './metrics';
+
+const request = require('supertest') as (app: express.Express) => any;
describe('normalizePath', () => {
it.each([
@@ -9,8 +23,17 @@ describe('normalizePath', () => {
['/api/messages/artifact/507f1f77bcf86cd799439012', '/api/messages/artifact/#id'],
['/api/convos/507f1f77bcf86cd799439011', '/api/convos/#id'],
['/api/files/507f1f77bcf86cd799439011', '/api/files/#id'],
+ ['/api/files/507f1f77bcf86cd799439011/preview', '/api/files/#id/preview'],
+ ['/api/files/download/user-123/file-456', '/api/files/download/#id/#id'],
+ ['/api/files/download-url/user-123/file-456', '/api/files/download-url/#id/#id'],
+ ['/api/files/code/download/session-123/file-456', '/api/files/code/download/#id/#id'],
['/api/agents/507f1f77bcf86cd799439011', '/api/agents/#id'],
+ ['/api/agents/chat/stream/stream-123', '/api/agents/chat/stream/#id'],
+ ['/api/agents/chat/status/convo-123', '/api/agents/chat/status/#id'],
+ ['/api/agents/v1/chat/completions', '/api/agents/v1/chat/completions'],
+ ['/api/agents/v1/responses', '/api/agents/v1/responses'],
['/api/assistants/507f1f77bcf86cd799439011', '/api/assistants/#id'],
+ ['/api/skills/507f1f77bcf86cd799439011/files/reference.md', '/api/skills/#id/files'],
['/api/share/some-token-value', '/api/share/#token'],
['/share/shareId-with_nanoidChars', '/share/#id'],
['/share/shareId-with_nanoidChars/edit', '/share/#id/edit'],
@@ -45,3 +68,316 @@ describe('normalizePath', () => {
expect(normalizePath(input)).toBe(normalized);
});
});
+
+describe('createMetrics', () => {
+ afterEach(() => {
+ delete process.env.METRICS_SECRET;
+ });
+
+ it('uses a no-op middleware and unauthorized router when metrics are not configured', async () => {
+ const app = express();
+ const { metricsMiddleware, metricsRouter } = createMetrics();
+ const res = new EventEmitter();
+ const next = jest.fn();
+
+ metricsMiddleware(
+ { headers: {}, method: 'GET', path: '/api/slow-response' } as Request,
+ res as unknown as Response,
+ next,
+ );
+ app.use('/metrics', metricsRouter);
+
+ await request(app).get('/metrics').expect(401);
+ expect(next).toHaveBeenCalledTimes(1);
+ expect(res.listenerCount('finish')).toBe(0);
+ expect(res.listenerCount('close')).toBe(0);
+ });
+
+ it('tracks request counts, in-flight gauges, and request body sizes', async () => {
+ const app = express();
+ process.env.METRICS_SECRET = 'test-secret';
+ const { metricsMiddleware, metricsRouter } = createMetrics();
+ app.use(metricsMiddleware);
+ app.use(express.text({ type: '*/*' }));
+ app.post('/api/files/:id', (_req, res) => {
+ res.status(201).send('ok');
+ });
+ app.use('/metrics', metricsRouter);
+
+ await request(app)
+ .post('/api/files/507f1f77bcf86cd799439011')
+ .set('Content-Type', 'text/plain')
+ .send('x'.repeat(42))
+ .expect(201);
+
+ const response = await request(app)
+ .get('/metrics')
+ .set('Authorization', 'Bearer test-secret')
+ .expect(200);
+
+ expect(response.text).toMatch(
+ /http_requests_total\{method="POST",path="\/api\/files\/#id",status="201"\} 1/,
+ );
+ expect(response.text).toMatch(
+ /http_requests_in_flight\{method="POST",path="\/api\/files\/#id"\} 0/,
+ );
+ expect(response.text).toMatch(
+ /http_request_body_bytes_count\{method="POST",path="\/api\/files\/#id"\} 1/,
+ );
+ expect(response.text).toMatch(
+ /http_request_body_bytes_sum\{method="POST",path="\/api\/files\/#id"\} 42/,
+ );
+ });
+
+ it('tracks SSE stream counts, active gauges, and stream duration', async () => {
+ const app = express();
+ process.env.METRICS_SECRET = 'test-secret';
+ const { metricsMiddleware, metricsRouter } = createMetrics();
+ app.use(metricsMiddleware);
+ app.get('/api/agents/chat/stream/:streamId', (_req, res) => {
+ res.setHeader('Content-Type', 'text/event-stream');
+ res.write('event: message\ndata: {}\n\n');
+ res.end();
+ });
+ app.use('/metrics', metricsRouter);
+
+ await request(app).get('/api/agents/chat/stream/stream-123').expect(200);
+
+ const response = await request(app)
+ .get('/metrics')
+ .set('Authorization', 'Bearer test-secret')
+ .expect(200);
+
+ expect(response.text).toMatch(
+ /sse_streams_total\{method="GET",path="\/api\/agents\/chat\/stream\/#id",status="200"\} 1/,
+ );
+ expect(response.text).toMatch(
+ /sse_streams_in_flight\{method="GET",path="\/api\/agents\/chat\/stream\/#id"\} 0/,
+ );
+ expect(response.text).toMatch(
+ /sse_stream_duration_seconds_count\{method="GET",path="\/api\/agents\/chat\/stream\/#id",status="200"\} 1/,
+ );
+ });
+
+ it('tracks upload counts, active gauges, duration, and upload bytes', async () => {
+ const app = express();
+ process.env.METRICS_SECRET = 'test-secret';
+ const { metricsMiddleware, metricsRouter } = createMetrics();
+ app.use(metricsMiddleware);
+ app.post('/api/files', (_req, res) => {
+ res.status(201).send('ok');
+ });
+ app.use('/metrics', metricsRouter);
+
+ await request(app)
+ .post('/api/files')
+ .attach('file', Buffer.from('uploaded body'), 'test.txt')
+ .expect(201);
+
+ const response = await request(app)
+ .get('/metrics')
+ .set('Authorization', 'Bearer test-secret')
+ .expect(200);
+
+ expect(response.text).toMatch(
+ /upload_requests_total\{method="POST",path="\/api\/files",status="201"\} 1/,
+ );
+ expect(response.text).toMatch(
+ /upload_requests_in_flight\{method="POST",path="\/api\/files"\} 0/,
+ );
+ expect(response.text).toMatch(
+ /upload_request_duration_seconds_count\{method="POST",path="\/api\/files",status="201"\} 1/,
+ );
+ expect(response.text).toMatch(/upload_bytes_total\{method="POST",path="\/api\/files"\} \d+/);
+ });
+
+ it('does not track non-upload methods on upload paths as uploads', async () => {
+ const app = express();
+ process.env.METRICS_SECRET = 'test-secret';
+ const { metricsMiddleware, metricsRouter } = createMetrics();
+ app.use(metricsMiddleware);
+ app.delete('/api/files', (_req, res) => {
+ res.status(204).end();
+ });
+ app.use('/metrics', metricsRouter);
+
+ await request(app).delete('/api/files').expect(204);
+
+ const response = await request(app)
+ .get('/metrics')
+ .set('Authorization', 'Bearer test-secret')
+ .expect(200);
+
+ expect(response.text).not.toMatch(/upload_requests_total\{method="DELETE"/);
+ expect(response.text).not.toMatch(/upload_bytes_total\{method="DELETE"/);
+ });
+
+ it('labels requests closed before finish as client-aborted', async () => {
+ const app = express();
+ process.env.METRICS_SECRET = 'test-secret';
+ const { metricsMiddleware, metricsRouter } = createMetrics();
+ const headers = new Map();
+ const res = Object.assign(new EventEmitter(), {
+ destroyed: false,
+ getHeader: (name: string) => headers.get(name.toLowerCase()),
+ setHeader(name: string, value: unknown) {
+ headers.set(name.toLowerCase(), value);
+ return this;
+ },
+ statusCode: 200,
+ writableEnded: false,
+ writeHead() {
+ return this;
+ },
+ });
+ const next = jest.fn();
+
+ metricsMiddleware(
+ { headers: {}, method: 'GET', path: '/api/slow-response' } as Request,
+ res as unknown as Response,
+ next,
+ );
+ res.emit('close');
+ app.use('/metrics', metricsRouter);
+
+ const response = await request(app)
+ .get('/metrics')
+ .set('Authorization', 'Bearer test-secret')
+ .expect(200);
+
+ expect(next).toHaveBeenCalledTimes(1);
+ expect(response.text).toMatch(
+ /http_requests_total\{method="GET",path="\/api\/#path",status="499"\} 1/,
+ );
+ expect(response.text).not.toMatch(
+ /http_requests_total\{method="GET",path="\/api\/#path",status="200"\}/,
+ );
+ });
+
+ it('tracks OpenID user lookup outcomes and latency', async () => {
+ const app = express();
+ process.env.METRICS_SECRET = 'test-secret';
+ const { metricsRouter } = createMetrics();
+ app.use('/metrics', metricsRouter);
+
+ recordOpenIDUserLookup('found', 0.2);
+
+ const response = await request(app)
+ .get('/metrics')
+ .set('Authorization', 'Bearer test-secret')
+ .expect(200);
+
+ expect(response.text).toMatch(/openid_user_lookup_total\{result="found"\} 1/);
+ expect(response.text).toMatch(/openid_user_lookup_duration_seconds_count\{result="found"\} 1/);
+ expect(response.text).toMatch(/openid_user_lookup_duration_seconds_sum\{result="found"\} 0.2/);
+ });
+
+ it('tracks mongoose query counts and latency by model and operation', async () => {
+ class FakeQuery {
+ model = { modelName: 'User' };
+ op = 'findOne';
+
+ exec() {
+ return Promise.resolve({ _id: 'user-1' });
+ }
+ }
+
+ const app = express();
+ process.env.METRICS_SECRET = 'test-secret';
+ const { metricsRouter } = createMetrics();
+ app.use('/metrics', metricsRouter);
+
+ instrumentMongooseQueryMetrics({ Query: FakeQuery } as never);
+ await new FakeQuery().exec();
+
+ const response = await request(app)
+ .get('/metrics')
+ .set('Authorization', 'Bearer test-secret')
+ .expect(200);
+
+ expect(response.text).toMatch(
+ /mongoose_queries_total\{model="User",operation="findOne",status="success"\} 1/,
+ );
+ expect(response.text).toMatch(
+ /mongoose_query_duration_seconds_count\{model="User",operation="findOne",status="success"\} 1/,
+ );
+ });
+
+ it('does not instrument mongoose queries when metrics are not configured', async () => {
+ class FakeQuery {
+ model = { modelName: 'User' };
+ op = 'findOne';
+
+ exec() {
+ return Promise.resolve({ _id: 'user-1' });
+ }
+ }
+
+ const originalExec = FakeQuery.prototype.exec;
+
+ instrumentMongooseQueryMetrics({ Query: FakeQuery } as never);
+
+ expect(FakeQuery.prototype.exec).toBe(originalExec);
+ });
+
+ it('tracks mongoose query errors thrown before a promise is returned', async () => {
+ class ThrowingQuery {
+ model = { modelName: 'User' };
+ op = 'findOne';
+
+ exec() {
+ throw new Error('sync database error');
+ }
+ }
+
+ const app = express();
+ process.env.METRICS_SECRET = 'test-secret';
+ const { metricsRouter } = createMetrics();
+ app.use('/metrics', metricsRouter);
+
+ instrumentMongooseQueryMetrics({ Query: ThrowingQuery } as never);
+ expect(() => new ThrowingQuery().exec()).toThrow('sync database error');
+
+ const response = await request(app)
+ .get('/metrics')
+ .set('Authorization', 'Bearer test-secret')
+ .expect(200);
+
+ expect(response.text).toMatch(
+ /mongoose_queries_total\{model="User",operation="findOne",status="error"\} 1/,
+ );
+ expect(response.text).toMatch(
+ /mongoose_query_duration_seconds_count\{model="User",operation="findOne",status="error"\} 1/,
+ );
+ });
+
+ it('tracks generation job and stream resume metrics', async () => {
+ const app = express();
+ process.env.METRICS_SECRET = 'test-secret';
+ const { metricsRouter } = createMetrics();
+ app.use('/metrics', metricsRouter);
+
+ recordGenerationJob('memory', 'created');
+ setGenerationJobsInFlight('memory', 2);
+ recordGenerationStreamSubscription('redis', 'resume', 'not_found');
+ recordGenerationStreamSubscription('redis', 'resume_state', 'missing');
+ recordGenerationStreamResumePendingEvents('memory', 3);
+
+ const response = await request(app)
+ .get('/metrics')
+ .set('Authorization', 'Bearer test-secret')
+ .expect(200);
+
+ expect(response.text).toMatch(/generation_jobs_total\{store="memory",result="created"\} 1/);
+ expect(response.text).toMatch(/generation_jobs_in_flight\{store="memory"\} 2/);
+ expect(response.text).toMatch(
+ /generation_stream_subscriptions_total\{store="redis",type="resume",result="not_found"\} 1/,
+ );
+ expect(response.text).toMatch(
+ /generation_stream_subscriptions_total\{store="redis",type="resume_state",result="missing"\} 1/,
+ );
+ expect(response.text).toMatch(
+ /generation_stream_resume_pending_events_total\{store="memory"\} 3/,
+ );
+ });
+});
diff --git a/packages/api/src/app/metrics.ts b/packages/api/src/app/metrics.ts
index 8089a9c3a1..4dd9fddac2 100644
--- a/packages/api/src/app/metrics.ts
+++ b/packages/api/src/app/metrics.ts
@@ -1,10 +1,18 @@
import { timingSafeEqual } from 'crypto';
import { Router } from 'express';
-import { Registry, collectDefaultMetrics, Counter, Histogram } from 'prom-client';
+import { Registry, collectDefaultMetrics, Counter, Gauge, Histogram } from 'prom-client';
import { logger } from '@librechat/data-schemas';
import type { Request, Response, NextFunction, RequestHandler } from 'express';
+import type { Mongoose } from 'mongoose';
const PATH_NORMALIZATIONS: [RegExp, string][] = [
+ [/^\/api\/agents\/chat\/stream\/[^/]+(?=\/|$)/, '/api/agents/chat/stream/#id'],
+ [/^\/api\/agents\/chat\/status\/[^/]+(?=\/|$)/, '/api/agents/chat/status/#id'],
+ [/^\/api\/files\/code\/download\/[^/]+\/[^/]+(?=\/|$)/, '/api/files/code/download/#id/#id'],
+ [/^\/api\/files\/download-url\/[^/]+\/[^/]+(?=\/|$)/, '/api/files/download-url/#id/#id'],
+ [/^\/api\/files\/download\/[^/]+\/[^/]+(?=\/|$)/, '/api/files/download/#id/#id'],
+ [/^\/api\/files\/[^/]+\/preview(?=\/|$)/, '/api/files/#id/preview'],
+ [/^\/api\/skills\/[^/]+\/files(?:\/.*)?(?=\/|$)/, '/api/skills/#id/files'],
[/^\/api\/messages\/artifact\/[^/]+(?=\/|$)/, '/api/messages/artifact/#id'],
[/^\/api\/messages\/[^/]+\/[^/]+(?=\/|$)/, '/api/messages/#id/#id'],
[/^\/api\/convos\/[^/]+\/messages\/[^/]+(?=\/|$)/, '/api/convos/#id/messages/#id'],
@@ -22,9 +30,39 @@ const PATH_NORMALIZATIONS: [RegExp, string][] = [
],
];
-const STATIC_PATHS = new Set(['/', '/health', '/metrics', '/api/auth/login', '/api/config']);
+const STATIC_PATHS = new Set([
+ '/',
+ '/health',
+ '/metrics',
+ '/api/auth/login',
+ '/api/config',
+ '/api/agents/chat/abort',
+ '/api/agents/chat/active',
+ '/api/agents/v1/chat/completions',
+ '/api/agents/v1/responses',
+ '/api/files',
+ '/api/files/config',
+ '/api/files/images',
+ '/api/files/images/avatar',
+ '/api/files/speech/stt',
+]);
+
+const UPLOAD_PATHS = new Set([
+ '/api/files',
+ '/api/files/images',
+ '/api/files/images/avatar',
+ '/api/files/speech/stt',
+ '/api/skills/#id/files',
+]);
+
+const UPLOAD_METHODS = new Set(['POST', 'PUT', 'PATCH']);
const LOW_CARDINALITY_PATHS: RegExp[] = [
+ /^\/api\/agents\/chat\/stream\/#id$/,
+ /^\/api\/agents\/chat\/status\/#id$/,
+ /^\/api\/files\/#id\/preview$/,
+ /^\/api\/files\/(code\/download|download-url|download)\/#id\/#id$/,
+ /^\/api\/skills\/#id\/files$/,
/^\/api\/messages\/#id$/,
/^\/api\/messages\/#id\/#id$/,
/^\/api\/messages\/artifact\/#id$/,
@@ -89,7 +127,215 @@ export interface PrometheusMetrics {
metricsRouter: Router;
}
+export type OpenIDUserLookupResult = 'found' | 'not_found' | 'migration' | 'auth_failed' | 'error';
+export type GenerationJobStore = 'memory' | 'redis';
+export type GenerationJobResult = 'created' | 'completed' | 'error' | 'aborted' | 'abort_failed';
+export type GenerationStreamSubscriptionType = 'initial' | 'resume' | 'resume_state';
+export type GenerationStreamSubscriptionResult =
+ | 'success'
+ | 'not_found'
+ | 'error'
+ | 'found'
+ | 'missing';
+
+type OpenIDUserLookupMetrics = {
+ recordLookup: (result: OpenIDUserLookupResult, durationSeconds: number) => void;
+};
+
+let openIDUserLookupMetrics: OpenIDUserLookupMetrics = {
+ recordLookup: () => undefined,
+};
+
+export function recordOpenIDUserLookup(
+ result: OpenIDUserLookupResult,
+ durationSeconds: number,
+): void {
+ openIDUserLookupMetrics.recordLookup(result, durationSeconds);
+}
+
+type MongooseQueryMetrics = {
+ recordQuery: (model: string, operation: string, status: string, durationSeconds: number) => void;
+};
+
+let mongooseQueryMetrics: MongooseQueryMetrics = {
+ recordQuery: () => undefined,
+};
+
+type GenerationJobMetrics = {
+ recordJob: (store: GenerationJobStore, result: GenerationJobResult) => void;
+ setJobsInFlight: (store: GenerationJobStore, count: number) => void;
+ recordSubscription: (
+ store: GenerationJobStore,
+ type: GenerationStreamSubscriptionType,
+ result: GenerationStreamSubscriptionResult,
+ ) => void;
+ recordResumePendingEvents: (store: GenerationJobStore, count: number) => void;
+};
+
+let generationJobMetrics: GenerationJobMetrics = {
+ recordJob: () => undefined,
+ setJobsInFlight: () => undefined,
+ recordSubscription: () => undefined,
+ recordResumePendingEvents: () => undefined,
+};
+
+const resetMetricRecorders = (): void => {
+ openIDUserLookupMetrics = {
+ recordLookup: () => undefined,
+ };
+ mongooseQueryMetrics = {
+ recordQuery: () => undefined,
+ };
+ generationJobMetrics = {
+ recordJob: () => undefined,
+ setJobsInFlight: () => undefined,
+ recordSubscription: () => undefined,
+ recordResumePendingEvents: () => undefined,
+ };
+};
+
+export function recordGenerationJob(store: GenerationJobStore, result: GenerationJobResult): void {
+ generationJobMetrics.recordJob(store, result);
+}
+
+export function setGenerationJobsInFlight(store: GenerationJobStore, count: number): void {
+ generationJobMetrics.setJobsInFlight(store, count);
+}
+
+export function recordGenerationStreamSubscription(
+ store: GenerationJobStore,
+ type: GenerationStreamSubscriptionType,
+ result: GenerationStreamSubscriptionResult,
+): void {
+ generationJobMetrics.recordSubscription(store, type, result);
+}
+
+export function recordGenerationStreamResumePendingEvents(
+ store: GenerationJobStore,
+ count: number,
+): void {
+ generationJobMetrics.recordResumePendingEvents(store, count);
+}
+
+const getElapsedSeconds = (startedAt: bigint): number =>
+ Number(process.hrtime.bigint() - startedAt) / 1_000_000_000;
+
+export const isMetricsConfigured = (): boolean => Boolean(process.env.METRICS_SECRET);
+
+const createUnauthorizedMetricsRouter = (): Router => {
+ const metricsRouter = Router();
+ metricsRouter.get('/', (_req, res) => {
+ res.status(401).end();
+ });
+ return metricsRouter;
+};
+
+const normalizeMongooseLabel = (value: unknown): string => {
+ if (typeof value !== 'string' || !value) return 'unknown';
+ return value.replace(/[^a-zA-Z0-9_:-]/g, '_').slice(0, 64) || 'unknown';
+};
+
+const getHeader = (headers: Record, name: string): unknown => {
+ const lowerName = name.toLowerCase();
+ for (const [key, value] of Object.entries(headers)) {
+ if (key.toLowerCase() === lowerName) return value;
+ }
+ return undefined;
+};
+
+const headerIncludes = (value: unknown, expected: string): boolean => {
+ if (Array.isArray(value)) {
+ return value.some((entry) => headerIncludes(entry, expected));
+ }
+ return typeof value === 'string' && value.toLowerCase().includes(expected);
+};
+
+const isEventStreamContentType = (value: unknown): boolean =>
+ headerIncludes(value, 'text/event-stream');
+
+const isMultipartContentType = (value: unknown): boolean =>
+ headerIncludes(value, 'multipart/form-data');
+
+const getRequestContentLength = (req: Request): number | null => {
+ const contentLength = req.headers['content-length'];
+ const rawContentLength = Array.isArray(contentLength) ? contentLength[0] : contentLength;
+ const bodyBytes = rawContentLength == null ? NaN : Number(rawContentLength);
+ return Number.isFinite(bodyBytes) && bodyBytes >= 0 ? bodyBytes : null;
+};
+
+const isUploadRequest = (req: Request, normalizedPath: string): boolean => {
+ if (!UPLOAD_METHODS.has(req.method)) return false;
+ if (isMultipartContentType(req.headers['content-type'])) return true;
+ return UPLOAD_PATHS.has(normalizedPath);
+};
+
+export function recordMongooseQuery(
+ model: string,
+ operation: string,
+ status: string,
+ durationSeconds: number,
+): void {
+ mongooseQueryMetrics.recordQuery(
+ normalizeMongooseLabel(model),
+ normalizeMongooseLabel(operation),
+ normalizeMongooseLabel(status),
+ durationSeconds,
+ );
+}
+
+export function instrumentMongooseQueryMetrics(mongoose: Mongoose): void {
+ if (!isMetricsConfigured()) return;
+
+ const instrumented = Symbol.for('librechat.mongooseQueryMetrics.instrumented');
+ const queryPrototype = mongoose.Query?.prototype as
+ | (typeof mongoose.Query.prototype & { [instrumented]?: boolean })
+ | undefined;
+
+ if (!queryPrototype || queryPrototype[instrumented]) return;
+
+ const originalExec = queryPrototype.exec;
+ queryPrototype.exec = function instrumentedExec(
+ this: typeof queryPrototype & {
+ model?: { modelName?: string };
+ op?: string;
+ },
+ ...args: Parameters
+ ) {
+ const startedAt = process.hrtime.bigint();
+ const model = normalizeMongooseLabel(this.model?.modelName);
+ const operation = normalizeMongooseLabel(this.op);
+
+ let result: ReturnType;
+ try {
+ result = originalExec.apply(this, args);
+ } catch (error) {
+ recordMongooseQuery(model, operation, 'error', getElapsedSeconds(startedAt));
+ throw error;
+ }
+
+ return Promise.resolve(result).then(
+ (result) => {
+ recordMongooseQuery(model, operation, 'success', getElapsedSeconds(startedAt));
+ return result;
+ },
+ (error) => {
+ recordMongooseQuery(model, operation, 'error', getElapsedSeconds(startedAt));
+ throw error;
+ },
+ );
+ } as typeof originalExec;
+ queryPrototype[instrumented] = true;
+}
+
export function createMetrics(): PrometheusMetrics {
+ if (!isMetricsConfigured()) {
+ resetMetricRecorders();
+ return {
+ metricsMiddleware: (_req: Request, _res: Response, next: NextFunction) => next(),
+ metricsRouter: createUnauthorizedMetricsRouter(),
+ };
+ }
+
const registry = new Registry();
collectDefaultMetrics({ register: registry });
@@ -108,13 +354,238 @@ export function createMetrics(): PrometheusMetrics {
registers: [registry],
});
+ const httpRequestsInFlight = new Gauge({
+ name: 'http_requests_in_flight',
+ help: 'HTTP requests currently being handled',
+ labelNames: ['method', 'path'] as const,
+ registers: [registry],
+ });
+
+ const httpRequestBodyBytes = new Histogram({
+ name: 'http_request_body_bytes',
+ help: 'HTTP request body size in bytes from the Content-Length header',
+ labelNames: ['method', 'path'] as const,
+ buckets: [1_000, 10_000, 100_000, 1_000_000, 5_000_000, 10_000_000, 25_000_000, 50_000_000],
+ registers: [registry],
+ });
+
+ const sseStreams = new Counter({
+ name: 'sse_streams_total',
+ help: 'Total SSE streams opened',
+ labelNames: ['method', 'path', 'status'] as const,
+ registers: [registry],
+ });
+
+ const sseStreamsInFlight = new Gauge({
+ name: 'sse_streams_in_flight',
+ help: 'SSE streams currently open',
+ labelNames: ['method', 'path'] as const,
+ registers: [registry],
+ });
+
+ const sseStreamDuration = new Histogram({
+ name: 'sse_stream_duration_seconds',
+ help: 'SSE stream open duration in seconds',
+ labelNames: ['method', 'path', 'status'] as const,
+ buckets: [1, 5, 10, 30, 60, 120, 300, 600, 1_200, 1_800],
+ registers: [registry],
+ });
+
+ const uploadRequests = new Counter({
+ name: 'upload_requests_total',
+ help: 'Total upload requests',
+ labelNames: ['method', 'path', 'status'] as const,
+ registers: [registry],
+ });
+
+ const uploadRequestsInFlight = new Gauge({
+ name: 'upload_requests_in_flight',
+ help: 'Upload requests currently being handled',
+ labelNames: ['method', 'path'] as const,
+ registers: [registry],
+ });
+
+ const uploadRequestDuration = new Histogram({
+ name: 'upload_request_duration_seconds',
+ help: 'Upload request duration in seconds',
+ labelNames: ['method', 'path', 'status'] as const,
+ buckets: [0.1, 0.3, 0.5, 1, 2, 5, 10, 30, 60, 120, 300],
+ registers: [registry],
+ });
+
+ const uploadBytes = new Counter({
+ name: 'upload_bytes_total',
+ help: 'Upload request bytes from the Content-Length header',
+ labelNames: ['method', 'path'] as const,
+ registers: [registry],
+ });
+
+ const openIDUserLookupTotal = new Counter({
+ name: 'openid_user_lookup_total',
+ help: 'OpenID user lookup attempts',
+ labelNames: ['result'] as const,
+ registers: [registry],
+ });
+
+ const openIDUserLookupDuration = new Histogram({
+ name: 'openid_user_lookup_duration_seconds',
+ help: 'OpenID user lookup latency in seconds',
+ labelNames: ['result'] as const,
+ buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60],
+ registers: [registry],
+ });
+
+ openIDUserLookupMetrics = {
+ recordLookup: (result, durationSeconds) => {
+ openIDUserLookupTotal.inc({ result });
+ openIDUserLookupDuration.observe({ result }, durationSeconds);
+ },
+ };
+
+ const mongooseQueries = new Counter({
+ name: 'mongoose_queries_total',
+ help: 'Mongoose queries by model, operation, and status',
+ labelNames: ['model', 'operation', 'status'] as const,
+ registers: [registry],
+ });
+
+ const mongooseQueryDuration = new Histogram({
+ name: 'mongoose_query_duration_seconds',
+ help: 'Mongoose query duration in seconds by model, operation, and status',
+ labelNames: ['model', 'operation', 'status'] as const,
+ buckets: [0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60],
+ registers: [registry],
+ });
+
+ mongooseQueryMetrics = {
+ recordQuery: (model, operation, status, durationSeconds) => {
+ const labels = { model, operation, status };
+ mongooseQueries.inc(labels);
+ mongooseQueryDuration.observe(labels, durationSeconds);
+ },
+ };
+
+ const generationJobs = new Counter({
+ name: 'generation_jobs_total',
+ help: 'Generation jobs by backing store and result',
+ labelNames: ['store', 'result'] as const,
+ registers: [registry],
+ });
+
+ const generationJobsInFlight = new Gauge({
+ name: 'generation_jobs_in_flight',
+ help: 'Generation jobs currently running in this process',
+ labelNames: ['store'] as const,
+ registers: [registry],
+ });
+
+ const generationStreamSubscriptions = new Counter({
+ name: 'generation_stream_subscriptions_total',
+ help: 'Generation stream subscription attempts by backing store, type, and result',
+ labelNames: ['store', 'type', 'result'] as const,
+ registers: [registry],
+ });
+
+ const generationStreamResumePendingEvents = new Counter({
+ name: 'generation_stream_resume_pending_events_total',
+ help: 'Pending events delivered while resuming generation streams',
+ labelNames: ['store'] as const,
+ registers: [registry],
+ });
+
+ generationJobMetrics = {
+ recordJob: (store, result) => generationJobs.inc({ store, result }),
+ setJobsInFlight: (store, count) => generationJobsInFlight.set({ store }, count),
+ recordSubscription: (store, type, result) =>
+ generationStreamSubscriptions.inc({ store, type, result }),
+ recordResumePendingEvents: (store, count) =>
+ generationStreamResumePendingEvents.inc({ store }, count),
+ };
+
const metricsMiddleware = (req: Request, res: Response, next: NextFunction): void => {
const end = httpDuration.startTimer();
- res.on('finish', () => {
- const labels = { method: req.method, path: normalizePath(req.path), status: res.statusCode };
- httpRequests.inc(labels);
- end(labels);
- });
+ const labels = { method: req.method, path: normalizePath(req.path) };
+ const uploadTracked = isUploadRequest(req, labels.path);
+ const uploadStartedAt = uploadTracked ? process.hrtime.bigint() : null;
+ let sseTracked = false;
+ let sseStartedAt: bigint | null = null;
+ let completed = false;
+
+ const markSSEStream = () => {
+ if (completed || res.writableEnded || res.destroyed) return;
+ if (sseTracked) return;
+ sseTracked = true;
+ sseStartedAt = process.hrtime.bigint();
+ sseStreamsInFlight.inc(labels);
+ };
+
+ const originalSetHeader = res.setHeader;
+ res.setHeader = function setHeader(this: Response, ...args: Parameters) {
+ const [name, value] = args;
+ const result = originalSetHeader.apply(this, args);
+ if (String(name).toLowerCase() === 'content-type' && isEventStreamContentType(value)) {
+ markSSEStream();
+ }
+ return result;
+ } as Response['setHeader'];
+
+ const originalWriteHead = res.writeHead;
+ res.writeHead = function writeHead(this: Response, ...args: [number, unknown?, unknown?]) {
+ const [, reasonPhrase, headers] = args;
+ const responseHeaders =
+ typeof reasonPhrase === 'object' && reasonPhrase != null ? reasonPhrase : headers;
+ if (
+ isEventStreamContentType(
+ getHeader((responseHeaders ?? {}) as Record, 'content-type'),
+ ) ||
+ isEventStreamContentType(res.getHeader('content-type'))
+ ) {
+ markSSEStream();
+ }
+ return originalWriteHead.apply(this, args as Parameters);
+ } as Response['writeHead'];
+
+ httpRequestsInFlight.inc(labels);
+ if (uploadTracked) {
+ uploadRequestsInFlight.inc(labels);
+ }
+
+ const complete = (completedBy: 'finish' | 'close') => {
+ if (completed) return;
+ completed = true;
+
+ const requestLabels = { ...labels, status: completedBy === 'close' ? 499 : res.statusCode };
+ httpRequests.inc(requestLabels);
+ end(requestLabels);
+ httpRequestsInFlight.dec(labels);
+
+ const bodyBytes = getRequestContentLength(req);
+ if (bodyBytes != null) {
+ httpRequestBodyBytes.observe(labels, bodyBytes);
+ }
+
+ if (sseTracked) {
+ sseStreams.inc(requestLabels);
+ sseStreamsInFlight.dec(labels);
+ if (sseStartedAt) {
+ sseStreamDuration.observe(requestLabels, getElapsedSeconds(sseStartedAt));
+ }
+ }
+
+ if (uploadTracked) {
+ uploadRequests.inc(requestLabels);
+ uploadRequestsInFlight.dec(labels);
+ if (uploadStartedAt) {
+ uploadRequestDuration.observe(requestLabels, getElapsedSeconds(uploadStartedAt));
+ }
+ if (bodyBytes != null) {
+ uploadBytes.inc(labels, bodyBytes);
+ }
+ }
+ };
+
+ res.once('finish', () => complete('finish'));
+ res.once('close', () => complete('close'));
next();
};
diff --git a/packages/api/src/auth/openid.spec.ts b/packages/api/src/auth/openid.spec.ts
index 975e736b72..543fa94fe3 100644
--- a/packages/api/src/auth/openid.spec.ts
+++ b/packages/api/src/auth/openid.spec.ts
@@ -5,6 +5,7 @@ import { ErrorTypes } from 'librechat-data-provider';
import type { IUser, UserMethods } from '@librechat/data-schemas';
import type { CommandStartedEvent } from 'mongodb';
import type { FilterQuery } from 'mongoose';
+import { recordOpenIDUserLookup } from '~/app/metrics';
import { findOpenIDUser, getOpenIdEmail, getOpenIdIssuer, normalizeOpenIdIssuer } from './openid';
function newId() {
@@ -19,6 +20,11 @@ jest.mock('@librechat/data-schemas', () => ({
},
}));
+jest.mock('~/app/metrics', () => ({
+ isMetricsConfigured: jest.fn(() => true),
+ recordOpenIDUserLookup: jest.fn(),
+}));
+
describe('normalizeOpenIdIssuer', () => {
it('normalizes blank, trailing-slash, and discovery-document issuers', () => {
expect(normalizeOpenIdIssuer('')).toBeUndefined();
@@ -111,6 +117,7 @@ describe('findOpenIDUser', () => {
error: null,
migration: false,
});
+ expect(recordOpenIDUserLookup).toHaveBeenCalledWith('found', expect.any(Number));
});
it('should find user by idOnTheSource', async () => {
@@ -734,6 +741,7 @@ describe('findOpenIDUser', () => {
findUser: mockFindUser,
}),
).rejects.toThrow('Database error');
+ expect(recordOpenIDUserLookup).toHaveBeenCalledWith('error', expect.any(Number));
});
it('should reject email fallback when openidId is empty and user has a stored openidId', async () => {
diff --git a/packages/api/src/auth/openid.ts b/packages/api/src/auth/openid.ts
index 8ee5cbd517..4d346a7591 100644
--- a/packages/api/src/auth/openid.ts
+++ b/packages/api/src/auth/openid.ts
@@ -2,6 +2,8 @@ import { logger } from '@librechat/data-schemas';
import { ErrorTypes } from 'librechat-data-provider';
import type { IUser, UserMethods } from '@librechat/data-schemas';
import type { FilterQuery } from 'mongoose';
+import { isMetricsConfigured, recordOpenIDUserLookup } from '~/app/metrics';
+import type { OpenIDUserLookupResult } from '~/app/metrics';
export type OpenIdEmailClaims = {
email?: unknown;
@@ -65,6 +67,17 @@ function hasOpenIdLookupValue(value: string | undefined): value is string {
return typeof value === 'string' && value.length > 0;
}
+function getElapsedSeconds(startedAt: bigint): number {
+ return Number(process.hrtime.bigint() - startedAt) / 1_000_000_000;
+}
+
+function getOpenIDUserLookupResult(resolution: OpenIdUserResolution): OpenIDUserLookupResult {
+ if (resolution.error) return 'auth_failed';
+ if (resolution.migration) return 'migration';
+ if (resolution.user) return 'found';
+ return 'not_found';
+}
+
function getIssuerExactCondition(
field: OpenIdLookupField,
value: string | undefined,
@@ -208,61 +221,79 @@ export async function findOpenIDUser({
idOnTheSource?: string;
strategyName?: string;
}): Promise {
- const normalizedIssuer = normalizeOpenIdIssuer(openidIssuer);
- const primaryConditions = getPrimaryLookupConditions(openidId, idOnTheSource, normalizedIssuer);
-
- let user: IUser | null = null;
- if (primaryConditions.length > 0) {
- user = await findFirstOpenIdUser(findUser, primaryConditions);
- }
-
- const primaryIssuerResolution = resolveIssuerBoundUser(
- user,
- normalizedIssuer,
- strategyName,
- 'OpenID lookup',
- );
- if (primaryIssuerResolution) return primaryIssuerResolution;
-
- if (!user && email) {
- user = await findUser({ email });
- logger.warn(
- `[${strategyName}] user ${user ? 'found' : 'not found'} with email: ${email} for openidId: ${openidId}`,
- );
-
- // If user found by email, check if they're allowed to use OpenID provider
- if (user && user.provider && user.provider !== 'openid') {
- logger.warn(
- `[${strategyName}] Attempted OpenID login by user ${user.email}, was registered with "${user.provider}" provider`,
+ const lookupStartedAt = isMetricsConfigured() ? process.hrtime.bigint() : null;
+ const finish = (resolution: OpenIdUserResolution): OpenIdUserResolution => {
+ if (lookupStartedAt != null) {
+ recordOpenIDUserLookup(
+ getOpenIDUserLookupResult(resolution),
+ getElapsedSeconds(lookupStartedAt),
);
- return { user: null, error: ErrorTypes.AUTH_FAILED, migration: false };
+ }
+ return resolution;
+ };
+
+ try {
+ const normalizedIssuer = normalizeOpenIdIssuer(openidIssuer);
+ const primaryConditions = getPrimaryLookupConditions(openidId, idOnTheSource, normalizedIssuer);
+
+ let user: IUser | null = null;
+ if (primaryConditions.length > 0) {
+ user = await findFirstOpenIdUser(findUser, primaryConditions);
}
- if (user?.openidId && user.openidId !== openidId) {
- logger.warn(
- `[${strategyName}] Rejected email fallback for ${user.email}: stored openidId does not match token sub`,
- );
- return { user: null, error: ErrorTypes.AUTH_FAILED, migration: false };
- }
-
- const emailIssuerResolution = resolveIssuerBoundUser(
+ const primaryIssuerResolution = resolveIssuerBoundUser(
user,
normalizedIssuer,
strategyName,
- 'email fallback',
+ 'OpenID lookup',
);
- if (emailIssuerResolution) return emailIssuerResolution;
+ if (primaryIssuerResolution) return finish(primaryIssuerResolution);
- if (user && !user.openidId) {
- logger.info(
- `[${strategyName}] Preparing user ${user.email} for migration to OpenID with sub: ${openidId}`,
+ if (!user && email) {
+ user = await findUser({ email });
+ logger.warn(
+ `[${strategyName}] user ${user ? 'found' : 'not found'} with email: ${email} for openidId: ${openidId}`,
);
- user.provider = 'openid';
- user.openidId = openidId;
- if (normalizedIssuer) user.openidIssuer = normalizedIssuer;
- return { user, error: null, migration: true };
- }
- }
- return { user, error: null, migration: false };
+ // If user found by email, check if they're allowed to use OpenID provider
+ if (user && user.provider && user.provider !== 'openid') {
+ logger.warn(
+ `[${strategyName}] Attempted OpenID login by user ${user.email}, was registered with "${user.provider}" provider`,
+ );
+ return finish({ user: null, error: ErrorTypes.AUTH_FAILED, migration: false });
+ }
+
+ if (user?.openidId && user.openidId !== openidId) {
+ logger.warn(
+ `[${strategyName}] Rejected email fallback for ${user.email}: stored openidId does not match token sub`,
+ );
+ return finish({ user: null, error: ErrorTypes.AUTH_FAILED, migration: false });
+ }
+
+ const emailIssuerResolution = resolveIssuerBoundUser(
+ user,
+ normalizedIssuer,
+ strategyName,
+ 'email fallback',
+ );
+ if (emailIssuerResolution) return finish(emailIssuerResolution);
+
+ if (user && !user.openidId) {
+ logger.info(
+ `[${strategyName}] Preparing user ${user.email} for migration to OpenID with sub: ${openidId}`,
+ );
+ user.provider = 'openid';
+ user.openidId = openidId;
+ if (normalizedIssuer) user.openidIssuer = normalizedIssuer;
+ return finish({ user, error: null, migration: true });
+ }
+ }
+
+ return finish({ user, error: null, migration: false });
+ } catch (error) {
+ if (lookupStartedAt != null) {
+ recordOpenIDUserLookup('error', getElapsedSeconds(lookupStartedAt));
+ }
+ throw error;
+ }
}
diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts
index de6232e85b..366445bb26 100644
--- a/packages/api/src/stream/GenerationJobManager.ts
+++ b/packages/api/src/stream/GenerationJobManager.ts
@@ -10,6 +10,13 @@ import type {
IJobStore,
} from './interfaces/IJobStore';
import type * as t from '~/types';
+import {
+ recordGenerationJob,
+ recordGenerationStreamResumePendingEvents,
+ recordGenerationStreamSubscription,
+ setGenerationJobsInFlight,
+} from '~/app/metrics';
+import type { GenerationJobStore } from '~/app/metrics';
import { InMemoryEventTransport } from './implementations/InMemoryEventTransport';
import { InMemoryJobStore } from './implementations/InMemoryJobStore';
@@ -87,6 +94,9 @@ class GenerationJobManagerClass {
/** Runtime state - always in-memory, not serializable */
private runtimeState = new Map();
+ /** Jobs actively generating in this process. */
+ private runningJobs = new Set();
+
private cleanupInterval: NodeJS.Timeout | null = null;
/** Whether we're using Redis stores */
@@ -144,6 +154,7 @@ class GenerationJobManagerClass {
isRedis?: boolean;
cleanupOnComplete?: boolean;
}): void {
+ const previousStore = this.storeLabel;
if (this.cleanupInterval) {
logger.warn(
'[GenerationJobManager] Reconfiguring after initialization - destroying existing services',
@@ -151,10 +162,14 @@ class GenerationJobManagerClass {
this.destroy();
}
+ this.runningJobs.clear();
+ setGenerationJobsInFlight(previousStore, 0);
+
this.jobStore = services.jobStore;
this.eventTransport = services.eventTransport;
this._isRedis = services.isRedis ?? false;
this._cleanupOnComplete = services.cleanupOnComplete ?? true;
+ this.syncRunningJobMetrics();
logger.info(
`[GenerationJobManager] Configured with ${this._isRedis ? 'Redis' : 'in-memory'} stores`,
@@ -168,6 +183,14 @@ class GenerationJobManagerClass {
return this._isRedis;
}
+ private get storeLabel(): GenerationJobStore {
+ return this._isRedis ? 'redis' : 'memory';
+ }
+
+ private syncRunningJobMetrics(store: GenerationJobStore = this.storeLabel): void {
+ setGenerationJobsInFlight(store, this.runningJobs.size);
+ }
+
/**
* Get the job store instance (for advanced use cases).
*/
@@ -226,6 +249,9 @@ class GenerationJobManagerClass {
hasSubscriber: false,
};
this.runtimeState.set(streamId, runtime);
+ this.runningJobs.add(streamId);
+ this.syncRunningJobMetrics();
+ recordGenerationJob(this.storeLabel, 'created');
// Resolve immediately - early event buffer handles late subscribers
resolveReady!();
@@ -554,6 +580,9 @@ class GenerationJobManagerClass {
completedAt: Date.now(),
error,
});
+ this.runningJobs.delete(streamId);
+ this.syncRunningJobMetrics();
+ recordGenerationJob(this.storeLabel, 'error');
// Keep runtime state so subscribe() can access errorEvent
logger.debug(
`[GenerationJobManager] Job completed with error (keeping for late subscribers): ${streamId}`,
@@ -575,6 +604,9 @@ class GenerationJobManagerClass {
});
}
+ this.runningJobs.delete(streamId);
+ this.syncRunningJobMetrics();
+ recordGenerationJob(this.storeLabel, 'completed');
logger.debug(`[GenerationJobManager] Job completed: ${streamId}`);
}
@@ -592,6 +624,7 @@ class GenerationJobManagerClass {
if (!jobData) {
logger.warn(`[GenerationJobManager] Cannot abort - job not found: ${streamId}`);
+ recordGenerationJob(this.storeLabel, 'abort_failed');
return {
text: '',
content: [],
@@ -682,6 +715,9 @@ class GenerationJobManagerClass {
});
}
+ this.runningJobs.delete(streamId);
+ this.syncRunningJobMetrics();
+ recordGenerationJob(this.storeLabel, 'aborted');
logger.debug(`[GenerationJobManager] Job aborted: ${streamId}`);
return {
@@ -723,9 +759,11 @@ class GenerationJobManagerClass {
onError?: t.ErrorHandler,
options?: t.SubscribeOptions,
): Promise<{ unsubscribe: t.UnsubscribeFn } | null> {
+ const subscriptionType = options?.skipBufferReplay ? 'resume' : 'initial';
// Use lazy initialization to support cross-replica subscriptions
const runtime = await this.getOrCreateRuntimeState(streamId);
if (!runtime) {
+ recordGenerationStreamSubscription(this.storeLabel, subscriptionType, 'not_found');
return null;
}
@@ -761,8 +799,14 @@ class GenerationJobManagerClass {
onError,
});
- if (subscription.ready) {
- await subscription.ready;
+ try {
+ if (subscription.ready) {
+ await subscription.ready;
+ }
+ recordGenerationStreamSubscription(this.storeLabel, subscriptionType, 'success');
+ } catch (err) {
+ recordGenerationStreamSubscription(this.storeLabel, subscriptionType, 'error');
+ throw err;
}
const isFirst = this.eventTransport.isFirstSubscriber(streamId);
@@ -864,6 +908,11 @@ class GenerationJobManagerClass {
: 0;
const resumeState = await this.getResumeState(streamId);
+ recordGenerationStreamSubscription(
+ this.storeLabel,
+ 'resume_state',
+ resumeState ? 'found' : 'missing',
+ );
let pendingEvents: t.ServerSentEvent[] = [];
if (!this._isRedis) {
@@ -872,6 +921,7 @@ class GenerationJobManagerClass {
pendingEvents = runtime.earlyEventBuffer.slice(bufferLengthAtSnapshot);
runtime.earlyEventBuffer = [];
if (pendingEvents.length > 0) {
+ recordGenerationStreamResumePendingEvents(this.storeLabel, pendingEvents.length);
logger.debug(
`[GenerationJobManager] Captured ${pendingEvents.length} gap events for ${streamId}`,
);
@@ -1179,11 +1229,13 @@ class GenerationJobManagerClass {
*/
private async cleanup(): Promise {
const count = await this.jobStore.cleanup();
+ let runningJobsChanged = false;
// Cleanup runtime state for deleted jobs
for (const streamId of this.runtimeState.keys()) {
if (!(await this.jobStore.hasJob(streamId))) {
this.runtimeState.delete(streamId);
+ runningJobsChanged = this.runningJobs.delete(streamId) || runningJobsChanged;
this.runStepBuffers?.delete(streamId);
this.jobStore.clearContentState(streamId);
this.eventTransport.cleanup(streamId);
@@ -1207,6 +1259,10 @@ class GenerationJobManagerClass {
}
}
+ if (runningJobsChanged) {
+ this.syncRunningJobMetrics();
+ }
+
if (count > 0) {
logger.debug(`[GenerationJobManager] Cleaned up ${count} expired jobs`);
}
@@ -1295,6 +1351,8 @@ class GenerationJobManagerClass {
await this.jobStore.destroy();
this.eventTransport.destroy();
this.runtimeState.clear();
+ this.runningJobs.clear();
+ this.syncRunningJobMetrics();
this.runStepBuffers?.clear();
logger.debug('[GenerationJobManager] Destroyed');