web: dedup MeshCore channel copies whose sender resolved differently (#880) (#914)

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)

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<N>` 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.
 

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';

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

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

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)
                      )

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 = []

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<Array>] +[id, rx_time, from_id, to_id, channel,
+  #   channel_name, text]+ tuples.
+  # @param boots [Integer] number of +ensure_schema_upgrades+ passes.
+  # @return [Array<Integer>] 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

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<N>`
+# 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<Hash>] 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<String>] 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<Hash>] 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<Hash>] 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<Hash>] 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

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.
   #

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