From cbf78a8a6c8d0f11da200eed0bcf9f4161c5dd2f Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Sun, 11 Oct 2026 16:18:39 +0000 Subject: [PATCH] fix(ingest): align replay identity and gate shared sessions (F03 F04 F14 F15) --- .env.example | 9 +- .../src/db/packetBatch.integration.test.ts | 23 +++++ backend/src/db/packetBatch.test.ts | 31 +++++++ backend/src/db/packetBatch.ts | 27 +++++- backend/src/metrics.test.ts | 6 ++ backend/src/metrics.ts | 2 + backend/src/mqtt/client.ts | 89 ++++++++----------- backend/src/mqtt/ingestConfig.test.ts | 9 ++ backend/src/mqtt/ingestConfig.ts | 6 ++ backend/src/mqtt/observationIdentity.test.ts | 55 ++++++++++++ backend/src/mqtt/observationIdentity.ts | 64 +++++++++++++ backend/src/mqtt/shutdownDeadline.test.ts | 10 +++ backend/src/mqtt/shutdownDeadline.ts | 20 +++++ 13 files changed, 293 insertions(+), 58 deletions(-) create mode 100644 backend/src/mqtt/ingestConfig.test.ts create mode 100644 backend/src/mqtt/ingestConfig.ts create mode 100644 backend/src/mqtt/observationIdentity.test.ts create mode 100644 backend/src/mqtt/observationIdentity.ts create mode 100644 backend/src/mqtt/shutdownDeadline.test.ts create mode 100644 backend/src/mqtt/shutdownDeadline.ts diff --git a/.env.example b/.env.example index 28c6b87..62b61e0 100644 --- a/.env.example +++ b/.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). diff --git a/backend/src/db/packetBatch.integration.test.ts b/backend/src/db/packetBatch.integration.test.ts index c443a46..768656a 100644 --- a/backend/src/db/packetBatch.integration.test.ts +++ b/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 diff --git a/backend/src/db/packetBatch.test.ts b/backend/src/db/packetBatch.test.ts index 810b368..8d39ff7 100644 --- a/backend/src/db/packetBatch.test.ts +++ b/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); +}); diff --git a/backend/src/db/packetBatch.ts b/backend/src/db/packetBatch.ts index a823f2d..cc97c09 100644 --- a/backend/src/db/packetBatch.ts +++ b/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(); 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; + 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') ) ),` : ''; diff --git a/backend/src/metrics.test.ts b/backend/src/metrics.test.ts index 8e430c9..9e6e898 100644 --- a/backend/src/metrics.test.ts +++ b/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/); +}); diff --git a/backend/src/metrics.ts b/backend/src/metrics.ts index 8866bb1..946a266 100644 --- a/backend/src/metrics.ts +++ b/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], }); diff --git a/backend/src/mqtt/client.ts b/backend/src/mqtt/client.ts index e4d983b..6ea65cd 100644 --- a/backend/src/mqtt/client.ts +++ b/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(); 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>(); 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 { - 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 { 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 { export async function stopMqttClient(timeoutMs = 20_000): Promise { 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 { 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; diff --git a/backend/src/mqtt/ingestConfig.test.ts b/backend/src/mqtt/ingestConfig.test.ts new file mode 100644 index 0000000..2c7ba05 --- /dev/null +++ b/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/); +}); diff --git a/backend/src/mqtt/ingestConfig.ts b/backend/src/mqtt/ingestConfig.ts new file mode 100644 index 0000000..b803668 --- /dev/null +++ b/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`); + } +} diff --git a/backend/src/mqtt/observationIdentity.test.ts b/backend/src/mqtt/observationIdentity.test.ts new file mode 100644 index 0000000..1a0eb35 --- /dev/null +++ b/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'); +}); diff --git a/backend/src/mqtt/observationIdentity.ts b/backend/src/mqtt/observationIdentity.ts new file mode 100644 index 0000000..d820ad1 --- /dev/null +++ b/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 { + 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); +} diff --git a/backend/src/mqtt/shutdownDeadline.test.ts b/backend/src/mqtt/shutdownDeadline.test.ts new file mode 100644 index 0000000..190ee1e --- /dev/null +++ b/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(() => {}); + 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/); +}); diff --git a/backend/src/mqtt/shutdownDeadline.ts b/backend/src/mqtt/shutdownDeadline.ts new file mode 100644 index 0000000..2774aa8 --- /dev/null +++ b/backend/src/mqtt/shutdownDeadline.ts @@ -0,0 +1,20 @@ +/** Every shutdown step consumes the same wall-clock budget. */ +export async function waitForShutdownStep( + operation: Promise, + deadline: number, + step: string, +): Promise { + 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((_resolve, reject) => { + timeout = setTimeout(() => reject(new Error(`MQTT_DRAIN_TIMEOUT:${step}`)), remainingMs); + }), + ]); + } finally { + if (timeout) clearTimeout(timeout); + } +}