nestjs-inflight-tracker

Drain in-flight work across NestJS transports before the process exits.

📦 nestjs-inflight-tracker on npm

What it does

Most graceful-shutdown patterns either (a) sleep a fixed timer and hope, or (b) close connections and lose mid-flight messages. This package gives you a single counter every async surface reports into, plus one shared drain promise that connection-managing services can await before closing their clients.

On SIGTERM / SIGINT, the tracker flips isShuttingDown() immediately, aborts its AbortSignal, and starts draining in-flight work. NestJS then runs onModuleDestroy; services that own connections should call tracker.drained() there before disconnecting. The drain resolves as soon as the counter reaches zero, or after the configured timeout.

Install

yarn add nestjs-inflight-tracker

Peer deps: @nestjs/common ^10, rxjs ^7. @nestjs/schedule ^4 is optional (only needed for @TrackedCron).

Usage

import { ShutdownModule } from 'nestjs-inflight-tracker';

@Module({
  imports: [ShutdownModule.forRoot()],
})
export class AppModule {}

forRoot() reads SHUTDOWN_DRAIN_TIMEOUT_MS from env (default 25000) and disables itself when NODE_ENV === 'dev'.

forRoot() is a process-wide singleton: importing it from multiple modules (e.g. a shared module and a TCP shared module) always yields the same tracker — one counter set, one drain promise, one pair of signal handlers. The first call's options win; a later call with different options logs a warning and is ignored. Tests can call ShutdownModule.resetForTesting() between cases.

Make sure the NestJS app enables shutdown hooks:

const app = await NestFactory.create(AppModule);
app.enableShutdownHooks();
await app.listen(3000);

Transport support

The tracker itself contains zero transport code — it is a counter, a drain promise, and an AbortSignal. That makes handler tracking work on every transport NestJS ships (Redis, MQTT, NATS, RabbitMQ, Kafka, gRPC, TCP, and custom transporters), because NestJS interceptors are transport-agnostic: register InflightInterceptor as an APP_INTERCEPTOR and it wraps every microservice handler, and its RxJS stream completes exactly when the handler truly finishes.

Why the interceptor, and not the transport server? On event-path messages (@EventPattern, or @MessagePattern invoked without a reply-to), a server's handleMessage resolves when the handler is dispatched, not when it completes. Counting there makes the drain see zero in-flight work while handlers are still mid-write — the drain resolves, your app closes its DB connections, and the still-running handlers fail against closed clients. The interceptor is the only completion-accurate hook.

What the library deliberately does not do is broker-side flow control (cancel/pause/unsubscribe on shutdown) and acknowledgement handling — those live in each transport's server/consumer, and whether an interrupted message is redelivered is a property of the broker, not of this package:

