From 189751c561b7a1b60f15f5e955cefdea119502cd Mon Sep 17 00:00:00 2001 From: Strycher Date: Sat, 18 Jul 2026 01:35:34 -0400 Subject: [PATCH] feat(#285): ingest RX time logging + persisted rxTime (arrival vs claimed) Every received DM/channel message now logs arrival wall-clock vs the sender-claimed payload timestamp (delta + TIME ANOMALY marker beyond 24h), routed directly to AppDebugLogService so the rotating on-disk trail is unconditional. Models persist a nullable rxTime set at every live ingest construction site; legacy stored records stay null. Diagnostic child of #280. Co-Authored-By: Claude Fable 5 --- lib/connector/meshcore_connector.dart | 52 ++++++++++++++++ lib/helpers/time_anomaly.dart | 46 ++++++++++++++ lib/models/channel_message.dart | 9 +++ lib/models/message.dart | 9 +++ lib/storage/channel_message_store.dart | 4 ++ lib/storage/message_store.dart | 4 ++ test/helpers/time_anomaly_test.dart | 83 ++++++++++++++++++++++++++ test/models/rx_time_test.dart | 58 ++++++++++++++++++ 8 files changed, 265 insertions(+) create mode 100644 lib/helpers/time_anomaly.dart create mode 100644 test/helpers/time_anomaly_test.dart create mode 100644 test/models/rx_time_test.dart diff --git a/lib/connector/meshcore_connector.dart b/lib/connector/meshcore_connector.dart index c06460e..76a449d 100644 --- a/lib/connector/meshcore_connector.dart +++ b/lib/connector/meshcore_connector.dart @@ -18,6 +18,7 @@ import '../models/offband_gps_status.dart'; import '../models/path_selection.dart'; import '../models/translation_support.dart'; import '../helpers/reaction_helper.dart'; +import '../helpers/time_anomaly.dart'; import '../helpers/cyr2lat.dart'; import '../helpers/smaz.dart'; import '../helpers/cayenne_lpp.dart'; @@ -5171,6 +5172,38 @@ class MeshCoreConnector extends ChangeNotifier { return false; } + /// Ingest RX log (#285): one line per received message recording local + /// arrival vs the sender-claimed timestamp, so time anomalies (#280) are + /// provable from the app debug log / rotating log files after the fact. + void _logMessageRx({ + required String kind, + required String sender, + required DateTime claimed, + required DateTime arrival, + required int textBytes, + Uint8List? pathBytes, + String? channel, + }) { + final path = (pathBytes == null || pathBytes.isEmpty) + ? 'none' + : pathBytes + .map((b) => b.toRadixString(16).padLeft(2, '0').toUpperCase()) + .join(','); + final line = + '$kind RX${channel != null ? ' ch=$channel' : ''} sender=$sender ' + 'len=$textBytes arrival=${arrival.toIso8601String()} ' + 'claimed=${claimed.toIso8601String()} ' + 'delta=${formatRxTimeDelta(arrival, claimed)} path=$path'; + // Direct service call (not appLogger): the file trail must not depend on + // the in-app debug-log toggle — the service always writes to disk and the + // toggle only gates the in-app ring buffer. + if (isRxTimeAnomaly(arrival, claimed)) { + _appDebugLogService?.warn('$line ** TIME ANOMALY **', tag: 'Ingest'); + } else { + _appDebugLogService?.info(line, tag: 'Ingest', noNotify: true); + } + } + Future _handleIncomingMessage(Uint8List frame) async { if (_selfPublicKey == null) return; @@ -5206,6 +5239,14 @@ class MeshCoreConnector extends ChangeNotifier { if (message != null) { if (!message.isOutgoing) { _lastContactMsgRxTime = DateTime.now(); + _logMessageRx( + kind: 'DM', + sender: message.senderKeyHex.substring(0, 12), + claimed: message.timestamp, + arrival: message.rxTime ?? _lastContactMsgRxTime, + textBytes: utf8.encode(message.text).length, + pathBytes: message.pathBytes, + ); } // Ignore messages from self (device hearing its own broadcast) // BUT allow repeated messages (pathLength indicates it went through repeater) @@ -5412,6 +5453,7 @@ class MeshCoreConnector extends ChangeNotifier { pathLength: pathLength == 0xFF ? 0 : pathLength, pathBytes: Uint8List(0), fourByteRoomContactKey: roomAuthorPrefix, + rxTime: DateTime.now(), ); } catch (e) { appLogger.warn('Error parsing contact direct message: $e'); @@ -5671,6 +5713,15 @@ class MeshCoreConnector extends ChangeNotifier { return; } _lastChannelMsgRxTime = DateTime.now(); + _logMessageRx( + kind: 'CHAN', + channel: '${parsed.channelIndex}', + sender: parsed.senderName, + claimed: parsed.timestamp, + arrival: parsed.rxTime ?? _lastChannelMsgRxTime, + textBytes: utf8.encode(parsed.text).length, + pathBytes: parsed.pathBytes, + ); final contentHash = _computeContentHash( parsed.channelIndex!, parsed.timestamp.millisecondsSinceEpoch ~/ 1000, @@ -5770,6 +5821,7 @@ class MeshCoreConnector extends ChangeNotifier { pathLength: packet.isFlood ? packet.hopCount : 0, pathBytes: packet.pathBytes, channelIndex: channel.index, + rxTime: DateTime.now(), packetHash: pktHash, ); diff --git a/lib/helpers/time_anomaly.dart b/lib/helpers/time_anomaly.dart new file mode 100644 index 0000000..9a77361 --- /dev/null +++ b/lib/helpers/time_anomaly.dart @@ -0,0 +1,46 @@ +/// Ingest time-anomaly detection (#285, child of #280). +/// +/// A received message carries the SENDER's claimed timestamp in its payload; +/// nothing guarantees that clock is sane. These helpers compare the claimed +/// time against the local arrival time so the ingest log can prove, after the +/// fact, whether an odd date came in on the wire. +library; + +/// Claimed-vs-arrival gap beyond which a received message is logged as a +/// time anomaly. Generous enough for mesh store-and-forward delays and +/// timezone-free radio clocks; months-off RTCs blow well past it. +const Duration rxTimeAnomalyThreshold = Duration(hours: 24); + +/// True when the sender-claimed [claimed] time is further than [threshold] +/// from the local [arrival] time in either direction. +bool isRxTimeAnomaly( + DateTime arrival, + DateTime claimed, { + Duration threshold = rxTimeAnomalyThreshold, +}) { + return claimed.difference(arrival).abs() > threshold; +} + +/// Compact human delta of claimed relative to arrival, e.g. `-64d5h` (claimed +/// 64 days in the past) or `+3m` (claimed 3 minutes ahead). `0s` when equal. +String formatRxTimeDelta(DateTime arrival, DateTime claimed) { + final delta = claimed.difference(arrival); + if (delta == Duration.zero) return '0s'; + final sign = delta.isNegative ? '-' : '+'; + var d = delta.abs(); + final parts = []; + if (d.inDays > 0) { + parts.add('${d.inDays}d'); + d -= Duration(days: d.inDays); + } + if (d.inHours > 0) { + parts.add('${d.inHours}h'); + d -= Duration(hours: d.inHours); + } + if (parts.length < 2 && d.inMinutes > 0) { + parts.add('${d.inMinutes}m'); + d -= Duration(minutes: d.inMinutes); + } + if (parts.isEmpty) parts.add('${d.inSeconds}s'); + return '$sign${parts.take(2).join()}'; +} diff --git a/lib/models/channel_message.dart b/lib/models/channel_message.dart index 7c95b89..886bab3 100644 --- a/lib/models/channel_message.dart +++ b/lib/models/channel_message.dart @@ -51,6 +51,11 @@ class ChannelMessage { final String? replyToText; final Map reactions; + /// Local wall-clock time this message's frame arrived, set at ingest. + /// Null for outgoing messages and records stored before #285 — never + /// fabricated. [timestamp] is the SENDER's claimed time; this is ours. + final DateTime? rxTime; + ChannelMessage({ this.senderKey, required this.senderName, @@ -75,6 +80,7 @@ class ChannelMessage { this.replyToSenderName, this.replyToText, Map? reactions, + this.rxTime, }) : messageId = messageId ?? '${timestamp.millisecondsSinceEpoch}_${senderName.hashCode}_${text.hashCode}', @@ -123,6 +129,7 @@ class ChannelMessage { MessageTranslationStatus? translationStatus, Object? translationModelId = _unset, Map? reactions, + DateTime? rxTime, }) { return ChannelMessage( senderKey: senderKey, @@ -156,6 +163,7 @@ class ChannelMessage { replyToSenderName: replyToSenderName ?? this.replyToSenderName, replyToText: replyToText ?? this.replyToText, reactions: reactions ?? this.reactions, + rxTime: rxTime ?? this.rxTime, ); } @@ -229,6 +237,7 @@ class ChannelMessage { pathLength: pathLen, pathBytes: pathBytes, channelIndex: channelIdx, + rxTime: DateTime.now(), ); } catch (e) { appLogger.error('Error parsing channel message frame: $e'); diff --git a/lib/models/message.dart b/lib/models/message.dart index e139561..e91fb56 100644 --- a/lib/models/message.dart +++ b/lib/models/message.dart @@ -34,6 +34,11 @@ class Message { final Map reactionStatuses; final Uint8List fourByteRoomContactKey; + /// Local wall-clock time this message's frame arrived, set at ingest. + /// Null for outgoing messages and records stored before #285 — never + /// fabricated. [timestamp] is the SENDER's claimed time; this is ours. + final DateTime? rxTime; + Message({ required this.senderKey, required this.text, @@ -58,6 +63,7 @@ class Message { Uint8List? fourByteRoomContactKey, Map? reactions, Map? reactionStatuses, + this.rxTime, }) : messageId = messageId ?? '${timestamp.millisecondsSinceEpoch}_${pubKeyToHex(senderKey)}_${text.hashCode}', @@ -87,6 +93,7 @@ class Message { Map? reactions, Map? reactionStatuses, Uint8List? fourByteRoomContactKey, + DateTime? rxTime, }) { return Message( senderKey: senderKey, @@ -121,6 +128,7 @@ class Message { reactionStatuses: reactionStatuses ?? this.reactionStatuses, fourByteRoomContactKey: fourByteRoomContactKey ?? this.fourByteRoomContactKey, + rxTime: rxTime ?? this.rxTime, ); } @@ -149,6 +157,7 @@ class Message { isCli: false, status: MessageStatus.delivered, pathBytes: Uint8List(0), + rxTime: DateTime.now(), ); } catch (e) { return null; diff --git a/lib/storage/channel_message_store.dart b/lib/storage/channel_message_store.dart index ee60050..dfef8c3 100644 --- a/lib/storage/channel_message_store.dart +++ b/lib/storage/channel_message_store.dart @@ -136,6 +136,7 @@ class ChannelMessageStore { 'timestamp': msg.timestamp.millisecondsSinceEpoch, 'isOutgoing': msg.isOutgoing, 'status': msg.status.index, + 'rxTime': msg.rxTime?.millisecondsSinceEpoch, 'channelIndex': msg.channelIndex, 'repeatCount': msg.repeatCount, 'pathLength': msg.pathLength, @@ -171,6 +172,9 @@ class ChannelMessageStore { timestamp: DateTime.fromMillisecondsSinceEpoch(json['timestamp'] as int), isOutgoing: json['isOutgoing'] as bool, status: ChannelMessageStatus.values[json['status'] as int], + rxTime: json['rxTime'] != null + ? DateTime.fromMillisecondsSinceEpoch(json['rxTime'] as int) + : null, repeatCount: (json['repeatCount'] as int?) ?? 0, pathLength: json['pathLength'] as int?, pathBytes: json['pathBytes'] != null diff --git a/lib/storage/message_store.dart b/lib/storage/message_store.dart index 3d23768..ccaf9ea 100644 --- a/lib/storage/message_store.dart +++ b/lib/storage/message_store.dart @@ -104,6 +104,7 @@ class MessageStore { (key, value) => MapEntry(key, value.index), ), 'fourByteRoomContactKey': base64Encode(msg.fourByteRoomContactKey), + 'rxTime': msg.rxTime?.millisecondsSinceEpoch, }; } @@ -157,6 +158,9 @@ class MessageStore { base64Decode(json['fourByteRoomContactKey'] as String), ) : null, + rxTime: json['rxTime'] != null + ? DateTime.fromMillisecondsSinceEpoch(json['rxTime'] as int) + : null, ); } } diff --git a/test/helpers/time_anomaly_test.dart b/test/helpers/time_anomaly_test.dart new file mode 100644 index 0000000..f029bd2 --- /dev/null +++ b/test/helpers/time_anomaly_test.dart @@ -0,0 +1,83 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:meshcore_open/helpers/time_anomaly.dart'; + +void main() { + final arrival = DateTime(2026, 7, 18, 22, 47, 3); + + group('isRxTimeAnomaly (#285)', () { + test('equal times are not anomalous', () { + expect(isRxTimeAnomaly(arrival, arrival), isFalse); + }); + + test('within 24h either direction is not anomalous', () { + expect( + isRxTimeAnomaly(arrival, arrival.subtract(const Duration(hours: 23))), + isFalse, + ); + expect( + isRxTimeAnomaly(arrival, arrival.add(const Duration(hours: 23))), + isFalse, + ); + }); + + test('beyond 24h either direction is anomalous', () { + expect( + isRxTimeAnomaly(arrival, arrival.subtract(const Duration(hours: 25))), + isTrue, + ); + expect( + isRxTimeAnomaly(arrival, arrival.add(const Duration(hours: 25))), + isTrue, + ); + }); + + test('the #280 case (claimed 64 days in the past) is anomalous', () { + final claimed = arrival.subtract(const Duration(days: 64, hours: 5)); + expect(isRxTimeAnomaly(arrival, claimed), isTrue); + }); + + test('custom threshold is honored', () { + expect( + isRxTimeAnomaly( + arrival, + arrival.subtract(const Duration(minutes: 10)), + threshold: const Duration(minutes: 5), + ), + isTrue, + ); + }); + }); + + group('formatRxTimeDelta (#285)', () { + test('zero delta', () { + expect(formatRxTimeDelta(arrival, arrival), '0s'); + }); + + test('claimed in the past is negative, days+hours', () { + final claimed = arrival.subtract(const Duration(days: 64, hours: 5)); + expect(formatRxTimeDelta(arrival, claimed), '-64d5h'); + }); + + test('claimed ahead is positive, minutes only', () { + final claimed = arrival.add(const Duration(minutes: 3)); + expect(formatRxTimeDelta(arrival, claimed), '+3m'); + }); + + test('sub-minute deltas fall back to seconds', () { + final claimed = arrival.subtract(const Duration(seconds: 45)); + expect(formatRxTimeDelta(arrival, claimed), '-45s'); + }); + + test('exact whole days omit smaller units', () { + final claimed = arrival.subtract(const Duration(days: 2)); + expect(formatRxTimeDelta(arrival, claimed), '-2d'); + }); + + test('caps at two units (no minute tail after days+hours)', () { + final claimed = arrival.subtract( + const Duration(days: 1, hours: 2, minutes: 30), + ); + expect(formatRxTimeDelta(arrival, claimed), '-1d2h'); + }); + }); +} diff --git a/test/models/rx_time_test.dart b/test/models/rx_time_test.dart new file mode 100644 index 0000000..d975e00 --- /dev/null +++ b/test/models/rx_time_test.dart @@ -0,0 +1,58 @@ +import 'dart:typed_data'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:meshcore_open/models/channel_message.dart'; +import 'package:meshcore_open/models/message.dart'; + +void main() { + final rx = DateTime(2026, 7, 18, 22, 47, 3); + final claimed = DateTime(2026, 5, 15, 17, 38, 12); + + group('Message.rxTime (#285)', () { + Message build({DateTime? rxTime}) => Message( + senderKey: Uint8List(32), + text: 'hello', + timestamp: claimed, + isOutgoing: false, + rxTime: rxTime, + ); + + test('defaults to null (never fabricated)', () { + expect(build().rxTime, isNull); + }); + + test('copyWith sets rxTime', () { + expect(build().copyWith(rxTime: rx).rxTime, rx); + }); + + test('copyWith of unrelated fields preserves rxTime', () { + final msg = build(rxTime: rx).copyWith(status: MessageStatus.delivered); + expect(msg.rxTime, rx); + expect(msg.timestamp, claimed); + }); + }); + + group('ChannelMessage.rxTime (#285)', () { + ChannelMessage build({DateTime? rxTime}) => ChannelMessage( + senderName: 'kd8zsp_tdeck', + text: 'hello', + timestamp: claimed, + isOutgoing: false, + rxTime: rxTime, + ); + + test('defaults to null (never fabricated)', () { + expect(build().rxTime, isNull); + }); + + test('copyWith sets rxTime', () { + expect(build().copyWith(rxTime: rx).rxTime, rx); + }); + + test('copyWith(packetHash:) — the live ingest path — preserves rxTime', () { + final msg = build(rxTime: rx).copyWith(packetHash: 'abc123'); + expect(msg.rxTime, rx); + expect(msg.timestamp, claimed); + }); + }); +}