Make review queue jobs resumable and lease-aware#11
Conversation
![codra-app-personal[bot]](https://avatars.githubusercontent.com/in/3347243?s=60&v=4){kind=small}
There was a problem hiding this comment.
Codra Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 35d06d55fd
ℹ️ About Codra in GitHub
Your team has set up Codra to review pull requests in this repo. Reviews are triggered when you:
- Open a pull request for review
- Mark a draft as ready
- Comment "@codra-app review"
If Codra has suggestions, it will comment; otherwise it will react with 👍.
Codra can also answer questions or update the PR. Try commenting "@codra-app address that feedback".
| return { | ||
| async query<T>(sqlText: string, params: unknown[] = []) { | ||
| return (await sql.unsafe(sqlText, params.map(normalizeParam) as any[], { prepare: true })) as T[]; | ||
| return (await sql.unsafe(sqlText, params.map(normalizeParam) as any[], { prepare: false })) as T[]; |
There was a problem hiding this comment.
The use of 'as any[]' when mapping parameters bypasses TypeScript's type checking. Since 'params' is already 'unknown[]', and 'normalizeParam' likely returns a type compatible with the database driver, casting to 'unknown[]' or omitting the cast (if the function return type allows) would be safer and more aligned with TypeScript best practices.
| return (await sql.unsafe(sqlText, params.map(normalizeParam) as any[], { prepare: false })) as T[]; | |
| return (await sql.unsafe(sqlText, params.map(normalizeParam), { prepare: false })) as T[]; |
| const app = new Hono<AppEnv>(); | ||
|
|
||
| app.get('/', async (c) => { | ||
| await runOpportunisticJobMaintenance(c.env); |
There was a problem hiding this comment.
The calls to runOpportunisticJobMaintenance(c.env) are awaited directly within the request handlers. If the maintenance logic fails (e.g., database timeout, lock contention, or internal error), it will throw an exception that results in a 500 Internal Server Error for the user. Since this maintenance is 'opportunistic' and not critical to the immediate request's success, it should be wrapped in a try-catch block to ensure that the API remains available even if maintenance fails.
| await runOpportunisticJobMaintenance(c.env); | |
| try { | |
| await runOpportunisticJobMaintenance(c.env); | |
| } catch (e) { | |
| console.error('Opportunistic job maintenance failed:', e); | |
| } |
| async send(message: any) { | ||
| this.sent.push(message); | ||
| async send(message: any, options?: { delaySeconds?: number }) { | ||
| this.sent.push({ ...message, options }); |
There was a problem hiding this comment.
The code uses the spread operator { ...message, options } to store the sent message. If message is a primitive (e.g., a string or number), the spread operator will not behave as expected (e.g., a string will be spread into indexed characters). Furthermore, if message contains a property named 'options', it will be overwritten by the options argument. It is safer to store the message and options as distinct properties in a wrapper object.
| this.sent.push({ ...message, options }); | |
| this.sent.push({ message, options }); |
| @@ -49,8 +49,8 @@ export class MockAssets { | |||
| export class MockQueue { | |||
| public readonly sent: any[] = []; | |||
There was a problem hiding this comment.
The sent array and the message parameter are typed as any. This bypasses TypeScript's type checking and can lead to runtime errors in tests. Since this is a Mock class, it should ideally use generics to allow the caller to specify the expected message type.
| public readonly sent: any[] = []; | |
| export class MockQueue<T = any> { | |
| public readonly sent: Array<{ message: T; options?: { delaySeconds?: number } }> = []; | |
| async send(message: T, options?: { delaySeconds?: number }) { | |
| this.sent.push({ message, options }); | |
| } | |
| } |
| ], | ||
| usage: { prompt_tokens: 1, completion_tokens: 4096 }, | ||
| }; | ||
| }, |
There was a problem hiding this comment.
The use of 'as any' to mock the Cloudflare AI binding (lines 45, 64, 82) bypasses TypeScript's type checking. While common in tests, it's better to use 'Partial' or a specific interface for the binding to ensure the mock remains compatible with the actual API as it evolves.
| }, | |
| // Example: Cast to a partial of the expected AI binding type | |
| AI: { | |
| async run() { | |
| // ... | |
| }, | |
| } as Partial<CloudflareAIBinding> |
There was a problem hiding this comment.
Codra Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 35d06d55fd
ℹ️ About Codra in GitHub
Your team has set up Codra to review pull requests in this repo. Reviews are triggered when you:
- Open a pull request for review
- Mark a draft as ready
- Comment "@codra-app review"
If Codra has suggestions, it will comment; otherwise it will react with 👍.
Codra can also answer questions or update the PR. Try commenting "@codra-app address that feedback".
| return { | ||
| async query<T>(sqlText: string, params: unknown[] = []) { | ||
| return (await sql.unsafe(sqlText, params.map(normalizeParam) as any[], { prepare: true })) as T[]; | ||
| return (await sql.unsafe(sqlText, params.map(normalizeParam) as any[], { prepare: false })) as T[]; |
There was a problem hiding this comment.
The use of 'as any[]' when mapping parameters bypasses TypeScript's type checking. Since 'params' is already 'unknown[]', and 'normalizeParam' likely returns a type compatible with the database driver, casting to 'unknown[]' or omitting the cast (if the function return type allows) would be safer and more aligned with TypeScript best practices.
| return (await sql.unsafe(sqlText, params.map(normalizeParam) as any[], { prepare: false })) as T[]; | |
| return (await sql.unsafe(sqlText, params.map(normalizeParam), { prepare: false })) as T[]; |
| ], | ||
| ); | ||
|
|
||
| await queryRows(env, 'DELETE FROM review_comments WHERE file_review_id = $1::uuid', [review.id]); |
There was a problem hiding this comment.
The function deletes existing comments before inserting new ones. If the subsequent insert operation fails (e.g., due to a network error or constraint violation), the database will be left in an inconsistent state where the review record exists but the comments are missing. This violates data integrity.
| await queryRows(env, 'DELETE FROM review_comments WHERE file_review_id = $1::uuid', [review.id]); | |
| Wrap the delete and insert operations in a database transaction to ensure atomicity. If the insert fails, the delete should be rolled back. |
| diffLineCount: number; | ||
| diffInput: string | null; | ||
| rawAiOutput: string | null; | ||
| parsedComments: ParsedReviewComment[]; |
There was a problem hiding this comment.
The type ParsedReviewComment is used in the function signature but is not imported from its source module. This will cause a TypeScript compilation error.
| parsedComments: ParsedReviewComment[]; | |
| Add the missing import statement at the top of the file, e.g., `import type { ParsedReviewComment } from './types';` |
| export async function upsertFileReview( | ||
| env: Pick<AppBindings, 'HYPERDRIVE'>, | ||
| jobId: string, | ||
| input: { |
There was a problem hiding this comment.
While the function signature defines the shape of input, the variable input inside the function is inferred. In strict TypeScript mode, relying on object literal inference can sometimes lead to issues if the literal doesn't match the interface exactly.
| input: { | |
| Explicitly type the parameter or the variable, e.g., `input: UpsertFileReviewInput` if a dedicated interface is created, or ensure the literal passed in matches the type definition strictly. |
| } catch (err) { | ||
| // Non-fatal: log and continue processing the batch. | ||
| logger.error('Failed to recover stale jobs', err instanceof Error ? err : new Error(String(err))); | ||
| } |
There was a problem hiding this comment.
The runOpportunisticJobMaintenance call was moved from the start of the queue function to the end. This is a critical logic error. If a job is stuck in 'running' state due to a previous crash, it will not be recovered until the next batch is processed. If the queue is empty, stale jobs will remain stuck indefinitely, violating the 'resumable' goal.
| } | |
| Move `await runOpportunisticJobMaintenance(env);` back to the beginning of the `queue` function, before processing messages, to ensure stale jobs are recovered immediately. |
| super(message); | ||
| this.name = 'RetryableModelError'; | ||
| if (cause !== undefined) { | ||
| (this as any).cause = cause; |
There was a problem hiding this comment.
RetryableModelError constructor
The code uses (this as any) to assign the cause property (line 26). This is unnecessary and unsafe. The Error class supports the cause property natively in modern JavaScript/TypeScript.
| (this as any).cause = cause; | |
| Object.defineProperty(this, 'cause', { value: cause, writable: true, configurable: true }); |
| } | ||
|
|
||
| export function isRetryableModelError(error: unknown) { | ||
| return Boolean(error && typeof error === 'object' && (error as any).retryable === true); |
There was a problem hiding this comment.
isRetryableModelError
The function uses (error as any) to check for the retryable property (line 32). This defeats the purpose of using a type-safe language. It should use instanceof or a type guard.
| return Boolean(error && typeof error === 'object' && (error as any).retryable === true); | |
| return error instanceof RetryableModelError; |
| async send(message: any) { | ||
| this.sent.push(message); | ||
| async send(message: any, options?: { delaySeconds?: number }) { | ||
| this.sent.push({ ...message, options }); |
There was a problem hiding this comment.
The code uses the spread operator { ...message, options } to store the sent message. If message is a primitive (e.g., a string or number), the spread operator will not behave as expected (e.g., a string will be spread into indexed characters). Furthermore, if message contains a property named 'options', it will be overwritten by the options argument. It is safer to store the message and options as distinct properties in a wrapper object.
| this.sent.push({ ...message, options }); | |
| this.sent.push({ message, options }); |
| @@ -49,8 +49,8 @@ export class MockAssets { | |||
| export class MockQueue { | |||
| public readonly sent: any[] = []; | |||
There was a problem hiding this comment.
The sent array and the message parameter are typed as any. This bypasses TypeScript's type checking and can lead to runtime errors in tests. Since this is a Mock class, it should ideally use generics to allow the caller to specify the expected message type.
| public readonly sent: any[] = []; | |
| export class MockQueue<T = any> { | |
| public readonly sent: Array<{ message: T; options?: { delaySeconds?: number } }> = []; | |
| async send(message: T, options?: { delaySeconds?: number }) { | |
| this.sent.push({ message, options }); | |
| } | |
| } |
| ], | ||
| usage: { prompt_tokens: 1, completion_tokens: 4096 }, | ||
| }; | ||
| }, |
There was a problem hiding this comment.
The use of 'as any' to mock the Cloudflare AI binding (lines 45, 64, 82) bypasses TypeScript's type checking. While common in tests, it's better to use 'Partial' or a specific interface for the binding to ensure the mock remains compatible with the actual API as it evolves.
| }, | |
| // Example: Cast to a partial of the expected AI binding type | |
| AI: { | |
| async run() { | |
| // ... | |
| }, | |
| } as Partial<CloudflareAIBinding> |
1 participant
Description
This PR makes review jobs resilient to worker crashes, duplicate queue deliveries, and transient model/provider failures.
Closes #9
Type of change
How Has This Been Tested?
Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration.
Checklist: