fix(ingest): align replay identity and gate shared sessions (F03 F04 F14 F15)

.env.example

@@ -36,12 +36,11 @@ MQTT_PASSWORD=change_me_strong_password
 # publications after a reconnect; the packet batch writer also retries transient
 # database failures.
 MQTT_MAX_PAYLOAD_BYTES=65536
-# Number of durable ingest connections (1-16). Each connection processes one
-# message at a time and acknowledges only after its database commit; values
-# above 1 distribute messages across the connections with a shared
-# subscription. 1 keeps single-connection behaviour exactly.
+# One durable ingest connection acknowledges only after its database commit.
+# Values above 1 are rejected until shared-session restart/resize recovery is qualified.
 MQTT_INGEST_CONNECTIONS=1
-# Shared-subscription group used when MQTT_INGEST_CONNECTIONS > 1.
+# N>1 is currently rejected: persistent shared sessions have not passed broker restart/resize recovery.
+# Reserved for a future qualified shared-subscription topology.
 MQTT_INGEST_SHARE_GROUP=meshcore-ingest
 # How long a packet batch may wait for more packets before flushing (ms;
 # 0 flushes immediately).

backend/src/db/packetBatch.integration.test.ts

@@ -21,6 +21,7 @@ import {
   type PacketBatchInput,
 } from './packetBatch.js';
 import { processNextPrivacyRematerialization } from '../workers/privacy-rematerialization.js';
+import { packetAlreadyPersisted } from '../mqtt/observationIdentity.js';
 
 const { Pool } = pg;
 const databaseUrl = process.env['TEST_INGEST_DATABASE_URL'];
@@ -638,6 +639,28 @@ test('packet batch atomically coalesces observer, sighting, and stats writes', {
     ]);
     assert.equal(statementCount, 1);
 
+    // F03: a DUP of the committed sighting is safe to acknowledge, while a
+    // distinct sighting of the same RF packet must still reach persistence.
+    const firstSighting = makePacket(0, 1);
+    const probe = (sql: string, params: unknown[]) => pool.query<{ present: boolean }>(sql, params);
+    const identity = {
+      packetHash: firstSighting.packetHash,
+      rxNodeId: firstSighting.rxNodeId,
+      topic: firstSighting.topic,
+      network: firstSighting.network,
+      hopCount: firstSighting.hopCount,
+      routeType: firstSighting.routeType,
+      pathHashes: firstSighting.pathHashes,
+      pathHashSizeBytes: firstSighting.pathHashSizeBytes,
+      direction: null,
+    };
+    assert.equal(await packetAlreadyPersisted(probe, identity), true);
+    assert.equal(await packetAlreadyPersisted(probe, { ...identity, hopCount: 3 }), false);
+    assert.equal(await packetAlreadyPersisted(probe, { ...identity, pathHashes: ['ABCD'] }), false);
+    assert.equal(await packetAlreadyPersisted(probe, { ...identity, rxNodeId: sourceId }), false);
+    assert.equal(await packetAlreadyPersisted(probe, { ...identity, network: 'teesside' }), false);
+    assert.equal(await packetAlreadyPersisted(probe, { ...identity, topic: 'meshcore/OTHER/packets' }), false);
+
     const packets = await pool.query<{ count: string }>(
       `SELECT COUNT(*)::text AS count
          FROM packets

backend/src/db/packetBatch.test.ts

@@ -71,3 +71,34 @@ test('packet batches retry transient database failures without losing the pendin
   assert.match(statements[1]!, /FOR KEY SHARE/);
   assert.doesNotMatch(statements[1]!, /FROM private_node_prefixes pp/);
 });
+
+test('F03 batch arbitration retains distinct same-hash paths at one hop count', async () => {
+  let batchSql = '';
+  let batchParams: unknown[] = [];
+  configurePacketBatch(async (sql, params) => {
+    if (sql.includes('jsonb_agg(jsonb_build_object')) {
+      return { rows: [{ generation: '7', prefixes: [] }] };
+    }
+    batchSql = sql;
+    batchParams = params ?? [];
+    return { rows: [0, 1].map((row_id) => ({
+      row_id, is_private: false, visibility_ok: true, prefix_cache_fresh: true,
+    })) };
+  });
+  const base: PacketBatchInput = {
+    time: new Date('2026-08-06T01:00:00.000Z'), packetHash: 'same-hash',
+    rxNodeId: 'A'.repeat(64), srcNodeId: 'B'.repeat(64),
+    topic: `meshcore/EMA/${'A'.repeat(64)}/packets`, topicPrefix: 'meshcore',
+    iata: 'EMA', packetType: 5, routeType: 0, hopCount: 2,
+    rssi: -80, snr: 5, payloadJson: JSON.stringify({ direction: 'rx' }),
+    companionSender: null, rawHex: '00', advertCount: null,
+    pathHashes: ['ABCD'], pathHashSizeBytes: 2, network: 'ukmesh',
+    transportCodes: null, regionScope: null,
+  };
+  const first = enqueuePacket(base);
+  const second = enqueuePacket({ ...base, pathHashes: ['EF01'] });
+  await flush();
+  await Promise.all([first, second]);
+  assert.match(batchSql, /incoming/);
+  assert.equal(batchParams.length, 50);
+});

backend/src/db/packetBatch.ts

@@ -13,6 +13,7 @@ import {
   PrivatePrefixCache,
   type PrivatePrefixRow,
 } from '../privacy/privatePrefixCache.js';
+import { observationIdentityKey } from '../mqtt/observationIdentity.js';
 
 const MAX_BATCH_SIZE = 50;
 /**
@@ -192,7 +193,26 @@ function collapseDuplicatePackets(batch: PendingPacket[]): PendingPacket[] {
   const groups = new Map<string, PendingPacket[]>();
   const kept: PendingPacket[] = [];
   for (const item of batch) {
-    const key = `${item.packet.packetHash}:${item.packet.rxNodeId ?? ''}:${item.packet.hopCount ?? '?'}`;
+    let direction: 'rx' | 'tx' | null = null;
+    if (item.packet.payloadJson) {
+      try {
+        const payload = JSON.parse(item.packet.payloadJson) as Record<string, unknown>;
+        if (payload['direction'] === 'rx' || payload['direction'] === 'tx') direction = payload['direction'];
+      } catch {
+        // Invalid JSON belongs to SQL validation, not batch arbitration.
+      }
+    }
+    const key = observationIdentityKey({
+      packetHash: item.packet.packetHash,
+      rxNodeId: item.packet.rxNodeId,
+      topic: item.packet.topic,
+      network: item.packet.network,
+      hopCount: item.packet.hopCount,
+      routeType: item.packet.routeType,
+      pathHashes: item.packet.pathHashes,
+      pathHashSizeBytes: item.packet.pathHashSizeBytes,
+      direction,
+    });
     const group = groups.get(key);
     if (group) group.push(item);
     else {
@@ -297,6 +317,11 @@ async function writeBatch(batch: PendingPacket[], idempotent: boolean): Promise<
              AND existing.rx_node_id IS NOT DISTINCT FROM c.rx_node_id
              AND existing.topic = c.topic
              AND existing.network = c.network
+             AND existing.hop_count IS NOT DISTINCT FROM c.hop_count
+             AND existing.route_type IS NOT DISTINCT FROM c.route_type
+             AND existing.path_hashes IS NOT DISTINCT FROM c.path_hashes
+             AND existing.path_hash_size_bytes IS NOT DISTINCT FROM c.path_hash_size_bytes
+             AND (existing.payload->>'direction') IS NOT DISTINCT FROM (c.payload::jsonb->>'direction')
          )
        ),`
     : '';

backend/src/metrics.test.ts

@@ -54,3 +54,9 @@ test('dynamic metric dimensions collapse unknown values into bounded buckets', (
   assert.equal(boundedNetworkMetricLabel('customer-created-network'), 'other');
   assert.equal(boundedAnalysisWorkloadLabel('run-123456'), 'other');
 });
+
+test('F14 follow-up inflight and drop metrics appear in the served registry', async () => {
+  const scrape = await metricsRegistry.metrics();
+  assert.match(scrape, /meshcore_mqtt_post_persist_inflight/);
+  assert.match(scrape, /meshcore_mqtt_post_persist_dropped_total/);
+});

backend/src/metrics.ts

@@ -503,9 +503,11 @@ export function updateDbPoolMetrics(
 export const postPersistInFlightGauge = new Gauge({
   name: 'meshcore_mqtt_post_persist_inflight',
   help: 'Best-effort post-persist follow-ups currently in flight (bounded).',
+  registers: [metricsRegistry],
 });
 
 export const postPersistDroppedTotal = new Counter({
   name: 'meshcore_mqtt_post_persist_dropped_total',
   help: 'Best-effort post-persist follow-ups dropped because the in-flight bound was reached.',
+  registers: [metricsRegistry],
 });

backend/src/mqtt/client.ts

@@ -16,6 +16,9 @@ import { parseMqttTopic } from './topic.js';
 import { createDurableMqttHandleMessage } from './durableHandler.js';
 import { waitForInFlightPacket } from './dedupeRegistry.js';
 import { persistStatusTelemetrySample } from './statusTelemetry.js';
+import { observationIdentityKey, packetAlreadyPersisted, type ObservationIdentity } from './observationIdentity.js';
+import { assertSupportedIngestConnections } from './ingestConfig.js';
+import { waitForShutdownStep } from './shutdownDeadline.js';
 import {
   boundedNetworkMetricLabel,
   mqttIngestAckDuration,
@@ -217,9 +220,8 @@ setInterval(() => {
 }, 60_000).unref();
 
 /**
- * Per-observer packet dedup — prevents relay copies of the same packet from being
- * ingested multiple times when they arrive at the same observer with the same hop count.
- * Keyed by "packetHash:observerKey:hopCount". Entries expire after 120 seconds.
+ * Process-local observation dedup. The same identity is used for claim arbitration,
+ * batch collapse and the persisted DUP probe. Entries expire after 120 seconds.
  */
 const seenPackets = new Map<string, number>();
 const SEEN_PACKETS_MAX = 50_000;
@@ -232,20 +234,16 @@ setInterval(() => {
   }
 }, 60_000).unref();
 
-function packetDedupeKey(packetHash: string, observerKey: string, hopCount: number | undefined): string {
-  return `${packetHash}:${observerKey}:${hopCount ?? '?'}`;
+function isDuplicatePacket(key: string): boolean {
+  return seenPackets.has(key);
 }
 
-function isDuplicatePacket(packetHash: string, observerKey: string, hopCount: number | undefined): boolean {
-  return seenPackets.has(packetDedupeKey(packetHash, observerKey, hopCount));
-}
-
-function rememberPacket(packetHash: string, observerKey: string, hopCount: number | undefined): void {
+function rememberPacket(key: string): void {
   if (seenPackets.size >= SEEN_PACKETS_MAX) {
     const oldest = seenPackets.keys().next().value;
     if (oldest !== undefined) seenPackets.delete(oldest);
   }
-  seenPackets.set(packetDedupeKey(packetHash, observerKey, hopCount), Date.now());
+  seenPackets.set(key, Date.now());
 }
 
 /**
@@ -262,32 +260,6 @@ const inFlightPackets = new Map<string, Promise<void>>();
 const POST_PERSIST_MAX_INFLIGHT = 64;
 let postPersistInFlight = 0;
 
-/**
- * Database-level duplicate confirmation for QoS 1 redeliveries — the in-memory
- * filters do not survive a restart. Bounded to the retained packet window so
- * the probe stays on the packet_hash index.
- */
-async function packetAlreadyPersisted(
-  packetHash: string,
-  rxNodeId: string,
-  topic: string,
-  network: string,
-): Promise<boolean> {
-  const result = await query<{ present: boolean }>(
-    `SELECT EXISTS (
-       SELECT 1
-         FROM packets
-        WHERE packet_hash = $1
-          AND rx_node_id = $2
-          AND topic = $3
-          AND network = $4
-          AND time > NOW() - INTERVAL '30 days'
-     ) AS present`,
-    [packetHash, rxNodeId, topic, network],
-  );
-  return Boolean(result.rows[0]?.present);
-}
-
 /**
  * Channel registry — built once at startup.
  * MESHCORE_CHANNEL_SECRETS supports 'name:hex' or bare 'hex' entries (comma-separated).
@@ -309,12 +281,9 @@ const MQTT_CLIENT_ID = String(
   process.env['MQTT_CLIENT_ID'] ?? 'meshcore-analytics-ingest',
 ).trim() || 'meshcore-analytics-ingest';
 /**
- * Number of ingest connections to the broker. Each connection is a strictly
- * serial durable pipeline (mqtt.js dispatches the next packet only after the
- * previous one is acknowledged), so this is the supported way to run more
- * than one persistence flow; messages are distributed between the connections
- * with a shared subscription. 1 preserves the historical single-connection
- * behaviour exactly.
+ * The default and only qualified topology is one durable connection. The
+ * configured value is checked before connecting; shared sessions need a
+ * successful restart and resize qualification before they can be enabled.
  */
 const MQTT_INGEST_CONNECTIONS = boundedEnvInteger('MQTT_INGEST_CONNECTIONS', 1, 1, 16);
 const MQTT_INGEST_SHARE_GROUP = String(
@@ -379,6 +348,7 @@ function reportMqttIngestFailure(client: MqttClient, error: Error): void {
 
 export async function startMqttClient(): Promise<void> {
   if (mqttStopping) return;
+  assertSupportedIngestConnections(MQTT_INGEST_CONNECTIONS);
   // Must complete before connecting to MQTT — the broker replays buffered
   // messages immediately on connect, and knownNodeIds must be populated first.
   await initSpamDetector();
@@ -454,14 +424,15 @@ export async function startMqttClient(): Promise<void> {
 
 export async function stopMqttClient(timeoutMs = 20_000): Promise<void> {
   mqttStopping = true;
+  const deadline = Date.now() + timeoutMs;
   const clients = mqttClients.splice(0, mqttClients.length);
   for (const client of clients) {
     client.removeAllListeners('reconnect');
-    await client.endAsync(false).catch((error: unknown) => {
+    await waitForShutdownStep(client.endAsync(false), deadline, 'disconnect').catch((error: unknown) => {
+      if (error instanceof Error && error.message.startsWith('MQTT_DRAIN_TIMEOUT:')) throw error;
       console.error('[mqtt] disconnect failed:', error instanceof Error ? error.message : error);
     });
   }
-  const deadline = Date.now() + timeoutMs;
   while (activeMqttIngests > 0 && Date.now() < deadline) {
     await new Promise((resolve) => setTimeout(resolve, 25));
   }
@@ -474,7 +445,7 @@ export async function stopMqttClient(timeoutMs = 20_000): Promise<void> {
     clearTimeout(nodeFlushTimer);
     flushPendingNodeEmits();
   }
-  await closePacketBatch();
+  await waitForShutdownStep(closePacketBatch(), deadline, 'batch');
   setMqttRuntimeStatus('disconnected');
 }
 
@@ -790,9 +761,20 @@ async function handleMessage(topic: string, rawPayload: Buffer, redelivered = fa
   // comes from validated route/path-invariant wire content.
   const finalHash = canonicalPacketId ?? upstreamPacketHash(json['hash']) ?? crypto.randomUUID();
 
-  const dedupeKey = packetDedupeKey(finalHash, observerKey, decodedHops);
+  const observationIdentity: ObservationIdentity = {
+    packetHash: finalHash,
+    rxNodeId: observerKey,
+    topic,
+    network,
+    hopCount: decodedHops ?? null,
+    routeType: decodedRouteType ?? null,
+    pathHashes: path ?? null,
+    pathHashSizeBytes: decodedPathHashSizeBytes ?? null,
+    direction: direction ?? null,
+  };
+  const dedupeKey = observationIdentityKey(observationIdentity);
 
-  if (isDuplicatePacket(finalHash, observerKey, decodedHops)) {
+  if (isDuplicatePacket(dedupeKey)) {
     // This marker is written only after the matching packet is committed.
     // A broker redelivery after a lost PUBACK confirms the durable outcome.
     markMqttPersistenceHealthy();
@@ -822,8 +804,8 @@ async function handleMessage(topic: string, rawPayload: Buffer, redelivered = fa
 
   // Decoded payloads omit mctomqtt's envelope direction. Keep that metadata in
   // the stored JSON so RX/TX counts do not classify every decoded packet as RX.
-  const packetPayload = innerPayload && direction
-    ? { ...innerPayload, direction }
+  const packetPayload = direction
+    ? { ...(innerPayload ?? json), direction }
     : (innerPayload ?? json);
 
   const livePacket: LivePacket = {
@@ -869,7 +851,7 @@ async function handleMessage(topic: string, rawPayload: Buffer, redelivered = fa
       transportCodes: decodedTransportCodes,
       regionScope:    decodedRegionScope,
     });
-    rememberPacket(finalHash, observerKey, decodedHops);
+    rememberPacket(dedupeKey);
     markMqttPersistenceHealthy();
     emitNode(observerKey, { network, observerId: observerKey });
     livePacket.isPrivate = visibility.isPrivate;
@@ -892,7 +874,10 @@ async function handleMessage(topic: string, rawPayload: Buffer, redelivered = fa
     // crash must not be persisted or counted a second time. Confirm against
     // the retained window before writing (inside the claim so the registered
     // claim promise covers this path too).
-    if (redelivered && await packetAlreadyPersisted(finalHash, observerKey, topic, network)) {
+    if (redelivered && await packetAlreadyPersisted(
+      (sql, params) => query<{ present: boolean }>(sql, params),
+      observationIdentity,
+    )) {
       markMqttPersistenceHealthy();
       mqttIngestOutcomesTotal.inc({ outcome: 'duplicate_packet' });
       return;

backend/src/mqtt/ingestConfig.test.ts

@@ -0,0 +1,9 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { assertSupportedIngestConnections } from './ingestConfig.js';
+
+test('F04 shared-subscription N>1 is gated until restart and resize recovery is qualified', () => {
+  assert.doesNotThrow(() => assertSupportedIngestConnections(1));
+  assert.throws(() => assertSupportedIngestConnections(2), /MQTT_INGEST_CONNECTIONS_UNSUPPORTED/);
+  assert.throws(() => assertSupportedIngestConnections(16), /MQTT_INGEST_CONNECTIONS_UNSUPPORTED/);
+});

backend/src/mqtt/ingestConfig.ts

@@ -0,0 +1,6 @@
+/** Persistent shared sessions need a proven broker restart/resize recovery path. */
+export function assertSupportedIngestConnections(connections: number): void {
+  if (connections !== 1) {
+    throw new Error(`MQTT_INGEST_CONNECTIONS_UNSUPPORTED:${connections}; use 1 until shared-session recovery is qualified`);
+  }
+}

backend/src/mqtt/observationIdentity.test.ts

@@ -0,0 +1,55 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {
+  observationIdentityKey,
+  packetAlreadyPersisted,
+  type ObservationIdentity,
+} from './observationIdentity.js';
+
+const base: ObservationIdentity = {
+  packetHash: 'ABCDEF',
+  rxNodeId: 'A'.repeat(64),
+  topic: `meshcore/EMA/${'A'.repeat(64)}/packets`,
+  network: 'ukmesh',
+  hopCount: 2,
+  routeType: 0,
+  pathHashes: ['ABCD', 'EF01'],
+  pathHashSizeBytes: 2,
+  direction: 'rx',
+};
+
+test('F03 canonical observation key distinguishes hop, path, direction, observer and scope', () => {
+  const key = observationIdentityKey(base);
+  assert.equal(key, observationIdentityKey({ ...base, pathHashes: ['abcd', 'ef01'] }));
+  for (const changed of [
+    { hopCount: 3 },
+    { pathHashes: ['ABCD', 'EF02'] },
+    { pathHashSizeBytes: 3 },
+    { routeType: 1 },
+    { direction: 'tx' as const },
+    { rxNodeId: 'B'.repeat(64) },
+    { topic: `meshcore/MME/${'A'.repeat(64)}/packets` },
+    { network: 'teesside' },
+  ]) {
+    assert.notEqual(key, observationIdentityKey({ ...base, ...changed }), JSON.stringify(changed));
+  }
+});
+
+test('F03 restart probe checks the same identity, including a prior same-hash different-hop sighting', async () => {
+  let statement = '';
+  let values: unknown[] = [];
+  const query = async (_sql: string, params: unknown[]) => {
+    statement = _sql;
+    values = params;
+    return { rows: [{ present: false }] };
+  };
+  assert.equal(await packetAlreadyPersisted(query, { ...base, hopCount: 3 }), false);
+  assert.match(statement, /hop_count IS NOT DISTINCT FROM/);
+  assert.match(statement, /route_type IS NOT DISTINCT FROM/);
+  assert.match(statement, /path_hashes IS NOT DISTINCT FROM/);
+  assert.match(statement, /path_hash_size_bytes IS NOT DISTINCT FROM/);
+  assert.match(statement, /payload->>'direction'/);
+  assert.equal(values[4], 3);
+  assert.deepEqual(values[6], ['ABCD', 'EF01']);
+  assert.equal(values[8], 'rx');
+});

backend/src/mqtt/observationIdentity.ts

@@ -0,0 +1,64 @@
+/** Fields that distinguish one observer's RF sighting from another. */
+export type ObservationIdentity = {
+  packetHash: string;
+  rxNodeId: string | null;
+  topic: string;
+  network: string;
+  hopCount: number | null;
+  routeType: number | null;
+  pathHashes: string[] | null;
+  pathHashSizeBytes: number | null;
+  direction: 'rx' | 'tx' | null;
+};
+
+function normalizedPath(pathHashes: string[] | null): string[] | null {
+  return pathHashes?.map((hash) => hash.trim().toUpperCase()) ?? null;
+}
+
+export function observationIdentityKey(observation: ObservationIdentity): string {
+  return JSON.stringify([
+    observation.packetHash,
+    observation.rxNodeId,
+    observation.topic,
+    observation.network,
+    observation.hopCount,
+    observation.routeType,
+    normalizedPath(observation.pathHashes),
+    observation.pathHashSizeBytes,
+    observation.direction,
+  ]);
+}
+
+/** The DUP probe must match the same sighting claimed by the live handler. */
+export async function packetAlreadyPersisted(
+  query: (sql: string, params: unknown[]) => Promise<{ rows: Array<{ present: boolean }> }>,
+  observation: ObservationIdentity,
+): Promise<boolean> {
+  const result = await query(
+    `SELECT EXISTS (
+       SELECT 1 FROM packets
+        WHERE packet_hash = $1
+          AND rx_node_id IS NOT DISTINCT FROM $2::text
+          AND topic = $3
+          AND network = $4
+          AND hop_count IS NOT DISTINCT FROM $5::int
+          AND route_type IS NOT DISTINCT FROM $6::int
+          AND path_hashes IS NOT DISTINCT FROM $7::text[]
+          AND path_hash_size_bytes IS NOT DISTINCT FROM $8::int
+          AND (payload->>'direction') IS NOT DISTINCT FROM $9::text
+          AND time > NOW() - INTERVAL '30 days'
+     ) AS present`,
+    [
+      observation.packetHash,
+      observation.rxNodeId,
+      observation.topic,
+      observation.network,
+      observation.hopCount,
+      observation.routeType,
+      normalizedPath(observation.pathHashes),
+      observation.pathHashSizeBytes,
+      observation.direction,
+    ],
+  );
+  return Boolean(result.rows[0]?.present);
+}

backend/src/mqtt/shutdownDeadline.test.ts

@@ -0,0 +1,10 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { waitForShutdownStep } from './shutdownDeadline.js';
+
+test('F15 one MQTT shutdown deadline includes a blocked disconnect', async () => {
+  const never = new Promise<void>(() => {});
+  const deadline = Date.now() + 20;
+  await assert.rejects(waitForShutdownStep(never, deadline, 'disconnect'), /MQTT_DRAIN_TIMEOUT:disconnect/);
+  await assert.rejects(waitForShutdownStep(Promise.resolve(), deadline - 1, 'batch'), /MQTT_DRAIN_TIMEOUT:batch/);
+});

backend/src/mqtt/shutdownDeadline.ts

@@ -0,0 +1,20 @@
+/** Every shutdown step consumes the same wall-clock budget. */
+export async function waitForShutdownStep<T>(
+  operation: Promise<T>,
+  deadline: number,
+  step: string,
+): Promise<T> {
+  const remainingMs = deadline - Date.now();
+  if (remainingMs <= 0) throw new Error(`MQTT_DRAIN_TIMEOUT:${step}`);
+  let timeout: NodeJS.Timeout | undefined;
+  try {
+    return await Promise.race([
+      operation,
+      new Promise<never>((_resolve, reject) => {
+        timeout = setTimeout(() => reject(new Error(`MQTT_DRAIN_TIMEOUT:${step}`)), remainingMs);
+      }),
+    ]);
+  } finally {
+    if (timeout) clearTimeout(timeout);
+  }
+}