From a6e251705bca51d51ec7a50f083f1124a894af3f Mon Sep 17 00:00:00 2001 From: X-Ryl669 Date: Fri, 21 Aug 2026 10:39:49 +0200 Subject: [PATCH] =?UTF-8?q?fix(files):=20make=20multi-file=20phone?= =?UTF-8?q?=E2=86=92laptop=20shares=20prompt=20and=20self-reporting?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Selecting more than two files to send from the phone took minutes, while a single file was instant. Measured on 2026-08-20: a five-file share sat 2m34s before its first byte moved, then completed in 8 s once a heartbeat round finally ran. Everything below came out of that one report; each part is a distinct defect on the same path, and the path only works once they are all fixed. The pull is piggybacked on the LAN heartbeat, ONE file per round, so the heartbeat's cadence IS the transfer rate: * the cadence knew nothing about the queue. Four failed rounds and it fell to the idle tick — 240 s while BLE is live — parking a half-received batch for minutes. It now stays at 2 s while files are queued AND the queue is moving (`file_pull_active()`), easing to 12 s, never 240, once it stalls: a queue that can never drain must not spin a TCP+IK every 2 s for the session; * the cached-IP probe got a single 2 s connect. `resolve_peer_addr` already retries 3× because a dozing Wi-Fi radio answers a cold connect late, and excused the heartbeat on the grounds that it "ticks again in a few seconds" — untrue for a round carrying a file. It now retries when work is queued; * the phone was often unreachable at the cached address at all. While BLE is up it releases the multicast lock and answers no mDNS, so a DHCP renew left the laptop dialing a dead lease with no way to learn better (observed: 2m34s spent on 192.168.0.15 while the phone sat on .125). `LanServer.keepLanHot()` now holds the throughput Wi-Fi lock AND the multicast lock for 60 s around a pull, re-announces over mDNS, and the first offer of a batch pushes AppState over BLE — that carries the live `wifi_ip`, which repoints the cache with no mDNS involved. The throughput lock was also per-blob, so the radio parked between rounds (one file pulled at 0.2 MB/s where warm rounds do 3.5-7.8); * a token the phone had evicted wedged the queue for ever. It answers `"nomatch"` in the bulk-sync done frame, which `exchange_bulk` logged and discarded, so the laptop re-requested a dead token every round, blocking every file behind it. `LanReconnectOutcome.bulk_status` now carries that map and the pull arm drops the entry. Silence is never treated as an answer, and the pop is guarded on the queue front still being the requested token. The offer that starts it all was fire-and-forget, and both ways it could fail were invisible from the phone: * `sealAndNotify` returning false (no BLE session — app restarted, out of range, laptop re-handshaking) was discarded and logged as success. Observed: three files shared 31 s before a BLE reconnect completed went nowhere while the phone logged "offered to laptop" and toasted "Sending 3 files…". Offers are now retried every 3 s for ~1 min, and immediately when the link returns; * returning TRUE only means the local stack queued the notify. Three offers 4-9 ms apart cost one outright — the laptop logged `resynced past dropped BLE frame(s) dropped=1` and queued 2 of 3 files. Offers are now paced 60 ms (the only bulk BLE path here that had no pacing at all), the LAN warming is deferred 250 ms past the burst so its own mDNS re-announce and STATE notify stop cutting in line, and a sent-but-unfetched offer is re-announced. The laptop dedups by content token, so a re-announce of one that DID arrive can't queue a second pull, write a spurious "name (1).ext", or re-prompt for consent. A dropped offer no longer waits out the 30 s resend timer either: the laptop's pull queue is FIFO in arrival order, so it fetching a LATER offer while an earlier one is outstanding PROVES the earlier one never arrived (`offersPresumedDropped()`), and that one is re-announced within ~1 s. Finally, feedback, because a share could sit a minute with nothing on screen after the share sheet's "Sending…": the phone reports "Laptop unreachable — keeping the file(s) queued" once per outage, "File sent: " as each file is FETCHED (the only moment this device can honestly call success), and names what was lost if an offer ends up nowhere. Verified on the live pair (Plasma 6 laptop + phone over BLE/Wi-Fi): a 3-file share that took 2m34s+ before now completes in ~6 s, with each file's "File sent:" toast; a force-stopped phone app recovers by itself — offers held through a 29 s BLE reconnect, delivered on retry, and a notify dropped in flight re-announced and fetched — where before all three files vanished silently. Tests: 141 daemon + 37 app (Rust), 88 Android unit tests, no new clippy warnings. Co-Authored-By: Claude Opus 5 --- .../java/com/vortex/a3/core/lan/LanServer.kt | 88 ++++- .../java/com/vortex/a3/service/VortexStack.kt | 29 ++ .../vortex/a3/service/VortexStackClipboard.kt | 22 +- .../a3/service/VortexStackOfferRetry.kt | 356 ++++++++++++++++++ .../com/vortex/a3/service/OfferVerdictTest.kt | 134 +++++++ linux/daemon/src/core/lan/tcp_client.rs | 91 ++++- .../ui-tauri/src-tauri/src/clipboard_sync.rs | 26 ++ linux/ui-tauri/src-tauri/src/lan.rs | 157 ++++++-- 8 files changed, 869 insertions(+), 34 deletions(-) create mode 100644 android/app/src/main/java/com/vortex/a3/service/VortexStackOfferRetry.kt create mode 100644 android/app/src/test/java/com/vortex/a3/service/OfferVerdictTest.kt diff --git a/android/app/src/main/java/com/vortex/a3/core/lan/LanServer.kt b/android/app/src/main/java/com/vortex/a3/core/lan/LanServer.kt index 88fd42e..593910c 100644 --- a/android/app/src/main/java/com/vortex/a3/core/lan/LanServer.kt +++ b/android/app/src/main/java/com/vortex/a3/core/lan/LanServer.kt @@ -101,6 +101,12 @@ class LanServer( * the stack can gate the redundant BLE burst. */ var onBulkDelivered: (key: String, hash: String) -> Unit = { _, _ -> } + /** Fired after an instant-share FILE blob has been written to the peer, + * with the content token it pulled by. Closes the loop the outgoing-offer + * watchdog waits on: an offer is only really done once the laptop has the + * bytes. */ + var onFileServed: (token: String) -> Unit = { } + /** * Watermark-dataset provider for bulk-sync (currently `sms_history`): * called with the peer's "I have everything up to [sinceMs]" watermark, @@ -196,9 +202,68 @@ class LanServer( } private fun releasePerfLock() { + // A batch window ([keepLanHot]) owns the lock for its whole duration — + // the per-blob `finally` must not drop it between the laptop's rounds, + // which is exactly what let the radio park mid-batch. + if (lanHotJob?.isActive == true) return try { if (perfLock?.isHeld == true) perfLock?.release() } catch (_: Exception) {} perfLock = null } + + /** Runs for the duration of the current "a file pull is imminent" window. */ + @Volatile + private var lanHotJob: Job? = null + + /** Bumped per window so a superseded expiry (the scope is multi-threaded: + * an old timer can resume just as a new window opens) can't release the + * locks the new window is holding. */ + @Volatile + private var lanHotGen: Int = 0 + + /** Whether the persistent BLE link is up, per [setBleLinked] — decides + * whether a closing hot window hands the multicast lock back or keeps it. */ + @Volatile + private var bleLinked: Boolean = false + + /** + * A phone→laptop file pull is imminent: file offers just went out over BLE + * and the laptop will dial us, ONE file per heartbeat round. Keep the LAN + * path hot for [ms] so those rounds land: + * + * - hold the throughput Wi-Fi lock, so the radio doesn't park between + * rounds and answer the laptop's cold TCP probe too late, + * - hold the multicast lock and re-announce, so the laptop can find our + * CURRENT address. While BLE is up we release that lock and answer no + * mDNS, which leaves the laptop's cached IP as its only guess — and a + * DHCP renew since the last successful handshake makes it a dead one. + * + * Repeat calls extend the window; only the first one re-announces. Returns + * true when this call OPENED the window, so a caller can pair it with a + * one-per-batch action (the BLE AppState push) instead of a per-file one. + */ + fun keepLanHot(ms: Long = HOT_WINDOW_MS): Boolean { + val first = lanHotJob?.isActive != true + acquirePerfLock() + acquireMulticast() + if (first) { + nudge() + Log.i(TAG, "LAN hot: radio + mDNS held for an incoming file pull") + } + val gen = ++lanHotGen + lanHotJob?.cancel() + lanHotJob = scope.launch { + kotlinx.coroutines.delay(ms) + if (gen != lanHotGen) return@launch + // Clear before releasing: [releasePerfLock] refuses to drop the + // lock while a window is live, and this one is over. + lanHotJob = null + releasePerfLock() + if (bleLinked) releaseMulticast() + Log.i(TAG, "LAN hot window over (no file pull for ${ms}ms)") + } + return first + } + /** Bound concurrent client handlers so a slow-loris attacker cannot * pin every coroutine + socket FD on the device. */ private val clientSlots = Semaphore(MAX_CONCURRENT_CLIENTS) @@ -310,8 +375,12 @@ class LanServer( * is up; re-acquire the moment BLE drops and mDNS matters again. */ fun setBleLinked(linked: Boolean) { + bleLinked = linked if (linked) { - releaseMulticast() + // Exception: a file pull in flight ([keepLanHot]) needs mDNS to + // answer, because the laptop's cached IP may be a dead lease and + // an unanswered browse leaves it nothing else to dial. + if (lanHotJob?.isActive != true) releaseMulticast() } else if (acceptJob != null) { acquireMulticast() } @@ -681,11 +750,16 @@ class LanServer( Log.i(TAG, "bulk-sync: clipboard_file token=$token not found") status.put(key, "nomatch") } else { - acquirePerfLock() - try { sendChunked(FrameType.CLIPBOARD_FILE, blob) } - finally { releasePerfLock() } + // Extends the hot window: the laptop + // comes back for the NEXT queued file + // in a fresh round moments from now. + keepLanHot() + sendChunked(FrameType.CLIPBOARD_FILE, blob) Log.i(TAG, "bulk-sync: clipboard_file sent (${blob.size} bytes)") status.put(key, "sent") + try { onFileServed(token) } catch (e: Exception) { + Log.w(TAG, "onFileServed listener threw: ${e.message}") + } } continue } @@ -1027,6 +1101,12 @@ class LanServer( /** Cap on concurrent client coroutines (handshake + idle). */ const val MAX_CONCURRENT_CLIENTS: Int = 16 + /** How long [keepLanHot] keeps the radio + mDNS up after the last file + * offer or served blob. Generous on purpose: the laptop needs one + * heartbeat round PER queued file, and a round that misses its window + * costs far more battery in retries than the lock costs held. */ + const val HOT_WINDOW_MS: Long = 60_000 + /** Bulk-sync chunk payload size over TCP. Far larger than the 450B * BLE chunks (TCP is reliable; only the 8KB frame cap binds) — * a 160KB contact list is ~40 frames instead of ~360 notifies. */ diff --git a/android/app/src/main/java/com/vortex/a3/service/VortexStack.kt b/android/app/src/main/java/com/vortex/a3/service/VortexStack.kt index 6225c88..680589b 100644 --- a/android/app/src/main/java/com/vortex/a3/service/VortexStack.kt +++ b/android/app/src/main/java/com/vortex/a3/service/VortexStack.kt @@ -80,6 +80,23 @@ class VortexStack(internal val service: Service) : VortexNotification.Host { * receive cipher. One-at-a-time keeps each burst at the proven 12ms rate. */ internal val companionSendMutex = kotlinx.coroutines.sync.Mutex() internal var lanServer: LanServer? = null + + /** Outgoing file offers not yet fetched by the laptop, keyed by content + * token. See [offerFileToLaptop] — the retry + "it never arrived" toast + * live there. */ + internal val pendingOffers = + java.util.concurrent.ConcurrentHashMap() + internal var offerRetryJob: kotlinx.coroutines.Job? = null + internal val offerRetryKick = newOfferRetryKick() + /** Debounced "warm the LAN for the incoming pull" job — deferred so it + * can't put an mDNS re-announce and a STATE notify between two offers. */ + internal var lanWarmJob: kotlinx.coroutines.Job? = null + /** True once this outage has been reported on screen, so a 5-file share + * doesn't stack 5 identical toasts. Cleared when a send gets through. */ + @Volatile internal var offerUnreachableToasted: Boolean = false + /** Monotonic offer counter — the sequence the laptop's FIFO pull queue is + * compared against to spot an offer that was dropped in flight. */ + @Volatile internal var offerSeq: Long = 0L private var pairingOrchestrator: PairingOrchestrator? = null private var reconnectOrchestrator: ReconnectOrchestrator? = null internal var callFlowOrchestrator: com.vortex.a3.core.call.CallFlowOrchestrator? = null @@ -120,6 +137,11 @@ class VortexStack(internal val service: Service) : VortexNotification.Host { /** Invoked when fresh peer state arrives, so the notification refreshes. */ internal var onStateChanged: () -> Unit = {} + /** `elapsedRealtime()` of the last successful BLE AppState push. Lets the + * file-offer path skip a redundant one that would only compete with the + * offers for the notify queue. */ + @Volatile internal var lastBleStatePushAtMs: Long = 0L + /** True once [start] has wired the stack (advertiser is up). */ fun isStarted(): Boolean = advertiser != null @@ -546,6 +568,9 @@ class VortexStack(internal val service: Service) : VortexNotification.Host { // Push our notes/todos set on (re)connect so the laptop merges + // replies — converges both sides after an offline edit. Debounced. com.vortex.a3.core.notes.NoteSync.markDirty() + // A file offer that couldn't go out while the link was down can go + // now — this is the whole reason it was kept. + kickOfferRetry() // Re-send app icons after a reconnect (until the laptop has them // cached) and flush any notifications buffered during the outage. sentIconPkgs.clear() @@ -703,6 +728,7 @@ class VortexStack(internal val service: Service) : VortexNotification.Host { try { val json = buildLocalAppState().toJsonBytes() if (server.sendStateEncrypted(peerPub, json)) { + lastBleStatePushAtMs = android.os.SystemClock.elapsedRealtime() Log.i(TAG, "state pushed over BLE") } } catch (e: Exception) { @@ -830,6 +856,9 @@ class VortexStack(internal val service: Service) : VortexNotification.Host { ).run(firstFrame) } } + // A blob the laptop pulled is an offer that landed — stop tracking it + // (and don't toast a failure for a file that plainly arrived). + lan.onFileServed = { token -> noteFileServed(token) } lanServer = lan } diff --git a/android/app/src/main/java/com/vortex/a3/service/VortexStackClipboard.kt b/android/app/src/main/java/com/vortex/a3/service/VortexStackClipboard.kt index 2f2bf25..c6bb77d 100644 --- a/android/app/src/main/java/com/vortex/a3/service/VortexStackClipboard.kt +++ b/android/app/src/main/java/com/vortex/a3/service/VortexStackClipboard.kt @@ -61,10 +61,20 @@ internal fun VortexStack.startClipboardOutbound() { o.put("token", token) o.put("bytes", png.size) val offer = o.toString().toByteArray(Charsets.UTF_8) + var delivered = false for (peer in peerStore.list()) { - gattServer?.sendClipboardImageOfferEncrypted(peer.peerStaticPub, offer) + if (gattServer?.sendClipboardImageOfferEncrypted(peer.peerStaticPub, offer) == true) { + delivered = true + } + } + // Not retried, unlike a file: a clipboard image is transient, and by + // the time the link is back the user has copied something else. But + // don't claim it was offered when it wasn't. + if (delivered) { + Log.i(VortexStack.TAG, "clipboard image offered to laptop (${png.size} bytes, token=$token)") + } else { + Log.w(VortexStack.TAG, "clipboard image offer couldn't go out (BLE link down?)") } - Log.i(VortexStack.TAG, "clipboard image offered to laptop (${png.size} bytes, token=$token)") } } @@ -82,10 +92,12 @@ internal fun VortexStack.startClipboardOutbound() { o.put("name", file.name) o.put("mime", file.mime) val offer = o.toString().toByteArray(Charsets.UTF_8) - for (peer in peerStore.list()) { - gattServer?.sendClipboardImageOfferEncrypted(peer.peerStaticPub, offer) - } Log.i(VortexStack.TAG, "clipboard file offered to laptop ('${file.name}', ${file.bytes.size} bytes, token=$token)") + // Tracked until the laptop has actually FETCHED the bytes: the OFFER + // is a fire-and-forget BLE notify that goes nowhere on a dead link, + // and even a delivered one can sit unfetched. Retries, warms the LAN + // path on delivery, and toasts here if it ends up nowhere. + offerFileToLaptop(token, file.name, offer) // Big file → bring up Wi-Fi Direct for a high-speed direct pull. Small // files stay on the router path (the ~6s Wi-Fi switch isn't worth it). if (file.bytes.size >= 4 * 1024 * 1024) maybeStartWifiDirect() diff --git a/android/app/src/main/java/com/vortex/a3/service/VortexStackOfferRetry.kt b/android/app/src/main/java/com/vortex/a3/service/VortexStackOfferRetry.kt new file mode 100644 index 0000000..bf4e233 --- /dev/null +++ b/android/app/src/main/java/com/vortex/a3/service/VortexStackOfferRetry.kt @@ -0,0 +1,356 @@ +package com.vortex.a3.service + +import android.util.Log +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.launch +import kotlinx.coroutines.withTimeoutOrNull + +/** + * Delivery tracking for phone→laptop FILE offers — the half of instant-share + * that used to fail in silence. + * + * A shared file is stashed locally and announced to the laptop as a small OFFER + * frame over BLE; the laptop then PULLS the bytes over LAN. Both steps can fail + * with nothing to show for it: + * + * - the OFFER is a fire-and-forget BLE notify. With no live session (the phone + * app restarted, the user walked out of range, the laptop is re-handshaking) + * it simply doesn't go out — and the send result was discarded, so the log + * claimed success and the user saw a "Sending…" toast for a file the laptop + * never heard about; + * - the offer can land while the LAN pull never happens (laptop asleep, no + * route, consent declined), which is equally invisible from the phone. + * + * So every offer is tracked until the laptop has actually FETCHED it: retried + * while it can't be delivered, watched for a pull once it has been, and + * surfaced as a toast when it ends up nowhere. The stashed blob is untouched + * either way — [com.vortex.a3.core.clipboard.ClipboardBlobStore] keeps it + * addressable, so a later re-share of the same file is free. + */ + +/** One outgoing file offer, tracked until the laptop fetches it. */ +internal class PendingOffer( + val token: String, + val name: String, + val offer: ByteArray, + /** Order this offer was queued in. The laptop pulls its queue FIFO, so + * comparing sequences tells us something its own reports can't — see + * [offersPresumedDropped]. */ + val seq: Long = 0L, +) { + /** BLE send attempts made so far. Only bounds the never-delivered case. */ + var attempts: Int = 0 + + /** `elapsedRealtime()` when the OFFER frame was last handed to the BLE + * stack successfully; 0 while it has never gone out. Note what this can + * NOT tell us: `sealAndNotify` returning true means the LOCAL stack + * queued the notify, not that the laptop received it — a notify can still + * be dropped in flight (observed: `resynced past dropped BLE frame(s)` + * swallowing one offer out of three). Hence [OFFER_RESEND_MS]: an offer + * that stays unfetched gets re-announced rather than assumed delivered. */ + var lastSentAtMs: Long = 0L + + /** The clock the give-up deadline runs from: first successful send, then + * slid forward by progress anywhere in the batch. Separate from + * [lastSentAtMs] on purpose — re-announcing must not postpone giving up + * for ever. */ + var deadlineFromMs: Long = 0L +} + +/** + * Offer [name] to the laptop and keep at it until the bytes are fetched: the + * send is retried while it can't go out, RE-announced while it has gone out but + * nothing came to collect it, and reported as lost if neither ever happens. + */ +internal suspend fun VortexStack.offerFileToLaptop(token: String, name: String, offer: ByteArray) { + val pending = PendingOffer(token, name, offer, seq = ++offerSeq) + pendingOffers[token] = pending + // First attempt inline: the common case is a live link, where queueing and + // waiting out a tick would add seconds to an otherwise instant share. + if (!tryDeliverOffer(pending)) { + Log.w( + VortexStack.TAG, + "file offer for '$name' couldn't go out (BLE link down?); retrying", + ) + // Say so ONCE per outage, not once per file: the wait that follows is + // the whole complaint. A share can otherwise sit silent for a minute + // (observed: 29 s for the BLE link to come back, then another 30 s for + // a dropped offer to be re-announced) with nothing on screen since the + // share sheet's "Sending…". + if (!offerUnreachableToasted) { + offerUnreachableToasted = true + toastOffer("Laptop unreachable — keeping the file(s) queued") + } + } + startOfferWatchdog() +} + +/** Push [pending]'s OFFER frame to every trusted peer. Returns true when the + * BLE stack took it, and marks it sent + warms the LAN path. */ +private suspend fun VortexStack.tryDeliverOffer(pending: PendingOffer): Boolean { + pending.attempts++ + val server = gattServer ?: return false + var delivered = false + for (peer in peerStore.list()) { + if (server.sendClipboardImageOfferEncrypted(peer.peerStaticPub, pending.offer)) { + delivered = true + } + } + if (!delivered) return false + val now = android.os.SystemClock.elapsedRealtime() + pending.lastSentAtMs = now + if (pending.deadlineFromMs == 0L) pending.deadlineFromMs = now + // Link is back — arm the "unreachable" notice again for the next outage. + offerUnreachableToasted = false + Log.i( + VortexStack.TAG, + "file offer for '${pending.name}' sent (attempt ${pending.attempts})", + ) + // PACE the burst, as every other bulk BLE path here does. A share of N + // files fires N offers back-to-back, and unpaced they overrun the notify + // queue: three offers 4-9 ms apart cost one of them outright (the laptop + // logged `resynced past dropped BLE frame(s) dropped=1` and queued 2 of 3 + // files) while the phone believed all three had landed. + kotlinx.coroutines.delay(OFFER_PACING_MS) + scheduleLanWarm() + return true +} + +/** + * Warm the LAN path for the imminent pull, once the offer burst has settled. + * + * The laptop has to REACH us, once per queued file, and two things stop it: + * while BLE is up we release the multicast lock (so mDNS goes unanswered and + * its cached IP — stale after any DHCP renew — is its only guess), and the + * Wi-Fi radio parks between its rounds. So hold the radio + mDNS open and + * re-announce, and push our AppState over BLE too: it carries our live + * `wifi_ip`, which repoints the laptop's cache with no mDNS involved at all. + * + * DEFERRED and debounced, because doing this inline per offer put an NSD + * re-announce and a STATE notify between offer 1 and offer 2 — and that notify + * is what cost us offer 1 (observed: sent at .474, `LAN hot` at .538, offer 2 + * at .549, and the laptop only ever saw offers 2 and 3). Pacing the offers + * against each other is pointless if something else cuts in line. + */ +private fun VortexStack.scheduleLanWarm() { + lanWarmJob?.cancel() + lanWarmJob = scope.launch { + kotlinx.coroutines.delay(LAN_WARM_SETTLE_MS) + // Re-announcing costs an NSD round-trip, so only the first offer of a + // batch does it; later ones just extend the window. + if (lanServer?.keepLanHot() == true) { + // Skip the redundant push right after a reconnect: the BLE + // re-subscribe handler has already sent one, and a second would be + // one more frame competing with the offers we just queued. + if (android.os.SystemClock.elapsedRealtime() - lastBleStatePushAtMs + >= STATE_PUSH_DEDUP_MS + ) { + pushStateViaBle() + } + } + } +} + +/** The laptop served itself the blob for [token] over LAN — the offer did its + * job. Wired to `LanServer.onFileServed`. */ +internal fun VortexStack.noteFileServed(token: String) { + val done = pendingOffers.remove(token) ?: return + Log.i(VortexStack.TAG, "file '${done.name}' fetched by the laptop") + // The one unambiguous "it worked" moment on this device: the laptop has the + // bytes. Per file rather than per batch, so a slow batch shows progress as + // it goes instead of one summary at the end. + toastOffer("File sent: ${done.name}") + // SLIDING deadline, like the daemon's bulk-sync idle budget: the laptop + // pulls one file per heartbeat round, so a big batch's last offer can + // legitimately wait many minutes for its turn. A fetch anywhere in the + // batch proves the link is working — restart the clock on the rest rather + // than reporting a failure for files that are simply still queued. + val now = android.os.SystemClock.elapsedRealtime() + for (still in pendingOffers.values) { + if (still.deadlineFromMs != 0L) still.deadlineFromMs = now + } + // Anything the laptop skipped over never reached it: re-announce at once + // rather than waiting out the resend timer. Clearing `lastSentAtMs` is the + // honest record — as far as the laptop is concerned this offer never + // happened — and puts it back in the DELIVER path on the next tick. + val dropped = offersPresumedDropped(pendingOffers.values, done.seq) + for (lost in dropped) { + lost.lastSentAtMs = 0L + // Reset the delivery budget too. It exists to bound a link that won't + // carry the offer at all, and we have just PROVED this one carries — + // the laptop fetched a file. Without this, an offer dropped after a + // long outage (say 19 of 20 attempts spent) would be declared lost on + // the spot instead of getting the one re-announce it needs. + lost.attempts = 0 + Log.w( + VortexStack.TAG, + "offer for '${lost.name}' was dropped in flight (the laptop fetched a " + + "later one first); re-announcing now", + ) + } + if (dropped.isNotEmpty()) kickOfferRetry() +} + +/** + * Offers we can PROVE the laptop never received, given that it just fetched + * [fetchedSeq]. Its pull queue is FIFO in offer-arrival order, so anything + * queued before the file it just took would have been fetched first — an + * earlier offer still sitting here was dropped in flight, not merely waiting + * its turn. + * + * This is the only way the phone can tell: the BLE notify was accepted locally, + * so nothing on this side reports the loss. Without it, the file waits out + * [OFFER_RESEND_MS] with the user watching nothing happen (observed: 2 of 3 + * files arrived in 3 s, the third took another 30 s for no reason but the + * timer). + */ +internal fun offersPresumedDropped( + pending: Collection, + fetchedSeq: Long, +): List = pending.filter { it.lastSentAtMs != 0L && it.seq < fetchedSeq } + +/** Wake the watchdog now instead of at its next tick — called when the BLE link + * comes back, which is exactly when a queued offer can finally go out. */ +internal fun VortexStack.kickOfferRetry() { + if (pendingOffers.isEmpty()) return + offerRetryKick.trySend(Unit) +} + +/** Start the watchdog if it isn't already running. Idempotent: one loop drains + * the whole map, and it exits when the map empties. */ +private fun VortexStack.startOfferWatchdog() { + if (offerRetryJob?.isActive == true) return + offerRetryJob = scope.launch { offerWatchdog() } +} + +/** + * Retry undelivered offers, time out delivered-but-unfetched ones, and toast + * whatever ends up nowhere. Runs only while offers are outstanding. + */ +private suspend fun VortexStack.offerWatchdog() { + while (pendingOffers.isNotEmpty()) { + withTimeoutOrNull(OFFER_RETRY_TICK_MS) { offerRetryKick.receive() } + val now = android.os.SystemClock.elapsedRealtime() + val lost = mutableListOf() + // Snapshot the values: `tryDeliverOffer` and `noteFileServed` both + // mutate the map while we walk it. + for (pending in pendingOffers.values.toList()) { + // Send first when it's due — `tryDeliverOffer` bumps `attempts`, so + // the second verdict sees the budget this attempt just consumed. + if (offerVerdict(pending, now) == OfferVerdict.DELIVER && tryDeliverOffer(pending)) { + continue + } + if (offerVerdict(pending, now) != OfferVerdict.GIVE_UP) continue + pendingOffers.remove(pending.token) + lost += pending.name + Log.w(VortexStack.TAG, "giving up on '${pending.name}': ${giveUpReason(pending)}") + } + if (lost.isNotEmpty()) toastOffersLost(lost) + } +} + +/** Tell the user, on this phone, that [lost] never made it. The log carries the + * reason; this only has to stop the share looking like it worked. */ +private fun VortexStack.toastOffersLost(lost: List) { + toastOffer( + if (lost.size == 1) { + "Laptop didn't get '${lost.first()}'" + } else { + "Laptop didn't get ${lost.size} files" + }, + ) +} + +/** Show [msg] on this phone. Transfer feedback only — the share leaves the + * device and nothing else here reports on it. */ +private fun VortexStack.toastOffer(msg: String) { + android.os.Handler(android.os.Looper.getMainLooper()).post { + try { + // English only, like the cast-failure toast: `ui/Strings.kt`'s + // `str()` is @Composable and unavailable from a service. + android.widget.Toast.makeText(ctx, msg, android.widget.Toast.LENGTH_LONG).show() + } catch (t: Throwable) { + // Best-effort (some ROMs suppress background toasts); the log line + // above is the durable record. + Log.w(VortexStack.TAG, "offer-failure toast suppressed: ${t.message}") + } + } +} + +/** What the watchdog should do with an offer right now. */ +internal enum class OfferVerdict { + /** Not delivered yet and still within budget — (re)send the OFFER frame. */ + DELIVER, + + /** Delivered; the laptop still has time to fetch the blob. */ + WAIT, + + /** Out of delivery attempts, or out of patience waiting for the pull. */ + GIVE_UP, +} + +/** + * The whole give-up policy, kept pure so it can be tested without a BLE stack + * or a clock: an offer that was never delivered gets [OFFER_MAX_ATTEMPTS] + * tries, and one that was gets [OFFER_PULL_GRACE_MS] from the batch's last + * progress to actually be fetched. + */ +internal fun offerVerdict(pending: PendingOffer, nowMs: Long): OfferVerdict = when { + pending.lastSentAtMs == 0L -> + if (pending.attempts >= OFFER_MAX_ATTEMPTS) OfferVerdict.GIVE_UP else OfferVerdict.DELIVER + nowMs - pending.deadlineFromMs >= OFFER_PULL_GRACE_MS -> OfferVerdict.GIVE_UP + // Sent, still unfetched: the notify may have been dropped in flight, which + // no amount of waiting recovers. Re-announce — the laptop dedups by token, + // so a duplicate that DID arrive costs nothing. + nowMs - pending.lastSentAtMs >= OFFER_RESEND_MS -> OfferVerdict.DELIVER + else -> OfferVerdict.WAIT +} + +/** Why we stopped tracking [pending] — the log's half of the toast. */ +internal fun giveUpReason(pending: PendingOffer): String = + if (pending.lastSentAtMs == 0L) { + "the offer never reached the laptop in ${pending.attempts} attempts" + } else { + "the laptop accepted the offer but never fetched it " + + "(asleep, no LAN route, or declined)" + } + +/** How often the watchdog re-tries delivery and re-checks the pull deadline. */ +internal const val OFFER_RETRY_TICK_MS = 3_000L + +/** Delivery attempts before giving up — ~1 min at [OFFER_RETRY_TICK_MS], which + * covers a BLE reconnect (observed: up to ~1m45s with an adapter power-cycle + * on the laptop, so a walk-away is still reported rather than waited out). */ +internal const val OFFER_MAX_ATTEMPTS = 20 + +/** How long the laptop gets to actually FETCH a delivered offer, measured from + * the last progress anywhere in the batch (see [noteFileServed]). Generous: it + * may have to re-find us on the LAN, and with consent prompts enabled a human + * has to click Accept (that banner itself times out at 45 s). */ +internal const val OFFER_PULL_GRACE_MS = 120_000L + +/** Gap between consecutive offer notifies. Wider than the 12-20 ms the chunk + * streams use, because offers are few (one per shared file, so ~0.6 s even for + * a ten-file share) and they go out at the worst possible moment: a BLE + * reconnect, where the state push, notes sync, icon re-send, live activities + * and companion mirrors all queue at once. The drop that prompted this had + * offers only 4-9 ms apart, so "no pacing at all" was well inside the danger + * zone; the re-announce below is what covers the rest. */ +internal const val OFFER_PACING_MS = 60L + +/** How long a sent-but-unfetched offer waits before being re-announced. Covers + * a notify dropped in flight, which the phone cannot otherwise detect. Long + * enough that a laptop working through a batch (one file per heartbeat round) + * isn't pestered mid-pull. */ +internal const val OFFER_RESEND_MS = 30_000L + +/** How long after the last offer went out to warm the LAN path. Long enough + * for the offer burst to drain the notify queue first. */ +internal const val LAN_WARM_SETTLE_MS = 250L + +/** A BLE AppState push within this window of the last one is redundant — the + * reconnect handler already sent it, and it would only crowd the offers. */ +internal const val STATE_PUSH_DEDUP_MS = 5_000L + +/** Conflated so a burst of BLE re-subscribes collapses into one wake-up. */ +internal fun newOfferRetryKick(): Channel = Channel(Channel.CONFLATED) diff --git a/android/app/src/test/java/com/vortex/a3/service/OfferVerdictTest.kt b/android/app/src/test/java/com/vortex/a3/service/OfferVerdictTest.kt new file mode 100644 index 0000000..21f7e1d --- /dev/null +++ b/android/app/src/test/java/com/vortex/a3/service/OfferVerdictTest.kt @@ -0,0 +1,134 @@ +package com.vortex.a3.service + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +/** + * The outgoing-file-offer policy. A phone→laptop share is announced over BLE + * and pulled over LAN, and every step can fail silently — including a notify + * the local stack accepts and the link then drops. This decides when to send, + * when to re-announce, and when to admit defeat; getting it wrong is + * user-visible both ways (a batch still in the queue reported as lost, or a + * share that went nowhere staying silent for ever). + */ +class OfferVerdictTest { + + /** [lastSentAtMs] doubles as "has it ever gone out"; [deadlineFromMs] is + * the give-up clock, which callers slide on batch progress. */ + private fun offer( + attempts: Int = 0, + lastSentAtMs: Long = 0L, + deadlineFromMs: Long = lastSentAtMs, + ) = PendingOffer("tok", "file.bin", ByteArray(0)).also { + it.attempts = attempts + it.lastSentAtMs = lastSentAtMs + it.deadlineFromMs = deadlineFromMs + } + + @Test + fun `an offer that never went out is retried until its budget runs out`() { + assertEquals(OfferVerdict.DELIVER, offerVerdict(offer(attempts = 0), 0L)) + assertEquals( + OfferVerdict.DELIVER, + offerVerdict(offer(attempts = OFFER_MAX_ATTEMPTS - 1), 0L), + ) + assertEquals( + OfferVerdict.GIVE_UP, + offerVerdict(offer(attempts = OFFER_MAX_ATTEMPTS), 0L), + ) + } + + @Test + fun `a sent offer waits for the pull`() { + val sent = offer(attempts = 1, lastSentAtMs = 1_000L) + assertEquals(OfferVerdict.WAIT, offerVerdict(sent, 1_000L)) + assertEquals(OfferVerdict.WAIT, offerVerdict(sent, 1_000L + OFFER_RESEND_MS - 1)) + } + + @Test + fun `a sent offer nothing came to collect is re-announced`() { + // The dropped-notify case: the local stack took the frame, the link + // lost it, so the laptop never knew to pull. Waiting can't fix that. + val sent = offer(attempts = 1, lastSentAtMs = 1_000L) + assertEquals(OfferVerdict.DELIVER, offerVerdict(sent, 1_000L + OFFER_RESEND_MS)) + } + + @Test + fun `re-announcing does not postpone giving up`() { + // Sent repeatedly (so `lastSentAtMs` keeps moving) while the give-up + // clock stays put: the deadline must still land. + val stubborn = offer(attempts = 5, lastSentAtMs = 119_000L, deadlineFromMs = 0L) + assertEquals(OfferVerdict.GIVE_UP, offerVerdict(stubborn, OFFER_PULL_GRACE_MS)) + } + + @Test + fun `progress in the batch slides the deadline and buys more time`() { + // What `noteFileServed` does to the rest of the batch: a fetch anywhere + // proves the link works, so the others aren't declared lost while they + // wait their turn (the laptop pulls one file per heartbeat round). + val queued = offer(attempts = 1, lastSentAtMs = 1_000L, deadlineFromMs = 1_000L) + val past = 1_000L + OFFER_PULL_GRACE_MS + assertEquals(OfferVerdict.GIVE_UP, offerVerdict(queued, past)) + queued.deadlineFromMs = past + queued.lastSentAtMs = past + assertEquals(OfferVerdict.WAIT, offerVerdict(queued, past)) + } + + @Test + fun `send attempts stop mattering once it has gone out`() { + val sent = offer(attempts = OFFER_MAX_ATTEMPTS, lastSentAtMs = 500L) + assertEquals(OfferVerdict.WAIT, offerVerdict(sent, 500L)) + } + + @Test + fun `the give-up reason distinguishes never-sent from never-fetched`() { + assertEquals( + "the offer never reached the laptop in 20 attempts", + giveUpReason(offer(attempts = 20)), + ) + assertTrue( + giveUpReason(offer(attempts = 1, lastSentAtMs = 5L)).contains("never fetched it"), + ) + } +} + +/** + * Spotting an offer the BLE link dropped in flight. The phone cannot see the + * loss directly — its own stack accepted the notify — so the only evidence is + * the laptop fetching a LATER offer while an earlier one is still outstanding, + * its pull queue being FIFO in arrival order. + */ +class PresumedDroppedTest { + + private fun offer(seq: Long, lastSentAtMs: Long) = + PendingOffer("tok$seq", "file$seq.bin", ByteArray(0), seq = seq).also { + it.lastSentAtMs = lastSentAtMs + it.deadlineFromMs = lastSentAtMs + } + + @Test + fun `an earlier offer skipped over was dropped in flight`() { + val first = offer(seq = 1, lastSentAtMs = 100L) + val third = offer(seq = 3, lastSentAtMs = 100L) + // The laptop fetched #2, so #1 would have come first had it arrived. + val dropped = offersPresumedDropped(listOf(first, third), fetchedSeq = 2) + assertEquals(listOf(first), dropped) + } + + @Test + fun `a later offer is simply waiting its turn`() { + // One file per heartbeat round: #3 outstanding after #2 was fetched is + // the normal case and must NOT be re-announced. + val third = offer(seq = 3, lastSentAtMs = 100L) + assertEquals(emptyList(), offersPresumedDropped(listOf(third), 2)) + } + + @Test + fun `an offer that never went out is left to the send retry`() { + // Nothing to infer: it isn't missing, it hasn't been sent. The delivery + // budget owns this one. + val unsent = offer(seq = 1, lastSentAtMs = 0L) + assertEquals(emptyList(), offersPresumedDropped(listOf(unsent), 2)) + } +} diff --git a/linux/daemon/src/core/lan/tcp_client.rs b/linux/daemon/src/core/lan/tcp_client.rs index 39e09b8..66e6c3d 100644 --- a/linux/daemon/src/core/lan/tcp_client.rs +++ b/linux/daemon/src/core/lan/tcp_client.rs @@ -29,6 +29,47 @@ pub struct LanReconnectOutcome { /// stale: (frame type, reassembled JSON bytes). Empty when everything /// matched, no request was made, or the peer predates BULK_SYNC. pub bulk: Vec<(u8, Vec)>, + /// Per-dataset outcome from the bulk-sync DONE frame, e.g. + /// `{"contacts":"match","clipboard_file":"nomatch"}`. `None` when no + /// request was made or the peer never sent a done frame (predates + /// BULK_SYNC, or the exchange broke off). + /// + /// Datasets report their own failures HERE and nowhere else: a "nomatch" + /// looks exactly like "nothing to send" in [`bulk`], so a caller holding a + /// pull request open (the instant-share file queue) needs this to learn + /// that what it asked for is never coming. + pub bulk_status: Option, +} + +/// The bulk-sync done frame's per-dataset outcome map. +#[derive(Debug, Clone, Default)] +pub struct BulkStatus(std::collections::HashMap); + +impl BulkStatus { + /// Parse the done frame's JSON body. Non-string values are ignored rather + /// than failing the whole map — one odd field must not blind the caller to + /// the rest. + fn parse(json: &[u8]) -> Option { + let v: serde_json::Value = serde_json::from_slice(json).ok()?; + let obj = v.as_object()?; + Some(Self( + obj.iter() + .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string()))) + .collect(), + )) + } + + /// What the phone reported for `dataset` ("sent" / "match" / "nomatch" / + /// "error" / "unknown"), or `None` if it said nothing about it. + pub fn get(&self, dataset: &str) -> Option<&str> { + self.0.get(dataset).map(String::as_str) + } + + /// True when the phone answered about `dataset` and it was NOT served — + /// i.e. asking again is pointless until something changes on its side. + pub fn unservable(&self, dataset: &str) -> bool { + matches!(self.get(dataset), Some(s) if s != "sent") + } } #[derive(Debug)] @@ -268,9 +309,13 @@ pub async fn run_lan_reconnect( // socket instead of a BLE notify burst. Skipped when the app-state // exchange already failed (transport unhealthy) or no request was made. let mut bulk: Vec<(u8, Vec)> = Vec::new(); + let mut bulk_status: Option = None; if let (Some(req), Some(_)) = (bulk_request, peer_state.as_ref()) { match exchange_bulk(&mut stream, &mut transport, req, wait_per_step).await { - Ok(datasets) => bulk = datasets, + Ok((datasets, status)) => { + bulk = datasets; + bulk_status = status; + } Err(e) => tracing::warn!("bulk-sync exchange failed: {e}"), } } @@ -296,6 +341,7 @@ pub async fn run_lan_reconnect( peer_counter, peer_state, bulk, + bulk_status, }) } @@ -416,7 +462,7 @@ async fn exchange_bulk( transport: &mut snow::TransportState, request_json: &str, wait: Duration, -) -> Result)>, LanError> { +) -> Result<(Vec<(u8, Vec)>, Option), LanError> { let plain = request_json.as_bytes(); let mut ct = vec![0u8; plain.len() + 16]; let n = transport.write_message(plain, &mut ct)?; @@ -427,6 +473,7 @@ async fn exchange_bulk( info!("→ bulk-sync request ({} bytes)", plain.len()); let mut out: Vec<(u8, Vec)> = Vec::new(); + let mut status: Option = None; let mut contacts = crate::core::contacts::ContactsAssembler::default(); let mut call_log = crate::core::call_log::CallLogAssembler::default(); let mut sms = crate::core::sms::SmsAssembler::default(); @@ -468,6 +515,10 @@ async fn exchange_bulk( match frame.ty { ty::BULK_SYNC if frame.sub == 0x02 => { info!("← bulk-sync done: {}", String::from_utf8_lossy(&pt)); + status = BulkStatus::parse(&pt); + if status.is_none() { + tracing::warn!("bulk-sync: done frame is not a JSON object; no status"); + } break; } ty::CONTACTS => { @@ -579,6 +630,40 @@ async fn exchange_bulk( } } } - Ok(out) + Ok((out, status)) } + +#[cfg(test)] +mod tests { + use super::BulkStatus; + + #[test] + fn nomatch_is_unservable_and_sent_is_not() { + let s = BulkStatus::parse(br#"{"contacts":"match","clipboard_file":"nomatch"}"#).unwrap(); + assert!(s.unservable("clipboard_file")); + assert_eq!(s.get("contacts"), Some("match")); + + let s = BulkStatus::parse(br#"{"clipboard_file":"sent"}"#).unwrap(); + assert!(!s.unservable("clipboard_file")); + } + + #[test] + fn a_dataset_the_phone_said_nothing_about_is_not_unservable() { + // Never drop a queued pull on silence — only on an explicit answer. + let s = BulkStatus::parse(br#"{"contacts":"match"}"#).unwrap(); + assert!(!s.unservable("clipboard_file")); + assert_eq!(s.get("clipboard_file"), None); + } + + #[test] + fn errors_count_as_unservable_and_junk_parses_to_none() { + let s = BulkStatus::parse(br#"{"clipboard_file":"error"}"#).unwrap(); + assert!(s.unservable("clipboard_file")); + // Non-string values are skipped, not fatal. + let s = BulkStatus::parse(br#"{"clipboard_file":"nomatch","n":7}"#).unwrap(); + assert!(s.unservable("clipboard_file")); + assert!(BulkStatus::parse(b"[]").is_none()); + assert!(BulkStatus::parse(b"not json").is_none()); + } +} diff --git a/linux/ui-tauri/src-tauri/src/clipboard_sync.rs b/linux/ui-tauri/src-tauri/src/clipboard_sync.rs index 996612d..1c09127 100644 --- a/linux/ui-tauri/src-tauri/src/clipboard_sync.rs +++ b/linux/ui-tauri/src-tauri/src/clipboard_sync.rs @@ -430,6 +430,31 @@ async fn flush_file_batch(batch: Vec) { if batch.is_empty() { return; } + // Drop offers we are already holding. The phone RE-ANNOUNCES an offer that + // hasn't been fetched, because a notify its own stack accepted can still be + // dropped in flight and it has no other way to tell. Without this the + // duplicate would queue the same file a second time — a redundant pull and + // a spurious "name (1).ext" beside the real one — and re-prompt for consent + // on a batch the user already accepted. Deduped by content token, so a + // deliberate re-share AFTER the pull completed still goes through (the + // entry is gone from the queue by then). + let batch: Vec = { + let queued: std::collections::HashSet = crate::PENDING_FILE_OFFERS + .get() + .and_then(|m| m.lock().ok().map(|g| g.iter().map(|(t, ..)| t.clone()).collect())) + .unwrap_or_default(); + let mut seen = std::collections::HashSet::new(); + batch + .into_iter() + // `seen` also collapses duplicates WITHIN the batch: a re-announce + // can land inside the same debounce window as the original. + .filter(|o| !queued.contains(&o.token) && seen.insert(o.token.clone())) + .collect() + }; + if batch.is_empty() { + tracing::info!("phone re-announced file offer(s) already queued; ignoring"); + return; + } let count = batch.len(); let total: u64 = batch.iter().map(|o| o.bytes).sum(); let label = if count == 1 { @@ -451,6 +476,7 @@ async fn flush_file_batch(batch: Vec) { } } } + crate::lan::note_queue_progress(); tracing::info!(count, "phone file offer(s) accepted → LAN pull nudged"); if let Some(nudge) = crate::SYNC_NUDGE.get() { nudge.notify_one(); diff --git a/linux/ui-tauri/src-tauri/src/lan.rs b/linux/ui-tauri/src-tauri/src/lan.rs index f24cbbf..400ff26 100644 --- a/linux/ui-tauri/src-tauri/src/lan.rs +++ b/linux/ui-tauri/src-tauri/src/lan.rs @@ -38,6 +38,51 @@ fn last_peer_ip_path() -> Option { Some(p) } +/// Is a phone→laptop file pull waiting on the next heartbeat round? The pull is +/// piggybacked on the heartbeat (one queued file per round), so the tick +/// cadence and the address probe both need to know when the queue is hot — +/// otherwise a half-received batch sits out the idle backoff. +pub(crate) fn files_queued() -> bool { + crate::PENDING_FILE_OFFERS + .get() + .and_then(|m| m.lock().ok().map(|g| !g.is_empty())) + .unwrap_or(false) +} + +/// When the pull queue last MOVED — an offer accepted onto it, or a file pulled +/// off it. See [`file_pull_active`]. +static QUEUE_PROGRESS_AT: std::sync::Mutex> = + std::sync::Mutex::new(None); + +/// A queued file batch can stop draining for reasons a retry will never fix — +/// notably a token the phone has evicted, which it answers `"nomatch"` while +/// the laptop keeps it queued and re-requests it every round. Past this, treat +/// the queue as cold: the brisk cadence is for a batch that is moving. +const QUEUE_STALL_GRACE: Duration = Duration::from_secs(60); + +/// Record that the pull queue moved. Call whenever an entry is pushed or popped. +pub(crate) fn note_queue_progress() { + if let Ok(mut g) = QUEUE_PROGRESS_AT.lock() { + *g = Some(std::time::Instant::now()); + } +} + +/// Is a file pull both queued AND still making progress? This — not bare +/// [`files_queued`] — is what may drive the heartbeat harder: a queue that is +/// permanently stuck must not spin a TCP+IK every 2 s for the rest of the +/// session, which is exactly what the unconditional form would do. +pub(crate) fn file_pull_active() -> bool { + if !files_queued() { + return false; + } + QUEUE_PROGRESS_AT + .lock() + .ok() + .and_then(|g| *g) + .map(|t| t.elapsed() < QUEUE_STALL_GRACE) + .unwrap_or(false) +} + /// Cache the peer IP that just resolved — in memory AND on disk — so after a /// daemon restart the very first heartbeat reuses it instead of the /// (wrong-on-a-shared-network) gateway guess that caused a transient @@ -280,15 +325,32 @@ pub(crate) async fn try_lan_reconnect( match cached { Some(ip) => { let sa = std::net::SocketAddr::new(ip, LAN_DEFAULT_PORT); - match tokio::time::timeout( - Duration::from_secs(2), - tokio::net::TcpStream::connect(sa), - ) - .await - { - Ok(Ok(_probe)) => Some(sa), // reachable — probe socket drops here - _ => None, + // One attempt normally: a dozing Wi-Fi radio can miss a cold + // 2 s connect, and an idle heartbeat gets away with that + // because it ticks again shortly. A heartbeat carrying a + // queued file pull does NOT — losing the probe costs it a 6 s + // mDNS browse plus a failed dial, and four of those in a row + // park the rest of the batch on the idle backoff. So retry + // like `resolve_peer_addr` does when there's work waiting. + let attempts = if file_pull_active() { 3 } else { 1 }; + let mut found = None; + for attempt in 0..attempts { + if attempt > 0 { + tokio::time::sleep(Duration::from_millis(400)).await; + } + if matches!( + tokio::time::timeout( + Duration::from_secs(2), + tokio::net::TcpStream::connect(sa), + ) + .await, + Ok(Ok(_probe)) // reachable — probe socket drops here + ) { + found = Some(sa); + break; + } } + found } None => None, } @@ -460,11 +522,11 @@ pub(crate) async fn try_lan_reconnect( } // Instant-share file pull: request the FRONT queued file this round (the rest // follow on subsequent nudged rounds). - if let Some(token) = crate::PENDING_FILE_OFFERS + let requested_file_token: Option = crate::PENDING_FILE_OFFERS .get() - .and_then(|m| m.lock().ok().and_then(|g| g.front().map(|(t, _, _, _)| t.clone()))) - { - bulk_obj["clipboard_file"] = serde_json::Value::String(token); + .and_then(|m| m.lock().ok().and_then(|g| g.front().map(|(t, _, _, _)| t.clone()))); + if let Some(token) = &requested_file_token { + bulk_obj["clipboard_file"] = serde_json::Value::String(token.clone()); } let bulk_request = bulk_obj.to_string(); match run_lan_reconnect( @@ -539,6 +601,7 @@ pub(crate) async fn try_lan_reconnect( .get() .and_then(|m| m.lock().ok().and_then(|mut g| g.pop_front())); if let Some((_, name, mime, id)) = meta { + note_queue_progress(); match crate::clipboard_sync::apply_synced_file( app, &name, @@ -551,11 +614,7 @@ pub(crate) async fn try_lan_reconnect( None => crate::transfers::fail(id), } } - let more = crate::PENDING_FILE_OFFERS - .get() - .and_then(|m| m.lock().ok().map(|g| !g.is_empty())) - .unwrap_or(false); - if more { + if files_queued() { if let Some(nudge) = crate::SYNC_NUDGE.get() { nudge.notify_one(); } @@ -581,14 +640,53 @@ pub(crate) async fn try_lan_reconnect( } } } + // We asked for a file and the phone said it couldn't serve it — + // its blob store keeps only the last 32, so a token can be + // evicted before we get to it. Nothing will ever arrive for that + // entry: drop it, fail its pill, and move to the next. Left + // queued it would be re-requested on every round for the rest of + // the session, blocking every file behind it (and, on the + // Wi-Fi Direct path, never letting us restore Wi-Fi). + if let Some(req) = &requested_file_token { + if outcome + .bulk_status + .as_ref() + .is_some_and(|s| s.unservable("clipboard_file")) + { + // Pop only if the front is still the entry we asked + // about, so a batch accepted mid-round is never dropped. + let dead = crate::PENDING_FILE_OFFERS.get().and_then(|m| { + m.lock().ok().and_then(|mut g| { + let front_matches = + g.front().is_some_and(|(t, _, _, _)| t == req); + if front_matches { g.pop_front() } else { None } + }) + }); + if let Some((_, name, _, id)) = dead { + note_queue_progress(); + crate::transfers::fail(id); + tracing::warn!( + name = %name, + status = outcome + .bulk_status + .as_ref() + .and_then(|s| s.get("clipboard_file")) + .unwrap_or("?"), + "phone can no longer serve this file (token evicted?); \ + dropping it from the pull queue" + ); + if files_queued() { + if let Some(nudge) = crate::SYNC_NUDGE.get() { + nudge.notify_one(); + } + } + } + } + } // Wi-Fi Direct: once every queued file is pulled over the group // link, hop back to the normal Wi-Fi; otherwise pull the next now. if wd_active() { - let empty = crate::PENDING_FILE_OFFERS - .get() - .and_then(|m| m.lock().ok().map(|g| g.is_empty())) - .unwrap_or(true); - if empty { + if !files_queued() { tracing::info!("Wi-Fi Direct: all files pulled → restoring Wi-Fi"); restore_wifi(app).await; } else if let Some(n) = crate::SYNC_NUDGE.get() { @@ -1080,6 +1178,21 @@ pub(crate) fn spawn_heartbeat( Duration::from_secs(2) } else if had_trust && !lan_synced && consec_lan_fail <= 3 { Duration::from_secs(2) + } else if file_pull_active() { + // A phone file batch is mid-pull. One file rides each + // round, so the cadence IS the transfer rate here, and + // the idle branches below are ruinous: with BLE live + // the 240 s tick parked a five-file share for 2m34s + // (then the whole batch landed in 8 s once a round + // finally ran). Stay brisk while there is queued work, + // easing off only after the phone has been unreachable + // for a while so an offer it can no longer serve does + // not spin a TCP+IK every 2 s forever. + if consec_lan_fail <= 15 { + Duration::from_secs(2) + } else { + Duration::from_secs(12) + } } else if auto_ble_writers.lock().await.is_empty() { Duration::from_secs(12) } else {