Retry transient sends and shared-profile downloads

lib/services/rustpush/profile_retry_policy.dart

@@ -0,0 +1,58 @@
+enum ProfileFailureKind { transient, permanent }
+
+class ProfileRetryPolicy {
+  ProfileRetryPolicy({
+    this.delays = const <Duration>[
+      Duration(seconds: 5),
+      Duration(seconds: 30),
+      Duration(minutes: 2),
+    ],
+  });
+
+  final List<Duration> delays;
+  final Map<String, int> _attempts = <String, int>{};
+  final Set<String> _scheduled = <String>{};
+
+  ProfileFailureKind classify(Object error) {
+    final description = error.toString().toLowerCase();
+    if (description.contains("profile service unavailable") ||
+        description.contains("timeout") ||
+        description.contains("connection") ||
+        description.contains("network") ||
+        description.contains("socket") ||
+        description.contains("dns")) {
+      return ProfileFailureKind.transient;
+    }
+    return ProfileFailureKind.permanent;
+  }
+
+  bool get isEmpty => _attempts.isEmpty && _scheduled.isEmpty;
+
+  bool isScheduled(String profileKey) => _scheduled.contains(profileKey);
+
+  Duration? schedule(String profileKey) {
+    if (!_scheduled.add(profileKey)) return null;
+    final attempt = _attempts[profileKey] ?? 0;
+    if (attempt >= delays.length) {
+      _scheduled.remove(profileKey);
+      _attempts.remove(profileKey);
+      return null;
+    }
+    _attempts[profileKey] = attempt + 1;
+    return delays[attempt];
+  }
+
+  void timerFired(String profileKey) {
+    _scheduled.remove(profileKey);
+  }
+
+  void complete(String profileKey) {
+    _scheduled.remove(profileKey);
+    _attempts.remove(profileKey);
+  }
+
+  void clear() {
+    _scheduled.clear();
+    _attempts.clear();
+  }
+}

lib/services/rustpush/rustpush_service.dart

@@ -19,6 +19,8 @@ import 'package:bluebubbles/src/rust/lib.dart' as lib;
 import 'package:bluebubbles/helpers/helpers.dart';
 import 'package:bluebubbles/database/models.dart';
 import 'package:bluebubbles/services/services.dart';
+import 'package:bluebubbles/services/rustpush/send_retry_policy.dart';
+import 'package:bluebubbles/services/rustpush/profile_retry_policy.dart';
 import 'package:bluebubbles/utils/crypto_utils.dart';
 import 'package:bluebubbles/utils/logger/logger.dart';
 import 'package:collection/collection.dart';
