Skip to content

Repository navigation

Commit 4c85fb4

Browse files
brendan-kellamclaude
brendan-kellam
and
claude
authored
fix(worker): bound BullMQ job retention to prevent Redis OOM (#1693)
* fix(worker): bound BullMQ job retention to prevent Redis OOM (SOU-2311) Completed and failed jobs were retained by age only (14 days) with no count cap, so Redis memory grew with repo count times reindex frequency. - Window retention now caps completed jobs at 1 day / 5,000 and failed jobs at 14 days / 10,000 per queue. - Queues whose latest job is resolved by the web app (repo-index, repo-permission-sync, connection-sync, account-permission-sync) use a new latestPerResource retention: when a job starts, the job it supersedes for the same resource is removed, keyed by the queue's deduplication id, with a 7-day age backstop for orphans. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * chore: add changelog entry for #1693 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(worker): publish repo-index job pointer in onStarted before superseding trackLatestJob removes a resource's previous job right after onStarted, which assumes the parent's latest...JobId pointer has already moved. repo-index wrote latestIndexingJobId inside process instead, so an abort or DB error before that write could leave the pointer on a removed job. - Move the pointer write to an onStarted hook (updateMany keyed by repo id), matching the other latestPerResource workloads. - Reject a latestPerResource workload without onStarted at registration so the contract cannot be silently skipped by a future workload. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
1 parent 5091d26 commit 4c85fb4

9 files changed

Lines changed: 400 additions & 84 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
1818
### Fixed
1919
- Made the default home page configurable with `DEFAULT_HOME_VIEW_PAGE`, defaulting to Code Search and supporting Ask. [#1677](https://github.com/sourcebot-dev/sourcebot/pull/1677)
2020
- Require authentication for the streaming and blocking Ask APIs in Public SaaS deployments. [#1679](https://github.com/sourcebot-dev/sourcebot/pull/1679)
21+
- Bounded BullMQ job retention to keep Redis memory from growing with repo count, retaining only the latest job per repo, connection, and account. [#1693](https://github.com/sourcebot-dev/sourcebot/pull/1693)
2122

2223
## [5.1.14] - 2026-09-17
2324

‎packages/backend/src/connectionSyncWorkload.test.ts‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,9 +42,12 @@ vi.mock("@sourcebot/shared", () => ({
4242
jobOptions: {
4343
attempts: 2,
4444
backoff: { type: "exponential", delayMs: 5000 },
45-
keepJobs: {
46-
completed: { count: 50 },
47-
failed: { count: 50 },
45+
retention: {
46+
mode: "window",
47+
keepJobs: {
48+
completed: { count: 50 },
49+
failed: { count: 50 },
50+
},
4851
},
4952
keepLogs: 500,
5053
},

‎packages/backend/src/jobManager.test.ts‎

Lines changed: 75 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ const mocks = vi.hoisted(() => {
2323
upsertJobScheduler: vi.fn(),
2424
getJobSchedulerIds: vi.fn(),
2525
removeJobScheduler: vi.fn(),
26+
trackLatestJob: vi.fn(async () => null),
2627
producerClose: vi.fn(),
2728
workerClose: vi.fn(),
2829
executionLockUsing: vi.fn(),
@@ -54,6 +55,7 @@ vi.mock("@sourcebot/shared", () => ({
5455
upsertJobScheduler = mocks.upsertJobScheduler;
5556
getJobSchedulerIds = mocks.getJobSchedulerIds;
5657
removeJobScheduler = mocks.removeJobScheduler;
58+
trackLatestJob = mocks.trackLatestJob;
5759
close = mocks.producerClose;
5860
getQueue = vi.fn(() => ({
5961
getJobCounts: vi.fn(),
@@ -113,9 +115,12 @@ const createWorkload = (
113115
jobOptions: {
114116
attempts: 2,
115117
backoff: { type: "exponential", delayMs: 5000 },
116-
keepJobs: {
117-
completed: { count: 50 },
118-
failed: { count: 50 },
118+
retention: {
119+
mode: "window",
120+
keepJobs: {
121+
completed: { count: 50 },
122+
failed: { count: 50 },
123+
},
119124
},
120125
keepLogs: 500,
121126
},
@@ -152,6 +157,27 @@ describe("BullMQJobManager lifecycle", () => {
152157
);
153158
});
154159

160+
test("rejects a latestPerResource workload that does not publish its job id in onStarted", () => {
161+
const manager = new BullMQJobManager({} as Redis);
162+
const workload = createWorkload();
163+
workload.queueSpec.jobOptions = {
164+
...workload.queueSpec.jobOptions,
165+
retention: { mode: "latestPerResource", maxAgeSeconds: 60 },
166+
};
167+
168+
expect(() => manager.register(workload)).toThrow(
169+
/must publish its job id to the parent resource in onStarted/,
170+
);
171+
expect(() =>
172+
manager.register(
173+
createWorkload({
174+
queueSpec: workload.queueSpec,
175+
onStarted: vi.fn(),
176+
}),
177+
),
178+
).not.toThrow();
179+
});
180+
155181
test("delegates enqueueing to BullMQClient and returns its job id", async () => {
156182
const manager = new BullMQJobManager({} as Redis);
157183
const workload = createWorkload();
@@ -297,6 +323,52 @@ describe("BullMQJobManager lifecycle", () => {
297323
});
298324
});
299325

326+
test("tracks the latest job for its resource after onStarted and before processing", async () => {
327+
const calls: string[] = [];
328+
mocks.trackLatestJob.mockImplementation(async () => {
329+
calls.push("tracked");
330+
return "job-0";
331+
});
332+
const workload = createWorkload({
333+
onStarted: vi.fn(async () => {
334+
calls.push("started");
335+
}),
336+
process: vi.fn(async () => {
337+
calls.push("processed");
338+
return { outcome: "SUCCESS" };
339+
}),
340+
});
341+
const manager = new BullMQJobManager({} as Redis);
342+
manager.register(workload);
343+
await manager.start();
344+
345+
await mocks.workers[0].processor({ ...job, attemptsMade: 0 });
346+
347+
expect(calls).toEqual(["started", "tracked", "processed"]);
348+
expect(mocks.trackLatestJob).toHaveBeenCalledWith(
349+
workload.queueSpec,
350+
data,
351+
"job-1",
352+
);
353+
});
354+
355+
test("still processes the job when tracking the latest job fails", async () => {
356+
mocks.trackLatestJob.mockRejectedValue(new Error("redis unavailable"));
357+
const workload = createWorkload();
358+
const manager = new BullMQJobManager({} as Redis);
359+
manager.register(workload);
360+
await manager.start();
361+
362+
await expect(
363+
mocks.workers[0].processor({ ...job, attemptsMade: 0 }),
364+
).resolves.toEqual({ outcome: "SUCCESS" });
365+
expect(workload.process).toHaveBeenCalledTimes(1);
366+
expect(mocks.logger.warn).toHaveBeenCalledWith(
367+
expect.stringContaining("Failed to track latest job"),
368+
expect.any(Error),
369+
);
370+
});
371+
300372
test("runs onStarted and processing while the execution lock is held", async () => {
301373
const calls: string[] = [];
302374
const workloadSignal = new AbortController().signal;

‎packages/backend/src/jobManager.ts‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import {
66
DataOf,
77
JobEnqueueOptions,
88
QueueName,
9+
QueueSpec,
910
ResultOf,
1011
Schedule,
1112
scheduleToMs,
@@ -41,6 +42,14 @@ export class BullMQJobManager implements JobManager {
4142
if (this.workloads.has(name)) {
4243
throw new Error(`Workload "${name}" is already registered`);
4344
}
45+
if (
46+
workload.queueSpec.jobOptions.retention.mode === "latestPerResource"
47+
&& !workload.onStarted
48+
) {
49+
throw new Error(
50+
`Workload "${name}" uses latestPerResource retention and must publish its job id to the parent resource in onStarted`,
51+
);
52+
}
4453
this.workloads.set(name, workload);
4554
}
4655

@@ -150,6 +159,11 @@ export class BullMQJobManager implements JobManager {
150159

151160
const process = async (signal: AbortSignal) => {
152161
await workload.onStarted?.(lifecycleContext);
162+
// Contract: `latestPerResource` workloads publish this job's id to
163+
// their parent's `latest...JobId` pointer in onStarted (enforced in
164+
// `register`), so the pointer has moved before the superseded job
165+
// is removed.
166+
await this.trackLatestJob(spec, job);
153167
return workload.process({
154168
...lifecycleContext,
155169
signal,
@@ -257,6 +271,33 @@ export class BullMQJobManager implements JobManager {
257271
}
258272
}
259273

274+
private async trackLatestJob<TName extends QueueName>(
275+
spec: QueueSpec<TName>,
276+
job: Job,
277+
): Promise<void> {
278+
if (!job.id) {
279+
return;
280+
}
281+
try {
282+
const removedJobId = await this.bullmqClient.trackLatestJob(
283+
spec,
284+
job.data,
285+
job.id,
286+
);
287+
if (removedJobId) {
288+
logger.debug(
289+
`Removed job ${removedJobId} superseded by ${job.id} on "${spec.name}"`,
290+
);
291+
}
292+
} catch (error) {
293+
// Retention is best-effort; the age backstop still bounds the queue.
294+
logger.warn(
295+
`Failed to track latest job ${job.id} on "${spec.name}"`,
296+
error,
297+
);
298+
}
299+
}
300+
260301
private getWorkload<TName extends QueueName>(
261302
workloadName: TName,
262303
): Workload<TName> {

‎packages/backend/src/repoIndexWorkload.test.ts‎

Lines changed: 12 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -19,18 +19,10 @@ const repoFindUnique = vi.fn();
1919
const repoUpdate = vi.fn();
2020
const repoUpdateMany = vi.fn();
2121

22-
const transaction = vi.fn(async (callback: (tx: unknown) => Promise<unknown>) =>
23-
callback({
24-
repo: {
25-
findUnique: repoFindUnique,
26-
update: repoUpdate,
27-
},
28-
}),
29-
);
30-
3122
const db = {
32-
$transaction: transaction,
3323
repo: {
24+
findUnique: repoFindUnique,
25+
update: repoUpdate,
3426
updateMany: repoUpdateMany,
3527
},
3628
} as unknown as PrismaClient;
@@ -121,10 +113,20 @@ describe("repoIndexWorkload", () => {
121113
});
122114
});
123115

116+
test("publishes the job as the repository's latest indexing job when it starts", async () => {
117+
await workload.onStarted!(lifecycleContext);
118+
119+
expect(repoUpdateMany).toHaveBeenCalledWith({
120+
where: { id: 42 },
121+
data: { latestIndexingJobId: "job-1" },
122+
});
123+
});
124+
124125
test("skips an INDEX job when the repository no longer exists", async () => {
125126
await workload.process(processContext);
126127

127128
expect(repoUpdate).not.toHaveBeenCalled();
129+
expect(repoUpdateMany).not.toHaveBeenCalled();
128130
expect(lifecycleLogger.debug).toHaveBeenCalledWith(
129131
"Skipping INDEX job for repo 42: repository no longer exists",
130132
);

‎packages/backend/src/repoIndexWorkload.ts‎

Lines changed: 30 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,22 @@ export const createRepoIndexWorkload = ({
2525
queueSpec: REPO_INDEX_QUEUE,
2626
concurrency: settings.maxRepoIndexingJobConcurrency,
2727
executionLock: REPOSITORY_EXECUTION_LOCK,
28-
process: async ({ data, jobId, signal }) => {
28+
onStarted: async ({ data: { repoId }, jobId }) => {
29+
await db.repo.updateMany({
30+
where: {
31+
id: repoId,
32+
},
33+
data: {
34+
latestIndexingJobId: jobId,
35+
},
36+
});
37+
},
38+
process: async ({ data, signal }) => {
2939
signal.throwIfAborted();
3040

3141
const start = await prepareRepoIndexJob({
3242
db,
3343
repoId: data.repoId,
34-
jobId,
3544
});
3645

3746
if (start.action === "skip") {
@@ -119,45 +128,33 @@ type RepoIndexStartDecision =
119128
const prepareRepoIndexJob = async ({
120129
db,
121130
repoId,
122-
jobId,
123131
}: {
124132
db: PrismaClient;
125133
repoId: number;
126-
jobId: string;
127-
}): Promise<RepoIndexStartDecision> =>
128-
db.$transaction(async (tx) => {
129-
const repo = await tx.repo.findUnique({
130-
where: { id: repoId },
131-
include: {
132-
connections: {
133-
include: {
134-
connection: true,
135-
},
134+
}): Promise<RepoIndexStartDecision> => {
135+
const repo = await db.repo.findUnique({
136+
where: { id: repoId },
137+
include: {
138+
connections: {
139+
include: {
140+
connection: true,
136141
},
137142
},
138-
});
139-
140-
if (!repo) {
141-
return {
142-
action: "skip",
143-
reason: "repository no longer exists",
144-
};
145-
}
146-
147-
await tx.repo.update({
148-
where: {
149-
id: repoId,
150-
},
151-
data: {
152-
latestIndexingJobId: jobId,
153-
},
154-
});
143+
},
144+
});
155145

146+
if (!repo) {
156147
return {
157-
action: "run",
158-
repo,
148+
action: "skip",
149+
reason: "repository no longer exists",
159150
};
160-
});
151+
}
152+
153+
return {
154+
action: "run",
155+
repo,
156+
};
157+
};
161158

162159
const indexRepository = async (
163160
db: PrismaClient,

0 commit comments

Comments
 (0)