diff --git a/ACCEPTANCE.md b/ACCEPTANCE.md index 93712897..710b92f6 100644 --- a/ACCEPTANCE.md +++ b/ACCEPTANCE.md @@ -2186,3 +2186,74 @@ are **updated or removed as dead**, never left dangling: `__tests__/config.test. `applyFiltersToAllTiles` hook), and the Ruby config/app specs that asserted `data-app-config` `tileFilters`. `main/__tests__/offline-tile-layer.test.js` stays green — the fallback layer is retained, now reached only per DM-A3. + +--- + +## Bugfix: MeshCore dedup window vs inter-ingestor clock skew; warm-cache chat gap + +Two chat defects found on production `potatomesh.net` (v0.7.1-rc0) with two +live MeshCore ingestors. **(2) Duplicates:** 28% of MeshCore rows were +distinct-id copies of the same transmission from two ingestors whose host +clocks differ by a consistent ~126 s (median 126 s, p90 133 s). The content +dedup (`data_processing/messages.rb`) keys correctly on `channel_name` (#825, +MD-A1) but bounded the match to `rx_time ± 30 s`, so 89.6% of dup pairs fell +outside the window and persisted; the one-shot #756 purge additionally keyed on +the per-receiver `channel` **index** (not `channel_name`), so it could not +collapse the cross-slot copies even when it ran. Fix: widen +`MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS` 30→300 (covers ~99.5% of the observed +skew; **accepted tradeoff:** a sender's *identical* text repeated within 300 s +collapses — chosen over a 28% dup rate; the one-shot purge applies this +**transitively**, so a chain of such repeats spanning longer than 300 s also +collapses — a deliberately aggressive one-time cleanup, gentler per-insert guard +governs new rows), key the purge on `channel_name`, and bump +`MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION` so the purge re-runs once to clear the +accumulated duplicates. **(1) Missing messages:** on a warm revisit +the cache (FC2) seeds an older contiguous block, but the delta `since`-fetch is +capped at `MESSAGE_LIMIT` and returns the **newest** page (`ORDER BY rx_time +DESC LIMIT`), which need not reach the cache — orphaning the window between the +cache's newest row and the newest page's oldest row. `backfillChatHistory` +anchored at the **global-oldest** loaded row and paged further into the past, so +it never bridged the gap. Fix: anchor the backfill at the **live frontier** (the +oldest row of the newest delta page). The duplicate inflation (defect 2) widened +the gap, so the two interact, but each has a distinct root cause. Web-only; no +wire/contract change; apex (I)/privacy (II) untouched. + +### MW-A1 — Dedup spans the observed inter-ingestor clock skew (runtime + purge) +```bash +( cd web && bundle exec rspec spec/data_processing_spec.rb -e "meshcore content dedup" \ + spec/database_spec.rb -e "cross-ingestor meshcore pair" ) +``` +**Expected:** pass. Runtime: two MeshCore copies with identical `from_id` / +`to_id` / `text` / `channel_name` ("#ping") but different `channel` slots +(10 vs 18) and `rx_time` **126 s apart** collapse to one row (was two — the +30 s window). The one-shot purge collapses the same cross-slot, clock-skewed +pair to a single row by keying on `channel_name` and spanning the widened +window. `MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS == 300` and +`MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION` is bumped so the purge re-runs once. +Companion #756/#825 examples still hold (different `channel_name` / `text` / +`to_id` stay separate; beyond-window — now `> 300 s` — stays separate). + +### MW-A2 — Warm-cache load bridges the orphaned middle gap +```bash +( cd web && node --test public/assets/js/app/__tests__/main-cache-refresh.test.js ) +``` +**Expected:** pass, including "warm cache + capped since-page bridges the +orphaned middle gap": with a seeded cache whose newest row predates the newest +`since`-page by more than one page, the background backfill fetches the +in-between rows (anchored at the live frontier) so **every** in-window message +loads — no orphaned hole. The cold-load path is unchanged (live frontier == +global-oldest when there is no cache), so the existing seed-then-delta examples +(FC-A2) and the progressive-load walk (PL-A1/PL-A2) stay green. + +### MW-R1 — Regression: prior acceptance still holds +```bash +( cd web && bundle exec rspec ) && ( cd web && npm test ) +( . .venv/bin/activate && pytest -q tests/ ) +``` +**Expected:** all green. At risk and required to remain green: **C5 / MD-A1** +(cross-ingestor dedup — strengthened, not weakened), the #756 backfill examples +(within-window collapse, beyond-window preserve — now measured against 300 s, +idempotent, `user_version`-gated), **FC-A2** (seed-then-delta — the warm delta +contract is unchanged; only the backfill anchor moved), **PL-A1/PL-A2** +(progressive load), and **B1**. No POST/GET/event contract change and no +ingestor change, so **C2**, `CONTRACTS.md`, and the Python suite are unaffected. diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index 76fb09f1..d4e0773b 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -95,7 +95,7 @@ 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. +- *Relay-rewritten `sender_timestamp` & cross-ingestor clock skew (#756 / #825).* MeshCore has been observed delivering the same physical packet twice with a rewritten `sender_timestamp`, which flips the v1 fingerprint and bypasses the `messages.id` PK collapse; and two ingestors hearing one packet stamp it with their own host clocks, which drift. To cover both, 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_name, text)` within ±`MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS` of `rx_time` is dropped. The match keys on the sender-stable **`channel_name`** (not the per-receiver `channel` slot index, which differs across ingestors for one logical channel — #825). The window is **300 s**: a production fleet of two live ingestors showed a consistent ~126 s inter-ingestor clock offset (median 126 s, p90 133 s), so a former 30 s window let 28 % of meshcore rows through as duplicates. **Accepted trade-off:** a sender repeating the *identical* text on the same channel **within 300 s** is 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 (re-run when the window or key changes via `MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION`). That one-shot purge is **transitive** — a chain of identical-content rows each within the window of the previous collapses to a single row even if the chain spans longer than the window — so it is deliberately more aggressive than the per-insert guard (which keeps ~one row per window gap), clearing repeated-identical-text backlogs in one historical pass; new rows are governed only by the gentler per-insert guard. - *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)`. - *Upstream `meshcore` reader crash on truncated advertisements (#754).* `meshcore-py` 2.3.6 (latest at the time of writing) raises `IndexError` from `MessageReader.handle_rx` at `reader.py:365` when a `DEVICE_INFO`/advertisement frame declares `fw_ver >= 10` but omits the trailing `path_hash_mode` byte. Because the frame is parsed inside a detached `asyncio.create_task(...)`, the exception surfaces as `Task exception was never retrieved` on stderr and the event for that frame is lost. The ingestor installs a runtime patch (`data/mesh_ingestor/protocols/_meshcore_patches.py`) that wraps `handle_rx`, logs one line with the first 32 bytes of the offending frame under `context=meshcore.reader.patch`, and lets the task exit cleanly; a loop-level handler (`context=asyncio.unhandled`) catches anything the targeted patch misses. Both shims are additive and will be removed once upstream ships a defensive length check. diff --git a/web/lib/potato_mesh/application/data_processing/coercions.rb b/web/lib/potato_mesh/application/data_processing/coercions.rb index 4f3c1631..8c9e03f9 100644 --- a/web/lib/potato_mesh/application/data_processing/coercions.rb +++ b/web/lib/potato_mesh/application/data_processing/coercions.rb @@ -21,18 +21,26 @@ module DataProcessing 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. + # in +insert_message+ and the matching one-shot backfill. Two + # co-operating ingestors timestamp the same physical packet with their + # own host clock, and those clocks drift: on potatomesh.net a fleet of + # two live ingestors showed a consistent ~126 s offset (median 126 s, + # p90 133 s, p99 221 s), so a 30 s window missed 89.6% of the duplicate + # 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 + # +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 + # #756 / #825 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 + MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS = 300 # Coerce a Ruby boolean into a SQLite integer (1/0) while passing through # any other value unchanged. Used when writing boolean node fields. diff --git a/web/lib/potato_mesh/application/database.rb b/web/lib/potato_mesh/application/database.rb index 856724c1..10f8b5e9 100644 --- a/web/lib/potato_mesh/application/database.rb +++ b/web/lib/potato_mesh/application/database.rb @@ -21,7 +21,12 @@ module Database # 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 + # Bumped 1→2: the purge now keys on the stable +channel_name+ (not the + # per-receiver +channel+ slot index) and inherits the widened + # +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 # Column definitions required for environment telemetry support. Each # entry pairs the column name with the SQL type used when backfilling @@ -229,6 +234,10 @@ def ensure_schema_upgrades unless message_columns.include?("channel_name") db.execute("ALTER TABLE messages ADD COLUMN channel_name TEXT") + # Keep the cached column list current so the meshcore content-dedup + # purge below (which keys on +channel_name+) runs on this same boot + # rather than waiting for the next restart. + message_columns << "channel_name" end unless message_columns.include?("reply_id") @@ -264,7 +273,10 @@ def ensure_schema_upgrades # 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] + # +channel_name+ is required because the purge keys on it (the stable + # cross-ingestor discriminator) rather than the per-receiver +channel+ + # slot index; a table lacking the column degrades to no purge. + meshcore_dedup_columns = %w[from_id to_id channel_name 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 @@ -273,11 +285,25 @@ def ensure_schema_upgrades 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. + # 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. + # + # NOTE — this EXISTS predicate is TRANSITIVE and so does NOT behave + # identically to the runtime guard: a chain of identical-content rows + # each within the window of the previous collapses to a single row + # even if the chain spans MORE than the window, whereas the runtime + # guard (which only compares a new row against already-persisted rows) + # keeps roughly one row per window gap. This is intentionally more + # aggressive for the one-shot historical sweep — it also clears + # repeated-identical-text backlogs (e.g. a beacon/bot re-posting the + # same string on a cadence ≤ the window) on top of the cross-ingestor + # duplicates — and is a one-time effect; the gentler runtime guard + # 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 @@ -303,7 +329,7 @@ def ensure_schema_upgrades 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.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 <= ? diff --git a/web/public/assets/js/app/__tests__/main-cache-refresh.test.js b/web/public/assets/js/app/__tests__/main-cache-refresh.test.js index 901a78ad..0a86ea72 100644 --- a/web/public/assets/js/app/__tests__/main-cache-refresh.test.js +++ b/web/public/assets/js/app/__tests__/main-cache-refresh.test.js @@ -190,3 +190,64 @@ test('a disabled cache (no IndexedDB) leaves the app on the cold network path', env.cleanup(); } }); + +test('warm cache + capped since-page bridges the orphaned middle gap', async () => { + // Regression (production: #ping Jun27 08:00 → Jun28 00:00 missing on revisit). + // A returning visitor's cache holds an older contiguous block; the delta + // `since`-fetch is capped at MESSAGE_LIMIT and returns only the NEWEST page + // (DESC+LIMIT), which need not reach the cache — leaving an orphaned middle + // gap [cache_newest .. livepage_oldest]. The background backfill must bridge + // that gap (anchor at the live frontier), not page below the global-oldest + // cached row. + const fake = createFakeIndexedDb(); + const seed = createIndexedDbBackend({ indexedDB: fake.factory, databaseName: 'potato-mesh-cache' }); + await seed.write('meta', 'meta', { schemaVersion: CACHE_SCHEMA_VERSION, instanceId: CONFIG.instanceDomain }); + const msg = (id, t, text) => ({ id, channel: 0, from_id: '!a', to_id: '^all', text, rx_time: t, protocol: 'meshcore' }); + // Cached: an older contiguous pair, within the 7-day retention so it seeds. + await seed.write('messages', 'c1', { value: msg('c1', NOW - 2 * DAY - 60, 'old1'), cachedAt: NOW }); + await seed.write('messages', 'c2', { value: msg('c2', NOW - 2 * DAY, 'old2'), cachedAt: NOW }); + + // Server: the newest page (since-fetch) is [L1@NOW-1h, L2@NOW] and does NOT + // reach the cache; the gap row G@NOW-1d is reachable only via `before=` paging. + const L1 = msg('L1', NOW - 3600, 'live1'); + const L2 = msg('L2', NOW, 'live2'); + const G = msg('G', NOW - DAY, 'gap'); + const history = [L2, L1, G, msg('c2', NOW - 2 * DAY, 'old2'), msg('c1', NOW - 2 * DAY - 60, 'old1')]; + const ok = body => Promise.resolve({ ok: true, status: 200, json: () => Promise.resolve(body) }); + const stubFetch = url => { + if (url.includes('/api/messages')) { + const q = url.split('?')[1] || ''; + if (/encrypted=true/.test(q)) return ok([]); + const before = Number((q.match(/before=(\d+)/) || [])[1] || 0); + if (before > 0) { + return ok(history.filter(m => m.rx_time <= before).sort((a, b) => b.rx_time - a.rx_time).slice(0, 1000)); + } + return ok([L2, L1]); // since/default: newest page only (cap doesn't reach the cache) + } + return ok([]); + }; + + const env = createDomEnvironment({ includeBody: true }); + env.registerElement('chat', env.createElement('div', 'chat')); + const of = globalThis.fetch; + const oi = globalThis.indexedDB; + globalThis.fetch = stubFetch; + globalThis.indexedDB = fake.factory; + try { + const { _testUtils } = initializeApp(CONFIG); + await _testUtils.initialLoad; + await _testUtils.flushBackfill(); + await _testUtils.flushCacheWrites(); + // After bridging, every distinct message is loaded: cache c1,c2 + live L1,L2 + // + the gap row G = 5. Pre-fix the backfill pages below the cached oldest and + // never fetches G, so only 4 load. + assert.equal( + _testUtils.getLoadedMessageCount(), 5, + `the orphaned middle gap (G@NOW-1d) must be backfilled; loaded=${_testUtils.getLoadedMessageCount()}`, + ); + } finally { + globalThis.fetch = of; + globalThis.indexedDB = oi; + env.cleanup(); + } +}); diff --git a/web/public/assets/js/app/main.js b/web/public/assets/js/app/main.js index cd7528db..4f24c968 100644 --- a/web/public/assets/js/app/main.js +++ b/web/public/assets/js/app/main.js @@ -336,6 +336,15 @@ export function initializeApp(config) { let chatBackfillRunning = false; /** One-shot guard: the chat-history backfill runs once after the first load. */ let chatHistoryBackfilled = false; + /** + * Oldest ``rx_time`` of the newest delta page (the "live frontier"). The + * background backfill pages backward from here, not from the global-oldest + * loaded row, so a warm-cache load bridges the gap between the newest page + * and the seeded cache instead of paging below the cache and orphaning the + * window in between. 0 until the first fetch; on a cold load it equals the + * global-oldest, so the cold backfill is unchanged. + */ + let chatLiveFrontier = 0; /** Settles when the one-shot chat-history backfill finishes (test hook). */ let backfillPromise = Promise.resolve(); @@ -3453,8 +3462,15 @@ export function initializeApp(config) { */ async function backfillChatHistory() { if (!CHAT_ENABLED || chatBackfillRunning) return; - // Resume paging from just past the oldest message the newest page rendered. - const before = minRecordTimestamp(allMessages, ['rx_time']); + // Page backward from the live frontier (oldest row of the newest delta + // page), not the global-oldest loaded row. On a cold load they are equal; + // on a warm-cache load the cache contributes older rows, so anchoring at the + // global min would page below the cache and never fill the gap between the + // cache's newest row and the newest page's oldest row (the orphaned middle + // window). Falls back to the global min when no live page was fetched. + const before = chatLiveFrontier > 0 + ? chatLiveFrontier + : minRecordTimestamp(allMessages, ['rx_time']); if (!(before > 0)) return; chatBackfillRunning = true; try { @@ -4495,6 +4511,11 @@ export function initializeApp(config) { // Update high-water marks for incremental fetching. const incomingNodeTs = maxRecordTimestamp(incomingNodes, ['last_heard']); const incomingMsgTs = maxRecordTimestamp(incomingMessages, ['rx_time']); + // Record the oldest row of this delta page as the backfill's live frontier + // (see {@link chatLiveFrontier}); the newest delta page is what the + // one-shot backfill must extend backward from. + const incomingMsgOldest = minRecordTimestamp(incomingMessages, ['rx_time']); + if (incomingMsgOldest > 0) chatLiveFrontier = incomingMsgOldest; const incomingEncMsgTs = maxRecordTimestamp(incomingEncryptedMessages, ['rx_time']); const incomingPosTs = maxRecordTimestamp(incomingPositions, ['rx_time', 'position_time']); const incomingTelTs = maxRecordTimestamp(incomingTelemetry, ['rx_time', 'telemetry_time']); diff --git a/web/spec/data_processing_spec.rb b/web/spec/data_processing_spec.rb index c4950753..c26972c7 100644 --- a/web/spec/data_processing_spec.rb +++ b/web/spec/data_processing_spec.rb @@ -1389,6 +1389,34 @@ def message_count(db) db&.close end + it "collapses cross-ingestor copies separated by the observed inter-ingestor clock skew" do + # Production reproduction (potatomesh.net: 28% meshcore duplicate rate). + # Two ingestors whose host clocks differ by ~126 s store the same physical + # #ping transmission at different local channel slots (10 vs 18) with + # rx_times ~126 s apart. The 30 s window let both rows persist; the dedup + # window must span the real-world inter-ingestor skew (median 126 s, p90 + # 133 s observed). A literal 126 s delta pins the behaviour independently + # of the window constant's exact value. + db = open_db + meshcore_harness.insert_message( + db, + base_message.merge( + "id" => 1_000_301, "channel" => 10, "channel_name" => "#ping", + "ingestor" => "!02294310", + ), + ) + meshcore_harness.insert_message( + db, + base_message.merge( + "id" => 1_000_302, "rx_time" => base_rx_time + 126, + "channel" => 18, "channel_name" => "#ping", "ingestor" => "!930d4a21", + ), + ) + expect(message_count(db)).to eq(1) + 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")) diff --git a/web/spec/database_spec.rb b/web/spec/database_spec.rb index 456dd5cb..e52f1767 100644 --- a/web/spec/database_spec.rb +++ b/web/spec/database_spec.rb @@ -499,7 +499,7 @@ def seed_meshcore_message_tables(db) 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, + from_id TEXT, to_id TEXT, channel INTEGER, channel_name TEXT, text TEXT, protocol TEXT NOT NULL DEFAULT 'meshtastic' ) SQL @@ -528,6 +528,69 @@ def seed_meshcore_message_tables(db) end end + it "collapses a cross-ingestor meshcore pair on different local channel slots with clock-skewed rx_time" do + # Production reproduction: two ingestors store the same #ping transmission + # at local channel slots 10 and 18 with rx_time ~126 s apart (host-clock + # skew). The one-shot purge must key on the stable channel *name* (not the + # per-receiver slot index) and span the observed skew, or the accumulated + # duplicates (28% of rows on potatomesh.net) never clear. + 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,channel_name,text,protocol) VALUES (?,?,?,?,?,?,?,?,?)", + [9_001, 1_776_750_000, "2026-04-20T00:00:00Z", "!0bcfcc74", "^all", 10, "#ping", "JO62SQ: Pong!", "meshcore"], + ) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,channel_name,text,protocol) VALUES (?,?,?,?,?,?,?,?,?)", + [9_002, 1_776_750_126, "2026-04-20T00:02:06Z", "!0bcfcc74", "^all", 18, "#ping", "JO62SQ: Pong!", "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([9_001]) + end + 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 + # see it (keying on channel_name) the same boot, or the accumulated + # duplicates linger until the next restart. Two cross-slot copies (channel + # 10 vs 18), clock-skewed 126 s, NULL channel_name → IS-match collapse. + SQLite3::Database.new(PotatoMesh::Config.db_path) do |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 + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [9_101, 1_776_751_000, "2026-04-20T01:00:00Z", "!0bcfcc74", "^all", 10, "JO62SQ: Ping", "meshcore"], + ) + db.execute( + "INSERT INTO messages(id,rx_time,rx_iso,from_id,to_id,channel,text,protocol) VALUES (?,?,?,?,?,?,?,?)", + [9_102, 1_776_751_126, "2026-04-20T01:02:06Z", "!0bcfcc74", "^all", 18, "JO62SQ: 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([9_101]) + 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)