mirror of
https://github.com/screentinker/screentinker.git
synced 2026-08-13 22:03:13 -06:00
feat(apk): v4 liveness contract + caching-cluster fix + reconnect-safe downloads
APK v4 — client-only (Android player). Brings the reference client up to the locked v4 liveness contract and fixes the "stuck downloading / offline in CMS" caching bug: - v4 liveness watchdog (LivenessWatchdog): half-open detection via server-silence, arm-ONLY-after device:heartbeat-ack (degrade-safe), 45s±10s jittered threshold, exp backoff 1/2/4/8/16→30s ±20%, no-poll; reconnect delegates to the #148 ConnectionGuard (teardown-before-reopen, single socket). - Caching two-root fix: callTimeout + .part+Content-Length integrity + atomic swap (CacheValidation), onPlayerError advance, re-ack cached content + reconnect re-ack. - Screen resilience (PlaylistSelection): a pending/failed download never blanks the screen — keep-current, swap only fully-valid content. - Reconnect-safe background downloads (DownloadCoordinator): single-flight per contentId + bounded pool + failure backoff + cancellation; a reconnect mid-fetch can't orphan/duplicate/storm. Refuse 206 partials. - v4 client identity block on register (client_type/version/platform/contract_version). Tests: 52 JVM unit tests (watchdog, cache validation, reproduce-then-prove download stall/truncation/reconnect-mid-download, screen selection, assembly soak). Depends on the server device:heartbeat-ack (core pass) — degrade-safe until then. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
ScreenTinker
parent
5c3d1a18c5
commit
57cdaf7e4e
|
|
@ -46,6 +46,13 @@ android {
|
|||
kotlinOptions {
|
||||
jvmTarget = "17"
|
||||
}
|
||||
|
||||
testOptions {
|
||||
// Let JVM unit tests exercise Android-dependent code paths (e.g. ContentCache's real
|
||||
// download logic, which logs via android.util.Log) without Robolectric — stubbed Android
|
||||
// APIs return defaults instead of throwing "not mocked".
|
||||
unitTests.isReturnDefaultValues = true
|
||||
}
|
||||
}
|
||||
|
||||
dependencies {
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ class MainActivity : AppCompatActivity() {
|
|||
|
||||
private lateinit var config: ServerConfig
|
||||
private lateinit var contentCache: ContentCache
|
||||
private lateinit var downloadCoordinator: com.remotedisplay.player.data.DownloadCoordinator
|
||||
private lateinit var screenshotCapture: ScreenshotCapture
|
||||
private lateinit var touchInjector: TouchInjector
|
||||
|
||||
|
|
@ -124,6 +125,13 @@ class MainActivity : AppCompatActivity() {
|
|||
window.addFlags(WindowManager.LayoutParams.FLAG_KEEP_SCREEN_ON)
|
||||
|
||||
contentCache = ContentCache(this)
|
||||
// Coordinated background downloads (single-flight + bounded pool + backoff) — reconnect-safe.
|
||||
downloadCoordinator = com.remotedisplay.player.data.DownloadCoordinator(
|
||||
cache = contentCache,
|
||||
serverUrl = { config.serverUrl },
|
||||
socketAlive = { wsService?.isConnected() == true },
|
||||
onAck = { cid, status -> ackContentOnce(cid, status) }
|
||||
)
|
||||
screenshotCapture = ScreenshotCapture()
|
||||
touchInjector = TouchInjector()
|
||||
|
||||
|
|
@ -171,8 +179,17 @@ class MainActivity : AppCompatActivity() {
|
|||
// #74/#75: clear the last frame when going idle (else a now-filtered item lingers on screen)
|
||||
onPlaylistEmpty = { if (::mediaPlayer.isInitialized) mediaPlayer.stop(); showStatus(getString(R.string.waiting_for_content)) },
|
||||
onRequestRefresh = { wsService?.requestPlaylistRefresh() },
|
||||
onNothingScheduled = { if (::mediaPlayer.isInitialized) mediaPlayer.stop(); showStatus(getString(R.string.nothing_scheduled)) }
|
||||
onNothingScheduled = { if (::mediaPlayer.isInitialized) mediaPlayer.stop(); showStatus(getString(R.string.nothing_scheduled)) },
|
||||
// Screen-resilience: the defined "waiting for content" state — ONLY on a fresh device
|
||||
// with nothing to show yet (never while content is on screen; that path keeps current).
|
||||
onWaitingForContent = { if (::mediaPlayer.isInitialized) mediaPlayer.stop(); showStatus(getString(R.string.waiting_for_content)) }
|
||||
)
|
||||
// Screen-resilience: an item is playable only when its content is actually available —
|
||||
// a widget, a remote stream, or a fully-downloaded local file. A not-yet/failed download is
|
||||
// skipped (kept in the background) instead of blanking the screen on a loading state.
|
||||
playlistController.setContentReadyCheck { item ->
|
||||
item.isWidget || item.isRemote || contentCache.isContentCached(item.contentId)
|
||||
}
|
||||
|
||||
// Setup media player
|
||||
mediaPlayer = MediaPlayerManager(
|
||||
|
|
@ -469,31 +486,23 @@ class MainActivity : AppCompatActivity() {
|
|||
|
||||
// Skip remote URL content - it streams directly
|
||||
if (!remoteUrl.isNullOrEmpty()) {
|
||||
wsService?.sendContentAck(contentId, "ready")
|
||||
ackContentOnce(contentId, "ready")
|
||||
continue
|
||||
}
|
||||
|
||||
if (!contentCache.isContentCached(contentId)) {
|
||||
Log.i("MainActivity", "Downloading content: $filename")
|
||||
var downloaded = false
|
||||
for (attempt in 1..3) {
|
||||
val file = contentCache.downloadContent(config.serverUrl, contentId, filename)
|
||||
if (file != null) {
|
||||
wsService?.sendContentAck(contentId, "ready")
|
||||
downloaded = true
|
||||
break
|
||||
}
|
||||
Log.w("MainActivity", "Download attempt $attempt failed for $filename")
|
||||
if (attempt < 3) Thread.sleep(2000L * attempt)
|
||||
}
|
||||
if (!downloaded) wsService?.sendContentAck(contentId, "failed")
|
||||
}
|
||||
// Background download is now COORDINATED: single-flight per contentId + a bounded
|
||||
// pool + failure backoff. So a watchdog/ConnectionGuard reconnect mid-fetch can't
|
||||
// spawn a duplicate racing the .part, orphan/pile up threads, or storm a failing
|
||||
// URL. ensure() is non-blocking and idempotent: it re-acks cached content (SEED-A),
|
||||
// defers when the socket is down (watchdog owns recovery), respects backoff, and
|
||||
// downloads at most once. It acks ready/failed itself (deduped via onAck).
|
||||
downloadCoordinator.ensure(contentId, filename)
|
||||
}
|
||||
|
||||
// Start or resume playback after downloads complete — but ONLY in
|
||||
// single-zone/fullscreen mode. In multi-zone, ZoneManager drives each
|
||||
// zone; restarting the fullscreen controller here made it keep playing
|
||||
// items behind the zones (wasted work + phantom audio for videos).
|
||||
// Start/resume playback immediately — do NOT wait on downloads (they're async now).
|
||||
// Screen-resilience plays whatever is cached and skips not-yet-ready items; the 3s
|
||||
// recheck swaps new content in once its download completes. Single-zone only; in
|
||||
// multi-zone, ZoneManager drives each zone.
|
||||
handler.post {
|
||||
if (zoneManager?.hasZones() != true) playlistController.startIfNeeded()
|
||||
}
|
||||
|
|
@ -505,6 +514,7 @@ class MainActivity : AppCompatActivity() {
|
|||
}
|
||||
|
||||
wsService?.onContentDelete = { contentId ->
|
||||
downloadCoordinator.forget(contentId) // drop in-flight/backoff state so a re-add re-downloads
|
||||
contentCache.deleteContent(contentId)
|
||||
playlistController.removeContent(contentId)
|
||||
// Update cached playlist to reflect deletion
|
||||
|
|
@ -631,6 +641,9 @@ class MainActivity : AppCompatActivity() {
|
|||
|
||||
wsService?.onRegistered = { _ ->
|
||||
hideStatus()
|
||||
// Root-2 (SEED-B): a disconnect may have dropped in-flight content-acks. Clear the
|
||||
// de-dup set so the next playlist-update re-acks all content and the CMS re-syncs.
|
||||
ackedContent.clear()
|
||||
}
|
||||
|
||||
wsService?.onUnpaired = {
|
||||
|
|
@ -645,6 +658,20 @@ class MainActivity : AppCompatActivity() {
|
|||
}
|
||||
}
|
||||
|
||||
// Root-2 content-ack de-dup. Re-acking content state (SEED-A) fixes the CMS "stuck downloading"
|
||||
// label, but we must not re-ack the same (content,status) every 60s playlist refresh. This set
|
||||
// is cleared on each (re)registration (see onRegistered) so a reconnect re-acks everything the
|
||||
// server may have missed while we were disconnected (SEED-B), but is quiet within a session.
|
||||
private val ackedContent = java.util.Collections.synchronizedSet(HashSet<String>())
|
||||
|
||||
private fun ackContentOnce(contentId: String, status: String) {
|
||||
if (ackedContent.add("$contentId:$status")) {
|
||||
// a status change for this content supersedes the opposite one
|
||||
ackedContent.remove("$contentId:${if (status == "ready") "failed" else "ready"}")
|
||||
wsService?.sendContentAck(contentId, status)
|
||||
}
|
||||
}
|
||||
|
||||
private fun playItem(item: PlaylistItem) {
|
||||
hideStatus()
|
||||
com.remotedisplay.player.util.DebugLog.i("Player", "playItem: ${item.filename} mime=${item.mimeType} widget=${item.widgetId ?: "-"} zone=fullscreen")
|
||||
|
|
@ -680,22 +707,17 @@ class MainActivity : AppCompatActivity() {
|
|||
return
|
||||
}
|
||||
|
||||
// Local content - download if not cached
|
||||
// Local content - play from cache. Screen-resilience: the controller only advances to
|
||||
// items whose content is READY, so reaching here uncached is a rare race (e.g. the file
|
||||
// was evicted between selection and play). NEVER blank or show a "Downloading…" screen —
|
||||
// keep whatever is on screen and move on; the background download loop (onPlaylistUpdate)
|
||||
// fetches it and it plays once fully + validly downloaded. Content update is a BACKGROUND
|
||||
// operation; we only ever SWAP to fully-downloaded content.
|
||||
val file = contentCache.getCachedFile(item.contentId)
|
||||
if (file == null) {
|
||||
Log.w("MainActivity", "Content not cached: ${item.contentId}, downloading...")
|
||||
showStatus("Downloading ${item.filename}...")
|
||||
thread {
|
||||
val downloaded = contentCache.downloadContent(config.serverUrl, item.contentId, item.filename)
|
||||
handler.post {
|
||||
if (downloaded != null) {
|
||||
playFile(item, downloaded)
|
||||
} else {
|
||||
showStatus("Download failed: ${item.filename}")
|
||||
handler.postDelayed({ playlistController.next() }, 3000)
|
||||
}
|
||||
}
|
||||
}
|
||||
Log.i("MainActivity", "Content not ready at play time (${item.filename}) — keeping screen, advancing (bg download continues)")
|
||||
downloadCoordinator.ensure(item.contentId, item.filename) // ensure it's being fetched (single-flight)
|
||||
handler.post { playlistController.next() }
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -812,6 +834,7 @@ class MainActivity : AppCompatActivity() {
|
|||
|
||||
override fun onDestroy() {
|
||||
remoteStreaming = false
|
||||
if (::downloadCoordinator.isInitialized) downloadCoordinator.shutdown() // cancel in-flight downloads (no orphan/leak)
|
||||
zoneManager?.cleanup()
|
||||
if (::pipOverlay.isInitialized) pipOverlay.clear(null) // #109: tear down overlay WebView
|
||||
if (::mediaPlayer.isInitialized) {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,20 @@
|
|||
package com.remotedisplay.player.data
|
||||
|
||||
/**
|
||||
* Root-2 caching fix — the pure, unit-testable "is this download complete?" rule (no Android deps,
|
||||
* same pattern as ConnectionGuard / OtaThrottle). [ContentCache] is the imperative shell (OkHttp +
|
||||
* filesystem); this owns the integrity decision so a truncated/partial body is never promoted to
|
||||
* the cache dir and later served as if it were a whole file (which previously wedged playback on a
|
||||
* corrupt asset with no error path).
|
||||
*/
|
||||
object CacheValidation {
|
||||
/**
|
||||
* Complete iff we wrote at least one byte AND — when the server declared a Content-Length
|
||||
* ([expectedBytes] > 0) — we wrote exactly that many. A truncated body
|
||||
* (bytesWritten < expectedBytes) is INCOMPLETE and must be discarded + re-fetched. When the
|
||||
* length is unknown (chunked / -1), fall back to ">0 bytes" (an interrupted copy throws and is
|
||||
* discarded upstream, so a silent zero-length truncation is the only residual gap).
|
||||
*/
|
||||
fun isComplete(bytesWritten: Long, expectedBytes: Long): Boolean =
|
||||
bytesWritten > 0 && (expectedBytes <= 0 || bytesWritten == expectedBytes)
|
||||
}
|
||||
|
|
@ -8,16 +8,31 @@ import java.io.File
|
|||
import java.io.FileOutputStream
|
||||
import java.util.concurrent.TimeUnit
|
||||
|
||||
class ContentCache(private val context: Context) {
|
||||
|
||||
private val cacheDir = File(context.filesDir, "content_cache").also { it.mkdirs() }
|
||||
private val client = OkHttpClient.Builder()
|
||||
.connectTimeout(30, TimeUnit.SECONDS)
|
||||
.readTimeout(5, TimeUnit.MINUTES)
|
||||
.build()
|
||||
/**
|
||||
* Root-2 caching fixes vs the "stuck downloading / frozen" bug:
|
||||
* - a hard OVERALL [callTimeout] so a slow-drip/stalled download on a HEALTHY socket can't hang
|
||||
* forever (the old client only had a per-read timeout, which a trickle never trips),
|
||||
* - download to a `.part` temp + integrity-check (Content-Length) via [CacheValidation] + atomic
|
||||
* rename, so a truncated/interrupted body is NEVER promoted to the cache and played as if whole,
|
||||
* - exact-prefix cache lookup that also excludes in-flight `.part` files.
|
||||
*
|
||||
* The primary constructor takes the cache dir + client directly so the real download logic is
|
||||
* unit-testable (see ContentDownloadTest) against a local server without an Android Context; the
|
||||
* [Context] convenience constructor is what the app uses.
|
||||
*/
|
||||
class ContentCache internal constructor(
|
||||
private val cacheDir: File,
|
||||
private val client: OkHttpClient
|
||||
) {
|
||||
constructor(context: Context) : this(
|
||||
File(context.filesDir, "content_cache").also { it.mkdirs() },
|
||||
defaultClient()
|
||||
)
|
||||
|
||||
fun getCachedFile(contentId: String): File? {
|
||||
val files = cacheDir.listFiles { _, name -> name.startsWith(contentId) }
|
||||
// Match "<id>.<ext>" exactly: the trailing dot stops an id that PREFIXES another id from
|
||||
// cross-matching, and `.part` temps (partial/in-flight downloads) are never returned.
|
||||
val files = cacheDir.listFiles { _, name -> name.startsWith("$contentId.") && !name.endsWith(PART_SUFFIX) }
|
||||
return files?.firstOrNull()?.takeIf { it.exists() && it.length() > 0 }
|
||||
}
|
||||
|
||||
|
|
@ -26,35 +41,61 @@ class ContentCache(private val context: Context) {
|
|||
}
|
||||
|
||||
fun downloadContent(serverUrl: String, contentId: String, filename: String): File? {
|
||||
val ext = filename.substringAfterLast('.', "mp4")
|
||||
val finalFile = File(cacheDir, "${contentId}.${ext}")
|
||||
val partFile = File(cacheDir, "${contentId}.${ext}${PART_SUFFIX}")
|
||||
try {
|
||||
val url = "${serverUrl}/api/content/${contentId}/file"
|
||||
val request = Request.Builder().url(url).build()
|
||||
val response = client.newCall(request).execute()
|
||||
|
||||
if (!response.isSuccessful) {
|
||||
Log.e("ContentCache", "Download failed: ${response.code}")
|
||||
return null
|
||||
}
|
||||
|
||||
val ext = filename.substringAfterLast('.', "mp4")
|
||||
val file = File(cacheDir, "${contentId}.${ext}")
|
||||
|
||||
response.body?.byteStream()?.use { input ->
|
||||
FileOutputStream(file).use { output ->
|
||||
input.copyTo(output)
|
||||
// .use closes the Response (and its body) on every path — also fixes the prior
|
||||
// error-path body/connection leak.
|
||||
client.newCall(request).execute().use { response ->
|
||||
if (!response.isSuccessful) {
|
||||
Log.e("ContentCache", "Download failed: ${response.code}")
|
||||
return null
|
||||
}
|
||||
// We issue a plain (no-Range) GET, so a 206 Partial Content means a proxy/CDN
|
||||
// returned a PARTIAL body whose Content-Length matches that partial — which would
|
||||
// pass the byte-count integrity check and promote a truncated file. Require a full 200.
|
||||
if (response.code == 206) {
|
||||
Log.e("ContentCache", "Refusing 206 Partial Content for $filename — not a complete file")
|
||||
return null
|
||||
}
|
||||
partFile.delete() // clear any earlier partial before writing
|
||||
val body = response.body ?: return null
|
||||
val expected = body.contentLength() // -1 when unknown (chunked)
|
||||
var written = 0L
|
||||
body.byteStream().use { input ->
|
||||
FileOutputStream(partFile).use { output -> written = input.copyTo(output) }
|
||||
}
|
||||
// Root-2: a truncated body must NOT be promoted to the cache and played as whole.
|
||||
if (!CacheValidation.isComplete(written, expected)) {
|
||||
Log.e("ContentCache", "Incomplete download ($written/$expected bytes) for $filename — discarding partial")
|
||||
partFile.delete()
|
||||
return null
|
||||
}
|
||||
finalFile.delete()
|
||||
if (!partFile.renameTo(finalFile)) {
|
||||
Log.e("ContentCache", "Rename failed for $filename")
|
||||
partFile.delete()
|
||||
return null
|
||||
}
|
||||
Log.i("ContentCache", "Downloaded: $filename -> ${finalFile.absolutePath} ($written bytes)")
|
||||
return finalFile
|
||||
}
|
||||
|
||||
Log.i("ContentCache", "Downloaded: $filename -> ${file.absolutePath}")
|
||||
return file
|
||||
} catch (e: Exception) {
|
||||
// Includes callTimeout / readTimeout (a stalled download on a healthy socket) and any
|
||||
// mid-stream break — never leave a partial at the real path.
|
||||
Log.e("ContentCache", "Download error: ${e.message}")
|
||||
partFile.delete()
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
fun deleteContent(contentId: String) {
|
||||
cacheDir.listFiles { _, name -> name.startsWith(contentId) }?.forEach { it.delete() }
|
||||
// Exact-prefix (with the dot) so we don't delete a different id's file — and this also
|
||||
// sweeps the "<id>.<ext>.part" temp.
|
||||
cacheDir.listFiles { _, name -> name.startsWith("$contentId.") }?.forEach { it.delete() }
|
||||
Log.i("ContentCache", "Deleted cached content: $contentId")
|
||||
}
|
||||
|
||||
|
|
@ -65,4 +106,15 @@ class ContentCache(private val context: Context) {
|
|||
fun getCacheSize(): Long {
|
||||
return cacheDir.listFiles()?.sumOf { it.length() } ?: 0L
|
||||
}
|
||||
|
||||
companion object {
|
||||
private const val PART_SUFFIX = ".part"
|
||||
|
||||
fun defaultClient(): OkHttpClient = OkHttpClient.Builder()
|
||||
.connectTimeout(30, TimeUnit.SECONDS)
|
||||
.readTimeout(30, TimeUnit.SECONDS) // Root-2: a stalled stream (no bytes 30s) aborts (was 5min)
|
||||
.writeTimeout(30, TimeUnit.SECONDS)
|
||||
.callTimeout(5, TimeUnit.MINUTES) // Root-2: hard OVERALL cap so a slow-drip can't hang forever
|
||||
.build()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,111 @@
|
|||
package com.remotedisplay.player.data
|
||||
|
||||
import android.os.SystemClock
|
||||
import android.util.Log
|
||||
import java.util.Collections
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
import java.util.concurrent.ExecutorService
|
||||
import java.util.concurrent.Executors
|
||||
|
||||
/**
|
||||
* Resolves the background-download × reconnect A-bucket bug. Previously every playlist sweep spawned
|
||||
* a detached, un-deduped download `thread{}` writing a deterministic `${id}.ext.part` path. On a
|
||||
* watchdog / ConnectionGuard reconnect the re-register drove a NEW sweep that started a DUPLICATE
|
||||
* download of the same content mid-fetch — two writers racing the shared `.part` (a corrupt file
|
||||
* whose byte-count still matched Content-Length, so it passed the integrity check and was promoted),
|
||||
* plus orphaned threads piling up across reconnects, resurrecting "stuck downloading" through the
|
||||
* reconnect path — the exact failure the caching fix was meant to eliminate.
|
||||
*
|
||||
* This coordinator makes background downloads reconnect-resilient:
|
||||
* - SINGLE-FLIGHT per contentId: a reconnect's re-sweep never starts a second download for content
|
||||
* already in flight. The in-flight fetch runs on its OWN OkHttp connection (independent of the
|
||||
* socket), so a reconnect underneath it does not interrupt it — it CONTINUES and completes; if
|
||||
* the network drop truncated it, ContentCache's partial detection discards the `.part` and it
|
||||
* cleanly RESTARTS on a later sweep. Never orphaned, never a duplicate racing the `.part`.
|
||||
* - BOUNDED executor: caps concurrent downloads (no thread pileup across reconnects); shutdown()
|
||||
* cancels in-flight work so nothing orphans or pins the Activity.
|
||||
* - BACKOFF with failure memory: a permanently-failing URL backs off exponentially instead of
|
||||
* re-fetching every 60s forever (no retry storm). The screen keeps serving cached content
|
||||
* (screen-resilience is untouched) and the CMS gets one "failed" ack, not a stuck "downloading".
|
||||
*
|
||||
* Atomic-swap + partial detection are ContentCache's job and are unchanged; this only guarantees a
|
||||
* SINGLE writer per `.part` so that guarantee can't be defeated by a concurrent duplicate.
|
||||
*/
|
||||
class DownloadCoordinator(
|
||||
private val cache: ContentCache,
|
||||
private val serverUrl: () -> String,
|
||||
private val socketAlive: () -> Boolean,
|
||||
private val onAck: (contentId: String, status: String) -> Unit,
|
||||
private val executor: ExecutorService = Executors.newFixedThreadPool(MAX_CONCURRENT),
|
||||
private val now: () -> Long = { SystemClock.elapsedRealtime() }
|
||||
) {
|
||||
private val inFlight = Collections.synchronizedSet(HashSet<String>())
|
||||
private val attempts = ConcurrentHashMap<String, Int>()
|
||||
private val nextAttemptAt = ConcurrentHashMap<String, Long>()
|
||||
|
||||
/**
|
||||
* Ensure [contentId] is (being) downloaded. Called for each local assignment on EVERY playlist
|
||||
* sweep — the 60s refresh and every post-reconnect re-register included. Idempotent and
|
||||
* non-blocking: it enqueues at most ONE download per contentId and returns immediately.
|
||||
*/
|
||||
fun ensure(contentId: String, filename: String) {
|
||||
if (contentId.isEmpty()) return
|
||||
if (cache.isContentCached(contentId)) { onAck(contentId, "ready"); return } // already have it — re-ack (SEED-A)
|
||||
// Socket down => the WATCHDOG owns recovery; don't hammer downloads over a dead connection.
|
||||
if (!socketAlive()) return
|
||||
if (now() < (nextAttemptAt[contentId] ?: 0L)) return // in failure backoff — don't storm
|
||||
if (!inFlight.add(contentId)) return // single-flight: already downloading
|
||||
try {
|
||||
executor.execute { runDownload(contentId, filename) }
|
||||
} catch (e: Throwable) {
|
||||
inFlight.remove(contentId) // executor rejected (shut down) — don't leak the guard
|
||||
}
|
||||
}
|
||||
|
||||
private fun runDownload(contentId: String, filename: String) {
|
||||
try {
|
||||
val file = cache.downloadContent(serverUrl(), contentId, filename)
|
||||
if (file != null) {
|
||||
attempts.remove(contentId); nextAttemptAt.remove(contentId)
|
||||
// Ack only reaches a live socket; if it dropped, the reconnect's re-register clears
|
||||
// the ack set and the next sweep re-acks the now-cached file.
|
||||
if (socketAlive()) onAck(contentId, "ready")
|
||||
} else {
|
||||
onFailure(contentId) // includes a reconnect-truncated .part (ContentCache returned null)
|
||||
}
|
||||
} catch (e: Throwable) {
|
||||
Log.w("DownloadCoordinator", "download $contentId failed: ${e.message}")
|
||||
onFailure(contentId)
|
||||
} finally {
|
||||
inFlight.remove(contentId)
|
||||
}
|
||||
}
|
||||
|
||||
private fun onFailure(contentId: String) {
|
||||
val n = (attempts[contentId] ?: 0) + 1
|
||||
attempts[contentId] = n
|
||||
val delay = minOf(BACKOFF_BASE_MS shl (n - 1).coerceIn(0, 20), BACKOFF_MAX_MS)
|
||||
nextAttemptAt[contentId] = now() + delay
|
||||
if (socketAlive()) onAck(contentId, "failed")
|
||||
}
|
||||
|
||||
/** Content deleted server-side — drop download/backoff state so a re-add downloads fresh. */
|
||||
fun forget(contentId: String) {
|
||||
inFlight.remove(contentId); attempts.remove(contentId); nextAttemptAt.remove(contentId)
|
||||
}
|
||||
|
||||
/** Teardown (onDestroy): cancel in-flight downloads so none orphan or pin the Activity. */
|
||||
fun shutdown() {
|
||||
try { executor.shutdownNow() } catch (_: Throwable) {}
|
||||
inFlight.clear()
|
||||
}
|
||||
|
||||
// test visibility
|
||||
internal fun isInFlight(contentId: String) = inFlight.contains(contentId)
|
||||
|
||||
companion object {
|
||||
const val MAX_CONCURRENT = 3
|
||||
const val BACKOFF_BASE_MS = 15_000L
|
||||
const val BACKOFF_MAX_MS = 5 * 60_000L
|
||||
}
|
||||
}
|
||||
|
|
@ -46,6 +46,14 @@ class MediaPlayerManager(
|
|||
onVideoComplete()
|
||||
}
|
||||
}
|
||||
// Root-2: a corrupt/undecodable video used to freeze the playlist forever — only
|
||||
// STATE_ENDED advanced, and an error goes to STATE_IDLE, so onVideoComplete never
|
||||
// fired. Treat a playback error like a completion so the loop moves on instead of
|
||||
// wedging on the broken item (mirrors the web/.wgt onerror -> advance).
|
||||
override fun onPlayerError(error: androidx.media3.common.PlaybackException) {
|
||||
Log.e("MediaPlayerManager", "Playback error (${error.errorCodeName}) — advancing: ${error.message}")
|
||||
onVideoComplete()
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -31,8 +31,13 @@ class PlaylistController(
|
|||
private val onItemChanged: (PlaylistItem?) -> Unit,
|
||||
private val onPlaylistEmpty: () -> Unit,
|
||||
private val onRequestRefresh: (() -> Unit)? = null,
|
||||
private val onNothingScheduled: (() -> Unit)? = null
|
||||
private val onNothingScheduled: (() -> Unit)? = null,
|
||||
// Screen-resilience: the defined "content isn't downloaded yet" waiting state, shown ONLY when
|
||||
// nothing has ever played (fresh device). Never used while content is on screen.
|
||||
private val onWaitingForContent: (() -> Unit)? = null
|
||||
) {
|
||||
private companion object { const val CONTENT_RECHECK_MS = 3000L }
|
||||
|
||||
private val items = mutableListOf<PlaylistItem>()
|
||||
private var currentIndex = -1
|
||||
private val handler = Handler(Looper.getMainLooper())
|
||||
|
|
@ -42,6 +47,14 @@ class PlaylistController(
|
|||
@Volatile private var effectiveTimezone: String? = null
|
||||
private var retryRunnable: Runnable? = null
|
||||
|
||||
// Screen-resilience: an item is playable only when its content is actually AVAILABLE
|
||||
// (widget / remote-stream / fully-downloaded local file). Injected by MainActivity (which owns
|
||||
// the cache); the default keeps every item playable so behavior is unchanged until it's set.
|
||||
private var contentReady: (PlaylistItem) -> Boolean = { true }
|
||||
fun setContentReadyCheck(f: (PlaylistItem) -> Boolean) { contentReady = f }
|
||||
// True while a valid item is rendered on screen — so we NEVER blank it for a pending download.
|
||||
private var hasContentOnScreen = false
|
||||
|
||||
// Video wall: followers don't self-advance — the leader's wall:sync drives the index.
|
||||
private var wallFollower = false
|
||||
// Wall-clock at which the current item started playing, for non-video sync position.
|
||||
|
|
@ -149,9 +162,13 @@ class PlaylistController(
|
|||
}
|
||||
}
|
||||
// Current item was removed or nothing was playing - start from the first
|
||||
// schedule-active item; idle if none are active right now.
|
||||
val idx = firstActiveIndex()
|
||||
if (idx >= 0) { currentIndex = idx; playCurrentItem() } else showNothingScheduled()
|
||||
// schedule-active AND downloaded item. Distinguish the two idle reasons: daypart
|
||||
// closed (nothing scheduled) => defined idle; scheduled-but-not-yet-downloaded =>
|
||||
// keep current content / waiting, never blank.
|
||||
val fp = PlaylistSelection.firstPlayableIndex(items.size) { playableNow(it) }
|
||||
if (fp >= 0) { currentIndex = fp; playCurrentItem() }
|
||||
else if (firstActiveIndex() < 0) showNothingScheduled()
|
||||
else onContentNotReady()
|
||||
} else {
|
||||
currentIndex = 0
|
||||
}
|
||||
|
|
@ -174,11 +191,12 @@ class PlaylistController(
|
|||
fun start() {
|
||||
isRunning = true
|
||||
if (items.isEmpty()) { onPlaylistEmpty(); return }
|
||||
// #74/#75: begin on the first schedule-active item; idle if none.
|
||||
val idx = firstActiveIndex()
|
||||
if (idx < 0) { showNothingScheduled(); return }
|
||||
currentIndex = idx
|
||||
playCurrentItem()
|
||||
// #74/#75: begin on the first schedule-active item; daypart-closed => defined idle.
|
||||
if (firstActiveIndex() < 0) { showNothingScheduled(); return }
|
||||
// Screen-resilience: only start on an item whose content is downloaded; if the scheduled
|
||||
// content isn't ready yet, keep current/wait (never blank on a loading state).
|
||||
val idx = PlaylistSelection.firstPlayableIndex(items.size) { playableNow(it) }
|
||||
if (idx >= 0) { currentIndex = idx; playCurrentItem() } else onContentNotReady()
|
||||
}
|
||||
|
||||
fun startIfNeeded() {
|
||||
|
|
@ -200,17 +218,21 @@ class PlaylistController(
|
|||
isRunning = false
|
||||
cancelAdvance()
|
||||
cancelRetry()
|
||||
hasContentOnScreen = false
|
||||
}
|
||||
|
||||
fun next() {
|
||||
if (items.isEmpty()) return
|
||||
// Request a playlist refresh between plays so new content gets picked up
|
||||
onRequestRefresh?.invoke()
|
||||
// #74/#75: advance to the next item the schedule allows now; idle if none.
|
||||
val idx = nextActiveIndex(currentIndex)
|
||||
if (idx < 0) { showNothingScheduled(); return }
|
||||
currentIndex = idx
|
||||
playCurrentItem()
|
||||
// #74/#75: daypart closed (nothing scheduled now) => defined idle.
|
||||
if (firstActiveIndex() < 0) { showNothingScheduled(); return }
|
||||
// Screen-resilience: advance to the next schedule-active AND downloaded item. Unready items
|
||||
// are SKIPPED (their background download continues), so a stalled/failed download of the
|
||||
// next content never blanks the screen — we keep looping the content we already have. If
|
||||
// NOTHING is downloaded yet, keep current content / show the waiting state, never blank.
|
||||
val idx = PlaylistSelection.nextPlayableIndex(items.size, currentIndex) { playableNow(it) }
|
||||
if (idx >= 0) { currentIndex = idx; playCurrentItem() } else onContentNotReady()
|
||||
}
|
||||
|
||||
fun onVideoComplete() {
|
||||
|
|
@ -227,6 +249,7 @@ class PlaylistController(
|
|||
itemStartedAt = System.currentTimeMillis()
|
||||
Log.i("PlaylistController", "Playing: ${item.filename} (index $currentIndex)")
|
||||
onItemChanged(item)
|
||||
hasContentOnScreen = true // a valid item is now rendered — protect it from being blanked
|
||||
|
||||
// For images and widgets, auto-advance after duration. For videos, wait
|
||||
// for the completion callback. Wall followers never auto-advance — the
|
||||
|
|
@ -257,6 +280,40 @@ class PlaylistController(
|
|||
item.schedules.isEmpty() ||
|
||||
ScheduleEval.isItemActiveNow(item.schedules, System.currentTimeMillis(), effectiveTimezone)
|
||||
|
||||
// Playable NOW = schedule-active AND its content is downloaded/available.
|
||||
private fun playableNow(i: Int): Boolean =
|
||||
i in items.indices && scheduleAllows(items[i]) && contentReady(items[i])
|
||||
|
||||
// Screen-resilience: the scheduled item(s) exist but their content isn't downloaded yet.
|
||||
// NEVER blank a screen that is already showing content — keep it and re-check soon (the
|
||||
// background download finishes, or the watchdog restores connectivity). Only show the defined
|
||||
// waiting/setup state on a fresh device that has never played anything.
|
||||
private fun onContentNotReady() {
|
||||
cancelAdvance()
|
||||
when (PlaylistSelection.whenNonePlayable(hasContentOnScreen)) {
|
||||
PlaylistSelection.NonePlayable.KEEP_CURRENT -> { /* leave current content on screen */ }
|
||||
PlaylistSelection.NonePlayable.SHOW_WAITING -> {
|
||||
hasContentOnScreen = false
|
||||
(onWaitingForContent ?: onNothingScheduled ?: onPlaylistEmpty)()
|
||||
}
|
||||
}
|
||||
scheduleContentRecheck()
|
||||
}
|
||||
|
||||
// Re-evaluate playability shortly (a pending download may have finished / a daypart opened).
|
||||
// Faster than the schedule re-check because a finished download should play promptly.
|
||||
private fun scheduleContentRecheck() {
|
||||
cancelRetry()
|
||||
retryRunnable = Runnable {
|
||||
if (isRunning && items.isNotEmpty()) {
|
||||
if (firstActiveIndex() < 0) { showNothingScheduled(); return@Runnable }
|
||||
val idx = PlaylistSelection.nextPlayableIndex(items.size, currentIndex) { playableNow(it) }
|
||||
if (idx >= 0) { currentIndex = idx; playCurrentItem() } else onContentNotReady()
|
||||
}
|
||||
}
|
||||
handler.postDelayed(retryRunnable!!, CONTENT_RECHECK_MS)
|
||||
}
|
||||
|
||||
private fun firstActiveIndex(): Int {
|
||||
for (i in items.indices) if (scheduleAllows(items[i])) return i
|
||||
return -1
|
||||
|
|
@ -275,6 +332,7 @@ class PlaylistController(
|
|||
// daypart may open. (Boundary re-evaluation otherwise happens on advance.)
|
||||
private fun showNothingScheduled() {
|
||||
cancelAdvance()
|
||||
hasContentOnScreen = false // the daypart genuinely closed — a defined idle, not a blank-bug
|
||||
(onNothingScheduled ?: onPlaylistEmpty)()
|
||||
cancelRetry()
|
||||
retryRunnable = Runnable {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,46 @@
|
|||
package com.remotedisplay.player.player
|
||||
|
||||
/**
|
||||
* Pure, unit-testable playlist SELECTION + screen-resilience decisions (no Android deps, same
|
||||
* pattern as ConnectionGuard / OtaThrottle). PlaylistController is the imperative shell (Handler /
|
||||
* playback); this owns "which item can play right now" and "what to do when none can", so the
|
||||
* viewer-visible invariant has real coverage:
|
||||
*
|
||||
* a pending/failed/stalled content download must NEVER blank or freeze a screen that is showing
|
||||
* content — the player keeps showing what it already has and only swaps to new content once it's
|
||||
* fully + validly downloaded.
|
||||
*/
|
||||
object PlaylistSelection {
|
||||
/** First index for which [isPlayable] holds, or -1 if none. */
|
||||
fun firstPlayableIndex(size: Int, isPlayable: (Int) -> Boolean): Int {
|
||||
for (i in 0 until size) if (isPlayable(i)) return i
|
||||
return -1
|
||||
}
|
||||
|
||||
/**
|
||||
* Next index after [from] (wrapping) for which [isPlayable] holds, or -1 if none. With a single
|
||||
* playable item it returns that item (loop), so a device keeps looping the content it HAS while
|
||||
* other items are still downloading.
|
||||
*/
|
||||
fun nextPlayableIndex(size: Int, from: Int, isPlayable: (Int) -> Boolean): Int {
|
||||
if (size <= 0) return -1
|
||||
for (i in 1..size) {
|
||||
val idx = (((from + i) % size) + size) % size
|
||||
if (isPlayable(idx)) return idx
|
||||
}
|
||||
return -1
|
||||
}
|
||||
|
||||
enum class NonePlayable { KEEP_CURRENT, SHOW_WAITING }
|
||||
|
||||
/**
|
||||
* When nothing is playable (e.g. the scheduled item's content isn't downloaded yet): NEVER
|
||||
* blank a screen that is showing content. Keep the current content if we have some on screen;
|
||||
* only fall to the defined waiting/setup state when there is genuinely nothing displayed yet
|
||||
* (a fresh device that has never successfully played anything). This is the one decision that
|
||||
* separates "nothing to show yet" (acceptable) from "had content but blanked while updating"
|
||||
* (the bug this fix forbids).
|
||||
*/
|
||||
fun whenNonePlayable(hasContentOnScreen: Boolean): NonePlayable =
|
||||
if (hasContentOnScreen) NonePlayable.KEEP_CURRENT else NonePlayable.SHOW_WAITING
|
||||
}
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
package com.remotedisplay.player.service
|
||||
|
||||
/**
|
||||
* v4 canonical liveness contract — the pure, unit-testable decision logic for the client-side
|
||||
* half-open watchdog (no Android / Socket.IO deps, same pattern as [ConnectionGuard] /
|
||||
* [OtaThrottle]). WebSocketService is the imperative shell: it tracks lastServerMessageAt (ANY
|
||||
* inbound refreshes it), arms only after a device:heartbeat-ack (degrade-safe), and on each
|
||||
* heartbeat tick asks THIS object whether a connected-but-silent socket is half-open and, if so,
|
||||
* whether the anti-thundering-herd backoff allows a reconnect now.
|
||||
*
|
||||
* ANTI-THUNDERING-HERD (must not become the flood #143/#149 fixed): the THRESHOLD is jittered
|
||||
* (45s ± up to 10s) so a fleet doesn't all declare half-open at once under a shared cause (server
|
||||
* load delaying acks), and the reconnect BACKOFF is exponential-with-jitter (1,2,4,8,16… capped,
|
||||
* ±20%) so repeated failures spread out and back off instead of hammering. Load-adaptation is by
|
||||
* the client's OWN ack-silence — there is deliberately NO status/health poll (that is itself a
|
||||
* second herd).
|
||||
*
|
||||
* Canonical defaults match the .wgt and /player so all three exhibit identical on-the-wire
|
||||
* behavior; a platform-driven deviation would be documented here.
|
||||
*/
|
||||
object LivenessWatchdog {
|
||||
const val THRESHOLD_BASE_MS = 45_000L // v4 canonical: 45s …
|
||||
const val THRESHOLD_JITTER_MS = 10_000L // … ± up to 10s
|
||||
const val BACKOFF_BASE_MS = 1_000L // v4 canonical: 1s, 2s, 4s, 8s, 16s …
|
||||
const val BACKOFF_CAP_MS = 30_000L // … capped (within the contract's ~30–60s band)
|
||||
private const val BACKOFF_JITTER_FRACTION = 0.2 // … each ± ~20%
|
||||
|
||||
/**
|
||||
* Watchdog threshold with jitter: 45s ± up to 10s. [rand] is a uniform value in [0, 1)
|
||||
* (injected so the pure logic is testable); the result is in [35s, 55s).
|
||||
*/
|
||||
fun thresholdMs(rand: Double): Long =
|
||||
THRESHOLD_BASE_MS + ((rand - 0.5) * 2 * THRESHOLD_JITTER_MS).toLong()
|
||||
|
||||
/**
|
||||
* Exponential reconnect backoff with ±20% jitter. [attempt] is 1-based (the 1st reconnect
|
||||
* uses the 1s step). Steps double (1s,2s,4s,8s,16s,…) and saturate at [BACKOFF_CAP_MS]; the
|
||||
* shift is clamped so a large attempt count can't overflow.
|
||||
*/
|
||||
fun backoffMs(attempt: Int, rand: Double): Long {
|
||||
val steps = (attempt - 1).coerceIn(0, 20)
|
||||
val base = minOf(BACKOFF_BASE_MS shl steps, BACKOFF_CAP_MS)
|
||||
val jitter = ((rand - 0.5) * 2 * BACKOFF_JITTER_FRACTION * base).toLong()
|
||||
return base + jitter
|
||||
}
|
||||
|
||||
/**
|
||||
* HALF-OPEN decision: reconnect only when a socket that is [connected] (Socket.IO still
|
||||
* reports it up — the state its own auto-reconnect can't see) and whose liveness we have
|
||||
* [armed] (seen ≥1 heartbeat-ack — degrade-safe: an ack-less/old server never arms us) has
|
||||
* been silent longer than [thresholdMs]. Pure; mirrors the .wgt's watchdogShouldReconnect.
|
||||
*/
|
||||
fun isHalfOpen(armed: Boolean, connected: Boolean, silenceMs: Long, thresholdMs: Long): Boolean =
|
||||
armed && connected && silenceMs > thresholdMs
|
||||
|
||||
/** Backoff gate: a new reconnect attempt is allowed only once the backoff for the current
|
||||
* attempt count has elapsed since the last one — spaces repeated failures fleet-wide. */
|
||||
fun mayReconnectNow(msSinceLastAttempt: Long, backoffMs: Long): Boolean =
|
||||
msSinceLastAttempt >= backoffMs
|
||||
}
|
||||
|
|
@ -8,7 +8,9 @@ import android.os.Binder
|
|||
import android.os.Handler
|
||||
import android.os.IBinder
|
||||
import android.os.Looper
|
||||
import android.os.SystemClock
|
||||
import android.util.Log
|
||||
import kotlin.random.Random
|
||||
import androidx.core.app.NotificationCompat
|
||||
import com.remotedisplay.player.MainActivity
|
||||
import com.remotedisplay.player.RemoteDisplayApp
|
||||
|
|
@ -35,6 +37,20 @@ class WebSocketService : Service() {
|
|||
private var heartbeatRunnable: Runnable? = null
|
||||
private val binder = LocalBinder()
|
||||
|
||||
// v4 liveness watchdog state (see LivenessWatchdog for the pure decision logic).
|
||||
// lastServerMessageAt: ANY inbound server message refreshes it (markAlive, wired into safeOn) —
|
||||
// uses the monotonic elapsedRealtime clock so an NTP/wall-clock jump can't false-fire or blind
|
||||
// the watchdog. livenessConfirmed: ARMED only after a device:heartbeat-ack (degrade-safe — an
|
||||
// ack-less/old server never arms us, so no false-fire storm). currentThresholdMs: per-connection
|
||||
// jittered 45s ± up to 10s. watchdogAttempt/lastWatchdogAttemptAt: exponential-backoff gate.
|
||||
@Volatile private var lastServerMessageAt = 0L
|
||||
@Volatile private var livenessConfirmed = false
|
||||
@Volatile private var currentThresholdMs = LivenessWatchdog.THRESHOLD_BASE_MS
|
||||
private var watchdogAttempt = 0
|
||||
private var lastWatchdogAttemptAt = 0L
|
||||
|
||||
private fun markAlive() { lastServerMessageAt = SystemClock.elapsedRealtime() }
|
||||
|
||||
companion object {
|
||||
// #148: backoff before re-opening the single socket after a disconnect that Socket.IO
|
||||
// does NOT auto-reconnect (io server/client disconnect) — never a blind immediate re-open.
|
||||
|
|
@ -101,6 +117,10 @@ class WebSocketService : Service() {
|
|||
// exception on the Socket.IO IO thread and crash the whole app.
|
||||
private fun Socket.safeOn(event: String, handler: (Array<Any?>) -> Unit): Socket {
|
||||
on(event) { args ->
|
||||
// v4: ANY inbound server message refreshes liveness (not just acks). The half-open
|
||||
// decision additionally requires socket.connected(), so refreshing on a disconnect
|
||||
// event is harmless. This is the single central receive-path hook.
|
||||
markAlive()
|
||||
try {
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
handler(args as Array<Any?>)
|
||||
|
|
@ -137,6 +157,14 @@ class WebSocketService : Service() {
|
|||
currentUrl = url
|
||||
socketActive = true
|
||||
|
||||
// v4 watchdog: a fresh socket is assumed alive; DIS-arm until it earns an ack again
|
||||
// (degrade-safe), and pick a new jittered threshold for this connection so a fleet doesn't
|
||||
// declare half-open in lockstep. watchdogAttempt is intentionally NOT reset here — it
|
||||
// tracks repeated reconnect failures across sockets and resets on a healthy ack.
|
||||
lastServerMessageAt = SystemClock.elapsedRealtime()
|
||||
livenessConfirmed = false
|
||||
currentThresholdMs = LivenessWatchdog.thresholdMs(Random.nextDouble())
|
||||
|
||||
try {
|
||||
val options = IO.Options().apply {
|
||||
forceNew = true
|
||||
|
|
@ -193,6 +221,17 @@ class WebSocketService : Service() {
|
|||
startHeartbeat()
|
||||
}
|
||||
|
||||
// v4 degrade-safe ARM: the watchdog arms ONLY after the first heartbeat-ack, so a
|
||||
// server that never acks (old/pre-contract) never arms us -> no false-fire storm.
|
||||
// markAlive already fired via safeOn (any inbound); a healthy ack also resets the
|
||||
// reconnect backoff. Known ack-gap (reconnecting-not-yet-re-registered) is benign:
|
||||
// arm-after-ack + any-inbound-refresh keep the watchdog from firing in that window.
|
||||
safeOn("device:heartbeat-ack") {
|
||||
if (!livenessConfirmed) Log.i("WebSocketService", "v4 watchdog: ARMED (first heartbeat-ack)")
|
||||
livenessConfirmed = true
|
||||
watchdogAttempt = 0
|
||||
}
|
||||
|
||||
safeOn("device:unpaired") {
|
||||
Log.w("WebSocketService", "Device not found on server - clearing credentials")
|
||||
config.clearDeviceCredentials()
|
||||
|
|
@ -370,6 +409,19 @@ class WebSocketService : Service() {
|
|||
}
|
||||
}
|
||||
|
||||
// v4 client identity block — additive, canonical snake_case (same field shape as the .wgt and
|
||||
// /player), piggybacked on the register message the client already sends. Backward-compatible:
|
||||
// an old server ignores unknown fields. Capture-don't-act — the server stores it; no client
|
||||
// logic is built on it here.
|
||||
private fun JSONObject.putIdentity() {
|
||||
try {
|
||||
put("client_type", "apk")
|
||||
put("client_version", deviceInfo.getAppVersion())
|
||||
put("platform", "Android " + android.os.Build.VERSION.RELEASE)
|
||||
put("contract_version", "v4")
|
||||
} catch (e: Throwable) { Log.w("WebSocketService", "identity: ${e.message}") }
|
||||
}
|
||||
|
||||
private fun register() {
|
||||
try {
|
||||
val data = JSONObject().apply {
|
||||
|
|
@ -388,6 +440,7 @@ class WebSocketService : Service() {
|
|||
}
|
||||
try { put("device_info", deviceInfo.getDeviceInfo()) } catch (e: Throwable) { Log.w("WebSocketService", "device_info: ${e.message}") }
|
||||
try { put("fingerprint", deviceInfo.getFingerprint()) } catch (e: Throwable) { Log.w("WebSocketService", "fingerprint: ${e.message}") }
|
||||
putIdentity()
|
||||
}
|
||||
socket?.emit("device:register", data)
|
||||
} catch (e: Throwable) {
|
||||
|
|
@ -407,18 +460,60 @@ class WebSocketService : Service() {
|
|||
heartbeatCount = 0
|
||||
heartbeatRunnable = object : Runnable {
|
||||
override fun run() {
|
||||
// A watchdog reconnect (below) tears down + restarts the heartbeat; if this is a
|
||||
// stale runnable superseded by that restart, stop — never let two loops run.
|
||||
if (heartbeatRunnable !== this) return
|
||||
sendHeartbeat()
|
||||
heartbeatCount++
|
||||
// Every 4th heartbeat (60s), request a fresh playlist
|
||||
if (heartbeatCount % 4 == 0) {
|
||||
requestPlaylistRefresh()
|
||||
}
|
||||
// v4 liveness watchdog: runs on the heartbeat tick (i.e. only while we've been
|
||||
// SENDING heartbeats). If it detects+acts on a half-open socket it tears this loop
|
||||
// down and a fresh one starts on re-register, so do NOT reschedule this one.
|
||||
if (checkHalfOpenAndReconnect()) return
|
||||
handler.postDelayed(this, 15000) // Every 15 seconds
|
||||
}
|
||||
}
|
||||
handler.post(heartbeatRunnable!!)
|
||||
}
|
||||
|
||||
/**
|
||||
* v4 half-open detection. On a HALF-OPEN socket Socket.IO still reports connected()==true
|
||||
* (its own auto-reconnect can't see server-silence), so we detect it here: armed (saw an ack)
|
||||
* + connected + silent past the jittered threshold. Returns true iff it triggered a reconnect
|
||||
* (so the heartbeat loop stops). The exponential backoff gate spaces repeated attempts so the
|
||||
* watchdog can't become the flood #143/#149 fixed. No status/health poll — load is read from
|
||||
* our own ack-silence.
|
||||
*/
|
||||
private fun checkHalfOpenAndReconnect(): Boolean {
|
||||
val connected = socket?.connected() == true
|
||||
val silenceMs = SystemClock.elapsedRealtime() - lastServerMessageAt
|
||||
if (!LivenessWatchdog.isHalfOpen(livenessConfirmed, connected, silenceMs, currentThresholdMs)) return false
|
||||
val sinceAttempt = SystemClock.elapsedRealtime() - lastWatchdogAttemptAt
|
||||
val backoff = LivenessWatchdog.backoffMs(watchdogAttempt + 1, Random.nextDouble())
|
||||
if (!LivenessWatchdog.mayReconnectNow(sinceAttempt, backoff)) return false
|
||||
watchdogAttempt++
|
||||
lastWatchdogAttemptAt = SystemClock.elapsedRealtime()
|
||||
Log.w("WebSocketService", "v4 watchdog: HALF-OPEN (silent ${silenceMs}ms > ${currentThresholdMs}ms, attempt=$watchdogAttempt) — teardown+reconnect (#148)")
|
||||
reconnectHalfOpen()
|
||||
return true
|
||||
}
|
||||
|
||||
/**
|
||||
* Teardown-before-reopen (#148) for a half-open socket. disconnect() kills the dead socket AND
|
||||
* its listeners/auto-reconnect FIRST (so we don't race Socket.IO's own reconnect), then
|
||||
* connect() — with socket=null + socketActive=false — opens exactly ONE fresh socket via the
|
||||
* ConnectionGuard. @Synchronized so it can't interleave with a racing connect()/openSocket().
|
||||
* The watchdog LAYERS ON the existing #148 guard; it does not replace it.
|
||||
*/
|
||||
@Synchronized
|
||||
private fun reconnectHalfOpen() {
|
||||
disconnect()
|
||||
connect()
|
||||
}
|
||||
|
||||
fun requestPlaylistRefresh() {
|
||||
if (socket?.connected() != true || config.deviceId.isEmpty()) return
|
||||
Log.i("WebSocketService", "Requesting playlist refresh")
|
||||
|
|
@ -428,6 +523,7 @@ class WebSocketService : Service() {
|
|||
val token = config.deviceToken
|
||||
if (token.isNotEmpty()) put("device_token", token)
|
||||
try { put("device_info", deviceInfo.getDeviceInfo()) } catch (e: Throwable) { Log.w("WebSocketService", "device_info: ${e.message}") }
|
||||
putIdentity()
|
||||
}
|
||||
socket?.emit("device:register", data)
|
||||
} catch (e: Throwable) {
|
||||
|
|
|
|||
|
|
@ -174,7 +174,7 @@ class DeviceInfo(private val context: Context) {
|
|||
return dm.widthPixels to dm.heightPixels
|
||||
}
|
||||
|
||||
private fun getAppVersion(): String {
|
||||
fun getAppVersion(): String {
|
||||
return try {
|
||||
context.packageManager.getPackageInfo(context.packageName, 0).versionName ?: "1.0.0"
|
||||
} catch (e: Exception) {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,30 @@
|
|||
package com.remotedisplay.player.data
|
||||
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
|
||||
/** Root-2: the partial/truncated-download detection rule that keeps a broken file out of the cache. */
|
||||
class CacheValidationTest {
|
||||
|
||||
@Test fun `full download with known length is complete`() {
|
||||
assertTrue(CacheValidation.isComplete(bytesWritten = 100, expectedBytes = 100))
|
||||
}
|
||||
|
||||
@Test fun `truncated download (fewer bytes than declared) is INCOMPLETE`() {
|
||||
assertFalse(CacheValidation.isComplete(bytesWritten = 40, expectedBytes = 100))
|
||||
}
|
||||
|
||||
@Test fun `over-read (more than declared) is treated as INCOMPLETE, not promoted`() {
|
||||
assertFalse(CacheValidation.isComplete(bytesWritten = 150, expectedBytes = 100))
|
||||
}
|
||||
|
||||
@Test fun `unknown length (chunked, -1) with bytes falls back to complete`() {
|
||||
assertTrue(CacheValidation.isComplete(bytesWritten = 100, expectedBytes = -1))
|
||||
}
|
||||
|
||||
@Test fun `zero bytes is never complete`() {
|
||||
assertFalse(CacheValidation.isComplete(bytesWritten = 0, expectedBytes = -1))
|
||||
assertFalse(CacheValidation.isComplete(bytesWritten = 0, expectedBytes = 100))
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,119 @@
|
|||
package com.remotedisplay.player.data
|
||||
|
||||
import okhttp3.OkHttpClient
|
||||
import org.junit.After
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertNotNull
|
||||
import org.junit.Assert.assertNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Before
|
||||
import org.junit.Test
|
||||
import java.io.OutputStream
|
||||
import java.net.ServerSocket
|
||||
import java.nio.file.Files
|
||||
import java.util.concurrent.TimeUnit
|
||||
|
||||
/**
|
||||
* Root-2 REPRODUCE-THEN-PROVE for the "stuck downloading / frozen" caching bug. Each test drives
|
||||
* the REAL ContentCache.downloadContent against a local HTTP server that reproduces a specific
|
||||
* failure mode on a HEALTHY socket (the socket is fine — the DOWNLOAD misbehaves), and proves the
|
||||
* fix: a stalled/trickling download aborts instead of hanging forever, and a truncated body is
|
||||
* never promoted to the cache (so it can't be played as if whole and wedge the playlist).
|
||||
*
|
||||
* The client uses short timeouts so the STALL reproduction is fast; the download/validation logic
|
||||
* exercised is identical to production (only the timeout VALUES differ — production is
|
||||
* callTimeout=5min / readTimeout=30s, verified by inspection in ContentCache.defaultClient()).
|
||||
*/
|
||||
class ContentDownloadTest {
|
||||
|
||||
private lateinit var dir: java.io.File
|
||||
private lateinit var cache: ContentCache
|
||||
private var server: ServerSocket? = null
|
||||
|
||||
private val client: OkHttpClient = OkHttpClient.Builder()
|
||||
.connectTimeout(2, TimeUnit.SECONDS)
|
||||
.readTimeout(1, TimeUnit.SECONDS) // a stalled stream aborts in ~1s => fast test
|
||||
.callTimeout(3, TimeUnit.SECONDS) // hard overall cap backstop
|
||||
.build()
|
||||
|
||||
@Before fun setUp() {
|
||||
dir = Files.createTempDirectory("cachetest").toFile()
|
||||
cache = ContentCache(dir, client) // internal ctor — real logic, no Android Context
|
||||
}
|
||||
|
||||
@After fun tearDown() {
|
||||
try { server?.close() } catch (_: Exception) {}
|
||||
dir.deleteRecursively()
|
||||
}
|
||||
|
||||
/** Accept ONE connection, drain the request, then let [respond] write a crafted response. */
|
||||
private fun serveOnce(respond: (OutputStream) -> Unit): String {
|
||||
val s = ServerSocket(0)
|
||||
server = s
|
||||
Thread {
|
||||
try {
|
||||
s.accept().use { sock ->
|
||||
val reader = sock.getInputStream().bufferedReader()
|
||||
while (true) { val line = reader.readLine() ?: break; if (line.isEmpty()) break }
|
||||
respond(sock.getOutputStream())
|
||||
}
|
||||
} catch (_: Exception) { /* client hung up on timeout — expected for the stall case */ }
|
||||
}.apply { isDaemon = true; start() }
|
||||
return "http://127.0.0.1:${s.localPort}"
|
||||
}
|
||||
|
||||
private fun OutputStream.writeHttp(contentLength: Int, body: ByteArray) {
|
||||
write("HTTP/1.1 200 OK\r\nContent-Length: $contentLength\r\nContent-Type: application/octet-stream\r\n\r\n".toByteArray())
|
||||
write(body)
|
||||
flush()
|
||||
}
|
||||
|
||||
private fun partFiles() = dir.listFiles { _, name -> name.endsWith(".part") }?.toList() ?: emptyList()
|
||||
|
||||
// ---- positive control: a complete download IS cached ----
|
||||
@Test fun `complete download is cached with the right size and no leftover part file`() {
|
||||
val url = serveOnce { it.writeHttp(5, "hello".toByteArray()) }
|
||||
val file = cache.downloadContent(url, "cidA", "clip.bin")
|
||||
assertNotNull("a complete download should be cached", file)
|
||||
assertEquals(5L, file!!.length())
|
||||
assertNotNull(cache.getCachedFile("cidA"))
|
||||
assertTrue("no .part temp should remain", partFiles().isEmpty())
|
||||
}
|
||||
|
||||
// ---- REPRODUCE: truncated body (declares 100 bytes, sends 40 then closes) on a healthy socket ----
|
||||
@Test fun `truncated download is NOT promoted to the cache — partial detected and discarded`() {
|
||||
val url = serveOnce {
|
||||
it.writeHttp(100, ByteArray(40) { 'x'.code.toByte() })
|
||||
// close after 40 of the declared 100 bytes -> truncation
|
||||
}
|
||||
val file = cache.downloadContent(url, "cidB", "clip.bin")
|
||||
assertNull("a truncated download must return null (not a usable file)", file)
|
||||
assertNull("a truncated file must NOT be served as cached", cache.getCachedFile("cidB"))
|
||||
assertTrue("the partial .part must be cleaned up, not left behind", partFiles().isEmpty())
|
||||
}
|
||||
|
||||
// ---- REPRODUCE: a STALLED download (headers + a trickle, then hang) on a healthy socket ----
|
||||
@Test fun `stalled download aborts within the timeout instead of hanging forever`() {
|
||||
val url = serveOnce {
|
||||
it.write("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n".toByteArray())
|
||||
it.write(ByteArray(10)); it.flush()
|
||||
Thread.sleep(10_000) // hang mid-stream — the OLD client (5min readTimeout) waited here
|
||||
}
|
||||
val start = System.currentTimeMillis()
|
||||
val file = cache.downloadContent(url, "cidC", "clip.bin")
|
||||
val elapsed = System.currentTimeMillis() - start
|
||||
assertNull("a stalled download must fail, not hang", file)
|
||||
assertTrue("must abort quickly via the timeout (was ~$elapsed ms)", elapsed < 5_000)
|
||||
assertNull(cache.getCachedFile("cidC"))
|
||||
assertTrue("no partial left behind after a stall", partFiles().isEmpty())
|
||||
}
|
||||
|
||||
// ---- prefix cross-match guard: an id that prefixes another must not match ----
|
||||
@Test fun `getCachedFile does not cross-match an id that is a prefix of another`() {
|
||||
serveOnce { it.writeHttp(3, "abc".toByteArray()) }.let { url ->
|
||||
assertNotNull(cache.downloadContent(url, "abc", "x.bin"))
|
||||
}
|
||||
assertNotNull(cache.getCachedFile("abc"))
|
||||
assertNull("id 'ab' must NOT match cached 'abc.x'", cache.getCachedFile("ab"))
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,194 @@
|
|||
package com.remotedisplay.player.data
|
||||
|
||||
import com.remotedisplay.player.player.PlaylistSelection
|
||||
import com.remotedisplay.player.service.ConnectionGuard
|
||||
import com.remotedisplay.player.service.LivenessWatchdog
|
||||
import okhttp3.OkHttpClient
|
||||
import org.junit.After
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertNotNull
|
||||
import org.junit.Assert.assertNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Assert.fail
|
||||
import org.junit.Before
|
||||
import org.junit.Test
|
||||
import java.io.OutputStream
|
||||
import java.net.ServerSocket
|
||||
import java.nio.file.Files
|
||||
import java.util.Collections
|
||||
import java.util.concurrent.CountDownLatch
|
||||
import java.util.concurrent.ExecutorService
|
||||
import java.util.concurrent.Executors
|
||||
import java.util.concurrent.TimeUnit
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
|
||||
/**
|
||||
* Reproduce-then-prove for the background-download × reconnect A-bucket bug. Drives the REAL
|
||||
* DownloadCoordinator + ContentCache against a local server that can be GATED (held mid-response)
|
||||
* so a download is genuinely in flight when a reconnect is simulated underneath it.
|
||||
*/
|
||||
class DownloadCoordinatorTest {
|
||||
private lateinit var dir: java.io.File
|
||||
private lateinit var cache: ContentCache
|
||||
private lateinit var executor: ExecutorService
|
||||
private lateinit var coord: DownloadCoordinator
|
||||
private var server: ServerSocket? = null
|
||||
|
||||
private val requests = AtomicInteger(0)
|
||||
private val acks = Collections.synchronizedList(mutableListOf<String>()) // "id:status"
|
||||
@Volatile private var socketUp = true
|
||||
@Volatile private var clock = 0L
|
||||
private var baseUrl = ""
|
||||
|
||||
private val client = OkHttpClient.Builder()
|
||||
.connectTimeout(2, TimeUnit.SECONDS).readTimeout(2, TimeUnit.SECONDS).callTimeout(4, TimeUnit.SECONDS).build()
|
||||
|
||||
@Before fun setUp() {
|
||||
dir = Files.createTempDirectory("coord").toFile()
|
||||
cache = ContentCache(dir, client)
|
||||
executor = Executors.newFixedThreadPool(2)
|
||||
requests.set(0); acks.clear(); socketUp = true; clock = 0L
|
||||
coord = DownloadCoordinator(cache, { baseUrl }, { socketUp }, { c, s -> acks.add("$c:$s") }, executor, { clock })
|
||||
}
|
||||
|
||||
@After fun tearDown() {
|
||||
try { coord.shutdown() } catch (_: Exception) {}
|
||||
try { executor.shutdownNow() } catch (_: Exception) {}
|
||||
server?.close(); dir.deleteRecursively()
|
||||
}
|
||||
|
||||
// A server that accepts connections in a loop; each: counts, fires `arrived` on the 1st, awaits
|
||||
// `gate` (if any), then writes `respond` and closes. Multiple connections are accepted, so a
|
||||
// buggy DUPLICATE download shows up as requests==2.
|
||||
private fun serve(gate: CountDownLatch?, arrived: CountDownLatch?, respond: (OutputStream) -> Unit) {
|
||||
val s = ServerSocket(0); server = s; baseUrl = "http://127.0.0.1:${s.localPort}"
|
||||
Thread {
|
||||
try {
|
||||
while (!s.isClosed) {
|
||||
val sock = s.accept()
|
||||
if (requests.incrementAndGet() == 1) arrived?.countDown()
|
||||
Thread {
|
||||
try {
|
||||
val r = sock.getInputStream().bufferedReader()
|
||||
while (true) { val l = r.readLine() ?: break; if (l.isEmpty()) break }
|
||||
gate?.await(5, TimeUnit.SECONDS)
|
||||
respond(sock.getOutputStream()); sock.close()
|
||||
} catch (_: Exception) {}
|
||||
}.apply { isDaemon = true; start() }
|
||||
}
|
||||
} catch (_: Exception) {}
|
||||
}.apply { isDaemon = true; start() }
|
||||
}
|
||||
private fun full(body: String): (OutputStream) -> Unit = { o ->
|
||||
o.write("HTTP/1.1 200 OK\r\nContent-Length: ${body.length}\r\n\r\n".toByteArray()); o.write(body.toByteArray()); o.flush()
|
||||
}
|
||||
private fun status(line: String): (OutputStream) -> Unit = { o ->
|
||||
o.write("HTTP/1.1 $line\r\nContent-Length: 0\r\n\r\n".toByteArray()); o.flush()
|
||||
}
|
||||
private fun partFiles() = dir.listFiles { _, n -> n.endsWith(".part") }?.size ?: 0
|
||||
private fun await(l: CountDownLatch, what: String) = assertTrue("timeout waiting for $what", l.await(5, TimeUnit.SECONDS))
|
||||
private fun waitAck(key: String) { for (i in 0..60) { if (acks.contains(key)) return; Thread.sleep(100) }; fail("no ack '$key' (got $acks)") }
|
||||
|
||||
// ===== THE REPRODUCE-THEN-PROVE: single-flight survives a reconnect mid-download =====
|
||||
@Test fun `reconnect mid-download starts NO duplicate — one request, in-flight fetch completes cleanly`() {
|
||||
val gate = CountDownLatch(1); val arrived = CountDownLatch(1)
|
||||
serve(gate, arrived, full("HELLObytes"))
|
||||
coord.ensure("X", "v.bin") // download starts, blocks on the gate — genuinely in flight
|
||||
await(arrived, "the download to reach the server")
|
||||
assertTrue("X is in flight", coord.isInFlight("X"))
|
||||
// SIMULATE THE RECONNECT: re-register drives more sweeps for the SAME content
|
||||
coord.ensure("X", "v.bin"); coord.ensure("X", "v.bin")
|
||||
Thread.sleep(300) // give any (buggy) duplicate time to connect
|
||||
assertEquals("single-flight: exactly ONE request despite reconnect re-sweeps", 1, requests.get())
|
||||
gate.countDown() // let the in-flight download finish on its own connection
|
||||
waitAck("X:ready")
|
||||
assertNotNull("the in-flight download completes and is cached (not orphaned/abandoned)", cache.getCachedFile("X"))
|
||||
assertEquals("no partial left", 0, partFiles())
|
||||
assertFalse("no longer in flight after completion", coord.isInFlight("X"))
|
||||
}
|
||||
|
||||
// ===== partial safety: a reconnect-truncated download is detected + not swapped =====
|
||||
@Test fun `reconnect-truncated download is detected as partial, acked failed, never swapped`() {
|
||||
serve(null, null) { o -> // declares 100, sends 40 then closes — as a mid-fetch network drop would
|
||||
o.write("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n".toByteArray()); o.write(ByteArray(40)); o.flush()
|
||||
}
|
||||
coord.ensure("Y", "v.bin"); waitAck("Y:failed")
|
||||
assertNull("a truncated file must NOT be cached/swapped", cache.getCachedFile("Y"))
|
||||
assertEquals("no partial left behind", 0, partFiles())
|
||||
}
|
||||
|
||||
// ===== permanent failure: backoff, no retry storm =====
|
||||
@Test fun `a permanently-failing download backs off instead of storming`() {
|
||||
serve(null, null, status("404 Not Found"))
|
||||
coord.ensure("Z", "v.bin"); waitAck("Z:failed")
|
||||
val after1 = requests.get()
|
||||
repeat(5) { coord.ensure("Z", "v.bin") }; Thread.sleep(300) // clock unchanged -> within backoff
|
||||
assertEquals("within backoff, NO new attempts (no storm)", after1, requests.get())
|
||||
clock = 20_000 // past the first 15s backoff step
|
||||
coord.ensure("Z", "v.bin"); Thread.sleep(300)
|
||||
assertEquals("after backoff elapses, exactly one more attempt", after1 + 1, requests.get())
|
||||
}
|
||||
|
||||
// ===== socket-down DEFER (ownership: watchdog owns recovery) =====
|
||||
@Test fun `socket down defers the download to the reconnect, then downloads when back`() {
|
||||
serve(null, null, full("data"))
|
||||
socketUp = false
|
||||
coord.ensure("W", "v.bin"); Thread.sleep(300)
|
||||
assertEquals("no download while the socket is down", 0, requests.get())
|
||||
socketUp = true
|
||||
coord.ensure("W", "v.bin"); waitAck("W:ready")
|
||||
assertNotNull("downloads once the socket is back", cache.getCachedFile("W"))
|
||||
}
|
||||
|
||||
// ===== teardown cancels in-flight work (no orphan / no Activity pin) =====
|
||||
@Test fun `shutdown cancels in-flight download and clears state`() {
|
||||
val gate = CountDownLatch(1); val arrived = CountDownLatch(1)
|
||||
serve(gate, arrived, full("data"))
|
||||
coord.ensure("Q", "v.bin"); await(arrived, "in-flight download")
|
||||
coord.shutdown()
|
||||
assertFalse("in-flight guard cleared on shutdown (no orphan)", coord.isInFlight("Q"))
|
||||
Thread.sleep(200)
|
||||
assertNull("nothing cached — the download was cancelled, not abandoned mid-promote", cache.getCachedFile("Q"))
|
||||
}
|
||||
|
||||
// ===== 206 Partial Content is refused (never promoted as whole) =====
|
||||
@Test fun `a 206 partial-content response is refused, not promoted`() {
|
||||
serve(null, null) { o -> // 206 with a matching partial length would pass a naive byte-count check
|
||||
o.write("HTTP/1.1 206 Partial Content\r\nContent-Length: 3\r\n\r\n".toByteArray()); o.write("abc".toByteArray()); o.flush()
|
||||
}
|
||||
coord.ensure("P", "v.bin"); waitAck("P:failed")
|
||||
assertNull("a 206 partial must not be cached as complete", cache.getCachedFile("P"))
|
||||
}
|
||||
|
||||
// ===== INTEGRATED RE-SOAK: playing -> bg download -> reconnect mid-download -> resume =====
|
||||
@Test fun `re-soak — one socket, no orphaned download, screen never blanks, download resumes`() {
|
||||
val onlyZeroReady: (Int) -> Boolean = { it == 0 } // item 0 cached & playing; item 1 downloading
|
||||
val threshold = LivenessWatchdog.thresholdMs(0.5)
|
||||
val gate = CountDownLatch(1); val arrived = CountDownLatch(1)
|
||||
serve(gate, arrived, full("item1data"))
|
||||
|
||||
// playing item 0; item 1's background download is in flight
|
||||
coord.ensure("item1", "v.bin"); await(arrived, "item1 download")
|
||||
assertEquals("screen keeps playing item 0 (skips the still-downloading item 1)", 0,
|
||||
PlaylistSelection.nextPlayableIndex(2, 0, onlyZeroReady))
|
||||
|
||||
// server goes silent -> watchdog fires (its own signal); screen must NOT blank
|
||||
assertTrue(LivenessWatchdog.isHalfOpen(armed = true, connected = true, silenceMs = 50_000, thresholdMs = threshold))
|
||||
assertEquals(PlaylistSelection.NonePlayable.KEEP_CURRENT, PlaylistSelection.whenNonePlayable(hasContentOnScreen = true))
|
||||
|
||||
// watchdog reconnect -> exactly ONE socket via ConnectionGuard; the re-sweep does NOT dup the download
|
||||
assertTrue(ConnectionGuard.shouldOpenNewSocket(hasSocket = false, sameUrl = true, socketActive = false))
|
||||
coord.ensure("item1", "v.bin"); Thread.sleep(200)
|
||||
assertEquals("reconnect re-sweep did not orphan/duplicate the in-flight download", 1, requests.get())
|
||||
assertEquals("screen still shows item 0 across the reconnect", 0,
|
||||
PlaylistSelection.nextPlayableIndex(2, 0, onlyZeroReady))
|
||||
|
||||
// download RESUMES (its own connection was never torn down) and completes cleanly
|
||||
gate.countDown(); waitAck("item1:ready")
|
||||
assertNotNull("item 1 downloaded (not orphaned) and is now swappable", cache.getCachedFile("item1"))
|
||||
assertFalse(coord.isInFlight("item1"))
|
||||
assertEquals("now both items ready -> screen advances to item 1 (only fully-valid content)", 1,
|
||||
PlaylistSelection.nextPlayableIndex(2, 0) { true })
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
package com.remotedisplay.player.player
|
||||
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Test
|
||||
|
||||
/**
|
||||
* Screen-resilience (FIX 2): a pending/failed/stalled content download must NEVER blank or freeze
|
||||
* a screen that is showing content. These prove the SELECTION rules that guarantee it — unready
|
||||
* items are skipped (the player keeps looping what it has), and "nothing playable" keeps the
|
||||
* current content whenever anything is on screen.
|
||||
*/
|
||||
class PlaylistSelectionTest {
|
||||
|
||||
// helper: items 0..n-1, `ready` are the indices whose content is downloaded/available
|
||||
private fun readyPredicate(vararg ready: Int): (Int) -> Boolean = { it in ready.toSet() }
|
||||
|
||||
// ===== the KEY viewer scenario: playing item 0, item 1's download is stalled =====
|
||||
@Test fun `a stalled download of the NEXT item does not interrupt current content — it loops what it has`() {
|
||||
// playlist [0 ready, 1 NOT ready(downloading)]; currently playing 0. Advancing must SKIP 1
|
||||
// and come back to 0, so the viewer keeps seeing content (no blank, no "Downloading…").
|
||||
val idx = PlaylistSelection.nextPlayableIndex(size = 2, from = 0, isPlayable = readyPredicate(0))
|
||||
assertEquals("must skip the un-downloaded item and keep playing item 0", 0, idx)
|
||||
}
|
||||
|
||||
@Test fun `once the download completes the next item is picked up on the following advance`() {
|
||||
// now item 1 is downloaded too -> advancing from 0 swaps to 1 (only fully-ready content).
|
||||
val idx = PlaylistSelection.nextPlayableIndex(size = 2, from = 0, isPlayable = readyPredicate(0, 1))
|
||||
assertEquals(1, idx)
|
||||
}
|
||||
|
||||
// ===== partial-file safety: an un-ready (partial/corrupt) item is never selected =====
|
||||
@Test fun `an un-ready (partial-download) item is never chosen to play`() {
|
||||
// items [0 ready, 1 partial/not-ready, 2 ready] -> selection never returns 1.
|
||||
assertEquals(0, PlaylistSelection.firstPlayableIndex(3, readyPredicate(0, 2)))
|
||||
assertEquals(2, PlaylistSelection.nextPlayableIndex(3, 0, readyPredicate(0, 2)))
|
||||
assertEquals(0, PlaylistSelection.nextPlayableIndex(3, 2, readyPredicate(0, 2)))
|
||||
}
|
||||
|
||||
// ===== nothing downloaded yet =====
|
||||
@Test fun `no item ready returns -1 (nothing to play)`() {
|
||||
assertEquals(-1, PlaylistSelection.firstPlayableIndex(3, readyPredicate()))
|
||||
assertEquals(-1, PlaylistSelection.nextPlayableIndex(3, 1, readyPredicate()))
|
||||
}
|
||||
|
||||
// ===== the invariant: never blank while content is on screen =====
|
||||
@Test fun `nothing-playable KEEPS current content whenever something is on screen (never blanks)`() {
|
||||
assertEquals(PlaylistSelection.NonePlayable.KEEP_CURRENT,
|
||||
PlaylistSelection.whenNonePlayable(hasContentOnScreen = true))
|
||||
}
|
||||
|
||||
@Test fun `nothing-playable shows the defined waiting state only when nothing is displayed yet`() {
|
||||
assertEquals(PlaylistSelection.NonePlayable.SHOW_WAITING,
|
||||
PlaylistSelection.whenNonePlayable(hasContentOnScreen = false))
|
||||
}
|
||||
|
||||
// ===== single downloaded item loops (doesn't blank waiting for others) =====
|
||||
@Test fun `a single downloaded item loops instead of blanking`() {
|
||||
assertEquals(0, PlaylistSelection.nextPlayableIndex(1, 0, readyPredicate(0)))
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,172 @@
|
|||
package com.remotedisplay.player.service
|
||||
|
||||
import com.remotedisplay.player.data.ContentCache
|
||||
import com.remotedisplay.player.player.PlaylistSelection
|
||||
import okhttp3.OkHttpClient
|
||||
import org.junit.After
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertNotNull
|
||||
import org.junit.Assert.assertNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Before
|
||||
import org.junit.Test
|
||||
import java.io.OutputStream
|
||||
import java.net.ServerSocket
|
||||
import java.nio.file.Files
|
||||
import java.util.concurrent.TimeUnit
|
||||
|
||||
/**
|
||||
* APK VERIFY pass — the ASSEMBLY soak. Per-part tests prove the pieces; this proves the
|
||||
* caching-state-machine × liveness-watchdog INTERACTION (the novel APK-only behavior) does not
|
||||
* step on itself, driving the REAL ContentCache download + the REAL LivenessWatchdog decisions
|
||||
* through the lifecycle sequence.
|
||||
*
|
||||
* Scope note (honest): the Handler/Looper/Socket.IO runtime can't run in a plain JVM unit test
|
||||
* (no Robolectric in this module), so socket-teardown/timer wiring is verified by code-trace +
|
||||
* ConnectionGuard's decisions; the DOWNLOAD path and every WATCHDOG DECISION here are the real
|
||||
* production code exercised with real time.
|
||||
*/
|
||||
class AssemblyInteractionTest {
|
||||
|
||||
private lateinit var dir: java.io.File
|
||||
private lateinit var cache: ContentCache
|
||||
private var server: ServerSocket? = null
|
||||
private val client = OkHttpClient.Builder()
|
||||
.connectTimeout(2, TimeUnit.SECONDS)
|
||||
.readTimeout(1, TimeUnit.SECONDS)
|
||||
.callTimeout(3, TimeUnit.SECONDS)
|
||||
.build()
|
||||
|
||||
@Before fun setUp() {
|
||||
dir = Files.createTempDirectory("assembly").toFile()
|
||||
cache = ContentCache(dir, client)
|
||||
}
|
||||
|
||||
@After fun tearDown() {
|
||||
try { server?.close() } catch (_: Exception) {}
|
||||
dir.deleteRecursively()
|
||||
}
|
||||
|
||||
private fun serveOnce(respond: (OutputStream) -> Unit): String {
|
||||
val s = ServerSocket(0); server = s
|
||||
Thread {
|
||||
try {
|
||||
s.accept().use { sock ->
|
||||
val r = sock.getInputStream().bufferedReader()
|
||||
while (true) { val l = r.readLine() ?: break; if (l.isEmpty()) break }
|
||||
respond(sock.getOutputStream())
|
||||
}
|
||||
} catch (_: Exception) {}
|
||||
}.apply { isDaemon = true; start() }
|
||||
return "http://127.0.0.1:${s.localPort}"
|
||||
}
|
||||
private fun serveComplete(body: String) = serveOnce {
|
||||
it.write("HTTP/1.1 200 OK\r\nContent-Length: ${body.length}\r\n\r\n".toByteArray())
|
||||
it.write(body.toByteArray()); it.flush()
|
||||
}
|
||||
private fun serveStall() = serveOnce {
|
||||
it.write("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n".toByteArray())
|
||||
it.write(ByteArray(10)); it.flush(); Thread.sleep(10_000)
|
||||
}
|
||||
private fun partFiles() = dir.listFiles { _, n -> n.endsWith(".part") }?.toList() ?: emptyList()
|
||||
|
||||
// ===== VERIFY 1: reconnect ownership — watchdog routes through ConnectionGuard (single owner) =====
|
||||
@Test fun `watchdog reconnect opens exactly one socket via ConnectionGuard, never its own`() {
|
||||
// reconnectHalfOpen() = disconnect() [socket=null, socketActive=false] then connect(), and
|
||||
// connect() consults ConnectionGuard. After teardown -> open exactly ONE.
|
||||
assertTrue(ConnectionGuard.shouldOpenNewSocket(hasSocket = false, sameUrl = true, socketActive = false))
|
||||
// and a live/self-healing socket is REUSED (single-owner: never a second socket alongside).
|
||||
assertFalse(ConnectionGuard.shouldOpenNewSocket(hasSocket = true, sameUrl = true, socketActive = true))
|
||||
}
|
||||
|
||||
// ===== VERIFY 3 CHECK A: a caching stall/retry must NOT look like server-silence =====
|
||||
@Test fun `a stalled download does NOT trip the watchdog while the socket keeps getting acks`() {
|
||||
// Real architecture: downloads run on their own thread; socket liveness is refreshed by
|
||||
// heartbeat-acks (any inbound) every 15s, INDEPENDENT of download activity. Walk a 60s
|
||||
// stall; an ack lands every 15s. Silence must never cross the (jitter-floor) threshold.
|
||||
val threshold = LivenessWatchdog.thresholdMs(0.0) // 35s — the tightest (worst-case) threshold
|
||||
var lastServerMs = 0L
|
||||
var spurious = false
|
||||
for (t in 0..60_000 step 1_000) {
|
||||
if (t % 15_000 == 0) lastServerMs = t.toLong() // ack refresh — decoupled from the download
|
||||
if (LivenessWatchdog.isHalfOpen(armed = true, connected = true, silenceMs = t - lastServerMs, thresholdMs = threshold))
|
||||
spurious = true
|
||||
}
|
||||
assertFalse("a download stall must NOT cause a watchdog reconnect while the socket is healthy", spurious)
|
||||
}
|
||||
|
||||
// ===== VERIFY 3 CHECK B: server-silence DURING a download -> watchdog fires; download unharmed =====
|
||||
@Test fun `server silence during a download fires the watchdog AND the download state stays consistent`() {
|
||||
// The watchdog (server-silence) and the download (its own thread + .part file) are separate
|
||||
// subsystems: the watchdog tears down the SOCKET, never the download or its files.
|
||||
val threshold = LivenessWatchdog.thresholdMs(0.5) // 45s
|
||||
assertTrue("50s of server silence is half-open", LivenessWatchdog.isHalfOpen(true, true, 50_000, threshold))
|
||||
val f = cache.downloadContent(serveComplete("payload!"), "cidB", "v.bin")
|
||||
assertNotNull("the download completes independently of a socket reconnect", f)
|
||||
assertEquals(8L, f!!.length())
|
||||
assertTrue("no partial left — download files untouched by any reconnect", partFiles().isEmpty())
|
||||
}
|
||||
|
||||
// ===== VERIFY 3 CHECK C: no double-action — retry then success leaves ONE consistent file =====
|
||||
@Test fun `caching-timeout retry and watchdog do not double-act — retry yields one consistent file`() {
|
||||
// 1st attempt stalls -> caching timeout -> null, partial cleaned (NOT a watchdog concern).
|
||||
assertNull(cache.downloadContent(serveStall(), "cidC", "v.bin"))
|
||||
assertTrue("stall leaves no orphan .part", partFiles().isEmpty())
|
||||
assertNull("stall does not promote a partial", cache.getCachedFile("cidC"))
|
||||
// retry (healthy) -> exactly one cached file, no duplicates/orphans.
|
||||
val f = cache.downloadContent(serveComplete("retry-ok"), "cidC", "v.bin")
|
||||
assertNotNull(f); assertEquals(8L, f!!.length())
|
||||
assertEquals("exactly one cached artifact for the id", 1,
|
||||
dir.listFiles { _, n -> n.startsWith("cidC.") }?.size ?: 0)
|
||||
}
|
||||
|
||||
// ===== VERIFY 4: degrade-safe + the deferred ack-gap window =====
|
||||
@Test fun `degrade-safe — an ack-less server never arms the watchdog, even when very silent`() {
|
||||
assertFalse(LivenessWatchdog.isHalfOpen(armed = false, connected = true, silenceMs = 10 * 60_000, thresholdMs = 45_000))
|
||||
}
|
||||
|
||||
// ===== FIX 1 + FIX 2 re-soak: through the whole sequence the SCREEN NEVER BLANKS =====
|
||||
@Test fun `re-soak — stall then ack-silence then watchdog reconnect then resume, screen never blanks`() {
|
||||
// Device is PLAYING item 0 (downloaded); item 1's download is stalled. Walk every stage.
|
||||
val onlyZeroReady: (Int) -> Boolean = { it == 0 }
|
||||
val bothReady: (Int) -> Boolean = { true }
|
||||
val threshold = LivenessWatchdog.thresholdMs(0.5)
|
||||
|
||||
// 1) healthy, item 1 download STALLS on a live socket -> advancing SKIPS 1 and keeps 0
|
||||
// (a playable index is always found, so the screen shows content — never blanks), and
|
||||
// the stall does NOT trip the watchdog.
|
||||
assertEquals("skip the stalled item, keep playing 0", 0,
|
||||
PlaylistSelection.nextPlayableIndex(2, 0, onlyZeroReady))
|
||||
assertFalse("download stall on a live socket must not trip the watchdog",
|
||||
LivenessWatchdog.isHalfOpen(armed = true, connected = true, silenceMs = 5_000, thresholdMs = threshold))
|
||||
|
||||
// 2) server goes SILENT -> the watchdog fires on ITS signal, but the screen keeps content
|
||||
assertTrue(LivenessWatchdog.isHalfOpen(true, true, 50_000, threshold))
|
||||
assertEquals("a reconnect must never blank a screen showing content",
|
||||
PlaylistSelection.NonePlayable.KEEP_CURRENT, PlaylistSelection.whenNonePlayable(hasContentOnScreen = true))
|
||||
|
||||
// 3) watchdog reconnect -> EXACTLY ONE socket via ConnectionGuard (single owner)
|
||||
assertTrue(ConnectionGuard.shouldOpenNewSocket(hasSocket = false, sameUrl = true, socketActive = false))
|
||||
|
||||
// 4) during the reconnect window, item 1 still not ready (downloads deferred, FIX 1) ->
|
||||
// advancing still finds item 0 -> screen still shows content
|
||||
assertEquals(0, PlaylistSelection.nextPlayableIndex(2, 0, onlyZeroReady))
|
||||
|
||||
// 5) reconnected + item 1 now fully downloaded -> swaps forward to item 1 (only ever swap
|
||||
// to fully-valid content). At no stage did selection return -1 while 0 was ready, so the
|
||||
// screen was never blanked.
|
||||
assertEquals(1, PlaylistSelection.nextPlayableIndex(2, 0, bothReady))
|
||||
}
|
||||
|
||||
@Test fun `ack-gap window — reconnecting-not-yet-re-registered does not false-fire`() {
|
||||
// openSocket resets armed=false + lastServerMessageAt=now. Before the first post-reconnect
|
||||
// ack, armed=false so NO fire even though the OLD connection's silence was large...
|
||||
assertFalse("must not fire before the first post-reconnect ack",
|
||||
LivenessWatchdog.isHalfOpen(armed = false, connected = true, silenceMs = 999_999, thresholdMs = 45_000))
|
||||
// ...and once any-inbound (the re-register response) refreshes liveness and an ack arms it,
|
||||
// a fresh connection with small silence is healthy (still no fire).
|
||||
assertFalse("healthy post-reconnect (armed, low silence) does not fire",
|
||||
LivenessWatchdog.isHalfOpen(armed = true, connected = true, silenceMs = 500, thresholdMs = 45_000))
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,76 @@
|
|||
package com.remotedisplay.player.service
|
||||
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
|
||||
/**
|
||||
* v4 liveness-contract conformance for the pure watchdog decision logic. Proves:
|
||||
* - degrade-safe (never fires unarmed) and the connected+silent half-open condition,
|
||||
* - the anti-thundering-herd guarantees: threshold jitter (45s ± up to 10s) and exponential
|
||||
* reconnect backoff with jitter (1,2,4,8,16… capped, ±20%) are PRESENT (not fixed values).
|
||||
*/
|
||||
class LivenessWatchdogTest {
|
||||
|
||||
// ---- degrade-safe + half-open decision ----
|
||||
|
||||
@Test fun `unarmed never fires — degrade-safe against an ack-less server`() {
|
||||
// Even a connected socket silent well past threshold must NOT reconnect until armed by an ack.
|
||||
assertFalse(LivenessWatchdog.isHalfOpen(armed = false, connected = true, silenceMs = 999_999, thresholdMs = 45_000))
|
||||
}
|
||||
|
||||
@Test fun `disconnected never fires — that is Socket_IO's job, not the watchdog's`() {
|
||||
assertFalse(LivenessWatchdog.isHalfOpen(armed = true, connected = false, silenceMs = 999_999, thresholdMs = 45_000))
|
||||
}
|
||||
|
||||
@Test fun `armed + connected + silent past threshold IS half-open`() {
|
||||
assertTrue(LivenessWatchdog.isHalfOpen(armed = true, connected = true, silenceMs = 46_000, thresholdMs = 45_000))
|
||||
}
|
||||
|
||||
@Test fun `armed + connected but within threshold is healthy`() {
|
||||
assertFalse(LivenessWatchdog.isHalfOpen(armed = true, connected = true, silenceMs = 30_000, thresholdMs = 45_000))
|
||||
}
|
||||
|
||||
// ---- anti-herd: threshold jitter 45s ± up to 10s ----
|
||||
|
||||
@Test fun `threshold is centered on 45s and stays within ±10s`() {
|
||||
assertEquals(45_000L, LivenessWatchdog.thresholdMs(0.5))
|
||||
assertEquals(35_000L, LivenessWatchdog.thresholdMs(0.0))
|
||||
assertEquals(54_999L, LivenessWatchdog.thresholdMs(0.99995)) // ~55s upper bound (exclusive)
|
||||
for (i in 0..100) {
|
||||
val t = LivenessWatchdog.thresholdMs(i / 100.0)
|
||||
assertTrue("threshold $t out of [35s,55s)", t in 35_000..55_000)
|
||||
}
|
||||
}
|
||||
|
||||
@Test fun `threshold has jitter — different rands give different thresholds (not fixed)`() {
|
||||
assertTrue(LivenessWatchdog.thresholdMs(0.1) != LivenessWatchdog.thresholdMs(0.9))
|
||||
}
|
||||
|
||||
// ---- anti-herd: exponential backoff 1,2,4,8,16… capped, ±20% jitter ----
|
||||
|
||||
@Test fun `backoff doubles per attempt then saturates at the cap`() {
|
||||
// rand=0.5 => no jitter, so we see the exact base curve.
|
||||
assertEquals(1_000L, LivenessWatchdog.backoffMs(1, 0.5))
|
||||
assertEquals(2_000L, LivenessWatchdog.backoffMs(2, 0.5))
|
||||
assertEquals(4_000L, LivenessWatchdog.backoffMs(3, 0.5))
|
||||
assertEquals(8_000L, LivenessWatchdog.backoffMs(4, 0.5))
|
||||
assertEquals(16_000L, LivenessWatchdog.backoffMs(5, 0.5))
|
||||
assertEquals(30_000L, LivenessWatchdog.backoffMs(6, 0.5)) // 32s -> capped at 30s
|
||||
assertEquals(30_000L, LivenessWatchdog.backoffMs(50, 0.5)) // no overflow at large attempts
|
||||
}
|
||||
|
||||
@Test fun `backoff jitter is ±20% of the step and present (not fixed)`() {
|
||||
// attempt 5 base = 16s; ±20% => [12.8s, 19.2s].
|
||||
assertEquals(12_800L, LivenessWatchdog.backoffMs(5, 0.0)) // -20%
|
||||
assertEquals(19_200L, LivenessWatchdog.backoffMs(5, 1.0)) // +20%
|
||||
assertTrue(LivenessWatchdog.backoffMs(5, 0.2) != LivenessWatchdog.backoffMs(5, 0.8))
|
||||
}
|
||||
|
||||
@Test fun `mayReconnectNow gates on the backoff having elapsed`() {
|
||||
assertTrue(LivenessWatchdog.mayReconnectNow(msSinceLastAttempt = 2_000, backoffMs = 1_000))
|
||||
assertTrue(LivenessWatchdog.mayReconnectNow(msSinceLastAttempt = 1_000, backoffMs = 1_000))
|
||||
assertFalse(LivenessWatchdog.mayReconnectNow(msSinceLastAttempt = 500, backoffMs = 1_000))
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue