Retry stale PTS snapshots without resetting account state

Author: decoder-dev

CHANGELOG.md

@@ -10,6 +10,8 @@
 - Failed TCP endpoints share monotonic cooldown and a single recovery probe across contexts; duplicate connects reuse the existing connection, and closed interfaces release read buffers.
 - WEB resume requests coalesce per carrier. Stale callbacks cannot restart disabled/replaced profiles; duplicate failures count once and short-lived sessions retain backoff history.
 - Removed the global WebSocket fallback cancellation and the mid-transaction state replay timeout. State replay finishes atomically instead of committing a prefix and resetting synchronization.
+- Stale PTS snapshots retry difference synchronization without a full account reset. WebSocket dial requests are one-shot across fallback gaps and use a monotonic fallback clock.
+- Fetch completion logs identify the location and episode; TCP attempts have IDs for correlation and log the effective connection deadline.
 - Download EOF handling accepts empty boundary responses and out-of-order partial responses, without treating requested ranges as received bytes.
 - Memory-pressure cache trimming retains contexts owned by open peer views.
 - Recovery tests exercise production EOF handling, endpoint health, WEB manager lifecycle and the Network.framework interface against a loopback echo server.

Tests/NetworkRecovery/run.py

@@ -19,6 +19,8 @@ def extract(path, start, end):
                  '        private func commitPendingReadyPart(', '\n        }').replace('private func', 'func', 1)
 trim = extract('submodules/TelegramCore/Sources/State/AccountViewTracker.swift',
                '    func trimCachedData()', '\n    }')
+ws_connect = extract('submodules/TelegramCore/Sources/Network/MTWebSocketConnectionInterface.swift',
+                     '        func connect(timeout: Double)', '\n        }')
 fixture = '''import Foundation
 final class Logger {
     static let shared = Logger()
@@ -44,6 +46,17 @@ def extract(path, start, end):
     let queue = ImmediateQueue()
     var cachedDataContexts: [Int: PeerContext] = [:]
 ''' + trim + '\n}\n'
+fixture += '''
+struct Selector { var resets = 0; mutating func reset() { resets += 1 } }
+final class WebSocketDial {
+    var connection: Int?
+    var connectTimeout: Double = 0
+    var didRequestConnect = false
+    var reportedDisconnection = false
+    var endpointSelector = Selector()
+    var dials = 0
+    func dialCurrentCandidate() { dials += 1 }
+''' + ws_connect + '\n}\n'
 
 tests = '''
 var health = NetworkEndpointHealth()
@@ -103,6 +116,14 @@ def extract(path, start, end):
 tracker.trimCachedData()
 precondition(tracker.cachedDataContexts.count == 1 && tracker.cachedDataContexts[1] === active)
 print("Endpoint cooldown, probe ownership, EOF permutations and active cache ownership: passed")
+let ws = WebSocketDial()
+for _ in 0..<100 { ws.connect(timeout: 12) }
+precondition(ws.dials == 1 && ws.endpointSelector.resets == 1)
+let closed = WebSocketDial()
+closed.reportedDisconnection = true
+closed.connect(timeout: 12)
+precondition(closed.dials == 0)
+print("WebSocket connect coalescing across candidate gaps and terminal close: passed")
 '''
 with tempfile.TemporaryDirectory(prefix='telegram-recovery-') as tmp:
     file = pathlib.Path(tmp) / 'main.swift'

submodules/FetchManagerImpl/Sources/FetchManagerImpl.swift

@@ -295,7 +295,7 @@ private final class FetchManagerCategoryContext {
                         return .single(type)
                     }
                     |> deliverOnMainQueue).start(next: { _ in
-                        Logger.shared.log("FetchManager", "Completed fetching \(entry.resourceReference.resource.id.stringRepresentation)")
+                        Logger.shared.log("FetchManager", "Completed fetching \(entry.resourceReference.resource.id.stringRepresentation) [\(entry.id.location), episode \(entry.episode)]")
                         entryCompleted(id)
                     })
                 } else {
@@ -460,7 +460,7 @@ private final class FetchManagerCategoryContext {
                             return .single(type)
                         }
                         |> deliverOnMainQueue).start(next: { _ in
-                            Logger.shared.log("FetchManager", "Completed fetching \(entry.resourceReference.resource.id.stringRepresentation)")
+                            Logger.shared.log("FetchManager", "Completed fetching \(entry.resourceReference.resource.id.stringRepresentation) [\(entry.id.location), episode \(entry.episode)]")
                             entryCompleted(topEntryId)
                         })
                     }

