The previous post was the infrastructure: a FIFO topic per service, a FIFO queue per worker, cross-region subscriptions, and a set of deny statements that took longer than all the rest of it. That got us a place to put messages. It did not get us a service that could send one.

Publishing to SNS from a Node service is about six lines with the AWS SDK. The problem is that it is six lines per service, and each copy has to decide the same eight things: how to find the topic ARN, whether to batch, what to do about the 256 KB limit, what happens to a message that is mid-flight when ECS sends SIGTERM, whether a failure throws or logs, what the group ID is, how to run any of it on a laptop. Left alone, that becomes eight different answers and one of them is wrong in a way nobody notices for a quarter.

So: a package. Two objects, a publisher and a consumer, one NestJS module each wrapping them for the services that want the framework version. The code below is written out fresh for this post, but the decisions are the ones we argued about.

publish returns nothing

The first and most contested decision. The publisher's method signature is:

publish(...events: DomainEvent[]): void

No promise. Nothing to await. The call adds the event to an in-memory batch and returns, and the batch goes to SNS later — when it reaches ten entries, when it reaches 256 KB, or when a flush timer fires, whichever happens first. Failures go to an onError callback supplied at construction, not to the caller.

The case for it: SNS's PublishBatch takes up to ten entries and 256 KB total. An HTTP handler that updates a subscription and emits three events should pay for one network round trip, not three, and it should not pay for any of them inside the request. Ten batched publishes cost one call; ten awaited publishes cost ten, serialised, in the middle of a request somebody is waiting on.

The case against it, which is real: an event that is sitting in a batch has not been published, and a process that dies between the two is a process that silently dropped it. There is no acknowledgement to the caller, so there is nothing the caller could have done differently.

We took the fire-and-forget version, with two mitigations. The flush interval is ten seconds, so the exposure window is bounded and small. And shutdown drains the batch properly, which turned out to be most of the interesting code.

I still re-litigate this one with myself. If I were starting again I would probably ship both — publish() for the common case and an awaitable variant for the caller who has just written to the database and genuinely needs to know. What I would not do is make the awaitable one the default and let every handler pay for it.

The two sets of promises

Here is the part that is genuinely fiddly, and the reason a naive await Promise.all(everything) at shutdown does not work.

There are two asynchronous stages, not one. Preparing an event can be async, because an oversized payload has to go to object storage first (below). Sending a batch is async. So at any moment the publisher has some events still being prepared, and some batches in flight. Draining is ordered: finish preparing, then flush what preparation produced, then wait for the sends.

Which needs a small structure that holds a set of in-flight promises and forgets each one as it settles, so it does not grow without bound on a long-lived process:

class PendingSet {
  private readonly inFlight = new Set<Promise<unknown>>()

  track(work: Promise<unknown>): void {
    this.inFlight.add(work)
    // `finally` rather than `then`, so a rejection still releases the slot.
    // The rejection itself is handled where the work was created.
    void work.finally(() => this.inFlight.delete(work))
  }

  async settle(): Promise<void> {
    await Promise.allSettled(this.inFlight)
  }
}

allSettled rather than all, because this is a drain and not a barrier: one failed send must not stop us waiting for the other four.

And then the drain itself:

async shutdown(): Promise<void> {
  this.accepting = false
  await this.preparing.settle() // everything has reached the batch
  this.flushNow() //             ...and the batch is on its way
  await this.sending.settle() //  ...and has arrived
}

The other thing worth saying out loud: the method that appends to the batch is deliberately not async.

private append(entry: BatchEntry): void {
  if (this.batch.length >= MAX_ENTRIES || this.bytes + entry.bytes > MAX_BYTES) {
    this.flushNow()
  }

  this.batch.push(entry)
  this.bytes += entry.bytes

  this.timer ??= setTimeout(() => this.flushNow(), this.flushIntervalMs)
}

Node is single-threaded, which people take to mean this kind of thing is safe. It means the opposite of that as soon as there is an await in the middle. Put one between the size check and the push and you have handed control back to the event loop between deciding there is room and taking it; two events prepared concurrently both see room, both push, and the batch goes out at eleven entries and is rejected whole. The check and the mutation have to be in the same synchronous block. It is the only invariant in this file I would put a comment on, and I did.

