Release v0.1.3

VERSION

@@ -1 +1 @@
-0.1.2
+0.1.3

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<BlobSyncState?>
 
+    /**
+     * 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<Map<BlobId, BlobSyncState>>
+
     /**
      * 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.
      *

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,11 +34,15 @@ 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,
     public val wanted: Boolean,
     public val size: Long?,
     public val transferred: Long,
     public val lastError: String?,
+    public val attempts: Int,
 )

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.
      *

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.
      *

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<BlobSyncState?> =
-        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<Map<BlobId, BlobSyncState>> =
+        stores.blobs.observeReferenced(scope, collection).map { records ->
+            records.associate { record -> record.blobId to record.toSyncState() }
         }
 
     override val conflicts: Flow<List<Conflict>> =
@@ -120,6 +116,10 @@ internal class DefaultCollectionHandle(
         worker.evictBlob(id)
     }
 
+    override suspend fun retry(id: BlobId) {
+        worker.retryBlob(id)
+    }
+
     override suspend fun <R> 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,

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.
      *