Appearance

MigrationDocs 1.3.x

events-amqp 1.3 Behavior Changes ​

Applies when upgrading @connectum/events-amqp from 1.2.x to 1.3.0.

Does this apply to you? ​

You need to act if your service uses AmqpAdapter and any of the following is true:

  • you use a headers exchange with topologyMode: 'assert' and declare selective bindings for a queue in topology.bindings;
  • you count or alert on onDisconnected calls;
  • you use recovery: false and handle onDisconnected;
  • you set a finite recovery.maxRetries;
  • you match connection errors by message text or by .code on the error itself;
  • a lifecycle callback throws on purpose;
  • you pin amqplib below 2.2.0 through overrides, resolutions, or a lockfile patch, or you alert on reconnecting delay values.

The new options (publishRetry, onLifecycle, initialConnectMaxRetries, treatTopologyErrorAsFatal, AmqpTopologyError.object, and the /testing subpath) are opt-in. They are described in Run the AMQP adapter reliably.

Required changes ​

Terminal > excludes the base routing key ​

In the published 1.2.0 adapter, user.> was bound as RabbitMQ's user.#, so the broker also routed the base key user to that subscription. The upcoming 1.3.0 adapter binds user.*.#: RabbitMQ's * requires one segment and # then accepts zero or more additional segments.

The adapter rejects a complete > segment outside the terminal position. It also rejects subscriptions containing complete * or > wildcard tokens on a direct exchange when topologyMode is assert, before creating subscription topology, because the binding would be a literal key that never matches; fanout and headers exchanges accept any pattern and deliver every message to the queue. Embedded characters such as user* and user> remain literal. A complete # segment is rejected in topic subscriptions, where RabbitMQ would treat it as a wildcard even though EventBus treats it as literal. Non-topic exchanges continue to accept # as a routing-key literal or ignore it according to their exchange type. Operator-declared bindings in topology.bindings, including a # topic binding, are not rewritten by the adapter.

Action: inspect named-group and externally managed queues for bindings created by the old adapter. The adapter adds the new binding but does not remove an old broad binding. Add the new binding first, then remove only the obsolete binding; keep the queue and its queued messages:

typescript
await channel.bindQueue(queue, exchange, 'user.*.#');
await channel.unbindQueue(queue, exchange, 'user.#');

Check whether other consumers depend on the old binding before removing it. Unbinding changes future routing and does not delete messages already queued.

A headers exchange honours declared bindings ​

With exchangeType: 'headers' and topologyMode: 'assert', the previous adapter bound every subscribed queue to the exchange with an argument-less binding. A headers exchange delivers a message when any binding of the queue matches, so that binding defeated a selective x-match binding declared in topology.bindings: the queue received every message. The 1.3.0 adapter skips its own binding for a queue for which topology.bindings declares a binding to the adapter's exchange, so the declared bindings alone select what the queue receives. Queues without declared bindings, the check and skip modes and topic, direct and fanout exchanges are unchanged. A binding made outside the topology is not seen by the adapter, which still adds its own next to it, and a binding reached through an intermediate exchange is not considered.

Action, if a queue relied on receiving everything: declare the binding without arguments, or drop the selective one.

