fix: roll back consumed blobs and quota on commit failure

Old rollbackMoves() only unlinked moved blob files; it did not release reservations or remove the upload sessions whose payloads had been consumed. A commit that failed after consumeAndMoveCompletedBlob (e.g. module.json rename throws) leaked reservedBytes and orphan sessions.

commitModule now tracks each consumed blob in a ConsumedBlob struct (meta, live blob path, session dir) and exposes an explicit commitPointReached flag set right after the module.json.tmp -> module.json rename. Before that point, rollbackConsumedBlobs() does the full reverse: delete payload, removeConsumedSession, releaseReservation. After the commit point, even an access.json write failure is logged-and-swallowed - the module is committed and must not turn into a quota rollback.

UploadSessionRepo.ConsumeResult.Ready now carries sessionDir so callers don't re-resolve it. New removeConsumedSession() deletes a session whose payload was already moved out, without touching quota (the caller owns the reserved -> used transition).

Adds a BlobQuotaBoundaryFlowTest case that corrupts module.json.tmp (creates a directory with that name so rename fails), asserts 500 + reservedBytes back to 0 + session removed + a fresh session can be created.

Author: d4rken

src/main/kotlin/eu/darken/octi/server/module/ModuleLifecycleService.kt

@@ -52,6 +52,12 @@ class ModuleLifecycleService @Inject constructor(
         data class EtagMismatch(val currentEtag: String) : DeleteBlobResult
     }
 