@@ -342,6 +344,8 @@ class RustPushBBUtils {
 }
 
 class RustPushBackend implements BackendService {
+  final InFlightSendRegistry _inFlightSends = InFlightSendRegistry();
+
   Future<String> getDefaultHandle() async {
     var myHandles = await api.getHandles(state: pushService.state!.client);
     var setHandle = ss.settings.defaultHandle.value;
@@ -398,22 +402,33 @@ class RustPushBackend implements BackendService {
     return const api.MessageType.iMessage();
   }
 
-  Future<void> sendMsg(api.MessageInst msg) async {
+  Future<void> sendMsg(api.MessageInst msg, {bool waitForResource = true}) {
+    return _inFlightSends.run(msg.id, () => _sendMsgWithRetry(msg, waitForResource: waitForResource));
+  }
+
+  Future<void> _sendMsgWithRetry(api.MessageInst msg, {required bool waitForResource}) async {
     var message = Message.findOne(guid: msg.id);
     if (message != null) {
       message.sendingServiceId = pushService.serviceId;
       message.save(updateSendingServiceId: true);
     }
     var stillRunning = false;
+    final retryPolicy = SendRetryPolicy();
     try {
-      stillRunning = await api.send(state: pushService.state!.client, local: pushService.state!.localBroadcast, msg: msg);
-    } catch (e) {
-      if (e is AnyhowException) {
-        if (e.message.contains("Failed to generate resource") && e.message.contains("not retrying")) {
-          pushService.markFailedToLogin();
+      while (true) {
+        try {
+          stillRunning = await api.send(state: pushService.state!.client, local: pushService.state!.localBroadcast, msg: msg);
+          break;
+        } catch (e) {
+          final decision = retryPolicy.next(e, waitForResource: waitForResource);
+          if (decision.markFailedLogin) {
+            pushService.markFailedToLogin();
+          }
+          if (!decision.retry) rethrow;
+          Logger.warn("Retrying send ${msg.id} in ${decision.delay.inSeconds}s (${decision.kind.name})");
+          await Future.delayed(decision.delay);
         }
       }
-      rethrow;
     } finally {
       if (!stillRunning) {
         message = Message.findOne(guid: msg.id);
@@ -1279,7 +1294,7 @@ class RustPushBackend implements BackendService {
         icon: base64Decode(appdata.appIcon!),
       ) : null)
     );
-    await sendMsg(msg);
+    await sendMsg(msg, waitForResource: false); // typing state is stale after a reconnect
   }
 
   @override
@@ -1290,7 +1305,7 @@ class RustPushBackend implements BackendService {
       sender: await c.ensureHandle(),
       message: const api.Message.typing(false)
     );
-    await sendMsg(msg);
+    await sendMsg(msg, waitForResource: false);
   }
 
   @override
@@ -2827,8 +2842,86 @@ class RustPushService extends GetxService {
     return null;
   }
 
-  List<String> profilesDownloading = [];
-  Future handleSharedProfile(api.ShareProfileMessage shared, String sender, List<Handle> targets) async {
+  final Set<String> profilesDownloading = {};
+  final Map<String, Timer> _profileRetryTimers = {};
+  final ProfileRetryPolicy _profileRetryPolicy = ProfileRetryPolicy();
+
+  String _profileFailureCategory(Object error) {
+    final description = error.toString().toLowerCase();
+    if (description.contains("profile service unavailable")) return "service_unavailable";
+    if (description.contains("timeout")) return "timeout";
+    if (description.contains("connection") ||
+        description.contains("network") ||
+        description.contains("socket") ||
+        description.contains("dns")) {
+      return "network";
+    }
+    if (description.contains("record") && description.contains("not found")) return "record_not_found";
+    if (description.contains("plist") || description.contains("serde")) return "plist";
+    if (description.contains("decrypt") ||
+        description.contains("hmac") ||
+        description.contains("crypto")) {
+      return "crypto";
+    }
+    if (description.contains("asset")) return "asset";
+    if (description.contains("panic")) return "panic";
+    return error.runtimeType.toString();
+  }
+
+  void _clearProfileRetry(String profileKey) {
+    _profileRetryTimers.remove(profileKey)?.cancel();
+    _profileRetryPolicy.complete(profileKey);
+  }
+
+  void _scheduleProfileRetry(
+    api.ShareProfileMessage shared,
+    String sender,
+    List<Handle> targets,
+    String category,
+  ) {
+    final profileKey = shared.cloudKitRecordKey;
+    final delay = _profileRetryPolicy.schedule(profileKey);
+    if (delay == null) {
+      Logger.warn("Shared profile fetch exhausted or already scheduled category=$category");
+      return;
+    }
+
+    Logger.warn(
+      "Shared profile fetch deferred category=$category "
+      "retry_in_seconds=${delay.inSeconds}",
+    );
+    _profileRetryTimers[profileKey] = Timer(delay, () {
+      _profileRetryTimers.remove(profileKey);
+      _profileRetryPolicy.timerFired(profileKey);
+      unawaited(handleSharedProfile(shared, sender, targets));
+    });
+  }
+
+  Future<void> handleSharedProfile(api.ShareProfileMessage shared, String sender, List<Handle> targets) async {
+    final profileKey = shared.cloudKitRecordKey;
+    if (_profileRetryPolicy.isScheduled(profileKey) || !profilesDownloading.add(profileKey)) return;
+
+    try {
+      await _handleSharedProfile(shared, sender, targets);
+      _clearProfileRetry(profileKey);
+    } catch (error) {
+      // Shared profile payloads are optional message metadata. A malformed
+      // CloudKit plist must not escape an unawaited profile task and disturb
+      // message delivery. Retry independently so the contact image can recover
+      // after transient CloudKit, network, or service-initialization failures.
+      final category = _profileFailureCategory(error);
+      if (_profileRetryPolicy.classify(error) == ProfileFailureKind.transient) {
+        _scheduleProfileRetry(shared, sender, targets, category);
+      } else {
+        Logger.warn("Shared profile fetch permanently failed category=$category");
+        _profileRetryPolicy.complete(profileKey);
+      }
+    } finally {
+      profilesDownloading.remove(profileKey);
+    }
+  }
+
+  Future<void> _handleSharedProfile(api.ShareProfileMessage shared, String sender, List<Handle> targets) async {
     var myHandles = await api.getHandles(state: pushService.state!.client);
     if (myHandles.contains(sender)) {
       for (var target in targets) {
@@ -5036,6 +5129,11 @@ class RustPushService extends GetxService {
 
   @override
   void onClose() {
+    for (final timer in _profileRetryTimers.values) {
+      timer.cancel();
+    }
+    _profileRetryTimers.clear();
+    _profileRetryPolicy.clear();
     if (state != null) disposeState(state!, true, false);
     super.onClose();
   }

lib/services/rustpush/send_retry_policy.dart

@@ -0,0 +1,116 @@
+import 'dart:async';
+
+enum SendFailureKind {
+  resourceUnavailable,
+  confirmationTimeout,
+  permanent,
+  unknown,
+}
+
+class SendRetryDecision {
+  const SendRetryDecision({
+    required this.kind,
+    required this.retry,
+    this.delay = Duration.zero,
+    this.markFailedLogin = false,
+  });
+
+  final SendFailureKind kind;
+  final bool retry;
+  final Duration delay;
+  final bool markFailedLogin;
+}
+
+class SendRetryPolicy {
+  SendRetryPolicy({
+    this.budget = const Duration(minutes: 3),
+    this.maxResourceWait = const Duration(seconds: 35),
+    this.timeoutRetryWait = const Duration(seconds: 2),
+    this.maxTimeoutRetries = 1,
+  });
+
+  static final RegExp resourceRetryRegex = RegExp(r"retrying in (\d+)s");
+
+  final Duration budget;
+  final Duration maxResourceWait;
+  final Duration timeoutRetryWait;
+  final int maxTimeoutRetries;
+  Duration elapsed = Duration.zero;
+  int timeoutRetries = 0;
+
+  SendFailureKind classify(Object error) {
+    final description = error.toString();
+    if (description.contains("Failed to generate resource")) {
+      return description.contains("not retrying")
+          ? SendFailureKind.permanent
+          : SendFailureKind.resourceUnavailable;
+    }
+    if (description.contains("Send timeout; try again")) {
+      return SendFailureKind.confirmationTimeout;
+    }
+    return SendFailureKind.unknown;
+  }
+
+  Duration? resourceRetryWait(Object error) {
+    final seconds = int.tryParse(
+      resourceRetryRegex.firstMatch(error.toString())?.group(1) ?? "",
+    );
+    if (seconds == null) return null;
+    final wait = Duration(seconds: seconds) + const Duration(seconds: 1);
+    return wait > maxResourceWait ? maxResourceWait : wait;
+  }
+
+  SendRetryDecision next(Object error, {bool waitForResource = true}) {
+    final kind = classify(error);
+    if (kind == SendFailureKind.permanent) {
+      return const SendRetryDecision(
+        kind: SendFailureKind.permanent,
+        retry: false,
+        markFailedLogin: true,
+      );
+    }
+
+    Duration? delay;
+    if (kind == SendFailureKind.confirmationTimeout &&
+        timeoutRetries < maxTimeoutRetries) {
+      timeoutRetries++;
+      delay = timeoutRetryWait;
+    } else if (kind == SendFailureKind.resourceUnavailable && waitForResource) {
+      delay = resourceRetryWait(error);
+    }
+
+    if (delay == null || elapsed + delay > budget) {
+      return SendRetryDecision(kind: kind, retry: false);
+    }
+
+    elapsed += delay;
+    return SendRetryDecision(kind: kind, retry: true, delay: delay);
+  }
+}
+
+class InFlightSendRegistry {
+  final Map<String, Future<void>> _active = <String, Future<void>>{};
+
+  Future<void> run(String messageId, Future<void> Function() operation) {
+    final existing = _active[messageId];
+    if (existing != null) return existing;
+
+    final completer = Completer<void>();
+    _active[messageId] = completer.future;
+    Future<void>.sync(operation).then<void>(
+      (_) {
+        completer.complete();
+        if (identical(_active[messageId], completer.future)) {
+          _active.remove(messageId);
+        }
+      },
+      onError: (Object error, StackTrace stackTrace) {
+        completer.completeError(error, stackTrace);
+        if (identical(_active[messageId], completer.future)) {
+          _active.remove(messageId);
+        }
+      },
+    );
+    return completer.future;
+  }
+}