diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index f65d8df..72133f7 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -78,6 +78,8 @@ The `v1:` prefix lets the format evolve (e.g. add a channel-secret hash) without - *Format-string ambiguity around `:`.* Components are joined with literal colons and not length-prefixed, so a colon embedded in `sender_identity` or `text` shifts the boundary between fields. In theory two distinct triples (e.g. `sender_identity="a:b"` vs `sender_identity="a"` with a leading `b:` in `text`) can produce the same fingerprint. In practice this is vanishingly rare — MeshCore sender names rarely contain colons and even then both senders would have to land on the same timestamp/channel — but a `v2` revision should switch to a delimiter that cannot appear in any component (e.g. `\x00`) or length-prefix each field. - *meshcore_py text-decoding inconsistency.* The upstream `meshcore_py` reader strips trailing `\0` bytes on the real-time `CHANNEL_MSG_RECV` path but not on the sync-replay path. If the same physical message is heard once in real-time and once via sync-replay, the byte sequences differ → different fingerprints → 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. Two messages from the same sender containing the same text within one second of power-on collapse into a single row. Acceptable trade-off given the alternative (no dedup at all). +- *Relay-rewritten `sender_timestamp` (#756).* MeshCore has been observed delivering the same physical packet twice with a rewritten `sender_timestamp` (≈10 s later, same `from_id`/`channel`/`text`), which flips the v1 fingerprint and bypasses the `messages.id` PK collapse. To cover this, the web app runs an additional content-level dedup on insert: for `protocol = "meshcore"` with non-empty `text` and a known `from_id`, a second row matching `(from_id, to_id, channel, text)` within ±30 s of `rx_time` is dropped (window lives in `MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS`). The window is ~3× the observed relay delta; legitimate rapid re-sends of identical short text (e.g. `hi`, `ack`, `ok`, `test`) from the same sender on the same channel **within 30 s** will be silently collapsed into one row. Ingestors MUST still produce deterministic v1 ids — this content-level layer is additive, not a replacement. Pre-existing duplicates are cleared once by a `PRAGMA user_version`-gated one-shot backfill on startup. +- *Concurrent-insert race (#756).* The content-dedup SELECT and the downstream INSERT are not currently wrapped in a shared transaction, so two concurrent Puma threads carrying the same content with different ids can both pass the pre-check and both insert. Duplicates produced this way are narrow (single-node multi-threaded ingest) and are not cleaned up on subsequent boots because the backfill is one-shot. If the race is ever observed in production, tighten `insert_message` to wrap the meshcore pre-check + id-PK path in `db.transaction(:immediate)`. #### `POST /api/positions` diff --git a/web/lib/potato_mesh/application/data_processing.rb b/web/lib/potato_mesh/application/data_processing.rb index d3145eb..cdaa1b2 100644 --- a/web/lib/potato_mesh/application/data_processing.rb +++ b/web/lib/potato_mesh/application/data_processing.rb @@ -20,6 +20,20 @@ module PotatoMesh # Allowed values for the +telemetry_type+ discriminator column. VALID_TELEMETRY_TYPES = %w[device environment power air_quality].freeze + # Half-window (seconds) for the meshcore content-level message dedup + # in +insert_message+ and the matching one-shot backfill. Set to + # roughly 3× the observed relay-retransmit delta (~10 s) so genuine + # clock skew across co-operating ingestors still collapses, while + # rapid legitimate re-sends ("ack", "ok", "test") ≥30 s apart remain + # distinct rows. See issue #756 and ``CONTRACTS.md`` for rationale. + # + # IMPORTANT: widening this value only takes effect at runtime — the + # one-shot backfill in +PotatoMesh::App::Database+ is frozen at + # +MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION+. To re-sweep pre-existing + # rows that newly fall within an expanded window, bump the backfill + # version so the migration re-runs on the next deploy. + MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS = 30 + # Coerce a Ruby boolean into a SQLite integer (1/0) while passing through # any other value unchanged. Used when writing boolean node fields. # @@ -1914,6 +1928,59 @@ module PotatoMesh ] with_busy_retry do + # 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 + # has been observed delivering the same physical packet twice with + # a rewritten ``sender_timestamp`` (relay/retransmit behaviour). + # The PK path below cannot catch that — two copies compute two + # different ids — so we add a narrow content+window pre-check here. + # + # Ruby integer ``0`` is truthy, so the ``channel_index`` guard + # passes for the broadcast channel intentionally; we only skip when + # the channel is absent/nil. ``from_id`` + non-empty ``text`` keep + # encrypted or anonymous traffic on the id-PK path. + # + # 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. + if protocol == "meshcore" && from_id && channel_index && text && !text.to_s.empty? + # ``channel = ?`` matches the ``channel_index`` bind cleanly + # because the guard above rejects nil; ``to_id`` may legitimately + # be nil (rare meshcore fallback), so it keeps ``IS ?`` for a + # NULL-safe compare. + duplicate_id = db.get_first_value( + <<~SQL, + SELECT id FROM messages + WHERE protocol = 'meshcore' + AND from_id = ? + AND to_id IS ? + AND channel = ? + AND text = ? + AND rx_time BETWEEN ? AND ? + AND id != ? + LIMIT 1 + SQL + [from_id, to_id, channel_index, text, + rx_time - MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS, + rx_time + MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS, msg_id], + ) + if duplicate_id + debug_log( + "Skipped meshcore message duplicate", + context: "data_processing.insert_message", + new_id: msg_id, + existing_id: duplicate_id, + from_id: from_id, + channel: channel_index, + ) + return + 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], diff --git a/web/lib/potato_mesh/application/database.rb b/web/lib/potato_mesh/application/database.rb index 22cda72..856724c 100644 --- a/web/lib/potato_mesh/application/database.rb +++ b/web/lib/potato_mesh/application/database.rb @@ -17,6 +17,12 @@ module PotatoMesh module App module Database + # Schema-version marker that gates the one-shot #756 meshcore message + # content-dedup backfill. Stored in SQLite's ``PRAGMA user_version``; + # bump this constant when a new one-shot migration is appended and + # check the previous value below to decide whether to skip. + MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION = 1 + # Column definitions required for environment telemetry support. Each # entry pairs the column name with the SQL type used when backfilling # legacy databases that pre-date the extended telemetry schema. @@ -251,6 +257,64 @@ module PotatoMesh unless reply_index_exists 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 + # 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. + meshcore_dedup_columns = %w[from_id to_id channel text rx_time protocol] + if meshcore_dedup_columns.all? { |column| message_columns.include?(column) } + db.execute(<<~SQL) + CREATE INDEX IF NOT EXISTS idx_messages_meshcore_content + ON messages(from_id, channel, rx_time) + WHERE protocol = 'meshcore' + SQL + + # #756 backfill — collapse pre-existing meshcore duplicate groups. + # Keep the earliest (min rx_time, min id) copy in each + # (from_id, to_id, channel, text) cluster where any two rows are + # within #{PotatoMesh::App::DataProcessing::MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS} s + # of each other. Window matches the runtime guard so runtime and + # backfill behave identically. + # + # 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. + 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 + db.transaction do + # Window bound via ``?`` to match the rest of the codebase's + # parameter-binding style; the value is a Ruby integer constant + # so SQL-injection was never at risk here — the switch is + # purely for consistency. ``PRAGMA user_version`` cannot + # accept bind params, so it keeps literal interpolation of + # an internal constant. + db.execute(<<~SQL, [window]) + DELETE FROM messages + WHERE protocol = 'meshcore' + AND text IS NOT NULL AND text != '' + AND from_id IS NOT NULL + AND EXISTS ( + SELECT 1 FROM messages AS earlier + WHERE earlier.protocol = 'meshcore' + AND earlier.from_id = messages.from_id + AND earlier.to_id IS messages.to_id + AND earlier.channel IS messages.channel + 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) + ) + SQL + db.execute("PRAGMA user_version = #{MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION}") + end + end + end end tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='instances'").flatten diff --git a/web/spec/data_processing_spec.rb b/web/spec/data_processing_spec.rb index eb8470c..120e307 100644 --- a/web/spec/data_processing_spec.rb +++ b/web/spec/data_processing_spec.rb @@ -804,4 +804,226 @@ RSpec.describe PotatoMesh::App::DataProcessing do db&.close end end + + # --------------------------------------------------------------------------- + # insert_message — meshcore content dedup (issue #756). + # --------------------------------------------------------------------------- + describe "#insert_message — meshcore content dedup" do + include_context "with isolated db" + + let(:now) { Time.now.to_i } + + # Shared builder for a minimal ``insert_message`` harness parameterised + # by the protocol it advertises for every POST. Keeping this in one + # place (rather than duplicating per-describe) matches CLAUDE.md's + # modularity guidance and makes it trivial to add a third protocol. + def self.build_protocol_harness(protocol_name) + Class.new do + include PotatoMesh::App::DataProcessing + include PotatoMesh::App::Helpers + + define_method(:resolve_protocol) do |_db, _ingestor, cache: nil| + protocol_name + end + + 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 touch_node_last_seen(*); end + + def ensure_unknown_node(*); end + end.new + end + + let(:meshcore_harness) { self.class.build_protocol_harness("meshcore") } + let(:meshtastic_harness) { self.class.build_protocol_harness("meshtastic") } + + # rx_time sits in the past so we can shift later copies forward (up to + # ``now``) without tripping the ``rx_time > now`` clamp in + # ``insert_message``. + let(:base_rx_time) { now - 1_000 } + let(:dedup_window) { PotatoMesh::App::DataProcessing::MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS } + + let(:base_message) do + { + "rx_time" => base_rx_time, + "from_id" => "!aabbccdd", + "to_id" => "^all", + "channel" => 5, + "text" => "hello from alice", + "portnum" => "TEXT_MESSAGE_APP", + "ingestor" => "!ingest01", + } + end + + def message_count(db) + db.get_first_value("SELECT COUNT(*) FROM messages").to_i + end + + it "skips a second meshcore message with identical content within the dedup window" do + db = open_db + meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_001)) + meshcore_harness.insert_message( + db, + base_message.merge("id" => 1_000_002, "rx_time" => base_rx_time + (dedup_window - 1)), + ) + expect(message_count(db)).to eq(1) + expect(db.get_first_value("SELECT id FROM messages").to_i).to eq(1_000_001) + ensure + db&.close + end + + it "treats the dedup window as inclusive on the upper boundary" do + # Pins the ``BETWEEN`` inclusivity: a row exactly ``dedup_window`` seconds + # later still collapses. One-second-past-the-window inserts below prove + # the other side of the boundary. + db = open_db + meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_021)) + meshcore_harness.insert_message( + db, + base_message.merge("id" => 1_000_022, "rx_time" => base_rx_time + dedup_window), + ) + expect(message_count(db)).to eq(1) + meshcore_harness.insert_message( + db, + base_message.merge("id" => 1_000_023, "rx_time" => base_rx_time + dedup_window + 1), + ) + expect(message_count(db)).to eq(2) + ensure + db&.close + end + + it "inserts both copies when rx_time delta exceeds the dedup window" do + db = open_db + meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_003)) + meshcore_harness.insert_message( + db, + base_message.merge("id" => 1_000_004, "rx_time" => base_rx_time + (dedup_window * 3)), + ) + expect(message_count(db)).to eq(2) + ensure + db&.close + end + + it "does not collapse two meshcore messages on different channels" do + db = open_db + meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_005, "channel" => 5)) + meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_006, "channel" => 6)) + expect(message_count(db)).to eq(2) + ensure + db&.close + end + + it "does not collapse two meshcore messages with different text" do + db = open_db + meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_007, "text" => "first")) + meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_008, "text" => "second")) + expect(message_count(db)).to eq(2) + ensure + db&.close + end + + it "does not collapse two meshcore DMs to different recipients sharing text" do + db = open_db + meshcore_harness.insert_message( + db, + base_message.merge("id" => 1_000_009, "to_id" => "!bbbbbbbb"), + ) + meshcore_harness.insert_message( + db, + base_message.merge("id" => 1_000_010, "to_id" => "!cccccccc", "rx_time" => base_rx_time + 5), + ) + expect(message_count(db)).to eq(2) + ensure + db&.close + end + + it "does not collapse when the incoming message has no text" do + db = open_db + meshcore_harness.insert_message( + db, + base_message.merge("id" => 1_000_011, "text" => "blob"), + ) + # Second payload has no text — the content-dedup branch must not fire, + # so this falls through to the normal id-PK path and inserts. + meshcore_harness.insert_message( + db, + base_message.merge("id" => 1_000_012, "text" => nil, "rx_time" => base_rx_time + 5), + ) + expect(message_count(db)).to eq(2) + ensure + db&.close + end + + it "leaves meshtastic traffic untouched" do + db = open_db + # Two meshtastic packets with the same logical content but distinct + # firmware-assigned packet ids must both land — the new guard is + # scoped to meshcore by design. + meshtastic_harness.insert_message(db, base_message.merge("id" => 1_000_013)) + meshtastic_harness.insert_message( + db, + base_message.merge("id" => 1_000_014, "rx_time" => base_rx_time + 5), + ) + expect(message_count(db)).to eq(2) + ensure + db&.close + end + + 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. + db = open_db + content_select_pattern = /SELECT\s+id\s+FROM\s+messages\s+WHERE\s+protocol\s*=\s*'meshcore'/im + captured_sql = [] + wrapped = db.method(:get_first_value) + allow(db).to receive(:get_first_value) do |sql, *rest| + captured_sql << sql + wrapped.call(sql, *rest) + end + meshtastic_harness.insert_message(db, base_message.merge("id" => 1_000_020)) + expect(captured_sql.any? { |s| s =~ content_select_pattern }).to be(false) + ensure + db&.close + end + + it "still merges on the id-PK path when sender_timestamps collide on the wire" do + db = open_db + # Same id, same content — the existing update-on-match code path should + # patch the stored row rather than insert a duplicate. This proves the + # new dedup guard does not short-circuit the id-match merge behaviour. + meshcore_harness.insert_message(db, base_message.merge("id" => 1_000_015, "ingestor" => nil)) + meshcore_harness.insert_message( + db, + base_message.merge("id" => 1_000_015, "ingestor" => "!ingest99"), + ) + expect(message_count(db)).to eq(1) + expect( + db.get_first_value("SELECT ingestor FROM messages WHERE id = 1000015"), + ).to eq("!ingest99") + ensure + db&.close + end + end end diff --git a/web/spec/database_spec.rb b/web/spec/database_spec.rb index 216347b..456dd5c 100644 --- a/web/spec/database_spec.rb +++ b/web/spec/database_spec.rb @@ -480,4 +480,180 @@ RSpec.describe PotatoMesh::App::Database do expect(msg["from_id"]).to eq("!realidmp") end end + + # --------------------------------------------------------------------------- + # #756 backfill — collapse pre-existing meshcore duplicate message groups. + # --------------------------------------------------------------------------- + + # Build the minimal messages + nodes schema we need for the backfill specs, + # matching the subset of columns the migration inspects. ``role`` and + # ``long_name`` are included because #747's backfill touches them. + def seed_meshcore_message_tables(db) + db.execute(<<~SQL) + CREATE TABLE nodes( + node_id TEXT PRIMARY KEY, long_name TEXT, role TEXT, + protocol TEXT NOT NULL DEFAULT 'meshtastic', + synthetic BOOLEAN NOT NULL DEFAULT 0 + ) + SQL + db.execute(<<~SQL) + CREATE TABLE messages( + id INTEGER PRIMARY KEY, rx_time INTEGER, rx_iso TEXT, + from_id TEXT, to_id TEXT, channel INTEGER, text TEXT, + protocol TEXT NOT NULL DEFAULT 'meshtastic' + ) + SQL + end + + it "collapses a meshcore duplicate pair within the content-dedup window" do + SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| + seed_meshcore_message_tables(db) + # Observed shape from local DB: same from_id/channel/text, rx_time 9s + # apart, ids differ because sender_timestamp was rewritten on relay. + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [3_436_613_256_067_934, 1_776_750_469, "2026-04-20T00:00:00Z", "!e81e448a", "^all", 20, "mirkosw: hi", "meshcore"], + ) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [4_439_171_486_877_153, 1_776_750_478, "2026-04-20T00:00:09Z", "!e81e448a", "^all", 20, "mirkosw: hi", "meshcore"], + ) + end + + harness_class.ensure_schema_upgrades + + SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true) do |db| + ids = db.execute("SELECT id FROM messages ORDER BY id").flatten + expect(ids).to eq([3_436_613_256_067_934]) + end + end + + it "preserves both copies when rx_time delta exceeds the window" do + SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| + seed_meshcore_message_tables(db) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [501, 1_000_000, "2026-04-20T00:00:00Z", "!aabbccdd", "^all", 0, "ping", "meshcore"], + ) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [502, 1_000_600, "2026-04-20T00:10:00Z", "!aabbccdd", "^all", 0, "ping", "meshcore"], + ) + end + + harness_class.ensure_schema_upgrades + + SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true) do |db| + expect(db.execute("SELECT id FROM messages ORDER BY id").flatten).to eq([501, 502]) + end + end + + it "leaves meshtastic duplicates alone even when the content matches" do + SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| + seed_meshcore_message_tables(db) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [601, 1_000_000, "2026-04-20T00:00:00Z", "!aabbccdd", "^all", 0, "pong", "meshtastic"], + ) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [602, 1_000_010, "2026-04-20T00:00:10Z", "!aabbccdd", "^all", 0, "pong", "meshtastic"], + ) + end + + harness_class.ensure_schema_upgrades + + SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true) do |db| + expect(db.execute("SELECT id FROM messages ORDER BY id").flatten).to eq([601, 602]) + end + end + + it "makes the #756 backfill idempotent across successive boots" do + SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| + seed_meshcore_message_tables(db) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [701, 2_000_000, "2026-04-21T00:00:00Z", "!aabbccdd", "^all", 3, "dup", "meshcore"], + ) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [702, 2_000_005, "2026-04-21T00:00:05Z", "!aabbccdd", "^all", 3, "dup", "meshcore"], + ) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [703, 2_000_010, "2026-04-21T00:00:10Z", "!aabbccdd", "^all", 3, "dup", "meshcore"], + ) + end + + 2.times { harness_class.ensure_schema_upgrades } + + SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true) do |db| + expect(db.execute("SELECT id FROM messages ORDER BY id").flatten).to eq([701]) + expect(db.get_first_value("PRAGMA user_version").to_i).to eq( + PotatoMesh::App::Database::MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION, + ) + end + end + + it "gates the #756 backfill behind PRAGMA user_version and does not re-sweep later data" 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. + SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| + seed_meshcore_message_tables(db) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [801, 3_000_000, "2026-04-22T00:00:00Z", "!aabbccdd", "^all", 1, "hi", "meshcore"], + ) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [802, 3_000_005, "2026-04-22T00:00:05Z", "!aabbccdd", "^all", 1, "hi", "meshcore"], + ) + end + + harness_class.ensure_schema_upgrades + + SQLite3::Database.new(PotatoMesh::Config.db_path) do |db| + expect(db.execute("SELECT id FROM messages WHERE id IN (801,802)").flatten).to eq([801]) + # Inject brand new duplicates AFTER the one-shot sweep has run. + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [803, 3_100_000, "2026-04-22T00:01:00Z", "!aabbccdd", "^all", 1, "bye", "meshcore"], + ) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [804, 3_100_003, "2026-04-22T00:01:03Z", "!aabbccdd", "^all", 1, "bye", "meshcore"], + ) + end + + harness_class.ensure_schema_upgrades + + SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true) do |db| + # Second pass must have been gated out by user_version — both new rows + # survive even though they would otherwise match the backfill predicate. + expect(db.execute("SELECT id FROM messages WHERE id IN (803,804) ORDER BY id").flatten).to eq([803, 804]) + end + end + + it "creates the partial index backing the runtime content-dedup lookup" 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 + end end