| Transport | Handler tracking | Unfinished message on pod kill | Shutdown gate you should add | | --- | --- | --- | --- | | RabbitMQ | InflightInterceptor | ✅ unacked → requeued by broker | skip handler in a custom ServerRMQ once isShuttingDown() (see below) | | Kafka | InflightInterceptor | ✅ uncommitted offset → redelivered after rebalance | consumer.pause() on tracker.signal() abort | | NATS (core) / Redis pub-sub | InflightInterceptor | ❌ fire-and-forget — the broker never redelivers | none possible at this layer; use JetStream / Redis Streams if loss matters | | MQTT | InflightInterceptor | depends on QoS (1/2 redeliver, 0 doesn't) | unsubscribe on shutdown | | gRPC | InflightInterceptor | ✅ caller receives ShuttingDownError; the client treats it as one failed call and re-resolves | built in — the interceptor rejects new RPCs | | TCP (ServerTCP) | InflightInterceptor | ⚠️ refusing is not enough — the caller holds one long-lived socket pinned to this pod, so it is refused for the whole grace period | subclass ServerTCP and destroy established sockets on close (see below) | | HTTP | manual (framework hooks) | ✅ LB retries once readiness fails | count requests in onRequest/onResponse hooks + fail readiness when isShuttingDown() | | Custom transporter | InflightInterceptor or track() | whatever your broker guarantees | mirror the RabbitMQ pattern |

Pattern: gating a message transport (RabbitMQ example)

Two rules, learned the hard way:

1. Skip — don't nack — deliveries that arrive after shutdown began. Client-buffered messages keep arriving between SIGTERM and consumer cancellation. Running them risks executing against connections your onModuleDestroy chain is about to close. But an explicit nack(requeue: true) is worse: with the consumer still attached the broker redelivers instantly, you nack again, and the loop burns through a quorum queue's delivery limit straight into the dead-letter queue. Leaving the message unacked requeues it exactly once when the channel closes.
2. Ack only after the handler completes (interceptor tap → ack), so an interrupted handler's message is redelivered instead of lost.

export class GracefulServerRMQ extends ServerRMQ {
  private tracker: InflightTracker | null = null;

  setInflightTracker(tracker: InflightTracker) {
    this.tracker = tracker; // call from main.ts: strategy.setInflightTracker(app.get(InflightTracker))
  }

  public async handleMessage(message: Record<string, any>, channel: Channel) {
    if (this.tracker?.isShuttingDown()) {
      return; // unacked → broker requeues it on channel close
    }
    return super.handleMessage(message, channel);
  }
}

The same shape ports to any transport: expose a setInflightTracker on your custom server, check isShuttingDown() before dispatching a delivery, and use tracker.signal()'s abort event to pause/cancel the consumer eagerly.

Pattern: hanging up a connection-based transport (TCP example)

Refusing new work is only half of a graceful stop on TCP, and the missing half loses messages. Nest's ClientProxy opens one long-lived socket per target service and reuses it for every call, and Kubernetes load-balances connections, not requests. A caller that connected minutes ago is therefore pinned to this pod: removing the pod from the Service's endpoints changes nothing for it, and every call it makes during the grace period comes back ShuttingDownError. That live socket is also a handle Node counts, so the process does not exit until the kubelet kills it — the refusal window is exactly as long as terminationGracePeriodSeconds, no matter how fast the drain finished.

super.close() does not fix it: that stops new connections only. Nor is closeAllConnections() an option — it is an http.Server method, and ServerTCP runs a raw net.Server. Hold the sockets yourself:

export class GracefulServerTCP extends ServerTCP {
  private readonly sockets = new Set<Socket>();
  private tracker: InflightTracker | null = null;

  setInflightTracker(tracker: InflightTracker) {
    this.tracker = tracker; // call from main.ts, as with GracefulServerRMQ
  }

  bindHandler(socket: Socket) {
    this.sockets.add(socket);
    socket.on('close', () => this.sockets.delete(socket));
    super.bindHandler(socket);
  }

  async close() {
    await this.tracker?.drained();
    for (const socket of this.sockets) socket.destroy(); // hang up
    this.sockets.clear();
    super.close();
  }
}

The caller's client reconnects on its next send and lands on a healthy pod. Pair it with an exception filter that maps ShuttingDownError to a retryable status (503), so calls refused inside the drain window are retried instead of being reported to the caller as a server fault.

Production contract

During shutdown, this package:

1. Flips isShuttingDown() and aborts tracker.signal() as soon as SIGTERM or SIGINT is received.
2. Stops new tracked work at supported entrypoints: track() returns undefined, TrackedCron skips the tick, and InflightInterceptor throws Error('shutting down').
3. Lets already tracked work finish while NestJS is still running onModuleDestroy.
4. Lets connection-owning services await tracker.drained() before closing SQL, Redis, RMQ, TCP, Cassandra, or other clients.

The drain is capped by SHUTDOWN_DRAIN_TIMEOUT_MS or timeoutMs. If work is still running after the timeout, the tracker logs the remaining counters and allows shutdown to continue. It does not cancel or kill the work itself.

Use a drain timeout below the platform grace period. For example, with Kubernetes terminationGracePeriodSeconds: 30, keep SHUTDOWN_DRAIN_TIMEOUT_MS below 30000, such as 25000.

Only work that reports into the tracker is counted. Use track(), InflightInterceptor, TrackedCron, or manual start() / end() calls at the async surfaces you want to drain.

⚠️ track() returns undefined during shutdown. Once isShuttingDown() is true, track(label, fn) skips fn entirely and resolves to undefined — it does not throw. If you do anything with the return value (const res = await tracker.track(...)), you MUST handle the undefined case, and if skipping the work is not acceptable at that call site, check isShuttingDown() yourself and fail explicitly instead. This is deliberate: entrypoints that must signal rejection to a caller should use InflightInterceptor (which throws ShuttingDownError for the caller to map to a retryable status).

Track an arbitrary async block

import { InflightTracker } from 'nestjs-inflight-tracker';

constructor(private readonly tracker: InflightTracker) {}

await this.tracker.track('postmark-pagination', async () => {
  while (hasMore) {
    if (this.tracker.isShuttingDown()) break;
    await this.fetchPage();
  }
});

Wait before closing connections

import { Injectable, OnModuleDestroy } from '@nestjs/common';
import { InflightTracker } from 'nestjs-inflight-tracker';

@Injectable()
export class DatabaseService implements OnModuleDestroy {
  constructor(private readonly tracker: InflightTracker) {}

  async onModuleDestroy() {
    await this.tracker.drained();
    await this.disconnect();
  }

  private async disconnect() {
    // close SQL / Redis / RMQ / TCP / Cassandra clients here
  }
}

drained() is idempotent. Every service awaits the same drain promise, so the app does not sleep once per dependency.

Track every RPC handler (TCP / RMQ controllers)

import { APP_INTERCEPTOR } from '@nestjs/core';
import { InflightInterceptor } from 'nestjs-inflight-tracker';

@Module({
  imports: [ShutdownModule.forRoot()],
  providers: [{ provide: APP_INTERCEPTOR, useClass: InflightInterceptor }],
})
export class AppModule {}

When shutting down, the interceptor short-circuits with ShuttingDownError. Map it to a retryable status in your exception filter — and on a connection-based transport, close established connections too (see the TCP pattern above), or the caller stays pinned to this pod and is refused for the whole grace period.

Track scheduled jobs

import { CronExpression } from '@nestjs/schedule';
import { TrackedCron, InflightTracker } from 'nestjs-inflight-tracker';

class CleanupService {
  constructor(public readonly inflightTracker: InflightTracker) {}

  @TrackedCron(CronExpression.EVERY_MINUTE, { label: 'cleanup' })
  async run() {
    // ...
  }
}

The decorator requires the host class to expose the tracker as inflightTracker (override via trackerProp). Ticks scheduled during shutdown are skipped; ticks already in flight are awaited.

Bail axios / fetch / loops on shutdown

const res = await axios.get(url, { signal: this.tracker.signal() });

The signal aborts as soon as shutdown begins.

Configuration

| Option | Default | Description | | --- | --- | --- | | SHUTDOWN_DRAIN_TIMEOUT_MS | 25000 | Environment fallback for timeoutMs. | | timeoutMs | 25000 | Maximum time to wait for in-flight work to reach zero. | | pollMs | 200 | Polling interval while waiting for the counter to drain. | | disabled | NODE_ENV === 'dev' | Disables signal handling and drain waiting. | | signals | ['SIGTERM', 'SIGINT'] | Process signals that begin shutdown. |

Dependency model

This is a NestJS library, so NestJS and RxJS are declared as peer dependencies. The consuming app should provide those packages:

{
  "peerDependencies": {
    "@nestjs/common": "^10.0.0",
    "@nestjs/schedule": "^4.0.0",
    "rxjs": "^7.0.0"
  }
}

That keeps the app on one NestJS dependency graph and avoids bundling a second copy of framework-level packages.

The same packages can also appear in devDependencies. That is expected: this repository needs local copies to compile, run Jest tests, and build dist/, but consumers should still resolve them from their own app install.

{
  "devDependencies": {
    "@nestjs/common": "^10.4.1",
    "@nestjs/core": "^10.4.6",
    "@nestjs/schedule": "^4.1.1",
    "@nestjs/testing": "^10.4.1",
    "rxjs": "^7.8.1",
    "typescript": "^5.4.4"
  }
}

@nestjs/schedule is marked optional because it is only needed when using TrackedCron. If an app imports TrackedCron, it must have @nestjs/schedule installed.

Why this exists

NestJS's enableShutdownHooks() fires onModuleDestroy before it disposes transports and connections. That is the useful window for graceful draining, but each module only sees its own lifecycle. Common workarounds like await delay(25000) in every connection module are slow when the app is idle and still unreliable when work runs longer than the sleep. This package centralizes the in-flight state so connection modules can wait for actual work, capped by a timeout below the platform grace period.