Bounding a Private Account Event Consumer

An asynchronous WebSocket listener is not automatically a sequential consumer. If its callback awaits a database operation, the emitter can invoke it again before that operation finishes. Two updates may then write state concurrently even though they arrived in order.

A private consumer needs an explicit processing path. Its connection can remain healthy while that path is falling behind.

The following module runs one asynchronous operation at a time. It stops acceptance after overflow or a failed operation. Save it as serial-consumer.mjs:

export function serialConsumer(apply, onFailure, limit = 256) {
  if (!Number.isSafeInteger(limit) || limit < 1) throw new Error('Invalid queue limit');
  const queue = [];
  let running = false;
  let failed = false;
  let active = 0;

  function fail(error) {
    if (failed) return;
    failed = true;
    queue.length = 0;
    onFailure(error);
  }

  async function drain() {
    if (running || failed) return;
    running = true;
    try {
      while (queue.length && !failed) {
        const event = queue.shift();
        active = 1;
        await apply(event);
        active = 0;
      }
    } catch (error) {
      fail(error);
    } finally {
      active = 0;
      running = false;
    }
  }

  return (event) => {
    if (failed) return false;
    if (queue.length + active >= limit) {
      fail(new Error('Private consumer queue overflow'));
      return false;
    }
    queue.push(event);
    void drain();
    return true;
  };
}

The limit includes the active operation. This bounds event count, not bytes or processing age. A real consumer should select additional limits according to its payloads and the delay it can tolerate.

For a Bybit consumer, attach an update handler before subscribeV5. Filter the required private topic, validate its fields, and enqueue the event for the corresponding account. The Bybit tutorial supplies the authenticated subscription; it should not be replaced by a generic Node.js stream wrapper.

The apply function must finish the write or state transition before resolving. Starting another unawaited write inside it would recreate the concurrency problem behind the queue.

onFailure must return synchronously without throwing. It should mark the affected consumer as needing repair and report the cause. Clearing queued events is not successful recovery. The active operation may also complete after failure, so its owner must prevent a late result from making an abandoned recovery attempt current again.

The account initializer establishes missing values and resolves snapshot/event races. The queue preserves processing order; it cannot determine which REST response is newer than a private event or reconstruct messages that never arrived.

Hold the first apply promise open and confirm the second event does not start. Release it and check arrival order. Then reject an operation and fill the queue past its limit. Each failure should be visible once, and further events should be refused until a new consumer is initialized.

During shutdown, stop accepting new work and decide whether pending operations can be drained. A process restart needs durable state and scoped recovery, not a larger in-memory queue.