Prevent paste pull cursor gaps

app/src/commonMain/kotlin/com/crosspaste/paste/PasteReleaseService.kt

@@ -17,6 +17,7 @@ import com.crosspaste.path.UserDataPathProvider
 import com.crosspaste.presist.FilesIndex
 import com.crosspaste.presist.buildFilesIndexForReceive
 import com.crosspaste.sync.FilePullService
+import com.crosspaste.sync.PastePullCursorManager
 import com.crosspaste.task.TaskBuilder
 import com.crosspaste.task.TaskSubmitter
 import com.crosspaste.utils.getFileUtils
@@ -44,6 +45,7 @@ class PasteReleaseService(
     private val pasteDao: PasteDao,
     private val pasteItemReader: PasteItemReader,
     private val pasteProcessPlugins: List<PasteProcessPlugin>,
+    private val pastePullCursorManager: PastePullCursorManager,
     private val searchContentService: SearchContentService,
     private val syncRuntimeInfoDao: SyncRuntimeInfoDao,
     private val taskSubmitter: TaskSubmitter,
@@ -230,9 +232,9 @@ class PasteReleaseService(
             "Discard oversized remote non-file paste from ${pasteData.appInstanceId}: " +
                 "size=${pasteData.size}, limit=$maxSize"
         }
-        pasteDao.upsertPastePullCursorMaxCreateTime(
+        pastePullCursorManager.persistDiscardedMaxCreateTime(
             appInstanceId = pasteData.appInstanceId,
-            maxCreateTime = pasteData.createTime,
+            createTime = pasteData.createTime,
         )
         val deviceName =
             syncRuntimeInfoDao
@@ -291,6 +293,20 @@ class PasteReleaseService(
         }
     }
 
+    suspend fun releaseRemotePasteDataList(
+        pasteDataList: List<PasteData>,
+        releaseLast: suspend (PasteData) -> Result<Unit?>,
+    ): Result<Unit?> {
+        if (pasteDataList.isEmpty()) return Result.success(null)
+
+        for (index in 0 until pasteDataList.lastIndex) {
+            releaseRemotePasteData(pasteDataList[index]) { _ -> }
+                .onFailure { return Result.failure(it) }
+        }
+
+        return releaseLast(pasteDataList.last())
+    }
+
     suspend fun releaseRemotePasteDataWithFile(
         id: Long,
         tryWritePasteboard: (PasteData) -> Unit,

app/src/commonMain/kotlin/com/crosspaste/sync/PastePullCursorManager.kt

@@ -0,0 +1,97 @@
+package com.crosspaste.sync
+
+import com.crosspaste.db.paste.PasteDao
+import com.crosspaste.db.sync.SyncRuntimeInfoDao
+import com.crosspaste.utils.StripedMutex
+import io.github.oshai.kotlinlogging.KotlinLogging
+import io.ktor.util.collections.ConcurrentMap
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+
+class PastePullCursorManager(
+    private val pasteDao: PasteDao,
+    private val syncRuntimeInfoDao: SyncRuntimeInfoDao,
+) {
+
+    private val logger = KotlinLogging.logger {}
+
+    private val deviceMaxCreateTime: MutableMap<String, Long> = ConcurrentMap()
+    private val deviceMutex = StripedMutex()
+    private val initMutex = Mutex()
+
+    suspend fun init() {
+        initMutex.withLock {
+            val storedPasteMaxCreateTimes = pasteDao.getMaxCreateTimeByRemoteAppInstanceId()
+            val persistedPullCursorMaxCreateTimes =
+                pasteDao.getPastePullCursorMaxCreateTimes()
+            val appInstanceIds =
+                storedPasteMaxCreateTimes.keys + persistedPullCursorMaxCreateTimes.keys
+            val maxCreateTimeMap =
+                appInstanceIds.associateWith { appInstanceId ->
+                    listOfNotNull(
+                        storedPasteMaxCreateTimes[appInstanceId],
+                        persistedPullCursorMaxCreateTimes[appInstanceId],
+                    ).max()
+                }
+
+            deviceMaxCreateTime.putAll(maxCreateTimeMap)
+            maxCreateTimeMap.forEach { (appInstanceId, maxCreateTime) ->
+                logger.debug { "Initialized sync state for $appInstanceId: maxCreateTime=$maxCreateTime" }
+            }
+            logger.info { "Paste pull cursor initialized with ${deviceMaxCreateTime.size} device(s)" }
+        }
+    }
+
+    fun getMaxCreateTime(appInstanceId: String): Long? = deviceMaxCreateTime[appInstanceId]
+
+    suspend fun updateMaxCreateTime(
+        appInstanceId: String,
+        createTime: Long,
+    ) {
+        deviceMutex.withLock(appInstanceId) {
+            updateMaxCreateTimeUnsafe(appInstanceId, createTime)
+        }
+    }
+
+    suspend fun persistDiscardedMaxCreateTime(
+        appInstanceId: String,
+        createTime: Long,
+    ): Boolean =
+        deviceMutex.withLock(appInstanceId) {
+            if (syncRuntimeInfoDao.getSyncRuntimeInfo(appInstanceId) == null) {
+                logger.debug { "Skip persisting paste pull cursor for removed device $appInstanceId" }
+                return@withLock false
+            }
+
+            pasteDao.upsertPastePullCursorMaxCreateTime(
+                appInstanceId = appInstanceId,
+                maxCreateTime = createTime,
+            )
+            updateMaxCreateTimeUnsafe(appInstanceId, createTime)
+            true
+        }
+
+    suspend fun removeDevice(appInstanceId: String) {
+        deviceMutex.withLock(appInstanceId) {
+            pasteDao.deletePastePullCursor(appInstanceId)
+            val storedPasteMaxCreateTime =
+                pasteDao.getMaxCreateTimeByRemoteAppInstanceId()[appInstanceId]
+            if (storedPasteMaxCreateTime == null) {
+                deviceMaxCreateTime.remove(appInstanceId)
+            } else {
+                deviceMaxCreateTime[appInstanceId] = storedPasteMaxCreateTime
+            }
+            logger.debug { "Removed paste pull cursor for $appInstanceId" }
+        }
+    }
+
+    private fun updateMaxCreateTimeUnsafe(
+        appInstanceId: String,
+        createTime: Long,
+    ) {
+        val current = deviceMaxCreateTime[appInstanceId]
+        if (current == null || createTime > current) {
+            deviceMaxCreateTime[appInstanceId] = createTime
+        }
+    }
+}

app/src/commonMain/kotlin/com/crosspaste/sync/PastePullService.kt

@@ -1,6 +1,5 @@
 package com.crosspaste.sync
 
-import com.crosspaste.db.paste.PasteDao
 import com.crosspaste.db.sync.SyncState
 import com.crosspaste.net.clientapi.PullClientApi
 import com.crosspaste.net.clientapi.SuccessResult
@@ -9,56 +8,27 @@ import com.crosspaste.paste.PasteboardService
 import com.crosspaste.utils.HostAndPort
 import com.crosspaste.utils.buildUrl
 import io.github.oshai.kotlinlogging.KotlinLogging
-import io.ktor.util.collections.ConcurrentMap
-import kotlinx.coroutines.sync.Mutex
-import kotlinx.coroutines.sync.withLock
 
 class PastePullService(
-    private val pasteDao: PasteDao,
+    private val pastePullCursorManager: PastePullCursorManager,
     private val pasteboardService: PasteboardService,
     private val pullClientApi: PullClientApi,
     private val syncManager: SyncManager,
 ) {
 
     private val logger = KotlinLogging.logger {}
 
-    private val deviceMaxCreateTime: MutableMap<String, Long> = ConcurrentMap()
-    private val mutex = Mutex()
-
     suspend fun init() {
-        val storedPasteMaxCreateTimes = pasteDao.getMaxCreateTimeByRemoteAppInstanceId()
-        val persistedPullCursorMaxCreateTimes = pasteDao.getPastePullCursorMaxCreateTimes()
-        val maxCreateTimeMap =
-            (storedPasteMaxCreateTimes.keys + persistedPullCursorMaxCreateTimes.keys).associateWith { appInstanceId ->
-                listOfNotNull(
-                    storedPasteMaxCreateTimes[appInstanceId],
-                    persistedPullCursorMaxCreateTimes[appInstanceId],
-                ).max()
-            }
-        deviceMaxCreateTime.putAll(maxCreateTimeMap)
-        maxCreateTimeMap.forEach { (appInstanceId, maxCreateTime) ->
-            logger.debug { "Initialized sync state for $appInstanceId: maxCreateTime=$maxCreateTime" }
-        }
-        logger.info { "PastePullService initialized with ${deviceMaxCreateTime.size} device(s)" }
+        pastePullCursorManager.init()
     }
 
-    fun getMaxCreateTime(appInstanceId: String): Long? = deviceMaxCreateTime[appInstanceId]
+    fun getMaxCreateTime(appInstanceId: String): Long? = pastePullCursorManager.getMaxCreateTime(appInstanceId)
 
     suspend fun updateMaxCreateTime(
         appInstanceId: String,
         createTime: Long,
     ) {
-        mutex.withLock {
-            val current = deviceMaxCreateTime[appInstanceId]
-            if (current == null || createTime > current) {
-                deviceMaxCreateTime[appInstanceId] = createTime
-            }
-        }
-    }
-
-    fun removeDevice(appInstanceId: String) {
-        deviceMaxCreateTime.remove(appInstanceId)
-        logger.debug { "Removed sync state for $appInstanceId" }
+        pastePullCursorManager.updateMaxCreateTime(appInstanceId, createTime)
     }
 
     suspend fun pullAllDevices(limit: Long = 10L) {
@@ -84,9 +54,13 @@ class PastePullService(
 
         allPasteData.sortBy { it.createTime }
 
-        pasteboardService.tryWriteRemotePasteboardList(allPasteData)
-
-        logger.info { "Pulled and wrote ${allPasteData.size} paste(s) from all devices" }
+        pasteboardService
+            .tryWriteRemotePasteboardList(allPasteData)
+            .onSuccess {
+                logger.info { "Pulled and wrote ${allPasteData.size} paste(s) from all devices" }
+            }.onFailure { e ->
+                logger.warn(e) { "Stopped writing pulled pastes after a release failure" }
+            }
     }
 
     suspend fun pullBatch(

app/src/commonMain/kotlin/com/crosspaste/sync/SyncDeviceManager.kt

@@ -1,6 +1,5 @@
 package com.crosspaste.sync
 
-import com.crosspaste.db.paste.PasteDao
 import com.crosspaste.db.sync.SyncRuntimeInfo
 import com.crosspaste.db.sync.SyncRuntimeInfoDao
 import com.crosspaste.db.sync.SyncState
@@ -16,7 +15,7 @@ import com.crosspaste.utils.buildUrl
 import io.github.oshai.kotlinlogging.KotlinLogging
 
 class SyncDeviceManager(
-    private val pasteDao: PasteDao,
+    private val pastePullCursorManager: PastePullCursorManager,
     private val pendingExchangeLedger: PendingExchangeLedger,
     private val secureStore: SecureStore,
     private val syncClientApi: SyncClientApi,
@@ -199,7 +198,7 @@ class SyncDeviceManager(
         wsSessionManager.closeSession(syncRuntimeInfo.appInstanceId)
         secureStore.deleteCryptPublicKey(syncRuntimeInfo.appInstanceId)
         syncRuntimeInfoDao.deleteSyncRuntimeInfo(syncRuntimeInfo.appInstanceId)
-        pasteDao.deletePastePullCursor(syncRuntimeInfo.appInstanceId)
+        pastePullCursorManager.removeDevice(syncRuntimeInfo.appInstanceId)
 
         // HTTP fallback notification (best-effort, after local cleanup)
         if (!notifiedViaWs) {

app/src/desktopMain/kotlin/com/crosspaste/DesktopNetworkModule.kt

@@ -252,7 +252,7 @@ fun desktopNetworkModule(marketingMode: Boolean): Module =
         single<SharePushOrchestrator> { SharePushOrchestrator(get(), get(), get()) }
         single<SyncDeviceManager> {
             SyncDeviceManager(
-                pasteDao = get(),
+                pastePullCursorManager = get(),
                 pendingExchangeLedger = get(),
                 secureStore = get(),
                 syncClientApi = get(),

app/src/desktopMain/kotlin/com/crosspaste/DesktopPasteComponentModule.kt

@@ -53,6 +53,7 @@ import com.crosspaste.paste.plugin.type.UrlTypePlugin
 import com.crosspaste.rendering.OpenGraphService
 import com.crosspaste.rendering.RenderingService
 import com.crosspaste.sync.FilePullService
+import com.crosspaste.sync.PastePullCursorManager
 import com.crosspaste.sync.PastePullService
 import com.crosspaste.task.CleanPasteTaskExecutor
 import com.crosspaste.task.CleanTaskTaskExecutor
@@ -169,6 +170,7 @@ fun desktopPasteComponentModule(headless: Boolean): Module =
                 notificationManager = get(),
                 pasteDao = get(),
                 pasteItemReader = get(),
+                pastePullCursorManager = get(),
                 // Plugin order is load-bearing (mobile assembles the same commonMain
                 // plugins and must keep these invariants):
                 // 1. RemoveInvalid must run first.
@@ -211,6 +213,7 @@ fun desktopPasteComponentModule(headless: Boolean): Module =
 
         // region Sync pull
         single<FilePullService> { FilePullService(get(), get(), get(), get(), get(), get()) }
+        single<PastePullCursorManager> { PastePullCursorManager(get(), get()) }
         single<PastePullService> { PastePullService(get(), get(), get(), get()) }
         // endregion
 

app/src/desktopMain/kotlin/com/crosspaste/headless/HeadlessPasteboardService.kt

@@ -71,15 +71,8 @@ class HeadlessPasteboardService(
             }
         }
 
-    override suspend fun tryWriteRemotePasteboardList(pasteDataList: List<PasteData>): Result<Unit?> {
-        if (pasteDataList.isEmpty()) return Result.success(null)
-
-        for (i in 0 until pasteDataList.size - 1) {
-            pasteReleaseService.releaseRemotePasteData(pasteDataList[i]) { _ -> }
-        }
-
-        return tryWriteRemotePasteboard(pasteDataList.last())
-    }
+    override suspend fun tryWriteRemotePasteboardList(pasteDataList: List<PasteData>): Result<Unit?> =
+        pasteReleaseService.releaseRemotePasteDataList(pasteDataList, ::tryWriteRemotePasteboard)
 
     override suspend fun tryWriteRemotePasteboardWithFile(pasteId: Long): Result<Unit?> =
         pasteReleaseService.releaseRemotePasteDataWithFile(pasteId) {

app/src/desktopMain/kotlin/com/crosspaste/paste/AbstractPasteboardService.kt

@@ -119,15 +119,8 @@ abstract class AbstractPasteboardService :
                 }
         }
 
-    override suspend fun tryWriteRemotePasteboardList(pasteDataList: List<PasteData>): Result<Unit?> {
-        if (pasteDataList.isEmpty()) return Result.success(null)
-
-        for (i in 0 until pasteDataList.size - 1) {
-            pasteReleaseService.releaseRemotePasteData(pasteDataList[i]) { _ -> }
-        }
-
-        return tryWriteRemotePasteboard(pasteDataList.last())
-    }
+    override suspend fun tryWriteRemotePasteboardList(pasteDataList: List<PasteData>): Result<Unit?> =
+        pasteReleaseService.releaseRemotePasteDataList(pasteDataList, ::tryWriteRemotePasteboard)
 
     override suspend fun tryWriteRemotePasteboardWithFile(pasteId: Long): Result<Unit?> =
         pasteReleaseService.releaseRemotePasteDataWithFile(pasteId) { storePasteData ->

app/src/desktopTest/kotlin/com/crosspaste/paste/PasteReleaseServicePushTest.kt

@@ -9,6 +9,7 @@ import com.crosspaste.paste.item.PasteItemReader
 import com.crosspaste.path.PlatformUserDataPathProvider
 import com.crosspaste.path.UserDataPathProvider
 import com.crosspaste.presist.SingleFileInfoTree
+import com.crosspaste.sync.PastePullCursorManager
 import com.crosspaste.task.TaskSubmitter
 import com.crosspaste.utils.getJsonUtils
 import io.mockk.coEvery
@@ -43,6 +44,7 @@ class PasteReleaseServicePushTest {
             pasteDao = pasteDao,
             pasteItemReader = mockk<PasteItemReader>(relaxed = true),
             pasteProcessPlugins = emptyList(),
+            pastePullCursorManager = mockk<PastePullCursorManager>(relaxed = true),
             searchContentService = mockk(relaxed = true),
             syncRuntimeInfoDao = mockk(relaxed = true),
             taskSubmitter = mockk<TaskSubmitter>(relaxed = true),