Action, for a durable queue the previous adapter already bound: the broker keeps the old argument-less bindings, one per subscription pattern, and the adapter never removes bindings (it cannot tell its own from an operator's), so until you remove them the queue still receives every message. The routing key of such a binding is the pattern as the previous adapter converted it: a > became #, so user.> was bound as user.#, and unbinding with the pattern as written removes nothing, silently. Read the exact keys from the broker (rabbitmqctl list_bindings, or the management API) and unbind each argument-less leftover with its key and no arguments:

typescript
await channel.unbindQueue(queue, exchange, 'user.#'); // the key the broker lists

Check the list again afterwards. Deleting the queue and letting the adapter declare it again also works, and loses its queued messages.

onDisconnected fires once per drop ​

In 1.2, a socket-level connection cut reported the loss twice: once from the raw connection error event and once from the recovery disconnect event. In 1.3 it is reported exactly once, on both onDisconnected and onLifecycle. A graceful close by the server was reported once before and still is.

Action: expect disconnect counters to drop, roughly by half when most drops are socket-level cuts. Adjust alert thresholds and dashboards that were calibrated on the doubled numbers.

recovery: false reports every close, once ​

In 1.2, with recovery: false, onDisconnected ran only on the connection error event, so a close forced by the server without an error (for example, an operator closing the connection) was not reported at all. In 1.3 the adapter reports disconnected once when the connection closes, with the preceding error as the cause when there was one, otherwise the close error from the broker (its code is the reply code, for example 320 for a forced close), and Error('Connection closed') only when the close carries no cause at all. It is not reported for your own disconnect() or when connect() discards a connection whose setup failed.

Action: make sure your onDisconnected handler is correct for server-forced closes too; it now runs for them.

After recovery gives up, subscribe again ​

When a finite recovery.maxRetries was exhausted, 1.2 kept the dead connection: connect() threw already connected until you called disconnect(), and subscribe() went to the dead connection and failed with an untyped amqplib error.

In 1.3, before reporting reconnect-failed, the adapter drops the dead connection, its publish channel, and all subscription records:

  • publish() and subscribe() reject at once with AmqpConnectionError;
  • connect() is accepted without a prior disconnect();
  • the new connection starts with no subscriptions.

Action: after reconnect-failed, subscribe again after connect(). With the EventBus, bus.stop() followed by bus.start() does this, because start() connects the adapter and subscribes the registered routes. A disconnect() call before connect() is no longer needed, and still does no harm.

The initial connect rejects with AmqpConnectionError ​

With recovery enabled and a finite maxRetries, a broker that stays unreachable for the whole initial budget used to reject connect() with the raw amqplib error (for example, ECONNREFUSED). In 1.3 connect() rejects with AmqpConnectionError, and the original error is its cause. With recovery: false the raw error is still thrown unchanged.

Action: catch AmqpConnectionError and read the original error from cause:

typescript
import { AmqpConnectionError } from '@connectum/events-amqp';

try {
  await bus.start();
} catch (err) {
  // 1.2: (err as NodeJS.ErrnoException).code === 'ECONNREFUSED'
  if (err instanceof AmqpConnectionError) {
    const cause = err.cause as NodeJS.ErrnoException | undefined;
    logger.error({ code: cause?.code }, 'AMQP broker unreachable at startup');
  }
  throw err;
}

A subscribe() cut off by a dying connection rejects typed ​

A subscribe() that was waiting for a channel when the connection cycle died used to reject with amqplib's Error('Connection closed') or a raw socket error. In 1.3 it rejects with AmqpConnectionError; when the channel could not be created, the original error is its cause. A subscription whose connection died while it was being set up is no longer recorded, so a later connect() does not restart it.

Action: match these failures with instanceof AmqpConnectionError instead of the message text.

Lifecycle callbacks no longer propagate exceptions ​

In 1.2 an exception thrown by a lifecycle callback propagated into amqplib's connection event handlers. In 1.3 the adapter catches it, so a faulty callback cannot interfere with recovery, and a throwing onLifecycle does not prevent the flat callbacks from running (or the reverse). A promise returned by a callback is attached to but not awaited, and a rejection is isolated the same way. Each isolated failure is reported to onLifecycle as a lifecycle-error event with the callback and the event it was handling; a failure of the callback handling lifecycle-error itself is dropped.

Action: if a callback threw to signal a failure, report it another way, such as logging, a metric, or a flag your health check reads. Handle lifecycle-error in onLifecycle to see callback failures.

amqplib 2.2.0 is the minimum ​

The dependency range is now amqplib@^2.2.0. Reconnect delays are capped so that no delay exceeds recovery.maxDelay; with the defaults a saturated delay is between 20 000 and 30 000 ms.

Action: remove any overrides, resolutions, or pins that hold amqplib below 2.2.0. If you alert on the delay of reconnecting events, check the thresholds against the bound above.

Move to onLifecycle ​

The flat callbacks onConnected, onDisconnected, onReconnecting, onReconnectFailed, and onSetupFailed are deprecated since 1.3 and kept until at least 2.0. They receive the same events as onLifecycle.

Setting onLifecycle enables the startup topology check, as onSetupFailed and failFastOnInitialSetupError already did: with recovery enabled and initialConnectMaxRetries unset, connect() opens one extra short-lived connection and validates the topology before connecting. A service that used only onConnected or onDisconnected in 1.2 did not run this check.

A connection lost while the setup pass runs (at startup or during recovery) is now reported as AmqpConnectionError, not AmqpTopologyError. With failFastOnInitialSetupError such a drop no longer rejects connect(): the adapter continues with recovery, and no setup-failed is reported for it. With recovery: false the class of that rare error changes to AmqpConnectionError. Broker replies (404, 406, 403, and so on) and the adapter's own configuration check still fail fast. This affects code that catches AmqpTopologyError from connect() with recovery: false, and code that relied on a dropped connection to fail startup under failFastOnInitialSetupError. Handle AmqpConnectionError there, or treat the startup as retryable.

Watch for consumer-lost ​

When the broker ends one subscription's consumer on a live connection (the queue was deleted, the consumer was cancelled, the channel was closed), 1.2 stayed silent and the subscription stopped receiving. From 1.3 the adapter reports consumer-lost and, with recovery enabled, restores the consumer. In assert mode a restoration declares the queue again, so a queue you deleted on purpose returns unless you also unsubscribe(), set recovery: false, or use topologyMode: 'check'. Alert on consumer-lost and consumer-restore-failed. See A consumer the broker ends.

Before and after ​

typescript
// 1.2
const adapter = AmqpAdapter({
  url,
  lifecycle: {
    onConnected: () => health.set('amqp', true),
    onDisconnected: (err) => {
      health.set('amqp', false);
      metrics.amqpDisconnects.inc(); // counted twice per socket-level cut
    },
    onReconnectFailed: (err) => logger.error({ err }, 'AMQP recovery gave up'),
  },
});

// 1.3
const adapter = AmqpAdapter({
  url,
  lifecycle: {
    // Also enables the startup topology check (one extra connection at connect()).
    onLifecycle: (event) => {
      if (event.type === 'connected') health.set('amqp', true);
      if (event.type === 'disconnected') {
        health.set('amqp', false);
        metrics.amqpDisconnects.inc(); // once per drop
      }
      if (event.type === 'reconnect-failed') {
        // Subscriptions are gone: restart the bus or the service.
        logger.error({ err: event.error }, 'AMQP recovery gave up');
      }
    },
  },
});

Verify the upgrade ​

  • Cut the broker connection (for example, restart RabbitMQ) and check that each drop increments your disconnect counter once.
  • With a finite maxRetries, start the service with the broker stopped and check that startup fails with AmqpConnectionError and the original error in cause.
  • In unit tests, use FakeAmqpAdapter from @connectum/events-amqp/testing: control.dropConnection() followed by control.exhaustRecovery() puts the fake in the terminal state, after which connect() succeeds without the old subscriptions.