Msg Retry fixes, channel message fixes. Notification fixes. Make more desktop friendly. Enhance retry algo. Fix predicted location clustering add retries to reactions and fix the reactions in private DMS centralize and cleanup code in var areas

2026-07-27 14:06:38 +10:00 · 2026-03-20 01:54:31 -07:00
parent 53caec3e14
commit 4962a48e64
61 changed files with 4509 additions and 900 deletions
@@ -9,6 +9,8 @@ class PathHistoryService extends ChangeNotifier {
  final Map<String, ContactPathHistory> _cache = {};
  final Map<String, int> _autoRotationIndex = {};
  final Map<String, _FloodStats> _floodStats = {};
+  final Set<String> _pendingLoads = {};
+  final Map<String, List<_DeferredPathRecord>> _deferredRecords = {};

  // LRU cache eviction tracking
  static const int _maxCachedContacts = 50;
@@ -18,7 +20,6 @@ class PathHistoryService extends ChangeNotifier {

  int _version = 0;
  int get version => _version;
-  static const int _autoRotationTopCount = 3;

  PathHistoryService(this._storage);

@@ -26,17 +27,21 @@ class PathHistoryService extends ChangeNotifier {
    // Load cached path histories on startup if needed
  }

-  void handlePathUpdated(Contact contact) {
-    if (contact.pathLength < 0) return;
-
+  void handlePathUpdated(Contact contact, {double initialWeight = 1.0}) {
+    if (contact.pathLength < 0 && contact.path.isEmpty) return;
+    final hopCount = contact.pathLength < 0
+        ? contact.path.length
+        : contact.pathLength;
    _addPathRecord(
      contactPubKeyHex: contact.publicKeyHex,
-      hopCount: contact.pathLength,
+      hopCount: hopCount,
      tripTimeMs: 0,
      wasFloodDiscovery: true,
      pathBytes: contact.path,
      successCount: 0,
      failureCount: 0,
+      routeWeight: initialWeight,
+      timestamp: null,
    );
  }

@@ -54,6 +59,44 @@ class PathHistoryService extends ChangeNotifier {
      pathBytes: selection.pathBytes,
      successCount: 0,
      failureCount: 0,
+      timestamp: null,
+    );
+  }
+
+  /// When a flood message is delivered, credit the contact's current device
+  /// path so that the route the ACK traveled back through gets a weight boost.
+  void recordFloodPathAttribution({
+    required String contactPubKeyHex,
+    required List<int> pathBytes,
+    required int hopCount,
+    int? tripTimeMs,
+    double successIncrement = 0.5,
+    double maxWeight = 5.0,
+  }) {
+    if (pathBytes.isEmpty || hopCount < 0) return;
+
+    final existing = _findPathRecord(contactPubKeyHex, pathBytes);
+    final successCount = (existing?.successCount ?? 0) + 1;
+    final failureCount = existing?.failureCount ?? 0;
+
+    final currentWeight = existing?.routeWeight ?? 1.0;
+    final newWeight = (currentWeight + successIncrement).clamp(0.0, maxWeight);
+
+    debugPrint(
+      'Flood path attribution: crediting path [${pathBytes.map((b) => b.toRadixString(16).padLeft(2, '0')).join(',')}] '
+      'for $contactPubKeyHex (weight $currentWeight → $newWeight)',
+    );
+
+    _addPathRecord(
+      contactPubKeyHex: contactPubKeyHex,
+      hopCount: hopCount,
+      tripTimeMs: tripTimeMs ?? existing?.tripTimeMs ?? 0,
+      wasFloodDiscovery: true,
+      pathBytes: pathBytes,
+      successCount: successCount,
+      failureCount: failureCount,
+      routeWeight: newWeight,
+      timestamp: DateTime.now(),
    );
  }

@@ -62,6 +105,9 @@ class PathHistoryService extends ChangeNotifier {
    PathSelection selection, {
    required bool success,
    int? tripTimeMs,
+    double successIncrement = 0.5,
+    double failureDecrement = 0.5,
+    double maxWeight = 5.0,
  }) {
    if (selection.useFlood) {
      final stats = _floodStats.putIfAbsent(
@@ -82,6 +128,18 @@ class PathHistoryService extends ChangeNotifier {
    final successCount = (existing?.successCount ?? 0) + (success ? 1 : 0);
    final failureCount = (existing?.failureCount ?? 0) + (success ? 0 : 1);

+    final currentWeight = existing?.routeWeight ?? 1.0;
+    double newWeight;
+    if (success) {
+      newWeight = (currentWeight + successIncrement).clamp(0.0, maxWeight);
+    } else {
+      newWeight = currentWeight - failureDecrement;
+      if (newWeight <= 0) {
+        removePathRecord(contactPubKeyHex, selection.pathBytes);
+        return;
+      }
+    }
+
    _addPathRecord(
      contactPubKeyHex: contactPubKeyHex,
      hopCount: selection.hopCount,
@@ -90,37 +148,68 @@ class PathHistoryService extends ChangeNotifier {
      pathBytes: selection.pathBytes,
      successCount: successCount,
      failureCount: failureCount,
+      routeWeight: newWeight,
+      timestamp: success ? DateTime.now() : existing?.timestamp,
    );
  }

-  PathSelection getNextAutoPathSelection(String contactPubKeyHex) {
-    final ranked = _getRankedPaths(
-      contactPubKeyHex,
-    ).take(_autoRotationTopCount).toList();
+  PathSelection selectPathForAttempt(
+    String contactPubKeyHex, {
+    required int attemptIndex,
+    required int maxRetries,
+    List<PathSelection> recentSelections = const [],
+  }) {
+    if (maxRetries <= 0 || attemptIndex >= maxRetries - 1) {
+      return const PathSelection(pathBytes: [], hopCount: -1, useFlood: true);
+    }
+
+    final ranked = _getRankedPaths(contactPubKeyHex);
    if (ranked.isEmpty) {
      return const PathSelection(pathBytes: [], hopCount: -1, useFlood: true);
    }

    _trackAccess(contactPubKeyHex);

-    final selections =
-        ranked
-            .map(
-              (path) => PathSelection(
-                pathBytes: path.pathBytes,
-                hopCount: path.hopCount,
-                useFlood: false,
-              ),
-            )
-            .toList()
-          ..add(
-            const PathSelection(pathBytes: [], hopCount: -1, useFlood: true),
-          );
+    final recentPaths = recentSelections
+        .where((selection) => !selection.useFlood)
+        .map((selection) => selection.pathBytes)
+        .toList();
+    final candidates = recentPaths.isEmpty
+        ? ranked
+        : ranked
+              .where(
+                (path) => !recentPaths.any(
+                  (recentPath) => _pathsEqual(path.pathBytes, recentPath),
+                ),
+              )
+              .toList();
+    final selected = candidates.isNotEmpty
+        ? (recentPaths.isEmpty
+              ? _selectRotatedCandidate(contactPubKeyHex, candidates)
+              : candidates.first)
+        : ranked.first;
+
+    return PathSelection(
+      pathBytes: selected.pathBytes,
+      hopCount: selected.hopCount,
+      useFlood: false,
+    );
+  }
+
+  PathRecord _selectRotatedCandidate(
+    String contactPubKeyHex,
+    List<PathRecord> candidates,
+  ) {
+    if (candidates.length <= 1) {
+      _autoRotationIndex[contactPubKeyHex] = 0;
+      return candidates.first;
+    }

    final currentIndex = _autoRotationIndex[contactPubKeyHex] ?? 0;
-    final selection = selections[currentIndex % selections.length];
-    _autoRotationIndex[contactPubKeyHex] = currentIndex + 1;
-    return selection;
+    final selectedIndex = currentIndex % candidates.length;
+    _autoRotationIndex[contactPubKeyHex] =
+        (selectedIndex + 1) % candidates.length;
+    return candidates[selectedIndex];
  }

  void _addPathRecord({
@@ -131,37 +220,68 @@ class PathHistoryService extends ChangeNotifier {
    required List<int> pathBytes,
    required int successCount,
    required int failureCount,
+    double routeWeight = 1.0,
+    DateTime? timestamp,
  }) {
    var history = _cache[contactPubKeyHex];

    if (history == null) {
+      // If a load is already in progress, defer this record
+      if (_pendingLoads.contains(contactPubKeyHex)) {
+        _deferredRecords.putIfAbsent(contactPubKeyHex, () => []);
+        _deferredRecords[contactPubKeyHex]!.add(
+          _DeferredPathRecord(
+            hopCount: hopCount,
+            tripTimeMs: tripTimeMs,
+            wasFloodDiscovery: wasFloodDiscovery,
+            pathBytes: pathBytes,
+            successCount: successCount,
+            failureCount: failureCount,
+            routeWeight: routeWeight,
+            timestamp: timestamp,
+          ),
+        );
+        return;
+      }
+
+      _pendingLoads.add(contactPubKeyHex);
      _loadHistoryFromStorage(contactPubKeyHex).then((loaded) {
-        if (loaded != null) {
-          _cache[contactPubKeyHex] = loaded;
-          _addPathRecordInternal(
-            contactPubKeyHex,
-            hopCount,
-            tripTimeMs,
-            wasFloodDiscovery,
-            pathBytes,
-            successCount,
-            failureCount,
-          );
-        } else {
-          _cache[contactPubKeyHex] = ContactPathHistory(
-            contactPubKeyHex: contactPubKeyHex,
-            recentPaths: [],
-          );
-          _addPathRecordInternal(
-            contactPubKeyHex,
-            hopCount,
-            tripTimeMs,
-            wasFloodDiscovery,
-            pathBytes,
-            successCount,
-            failureCount,
-          );
+        _cache[contactPubKeyHex] =
+            loaded ??
+            ContactPathHistory(
+              contactPubKeyHex: contactPubKeyHex,
+              recentPaths: [],
+            );
+        _addPathRecordInternal(
+          contactPubKeyHex,
+          hopCount,
+          tripTimeMs,
+          wasFloodDiscovery,
+          pathBytes,
+          successCount,
+          failureCount,
+          routeWeight,
+          timestamp,
+        );
+
+        // Apply any deferred records
+        final deferred = _deferredRecords.remove(contactPubKeyHex);
+        if (deferred != null) {
+          for (final record in deferred) {
+            _addPathRecordInternal(
+              contactPubKeyHex,
+              record.hopCount,
+              record.tripTimeMs,
+              record.wasFloodDiscovery,
+              record.pathBytes,
+              record.successCount,
+              record.failureCount,
+              record.routeWeight,
+              record.timestamp,
+            );
+          }
        }
+        _pendingLoads.remove(contactPubKeyHex);
      });
      return;
    }
@@ -174,6 +294,8 @@ class PathHistoryService extends ChangeNotifier {
      pathBytes,
      successCount,
      failureCount,
+      routeWeight,
+      timestamp,
    );
  }

@@ -185,6 +307,8 @@ class PathHistoryService extends ChangeNotifier {
    List<int> pathBytes,
    int successCount,
    int failureCount,
+    double routeWeight,
+    DateTime? timestamp,
  ) {
    var history = _cache[contactPubKeyHex];
    if (history == null) return;
@@ -198,16 +322,18 @@ class PathHistoryService extends ChangeNotifier {
        tripTimeMs = existing.tripTimeMs;
      }
      wasFloodDiscovery = existing.wasFloodDiscovery || wasFloodDiscovery;
+      timestamp ??= existing.timestamp;
    }

    final newRecord = PathRecord(
      hopCount: hopCount,
      tripTimeMs: tripTimeMs,
-      timestamp: DateTime.now(),
+      timestamp: timestamp,
      wasFloodDiscovery: wasFloodDiscovery,
      pathBytes: pathBytes,
      successCount: successCount,
      failureCount: failureCount,
+      routeWeight: routeWeight,
    );

    final updatedPaths = List<PathRecord>.from(history.recentPaths);
@@ -275,6 +401,23 @@ class PathHistoryService extends ChangeNotifier {
    return history?.mostRecent;
  }

+  ({
+    int successCount,
+    int failureCount,
+    int lastTripTimeMs,
+    DateTime? lastUsed,
+  })?
+  getFloodStats(String contactPubKeyHex) {
+    final stats = _floodStats[contactPubKeyHex];
+    if (stats == null) return null;
+    return (
+      successCount: stats.successCount,
+      failureCount: stats.failureCount,
+      lastTripTimeMs: stats.lastTripTimeMs,
+      lastUsed: stats.lastUsed,
+    );
+  }
+
  Future<void> clearPathHistory(String contactPubKeyHex) async {
    _cache.remove(contactPubKeyHex);
    _cacheAccessOrder.remove(contactPubKeyHex);
@@ -322,26 +465,81 @@ class PathHistoryService extends ChangeNotifier {

    final ranked = List<PathRecord>.from(history.recentPaths)
      ..removeWhere((p) => p.pathBytes.isEmpty);
+    final fastestTripMs = _getFastestKnownTripMs(ranked);
+    final highestRouteWeight = _getHighestKnownRouteWeight(ranked);

    ranked.sort((a, b) {
-      final aRate =
-          (a.successCount + 1) / (a.successCount + a.failureCount + 2);
-      final bRate =
-          (b.successCount + 1) / (b.successCount + b.failureCount + 2);
-      if (aRate != bRate) return bRate.compareTo(aRate);
-      if (a.successCount != b.successCount) {
-        return b.successCount.compareTo(a.successCount);
+      final scoreCompare = _scorePathRecord(
+        b,
+        fastestTripMs: fastestTripMs,
+        highestRouteWeight: highestRouteWeight,
+      ).compareTo(
+        _scorePathRecord(
+          a,
+          fastestTripMs: fastestTripMs,
+          highestRouteWeight: highestRouteWeight,
+        ),
+      );
+      if (scoreCompare != 0) {
+        return scoreCompare;
+      }
+      if (a.routeWeight != b.routeWeight) {
+        return b.routeWeight.compareTo(a.routeWeight);
      }
-
      final aTrip = a.tripTimeMs == 0 ? 999999 : a.tripTimeMs;
      final bTrip = b.tripTimeMs == 0 ? 999999 : b.tripTimeMs;
      if (aTrip != bTrip) return aTrip.compareTo(bTrip);
-      return b.timestamp.compareTo(a.timestamp);
+      final aTime = a.timestamp ?? DateTime.fromMillisecondsSinceEpoch(0);
+      final bTime = b.timestamp ?? DateTime.fromMillisecondsSinceEpoch(0);
+      return bTime.compareTo(aTime);
    });

    return ranked;
  }

+  int? _getFastestKnownTripMs(List<PathRecord> paths) {
+    final knownTrips = paths
+        .where((path) => path.tripTimeMs > 0)
+        .map((path) => path.tripTimeMs)
+        .toList();
+    if (knownTrips.isEmpty) return null;
+    return knownTrips.reduce((a, b) => a < b ? a : b);
+  }
+
+  double _getHighestKnownRouteWeight(List<PathRecord> paths) {
+    if (paths.isEmpty) return 1.0;
+    final highestWeight = paths
+        .map((path) => path.routeWeight)
+        .reduce((a, b) => a > b ? a : b);
+    return highestWeight <= 0 ? 1.0 : highestWeight;
+  }
+
+  double _scorePathRecord(
+    PathRecord path, {
+    required int? fastestTripMs,
+    required double highestRouteWeight,
+  }) {
+    final totalAttempts = path.successCount + path.failureCount;
+    final reliability = (path.successCount + 1) / (totalAttempts + 2);
+    final latency = fastestTripMs == null || path.tripTimeMs <= 0
+        ? 0.6
+        : (fastestTripMs / path.tripTimeMs).clamp(0.0, 1.0);
+    final freshness = path.timestamp == null
+        ? 0.0
+        : 1.0 /
+              (1.0 +
+                  (DateTime.now().difference(path.timestamp!).inMinutes /
+                      60.0 /
+                      24.0));
+    final routeWeight =
+        (path.routeWeight / highestRouteWeight).clamp(0.0, 1.0);
+
+    return (reliability * 0.45) +
+        (latency * 0.25) +
+        (freshness * 0.1) +
+        (routeWeight * 0.2);
+  }
+
  bool _pathsEqual(List<int> a, List<int> b) {
    return listEquals(a, b);
  }
@@ -369,6 +567,28 @@ class PathHistoryService extends ChangeNotifier {
  }
 }

+class _DeferredPathRecord {
+  final int hopCount;
+  final int tripTimeMs;
+  final bool wasFloodDiscovery;
+  final List<int> pathBytes;
+  final int successCount;
+  final int failureCount;
+  final double routeWeight;
+  final DateTime? timestamp;
+
+  _DeferredPathRecord({
+    required this.hopCount,
+    required this.tripTimeMs,
+    required this.wasFloodDiscovery,
+    required this.pathBytes,
+    required this.successCount,
+    required this.failureCount,
+    this.routeWeight = 1.0,
+    this.timestamp,
+  });
+}
+
 class _FloodStats {
  int successCount = 0;
  int failureCount = 0;