fix: stream gateway downloads to disk instead of slurping into RAM (#914)
* fix: stream gateway downloads to disk instead of slurping into RAM Opening or saving a ~100MB MEDIA file OOMs the app: fetch() read the whole response body into a single ByteArray (peak heap ~2x file size) before writing it to cache. A 102MB docx (~97MiB) passed the server's 100MiB cap but blew the phone heap (192MiB growth limit) on both open and save. fetch() now streams the body to a cache file in 64KiB chunks and GatewayFile carries the written file instead of bytes; save copies the cache file to the destination chunked. Cancellation-safe (partial file deleted, CancellationException never swallowed), UUID-named cache files, 24h sweep. Closes #913 * fix: stub app.cacheDir/applicationContext in attachment tests The streaming fetch builds File(cacheDir, ...) from the application context's cacheDir; the affected tests never stubbed it, so the mock returned a bogus File (null path) and java.io.File's constructor NPE'd before fetch was even called. Stub cacheDir + applicationContext (returning app itself, matching the existing success-path test) in the five open/save tests. copyChunked is also stubbed to copy for real in the save-writes test since mockkObject mocks the whole object. --------- Co-authored-by: M57 <hy4ri@users.noreply.github.com>
This commit is contained in:
1 parent
dadcc3e588
commit
b653903fa8
4 files changed
+225
-55
No files matched your search
@@ -1,12 +1,17 @@
|
||||
package com.m57.hermescontrol.data.remote
|
||||
|
||||
import com.m57.hermescontrol.data.local.AuthManager
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.ensureActive
|
||||
import okhttp3.Request
|
||||
import java.io.File
|
||||
import java.io.IOException
|
||||
import java.net.URLDecoder
|
||||
import java.net.URLEncoder
|
||||
import java.nio.charset.StandardCharsets
|
||||
import java.util.UUID
|
||||
import java.util.regex.Pattern
|
||||
import kotlin.coroutines.coroutineContext
|
||||
|
||||
/**
|
||||
* Client for the gateway's managed-files download endpoint
|
||||
@@ -98,35 +103,103 @@ object GatewayFileClient {
|
||||
else -> null
|
||||
}
|
||||
|
||||
/** Fetch a gateway-hosted file using the current [AuthManager] credentials. */
|
||||
suspend fun fetch(path: String): GatewayFileResult =
|
||||
fetch(path, AuthManager.getBaseUrl(), AuthManager.getToken().orEmpty())
|
||||
/** Fetch a gateway-hosted file using the current [AuthManager] credentials,
|
||||
* streaming the response body straight to [cacheDir] so large files never
|
||||
* sit in the app heap (a ~100 MB download used to OOM: the whole body was
|
||||
* slurped into one [ByteArray] first). */
|
||||
suspend fun fetch(
|
||||
path: String,
|
||||
cacheDir: File,
|
||||
): GatewayFileResult = fetch(path, AuthManager.getBaseUrl(), AuthManager.getToken().orEmpty(), cacheDir)
|
||||
|
||||
/** Fetch a gateway-hosted file with explicit credentials (testable). */
|
||||
suspend fun fetch(
|
||||
path: String,
|
||||
baseUrl: String,
|
||||
token: String,
|
||||
cacheDir: File,
|
||||
): GatewayFileResult {
|
||||
val url =
|
||||
buildDownloadUrl(baseUrl, token, path)
|
||||
?: return GatewayFileResult.Failure(IllegalArgumentException("not an absolute gateway path: $path"))
|
||||
val call = OkHttpProvider.base.newCall(Request.Builder().url(url).build())
|
||||
val resp = call.execute()
|
||||
return try {
|
||||
val request = Request.Builder().url(url).build()
|
||||
OkHttpProvider.base.newCall(request).execute().use { resp ->
|
||||
classifyStatus(resp.code)?.let { return it }
|
||||
if (!resp.isSuccessful) {
|
||||
return GatewayFileResult.Failure(IOException("HTTP ${resp.code}"))
|
||||
}
|
||||
val body = resp.body.bytes()
|
||||
val name =
|
||||
resp.header("Content-Disposition")?.let { parseFilename(it) }
|
||||
?: fileNameFromPath(path)
|
||||
val mime = resp.header("Content-Type") ?: "application/octet-stream"
|
||||
GatewayFileResult.Success(GatewayFile(name, mime, body))
|
||||
classifyStatus(resp.code)?.let { return it }
|
||||
if (!resp.isSuccessful) {
|
||||
return GatewayFileResult.Failure(IOException("HTTP ${resp.code}"))
|
||||
}
|
||||
val name =
|
||||
resp.header("Content-Disposition")?.let { parseFilename(it) }
|
||||
?: fileNameFromPath(path)
|
||||
val mime = resp.header("Content-Type") ?: "application/octet-stream"
|
||||
val cacheFile =
|
||||
streamBodyToCache(resp, cacheDir, name)
|
||||
?: return GatewayFileResult.Failure(IOException("failed to write $name to cache"))
|
||||
GatewayFileResult.Success(GatewayFile(name, mime, cacheFile))
|
||||
} catch (e: CancellationException) {
|
||||
// Never swallow cancellation — the caller's job was cancelled.
|
||||
throw e
|
||||
} catch (e: Throwable) {
|
||||
GatewayFileResult.Failure(e)
|
||||
} finally {
|
||||
resp.close()
|
||||
}
|
||||
}
|
||||
|
||||
/** Stream the response body into a fresh file under [cacheDir] in 64 KiB
|
||||
* chunks (peak heap stays ~constant no matter the file size), sweeping
|
||||
* cache files older than a day first. Returns the written file, or null
|
||||
* on any I/O failure. Cancellation-safe: the copy loop checks the calling
|
||||
* coroutine's job, and a cancelled fetch deletes the partial file. */
|
||||
private suspend fun streamBodyToCache(
|
||||
resp: okhttp3.Response,
|
||||
cacheDir: File,
|
||||
name: String,
|
||||
): File? {
|
||||
val body = resp.body ?: return null
|
||||
val safeName = name.replace(Regex("[/\\\\]"), "_").ifBlank { "file" }
|
||||
val dir = cacheDir.also { it.mkdirs() }
|
||||
sweepOldCacheFiles(dir)
|
||||
val target = File(dir, "${UUID.randomUUID().toString().take(8)}-$safeName")
|
||||
return try {
|
||||
body.byteStream().use { input ->
|
||||
target.outputStream().use { output ->
|
||||
copyChunked(input, output)
|
||||
}
|
||||
}
|
||||
target
|
||||
} catch (e: CancellationException) {
|
||||
target.delete()
|
||||
throw e
|
||||
} catch (e: Throwable) {
|
||||
target.delete()
|
||||
null
|
||||
}
|
||||
}
|
||||
|
||||
/** Copy [input] to [output] in 64 KiB chunks so peak heap stays constant
|
||||
* regardless of file size, aborting if the calling coroutine is cancelled
|
||||
* (used by both the download stream and save-to-document). */
|
||||
internal suspend fun copyChunked(
|
||||
input: java.io.InputStream,
|
||||
output: java.io.OutputStream,
|
||||
) {
|
||||
val buffer = ByteArray(64 * 1024)
|
||||
while (true) {
|
||||
coroutineContext.ensureActive()
|
||||
val read = input.read(buffer)
|
||||
if (read < 0) break
|
||||
output.write(buffer, 0, read)
|
||||
}
|
||||
}
|
||||
|
||||
/** Best-effort cleanup of gateway cache files older than a day (the cache
|
||||
* dir is evictable by the OS anyway; this keeps it from growing forever). */
|
||||
private fun sweepOldCacheFiles(dir: File) {
|
||||
val cutoff = System.currentTimeMillis() - 24 * 60 * 60 * 1000L
|
||||
dir.listFiles()?.forEach { file ->
|
||||
if (file.isFile && file.lastModified() < cutoff) file.delete()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -148,25 +221,13 @@ object GatewayFileClient {
|
||||
private val FILENAME_RE = Regex("""filename\*?=(?:UTF-8'')?"?([^";]+)"?""", RegexOption.IGNORE_CASE)
|
||||
}
|
||||
|
||||
/** A file fetched from the gateway. */
|
||||
/** A file fetched from the gateway, already streamed to disk under
|
||||
* [cacheFile] (never held in memory — see [GatewayFileClient.fetch]). */
|
||||
data class GatewayFile(
|
||||
val name: String,
|
||||
val mimeType: String,
|
||||
val bytes: ByteArray,
|
||||
) {
|
||||
override fun equals(other: Any?): Boolean {
|
||||
if (this === other) return true
|
||||
if (other !is GatewayFile) return false
|
||||
return name == other.name && mimeType == other.mimeType && bytes.contentEquals(other.bytes)
|
||||
}
|
||||
|
||||
override fun hashCode(): Int {
|
||||
var r = name.hashCode()
|
||||
r = 31 * r + mimeType.hashCode()
|
||||
r = 31 * r + bytes.contentHashCode()
|
||||
return r
|
||||
}
|
||||
}
|
||||
val cacheFile: File,
|
||||
)
|
||||
|
||||
/** Outcome of [GatewayFileClient.fetch]. */
|
||||
sealed interface GatewayFileResult {
|
||||
|
||||
@@ -3023,8 +3023,8 @@ class ChatViewModel(
|
||||
* already grants read access for the picked document. If that fails
|
||||
* (e.g. the permission lapsed), we copy to cache and retry via
|
||||
* FileProvider so the tap is never a silent no-op.
|
||||
* - GATEWAY (agent `MEDIA:`) files: fetch the bytes via
|
||||
* [GatewayFileClient], write them to a cache file, and open with
|
||||
* - GATEWAY (agent `MEDIA:`) files: stream the file to cache via
|
||||
* [GatewayFileClient] (chunked — never held in memory), then open with
|
||||
* [android.content.Intent.ACTION_VIEW] through FileProvider — so a
|
||||
* remote phone can view agent-delivered files in-place.
|
||||
*
|
||||
@@ -3041,8 +3041,9 @@ class ChatViewModel(
|
||||
}
|
||||
// GATEWAY, or LOCAL direct-open failed → fetch/copy then open.
|
||||
val path = gatewayPathFor(attachment)
|
||||
val cacheDir = java.io.File(ctx.cacheDir, "gateway_files")
|
||||
viewModelScope.launch(ioDispatcher) {
|
||||
when (val result = GatewayFileClient.fetch(path)) {
|
||||
when (val result = GatewayFileClient.fetch(path, cacheDir)) {
|
||||
is GatewayFileResult.Success -> {
|
||||
openBytes(ctx, result.file)
|
||||
}
|
||||
@@ -3077,16 +3078,22 @@ class ChatViewModel(
|
||||
) {
|
||||
if (_uiState.value.savingAttachmentPath != null) return
|
||||
val path = gatewayPathFor(attachment)
|
||||
val cacheDir =
|
||||
java.io.File(getApplication<Application>().applicationContext.cacheDir, "gateway_files")
|
||||
_uiState.update { it.copy(savingAttachmentPath = path) }
|
||||
viewModelScope.launch(ioDispatcher) {
|
||||
try {
|
||||
when (val result = GatewayFileClient.fetch(path)) {
|
||||
when (val result = GatewayFileClient.fetch(path, cacheDir)) {
|
||||
is GatewayFileResult.Success -> {
|
||||
val resolver = getApplication<Application>().contentResolver
|
||||
runCatching {
|
||||
resolver
|
||||
.openOutputStream(destination, "wt")
|
||||
?.use { it.write(result.file.bytes) }
|
||||
?.use { output ->
|
||||
result.file.cacheFile.inputStream().use { input ->
|
||||
GatewayFileClient.copyChunked(input, output)
|
||||
}
|
||||
}
|
||||
?: error("destination is unavailable")
|
||||
}.onSuccess {
|
||||
showOpenError("Saved ${result.file.name}")
|
||||
@@ -3122,21 +3129,17 @@ class ChatViewModel(
|
||||
?: attachment.uri.removePrefix("gateway:").takeIf { it != attachment.uri }
|
||||
?: attachment.name
|
||||
|
||||
/** Open bytes written to a cache file via FileProvider + ACTION_VIEW. */
|
||||
/** Open a gateway file already streamed to cache via FileProvider + ACTION_VIEW. */
|
||||
private fun openBytes(
|
||||
ctx: android.content.Context,
|
||||
file: GatewayFile,
|
||||
) {
|
||||
runCatching {
|
||||
val dir = java.io.File(ctx.cacheDir, "gateway_files").also { it.mkdirs() }
|
||||
val safeName = file.name.replace(Regex("[/\\\\]"), "_").ifBlank { "file" }
|
||||
val out = java.io.File(dir, safeName)
|
||||
out.writeBytes(file.bytes)
|
||||
val uri =
|
||||
androidx.core.content.FileProvider.getUriForFile(
|
||||
ctx,
|
||||
"${ctx.packageName}.fileprovider",
|
||||
out,
|
||||
file.cacheFile,
|
||||
)
|
||||
openWithView(ctx, uri, file.mimeType)
|
||||
}.onFailure { showOpenError("Could not open ${file.name}: ${it.message}") }
|
||||
|
||||
@@ -5,6 +5,12 @@ import com.m57.hermescontrol.data.config.ServerStoreState
|
||||
import com.m57.hermescontrol.data.local.AuthManager
|
||||
import io.mockk.every
|
||||
import io.mockk.mockk
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.test.runCurrent
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertNull
|
||||
@@ -101,4 +107,50 @@ class GatewayFileClientTest {
|
||||
assertEquals("report.pdf", GatewayFileClient.parseFilename("attachment; filename=\"report.pdf\""))
|
||||
assertEquals("a b.png", GatewayFileClient.parseFilename("inline; filename*=UTF-8''a%20b.png"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `copyChunked copies input to output byte-for-byte`() =
|
||||
runTest {
|
||||
// >64 KiB so the copy exercises multiple chunks (buffer is 64 KiB).
|
||||
val payload = ByteArray(200_000) { (it % 251).toByte() }
|
||||
val output = java.io.ByteArrayOutputStream()
|
||||
GatewayFileClient.copyChunked(java.io.ByteArrayInputStream(payload), output)
|
||||
assertTrue(output.toByteArray().contentEquals(payload))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `copyChunked aborts mid-copy when the job is cancelled`() =
|
||||
runTest {
|
||||
// Input that parks on its 2nd chunk read until the gate opens, so
|
||||
// the test can cancel the job *while* the copy is in flight.
|
||||
val gate = CompletableDeferred<Unit>()
|
||||
val blockingInput =
|
||||
object : java.io.InputStream() {
|
||||
var reads = 0
|
||||
|
||||
override fun read(): Int = 1
|
||||
|
||||
override fun read(
|
||||
b: ByteArray,
|
||||
off: Int,
|
||||
len: Int,
|
||||
): Int {
|
||||
reads++
|
||||
if (reads == 2) kotlinx.coroutines.runBlocking { gate.await() }
|
||||
return if (reads <= 3) 1 else -1
|
||||
}
|
||||
}
|
||||
val output = java.io.ByteArrayOutputStream()
|
||||
val job =
|
||||
CoroutineScope(Dispatchers.Default).launch {
|
||||
GatewayFileClient.copyChunked(blockingInput, output)
|
||||
}
|
||||
runCurrent()
|
||||
job.cancel()
|
||||
gate.complete(Unit)
|
||||
job.join()
|
||||
// If copyChunked swallowed the cancellation, the job would have
|
||||
// completed normally; a cancelled completion proves propagation.
|
||||
assertTrue(job.isCancelled)
|
||||
}
|
||||
}
|
||||
@@ -3396,10 +3396,10 @@ class ChatViewModelTest {
|
||||
every { app.cacheDir } returns cacheDir
|
||||
|
||||
mockkObject(GatewayFileClient)
|
||||
val bytes = "hello".toByteArray()
|
||||
val file = java.io.File(cacheDir, "note.txt").apply { writeBytes("hello".toByteArray()) }
|
||||
coEvery {
|
||||
GatewayFileClient.fetch(any())
|
||||
} returns GatewayFileResult.Success(GatewayFile("note.txt", "text/plain", bytes))
|
||||
GatewayFileClient.fetch(any(), any())
|
||||
} returns GatewayFileResult.Success(GatewayFile("note.txt", "text/plain", file))
|
||||
|
||||
val intentSlot = slot<Intent>()
|
||||
// android.jar stubs Intent ctor/setters to throw "not mocked" in unit tests;
|
||||
@@ -3430,7 +3430,7 @@ class ChatViewModelTest {
|
||||
advanceUntilIdle()
|
||||
|
||||
// fetch was invoked (proves we entered the IO launch)
|
||||
coVerify { GatewayFileClient.fetch(any()) }
|
||||
coVerify { GatewayFileClient.fetch(any(), any()) }
|
||||
// ACTION_VIEW intent delivered, with the right type + grant flag.
|
||||
verify { app.startActivity(any()) }
|
||||
verify { intentSlot.captured.setDataAndType(any(), eq("text/plain")) }
|
||||
@@ -3443,9 +3443,24 @@ class ChatViewModelTest {
|
||||
fun `saveAttachment writes gateway file to selected document`() =
|
||||
runTest {
|
||||
mockkObject(GatewayFileClient)
|
||||
val bytes = "downloaded".toByteArray()
|
||||
coEvery { GatewayFileClient.fetch("/tmp/note.txt") } returns
|
||||
GatewayFileResult.Success(GatewayFile("note.txt", "text/plain", bytes))
|
||||
val cacheDir =
|
||||
java.io.File(
|
||||
System.getProperty("java.io.tmpdir"),
|
||||
"gw_save_${System.nanoTime()}",
|
||||
).apply { mkdirs() }
|
||||
every { app.cacheDir } returns cacheDir
|
||||
every { app.applicationContext } returns app
|
||||
every { app.getApplicationContext() } returns app
|
||||
val file =
|
||||
java.io.File(System.getProperty("java.io.tmpdir"), "note_${System.nanoTime()}.txt")
|
||||
.apply { writeText("downloaded") }
|
||||
coEvery { GatewayFileClient.fetch("/tmp/note.txt", any()) } returns
|
||||
GatewayFileResult.Success(GatewayFile("note.txt", "text/plain", file))
|
||||
// copyChunked is mocked with the object — restore real copy so the
|
||||
// saved document actually receives the bytes.
|
||||
coEvery { GatewayFileClient.copyChunked(any(), any()) } coAnswers {
|
||||
firstArg<java.io.InputStream>().copyTo(secondArg<java.io.OutputStream>())
|
||||
}
|
||||
val resolver = mockk<android.content.ContentResolver>()
|
||||
val output = java.io.ByteArrayOutputStream()
|
||||
val destination = mockk<android.net.Uri>()
|
||||
@@ -3465,7 +3480,7 @@ class ChatViewModelTest {
|
||||
vm.saveAttachment(attachment, destination)
|
||||
advanceUntilIdle()
|
||||
|
||||
assertArrayEquals(bytes, output.toByteArray())
|
||||
assertArrayEquals("downloaded".toByteArray(), output.toByteArray())
|
||||
assertNull(vm.uiState.value.savingAttachmentPath)
|
||||
assertEquals("Saved note.txt", vm.uiState.value.openError)
|
||||
}
|
||||
@@ -3474,11 +3489,19 @@ class ChatViewModelTest {
|
||||
fun `saveAttachment never deletes the selected document when download fails`() =
|
||||
runTest {
|
||||
mockkObject(GatewayFileClient)
|
||||
val cacheDir =
|
||||
java.io.File(
|
||||
System.getProperty("java.io.tmpdir"),
|
||||
"gw_save_${System.nanoTime()}",
|
||||
).apply { mkdirs() }
|
||||
every { app.cacheDir } returns cacheDir
|
||||
every { app.applicationContext } returns app
|
||||
every { app.getApplicationContext() } returns app
|
||||
val resolver = mockk<android.content.ContentResolver>(relaxed = true)
|
||||
val destination = mockk<android.net.Uri>()
|
||||
every { app.contentResolver } returns resolver
|
||||
coEvery {
|
||||
GatewayFileClient.fetch("/tmp/missing.pdf")
|
||||
GatewayFileClient.fetch("/tmp/missing.pdf", any())
|
||||
} returns GatewayFileResult.NotFound
|
||||
|
||||
val (viewModel, _) = createViewModelWithSession()
|
||||
@@ -3504,12 +3527,27 @@ class ChatViewModelTest {
|
||||
fun `saveAttachment never deletes the selected document when writing fails`() =
|
||||
runTest {
|
||||
mockkObject(GatewayFileClient)
|
||||
val cacheDir =
|
||||
java.io.File(
|
||||
System.getProperty("java.io.tmpdir"),
|
||||
"gw_save_${System.nanoTime()}",
|
||||
).apply { mkdirs() }
|
||||
every { app.cacheDir } returns cacheDir
|
||||
every { app.applicationContext } returns app
|
||||
every { app.getApplicationContext() } returns app
|
||||
val resolver = mockk<android.content.ContentResolver>(relaxed = true)
|
||||
val destination = mockk<android.net.Uri>()
|
||||
every { app.contentResolver } returns resolver
|
||||
every { resolver.openOutputStream(destination, "wt") } throws IllegalStateException("write failed")
|
||||
coEvery { GatewayFileClient.fetch("/tmp/report.pdf") } returns
|
||||
GatewayFileResult.Success(GatewayFile("report.pdf", "application/pdf", byteArrayOf(1)))
|
||||
coEvery { GatewayFileClient.fetch("/tmp/report.pdf", any()) } returns
|
||||
GatewayFileResult.Success(
|
||||
GatewayFile(
|
||||
"report.pdf",
|
||||
"application/pdf",
|
||||
java.io.File(System.getProperty("java.io.tmpdir"), "report_${System.nanoTime()}.pdf")
|
||||
.apply { writeBytes(byteArrayOf(1)) },
|
||||
),
|
||||
)
|
||||
|
||||
val viewModel = createViewModel()
|
||||
viewModel.saveAttachment(
|
||||
@@ -3532,9 +3570,17 @@ class ChatViewModelTest {
|
||||
fun `saveAttachment ignores overlapping saves`() =
|
||||
runTest {
|
||||
mockkObject(GatewayFileClient)
|
||||
val cacheDir =
|
||||
java.io.File(
|
||||
System.getProperty("java.io.tmpdir"),
|
||||
"gw_save_${System.nanoTime()}",
|
||||
).apply { mkdirs() }
|
||||
every { app.cacheDir } returns cacheDir
|
||||
every { app.applicationContext } returns app
|
||||
every { app.getApplicationContext() } returns app
|
||||
val firstResult = CompletableDeferred<GatewayFileResult>()
|
||||
val fetchedPaths = mutableListOf<String>()
|
||||
coEvery { GatewayFileClient.fetch(capture(fetchedPaths)) } coAnswers { firstResult.await() }
|
||||
coEvery { GatewayFileClient.fetch(capture(fetchedPaths), any()) } coAnswers { firstResult.await() }
|
||||
val resolver = mockk<android.content.ContentResolver>(relaxed = true)
|
||||
every { app.contentResolver } returns resolver
|
||||
val viewModel = createViewModel()
|
||||
@@ -3566,8 +3612,16 @@ class ChatViewModelTest {
|
||||
fun `openAttachment GATEWAY not-found surfaces openError`() =
|
||||
runTest {
|
||||
mockkObject(GatewayFileClient)
|
||||
val cacheDir =
|
||||
java.io.File(
|
||||
System.getProperty("java.io.tmpdir"),
|
||||
"gw_open_${System.nanoTime()}",
|
||||
).apply { mkdirs() }
|
||||
every { app.cacheDir } returns cacheDir
|
||||
every { app.applicationContext } returns app
|
||||
every { app.getApplicationContext() } returns app
|
||||
coEvery {
|
||||
GatewayFileClient.fetch(any())
|
||||
GatewayFileClient.fetch(any(), any())
|
||||
} returns GatewayFileResult.NotFound
|
||||
|
||||
val vm = createViewModel()
|
||||
|
||||
Reference in new issue
Block a user