Enhance job removal logic and scheduler handling
- Updated the `safeRemoveBullJob` function to handle jobs belonging to a Job Scheduler, ensuring they are removed via `removeJobScheduler` instead of direct removal. - Introduced `removeScheduleJobSchedulers` to streamline the removal of job schedulers associated with scheduled jobs. - Improved error handling and logging for job removal operations, particularly for legacy and repeatable jobs. - Refactored related functions in both the bot and WebUI to ensure consistent handling of job scheduling and removal across the application.
This commit is contained in:
@@ -5,6 +5,8 @@ import { logger } from '../logger.js';
|
||||
* Removes a BullMQ job when present. Locked (active) jobs cannot be removed —
|
||||
* those errors are swallowed so callers can still schedule a replacement jobId
|
||||
* or rely on removeOnComplete after the worker finishes.
|
||||
* Jobs that belong to a Job Scheduler must be removed via removeJobScheduler;
|
||||
* direct remove is ignored here.
|
||||
*/
|
||||
export async function safeRemoveBullJob(
|
||||
job: Job | null | undefined,
|
||||
@@ -17,10 +19,14 @@ export async function safeRemoveBullJob(
|
||||
await job.remove();
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
if (message.includes('locked') || message.includes('could not be removed')) {
|
||||
if (
|
||||
message.includes('locked') ||
|
||||
message.includes('could not be removed') ||
|
||||
message.includes('belongs to a job scheduler')
|
||||
) {
|
||||
logger.warn(
|
||||
{ error, jobId: context?.jobId ?? job.id, label: context?.label },
|
||||
'BullMQ job is locked; skipping remove'
|
||||
'BullMQ job could not be removed directly; skipping'
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -78,6 +78,58 @@ export function scheduleJobId(scheduleId: string): string {
|
||||
return `schedule-${scheduleId}`;
|
||||
}
|
||||
|
||||
/** BullMQ delayed instance ids look like `repeat:{schedulerId}:{millis}`. */
|
||||
const REPEAT_INSTANCE_ID_RE = /^repeat:(.+):(\d+)$/;
|
||||
|
||||
function collectSchedulerIds(scheduleId: string, jobId: string | null | undefined): string[] {
|
||||
const ids = new Set<string>();
|
||||
ids.add(scheduleJobId(scheduleId));
|
||||
if (!jobId) {
|
||||
return [...ids];
|
||||
}
|
||||
|
||||
const repeatMatch = jobId.match(REPEAT_INSTANCE_ID_RE);
|
||||
if (repeatMatch?.[1]) {
|
||||
ids.add(repeatMatch[1]);
|
||||
} else if (!jobId.startsWith('repeat:')) {
|
||||
ids.add(jobId);
|
||||
}
|
||||
return [...ids];
|
||||
}
|
||||
|
||||
async function removeScheduleJobSchedulers(
|
||||
queue: typeof scheduleQueue,
|
||||
scheduleId: string,
|
||||
jobId: string | null | undefined
|
||||
): Promise<void> {
|
||||
for (const schedulerId of collectSchedulerIds(scheduleId, jobId)) {
|
||||
try {
|
||||
await queue.removeJobScheduler(schedulerId);
|
||||
} catch (error) {
|
||||
logger.warn({ schedulerId, error }, 'Failed to remove schedule job scheduler');
|
||||
}
|
||||
}
|
||||
|
||||
// Legacy / hash-keyed schedulers: match by payload scheduleId.
|
||||
try {
|
||||
const schedulers = await queue.getJobSchedulers();
|
||||
for (const scheduler of schedulers) {
|
||||
const data = scheduler.template?.data as { scheduleId?: string } | undefined;
|
||||
if (data?.scheduleId !== scheduleId) {
|
||||
continue;
|
||||
}
|
||||
const id = scheduler.id ?? scheduler.key;
|
||||
try {
|
||||
await queue.removeJobScheduler(id);
|
||||
} catch (error) {
|
||||
logger.warn({ schedulerId: id, error }, 'Failed to remove matched schedule job scheduler');
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
logger.warn({ scheduleId, error }, 'Failed to scan job schedulers for schedule');
|
||||
}
|
||||
}
|
||||
|
||||
function isValidCron(pattern: string): boolean {
|
||||
const parts = pattern.trim().split(/\s+/);
|
||||
return parts.length >= 5 && parts.length <= 6;
|
||||
@@ -230,25 +282,36 @@ export async function removeScheduleJob(
|
||||
jobId: string | null,
|
||||
options?: { scheduleId?: string; cron?: string | null }
|
||||
): Promise<void> {
|
||||
const schedulerId = options?.scheduleId ? scheduleJobId(options.scheduleId) : jobId;
|
||||
if (schedulerId && options?.cron) {
|
||||
try {
|
||||
await scheduleQueue.removeJobScheduler(schedulerId);
|
||||
} catch (error) {
|
||||
logger.warn({ schedulerId, error }, 'Failed to remove schedule job scheduler');
|
||||
if (options?.scheduleId) {
|
||||
await removeScheduleJobSchedulers(scheduleQueue, options.scheduleId, jobId);
|
||||
} else if (options?.cron && jobId) {
|
||||
// Best-effort for callers without scheduleId: extract scheduler from repeat instance id.
|
||||
for (const schedulerId of collectSchedulerIds('unknown', jobId)) {
|
||||
if (schedulerId === 'schedule-unknown') {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
await scheduleQueue.removeJobScheduler(schedulerId);
|
||||
} catch (error) {
|
||||
logger.warn({ schedulerId, error }, 'Failed to remove schedule job scheduler');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const candidateIds = new Set<string>();
|
||||
if (jobId) {
|
||||
candidateIds.add(jobId);
|
||||
// One-shot delayed jobs only — never job.remove() on repeat:* scheduler children.
|
||||
const oneShotIds = new Set<string>();
|
||||
if (jobId && !jobId.startsWith('repeat:')) {
|
||||
oneShotIds.add(jobId);
|
||||
}
|
||||
if (schedulerId) {
|
||||
candidateIds.add(schedulerId);
|
||||
if (options?.scheduleId) {
|
||||
oneShotIds.add(scheduleJobId(options.scheduleId));
|
||||
}
|
||||
|
||||
for (const id of candidateIds) {
|
||||
for (const id of oneShotIds) {
|
||||
const job = await scheduleQueue.getJob(id);
|
||||
if (job?.id && String(job.id).startsWith('repeat:')) {
|
||||
continue;
|
||||
}
|
||||
await safeRemoveBullJob(job, { label: 'schedule', jobId: id });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -48,6 +48,81 @@ function scheduleJobId(scheduleId: string): string {
|
||||
return `schedule-${scheduleId}`;
|
||||
}
|
||||
|
||||
/** BullMQ delayed instance ids look like `repeat:{schedulerId}:{millis}`. */
|
||||
const REPEAT_INSTANCE_ID_RE = /^repeat:(.+):(\d+)$/;
|
||||
|
||||
function collectSchedulerIds(scheduleId: string, jobId: string | null | undefined): string[] {
|
||||
const ids = new Set<string>();
|
||||
ids.add(scheduleJobId(scheduleId));
|
||||
if (!jobId) {
|
||||
return [...ids];
|
||||
}
|
||||
|
||||
const repeatMatch = jobId.match(REPEAT_INSTANCE_ID_RE);
|
||||
if (repeatMatch?.[1]) {
|
||||
ids.add(repeatMatch[1]);
|
||||
} else if (!jobId.startsWith('repeat:')) {
|
||||
ids.add(jobId);
|
||||
}
|
||||
return [...ids];
|
||||
}
|
||||
|
||||
async function removeQueueSchedulersForSchedule(
|
||||
scheduleId: string,
|
||||
jobId: string | null | undefined
|
||||
): Promise<void> {
|
||||
const queue = getScheduleQueue();
|
||||
|
||||
for (const schedulerId of collectSchedulerIds(scheduleId, jobId)) {
|
||||
try {
|
||||
await queue.removeJobScheduler(schedulerId);
|
||||
} catch {
|
||||
// Already gone or legacy key — keep trying other candidates.
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const schedulers = await queue.getJobSchedulers();
|
||||
for (const scheduler of schedulers) {
|
||||
const data = scheduler.template?.data as { scheduleId?: string } | undefined;
|
||||
if (data?.scheduleId !== scheduleId) {
|
||||
continue;
|
||||
}
|
||||
const id = scheduler.id ?? scheduler.key;
|
||||
try {
|
||||
await queue.removeJobScheduler(id);
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// ignore scan failures
|
||||
}
|
||||
|
||||
// Older BullMQ repeatables (pre job-scheduler API).
|
||||
try {
|
||||
const repeatables = await queue.getRepeatableJobs();
|
||||
for (const repeatable of repeatables) {
|
||||
const matchesId =
|
||||
repeatable.id === scheduleJobId(scheduleId) ||
|
||||
(jobId != null &&
|
||||
(repeatable.key === jobId ||
|
||||
(REPEAT_INSTANCE_ID_RE.test(jobId) &&
|
||||
repeatable.key.includes(jobId.match(REPEAT_INSTANCE_ID_RE)?.[1] ?? ''))));
|
||||
if (!matchesId) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
await queue.removeRepeatableByKey(repeatable.key);
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
|
||||
export async function listScheduledMessagesDashboard(guildId: string): Promise<ScheduledMessageDashboard[]> {
|
||||
const schedules = await prisma.scheduledMessage.findMany({
|
||||
where: { guildId, enabled: true },
|
||||
@@ -168,34 +243,33 @@ export async function deleteScheduledMessageDashboard(guildId: string, scheduleI
|
||||
}
|
||||
|
||||
const queue = getScheduleQueue();
|
||||
const schedulerId = scheduleJobId(existing.id);
|
||||
await removeQueueSchedulersForSchedule(existing.id, existing.jobId);
|
||||
|
||||
if (existing.cron) {
|
||||
try {
|
||||
await queue.removeJobScheduler(schedulerId);
|
||||
} catch {
|
||||
// Scheduler may already be gone (legacy repeatables / one-shots).
|
||||
}
|
||||
// One-shot delayed jobs only — never job.remove() on repeat:* scheduler children
|
||||
// (BullMQ throws code -8: belongs to a job scheduler).
|
||||
const oneShotIds = new Set<string>();
|
||||
oneShotIds.add(scheduleJobId(existing.id));
|
||||
if (existing.jobId && !existing.jobId.startsWith('repeat:')) {
|
||||
oneShotIds.add(existing.jobId);
|
||||
}
|
||||
|
||||
const candidateIds = new Set<string>();
|
||||
candidateIds.add(schedulerId);
|
||||
if (existing.jobId) {
|
||||
candidateIds.add(existing.jobId);
|
||||
}
|
||||
|
||||
for (const id of candidateIds) {
|
||||
for (const id of oneShotIds) {
|
||||
const job = await queue.getJob(id);
|
||||
if (!job) {
|
||||
if (!job || (job.id && String(job.id).startsWith('repeat:'))) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
await job.remove();
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
if (!message.includes('locked') && !message.includes('could not be removed')) {
|
||||
throw error;
|
||||
if (
|
||||
message.includes('locked') ||
|
||||
message.includes('could not be removed') ||
|
||||
message.includes('belongs to a job scheduler')
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user