mirror of
https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git
synced 2026-07-31 14:02:26 +02:00
Fix dedupe for frontend raw packet delivery
This commit is contained in:
@@ -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
|
||||
|
||||
```
|
||||
|
||||
+5
-2
@@ -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
|
||||
|
||||
|
||||
@@ -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)")
|
||||
|
||||
@@ -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,
|
||||
|
||||
+5
-1
@@ -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`)
|
||||
|
||||
|
||||
+2
-10
@@ -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);
|
||||
|
||||
@@ -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<Map<string, GraphLink>>(new Map());
|
||||
const particlesRef = useRef<Particle[]>([]);
|
||||
const simulationRef = useRef<Simulation3D<GraphNode, GraphLink> | null>(null);
|
||||
const processedRef = useRef<Set<number>>(new Set());
|
||||
const processedRef = useRef<Set<string>>(new Set());
|
||||
const pendingRef = useRef<Map<string, PendingPacket>>(new Map());
|
||||
const timersRef = useRef<Map<string, ReturnType<typeof setTimeout>>>(new Map());
|
||||
const trafficPatternsRef = useRef<Map<string, RepeaterTrafficData>>(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) {
|
||||
|
||||
@@ -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 (
|
||||
<div className="h-full overflow-y-auto p-4 flex flex-col gap-2" ref={listRef}>
|
||||
{sortedPackets.map(({ packet, decoded }) => (
|
||||
<div key={packet.id} className="py-2 px-3 bg-card rounded-md border border-border/50">
|
||||
<div
|
||||
key={getRawPacketObservationKey(packet)}
|
||||
className="py-2 px-3 bg-card rounded-md border border-border/50"
|
||||
>
|
||||
<div className="flex items-center gap-2">
|
||||
{/* Route type badge */}
|
||||
<span
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import type { RawPacket } from '../types';
|
||||
import { appendRawPacketUnique, getRawPacketObservationKey } from '../utils/rawPacketIdentity';
|
||||
|
||||
function createPacket(overrides: Partial<RawPacket> = {}): 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);
|
||||
});
|
||||
});
|
||||
@@ -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;
|
||||
|
||||
@@ -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<RawPacket, 'id' | 'observation_id'>
|
||||
): 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;
|
||||
}
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user