The event that does not fit

SNS caps a message at 256 KB. Most of ours are a few hundred bytes. A small number — an import summary, a bulk operation result — are not, and the answer of "don't do that" was not going to survive contact with the roadmap.

So: anything over the limit is written to a bucket, and what gets published is the key. A message attribute tells the consumer which of the two it is holding.

const enum Payload {
  Inline = 'inline',
  Stored = 'stored',
}

private async prepare(event: DomainEvent): Promise<BatchEntry> {
  const id = randomUUID()
  const body = JSON.stringify(event)

  const entry: PublishBatchRequestEntry = {
    Id: id,
    Message: body,
    // A FIFO topic requires both. A fresh UUID for the group means this
    // message is ordered against nothing, which is the default we chose and
    // the caller can override with something meaningful.
    MessageGroupId: event.groupId ?? id,
    MessageDeduplicationId: id,
    MessageAttributes: {
      payload: { DataType: 'String', StringValue: Payload.Inline },
    },
  }

  if (Buffer.byteLength(body) <= MAX_BYTES) {
    return { entry, bytes: Buffer.byteLength(body) }
  }

  if (Buffer.byteLength(body) > MAX_STORED_BYTES) {
    throw new PublishError(`event exceeds the ${MAX_STORED_BYTES} byte ceiling`)
  }

  const key = `${this.topicName}/${id}`
  await this.store.put(key, body)

  entry.Message = key
  entry.MessageAttributes.payload.StringValue = Payload.Stored

  return { entry, bytes: Buffer.byteLength(key) }
}

Two things about that.

There is still a ceiling — ten megabytes — and going over it throws rather than uploading. A queue is not a file transfer mechanism, and a service that wants to move a hundred megabytes between two places should be told so at the point it tries, not by an operator reading a bucket's lifecycle metrics in six months.

And the whole scheme depends on raw_message_delivery = true on every subscription, which was the previous post's one non-negotiable. Without raw delivery, SNS wraps the payload in an envelope and puts the message attributes inside it, so the consumer would have to parse the body to find out how to parse the body. With it, the attribute is an SQS message attribute and the decision is made before anything is deserialised.

The consumer, and the two hours it avoids

The consumer side is a long-poll loop — twenty second wait, ten messages a receive — and a table of handlers keyed by event type. Registration is a decorator, so a handler is a class with a method and no wiring:

@Handles(['subscription.upgraded', 'subscription.cancelled'])
async onSubscriptionChange(event: DomainEvent<SubscriptionPayload>) { ... }

Dispatch runs every registered handler for the type and collects results rather than short-circuiting:

const handlers = this.table.for(event.type)
if (handlers.length === 0) {
  // Not an error. A queue subscribed to a whole topic will receive types it
  // was never interested in, and treating that as a failure means a service
  // dead-letters its neighbours' events.
  this.log.debug(`no handler for ${event.type}`)
  return
}

const results = await Promise.allSettled(handlers.map((h) => h.handle(event)))
if (results.some((r) => r.status === 'rejected')) {
  throw new ConsumeError(`one or more handlers failed for ${event.type}`)
}

allSettled again, and for a sharper reason than in the publisher. With Promise.all, the first rejection resolves the outer promise while the other handlers are still running, and their eventual rejections become unhandled. You lose the error you needed to read in order to understand the one you got.

Now the part I like. Recall from the infrastructure post that the visibility timeout is two hours and maxReceiveCount is 1. A handler that throws should end up in the dead-letter queue — but if we simply do not delete the message, it sits invisible for two hours before SQS notices the receive count and moves it. Two hours in which the failure exists but nothing anywhere shows it.

So the consumer resets the visibility of anything it failed to process, in a batch, before it acknowledges the rest:

private async onBatch(messages: Message[]): Promise<Message[]> {
  const done: Message[] = []

  for (const message of messages) {
    try {
      await this.dispatch(await this.read(message))
      done.push(message)
    } catch (error) {
      this.onError?.(error as ConsumeError)
    }
  }

  const failed = messages.filter((m) => !done.includes(m))
  if (failed.length > 0) {
    // Hand the lease back immediately. With a receive count of one, the next
    // receive is the redrive, so the DLQ alarm fires in seconds rather than
    // after the visibility timeout expires.
    await this.releaseVisibility(failed)
  }

  return done // the polling loop deletes exactly these
}

