diff --git a/changes/unreleased/windows-cloud-files-streamed-hydration.md b/changes/unreleased/windows-cloud-files-streamed-hydration.md new file mode 100644 index 000000000..f106aaa00 --- /dev/null +++ b/changes/unreleased/windows-cloud-files-streamed-hydration.md @@ -0,0 +1,7 @@ +category: fix +issue: none +pull: 519 +platforms: windows +user-facing: yes + +Opening a cloud-only file in File Explorer now downloads it with one streamed request instead of one server request per 4 MiB, so large photos open about as fast as a normal download. diff --git a/tools/kotlin-file-size-baseline.txt b/tools/kotlin-file-size-baseline.txt index c874373b9..e7bc08d60 100644 --- a/tools/kotlin-file-size-baseline.txt +++ b/tools/kotlin-file-size-baseline.txt @@ -48,11 +48,11 @@ ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopAppUpdates.kt|9 ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopFileReadCache.kt|1536 ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopFileSyncEngine.kt|876 ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopFileSyncRemoteTree.kt|857 -ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopNextcloudServices.kt|6161 +ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopNextcloudServices.kt|6115 ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopVirtualRangeCache.kt|2759 ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/LinuxVirtualFileSystem.kt|1697 ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesJna.kt|1023 -ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesProvider.kt|2319 +ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesProvider.kt|2275 ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/DesktopVirtualRangeCacheTest.kt|2479 ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/JvmSupportIntakeTest.kt|2888 ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/LinuxVirtualFileSystemTest.kt|2149 diff --git a/ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudFileRange.kt b/ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudFileRange.kt index 32c6037ec..61dae74ac 100644 --- a/ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudFileRange.kt +++ b/ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudFileRange.kt @@ -11,6 +11,9 @@ class NextcloudFileRangeSession( private val readBlock: suspend (offset: Long, length: Int) -> ByteArray, private val closeBlock: () -> Unit = {}, private val beginUseBlock: () -> (() -> Unit)? = { {} }, + private val streamBlock: ( + suspend (offset: Long, length: Long, chunkBytes: Int, onChunk: (Long, ByteArray) -> Unit) -> Unit + )? = null, ) : AutoCloseable { init { require(size > 0L) { "A file range session must have a positive size." } @@ -18,6 +21,23 @@ class NextcloudFileRangeSession( suspend fun read(offset: Long, length: Int): ByteArray = readBlock(offset, length) + /** + * Delivers [length] bytes from [offset] in order, as chunks of [chunkBytes] except a shorter + * final chunk. A platform with streaming support serves the whole range with one request; + * otherwise this falls back to one bounded read per chunk. + */ + suspend fun stream(offset: Long, length: Long, chunkBytes: Int, onChunk: (offset: Long, bytes: ByteArray) -> Unit) { + require(offset >= 0L && length > 0L && chunkBytes > 0) { "The file range stream is invalid." } + require(offset <= size && length <= size - offset) { "The requested file range exceeds the file size." } + streamBlock?.let { stream -> return stream(offset, length, chunkBytes, onChunk) } + var position = 0L + while (position < length) { + val chunk = minOf(chunkBytes.toLong(), length - position).toInt() + onChunk(offset + position, read(offset + position, chunk)) + position += chunk + } + } + fun beginUse(): (() -> Unit)? = beginUseBlock() override fun close() = closeBlock() diff --git a/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopFileRangeTransport.kt b/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopFileRangeTransport.kt new file mode 100644 index 000000000..2d7cf98b2 --- /dev/null +++ b/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopFileRangeTransport.kt @@ -0,0 +1,108 @@ +package dev.obiente.nextcloudnative.app + +import java.io.InputStream +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicReference +import okhttp3.Call +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.ResponseBody + +/** + * Generation-pinned Files WebDAV range requests for one [NextcloudFileRangeSession]. + * + * Every request sends If-Match with the pinned ETag, requires HTTP 206 with exactly the requested + * Content-Range, and fails when the body is shorter or longer than that range. [stream] serves a + * whole range with one response and hands it over chunk by chunk as bytes arrive, so a large + * hydration costs one server round trip instead of one per chunk. + */ +internal class DesktopFileRangeTransport( + private val client: OkHttpClient, + private val url: String, + private val authorization: String, + private val userAgent: String, + private val etag: String, + private val size: Long, + private val closed: AtomicBoolean, + private val recordFailure: (startedNanos: Long, attempt: JvmNetworkRequestAttempt, failure: Throwable) -> Unit, +) { + private val activeCall = AtomicReference(null) + + fun cancel() { + activeCall.get()?.cancel() + } + + fun read(offset: Long, length: Int): ByteArray = execute(offset, length.toLong()) { body -> + body.byteStream().use { input -> input.readExactly(length).also { input.requireEnd() } } + } + + fun stream(offset: Long, length: Long, chunkBytes: Int, onChunk: (offset: Long, bytes: ByteArray) -> Unit) { + require(chunkBytes > 0) { "The file range chunk size must be greater than zero." } + execute(offset, length) { body -> + body.byteStream().use { input -> + var position = 0L + while (position < length) { + val chunk = input.readExactly(minOf(chunkBytes.toLong(), length - position).toInt()) + onChunk(offset + position, chunk) + position += chunk.size + } + input.requireEnd() + } + } + } + + private fun execute(offset: Long, length: Long, consume: (ResponseBody) -> T): T { + check(!closed.get()) { "The file range session is closed." } + require(offset >= 0L) { "The file range offset must not be negative." } + require(length > 0L) { "The file range length must be greater than zero." } + require(offset <= size && length <= size - offset) { "The requested file range exceeds the file size." } + val endInclusive = Math.addExact(offset, length - 1L) + val started = System.nanoTime() + val networkAttempt = JvmNetworkRequestAttempt() + val request = Request.Builder() + .url(url) + .get() + .header("Accept", "application/octet-stream") + .header("Range", "bytes=$offset-$endInclusive") + .header("If-Match", etag) + .header("User-Agent", userAgent) + .header("Authorization", authorization) + .tag(JvmNetworkRequestAttempt::class.java, networkAttempt) + .build() + val call = client.newCall(request) + check(activeCall.compareAndSet(null, call)) { "Only one file range read can run at a time." } + if (closed.get()) call.cancel() + try { + call.execute().use { response -> + if (response.code != 206) throw NextcloudFileRangeHttpException(response.code) + check(isExactHttpByteContentRange(response.header("Content-Range"), offset, endInclusive)) { + "The server returned a different file range than requested." + } + val contentLength = response.body.contentLength() + if (contentLength in 0 until length) throw JvmNetworkResponseTruncatedIOException() + check(contentLength == -1L || contentLength == length) { "The server returned an incomplete file range." } + return consume(response.body) + } + } catch (failure: Throwable) { + recordFailure(started, networkAttempt, failure) + throw failure + } finally { + activeCall.compareAndSet(call, null) + } + } + + private fun InputStream.readExactly(length: Int): ByteArray { + val bytes = ByteArray(length) + var filled = 0 + while (filled < length) { + val read = read(bytes, filled, length - filled) + if (read < 0) throw JvmNetworkResponseTruncatedIOException() + filled += read + } + return bytes + } + + private fun InputStream.requireEnd() { + check(read() == -1) { "The server returned more data than the requested file range." } + } +} diff --git a/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopNextcloudServices.kt b/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopNextcloudServices.kt index 65a6da420..948f8edd2 100644 --- a/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopNextcloudServices.kt +++ b/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopNextcloudServices.kt @@ -28,7 +28,6 @@ import java.util.UUID import java.util.concurrent.TimeUnit import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicBoolean -import java.util.concurrent.atomic.AtomicReference import java.util.prefs.Preferences import javax.swing.JFileChooser import javax.xml.parsers.DocumentBuilderFactory @@ -53,7 +52,6 @@ import kotlinx.coroutines.withContext import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import okhttp3.HttpUrl -import okhttp3.Call import okhttp3.HttpUrl.Companion.toHttpUrlOrNull import okhttp3.MediaType.Companion.toMediaType import okhttp3.OkHttpClient @@ -4445,73 +4443,29 @@ class DesktopNextcloudServices( synchronized(fileRangeSessionLock) { check(!sessionClearing) { "The account session is closing." } } - val safeEtag = requireSafeFileRangeEtag(expectedEtag) val closed = AtomicBoolean(false) - val activeCall = AtomicReference(null) + val transport = DesktopFileRangeTransport( + client = noRedirectHttpClient, + url = buildNextcloudFileUrl(session.serverUrl, userId, path), + authorization = "Basic " + Base64.getEncoder() + .encodeToString("${session.loginName}:${session.appPassword}".toByteArray()), + userAgent = USER_AGENT, + etag = requireSafeFileRangeEtag(expectedEtag), + size = size, + closed = closed, + ) { started, attempt, failure -> + recordDesktopStreamingFailure(session, "file_range", started, attempt, failure) + } lateinit var rangeSession: NextcloudFileRangeSession rangeSession = NextcloudFileRangeSession( size = size, - readBlock = { offset, length -> - withContext(Dispatchers.IO) { - check(!closed.get()) { "The file range session is closed." } - require(offset >= 0L) { "The file range offset must not be negative." } - require(length > 0) { "The file range length must be greater than zero." } - require(offset <= size && length.toLong() <= size - offset) { - "The requested file range exceeds the file size." - } - val endInclusive = Math.addExact(offset, length.toLong() - 1L) - val builder = Request.Builder() - .url(buildNextcloudFileUrl(session.serverUrl, userId, path)) - .get() - .header("Accept", "application/octet-stream") - .header("Range", "bytes=$offset-$endInclusive") - .header("If-Match", safeEtag) - .header("User-Agent", USER_AGENT) - val started = System.nanoTime() - val networkAttempt = JvmNetworkRequestAttempt() - builder.tag(JvmNetworkRequestAttempt::class.java, networkAttempt) - val encoded = Base64.getEncoder() - .encodeToString("${session.loginName}:${session.appPassword}".toByteArray()) - builder.header("Authorization", "Basic $encoded") - val call = noRedirectHttpClient.newCall(builder.build()) - check(activeCall.compareAndSet(null, call)) { "Only one file range read can run at a time." } - if (closed.get()) call.cancel() - try { - call.execute().use { response -> - if (response.code != 206) { - throw NextcloudFileRangeHttpException(response.code) - } - check(isExactHttpByteContentRange(response.header("Content-Range"), offset, endInclusive)) { - "The server returned a different file range than requested." - } - val responseBody = response.body - val contentLength = responseBody.contentLength() - if (contentLength in 0 until length.toLong()) { - throw JvmNetworkResponseTruncatedIOException() - } - check(contentLength == -1L || contentLength == length.toLong()) { - "The server returned an incomplete file range." - } - responseBody.byteStream().readBounded(length.toLong()) - .requireExactJvmNetworkResponseBytes(length) - } - } catch (failure: Throwable) { - recordDesktopStreamingFailure( - session = session, - streamKind = "file_range", - startedNanos = started, - attempt = networkAttempt, - failure = failure, - ) - throw failure - } finally { - activeCall.compareAndSet(call, null) - } - } + readBlock = { offset, length -> withContext(Dispatchers.IO) { transport.read(offset, length) } }, + streamBlock = { offset, length, chunkBytes, onChunk -> + withContext(Dispatchers.IO) { transport.stream(offset, length, chunkBytes, onChunk) } }, closeBlock = { if (closed.compareAndSet(false, true)) { - activeCall.get()?.cancel() + transport.cancel() synchronized(fileRangeSessionLock) { activeFileRangeSessions.remove(rangeSession) } diff --git a/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesHydration.kt b/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesHydration.kt new file mode 100644 index 000000000..0c35859ad --- /dev/null +++ b/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesHydration.kt @@ -0,0 +1,125 @@ +package dev.obiente.nextcloudnative.app + +import java.util.concurrent.CancellationException +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicReference + +internal const val WINDOWS_CLOUD_ALIGNMENT = 4 * 1024L + +/** Bytes handed to CfExecute at a time; small enough that an opening app sees data early. */ +internal const val WINDOWS_CLOUD_HYDRATION_CHUNK_BYTES = 1024 * 1024 + +/** + * Cancellation for one Cloud Files callback request. [cancel] also closes the read handle that + * the request has open, so a hydration blocked on a slow or stalled response stops at once + * instead of waiting for the next chunk or a network timeout. + */ +internal class WindowsCloudRequestCancellation { + private val cancelled = AtomicBoolean(false) + private val activeHandle = AtomicReference(null) + + fun get(): Boolean = cancelled.get() + + fun cancel() { + cancelled.set(true) + activeHandle.getAndSet(null)?.close() + } + + /** Runs [block] with [handle] closable by [cancel]; a request cancelled first never starts. */ + fun whileOpen(handle: AutoCloseable, block: () -> T): T { + activeHandle.set(handle) + try { + if (cancelled.get()) throw CancellationException("Windows cancelled the request.") + return block() + } finally { + activeHandle.compareAndSet(handle, null) + } + } +} + +/** Read access to one pinned remote file generation for hydration. */ +internal interface WindowsCloudFileReadHandle : AutoCloseable { + val size: Long + fun read(offset: Long, length: Int): ByteArray + + /** + * Delivers a range in order as [chunkBytes] chunks. Backends that can stream serve it with one + * request; this fallback costs one request per chunk. + */ + fun stream(offset: Long, length: Long, chunkBytes: Int, onChunk: (offset: Long, bytes: ByteArray) -> Unit) { + var position = 0L + while (position < length) { + val chunk = minOf(chunkBytes.toLong(), length - position).toInt() + onChunk(offset + position, read(offset + position, chunk)) + position += chunk + } + } +} + +internal data class WindowsCloudHydrationRange(val offset: Long, val length: Int) { + init { + require(offset >= 0L && offset % WINDOWS_CLOUD_ALIGNMENT == 0L) + require(length > 0) + } +} + +/** Produces CfExecute-compatible chunks; only the final range may end unaligned at EOF. */ +internal fun planWindowsCloudHydration( + requiredOffset: Long, + requiredLength: Long, + fileSize: Long, + maximumChunkBytes: Int = WINDOWS_CLOUD_HYDRATION_CHUNK_BYTES, +): List { + require(requiredOffset >= 0L && requiredLength > 0L && fileSize > 0L) + require( + maximumChunkBytes.toLong() >= WINDOWS_CLOUD_ALIGNMENT && + maximumChunkBytes.toLong() % WINDOWS_CLOUD_ALIGNMENT == 0L, + ) + val requiredEnd = minOf(fileSize, Math.addExact(requiredOffset, requiredLength)) + require(requiredOffset < requiredEnd) + val transferEnd = if (requiredEnd == fileSize) { + fileSize + } else { + minOf(fileSize, Math.addExact(requiredEnd, WINDOWS_CLOUD_ALIGNMENT - 1L) / WINDOWS_CLOUD_ALIGNMENT * WINDOWS_CLOUD_ALIGNMENT) + } + var cursor = requiredOffset - requiredOffset % WINDOWS_CLOUD_ALIGNMENT + return buildList { + while (cursor < transferEnd) { + val remaining = transferEnd - cursor + val length = minOf(maximumChunkBytes.toLong(), remaining).toInt() + add(WindowsCloudHydrationRange(cursor, length)) + cursor += length + } + } +} + +/** + * Serves one FETCH_DATA request. Windows asks for a whole file when an application opens it, so + * the aligned span is fetched as one streamed range rather than one server request per chunk, + * and each chunk is transferred as soon as it arrives. Cancelling closes [source], which stops a + * blocked read; a chunk that arrives after cancellation is never transferred. Any other mismatch + * fails the request. + */ +internal fun hydrateWindowsCloudRange( + source: WindowsCloudFileReadHandle, + requiredOffset: Long, + requiredLength: Long, + cancellation: WindowsCloudRequestCancellation, + transfer: (offset: Long, bytes: ByteArray) -> Unit, +) { + val plan = planWindowsCloudHydration(requiredOffset, requiredLength, source.size) + val start = plan.first().offset + val end = plan.last().let { it.offset + it.length } + var next = 0 + cancellation.whileOpen(source) { + source.stream(start, end - start, WINDOWS_CLOUD_HYDRATION_CHUNK_BYTES) { offset, bytes -> + if (cancellation.get()) throw CancellationException("Windows cancelled the hydration request.") + val expected = plan.getOrNull(next++) + check(expected != null && offset == expected.offset && bytes.size == expected.length) { + "The remote file returned an incomplete range." + } + transfer(offset, bytes) + } + } + check(next == plan.size) { "The remote file returned an incomplete range." } +} diff --git a/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesProvider.kt b/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesProvider.kt index 09c00abd1..fe14027fc 100644 --- a/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesProvider.kt +++ b/ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesProvider.kt @@ -24,43 +24,6 @@ import kotlinx.coroutines.runBlocking private const val MAX_WINDOWS_CLOUD_FILES_RECOVERY_ROOT_ATTEMPTS = 16 -internal data class WindowsCloudHydrationRange(val offset: Long, val length: Int) { - init { - require(offset >= 0L && offset % WINDOWS_CLOUD_ALIGNMENT == 0L) - require(length > 0) - } -} - -/** Produces CfExecute-compatible chunks; only the final range may end unaligned at EOF. */ -internal fun planWindowsCloudHydration( - requiredOffset: Long, - requiredLength: Long, - fileSize: Long, - maximumChunkBytes: Int = 4 * 1024 * 1024, -): List { - require(requiredOffset >= 0L && requiredLength > 0L && fileSize > 0L) - require( - maximumChunkBytes.toLong() >= WINDOWS_CLOUD_ALIGNMENT && - maximumChunkBytes.toLong() % WINDOWS_CLOUD_ALIGNMENT == 0L, - ) - val requiredEnd = minOf(fileSize, Math.addExact(requiredOffset, requiredLength)) - require(requiredOffset < requiredEnd) - val transferEnd = if (requiredEnd == fileSize) { - fileSize - } else { - minOf(fileSize, Math.addExact(requiredEnd, WINDOWS_CLOUD_ALIGNMENT - 1L) / WINDOWS_CLOUD_ALIGNMENT * WINDOWS_CLOUD_ALIGNMENT) - } - var cursor = requiredOffset - requiredOffset % WINDOWS_CLOUD_ALIGNMENT - return buildList { - while (cursor < transferEnd) { - val remaining = transferEnd - cursor - val length = minOf(maximumChunkBytes.toLong(), remaining).toInt() - add(WindowsCloudHydrationRange(cursor, length)) - cursor += length - } - } -} - internal data class WindowsCloudCallbackInfo( val connectionKey: Long, val transferKey: Long, @@ -231,11 +194,6 @@ internal interface WindowsCloudFilesApi : AutoCloseable { fun dehydrate(path: Path): Long } -internal interface WindowsCloudFileReadHandle : AutoCloseable { - val size: Long - fun read(offset: Long, length: Int): ByteArray -} - internal data class WindowsCloudFilesSummary( val cachedBytes: Long, val reclaimableBytes: Long, @@ -287,6 +245,8 @@ internal class DesktopNextcloudWindowsCloudFilesBackend( override val size: Long = source.size override fun read(offset: Long, length: Int): ByteArray = runBlocking(Dispatchers.IO) { source.read(offset, length) } + override fun stream(offset: Long, length: Long, chunkBytes: Int, onChunk: (Long, ByteArray) -> Unit) = + runBlocking(Dispatchers.IO) { source.stream(offset, length, chunkBytes, onChunk) } override fun close() = source.close() } } @@ -353,7 +313,7 @@ internal class WindowsCloudFilesProvider( ) : AutoCloseable, WindowsCloudFilesCallbacks { private val connection = AtomicLongState() private val apiClosed = AtomicBoolean(false) - private val cancelledRequests = ConcurrentHashMap() + private val cancelledRequests = ConcurrentHashMap() private val knownIdentities = ConcurrentHashMap() private val pathOperations = ConcurrentHashMap.newKeySet() private val queuedPathOperations = ConcurrentHashMap Unit>() @@ -836,7 +796,7 @@ internal class WindowsCloudFilesProvider( override fun fetchData(info: WindowsCloudCallbackInfo, requiredOffset: Long, requiredLength: Long) { if (callbacksPaused.get()) return - val cancellation = AtomicBoolean(false) + val cancellation = WindowsCloudRequestCancellation() cancelledRequests[info.requestKey] = cancellation executor.execute { val identity = runCatching { requireIdentity(info, expectDirectory = false) }.getOrElse { failure -> @@ -847,11 +807,8 @@ internal class WindowsCloudFilesProvider( try { backend.open(identity).use { source -> check(source.size == identity.size) { "The remote file generation has a different size." } - planWindowsCloudHydration(requiredOffset, requiredLength, identity.size).forEach { range -> - if (cancellation.get()) return@execute - val bytes = source.read(range.offset, range.length) - check(bytes.size == range.length) { "The remote file returned an incomplete range." } - api.transferData(info, range.offset, bytes) + hydrateWindowsCloudRange(source, requiredOffset, requiredLength, cancellation) { offset, bytes -> + api.transferData(info, offset, bytes) } } } catch (failure: Throwable) { @@ -865,12 +822,12 @@ internal class WindowsCloudFilesProvider( } override fun cancelFetchData(info: WindowsCloudCallbackInfo, offset: Long, length: Long) { - cancelledRequests[info.requestKey]?.set(true) + cancelledRequests[info.requestKey]?.cancel() } override fun fetchPlaceholders(info: WindowsCloudCallbackInfo, pattern: String?) { if (callbacksPaused.get()) return - val cancellation = AtomicBoolean(false) + val cancellation = WindowsCloudRequestCancellation() cancelledRequests[info.requestKey] = cancellation executor.execute { try { @@ -902,7 +859,7 @@ internal class WindowsCloudFilesProvider( } override fun cancelFetchPlaceholders(info: WindowsCloudCallbackInfo) { - cancelledRequests[info.requestKey]?.set(true) + cancelledRequests[info.requestKey]?.cancel() } override fun closed(info: WindowsCloudCallbackInfo, deleted: Boolean) { @@ -1147,7 +1104,7 @@ internal class WindowsCloudFilesProvider( api.disconnect(key) connection.compareAndSet(key, 0L) } - cancelledRequests.values.forEach { it.set(true) } + cancelledRequests.values.forEach { it.cancel() } cancelledRequests.clear() stopLocalWatcher() queuedPathOperations.clear() @@ -2292,7 +2249,6 @@ internal class WindowsCloudFilesProvider( } } -private const val WINDOWS_CLOUD_ALIGNMENT = 4 * 1024L private const val MAX_WINDOWS_WRITEBACK_ATTEMPTS = 5 private const val MAX_WINDOWS_DIRECTORY_REFRESH_ATTEMPTS = 4 private const val DELAYED_CORRUPT_ROOT_RECOVERY_POLL_MILLIS = 25L diff --git a/ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/DesktopFileRangeTransportTest.kt b/ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/DesktopFileRangeTransportTest.kt new file mode 100644 index 000000000..34e59db7c --- /dev/null +++ b/ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/DesktopFileRangeTransportTest.kt @@ -0,0 +1,104 @@ +package dev.obiente.nextcloudnative.app + +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertNotNull +import kotlin.test.assertTrue +import mockwebserver3.MockResponse +import mockwebserver3.MockWebServer +import okhttp3.OkHttpClient +import okio.Buffer + +class DesktopFileRangeTransportTest { + private val content = ByteArray(3 * 1024 * 1024 + 100) { (it % 251).toByte() } + + private fun partial(offset: Long, endInclusive: Long, body: ByteArray = content.copyOfRange(offset.toInt(), endInclusive.toInt() + 1)) = + MockResponse.Builder() + .code(206) + .addHeader("Content-Range", "bytes $offset-$endInclusive/${content.size}") + .body(Buffer().write(body)) + .build() + + private fun MockWebServer.transport(failures: MutableList = mutableListOf()) = DesktopFileRangeTransport( + client = OkHttpClient(), + url = url("/remote.php/dav/files/synthetic/photo.jpg").toString(), + authorization = "Basic c3ludGhldGljOnNlY3JldA==", + userAgent = "test", + etag = "\"etag-01\"", + size = content.size.toLong(), + closed = AtomicBoolean(false), + ) { _, _, failure -> failures += failure } + + @Test + fun `a streamed range is one pinned request delivered in ordered chunks`() { + MockWebServer().use { server -> + server.enqueue(partial(0L, content.size - 1L)) + server.start() + val chunks = mutableListOf>() + + server.transport().stream(0L, content.size.toLong(), 1024 * 1024) { offset, bytes -> chunks += offset to bytes } + + val request = assertNotNull(server.takeRequest(5, TimeUnit.SECONDS)) + assertEquals("bytes=0-${content.size - 1}", request.headers["Range"]) + assertEquals("\"etag-01\"", request.headers["If-Match"]) + assertEquals(1, server.requestCount) + assertEquals(listOf(0L, 1_048_576L, 2_097_152L, 3_145_728L), chunks.map { it.first }) + assertEquals(100, chunks.last().second.size) + assertContentEquals(content, chunks.flatMap { it.second.asIterable() }.toByteArray()) + } + } + + @Test + fun `a bounded read still returns the exact range`() { + MockWebServer().use { server -> + server.enqueue(partial(4_096L, 8_191L)) + server.start() + + assertContentEquals(content.copyOfRange(4_096, 8_192), server.transport().read(4_096L, 4_096)) + } + } + + @Test + fun `responses that do not match the pinned range fail before any chunk is delivered`() { + MockWebServer().use { server -> + server.enqueue(MockResponse.Builder().code(412).build()) + server.enqueue(partial(0L, 4_095L).newBuilder().setHeader("Content-Range", "bytes 0-4095/1").build()) + server.enqueue(partial(0L, 8_191L, body = content.copyOfRange(0, 4_096))) + server.enqueue(partial(0L, 4_095L, body = content.copyOfRange(0, 8_192))) + server.start() + val failures = mutableListOf() + val transport = server.transport(failures) + val delivered = mutableListOf() + + assertEquals(412, assertFailsWith { + transport.stream(0L, 4_096L, 4_096) { offset, _ -> delivered += offset } + }.status) + assertFailsWith { transport.stream(0L, 4_096L, 4_096) { offset, _ -> delivered += offset } } + assertFailsWith { + transport.stream(0L, 8_192L, 4_096) { offset, _ -> delivered += offset } + } + assertFailsWith { transport.stream(0L, 4_096L, 4_096) { offset, _ -> delivered += offset } } + + assertEquals(emptyList(), delivered) + assertEquals(4, failures.size) + } + } + + @Test + fun `a closed transport refuses new requests`() { + MockWebServer().use { server -> + server.start() + val closed = AtomicBoolean(true) + val transport = DesktopFileRangeTransport( + OkHttpClient(), server.url("/f").toString(), "Basic x", "test", "\"e\"", content.size.toLong(), closed, + ) { _, _, _ -> } + + assertFailsWith { transport.stream(0L, 4_096L, 4_096) { _, _ -> } } + assertTrue(server.requestCount == 0) + } + } +} diff --git a/ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesHydrationTest.kt b/ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesHydrationTest.kt new file mode 100644 index 000000000..68e505887 --- /dev/null +++ b/ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/WindowsCloudFilesHydrationTest.kt @@ -0,0 +1,155 @@ +package dev.obiente.nextcloudnative.app + +import java.util.concurrent.CancellationException +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import kotlin.concurrent.thread +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertTrue + +class WindowsCloudFilesHydrationTest { + private val content = ByteArray(5 * WINDOWS_CLOUD_HYDRATION_CHUNK_BYTES + 12_345) { (it % 251).toByte() } + + /** Records each backend request; [streaming] backends answer a whole range at once. */ + private inner class RecordingHandle(private val streaming: Boolean) : WindowsCloudFileReadHandle { + val requests = mutableListOf>() + var shortenChunk: Int? = null + override val size = content.size.toLong() + override fun read(offset: Long, length: Int): ByteArray { + requests += offset to length.toLong() + return content.copyOfRange(offset.toInt(), offset.toInt() + length) + } + override fun stream(offset: Long, length: Long, chunkBytes: Int, onChunk: (Long, ByteArray) -> Unit) { + if (!streaming) return super.stream(offset, length, chunkBytes, onChunk) + requests += offset to length + var position = 0L + var index = 0 + while (position < length) { + val chunk = minOf(chunkBytes.toLong(), length - position).toInt() + val start = (offset + position).toInt() + val delivered = if (shortenChunk == index++) chunk - 1 else chunk + onChunk(offset + position, content.copyOfRange(start, start + delivered)) + position += chunk + } + } + override fun close() = Unit + } + + private fun hydrate( + handle: WindowsCloudFileReadHandle, + offset: Long = 0L, + length: Long = content.size.toLong(), + ): List> = buildList { + hydrateWindowsCloudRange(handle, offset, length, WindowsCloudRequestCancellation()) { at, bytes -> add(at to bytes) } + } + + @Test + fun `a whole-file request is served with one streamed backend request`() { + val handle = RecordingHandle(streaming = true) + + val transfers = hydrate(handle) + + assertEquals(listOf(0L to content.size.toLong()), handle.requests) + assertEquals(6, transfers.size) + assertTrue(transfers.dropLast(1).all { (_, bytes) -> bytes.size == WINDOWS_CLOUD_HYDRATION_CHUNK_BYTES }) + assertTrue(transfers.all { (offset, _) -> offset % WINDOWS_CLOUD_ALIGNMENT == 0L }) + assertContentEquals(content, transfers.flatMap { it.second.asIterable() }.toByteArray()) + } + + @Test + fun `an interior request streams only its aligned span`() { + val handle = RecordingHandle(streaming = true) + + val transfers = hydrate(handle, offset = 2_000_123L, length = 10L) + + assertEquals(listOf(1_998_848L to 4_096L), handle.requests) + assertEquals(listOf(1_998_848L), transfers.map { it.first }) + } + + @Test + fun `a backend without streaming still transfers the exact bytes`() { + val handle = RecordingHandle(streaming = false) + + val transfers = hydrate(handle) + + assertEquals(6, handle.requests.size) + assertContentEquals(content, transfers.flatMap { it.second.asIterable() }.toByteArray()) + } + + @Test + fun `cancellation stops the stream before the next chunk is transferred`() { + val handle = RecordingHandle(streaming = true) + val cancellation = WindowsCloudRequestCancellation() + var transferred = 0 + + assertFailsWith { + hydrateWindowsCloudRange(handle, 0L, content.size.toLong(), cancellation) { _, _ -> + if (++transferred == 2) cancellation.cancel() + } + } + + assertEquals(2, transferred) + } + + @Test + fun `cancellation closes a stream that is blocked waiting for the server`() { + val streaming = CountDownLatch(1) + val closed = CountDownLatch(1) + val stalled = object : WindowsCloudFileReadHandle { + override val size = content.size.toLong() + override fun read(offset: Long, length: Int) = error("Hydration must stream.") + override fun stream(offset: Long, length: Long, chunkBytes: Int, onChunk: (Long, ByteArray) -> Unit) { + streaming.countDown() + // Models a response whose next chunk never arrives until the transport is closed. + check(closed.await(30, TimeUnit.SECONDS)) { "The stalled read was never interrupted." } + throw java.io.IOException("Canceled") + } + override fun close() = closed.countDown() + } + val cancellation = WindowsCloudRequestCancellation() + val transfers = mutableListOf() + val canceller = thread { + check(streaming.await(5, TimeUnit.SECONDS)) + cancellation.cancel() + } + val started = System.nanoTime() + + assertFailsWith { + hydrateWindowsCloudRange(stalled, 0L, content.size.toLong(), cancellation) { offset, _ -> transfers += offset } + } + + canceller.join() + assertTrue(TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - started) < 5) + assertTrue(cancellation.get()) + assertEquals(emptyList(), transfers) + } + + @Test + fun `a request cancelled before it starts never opens a stream`() { + val handle = RecordingHandle(streaming = true) + val cancellation = WindowsCloudRequestCancellation().apply { cancel() } + + assertFailsWith { + hydrateWindowsCloudRange(handle, 0L, content.size.toLong(), cancellation) { _, _ -> error("Nothing may transfer.") } + } + + assertEquals(emptyList(), handle.requests) + } + + @Test + fun `a short chunk fails the request without transferring it`() { + val handle = RecordingHandle(streaming = true).apply { shortenChunk = 1 } + val transfers = mutableListOf() + + assertFailsWith { + hydrateWindowsCloudRange(handle, 0L, content.size.toLong(), WindowsCloudRequestCancellation()) { offset, _ -> + transfers += offset + } + } + + assertEquals(listOf(0L), transfers) + } +} diff --git a/website/public/screenshots/capture-manifest.json b/website/public/screenshots/capture-manifest.json index 4225e8b2b..a1be74548 100644 --- a/website/public/screenshots/capture-manifest.json +++ b/website/public/screenshots/capture-manifest.json @@ -757,7 +757,7 @@ "ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudDocumentPreview.kt": "1c503e08e12170b9d211cdc0dbc04b1d0f0f7a97354e837ad18bd36574b2b5ad", "ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudEntryNavigation.kt": "b558f3d0b663539c37e4d9ba055829fa99cdb03900d88461093aa3b884128dcf", "ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudFileListing.kt": "1f3de4f31d44d3ead139686d3fbcfbd27ccb520a9dab06ccc1d3e582ee25fb71", - "ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudFileRange.kt": "0d0464daf9359af5ec0ef4b1930ab04856d52f73f6c472e47b66ceaa5843283b", + "ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudFileRange.kt": "a0cc098f7fb0a0cced2dfc4141faf6cb38457896524d13df6cb3dc8a1cd3c852", "ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudLinkRouting.kt": "d1a9fbf7d82b48ed227708221eec71c27dcc9828fb12467d132bde9a6a00f6f8", "ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudMediaViewer.kt": "f6b7bbbaf9269e1b22ae43aee40cd765e140f5f5fe2444894bcc79b2140bccad", "ui/src/commonMain/kotlin/dev/obiente/nextcloudnative/app/NextcloudMediaViewerActions.kt": "5c627c3980c086199aa583191d9623e3adc1f0dbbb9dc753cf6cbe176fe31667",