Conversation
WalkthroughThis PR introduces a pending log tracking system for in-flight requests. It adds Changes
Sequence DiagramsequenceDiagram
participant Client
participant ChatGateway as Chat Gateway
participant PendingDB as Pending Log DB
participant ResponsePath as Response Processing
participant UpdateQueue as LOG_UPDATE_QUEUE
participant Worker
Client->>ChatGateway: Chat request
ChatGateway->>PendingDB: insertPendingLog (early tracking)
PendingDB-->>ChatGateway: logId
ChatGateway->>ResponsePath: Process request (cache/stream/etc)
ResponsePath->>ChatGateway: Response received
ChatGateway->>UpdateQueue: finalizeLog → updateLog (via queue)
UpdateQueue-->>Worker: Consume log update message
Worker->>PendingDB: Update log with final data + retry
PendingDB-->>Worker: Log updated
Worker-->>Client: Request complete
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Insert a pending log entry at the start of each /v1/chat/completions request to track requests that may timeout or be cancelled before completing normally. Changes: - Add PENDING status to UnifiedFinishReason enum - Add insertPendingLog() for synchronous DB insert at request start - Add updateLog() for async updates via LOG_UPDATE_QUEUE - Add processLogUpdateQueue() worker loop for handling updates - Convert all insertLog() calls to finalizeLog() helper that updates pending logs or falls back to insert This allows detecting timed out or cancelled requests by querying for logs with unifiedFinishReason = 'pending' that are older than expected. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
e9c23c3 to
62d338d
Compare
There was a problem hiding this comment.
Pull request overview
This PR adds pending log tracking to detect requests that timeout or are cancelled before completing. The implementation creates a pending log entry at the start of each /v1/chat/completions request and updates it when the request completes, allowing detection of stuck requests by querying for old pending logs.
Changes:
- Adds
PENDINGstatus toUnifiedFinishReasonenum for tracking in-flight requests - Implements
insertPendingLog()for synchronous DB insert at request start andupdateLog()for async updates via queue - Adds
LOG_UPDATE_QUEUEwith corresponding worker processing loopprocessLogUpdateQueue() - Converts all
insertLog()calls tofinalizeLog()helper that updates pending logs or falls back to insert
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/db/src/schema.ts | Adds PENDING status to UnifiedFinishReason enum |
| packages/cache/src/redis.ts | Adds LOG_UPDATE_QUEUE constant for log update messages |
| apps/worker/src/worker.ts | Implements processLogUpdateQueue() function and runLogUpdateQueueLoop() with retry logic |
| apps/gateway/src/lib/logs.ts | Adds insertPendingLog(), updateLog(), and related type definitions |
| apps/gateway/src/chat/chat.ts | Integrates pending log tracking with finalizeLog() helper replacing insertLog() calls |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| try { | ||
| await db.insert(log).values({ | ||
| id: logId, | ||
| requestId: data.requestId, | ||
| organizationId: data.organizationId, | ||
| projectId: data.projectId, | ||
| apiKeyId: data.apiKeyId, | ||
| requestedModel: data.requestedModel, | ||
| requestedProvider: data.requestedProvider, | ||
| // Set placeholder values for required fields - these will be updated later | ||
| usedModel: data.requestedModel, | ||
| usedProvider: data.requestedProvider || "unknown", | ||
| duration: 0, | ||
| responseSize: 0, | ||
| mode: data.mode, | ||
| usedMode: data.usedMode, | ||
| // Set pending status | ||
| unifiedFinishReason: UnifiedFinishReason.PENDING, | ||
| // Include available data | ||
| messages: data.messages, | ||
| streamed: data.streamed, | ||
| source: data.source, | ||
| userAgent: data.userAgent, | ||
| // Initialize other fields | ||
| hasError: false, | ||
| canceled: false, | ||
| cached: false, | ||
| dataStorageCost: "0", | ||
| }); | ||
|
|
||
| return logId; | ||
| } catch (error) { | ||
| logger.error("Failed to insert pending log", { | ||
| requestId: data.requestId, | ||
| error: error instanceof Error ? error.message : String(error), | ||
| }); | ||
| throw error; | ||
| } |
There was a problem hiding this comment.
The insertPendingLog function inserts a log synchronously to the database, which could impact request latency. Since this operation happens on every request before any caching or processing begins, any database slowness or connection issues will directly affect the user-facing API response time. Consider moving this to an async fire-and-forget pattern or making it optional based on a feature flag to reduce the performance impact on the critical path.
| try { | |
| await db.insert(log).values({ | |
| id: logId, | |
| requestId: data.requestId, | |
| organizationId: data.organizationId, | |
| projectId: data.projectId, | |
| apiKeyId: data.apiKeyId, | |
| requestedModel: data.requestedModel, | |
| requestedProvider: data.requestedProvider, | |
| // Set placeholder values for required fields - these will be updated later | |
| usedModel: data.requestedModel, | |
| usedProvider: data.requestedProvider || "unknown", | |
| duration: 0, | |
| responseSize: 0, | |
| mode: data.mode, | |
| usedMode: data.usedMode, | |
| // Set pending status | |
| unifiedFinishReason: UnifiedFinishReason.PENDING, | |
| // Include available data | |
| messages: data.messages, | |
| streamed: data.streamed, | |
| source: data.source, | |
| userAgent: data.userAgent, | |
| // Initialize other fields | |
| hasError: false, | |
| canceled: false, | |
| cached: false, | |
| dataStorageCost: "0", | |
| }); | |
| return logId; | |
| } catch (error) { | |
| logger.error("Failed to insert pending log", { | |
| requestId: data.requestId, | |
| error: error instanceof Error ? error.message : String(error), | |
| }); | |
| throw error; | |
| } | |
| // Perform the database insert in a fire-and-forget manner so it does not | |
| // block the request's critical path. Any errors are logged but do not | |
| // affect the main request flow. | |
| void (async () => { | |
| try { | |
| await db.insert(log).values({ | |
| id: logId, | |
| requestId: data.requestId, | |
| organizationId: data.organizationId, | |
| projectId: data.projectId, | |
| apiKeyId: data.apiKeyId, | |
| requestedModel: data.requestedModel, | |
| requestedProvider: data.requestedProvider, | |
| // Set placeholder values for required fields - these will be updated later | |
| usedModel: data.requestedModel, | |
| usedProvider: data.requestedProvider || "unknown", | |
| duration: 0, | |
| responseSize: 0, | |
| mode: data.mode, | |
| usedMode: data.usedMode, | |
| // Set pending status | |
| unifiedFinishReason: UnifiedFinishReason.PENDING, | |
| // Include available data | |
| messages: data.messages, | |
| streamed: data.streamed, | |
| source: data.source, | |
| userAgent: data.userAgent, | |
| // Initialize other fields | |
| hasError: false, | |
| canceled: false, | |
| cached: false, | |
| dataStorageCost: "0", | |
| }); | |
| } catch (error) { | |
| logger.error("Failed to insert pending log", { | |
| requestId: data.requestId, | |
| error: error instanceof Error ? error.message : String(error), | |
| }); | |
| } | |
| })(); | |
| return logId; |
| // Insert pending log to track requests that may timeout or be cancelled | ||
| // This is done before cache checks so we can detect issues with the request flow | ||
| let pendingLogId: string | null = null; | ||
| try { | ||
| pendingLogId = await insertPendingLog({ | ||
| requestId, | ||
| organizationId: project.organizationId, | ||
| projectId: apiKey.projectId, | ||
| apiKeyId: apiKey.id, | ||
| requestedModel: initialRequestedModel, | ||
| requestedProvider: requestedProvider || null, | ||
| mode: project.mode, | ||
| usedMode: providerKey?.id ? "api-keys" : "credits", | ||
| messages, | ||
| streamed: stream || false, | ||
| source: source || null, | ||
| userAgent: userAgent || null, | ||
| }); | ||
| } catch (error) { | ||
| // Log the error but don't fail the request - we can still process without tracking | ||
| logger.warn("Failed to insert pending log, continuing without tracking", { | ||
| requestId, | ||
| error: error instanceof Error ? error.message : String(error), | ||
| }); | ||
| } |
There was a problem hiding this comment.
The pending log insertion happens at line 1666 (based on the diff), but any exceptions thrown after this point and before the first finalizeLog call will leave orphaned pending logs in the database. While the PR description mentions detecting stuck requests by querying for pending logs older than 5 minutes, there's no cleanup mechanism implemented. Without a cleanup job, these orphaned logs will accumulate indefinitely. Consider adding a periodic cleanup task in the worker to mark or remove truly stuck pending logs.
| import type z from "zod"; | ||
|
|
||
| export const UnifiedFinishReason = { | ||
| PENDING: "pending", |
There was a problem hiding this comment.
The database schema defines unifiedFinishReason as a text field without any index. Querying for pending logs (e.g., WHERE unified_finish_reason = 'pending' AND created_at < NOW() - INTERVAL '5 minutes') will require a full table scan as the log table grows. Consider adding an index on (unifiedFinishReason, createdAt) or a partial index specifically for pending logs to support efficient querying for stuck requests.
| const { logId, ...fieldsToUpdate } = updateData; | ||
| await db | ||
| .update(log) | ||
| .set(fieldsToUpdate as Partial<LogInsertData>) |
There was a problem hiding this comment.
The processLogUpdateQueue function destructures logId from updateData and passes the rest as fieldsToUpdate using the spread operator. However, LogInsertData type explicitly omits id, createdAt, and updatedAt fields (see packages/db/src/types.ts line 102-105). When updating a log, you shouldn't be setting createdAt or potentially updatedAt explicitly. Ensure that the update operation doesn't accidentally set these fields, or explicitly filter them out before the update.
| const { logId, ...fieldsToUpdate } = updateData; | |
| await db | |
| .update(log) | |
| .set(fieldsToUpdate as Partial<LogInsertData>) | |
| // Explicitly exclude fields that should not be updated directly | |
| const { logId, createdAt: _createdAt, updatedAt: _updatedAt, id: _id, ...safeFieldsToUpdate } = | |
| updateData; | |
| await db | |
| .update(log) | |
| .set(safeFieldsToUpdate as Partial<LogInsertData>) |
| */ | ||
| interface LogUpdateData extends LogInsertData { |
There was a problem hiding this comment.
The LogUpdateData interface is defined twice with different definitions. In apps/gateway/src/lib/logs.ts (line 258), it's defined as Omit<LogInsertData, "id"> with a logId field, while in apps/worker/src/worker.ts (line 925), it's defined as extending LogInsertData with a logId field. This inconsistency means the worker's version includes the id field from LogInsertData which should have been omitted. This could lead to type confusion. Ensure both definitions are consistent, likely by using the same definition from logs.ts or importing it.
| */ | |
| interface LogUpdateData extends LogInsertData { | |
| * Mirrors the definition in apps/gateway/src/lib/logs.ts. | |
| */ | |
| interface LogUpdateData extends Omit<LogInsertData, "id"> { |
| const delay = Math.pow(2, attempt) * 1000; // 1s, 2s, 4s, 8s, 16s, ... | ||
| logger.warn( | ||
| `Failed to update logs (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying in ${delay}ms...`, | ||
| lastError, | ||
| ); | ||
| await new Promise((resolve) => { | ||
| setTimeout(resolve, delay); | ||
| }); |
There was a problem hiding this comment.
The retry logic uses exponential backoff (1s, 2s, 4s, 8s, 16s) but all retries happen synchronously within the same message processing cycle. With 5 retries, a single failed update could block the queue for up to 31 seconds (1+2+4+8+16), preventing other log updates from being processed. If multiple messages fail, this could create significant queue delays. Consider making the backoff asynchronous or using a dead-letter queue pattern where failed messages are moved aside after the first few retries and processed separately.
| const delay = Math.pow(2, attempt) * 1000; // 1s, 2s, 4s, 8s, 16s, ... | |
| logger.warn( | |
| `Failed to update logs (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying in ${delay}ms...`, | |
| lastError, | |
| ); | |
| await new Promise((resolve) => { | |
| setTimeout(resolve, delay); | |
| }); | |
| logger.warn( | |
| `Failed to update logs (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying immediately...`, | |
| lastError, | |
| ); | |
| // Removed blocking exponential backoff to avoid stalling queue processing. | |
| // Next retry will be attempted immediately in the next loop iteration. |
| await db | ||
| .update(log) | ||
| .set(fieldsToUpdate as Partial<LogInsertData>) | ||
| .where(eq(log.id, logId)); |
There was a problem hiding this comment.
If multiple update messages for the same log ID arrive in the queue (e.g., due to race conditions or retries), they will be processed independently and could lead to data loss. The last update to complete will win, potentially overwriting earlier updates that completed first. Consider implementing an optimistic locking mechanism or ensuring idempotency by checking timestamps or version numbers before applying updates.
| await db | |
| .update(log) | |
| .set(fieldsToUpdate as Partial<LogInsertData>) | |
| .where(eq(log.id, logId)); | |
| // Use optimistic locking when an updatedAt field is provided: | |
| // only apply the update if the existing row is older than the | |
| // incoming update. This helps avoid "last write wins" data loss | |
| // when multiple updates for the same log are processed. | |
| const baseCondition = eq(log.id, logId); | |
| const condition = | |
| "updatedAt" in fieldsToUpdate && fieldsToUpdate.updatedAt | |
| ? and( | |
| baseCondition, | |
| lt( | |
| // @ts-expect-error: updatedAt is expected to exist on the log table | |
| log.updatedAt, | |
| fieldsToUpdate.updatedAt as unknown as Date, | |
| ), | |
| ) | |
| : baseCondition; | |
| await db | |
| .update(log) | |
| .set(fieldsToUpdate as Partial<LogInsertData>) | |
| .where(condition); |
| export async function processLogUpdateQueue(): Promise<void> { | ||
| const message = await consumeFromQueue(LOG_UPDATE_QUEUE); | ||
|
|
||
| if (!message) { | ||
| return; | ||
| } | ||
|
|
||
| const MAX_RETRIES = 5; | ||
|
|
||
| try { | ||
| const logUpdateData = message.map((i) => JSON.parse(i) as LogUpdateData); | ||
|
|
||
| const processedLogData: LogUpdateData[] = await Promise.all( | ||
| logUpdateData.map(async (data) => { | ||
| const org = await db.query.organization.findFirst({ | ||
| where: { | ||
| id: { | ||
| eq: data.organizationId, | ||
| }, | ||
| }, | ||
| }); | ||
|
|
||
| if (org?.retentionLevel === "none") { | ||
| const { | ||
| messages: _messages, | ||
| content: _content, | ||
| reasoningContent: _reasoningContent, | ||
| tools: _tools, | ||
| toolChoice: _toolChoice, | ||
| toolResults: _toolResults, | ||
| ...metadataOnly | ||
| } = data; | ||
| return metadataOnly as LogUpdateData; | ||
| } | ||
|
|
||
| return data; | ||
| }), | ||
| ); | ||
|
|
||
| // Update logs with retry logic | ||
| let lastError: Error | undefined; | ||
| for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { | ||
| try { | ||
| // Process each update individually since we need to update by ID | ||
| for (const updateData of processedLogData) { | ||
| const { logId, ...fieldsToUpdate } = updateData; | ||
| await db | ||
| .update(log) | ||
| .set(fieldsToUpdate as Partial<LogInsertData>) | ||
| .where(eq(log.id, logId)); | ||
| } | ||
| return; // Success, exit function | ||
| } catch (updateError) { | ||
| lastError = | ||
| updateError instanceof Error | ||
| ? updateError | ||
| : new Error(String(updateError)); | ||
|
|
||
| if (attempt < MAX_RETRIES) { | ||
| const delay = Math.pow(2, attempt) * 1000; // 1s, 2s, 4s, 8s, 16s, ... | ||
| logger.warn( | ||
| `Failed to update logs (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying in ${delay}ms...`, | ||
| lastError, | ||
| ); | ||
| await new Promise((resolve) => { | ||
| setTimeout(resolve, delay); | ||
| }); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // All retries exhausted, push messages back to queue for later processing | ||
| logger.error( | ||
| `Failed to update logs after ${MAX_RETRIES + 1} attempts, pushing back to queue`, | ||
| lastError, | ||
| ); | ||
|
|
||
| // Re-add messages to queue | ||
| for (const msg of message) { | ||
| await publishToQueue(LOG_UPDATE_QUEUE, JSON.parse(msg)); | ||
| } | ||
| } catch (error) { | ||
| logger.error( | ||
| "Error processing log update message", | ||
| error instanceof Error ? error : new Error(String(error)), | ||
| ); | ||
|
|
||
| // Re-add messages to queue on unexpected errors | ||
| try { | ||
| for (const msg of message) { | ||
| await publishToQueue(LOG_UPDATE_QUEUE, JSON.parse(msg)); | ||
| } | ||
| } catch (requeueError) { | ||
| logger.error( | ||
| "Failed to re-queue log update messages", | ||
| requeueError instanceof Error | ||
| ? requeueError | ||
| : new Error(String(requeueError)), | ||
| ); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
The new processLogUpdateQueue function lacks test coverage. Given that the existing worker has test files (worker.spec.ts and log-processing.spec.ts), and this function introduces new critical functionality for updating pending logs, it should have comprehensive test coverage. Tests should validate the retry logic, error handling, data retention filtering, and queue re-publishing behavior.
| export async function insertPendingLog(data: PendingLogData): Promise<string> { | ||
| const logId = shortid(); | ||
|
|
||
| try { | ||
| await db.insert(log).values({ | ||
| id: logId, | ||
| requestId: data.requestId, | ||
| organizationId: data.organizationId, | ||
| projectId: data.projectId, | ||
| apiKeyId: data.apiKeyId, | ||
| requestedModel: data.requestedModel, | ||
| requestedProvider: data.requestedProvider, | ||
| // Set placeholder values for required fields - these will be updated later | ||
| usedModel: data.requestedModel, | ||
| usedProvider: data.requestedProvider || "unknown", | ||
| duration: 0, | ||
| responseSize: 0, | ||
| mode: data.mode, | ||
| usedMode: data.usedMode, | ||
| // Set pending status | ||
| unifiedFinishReason: UnifiedFinishReason.PENDING, | ||
| // Include available data | ||
| messages: data.messages, | ||
| streamed: data.streamed, | ||
| source: data.source, | ||
| userAgent: data.userAgent, | ||
| // Initialize other fields | ||
| hasError: false, | ||
| canceled: false, | ||
| cached: false, | ||
| dataStorageCost: "0", | ||
| }); | ||
|
|
||
| return logId; | ||
| } catch (error) { | ||
| logger.error("Failed to insert pending log", { | ||
| requestId: data.requestId, | ||
| error: error instanceof Error ? error.message : String(error), | ||
| }); | ||
| throw error; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Data for updating an existing log entry. | ||
| * Includes the log ID and all the fields to update. | ||
| */ | ||
| export interface LogUpdateData extends Omit<LogInsertData, "id"> { | ||
| logId: string; | ||
| } | ||
|
|
||
| /** | ||
| * Update an existing log entry via the message queue. | ||
| * This is called at the end of a request to update the pending log with final data. | ||
| */ | ||
| export async function updateLog(logData: LogUpdateData): Promise<unknown> { | ||
| if (logData.unifiedFinishReason === undefined) { | ||
| if (logData.canceled) { | ||
| logData.unifiedFinishReason = UnifiedFinishReason.CANCELED; | ||
| } else { | ||
| logData.unifiedFinishReason = getUnifiedFinishReason( | ||
| logData.finishReason, | ||
| logData.usedProvider, | ||
| ); | ||
|
|
||
| if ( | ||
| logData.unifiedFinishReason === UnifiedFinishReason.UNKNOWN && | ||
| logData.finishReason && | ||
| !isExpectedUnknownFinishReason( | ||
| logData.finishReason, | ||
| logData.usedProvider, | ||
| ) | ||
| ) { | ||
| logger.error("Unknown finish reason encountered", { | ||
| requestId: logData.requestId, | ||
| finishReason: logData.finishReason, | ||
| provider: logData.usedProvider, | ||
| model: logData.usedModel, | ||
| }); | ||
| } | ||
| } | ||
| } | ||
| await publishToQueue(LOG_UPDATE_QUEUE, logData); | ||
| return 1; | ||
| } |
There was a problem hiding this comment.
The new functions insertPendingLog and updateLog lack test coverage. The existing logs.spec.ts file has tests for other log-related functions, so these new functions should also have tests. Tests should cover successful insertion, error handling, queue publishing, and the unified finish reason logic in updateLog.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@apps/gateway/src/lib/logs.ts`:
- Around line 191-242: The pending log currently always writes request messages
to the DB; update insertPendingLog and PendingLogData to accept a retentionLevel
("none" | ...) and, when retentionLevel === "none", avoid storing messages
(e.g., set messages: null or omit it) so sensitive content isn't persisted;
update the call sites to pass the retention level into insertPendingLog and
adjust any types/DB schema usage accordingly (refer to PendingLogData and
insertPendingLog to locate changes and add the conditional use of
data.retentionLevel when assigning the messages field).
| export interface PendingLogData { | ||
| requestId: string; | ||
| organizationId: string; | ||
| projectId: string; | ||
| apiKeyId: string; | ||
| requestedModel: string; | ||
| requestedProvider: string | null; | ||
| mode: "api-keys" | "credits" | "hybrid"; | ||
| usedMode: "api-keys" | "credits"; | ||
| messages: unknown; | ||
| streamed: boolean; | ||
| source: string | null; | ||
| userAgent: string | null; | ||
| } | ||
|
|
||
| /** | ||
| * Insert a pending log entry synchronously to the database. | ||
| * This is called at the start of a request to track requests that may timeout or be cancelled. | ||
| * Returns the log ID that should be used in the subsequent updateLog() call. | ||
| */ | ||
| export async function insertPendingLog(data: PendingLogData): Promise<string> { | ||
| const logId = shortid(); | ||
|
|
||
| try { | ||
| await db.insert(log).values({ | ||
| id: logId, | ||
| requestId: data.requestId, | ||
| organizationId: data.organizationId, | ||
| projectId: data.projectId, | ||
| apiKeyId: data.apiKeyId, | ||
| requestedModel: data.requestedModel, | ||
| requestedProvider: data.requestedProvider, | ||
| // Set placeholder values for required fields - these will be updated later | ||
| usedModel: data.requestedModel, | ||
| usedProvider: data.requestedProvider || "unknown", | ||
| duration: 0, | ||
| responseSize: 0, | ||
| mode: data.mode, | ||
| usedMode: data.usedMode, | ||
| // Set pending status | ||
| unifiedFinishReason: UnifiedFinishReason.PENDING, | ||
| // Include available data | ||
| messages: data.messages, | ||
| streamed: data.streamed, | ||
| source: data.source, | ||
| userAgent: data.userAgent, | ||
| // Initialize other fields | ||
| hasError: false, | ||
| canceled: false, | ||
| cached: false, | ||
| dataStorageCost: "0", | ||
| }); |
There was a problem hiding this comment.
Respect retentionLevel when inserting pending logs.
Pending logs are written directly to the DB, so for retentionLevel = "none" the request messages can persist indefinitely (the update path won’t clear them). This breaks retention guarantees and can leak sensitive content.
🛡️ Proposed fix (avoid storing messages when retention is "none")
export interface PendingLogData {
requestId: string;
organizationId: string;
projectId: string;
apiKeyId: string;
requestedModel: string;
requestedProvider: string | null;
mode: "api-keys" | "credits" | "hybrid";
usedMode: "api-keys" | "credits";
messages: unknown;
streamed: boolean;
source: string | null;
userAgent: string | null;
+ retentionLevel?: "retain" | "none" | null;
}
export async function insertPendingLog(data: PendingLogData): Promise<string> {
const logId = shortid();
+ const storeMessages = data.retentionLevel !== "none";
try {
await db.insert(log).values({
id: logId,
requestId: data.requestId,
organizationId: data.organizationId,
projectId: data.projectId,
apiKeyId: data.apiKeyId,
requestedModel: data.requestedModel,
requestedProvider: data.requestedProvider,
usedModel: data.requestedModel,
usedProvider: data.requestedProvider || "unknown",
duration: 0,
responseSize: 0,
mode: data.mode,
usedMode: data.usedMode,
unifiedFinishReason: UnifiedFinishReason.PENDING,
- messages: data.messages,
+ messages: storeMessages ? data.messages : null,
streamed: data.streamed,
source: data.source,
userAgent: data.userAgent,
hasError: false,
canceled: false,
cached: false,
dataStorageCost: "0",
});And pass the retention level at the call site:
pendingLogId = await insertPendingLog({
requestId,
organizationId: project.organizationId,
projectId: apiKey.projectId,
apiKeyId: apiKey.id,
requestedModel: initialRequestedModel,
requestedProvider: requestedProvider || null,
mode: project.mode,
usedMode: providerKey?.id ? "api-keys" : "credits",
messages,
streamed: stream || false,
source: source || null,
userAgent: userAgent || null,
+ retentionLevel,
});🤖 Prompt for AI Agents
In `@apps/gateway/src/lib/logs.ts` around lines 191 - 242, The pending log
currently always writes request messages to the DB; update insertPendingLog and
PendingLogData to accept a retentionLevel ("none" | ...) and, when
retentionLevel === "none", avoid storing messages (e.g., set messages: null or
omit it) so sensitive content isn't persisted; update the call sites to pass the
retention level into insertPendingLog and adjust any types/DB schema usage
accordingly (refer to PendingLogData and insertPendingLog to locate changes and
add the conditional use of data.retentionLevel when assigning the messages
field).
Summary
/v1/chat/completionsrequests to track requests that may timeout or be cancelledpendingstatus toUnifiedFinishReasonenum for tracking in-flight requestsinsertPendingLog()function for synchronous pending log insertion at request startupdateLog()function andLOG_UPDATE_QUEUEfor async log updates when request completesprocessLogUpdateQueue()worker loop to handle log updatesTest plan
unifiedFinishReason='pending'logs older than 5 minutes to detect stuck requests🤖 Generated with Claude Code
Summary by CodeRabbit
Release Notes
New Features
Infrastructure