Sequential rather than concurrent inside the batch, which is a deliberate throughput sacrifice: ten messages processed in parallel means ten handlers touching the same connection pool, and the thing that used to be one job at a time under BullMQ's default concurrency should not quietly become ten.

Shutdown, which is two different problems

ECS sends SIGTERM and then waits out the stop timeout before SIGKILL. Both container types have to use that window, and they have to use it differently.

A web task stops accepting connections, finishes what it is serving, and then drains the publisher. If it drains the publisher first, the requests still in flight publish into a closed batch.

A worker stops polling first, finishes the messages it already has, and only then drains its own publisher — because workers publish too. The long-running jobs are the ones that spawn more work, so a worker that closes its publisher before its handlers finish will lose the events those handlers emit, which is the most annoying possible bug: intermittent, only under deploy, and invisible in every test.

Nest's lifecycle hooks give you the ordering for free if you put the two concerns in the right ones, and if you remember to turn them on at all:

app.enableShutdownHooks() // without this, none of the below ever runs

@Injectable()
export class ConsumerLifecycle implements OnApplicationBootstrap, OnModuleDestroy {
  async onApplicationBootstrap() {
    await this.consumer.listen()
  }

  async onModuleDestroy() {
    // Stops polling, waits for in-flight handlers, then returns.
    await this.consumer.stop()
  }
}

The publisher's drain goes in onApplicationShutdown, which Nest runs after every onModuleDestroy, so the ordering falls out of the framework rather than out of a comment telling the next person not to reorder two lines.

And stopTimeout on the task definition has to be longer than the worst-case drain, or the whole thing is theatre. Ours is ninety seconds against a ten second flush interval, which is enough slack that I have never seen it matter — and it is exactly the sort of number that gets copied into a new template with the wrong value, so it lives in the module and not in the service.

Running it on a laptop

This one is a consequence of the infrastructure post's IAM, and it is the piece I spent the most time on personally.

The topic policy denies sns:Publish to anything whose principal ARN is not a task role in the right stage. That is exactly the fence we wanted. It also means a developer running the service on their machine, holding perfectly valid credentials, is on the outside of it — and the failure is an AuthorizationError with no hint about why.

The options were to fake the infrastructure locally, or to let laptops in. Faking it means the thing under test is not the thing that runs, and the whole point of the deny statement is that it is difficult to get right; a local emulator would have cheerfully accepted the version of the policy I had written backwards. So: let laptops in, narrowly.

A small internal service mints short-lived credentials for a single role that the topic policy names, and only for the staging stage. The SDK asks for them when it is told it is running locally, caches them, and refreshes a minute before they expire:

async withCredentials<T>(call: () => Promise<T>): Promise<T> {
  if (!this.local) return call()

  if (!this.cached || this.cached.expiresAt <= new Date()) {
    await this.refresh()
  }

  return call()
}

Every AWS call in the SDK goes through that wrapper. It is a no-op in production — one boolean check — and on a laptop it is the difference between "works against real infrastructure" and "works against a mock and then does not work". The background refresh is a setTimeout that installs a setInterval, so the first refresh lands a minute before the first expiry and every one after it stays on that cadence rather than drifting.

The thing it does not solve is that credentials are handed to a machine, and a machine is not a person. Revocation is by role, so it is all-or-nothing. That is fine for a staging-only path and it would not be fine for anything else, and I would rather write that sentence down than let the next person assume the pattern generalises.

What I would change

The fire-and-forget publish, as above — not remove it, but stop making it the only option.

And the handler table is module-level state, populated by a decorator at import time. It works, it makes registration a one-liner, and it makes two test files that register overlapping handlers interfere with each other in an order that depends on the import graph. Every framework that has ever done this has eventually grown a way to scope it, and we will too.

Neither of those is the migration's difficulty, though. The difficulty was never the client library. It was the fifteen lines of deny statement in the previous post, and the fact that getting them wrong looks exactly like getting the code wrong.