From b669f9c8d9f550e450bb4f980a855f9f494231f1 Mon Sep 17 00:00:00 2001 From: voir-sw Date: Wed, 7 Oct 2026 09:12:20 +0000 Subject: [PATCH 1/3] Release v0.1.3 --- VERSION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VERSION b/VERSION index d917d3e..b1e80bb 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.1.2 +0.1.3 From aa2dbb83c597423d9ffe34250a84dee292145ec1 Mon Sep 17 00:00:00 2001 From: Gary Bezruchko Date: Wed, 7 Oct 2026 19:31:34 +0200 Subject: [PATCH 2/3] Fix file sync --- .../reflector/sync/core/CollectionHandle.kt | 30 ++++ .../reflector/sync/core/blob/BlobSyncState.kt | 9 +- .../reflector/sync/core/log/SyncLogEvent.kt | 17 ++ .../reflector/sync/engine/CollectionWorker.kt | 12 ++ .../sync/engine/DefaultCollectionHandle.kt | 30 ++-- .../sync/engine/blob/BlobReconciler.kt | 47 ++++++ .../reflector/sync/engine/blob/BlobWorker.kt | 149 +++++++++++++++++- .../sync/network/KtorBlobTransport.kt | 92 ++++++++++- .../sync/persistence/blob/BlobRecordStore.kt | 61 +++++++ .../sync/persistence/blob/SyncBlobDao.kt | 90 +++++++++++ .../sync/network/KtorBlobTransportTest.kt | 68 +++++++- .../reflector/sync/engine/SyncEngineTest.kt | 20 ++- .../sync/engine/blob/BlobWorkerTest.kt | 113 ++++++++++++- 13 files changed, 707 insertions(+), 31 deletions(-) diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/CollectionHandle.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/CollectionHandle.kt index 338020f..de087bd 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/CollectionHandle.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/CollectionHandle.kt @@ -1,6 +1,7 @@ package dev.voir.reflector.sync.core import dev.voir.reflector.sync.core.blob.BlobSyncState +import dev.voir.reflector.sync.core.blob.BlobTransferState import dev.voir.reflector.sync.core.conflict.Conflict import dev.voir.reflector.sync.core.conflict.ConflictId import dev.voir.reflector.sync.core.conflict.Resolution @@ -62,6 +63,20 @@ public interface CollectionHandle { */ public fun blob(id: BlobId): Flow + /** + * Publishes the progress of every file some document of this collection still points at. + * + * For a screen that reports on files rather than drawing one — what is still on its way out, + * what is arriving, what keeps failing. Files nothing references any more are left out: they + * are on their way to being offered back to the application and are not the owner's to worry + * about. + * + * The same caveats as [blob] apply to each entry: [BlobTransferState.REMOTE] and + * [BlobTransferState.DOWNLOADING] are ordinary, and a [BlobSyncState.lastError] on a file that + * is still moving is a failed attempt with another one coming. + */ + public val blobs: Flow> + /** * Asks for a file's bytes to be brought to this device. * @@ -109,6 +124,21 @@ public interface CollectionHandle { */ public suspend fun evict(id: BlobId) + /** + * Tries a file again now, whichever way it was moving. + * + * The call behind a "try again" button on a screen listing files. A file given up on — + * [BlobTransferState.UNAVAILABLE] — starts again from zero attempts: as an upload when this + * device holds its bytes, as a download otherwise. A file still moving has its backoff cut short + * and its attempts forgotten, which is the difference from [fetch]: a person asking is a reason + * to try sooner, a screen redrawing itself is not. + * + * A file the library has never heard of, or one already where it belongs, is left alone. + * + * @param id File to try again. + */ + public suspend fun retry(id: BlobId) + /** * Runs a block of local changes as one transaction of the application's database. * diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/blob/BlobSyncState.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/blob/BlobSyncState.kt index b2dad8c..6cef97f 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/blob/BlobSyncState.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/blob/BlobSyncState.kt @@ -3,9 +3,8 @@ package dev.voir.reflector.sync.core.blob /** * Progress of one file, as a user interface reads it. * - * Published per file rather than as a list of every file in a collection, because the one screen - * that needs this already knows which identifier it is drawing: a photograph in a list binds to its - * own state and nothing else. + * Published per file for a screen drawing one, and as a whole collection's worth through + * [dev.voir.reflector.sync.core.CollectionHandle.blobs] for a screen that reports on all of them. * * Between them [state] and [wanted] cover every case a screen has to draw, which is why both are * here: the state says where the bytes are, and the flag says whether this device is trying to get @@ -35,6 +34,9 @@ package dev.voir.reflector.sync.core.blob * a value here while the state is still moving means an attempt failed and another is coming, and * a screen that treated it as final would tell the user a photograph was lost every time a train * went into a tunnel. + * @property attempts Attempts that have failed since the transfer last started from zero. An upload + * is never given up on while its bytes are here, so this — not the state — is what says one keeps + * failing. Waiting for a file that another device has not finished sending is not counted. */ public data class BlobSyncState( public val state: BlobTransferState, @@ -42,4 +44,5 @@ public data class BlobSyncState( public val size: Long?, public val transferred: Long, public val lastError: String?, + public val attempts: Int, ) diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/log/SyncLogEvent.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/log/SyncLogEvent.kt index 8f8bc65..c5ce114 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/log/SyncLogEvent.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/log/SyncLogEvent.kt @@ -143,6 +143,23 @@ public enum class SyncLogEvent( /** A transfer did not complete and will be tried again. */ BLOB_TRANSFER_FAILED(SyncLogSource.BLOBS), + /** + * An upload has failed as many times as a download is allowed to before being given up on, and + * is still being tried. + * + * Recorded once, as a warning, because it is the moment a failure stops looking like a passing + * network and starts looking like something nobody will notice otherwise: an upload is never + * given up on while its bytes are here, so without this line a bucket refusing every request + * would be visible only at debug level. + */ + BLOB_UPLOAD_STUCK(SyncLogSource.BLOBS), + + /** + * A file was put back to work: asked to be tried again, or found given up on while its bytes are + * still on this device. + */ + BLOB_RETRIED(SyncLogSource.BLOBS), + /** * A file will not arrive, and the application has been told. * diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/CollectionWorker.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/CollectionWorker.kt index 7e54ed7..81db4a7 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/CollectionWorker.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/CollectionWorker.kt @@ -499,6 +499,18 @@ internal class CollectionWorker( reconciler?.evict(blobId) } + /** + * Puts a file back to work now and wakes the worker for it. + * + * @param blobId File to try again. + */ + suspend fun retryBlob(blobId: BlobId) { + val reconciler = reconciler ?: return + if (reconciler.retry(blobId)) { + blobWorker?.requestTransfers() + } + } + /** * Compares the shape the adapter declares with the one this collection was synchronised under. * diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/DefaultCollectionHandle.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/DefaultCollectionHandle.kt index e534cd1..6828508 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/DefaultCollectionHandle.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/DefaultCollectionHandle.kt @@ -18,6 +18,7 @@ import dev.voir.reflector.sync.engine.conflict.ConflictCoordinator import dev.voir.reflector.sync.engine.mutation.MutationCoordinator import dev.voir.reflector.sync.persistence.SyncStores import dev.voir.reflector.sync.persistence.SyncTransactionRunner +import dev.voir.reflector.sync.persistence.blob.BlobRecord import dev.voir.reflector.sync.persistence.conflict.StoredConflict import dev.voir.reflector.sync.persistence.group.PendingGroup import dev.voir.reflector.sync.protocol.BlobId @@ -95,16 +96,11 @@ internal class DefaultCollectionHandle( ) override fun blob(id: BlobId): Flow = - stores.blobs.observe(scope, collection, id).map { record -> - record?.let { - BlobSyncState( - state = it.state, - wanted = it.wanted, - size = it.stat?.size, - transferred = it.transferred, - lastError = it.lastError, - ) - } + stores.blobs.observe(scope, collection, id).map { record -> record?.toSyncState() } + + override val blobs: Flow> = + stores.blobs.observeReferenced(scope, collection).map { records -> + records.associate { record -> record.blobId to record.toSyncState() } } override val conflicts: Flow> = @@ -120,6 +116,10 @@ internal class DefaultCollectionHandle( worker.evictBlob(id) } + override suspend fun retry(id: BlobId) { + worker.retryBlob(id) + } + override suspend fun mutate(block: suspend MutationScope.() -> R): R { val result = mutations.mutate(scope, collection) { block() } // A local change is a reason to synchronise, and the user expects it to leave as soon as the @@ -244,6 +244,16 @@ internal class DefaultCollectionHandle( stored } + private fun BlobRecord.toSyncState(): BlobSyncState = + BlobSyncState( + state = state, + wanted = wanted, + size = stat?.size, + transferred = transferred, + lastError = lastError, + attempts = attempts, + ) + private fun StoredConflict.toConflict(): Conflict = Conflict( id = conflictId, diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/blob/BlobReconciler.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/blob/BlobReconciler.kt index 04cb199..5ace0a1 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/blob/BlobReconciler.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/blob/BlobReconciler.kt @@ -147,6 +147,53 @@ internal class BlobReconciler( } } + /** + * Puts a file back to work now, whichever way it was moving. + * + * A file given up on goes back to where its direction starts, with no attempts and no error: an + * upload when the application still holds the bytes, a download otherwise, wanted either way — + * somebody asked. A file still moving keeps its state and loses its attempts and its backoff, so + * the worker takes it on its next pass with a first attempt's short delays. A file already where + * it belongs, or one the library does not track, is left alone. + * + * @param blobId File to try again. + * @return Whether anything was put back to work, so the caller knows to wake the worker. + */ + suspend fun retry(blobId: BlobId): Boolean { + // Outside the transaction, as everywhere else this asks the application about its own store. + val stat = runCatching { blobStore.stat(blobId) }.getOrNull() + val retried = + transactions.transaction { + val record = stores.blobs.find(scope, collection, blobId) + when (record?.state) { + BlobTransferState.UNAVAILABLE -> { + val restart = if (stat != null) BlobTransferState.LOCAL else BlobTransferState.REMOTE + stores.blobs.setState(scope, collection, blobId, restart) + stores.blobs.setWanted(scope, collection, blobId, wanted = true) + true + } + + BlobTransferState.LOCAL, + BlobTransferState.UPLOADING, + BlobTransferState.REMOTE, + BlobTransferState.DOWNLOADING, + -> { + stores.blobs.resetAttempts(scope, collection, blobId) + true + } + + BlobTransferState.UPLOADED, BlobTransferState.READY, null -> false + } + } + if (retried) { + log.debug( + SyncLogEvent.BLOB_RETRIED, + context = { mapOf("blob" to blobId.value.toString()) }, + ) { "the application asked for a file to be tried again" } + } + return retried + } + /** * Records that the application wants a file's bytes on this device. * diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/blob/BlobWorker.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/blob/BlobWorker.kt index 7eab878..36204de 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/blob/BlobWorker.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/blob/BlobWorker.kt @@ -22,13 +22,16 @@ import dev.voir.reflector.sync.protocol.blob.BlobDescriptor import dev.voir.reflector.sync.protocol.blob.BlobState import dev.voir.reflector.sync.protocol.config.BlobLimits import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.launch import kotlin.time.Clock +import kotlin.time.Duration.Companion.minutes /** * Moves the bytes, beside the cycle rather than inside it. @@ -59,7 +62,8 @@ import kotlin.time.Clock * without this a record would sit until the next timer, which on a device nobody touches is * however long that is. * @property backoff Delay policy for transfers that failed. - * @property maxAttempts Transfers tried before a file is given up on. + * @property maxAttempts Failed attempts before a download is given up on, and after which an upload + * that keeps failing is reported. */ internal class BlobWorker( private val scope: ScopeId, @@ -80,9 +84,18 @@ internal class BlobWorker( /** Pending requests to look for work; conflated, since one pass answers any number of them. */ private val requests = Channel(Channel.CONFLATED) + /** The sleep until the next backoff runs out; replaced each time the worker runs dry. */ + private var wake: Job? = null + /** Starts the loop. Cancelling the coroutine scope stops it. */ fun start() { coroutineScope.launch { + runCatching { resumeGivenUpUploads() } + .onFailure { failure -> + log.error(SyncLogEvent.BLOB_TRANSFER_FAILED, failure) { + "files given up on could not be looked at; they wait for somebody to ask again" + } + } for (request in requests) { runCatching { drain() } .onFailure { failure -> @@ -99,6 +112,39 @@ internal class BlobWorker( requests.trySend(Unit) } + /** + * Puts back to work every upload an earlier version of this library gave up on. + * + * Uploads used to be given up on after a handful of attempts, which a minute without a network + * was enough to use up, and nothing ever tried them again: the record had long since been + * published naming the file, so every other device saw an attachment whose bytes would never + * arrive. A file given up on whose bytes are still here can only be one of those — a download + * that was given up on has no bytes — so it is sent again. One the server will not take for its + * size is left where it is, since sending it again would only be refused again. + */ + private suspend fun resumeGivenUpUploads() { + val givenUp = + transactions.transaction { + stores.blobs.all(scope, collection).filter { it.state == BlobTransferState.UNAVAILABLE } + } + for (record in givenUp) { + val stat = blobStore.statOrNull(record.blobId) ?: continue + if (stat.size > limits.maxBlobBytes) { + continue + } + transactions.transaction { + stores.blobs.setState(scope, collection, record.blobId, BlobTransferState.LOCAL) + } + log.info( + SyncLogEvent.BLOB_RETRIED, + context = { mapOf("blob" to record.blobId.value.toString()) }, + ) { "an upload that was given up on still has its bytes on this device; sending it again" } + } + if (givenUp.isNotEmpty()) { + requestTransfers() + } + } + /** * Sends and fetches until nothing is ready to move. * @@ -111,11 +157,33 @@ internal class BlobWorker( while (true) { val moved = uploadBatch() || downloadBatch() if (!moved) { + wakeWhenDue() return } } } + /** + * Sleeps until the next file serving a backoff becomes due, then looks again. + * + * Without it a retry happened whenever the next cycle ran, which on a device nobody touches is + * the periodic timer — a quarter of an hour — so the backoff said one thing and the worker did + * another: a bucket fixed a minute ago left a receipt unsent for fourteen more, and a file + * waiting for another device was looked for far less often than it was meant to be. + */ + private suspend fun wakeWhenDue() { + val now = clock.now().toEpochMilliseconds() + val due = transactions.transaction { stores.blobs.nextRetryAt(scope, collection, now) } + wake?.cancel() + wake = + due?.let { + coroutineScope.launch { + delay(due - now) + requestTransfers() + } + } + } + /** * Sends the next few files that are due. * @@ -199,7 +267,7 @@ internal class BlobWorker( ) transactions.transaction { stores.blobs.setDeclaration(scope, collection, record.blobId, stat) - stores.blobs.setState(scope, collection, record.blobId, BlobTransferState.UPLOADING) + enter(record, BlobTransferState.UPLOADING) } // The server has heard of the file now, which is all a deferred record was waiting for. // Its bytes may still take minutes, and the record should not wait them out. @@ -226,8 +294,24 @@ internal class BlobWorker( log, ) onProgressed() - } catch (failure: SyncTransportFailure) { + } catch (failure: SyncTransportFailure.BlobRefused) { + // The server refuses the file itself — its identifier names other bytes, or it is over + // the ceiling. Sending the same bytes again cannot change that answer. retryOrGiveUp(record, failure) + } catch (failure: SyncTransportFailure) { + // Never given up on while the bytes are here. The record naming this file has been on + // every other device since registration, and those devices can do nothing but wait for + // it; a network that is gone for a minute, or a bucket that is misconfigured for a day, + // is something to outlast rather than a reason to abandon the owner's receipt. + schedule(record, failure.message ?: "the transfer did not complete") + if (record.attempts + 1 == maxAttempts) { + log.warn( + SyncLogEvent.BLOB_UPLOAD_STUCK, + context = { + mapOf("blob" to record.blobId.value.toString(), "attempts" to maxAttempts.toString()) + }, + ) { "this file keeps failing to upload and is still being tried: ${failure.message}" } + } } } @@ -250,14 +334,14 @@ internal class BlobWorker( SyncLogEvent.BLOB_NOT_READY, context = { mapOf("blob" to record.blobId.value.toString()) }, ) { "the file is registered and still arriving at the server; this device waits for it" } - schedule(record, "the server does not have the bytes yet") + wait(record, "the server does not have the bytes yet") return } val stat = BlobStat(answer.blob.size, answer.blob.contentType, answer.blob.checksum) transactions.transaction { stores.blobs.setDeclaration(scope, collection, record.blobId, stat) - stores.blobs.setState(scope, collection, record.blobId, BlobTransferState.DOWNLOADING) + enter(record, BlobTransferState.DOWNLOADING) } val written = reportingProgress(record.blobId) { onProgress -> @@ -292,6 +376,28 @@ internal class BlobWorker( } } + /** + * Moves a file into the state its transfer runs in, unless a previous attempt already did. + * + * Moving to a state forgets the attempts, which is right the first time and wrong every time + * after: a transfer that failed while already in this state would start each retry from zero, + * so its backoff never grew — a refused upload was sent again every second or two for as long + * as the refusal lasted — and a download never reached the attempt it is given up on at. + * + * Runs inside the caller's transaction. + * + * @param record File as it was read for this attempt. + * @param state State the transfer runs in. + */ + private suspend fun enter( + record: BlobRecord, + state: BlobTransferState, + ) { + if (record.state != state) { + stores.blobs.setState(scope, collection, record.blobId, state) + } + } + /** * Runs one transfer while writing its progress to the file's row. * @@ -380,6 +486,31 @@ internal class BlobWorker( ) { "the transfer did not complete and will be tried again: $reason" } } + /** + * Looks again later for a file another device has not finished sending, without counting it. + * + * Not an attempt: nothing failed here, and spending attempts on it gave a receipt up a minute + * after it was attached. The delay is fixed rather than growing because nothing is being backed + * off from, and the scope's channel wakes the worker sooner when the file becomes usable. + * + * @param record File that is not on the server yet. + * @param reason What to show for it meanwhile. + */ + private suspend fun wait( + record: BlobRecord, + reason: String, + ) { + transactions.transaction { + stores.blobs.recordWait( + scope = scope, + collection = collection, + blobId = record.blobId, + error = reason, + nextRetryAt = (clock.now() + NOT_READY_DELAY).toEpochMilliseconds(), + ) + } + } + /** * Stops trying and tells the application, which is the only party that can do anything about it. * @@ -429,7 +560,13 @@ internal class BlobWorker( /** Transfers running at once. Two keeps a connection busy without monopolising a phone's. */ const val CONCURRENCY = 2 - /** Attempts before a file is given up on and the application is told. */ + /** + * Attempts before a download is given up on and the application is told. An upload is never + * given up on for a failed transfer; reaching this many is only reported. + */ const val DEFAULT_MAX_ATTEMPTS = 5 + + /** How long to wait before asking again for a file another device is still sending. */ + val NOT_READY_DELAY = 1.minutes } } diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/network/KtorBlobTransport.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/network/KtorBlobTransport.kt index ca90689..e5fffd7 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/network/KtorBlobTransport.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/network/KtorBlobTransport.kt @@ -17,6 +17,7 @@ import dev.voir.reflector.sync.protocol.blob.BlobTicket import dev.voir.reflector.sync.protocol.blob.BlobTicketMethod import io.ktor.client.HttpClient import io.ktor.client.call.body +import io.ktor.client.request.HttpRequestBuilder import io.ktor.client.request.get import io.ktor.client.request.header import io.ktor.client.request.post @@ -25,18 +26,22 @@ import io.ktor.client.request.setBody import io.ktor.client.statement.HttpResponse import io.ktor.client.statement.bodyAsChannel import io.ktor.http.ContentType +import io.ktor.http.HttpHeaders import io.ktor.http.HttpMethod import io.ktor.http.HttpStatusCode import io.ktor.http.appendPathSegments import io.ktor.http.content.OutgoingContent import io.ktor.http.contentType +import io.ktor.http.isSuccess import io.ktor.utils.io.ByteWriteChannel import io.ktor.utils.io.readAvailable +import io.ktor.utils.io.readRemaining import io.ktor.utils.io.writeFully import kotlinx.io.Buffer import kotlinx.io.RawSink import kotlinx.io.RawSource import kotlinx.io.readByteArray +import kotlinx.io.readString import kotlin.coroutines.cancellation.CancellationException /** @@ -160,7 +165,7 @@ public class KtorBlobTransport( method = ticket.method.toHttpMethod() // Exactly what the ticket names and nothing else. In particular no credential of // this library's: the destination is not the synchronisation server. - ticket.headers.forEach { (name, value) -> header(name, value) } + ticketHeaders(ticket) setBody( object : OutgoingContent.WriteChannelContent() { override val contentLength: Long = stat.size @@ -171,7 +176,7 @@ public class KtorBlobTransport( }, ) }.execute { response -> - requests.ensureSuccess(response) + ensureStored(response, "upload") } } } @@ -187,14 +192,65 @@ public class KtorBlobTransport( client .prepareRequest(ticket.url) { method = ticket.method.toHttpMethod() - ticket.headers.forEach { (name, value) -> header(name, value) } + ticketHeaders(ticket) }.execute { response -> - requests.ensureSuccess(response) + ensureStored(response, "download") copy(response.bodyAsChannel(), sink, onProgress) } } } + /** + * Checks the storage's answer to a presigned request, in the storage's own terms. + * + * The synchronisation server's status mapping does not apply here, and used to be applied: a + * bucket's `403` is a refused signature or a missing permission on the host's side, not a + * revoked scope, and reporting it as one sent every diagnosis of a misconfigured bucket in the + * wrong direction. Only `404` keeps a meaning of its own — the object is not there — and only + * for a download. + * + * What is reported is the status and the S3 error code (`SignatureDoesNotMatch`, + * `AccessDenied`, `RequestTimeTooSkewed`…), because that code is the whole of the diagnosis. + * Nothing else from the body is: an S3 error document quotes the canonical request and the + * string that was signed, which carry the object's key and therefore the scope. + * + * @param response Answer from the storage. + * @param operation What was attempted, for the message. + * @throws SyncTransportFailure When the storage did not accept the request. + */ + private suspend fun ensureStored( + response: HttpResponse, + operation: String, + ) { + if (response.status.isSuccess()) { + return + } + val status = response.status.value + val code = storageErrorCode(response) + val message = "the storage refused the $operation: $status ${code ?: response.status.description}" + if (operation == "download" && response.status == HttpStatusCode.NotFound) { + throw SyncTransportFailure.BlobGone(message) + } + throw SyncTransportFailure.ServerError(status, message) + } + + /** + * Reads the error code out of an S3-style error document, if the storage sent one. + * + * Bounded, because the body belongs to whoever answered — a proxy's HTML page is as likely as an + * S3 document — and restricted to the characters an S3 code is made of, so that nothing else from + * the body can reach a log through it. + * + * @param response Answer that was not a success. + * @return The code, or `null` when the body carries none. + */ + private suspend fun storageErrorCode(response: HttpResponse): String? { + val body = + runCatching { response.bodyAsChannel().readRemaining(ERROR_BODY_LIMIT).readString() } + .getOrNull() ?: return null + return STORAGE_ERROR_CODE.find(body)?.groupValues?.get(1) + } + /** * Runs a transfer, reporting anything that stopped it as a transport failure. * @@ -272,6 +328,25 @@ public class KtorBlobTransport( return total } + /** + * Sets the headers a ticket names, each exactly once. + * + * The body headers are respelled the way Ktor spells them. Every engine merges a request's + * headers with its body's and writes `Content-Type` and `Content-Length` itself, skipping the + * request's own copies only when the name matches its constant exactly. A signer that names the + * header in lower case — the AWS SDK does — therefore had it sent twice, and S3 joins duplicates + * into `image/jpeg,image/jpeg` before checking a signature computed over one value: every + * upload was refused with `SignatureDoesNotMatch`. The value is left untouched; it is signed. + * + * @param ticket Ticket whose headers to send. + */ + private fun HttpRequestBuilder.ticketHeaders(ticket: BlobTicket) { + ticket.headers.forEach { (name, value) -> + val spelling = ENGINE_HEADERS.firstOrNull { it.equals(name, ignoreCase = true) } ?: name + header(spelling, value) + } + } + private fun BlobTicketMethod.toHttpMethod(): HttpMethod = when (this) { BlobTicketMethod.GET -> HttpMethod.Get @@ -285,5 +360,14 @@ public class KtorBlobTransport( /** Large enough to keep the syscalls cheap, small enough that a phone does not feel it. */ const val CHUNK = 64L * 1024 + + /** More than any S3 error document needs; the rest of a refusal's body is never read. */ + const val ERROR_BODY_LIMIT = 4L * 1024 + + /** Headers the engine writes from the body, recognised only in exactly this spelling. */ + val ENGINE_HEADERS = listOf(HttpHeaders.ContentType, HttpHeaders.ContentLength) + + /** An S3 error code, and only the characters such a code is made of. */ + val STORAGE_ERROR_CODE = Regex("([A-Za-z0-9.]{1,64})") } } diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/persistence/blob/BlobRecordStore.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/persistence/blob/BlobRecordStore.kt index 8567afb..e5d0727 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/persistence/blob/BlobRecordStore.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/persistence/blob/BlobRecordStore.kt @@ -173,6 +173,40 @@ internal class BlobRecordStore( dao.recordFailure(scope.value, collection.value, blobId.value, error, nextRetryAt) } + /** + * Records that a file's bytes are not there yet and when to look again, without counting it. + * + * @param scope Scope of the file. + * @param collection Collection of the file. + * @param blobId Identifier of the file. + * @param error Why the file could not be moved yet. + * @param nextRetryAt When to look again. + */ + suspend fun recordWait( + scope: ScopeId, + collection: CollectionId, + blobId: BlobId, + error: String, + nextRetryAt: Long, + ) { + dao.recordWait(scope.value, collection.value, blobId.value, error, nextRetryAt) + } + + /** + * Forgets the attempts a file has used and any backoff it is serving, keeping its state. + * + * @param scope Scope of the file. + * @param collection Collection of the file. + * @param blobId Identifier of the file. + */ + suspend fun resetAttempts( + scope: ScopeId, + collection: CollectionId, + blobId: BlobId, + ) { + dao.resetAttempts(scope.value, collection.value, blobId.value) + } + /** * Records how much of a transfer has moved. * @@ -208,6 +242,20 @@ internal class BlobRecordStore( limit: Int, ): List = dao.waiting(scope.value, collection.value, state, now, limit).map { it.toRecord() } + /** + * Finds when the next file serving a backoff becomes due. + * + * @param scope Scope to read. + * @param collection Collection to read. + * @param now Current moment, in epoch milliseconds. + * @return The earliest moment one becomes due, or `null` when none is waiting on a backoff. + */ + suspend fun nextRetryAt( + scope: ScopeId, + collection: CollectionId, + now: Long, + ): Long? = dao.nextRetryAt(scope.value, collection.value, now) + /** * Counts files waiting for a transfer in one direction. * @@ -222,6 +270,19 @@ internal class BlobRecordStore( state: BlobTransferState, ): Flow = dao.observeCount(scope.value, collection.value, state) + /** + * Publishes every file some document still points at. + * + * @param scope Scope to read. + * @param collection Collection to read. + * @return Referenced files, in the order the library learned of them. + */ + fun observeReferenced( + scope: ScopeId, + collection: CollectionId, + ): Flow> = + dao.observeReferenced(scope.value, collection.value).map { rows -> rows.map { it.toRecord() } } + /** * Reads files no document points at any more. * diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/persistence/blob/SyncBlobDao.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/persistence/blob/SyncBlobDao.kt index 1f64960..06402c9 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/persistence/blob/SyncBlobDao.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/persistence/blob/SyncBlobDao.kt @@ -109,6 +109,51 @@ public interface SyncBlobDao { nextRetryAt: Long, ) + /** + * Records that the bytes are not there yet and when to look again, without counting it. + * + * A file the server holds as still arriving is waiting on another device, not failing on this + * one. Counting those looks as attempts let a device give up on a receipt a minute after it was + * attached, while its owner's other phone was still on the way to a network. + * + * @param scopeId Scope of the file. + * @param collectionId Collection of the file. + * @param blobId Identifier of the file. + * @param error Why the file could not be moved yet. + * @param nextRetryAt When to look again. + */ + @Query( + "UPDATE sync_blob SET last_error = :error, next_retry_at = :nextRetryAt, transferred = 0 " + + "WHERE scope_id = :scopeId AND collection_id = :collectionId AND blob_id = :blobId", + ) + public suspend fun recordWait( + scopeId: String, + collectionId: String, + blobId: Uuid, + error: String, + nextRetryAt: Long, + ) + + /** + * Forgets the attempts a file has used and any backoff it is serving, keeping its state. + * + * What a user asking to try again means for a transfer that has not been given up on: start now, + * and with the short delays a first attempt gets. + * + * @param scopeId Scope of the file. + * @param collectionId Collection of the file. + * @param blobId Identifier of the file. + */ + @Query( + "UPDATE sync_blob SET attempts = 0, next_retry_at = NULL, last_error = NULL " + + "WHERE scope_id = :scopeId AND collection_id = :collectionId AND blob_id = :blobId", + ) + public suspend fun resetAttempts( + scopeId: String, + collectionId: String, + blobId: Uuid, + ) + /** * Records whether this device is trying to hold a file's bytes. * @@ -231,6 +276,31 @@ public interface SyncBlobDao { limit: Int, ): List + /** + * Finds when the next file serving a backoff becomes due. + * + * The same files [waiting] would return, minus the condition that they are due already: what + * the worker sleeps until when nothing can move now, so a retry happens when its backoff says + * rather than whenever the next cycle happens to run. + * + * @param scopeId Scope to read. + * @param collectionId Collection to read. + * @param now Current moment; only later retries are considered. + * @return The earliest moment one becomes due, or `null` when none is waiting on a backoff. + */ + @Query( + "SELECT MIN(b.next_retry_at) FROM sync_blob b WHERE b.scope_id = :scopeId " + + "AND b.collection_id = :collectionId AND b.wanted = 1 AND b.next_retry_at > :now " + + "AND b.state IN ('LOCAL', 'UPLOADING', 'REMOTE', 'DOWNLOADING') " + + "AND EXISTS (SELECT 1 FROM sync_blob_ref r WHERE r.scope_id = b.scope_id " + + "AND r.collection_id = b.collection_id AND r.blob_id = b.blob_id)", + ) + public suspend fun nextRetryAt( + scopeId: String, + collectionId: String, + now: Long, + ): Long? + /** * Counts files in a given state that this device wants and some document still points at. * @@ -253,6 +323,26 @@ public interface SyncBlobDao { state: BlobTransferState, ): Flow + /** + * Publishes every file of a collection that some document still points at. + * + * What a screen listing a collection's files binds to. A file nothing references is on its way + * out and is not the owner's any more, so it is left out. + * + * @param scopeId Scope to read. + * @param collectionId Collection to read. + * @return Referenced files, in the order the library learned of them. + */ + @Query( + "SELECT b.* FROM sync_blob b WHERE b.scope_id = :scopeId AND b.collection_id = :collectionId " + + "AND EXISTS (SELECT 1 FROM sync_blob_ref r WHERE r.scope_id = b.scope_id " + + "AND r.collection_id = b.collection_id AND r.blob_id = b.blob_id) ORDER BY b.rowid", + ) + public fun observeReferenced( + scopeId: String, + collectionId: String, + ): Flow> + /** * Reads every file of a collection that no document points at any more. * diff --git a/client-sdk/src/commonTest/kotlin/dev/voir/reflector/sync/network/KtorBlobTransportTest.kt b/client-sdk/src/commonTest/kotlin/dev/voir/reflector/sync/network/KtorBlobTransportTest.kt index 75aa41e..7b2f881 100644 --- a/client-sdk/src/commonTest/kotlin/dev/voir/reflector/sync/network/KtorBlobTransportTest.kt +++ b/client-sdk/src/commonTest/kotlin/dev/voir/reflector/sync/network/KtorBlobTransportTest.kt @@ -10,6 +10,7 @@ import dev.voir.reflector.sync.protocol.blob.BlobContentType import dev.voir.reflector.sync.protocol.blob.BlobDescriptor import dev.voir.reflector.sync.protocol.blob.BlobTicket import dev.voir.reflector.sync.protocol.blob.BlobTicketMethod +import io.ktor.client.engine.mergeHeaders import io.ktor.client.engine.mock.MockEngine import io.ktor.client.engine.mock.MockRequestHandleScope import io.ktor.client.engine.mock.respond @@ -20,6 +21,7 @@ import io.ktor.http.HttpHeaders import io.ktor.http.HttpMethod import io.ktor.http.HttpStatusCode import io.ktor.http.headersOf +import io.ktor.utils.io.InternalAPI import io.ktor.utils.io.ByteReadChannel import kotlinx.coroutines.test.runTest import kotlinx.io.Buffer @@ -92,6 +94,37 @@ class KtorBlobTransportTest { assertNull(request.headers[HttpHeaders.Authorization]) } + /** + * The AWS SDK names signed headers in lower case. Every engine writes `Content-Type` from the + * body and skipped the request's copy only when spelled exactly like that, so the header went + * out twice and S3 refused every upload with `SignatureDoesNotMatch`. + */ + @OptIn(InternalAPI::class) + @Test + fun `a signed header spelled in lower case is sent once`() = + runTest { + val transport = transport { respond("", HttpStatusCode.OK) } + + transport.upload( + ticket(BlobTicketMethod.PUT).copy(headers = mapOf("content-type" to "image/jpeg")), + Buffer().apply { write(BYTES) }, + stat(), + ) + + val request = requests.single() + // What an engine actually writes, rather than the builder's case-insensitive view of it. + val sent = mutableListOf>() + mergeHeaders(request.headers, request.body) { name, value -> sent += name to value } + assertEquals( + listOf("image/jpeg"), + sent.filter { it.first.equals(HttpHeaders.ContentType, ignoreCase = true) }.map { it.second }, + ) + assertEquals( + listOf(BYTES.size.toString()), + sent.filter { it.first.equals(HttpHeaders.ContentLength, ignoreCase = true) }.map { it.second }, + ) + } + @Test fun `a request to the synchronisation server does carry them`() = runTest { @@ -157,7 +190,40 @@ class KtorBlobTransportTest { } } - private fun stat() = BlobStat(size = BYTES.size.toLong(), contentType = BlobContentType("image/jpeg")) + @Test + fun `a storage refusal names the storage's status and error code, never the scope`() = + runTest { + val transport = + transport { + respond( + "SignatureDoesNotMatchAWS4 user-1/ledger", + HttpStatusCode.Forbidden, + ) + } + + // A bucket's 403 is a refused signature or a missing permission on the host's side. Read + // with the synchronisation server's meaning it said the scope was revoked, which sent + // the diagnosis of a misconfigured bucket in the wrong direction. + val failure = + assertFailsWith { + transport.upload(ticket(BlobTicketMethod.PUT), Buffer().apply { write(BYTES) }, stat()) + } + + assertEquals(403, failure.statusCode) + assertEquals("the storage refused the upload: 403 SignatureDoesNotMatch", failure.message) + } + + @Test + fun `a download the storage has no object for is a gone file`() = + runTest { + val transport = transport { respond("NoSuchKey", HttpStatusCode.NotFound) } + + assertFailsWith { + transport.download(ticket(BlobTicketMethod.GET), Buffer()) + } + } + + private fun stat() =BlobStat(size = BYTES.size.toLong(), contentType = BlobContentType("image/jpeg")) private companion object { val BYTES = byteArrayOf(1, 2, 3, 4) diff --git a/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/SyncEngineTest.kt b/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/SyncEngineTest.kt index 3fe140f..4c372b1 100644 --- a/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/SyncEngineTest.kt +++ b/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/SyncEngineTest.kt @@ -308,7 +308,15 @@ class SyncEngineTest { markUpserted(wallet, walletId) } transport.onPush = { throw SyncTransportFailure.CollectionReset("purged") } - transport.onChanges = { throw SyncTransportFailure.CollectionReset("purged") } + // Only a device still reading the purged collection is told so; one that has rebuilt reads + // the new one like any other. A server that kept refusing would reset it on every cycle, + // and the test would read the store halfway through the next rebuild. + transport.onChanges = { cursor -> + if (cursor?.value?.startsWith(NEW_EPOCH_VALUE) != true) { + throw SyncTransportFailure.CollectionReset("purged") + } + ChangesPage(emptyList(), null, hasMore = false, epoch = CollectionEpoch(NEW_EPOCH_VALUE)) + } transport.onSnapshot = { snapshots++ SnapshotPage( @@ -321,7 +329,15 @@ class SyncEngineTest { } collection.requestSync() - awaitRecord("the collection being discarded") { it.event == SyncLogEvent.COLLECTION_RESET } + // The published state follows the database asynchronously, so right after the reset it + // can still read live with nothing queued from before it — the edit may even have reached + // the old collection before it was purged. Only a bootstrap finished after the reset says + // the new incarnation is in place. + awaitRecords("the collection being discarded and rebuilt") { + val events = logs.records.value.map { it.event } + val reset = events.indexOf(SyncLogEvent.COLLECTION_RESET) + reset >= 0 && events.subList(reset, events.size).contains(SyncLogEvent.BOOTSTRAP_FINISHED) + } collection.state.await("a live collection rebuilt from the new incarnation") { it.phase == SyncPhase.LIVE && it.pendingCount == 0 } diff --git a/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/blob/BlobWorkerTest.kt b/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/blob/BlobWorkerTest.kt index aa17410..be93753 100644 --- a/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/blob/BlobWorkerTest.kt +++ b/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/blob/BlobWorkerTest.kt @@ -141,8 +141,10 @@ class BlobWorkerTest { ) } - private suspend fun stateOf(blobId: BlobId) = - transactions.transaction { stores.blobs.find(scopeId, ledger, blobId)?.state } + private suspend fun stateOf(blobId: BlobId) = recordOf(blobId)?.state + + private suspend fun recordOf(blobId: BlobId) = + transactions.transaction { stores.blobs.find(scopeId, ledger, blobId) } private suspend fun attach( binding: BlobBinding = BlobBinding.DEFERRED, @@ -342,13 +344,114 @@ class BlobWorkerTest { // The ordinary state of an attachment whose record arrived ahead of its bytes. It is // answered by waiting, and must never be reported to the application as a failure. - await("the file being waited for") { - transactions.transaction { stores.blobs.find(scopeId, ledger, photo)?.attempts ?: 0 } > 0 + await("the file being waited for") { recordOf(photo)?.nextRetryAt != null } + val record = assertNotNull(recordOf(photo)) + assertEquals(BlobTransferState.REMOTE, record.state) + assertEquals("the server does not have the bytes yet", record.lastError) + // Not an attempt. Counting these let a device give a receipt up a minute after it was + // attached, while the phone that took it was still on its way to a network. + assertEquals(0, record.attempts) + assertTrue(files.failures.isEmpty()) + } + + @Test + fun `an upload that keeps failing is never given up on while its bytes are here`() = + runBlocking { + blobs.failWith = SyncTransportFailure.ServerError(403, "the storage refused the upload: 403 AccessDenied") + attach() + await("the first failure being recorded") { (recordOf(photo)?.attempts ?: 0) > 0 } + + // Brought to the attempt that used to be the last one, and made due at once rather than + // waiting out the backoff that got it there. + transactions.transaction { + repeat(3) { stores.blobs.recordFailure(scopeId, ledger, photo, "failed", nextRetryAt = 0) } } - assertEquals(BlobTransferState.REMOTE, stateOf(photo)) + collection().requestSync() + + await("the attempt that used to give up") { (recordOf(photo)?.attempts ?: 0) >= 5 } + // Still queued for another attempt. Every other device has had the record naming this + // file since it was registered, and giving up here would leave them waiting forever. + assertEquals(BlobTransferState.LOCAL, stateOf(photo)) assertTrue(files.failures.isEmpty()) } + @Test + fun `an upload refused after registering keeps counting its attempts`() = + runBlocking { + // Registered every time, refused by the storage every time — a misconfigured bucket. + // Each retry used to move the row into UPLOADING again, which forgot the attempts, so + // the backoff never grew and the file was sent again every second or two. + blobs.holdUploads = true + attach() + + // No cycle runs after the first here — there is no timer and nobody asks — so the + // second attempt also shows the worker waking itself when its backoff runs out. + await("a second failure being counted") { (recordOf(photo)?.attempts ?: 0) >= 2 } + assertEquals(BlobTransferState.UPLOADING, stateOf(photo)) + } + + @Test + fun `asking to try again restarts an upload that was given up on`() = + runBlocking { + blobs.failWith = SyncTransportFailure.Unreachable("the connection dropped") + attach() + await("the first failure being recorded") { (recordOf(photo)?.attempts ?: 0) > 0 } + transactions.transaction { + stores.blobs.setState(scopeId, ledger, photo, BlobTransferState.UNAVAILABLE) + } + blobs.failWith = null + + collection().retry(photo) + + await("the file reaching the server") { stateOf(photo) == BlobTransferState.UPLOADED } + } + + @Test + fun `an upload given up on earlier is sent again when the worker starts`() = + runBlocking { + blobs.failWith = SyncTransportFailure.Unreachable("the connection dropped") + attach() + await("the first failure being recorded") { (recordOf(photo)?.attempts ?: 0) > 0 } + // What an earlier version of the library left behind: an upload abandoned after a few + // attempts with its bytes still on the device, and nothing that would ever try it again. + workers.cancel() + transactions.transaction { + stores.blobs.setState(scopeId, ledger, photo, BlobTransferState.UNAVAILABLE) + } + blobs.failWith = null + + val restarted = CoroutineScope(Dispatchers.Default + SupervisorJob()) + try { + SyncEngine( + database = database, + transactions = transactions, + transport = transport, + adapters = mapOf(ledger to adapter), + coroutineScope = restarted, + blobStore = files, + blobTransport = blobs, + triggerSources = emptyList(), + ).scope(scopeId).collection(ledger).requestSync() + + await("the file reaching the server") { stateOf(photo) == BlobTransferState.UPLOADED } + } finally { + restarted.cancel() + } + } + + @Test + fun `a screen listing files sees every file a document names`() = + runBlocking { + attach() + await("the file reaching the server") { stateOf(photo) == BlobTransferState.UPLOADED } + + val listed = collection().blobs.first() + + assertEquals(setOf(photo), listed.keys) + assertEquals(BlobTransferState.UPLOADED, listed.getValue(photo).state) + assertEquals(0, listed.getValue(photo).attempts) + } + private fun changesNamingPhoto() = dev.voir.reflector.sync.protocol.changes.ChangesPage( batches = From 5b8a6c553e2003f274e5f8f1c6e8ca3184fd47e1 Mon Sep 17 00:00:00 2001 From: Gary Bezruchko Date: Wed, 7 Oct 2026 19:45:13 +0200 Subject: [PATCH 3/3] fix test name --- .../dev/voir/reflector/sync/network/KtorBlobTransportTest.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client-sdk/src/commonTest/kotlin/dev/voir/reflector/sync/network/KtorBlobTransportTest.kt b/client-sdk/src/commonTest/kotlin/dev/voir/reflector/sync/network/KtorBlobTransportTest.kt index 7b2f881..159e839 100644 --- a/client-sdk/src/commonTest/kotlin/dev/voir/reflector/sync/network/KtorBlobTransportTest.kt +++ b/client-sdk/src/commonTest/kotlin/dev/voir/reflector/sync/network/KtorBlobTransportTest.kt @@ -191,7 +191,7 @@ class KtorBlobTransportTest { } @Test - fun `a storage refusal names the storage's status and error code, never the scope`() = + fun `a storage refusal names the storage's status and error code never the scope`() = runTest { val transport = transport {