submodules/TelegramCore/Sources/Network/MTWebSocketConnectionInterface.swift

@@ -51,7 +51,7 @@ final class MTWebSocketFallbackCoordinator {
     /// fallback is engaged it answers true exactly once per wait, and false to everything else until
     /// that attempt has reported success or failure.
     func shouldAttemptWebSocket() -> Bool {
-        let timestamp = CFAbsoluteTimeGetCurrent()
+        let timestamp = ProcessInfo.processInfo.systemUptime
         var grantedProbe = false
         let state = self.state.modify { state in
             var state = state
@@ -86,7 +86,7 @@ final class MTWebSocketFallbackCoordinator {
     }
 
     func recordAllEndpointsFailed() {
-        let timestamp = CFAbsoluteTimeGetCurrent()
+        let timestamp = ProcessInfo.processInfo.systemUptime
         var engagedFallback = false
         var probeFailed = false
         let state = self.state.modify { state in
@@ -200,6 +200,7 @@ final class MTWebSocketConnectionInterface: NSObject, MTTcpConnectionInterface {
 
         private var connection: NWConnection?
         private var reportedDisconnection = false
+        private var didRequestConnect = false
         private var currentInterfaceIsWifi = true
 
         private var connectTimeout: Double = 12.0
@@ -316,10 +317,10 @@ final class MTWebSocketConnectionInterface: NSObject, MTTcpConnectionInterface {
         }
 
         func connect(timeout: Double) {
-            if self.connection != nil {
-                assertionFailure("A connection already exists")
-                return
-            }
+            // The interface owns one MTProto stream, including the jitter gap where its
+            // NWConnection is temporarily nil. Repeated requests must not reset candidates.
+            guard !self.didRequestConnect, !self.reportedDisconnection else { return }
+            self.didRequestConnect = true
             self.connectTimeout = timeout
             self.endpointSelector.reset()
             self.dialCurrentCandidate()

submodules/TelegramCore/Sources/Network/NetworkFrameworkTcpConnectionInterface.swift

@@ -278,7 +278,7 @@ final class NetworkFrameworkTcpConnectionInterface: NSObject, MTTcpConnectionInt
                     return
                 }
                 self.connectTimeoutTimer = nil
-                Logger.shared.log("Network", "NW connect to \(self.endpointDescription) timed out after \(timeout)s")
+                Logger.shared.log("Network", "NW connect to \(self.endpointDescription) timed out after \(connectTimeout)s (attempt \(self.attempt))")
                 if self.isCountedInFlight {
                     Impl.endpointHealth.failed(endpoint: self.endpointDescription, attempt: self.attempt, now: ProcessInfo.processInfo.systemUptime)
                 }
@@ -306,7 +306,7 @@ final class NetworkFrameworkTcpConnectionInterface: NSObject, MTTcpConnectionInt
             }
             self.isCountedInFlight = true
             Impl.inFlightConnectCount += 1
-            Logger.shared.log("Network", "NW connect starting to \(self.endpointDescription), \(Impl.inFlightConnectCount) in flight")
+            Logger.shared.log("Network", "NW connect starting to \(self.endpointDescription), \(Impl.inFlightConnectCount) in flight (attempt \(self.attempt))")
 
             connection.start(queue: self.queue.queue)
 
@@ -545,7 +545,7 @@ final class NetworkFrameworkTcpConnectionInterface: NSObject, MTTcpConnectionInt
             // connection that dies after minutes of traffic look identical in the count above, and
             // they are not the same problem — the first is a retry loop with nothing throttling it.
             if let error = error, self.isCountedInFlight {
-                Logger.shared.log("Network", "NW connect to \(self.endpointDescription) failed: \(error)")
+                Logger.shared.log("Network", "NW connect to \(self.endpointDescription) failed: \(error) (attempt \(self.attempt))")
             }
             self.leaveInFlight()
 

submodules/TelegramCore/Sources/State/AccountStateManager.swift

@@ -915,8 +915,11 @@ public final class AccountStateManager {
 
                                                     return (difference, replayedState, false, false)
                                                 } else {
-                                                    Logger.shared.log("State", "replayFinalState returned nil, triggering state reset")
-                                                    return (nil, nil, true, true)
+                                                    // A concurrent state/channel update can invalidate the snapshot.
+                                                    // Retry the difference from current PTS; this is not a reason to
+                                                    // discard the chat list and advance the account to updates.getState.
+                                                    Logger.shared.log("State", "replayFinalState snapshot changed, retrying difference")
+                                                    return (nil, nil, false, false)
                                                 }
                                             }
                                         }