Stopping background tasks in NestJS: scheduler, BullMQ, outbox

Node Backend

How to correctly stop background tasks on SIGTERM in NestJS: the scheduler finishes its iteration, BullMQ waits for active jobs, the outbox finishes the current batch via SKIP LOCKED and checks the draining flag.

Published 28 июня 2026 ~6 min read Author: Вадим Викулин

← Back to the section

When Kubernetes sends SIGTERM, NestJS starts stopping. For HTTP requests this looks tidy: the server stops accepting new connections and waits for the old ones to finish. But background tasks — @nestjs/schedule, BullMQ, the outbox relay — do not stop on their own. If you kill them in the middle, you get an inconsistency: the transaction rolled back, but the Kafka message has already gone out.

The right approach: let each task finish its current iteration and not start the next one.

Scheduler: NestJS has no built-in waiting

@nestjs/schedule runs methods on a schedule, but does not provide a way to know whether a method is running right now. SchedulerRegistry.deleteInterval() simply cancels the next tick — if the method is already running, it will continue executing, and no one will wait for it.

The solution: store the Promise of the current iteration in a class field and wait for it in the beforeApplicationShutdown hook.

@Injectable()
export class OutboxRelayService implements BeforeApplicationShutdown {
  private runningJob: Promise<void> | null = null;

  constructor(
    private readonly schedulerRegistry: SchedulerRegistry,
    private readonly shutdownState: ShutdownStateService,
    private readonly pool: Pool,
    private readonly producer: KafkaProducerService,
  ) {}

  @Interval('outbox-relay', 500)
  async relay(): Promise<void> {
    if (this.shutdownState.isDraining()) return;

    this.runningJob = this.processBatch();
    await this.runningJob;
    this.runningJob = null;
  }

  async beforeApplicationShutdown(): Promise<void> {
    if (this.schedulerRegistry.doesExist('interval', 'outbox-relay')) {
      this.schedulerRegistry.deleteInterval('outbox-relay');
    }
    if (this.runningJob) {
      await Promise.race([this.runningJob, timeout(25_000)]);
    }
  }
}

What happens on SIGTERM:

NestJS calls beforeApplicationShutdown on all Injectables.
deleteInterval — the next tick won't run anymore.
If processBatch() is running right now — await this.runningJob waits for it to finish (25 seconds at most).
The current batch runs to the end; the next one does not start.

Why you can't do while(true) inside @Interval

It is tempting to write this:

@Interval('relay', 100)
async relay(): Promise<void> {
  while (true) {
    await this.processBatch();
  }
}

The problem: such a loop never finishes. beforeApplicationShutdown will hang on await this.runningJob until the timeout expires, and the batch will be interrupted in the middle.

The right approach is short iterations with a stop-flag check:

@Interval('relay', 500)
async relay(): Promise<void> {
  if (this.shutdownState.isDraining()) return;
  this.runningJob = this.processBatch();
  await this.runningJob;
  this.runningJob = null;
}

After deleteInterval the next tick won't run. The current batch finishes completely.

Outbox relay: transaction and SKIP LOCKED

The outbox relay reads events from the database and publishes them to Kafka. For a correct stop it is important that the current batch finishes atomically.