+    private data class ConsumedBlob(
+        val meta: UploadSessionMeta,
+        val liveBlobFile: Path,
+        val sessionDir: Path,
+    )
+
     /**
      * Legacy POST write — under device lock, loads old meta, rejects if blob-backed,
      * pre-checks the document quota delta, writes payload, settles quota, aborts
@@ -235,6 +241,24 @@ class ModuleLifecycleService @Inject constructor(
                 return@withLock CommitResult.QuotaExceeded to emptyList<Path>()
             }
             var rollbackDocDelta = docDelta > 0
+            var commitPointReached = false
+            val consumedBlobs = mutableListOf<ConsumedBlob>()
+
+            fun rollbackConsumedBlobs() {
+                if (consumedBlobs.isEmpty()) return
+                consumedBlobs.asReversed().forEach { consumed ->
+                    try {
+                        consumed.liveBlobFile.parent?.deleteRecursively()
+                    } catch (e: Exception) {
+                        log(TAG, WARN) { "commitModule: failed to rollback consumed blob ${consumed.liveBlobFile}: ${e.message}" }
+                    }
+                    sessionRepo.removeConsumedSession(consumed.meta.sessionId, consumed.sessionDir)
+                    if (consumed.meta.expectedSizeBytes > 0) {
+                        storageTracker.releaseReservation(caller.accountId, consumed.meta.expectedSizeBytes)
+                    }
+                }
+                consumedBlobs.clear()
+            }
 
             try {
                 val modulePath = moduleRepo.resolveModulePath(target, moduleId)
@@ -265,60 +289,54 @@ class ModuleLifecycleService @Inject constructor(
 
                 // Pass 2 — move staged blobs into live storage. peek+consume is TOCTOU
                 // (GC can terminate a peeked session before consume runs); on any failure
-                // here, roll back already-moved blobs by deleting their storageKey dirs.
+                // before module.json becomes the commit point, rollbackConsumedBlobs()
+                // deletes already-moved payloads and releases their reservations.
                 val newBlobRefs = mutableListOf<BlobRef>()
-                val movedDestPaths = mutableListOf<Path>()
-                fun rollbackMoves() {
-                    movedDestPaths.forEach { dest ->
-                        runCatching { dest.parent?.deleteRecursively() }
+                for (blobId in blobRefIds) {
+                    val existingRef = existingMeta?.blobRefs?.find { it.blobId == blobId }
+                    if (existingRef != null) {
+                        newBlobRefs.add(existingRef)
+                        continue
                     }
-                }
-                try {
-                    for (blobId in blobRefIds) {
-                        val existingRef = existingMeta?.blobRefs?.find { it.blobId == blobId }
-                        if (existingRef != null) {
-                            newBlobRefs.add(existingRef)
-                            continue
-                        }
-                        var capturedDest: Path? = null
-                        val consumed = sessionRepo.consumeAndMoveCompletedBlob(
-                            blobId = blobId,
-                            accountId = caller.accountId,
-                            deviceId = target.id,
-                            moduleId = moduleId,
-                        ) { meta ->
-                            val prefix = meta.storageKey.take(4)
-                            val destPath = modulePath.resolve("blobs").resolve(prefix).resolve(meta.storageKey).resolve("payload.blob")
-                            capturedDest = destPath
-                            destPath
-                        }
-                        val sessionMeta = when (consumed) {
-                            is UploadSessionRepo.ConsumeResult.Ready -> {
-                                capturedDest?.let { movedDestPaths.add(it) }
-                                consumed.meta
-                            }
-                            UploadSessionRepo.ConsumeResult.SessionNotFound -> {
-                                rollbackMoves()
-                                return@withLock CommitResult.BadRequest("Referenced blobId not found: $blobId") to emptyList<Path>()
-                            }
-                            UploadSessionRepo.ConsumeResult.PayloadMissing -> {
-                                rollbackMoves()
-                                return@withLock CommitResult.BadRequest("Staged blob payload missing for $blobId") to emptyList<Path>()
-                            }
-                        }
-                        newBlobRefs.add(
-                            BlobRef(
-                                blobId = sessionMeta.blobId,
-                                storageKey = sessionMeta.storageKey,
-                                sizeBytes = sessionMeta.expectedSizeBytes,
-                                hashAlgorithm = sessionMeta.hashAlgorithm,
-                                hashHex = sessionMeta.hashHex,
+                    var capturedDest: Path? = null
+                    val consumed = sessionRepo.consumeAndMoveCompletedBlob(
+                        blobId = blobId,
+                        accountId = caller.accountId,
+                        deviceId = target.id,
+                        moduleId = moduleId,
+                    ) { meta ->
+                        val prefix = meta.storageKey.take(4)
+                        val destPath = modulePath.resolve("blobs").resolve(prefix).resolve(meta.storageKey).resolve("payload.blob")
+                        capturedDest = destPath
+                        destPath
+                    }
+                    val sessionMeta = when (consumed) {
+                        is UploadSessionRepo.ConsumeResult.Ready -> {
+                            val liveBlobFile = capturedDest
+                                ?: throw IllegalStateException("Consumed blob destination was not captured")
+                            consumedBlobs.add(
+                                ConsumedBlob(
+                                    meta = consumed.meta,
+                                    liveBlobFile = liveBlobFile,
+                                    sessionDir = consumed.sessionDir,
+                                )
                             )
-                        )
+                            consumed.meta
+                        }
+                        UploadSessionRepo.ConsumeResult.SessionNotFound ->
+                            return@withLock CommitResult.BadRequest("Referenced blobId not found: $blobId") to emptyList<Path>()
+                        UploadSessionRepo.ConsumeResult.PayloadMissing ->
+                            return@withLock CommitResult.BadRequest("Staged blob payload missing for $blobId") to emptyList<Path>()
                     }
-                } catch (e: Exception) {
-                    rollbackMoves()
-                    throw e
+                    newBlobRefs.add(
+                        BlobRef(
+                            blobId = sessionMeta.blobId,
+                            storageKey = sessionMeta.storageKey,
+                            sizeBytes = sessionMeta.expectedSizeBytes,
+                            hashAlgorithm = sessionMeta.hashAlgorithm,
+                            hashHex = sessionMeta.hashHex,
+                        )
+                    )
                 }
 
                 // Write payload.blob first, then module.json as commit point
@@ -343,25 +361,29 @@ class ModuleLifecycleService @Inject constructor(
                 val tempMeta = modulePath.resolve("module.json.tmp")
                 tempMeta.writeText(json.encodeToString(meta))
                 tempMeta.moveTo(metaFile, overwrite = true)
+                commitPointReached = true
+                rollbackDocDelta = false
 
-                // Update access metadata
-                val accessFile = modulePath.resolve("access.json")
-                val tempAccess = modulePath.resolve("access.json.tmp")
-                tempAccess.writeText(json.encodeToString(AccessMeta(lastAccessedAt = now)))
-                tempAccess.moveTo(accessFile, overwrite = true)
+                // Update access metadata. module.json is the commit point; an access
+                // write failure must not turn a committed module into a quota rollback.
+                try {
+                    val accessFile = modulePath.resolve("access.json")
+                    val tempAccess = modulePath.resolve("access.json.tmp")
+                    tempAccess.writeText(json.encodeToString(AccessMeta(lastAccessedAt = now)))
+                    tempAccess.moveTo(accessFile, overwrite = true)
+                } catch (e: Exception) {
+                    log(TAG, WARN) { "commitModule: failed to persist access metadata for $moduleId: ${e.message}" }
+                }
 
-                // Clean up committed sessions
-                for (ref in newBlobRefs) {
-                    if (existingMeta?.blobRefs?.any { it.blobId == ref.blobId } != true) {
-                        sessionRepo.removeCommittedSessionByBlobId(ref.blobId)
-                    }
+                // Clean up sessions whose payloads were consumed by this commit. Quota is
+                // settled below; this deletion deliberately does not release reservations.
+                for (consumed in consumedBlobs) {
+                    sessionRepo.removeConsumedSession(consumed.meta.sessionId, consumed.sessionDir)
                 }
 
                 // Quota update for blobs and shrinking documents. Positive doc delta was
                 // already applied by tryAdjustUsed above.
-                val newReferencedBytes = newBlobRefs.filter { ref ->
-                    existingMeta?.blobRefs?.any { it.blobId == ref.blobId } != true
-                }.sumOf { it.sizeBytes }
+                val newReferencedBytes = consumedBlobs.sumOf { it.meta.expectedSizeBytes }
                 val orphanedBlobRefs = existingMeta?.blobRefs
                     ?.filter { old -> newBlobRefs.none { it.blobId == old.blobId } }
                     ?: emptyList()
@@ -373,7 +395,6 @@ class ModuleLifecycleService @Inject constructor(
                 if (docDelta < 0) {
                     storageTracker.adjustUsed(caller.accountId, docDelta)
                 }
-                rollbackDocDelta = false
 
                 // Collect orphaned blob paths for async deletion outside the lock —
                 // deleting here would block every concurrent read/write for large orphans.
@@ -385,6 +406,9 @@ class ModuleLifecycleService @Inject constructor(
                 log(TAG) { "commitModule(${caller.id.shortId()}): $moduleId committed, etag=$newEtag" }
                 CommitResult.Success(newEtag) to orphansToDelete
             } finally {
+                if (!commitPointReached) {
+                    rollbackConsumedBlobs()
+                }
                 if (rollbackDocDelta) {
                     storageTracker.adjustUsed(caller.accountId, -docDelta)
                 }

src/main/kotlin/eu/darken/octi/server/module/UploadSessionRepo.kt

@@ -451,7 +451,7 @@ class UploadSessionRepo @Inject constructor(
     // region Lookups (scoped)
 
     sealed interface ConsumeResult {
-        data class Ready(val meta: UploadSessionMeta) : ConsumeResult
+        data class Ready(val meta: UploadSessionMeta, val sessionDir: Path) : ConsumeResult
         data object SessionNotFound : ConsumeResult
         data object PayloadMissing : ConsumeResult
     }
@@ -501,7 +501,7 @@ class UploadSessionRepo @Inject constructor(
             // cleans it up after the module.json rename. terminateSessionLocked sees the empty
             // session dir (staged file is gone) and skips releaseReservation — commitReservation
             // owns the reserved→used transition.
-            ConsumeResult.Ready(meta = current.meta)
+            ConsumeResult.Ready(meta = current.meta, sessionDir = current.sessionDir)
         }
     }
 
@@ -552,6 +552,21 @@ class UploadSessionRepo @Inject constructor(
         }
     }
 
+    /**
+     * Drops a session whose staged payload has already been consumed by commit logic.
+     * Does not release quota; the caller owns the reserved -> used or reserved -> free
+     * transition after moving the staged payload out of the session directory.
+     */
+    fun removeConsumedSession(sessionId: String, sessionDir: Path) {
+        sessions.remove(sessionId)
+        try {
+            sessionDir.deleteRecursively()
+        } catch (e: Exception) {
+            log(TAG, WARN) { "removeConsumedSession: failed to clean up $sessionId: ${e.message}" }
+        }
+        cleanupEmptyParentsAfterSession(sessionDir)
+    }
+
     /**
      * Checks whether a module has any active non-expired sessions (for GC protection).
      */

