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 {