From 15e5790d35cc63aecf26ee910b015af50bcb0e65 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Mon, 5 Oct 2026 19:35:00 +0000 Subject: [PATCH 01/10] fix(upload): resume chunked uploads from the parts the server already has A chunked upload restarted from zero after every interruption, although the parts it had sent were still on the server: - The PROPFIND on the upload folder named no properties. isFile is set only from a resourcetype property, so it stayed false for every member and the list of uploaded parts was always empty. The request now asks for resourcetype, getcontentlength and displayname. - Parts are accepted only when their name is well formed and their size matches the name. A wrong or foreign entry is uploaded again or ignored. - Only the missing parts are sent. - The assembling MOVE sends OC-Total-Length, so the server can check that the parts add up to the file. - A MOVE the server rejects with 400 removes the parts once and sends the file again. The caller supplies the "already restarted" mark. - stop() interrupts the upload between two parts and keeps the parts on the server. Only abortUpload() removes them. - upload() no longer swallows exceptions. A failed request is thrown (an HttpException for an HTTP error, an IOException for a network error) and the parts stay on the server. A lost network while listing the parts is thrown too instead of being taken for "no parts", which would send the whole file again. Signed-off-by: Krainov Gleb Assisted-by: Claude-Code:claude-sonnet-5-5 --- .../upload/chunked/ChunkedFileUploader.kt | 175 +++++--- .../chunked/ChunkedFileUploaderAbortTest.kt | 2 +- .../chunked/ChunkedFileUploaderResumeTest.kt | 382 ++++++++++++++++++ .../upload/chunked/ChunkedFileUploaderTest.kt | 16 +- 4 files changed, 511 insertions(+), 64 deletions(-) create mode 100644 app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt diff --git a/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt b/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt index c6b2db38be1..b51ee00299f 100644 --- a/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt +++ b/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt @@ -16,6 +16,7 @@ import at.bitfire.dav4jvm.exception.DavException import at.bitfire.dav4jvm.exception.HttpException import at.bitfire.dav4jvm.exception.NotFoundException import at.bitfire.dav4jvm.property.DisplayName +import at.bitfire.dav4jvm.property.GetContentLength import at.bitfire.dav4jvm.property.GetContentType import at.bitfire.dav4jvm.property.GetLastModified import at.bitfire.dav4jvm.property.ResourceType @@ -46,17 +47,22 @@ import java.io.RandomAccessFile import java.nio.channels.FileChannel import java.util.Locale +@Suppress("TooManyFunctions") @AutoInjector(NextcloudTalkApplication::class) class ChunkedFileUploader( okHttpClient: OkHttpClient, val currentUser: User, val listener: OnDataTransferProgressListener, - val ncApiCoroutines: NcApiCoroutines + val ncApiCoroutines: NcApiCoroutines, + private val isRestarted: () -> Boolean = { false }, + private val markRestarted: () -> Unit = {} ) { private var okHttpClientNoRedirects: OkHttpClient? = null private var remoteChunkUrl: String private var uploadFolderUri: String = "" + + @Volatile private var isUploadAborted = false init { @@ -65,50 +71,60 @@ class ChunkedFileUploader( } /** - * Uploads [localFile] in chunks, skipping chunks that are already on the server, and assembles them at - * [targetPath]. + * Uploads the parts of [localFile] that the server does not have yet, then assembles them into [targetPath]. + * The upload folder on the server is keyed by the file, so a repeated call resumes a previous one. * - * @return false if the upload was aborted or the server answered with an HTTP error - * @throws IOException on I/O failures (e.g. connection loss or timeout) and chunk folder creation failures, - * so the caller can retry; chunks already on the server are skipped as long as [localFile] is unchanged + * @return true when the file was assembled, false when the upload was stopped or aborted + * @throws IOException on I/O failures (e.g. connection loss or timeout), so the caller can retry; the parts + * already on the server are kept and skipped by the retry as long as [localFile] is unchanged + * @throws Exception when another request fails; the parts already on the server are kept */ - @Suppress("Detekt.TooGenericExceptionCaught") @Throws(IOException::class) fun upload(localFile: File, mimeType: MediaType?, targetPath: String): Boolean { - try { - var isUploadSuccessful = true - uploadFolderUri = remoteChunkUrl + "/" + FileUtils.md5Sum(localFile) - val davResource = DavResource( - okHttpClientNoRedirects!!, - uploadFolderUri.toHttpUrlOrNull()!! - ) + uploadFolderUri = folderUriOf(localFile) + return try { + uploadParts(localFile, mimeType, targetPath) + } catch (e: HttpException) { + // The server refused the assembly (e.g. the length does not match its parts): the parts it holds are + // not usable, so remove them once per upload (the worker remembers it) and send the file again. + if (e.code != HTTP_BAD_REQUEST || isRestarted()) { + throw e + } + Log.w(TAG, "Server rejected the assembly, uploading the file again", e) + markRestarted() + deleteUploadFolder() + uploadParts(localFile, mimeType, targetPath) + } + } + + private fun folderUriOf(localFile: File) = remoteChunkUrl + "/" + FileUtils.md5Sum(localFile) + + private fun uploadParts(localFile: File, mimeType: MediaType?, targetPath: String): Boolean { + val davResource = DavResource( + okHttpClientNoRedirects!!, + uploadFolderUri.toHttpUrlOrNull()!! + ) - createFolder(davResource) + createFolder(davResource) - val chunksOnServer: MutableList = getUploadedChunks(davResource, uploadFolderUri) - Log.d(TAG, "chunksOnServer: " + chunksOnServer.size) + val chunksOnServer: MutableList = getUploadedChunks(davResource, uploadFolderUri, localFile.length()) + Log.d(TAG, "chunksOnServer: " + chunksOnServer.size) - val missingChunks: List = checkMissingChunks(chunksOnServer, localFile.length()) - Log.d(TAG, "missingChunks: " + missingChunks.size) + val missingChunks: List = checkMissingChunks(chunksOnServer, localFile.length()) + Log.d(TAG, "missingChunks: " + missingChunks.size) - for (missingChunk in missingChunks) { - if (isUploadAborted) { - isUploadSuccessful = false - break - } - uploadChunk(localFile, uploadFolderUri, mimeType, missingChunk, missingChunk.length()) + for (missingChunk in missingChunks) { + if (isUploadAborted) { + break } + uploadChunk(localFile, uploadFolderUri, mimeType, missingChunk, missingChunk.length()) + } - if (isUploadSuccessful) { - assembleChunks(uploadFolderUri, targetPath) - } - return isUploadSuccessful - } catch (e: IOException) { - throw e - } catch (e: Exception) { - Log.e(TAG, "Something went wrong in ChunkedFileUploader", e) - return false + val isComplete = !isUploadAborted + if (isComplete) { + assembleChunks(uploadFolderUri, targetPath, localFile.length()) } + return isComplete } @Suppress("Detekt.ThrowsCount") @@ -134,14 +150,21 @@ class ChunkedFileUploader( } @Suppress("Detekt.ComplexMethod", "Detekt.ReturnCount") - private fun getUploadedChunks(davResource: DavResource, uploadFolderUri: String): MutableList { + private fun getUploadedChunks( + davResource: DavResource, + uploadFolderUri: String, + fileLength: Long + ): MutableList { val davResponse = DavResponse() val memberElements: MutableList = ArrayList() val rootElement = arrayOfNulls(1) val remoteFiles: MutableList = ArrayList() try { davResource.propfind( - 1 + 1, + ResourceType.NAME, + GetContentLength.NAME, + DisplayName.NAME ) { response: at.bitfire.dav4jvm.Response, hrefRelation: at.bitfire.dav4jvm.Response.HrefRelation? -> davResponse.setResponse(response) when (hrefRelation) { @@ -152,13 +175,14 @@ class ChunkedFileUploader( } Unit } - } catch (e: IOException) { - // PROPFIND on Nextcloud chunked-upload folders can return unexpected responses - // (e.g. 200 instead of 207). Treat any failure as "no chunks uploaded yet" so - // we fall back to a full upload rather than aborting entirely. - Log.w(TAG, "PROPFIND failed — assuming no chunks on server, will upload from scratch: ${e.message}") - return ArrayList() } catch (e: DavException) { + if (e is HttpException && e.code >= HTTP_SERVER_ERROR) { + throw e + } + // An IOException is not caught on purpose: a lost network must repeat the request, + // not send the whole file again. + // PROPFIND on Nextcloud chunked-upload folders can return unexpected responses + // (e.g. 200 instead of 207). Treat any such answer as "no chunks uploaded yet". Log.w(TAG, "PROPFIND failed — assuming no chunks on server, will upload from scratch: ${e.message}") return ArrayList() } @@ -177,20 +201,29 @@ class ChunkedFileUploader( val chunksOnServer: MutableList = ArrayList() for (remoteFile in remoteFiles) { - if (!".file".equals(remoteFile.displayName, ignoreCase = true) && remoteFile.isFile) { - val part: List = remoteFile.displayName!!.split("-") - chunksOnServer.add( - Chunk( - part[0].toLong(), - part[1].toLong() - ) - ) + if (remoteFile.isFile) { + parseUploadedChunk(remoteFile.displayName, remoteFile.size, fileLength)?.let { chunksOnServer.add(it) } } } return chunksOnServer } - private fun checkMissingChunks(chunks: List, length: Long): List { + /** + * Returns the part a file on the server stands for, or null when it is not a part (e.g. `.file`), or its size + * differs from what its name says, so it has to be uploaded again. The last part is named one byte longer than + * its content. + */ + internal fun parseUploadedChunk(name: String?, size: Long?, fileLength: Long): Chunk? { + if (name == null || !CHUNK_NAME_REGEX.matches(name)) { + return null + } + val chunk = + Chunk(name.substring(0, CHUNK_NUMBER_LENGTH).toLong(), name.substring(CHUNK_NUMBER_LENGTH + 1).toLong()) + val expectedSize = minOf(chunk.length(), fileLength - chunk.start) + return if (size == expectedSize) chunk else null + } + + internal fun checkMissingChunks(chunks: List, length: Long): List { val missingChunks: MutableList = java.util.ArrayList() var start: Long = 0 while (start <= length) { @@ -306,7 +339,7 @@ class ChunkedFileUploader( this.okHttpClientNoRedirects = builder.build() } - private fun assembleChunks(uploadFolderUri: String, targetPath: String) { + private fun assembleChunks(uploadFolderUri: String, targetPath: String, totalLength: Long) { val destinationUri = ApiUtils.getUrlForFileUpload( currentUser.baseUrl!!, currentUser.userId!!, @@ -316,8 +349,13 @@ class ChunkedFileUploader( createRemoteFolder(targetPath) val originUri = "$uploadFolderUri/.file" + val client = okHttpClientNoRedirects!!.newBuilder() + .addInterceptor { chain -> + chain.proceed(chain.request().newBuilder().header(TOTAL_LENGTH_HEADER, totalLength.toString()).build()) + } + .build() DavResource( - okHttpClientNoRedirects!!, + client, originUri.toHttpUrlOrNull()!! ).move( destinationUri.toHttpUrlOrNull()!!, @@ -347,12 +385,20 @@ class ChunkedFileUploader( } /** - * Called from the worker's onStopped, which WorkManager runs inside a coroutine cancellation handler: - * anything thrown here crashes the process, so this function must never throw. + * Interrupts a running [upload] between two parts, without touching the parts on the server, so a later + * [upload] can resume. Called from the worker's onStopped, which WorkManager runs inside a coroutine + * cancellation handler: anything thrown here crashes the process, so this function must never throw. + */ + fun stop() { + isUploadAborted = true + } + + /** + * Interrupts a running [upload] and removes its parts from the server. Same contract as [stop]: never throws. */ @Suppress("Detekt.TooGenericExceptionCaught") fun abortUpload(onSuccess: () -> Unit) { - isUploadAborted = true + stop() val client = okHttpClientNoRedirects val folderUrl = uploadFolderUri.toHttpUrlOrNull() if (client == null || folderUrl == null) { @@ -378,6 +424,14 @@ class ChunkedFileUploader( } } + private fun deleteUploadFolder() { + try { + DavResource(okHttpClientNoRedirects!!, uploadFolderUri.toHttpUrlOrNull()!!).delete { _ -> } + } catch (e: NotFoundException) { + Log.i(TAG, "Chunk upload folder is already gone", e) + } + } + private fun getModelFromResponse(response: at.bitfire.dav4jvm.Response, remotePath: String): RemoteFileBrowserItem { val remoteFileBrowserItem = RemoteFileBrowserItem() remoteFileBrowserItem.path = Uri.decode(remotePath) @@ -413,6 +467,10 @@ class ChunkedFileUploader( remoteFileBrowserItem.modifiedTimestamp = property.lastModified } + is GetContentLength -> { + remoteFileBrowserItem.size = property.contentLength + } + is GetContentType -> { remoteFileBrowserItem.mimeType = property.type } @@ -448,5 +506,10 @@ class ChunkedFileUploader( private const val READ_PERMISSION = "R" private const val CHUNK_SIZE: Long = 1024000 private const val METHOD_NOT_ALLOWED_CODE: Int = 405 + private const val HTTP_BAD_REQUEST: Int = 400 + private const val HTTP_SERVER_ERROR: Int = 500 + private const val CHUNK_NUMBER_LENGTH = 16 + private val CHUNK_NAME_REGEX = Regex("^\\d{16}-\\d{16}$") + private const val TOTAL_LENGTH_HEADER = "OC-Total-Length" } } diff --git a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderAbortTest.kt b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderAbortTest.kt index 6bfbbd747e9..31df335f085 100644 --- a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderAbortTest.kt +++ b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderAbortTest.kt @@ -139,7 +139,7 @@ class ChunkedFileUploaderAbortTest { server.enqueue(MockResponse().setResponseCode(NOT_FOUND)) server.enqueue(MockResponse().setResponseCode(FORBIDDEN)) val uploader = uploader() - uploader.upload(file, "text/plain".toMediaType(), "/Talk/test.txt") + runCatching { uploader.upload(file, "text/plain".toMediaType(), "/Talk/test.txt") } repeat(server.requestCount) { server.takeRequest(1, TimeUnit.SECONDS) } return uploader } diff --git a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt new file mode 100644 index 00000000000..0fe6309b910 --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt @@ -0,0 +1,382 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: GPL-3.0-or-later + */ +package com.nextcloud.talk.upload.chunked + +import android.app.Application +import at.bitfire.dav4jvm.exception.HttpException +import com.nextcloud.talk.api.NcApiCoroutines +import com.nextcloud.talk.data.user.model.User +import okhttp3.OkHttpClient +import okhttp3.mockwebserver.Dispatcher +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import okhttp3.mockwebserver.RecordedRequest +import okhttp3.mockwebserver.SocketPolicy +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Assert.fail +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import org.junit.rules.TemporaryFolder +import org.junit.runner.RunWith +import org.mockito.kotlin.mock +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config +import java.io.File +import java.io.IOException +import java.util.Collections +import java.util.concurrent.TimeUnit + +@Suppress("TooManyFunctions") +@RunWith(RobolectricTestRunner::class) +@Config(application = Application::class, sdk = [33]) +class ChunkedFileUploaderResumeTest { + + @get:Rule + val tempFolder = TemporaryFolder() + + private val server = MockWebServer() + private val requests: MutableList = Collections.synchronizedList(mutableListOf()) + private lateinit var client: OkHttpClient + private lateinit var user: User + private lateinit var uploader: ChunkedFileUploader + private lateinit var file: File + + /** Parts the fake server already holds: name -> size. */ + private var partsOnServer: Map = emptyMap() + private var putResponseCode = CREATED + private var propfindResponseCode: Int? = null + private var propfindDisconnects = false + private var putDisconnects = false + + /** Answers of the next MOVE requests; 201 when empty. */ + private val moveResponseCodes: MutableList = Collections.synchronizedList(mutableListOf()) + + @Before + fun setUp() { + server.dispatcher = object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + requests.add(request) + return when (request.method) { + "PROPFIND" -> when { + propfindDisconnects -> MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AFTER_REQUEST) + propfindResponseCode != null -> MockResponse().setResponseCode(propfindResponseCode!!) + else -> propfindResponse(request.path!!) + } + "MOVE" -> MockResponse().setResponseCode( + if (moveResponseCodes.isEmpty()) CREATED else moveResponseCodes.removeAt(0) + ) + "PUT" -> if (putDisconnects) { + MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AFTER_REQUEST) + } else { + MockResponse().setResponseCode(putResponseCode) + } + "MKCOL" -> MockResponse().setResponseCode(METHOD_NOT_ALLOWED) + else -> MockResponse().setResponseCode(CREATED) + } + } + } + server.start() + client = OkHttpClient.Builder() + .retryOnConnectionFailure(false) + .connectTimeout(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .readTimeout(TIMEOUT_SECONDS, TimeUnit.SECONDS) + .build() + user = User( + id = 1, + userId = "alice", + username = "alice", + token = "token", + baseUrl = server.url("/").toString().trimEnd('/') + ) + uploader = ChunkedFileUploader(client, user, mock(), mock()) + file = tempFolder.newFile("video.mp4").apply { writeBytes(ByteArray(FILE_SIZE)) } + } + + @After + fun tearDown() { + server.shutdown() + } + + @Test + fun `uploads only the parts missing on the server`() { + partsOnServer = mapOf(PART_0 to CHUNK, PART_1 to CHUNK) + + assertTrue(uploader.upload(file, null, "/Talk/video.mp4")) + + assertEquals(listOf("/$PART_2"), putPaths()) + } + + @Test + fun `uploads everything when the server has no parts`() { + assertTrue(uploader.upload(file, null, "/Talk/video.mp4")) + + assertEquals(listOf("/$PART_0", "/$PART_1", "/$PART_2"), putPaths()) + } + + @Test + fun `uploads from scratch when the server removed the upload folder`() { + propfindResponseCode = NOT_FOUND + + assertTrue(uploader.upload(file, null, "/Talk/video.mp4")) + + assertEquals(listOf("/$PART_0", "/$PART_1", "/$PART_2"), putPaths()) + } + + @Test + fun `uploads nothing but the assembly when the server has all parts`() { + partsOnServer = mapOf(PART_0 to CHUNK, PART_1 to CHUNK, PART_2 to LAST_CHUNK) + + assertTrue(uploader.upload(file, null, "/Talk/video.mp4")) + + assertEquals(emptyList(), putPaths()) + assertEquals(1, requests.count { it.method == "MOVE" }) + } + + @Test + fun `a part with a wrong size on the server is uploaded again`() { + partsOnServer = mapOf(PART_0 to CHUNK, PART_1 to CHUNK - 1, PART_2 to LAST_CHUNK) + + uploader.upload(file, null, "/Talk/video.mp4") + + assertEquals(listOf("/$PART_1"), putPaths()) + } + + @Test + fun `foreign names in the upload folder do not break the upload`() { + partsOnServer = mapOf(PART_0 to CHUNK, "readme.txt" to 3, "1-2" to 2, ".file" to FILE_SIZE) + + assertTrue(uploader.upload(file, null, "/Talk/video.mp4")) + + assertEquals(listOf("/$PART_1", "/$PART_2"), putPaths()) + } + + @Test + fun `assembly tells the server the total length`() { + uploader.upload(file, null, "/Talk/video.mp4") + + val move = requests.single { it.method == "MOVE" } + assertEquals(FILE_SIZE.toString(), move.getHeader("OC-Total-Length")) + assertTrue(move.path!!.endsWith("/.file")) + } + + @Test + fun `a failing request throws and keeps the parts on the server`() { + putResponseCode = SERVER_ERROR + + try { + uploader.upload(file, null, "/Talk/video.mp4") + fail("expected the upload to throw") + } catch (e: HttpException) { + assertEquals(SERVER_ERROR, e.code) + } + assertEquals(0, requests.count { it.method == "DELETE" }) + } + + @Test + fun `a lost network while sending a part is an error and keeps the parts on the server`() { + putDisconnects = true + + try { + uploader.upload(file, null, "/Talk/video.mp4") + fail("expected the upload to throw") + } catch (e: IOException) { + // retried later by the worker + } + assertEquals(0, requests.count { it.method == "DELETE" || it.method == "MOVE" }) + } + + @Test + fun `stop interrupts the upload without removing the parts`() { + uploader.stop() + + assertFalse(uploader.upload(file, null, "/Talk/video.mp4")) + + assertEquals(0, requests.count { it.method == "DELETE" || it.method == "MOVE" }) + assertEquals(emptyList(), putPaths()) + } + + @Test + fun `abort removes the parts from the server`() { + uploader.upload(file, null, "/Talk/video.mp4") + requests.clear() + + uploader.abortUpload {} + + assertEquals(1, requests.count { it.method == "DELETE" }) + } + + @Test + fun `a rejected assembly removes the parts once and uploads the file again`() { + moveResponseCodes.add(BAD_REQUEST) + + assertTrue(uploader.upload(file, null, "/Talk/video.mp4")) + + assertEquals(1, requests.count { it.method == "DELETE" }) + assertEquals(2, requests.count { it.method == "MOVE" }) + assertEquals(2, requests.count { it.method == "PROPFIND" }) + } + + @Test + fun `an assembly rejected after an earlier restart is an error at once`() { + moveResponseCodes.add(BAD_REQUEST) + var marked = 0 + val restarted = ChunkedFileUploader( + client, + user, + mock(), + mock(), + isRestarted = { true }, + markRestarted = { marked++ } + ) + + try { + restarted.upload(file, null, "/Talk/video.mp4") + fail("expected the upload to throw") + } catch (e: HttpException) { + assertEquals(BAD_REQUEST, e.code) + } + assertEquals(0, requests.count { it.method == "DELETE" }) + assertEquals(0, marked) + } + + @Test + fun `a restart after a rejected assembly is reported once`() { + moveResponseCodes.add(BAD_REQUEST) + var marked = 0 + val first = ChunkedFileUploader( + client, + user, + mock(), + mock(), + markRestarted = { marked++ } + ) + + assertTrue(first.upload(file, null, "/Talk/video.mp4")) + + assertEquals(1, marked) + } + + @Test + fun `a server error while listing the parts is an error`() { + propfindResponseCode = SERVER_ERROR + + try { + uploader.upload(file, null, "/Talk/video.mp4") + fail("expected the upload to throw") + } catch (e: HttpException) { + assertEquals(SERVER_ERROR, e.code) + } + assertEquals(emptyList(), putPaths()) + } + + @Test + fun `an assembly rejected twice is an error`() { + moveResponseCodes.addAll(listOf(BAD_REQUEST, BAD_REQUEST)) + + try { + uploader.upload(file, null, "/Talk/video.mp4") + fail("expected the upload to throw") + } catch (e: HttpException) { + assertEquals(BAD_REQUEST, e.code) + } + assertEquals(1, requests.count { it.method == "DELETE" }) + } + + @Test + fun `a lost network while listing the parts is an error and does not restart the upload`() { + propfindDisconnects = true + + try { + uploader.upload(file, null, "/Talk/video.mp4") + fail("expected the upload to throw") + } catch (e: IOException) { + // retried later by the worker + } + assertEquals(emptyList(), putPaths()) + } + + @Test + fun `missing parts are planned around existing ones`() { + val onServer = listOf(Chunk(0, CHUNK - 1L), Chunk(2L * CHUNK, FILE_SIZE.toLong())) + + val missing = uploader.checkMissingChunks(onServer, FILE_SIZE.toLong()) + + assertEquals(listOf(Chunk(CHUNK.toLong(), 2L * CHUNK - 1)), missing) + } + + @Test + fun `all parts are planned when nothing is on the server`() { + val missing = uploader.checkMissingChunks(emptyList(), FILE_SIZE.toLong()) + + assertEquals( + listOf(Chunk(0, CHUNK - 1L), Chunk(CHUNK.toLong(), 2L * CHUNK - 1), Chunk(2L * CHUNK, FILE_SIZE.toLong())), + missing + ) + } + + @Test + fun `only well formed parts of the right size are accepted`() { + val length = FILE_SIZE.toLong() + assertEquals(Chunk(0, CHUNK - 1L), uploader.parseUploadedChunk(PART_0, CHUNK.toLong(), length)) + assertEquals(Chunk(2L * CHUNK, length), uploader.parseUploadedChunk(PART_2, LAST_CHUNK.toLong(), length)) + assertNull(uploader.parseUploadedChunk(PART_0, CHUNK - 1L, length)) + assertNull(uploader.parseUploadedChunk(".file", length, length)) + assertNull(uploader.parseUploadedChunk("readme.txt", 3, length)) + assertNull(uploader.parseUploadedChunk("0-1", 2, length)) + assertNull(uploader.parseUploadedChunk(null, 1, length)) + } + + private fun putPaths(): List = + requests.filter { it.method == "PUT" }.map { "/" + it.path!!.substringAfterLast('/') } + + private fun propfindResponse(path: String): MockResponse { + val folder = path.trimEnd('/') + val members = partsOnServer.entries.joinToString("") { (name, size) -> + """ + + $folder/$name + $size + HTTP/1.1 200 OK + + """ + } + val body = """ + + + $folder/ + + HTTP/1.1 200 OK + + $members + """ + return MockResponse() + .setResponseCode(MULTI_STATUS) + .setHeader("Content-Type", "application/xml; charset=utf-8") + .setBody(body) + } + + companion object { + private const val CHUNK = 1_024_000 + private const val FILE_SIZE = 2 * CHUNK + 952_000 + private const val LAST_CHUNK = 952_000 + private const val PART_0 = "0000000000000000-0000000001023999" + private const val PART_1 = "0000000001024000-0000000002047999" + private const val PART_2 = "0000000002048000-0000000003000000" + private const val TIMEOUT_SECONDS = 10L + private const val CREATED = 201 + private const val MULTI_STATUS = 207 + private const val METHOD_NOT_ALLOWED = 405 + private const val NOT_FOUND = 404 + private const val BAD_REQUEST = 400 + private const val SERVER_ERROR = 500 + } +} diff --git a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderTest.kt b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderTest.kt index 0e2e0613d50..d965419a0b8 100644 --- a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderTest.kt +++ b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderTest.kt @@ -7,6 +7,7 @@ package com.nextcloud.talk.upload.chunked import android.app.Application +import at.bitfire.dav4jvm.exception.HttpException import com.nextcloud.talk.api.NcApiCoroutines import com.nextcloud.talk.data.user.model.User import okhttp3.MediaType.Companion.toMediaType @@ -16,7 +17,6 @@ import okhttp3.mockwebserver.MockWebServer import okhttp3.mockwebserver.SocketPolicy import org.junit.After import org.junit.Assert.assertEquals -import org.junit.Assert.assertFalse import org.junit.Assert.assertThrows import org.junit.Assert.assertTrue import org.junit.Before @@ -101,20 +101,21 @@ class ChunkedFileUploaderTest { } @Test - fun `returns false when the server rejects the chunk with an HTTP error`() { + fun `throws the HTTP error when the server rejects the chunk`() { val file = localFile(SMALL_FILE_SIZE) enqueueFolderCreated() enqueueNoChunksOnServer() server.enqueue(MockResponse().setResponseCode(FORBIDDEN)) - val result = uploader().upload(file, MIME_TYPE, TARGET_PATH) + assertThrows(HttpException::class.java) { + uploader().upload(file, MIME_TYPE, TARGET_PATH) + } - assertFalse(result) assertEquals(listOf("MKCOL", "PROPFIND", "PUT"), takeRequests().map { it.method }) } @Test - fun `returns false when the server rejects assembling the chunks with an HTTP error`() { + fun `throws the HTTP error when the server rejects assembling the chunks`() { val file = localFile(SMALL_FILE_SIZE) enqueueFolderCreated() enqueueNoChunksOnServer() @@ -122,9 +123,10 @@ class ChunkedFileUploaderTest { server.enqueue(MockResponse().setResponseCode(CREATED)) server.enqueue(MockResponse().setResponseCode(FORBIDDEN)) - val result = uploader().upload(file, MIME_TYPE, TARGET_PATH) + assertThrows(HttpException::class.java) { + uploader().upload(file, MIME_TYPE, TARGET_PATH) + } - assertFalse(result) assertEquals(listOf("MKCOL", "PROPFIND", "PUT", "MKCOL", "MOVE"), takeRequests().map { it.method }) } From bfe836629318e9958ac6ccc4d2d5b1db9e48c86b Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Mon, 5 Oct 2026 19:35:00 +0000 Subject: [PATCH 02/10] fix(upload): prepare the file of an upload once and keep it between runs The folder of the parts on the server is keyed by the name, the modification time and the length of the file. UploadAndShareFilesWorker copied the content uri to the cache and compressed it again on every run. Each run produced a file with a new modification time, so a retry used another folder and sent the whole file again. The worker now prepares the file once into a directory of its own work (cacheDir/uploads/) and uploads the same file on every run: - UploadWorkspace keeps the prepared file and the "parts were removed once" mark, and holds a lock, because WorkManager does not interrupt the thread of a stopped worker and an old run can still be busy. - A workspace stays after a stop or before a retry. It is removed when the work ends, and the workspaces of finished or unknown works are removed before the next upload. - A run that is stopped while the file is prepared does not store the half-finished file. - A copy of a content uri that could not be read completely is not used as the file to upload. - onStopped() keeps the parts on the server when the system or the network stopped the work. The parts are removed only when the user cancelled it. Signed-off-by: Krainov Gleb Assisted-by: Claude-Code:claude-sonnet-5-5 --- .../talk/jobs/UploadAndShareFilesWorker.kt | 135 +++++++++++++--- .../nextcloud/talk/upload/UploadWorkspace.kt | 151 +++++++++++++++++ .../com/nextcloud/talk/utils/FileUtils.kt | 70 ++++++-- .../nextcloud/talk/utils/ImageCompressor.kt | 8 +- .../nextcloud/talk/utils/VideoCompressor.kt | 11 +- .../talk/upload/UploadWorkspaceTest.kt | 152 ++++++++++++++++++ .../utils/FileUtilsContentResolverTest.kt | 77 +++++++++ 7 files changed, 557 insertions(+), 47 deletions(-) create mode 100644 app/src/main/java/com/nextcloud/talk/upload/UploadWorkspace.kt create mode 100644 app/src/test/java/com/nextcloud/talk/upload/UploadWorkspaceTest.kt diff --git a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt index 0fdda7cb244..c9bf85ffe6d 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt +++ b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt @@ -39,6 +39,8 @@ import com.nextcloud.talk.data.user.model.User import com.nextcloud.talk.models.json.chatpostattachment.PostConversationAttachmentRequestDto import com.nextcloud.talk.models.json.chatprobeattachmentfolder.ChatProbeAttachmentDataDto import com.nextcloud.talk.models.json.chatprobeattachmentfolder.ProbeConversationAttachmentRequestDto +import com.nextcloud.talk.upload.PreparedUpload +import com.nextcloud.talk.upload.UploadWorkspace import com.nextcloud.talk.upload.chunked.ChunkedFileUploader import com.nextcloud.talk.upload.chunked.OnDataTransferProgressListener import com.nextcloud.talk.upload.normal.FileUploader @@ -70,6 +72,7 @@ import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit import javax.inject.Inject +@Suppress("TooManyFunctions") @AutoInjector(NextcloudTalkApplication::class) class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerParameters) : Worker(context, workerParameters), @@ -112,6 +115,8 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa private var internalConversationId: String? = null private var uploadDisposable: Disposable? = null private var uploadLatch: CountDownLatch? = null + private lateinit var workspace: UploadWorkspace + private var keepWorkspace = false /** * WorkManager's own `isStopped`/`onStopped()` only becomes true/fires after @@ -128,10 +133,21 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa NextcloudTalkApplication.sharedApplication!!.componentApplication.inject(this) } + workspace = UploadWorkspace(workspaceDir(context, id)) + // WorkManager does not interrupt the thread of a stopped worker, so an old run can still be busy. + if (!workspace.tryLock()) { + Log.w(TAG, "Another run of this upload is still active, trying again later") + return Result.retry() + } try { return doUpload() } finally { referenceId?.let { cancelledReferenceIds.remove(it) } + // A run that is stopped, or will be tried again, resumes from the prepared file. + if (!keepWorkspace) { + workspace.delete() + } + workspace.unlock() } } @@ -162,12 +178,16 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa require(sourceFile.isNotEmpty()) checkNotNull(roomToken) - var sourceFileUri = sourceFile.toUri() - fileName = FileUtils.getFileName(sourceFileUri, context) - file = FileUtils.getFileFromUri(context, sourceFileUri) + val originalUri = sourceFile.toUri() + fileName = FileUtils.getFileName(originalUri, context) + deleteFinishedWorkspaces() - if (inputData.getBoolean(COMPRESS_IMAGES, false)) { - sourceFileUri = compressMediaIfPossible(sourceFileUri) + val prepared = workspace.prepareOnce { dir -> prepareFile(originalUri, dir) } ?: return failUpload() + file = prepared.file + fileName = prepared.fileName + val sourceFileUri = Uri.fromFile(prepared.file) + if (isStopped || isCancelled()) { + return stoppedResult() } val remotePath = getRemotePath(currentUser) @@ -187,7 +207,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa if (uploadSuccess && (isStopped || isCancelled())) { // Cancelled right as the upload finished - don't share a cancelled upload. - return Result.failure() + return stoppedResult() } if (uploadSuccess) { @@ -202,8 +222,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa Log.e(TAG, "Share operation failed after upload") return failUpload() } else if (isStopped || isCancelled()) { - // since work is cancelled the result would be ignored anyways - return Result.failure() + return stoppedResult() } Log.e(TAG, "Something went wrong when trying to upload file") @@ -218,13 +237,17 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa e ) if (isStopped || isCancelled()) { - Result.failure() + stoppedResult() } else if (runAttemptCount < MAX_UPLOAD_ATTEMPTS - 1) { + keepWorkspace = true Result.retry() } else { failUpload() } } catch (e: Exception) { + if (isStopped || isCancelled()) { + return stoppedResult() + } Log.e(TAG, "Something went wrong when trying to upload file", e) failUpload() } @@ -245,23 +268,71 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa } /** - * Replaces [file] and [fileName] with a compressed copy if [sourceFileUri] points to a - * compressible image or video, returning the [Uri] that should be uploaded. + * Copies the source into [dir] and, when the user asked for it, compresses it there. The result is stored by + * the workspace, so every later run of this work uploads the same file: the chunk folder on the server is keyed + * by the file, and a fresh copy or compression would change that key and start the upload from zero. */ - @Suppress("ReturnCount") - private fun compressMediaIfPossible(sourceFileUri: Uri): Uri { - val originalFile = file ?: return sourceFileUri + private fun prepareFile(sourceFileUri: Uri, dir: File): PreparedUpload? { + val original = FileUtils.getFileFromUri(context, sourceFileUri, dir) ?: return null + throwIfStopped() + val compressed = if (inputData.getBoolean(COMPRESS_IMAGES, false)) { + compressMediaIfPossible(sourceFileUri, original, dir) + } else { + null + } + throwIfStopped() + return compressed ?: PreparedUpload(original, fileName) + } + + /** Keeps a stopped run from storing a half-finished file as the prepared one (see [PreparationStopped]). */ + private fun throwIfStopped() { + if (isStopped || isCancelled()) { + throw PreparationStopped() + } + } + + private class PreparationStopped : RuntimeException("Stopped while preparing the file") + + /** Removes the workspaces of works that are finished or unknown to WorkManager, e.g. after a killed process. */ + @Suppress("Detekt.TooGenericExceptionCaught") + private fun deleteFinishedWorkspaces() { + UploadWorkspace.deleteFinished(File(context.cacheDir, WORKSPACE_DIR)) { name -> + try { + val info = WorkManager.getInstance(context).getWorkInfoById(UUID.fromString(name)).get() + info != null && !info.state.isFinished + } catch (e: Exception) { + Log.w(TAG, "Could not look up work $name, keeping its files", e) + true + } + } + } + + /** + * Returns a compressed copy of [originalFile] inside [dir] if [sourceFileUri] points to a compressible image or + * video, or null when there is nothing to compress. + */ + private fun compressMediaIfPossible(sourceFileUri: Uri, originalFile: File, dir: File): PreparedUpload? { val mimeType = FileUtils.resolveMimeType(context, sourceFileUri) val compressedFile = when { - ImageCompressor.isCompressible(mimeType) -> ImageCompressor.compress(context, originalFile) - VideoCompressor.isCompressible(mimeType) -> VideoCompressor.compress(context, originalFile) + ImageCompressor.isCompressible(mimeType) -> ImageCompressor.compress(context, originalFile, dir) + VideoCompressor.isCompressible(mimeType) -> VideoCompressor.compress(context, originalFile, dir) else -> null - } ?: return sourceFileUri + } ?: return null - file = compressedFile - fileName = compressedFile.name - return Uri.fromFile(compressedFile) + if (originalFile.parentFile == dir) { + originalFile.delete() + } + return PreparedUpload(compressedFile, compressedFile.name) + } + + /** + * A stop by the system keeps the parts on the server and the prepared file, so the next run resumes. + * The result of a work that WorkManager stopped is ignored; WorkManager schedules it again by itself. + */ + private fun stoppedResult(): Result { + keepWorkspace = !isCancelled() + return Result.failure() } private fun uploadFile( @@ -277,12 +348,14 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa uploadUsingConversationSubfolders(sourceFileUri, metaData, allowUpdate) } else if (isChunkedUploading) { Log.d(TAG, "starting chunked upload because size is " + file!!.length()) - val mimeType = context.contentResolver.getType(sourceFileUri)?.toMediaTypeOrNull() + val mimeType = FileUtils.resolveMimeType(context, sourceFileUri)?.toMediaTypeOrNull() chunkedFileUploader = ChunkedFileUploader( okHttpClient, currentUser, this, - ncApiCoroutines + ncApiCoroutines, + isRestarted = workspace::isRestarted, + markRestarted = workspace::markRestarted ) chunkedFileUploader!!.upload(file!!, mimeType, remotePath) } else { @@ -359,12 +432,14 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa val tempRemotePath = "/$draftFolderPath/$uploadId-$fileName" val uploadSuccess = if (isChunkedUploading) { - val mimeType = context.contentResolver.getType(sourceFileUri)?.toMediaTypeOrNull() + val mimeType = FileUtils.resolveMimeType(context, sourceFileUri)?.toMediaTypeOrNull() chunkedFileUploader = ChunkedFileUploader( okHttpClient, currentUser, this@UploadAndShareFilesWorker, - ncApiCoroutines + ncApiCoroutines, + isRestarted = workspace::isRestarted, + markRestarted = workspace::markRestarted ) chunkedFileUploader!!.upload(file!!, mimeType, tempRemotePath) } else { @@ -433,9 +508,15 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa chatDao.updateTempMessageSendStatus(convId, refId, status) } + /** + * The user's cancel removes the parts from the server. Any other stop (the network is gone, the system stops + * the work) only interrupts the upload and keeps them, so the next run uploads what is still missing. + */ override fun onStopped() { - if (file != null && isChunkedUploading) { + if (isCancelled()) { chunkedFileUploader?.abortUpload {} + } else { + chunkedFileUploader?.stop() } uploadDisposable?.dispose() uploadLatch?.countDown() @@ -485,6 +566,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa private const val COMPRESS_IMAGES = "COMPRESS_IMAGES" private const val ALLOW_UPDATE = "ALLOW_UPDATE" private const val CHUNK_UPLOAD_THRESHOLD_SIZE: Long = 1024 * 1024 + private const val WORKSPACE_DIR = "uploads" // Total attempts allowed for a single upload (1 initial run + retries) before giving up on a // transient network failure and marking the placeholder FAILED. @@ -596,6 +678,9 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa private fun uploadQueueName(internalConversationId: String) = "upload_queue_$internalConversationId" + private fun workspaceDir(context: Context, workId: UUID) = + File(File(context.cacheDir, WORKSPACE_DIR), workId.toString()) + // Cancellation is by the WorkRequest's own id (not enqueueUniqueWork's name) since that name // is now shared by every file queued in the same conversation - cancelling by name would // cancel the whole queue instead of just this one upload. diff --git a/app/src/main/java/com/nextcloud/talk/upload/UploadWorkspace.kt b/app/src/main/java/com/nextcloud/talk/upload/UploadWorkspace.kt new file mode 100644 index 00000000000..88bb86b5a95 --- /dev/null +++ b/app/src/main/java/com/nextcloud/talk/upload/UploadWorkspace.kt @@ -0,0 +1,151 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: GPL-3.0-or-later + */ +package com.nextcloud.talk.upload + +import java.io.File +import java.io.IOException +import java.io.RandomAccessFile +import java.nio.channels.FileLock +import java.nio.channels.OverlappingFileLockException + +/** The file an upload sends and the name it gets on the server. */ +data class PreparedUpload(val file: File, val fileName: String) + +/** + * Holds what one upload needs across runs of its worker: the prepared file (copy of the content uri, compressed + * media), whether the parts on the server were already removed once, and a lock that keeps two runs of the same + * work and the cleanup of dead workspaces apart. + * + * The prepared file is created once. Its name, size and modification time stay the same on every run, so the + * chunk folder on the server, which is keyed by them, stays the same too. + */ +class UploadWorkspace(private val dir: File) { + + private val preparedFile = File(dir, PREPARED_FILE) + private val lockFile = File(dir, LOCK_FILE) + private val restartedFile = File(dir, RESTARTED_FILE) + + private var lockChannel: RandomAccessFile? = null + private var lock: FileLock? = null + + /** + * Takes the lock of this workspace. WorkManager does not interrupt a worker thread it stopped, so the old run + * can still be busy when a new run of the same work starts. Returns false when somebody else holds the lock. + * + * The lock protects only inside this process: within one process a [FileLock] is refused with + * [OverlappingFileLockException], and that is what this relies on. Do not rely on it between processes. + */ + @Synchronized + fun tryLock(): Boolean { + if (lock == null) { + dir.mkdirs() + val channel = runCatching { RandomAccessFile(lockFile, "rw") }.getOrNull() + val acquired = channel?.let { acquire(it) } + if (acquired == null) { + channel?.close() + } else { + lockChannel = channel + lock = acquired + } + } + return lock != null + } + + private fun acquire(channel: RandomAccessFile): FileLock? = + try { + channel.channel.tryLock() + } catch (e: OverlappingFileLockException) { + // The lock is held by another workspace object of this process. + null + } catch (e: IOException) { + null + } + + @Synchronized + fun unlock() { + runCatching { lock?.release() } + runCatching { lockChannel?.close() } + lock = null + lockChannel = null + } + + /** + * Returns the prepared file of an earlier run, or runs [prepare] once and keeps its result. + * [prepare] gets an empty directory it may create files in, and returns null when the file cannot be prepared. + */ + fun prepareOnce(prepare: (File) -> PreparedUpload?): PreparedUpload? { + stored()?.let { return it } + resetFiles() + dir.mkdirs() + return prepare(dir)?.also { write(preparedFile, "${it.file.absolutePath}\n${it.fileName}") } + } + + /** Whether the parts were already removed once after the server rejected the assembly. */ + fun isRestarted(): Boolean = restartedFile.exists() + + fun markRestarted() { + dir.mkdirs() + write(restartedFile, "1") + } + + /** Call only while holding the lock. */ + fun delete() { + dir.deleteRecursively() + } + + private fun stored(): PreparedUpload? { + val lines = runCatching { preparedFile.readLines() }.getOrNull() + val file = lines?.getOrNull(0)?.let(::File) + val name = lines?.getOrNull(1) + return if (file != null && name != null && file.isFile) PreparedUpload(file, name) else null + } + + private fun resetFiles() { + dir.listFiles()?.filter { it.name !in KEPT_FILES }?.forEach { + it.deleteRecursively() + } + } + + private fun write(file: File, text: String) { + val tmp = File(dir, file.name + TMP_SUFFIX) + tmp.writeText(text) + tmp.renameTo(file) + } + + companion object { + private const val PREPARED_FILE = "prepared" + private const val LOCK_FILE = "lock" + private const val RESTARTED_FILE = "restarted" + private const val TMP_SUFFIX = ".tmp" + + /** Not part of the prepared file: survive a new preparation. */ + private val KEPT_FILES = setOf(LOCK_FILE, RESTARTED_FILE) + + /** + * Removes the workspaces of uploads that are no longer alive, e.g. left behind by a killed process. + * [isAlive] gets the name of a workspace directory, which is the id of its work. A workspace whose lock is + * held belongs to a running work and is left alone. + */ + fun deleteFinished(root: File, isAlive: (String) -> Boolean) { + root.listFiles()?.filter { it.isDirectory }?.forEach { deleteIfDead(it, isAlive) } + } + + private fun deleteIfDead(dir: File, isAlive: (String) -> Boolean) { + val workspace = UploadWorkspace(dir) + if (!workspace.tryLock()) { + return + } + try { + if (!isAlive(dir.name)) { + workspace.delete() + } + } finally { + workspace.unlock() + } + } + } +} diff --git a/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt b/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt index 9e44d9f1402..756630e63ae 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt +++ b/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt @@ -19,10 +19,10 @@ import java.io.File import java.io.FileNotFoundException import java.io.FileOutputStream import java.io.IOException -import java.io.InputStream import java.math.BigInteger import java.security.MessageDigest +@Suppress("TooManyFunctions") object FileUtils { private val TAG = FileUtils::class.java.simpleName private const val RADIX: Int = 16 @@ -155,7 +155,7 @@ object FileUtils { } @Suppress("ThrowsCount") - fun getFileFromUri(context: Context, sourceFileUri: Uri): File? { + fun getFileFromUri(context: Context, sourceFileUri: Uri, targetDir: File = context.cacheDir): File? { val fileName = getFileName(sourceFileUri, context) val scheme = sourceFileUri.scheme @@ -163,7 +163,7 @@ object FileUtils { Log.d(TAG, "relative uri: " + sourceFileUri.path) throw IllegalArgumentException("relative paths are not supported") } else if (ContentResolver.SCHEME_CONTENT == scheme) { - copyFileToCache(context, sourceFileUri, fileName) + copyFileToCache(context, sourceFileUri, fileName, targetDir) } else if (ContentResolver.SCHEME_FILE == scheme) { val path = sourceFileUri.path ?: throw IllegalArgumentException("uri does not contain path") @@ -182,27 +182,43 @@ object FileUtils { } @Suppress("NestedBlockDepth") - fun copyFileToCache(context: Context, sourceFileUri: Uri, filename: String): File? { - val cachedFile = File(context.cacheDir, filename) + fun copyFileToCache( + context: Context, + sourceFileUri: Uri, + filename: String, + targetDir: File = context.cacheDir + ): File? { + val cachedFile = File(targetDir, filename) - if (!cachedFile.toPath().normalize().startsWith(context.cacheDir.toPath())) { + if (!cachedFile.toPath().normalize().startsWith(targetDir.toPath())) { Log.w(TAG, "cachedFile was not created in cacheDir. Aborting for security reasons.") cachedFile.delete() return null } - if (cachedFile.exists()) { + val ready = if (cachedFile.exists()) { Log.d(TAG, "file is already in cache") + true } else { - val outputStream = FileOutputStream(cachedFile) + copyToFile(context, sourceFileUri, cachedFile) + } + if (!ready) { + cachedFile.delete() + } + return cachedFile.takeIf { ready } + } + + /** Returns false when the source could not be read completely, so the copy is not a usable file. */ + @Suppress("NestedBlockDepth") + private fun copyToFile(context: Context, sourceFileUri: Uri, target: File): Boolean { + var copied = false + FileOutputStream(target).use { output -> try { - val inputStream: InputStream? = context.contentResolver.openInputStream(sourceFileUri) - inputStream?.use { input -> - outputStream.use { output -> - input.copyTo(output) - } + context.contentResolver.openInputStream(sourceFileUri)?.use { input -> + input.copyTo(output) + copied = true } - outputStream.flush() + output.flush() } catch (e: FileNotFoundException) { Log.w(TAG, "failed to copy file to cache", e) } catch (e: SecurityException) { @@ -212,9 +228,33 @@ object FileUtils { Log.w(TAG, "no longer permitted to read $sourceFileUri", e) } } - return cachedFile + if (copied) { + // The size a provider reports is not reliable (transcoding, cloud providers), so it is only logged. + val reported = querySize(context, sourceFileUri) + if (reported != null && reported > 0 && reported != target.length()) { + Log.w(TAG, "copy of $sourceFileUri has ${target.length()} bytes, the provider reports $reported") + } + } + return copied && isCompleteCopy(target.length()) } + /** An empty copy is a broken copy; a stream that broke off throws and never gets here. */ + internal fun isCompleteCopy(copiedLength: Long): Boolean = copiedLength > 0 + + private fun querySize(context: Context, uri: Uri): Long? = + try { + context.contentResolver.query(uri, arrayOf(OpenableColumns.SIZE), null, null, null)?.use { cursor -> + val index = cursor.getColumnIndex(OpenableColumns.SIZE) + if (index >= 0 && cursor.moveToFirst() && !cursor.isNull(index)) cursor.getLong(index) else null + } + } catch (e: SecurityException) { + Log.w(TAG, "could not query the size of $uri", e) + null + } catch (e: IllegalArgumentException) { + Log.w(TAG, "could not query the size of $uri", e) + null + } + fun getFileName(uri: Uri, context: Context?): String { var filename: String? = null if (uri.scheme == "content" && context != null) { diff --git a/app/src/main/java/com/nextcloud/talk/utils/ImageCompressor.kt b/app/src/main/java/com/nextcloud/talk/utils/ImageCompressor.kt index 7f7807ffb13..263b205b670 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/ImageCompressor.kt +++ b/app/src/main/java/com/nextcloud/talk/utils/ImageCompressor.kt @@ -52,13 +52,13 @@ object ImageCompressor { } /** - * Returns a downscaled, JPEG re-encoded copy of [sourceFile] in the app cache directory, - * or null if the image could not be decoded or written. + * Returns a downscaled, JPEG re-encoded copy of [sourceFile] in [outputDir] (the app cache directory by + * default), or null if the image could not be decoded or written. */ - fun compress(context: Context, sourceFile: File): File? { + fun compress(context: Context, sourceFile: File, outputDir: File = context.cacheDir): File? { val (bitmap, bytes) = encode(sourceFile) ?: return null return try { - val compressedFile = File(context.cacheDir, sourceFile.nameWithoutExtension + COMPRESSED_FILE_SUFFIX) + val compressedFile = File(outputDir, sourceFile.nameWithoutExtension + COMPRESSED_FILE_SUFFIX) compressedFile.writeBytes(bytes) compressedFile } catch (e: IOException) { diff --git a/app/src/main/java/com/nextcloud/talk/utils/VideoCompressor.kt b/app/src/main/java/com/nextcloud/talk/utils/VideoCompressor.kt index 428bc466aec..7d9d43ad815 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/VideoCompressor.kt +++ b/app/src/main/java/com/nextcloud/talk/utils/VideoCompressor.kt @@ -93,18 +93,23 @@ object VideoCompressor { } /** - * Returns a downscaled, re-encoded copy of [sourceFile] in the app cache directory, or null if + * Returns a downscaled, re-encoded copy of [sourceFile] in [outputDir] (default: the app cache dir), or null if * the video was already small enough, could not be read, or transcoding failed/timed out. * * [onProgress] is invoked with a 0-100 value from a background thread while transcoding runs. */ @Suppress("ReturnCount") - fun compress(context: Context, sourceFile: File, onProgress: (Int) -> Unit = {}): File? { + fun compress( + context: Context, + sourceFile: File, + outputDir: File = context.cacheDir, + onProgress: (Int) -> Unit = {} + ): File? { Log.d(TAG, "compressing ${sourceFile.name}") val original = readVideoInfo(sourceFile) ?: return null if (minOf(original.width, original.height) <= TARGET_SHORT_SIDE) return null - val outputFile = File(context.cacheDir, sourceFile.nameWithoutExtension + COMPRESSED_FILE_SUFFIX) + val outputFile = File(outputDir, sourceFile.nameWithoutExtension + COMPRESSED_FILE_SUFFIX) outputFile.delete() val success = runTransform(context, sourceFile, outputFile, original.durationMs, onProgress) diff --git a/app/src/test/java/com/nextcloud/talk/upload/UploadWorkspaceTest.kt b/app/src/test/java/com/nextcloud/talk/upload/UploadWorkspaceTest.kt new file mode 100644 index 00000000000..c89fb0e333c --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/upload/UploadWorkspaceTest.kt @@ -0,0 +1,152 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: GPL-3.0-or-later + */ +package com.nextcloud.talk.upload + +import com.nextcloud.talk.utils.FileUtils +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Rule +import org.junit.Test +import org.junit.rules.TemporaryFolder +import java.io.File + +class UploadWorkspaceTest { + + @get:Rule + val tempFolder = TemporaryFolder() + + private fun workspace(name: String = "work") = UploadWorkspace(File(tempFolder.root, name)) + + private fun create(dir: File, name: String, lastModified: Long): PreparedUpload { + val file = File(dir, name).apply { + writeBytes(ByteArray(SIZE)) + setLastModified(lastModified) + } + return PreparedUpload(file, name) + } + + @Test + fun `file is prepared once and reused by later runs`() { + var calls = 0 + val first = workspace().prepareOnce { dir -> calls++.let { create(dir, "photo.jpg", 1_000) } } + val second = workspace().prepareOnce { dir -> calls++.let { create(dir, "photo.jpg", 2_000) } } + + assertEquals(1, calls) + assertEquals(first, second) + } + + @Test + fun `chunk folder key stays the same between runs`() { + val first = workspace().prepareOnce { create(it, "video.mp4", 1_000) }!! + val keyBefore = FileUtils.md5Sum(first.file) + val second = workspace().prepareOnce { create(it, "video.mp4", 9_000) }!! + + assertEquals(keyBefore, FileUtils.md5Sum(second.file)) + } + + @Test + fun `a new preparation of the same file gets another chunk folder key`() { + val first = workspace("a").prepareOnce { create(it, "video.mp4", 1_000) }!! + val second = workspace("b").prepareOnce { create(it, "video.mp4", 2_000) }!! + + assertTrue(FileUtils.md5Sum(first.file) != FileUtils.md5Sum(second.file)) + } + + @Test + fun `file is prepared again when it disappeared`() { + val first = workspace().prepareOnce { create(it, "photo.jpg", 1_000) }!! + first.file.delete() + var calls = 0 + val second = workspace().prepareOnce { dir -> calls++.let { create(dir, "photo.jpg", 3_000) } } + + assertEquals(1, calls) + assertNotNull(second) + assertTrue(second!!.file.isFile) + } + + @Test + fun `leftovers of an interrupted preparation are removed before the next one`() { + val ws = workspace() + ws.prepareOnce { dir -> + File(dir, "half.jpg").writeText("partial") + null + } + ws.prepareOnce { dir -> + assertEquals(emptyList(), dir.list()!!.toList()) + create(dir, "photo.jpg", 1_000) + } + } + + @Test + fun `delete removes the prepared file and the marks`() { + val ws = workspace() + val prepared = ws.prepareOnce { create(it, "photo.jpg", 1_000) }!! + ws.markRestarted() + ws.delete() + + assertFalse(prepared.file.exists()) + assertFalse(workspace().isRestarted()) + assertNull(workspace().prepareOnce { null }) + } + + @Test + fun `workspaces of finished uploads are removed and living ones kept`() { + workspace("finished").prepareOnce { create(it, "a.jpg", 1_000) } + workspace("waiting-for-network").prepareOnce { create(it, "b.jpg", 1_000) } + + UploadWorkspace.deleteFinished(tempFolder.root) { it == "waiting-for-network" } + + assertFalse(File(tempFolder.root, "finished").exists()) + assertTrue(File(tempFolder.root, "waiting-for-network").exists()) + } + + @Test + fun `a locked workspace is not removed even when its work is unknown`() { + val running = workspace("running") + running.prepareOnce { create(it, "a.jpg", 1_000) } + assertTrue(running.tryLock()) + + UploadWorkspace.deleteFinished(tempFolder.root) { false } + + assertTrue(File(tempFolder.root, "running").exists()) + running.unlock() + UploadWorkspace.deleteFinished(tempFolder.root) { false } + assertFalse(File(tempFolder.root, "running").exists()) + } + + @Test + fun `only one run can hold the lock of a workspace`() { + val first = workspace() + val second = workspace() + + assertTrue(first.tryLock()) + assertFalse(second.tryLock()) + first.unlock() + assertTrue(second.tryLock()) + second.unlock() + } + + @Test + fun `the restart mark survives a new preparation after the prepared file is gone`() { + val ws = workspace() + val prepared = ws.prepareOnce { create(it, "a.jpg", 1_000) }!! + assertFalse(ws.isRestarted()) + ws.markRestarted() + prepared.file.delete() + + workspace().prepareOnce { create(it, "a.jpg", 2_000) } + + assertTrue(workspace().isRestarted()) + } + + companion object { + private const val SIZE = 16 + } +} diff --git a/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt b/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt index 600713ab7fd..dc7ac8147a9 100644 --- a/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt +++ b/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt @@ -9,14 +9,26 @@ package com.nextcloud.talk.utils import android.app.Application import android.content.ContentResolver import android.content.Context +import android.database.MatrixCursor import android.net.Uri +import android.provider.OpenableColumns import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Rule import org.junit.Test +import org.junit.rules.TemporaryFolder import org.junit.runner.RunWith import org.mockito.Mockito.mock import org.mockito.Mockito.`when` +import org.mockito.kotlin.any +import org.mockito.kotlin.anyOrNull +import org.mockito.kotlin.eq import org.robolectric.RobolectricTestRunner import org.robolectric.annotation.Config +import java.io.ByteArrayInputStream // Regression coverage for a crash: the system Photo Picker's ephemeral content:// uris can have // their read grant revoked at any point after being handed to us (e.g. while a batch upload is @@ -26,6 +38,11 @@ import org.robolectric.annotation.Config @Config(application = Application::class, sdk = [33]) class FileUtilsContentResolverTest { + @get:Rule + val tempFolder = TemporaryFolder() + + private val contentUri = Uri.parse("content://media/picker/0/media/42") + @Test fun `getFileName falls back to the uri path when the content provider query throws`() { val uri = Uri.parse("content://media/picker/0/com.android.providers.media.photopicker/media/123") @@ -49,6 +66,62 @@ class FileUtilsContentResolverTest { assertEquals("image/jpeg", mimeType) } + @Test + fun `copyFileToCache keeps a complete copy`() { + val context = contextWithStream(ByteArray(COPY_SIZE)) + + val copy = FileUtils.copyFileToCache(context, contentUri, "photo.jpg", tempFolder.root) + + assertNotNull(copy) + assertEquals(COPY_SIZE.toLong(), copy!!.length()) + } + + @Test + fun `copyFileToCache returns null and leaves no file when the source is empty`() { + val context = contextWithStream(ByteArray(0)) + + assertNull(FileUtils.copyFileToCache(context, contentUri, "photo.jpg", tempFolder.root)) + assertFalse(java.io.File(tempFolder.root, "photo.jpg").exists()) + } + + @Test + fun `copyFileToCache returns null when the read grant was revoked`() { + val context = mock(Context::class.java) + val resolver = mock(ContentResolver::class.java) + `when`(context.contentResolver).thenReturn(resolver) + `when`(resolver.openInputStream(contentUri)).thenThrow(SecurityException("revoked")) + + assertNull(FileUtils.copyFileToCache(context, contentUri, "photo.jpg", tempFolder.root)) + assertFalse(java.io.File(tempFolder.root, "photo.jpg").exists()) + } + + @Test + fun `copyFileToCache keeps a complete copy whose size differs from the one the provider reports`() { + val context = contextWithStream(ByteArray(COPY_SIZE), reportedSize = COPY_SIZE * 2L) + + val copy = FileUtils.copyFileToCache(context, contentUri, "photo.jpg", tempFolder.root) + + assertEquals(COPY_SIZE.toLong(), copy!!.length()) + } + + @Test + fun `a copy is complete when it is not empty`() { + assertTrue(FileUtils.isCompleteCopy(10)) + assertFalse(FileUtils.isCompleteCopy(0)) + } + + private fun contextWithStream(bytes: ByteArray, reportedSize: Long? = null): Context { + val context = mock(Context::class.java) + val resolver = mock(ContentResolver::class.java) + `when`(context.contentResolver).thenReturn(resolver) + `when`(resolver.openInputStream(contentUri)).thenReturn(ByteArrayInputStream(bytes)) + if (reportedSize != null) { + val cursor = MatrixCursor(arrayOf(OpenableColumns.SIZE)).apply { addRow(arrayOf(reportedSize)) } + `when`(resolver.query(eq(contentUri), any(), anyOrNull(), anyOrNull(), anyOrNull())).thenReturn(cursor) + } + return context + } + private fun mockContextWhoseResolverThrows(uri: Uri): Context { val context = mock(Context::class.java) val resolver = mock(ContentResolver::class.java) @@ -57,4 +130,8 @@ class FileUtilsContentResolverTest { .thenThrow(SecurityException("permission revoked")) return context } + + companion object { + private const val COPY_SIZE = 64 + } } From a40ecad37220863c741eef0736e651a4bfa5ed97 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Mon, 5 Oct 2026 20:03:38 +0000 Subject: [PATCH 03/10] fix(upload): keep the restart mark and the upload files safe Remove the rejected parts before marking the upload as restarted: when the DELETE fails, the retry must still be allowed to remove them. Keep the upload workspace in noBackupFilesDir instead of cacheDir. The system may clear the cache under storage pressure while an upload waits for the network. A new copy of the file changes its resume key and leaves the parts on the server orphaned, and after a process restart the content URI may no longer be readable. Finished workspaces are still removed by the worker. Signed-off-by: Krainov Gleb Assisted-by: Claude-Code:claude-opus-5-5 --- .../talk/jobs/UploadAndShareFilesWorker.kt | 4 +-- .../upload/chunked/ChunkedFileUploader.kt | 2 +- .../chunked/ChunkedFileUploaderResumeTest.kt | 27 +++++++++++++++++++ 3 files changed, 30 insertions(+), 3 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt index c9bf85ffe6d..8efeeced5d8 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt +++ b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt @@ -296,7 +296,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa /** Removes the workspaces of works that are finished or unknown to WorkManager, e.g. after a killed process. */ @Suppress("Detekt.TooGenericExceptionCaught") private fun deleteFinishedWorkspaces() { - UploadWorkspace.deleteFinished(File(context.cacheDir, WORKSPACE_DIR)) { name -> + UploadWorkspace.deleteFinished(File(context.noBackupFilesDir, WORKSPACE_DIR)) { name -> try { val info = WorkManager.getInstance(context).getWorkInfoById(UUID.fromString(name)).get() info != null && !info.state.isFinished @@ -679,7 +679,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa private fun uploadQueueName(internalConversationId: String) = "upload_queue_$internalConversationId" private fun workspaceDir(context: Context, workId: UUID) = - File(File(context.cacheDir, WORKSPACE_DIR), workId.toString()) + File(File(context.noBackupFilesDir, WORKSPACE_DIR), workId.toString()) // Cancellation is by the WorkRequest's own id (not enqueueUniqueWork's name) since that name // is now shared by every file queued in the same conversation - cancelling by name would diff --git a/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt b/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt index b51ee00299f..c8425910456 100644 --- a/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt +++ b/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt @@ -91,8 +91,8 @@ class ChunkedFileUploader( throw e } Log.w(TAG, "Server rejected the assembly, uploading the file again", e) - markRestarted() deleteUploadFolder() + markRestarted() uploadParts(localFile, mimeType, targetPath) } } diff --git a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt index 0fe6309b910..f7004506254 100644 --- a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt +++ b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt @@ -56,6 +56,7 @@ class ChunkedFileUploaderResumeTest { private var propfindResponseCode: Int? = null private var propfindDisconnects = false private var putDisconnects = false + private var deleteDisconnects = false /** Answers of the next MOVE requests; 201 when empty. */ private val moveResponseCodes: MutableList = Collections.synchronizedList(mutableListOf()) @@ -80,6 +81,11 @@ class ChunkedFileUploaderResumeTest { MockResponse().setResponseCode(putResponseCode) } "MKCOL" -> MockResponse().setResponseCode(METHOD_NOT_ALLOWED) + "DELETE" -> if (deleteDisconnects) { + MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AFTER_REQUEST) + } else { + MockResponse().setResponseCode(CREATED) + } else -> MockResponse().setResponseCode(CREATED) } } @@ -265,6 +271,27 @@ class ChunkedFileUploaderResumeTest { assertEquals(1, marked) } + @Test + fun `a restart is not reported when the parts could not be removed`() { + moveResponseCodes.add(BAD_REQUEST) + deleteDisconnects = true + var marked = 0 + val first = ChunkedFileUploader( + client, + user, + mock(), + mock(), + markRestarted = { marked++ } + ) + + try { + first.upload(file, null, "/Talk/video.mp4") + fail("expected the upload to throw") + } catch (e: IOException) { + assertEquals(0, marked) + } + } + @Test fun `a server error while listing the parts is an error`() { propfindResponseCode = SERVER_ERROR From cce7ce7d228aecf6b91c00cce13eb7321c141114 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Mon, 5 Oct 2026 20:41:52 +0000 Subject: [PATCH 04/10] fix(upload): do not start a chunked upload after a stop A stop that came after the last check of doUpload() and before the chunked uploader was assigned found no uploader in onStopped(), so the upload ran on to the end. Assign the uploader first and check for a stop after that: an earlier stop is seen by the check, a later one reaches the uploader. Signed-off-by: Krainov Gleb Assisted-by: Claude-Code:claude-opus-5-5 --- .../talk/jobs/UploadAndShareFilesWorker.kt | 46 +++++++++++-------- 1 file changed, 28 insertions(+), 18 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt index 8efeeced5d8..70ecfe2c26d 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt +++ b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt @@ -61,6 +61,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.runBlocking +import okhttp3.MediaType import okhttp3.MediaType.Companion.toMediaTypeOrNull import okhttp3.OkHttpClient import java.io.File @@ -110,7 +111,10 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa lateinit var currentUser: User private var isChunkedUploading = false private var file: File? = null + + @Volatile private var chunkedFileUploader: ChunkedFileUploader? = null + private var referenceId: String? = null private var internalConversationId: String? = null private var uploadDisposable: Disposable? = null @@ -335,6 +339,28 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa return Result.failure() } + private fun newChunkedUploader() = + ChunkedFileUploader( + okHttpClient, + currentUser, + this, + ncApiCoroutines, + isRestarted = workspace::isRestarted, + markRestarted = workspace::markRestarted + ) + + /** + * Publishes [uploader] before the last stop check: a stop that comes earlier is seen here, a later one reaches + * the uploader through [onStopped]. + */ + private fun startChunkedUpload(uploader: ChunkedFileUploader, mimeType: MediaType?, path: String): Boolean { + chunkedFileUploader = uploader + if (isStopped || isCancelled()) { + return false + } + return uploader.upload(file!!, mimeType, path) + } + private fun uploadFile( sourceFileUri: Uri, metaData: String?, @@ -349,15 +375,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa } else if (isChunkedUploading) { Log.d(TAG, "starting chunked upload because size is " + file!!.length()) val mimeType = FileUtils.resolveMimeType(context, sourceFileUri)?.toMediaTypeOrNull() - chunkedFileUploader = ChunkedFileUploader( - okHttpClient, - currentUser, - this, - ncApiCoroutines, - isRestarted = workspace::isRestarted, - markRestarted = workspace::markRestarted - ) - chunkedFileUploader!!.upload(file!!, mimeType, remotePath) + startChunkedUpload(newChunkedUploader(), mimeType, remotePath) } else { Log.d(TAG, "starting normal upload (not chunked) of $fileName") val observable = FileUploader( @@ -433,15 +451,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa val uploadSuccess = if (isChunkedUploading) { val mimeType = FileUtils.resolveMimeType(context, sourceFileUri)?.toMediaTypeOrNull() - chunkedFileUploader = ChunkedFileUploader( - okHttpClient, - currentUser, - this@UploadAndShareFilesWorker, - ncApiCoroutines, - isRestarted = workspace::isRestarted, - markRestarted = workspace::markRestarted - ) - chunkedFileUploader!!.upload(file!!, mimeType, tempRemotePath) + startChunkedUpload(newChunkedUploader(), mimeType, tempRemotePath) } else { FileUploader(okHttpClient, context, currentUser, roomToken, ncApi, file!!, ncApiCoroutines) .uploadToConversationSubfolder(sourceFileUri, tempRemotePath) From e6fb95afa062390f74d49ad55822e27f64a4975a Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Mon, 5 Oct 2026 21:13:23 +0000 Subject: [PATCH 05/10] fix(upload): remember a cancel after doWork() removed its marker doWork() removes the cancel marker of its reference id when it ends, but WorkManager may call onStopped() after that. onStopped() then saw no cancel and only stopped the uploader, so the parts stayed on the server. Latch the cancel once it is seen, and check it once more right before the marker is removed. Signed-off-by: Krainov Gleb Assisted-by: Claude-Code:claude-opus-5-5 --- .../nextcloud/talk/jobs/UploadAndShareFilesWorker.kt | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt index 70ecfe2c26d..eced926f6e6 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt +++ b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt @@ -122,14 +122,23 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa private lateinit var workspace: UploadWorkspace private var keepWorkspace = false + @Volatile + private var cancelSeen = false + /** * WorkManager's own `isStopped`/`onStopped()` only becomes true/fires after * cancelUniqueWork() round-trips through WorkManager's internal executor and Room DB - for a * small/compressed image the whole upload+share can finish faster than that round-trip, so * isStopped alone arrives too late. [cancelledReferenceIds] is set synchronously by the UI * before cancelUniqueWork() is even called, so it's visible to doWork() immediately. + * Once seen, the cancel is latched: doWork() removes the marker, and onStopped() can still come after that. */ - private fun isCancelled(): Boolean = referenceId?.let { cancelledReferenceIds.contains(it) } == true + private fun isCancelled(): Boolean { + if (referenceId?.let { cancelledReferenceIds.contains(it) } == true) { + cancelSeen = true + } + return cancelSeen + } override fun doWork(): Result { // Invariant: dependencies that are already set (a test worker factory) are not overwritten. @@ -146,6 +155,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa try { return doUpload() } finally { + isCancelled() // latches a cancel that came after the last check, before its marker is removed referenceId?.let { cancelledReferenceIds.remove(it) } // A run that is stopped, or will be tried again, resumes from the prepared file. if (!keepWorkspace) { From b9d524552d326068118cf71940320535a3cad4d9 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Tue, 13 Oct 2026 05:13:46 +0000 Subject: [PATCH 06/10] fix(upload): address review findings on resume and cancel - Delete the workspace of a cancelled run even when it was marked to be kept. - Copy a file:// source into the workspace, so the prepared upload does not depend on the caller's cache file. - Fix the upload folder before the uploader is published, so a cancel that arrives early can still remove the parts; abort at once if it already came. - Restart after HTTP 400 only when the assembly was rejected, not a part. - Sort the listed parts before looking for gaps. - Treat 401, 403, 408 and 429 while listing the parts as errors, not as an empty folder. - Delete a partial copy when the stream breaks off. Signed-off-by: Krainov Gleb Assisted-by: Claude-Code:claude-sonnet-5-5 --- .../talk/jobs/UploadAndShareFilesWorker.kt | 14 ++++-- .../upload/chunked/ChunkedFileUploader.kt | 47 ++++++++++++++----- .../com/nextcloud/talk/utils/FileUtils.kt | 25 +++++++++- .../chunked/ChunkedFileUploaderAbortTest.kt | 13 +++++ .../chunked/ChunkedFileUploaderResumeTest.kt | 40 ++++++++++++++++ .../utils/FileUtilsContentResolverTest.kt | 41 ++++++++++++++++ 6 files changed, 163 insertions(+), 17 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt index eced926f6e6..4be4d9c841f 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt +++ b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt @@ -155,10 +155,12 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa try { return doUpload() } finally { - isCancelled() // latches a cancel that came after the last check, before its marker is removed + // Latches a cancel that came after the last check, before its marker is removed. + val cancelled = isCancelled() referenceId?.let { cancelledReferenceIds.remove(it) } // A run that is stopped, or will be tried again, resumes from the prepared file. - if (!keepWorkspace) { + // A cancelled one never does. + if (!keepWorkspace || cancelled) { workspace.delete() } workspace.unlock() @@ -287,7 +289,8 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa * by the file, and a fresh copy or compression would change that key and start the upload from zero. */ private fun prepareFile(sourceFileUri: Uri, dir: File): PreparedUpload? { - val original = FileUtils.getFileFromUri(context, sourceFileUri, dir) ?: return null + val original = FileUtils.getFileFromUri(context, sourceFileUri, dir)?.let { FileUtils.ensureInDir(it, dir) } + ?: return null throwIfStopped() val compressed = if (inputData.getBoolean(COMPRESS_IMAGES, false)) { compressMediaIfPossible(sourceFileUri, original, dir) @@ -364,8 +367,13 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa * the uploader through [onStopped]. */ private fun startChunkedUpload(uploader: ChunkedFileUploader, mimeType: MediaType?, path: String): Boolean { + // The folder has to be known before onStopped can reach the uploader, or an abort has nothing to remove. + uploader.prepare(file!!) chunkedFileUploader = uploader if (isStopped || isCancelled()) { + if (isCancelled()) { + uploader.abortUpload {} + } return false } return uploader.upload(file!!, mimeType, path) diff --git a/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt b/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt index c8425910456..db673dffbce 100644 --- a/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt +++ b/app/src/main/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploader.kt @@ -60,6 +60,8 @@ class ChunkedFileUploader( private var okHttpClientNoRedirects: OkHttpClient? = null private var remoteChunkUrl: String + + @Volatile private var uploadFolderUri: String = "" @Volatile @@ -81,25 +83,46 @@ class ChunkedFileUploader( */ @Throws(IOException::class) fun upload(localFile: File, mimeType: MediaType?, targetPath: String): Boolean { - uploadFolderUri = folderUriOf(localFile) - return try { - uploadParts(localFile, mimeType, targetPath) + prepare(localFile) + return uploadMissingParts(localFile, mimeType) && assembleOrRestart(localFile, mimeType, targetPath) + } + + /** @return false when the upload was stopped while it was sent again after a rejected assembly */ + private fun assembleOrRestart(localFile: File, mimeType: MediaType?, targetPath: String): Boolean { + try { + assembleChunks(uploadFolderUri, targetPath, localFile.length()) } catch (e: HttpException) { // The server refused the assembly (e.g. the length does not match its parts): the parts it holds are // not usable, so remove them once per upload (the worker remembers it) and send the file again. + // A rejected part is not this case and is thrown by uploadMissingParts. if (e.code != HTTP_BAD_REQUEST || isRestarted()) { throw e } Log.w(TAG, "Server rejected the assembly, uploading the file again", e) deleteUploadFolder() markRestarted() - uploadParts(localFile, mimeType, targetPath) + if (!uploadMissingParts(localFile, mimeType)) { + return false + } + assembleChunks(uploadFolderUri, targetPath, localFile.length()) + } + return true + } + + /** + * Fixes the upload folder of [localFile], so that [abortUpload] can remove it. Done by [upload] when not called + * before; hashes the whole file once. + */ + fun prepare(localFile: File) { + if (uploadFolderUri.isEmpty()) { + uploadFolderUri = folderUriOf(localFile) } } private fun folderUriOf(localFile: File) = remoteChunkUrl + "/" + FileUtils.md5Sum(localFile) - private fun uploadParts(localFile: File, mimeType: MediaType?, targetPath: String): Boolean { + /** @return false when the upload was stopped before every missing part was sent */ + private fun uploadMissingParts(localFile: File, mimeType: MediaType?): Boolean { val davResource = DavResource( okHttpClientNoRedirects!!, uploadFolderUri.toHttpUrlOrNull()!! @@ -119,12 +142,7 @@ class ChunkedFileUploader( } uploadChunk(localFile, uploadFolderUri, mimeType, missingChunk, missingChunk.length()) } - - val isComplete = !isUploadAborted - if (isComplete) { - assembleChunks(uploadFolderUri, targetPath, localFile.length()) - } - return isComplete + return !isUploadAborted } @Suppress("Detekt.ThrowsCount") @@ -176,9 +194,10 @@ class ChunkedFileUploader( Unit } } catch (e: DavException) { - if (e is HttpException && e.code >= HTTP_SERVER_ERROR) { + if (e is HttpException && (e.code >= HTTP_SERVER_ERROR || e.code in LISTING_ERROR_CODES)) { throw e } + // Failures of the request itself (login, rate limit, server) are no answer about the parts either. // An IOException is not caught on purpose: a lost network must repeat the request, // not send the whole file again. // PROPFIND on Nextcloud chunked-upload folders can return unexpected responses @@ -225,9 +244,10 @@ class ChunkedFileUploader( internal fun checkMissingChunks(chunks: List, length: Long): List { val missingChunks: MutableList = java.util.ArrayList() + val sortedChunks = chunks.sortedBy { it.start } var start: Long = 0 while (start <= length) { - val nextChunk: Chunk? = findNextFittingChunk(chunks, start) + val nextChunk: Chunk? = findNextFittingChunk(sortedChunks, start) if (nextChunk == null) { // create new chunk val end: Long = if (start + CHUNK_SIZE <= length) { @@ -508,6 +528,7 @@ class ChunkedFileUploader( private const val METHOD_NOT_ALLOWED_CODE: Int = 405 private const val HTTP_BAD_REQUEST: Int = 400 private const val HTTP_SERVER_ERROR: Int = 500 + private val LISTING_ERROR_CODES = setOf(401, 403, 408, 429) private const val CHUNK_NUMBER_LENGTH = 16 private val CHUNK_NAME_REGEX = Regex("^\\d{16}-\\d{16}$") private const val TOTAL_LENGTH_HEADER = "OC-Total-Length" diff --git a/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt b/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt index 756630e63ae..e1e531835a4 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt +++ b/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt @@ -181,6 +181,23 @@ object FileUtils { return file } + /** + * Returns [file] when it already lies in [dir], otherwise a copy of it there. A file:// source is the caller's + * own file in the cache; a copy in [dir] keeps the prepared upload independent of it. + */ + fun ensureInDir(file: File, dir: File): File { + if (file.canonicalFile.parentFile == dir.canonicalFile) { + return file + } + val copy = File(dir, file.name) + try { + return file.copyTo(copy, overwrite = true) + } catch (e: IOException) { + copy.delete() + throw e + } + } + @Suppress("NestedBlockDepth") fun copyFileToCache( context: Context, @@ -200,7 +217,13 @@ object FileUtils { Log.d(TAG, "file is already in cache") true } else { - copyToFile(context, sourceFileUri, cachedFile) + try { + copyToFile(context, sourceFileUri, cachedFile) + } catch (e: IOException) { + // A later call would take the partial file for a finished copy. + cachedFile.delete() + throw e + } } if (!ready) { cachedFile.delete() diff --git a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderAbortTest.kt b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderAbortTest.kt index 31df335f085..7a47c2474d7 100644 --- a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderAbortTest.kt +++ b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderAbortTest.kt @@ -104,6 +104,19 @@ class ChunkedFileUploaderAbortTest { assertEquals(1, successCalls) } + @Test + fun `removes the folder of an uploader that is prepared but not yet uploading`() { + val file = tempFolder.newFile("prepared.txt").apply { writeBytes(ByteArray(FILE_SIZE)) } + val uploader = uploader() + uploader.prepare(file) + server.enqueue(MockResponse().setResponseCode(NO_CONTENT)) + + uploader.abortUpload { successCalls++ } + + assertEquals(1, successCalls) + assertEquals("DELETE", server.takeRequest(1, TimeUnit.SECONDS)?.method) + } + @Test fun `reports success when the upload folder is deleted`() { val uploader = startedUploader() diff --git a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt index f7004506254..b2a5f39a129 100644 --- a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt +++ b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt @@ -305,6 +305,37 @@ class ChunkedFileUploaderResumeTest { assertEquals(emptyList(), putPaths()) } + @Test + fun `a rejected part is an error and keeps the parts on the server`() { + putResponseCode = BAD_REQUEST + + try { + uploader.upload(file, null, "/Talk/video.mp4") + fail("expected the upload to throw") + } catch (e: HttpException) { + assertEquals(BAD_REQUEST, e.code) + } + assertEquals(0, requests.count { it.method == "DELETE" }) + assertEquals(0, requests.count { it.method == "MOVE" }) + } + + @Test + fun `login, rate limit and timeout answers while listing the parts are errors`() { + for (code in listOf(401, 403, 408, 429)) { + requests.clear() + propfindResponseCode = code + val fresh = ChunkedFileUploader(client, user, mock(), mock()) + + try { + fresh.upload(file, null, "/Talk/video.mp4") + fail("expected the upload to throw for $code") + } catch (e: HttpException) { + assertEquals(code, e.code) + } + assertEquals(emptyList(), putPaths()) + } + } + @Test fun `an assembly rejected twice is an error`() { moveResponseCodes.addAll(listOf(BAD_REQUEST, BAD_REQUEST)) @@ -340,6 +371,15 @@ class ChunkedFileUploaderResumeTest { assertEquals(listOf(Chunk(CHUNK.toLong(), 2L * CHUNK - 1)), missing) } + @Test + fun `parts listed out of order are not uploaded again`() { + val onServer = listOf(Chunk(2L * CHUNK, FILE_SIZE.toLong()), Chunk(0, CHUNK - 1L)) + + val missing = uploader.checkMissingChunks(onServer, FILE_SIZE.toLong()) + + assertEquals(listOf(Chunk(CHUNK.toLong(), 2L * CHUNK - 1)), missing) + } + @Test fun `all parts are planned when nothing is on the server`() { val missing = uploader.checkMissingChunks(emptyList(), FILE_SIZE.toLong()) diff --git a/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt b/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt index dc7ac8147a9..3f858be3bc1 100644 --- a/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt +++ b/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt @@ -17,6 +17,7 @@ import org.junit.Assert.assertFalse import org.junit.Assert.assertNotNull import org.junit.Assert.assertNull import org.junit.Assert.assertTrue +import org.junit.Assert.fail import org.junit.Rule import org.junit.Test import org.junit.rules.TemporaryFolder @@ -29,6 +30,8 @@ import org.mockito.kotlin.eq import org.robolectric.RobolectricTestRunner import org.robolectric.annotation.Config import java.io.ByteArrayInputStream +import java.io.IOException +import java.io.InputStream // Regression coverage for a crash: the system Photo Picker's ephemeral content:// uris can have // their read grant revoked at any point after being handed to us (e.g. while a batch upload is @@ -104,6 +107,36 @@ class FileUtilsContentResolverTest { assertEquals(COPY_SIZE.toLong(), copy!!.length()) } + @Test + fun `copyFileToCache leaves no partial file when the stream breaks off`() { + val context = mock(Context::class.java) + val resolver = mock(ContentResolver::class.java) + `when`(context.contentResolver).thenReturn(resolver) + `when`(resolver.openInputStream(contentUri)).thenReturn(BrokenStream(COPY_SIZE)) + + try { + FileUtils.copyFileToCache(context, contentUri, "photo.jpg", tempFolder.root) + fail("expected the copy to throw") + } catch (e: IOException) { + // the caller retries + } + assertFalse(java.io.File(tempFolder.root, "photo.jpg").exists()) + } + + @Test + fun `ensureInDir copies a file from outside and keeps one that is already inside`() { + val dir = tempFolder.newFolder("workspace") + val outside = tempFolder.newFile("camera.jpg").apply { writeBytes(ByteArray(COPY_SIZE)) } + val inside = java.io.File(dir, "inside.jpg").apply { writeBytes(ByteArray(COPY_SIZE)) } + + val copy = FileUtils.ensureInDir(outside, dir) + + assertEquals(dir.canonicalFile, copy.canonicalFile.parentFile) + assertEquals(COPY_SIZE.toLong(), copy.length()) + assertTrue(outside.exists()) + assertEquals(inside, FileUtils.ensureInDir(inside, dir)) + } + @Test fun `a copy is complete when it is not empty`() { assertTrue(FileUtils.isCompleteCopy(10)) @@ -131,6 +164,14 @@ class FileUtilsContentResolverTest { return context } + /** Delivers [size] bytes, then fails like a dropped connection. */ + private class BrokenStream(private var size: Int) : InputStream() { + override fun read(): Int { + if (size-- <= 0) throw IOException("stream broke off") + return 0 + } + } + companion object { private const val COPY_SIZE = 64 } From 4107b52ed51af508d8ca7c3f1ed4dcf9b4bbde64 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Tue, 13 Oct 2026 05:52:40 +0000 Subject: [PATCH 07/10] fix(upload): copy a content uri aside and rename it after success copyFileToCache returned any existing target, including the leftover of a copy that died half way, and removed the target only on IOException. - Copy into a temporary file next to the target and rename it after a complete copy; remove the temporary file on any exception. - Treat an empty existing target as no cache hit and copy again. Assisted-by: Claude-Code:claude-sonnet-5-5 Signed-off-by: Krainov Gleb --- .../com/nextcloud/talk/utils/FileUtils.kt | 21 +++++++----- .../utils/FileUtilsContentResolverTest.kt | 34 +++++++++++++++++++ 2 files changed, 47 insertions(+), 8 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt b/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt index e1e531835a4..7541fd3a1b7 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt +++ b/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt @@ -28,6 +28,7 @@ object FileUtils { private const val RADIX: Int = 16 private const val MD5_LENGTH: Int = 32 private const val SHARED_ATTACHMENTS_DIRECTORY = "shared_attachments" + private const val PARTIAL_SUFFIX = ".part" /** * Returns the dedicated cache directory used for externally shared attachments. @@ -213,17 +214,11 @@ object FileUtils { return null } - val ready = if (cachedFile.exists()) { + val ready = if (cachedFile.exists() && isCompleteCopy(cachedFile.length())) { Log.d(TAG, "file is already in cache") true } else { - try { - copyToFile(context, sourceFileUri, cachedFile) - } catch (e: IOException) { - // A later call would take the partial file for a finished copy. - cachedFile.delete() - throw e - } + copyAside(context, sourceFileUri, cachedFile) } if (!ready) { cachedFile.delete() @@ -231,6 +226,16 @@ object FileUtils { return cachedFile.takeIf { ready } } + /** Copies aside and renames after success: a copy that dies half way never becomes the cached file. */ + private fun copyAside(context: Context, sourceFileUri: Uri, target: File): Boolean { + val partial = File(target.parentFile, "${target.name}.${System.nanoTime()}$PARTIAL_SUFFIX") + return try { + copyToFile(context, sourceFileUri, partial) && partial.renameTo(target) + } finally { + partial.delete() + } + } + /** Returns false when the source could not be read completely, so the copy is not a usable file. */ @Suppress("NestedBlockDepth") private fun copyToFile(context: Context, sourceFileUri: Uri, target: File): Boolean { diff --git a/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt b/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt index 3f858be3bc1..d78490f26a3 100644 --- a/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt +++ b/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt @@ -123,6 +123,33 @@ class FileUtilsContentResolverTest { assertFalse(java.io.File(tempFolder.root, "photo.jpg").exists()) } + @Test + fun `copyFileToCache leaves no file when the copy dies with a runtime exception`() { + val context = mock(Context::class.java) + val resolver = mock(ContentResolver::class.java) + `when`(context.contentResolver).thenReturn(resolver) + `when`(resolver.openInputStream(contentUri)).thenReturn(CrashingStream(COPY_SIZE)) + + try { + FileUtils.copyFileToCache(context, contentUri, "photo.jpg", tempFolder.root) + fail("expected the copy to throw") + } catch (e: IllegalStateException) { + // the caller retries + } + assertEquals(0, tempFolder.root.list()!!.size) + } + + @Test + fun `copyFileToCache copies again over an empty file left by an earlier run`() { + java.io.File(tempFolder.root, "photo.jpg").writeBytes(ByteArray(0)) + val context = contextWithStream(ByteArray(COPY_SIZE)) + + val copy = FileUtils.copyFileToCache(context, contentUri, "photo.jpg", tempFolder.root) + + assertEquals(COPY_SIZE.toLong(), copy!!.length()) + assertEquals(listOf("photo.jpg"), tempFolder.root.list()!!.toList()) + } + @Test fun `ensureInDir copies a file from outside and keeps one that is already inside`() { val dir = tempFolder.newFolder("workspace") @@ -172,6 +199,13 @@ class FileUtilsContentResolverTest { } } + private class CrashingStream(private var size: Int) : InputStream() { + override fun read(): Int { + if (size-- <= 0) throw IllegalStateException("provider crashed") + return 0 + } + } + companion object { private const val COPY_SIZE = 64 } From 26fc4ab4389cd943ea06ad630845a89669d8282a Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Fri, 16 Oct 2026 15:12:44 +0000 Subject: [PATCH 08/10] fix(upload): prepare the file in a subdirectory of the workspace A file named like a file of the workspace itself (prepared, lock, restarted, or a name ending in .tmp) was copied over that file or was replaced by it, so the upload sent a damaged file or lost its marks. - Hand the prepare step a subdirectory of the workspace, so the user's file name cannot meet a file of the workspace. Assisted-by: Claude-Code:claude-sonnet-5-5 Signed-off-by: Krainov Gleb --- .../nextcloud/talk/upload/UploadWorkspace.kt | 7 +++++-- .../talk/upload/UploadWorkspaceTest.kt | 20 +++++++++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/upload/UploadWorkspace.kt b/app/src/main/java/com/nextcloud/talk/upload/UploadWorkspace.kt index 88bb86b5a95..8b47cc9148f 100644 --- a/app/src/main/java/com/nextcloud/talk/upload/UploadWorkspace.kt +++ b/app/src/main/java/com/nextcloud/talk/upload/UploadWorkspace.kt @@ -76,12 +76,14 @@ class UploadWorkspace(private val dir: File) { /** * Returns the prepared file of an earlier run, or runs [prepare] once and keeps its result. * [prepare] gets an empty directory it may create files in, and returns null when the file cannot be prepared. + * The directory is a subdirectory of the workspace, so a file named like a file of the workspace itself + * ("prepared", "lock", "restarted", anything ending in ".tmp") cannot replace it. */ fun prepareOnce(prepare: (File) -> PreparedUpload?): PreparedUpload? { stored()?.let { return it } resetFiles() - dir.mkdirs() - return prepare(dir)?.also { write(preparedFile, "${it.file.absolutePath}\n${it.fileName}") } + val filesDir = File(dir, FILES_DIR).apply { mkdirs() } + return prepare(filesDir)?.also { write(preparedFile, "${it.file.absolutePath}\n${it.fileName}") } } /** Whether the parts were already removed once after the server rejected the assembly. */ @@ -121,6 +123,7 @@ class UploadWorkspace(private val dir: File) { private const val LOCK_FILE = "lock" private const val RESTARTED_FILE = "restarted" private const val TMP_SUFFIX = ".tmp" + private const val FILES_DIR = "file" /** Not part of the prepared file: survive a new preparation. */ private val KEPT_FILES = setOf(LOCK_FILE, RESTARTED_FILE) diff --git a/app/src/test/java/com/nextcloud/talk/upload/UploadWorkspaceTest.kt b/app/src/test/java/com/nextcloud/talk/upload/UploadWorkspaceTest.kt index c89fb0e333c..02a57024183 100644 --- a/app/src/test/java/com/nextcloud/talk/upload/UploadWorkspaceTest.kt +++ b/app/src/test/java/com/nextcloud/talk/upload/UploadWorkspaceTest.kt @@ -84,6 +84,26 @@ class UploadWorkspaceTest { } } + @Test + fun `a file named like a file of the workspace stays intact and is found again`() { + for (name in listOf("prepared", "lock", "restarted", "prepared.tmp", "restarted.tmp", "file")) { + val ws = workspace(name) + // The worker takes the lock before it prepares the file. + assertTrue(ws.tryLock()) + val prepared = ws.prepareOnce { create(it, name, 1_000) }!! + ws.markRestarted() + assertFalse(workspace(name).tryLock()) + + assertEquals(SIZE.toLong(), prepared.file.length()) + assertEquals(prepared, workspace(name).prepareOnce { null }) + assertEquals(SIZE.toLong(), prepared.file.length()) + assertTrue(workspace(name).isRestarted()) + assertFalse(workspace(name).tryLock()) + ws.unlock() + assertTrue(workspace(name).tryLock()) + } + } + @Test fun `delete removes the prepared file and the marks`() { val ws = workspace() From 0f1e2987b9be1c689cf706720c371505a3fbdbb4 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Fri, 16 Oct 2026 15:24:52 +0000 Subject: [PATCH 09/10] fix(upload): copy an empty file instead of failing it isCompleteCopy treated a copy of 0 bytes as broken, so an empty file could not be sent. A stream that breaks off throws, so a finished copy is complete whatever its size. - Accept an empty copy of an empty source. - Still refuse an empty copy of a source that reports content (a cloud placeholder that was not loaded). - Still copy again over an empty file of an earlier run. Assisted-by: Claude-Code:claude-sonnet-5-5 Signed-off-by: Krainov Gleb --- .../com/nextcloud/talk/utils/FileUtils.kt | 22 +++++++++--------- .../utils/FileUtilsContentResolverTest.kt | 23 +++++++++++-------- 2 files changed, 25 insertions(+), 20 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt b/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt index 7541fd3a1b7..08cb56a6fb4 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt +++ b/app/src/main/java/com/nextcloud/talk/utils/FileUtils.kt @@ -214,7 +214,7 @@ object FileUtils { return null } - val ready = if (cachedFile.exists() && isCompleteCopy(cachedFile.length())) { + val ready = if (cachedFile.exists() && cachedFile.length() > 0) { Log.d(TAG, "file is already in cache") true } else { @@ -256,19 +256,19 @@ object FileUtils { Log.w(TAG, "no longer permitted to read $sourceFileUri", e) } } - if (copied) { - // The size a provider reports is not reliable (transcoding, cloud providers), so it is only logged. - val reported = querySize(context, sourceFileUri) - if (reported != null && reported > 0 && reported != target.length()) { - Log.w(TAG, "copy of $sourceFileUri has ${target.length()} bytes, the provider reports $reported") - } + if (!copied) { + return false + } + // The size a provider reports is not reliable (transcoding, cloud providers), so it is only logged. + val reported = querySize(context, sourceFileUri) + if (reported != null && reported > 0 && reported != target.length()) { + Log.w(TAG, "copy of $sourceFileUri has ${target.length()} bytes, the provider reports $reported") } - return copied && isCompleteCopy(target.length()) + // A stream that broke off throws and never gets here, so an empty copy of an empty source is complete. + // An empty copy of a source that reports content is a placeholder that was not loaded: not a usable file. + return target.length() > 0 || reported == null || reported <= 0 } - /** An empty copy is a broken copy; a stream that broke off throws and never gets here. */ - internal fun isCompleteCopy(copiedLength: Long): Boolean = copiedLength > 0 - private fun querySize(context: Context, uri: Uri): Long? = try { context.contentResolver.query(uri, arrayOf(OpenableColumns.SIZE), null, null, null)?.use { cursor -> diff --git a/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt b/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt index d78490f26a3..48c20efcc6e 100644 --- a/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt +++ b/app/src/test/java/com/nextcloud/talk/utils/FileUtilsContentResolverTest.kt @@ -80,11 +80,22 @@ class FileUtilsContentResolverTest { } @Test - fun `copyFileToCache returns null and leaves no file when the source is empty`() { + fun `copyFileToCache copies an empty source to an empty file`() { val context = contextWithStream(ByteArray(0)) - assertNull(FileUtils.copyFileToCache(context, contentUri, "photo.jpg", tempFolder.root)) - assertFalse(java.io.File(tempFolder.root, "photo.jpg").exists()) + val copy = FileUtils.copyFileToCache(context, contentUri, "empty.txt", tempFolder.root) + + assertNotNull(copy) + assertEquals(0L, copy!!.length()) + assertEquals(listOf("empty.txt"), tempFolder.root.list()!!.toList()) + } + + @Test + fun `copyFileToCache returns null for an empty copy of a source that reports content`() { + val context = contextWithStream(ByteArray(0), reportedSize = COPY_SIZE.toLong()) + + assertNull(FileUtils.copyFileToCache(context, contentUri, "placeholder.jpg", tempFolder.root)) + assertEquals(0, tempFolder.root.list()!!.size) } @Test @@ -164,12 +175,6 @@ class FileUtilsContentResolverTest { assertEquals(inside, FileUtils.ensureInDir(inside, dir)) } - @Test - fun `a copy is complete when it is not empty`() { - assertTrue(FileUtils.isCompleteCopy(10)) - assertFalse(FileUtils.isCompleteCopy(0)) - } - private fun contextWithStream(bytes: ByteArray, reportedSize: Long? = null): Context { val context = mock(Context::class.java) val resolver = mock(ContentResolver::class.java) From 6a17889117b83bf7d8f948e4b8f70733448edca9 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Fri, 16 Oct 2026 15:24:53 +0000 Subject: [PATCH 10/10] fix(upload): try an upload again after a short server outage dav4jvm and Retrofit report a refused request as HttpException, which is not an IOException, so a 503 (maintenance mode) ended the upload as failed at once. - Retry on 408, 429 and 5xx like on a network error, up to the same number of attempts; not on 501, 505 and 507, which do not go away. - Other refusals still fail the upload. Assisted-by: Claude-Code:claude-sonnet-5-5 Signed-off-by: Krainov Gleb --- .../talk/jobs/UploadAndShareFilesWorker.kt | 76 ++++++++++++++----- .../jobs/UploadAndShareFilesWorkerTest.kt | 22 ++++++ .../chunked/ChunkedFileUploaderResumeTest.kt | 14 ++++ 3 files changed, 92 insertions(+), 20 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt index 4be4d9c841f..4127c37acbb 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt +++ b/app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt @@ -28,6 +28,7 @@ import androidx.work.WorkManager import androidx.work.WorkRequest import androidx.work.Worker import androidx.work.WorkerParameters +import at.bitfire.dav4jvm.exception.HttpException import autodagger.AutoInjector import com.nextcloud.talk.R import com.nextcloud.talk.api.NcApi @@ -243,32 +244,42 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa Log.e(TAG, "Something went wrong when trying to upload file") failUpload() - } catch (e: IOException) { - // Transient network failures (connection reset, timeout, dropped Wi-Fi, ...) shouldn't - // require the user to manually resend - retry a few times with backoff instead, and only - // give up once we've exhausted the allowed attempts. - Log.w( - TAG, - "Network error while uploading file (attempt ${runAttemptCount + 1}/$MAX_UPLOAD_ATTEMPTS)", - e - ) - if (isStopped || isCancelled()) { - stoppedResult() - } else if (runAttemptCount < MAX_UPLOAD_ATTEMPTS - 1) { - keepWorkspace = true - Result.retry() - } else { - failUpload() - } } catch (e: Exception) { - if (isStopped || isCancelled()) { - return stoppedResult() + if (isRetryable(e)) { + // Transient failures (connection reset, timeout, dropped Wi-Fi, a short outage of the server, ...) + // shouldn't require the user to manually resend - retry a few times with backoff instead, and only + // give up once we've exhausted the allowed attempts. + retryOrFail(e) + } else { + failOnException(e) } - Log.e(TAG, "Something went wrong when trying to upload file", e) + } + } + + private fun retryOrFail(e: Exception): Result { + Log.w( + TAG, + "Upload interrupted, trying again later (attempt ${runAttemptCount + 1}/$MAX_UPLOAD_ATTEMPTS)", + e + ) + return if (isStopped || isCancelled()) { + stoppedResult() + } else if (runAttemptCount < MAX_UPLOAD_ATTEMPTS - 1) { + keepWorkspace = true + Result.retry() + } else { failUpload() } } + private fun failOnException(e: Exception): Result { + if (isStopped || isCancelled()) { + return stoppedResult() + } + Log.e(TAG, "Something went wrong when trying to upload file", e) + return failUpload() + } + // Result.success(), not failure(): the failure is recorded on the placeholder row, and a failed work // request would cascade to every upload already queued behind it in the conversation. @Suppress("Detekt.TooGenericExceptionCaught") @@ -599,8 +610,33 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa // Total attempts allowed for a single upload (1 initial run + retries) before giving up on a // transient network failure and marking the placeholder FAILED. private const val MAX_UPLOAD_ATTEMPTS = 4 + private const val HTTP_REQUEST_TIMEOUT = 408 + private const val HTTP_TOO_MANY_REQUESTS = 429 + private const val HTTP_SERVER_ERROR = 500 + private val PERMANENT_SERVER_ERRORS = setOf(501, 505, 507) const val REQUEST_PERMISSION = 3123 + /** + * Whether [e] is a lost connection or a refusal that says "not now" rather than "no". dav4jvm and Retrofit + * report a refused request as an HttpException, which is not an IOException. + */ + internal fun isRetryable(e: Throwable): Boolean = + when (e) { + is IOException -> true + is HttpException -> isTransientHttpError(e.code) + is retrofit2.HttpException -> isTransientHttpError(e.code()) + else -> false + } + + /** + * 5xx, 408 and 429 are tried again. Not 501, 505 and 507: a quota that is full or a server that cannot + * do the request does not change within the few attempts. + */ + internal fun isTransientHttpError(code: Int): Boolean = + (code >= HTTP_SERVER_ERROR && code !in PERMANENT_SERVER_ERRORS) || + code == HTTP_REQUEST_TIMEOUT || + code == HTTP_TOO_MANY_REQUESTS + // referenceIds the user cancelled - set synchronously here, before cancelUniqueWork() is // even called, so doWork() can see it immediately instead of waiting for isStopped, which // only becomes true once WorkManager's own async cancellation dispatch completes. diff --git a/app/src/test/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorkerTest.kt b/app/src/test/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorkerTest.kt index bdd55d53dfa..c365ba4be3c 100644 --- a/app/src/test/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorkerTest.kt +++ b/app/src/test/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorkerTest.kt @@ -20,13 +20,19 @@ import androidx.work.testing.WorkManagerTestInitHelper import com.nextcloud.talk.data.database.dao.ChatMessagesDao import com.nextcloud.talk.data.database.model.SendStatus import com.nextcloud.talk.users.UserManager +import okhttp3.ResponseBody.Companion.toResponseBody import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue import org.junit.Test import org.junit.runner.RunWith import org.mockito.kotlin.mock import org.mockito.kotlin.verify import org.robolectric.RobolectricTestRunner import org.robolectric.annotation.Config +import retrofit2.HttpException +import retrofit2.Response +import java.io.IOException import java.util.UUID /** @@ -56,6 +62,22 @@ class UploadAndShareFilesWorkerTest { verify(chatDao).updateTempMessageSendStatus(CONVERSATION_ID, "ref-b", SendStatus.FAILED) } + @Test + fun `a short outage of the server and a lost connection are tried again and a refusal is not`() { + for (code in listOf(408, 429, 500, 502, 503, 504)) { + assertTrue("$code", UploadAndShareFilesWorker.isTransientHttpError(code)) + assertTrue("retrofit $code", UploadAndShareFilesWorker.isRetryable(retrofitError(code))) + } + for (code in listOf(400, 401, 403, 404, 409, 412, 501, 505, 507)) { + assertFalse("$code", UploadAndShareFilesWorker.isTransientHttpError(code)) + assertFalse("retrofit $code", UploadAndShareFilesWorker.isRetryable(retrofitError(code))) + } + assertTrue(UploadAndShareFilesWorker.isRetryable(IOException("connection reset"))) + assertFalse(UploadAndShareFilesWorker.isRetryable(IllegalStateException("bug"))) + } + + private fun retrofitError(code: Int) = HttpException(Response.error(code, "".toResponseBody())) + private fun upload(referenceId: String): UUID = UploadAndShareFilesWorker.upload( userId = ACCOUNT_ID, diff --git a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt index b2a5f39a129..7021c9f9c02 100644 --- a/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt +++ b/app/src/test/java/com/nextcloud/talk/upload/chunked/ChunkedFileUploaderResumeTest.kt @@ -10,6 +10,7 @@ import android.app.Application import at.bitfire.dav4jvm.exception.HttpException import com.nextcloud.talk.api.NcApiCoroutines import com.nextcloud.talk.data.user.model.User +import com.nextcloud.talk.jobs.UploadAndShareFilesWorker import okhttp3.OkHttpClient import okhttp3.mockwebserver.Dispatcher import okhttp3.mockwebserver.MockResponse @@ -20,6 +21,7 @@ import org.junit.After import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertNull +import org.junit.Assert.assertThrows import org.junit.Assert.assertTrue import org.junit.Assert.fail import org.junit.Before @@ -187,6 +189,17 @@ class ChunkedFileUploaderResumeTest { assertEquals(0, requests.count { it.method == "DELETE" }) } + @Test + fun `the worker tries again after a service unavailable answer to a part but not after a rejected part`() { + putResponseCode = SERVICE_UNAVAILABLE + val unavailable = assertThrows(HttpException::class.java) { uploader.upload(file, null, "/Talk/video.mp4") } + assertTrue(UploadAndShareFilesWorker.isRetryable(unavailable)) + + putResponseCode = BAD_REQUEST + val rejected = assertThrows(HttpException::class.java) { uploader.upload(file, null, "/Talk/video.mp4") } + assertFalse(UploadAndShareFilesWorker.isRetryable(rejected)) + } + @Test fun `a lost network while sending a part is an error and keeps the parts on the server`() { putDisconnects = true @@ -445,5 +458,6 @@ class ChunkedFileUploaderResumeTest { private const val NOT_FOUND = 404 private const val BAD_REQUEST = 400 private const val SERVER_ERROR = 500 + private const val SERVICE_UNAVAILABLE = 503 } }