mirror of
https://github.com/screentinker/screentinker.git
synced 2026-08-13 22:03:13 -06:00
Resume interrupted content downloads instead of restarting from zero
A site on a marginal link (the report came from a one-bar 5G install) could never fill its cache. Every attempt started at byte 0 and the .part was deleted on any interruption, so an asset larger than one call's worth of transfer was discarded and re-fetched forever — five minutes of progress thrown away, back off, five more minutes, thrown away. With nothing cached, the player showed the waiting state, which is what got reported as "the screens go black instead of playing cached content". The offline playback path was never the problem; the cache simply could not be filled. An interrupted download now keeps its .part and the next attempt asks for the rest with Range. Two ways that could corrupt the cache, both closed: If-Range with a stored validator makes a changed asset come back as a full 200 (restart) rather than a spliceable tail, and a partial longer than the asset gets a 416 and is discarded. Bytes are kept only when they can be built upon — with no validator there is no safe resume, so the partial is dropped and the attempt backs off as the failure it is, rather than re-fetching the same prefix forever. DownloadCoordinator now distinguishes progress from failure: attempts chain while bytes are landing (bounded, single-flight held throughout) and only a no-progress attempt escalates the exponential backoff or acks "failed" — an advancing download is not a failed one and should not be shown as such. Server side is unchanged; res.sendFile already serves Range/If-Range, and content-range-resume.test.js pins that since it is load-bearing and a future middleware could silently remove it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Uaeo9MvzKoyXuN6ZsbhtkL
This commit is contained in:
ScreenTinker
parent
3e6c97ba10
commit
cf3b2e62af
|
|
@ -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 "<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) }
|
||||
// cross-matching. `contains` rather than `endsWith` for the temp check because the resume
|
||||
// validator sidecar is "<id>.<ext>.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 "<id>.<ext>.part" temp.
|
||||
// sweeps the "<id>.<ext>.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>-<end>/<total>" -> (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<Long, Long>? {
|
||||
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()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Map<String, String>>())
|
||||
|
||||
/**
|
||||
* 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<String, String>()
|
||||
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))
|
||||
}
|
||||
}
|
||||
|
|
|
|||
124
server/test/content-range-resume.test.js
Normal file
124
server/test/content-range-resume.test.js
Normal file
|
|
@ -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);
|
||||
});
|
||||
Loading…
Reference in a new issue