src/test/kotlin/eu/darken/octi/server/module/BlobQuotaBoundaryFlowTest.kt

@@ -8,6 +8,7 @@ import io.ktor.http.*
 import kotlinx.serialization.Serializable
 import org.junit.jupiter.api.Test
 import java.util.*
+import kotlin.io.path.createDirectories
 
 private fun base64Encode(data: ByteArray): String = Base64.getEncoder().encodeToString(data)
 
@@ -26,6 +27,39 @@ class BlobQuotaBoundaryFlowTest : TestRunner() {
         val sessionId: String = "",
     )
 
+    private suspend fun TestEnvironment.createFinalizedSession(
+        creds: Credentials,
+        moduleId: String,
+        payload: ByteArray,
+    ): SessionInfo {
+        val hash = payload.sha256Hex()
+        val session = http.post("/v1/module/$moduleId/blob-sessions") {
+            url { parameters.append("device-id", creds.deviceId.toString()) }
+            addCredentials(creds)
+            contentType(ContentType.Application.Json)
+            setBody("""{"sizeBytes": ${payload.size}, "hashAlgorithm": "sha256", "hashHex": "$hash"}""")
+        }.body<SessionInfo>()
+
+        if (payload.isNotEmpty()) {
+            http.patch("/v1/module/$moduleId/blob-sessions/${session.sessionId}") {
+                url { parameters.append("device-id", creds.deviceId.toString()) }
+                addCredentials(creds)
+                header("Upload-Offset", "0")
+                contentType(ContentType.Application.OctetStream)
+                setBody(payload)
+            }.status shouldBe HttpStatusCode.NoContent
+        }
+
+        http.post("/v1/module/$moduleId/blob-sessions/${session.sessionId}/finalize") {
+            url { parameters.append("device-id", creds.deviceId.toString()) }
+            addCredentials(creds)
+            contentType(ContentType.Application.Json)
+            setBody("{}")
+        }.status shouldBe HttpStatusCode.OK
+
+        return session
+    }
+
     @Test
     fun `session at exact quota succeeds, one more byte returns 507`() {
         runTest2(appConfig = baseConfig.copy(accountQuotaBytes = 1024, maxBlobBytes = 10_000)) {
@@ -76,6 +110,40 @@ class BlobQuotaBoundaryFlowTest : TestRunner() {
         }
     }
 
+    @Test
+    fun `failed PUT after consuming blob releases reservation`() {
+        runTest2(appConfig = baseConfig.copy(accountQuotaBytes = 1024, maxBlobBytes = 10_000)) {
+            val creds = createDevice()
+            val accountId = UUID.fromString(creds.account)
+            val failingModuleId = "eu.darken.octi.quota.failedput"
+            val payload = ByteArray(1024) { 7 }
+            val session = createFinalizedSession(creds, failingModuleId, payload)
+
+            component.storageTracker().getUsage(accountId).reservedBytes shouldBe 1024
+
+            val moduleDir = BlobFixtures.moduleDir(config.dataPath, accountId, creds.deviceId, failingModuleId)
+            moduleDir.resolve("module.json.tmp").createDirectories()
+
+            http.put("/v1/module/$failingModuleId") {
+                url { parameters.append("device-id", creds.deviceId.toString()) }
+                addCredentials(creds)
+                header("If-None-Match", "*")
+                contentType(ContentType.Application.Json)
+                setBody("""{"documentBase64": "${base64Encode(ByteArray(0))}", "blobRefs": [{"blobId": "${session.blobId}"}]}""")
+            }.status shouldBe HttpStatusCode.InternalServerError
+
+            component.storageTracker().getUsage(accountId).reservedBytes shouldBe 0
+            component.sessionRepo().getSession(session.sessionId, accountId, creds.deviceId, failingModuleId) shouldBe null
+
+            http.post("/v1/module/eu.darken.octi.quota.retry/blob-sessions") {
+                url { parameters.append("device-id", creds.deviceId.toString()) }
+                addCredentials(creds)
+                contentType(ContentType.Application.Json)
+                setBody("""{"sizeBytes": 1024}""")
+            }.status shouldBe HttpStatusCode.Created
+        }
+    }
+
     // Documents current behaviour: zero-byte sessions skip the tryReserve call entirely (the
     // `sizeBytes > 0` branch in BlobRoute.createSession). They can be created even when quota
     // is fully reserved. If production later gates zero-byte sessions for consistency, update