|
|
|
|
@ -44,6 +44,12 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
[]; // Rolling buffer of recent ACK hashes
|
|
|
|
|
final Map<String, List<String>> _pendingMessageQueuePerContact =
|
|
|
|
|
{}; // contactPubKeyHex → FIFO queue of messageIds (DEPRECATED - will be removed)
|
|
|
|
|
final Map<String, List<String>> _sendQueue =
|
|
|
|
|
{}; // contactPubKeyHex → ordered list of messageIds awaiting send
|
|
|
|
|
final Set<String> _activeMessages =
|
|
|
|
|
{}; // messageIds currently in-flight (sent/retrying)
|
|
|
|
|
final Set<String> _resolvedMessages =
|
|
|
|
|
{}; // messageIds already resolved (prevents double _onMessageResolved)
|
|
|
|
|
final Map<String, String> _expectedHashToMessageId =
|
|
|
|
|
{}; // expectedAckHashHex → messageId (for matching RESP_CODE_SENT by hash)
|
|
|
|
|
|
|
|
|
|
@ -156,7 +162,49 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
_addMessageCallback!(contact.publicKeyHex, message);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
await _attemptSend(messageId);
|
|
|
|
|
// Queue per contact — only one message in-flight at a time to avoid
|
|
|
|
|
// overflowing the firmware's 8-entry expected_ack_table.
|
|
|
|
|
final contactKey = contact.publicKeyHex;
|
|
|
|
|
_sendQueue[contactKey] ??= [];
|
|
|
|
|
_sendQueue[contactKey]!.add(messageId);
|
|
|
|
|
|
|
|
|
|
if (!_activeMessages.any(
|
|
|
|
|
(id) => _pendingContacts[id]?.publicKeyHex == contactKey,
|
|
|
|
|
)) {
|
|
|
|
|
_sendNextForContact(contactKey);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
void _sendNextForContact(String contactKey) {
|
|
|
|
|
final queue = _sendQueue[contactKey];
|
|
|
|
|
if (queue == null) return;
|
|
|
|
|
|
|
|
|
|
// Drain stale entries iteratively instead of recursing.
|
|
|
|
|
while (queue.isNotEmpty) {
|
|
|
|
|
final messageId = queue.removeAt(0);
|
|
|
|
|
if (_pendingMessages.containsKey(messageId)) {
|
|
|
|
|
_activeMessages.add(messageId);
|
|
|
|
|
_attemptSend(messageId).catchError((e) {
|
|
|
|
|
debugPrint('_attemptSend threw for $messageId: $e');
|
|
|
|
|
final msg = _pendingMessages[messageId];
|
|
|
|
|
if (msg != null) {
|
|
|
|
|
final failed = msg.copyWith(status: MessageStatus.failed);
|
|
|
|
|
_pendingMessages[messageId] = failed;
|
|
|
|
|
_updateMessageCallback?.call(failed);
|
|
|
|
|
}
|
|
|
|
|
_onMessageResolved(messageId, contactKey);
|
|
|
|
|
});
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
// Message was cancelled/cleaned up while queued — try next
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
void _onMessageResolved(String messageId, String contactKey) {
|
|
|
|
|
if (_resolvedMessages.contains(messageId)) return;
|
|
|
|
|
_resolvedMessages.add(messageId);
|
|
|
|
|
_activeMessages.remove(messageId);
|
|
|
|
|
_sendNextForContact(contactKey);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
Future<void> _attemptSend(String messageId) async {
|
|
|
|
|
@ -169,13 +217,11 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
// Use the path that was captured when the message was first sent
|
|
|
|
|
if (_setContactPathCallback != null && _clearContactPathCallback != null) {
|
|
|
|
|
if (message.pathLength != null && message.pathLength! < 0) {
|
|
|
|
|
// Flood mode - clear the path
|
|
|
|
|
debugPrint(
|
|
|
|
|
'Setting flood mode for retry attempt ${message.retryCount}',
|
|
|
|
|
);
|
|
|
|
|
_clearContactPathCallback!(contact);
|
|
|
|
|
await _clearContactPathCallback!(contact);
|
|
|
|
|
} else if (message.pathLength != null && message.pathLength! >= 0) {
|
|
|
|
|
// Specific path (including direct neighbor with pathLength=0)
|
|
|
|
|
final pathStr = message.pathBytes.isEmpty
|
|
|
|
|
? 'direct'
|
|
|
|
|
: message.pathBytes
|
|
|
|
|
@ -192,6 +238,24 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Re-validate after async gap — a timer or ACK could have resolved/retried
|
|
|
|
|
// this message while we were awaiting the path callback.
|
|
|
|
|
final currentMessage = _pendingMessages[messageId];
|
|
|
|
|
if (currentMessage == null || _resolvedMessages.contains(messageId)) {
|
|
|
|
|
debugPrint(
|
|
|
|
|
'_attemptSend: message $messageId resolved during path sync, aborting',
|
|
|
|
|
);
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
// If the message was retried by a timer during our await, the retryCount
|
|
|
|
|
// will have advanced. Only proceed if it still matches the attempt we started.
|
|
|
|
|
if (currentMessage.retryCount != message.retryCount) {
|
|
|
|
|
debugPrint(
|
|
|
|
|
'_attemptSend: message $messageId retryCount changed during path sync, aborting',
|
|
|
|
|
);
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
final attempt = message.retryCount.clamp(0, 3);
|
|
|
|
|
final timestampSeconds = message.timestamp.millisecondsSinceEpoch ~/ 1000;
|
|
|
|
|
|
|
|
|
|
@ -231,6 +295,15 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
|
|
|
|
|
if (_sendMessageCallback != null) {
|
|
|
|
|
_sendMessageCallback!(contact, message.text, attempt, timestampSeconds);
|
|
|
|
|
} else {
|
|
|
|
|
// No send callback — message would be stuck forever. Fail it immediately.
|
|
|
|
|
debugPrint(
|
|
|
|
|
'_attemptSend: no sendMessageCallback, failing message $messageId',
|
|
|
|
|
);
|
|
|
|
|
final failedMessage = message.copyWith(status: MessageStatus.failed);
|
|
|
|
|
_pendingMessages[messageId] = failedMessage;
|
|
|
|
|
_updateMessageCallback?.call(failedMessage);
|
|
|
|
|
_onMessageResolved(messageId, contact.publicKeyHex);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -281,6 +354,7 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// FALLBACK: Old queue-based matching (for messages sent before hash computation was added)
|
|
|
|
|
// Only match within a single contact's queue to avoid cross-contact mismatches.
|
|
|
|
|
if (messageId == null && allowQueueFallback) {
|
|
|
|
|
_debugLogService?.warn(
|
|
|
|
|
'RESP_CODE_SENT: ACK hash $ackHashHex not found in hash table, falling back to queue',
|
|
|
|
|
@ -290,13 +364,16 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
'Hash-based match failed for $ackHashHex, falling back to queue-based matching',
|
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
for (var entry in _pendingMessageQueuePerContact.entries) {
|
|
|
|
|
// Search all contact queues so concurrent chats don't miss matches.
|
|
|
|
|
final queuesToSearch = _pendingMessageQueuePerContact;
|
|
|
|
|
|
|
|
|
|
for (var entry in queuesToSearch.entries) {
|
|
|
|
|
final contactKey = entry.key;
|
|
|
|
|
final queue = entry.value;
|
|
|
|
|
|
|
|
|
|
if (queue.isNotEmpty) {
|
|
|
|
|
// Drain stale entries until we find a valid one or exhaust the queue.
|
|
|
|
|
while (queue.isNotEmpty) {
|
|
|
|
|
final candidateMessageId = queue.removeAt(0);
|
|
|
|
|
|
|
|
|
|
if (_pendingMessages.containsKey(candidateMessageId)) {
|
|
|
|
|
messageId = candidateMessageId;
|
|
|
|
|
contact = _pendingContacts[candidateMessageId];
|
|
|
|
|
@ -304,21 +381,10 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
'Queue-based match (fallback): $ackHashHex → message $messageId for $contactKey',
|
|
|
|
|
);
|
|
|
|
|
break;
|
|
|
|
|
} else {
|
|
|
|
|
debugPrint('Dequeued stale message $candidateMessageId - skipping');
|
|
|
|
|
if (queue.isNotEmpty) {
|
|
|
|
|
final nextMessageId = queue.removeAt(0);
|
|
|
|
|
if (_pendingMessages.containsKey(nextMessageId)) {
|
|
|
|
|
messageId = nextMessageId;
|
|
|
|
|
contact = _pendingContacts[nextMessageId];
|
|
|
|
|
debugPrint(
|
|
|
|
|
'Queue-based match (fallback): $ackHashHex → message $messageId',
|
|
|
|
|
);
|
|
|
|
|
break;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
debugPrint('Dequeued stale message $candidateMessageId - skipping');
|
|
|
|
|
}
|
|
|
|
|
if (messageId != null) break;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -463,22 +529,7 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
} else {
|
|
|
|
|
// Max retries reached - mark as failed
|
|
|
|
|
final failedMessage = message.copyWith(status: MessageStatus.failed);
|
|
|
|
|
|
|
|
|
|
// Move ACK hashes to history before removing
|
|
|
|
|
_moveAckHashesToHistory(messageId);
|
|
|
|
|
|
|
|
|
|
_pendingMessages.remove(messageId);
|
|
|
|
|
_pendingContacts.remove(messageId);
|
|
|
|
|
_pendingPathSelections.remove(messageId);
|
|
|
|
|
_timeoutTimers[messageId]?.cancel();
|
|
|
|
|
_timeoutTimers.remove(messageId);
|
|
|
|
|
|
|
|
|
|
// Clean up the queue entry for this contact
|
|
|
|
|
_pendingMessageQueuePerContact[contact.publicKeyHex]?.remove(messageId);
|
|
|
|
|
if (_pendingMessageQueuePerContact[contact.publicKeyHex]?.isEmpty ??
|
|
|
|
|
false) {
|
|
|
|
|
_pendingMessageQueuePerContact.remove(contact.publicKeyHex);
|
|
|
|
|
}
|
|
|
|
|
_pendingMessages[messageId] = failedMessage;
|
|
|
|
|
|
|
|
|
|
// Check if we should clear the path on max retry
|
|
|
|
|
if (_appSettingsService?.settings.clearPathOnMaxRetry == true &&
|
|
|
|
|
@ -499,6 +550,30 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
notifyListeners();
|
|
|
|
|
|
|
|
|
|
// Message is done retrying — send next queued message for this contact
|
|
|
|
|
_onMessageResolved(messageId, contact.publicKeyHex);
|
|
|
|
|
|
|
|
|
|
// Keep message in pending maps for 30s grace period so late ACKs
|
|
|
|
|
// can still match and update the message to delivered.
|
|
|
|
|
_timeoutTimers[messageId] = Timer(const Duration(seconds: 30), () {
|
|
|
|
|
_moveAckHashesToHistory(messageId);
|
|
|
|
|
// Clean up ALL hash mappings for this message
|
|
|
|
|
_ackHashToMessageId.removeWhere(
|
|
|
|
|
(_, mapping) => mapping.messageId == messageId,
|
|
|
|
|
);
|
|
|
|
|
_expectedHashToMessageId.removeWhere((_, msgId) => msgId == messageId);
|
|
|
|
|
_pendingMessages.remove(messageId);
|
|
|
|
|
_pendingContacts.remove(messageId);
|
|
|
|
|
_pendingPathSelections.remove(messageId);
|
|
|
|
|
_timeoutTimers.remove(messageId);
|
|
|
|
|
_resolvedMessages.remove(messageId);
|
|
|
|
|
final contactKey = contact.publicKeyHex;
|
|
|
|
|
_pendingMessageQueuePerContact[contactKey]?.remove(messageId);
|
|
|
|
|
if (_pendingMessageQueuePerContact[contactKey]?.isEmpty ?? false) {
|
|
|
|
|
_pendingMessageQueuePerContact.remove(contactKey);
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -594,7 +669,15 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (matchedMessageId != null) {
|
|
|
|
|
final message = _pendingMessages[matchedMessageId]!;
|
|
|
|
|
final message = _pendingMessages[matchedMessageId];
|
|
|
|
|
if (message == null) {
|
|
|
|
|
// Message was already cleaned up (e.g. grace period expired)
|
|
|
|
|
_ackHashToMessageId.remove(ackHashHex);
|
|
|
|
|
debugPrint(
|
|
|
|
|
'ACK matched $matchedMessageId but message already cleaned up',
|
|
|
|
|
);
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
final contact = _pendingContacts[matchedMessageId];
|
|
|
|
|
final selection = _pendingPathSelections[matchedMessageId];
|
|
|
|
|
|
|
|
|
|
@ -616,12 +699,21 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
tripTimeMs: tripTimeMs,
|
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
// Clean up ALL hash mappings for this message (from all retry attempts)
|
|
|
|
|
_ackHashToMessageId.removeWhere(
|
|
|
|
|
(_, mapping) => mapping.messageId == matchedMessageId,
|
|
|
|
|
);
|
|
|
|
|
_expectedHashToMessageId.removeWhere(
|
|
|
|
|
(_, msgId) => msgId == matchedMessageId,
|
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
// Move ACK hashes to history before removing
|
|
|
|
|
_moveAckHashesToHistory(matchedMessageId);
|
|
|
|
|
|
|
|
|
|
_pendingMessages.remove(matchedMessageId);
|
|
|
|
|
_pendingContacts.remove(matchedMessageId);
|
|
|
|
|
_pendingPathSelections.remove(matchedMessageId);
|
|
|
|
|
_resolvedMessages.remove(matchedMessageId);
|
|
|
|
|
|
|
|
|
|
// Clean up the queue entry for this contact (remove any remaining references to this message)
|
|
|
|
|
if (contact != null) {
|
|
|
|
|
@ -646,6 +738,7 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
true,
|
|
|
|
|
tripTimeMs,
|
|
|
|
|
);
|
|
|
|
|
_onMessageResolved(matchedMessageId, contact.publicKeyHex);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
notifyListeners();
|
|
|
|
|
@ -783,6 +876,9 @@ class MessageRetryService extends ChangeNotifier {
|
|
|
|
|
_ackHistory.clear();
|
|
|
|
|
_ackHashToMessageId.clear();
|
|
|
|
|
_pendingMessageQueuePerContact.clear();
|
|
|
|
|
_sendQueue.clear();
|
|
|
|
|
_activeMessages.clear();
|
|
|
|
|
_resolvedMessages.clear();
|
|
|
|
|
super.dispose();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
zjs81
4 months ago
committed by
GitHub