diff --git a/android/app/src/main/java/com/remotedisplay/player/data/ContentCache.kt b/android/app/src/main/java/com/remotedisplay/player/data/ContentCache.kt index afb22a2..d101d45 100644 --- a/android/app/src/main/java/com/remotedisplay/player/data/ContentCache.kt +++ b/android/app/src/main/java/com/remotedisplay/player/data/ContentCache.kt @@ -12,10 +12,29 @@ import java.util.concurrent.TimeUnit * 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, + * - download to a `.part` temp + integrity-check 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. * + * RESUME. Those fixes made a bad download safe; they did not make it possible. Every attempt began + * at byte 0 and the `.part` was deleted on failure, so on a link that cannot carry a whole asset in + * one unbroken call — a one-bar 5G site, the case this was reported from — the file NEVER lands. + * The player then has nothing cached, and a screen with nothing cached shows the waiting state: + * reported as "the screens go black instead of playing from cache", when the real failure was that + * the cache could never be filled in the first place. Five minutes of transfer, discarded; back + * off; five more minutes, discarded; forever. + * + * So an interrupted download now KEEPS its `.part` and the next attempt asks for the rest with a + * Range header. Progress accumulates across attempts and across reboots instead of being thrown + * away, which is the whole difference between "eventually plays" and "never plays". + * + * Two ways a resume could corrupt the cache, both closed: + * - the asset changed under us — `If-Range` with the stored validator makes the server answer 200 + * with the whole body instead of a tail, and we restart from zero. + * - the `.part` is longer than the asset — the server answers 416 and we discard it. + * The completeness check is unchanged in spirit but now counts TOTAL bytes on disk against the + * total the server declared in Content-Range, not bytes received this attempt. + * * 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. @@ -29,12 +48,27 @@ class ContentCache internal constructor( defaultClient() ) + /** + * What one download attempt achieved. The distinction that matters is [Partial] vs [Failed]: + * an attempt that moved bytes onto disk is PROGRESS, and backing that off exponentially the way + * a hard failure is backed off is what turns a slow site into a dead one. + */ + sealed class Result { + data class Done(val file: File) : Result() + /** Bytes are on disk and the next attempt resumes from there. */ + data class Partial(val bytesOnDisk: Long, val totalBytes: Long, val progressed: Boolean) : Result() + /** Nothing usable happened: refused, unreachable, or a stale partial we had to discard. */ + object Failed : Result() + } + fun getCachedFile(contentId: String): File? { // Match "." 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) } + // cross-matching. `contains` rather than `endsWith` for the temp check because the resume + // validator sidecar is "..part.tag" — it does not END with ".part", and returning + // THAT as the cached asset would hand the player a short ETag file to play. + val files = cacheDir.listFiles { _, name -> name.startsWith("$contentId.") && !name.contains(PART_SUFFIX) } val hit = files?.firstOrNull()?.takeIf { it.exists() && it.length() > 0 } - com.remotedisplay.player.util.DebugLog.v("ContentCache", "getCachedFile($contentId): dir=${cacheDir.absolutePath} listFiles=${files?.size} -> ${hit?.name ?: "MISS"}") + com.remotedisplay.player.util.DebugLog.v("ContentCache", "getCachedFile($contentId): dir=${cacheDir.absolutePath} listFiles=${files?.size ?: -1} hit=${hit?.name}") return hit } @@ -42,61 +76,138 @@ class ContentCache internal constructor( return getCachedFile(contentId) != null } - fun downloadContent(serverUrl: String, contentId: String, filename: String): File? { + /** + * Fetch (or continue fetching) [contentId]. Safe to call repeatedly: each call transfers what + * the link allows and leaves the rest for the next one. + */ + fun fetch(serverUrl: String, contentId: String, filename: String): Result { val ext = filename.substringAfterLast('.', "mp4") - val finalFile = File(cacheDir, "${contentId}.${ext}") - val partFile = File(cacheDir, "${contentId}.${ext}${PART_SUFFIX}") + val finalFile = File(cacheDir, "$contentId.$ext") + val partFile = File(cacheDir, "$contentId.$ext$PART_SUFFIX") + val tagFile = File(cacheDir, "$contentId.$ext$PART_SUFFIX$TAG_SUFFIX") + + // Only resume when we also hold the validator that was current when those bytes were + // fetched. Without it there is no way to know the asset is still the same one, and a silent + // splice of two files is worse than re-downloading. + val validator = readValidator(tagFile) + val resumeFrom = if (validator != null && partFile.exists()) partFile.length() else 0L + if (resumeFrom == 0L) { partFile.delete(); tagFile.delete() } + try { - val url = "${serverUrl}/api/content/${contentId}/file" - val request = Request.Builder().url(url).build() + val builder = Request.Builder().url("$serverUrl/api/content/$contentId/file") + if (resumeFrom > 0) { + builder.header("Range", "bytes=$resumeFrom-") + builder.header("If-Range", validator!!) + } + // .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 -> + client.newCall(builder.build()).execute().use { response -> + // Our partial is at or past the end of the asset: it belongs to something else, or + // to a truncated earlier life of this file. Discard and start clean next time. + if (response.code == 416) { + Log.w("ContentCache", "Server refused resume at $resumeFrom for $filename (416) — discarding stale partial") + partFile.delete(); tagFile.delete() + return Result.Failed + } if (!response.isSuccessful) { Log.e("ContentCache", "Download failed: ${response.code}") - return null + // The partial is kept: a 5xx or a captive-portal interception says nothing about + // the bytes we already hold. + return if (resumeFrom > 0) Result.Partial(resumeFrom, -1L, false) else Result.Failed } - // 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. + val body = response.body ?: return Result.Failed + + val appendAt: Long + val total: Long 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") + // Content-Length on a 206 is the length of the CHUNK, so the only trustworthy + // source for the full size is the total in Content-Range. + val range = parseContentRange(response.header("Content-Range")) + if (range == null || range.first != resumeFrom || range.second <= 0L) { + // A 206 we cannot verify, or one starting somewhere we did not ask for. + // Appending it blind would corrupt the file at exactly the byte count that + // makes it look complete. + Log.e("ContentCache", "Unusable 206 for $filename (Content-Range=${response.header("Content-Range")}, wanted $resumeFrom) — restarting") + partFile.delete(); tagFile.delete() + return Result.Failed + } + appendAt = resumeFrom + total = range.second + if (!tagFile.exists()) writeValidator(tagFile, response.header("ETag") ?: response.header("Last-Modified")) + } else { + // 200. Either we sent no Range, or If-Range told the server the asset changed + // and it sent the whole thing instead of a tail. Both mean: start from zero. + if (resumeFrom > 0) Log.i("ContentCache", "Asset changed under a resume for $filename — restarting from 0") + appendAt = 0L partFile.delete() - return null + total = body.contentLength() // -1 when unknown (chunked) + writeValidator(tagFile, response.header("ETag") ?: response.header("Last-Modified")) } + + var onDisk = appendAt + try { + body.byteStream().use { input -> + FileOutputStream(partFile, appendAt > 0).use { output -> + val buf = ByteArray(64 * 1024) + while (true) { + val n = input.read(buf) + if (n < 0) break + output.write(buf, 0, n) + onDisk += n + } + output.flush() + // Durable, so a power cut on a signage panel costs the last buffer + // rather than the whole partial. These are big files on bad links; the + // sync is cheap next to re-fetching 200MB. + try { output.fd.sync() } catch (_: Exception) {} + } + } + } catch (e: Exception) { + // The break we now RECOVER from instead of restarting after. Includes the read + // timeout (a stalled stream) and the call timeout (a slow drip that ran out of + // its attempt budget) — on a bad link these are the normal case, not the + // exception, and everything written so far stays. + Log.i("ContentCache", "Download interrupted for $filename at ${partFile.length()} bytes (${e.message})") + return partialOrDiscard(partFile, tagFile, partFile.length(), total, resumeFrom) + } + + // Total bytes on disk against the declared total — NOT bytes received this attempt, + // which on a resume is only the tail. + if (!CacheValidation.isComplete(onDisk, total)) { + Log.i("ContentCache", "Incomplete after this attempt ($onDisk/$total) for $filename") + return partialOrDiscard(partFile, tagFile, onDisk, total, resumeFrom) + } + finalFile.delete() if (!partFile.renameTo(finalFile)) { Log.e("ContentCache", "Rename failed for $filename") - partFile.delete() - return null + partFile.delete(); tagFile.delete() + return Result.Failed } - Log.i("ContentCache", "Downloaded: $filename -> ${finalFile.absolutePath} ($written bytes)") - return finalFile + tagFile.delete() + Log.i("ContentCache", "Downloaded: $filename -> ${finalFile.absolutePath} ($onDisk bytes)") + return Result.Done(finalFile) } } 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. + // Connect-time failures (no route, DNS, TLS) — nothing was transferred, so whatever is + // already on disk is still valid to resume from. This also catches a throw from closing + // the response after a partial body, which is why `progressed` is measured against the + // bytes on disk rather than assumed false: an attempt that advanced must not be + // reported as a stall just because the connection objected on the way out. Log.e("ContentCache", "Download error: ${e.message}") - partFile.delete() - return null + val kept = partFile.length() + return if (kept > 0) partialOrDiscard(partFile, tagFile, kept, -1L, resumeFrom) else Result.Failed } } + /** Complete-or-nothing wrapper: returns the cached file only when the asset is whole. */ + fun downloadContent(serverUrl: String, contentId: String, filename: String): File? = + (fetch(serverUrl, contentId, filename) as? Result.Done)?.file + fun deleteContent(contentId: String) { // Exact-prefix (with the dot) so we don't delete a different id's file — and this also - // sweeps the "..part" temp. + // sweeps the "..part" temp and its ".part.tag" validator. cacheDir.listFiles { _, name -> name.startsWith("$contentId.") }?.forEach { it.delete() } Log.i("ContentCache", "Deleted cached content: $contentId") } @@ -109,14 +220,55 @@ class ContentCache internal constructor( return cacheDir.listFiles()?.sumOf { it.length() } ?: 0L } + /** + * Report an unfinished attempt — keeping the bytes only if the NEXT attempt can build on them. + * + * Without a validator there is no safe resume, so the partial is dead weight: the next attempt + * would restart from zero, re-fetch the same prefix, and land in exactly the same place. Worse, + * counting that as progress would make the coordinator chain attempts against a link that is + * getting nowhere. Discard it and report no progress, so it backs off like the failure it is. + */ + private fun partialOrDiscard(partFile: File, tagFile: File, onDisk: Long, total: Long, resumeFrom: Long): Result { + if (!tagFile.exists()) { + partFile.delete() + return Result.Partial(0L, total, false) + } + return Result.Partial(onDisk, total, onDisk > resumeFrom) + } + + private fun readValidator(tagFile: File): String? = + try { if (tagFile.exists()) tagFile.readText().trim().ifEmpty { null } else null } catch (_: Exception) { null } + + private fun writeValidator(tagFile: File, value: String?) { + // No validator (a server that sends neither ETag nor Last-Modified) means no safe resume: + // leave the sidecar absent and the next attempt starts over rather than splicing blind. + try { if (value.isNullOrBlank()) tagFile.delete() else tagFile.writeText(value) } catch (_: Exception) {} + } + companion object { private const val PART_SUFFIX = ".part" + private const val TAG_SUFFIX = ".tag" + + /** + * "bytes -/" -> (start, total). Null for anything else, including the + * "*" total a server may send, which gives us nothing to validate completeness against. + */ + internal fun parseContentRange(header: String?): Pair? { + val m = Regex("""^\s*bytes\s+(\d+)-(\d+)/(\d+)\s*$""").find(header ?: return null) ?: return null + val start = m.groupValues[1].toLongOrNull() ?: return null + val total = m.groupValues[3].toLongOrNull() ?: return null + return start to total + } 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 + // Root-2 gave this a hard OVERALL cap so a slow drip could not hang forever. With resume + // it caps ONE ATTEMPT rather than the whole asset: a link that only manages 20MB per + // call now keeps those 20MB and continues, where before the cap was the reason a large + // file could never finish. + .callTimeout(5, TimeUnit.MINUTES) .build() } } diff --git a/android/app/src/main/java/com/remotedisplay/player/data/DownloadCoordinator.kt b/android/app/src/main/java/com/remotedisplay/player/data/DownloadCoordinator.kt index a0c4b10..b457731 100644 --- a/android/app/src/main/java/com/remotedisplay/player/data/DownloadCoordinator.kt +++ b/android/app/src/main/java/com/remotedisplay/player/data/DownloadCoordinator.kt @@ -64,17 +64,52 @@ class DownloadCoordinator( } } + /** + * Run attempts back-to-back for as long as each one is putting bytes on disk. + * + * A site whose link only carries part of an asset per call needs many attempts to finish one + * file. Handing each attempt back to the 60s playlist sweep would stretch a 200MB video over + * hours of mostly-idle waiting, and running to the exponential backoff would stretch it to + * never — which is the failure this whole change is about. Progress is the signal: keep going + * while bytes are landing, stop the moment one attempt achieves nothing. + * + * inFlight is held across the whole chain, so this is still exactly one writer per `.part` and + * a concurrent sweep still finds the item busy rather than starting a duplicate. + */ 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) + var link = 0 + while (link < MAX_RESUME_CHAIN) { + link++ + val result = cache.fetch(serverUrl(), contentId, filename) + when (result) { + is ContentCache.Result.Done -> { + 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") + return + } + is ContentCache.Result.Partial -> { + if (!result.progressed) { + // Bytes are held but this attempt added none: the link is down rather + // than slow, so back off properly instead of spinning on it. + onFailure(contentId) + return + } + DebugLog.v("DownloadCoordinator", "resume $contentId: ${result.bytesOnDisk}/${result.totalBytes} after attempt $link") + // Deliberately NOT acked as failed. A download that is advancing is not a + // failure, and telling the CMS otherwise is what put items in the dashboard + // showing "failed" while they were in fact still arriving. + } + is ContentCache.Result.Failed -> { onFailure(contentId); return } + } } + // Still going when the chain ran out. Not a failure — hand it back to the next sweep + // with a short, FIXED delay rather than the exponential one, so a large asset over a + // slow link keeps advancing instead of decaying into the 5-minute cap. + nextAttemptAt[contentId] = now() + RESUME_HANDBACK_MS + DebugLog.v("DownloadCoordinator", "resume chain exhausted for $contentId — continuing on the next sweep") } catch (e: Throwable) { Log.w("DownloadCoordinator", "download $contentId failed: ${e.message}") onFailure(contentId) @@ -125,5 +160,13 @@ class DownloadCoordinator( const val MAX_CONCURRENT = 3 const val BACKOFF_BASE_MS = 15_000L const val BACKOFF_MAX_MS = 5 * 60_000L + /** + * Consecutive resuming attempts before the item goes back on the sweep. Bounded so one + * enormous asset on a slow link cannot hold an executor slot indefinitely and starve the + * other two items in a playlist — with a 5-minute attempt cap this is still up to an hour + * of continuous transfer per dispatch. + */ + const val MAX_RESUME_CHAIN = 12 + const val RESUME_HANDBACK_MS = 5_000L } } diff --git a/android/app/src/test/java/com/remotedisplay/player/data/ContentDownloadTest.kt b/android/app/src/test/java/com/remotedisplay/player/data/ContentDownloadTest.kt index c9ac893..d2375f2 100644 --- a/android/app/src/test/java/com/remotedisplay/player/data/ContentDownloadTest.kt +++ b/android/app/src/test/java/com/remotedisplay/player/data/ContentDownloadTest.kt @@ -2,7 +2,9 @@ package com.remotedisplay.player.data import okhttp3.OkHttpClient import org.junit.After +import org.junit.Assert.assertArrayEquals 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 @@ -14,11 +16,13 @@ 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). + * Root-2 REPRODUCE-THEN-PROVE for the "stuck downloading / frozen" caching bug, extended with the + * RESUME case that came out of a customer on an unstable one-bar 5G link: their screens showed the + * waiting state instead of playing, and the reason was not the playback path at all — the asset + * could never finish downloading, so there was never anything cached to play. + * + * Each test drives the REAL ContentCache against a local HTTP server that reproduces a specific + * failure mode on a HEALTHY socket (the socket is fine — the DOWNLOAD misbehaves). * * 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 @@ -62,13 +66,79 @@ class ContentDownloadTest { return "http://127.0.0.1:${s.localPort}" } + /** The request headers of each call the client made, in order. */ + private val seen = java.util.Collections.synchronizedList(ArrayList>()) + + /** + * A server that behaves like a bad link: it honours Range/If-Range correctly, but never sends + * more than [bytesPerCall] before dropping the connection mid-body. Nothing is wrong with the + * server or the file — the transfer simply cannot complete in one call, which is the whole + * shape of the reported fault. + * + * [etagOf] is read per request so a test can change the asset underneath a resume. + */ + private fun serveFlaky(body: () -> ByteArray, bytesPerCall: Int, etagOf: () -> String): String { + val s = ServerSocket(0) + server = s + Thread { + while (!s.isClosed) { + try { + s.accept().use { sock -> + val headers = HashMap() + val reader = sock.getInputStream().bufferedReader() + reader.readLine() // request line + while (true) { + val line = reader.readLine() ?: break + if (line.isEmpty()) break + val i = line.indexOf(':') + if (i > 0) headers[line.substring(0, i).trim().lowercase()] = line.substring(i + 1).trim() + } + seen.add(headers) + + val content = body() + val etag = etagOf() + val out = sock.getOutputStream() + val range = headers["range"] + val ifRange = headers["if-range"] + // If-Range with a stale validator means "send the whole thing" — the + // mechanism that stops a resume splicing two different assets together. + val honourRange = range != null && (ifRange == null || ifRange == etag) + val start = if (honourRange) Regex("""bytes=(\d+)-""").find(range!!)?.groupValues?.get(1)?.toInt() ?: 0 else 0 + + if (honourRange && start >= content.size) { + out.write("HTTP/1.1 416 Range Not Satisfiable\r\nContent-Range: bytes */${content.size}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".toByteArray()) + out.flush() + return@use + } + + val remaining = content.size - start + if (honourRange) { + out.write(("HTTP/1.1 206 Partial Content\r\n" + + "Content-Range: bytes $start-${content.size - 1}/${content.size}\r\n" + + "Content-Length: $remaining\r\nETag: $etag\r\nConnection: close\r\n\r\n").toByteArray()) + } else { + out.write(("HTTP/1.1 200 OK\r\nContent-Length: ${content.size}\r\n" + + "ETag: $etag\r\nConnection: close\r\n\r\n").toByteArray()) + } + // ...then send only part of what we just declared, and hang up. + val send = minOf(bytesPerCall, remaining) + out.write(content, start, send) + out.flush() + } + } catch (_: Exception) { /* closed between tests */ } + } + }.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("HTTP/1.1 200 OK\r\nContent-Length: $contentLength\r\nETag: \"w1\"\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() + private fun tagFiles() = dir.listFiles { _, name -> name.endsWith(".part.tag") }?.toList() ?: emptyList() // ---- positive control: a complete download IS cached ---- @Test fun `complete download is cached with the right size and no leftover part file`() { @@ -78,10 +148,11 @@ class ContentDownloadTest { assertEquals(5L, file!!.length()) assertNotNull(cache.getCachedFile("cidA")) assertTrue("no .part temp should remain", partFiles().isEmpty()) + assertTrue("no validator sidecar should remain", tagFiles().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`() { + @Test fun `truncated download is NOT promoted to the cache — but its bytes are KEPT to resume from`() { val url = serveOnce { it.writeHttp(100, ByteArray(40) { 'x'.code.toByte() }) // close after 40 of the declared 100 bytes -> truncation @@ -89,13 +160,17 @@ class ContentDownloadTest { 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()) + // The bytes stay. Deleting them was correct while there was no way to continue from them + // and catastrophic once the link is the limiting factor: it made every attempt start at + // zero, so an asset larger than one call's worth could never be cached at all. + assertEquals("the 40 received bytes must be kept for the next attempt", 1, partFiles().size) + assertEquals(40L, partFiles().first().length()) } // ---- 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("HTTP/1.1 200 OK\r\nContent-Length: 100\r\nETag: \"v1\"\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 } @@ -105,7 +180,115 @@ class ContentDownloadTest { 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()) + assertEquals("the 10 bytes that did arrive are kept", 10L, partFiles().first().length()) + } + + // ---- THE BUG: a link that can never carry the whole asset in one call ---- + @Test fun `an asset larger than any single call still completes, one resumed attempt at a time`() { + val body = ByteArray(1000) { (it % 251).toByte() } + val url = serveFlaky({ body }, bytesPerCall = 300, etagOf = { "\"v1\"" }) + + // Each attempt gets 300 bytes and the connection dies. Restart-from-zero would loop here + // forever, cache nothing, and leave the screen on "waiting for content" — the customer's + // report. Resume needs four. + var result: ContentCache.Result = ContentCache.Result.Failed + var attempts = 0 + while (attempts < 10) { + attempts++ + result = cache.fetch(url, "big", "movie.bin") + if (result is ContentCache.Result.Done) break + assertTrue("every attempt must make progress", (result as ContentCache.Result.Partial).progressed) + } + + assertTrue("the asset must eventually be cached, not retried forever", result is ContentCache.Result.Done) + assertEquals(4, attempts) + assertArrayEquals("the reassembled file must be byte-identical to the original", + body, cache.getCachedFile("big")!!.readBytes()) + assertTrue("no temp files survive completion", partFiles().isEmpty() && tagFiles().isEmpty()) + } + + @Test fun `each attempt asks for exactly the bytes it does not have yet`() { + seen.clear() + val body = ByteArray(1000) { (it % 251).toByte() } + val url = serveFlaky({ body }, bytesPerCall = 400, etagOf = { "\"v1\"" }) + repeat(3) { cache.fetch(url, "big", "movie.bin") } + + assertNull("the first call has nothing to resume from", seen[0]["range"]) + assertEquals("bytes=400-", seen[1]["range"]) + assertEquals("bytes=800-", seen[2]["range"]) + // Without If-Range the server cannot tell us the asset changed, and a resume would append + // the tail of a new file to the head of an old one. + assertEquals("\"v1\"", seen[1]["if-range"]) + } + + // ---- the corruption a resume could cause, and the guard that stops it ---- + @Test fun `an asset that changes under a resume restarts from zero instead of splicing`() { + val v1 = ByteArray(1000) { 'a'.code.toByte() } + val v2 = ByteArray(1000) { 'b'.code.toByte() } + var current = v1 + var etag = "\"v1\"" + val url = serveFlaky({ current }, bytesPerCall = 400, etagOf = { etag }) + + cache.fetch(url, "swap", "movie.bin") // 400 bytes of v1 on disk + assertEquals(400L, partFiles().first().length()) + + current = v2; etag = "\"v2\"" // replaced between attempts + val second = cache.fetch(url, "swap", "movie.bin") + assertTrue(second is ContentCache.Result.Partial) + // If-Range mismatch -> the server sent the WHOLE new asset, so we started over and hold + // 400 bytes of v2, not 400 of v1 with a v2 tail to come. A splice would have been exactly + // 1000 bytes and passed every completeness check we have. + assertEquals(400L, partFiles().first().length()) + + repeat(3) { cache.fetch(url, "swap", "movie.bin") } + assertArrayEquals("the cached asset must be all-v2, with no v1 bytes spliced in", + v2, cache.getCachedFile("swap")!!.readBytes()) + } + + @Test fun `a partial longer than the asset is discarded rather than resumed forever`() { + // The server answers 416. Keeping the partial would mean asking for a range past the end on + // every future attempt and never recovering. + val body = ByteArray(100) { 'z'.code.toByte() } + val url = serveFlaky({ body }, bytesPerCall = 500, etagOf = { "\"v1\"" }) + java.io.File(dir, "over.bin.part").writeBytes(ByteArray(400)) + java.io.File(dir, "over.bin.part.tag").writeText("\"v1\"") + + val first = cache.fetch(url, "over", "movie.bin") + assertTrue("an over-long partial is a hard failure, not a resume", first is ContentCache.Result.Failed) + assertTrue("the stale partial must be discarded", partFiles().isEmpty()) + + assertTrue(cache.fetch(url, "over", "movie.bin") is ContentCache.Result.Done) + assertArrayEquals(body, cache.getCachedFile("over")!!.readBytes()) + } + + @Test fun `a server that offers no validator discards the partial rather than hoarding it`() { + // No ETag and no Last-Modified: there is nothing to detect a changed asset with, so a + // resume would be a guess and the next attempt has to start over anyway. Bytes that cannot + // be built upon are not progress — keeping them would leave dead weight on disk, and + // COUNTING them as progress would make the coordinator chain attempts against a link that + // is getting nowhere. It backs off like the failure it is. + val s = ServerSocket(0) + server = s + Thread { + while (!s.isClosed) { + try { + s.accept().use { sock -> + val reader = sock.getInputStream().bufferedReader() + while (true) { val line = reader.readLine() ?: break; if (line.isEmpty()) break } + val out = sock.getOutputStream() + out.write("HTTP/1.1 200 OK\r\nContent-Length: 100\r\nConnection: close\r\n\r\n".toByteArray()) + out.write(ByteArray(40)); out.flush() + } + } catch (_: Exception) {} + } + }.apply { isDaemon = true; start() } + val url = "http://127.0.0.1:${s.localPort}" + + val r = cache.fetch(url, "noval", "movie.bin") + assertTrue(r is ContentCache.Result.Partial) + assertFalse("re-fetching the same prefix forever is not progress", (r as ContentCache.Result.Partial).progressed) + assertTrue("an unusable partial must not be left on disk", partFiles().isEmpty()) + assertNull("and nothing incomplete is ever served as cached", cache.getCachedFile("noval")) } // ---- prefix cross-match guard: an id that prefixes another must not match ---- @@ -116,4 +299,20 @@ class ContentDownloadTest { assertNotNull(cache.getCachedFile("abc")) assertNull("id 'ab' must NOT match cached 'abc.x'", cache.getCachedFile("ab")) } + + @Test fun `the validator sidecar is never mistaken for the cached asset`() { + // ".part.tag" does not END with ".part", so the old endsWith() exclusion would have handed + // the player a few bytes of ETag to decode as a video. + java.io.File(dir, "sid.bin.part").writeBytes(ByteArray(10)) + java.io.File(dir, "sid.bin.part.tag").writeText("\"v1\"") + assertNull("neither temp may be served as content", cache.getCachedFile("sid")) + } + + @Test fun `Content-Range parsing rejects anything it cannot verify a total from`() { + assertEquals(400L to 1000L, ContentCache.parseContentRange("bytes 400-999/1000")) + assertNull("an unknown total gives nothing to check completeness against", + ContentCache.parseContentRange("bytes 400-999/*")) + assertNull(ContentCache.parseContentRange("items 0-1/2")) + assertNull(ContentCache.parseContentRange(null)) + } } diff --git a/server/test/content-range-resume.test.js b/server/test/content-range-resume.test.js new file mode 100644 index 0000000..00557c4 --- /dev/null +++ b/server/test/content-range-resume.test.js @@ -0,0 +1,124 @@ +'use strict'; + +// A player on a bad link cannot download a large asset in one unbroken call. It has to be able to +// come back and ask for "the rest", which means GET /api/content/:id/file must honour Range — and +// must reject a resume against an asset that changed underneath it, or the player would splice two +// different files together and cache the result as whole. +// +// Both behaviours come from res.sendFile / the `send` module rather than from code in this repo, +// which is exactly why they are worth pinning: they are load-bearing for offline resilience and a +// future middleware (compression, a custom Content-Length, a stream wrapper) would silently take +// them away. Nothing in the route would look wrong afterwards — downloads would simply start over +// from zero forever on the sites that need resume most. + +const os = require('node:os'); +const path = require('node:path'); +const fs = require('node:fs'); +const crypto = require('node:crypto'); +process.env.DATA_DIR = path.join(os.tmpdir(), 'st-range-' + crypto.randomBytes(4).toString('hex')); +process.env.SELF_HOSTED = 'true'; +process.env.NODE_ENV = 'test'; + +const { test, before, after } = require('node:test'); +const assert = require('node:assert/strict'); +const http = require('node:http'); +const express = require('express'); +const { db } = require('../db/database'); +const config = require('../config'); + +const ID = 'range-fixture'; +const BODY = Buffer.from(Array.from({ length: 1000 }, (_, i) => i % 251)); // 1000 known bytes +let server, base, filePath; + +/** Raw request so we can see the status line and headers, not just a parsed body. */ +function req(headers) { + return new Promise((resolve, reject) => { + http.get(`${base}/${ID}/file`, { headers }, (res) => { + const chunks = []; + res.on('data', (c) => chunks.push(c)); + res.on('end', () => resolve({ status: res.statusCode, headers: res.headers, body: Buffer.concat(chunks) })); + }).on('error', reject); + }); +} + +before(async () => { + fs.mkdirSync(config.contentDir, { recursive: true }); + filePath = path.join(config.contentDir, 'range-fixture.bin'); + fs.writeFileSync(filePath, BODY); + // workspace_id NULL is readable by any authenticated caller (checkContentRead), which keeps the + // fixture to one row. + db.prepare('INSERT INTO content (id, filename, mime_type, file_size, filepath) VALUES (?,?,?,?,?)') + .run(ID, 'range-fixture.bin', 'application/octet-stream', BODY.length, 'range-fixture.bin'); + + const app = express(); + app.use((r, _res, next) => { r.workspaceId = 'ws'; r.user = { id: 'u', role: 'admin' }; next(); }); + app.use('/', require('../routes/content')); + server = http.createServer(app); + await new Promise((r) => server.listen(0, r)); + base = `http://127.0.0.1:${server.address().port}`; +}); + +after(() => new Promise((r) => server.close(r))); + +test('the route advertises range support', async () => { + const r = await req({}); + assert.equal(r.status, 200); + assert.equal(r.headers['accept-ranges'], 'bytes'); + assert.equal(r.body.length, BODY.length); +}); + +test('a resume request returns 206 with exactly the remaining bytes', async () => { + // This is the player's second attempt: 400 bytes already on disk in the .part, ask for the rest. + const r = await req({ Range: 'bytes=400-' }); + assert.equal(r.status, 206, 'a resume must not be answered with the whole file'); + assert.equal(r.headers['content-range'], `bytes 400-999/${BODY.length}`); + assert.deepEqual(r.body, BODY.subarray(400), 'the tail must line up byte-for-byte with the head already cached'); +}); + +test('the declared total in Content-Range is what the player validates completeness against', async () => { + // The player has no other trustworthy source for the full size: Content-Length on a 206 is the + // length of the CHUNK. Parsing the total out of Content-Range is what stops a resumed download + // being promoted while still short. + const r = await req({ Range: 'bytes=999-' }); + assert.equal(r.status, 206); + assert.match(r.headers['content-range'], /\/1000$/); + assert.equal(r.body.length, 1); +}); + +test('a request starting past the end is refused, not answered with an empty body', async () => { + // The player treats this as "my .part is stale or longer than the asset" and starts over. If the + // server answered 200/206-with-nothing instead, the bad .part would survive every retry. + const r = await req({ Range: 'bytes=5000-' }); + assert.equal(r.status, 416); +}); + +test('If-Range with a stale validator falls back to the WHOLE file instead of a splice', async () => { + // The one that actually protects the cache. If the asset changed since the .part was started, + // appending the tail of the new file to the head of the old one produces a corrupt asset whose + // byte count is nonetheless exactly right — it would pass the completeness check and be played. + // If-Range makes the server answer 200, and the player restarts from zero. + const r = await req({ Range: 'bytes=400-', 'If-Range': '"not-the-current-etag"' }); + assert.equal(r.status, 200, 'a changed entity must yield the full body, not a 206 tail'); + assert.equal(r.body.length, BODY.length); +}); + +test('If-Range with the CURRENT validator still resumes', async () => { + const head = await req({}); + const etag = head.headers.etag; + assert.ok(etag, 'no ETag means the player has no validator to send and every resume restarts'); + const r = await req({ Range: 'bytes=400-', 'If-Range': etag }); + assert.equal(r.status, 206); + assert.deepEqual(r.body, BODY.subarray(400)); +}); + +test('remote-url content still 404s rather than serving a range of nothing', async () => { + db.prepare('INSERT INTO content (id, filename, mime_type, file_size, remote_url) VALUES (?,?,?,?,?)') + .run('range-remote', 'feed', 'text/html', 0, 'https://example.com/'); + const r = await new Promise((resolve, reject) => { + http.get(`${base}/range-remote/file`, { headers: { Range: 'bytes=0-' } }, (res) => { + res.resume(); + res.on('end', () => resolve({ status: res.statusCode })); + }).on('error', reject); + }); + assert.equal(r.status, 404); +});