fix(sync): preserve upload checkpoint recovery evidence

changes/unreleased/113-upload-checkpoint-recovery.md

@@ -0,0 +1,7 @@
+category: fix
+issue: 113
+pull: 431
+platforms: desktop
+user-facing: yes
+
+Superseded resumable uploads now retain durable size, content hash, and publication state during cleanup, so an ambiguous server result cannot mistake an already published directory replacement for an abandoned stage.

ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopFileSyncRemoteTypeReplacement.kt

@@ -143,7 +143,9 @@ internal fun DesktopFileSyncRemoteTree.reconcilePublishedReplacement(
     val expectedBackupEtag = ownedReplacementBackupEtags[uploadId] ?: return null
     val destination = resolvePhysical(relativePath, shouldContinue) ?: return null
     if (destination.isDirectory) return null
-    if (expectedSizeBytes == null || expectedContentHash == null) return false
+    // Older checkpoints predate durable size/hash evidence. Let the caller fall back to the
+    // recorded stage ETag so it can restore the protected directory without trusting the file.
+    if (expectedSizeBytes == null || expectedContentHash == null) return null
     if (destination.entry.size != expectedSizeBytes) {
         return discardReplacementBackup(relativePath, uploadId, assembledStageEtag = null, shouldContinue)
     }

ui/src/desktopMain/kotlin/dev/obiente/nextcloudnative/app/DesktopFileSyncUploadExecution.kt

@@ -130,12 +130,20 @@ internal fun executeDesktopFileSyncUpload(
     val expectedDirectoryEtag = requireNotNull(expectedRemoteEtag)
     val transferPlan = nextcloudUploadTransferPlan(source.length())
     if (transferPlan is NextcloudUploadTransferPlan.Chunked) {
-        val recoveringPublication = checkpoint?.let {
-            it.commitInFlight && it.localRevision == exactLocal.revision && it.transferPlan == transferPlan
-        } == true
+        val matchingCheckpoint = checkpoint?.takeIf {
+            it.localRevision == exactLocal.revision && it.contentRevision == exactLocal.revision &&
+                it.contentHash == exactLocal.contentHash && it.transferPlan == transferPlan
+        }
+        if (checkpoint != null && matchingCheckpoint == null) {
+            check(
+                remote.resumableUploadRemote(shouldContinue, expectedDirectoryEtag)
+                    .discardCheckpointUpload(checkpoint, relativePath),
+            ) { "An unverified upload stage still requires recovery." }
+        }
+        val recoveringPublication = matchingCheckpoint?.commitInFlight == true
         if (!recoveringPublication) remote.requireDirectoryGeneration(relativePath, expectedDirectoryEtag)
         return resumeDesktopFileSyncUpload(
-            source, relativePath, exactLocal, expectedDirectoryEtag, checkpoint,
+            source, relativePath, exactLocal, expectedDirectoryEtag, matchingCheckpoint,
             persistCheckpoint, remote, shouldContinue,
             replacingDirectoryEtag = expectedDirectoryEtag,
         )

ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/DesktopFileSyncCleanupCancellationTest.kt

@@ -99,6 +99,7 @@ class DesktopFileSyncCleanupCancellationTest {
                         """.trimIndent(),
                     ).build(),
             )
+            server.enqueue(MockResponse.Builder().code(412).build())
             val directory = Files.createTempDirectory("desktop-sync-cleanup-block-").toFile()
             val localRoot = directory.resolve("local").apply { mkdirs() }
             val session = NextcloudSession(server.url("/").toString(), "alice", "secret")
@@ -134,7 +135,7 @@ class DesktopFileSyncCleanupCancellationTest {
 
                 assertIs<FileSyncCenterActionResult.Rejected>(result)
                 assertEquals(FileSyncRejectionScope.Preflight, result.scope)
-                assertEquals(2, server.requestCount)
+                assertEquals(3, server.requestCount)
                 assertEquals(listOf(cleanup), store.loadPair(pair.id).coordinator.pairs.single().pendingUploadCleanups)
             } finally {
                 directory.deleteRecursively()

ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/DesktopFileSyncReplacementPublicationTest.kt

@@ -1,8 +1,10 @@
 package dev.obiente.nextcloudnative.app
 
+import java.io.RandomAccessFile
 import java.nio.file.Files
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertTrue
 import okhttp3.OkHttpClient
 import okhttp3.Protocol
@@ -154,6 +156,46 @@ class DesktopFileSyncReplacementPublicationTest {
         assertTrue(requests.none { it.method == "DELETE" || it.method == "GET" })
     }
 
+    @Test
+    fun `legacy checkpoint restores its directory backup using the recorded stage etag`() {
+        val uploadId = "01234567-89ab-cdef-0123-456789abcdef"
+        val requests = mutableListOf<Request>()
+        val client = OkHttpClient.Builder().addInterceptor { chain ->
+            requests += chain.request()
+            when (chain.request().method) {
+                "PROPFIND" -> response(chain.request(), 207, publishedListing(uploadId, sizeBytes = 5))
+                "DELETE" -> response(chain.request(), if (".upload" in chain.request().url.encodedPath) 404 else 204)
+                "MOVE" -> response(chain.request(), 201)
+                else -> error("Legacy recovery must not ${chain.request().method} either generation")
+            }
+        }.build()
+        val tree = DesktopFileSyncRemoteTree(
+            NextcloudSession("https://cloud.example.test", "alice", "secret"),
+            "alice",
+            "Vault",
+            client,
+            ownedUploadIds = setOf(uploadId),
+            ownedUploadPaths = mapOf(uploadId to "archive.bin"),
+            ownedReplacementBackupEtags = mapOf(uploadId to "directory-etag"),
+        )
+
+        val cleaned = tree.resumableUploadRemote(shouldContinue = { true }).discardOwnedUpload(
+            uploadId = uploadId,
+            relativePath = "archive.bin",
+            assembledStageEtag = "published-etag",
+            expectedStageSizeBytes = null,
+            expectedStageContentHash = null,
+            publicationInFlight = true,
+        )
+
+        assertTrue(cleaned)
+        assertTrue(requests.any { it.method == "DELETE" && it.url.encodedPath.endsWith("/archive.bin") })
+        val restore = requests.single { it.method == "MOVE" }
+        assertTrue(restore.url.encodedPath.endsWith(".nextcloud-native-backup-$uploadId"))
+        assertTrue(restore.header("Destination").orEmpty().endsWith("/archive.bin"))
+        assertTrue(requests.none { it.method == "GET" })
+    }
+
     @Test
     fun `recovery scan traverses an owned backup at its physical path`() {
         val uploadId = "01234567-89ab-cdef-0123-456789abcdef"
@@ -185,6 +227,82 @@ class DesktopFileSyncReplacementPublicationTest {
         assertTrue(requestedPaths.none { it.endsWith("/archive.bin") })
     }
 
+    @Test
+    fun `superseded published replacement is reconciled before directory preflight`() {
+        val uploadId = "01234567-89ab-cdef-0123-456789abcdef"
+        val oldPayload = ByteArray(21 * 1024 * 1024) { 1 }
+        val oldHash = hashExactJvmFileSyncContent(oldPayload.inputStream(), oldPayload.size.toLong())
+        val requests = mutableListOf<Request>()
+        val client = OkHttpClient.Builder().addInterceptor { chain ->
+            requests += chain.request()
+            when (chain.request().method) {
+                "GET" -> response(chain.request(), 200, oldPayload)
+                "PROPFIND" -> response(chain.request(), 207, publishedListing(uploadId, oldPayload.size.toLong()))
+                "DELETE" -> response(chain.request(), 204)
+                else -> error("Superseded recovery must not ${chain.request().method} a new upload")
+            }
+        }.build()
+        val tree = DesktopFileSyncRemoteTree(
+            NextcloudSession("https://cloud.example.test", "alice", "secret"),
+            "alice",
+            "Vault",
+            client,
+            ownedUploadIds = setOf(uploadId),
+            ownedStageEtags = mapOf(uploadId to "stage-etag"),
+            ownedUploadPaths = mapOf(uploadId to "archive.bin"),
+            ownedReplacementBackupEtags = mapOf(uploadId to "directory-etag"),
+        )
+        val source = Files.createTempFile("nextcloud-sync-superseded-replacement", ".tmp").toFile()
+        RandomAccessFile(source, "rw").use { it.setLength(oldPayload.size.toLong()) }
+        val newHash = source.inputStream().buffered().use { input ->
+            hashExactJvmFileSyncContent(input, source.length())
+        }
+        val plan = nextcloudUploadTransferPlan(source.length()) as NextcloudUploadTransferPlan.Chunked
+        val checkpoint = newFileSyncUploadCheckpoint(
+            uploadId,
+            "local-1",
+            plan,
+            contentHash = oldHash,
+        ).copy(
+            uploadedChunks = plan.chunkCount,
+            commitInFlight = true,
+            assembledStageEtag = "stage-etag",
+        )
+        try {
+            assertFailsWith<IllegalArgumentException> {
+                executeDesktopFileSyncUpload(
+                    source = source,
+                    relativePath = "archive.bin",
+                    exactLocal = LocalSyncEntry(
+                        "archive.bin",
+                        SyncEntryKind.File,
+                        "local-2",
+                        source.length(),
+                        contentHash = newHash,
+                    ),
+                    expectedRemoteEtag = "directory-etag",
+                    checkpoint = checkpoint,
+                    replacingType = true,
+                    persistCheckpoint = {},
+                    retainCleanup = {},
+                    completeCleanup = {},
+                    remote = tree,
+                    shouldContinue = { true },
+                )
+            }
+
+            assertEquals(
+                listOf("DELETE", "PROPFIND", "GET", "PROPFIND", "PROPFIND", "DELETE", "PROPFIND"),
+                requests.map { it.method },
+            )
+            assertTrue(requests.none { it.method == "PUT" || it.method == "MOVE" })
+            assertTrue(requests[5].url.encodedPath.endsWith(".nextcloud-native-backup-$uploadId"))
+            assertEquals(requests[1].url.encodedPath, requests.last().url.encodedPath)
+        } finally {
+            assertTrue(source.delete())
+        }
+    }
+
     private fun stagedListing(uploadId: String?, sizeBytes: Long): String =
         """
         <d:multistatus xmlns:d="DAV:">

ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/JvmResumableNextcloudUploadTest.kt

@@ -130,6 +130,130 @@ class JvmResumableNextcloudUploadTest {
         }
     }
 
+    @Test
+    fun `superseded ambiguous publication is reconciled with its durable generation evidence`() {
+        val source = sparseFile(25L * 1024L * 1024L)
+        val plan = nextcloudUploadTransferPlan(source.length()) as NextcloudUploadTransferPlan.Chunked
+        val oldHash = "sha256:" + "11".repeat(32)
+        val checkpoint = newFileSyncUploadCheckpoint(
+            UPLOAD_ID,
+            "local-1",
+            plan,
+            contentRevision = "content-1",
+            contentHash = oldHash,
+        ).copy(
+            uploadedChunks = plan.chunkCount,
+            commitInFlight = true,
+            assembledStageEtag = "published-stage",
+        )
+        val remote = RecordingUploadRemote(collectionCreated = true)
+        val persisted = mutableListOf<FileSyncUploadCheckpoint>()
+        try {
+            jvmResumableNextcloudUpload(
+                source, "archive.bin", "local-2", "directory-etag", checkpoint,
+                newUploadId = { "fedcba98-7654-3210-fedc-ba9876543210" },
+                persistCheckpoint = persisted::add,
+                remote = remote,
+                contentRevision = "content-2",
+                contentHash = "sha256:" + "22".repeat(32),
+            )
+
+            assertEquals(
+                listOf(
+                    DiscardedUpload(
+                        assembledStageEtag = "published-stage",
+                        expectedStageSizeBytes = checkpoint.sizeBytes,
+                        expectedStageContentHash = oldHash,
+                        publicationInFlight = true,
+                    ),
+                ),
+                remote.discardedUploads,
+            )
+            assertEquals("fedcba98-7654-3210-fedc-ba9876543210", persisted.first().uploadId)
+        } finally {
+            source.delete()
+        }
+    }
+
+    @Test
+    fun `superseded ambiguous assembly is discarded only with its durable content evidence`() {
+        val source = sparseFile(25L * 1024L * 1024L)
+        val plan = nextcloudUploadTransferPlan(source.length()) as NextcloudUploadTransferPlan.Chunked
+        val oldHash = "sha256:" + "33".repeat(32)
+        val checkpoint = newFileSyncUploadCheckpoint(
+            UPLOAD_ID,
+            "local-1",
+            plan,
+            contentHash = oldHash,
+        ).copy(
+            uploadedChunks = plan.chunkCount,
+            commitInFlight = true,
+        )
+        val remote = RecordingUploadRemote(collectionCreated = true)
+        try {
+            jvmResumableNextcloudUpload(
+                source, "large.bin", "local-2", null, checkpoint,
+                newUploadId = { "fedcba98-7654-3210-fedc-ba9876543210" },
+                persistCheckpoint = {},
+                remote = remote,
+                contentHash = "sha256:" + "44".repeat(32),
+            )
+
+            assertEquals(
+                listOf(
+                    DiscardedUpload(
+                        assembledStageEtag = null,
+                        expectedStageSizeBytes = checkpoint.sizeBytes,
+                        expectedStageContentHash = oldHash,
+                        publicationInFlight = false,
+                    ),
+                ),
+                remote.discardedUploads,
+            )
+        } finally {
+            source.delete()
+        }
+    }
+
+    @Test
+    fun `failed superseded checkpoint cleanup blocks a replacement upload`() {
+        val source = sparseFile(25L * 1024L * 1024L)
+        val plan = nextcloudUploadTransferPlan(source.length()) as NextcloudUploadTransferPlan.Chunked
+        val checkpoint = newFileSyncUploadCheckpoint(
+            UPLOAD_ID,
+            "local-1",
+            plan,
+            contentHash = "sha256:" + "55".repeat(32),
+        ).copy(
+            uploadedChunks = plan.chunkCount,
+            commitInFlight = true,
+        )
+        val remote = RecordingUploadRemote(collectionCreated = true, cleanupComplete = false)
+        var allocatedReplacement = false
+        val persisted = mutableListOf<FileSyncUploadCheckpoint>()
+        try {
+            assertFailsWith<IllegalStateException> {
+                jvmResumableNextcloudUpload(
+                    source, "large.bin", "local-2", null, checkpoint,
+                    newUploadId = {
+                        allocatedReplacement = true
+                        "fedcba98-7654-3210-fedc-ba9876543210"
+                    },
+                    persistCheckpoint = persisted::add,
+                    remote = remote,
+                    contentHash = "sha256:" + "66".repeat(32),
+                )
+            }
+
+            assertFalse(allocatedReplacement)
+            assertTrue(persisted.isEmpty())
+            assertTrue(remote.uploadedChunkNumbers.isEmpty())
+            assertEquals(1, remote.discardedUploads.size)
+        } finally {
+            source.delete()
+        }
+    }
+
     @Test
     fun `expired collection resets progress before any bytes are skipped`() {
         val source = sparseFile(25L * 1024L * 1024L)
@@ -480,6 +604,7 @@ class JvmResumableNextcloudUploadTest {
     ) : JvmResumableNextcloudUploadRemote {
         val uploadedChunkNumbers = mutableListOf<Int>()
         val discardedStageEtags = mutableListOf<String?>()
+        val discardedUploads = mutableListOf<DiscardedUpload>()
         val finalizationEvents = mutableListOf<String>()
         var discardCount = 0
         var resolvePublishedCount = 0
@@ -589,10 +714,23 @@ class JvmResumableNextcloudUploadTest {
         ): Boolean {
             discardCount += 1
             discardedStageEtags += assembledStageEtag
+            discardedUploads += DiscardedUpload(
+                assembledStageEtag,
+                expectedStageSizeBytes,
+                expectedStageContentHash,
+                publicationInFlight,
+            )
             return cleanupComplete
         }
     }
 
+    private data class DiscardedUpload(
+        val assembledStageEtag: String?,
+        val expectedStageSizeBytes: Long?,
+        val expectedStageContentHash: String?,
+        val publicationInFlight: Boolean,
+    )
+
     private companion object {
         const val UPLOAD_ID = "01234567-89ab-cdef-0123-456789abcdef"
     }

ui/src/desktopTest/kotlin/dev/obiente/nextcloudnative/app/JvmSupportIntakeTest.kt

@@ -2219,7 +2219,9 @@ class JvmSupportIntakeTest {
             val submission = launch(Dispatchers.Default) {
                 fixture.intake.submit("A refresh failed.", "nightly", emptyList())
             }
-            val upload = requireNotNull(fixture.server.takeRequest(2, TimeUnit.SECONDS))
+            val upload = requireNotNull(
+                fixture.server.takeRequest(WINDOWS_REQUEST_START_TIMEOUT_SECONDS, TimeUnit.SECONDS),
+            )
             assertTrue(fixture.intake.cancel())
             submission.join()