From 7b0176df93b2e00bb92504fe57b73bc88ee0d0fe Mon Sep 17 00:00:00 2001 From: Afri <220195275+l5yth@users.noreply.github.com> Date: Sun, 4 Oct 2026 21:25:57 +0200 Subject: [PATCH] web: dedup MeshCore channel copies whose sender resolved differently (#880) (#914) --- CHANGELOG.md | 2 + data/mesh_ingestor/CONTRACTS.md | 2 +- data/messages.sql | 4 + .../application/data_processing/coercions.rb | 4 +- .../application/data_processing/messages.rb | 82 ++- web/lib/potato_mesh/application/database.rb | 82 ++- web/spec/data_processing_spec.rb | 99 +--- web/spec/database_spec.rb | 129 ++++- web/spec/meshcore_dedup_spec.rb | 468 ++++++++++++++++++ web/spec/meshcore_ghost_nodes_spec.rb | 58 +-- web/spec/support/data_processing_harness.rb | 149 ++++++ 11 files changed, 881 insertions(+), 198 deletions(-) create mode 100644 web/spec/meshcore_dedup_spec.rb create mode 100644 web/spec/support/data_processing_harness.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c9c36fa..5aaab055 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -52,6 +52,8 @@ * Data: new `DROP_VIA_MQTT=1` drops Meshtastic packets and nodes relayed via MQTT, on the API and UDP transports (#884, SPEC VM1-VM2) * Web: the web service no longer receives the unused `ALLOWED_CHANNELS`/`HIDDEN_CHANNELS` settings (#784, SPEC CF6) * Docs: `CHANNEL_INDEX` documented as the announcement channel, not an ingest filter (#884, SPEC CF7) +* Web: a MeshCore channel message heard by two ingestors that resolved its sender to different node ids is stored once (#880) +* Web: the first boot after upgrading deletes duplicate MeshCore messages once; on instances with several MeshCore ingestors this can remove about a fifth of the stored MeshCore messages (#880) ### Features * Data/Web: Reticulum protocol support - `PROTOCOL=reticulum` ingests announces as `protocol="reticulum"` nodes end-to-end (ingest, stats, federation wire, UI) (SPEC S6/FS2/MA5/MA-F2/F2-2 as amended) diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index caed7cef..0a4f5775 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -192,7 +192,7 @@ Known limitations of the v1 fingerprint: - *Format-string ambiguity around `:`.* Components are joined with literal colons and not length-prefixed, so a colon embedded in `sender_identity` or `text` can shift the field boundary and collide two distinct triples. Rare in practice. A `v2` revision should use a delimiter that cannot appear in any component (e.g. `\x00`), or length-prefix each field. - *meshcore_py text-decoding inconsistency.* The upstream reader strips trailing `\0` bytes on the real-time path but not on sync-replay, so one physical message heard both ways can produce two fingerprints (a duplicate row). Out of scope for the ingestor; track upstream. - *Sender-side clock reset.* MeshCore nodes without an RTC start `sender_timestamp` from `0` after reboot, so two same-text messages from one sender within a second of power-on collapse into one row. Accepted trade-off (the alternative is no dedup at all). -- *Relay-rewritten `sender_timestamp` & cross-ingestor clock skew.* A rewritten `sender_timestamp`, or clock drift between two ingestors hearing the same packet, can bypass the `messages.id` PK collapse. The web app additionally dedups on insert: for `protocol = "meshcore"` with non-empty `text` and a known `from_id`, a row matching `(from_id, to_id, channel_name, text)` within ±`MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS` (default 300 s) of `rx_time` is dropped - keyed on the sender-stable `channel_name`, not the per-receiver `channel` slot index. Accepted trade-off: identical text repeated on the same channel within the window silently collapses to one row. Ingestors MUST still produce deterministic v1 ids - this content-level layer is additive, not a replacement. A one-shot, `PRAGMA user_version`-gated backfill (re-run via `MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION`) clears pre-existing duplicates on startup, transitively (a chain of identical-content rows collapses even if the chain spans longer than the window); new rows are governed only by the per-insert guard. +- *Relay-rewritten `sender_timestamp`, differing local channel slots & cross-ingestor clock skew.* A rewritten `sender_timestamp`, or two ingestors holding one channel at different local slots (the `c` discriminator is the receiver's slot index), gives one physical transmission two ids, so the `messages.id` PK collapse cannot merge the copies; clock drift between the two ingestors then separates their `rx_time`. The web app additionally dedups on insert: for `protocol = "meshcore"` with non-empty `text` and a known `from_id`, an earlier row that also has a `from_id`, within ±`MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS` (default 300 s) of `rx_time`, absorbs the copy when it matches `(to_id, channel_name, text)` for a channel broadcast (`to_id = "^all"`) whose text carries a `SenderName:` prefix, and `(from_id, to_id, channel_name, text)` otherwise. Broadcasts skip `from_id` because each ingestor resolves the sender against its own roster (its pubkey id, or the name-derived synthetic id), while the prefix in `text` already names the sender. The key uses the sender-stable `channel_name`, not the per-receiver `channel` slot index. The copy is applied to the earliest such row as if it carried that row's `id` (sender ranking per SPEC MR3, columns the stored row lacks), not stored as a second row. Accepted trade-off: identical text repeated on the same channel within the window silently collapses to one row, including identical text from two devices that share a display name. Ingestors MUST still produce deterministic v1 ids - this content-level layer is additive, not a replacement. A one-shot, `PRAGMA user_version`-gated backfill (re-run by bumping `MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION`, currently 3) clears pre-existing duplicates on startup with the same key, transitively (a chain of identical-content rows collapses even if the chain spans longer than the window). New rows are governed only by the per-insert guard, which does not catch copies further apart than the window, a copy without `channel_name` next to a named one, one channel stored under two different `channel_name` labels, or the concurrent-insert race below. - *Concurrent-insert race.* The content-dedup check and the insert are not wrapped in a shared transaction, so two concurrent threads carrying the same content with different ids can both pass the check and both insert. Narrow (single-node, multi-threaded ingest) and not cleaned up retroactively, since the backfill above is one-shot. A future fix would wrap the meshcore pre-check + id-PK path in `db.transaction(:immediate)`. - *Upstream `meshcore` reader crash on truncated advertisements.* `meshcore-py` can raise from `MessageReader.handle_rx` when an advertisement frame omits its trailing `path_hash_mode` byte, silently dropping the event (`Task exception was never retrieved`). The ingestor patches around it (`data/mesh_ingestor/protocols/_meshcore_patches.py`): logs the offending frame under `context=meshcore.reader.patch` and lets the task exit cleanly, with a loop-level `context=asyncio.unhandled` handler as backstop. Additive; removable once upstream ships a defensive length check. diff --git a/data/messages.sql b/data/messages.sql index 7e1b62fa..19592647 100644 --- a/data/messages.sql +++ b/data/messages.sql @@ -42,3 +42,7 @@ CREATE INDEX IF NOT EXISTS idx_messages_to_id ON messages(to_id); CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel); CREATE INDEX IF NOT EXISTS idx_messages_portnum ON messages(portnum); CREATE INDEX IF NOT EXISTS idx_messages_reply_id ON messages(reply_id); +-- MeshCore content dedup (#880): the insert-time duplicate lookups and the +-- one-shot purge search this index on text equality plus an rx_time window. +-- ensure_schema_upgrades creates the same index on upgraded databases. +CREATE INDEX IF NOT EXISTS idx_messages_meshcore_text ON messages(text, rx_time) WHERE protocol = 'meshcore'; diff --git a/web/lib/potato_mesh/application/data_processing/coercions.rb b/web/lib/potato_mesh/application/data_processing/coercions.rb index 9057134d..1b81efb9 100644 --- a/web/lib/potato_mesh/application/data_processing/coercions.rb +++ b/web/lib/potato_mesh/application/data_processing/coercions.rb @@ -38,7 +38,9 @@ module PotatoMesh # pairs and 28% of all meshcore rows were duplicates. 300 s covers # ~99.5% of the observed skew. **Accepted tradeoff:** a sender repeating # the *identical* text to the same channel within 300 s collapses to one - # row — chosen over the 28% duplicate rate. (The one-shot purge in + # row, and since channel broadcasts match without ``from_id`` (#880) so + # do two devices sharing a display name that send identical text there + # — chosen over the 28% duplicate rate. (The one-shot purge in # +PotatoMesh::App::Database+ applies this transitively, so a chain of such # repeats spanning longer than 300 s also collapses — a deliberately # aggressive one-time cleanup; see that file's note.) See issues diff --git a/web/lib/potato_mesh/application/data_processing/messages.rb b/web/lib/potato_mesh/application/data_processing/messages.rb index a1828757..bec58fd8 100644 --- a/web/lib/potato_mesh/application/data_processing/messages.rb +++ b/web/lib/potato_mesh/application/data_processing/messages.rb @@ -220,6 +220,12 @@ module PotatoMesh resolved_from_id = from_id with_busy_retry do + # Each attempt starts from this copy's own id and sender, so a retry + # after SQLite3::BusyException re-derives the survivor instead of + # inheriting the previous attempt's choice. + target_id = msg_id + resolved_from_id = from_id + # Meshcore-only content-level dedup (issue #756). The deterministic # message id (``_derive_message_id`` in the Python ingestor) hashes # ``sender_timestamp`` among other fields, but the MeshCore library @@ -236,9 +242,10 @@ module PotatoMesh # Known race: the SELECT and the downstream INSERT do not share a # transaction, so two Puma threads carrying the same content with # different ids can both pass the pre-check and both insert. The - # deploy-time backfill sweeps the survivors; wrapping the pair in - # ``db.transaction(:immediate)`` is a future tightening if the race - # is ever observed in production. + # survivors are not cleaned up retroactively, since the backfill in + # +Database#ensure_schema_upgrades+ is one-shot (``CONTRACTS.md``); + # wrapping the pair in ``db.transaction(:immediate)`` is a future + # tightening if the race is ever observed in production. if protocol == "meshcore" && from_id && channel_index && text && !text.to_s.empty? # Match on the sender-stable ``channel_name`` (NULL-safe ``IS ?``) # rather than the per-receiver ``channel`` slot index. Two @@ -249,38 +256,67 @@ module PotatoMesh # carried in the message text/contact roster identically across # receivers, so it is the stable discriminator. ``to_id`` is also # ``IS ?`` (rare meshcore nil fallback). - duplicate_id = db.get_first_value( - <<~SQL, - SELECT id FROM messages - WHERE protocol = 'meshcore' - AND from_id = ? - AND to_id IS ? - AND channel_name IS ? - AND text = ? - AND rx_time BETWEEN ? AND ? - AND id != ? - LIMIT 1 - SQL - [from_id, to_id, channel_name, text, - rx_time - MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS, - rx_time + MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS, msg_id], - ) + # + # A channel broadcast names its sender in the text ("Name: body"), + # so ``text =`` already pins the claimed sender. ``from_id`` is + # each ingestor's own roster resolution of that name (the pubkey id, + # or the name-derived synthetic id when the roster lacks the + # sender; SPEC MR3), so it differs between copies of one + # transmission and must not split them (#880). Direct messages + # carry no sender prefix and keep the ``from_id`` leg. + # + # Both forms constrain ``text = ?`` and an ``rx_time`` window, so + # they search +idx_messages_meshcore_text+ (SPEC MX6) and come back + # in ``rx_time, id`` order without a sort, however many channel + # rows the table holds. + rx_window = [rx_time - MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS, rx_time + MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS] + duplicate_id = if to_id == "^all" && parse_meshcore_sender_name(text) + db.get_first_value(<<~SQL, [channel_name, text, *rx_window, msg_id]) + SELECT id FROM messages + WHERE protocol = 'meshcore' + AND to_id = '^all' + AND from_id IS NOT NULL + AND channel_name IS ? + AND text = ? + AND rx_time BETWEEN ? AND ? + AND id != ? + ORDER BY rx_time, id + LIMIT 1 + SQL + else + db.get_first_value(<<~SQL, [from_id, to_id, channel_name, text, *rx_window, msg_id]) + SELECT id FROM messages + WHERE protocol = 'meshcore' + AND from_id = ? + AND to_id IS ? + AND channel_name IS ? + AND text = ? + AND rx_time BETWEEN ? AND ? + AND id != ? + ORDER BY rx_time, id + LIMIT 1 + SQL + end if duplicate_id debug_log( - "Skipped meshcore message duplicate", + "Collapsed meshcore message duplicate onto the stored copy", context: "data_processing.insert_message", new_id: msg_id, existing_id: duplicate_id, from_id: from_id, channel: channel_index, ) - return + # Apply this copy to the earliest matching row exactly as an id + # hit: the update path below ranks the two senders (SPEC MR3), + # fills columns the stored copy lacks, and credits the reception + # to the resolved winner, as if both ingestors had derived one id. + target_id = duplicate_id 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 FROM messages WHERE id = ?", - [msg_id], + [target_id], ) if existing updates = {} @@ -407,7 +443,7 @@ module PotatoMesh unless updates.empty? assignments = updates.keys.map { |column| "#{column} = ?" }.join(", ") - db.execute("UPDATE messages SET #{assignments} WHERE id = ?", updates.values + [msg_id]) + db.execute("UPDATE messages SET #{assignments} WHERE id = ?", updates.values + [target_id]) end else PotatoMesh::App::Prometheus::MESSAGES_TOTAL.increment diff --git a/web/lib/potato_mesh/application/database.rb b/web/lib/potato_mesh/application/database.rb index 606abd45..10bb8730 100644 --- a/web/lib/potato_mesh/application/database.rb +++ b/web/lib/potato_mesh/application/database.rb @@ -26,7 +26,12 @@ module PotatoMesh # +MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS+ (30→300 s), so it must re-run # once to clear the cross-slot, clock-skewed duplicates that accumulated # under the old window/key. - MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION = 2 + # Bumped 2→3 (#880): a MeshCore channel broadcast with a "Name:" prefix + # no longer keys on +from_id+, which each ingestor resolves against its + # own roster (pubkey id or name-derived synthetic id), so the purge + # re-runs once to clear the cross-slot copies that the old guard let + # through when the two ingestors resolved the sender differently. + MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION = 3 # Column definitions required for environment telemetry support. Each # entry pairs the column name with the SQL type used when backfilling @@ -358,9 +363,13 @@ module PotatoMesh db.execute("CREATE INDEX IF NOT EXISTS idx_messages_reply_id ON messages(reply_id)") end - # #756 — partial index backing the meshcore content-dedup lookup in - # insert_message. Scoped to meshcore so the index stays small even - # on meshtastic-heavy deployments. ``CREATE … IF NOT EXISTS`` is + # #756 — meshcore-scoped partial indexes, small even on + # meshtastic-heavy deployments. +idx_messages_meshcore_content+ + # (from_id, channel, rx_time) is the original #756 index; since #880 + # the content-dedup lookups in insert_message and both purge + # statements below search +idx_messages_meshcore_text+ (text, + # rx_time) instead (SPEC MX6), which +data/messages.sql+ also + # creates for a fresh database. ``CREATE … IF NOT EXISTS`` is # cheap enough to run on every boot; the one-shot backfill below # is gated separately via ``PRAGMA user_version`` so it does not # repeat after the first successful pass. @@ -374,15 +383,25 @@ module PotatoMesh ON messages(from_id, channel, rx_time) WHERE protocol = 'meshcore' SQL + db.execute(<<~SQL) + CREATE INDEX IF NOT EXISTS idx_messages_meshcore_text + ON messages(text, rx_time) + WHERE protocol = 'meshcore' + SQL # #756 backfill — collapse pre-existing meshcore duplicate groups. # Delete any row that has an EARLIER same-content row # (from_id, to_id, channel_name, text) within # #{PotatoMesh::App::DataProcessing::MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS} s, - # keeping the earliest (min rx_time, min id) copy. The key + window - # match the runtime guard (+insert_message+): the per-receiver - # +channel+ index differs across ingestors for one logical channel, - # so the stable +channel_name+ is what clusters cross-ingestor copies. + # keeping the earliest (min rx_time, min id) copy; a channel + # broadcast whose text carries a "Name:" prefix drops the +from_id+ + # leg (#880), so the sweep runs as two statements, one per key. + # The keys + window match the runtime guard (+insert_message+): the + # per-receiver +channel+ index differs across ingestors for one + # logical channel, so the stable +channel_name+ is what clusters + # cross-ingestor copies. Each lookup of an earlier copy searches + # +idx_messages_meshcore_text+ on ``text = … AND rx_time BETWEEN``, + # so the sweep stays near-linear in the number of meshcore rows. # # NOTE — this EXISTS predicate is TRANSITIVE and so does NOT behave # identically to the runtime guard: a chain of identical-content rows @@ -397,9 +416,13 @@ module PotatoMesh # governs every new row. See #756 and ``CONTRACTS.md``. # # Gated via ``PRAGMA user_version`` so this expensive self-join - # runs exactly once after deploy. Post-fix the runtime guard - # prevents new duplicates from accumulating, so re-running on - # every boot would scan ``messages`` for no reason. + # runs once per backfill version, not on every boot. The runtime + # guard alone governs rows inserted later and does not stop every + # duplicate: until #880 it let through cross-slot copies whose + # sender two ingestors resolved differently, and copies further + # apart than the window or caught in the check-then-insert race + # still pass (``CONTRACTS.md``). A guard change that should also + # clear history bumps +MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION+. current_version = db.get_first_value("PRAGMA user_version").to_i if current_version < MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION window = PotatoMesh::App::DataProcessing::MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS @@ -410,20 +433,51 @@ module PotatoMesh # purely for consistency. ``PRAGMA user_version`` cannot # accept bind params, so it keeps literal interpolation of # an internal constant. + # + # The broadcast test mirrors +parse_meshcore_sender_name+: a + # colon with a non-blank name before it. The +trim+ set (tab, + # LF, VT, FF, CR, space) is what +String#strip+ removes, less + # NUL, which SQLite would read as the end of the set. The + # second statement takes every other row (``to_id IS`` keeps a + # NULL recipient in it) and keeps the +from_id+ leg. + sender_prefix = <<~SQL.strip + instr(messages.text, ':') > 1 + AND trim(substr(messages.text, 1, instr(messages.text, ':') - 1), + char(9, 10, 11, 12, 13, 32)) != '' + SQL + db.execute(<<~SQL, [window]) + DELETE FROM messages + WHERE protocol = 'meshcore' + AND to_id = '^all' + AND text IS NOT NULL AND text != '' + AND from_id IS NOT NULL + AND #{sender_prefix} + AND EXISTS ( + SELECT 1 FROM messages AS earlier + WHERE earlier.protocol = 'meshcore' + AND earlier.text = messages.text + AND earlier.rx_time BETWEEN messages.rx_time - ? AND messages.rx_time + AND earlier.to_id = '^all' + AND earlier.from_id IS NOT NULL + AND earlier.channel_name IS messages.channel_name + AND (earlier.rx_time < messages.rx_time + OR earlier.id < messages.id) + ) + SQL db.execute(<<~SQL, [window]) DELETE FROM messages WHERE protocol = 'meshcore' AND text IS NOT NULL AND text != '' AND from_id IS NOT NULL + AND NOT (to_id IS '^all' AND #{sender_prefix}) AND EXISTS ( SELECT 1 FROM messages AS earlier WHERE earlier.protocol = 'meshcore' + AND earlier.text = messages.text + AND earlier.rx_time BETWEEN messages.rx_time - ? AND messages.rx_time AND earlier.from_id = messages.from_id AND earlier.to_id IS messages.to_id AND earlier.channel_name IS messages.channel_name - AND earlier.text = messages.text - AND messages.rx_time - earlier.rx_time >= 0 - AND messages.rx_time - earlier.rx_time <= ? AND (earlier.rx_time < messages.rx_time OR earlier.id < messages.id) ) diff --git a/web/spec/data_processing_spec.rb b/web/spec/data_processing_spec.rb index ae125294..84d0487d 100644 --- a/web/spec/data_processing_spec.rb +++ b/web/spec/data_processing_spec.rb @@ -16,42 +16,11 @@ require "spec_helper" require "sqlite3" +require_relative "support/data_processing_harness" RSpec.describe PotatoMesh::App::DataProcessing do # Build a minimal host class so we can call the module methods in isolation. - let(:harness_class) do - Class.new do - include PotatoMesh::App::DataProcessing - include PotatoMesh::App::Helpers - - def debug_log(message, **); end - - def warn_log(message, **); end - - def with_busy_retry - yield - end - - def update_prometheus_metrics(*); end - - def prom_report_ids - [] - end - - def private_mode? - false - end - - def normalize_node_id(_db, node_ref) - parts = canonical_node_parts(node_ref) - parts ? parts[0] : nil - end - - def resolve_protocol(_db, _ingestor, cache: nil) - "meshtastic" - end - end - end + let(:harness_class) { DataProcessingHarness.build } subject(:dp) { harness_class.new } @@ -218,54 +187,6 @@ RSpec.describe PotatoMesh::App::DataProcessing do end end - shared_context "with isolated db" do - around do |example| - Dir.mktmpdir("dp-spec-") do |dir| - db_path = File.join(dir, "mesh.db") - RSpec::Mocks.with_temporary_scope do - allow(PotatoMesh::Config).to receive(:db_path).and_return(db_path) - allow(PotatoMesh::Config).to receive(:db_busy_timeout_ms).and_return(5000) - allow(PotatoMesh::Config).to receive(:week_seconds).and_return(604_800) - allow(PotatoMesh::Config).to receive(:four_weeks_seconds).and_return(604_800) - allow(PotatoMesh::Config).to receive(:debug?).and_return(false) - db_helper = Object.new.extend(PotatoMesh::App::Database) - db_helper.init_db - db_helper.ensure_schema_upgrades - example.run - end - end - end - - def open_db - db = SQLite3::Database.new(PotatoMesh::Config.db_path) - db.results_as_hash = true - db - end - - # Return the full node row for the canonical test node ID. - def read_node(db) - db.execute("SELECT * FROM nodes WHERE node_id = '!aabbccdd'").first - end - - # Insert the canonical test node with full user info and CLIENT_BASE role. - def seed_node(db) - dp.upsert_node(db, "!aabbccdd", { - "lastHeard" => now - 100, - "num" => 0xaabbccdd, - "user" => { - "role" => "CLIENT_BASE", - "longName" => "Real Long Name", - "shortName" => "RLN", - "macaddr" => "aa:bb:cc:dd:ee:ff", - "hwModel" => "TBEAM", - "publicKey" => "abc123", - }, - }) - end - - let(:now) { Time.now.to_i } - end - # --------------------------------------------------------------------------- # insert_telemetry (telemetry_type validation path) # --------------------------------------------------------------------------- @@ -2260,19 +2181,10 @@ RSpec.describe PotatoMesh::App::DataProcessing do # --------------------------------------------------------------------------- describe "#insert_message — meshcore duplicate-copy sender resolution" do include_context "with isolated db" + include MeshcoreNodeSeeds let(:now) { Time.now.to_i } - # Seed a keyed (synthetic=0) meshcore node whose evidence age is - # controlled by +heard+. - def seed_keyed_node(db, node_id, name, key_byte, heard) - dp.upsert_node(db, node_id, { - "lastHeard" => heard, - "protocol" => "meshcore", - "user" => { "longName" => name, "shortName" => key_byte, "role" => "COMPANION", "publicKey" => key_byte * 32 }, - }, protocol: "meshcore") - end - def duplicate_copy(from_id, rx_time) { "id" => 7001, @@ -2614,8 +2526,9 @@ RSpec.describe PotatoMesh::App::DataProcessing do it "never issues the content-dedup SELECT for non-meshcore traffic" do # Pins the performance contract: meshtastic traffic must skip the - # partial-index lookup entirely so any future regression that makes - # the pre-check unconditional surfaces as a failing test. + # meshcore content-dedup lookup (an +idx_messages_meshcore_text+ + # search, SPEC MX6) entirely so any future regression that makes the + # pre-check unconditional surfaces as a failing test. db = open_db content_select_pattern = /SELECT\s+id\s+FROM\s+messages\s+WHERE\s+protocol\s*=\s*'meshcore'/im captured_sql = [] diff --git a/web/spec/database_spec.rb b/web/spec/database_spec.rb index 6ec34b83..7f46dffc 100644 --- a/web/spec/database_spec.rb +++ b/web/spec/database_spec.rb @@ -617,6 +617,91 @@ RSpec.describe PotatoMesh::App::Database do end end + # Seed meshcore message rows into a database that already ran the + # version-2 sweep, boot it, and return the ids the #880 purge leaves. + # + # @param rows [Array] +[id, rx_time, from_id, to_id, channel, + # channel_name, text]+ tuples. + # @param boots [Integer] number of +ensure_schema_upgrades+ passes. + # @return [Array] surviving message ids, ascending. + def purge_from_version_2(rows, boots: 1) + SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| + seed_meshcore_message_tables(db) + db.execute("PRAGMA user_version = 2") + rows.each do |id, rx_time, from_id, to_id, channel, channel_name, text| + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,channel_name,text,protocol) VALUES (?,?,?,?,?,?,?,?,?)", + [id, rx_time, Time.at(rx_time).utc.iso8601, from_id, to_id, channel, channel_name, text, "meshcore"], + ) + end + end + boots.times { harness_class.ensure_schema_upgrades } + db = SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true) + db.execute("SELECT id FROM messages ORDER BY id").flatten + ensure + db&.close + end + + it "re-runs the purge once at backfill version 3 for copies whose sender resolved differently (#880)" do + # A database that already ran the version-2 sweep. Two ingestors stored + # one #bot transmission at local slots 4 and 6; one resolved Alice to her + # pubkey id, the other to the name-derived synthetic id. The 91xx pair was + # converged afterwards by the synthetic->real merge; the 92xx pair never + # merged. Direct messages and broadcasts without a usable "Name:" prefix + # keep their from_id leg, so the 93xx-95xx pairs survive. + survivors = purge_from_version_2([ + [9_101, 1_776_760_000, "!a1a2a3a4", "^all", 4, "#bot", "Alice: dedup me"], + [9_102, 1_776_760_002, "!a1a2a3a4", "^all", 6, "#bot", "Alice: dedup me"], + [9_201, 1_776_761_000, "!a1a2a3a4", "^all", 4, "#bot", "Alice: again"], + [9_202, 1_776_761_003, "!3bc51062", "^all", 6, "#bot", "Alice: again"], + [9_301, 1_776_762_000, "!b1b1b1b1", "!0000aaaa", 0, nil, "re: meet at five"], + [9_302, 1_776_762_002, "!c2c2c2c2", "!0000aaaa", 0, nil, "re: meet at five"], + [9_401, 1_776_763_000, "!b1b1b1b1", "^all", 4, "#bot", "no sender prefix"], + [9_402, 1_776_763_002, "!c2c2c2c2", "^all", 6, "#bot", "no sender prefix"], + [9_501, 1_776_764_000, "!b1b1b1b1", "^all", 4, "#bot", " \t: blank name"], + [9_502, 1_776_764_002, "!c2c2c2c2", "^all", 6, "#bot", " \t: blank name"], + ], boots: 2) + + expect(survivors).to eq([9_101, 9_201, 9_301, 9_302, 9_401, 9_402, 9_501, 9_502]) + SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true) do |db| + expect(db.get_first_value("PRAGMA user_version").to_i).to eq(3) + end + end + + it "keeps an attributed broadcast when the only earlier copy has no sender (#880)" do + # The earlier copy cannot vouch for who sent the text, so it must not + # delete the later one that names a sender. + survivors = purge_from_version_2([ + [9_601, 1_776_765_000, nil, "^all", 4, "#bot", "Alice: unattributed first"], + [9_602, 1_776_765_002, "!a1a2a3a4", "^all", 6, "#bot", "Alice: unattributed first"], + ]) + + expect(survivors).to eq([9_601, 9_602]) + end + + it "keeps a broadcast whose only earlier match is a direct message with the same text (#880)" do + # Neither row knows its channel name, so only the recipient tells the + # direct message apart from the broadcast. + survivors = purge_from_version_2([ + [9_701, 1_776_766_000, "!b1b1b1b1", "!0000aaaa", 0, nil, "Alice: same words"], + [9_702, 1_776_766_002, "!a1a2a3a4", "^all", 4, nil, "Alice: same words"], + ]) + + expect(survivors).to eq([9_701, 9_702]) + end + + it "purges a sender-keyed duplicate that has no recipient (#880)" do + # A row with no recipient belongs to the sender-keyed sweep. Its text has + # a "Name:"-like prefix, so comparing the NULL to_id with "=" instead of + # "IS" would yield NULL and drop the row from both statements. + survivors = purge_from_version_2([ + [9_801, 1_776_767_000, "!a1a2a3a4", nil, 0, nil, "re: no recipient"], + [9_802, 1_776_767_002, "!a1a2a3a4", nil, 0, nil, "re: no recipient"], + ]) + + expect(survivors).to eq([9_801]) + end + it "purges cross-ingestor duplicates on the same boot that adds the channel_name column" do # Upgrade path from a pre-#825 schema (no channel_name column): the same # ensure_schema_upgrades pass adds the column AND the one-shot purge must @@ -726,8 +811,9 @@ RSpec.describe PotatoMesh::App::Database do # First boot seeds the backfill target; migration collapses the pair and # sets user_version. Second boot seeds NEW duplicates post-bump — the # gated migration must leave them alone so we are not paying for a - # self-join on every single startup. The runtime guard in - # insert_message is what keeps new duplicates from piling up in prod. + # self-join on every single startup. Rows inserted later are left to the + # runtime guard in insert_message, which does not stop every duplicate + # (#880), so clearing history again takes a backfill version bump. SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| seed_meshcore_message_tables(db) db.execute( @@ -764,23 +850,38 @@ RSpec.describe PotatoMesh::App::Database do end end - it "creates the partial index backing the runtime content-dedup lookup" do + # Read the CREATE statement SQLite stored for an index. + # + # @param name [String] index name. + # @return [String, nil] the statement, or nil when the index does not exist. + def index_sql(name) + db = SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true) + db.get_first_value("SELECT sql FROM sqlite_master WHERE type='index' AND name = ?", [name]) + ensure + db&.close + end + + it "creates the meshcore partial indexes when upgrading an existing schema" do SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| seed_meshcore_message_tables(db) end harness_class.ensure_schema_upgrades - SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true) do |db| - row = db.get_first_row( - "SELECT sql FROM sqlite_master WHERE type='index' AND name='idx_messages_meshcore_content'", - ) - expect(row).not_to be_nil - index_sql = row.first - expect(index_sql).to include("meshcore") - expect(index_sql).to include("from_id") - expect(index_sql).to include("channel") - expect(index_sql).to include("rx_time") - end + content_index = index_sql("idx_messages_meshcore_content") + expect(content_index).not_to be_nil + expect(content_index).to include("meshcore") + expect(content_index).to include("from_id") + expect(content_index).to include("channel") + expect(content_index).to include("rx_time") + # The index every content-dedup lookup and both purge statements search + # (#880, SPEC MX6). + expect(index_sql("idx_messages_meshcore_text")).to include("(text, rx_time)").and include("protocol = 'meshcore'") + end + + it "creates the meshcore text index with the schema of a fresh database (#880)" do + harness_class.init_db + + expect(index_sql("idx_messages_meshcore_text")).to include("(text, rx_time)").and include("protocol = 'meshcore'") end end diff --git a/web/spec/meshcore_dedup_spec.rb b/web/spec/meshcore_dedup_spec.rb new file mode 100644 index 00000000..83f9e9ad --- /dev/null +++ b/web/spec/meshcore_dedup_spec.rb @@ -0,0 +1,468 @@ +# 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 "sqlite3" +require "json" +require "digest" +require_relative "support/data_processing_harness" + +# MeshCore cross-ingestor message dedup (issue #880; SPEC MX1/MX2/MX6, +# ACCEPTANCE MX-A1/MX-A2/MX-A4). Two ingestors that hold one channel at +# different local slots derive two v1 ids for one transmission (the `c` +# fingerprint discriminator is the receiver's slot index), and each resolves +# the "Name:" sender against its own roster: the pubkey id on one, the +# name-derived synthetic id on the other. The insert guard must collapse such +# copies on the broadcast's text rather than on `from_id`, apply the late copy +# to the earliest matching row exactly as an id hit would (MR3 ranking, +# filling columns the stored copy lacks, liveness credited to the resolved +# winner), and stay an index search as the table grows. +RSpec.describe "MeshCore cross-ingestor message dedup" do + include_context "with isolated db" + + # One #bot transmission ("Alice: dedup me", sender_timestamp 1_790_000_000) + # as the ingestor derives its v1 id on local slot 4 and on local slot 6 + # (+_derive_message_id("alice", 1_790_000_000, "c4" / "c6", text)+). + let(:text) { "Alice: dedup me" } + let(:slot_ids) { { 4 => 5_007_131_281_259_980, 6 => 7_405_791_464_985_531 } } + let(:real_id) { "!a1a2a3a4" } + let(:synthetic_id) { "!#{Digest::SHA256.hexdigest("Alice")[0, 8]}" } + let(:stale_id) { "!a9a9a9a9" } + # rx_time sits in the past so later copies can be shifted forward without + # tripping the +rx_time > now+ clamp in +insert_message+. + let(:base_rx) { now - 1_000 } + # The content-dedup lookups and both purge statements must search this + # partial index on (text, rx_time) (SPEC MX6). + let(:text_index_search) { /\ASEARCH \w+ USING (COVERING )?INDEX idx_messages_meshcore_text \(text=\? AND rx_time>\? AND rx_time<\?\)\z/ } + + # Read every stored message, oldest first. + # + # @return [Array] message rows with the columns these examples assert. + def stored_messages + db = open_db + db.execute("SELECT id, from_id, channel, channel_name, lora_freq, modem_preset FROM messages ORDER BY rx_time, id") + ensure + db&.close + end + + # Read one node's +last_heard+. + # + # @param node_id [String] canonical node id. + # @return [Integer, nil] stored last-heard timestamp. + def last_heard_of(node_id) + db = open_db + db.get_first_value("SELECT last_heard FROM nodes WHERE node_id = ?", [node_id]) + ensure + db&.close + end + + # Build the POST /api/messages payload one ingestor emits for the #bot + # transmission heard on its local +slot+. + # + # @param slot [Integer] receiver-local channel slot (4 or 6). + # @param from_id [String] sender id as that ingestor resolved it. + # @param dt [Integer] receive-time offset from +base_rx+ in seconds. + # @param overrides [Hash] extra or replacing payload fields. + # @return [Hash] message payload. + def channel_copy(slot:, from_id:, dt: 0, **overrides) + rx_time = base_rx + dt + { + "id" => slot_ids.fetch(slot), + "rx_time" => rx_time, + "rx_iso" => Time.at(rx_time).utc.iso8601, + "from_id" => from_id, + "to_id" => "^all", + "channel" => slot, + "channel_name" => "#bot", + "portnum" => "TEXT_MESSAGE_APP", + "text" => text, + "protocol" => "meshcore", + "ingestor" => "!0000000#{slot}", + }.merge(overrides.transform_keys(&:to_s)) + end + + # The direct-message shape of +channel_copy+: addressed to an ingestor host + # on channel 0, no channel name, a colon in the body but no sender prefix + # semantics. + # + # @return [Hash] overrides for +channel_copy+. + def direct_message + { to_id: "!0000aaaa", channel: 0, channel_name: nil, text: "re: meet at five" } + end + + # EXPLAIN QUERY PLAN detail lines for one statement. + # + # @param sql [String] statement text. + # @param binds [Array] bind values. + # @return [Array] plan detail column, in plan order. + def plan_of(sql, binds) + db = open_db + db.execute("EXPLAIN QUERY PLAN #{sql}", binds).map { |row| row["detail"] } + ensure + db&.close + end + + describe "insert guard" do + # The synthetic->real merge machinery behind process_meshcore_chat_nodes + # has its own specs; stubbing it keeps these examples on the guard and the + # id-hit path alone. + let(:harness) { DataProcessingHarness.build(protocol: "meshcore", stub_chat_nodes: true).new } + + # Seed one "Alice" node row directly, bypassing +upsert_node+ so no + # synthetic/real merge runs. + # + # @param node_id [String] canonical node id. + # @param synthetic [Boolean] whether the row is a name-derived placeholder. + # @param heard [Integer] last-heard time; doubles as the keyed evidence + # (+last_advert_heard+) of a real row. + # @return [void] + def seed_alice(node_id, synthetic:, heard:) + db = open_db + db.execute( + "INSERT INTO nodes(node_id,long_name,protocol,synthetic,last_heard,first_heard,last_advert_heard) VALUES (?,?,?,?,?,?,?)", + [node_id, "Alice", "meshcore", synthetic ? 1 : 0, heard, heard, synthetic ? nil : heard], + ) + ensure + db&.close + end + + # Store message rows directly, bypassing the insert guard, so an example + # can start from copies the guard would have collapsed. + # + # @param messages [Array] message payloads from +channel_copy+. + # @return [void] + def store_directly(*messages) + db = open_db + columns = %w[id rx_time rx_iso from_id to_id channel channel_name portnum text protocol] + messages.each do |message| + db.execute( + "INSERT INTO messages(#{columns.join(",")}) VALUES (#{Array.new(columns.length, "?").join(",")})", + message.values_at(*columns), + ) + end + ensure + db&.close + end + + # Run each payload through +insert_message+, in order. + # + # @param messages [Array] message payloads. + # @return [void] + def insert(*messages) + db = open_db + messages.each { |message| harness.insert_message(db, message) } + ensure + db&.close + end + + # Capture the content-dedup lookup +insert_message+ runs for +message+. + # + # @param message [Hash] message payload. + # @return [Array(String, Array)] the lookup's SQL and bind values. + def captured_lookup(message) + db = open_db + captured = nil + lookup = db.method(:get_first_value) + db.define_singleton_method(:get_first_value) do |sql, *args| + captured ||= [sql, args.first] if sql.include?("FROM messages") && sql.include?("text = ?") + lookup.call(sql, *args) + end + harness.insert_message(db, message) + captured + ensure + db&.close + end + + it "collapses cross-slot copies whose sender resolved to the pubkey id on one ingestor and the synthetic id on the other" do + seed_alice(real_id, synthetic: false, heard: base_rx - 500) + seed_alice(synthetic_id, synthetic: true, heard: base_rx - 500) + + insert( + channel_copy(slot: 4, from_id: real_id), + channel_copy(slot: 6, from_id: synthetic_id, dt: 2), + ) + + expect(stored_messages.map { |row| [row["id"], row["from_id"]] }).to eq([[slot_ids[4], real_id]]) + end + + it "upgrades a synthetic-attributed survivor when the pubkey copy arrives later (MR3 rank)" do + seed_alice(real_id, synthetic: false, heard: base_rx - 500) + seed_alice(synthetic_id, synthetic: true, heard: base_rx - 500) + + insert( + channel_copy(slot: 6, from_id: synthetic_id), + channel_copy(slot: 4, from_id: real_id, dt: 2), + ) + + # The survivor keeps its own id; only the attribution moves. + expect(stored_messages.map { |row| [row["id"], row["from_id"]] }).to eq([[slot_ids[6], real_id]]) + end + + it "credits the reception to the survivor's sender, not to a positively-stale copy" do + seed_alice(real_id, synthetic: false, heard: base_rx - 500) + stale_heard = now - 40 * 86_400 + seed_alice(stale_id, synthetic: false, heard: stale_heard) + + insert( + channel_copy(slot: 4, from_id: real_id), + channel_copy(slot: 6, from_id: stale_id, dt: 2), + ) + + expect(stored_messages.map { |row| row["from_id"] }).to eq([real_id]) + expect(last_heard_of(stale_id)).to eq(stale_heard) + expect(last_heard_of(real_id)).to eq(base_rx + 2) + end + + it "fills radio metadata the stored copy lacks from the later copy" do + insert( + channel_copy(slot: 4, from_id: synthetic_id), + channel_copy(slot: 6, from_id: synthetic_id, dt: 2, lora_freq: 869, modem_preset: "SF8/BW62/CR8"), + ) + + metadata = stored_messages.map { |row| [row["id"], row["lora_freq"], row["modem_preset"]] } + expect(metadata).to eq([[slot_ids[4], 869, "SF8/BW62/CR8"]]) + end + + it "applies a channel copy to the earliest of two stored matching rows" do + # The earlier copy carries the larger id, so neither id order nor + # insertion order can stand in for receive-time order. + store_directly( + channel_copy(slot: 4, from_id: real_id, dt: 5, id: 2_000_001), + channel_copy(slot: 6, from_id: synthetic_id, id: 2_000_009), + ) + + insert(channel_copy(slot: 4, from_id: real_id, dt: 10, lora_freq: 869)) + + expect(stored_messages.map { |row| [row["id"], row["lora_freq"]] }).to eq([[2_000_009, 869], [2_000_001, nil]]) + end + + it "applies a direct-message copy to the earliest of two stored matching rows" do + store_directly( + channel_copy(slot: 4, from_id: real_id, dt: 5, id: 3_000_001, **direct_message), + channel_copy(slot: 6, from_id: real_id, id: 3_000_009, **direct_message), + ) + + insert(channel_copy(slot: 4, from_id: real_id, dt: 10, lora_freq: 869, **direct_message)) + + expect(stored_messages.map { |row| [row["id"], row["lora_freq"]] }).to eq([[3_000_009, 869], [3_000_001, nil]]) + end + + it "re-derives the survivor from the copy's own id when a busy database forces a retry" do + # The real retrying +with_busy_retry+ replaces the harness's single pass + # (the harness class already includes Database, so +extend+ would not). + harness.define_singleton_method(:with_busy_retry, PotatoMesh::App::Database.instance_method(:with_busy_retry)) + store_directly( + channel_copy(slot: 6, from_id: synthetic_id, id: 4_000_009), + channel_copy(slot: 6, from_id: synthetic_id, dt: 5, id: 4_000_001), + ) + db = open_db + busy = true + row_lookup = db.method(:get_first_row) + # Fail the first stored-row lookup, after the guard has picked the + # survivor, so the whole attempt re-runs. + db.define_singleton_method(:get_first_row) do |sql, *args| + if busy && sql.start_with?("SELECT from_id, to_id, text, encrypted") + busy = false + raise SQLite3::BusyException, "database is locked" + end + row_lookup.call(sql, *args) + end + + harness.insert_message(db, channel_copy(slot: 4, from_id: real_id, dt: 10, lora_freq: 869)) + db.close + + expect(busy).to be(false) + expect(stored_messages.map { |row| [row["id"], row["lora_freq"]] }).to eq([[4_000_009, 869], [4_000_001, nil]]) + end + + it "does not let a stored channel copy without a sender absorb an attributed copy" do + store_directly(channel_copy(slot: 6, from_id: nil, id: 5_000_001)) + + insert(channel_copy(slot: 4, from_id: real_id, dt: 2)) + + expect(stored_messages.map { |row| [row["id"], row["from_id"]] }).to eq([[5_000_001, nil], [slot_ids[4], real_id]]) + end + + it "does not let a stored direct message absorb a channel copy with the same text" do + # Neither row knows its channel name, so only the recipient tells the + # direct message apart from the broadcast. + store_directly(channel_copy(slot: 4, from_id: stale_id, id: 6_000_001, **direct_message.merge(text: text))) + + insert(channel_copy(slot: 4, from_id: real_id, dt: 2, channel_name: nil)) + + expect(stored_messages.map { |row| row["id"] }).to eq([6_000_001, slot_ids[4]]) + end + + it "keeps identical text on two differently named channels apart when the senders resolved differently" do + insert( + channel_copy(slot: 4, from_id: real_id), + channel_copy(slot: 6, from_id: synthetic_id, dt: 2, channel_name: "#other"), + ) + + expect(stored_messages.length).to eq(2) + end + + it "keeps the from_id leg for direct messages, even when the text has a colon" do + insert( + channel_copy(slot: 4, from_id: real_id, **direct_message), + channel_copy(slot: 6, from_id: stale_id, dt: 2, **direct_message), + ) + + expect(stored_messages.map { |row| row["from_id"] }).to eq([real_id, stale_id]) + end + + it "keeps the from_id leg for a broadcast without a sender prefix" do + insert( + channel_copy(slot: 4, from_id: real_id, text: "no sender prefix"), + channel_copy(slot: 6, from_id: synthetic_id, dt: 2, text: "no sender prefix"), + ) + + expect(stored_messages.length).to eq(2) + end + + it "searches the channel-broadcast lookup on idx_messages_meshcore_text without a scan or sort" do + plan = plan_of(*captured_lookup(channel_copy(slot: 4, from_id: real_id))) + + expect(plan).to match([text_index_search]) + end + + it "searches the sender-keyed lookup on idx_messages_meshcore_text without a scan or sort" do + plan = plan_of(*captured_lookup(channel_copy(slot: 4, from_id: real_id, **direct_message))) + + expect(plan).to match([text_index_search]) + end + end + + describe "one-time purge" do + it "searches every purge statement's earlier copies on idx_messages_meshcore_text" do + helper = Object.new.extend(PotatoMesh::App::Database) + purges = [] + # Record each DELETE the upgrade issues, with its bind values. + allow(helper).to receive(:open_database).and_wrap_original do |original, *args, **kwargs| + original.call(*args, **kwargs).tap do |db| + statement = db.method(:execute) + db.define_singleton_method(:execute) do |sql, *rest, &block| + purges << [sql, rest.first || []] if sql.include?("DELETE FROM messages") + statement.call(sql, *rest, &block) + end + end + end + db = open_db + db.execute("PRAGMA user_version = 2") + db.close + + helper.ensure_schema_upgrades + + expect(purges).not_to be_empty + purges.each do |sql, binds| + plan = plan_of(sql, binds) + expect(plan.grep(/\ASCAN /)).to be_empty + expect(plan.grep(/\A(SCAN|SEARCH) earlier /)).to match([text_index_search]) + end + end + end + + describe "two ingestors through the POST routes" do + let(:app) { Sinatra::Application } + let(:auth_headers) do + { "CONTENT_TYPE" => "application/json", "HTTP_AUTHORIZATION" => "Bearer meshcore-dedup-token" } + end + + before do + @original_token = ENV["API_TOKEN"] + ENV["API_TOKEN"] = "meshcore-dedup-token" + PotatoMesh::App::ApiCache.invalidate_all + end + + after do + if @original_token.nil? + ENV.delete("API_TOKEN") + else + ENV["API_TOKEN"] = @original_token + end + end + + # POST one JSON body and require the ingest route to accept it. + # + # @param path [String] ingest route. + # @param body [Hash] JSON payload. + # @return [void] + def ingest(path, body) + post path, body.to_json, auth_headers + expect(last_response.status).to eq(201) + end + + # Ingestor A has Alice in its contact roster: it posts her pubkey-derived + # node, then its slot-4 copy attributed to that id. + # + # @param dt [Integer] receive-time offset of the copy. + # @return [void] + def roster_ingestor_posts(dt: 0) + ingest("/api/nodes", { + real_id => { + "lastHeard" => now - 2_000, + "protocol" => "meshcore", + "user" => { "longName" => "Alice", "shortName" => "a1a2", "role" => "COMPANION", "publicKey" => "a1a2a3a4" + "00" * 28 }, + }, + "ingestor" => "!00000004", + "protocol" => "meshcore", + }) + ingest("/api/messages", channel_copy(slot: 4, from_id: real_id, dt: dt)) + end + + # Ingestor B has no roster entry for Alice: it posts the synthetic + # placeholder, then its slot-6 copy attributed to the synthetic id. + # + # @param dt [Integer] receive-time offset of the copy. + # @return [void] + def placeholder_ingestor_posts(dt: 0) + ingest("/api/nodes", { + synthetic_id => { + "lastHeard" => base_rx + dt, + "protocol" => "meshcore", + "user" => { "longName" => "Alice", "shortName" => "", "role" => "COMPANION", "synthetic" => true }, + }, + "ingestor" => "!00000006", + "protocol" => "meshcore", + }) + ingest("/api/messages", channel_copy(slot: 6, from_id: synthetic_id, dt: dt)) + end + + # The chat feed every reader (dashboard, Matrix bridge, mobile app) sees. + # + # @return [Array] GET /api/messages rows for MeshCore. + def served_messages + get "/api/messages?protocol=meshcore" + JSON.parse(last_response.body) + end + + it "stores and serves one row when the pubkey-attributed copy arrives first" do + roster_ingestor_posts + placeholder_ingestor_posts(dt: 2) + + expect(stored_messages.map { |row| [row["from_id"], row["channel_name"]] }).to eq([[real_id, "#bot"]]) + expect(served_messages.map { |row| row["text"] }).to eq([text]) + end + + it "stores and serves one row when the synthetic-attributed copy arrives first" do + placeholder_ingestor_posts + roster_ingestor_posts(dt: 2) + + expect(stored_messages.map { |row| row["from_id"] }).to eq([real_id]) + expect(served_messages.map { |row| row["text"] }).to eq([text]) + end + end +end diff --git a/web/spec/meshcore_ghost_nodes_spec.rb b/web/spec/meshcore_ghost_nodes_spec.rb index 9db60253..dfee26f4 100644 --- a/web/spec/meshcore_ghost_nodes_spec.rb +++ b/web/spec/meshcore_ghost_nodes_spec.rb @@ -18,6 +18,7 @@ require "spec_helper" require "sqlite3" require "json" require "digest" +require_relative "support/data_processing_harness" # MeshCore ghost nodes from chat text (issue #883; SPEC GN1-GN4, ACCEPTANCE # GN-A1 - GN-A5). A MeshCore channel message names its sender only in a @@ -28,41 +29,9 @@ require "digest" # sole copy name-resolved to a retired key credits the live key (GN3), and a # POSTed name-derived placeholder records nothing (GN4). RSpec.describe "MeshCore ghost nodes from chat text (issue #883)" do - # Minimal host mixing in the data-processing pipeline, stubbed like the - # harness in data_processing_spec.rb. - let(:harness_class) do - Class.new do - include PotatoMesh::App::DataProcessing - include PotatoMesh::App::Helpers + include MeshcoreNodeSeeds - def debug_log(message, **); end - - def warn_log(message, **); end - - def with_busy_retry - yield - end - - def update_prometheus_metrics(*); end - - def prom_report_ids - [] - end - - def private_mode? - false - end - - def normalize_node_id(_db, node_ref) - parts = canonical_node_parts(node_ref) - parts ? parts[0] : nil - end - - def resolve_protocol(_db, _ingestor, cache: nil) - "meshtastic" - end - end - end + let(:harness_class) { DataProcessingHarness.build } subject(:dp) { harness_class.new } @@ -71,7 +40,9 @@ RSpec.describe "MeshCore ghost nodes from chat text (issue #883)" do # An unrostered sender: the ingestor posts her name-derived id. let(:erin) { derived_id("Erin") } - # Fresh database per example. The production windows are kept (7-day node + # Fresh database per example. Not the shared "with isolated db" context: + # that one shortens +four_weeks_seconds+ to a week, and here the production + # windows are kept (7-day node # list, 28-day keyed-evidence horizon), so "positively stale" means what it # means in the field. The Rack routes open the same stubbed path. around do |example| @@ -106,23 +77,6 @@ RSpec.describe "MeshCore ghost nodes from chat text (issue #883)" do "!" + Digest::SHA256.hexdigest(name)[0, 8] end - # Store a keyed MeshCore contact (roster or advert record) last heard at - # +heard+; its keyed evidence (+last_advert_heard+) is +heard+ as well. - # - # @param db [SQLite3::Database] open database handle. - # @param node_id [String] canonical node id. - # @param name [String] advertised long name. - # @param key_byte [String] two hex digits repeated into the public key. - # @param heard [Integer] unix seconds of the record. - # @return [void] - def seed_keyed_node(db, node_id, name, key_byte, heard) - dp.upsert_node(db, node_id, { - "lastHeard" => heard, - "protocol" => "meshcore", - "user" => { "longName" => name, "shortName" => key_byte, "role" => "COMPANION", "publicKey" => key_byte * 32 }, - }, protocol: "meshcore") - end - # Store a MeshCore row with no keyed evidence and no position: a legacy real # row (MR2 "unknown") or, with +synthetic: 1+, a chat placeholder. # diff --git a/web/spec/support/data_processing_harness.rb b/web/spec/support/data_processing_harness.rb new file mode 100644 index 00000000..59a5367f --- /dev/null +++ b/web/spec/support/data_processing_harness.rb @@ -0,0 +1,149 @@ +# 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 "sqlite3" +require "tmpdir" + +# Shared helpers for specs that call +PotatoMesh::App::DataProcessing+ methods +# in isolation (+data_processing_spec.rb+, +meshcore_dedup_spec.rb+). +module DataProcessingHarness + # Build a minimal host class so the module methods can be called in + # isolation: logging, retry, metrics and privacy collaborators are stubbed, + # and node references normalize to their canonical id without a lookup. + # + # @param protocol [String] value the stubbed +resolve_protocol+ returns. + # @param stub_chat_nodes [Boolean] when true, +process_meshcore_chat_nodes+ + # becomes a no-op that reports the sender as named, so an example drives + # +insert_message+ without the synthetic->real merge machinery (covered by + # its own specs). + # @return [Class] the harness class; instantiate it with +.new+. + def self.build(protocol: "meshtastic", stub_chat_nodes: false) + Class.new do + include PotatoMesh::App::DataProcessing + include PotatoMesh::App::Helpers + + def debug_log(message, **); end + + def warn_log(message, **); end + + def with_busy_retry + yield + end + + def update_prometheus_metrics(*); end + + def prom_report_ids + [] + end + + def private_mode? + false + end + + def normalize_node_id(_db, node_ref) + parts = canonical_node_parts(node_ref) + parts ? parts[0] : nil + end + + define_method(:resolve_protocol) { |_db, _ingestor, cache: nil| protocol } + define_method(:process_meshcore_chat_nodes) { |*| true } if stub_chat_nodes + end + end +end + +# Node-seeding helpers shared by the MeshCore specs +# (+data_processing_spec.rb+, +meshcore_ghost_nodes_spec.rb+). The including +# group must define +dp+, an instance of a +DataProcessingHarness.build+ class. +module MeshcoreNodeSeeds + # Store a keyed MeshCore contact (roster or advert record) last heard at + # +heard+; its keyed evidence (+last_advert_heard+) is +heard+ as well. + # + # @param db [SQLite3::Database] open database handle. + # @param node_id [String] canonical node id. + # @param name [String] advertised long name. + # @param key_byte [String] two hex digits repeated into the public key. + # @param heard [Integer] unix seconds of the record. + # @return [void] + def seed_keyed_node(db, node_id, name, key_byte, heard) + dp.upsert_node(db, node_id, { + "lastHeard" => heard, + "protocol" => "meshcore", + "user" => { "longName" => name, "shortName" => key_byte, "role" => "COMPANION", "publicKey" => key_byte * 32 }, + }, protocol: "meshcore") + end +end + +# A fresh SQLite database per example, built by +init_db+ and +# +ensure_schema_upgrades+ like a booted deployment, with the timing knobs +# pinned. +seed_node+ expects the including group to define +dp+. +RSpec.shared_context "with isolated db" do + around do |example| + Dir.mktmpdir("dp-spec-") do |dir| + db_path = File.join(dir, "mesh.db") + RSpec::Mocks.with_temporary_scope do + allow(PotatoMesh::Config).to receive(:db_path).and_return(db_path) + allow(PotatoMesh::Config).to receive(:db_busy_timeout_ms).and_return(5000) + allow(PotatoMesh::Config).to receive(:week_seconds).and_return(604_800) + allow(PotatoMesh::Config).to receive(:four_weeks_seconds).and_return(604_800) + allow(PotatoMesh::Config).to receive(:debug?).and_return(false) + db_helper = Object.new.extend(PotatoMesh::App::Database) + db_helper.init_db + db_helper.ensure_schema_upgrades + example.run + end + end + end + + # Open a handle on the example's isolated database that returns rows as + # hashes keyed by column name. The caller closes it. + # + # @return [SQLite3::Database] open database handle. + def open_db + db = SQLite3::Database.new(PotatoMesh::Config.db_path) + db.results_as_hash = true + db + end + + # Return the full node row for the canonical test node ID. + # + # @param db [SQLite3::Database] open database handle. + # @return [Hash, Array, nil] the +nodes+ row for +!aabbccdd+, or nil when it + # does not exist. + def read_node(db) + db.execute("SELECT * FROM nodes WHERE node_id = '!aabbccdd'").first + end + + # Insert the canonical test node with full user info and CLIENT_BASE role. + # + # @param db [SQLite3::Database] open database handle. + # @return [void] + def seed_node(db) + dp.upsert_node(db, "!aabbccdd", { + "lastHeard" => now - 100, + "num" => 0xaabbccdd, + "user" => { + "role" => "CLIENT_BASE", + "longName" => "Real Long Name", + "shortName" => "RLN", + "macaddr" => "aa:bb:cc:dd:ee:ff", + "hwModel" => "TBEAM", + "publicKey" => "abc123", + }, + }) + end + + let(:now) { Time.now.to_i } +end