@ -44,6 +44,12 @@ class MessageRetryService extends ChangeNotifier {
[ ] ; / / Rolling buffer of recent ACK hashes
[ ] ; / / Rolling buffer of recent ACK hashes
final Map < String , List < String > > _pendingMessageQueuePerContact =
final Map < String , List < String > > _pendingMessageQueuePerContact =
{ } ; / / contactPubKeyHex → FIFO queue of messageIds ( DEPRECATED - will be removed )
{ } ; / / 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 =
final Map < String , String > _expectedHashToMessageId =
{ } ; / / expectedAckHashHex → messageId ( for matching RESP_CODE_SENT by hash )
{ } ; / / expectedAckHashHex → messageId ( for matching RESP_CODE_SENT by hash )
@ -156,7 +162,40 @@ class MessageRetryService extends ChangeNotifier {
_addMessageCallback ! ( contact . publicKeyHex , message ) ;
_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 ) ;
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 {
Future < void > _attemptSend ( String messageId ) async {
@ -169,13 +208,11 @@ class MessageRetryService extends ChangeNotifier {
/ / Use the path that was captured when the message was first sent
/ / Use the path that was captured when the message was first sent
if ( _setContactPathCallback ! = null & & _clearContactPathCallback ! = null ) {
if ( _setContactPathCallback ! = null & & _clearContactPathCallback ! = null ) {
if ( message . pathLength ! = null & & message . pathLength ! < 0 ) {
if ( message . pathLength ! = null & & message . pathLength ! < 0 ) {
/ / Flood mode - clear the path
debugPrint (
debugPrint (
' Setting flood mode for retry attempt ${ message . retryCount } ' ,
' Setting flood mode for retry attempt ${ message . retryCount } ' ,
) ;
) ;
_clearContactPathCallback ! ( contact ) ;
await _clearContactPathCallback ! ( contact ) ;
} else if ( message . pathLength ! = null & & message . pathLength ! > = 0 ) {
} else if ( message . pathLength ! = null & & message . pathLength ! > = 0 ) {
/ / Specific path ( including direct neighbor with pathLength = 0 )
final pathStr = message . pathBytes . isEmpty
final pathStr = message . pathBytes . isEmpty
? ' direct '
? ' direct '
: message . pathBytes
: message . pathBytes
@ -192,6 +229,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 attempt = message . retryCount . clamp ( 0 , 3 ) ;
final timestampSeconds = message . timestamp . millisecondsSinceEpoch ~ / 1000 ;
final timestampSeconds = message . timestamp . millisecondsSinceEpoch ~ / 1000 ;
@ -231,6 +286,15 @@ class MessageRetryService extends ChangeNotifier {
if ( _sendMessageCallback ! = null ) {
if ( _sendMessageCallback ! = null ) {
_sendMessageCallback ! ( contact , message . text , attempt , timestampSeconds ) ;
_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 +345,7 @@ class MessageRetryService extends ChangeNotifier {
}
}
/ / FALLBACK: Old queue - based matching ( for messages sent before hash computation was added )
/ / 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 ) {
if ( messageId = = null & & allowQueueFallback ) {
_debugLogService ? . warn (
_debugLogService ? . warn (
' RESP_CODE_SENT: ACK hash $ ackHashHex not found in hash table, falling back to queue ' ,
' RESP_CODE_SENT: ACK hash $ ackHashHex not found in hash table, falling back to queue ' ,
@ -290,13 +355,28 @@ class MessageRetryService extends ChangeNotifier {
' Hash-based match failed for $ ackHashHex , falling back to queue-based matching ' ,
' Hash-based match failed for $ ackHashHex , falling back to queue-based matching ' ,
) ;
) ;
for ( var entry in _pendingMessageQueuePerContact . entries ) {
/ / Try to identify the correct contact from _activeMessages first .
String ? targetContactKey ;
for ( final activeId in _activeMessages ) {
final activeContact = _pendingContacts [ activeId ] ;
if ( activeContact ! = null ) {
targetContactKey = activeContact . publicKeyHex ;
break ;
}
}
final queuesToSearch = targetContactKey ! = null
? { targetContactKey: _pendingMessageQueuePerContact [ targetContactKey ] }
: _pendingMessageQueuePerContact ;
for ( var entry in queuesToSearch . entries ) {
final contactKey = entry . key ;
final contactKey = entry . key ;
final queue = entry . value ;
final queue = entry . value ;
if ( queue = = null ) continue ;
if ( queue . isNotEmpty ) {
/ / Drain stale entries until we find a valid one or exhaust the queue .
while ( queue . isNotEmpty ) {
final candidateMessageId = queue . removeAt ( 0 ) ;
final candidateMessageId = queue . removeAt ( 0 ) ;
if ( _pendingMessages . containsKey ( candidateMessageId ) ) {
if ( _pendingMessages . containsKey ( candidateMessageId ) ) {
messageId = candidateMessageId ;
messageId = candidateMessageId ;
contact = _pendingContacts [ candidateMessageId ] ;
contact = _pendingContacts [ candidateMessageId ] ;
@ -304,21 +384,10 @@ class MessageRetryService extends ChangeNotifier {
' Queue-based match (fallback): $ ackHashHex → message $ messageId for $ contactKey ' ,
' Queue-based match (fallback): $ ackHashHex → message $ messageId for $ contactKey ' ,
) ;
) ;
break ;
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 +532,7 @@ class MessageRetryService extends ChangeNotifier {
} else {
} else {
/ / Max retries reached - mark as failed
/ / Max retries reached - mark as failed
final failedMessage = message . copyWith ( status: MessageStatus . failed ) ;
final failedMessage = message . copyWith ( status: MessageStatus . failed ) ;
_pendingMessages [ messageId ] = failedMessage ;
/ / 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 ) ;
}
/ / Check if we should clear the path on max retry
/ / Check if we should clear the path on max retry
if ( _appSettingsService ? . settings . clearPathOnMaxRetry = = true & &
if ( _appSettingsService ? . settings . clearPathOnMaxRetry = = true & &
@ -499,6 +553,30 @@ class MessageRetryService extends ChangeNotifier {
}
}
notifyListeners ( ) ;
notifyListeners ( ) ;
/ / Message is done retrying — send next queued message for this contact
_onMessageResolved ( messageId , contact . publicKeyHex ) ;
/ / Keep message in pending maps for 30 s 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 +672,15 @@ class MessageRetryService extends ChangeNotifier {
}
}
if ( matchedMessageId ! = null ) {
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 contact = _pendingContacts [ matchedMessageId ] ;
final selection = _pendingPathSelections [ matchedMessageId ] ;
final selection = _pendingPathSelections [ matchedMessageId ] ;
@ -616,12 +702,21 @@ class MessageRetryService extends ChangeNotifier {
tripTimeMs: tripTimeMs ,
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
/ / Move ACK hashes to history before removing
_moveAckHashesToHistory ( matchedMessageId ) ;
_moveAckHashesToHistory ( matchedMessageId ) ;
_pendingMessages . remove ( matchedMessageId ) ;
_pendingMessages . remove ( matchedMessageId ) ;
_pendingContacts . remove ( matchedMessageId ) ;
_pendingContacts . remove ( matchedMessageId ) ;
_pendingPathSelections . remove ( matchedMessageId ) ;
_pendingPathSelections . remove ( matchedMessageId ) ;
_resolvedMessages . remove ( matchedMessageId ) ;
/ / Clean up the queue entry for this contact ( remove any remaining references to this message )
/ / Clean up the queue entry for this contact ( remove any remaining references to this message )
if ( contact ! = null ) {
if ( contact ! = null ) {
@ -646,6 +741,7 @@ class MessageRetryService extends ChangeNotifier {
true ,
true ,
tripTimeMs ,
tripTimeMs ,
) ;
) ;
_onMessageResolved ( matchedMessageId , contact . publicKeyHex ) ;
}
}
notifyListeners ( ) ;
notifyListeners ( ) ;
@ -783,6 +879,9 @@ class MessageRetryService extends ChangeNotifier {
_ackHistory . clear ( ) ;
_ackHistory . clear ( ) ;
_ackHashToMessageId . clear ( ) ;
_ackHashToMessageId . clear ( ) ;
_pendingMessageQueuePerContact . clear ( ) ;
_pendingMessageQueuePerContact . clear ( ) ;
_sendQueue . clear ( ) ;
_activeMessages . clear ( ) ;
_resolvedMessages . clear ( ) ;
super . dispose ( ) ;
super . dispose ( ) ;
}
}
} }
zjs81
4 months ago