From 0e25bd2281bc9e0be6a5611e9e481fec621a9bd1 Mon Sep 17 00:00:00 2001 From: Jack Kingsman Date: Mon, 16 Feb 2026 20:46:43 -0800 Subject: [PATCH] Fix dedupe for frontend raw packet delivery --- AGENTS.md | 10 ++- app/AGENTS.md | 7 +- app/models.py | 5 ++ app/packet_processor.py | 5 ++ frontend/AGENTS.md | 6 +- frontend/src/App.tsx | 12 +-- .../src/components/PacketVisualizer3D.tsx | 8 +- frontend/src/components/RawPacketList.tsx | 6 +- frontend/src/test/rawPacketIdentity.test.ts | 79 +++++++++++++++++++ frontend/src/types.ts | 2 + frontend/src/utils/rawPacketIdentity.ts | 31 ++++++++ tests/test_packet_pipeline.py | 37 +++++++++ 12 files changed, 188 insertions(+), 20 deletions(-) create mode 100644 frontend/src/test/rawPacketIdentity.test.ts create mode 100644 frontend/src/utils/rawPacketIdentity.ts diff --git a/AGENTS.md b/AGENTS.md index 1eb5029..b9b9c27 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -104,9 +104,11 @@ The following are **deliberate design choices**, not bugs. They are documented i ## Intentional Packet Handling Decision -Raw packet deduplication is path-aware by design: -- Keep repeats that represent the same payload arriving via different mesh paths (this path diversity is useful). -- Drop packets that are truly identical observations (same payload and same effective pathing observation) to avoid redundant noise. +Raw packet handling uses two identities by design: +- **`id` (DB packet row ID)**: storage identity from payload-hash deduplication (path bytes are excluded), so repeated payloads share one stored raw-packet row. +- **`observation_id` (WebSocket only)**: realtime observation identity, unique per RF arrival, so path-diverse repeats are still visible in-session. + +Frontend packet-feed consumers should treat `observation_id` as the dedup/render key, while `id` remains the storage reference. ## Data Flow @@ -131,6 +133,8 @@ Raw packet deduplication is path-aware by design: **Channel messages**: Flood messages echo back through repeaters. Repeats are identified by the database UNIQUE constraint on `(type, conversation_key, text, sender_timestamp)` — when an INSERT hits a duplicate, `_handle_duplicate_message()` in `packet_processor.py` increments the ack count on the original and adds the new path. There is no timestamp-windowed matching; deduplication is exact-match only. +This message-layer echo/path handling is independent of raw-packet storage deduplication. + ## Directory Structure ``` diff --git a/app/AGENTS.md b/app/AGENTS.md index dc9fd34..1e95c7f 100644 --- a/app/AGENTS.md +++ b/app/AGENTS.md @@ -81,8 +81,11 @@ app/ ### Raw packet dedup policy -- Path diversity is meaningful: same payload observed across different paths should be retained/represented. -- Only truly identical packet observations should be dropped as duplicates. +- Raw packet storage deduplicates by payload hash (`RawPacketRepository.create`), excluding routing/path bytes. +- Stored packet `id` is therefore a payload identity, not a per-arrival identity. +- Realtime raw-packet WS broadcasts include `observation_id` (unique per RF arrival) in addition to `id`. +- Frontend packet-feed features should key/dedupe by `observation_id`; use `id` only as the storage reference. +- Message-layer repeat handling (`_handle_duplicate_message` + `MessageRepository.add_path`) is separate from raw-packet storage dedup. ### Periodic advertisement diff --git a/app/models.py b/app/models.py index fd4f952..0b731e0 100644 --- a/app/models.py +++ b/app/models.py @@ -118,6 +118,11 @@ class RawPacketBroadcast(BaseModel): """ id: int + observation_id: int = Field( + description=( + "Monotonic per-process ID for this RF observation (distinct from the DB packet row ID)" + ) + ) timestamp: int data: str = Field(description="Hex-encoded packet data") payload_type: str = Field(description="Packet type name (e.g., GROUP_TEXT, ADVERT)") diff --git a/app/packet_processor.py b/app/packet_processor.py index 1aa9d86..0d2a218 100644 --- a/app/packet_processor.py +++ b/app/packet_processor.py @@ -15,6 +15,7 @@ are offloaded from the radio to the server. import asyncio import logging import time +from itertools import count from app.decoder import ( DecryptedDirectMessage, @@ -38,6 +39,8 @@ from app.websocket import broadcast_error, broadcast_event logger = logging.getLogger(__name__) +_raw_observation_counter = count(1) + async def _handle_duplicate_message( packet_id: int, @@ -476,6 +479,7 @@ async def process_raw_packet( since the original packet was already processed. """ ts = timestamp or int(time.time()) + observation_id = next(_raw_observation_counter) packet_id, is_new_packet = await RawPacketRepository.create(raw_bytes, ts) raw_hex = raw_bytes.hex() @@ -531,6 +535,7 @@ async def process_raw_packet( # This enables the frontend cracker to see all incoming packets in real-time broadcast_payload = RawPacketBroadcast( id=packet_id, + observation_id=observation_id, timestamp=ts, data=raw_hex, payload_type=payload_type_name, diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 1504944..620aabe 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -107,13 +107,17 @@ Specialized logic is delegated to hooks: - `VisualizerView.tsx` hosts `PacketVisualizer3D.tsx` (desktop split-pane and mobile tabs). - `PacketVisualizer3D` uses persistent Three.js geometries for links/highlights/particles and updates typed-array buffers in-place per frame. - Packet repeat aggregation keys prefer decoder `messageHash` (path-insensitive), with hash fallback for malformed packets. -- Keep packet repeats that add distinct path observations; only drop truly identical duplicate observations. +- Raw packet events carry both: + - `id`: backend storage row identity (payload-level dedup) + - `observation_id`: realtime per-arrival identity (session fidelity) +- Packet feed/visualizer render keys and dedup logic should use `observation_id` (fallback to `id` only for older payloads). ## WebSocket (`useWebSocket.ts`) - Auto reconnect (3s) with cleanup guard on unmount. - Heartbeat ping every 30s. - Event handlers: `health`, `message`, `contact`, `raw_packet`, `message_acked`, `error`, `success`, `pong` (ignored). +- For `raw_packet` events, use `observation_id` as event identity; `id` is a storage reference and may repeat. ## URL Hash Navigation (`utils/urlHash.ts`) diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 10795da..c4b023e 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -49,6 +49,7 @@ const CrackerPanel = lazy(() => import { Sheet, SheetContent, SheetHeader, SheetTitle } from './components/ui/sheet'; import { Toaster, toast } from './components/ui/sonner'; import { getStateKey } from './utils/conversationState'; +import { appendRawPacketUnique } from './utils/rawPacketIdentity'; import { cn } from '@/lib/utils'; import type { Contact, Conversation, HealthStatus, Message, MessagePath, RawPacket } from './types'; @@ -297,16 +298,7 @@ export function App() { }); }, onRawPacket: (packet: RawPacket) => { - setRawPackets((prev) => { - if (prev.some((p) => p.id === packet.id)) { - return prev; - } - const updated = [...prev, packet]; - if (updated.length > MAX_RAW_PACKETS) { - return updated.slice(-MAX_RAW_PACKETS); - } - return updated; - }); + setRawPackets((prev) => appendRawPacketUnique(prev, packet, MAX_RAW_PACKETS)); }, onMessageAcked: (messageId: number, ackCount: number, paths?: MessagePath[]) => { updateMessageAck(messageId, ackCount, paths); diff --git a/frontend/src/components/PacketVisualizer3D.tsx b/frontend/src/components/PacketVisualizer3D.tsx index f35f3bc..5d0f6d0 100644 --- a/frontend/src/components/PacketVisualizer3D.tsx +++ b/frontend/src/components/PacketVisualizer3D.tsx @@ -17,6 +17,7 @@ import { import type { SimulationLinkDatum } from 'd3-force'; import { PayloadType } from '@michaelhart/meshcore-decoder'; import { CONTACT_TYPE_REPEATER, type Contact, type RawPacket, type RadioConfig } from '../types'; +import { getRawPacketObservationKey } from '../utils/rawPacketIdentity'; import { Checkbox } from './ui/checkbox'; import { type NodeType, @@ -145,7 +146,7 @@ function useVisualizerData3D({ const linksRef = useRef>(new Map()); const particlesRef = useRef([]); const simulationRef = useRef | null>(null); - const processedRef = useRef>(new Set()); + const processedRef = useRef>(new Set()); const pendingRef = useRef>(new Map()); const timersRef = useRef>>(new Map()); const trafficPatternsRef = useRef>(new Map()); @@ -592,8 +593,9 @@ function useVisualizerData3D({ const myPrefix = config?.public_key?.slice(0, 12).toLowerCase() || null; for (const packet of packets) { - if (processedRef.current.has(packet.id)) continue; - processedRef.current.add(packet.id); + const observationKey = getRawPacketObservationKey(packet); + if (processedRef.current.has(observationKey)) continue; + processedRef.current.add(observationKey); newProcessed++; if (processedRef.current.size > 1000) { diff --git a/frontend/src/components/RawPacketList.tsx b/frontend/src/components/RawPacketList.tsx index 83ccfdf..3f08e3e 100644 --- a/frontend/src/components/RawPacketList.tsx +++ b/frontend/src/components/RawPacketList.tsx @@ -1,6 +1,7 @@ import { useEffect, useRef, useMemo } from 'react'; import { MeshCoreDecoder, PayloadType, Utils } from '@michaelhart/meshcore-decoder'; import type { RawPacket } from '../types'; +import { getRawPacketObservationKey } from '../utils/rawPacketIdentity'; import { cn } from '@/lib/utils'; interface RawPacketListProps { @@ -207,7 +208,10 @@ export function RawPacketList({ packets }: RawPacketListProps) { return (
{sortedPackets.map(({ packet, decoded }) => ( -
+
{/* Route type badge */} = {}): RawPacket { + return { + id: 1, + timestamp: 1700000000, + data: '010203', + payload_type: 'ACK', + snr: null, + rssi: null, + decrypted: false, + decrypted_info: null, + ...overrides, + }; +} + +describe('getRawPacketObservationKey', () => { + it('uses observation_id when present', () => { + const packet = createPacket({ id: 99, observation_id: 7 }); + expect(getRawPacketObservationKey(packet)).toBe('obs-7'); + }); + + it('falls back to db id when observation_id is missing', () => { + const packet = createPacket({ id: 42 }); + expect(getRawPacketObservationKey(packet)).toBe('db-42'); + }); +}); + +describe('appendRawPacketUnique', () => { + it('keeps path-diverse observations with same db id', () => { + const first = createPacket({ id: 5, observation_id: 100, data: 'aa' }); + const second = createPacket({ id: 5, observation_id: 101, data: 'bb' }); + + const afterFirst = appendRawPacketUnique([], first, 500); + const afterSecond = appendRawPacketUnique(afterFirst, second, 500); + + expect(afterSecond).toHaveLength(2); + expect(afterSecond[0].observation_id).toBe(100); + expect(afterSecond[1].observation_id).toBe(101); + }); + + it('drops exact duplicate observations', () => { + const packet = createPacket({ id: 5, observation_id: 100 }); + + const afterFirst = appendRawPacketUnique([], packet, 500); + const afterSecond = appendRawPacketUnique(afterFirst, packet, 500); + + expect(afterSecond).toHaveLength(1); + }); + + it('dedupes by db id when observation_id is absent', () => { + const first = createPacket({ id: 11, observation_id: undefined }); + const second = createPacket({ id: 11, observation_id: undefined, timestamp: 1700000001 }); + + const afterFirst = appendRawPacketUnique([], first, 500); + const afterSecond = appendRawPacketUnique(afterFirst, second, 500); + + expect(afterSecond).toHaveLength(1); + }); + + it('enforces max packet cap', () => { + const packets = [ + createPacket({ id: 1, observation_id: 1 }), + createPacket({ id: 2, observation_id: 2 }), + createPacket({ id: 3, observation_id: 3 }), + ]; + + let state: RawPacket[] = []; + state = appendRawPacketUnique(state, packets[0], 2); + state = appendRawPacketUnique(state, packets[1], 2); + state = appendRawPacketUnique(state, packets[2], 2); + + expect(state).toHaveLength(2); + expect(state[0].observation_id).toBe(2); + expect(state[1].observation_id).toBe(3); + }); +}); diff --git a/frontend/src/types.ts b/frontend/src/types.ts index ec52d4b..aa91aa5 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -98,6 +98,8 @@ export interface Conversation { export interface RawPacket { id: number; + /** Per-observation WS identity (unique per RF arrival, may be absent in older payloads) */ + observation_id?: number; timestamp: number; data: string; // hex payload_type: string; diff --git a/frontend/src/utils/rawPacketIdentity.ts b/frontend/src/utils/rawPacketIdentity.ts new file mode 100644 index 0000000..5b08a68 --- /dev/null +++ b/frontend/src/utils/rawPacketIdentity.ts @@ -0,0 +1,31 @@ +import type { RawPacket } from '../types'; + +/** + * Distinguish real-time RF observations from storage identity. + * observation_id is emitted per WS event; id is the DB row identity fallback. + */ +export function getRawPacketObservationKey( + packet: Pick +): string { + if (packet.observation_id !== undefined && packet.observation_id !== null) { + return `obs-${packet.observation_id}`; + } + return `db-${packet.id}`; +} + +export function appendRawPacketUnique( + prev: RawPacket[], + packet: RawPacket, + maxPackets: number +): RawPacket[] { + const packetKey = getRawPacketObservationKey(packet); + if (prev.some((p) => getRawPacketObservationKey(p) === packetKey)) { + return prev; + } + + const updated = [...prev, packet]; + if (updated.length > maxPackets) { + return updated.slice(-maxPackets); + } + return updated; +} diff --git a/tests/test_packet_pipeline.py b/tests/test_packet_pipeline.py index 95cd013..c2c90b9 100644 --- a/tests/test_packet_pipeline.py +++ b/tests/test_packet_pipeline.py @@ -1638,6 +1638,43 @@ class TestProcessRawPacketIntegration: raw_broadcasts = [b for b in broadcasts if b["type"] == "raw_packet"] assert len(raw_broadcasts) == 1 assert raw_broadcasts[0]["data"]["payload_type"] == "ACK" + assert isinstance(raw_broadcasts[0]["data"]["observation_id"], int) + assert raw_broadcasts[0]["data"]["observation_id"] > 0 + + @pytest.mark.asyncio + async def test_duplicate_payload_has_same_packet_id_but_unique_observation_ids( + self, test_db, captured_broadcasts + ): + """Path-diverse duplicates share storage id but retain unique observation ids.""" + from app.packet_processor import process_raw_packet + + broadcasts, mock_broadcast = captured_broadcasts + + ack_info = PacketInfo( + route_type=1, + payload_type=PayloadType.ACK, + payload_version=0, + path_length=0, + path=b"", + payload=b"\x00" * 10, + ) + + # Same payload bytes, different path bytes in packet header/path region. + raw_1 = bytes([0x01, 0x01, 0xAA]) + b"PAYLOAD-1234" + raw_2 = bytes([0x01, 0x02, 0xBB, 0xCC]) + b"PAYLOAD-1234" + + with patch("app.packet_processor.broadcast_event", mock_broadcast): + with patch("app.packet_processor.parse_packet", return_value=ack_info): + await process_raw_packet(raw_1, timestamp=5001) + await process_raw_packet(raw_2, timestamp=5002) + + raw_broadcasts = [b for b in broadcasts if b["type"] == "raw_packet"] + assert len(raw_broadcasts) == 2 + + first = raw_broadcasts[0]["data"] + second = raw_broadcasts[1]["data"] + assert first["id"] == second["id"] # Same DB packet row + assert first["observation_id"] != second["observation_id"] # Distinct RF observations @pytest.mark.asyncio async def test_result_structure(self, test_db, captured_broadcasts):