fix(queue): remove the dead {search-queue} — producers with no consumer (#379)
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>
This commit is contained in:
@@ -4,7 +4,6 @@ import { EventName } from '../../common/events/event.contants';
|
||||
import { InjectQueue } from '@nestjs/bullmq';
|
||||
import { QueueJob, QueueName } from '../../integrations/queue/constants';
|
||||
import { Queue } from 'bullmq';
|
||||
import { EnvironmentService } from '../../integrations/environment/environment.service';
|
||||
|
||||
/**
|
||||
* Thin snapshot of a page node carried inside domain events so the WebSocket
|
||||
@@ -112,48 +111,24 @@ export class PageListener {
|
||||
private readonly logger = new Logger(PageListener.name);
|
||||
|
||||
constructor(
|
||||
private readonly environmentService: EnvironmentService,
|
||||
@InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
|
||||
@InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
|
||||
) {}
|
||||
|
||||
@OnEvent(EventName.PAGE_CREATED)
|
||||
async handlePageCreated(event: PageEvent) {
|
||||
const { pageIds, workspaceId } = event;
|
||||
if (this.isTypesense()) {
|
||||
await this.searchQueue.add(QueueJob.PAGE_CREATED, {
|
||||
pageIds,
|
||||
});
|
||||
}
|
||||
|
||||
await this.aiQueue.add(QueueJob.PAGE_CREATED, { pageIds, workspaceId });
|
||||
}
|
||||
|
||||
@OnEvent(EventName.PAGE_UPDATED)
|
||||
async handlePageUpdated(event: PageEvent) {
|
||||
const { pageIds } = event;
|
||||
|
||||
await this.searchQueue.add(QueueJob.PAGE_UPDATED, { pageIds });
|
||||
}
|
||||
|
||||
@OnEvent(EventName.PAGE_DELETED)
|
||||
async handlePageDeleted(event: PageEvent) {
|
||||
const { pageIds, workspaceId } = event;
|
||||
if (this.isTypesense()) {
|
||||
await this.searchQueue.add(QueueJob.PAGE_DELETED, { pageIds });
|
||||
}
|
||||
|
||||
await this.aiQueue.add(QueueJob.PAGE_DELETED, { pageIds, workspaceId });
|
||||
}
|
||||
|
||||
@OnEvent(EventName.PAGE_SOFT_DELETED)
|
||||
async handlePageSoftDeleted(event: PageEvent) {
|
||||
const { pageIds, workspaceId } = event;
|
||||
|
||||
if (this.isTypesense()) {
|
||||
await this.searchQueue.add(QueueJob.PAGE_SOFT_DELETED, { pageIds });
|
||||
}
|
||||
|
||||
await this.aiQueue.add(QueueJob.PAGE_SOFT_DELETED, {
|
||||
pageIds,
|
||||
workspaceId,
|
||||
@@ -163,14 +138,6 @@ export class PageListener {
|
||||
@OnEvent(EventName.PAGE_RESTORED)
|
||||
async handlePageRestored(event: PageEvent) {
|
||||
const { pageIds, workspaceId } = event;
|
||||
if (this.isTypesense()) {
|
||||
await this.searchQueue.add(QueueJob.PAGE_RESTORED, { pageIds });
|
||||
}
|
||||
|
||||
await this.aiQueue.add(QueueJob.PAGE_RESTORED, { pageIds, workspaceId });
|
||||
}
|
||||
|
||||
isTypesense(): boolean {
|
||||
return this.environmentService.getSearchDriver() === 'typesense';
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import { EventName } from '../../common/events/event.contants';
|
||||
import { InjectQueue } from '@nestjs/bullmq';
|
||||
import { QueueJob, QueueName } from '../../integrations/queue/constants';
|
||||
import { Queue } from 'bullmq';
|
||||
import { EnvironmentService } from '../../integrations/environment/environment.service';
|
||||
|
||||
export class SpaceEvent {
|
||||
spaceId: string;
|
||||
@@ -15,22 +14,12 @@ export class SpaceListener {
|
||||
private readonly logger = new Logger(SpaceListener.name);
|
||||
|
||||
constructor(
|
||||
private readonly environmentService: EnvironmentService,
|
||||
@InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
|
||||
@InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
|
||||
) {}
|
||||
|
||||
@OnEvent(EventName.SPACE_DELETED)
|
||||
async handleSpaceDeleted(event: SpaceEvent) {
|
||||
const { spaceId } = event;
|
||||
if (this.isTypesense()) {
|
||||
await this.searchQueue.add(QueueJob.SPACE_DELETED, { spaceId });
|
||||
}
|
||||
|
||||
await this.aiQueue.add(QueueJob.SPACE_DELETED, { spaceId });
|
||||
}
|
||||
|
||||
isTypesense(): boolean {
|
||||
return this.environmentService.getSearchDriver() === 'typesense';
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import { EventName } from '../../common/events/event.contants';
|
||||
import { InjectQueue } from '@nestjs/bullmq';
|
||||
import { QueueJob, QueueName } from '../../integrations/queue/constants';
|
||||
import { Queue } from 'bullmq';
|
||||
import { EnvironmentService } from '../../integrations/environment/environment.service';
|
||||
|
||||
export class WorkspaceEvent {
|
||||
workspaceId: string;
|
||||
@@ -15,22 +14,12 @@ export class WorkspaceListener {
|
||||
private readonly logger = new Logger(WorkspaceListener.name);
|
||||
|
||||
constructor(
|
||||
private readonly environmentService: EnvironmentService,
|
||||
@InjectQueue(QueueName.SEARCH_QUEUE) private searchQueue: Queue,
|
||||
@InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
|
||||
) {}
|
||||
|
||||
@OnEvent(EventName.WORKSPACE_DELETED)
|
||||
async handlePageDeleted(event: WorkspaceEvent) {
|
||||
const { workspaceId } = event;
|
||||
if (this.isTypesense()) {
|
||||
await this.searchQueue.add(QueueJob.WORKSPACE_DELETED, { workspaceId });
|
||||
}
|
||||
|
||||
await this.aiQueue.add(QueueJob.WORKSPACE_DELETED, { workspaceId });
|
||||
}
|
||||
|
||||
isTypesense(): boolean {
|
||||
return this.environmentService.getSearchDriver() === 'typesense';
|
||||
}
|
||||
}
|
||||
|
||||
@@ -46,7 +46,6 @@ export class MetricsBullService implements OnModuleInit, OnModuleDestroy {
|
||||
@InjectQueue(QueueName.GENERAL_QUEUE) generalQueue: Queue,
|
||||
@InjectQueue(QueueName.BILLING_QUEUE) billingQueue: Queue,
|
||||
@InjectQueue(QueueName.FILE_TASK_QUEUE) fileTaskQueue: Queue,
|
||||
@InjectQueue(QueueName.SEARCH_QUEUE) searchQueue: Queue,
|
||||
@InjectQueue(QueueName.AI_QUEUE) aiQueue: Queue,
|
||||
@InjectQueue(QueueName.HISTORY_QUEUE) historyQueue: Queue,
|
||||
@InjectQueue(QueueName.NOTIFICATION_QUEUE) notificationQueue: Queue,
|
||||
@@ -58,7 +57,6 @@ export class MetricsBullService implements OnModuleInit, OnModuleDestroy {
|
||||
{ label: 'general', queue: generalQueue },
|
||||
{ label: 'billing', queue: billingQueue },
|
||||
{ label: 'file-task', queue: fileTaskQueue },
|
||||
{ label: 'search', queue: searchQueue },
|
||||
{ label: 'ai', queue: aiQueue },
|
||||
{ label: 'history', queue: historyQueue },
|
||||
{ label: 'notification', queue: notificationQueue },
|
||||
|
||||
@@ -4,7 +4,6 @@ export enum QueueName {
|
||||
GENERAL_QUEUE = '{general-queue}',
|
||||
BILLING_QUEUE = '{billing-queue}',
|
||||
FILE_TASK_QUEUE = '{file-task-queue}',
|
||||
SEARCH_QUEUE = '{search-queue}',
|
||||
AI_QUEUE = '{ai-queue}',
|
||||
HISTORY_QUEUE = '{history-queue}',
|
||||
NOTIFICATION_QUEUE = '{notification-queue}',
|
||||
@@ -32,12 +31,6 @@ export enum QueueJob {
|
||||
IMPORT_TASK = 'import-task',
|
||||
EXPORT_TASK = 'export-task',
|
||||
|
||||
SEARCH_INDEX_PAGE = 'search-index-page',
|
||||
SEARCH_INDEX_PAGES = 'search-index-pages',
|
||||
SEARCH_INDEX_COMMENT = 'search-index-comment',
|
||||
SEARCH_INDEX_COMMENTS = 'search-index-comments',
|
||||
SEARCH_INDEX_ATTACHMENT = 'search-index-attachment',
|
||||
SEARCH_INDEX_ATTACHMENTS = 'search-index-attachments',
|
||||
SEARCH_REMOVE_PAGE = 'search-remove-page',
|
||||
SEARCH_REMOVE_ASSET = 'search-remove-attachment',
|
||||
SEARCH_REMOVE_FACE = 'search-remove-comment',
|
||||
|
||||
@@ -57,14 +57,6 @@ import { GeneralQueueProcessor } from './processors/general-queue.processor';
|
||||
attempts: 1,
|
||||
},
|
||||
}),
|
||||
BullModule.registerQueue({
|
||||
name: QueueName.SEARCH_QUEUE,
|
||||
defaultJobOptions: {
|
||||
removeOnComplete: true,
|
||||
removeOnFail: true,
|
||||
attempts: 2,
|
||||
},
|
||||
}),
|
||||
BullModule.registerQueue({
|
||||
name: QueueName.AI_QUEUE,
|
||||
defaultJobOptions: {
|
||||
|
||||
Reference in New Issue
Block a user