private async processBatch(): Promise<void> {
  const client = await this.pool.connect();
  try {
    await client.query('BEGIN');
    const { rows: batch } = await client.query<OutboxEvent>(
      `SELECT id, topic, partition_key AS "partitionKey", payload
         FROM order_outbox
        WHERE published_at IS NULL
        ORDER BY id
        LIMIT 50
          FOR UPDATE SKIP LOCKED`,
    );
    const publishedIds: string[] = [];
    for (const event of batch) {
      await this.producer.send(event.topic, event.partitionKey, event.payload);
      publishedIds.push(event.id);
    }
    if (publishedIds.length > 0) {
      await client.query(
        `UPDATE order_outbox SET published_at = NOW() WHERE id = ANY($1::uuid[])`,
        [publishedIds],
      );
    }
    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

FOR UPDATE SKIP LOCKED inside an explicit transaction (BEGIN/COMMIT) locks the rows until COMMIT: another pod doing a parallel SELECT will skip them via SKIP LOCKED. If you do a pool.query() without BEGIN, that is autocommit — the lock is released right after the SELECT, and another pod can take the same rows.

BullMQ: wait for active jobs, don't interrupt

BullMQ provides two ways to close a worker:

worker.close() — a soft close: waits for all active jobs.
worker.close(true) — an immediate interrupt: the job is stopped in the middle, its state is unknown.

Always use worker.close() with no arguments:

@Injectable()
export class PaymentWorkerService implements OnApplicationBootstrap, BeforeApplicationShutdown {
  private worker!: Worker;

  constructor(
    private readonly shutdownState: ShutdownStateService,
    private readonly paymentService: PaymentService,
  ) {}

  onApplicationBootstrap(): void {
    this.worker = new Worker(
      'payment-jobs',
      async (job: Job<PaymentJobData>) => {
        if (this.shutdownState.isDraining()) {
          await job.moveToFailed(new Error('draining'), job.token ?? '');
          return;
        }
        await this.paymentService.processCharge(job.data);
      },
      { connection: redisConnection, concurrency: 5 },
    );
  }

  async beforeApplicationShutdown(): Promise<void> {
    await Promise.race([
      this.worker.close(),
      timeout(20_000),
    ]);
  }
}

If the application has started stopping (isDraining()), a new job is immediately moved to the Failed state — another pod will pick it up. Active jobs run to the end.

Idempotency of jobs

BullMQ may retry a job after a restart. If a job charges money, then without an idempotent key a double charge occurs. An idempotencyKey field in the job data is mandatory:

interface PaymentJobData {
  customerId: string;
  orderId: string;
  amountKopecks: number;
  idempotencyKey: string;
}

The payment service checks this key before executing — a repeated call with the same key returns the result without charging again.

A long process: AbortSignal as a stop flag

Sometimes a task iterates over a large list of records — a catalog sync, a data migration. If you don't react to the stop signal, the task will keep working until the timeout.

@Injectable()
export class ProductSyncService implements BeforeApplicationShutdown {
  private abortController = new AbortController();

  async beforeApplicationShutdown(): Promise<void> {
    this.abortController.abort();
  }

  async syncProductCatalog(): Promise<void> {
    const products = await this.fetchExternalProducts();

    for (const product of products) {
      if (this.abortController.signal.aborted) break;

      await this.productRepository.upsert(product);
      await this.searchIndexer.index(product);
    }
  }
}

On SIGTERM: abort() is set, the next loop iteration is interrupted. A new pod will continue from the first unprocessed record.

Time budget

Kubernetes gives 60 seconds for a full pod stop (by default). The phases overlap:

SIGTERM
  │
  ├─ preStop sleep 10s        (k8s endpoint drain)
  ├─ HTTP drain       ≤ 25s   (server.close, wait for requests)
  ├─ Scheduler/BullMQ ≤ 20s   (await runningJob / worker.close)
  ├─ Kafka consumer   ≤ 20s   (consumer.disconnect)
  └─ pool.end()               (after the drain, in onApplicationShutdown)
                      ──────
                      ≤ 60s = terminationGracePeriodSeconds

HTTP drain and scheduler/BullMQ run in parallel — the actual time is not summed. But each phase must not exceed its own limit.

Common mistakes

clearInterval without waiting for the iteration. The task is killed in the middle, the transaction rolled back, but the Kafka message has already gone out. You need: deleteInterval plus await runningJob.

worker.close(true) instead of worker.close(). An immediate interrupt leaves the job in an unknown state. Use the soft close.

while (true) inside an @Interval method. The shutdown will hang on the timeout, the batch interrupted in the middle. Use short iterations with an isDraining() check.

pool.end() in beforeApplicationShutdown. The database is closed before the HTTP requests are drained — active transactions are cut off. Close the pool in onApplicationShutdown, which is called after beforeApplicationShutdown.

In short

@nestjs/schedule does not wait for the current iteration on its own — you need to store the Promise and wait for it in beforeApplicationShutdown.
SchedulerRegistry.deleteInterval() cancels the next tick but does not interrupt the current one.
BullMQ: worker.close() waits for active jobs; worker.close(true) is an immediate interrupt and must not be used.
The outbox relay finishes the current batch atomically via FOR UPDATE SKIP LOCKED; it does not start the next one.
The loop in the relay method checks isDraining() — not while (true).
Long processes react to AbortSignal and interrupt at checkpoints.
Jobs with money operations require an idempotencyKey — a repeat must not produce a double action.
The total stop budget is 60 seconds; the database pool is closed last, in onApplicationShutdown.