Conversation
Removed internal benchmark command examples from the implementation PR requirements.
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
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 |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d8e67707a8
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| const em = { upsert: jest.fn().mockResolvedValue(undefined) } | ||
| await recordDataSyncRepairCell(em as never, 'progress-1', { tenantId: 'tenant-1', organizationId: 'org-1' }, 'import', 'upstream timeout') |
There was a problem hiding this comment.
This test constructs an entity manager with only upsert, but recordDataSyncRepairCell() delegates to upsertRepairCell(), which calls em.getConnection().execute(...). Consequently, running this new test throws TypeError: em.getConnection is not a function before reaching the assertion; mock the connection API and assert the executed statement instead.
Useful? React with 👍 / 👎.
| queueWatcher = watcher | ||
| queueWatcherIdentity = nextIdentity | ||
| watcherRetryAttempts = 0 |
There was a problem hiding this comment.
When fs.watch() returns a watcher but that watcher subsequently emits a persistent error, every retry successfully constructs another watcher and resets watcherRetryAttempts here before the next error. The error handler therefore never reaches the three-attempt limit, producing an unbounded 100 ms re-arm and logging loop in exactly the asynchronous watcher-error scenario this recovery path handles; reset the counter only after a stable period or preserve the accumulated error count across re-arms.
Useful? React with 👍 / 👎.
| queueDirectoryProbe = setInterval(() => { | ||
| void fsp.stat(queueDir).then(() => { |
There was a problem hiding this comment.
Every continuous local worker now starts this interval permanently, so even with a healthy watcher and an idle queue it performs ten filesystem stat calls per second per queue. A worker --all process therefore creates continuous filesystem load proportional to the number of queues, contradicting the accompanying recovery design that limits probes to periods when the directory is absent or the watcher is unavailable; arm the fast probe only during recovery and stop it after the watcher is restored.
Useful? React with 👍 / 👎.
| "lease_epoch" = excluded."lease_epoch", | ||
| "reason" = excluded."reason", | ||
| "updated_at" = now() | ||
| where excluded."lease_epoch" >= "progress_job_repair_cells"."lease_epoch" |
There was a problem hiding this comment.
When an upsert with an equal or newer leaseEpoch is accepted while the cell is already leased, this update replaces the cell's due time and reason but leaves the previous lease_token and lease_until intact. The worker holding that old token can then call acknowledgeRepairCell() and delete the newly recorded repair condition, so a failure that races with an active repair is lost; invalidate the old lease on an accepted replacement or include the epoch in acknowledgement and release predicates.
Useful? React with 👍 / 👎.
| const syncRunService = ctx.resolve<SyncRunService>('dataSyncRunService') | ||
| const run = await syncRunService.getRun(job.payload.runId, job.payload.scope) | ||
| await clearDataSyncRepairCell(em, run?.progressJobId, job.payload.scope) |
There was a problem hiding this comment.
On a queue retry after this handler has marked the run failed, runImport() can resolve the transient dependency and then return early because its failed → running transition matches zero rows. The worker nevertheless reloads that still-failed run and clears its repair cell, so the delivery is reported successful and the only repair marker disappears without the sync being rerun; verify that the run actually reached the intended recovered/terminal state before clearing the cell. The export worker contains the same sequence.
Useful? React with 👍 / 👎.
| const em = ctx.resolve<import('@mikro-orm/postgresql').EntityManager>('em') | ||
| const progressService = ctx.resolve<ProgressService>('progressService') | ||
| const run = await syncRunService.getRun(job.payload.runId, job.payload.scope) | ||
| await recordDataSyncRepairCell(em, run?.progressJobId, job.payload.scope, 'import', message) |
There was a problem hiding this comment.
If the new repair table is unavailable—for example during schema drift or a rolling deployment—or its conditional upsert otherwise fails, this awaited auxiliary write aborts the entire nested finalization block before markStatus(..., 'failed') and progressService.failJob() run. Every retry can therefore leave both the sync run and its progress job non-terminal solely because diagnostic repair bookkeeping failed; isolate/log the repair-cell error and always execute the existing terminal transitions. The export worker has the same ordering.
AGENTS.md reference: packages/core/src/modules/progress/AGENTS.md:L7-L10
Useful? React with 👍 / 👎.
1 participant
PR 5450 integration candidate
This branch integrates the two review lanes onto the current PR open-mercato#5450 base (
60e0bac3b7):Validation
3/3.35/35.104/104with--forceExit; normal Jest shutdown still has an open-handle cleanup gate.This is an integration candidate for review, not a direct mutation of the upstream PR branch. The remaining live Postgres concurrent-claim/crash-recovery, Valkey worker-loss, workflow latency/RSS, and full upstream validation gates remain explicit.