diff --git a/CHANGELOG.md b/CHANGELOG.md index 01e29013..f738ae19 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -164,6 +164,11 @@ * Web: a new node's long name in the chat is underlined only on hover or keyboard focus (SPEC LA6) * Web: long names and hardware models no longer widen the page past the nodes table's column: while the table would overflow, those cells break inside a word (SPEC PO1) * Web: static pages fit phone screens with long inline code and wide tables: inline code breaks anywhere, and a table wider than the page scrolls inside its own box (SPEC PO2/PO3) +* Web: a later message from another sender under a stored message's id no longer replaces that message's sender, recipient and text; it is dropped with a warning (SPEC KC1, KC3) +* Web: a later copy of a stored message fills a missing recipient, text, reply, emoji or portnum and no longer replaces them (SPEC KC2) +* Web: a position, telemetry reading or trace under an id another node already holds no longer replaces that node's row; it is dropped with a warning (SPEC KC4) +* Web: a message naming its sender only by `from_num` stores that number's `!xxxxxxxx` id as `from_id`; such a copy of another sender's message is dropped (SPEC KC1) +* Web: the warnings of dropped messages and records log at most 10 lines a minute per writer, then one line with the number suppressed (SPEC KC5) * Web: HTML pages send a `Content-Security-Policy` and every response `Referrer-Policy: strict-origin-when-cross-origin`; basemap tiles and images on custom pages must use `https://` (SPEC HD1-HD3) * Web: the canonical link, `og:image`, `robots.txt`, `sitemap.xml` and the `/og-image.png` capture keep the scheme `INSTANCE_DOMAIN` names, and otherwise take only `http` or `https` from forwarded headers (SPEC HD4) * Web: the Chromium that renders `/og-image.png` starts without `API_TOKEN` or the app's other environment variables (SPEC HD5) diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index 9e588e83..689f1a7c 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -199,7 +199,7 @@ Single message payload: - Meta: `channel_name` (string; only when not encrypted and known), `ingestor` (canonical host id), `lora_freq`, `modem_preset` - `protocol` (optional string; `"meshtastic"`, `"meshcore"`, or `"reticulum"`) - explicit per-record protocol stamp. Takes precedence over the value inherited from the registered ingestor; values outside the whitelist fall back to the ingestor lookup, then to `"meshtastic"`. Ingestors SHOULD stamp this on every message so the web app classifies senders correctly even before the ingestor heartbeat is processed. -Cross-ingestor deduplication. The `id` field is the sole dedup key - the server collapses repeat POSTs on the `messages.id` PRIMARY KEY. Protocols that lack a firmware-assigned packet ID MUST derive a stable, sender-side fingerprint so that the same physical transmission heard by multiple ingestors produces the same `id`. The id MUST fit in 53 bits (`0 <= id <= (1 << 53) - 1`) to round-trip through the JavaScript frontend without precision loss. +Cross-ingestor deduplication. The `id` field is the sole dedup key - the server collapses repeat POSTs on the `messages.id` PRIMARY KEY. A later copy fills the `to_id`, `text`, `reply_id`, `emoji` and `portnum` the stored row lacks and never replaces them (SPEC KC2). A copy of any protocol but MeshCore that names a `from_id` other than the stored row's is another message under a reused `id`: the stored row stays, and the copy is dropped and logged at `warn`, with no decrypted payload stored and no node created or refreshed for it (SPEC KC1). MeshCore copies rank their senders instead (SPEC MR3). Protocols that lack a firmware-assigned packet ID MUST derive a stable, sender-side fingerprint so that the same physical transmission heard by multiple ingestors produces the same `id`. The id MUST fit in 53 bits (`0 <= id <= (1 << 53) - 1`) to round-trip through the JavaScript frontend without precision loss. For MeshCore the canonical fingerprint is: @@ -231,7 +231,7 @@ Single position payload: - Required: `id` (int), `rx_time` (int), `rx_iso` (string) - Node: `node_id` (canonical string), `node_num` (int|nil), `num` (int|nil), `from_id` (canonical string), `to_id` (string|nil) -- Key: `public_key` (string|absent) - the full public key of the MeshCore advert or contact the position came from. A position carrying one moves the node row only when it is the key the row is bound to; the position row itself is stored either way (SPEC NI3). Meshtastic and Reticulum positions carry none. +- Key: `public_key` (string|absent) - the full public key of the MeshCore advert or contact the position came from. A position carrying one moves the node row only when it is the key the row is bound to; the position row itself is stored either way (SPEC NI3), unless its `id` is stored for another node (SPEC KC4). Meshtastic and Reticulum positions carry none. - Position: `latitude`, `longitude`, `altitude` (floats|nil) - Position time: `position_time` (int|nil) - Quality: `location_source` (string|nil), `precision_bits` (int|nil), `sats_in_view` (int|nil), `pdop` (float|nil) @@ -246,6 +246,8 @@ Sentinel handling (issue #782). The same rules as `POST /api/nodes` apply here: MeshCore providers that obtain a contact advertisement with `(0, 0)` SHOULD drop the entire advertisement rather than queue a coordinate-less position row. +Reused `id` (SPEC KC4). A position whose `id` is stored under a `node_id` it does not name is not stored: the stored row stays, the web app logs it at `warn`, and the position still updates its own node. A stored row without a `node_id` takes the position's. + #### `POST /api/telemetry` Single telemetry payload: @@ -293,6 +295,8 @@ Single telemetry payload: - Meta: `ingestor`, `lora_freq`, `modem_preset` - `protocol` (optional string; `"meshtastic"`, `"meshcore"`, or `"reticulum"`) - explicit per-record protocol stamp; same semantics as on `POST /api/messages`. +Reused `id` (SPEC KC4). As for positions: a reading whose `id` is stored under a `node_id` it does not name is not stored and is logged at `warn`; it still updates its own node, and a stored row without a `node_id` takes the reading's. + MeshCore telemetry sourcing (TI-A3). MeshCore exposes other nodes' telemetry only as on-air *pull* requests (there is no unsolicited telemetry broadcast the companion library surfaces), so the MeshCore provider collects it three ways and normalises every reading into this same payload shape with `protocol="meshcore"`: (1) host self-telemetry over the local companion link (`get_bat` → battery millivolts as `voltage`; `get_self_telemetry` → the host's CayenneLPP sensor list), no LoRa airtime, cadence `MESHCORE_SELF_TELEMETRY_SECONDS` (default 3600 s, matching the host-telemetry suppression window; `<= 0` disables); (2) round-robin contact polling (`req_telemetry_sync`, falling back to `req_status_sync` when a node reports no sensors) at one on-air request per `MESHCORE_TELEMETRY_POLL_SECONDS` (default 300 s; `<= 0` disables) regardless of roster size, with each contact additionally capped at one poll per 24 h (a fixed per-node cooldown, stamped at the poll attempt so unreachable nodes are not hammered; when every contact is fresh the tick transmits nothing) - and the transmit policy gates these on-air polls entirely - they require `TX_ENABLED=1` (default `0`, so an ingestor polls no other node unless its operator opts in), and the legacy `RX_ONLY=1` vetoes them regardless; the local self reads in (1) cost no airtime and are unaffected; (3) unsolicited/tag-matched events (`TELEMETRY_RESPONSE`, `STATUS_RESPONSE`, `BATTERY`) whenever the radio surfaces them. CayenneLPP types map to canonical keys (`temperature`, `humidity`→`relative_humidity`, `barometer`→`barometric_pressure`, `voltage`, `current` - scaled A→mA to match the Meshtastic column convention, `illuminance`→`lux`, `percentage`→`battery_level`); status `bat`/`level` millivolt gauges map to `voltage` (V). MeshCore assigns no firmware packet id, so the record `id` is the deterministic 53-bit fingerprint of *(node id, receive second, source kind)* - re-reads of the same source in the same second collapse into one row via the `telemetry.id` upsert. #### `POST /api/neighbors` @@ -318,7 +322,7 @@ Single trace payload: - Meta: `ingestor`, `lora_freq`, `modem_preset` - `protocol` (optional string; `"meshtastic"`, `"meshcore"`, or `"reticulum"`) - explicit per-record protocol stamp; same semantics as on `POST /api/messages`. -A trace is stored once per `id`. A trace posted without an `id` gets a new row on every POST, so a copy the ingestor sends again after a lost reply (Client delivery, below) is stored twice. Meshtastic traces always carry their packet id. +A trace is stored once per `id`. A trace whose `id` is stored under a `src` it does not name is not stored: the stored trace keeps its hops, and the web app logs it at `warn`; a stored trace without a `src` takes the trace's (SPEC KC4). A trace posted without an `id` gets a new row on every POST, so a copy the ingestor sends again after a lost reply (Client delivery, below) is stored twice. Meshtastic traces always carry their packet id. #### `POST /api/waypoints` diff --git a/web/lib/potato_mesh/application/data_processing.rb b/web/lib/potato_mesh/application/data_processing.rb index e0d6b517..b86ca1f3 100644 --- a/web/lib/potato_mesh/application/data_processing.rb +++ b/web/lib/potato_mesh/application/data_processing.rb @@ -30,6 +30,7 @@ require_relative "data_processing/field_limits" require_relative "data_processing/ingestors" require_relative "data_processing/node_writes" require_relative "data_processing/node_binding" +require_relative "data_processing/record_collisions" require_relative "data_processing/positions" require_relative "data_processing/waypoints" require_relative "data_processing/neighbors" @@ -39,4 +40,5 @@ require_relative "data_processing/telemetry" require_relative "data_processing/decrypted_payloads" require_relative "data_processing/meshcore_chat" require_relative "data_processing/message_scope" +require_relative "data_processing/message_merge" require_relative "data_processing/messages" diff --git a/web/lib/potato_mesh/application/data_processing/message_merge.rb b/web/lib/potato_mesh/application/data_processing/message_merge.rb new file mode 100644 index 00000000..64363a7d --- /dev/null +++ b/web/lib/potato_mesh/application/data_processing/message_merge.rb @@ -0,0 +1,157 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# frozen_string_literal: true + +module PotatoMesh + module App + module DataProcessing + # Columns of a stored message that a later copy is merged against, in + # the order {#stored_message_for_merge} selects them. + MESSAGE_MERGE_COLUMNS = %w[ + from_id to_id text encrypted lora_freq modem_preset channel_name + reply_id emoji portnum ingestor protocol scope + ].freeze + + # Read the stored message that a later copy merges into. + # + # The id hit, the MeshCore content-dedup hit and the INSERT-race + # recovery all read the row through this one lookup (SPEC KC3). + # + # @param db [SQLite3::Database] open database handle. + # @param id [Integer] message id. + # @return [Hash{String=>Object}, nil] the {MESSAGE_MERGE_COLUMNS} keyed by + # name, whether the handle returns rows as hashes or as arrays; nil + # when no row holds +id+. + def stored_message_for_merge(db, id) + row = db.get_first_row( + "SELECT #{MESSAGE_MERGE_COLUMNS.join(", ")} FROM messages WHERE id = ?", + [id], + ) + return row if row.nil? || row.is_a?(Hash) + + MESSAGE_MERGE_COLUMNS.zip(row).to_h + end + + # Merge a later copy of a message into the row stored under its id. + # + # One rule for the id hit, the MeshCore content-dedup hit and the + # INSERT-race recovery (SPEC KC3). The copy is dropped, and the row + # left as it is, when neither names a sender, when the row belongs to + # another protocol than the copy (a +meshtastic+ row, the default, takes + # any), and when a copy of any protocol but MeshCore names another + # sender than the row: its packet names its own sender, so the id was + # reused, by chance or on purpose, and the message stored first stays + # (SPEC KC1, logged at warn as SPEC KC5 limits). A copy naming its + # sender only by number reaches here with the id that number names. + # MeshCore copies name the sender each ingestor's roster resolved, so + # a strictly better-evidenced id replaces the stored one instead (SPEC + # MR3). + # + # Otherwise the copy fills a sender, recipient, text, reply, emoji and + # portnum the row lacks and never replaces one (SPEC KC2). A copy with + # text for an encrypted row is its decrypted form: it clears the + # ciphertext and its signal fields and receive time replace the + # stored ones. +lora_freq+, +modem_preset+ and +channel_name+ follow + # the copy, the first ingestor stays, and +scope+ follows SPEC SC6. + # + # @param db [SQLite3::Database] open database handle. + # @param id [Integer] id of the stored row. + # @param stored [Hash{String=>Object}] the row as + # {#stored_message_for_merge} reads it. + # @param copy [Hash{Symbol=>Object}] the copy's normalised fields: + # +:from_id+, +:to_id+, +:text+, +:encrypted+, +:portnum+, + # +:lora_freq+, +:modem_preset+, +:channel_name+, +:reply_id+, + # +:emoji+, +:ingestor+, +:protocol+, +:scope+, +:hops+, +:path+, + # +:rx_time+, +:rx_iso+, +:sender_present+ (any sender reference, + # numeric included) and +:message+ (the raw payload, read for its + # +channel+, +snr+, +rssi+ and +hop_limit+ keys). + # @return [Array(Boolean, String)] whether the copy merged, and the + # sender the reception is credited to (nil when the copy names none). + def merge_message_copy(db, id, stored, copy) + stored_from = string_or_nil(stored["from_id"]) + protocol = copy[:protocol] + stored_protocol = stored["protocol"] + # Nothing ties a copy without a sender to a row without one. + return [false, nil] if !copy[:sender_present] && stored_from.nil? + # Never merge across protocols (the cross-protocol guard). + return [false, nil] if stored_protocol && stored_protocol != "meshtastic" && stored_protocol != protocol + + updates = {} + from_id = copy[:from_id] + resolved_from_id = from_id + if from_id + fill = stored_from.nil? + if !fill && stored["from_id"] != from_id + if protocol == "meshcore" + fill = meshcore_sender_supersedes?(db, stored["from_id"], from_id) + else + warn_collision( + "Dropped a message copy naming another sender", + context: "data_processing.insert_message", + message_id: id, + stored_from_id: stored_from, + from_id: from_id, + ) + return [false, nil] + end + end + updates["from_id"] = from_id if fill + # Placeholder synthesis and the last-heard touch follow the id that + # survived, so a losing copy grants no liveness to the one it names. + resolved_from_id = fill ? from_id : stored_from + end + + updates["to_id"] = copy[:to_id] if copy[:to_id] && string_or_nil(stored["to_id"]).nil? + + text = copy[:text] + stored_has_text = !string_or_nil(stored["text"]).nil? + # A copy with text for an encrypted row is that packet decrypted. + decrypted = text && !string_or_nil(stored["encrypted"]).nil? + if decrypted + updates["encrypted"] = nil + elsif copy[:encrypted] && !stored_has_text && stored["encrypted"] != copy[:encrypted] + updates["encrypted"] = copy[:encrypted] + end + updates["text"] = text if text && !stored_has_text + + if decrypted + message = copy[:message] + %w[channel snr rssi hop_limit].each { |column| updates[column] = message[column] if message.key?(column) } + updates["hops"] = copy[:hops] unless copy[:hops].nil? + updates["path"] = copy[:path] if copy[:path] + updates["rx_time"] = copy[:rx_time] + updates["rx_iso"] = copy[:rx_iso] + end + + updates["portnum"] = copy[:portnum] if copy[:portnum] && string_or_nil(stored["portnum"]).nil? + %i[lora_freq modem_preset channel_name].each do |column| + value = copy[column] + updates[column.to_s] = value if !value.nil? && stored[column.to_s] != value + end + updates["reply_id"] = copy[:reply_id] if !copy[:reply_id].nil? && stored["reply_id"].nil? + updates["emoji"] = copy[:emoji] if copy[:emoji] && string_or_nil(stored["emoji"]).nil? + updates["ingestor"] = copy[:ingestor] if copy[:ingestor] && string_or_nil(stored["ingestor"]).nil? + updates["scope"] = copy[:scope] if message_scope_supersedes?(stored["scope"], copy[:scope]) + updates["protocol"] = protocol if (stored_protocol.nil? || stored_protocol == "meshtastic") && protocol != "meshtastic" + + unless updates.empty? + assignments = updates.keys.map { |column| "#{column} = ?" }.join(", ") + db.execute("UPDATE messages SET #{assignments} WHERE id = ?", updates.values + [id]) + end + [true, resolved_from_id] + end + end + end +end diff --git a/web/lib/potato_mesh/application/data_processing/messages.rb b/web/lib/potato_mesh/application/data_processing/messages.rb index a4bc2a96..9730c6c1 100644 --- a/web/lib/potato_mesh/application/data_processing/messages.rb +++ b/web/lib/potato_mesh/application/data_processing/messages.rb @@ -81,7 +81,9 @@ module PotatoMesh end # Persist a chat-layer message payload, performing meshcore content - # dedup, decryption, and per-protocol bookkeeping. + # dedup, decryption, and per-protocol bookkeeping. A copy of a stored + # message merges into it through {#merge_message_copy}, which drops a + # copy that names another sender (SPEC KC1-KC3). # # @param db [SQLite3::Database] open database handle. # @param message [Hash] inbound message payload. @@ -125,6 +127,10 @@ module PotatoMesh message["from_num"] ||= canonical_parts[1] end end + # A copy naming its sender only by number names the node that number + # canonicalises to, the node the writes below touch, so the row and + # the merge into a stored copy hold it to that sender (SPEC KC1). + from_id ||= canonical_node_parts(nil, message["from_num"])&.first sender_present = !from_id.nil? || !coerce_integer(message["from_num"]).nil? || !trimmed_from_id.nil? raw_to_id = message["to_id"] @@ -218,6 +224,15 @@ module PotatoMesh protocol, ] + # This copy's fields as the merge into a stored copy reads them. + copy = { + from_id: from_id, to_id: to_id, text: text, encrypted: encrypted, portnum: portnum, + lora_freq: lora_freq, modem_preset: modem_preset, channel_name: channel_name, + reply_id: reply_id, emoji: emoji, ingestor: ingestor, protocol: protocol, scope: scope, + hops: hops, path: path, rx_time: rx_time, rx_iso: rx_iso, + sender_present: sender_present, message: message, + } + # Sender id that survives collapse with any copy already stored (SPEC # MR3). Starts as this copy's own sender and is narrowed below when an # existing row turns out to carry a better-evidenced identity. @@ -318,146 +333,8 @@ module PotatoMesh end end - existing = db.get_first_row( - "SELECT from_id, to_id, text, encrypted, lora_freq, modem_preset, channel_name, reply_id, emoji, portnum, ingestor, protocol, scope FROM messages WHERE id = ?", - [target_id], - ) - if existing - updates = {} - existing_text = existing.is_a?(Hash) ? existing["text"] : existing[2] - existing_text_str = existing_text&.to_s - existing_has_text = existing_text_str && !existing_text_str.strip.empty? - existing_from = existing.is_a?(Hash) ? existing["from_id"] : existing[0] - existing_from_str = existing_from&.to_s - return if !sender_present && (existing_from_str.nil? || existing_from_str.strip.empty?) - existing_encrypted = existing.is_a?(Hash) ? existing["encrypted"] : existing[3] - existing_encrypted_str = existing_encrypted&.to_s - decrypted_precedence = text && existing_encrypted_str && !existing_encrypted_str.strip.empty? - - if from_id - should_update = existing_from_str.nil? || existing_from_str.strip.empty? - if !should_update && existing_from != from_id - # A second copy of the same physical message disagrees about the - # sender. For MeshCore that disagreement is expected (each - # ingestor resolves the sender against its own roster), so the - # better-evidenced id wins rather than the last writer (SPEC - # MR3). Other protocols carry a firmware-assigned sender and - # keep the historical overwrite. - should_update = if protocol == "meshcore" - meshcore_sender_supersedes?(db, existing_from, from_id) - else - true - end - end - updates["from_id"] = from_id if should_update - # Everything downstream (placeholder synthesis, last-heard touch) - # must follow the id that actually survived, so a losing copy - # never grants liveness to the identity it names. - resolved_from_id = should_update ? from_id : (string_or_nil(existing_from) || from_id) - end - - if to_id - existing_to = existing.is_a?(Hash) ? existing["to_id"] : existing[1] - existing_to_str = existing_to&.to_s - should_update = existing_to_str.nil? || existing_to_str.strip.empty? - should_update ||= existing_to != to_id - updates["to_id"] = to_id if should_update - end - - if decrypted_precedence && existing_encrypted_str && !existing_encrypted_str.strip.empty? - updates["encrypted"] = nil if existing_encrypted - elsif encrypted && !existing_has_text - should_update = existing_encrypted_str.nil? || existing_encrypted_str.strip.empty? - should_update ||= existing_encrypted != encrypted - updates["encrypted"] = encrypted if should_update - end - - if text - should_update = existing_text_str.nil? || existing_text_str.strip.empty? - should_update ||= existing_text != text - updates["text"] = text if should_update - end - - if decrypted_precedence - updates["channel"] = message["channel"] if message.key?("channel") - updates["snr"] = message["snr"] if message.key?("snr") - updates["rssi"] = message["rssi"] if message.key?("rssi") - updates["hop_limit"] = message["hop_limit"] if message.key?("hop_limit") - updates["hops"] = hops unless hops.nil? - updates["path"] = path if path - updates["lora_freq"] = lora_freq unless lora_freq.nil? - updates["modem_preset"] = modem_preset if modem_preset - updates["channel_name"] = channel_name if channel_name - updates["rx_time"] = rx_time if rx_time - updates["rx_iso"] = rx_iso if rx_iso - end - - if portnum - existing_portnum = existing.is_a?(Hash) ? existing["portnum"] : existing[9] - existing_portnum_str = existing_portnum&.to_s - should_update = existing_portnum_str.nil? || existing_portnum_str.strip.empty? - should_update ||= existing_portnum != portnum - should_update ||= decrypted_precedence - updates["portnum"] = portnum if should_update - end - - unless lora_freq.nil? - existing_lora = existing.is_a?(Hash) ? existing["lora_freq"] : existing[4] - updates["lora_freq"] = lora_freq if existing_lora != lora_freq - end - - if modem_preset - existing_preset = existing.is_a?(Hash) ? existing["modem_preset"] : existing[5] - existing_preset_str = existing_preset&.to_s - should_update = existing_preset_str.nil? || existing_preset_str.strip.empty? - should_update ||= existing_preset != modem_preset - updates["modem_preset"] = modem_preset if should_update - end - - if channel_name - existing_channel = existing.is_a?(Hash) ? existing["channel_name"] : existing[6] - existing_channel_str = existing_channel&.to_s - should_update = existing_channel_str.nil? || existing_channel_str.strip.empty? - should_update ||= existing_channel != channel_name - updates["channel_name"] = channel_name if should_update - end - - unless reply_id.nil? - existing_reply = existing.is_a?(Hash) ? existing["reply_id"] : existing[7] - updates["reply_id"] = reply_id if existing_reply != reply_id - end - - if emoji - existing_emoji = existing.is_a?(Hash) ? existing["emoji"] : existing[8] - existing_emoji_str = existing_emoji&.to_s - should_update = existing_emoji_str.nil? || existing_emoji_str.strip.empty? - should_update ||= existing_emoji != emoji - updates["emoji"] = emoji if should_update - end - - if ingestor - existing_ingestor = existing.is_a?(Hash) ? existing["ingestor"] : existing[10] - existing_ingestor = string_or_nil(existing_ingestor) - updates["ingestor"] = ingestor if existing_ingestor.nil? - end - - # A later copy fills a NULL scope or names a stored "?", nothing - # more; hops, path, snr and rssi stay with the first ingestor - # (SPEC SC6). - if scope - existing_scope = existing.is_a?(Hash) ? existing["scope"] : existing[12] - updates["scope"] = scope if message_scope_supersedes?(existing_scope, scope) - end - - existing_protocol = existing.is_a?(Hash) ? existing["protocol"] : existing[11] - return if existing_protocol && existing_protocol != "meshtastic" && existing_protocol != protocol - updates["protocol"] = protocol if (existing_protocol.nil? || existing_protocol == "meshtastic") && protocol != "meshtastic" - - unless updates.empty? - assignments = updates.keys.map { |column| "#{column} = ?" }.join(", ") - db.execute("UPDATE messages SET #{assignments} WHERE id = ?", updates.values + [target_id]) - end - else + existing = stored_message_for_merge(db, target_id) + unless existing PotatoMesh::App::Prometheus::MESSAGES_TOTAL.increment begin @@ -466,74 +343,18 @@ module PotatoMesh VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) SQL rescue SQLite3::ConstraintException - existing_row = db.get_first_row( - "SELECT text, encrypted, ingestor, protocol, from_id, scope FROM messages WHERE id = ?", - [msg_id], - ) - existing_text = existing_row.is_a?(Hash) ? existing_row["text"] : existing_row&.[](0) - existing_text_str = existing_text&.to_s - allow_encrypted_update = existing_text_str.nil? || existing_text_str.strip.empty? - existing_encrypted = existing_row.is_a?(Hash) ? existing_row["encrypted"] : existing_row&.[](1) - existing_encrypted_str = existing_encrypted&.to_s - existing_ingestor = existing_row.is_a?(Hash) ? existing_row["ingestor"] : existing_row&.[](2) - existing_ingestor = string_or_nil(existing_ingestor) - existing_fallback_protocol = existing_row.is_a?(Hash) ? existing_row["protocol"] : existing_row&.[](3) - # Guard against cross-protocol contamination in the constraint fallback path, - # mirroring the same guard applied in the primary update path above. - return if existing_fallback_protocol && existing_fallback_protocol != "meshtastic" && existing_fallback_protocol != protocol - decrypted_precedence = text && existing_encrypted_str && !existing_encrypted_str.strip.empty? - - fallback_updates = {} - if from_id - # Same sender-resolution rule as the primary update path (SPEC - # MR3); this branch is the INSERT race between two ingestors - # posting the same message, exactly where competing sender ids - # meet. - existing_fallback_from = existing_row.is_a?(Hash) ? existing_row["from_id"] : existing_row&.[](4) - existing_fallback_from_str = string_or_nil(existing_fallback_from) - supersedes = - existing_fallback_from_str.nil? || - existing_fallback_from_str == from_id || - protocol != "meshcore" || - meshcore_sender_supersedes?(db, existing_fallback_from_str, from_id) - fallback_updates["from_id"] = from_id if supersedes - resolved_from_id = supersedes ? from_id : existing_fallback_from_str - end - fallback_updates["to_id"] = to_id if to_id - fallback_updates["text"] = text if text - fallback_updates["encrypted"] = encrypted if encrypted && allow_encrypted_update - fallback_updates["portnum"] = portnum if portnum - if decrypted_precedence - fallback_updates["channel"] = message["channel"] if message.key?("channel") - fallback_updates["snr"] = message["snr"] if message.key?("snr") - fallback_updates["rssi"] = message["rssi"] if message.key?("rssi") - fallback_updates["hop_limit"] = message["hop_limit"] if message.key?("hop_limit") - fallback_updates["hops"] = hops unless hops.nil? - fallback_updates["path"] = path if path - fallback_updates["portnum"] = portnum if portnum - fallback_updates["lora_freq"] = lora_freq unless lora_freq.nil? - fallback_updates["modem_preset"] = modem_preset if modem_preset - fallback_updates["channel_name"] = channel_name if channel_name - fallback_updates["rx_time"] = rx_time if rx_time - fallback_updates["rx_iso"] = rx_iso if rx_iso - else - fallback_updates["lora_freq"] = lora_freq unless lora_freq.nil? - fallback_updates["modem_preset"] = modem_preset if modem_preset - fallback_updates["channel_name"] = channel_name if channel_name - end - fallback_updates["reply_id"] = reply_id unless reply_id.nil? - fallback_updates["emoji"] = emoji if emoji - fallback_updates["ingestor"] = ingestor if ingestor && existing_ingestor.nil? - # Same flood-scope precedence as the update path (SPEC SC6). - existing_fallback_scope = existing_row.is_a?(Hash) ? existing_row["scope"] : existing_row&.[](5) - fallback_updates["scope"] = scope if message_scope_supersedes?(existing_fallback_scope, scope) - fallback_updates["protocol"] = protocol if (existing_fallback_protocol.nil? || existing_fallback_protocol == "meshtastic") && protocol != "meshtastic" - unless fallback_updates.empty? - assignments = fallback_updates.keys.map { |column| "#{column} = ?" }.join(", ") - db.execute("UPDATE messages SET #{assignments} WHERE id = ?", fallback_updates.values + [msg_id]) - end + # Another ingestor's copy landed between the lookup and the + # INSERT, precisely where two ingestors' copies meet (SPEC MR3): + # merge into it as an id hit does (SPEC KC3). + target_id = msg_id + existing = stored_message_for_merge(db, target_id) end end + + if existing + merged, resolved_from_id = merge_message_copy(db, target_id, existing, copy) + return unless merged + end end stored_decrypted = nil diff --git a/web/lib/potato_mesh/application/data_processing/positions.rb b/web/lib/potato_mesh/application/data_processing/positions.rb index ebc99332..f8d133ea 100644 --- a/web/lib/potato_mesh/application/data_processing/positions.rb +++ b/web/lib/potato_mesh/application/data_processing/positions.rb @@ -20,7 +20,8 @@ module PotatoMesh # Persist a position payload, populate the +nodes+ table for newly seen # senders, and update node rows with the freshest GPS fields. A payload # carrying +public_key+ updates the node row only under the key the row - # is bound to (SPEC NI3). + # is bound to (SPEC NI3), and a position row stored under its id for + # another node is left as it is (SPEC KC4). # # @param db [SQLite3::Database] open database handle. # @param payload [Hash] inbound position payload. @@ -220,7 +221,11 @@ module PotatoMesh payload_b64=COALESCE(excluded.payload_b64,positions.payload_b64), ingestor=COALESCE(NULLIF(positions.ingestor,''), excluded.ingestor), protocol=COALESCE(NULLIF(positions.protocol,'meshtastic'), excluded.protocol) + WHERE positions.node_id IS NULL OR positions.node_id = excluded.node_id SQL + # Another node's position under a stored id leaves that row alone; + # the record still refreshes its own node below (SPEC KC4). + warn_dropped_record(db, "positions", "node_id", pos_id, node_id, context: "data_processing.insert_position") if db.changes.zero? end # A MeshCore advert's position carries the advert's full key: under diff --git a/web/lib/potato_mesh/application/data_processing/record_collisions.rb b/web/lib/potato_mesh/application/data_processing/record_collisions.rb new file mode 100644 index 00000000..d5dca5b1 --- /dev/null +++ b/web/lib/potato_mesh/application/data_processing/record_collisions.rb @@ -0,0 +1,176 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# frozen_string_literal: true + +module PotatoMesh + module App + module DataProcessing + # Rate limit of the warnings that dropped message copies and records + # log (SPEC KC5). + # + # A token holder or a compromised ingestor can post a whole batch under + # ids other nodes hold, and each record would log one warning. Per + # logging context the limiter lets {LIMIT} warnings through in a window + # of {WINDOW_SECONDS} that opens with the first, and counts the rest. + # The context's first warning after the window ends opens the next + # window and carries that count, so a flood costs one line more. No + # timer runs: a count waits for the context's next warning. All + # methods are thread-safe. + class CollisionWarningLimiter + # Warnings logged per context in one window. + LIMIT = 10 + + # Length of a window, in seconds. + WINDOW_SECONDS = 60 + + # @return [Integer] warnings logged per context in one window. + attr_reader :limit + + # @return [Numeric] length of a window, in seconds. + attr_reader :window_seconds + + # Start with no window open. + # + # @param limit [Integer] warnings logged per context in one window. + # @param window_seconds [Numeric] length of a window, in seconds. + # @param clock [#call] returns the current time in seconds; the + # monotonic clock unless a test injects its own. + # @return [void] + def initialize(limit: LIMIT, window_seconds: WINDOW_SECONDS, clock: -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) }) + @limit = limit + @window_seconds = window_seconds + @clock = clock + @mutex = Mutex.new + @windows = {} + end + + # Count one warning of +context+. + # + # @param context [String] logging context of the warning. + # @return [Array(Boolean, Integer)] whether to log the warning, and + # the count the context's previous window suppressed when this + # warning opens a new window (0 otherwise). + def admit(context) + @mutex.synchronize do + now = @clock.call + window = @windows[context] + reported = 0 + unless window && now - window[:opened] < @window_seconds + reported = window ? window[:suppressed] : 0 + window = @windows[context] = { opened: now, logged: 0, suppressed: 0 } + end + if window[:logged] < @limit + window[:logged] += 1 + [true, reported] + else + window[:suppressed] += 1 + [false, 0] + end + end + end + + # Forget every window, as a new process starts with none. + # + # @return [void] + def reset! + @mutex.synchronize { @windows.clear } + end + end + + # Sinatra builds an app instance per request, so the limiter lives on + # the module for the lifetime of the process. + @collision_warning_limiter = CollisionWarningLimiter.new + + # @return [CollisionWarningLimiter] the limiter every collision warning + # of this process passes (SPEC KC5). + def self.collision_warning_limiter + @collision_warning_limiter + end + + # The process-wide {CollisionWarningLimiter}. + # + # @return [CollisionWarningLimiter] the shared limiter. + def collision_warning_limiter + PotatoMesh::App::DataProcessing.collision_warning_limiter + end + + # Log the warning of a dropped message copy or record unless its + # context logged {CollisionWarningLimiter::LIMIT} in the current window + # (SPEC KC5). A warning that opens a new window is preceded by one + # +Suppressed collision warnings+ line carrying the count the previous + # window suppressed, when it suppressed any. + # + # @param text [String] warning text. + # @param context [String] logging context, the unit of the limit. + # @param fields [Hash] structured fields of the warning. + # @return [void] + def warn_collision(text, context:, **fields) + limiter = collision_warning_limiter + logged, suppressed = limiter.admit(context) + if suppressed.positive? + warn_log( + "Suppressed collision warnings", + context: context, + suppressed: suppressed, + window_seconds: limiter.window_seconds, + ) + end + warn_log(text, context: context, **fields) if logged + end + + # Log a position, telemetry reading or trace that left the stored row + # alone because its id is stored for another node (SPEC KC4, limited + # by SPEC KC5). + # + # Each of the three upserts updates a stored row only when the row + # names no node or the record's own (+src+ for traces), so an upsert + # that changed no row met another node's record under the same id. A + # trace's +src+ is a node number and is logged as the id it names. + # + # @param db [SQLite3::Database] open database handle. + # @param table [String] +positions+, +telemetry+ or +traces+. + # @param column [String] the column naming the node: +node_id+, or + # +src+ for traces. + # @param id [Integer] id the dropped record shares with the stored row. + # @param incoming [String, Integer, nil] the node the dropped record names. + # @param context [String] logging context of the writer. + # @return [void] + def warn_dropped_record(db, table, column, id, incoming, context:) + stored = db.get_first_value("SELECT #{column} FROM #{table} WHERE id = ?", [id]) + warn_collision( + "Dropped a record whose id another node holds", + context: context, + table: table, + id: id, + "stored_#{column}": collision_node_label(stored), + column.to_sym => collision_node_label(incoming), + ) + end + + # How a collision warning names a node: a node number as the + # canonical +!xxxxxxxx+ id it names (a negative one names none and + # stays a number), a node id as stored. + # + # @param node_ref [String, Integer, nil] node id, node number or nil. + # @return [String, Integer, nil] the id a node number names, otherwise + # +node_ref+. + def collision_node_label(node_ref) + return node_ref unless node_ref.is_a?(Integer) + + canonical_node_parts(node_ref)&.first || node_ref + end + end + end +end diff --git a/web/lib/potato_mesh/application/data_processing/telemetry.rb b/web/lib/potato_mesh/application/data_processing/telemetry.rb index a53986a5..f68ddfab 100644 --- a/web/lib/potato_mesh/application/data_processing/telemetry.rb +++ b/web/lib/potato_mesh/application/data_processing/telemetry.rb @@ -335,7 +335,9 @@ module PotatoMesh private :coerce_float_array_json - # Persist a telemetry packet and refresh the related node row. + # Persist a telemetry packet and refresh the related node row. A + # telemetry row stored under its id for another node is left as it is + # (SPEC KC4). # # @param db [SQLite3::Database] open database handle. # @param payload [Hash] inbound telemetry payload. @@ -575,7 +577,11 @@ module PotatoMesh ingestor=COALESCE(NULLIF(telemetry.ingestor,''), excluded.ingestor), protocol=COALESCE(NULLIF(telemetry.protocol,'meshtastic'), excluded.protocol), telemetry_type=COALESCE(excluded.telemetry_type,telemetry.telemetry_type)#{EXTENDED_TELEMETRY_UPSERT_SQL} + WHERE telemetry.node_id IS NULL OR telemetry.node_id = excluded.node_id SQL + # Another node's reading under a stored id leaves that row alone; + # the record still refreshes its own node below (SPEC KC4). + warn_dropped_record(db, "telemetry", "node_id", telemetry_id, node_id, context: "data_processing.insert_telemetry") if db.changes.zero? end update_node_from_telemetry( diff --git a/web/lib/potato_mesh/application/data_processing/traces.rb b/web/lib/potato_mesh/application/data_processing/traces.rb index 33279016..5519adcb 100644 --- a/web/lib/potato_mesh/application/data_processing/traces.rb +++ b/web/lib/potato_mesh/application/data_processing/traces.rb @@ -52,7 +52,8 @@ module PotatoMesh hop_entries.filter_map { |entry| coerce_trace_node_id(entry) } end - # Persist a traceroute observation and its hop path. + # Persist a traceroute observation and its hop path. A trace stored + # under its id for another source keeps its row and hops (SPEC KC4). # # @param db [SQLite3::Database] open database handle. # @param payload [Hash] traceroute payload as produced by the ingestor. @@ -111,7 +112,14 @@ module PotatoMesh elapsed_ms=COALESCE(excluded.elapsed_ms,traces.elapsed_ms), ingestor=COALESCE(NULLIF(traces.ingestor,''), excluded.ingestor), protocol=COALESCE(NULLIF(traces.protocol,'meshtastic'), excluded.protocol) + WHERE traces.src IS NULL OR traces.src = excluded.src SQL + # Another source's trace under a stored id leaves that trace and its + # hops alone (SPEC KC4). + if db.changes.zero? + warn_dropped_record(db, "traces", "src", trace_identifier, src, context: "data_processing.insert_trace") + return + end trace_id = trace_identifier || db.last_insert_row_id return unless trace_id diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index af03be54..04808a46 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -3872,7 +3872,8 @@ RSpec.describe "Potato Mesh Sinatra app" do with_db(readonly: true) do |db| db.results_as_hash = true row = db.get_first_row("SELECT text, ingestor FROM messages WHERE id = ?", [first_payload["id"]]) - expect(row["text"]).to eq("updated text") + # A later copy never replaces the stored text (SPEC KC2). + expect(row["text"]).to eq("first reporter") expect(row["ingestor"]).to eq(FIRST_MESSAGE_INGESTOR_ID) end diff --git a/web/spec/data_processing_spec.rb b/web/spec/data_processing_spec.rb index 7b47790a..22b4f125 100644 --- a/web/spec/data_processing_spec.rb +++ b/web/spec/data_processing_spec.rb @@ -2370,8 +2370,8 @@ RSpec.describe PotatoMesh::App::DataProcessing do end # The rank-based resolution is MeshCore-specific: a firmware-assigned - # Meshtastic sender keeps the historical last-writer-wins overwrite, so the - # rule never changes attribution for the other protocol. + # Meshtastic sender that differs means a reused packet id, so that copy is + # dropped (SPEC KC1) and the rank never decides a Meshtastic attribution. it "ranks a nil, blank, or unknown sender id as zero and never supersedes on it" do db = open_db expect(dp.meshcore_sender_rank(db, nil)).to eq(0) @@ -2383,7 +2383,7 @@ RSpec.describe PotatoMesh::App::DataProcessing do db&.close end - it "keeps last-writer-wins for a meshtastic message with a differing sender" do + it "drops a meshtastic copy with a differing sender and keeps the stored one (SPEC KC1)" do db = open_db first_sender = "!aaaa1111" second_sender = "!bbbb2222" @@ -2400,7 +2400,7 @@ RSpec.describe PotatoMesh::App::DataProcessing do dp.insert_message(db, base.merge("from_id" => first_sender)) dp.insert_message(db, base.merge("from_id" => second_sender)) - expect(db.get_first_value("SELECT from_id FROM messages WHERE id = 7101")).to eq(second_sender) + expect(db.get_first_value("SELECT from_id FROM messages WHERE id = 7101")).to eq(first_sender) ensure db&.close end @@ -3138,7 +3138,7 @@ RSpec.describe PotatoMesh::App::DataProcessing do end.new end - it "updates reply_id when the existing row references a different reply_id" do + it "keeps the stored reply_id when a later copy references a different one (SPEC KC2)" do db = open_db base = { "from_id" => "!aabbccdd", @@ -3149,7 +3149,7 @@ RSpec.describe PotatoMesh::App::DataProcessing do } msg_update_harness.insert_message(db, base.merge("id" => 8001)) msg_update_harness.insert_message(db, base.merge("id" => 8001, "reply_id" => 200)) - expect(db.get_first_value("SELECT reply_id FROM messages WHERE id = 8001")).to eq(200) + expect(db.get_first_value("SELECT reply_id FROM messages WHERE id = 8001")).to eq(100) ensure db&.close end @@ -3222,23 +3222,18 @@ RSpec.describe PotatoMesh::App::DataProcessing do # The fallback path is taken when the SELECT-before-INSERT misses but the # INSERT itself trips the PK constraint — i.e., a concurrent ingestor has # already inserted the row in between. We simulate that race by seeding a - # row, then forcing the existing-row SELECT to return nil so the INSERT - # path runs and trips the constraint deterministically. + # row, then hiding it from the copy's first lookup so the INSERT path runs + # and trips the constraint deterministically (+hide_stored_message_once+). it "applies fallback updates when INSERT trips a constraint violation" do db = open_db fb_harness.insert_message(db, base_msg.merge("id" => 9001, "text" => "first", "ingestor" => "!0000000a")) + hide_stored_message_once(db) - allow(db).to receive(:get_first_row).and_wrap_original do |original, sql, *args| - if sql.include?("SELECT from_id, to_id, text, encrypted, lora_freq") - nil - else - original.call(sql, *args) - end - end + fb_harness.insert_message(db, base_msg.merge("id" => 9001, "text" => "second", "emoji" => "e", "ingestor" => "!0000000b")) - fb_harness.insert_message(db, base_msg.merge("id" => 9001, "text" => "second", "ingestor" => "!0000000b")) - - expect(db.get_first_value("SELECT text FROM messages WHERE id = 9001")).to eq("second") + # A later copy only fills (SPEC KC2): the stored text stays, the + # missing emoji is filled. + expect(db.get_first_row("SELECT text, emoji FROM messages WHERE id = 9001")).to eq("text" => "first", "emoji" => "e") # First-write-wins for ingestor: the existing value (!0000000a) is preserved. expect(db.get_first_value("SELECT ingestor FROM messages WHERE id = 9001")).to eq("!0000000a") ensure @@ -3252,14 +3247,7 @@ RSpec.describe PotatoMesh::App::DataProcessing do "encrypted" => "BLOB", "text" => nil, )) - - allow(db).to receive(:get_first_row).and_wrap_original do |original, sql, *args| - if sql.include?("SELECT from_id, to_id, text, encrypted, lora_freq") - nil - else - original.call(sql, *args) - end - end + hide_stored_message_once(db) fb_harness.insert_message(db, base_msg.merge( "id" => 9002, @@ -3272,6 +3260,9 @@ RSpec.describe PotatoMesh::App::DataProcessing do expect(db.get_first_value("SELECT text FROM messages WHERE id = 9002")).to eq("decrypted") expect(db.get_first_value("SELECT lora_freq FROM messages WHERE id = 9002")).to eq(869525) expect(db.get_first_value("SELECT modem_preset FROM messages WHERE id = 9002")).to eq("MEDIUM_SLOW") + # The recovery merges as the id hit does (SPEC KC3), so the decrypted + # copy clears the stored ciphertext there too. + expect(db.get_first_value("SELECT encrypted FROM messages WHERE id = 9002")).to be_nil ensure db&.close end @@ -3279,16 +3270,7 @@ RSpec.describe PotatoMesh::App::DataProcessing do # SPEC MR3 names the insert-race fallback as "precisely where two ingestors' # copies meet", so the sender-resolution rule must hold on this path too, in # both DB result modes. Force the fallback the same way as above: seed the - # row, then stub the existing-row SELECT to nil so the INSERT trips the PK. - def stub_missing_existing_row(db) - allow(db).to receive(:get_first_row).and_wrap_original do |original, sql, *args| - if sql.include?("SELECT from_id, to_id, text, encrypted, lora_freq") - nil - else - original.call(sql, *args) - end - end - end + # row, then hide it from the copy's first lookup so the INSERT trips the PK. # Two hex node ids so canonical normalisation keeps them verbatim; a synth # (rank 0) and a key-backed real (rank 2) so the copies disagree by evidence. @@ -3315,7 +3297,7 @@ RSpec.describe PotatoMesh::App::DataProcessing do db = open_db seed_rank_nodes(db) fb_harness.insert_message(db, meshcore_copy(9201, synth_sender)) - stub_missing_existing_row(db) + hide_stored_message_once(db) fb_harness.insert_message(db, meshcore_copy(9201, real_sender)) @@ -3331,7 +3313,7 @@ RSpec.describe PotatoMesh::App::DataProcessing do db = SQLite3::Database.new(PotatoMesh::Config.db_path) seed_rank_nodes(db) fb_harness.insert_message(db, meshcore_copy(9202, real_sender)) - stub_missing_existing_row(db) + hide_stored_message_once(db) fb_harness.insert_message(db, meshcore_copy(9202, synth_sender)) @@ -3344,14 +3326,14 @@ RSpec.describe PotatoMesh::App::DataProcessing do it "skips fallback from_id handling when the racing copy carries no sender" do db = open_db fb_harness.insert_message(db, meshcore_copy(9203, real_sender)) - stub_missing_existing_row(db) + hide_stored_message_once(db) fb_harness.insert_message(db, { "id" => 9203, "to_id" => "^all", "channel" => 3, "text" => "senderless", "protocol" => "meshcore" }) - # No incoming sender → the stored attribution is untouched, and the text - # still reconciles through the fallback. + # No incoming sender → the stored attribution is untouched, and so is + # the stored text: a later copy only fills (SPEC KC2). expect(db.get_first_value("SELECT from_id FROM messages WHERE id = 9203")).to eq(real_sender) - expect(db.get_first_value("SELECT text FROM messages WHERE id = 9203")).to eq("senderless") + expect(db.get_first_value("SELECT text FROM messages WHERE id = 9203")).to eq("shared") ensure db&.close end diff --git a/web/spec/ingest_collisions_spec.rb b/web/spec/ingest_collisions_spec.rb new file mode 100644 index 00000000..3f292de7 --- /dev/null +++ b/web/spec/ingest_collisions_spec.rb @@ -0,0 +1,519 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# frozen_string_literal: true + +require "spec_helper" +require "json" +require "sqlite3" +require_relative "support/data_processing_harness" + +# A packet id that another sender or node reuses (SPEC KC1-KC5; ACCEPTANCE +# KC-A1-KC-A6). A message copy naming another sender than the stored row, +# by id or only by number, and a position, telemetry reading or trace from +# another node, leave the stored row as it is and log one warning, at most +# ten a minute per writer; a later copy of a message fills the recipient, +# text, reply, emoji and portnum the stored row lacks and never replaces +# them. Copies of one packet still merge. +RSpec.describe "Ingest identity collisions" do + let(:app) { Sinatra::Application } + let(:api_token) { "ingest-collisions-token" } + let(:auth_headers) do + { "CONTENT_TYPE" => "application/json", "HTTP_AUTHORIZATION" => "Bearer #{api_token}" } + end + let(:now) { Time.now.to_i } + # The warning a dropped message copy logs (SPEC KC1). + let(:message_drop) { "Dropped a message copy naming another sender" } + # The warning a dropped position, reading or trace logs (SPEC KC4). + let(:record_drop) { "Dropped a record whose id another node holds" } + + # Yield a handle on the spec database, rows as hashes, and close it. + # + # @yieldparam db [SQLite3::Database] open database handle. + # @return [Object] the block's result. + def with_db + db = SQLite3::Database.new(PotatoMesh::Config.db_path) + db.results_as_hash = true + db.busy_timeout = PotatoMesh::Config.db_busy_timeout_ms + yield db + ensure + db&.close + end + + # The first row +sql+ selects. + # + # @param sql [String] query. + # @param params [Array] bind values. + # @return [Hash, nil] the row, or nil without one. + def stored(sql, params = []) + with_db { |db| db.get_first_row(sql, params) } + end + + # Every row +sql+ selects. + # + # @param sql [String] query. + # @param params [Array] bind values. + # @return [Array] the rows. + def stored_rows(sql, params = []) + with_db { |db| db.execute(sql, params) } + end + + # POST +body+ as JSON to +path+ and expect the route to accept it. + # + # @param path [String] ingest route. + # @param body [Object] request body. + # @return [void] + def post_ok(path, body) + post path, body.to_json, auth_headers + expect(last_response.status).to eq(201), "#{path} answered #{last_response.status}: #{last_response.body}" + end + + # A Meshtastic text message on the broadcast channel. + # + # @param fields [Hash] fields merged over the defaults. + # @return [Hash] message record. + def message(fields) + { "rx_time" => now, "channel" => 0, "portnum" => "TEXT_MESSAGE_APP", "protocol" => "meshtastic", "to_id" => "^all" }.merge(fields) + end + + # Record the app's warnings from here on. + # + # @return [Array] each warning's text and metadata, + # appended as they are logged. + def capture_warnings + lines = [] + allow_any_instance_of(Sinatra::Application).to receive(:warn_log) { |_app, text, **metadata| lines << [text, metadata] } + lines + end + + before do + @original_token = ENV["API_TOKEN"] + ENV["API_TOKEN"] = api_token + with_db do |db| + %w[messages nodes positions telemetry neighbors traces trace_hops waypoints ingestors].each do |table| + db.execute("DELETE FROM #{table}") + end + end + PotatoMesh::App::ApiCache.invalidate_all + end + + after do + @original_token.nil? ? ENV.delete("API_TOKEN") : ENV["API_TOKEN"] = @original_token + PotatoMesh::App::ApiCache.invalidate_all + end + + describe "a message copy naming another sender (KC1)" do + it "leaves the stored message alone and logs the id and both senders" do + warnings = capture_warnings + post_ok("/api/messages", message("id" => 1_111_111_111, "rx_time" => now - 600, "from_id" => "!aaaa0001", "text" => "original")) + post_ok("/api/messages", message( + "id" => 1_111_111_111, "from_id" => "!bbbb0002", "to_id" => "!cccc0003", "text" => "forged", "reply_id" => 7, "emoji" => "x", + )) + + expect(stored("SELECT from_id, to_id, text, rx_time, reply_id, emoji FROM messages WHERE id = 1111111111")).to eq( + "from_id" => "!aaaa0001", "to_id" => "^all", "text" => "original", "rx_time" => now - 600, "reply_id" => nil, "emoji" => nil, + ) + expect(warnings.select { |text, _| text == message_drop }).to eq([[ + message_drop, + { context: "data_processing.insert_message", message_id: 1_111_111_111, stored_from_id: "!aaaa0001", from_id: "!bbbb0002" }, + ]]) + end + + it "creates and refreshes no node for the dropped copy" do + post_ok("/api/messages", message("id" => 1_111_111_112, "rx_time" => now - 600, "from_id" => "!aaaa0001", "text" => "original")) + post_ok("/api/messages", message("id" => 1_111_111_112, "from_id" => "!bbbb0002", "to_id" => "!cccc0003", "text" => "forged")) + + expect(stored_rows("SELECT node_id, last_heard FROM nodes ORDER BY node_id")).to eq([{ "node_id" => "!aaaa0001", "last_heard" => now - 600 }]) + end + + it "keeps the first message of a natural collision and drops the later one" do + eight_days_ago = now - (8 * 86_400) + post_ok("/api/messages", message("id" => 2_222_222_222, "rx_time" => eight_days_ago, "from_id" => "!cccc0003", "text" => "C, eight days ago")) + post_ok("/api/messages", message("id" => 2_222_222_222, "from_id" => "!dddd0004", "text" => "D, just now")) + + expect(stored("SELECT from_id, text, rx_time FROM messages WHERE id = 2222222222")).to eq( + "from_id" => "!cccc0003", "text" => "C, eight days ago", "rx_time" => eight_days_ago, + ) + end + + it "still completes a copy without a sender with a later copy naming one (guard, #108)" do + post_ok("/api/messages", message("id" => 1_111_111_114, "from_id" => nil, "text" => "anon first")) + post_ok("/api/messages", message("id" => 1_111_111_114, "from_id" => "!aaaa0001", "text" => "anon first")) + + expect(stored("SELECT from_id FROM messages WHERE id = 1111111114")).to eq("from_id" => "!aaaa0001") + end + + it "still fills an encrypted copy from a decrypted copy of the same packet (guard)" do + post_ok("/api/messages", message("id" => 1_111_111_115, "from_id" => "!eeee0005", "encrypted" => "AAECAwQ=", "portnum" => nil, "hops" => 2)) + post_ok("/api/messages", message("id" => 1_111_111_115, "from_id" => "!eeee0005", "text" => "decrypted", "hops" => 1)) + + expect(stored("SELECT text, encrypted, portnum, hops FROM messages WHERE id = 1111111115")).to eq( + "text" => "decrypted", "encrypted" => nil, "portnum" => "TEXT_MESSAGE_APP", "hops" => 1, + ) + end + end + + describe "a copy naming its sender only by number (KC1)" do + it "drops a copy whose number names another sender than the stored row, and logs it" do + warnings = capture_warnings + post_ok("/api/messages", message("id" => 1_111_111_116, "rx_time" => now - 600, "from_id" => "!e0e00007", "encrypted" => "AAECAwQ=", "portnum" => nil)) + post_ok("/api/messages", message("id" => 1_111_111_116, "from_num" => 0xf0f00008, "text" => "forged by number")) + + expect(stored("SELECT from_id, text, encrypted, rx_time FROM messages WHERE id = 1111111116")).to eq( + "from_id" => "!e0e00007", "text" => nil, "encrypted" => "AAECAwQ=", "rx_time" => now - 600, + ) + expect(stored_rows("SELECT node_id, last_heard FROM nodes ORDER BY node_id")).to eq([{ "node_id" => "!e0e00007", "last_heard" => now - 600 }]) + expect(warnings.select { |text, _| text == message_drop }).to eq([[ + message_drop, + { context: "data_processing.insert_message", message_id: 1_111_111_116, stored_from_id: "!e0e00007", from_id: "!f0f00008" }, + ]]) + end + + it "stores the id the number names and holds later copies to it" do + post_ok("/api/messages", message("id" => 1_111_111_117, "from_num" => 0xf0f00008, "text" => "by number")) + post_ok("/api/messages", message("id" => 1_111_111_117, "from_id" => "!e0e00007", "text" => "by number")) + + expect(stored("SELECT from_id FROM messages WHERE id = 1111111117")).to eq("from_id" => "!f0f00008") + end + + it "still fills an encrypted copy from its decrypted copy (guard)" do + post_ok("/api/messages", message("id" => 1_111_111_118, "from_id" => "!e0e00007", "encrypted" => "AAECAwQ=", "portnum" => nil)) + post_ok("/api/messages", message("id" => 1_111_111_118, "from_num" => 0xe0e00007, "text" => "decrypted")) + + expect(stored("SELECT from_id, text, encrypted FROM messages WHERE id = 1111111118")).to eq( + "from_id" => "!e0e00007", "text" => "decrypted", "encrypted" => nil, + ) + end + + it "checks a payload it decrypts against the sender the number names (NI1)" do + warnings = capture_warnings + allow_any_instance_of(Sinatra::Application).to receive(:decrypt_meshtastic_message).and_return( + { text: nil, portnum: 4, payload: "decrypted".b, channel_name: nil }, + ) + allow(PotatoMesh::App::Meshtastic::PayloadDecoder).to receive(:decode).and_return( + "type" => "NODEINFO_APP", + "payload" => { "user" => { "id" => "!b2b2b2b2", "long_name" => "Mallory", "short_name" => "MLRY" } }, + ) + post_ok("/api/messages", message("id" => 1_111_111_119, "from_num" => 0xf0f00008, "encrypted" => "AAECAwQ=", "portnum" => nil)) + + expect(warnings).to include([ + "Dropped decrypted payload naming another node", + hash_including(message_id: 1_111_111_119, from_id: "!f0f00008", node_id: "!b2b2b2b2"), + ]) + expect(stored_rows("SELECT node_id, long_name FROM nodes")).to eq([{ "node_id" => "!f0f00008", "long_name" => "Meshtastic 0008" }]) + expect(stored("SELECT from_id, encrypted FROM messages WHERE id = 1111111119")).to eq("from_id" => "!f0f00008", "encrypted" => "AAECAwQ=") + end + end + + describe "a later copy of a stored message (KC2)" do + it "never replaces the stored recipient, text, reply, emoji or portnum" do + post_ok("/api/messages", message( + "id" => 1_111_111_121, "from_id" => "!aaaa0001", "text" => "original", "reply_id" => 100, "emoji" => "a", + )) + post_ok("/api/messages", message( + "id" => 1_111_111_121, "from_id" => "!aaaa0001", "to_id" => "!cccc0003", "text" => "rewritten", "reply_id" => 200, + "emoji" => "b", "portnum" => "REACTION_APP", + )) + + expect(stored("SELECT to_id, text, reply_id, emoji, portnum FROM messages WHERE id = 1111111121")).to eq( + "to_id" => "^all", "text" => "original", "reply_id" => 100, "emoji" => "a", "portnum" => "TEXT_MESSAGE_APP", + ) + end + + it "fills the recipient, text, reply, emoji and portnum the stored copy lacks" do + post_ok("/api/messages", message("id" => 1_111_111_122, "from_id" => "!aaaa0001", "to_id" => nil, "portnum" => nil)) + post_ok("/api/messages", message( + "id" => 1_111_111_122, "from_id" => "!aaaa0001", "text" => "filled", "reply_id" => 5, "emoji" => "e", + )) + + expect(stored("SELECT to_id, text, reply_id, emoji, portnum FROM messages WHERE id = 1111111122")).to eq( + "to_id" => "^all", "text" => "filled", "reply_id" => 5, "emoji" => "e", "portnum" => "TEXT_MESSAGE_APP", + ) + end + + it "never replaces the stored text of a MeshCore message" do + meshcore = { "from_id" => "!aaaa0001", "protocol" => "meshcore", "channel" => 1, "channel_name" => "#test" } + post_ok("/api/messages", message(meshcore.merge("id" => 1_111_111_123, "text" => "first words"))) + post_ok("/api/messages", message(meshcore.merge("id" => 1_111_111_123, "text" => "other words", "emoji" => "m"))) + + expect(stored("SELECT text, emoji FROM messages WHERE id = 1111111123")).to eq("text" => "first words", "emoji" => "m") + end + end + + describe "the INSERT race (KC1, KC3)" do + include_context "with isolated db" + + let(:dp) { DataProcessingHarness.build.new } + + it "leaves the stored row alone when the racing copy names another sender, and logs it" do + db = open_db + allow(dp).to receive(:warn_log) + dp.insert_message(db, message("id" => 1_111_111_131, "from_id" => "!aaaa0001", "text" => "original")) + hide_stored_message_once(db) + + dp.insert_message(db, message("id" => 1_111_111_131, "from_id" => "!bbbb0002", "to_id" => "!cccc0003", "text" => "forged")) + + expect(db.get_first_row("SELECT from_id, to_id, text FROM messages WHERE id = 1111111131")).to eq( + "from_id" => "!aaaa0001", "to_id" => "^all", "text" => "original", + ) + expect(dp).to have_received(:warn_log).with( + message_drop, + hash_including(message_id: 1_111_111_131, stored_from_id: "!aaaa0001", from_id: "!bbbb0002"), + ) + ensure + db&.close + end + + it "applies the same rule over a connection that returns rows as arrays" do + db = SQLite3::Database.new(PotatoMesh::Config.db_path) + dp.insert_message(db, message("id" => 1_111_111_133, "from_id" => "!aaaa0001", "text" => "original")) + hide_stored_message_once(db) + + dp.insert_message(db, message("id" => 1_111_111_133, "from_id" => "!bbbb0002", "text" => "forged")) + + expect(db.get_first_row("SELECT from_id, text FROM messages WHERE id = 1111111133")).to eq(["!aaaa0001", "original"]) + ensure + db&.close + end + end + + describe "merge_message_copy, the one merge of every collapse path (KC3)" do + include_context "with isolated db" + + let(:dp) { DataProcessingHarness.build.new } + let(:row_id) { 1_111_111_141 } + + # Store one message row and read it back as the merge does. + # + # @param db [SQLite3::Database] open database handle. + # @param columns [Hash] stored columns merged over the required ones. + # @return [Hash] the row as +stored_message_for_merge+ returns it. + def seed_row(db, columns) + row = { "id" => row_id, "rx_time" => now - 60, "rx_iso" => Time.at(now - 60).utc.iso8601, "protocol" => "meshtastic" }.merge(columns) + db.execute("INSERT INTO messages(#{row.keys.join(",")}) VALUES (#{Array.new(row.size, "?").join(",")})", row.values) + dp.stored_message_for_merge(db, row_id) + end + + # A later copy from +!aaaa0001+, every field absent unless given. + # + # @param fields [Hash{Symbol=>Object}] copy fields merged over the defaults. + # @return [Hash{Symbol=>Object}] the copy as +insert_message+ builds it. + def later_copy(fields = {}) + { + from_id: "!aaaa0001", to_id: nil, text: nil, encrypted: nil, portnum: nil, lora_freq: nil, modem_preset: nil, + channel_name: nil, reply_id: nil, emoji: nil, ingestor: nil, protocol: "meshtastic", scope: nil, hops: nil, + path: nil, rx_time: now, rx_iso: Time.at(now).utc.iso8601, sender_present: true, message: {}, + }.merge(fields) + end + + it "drops a copy of another protocol than the row's" do + db = open_db + stored = seed_row(db, "from_id" => "!aaaa0001", "text" => "meshcore row", "protocol" => "meshcore") + + expect(dp.merge_message_copy(db, row_id, stored, later_copy(text: "other row"))).to eq([false, nil]) + expect(db.get_first_row("SELECT text, protocol FROM messages WHERE id = ?", [row_id])).to eq("text" => "meshcore row", "protocol" => "meshcore") + ensure + db&.close + end + + it "fills the ciphertext of a row that holds no text" do + db = open_db + stored = seed_row(db, "from_id" => "!aaaa0001") + + expect(dp.merge_message_copy(db, row_id, stored, later_copy(encrypted: "AAECAwQ="))).to eq([true, "!aaaa0001"]) + expect(db.get_first_value("SELECT encrypted FROM messages WHERE id = ?", [row_id])).to eq("AAECAwQ=") + ensure + db&.close + end + + it "takes the route and receive time of a decrypted copy of an encrypted row" do + db = open_db + stored = seed_row(db, "from_id" => "!aaaa0001", "encrypted" => "AAECAwQ=", "hops" => 3, "path" => "f0bf44") + + dp.merge_message_copy(db, row_id, stored, later_copy(text: "decrypted", hops: 1, path: "a1b2", message: { "snr" => 4.5 })) + + expect(db.get_first_row("SELECT text, encrypted, hops, path, snr, rx_time FROM messages WHERE id = ?", [row_id])).to eq( + "text" => "decrypted", "encrypted" => nil, "hops" => 1, "path" => "a1b2", "snr" => 4.5, "rx_time" => now, + ) + ensure + db&.close + end + + it "moves a meshtastic row to the protocol of a MeshCore copy (#747)" do + db = open_db + stored = seed_row(db, "from_id" => "!aaaa0001", "text" => "default row") + + expect(dp.merge_message_copy(db, row_id, stored, later_copy(protocol: "meshcore"))).to eq([true, "!aaaa0001"]) + expect(db.get_first_value("SELECT protocol FROM messages WHERE id = ?", [row_id])).to eq("meshcore") + ensure + db&.close + end + end + + describe "positions, telemetry and traces from another node (KC4)" do + it "leaves a stored position alone and logs the id and both nodes" do + warnings = capture_warnings + post_ok("/api/positions", { "id" => 4_444_444_444, "node_id" => "!aaaa0001", "rx_time" => now - 3600, "latitude" => 52.52, "longitude" => 13.4 }) + post_ok("/api/positions", { "id" => 4_444_444_444, "node_id" => "!bbbb0002", "rx_time" => now, "sats_in_view" => 3 }) + + expect(stored("SELECT node_id, rx_time, latitude, longitude, sats_in_view FROM positions WHERE id = 4444444444")).to eq( + "node_id" => "!aaaa0001", "rx_time" => now - 3600, "latitude" => 52.52, "longitude" => 13.4, "sats_in_view" => nil, + ) + expect(warnings.select { |text, _| text == record_drop }).to eq([[ + record_drop, + { context: "data_processing.insert_position", table: "positions", id: 4_444_444_444, stored_node_id: "!aaaa0001", node_id: "!bbbb0002" }, + ]]) + get "/api/positions/!aaaa0001?since=1" + expect(JSON.parse(last_response.body).map { |row| row["id"] }).to eq([4_444_444_444]) + end + + it "leaves stored telemetry alone and logs the id and both nodes" do + warnings = capture_warnings + post_ok("/api/telemetry", { "id" => 5_555_555_555, "node_id" => "!aaaa0001", "rx_time" => now - 3600, "temperature" => 21.5 }) + post_ok("/api/telemetry", { "id" => 5_555_555_555, "node_id" => "!bbbb0002", "rx_time" => now, "battery_level" => 88 }) + + expect(stored("SELECT node_id, from_id, rx_time, temperature, battery_level FROM telemetry WHERE id = 5555555555")).to eq( + "node_id" => "!aaaa0001", "from_id" => "!aaaa0001", "rx_time" => now - 3600, "temperature" => 21.5, "battery_level" => nil, + ) + expect(warnings.select { |text, _| text == record_drop }).to eq([[ + record_drop, + { context: "data_processing.insert_telemetry", table: "telemetry", id: 5_555_555_555, stored_node_id: "!aaaa0001", node_id: "!bbbb0002" }, + ]]) + end + + it "leaves a stored trace and its hops alone and logs the id and both sources" do + warnings = capture_warnings + post_ok("/api/traces", { "id" => 6_666_666_666, "src" => 0xaaaa0001, "dest" => 0xcccc0003, "hops" => [0xdddd0004], "rx_time" => now - 3600 }) + post_ok("/api/traces", { "id" => 6_666_666_666, "src" => 0xbbbb0002, "hops" => [0xeeee0005, 0xffff0006], "rx_time" => now }) + + expect(stored("SELECT src, dest, rx_time FROM traces WHERE id = 6666666666")).to eq( + "src" => 0xaaaa0001, "dest" => 0xcccc0003, "rx_time" => now - 3600, + ) + expect(stored_rows("SELECT node_id FROM trace_hops WHERE trace_id = 6666666666 ORDER BY hop_index").map { |row| row["node_id"] }).to eq([0xdddd0004]) + expect(warnings.select { |text, _| text == record_drop }).to eq([[ + record_drop, + { context: "data_processing.insert_trace", table: "traces", id: 6_666_666_666, stored_src: "!aaaa0001", src: "!bbbb0002" }, + ]]) + end + + it "names a node number in the warning by the id it names, a negative one as sent" do + dp = DataProcessingHarness.build.new + + expect([0xbbbb0002, -1, "!aaaa0001", nil].map { |ref| dp.collision_node_label(ref) }).to eq(["!bbbb0002", -1, "!aaaa0001", nil]) + end + + it "still merges copies from the same node (guard)" do + warnings = capture_warnings + post_ok("/api/positions", { "id" => 4_444_444_445, "node_id" => "!aaaa0001", "rx_time" => now - 5, "latitude" => 52.52, "longitude" => 13.4, "ingestor" => "!0000beef" }) + post_ok("/api/positions", { "id" => 4_444_444_445, "node_id" => "!aaaa0001", "rx_time" => now, "latitude" => 52.52, "longitude" => 13.4, "snr" => 4.5, "ingestor" => "!0000cafe" }) + post_ok("/api/telemetry", { "id" => 5_555_555_556, "node_id" => "!aaaa0001", "rx_time" => now - 5, "temperature" => 21.5 }) + post_ok("/api/telemetry", { "id" => 5_555_555_556, "node_id" => "!aaaa0001", "rx_time" => now, "relative_humidity" => 40.0 }) + post_ok("/api/traces", { "id" => 6_666_666_667, "src" => 0xaaaa0001, "hops" => [0xdddd0004], "rx_time" => now - 5 }) + post_ok("/api/traces", { "id" => 6_666_666_667, "src" => 0xaaaa0001, "hops" => [0xeeee0005], "rx_time" => now, "snr" => 3.25 }) + + expect(stored("SELECT COUNT(*) AS n, MAX(snr) AS snr, MAX(rx_time) AS rx_time, MAX(ingestor) AS ingestor FROM positions WHERE id = 4444444445")).to eq( + "n" => 1, "snr" => 4.5, "rx_time" => now, "ingestor" => "!0000beef", + ) + expect(stored("SELECT temperature, relative_humidity, rx_time FROM telemetry WHERE id = 5555555556")).to eq( + "temperature" => 21.5, "relative_humidity" => 40.0, "rx_time" => now, + ) + expect(stored("SELECT snr, rx_time FROM traces WHERE id = 6666666667")).to eq("snr" => 3.25, "rx_time" => now) + expect(stored_rows("SELECT node_id FROM trace_hops WHERE trace_id = 6666666667").map { |row| row["node_id"] }).to eq([0xeeee0005]) + expect(warnings.map(&:first)).not_to include(record_drop) + end + + it "fills the node of a stored record that names none" do + post_ok("/api/positions", { "id" => 4_444_444_446, "rx_time" => now - 60, "latitude" => 1.5, "longitude" => 2.5 }) + post_ok("/api/positions", { "id" => 4_444_444_446, "node_id" => "!aaaa0001", "rx_time" => now }) + post_ok("/api/telemetry", { "id" => 5_555_555_557, "rx_time" => now - 60, "temperature" => 20.0 }) + post_ok("/api/telemetry", { "id" => 5_555_555_557, "node_id" => "!aaaa0001", "rx_time" => now, "battery_level" => 50 }) + post_ok("/api/traces", { "id" => 6_666_666_668, "dest" => 0xcccc0003, "rx_time" => now - 60 }) + post_ok("/api/traces", { "id" => 6_666_666_668, "src" => 0xaaaa0001, "rx_time" => now }) + + expect(stored("SELECT node_id, latitude FROM positions WHERE id = 4444444446")).to eq("node_id" => "!aaaa0001", "latitude" => 1.5) + expect(stored("SELECT node_id, temperature, battery_level FROM telemetry WHERE id = 5555555557")).to eq( + "node_id" => "!aaaa0001", "temperature" => 20.0, "battery_level" => 50.0, + ) + expect(stored("SELECT src, dest FROM traces WHERE id = 6666666668")).to eq("src" => 0xaaaa0001, "dest" => 0xcccc0003) + end + end + + describe "collision warnings (KC5)" do + include_context "with isolated db" + + let(:dp) { DataProcessingHarness.build.new } + # A limiter on a clock the example sets, in whole seconds. + let(:limiter) { PotatoMesh::App::DataProcessing::CollisionWarningLimiter.new(clock: -> { @clock }) } + # The warning that reports a window's suppressed count. + let(:suppressed) { "Suppressed collision warnings" } + + before { @clock = 1_000 } + + it "logs ten warnings of a context in 60 seconds and reports the rest once the window ends" do + expect(Array.new(12) { limiter.admit("ctx") }).to eq([[true, 0]] * 10 + [[false, 0]] * 2) + @clock += 59 + expect(limiter.admit("ctx")).to eq([false, 0]) + @clock += 1 + expect(limiter.admit("ctx")).to eq([true, 3]) + expect(limiter.admit("ctx")).to eq([true, 0]) + end + + it "limits each context on its own and reports nothing for a window that suppressed nothing" do + 10.times { limiter.admit("first") } + + expect([limiter.admit("first"), limiter.admit("second")]).to eq([[false, 0], [true, 0]]) + @clock += 60 + expect(limiter.admit("second")).to eq([true, 0]) + end + + it "forgets every window when reset" do + 11.times { limiter.admit("ctx") } + limiter.reset! + + expect(limiter.admit("ctx")).to eq([true, 0]) + end + + it "is one limiter of 10 lines per 60 seconds for the whole process" do + shared = PotatoMesh::App::DataProcessing.collision_warning_limiter + + expect([shared.limit, shared.window_seconds]).to eq([10, 60]) + expect([dp.collision_warning_limiter, DataProcessingHarness.build.new.collision_warning_limiter]).to all(be(shared)) + end + + it "passes the warnings of dropped message copies and records through it" do + db = open_db + lines = [] + allow(dp).to receive(:collision_warning_limiter).and_return(limiter) + allow(dp).to receive(:warn_log) { |text, **fields| lines << [text, fields[:context], fields[:suppressed]] } + dp.insert_message(db, message("id" => 1_111_111_151, "from_id" => "!aaaa0001", "text" => "original")) + dp.insert_position(db, { "id" => 4_444_444_451, "node_id" => "!aaaa0001", "rx_time" => now, "latitude" => 52.52, "longitude" => 13.4 }) + 12.times do |n| + dp.insert_message(db, message("id" => 1_111_111_151, "from_id" => format("!bbbb%04x", n), "text" => "forged")) + dp.insert_position(db, { "id" => 4_444_444_451, "node_id" => format("!bbbb%04x", n), "rx_time" => now }) + end + @clock += 60 + dp.insert_message(db, message("id" => 1_111_111_151, "from_id" => "!cccc0003", "text" => "forged")) + + expect(lines.count { |text, context, _| text == message_drop && context == "data_processing.insert_message" }).to eq(11) + expect(lines.count { |text, context, _| text == record_drop && context == "data_processing.insert_position" }).to eq(10) + expect(lines.last(2)).to eq([ + [suppressed, "data_processing.insert_message", 2], + [message_drop, "data_processing.insert_message", nil], + ]) + expect(dp).to have_received(:warn_log).with(suppressed, context: "data_processing.insert_message", suppressed: 2, window_seconds: 60) + ensure + db&.close + end + end +end diff --git a/web/spec/message_scope_spec.rb b/web/spec/message_scope_spec.rb index 8cc1d36b..da049da8 100644 --- a/web/spec/message_scope_spec.rb +++ b/web/spec/message_scope_spec.rb @@ -204,21 +204,10 @@ RSpec.describe "MeshCore message scope" do db&.close end - # Force the INSERT race: the existing-row lookup misses, so the INSERT - # trips the primary key and the constraint-recovery path runs. - # - # @param db [SQLite3::Database] open database handle. - # @return [void] - def hide_existing_row(db) - allow(db).to receive(:get_first_row).and_wrap_original do |original, sql, *args| - sql.include?("SELECT from_id, to_id, text, encrypted, lora_freq") ? nil : original.call(sql, *args) - end - end - it "fills a NULL scope on the INSERT-race path" do db = open_db dp.insert_message(db, scope_message("ingestor" => "!11111111")) - hide_existing_row(db) + hide_stored_message_once(db) dp.insert_message(db, scope_message("ingestor" => "!22222222", "scope" => "de-be")) expect(db.get_first_value("SELECT scope FROM messages WHERE id = 765001")).to eq("de-be") ensure @@ -247,7 +236,7 @@ RSpec.describe "MeshCore message scope" do it "never overwrites a stored name on the INSERT-race path" do db = open_db dp.insert_message(db, scope_message("ingestor" => "!11111111", "scope" => "de-be")) - hide_existing_row(db) + hide_stored_message_once(db) dp.insert_message(db, scope_message("ingestor" => "!22222222", "scope" => "eu")) expect(db.get_first_value("SELECT scope FROM messages WHERE id = 765001")).to eq("de-be") ensure @@ -285,7 +274,7 @@ RSpec.describe "MeshCore message scope" do db = open_db dp.insert_message(db, scope_message("id" => 765_301, "text" => "Alice: race unknown", "scope" => "?")) dp.insert_message(db, scope_message("id" => 765_302, "text" => "Alice: race plain", "scope" => "*")) - hide_existing_row(db) + hide_stored_message_once(db) dp.insert_message(db, scope_message("id" => 765_301, "text" => "Alice: race unknown", "scope" => "de-be")) dp.insert_message(db, scope_message("id" => 765_302, "text" => "Alice: race plain", "scope" => "de-be")) expect(db.get_first_value("SELECT scope FROM messages WHERE id = 765301")).to eq("de-be") diff --git a/web/spec/spec_helper.rb b/web/spec/spec_helper.rb index 73bd99b9..f8893e72 100644 --- a/web/spec/spec_helper.rb +++ b/web/spec/spec_helper.rb @@ -62,10 +62,12 @@ RSpec.configure do |config| config.include Rack::Test::Methods # Federation keeps process-wide peer state: the fetch cooldown, backoff and - # revalidation store per host, and the registrations in flight. Every - # example starts without it. + # revalidation store per host, and the registrations in flight. Ingest + # keeps the windows of the collision-warning limiter (SPEC KC5). Every + # example starts without them. config.before do PotatoMesh::Application.clear_federation_crawl_state! + PotatoMesh::App::DataProcessing.collision_warning_limiter.reset! end config.after(:suite) do diff --git a/web/spec/support/data_processing_harness.rb b/web/spec/support/data_processing_harness.rb index 59a5367f..7c8a2c20 100644 --- a/web/spec/support/data_processing_harness.rb +++ b/web/spec/support/data_processing_harness.rb @@ -117,6 +117,26 @@ RSpec.shared_context "with isolated db" do db end + # Force the INSERT race of +insert_message+ on +db+: the first stored-row + # lookup of each message id misses, as when another ingestor's copy lands + # between this copy's lookup and its INSERT. The INSERT then trips the + # primary key, and the race recovery's own lookup sees the row (SPEC KC3). + # + # @param db [SQLite3::Database] open database handle. + # @return [void] + def hide_stored_message_once(db) + lookup = "SELECT #{PotatoMesh::App::DataProcessing::MESSAGE_MERGE_COLUMNS.join(", ")} FROM messages" + looked_up = {} + allow(db).to receive(:get_first_row).and_wrap_original do |original, sql, *args| + if sql.start_with?(lookup) && !looked_up.key?(args.first) + looked_up[args.first] = true + nil + else + original.call(sql, *args) + end + end + end + # Return the full node row for the canonical test node ID. # # @param db [SQLite3::Database] open database handle.