agent_coder
97bd554cb5
The {search-queue} BullMQ queue had producers on every page/space/workspace change
but NO consumer/@Processor — it lives in Docmost's EE edition (external search
drivers), never in this fork. Search works independently via the pages tsvector DB
trigger; the queue was pure dead weight, growing forever (1902 stuck jobs in Redis,
the first real hit of the #355 queue-growing alert — which fired correctly).
Remove the plumbing:
- the 3 producers: drop @InjectQueue(SEARCH_QUEUE) + every searchQueue.add(...) from
page/space/workspace.listener.ts (and the isTypesense() gate that only wrapped the
search enqueue). page.listener.handlePageUpdated did ONLY the search enqueue, so its
@OnEvent(PAGE_UPDATED) handler is removed; the other handlers keep their aiQueue.add.
- the registration (queue.module.ts) and the metrics injection + 'search' depth-metric
entry (metrics-bull.service.ts).
- the constants: QueueName.SEARCH_QUEUE + the 6 unused SEARCH_INDEX_* QueueJob members
(grep-confirmed unreferenced). Shared QueueJob members (PAGE_*, SPACE_DELETED,
WORKSPACE_DELETED — used by the AI queue / embedding/history/notification processors)
are kept.
Search (tsvector trigger, search.service) and the PAGE_UPDATED websocket listener are
untouched. Verified: apps/server tsc --noEmit 0 errors; 6 spec suites / 52 tests green.
Ops (maintainer, out of code scope): one-time Redis cleanup
`redis-cli --scan --pattern "bull:{search-queue}:*" | xargs -r redis-cli del`, then
confirm bullmq_queue_depth{queue="search"} stays 0 (the metric no longer injects it).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
92 lines
2.4 KiB
TypeScript
92 lines
2.4 KiB
TypeScript
import { Global, Module } from '@nestjs/common';
|
|
import { BullModule } from '@nestjs/bullmq';
|
|
import { EnvironmentService } from '../environment/environment.service';
|
|
import { createRetryStrategy, parseRedisUrl } from '../../common/helpers';
|
|
import { QueueName } from './constants';
|
|
import { GeneralQueueProcessor } from './processors/general-queue.processor';
|
|
|
|
@Global()
|
|
@Module({
|
|
imports: [
|
|
BullModule.forRootAsync({
|
|
useFactory: (environmentService: EnvironmentService) => {
|
|
const redisConfig = parseRedisUrl(environmentService.getRedisUrl());
|
|
return {
|
|
connection: {
|
|
host: redisConfig.host,
|
|
port: redisConfig.port,
|
|
password: redisConfig.password,
|
|
db: redisConfig.db,
|
|
family: redisConfig.family,
|
|
retryStrategy: createRetryStrategy(),
|
|
},
|
|
defaultJobOptions: {
|
|
attempts: 3,
|
|
backoff: {
|
|
type: 'exponential',
|
|
delay: 20 * 1000,
|
|
},
|
|
removeOnComplete: {
|
|
count: 200,
|
|
},
|
|
removeOnFail: {
|
|
count: 100,
|
|
},
|
|
},
|
|
};
|
|
},
|
|
inject: [EnvironmentService],
|
|
}),
|
|
BullModule.registerQueue({
|
|
name: QueueName.EMAIL_QUEUE,
|
|
}),
|
|
BullModule.registerQueue({
|
|
name: QueueName.ATTACHMENT_QUEUE,
|
|
}),
|
|
BullModule.registerQueue({
|
|
name: QueueName.GENERAL_QUEUE,
|
|
}),
|
|
BullModule.registerQueue({
|
|
name: QueueName.BILLING_QUEUE,
|
|
}),
|
|
BullModule.registerQueue({
|
|
name: QueueName.FILE_TASK_QUEUE,
|
|
defaultJobOptions: {
|
|
removeOnComplete: true,
|
|
removeOnFail: true,
|
|
attempts: 1,
|
|
},
|
|
}),
|
|
BullModule.registerQueue({
|
|
name: QueueName.AI_QUEUE,
|
|
defaultJobOptions: {
|
|
removeOnComplete: true,
|
|
removeOnFail: true,
|
|
attempts: 1,
|
|
},
|
|
}),
|
|
BullModule.registerQueue({
|
|
name: QueueName.HISTORY_QUEUE,
|
|
defaultJobOptions: {
|
|
removeOnComplete: true,
|
|
removeOnFail: true,
|
|
attempts: 2,
|
|
},
|
|
}),
|
|
BullModule.registerQueue({
|
|
name: QueueName.NOTIFICATION_QUEUE,
|
|
}),
|
|
BullModule.registerQueue({
|
|
name: QueueName.AUDIT_QUEUE,
|
|
defaultJobOptions: {
|
|
removeOnComplete: true,
|
|
removeOnFail: true,
|
|
attempts: 3,
|
|
},
|
|
}),
|
|
],
|
|
exports: [BullModule],
|
|
providers: [GeneralQueueProcessor],
|
|
})
|
|
export class QueueModule {}
|