Merge pull request #1017 from synonymdev/fix/cjit-channel-false-match

fix: match cjit entry to channel by funding tx

Author: ovitrif

AGENTS.md

@@ -153,7 +153,7 @@ fun updateState(action: Action) {
 
 ```kotlin
 suspend fun getData(): Result<Data> = withContext(Dispatchers.IO) {
-  runCatching {
+  runSuspendCatching {
     apiService.fetchData()
   }.onFailure {
     Logger.error("Failed", it, context = TAG)
@@ -196,6 +196,8 @@ suspend fun getData(): Result<Data> = withContext(Dispatchers.IO) {
 - ALWAYS log errors at the final handling layer where the error is acted upon, not in intermediate layers that just propagate it
 - ALWAYS use the Result API instead of try-catch
 - NEVER wrap methods returning `Result<T>` in try-catch
+- ALWAYS use `runSuspendCatching` (from `ext/Coroutines.kt`) instead of `runCatching` when the block calls suspend functions or runs in a coroutine — it re-throws `CancellationException` so structured-concurrency cancellation is preserved; plain `runCatching` catches `Throwable` and swallows it. NEVER log a `CancellationException` as an error
+- EXCEPTION: when a `TimeoutCancellationException` from `withTimeout` must be treated as a retriable failure, use `runCatching` with an explicit `if (it is CancellationException && it !is TimeoutCancellationException) throw it` guard (e.g. `BlocktankRepo.refreshCjitEntries`)
 - PREFER to use `it` instead of explicit named parameters in lambdas e.g. `fn().onSuccess { log(it) }.onFailure { log(it) }`
 - NEVER inject ViewModels as dependencies - Only android activities and composable functions can use viewmodels
 - ALWAYS co-locate screen-specific ViewModels in the same package as their screen; only place ViewModels in `viewmodels/` when shared across multiple screens

app/src/main/java/to/bitkit/domain/commands/NotifyChannelReadyHandler.kt

@@ -4,6 +4,7 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.withContext
 import to.bitkit.di.IoDispatcher
 import to.bitkit.ext.amountOnClose
+import to.bitkit.ext.runSuspendCatching
 import to.bitkit.models.NewTransactionSheetDetails
 import to.bitkit.models.NewTransactionSheetDirection
 import to.bitkit.models.NewTransactionSheetType
@@ -30,17 +31,17 @@ class NotifyChannelReadyHandler @Inject constructor(
     suspend operator fun invoke(
         command: NotifyChannelReady.Command,
     ): Result<NotifyChannelReady.Result> = withContext(ioDispatcher) {
-        runCatching {
+        runSuspendCatching {
             val channel = lightningRepo.getChannels()
                 ?.find { it.channelId == command.event.channelId }
-                ?: return@runCatching NotifyChannelReady.Result.Skip
+                ?: return@runSuspendCatching NotifyChannelReady.Result.Skip
 
             val cjitEntry = blocktankRepo.getCjitEntry(channel)
-                ?: return@runCatching NotifyChannelReady.Result.Skip
+                ?: return@runSuspendCatching NotifyChannelReady.Result.Skip
 
             val inserted = activityRepo.insertActivityFromCjit(cjitEntry = cjitEntry, channel = channel)
                 .getOrDefault(false)
-            if (!inserted) return@runCatching NotifyChannelReady.Result.Duplicate
+            if (!inserted) return@runSuspendCatching NotifyChannelReady.Result.Duplicate
 
             val sats = channel.amountOnClose.toLong()
 

app/src/main/java/to/bitkit/ext/Coroutines.kt

@@ -1,8 +1,19 @@
 package to.bitkit.ext
 
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Job
 import to.bitkit.utils.Logger
 
+@Suppress("TooGenericExceptionCaught")
+suspend inline fun <R> runSuspendCatching(block: () -> R): Result<R> =
+    try {
+        Result.success(block())
+    } catch (c: CancellationException) {
+        throw c
+    } catch (e: Throwable) {
+        Result.failure(e)
+    }
+
 fun Job.logCompletion(name: String = "") = invokeOnCompletion { err ->
     if (err != null) {
         Logger.verbose("Coroutine '$name' error: ${err.message}")

app/src/main/java/to/bitkit/repositories/BlocktankRepo.kt

@@ -2,6 +2,7 @@ package to.bitkit.repositories
 
 import androidx.compose.runtime.Stable
 import com.synonym.bitkitcore.BtOrderState2
+import com.synonym.bitkitcore.CJitStateEnum
 import com.synonym.bitkitcore.ChannelLiquidityOptions
 import com.synonym.bitkitcore.ChannelLiquidityParams
 import com.synonym.bitkitcore.CreateCjitOptions
@@ -18,10 +19,12 @@ import com.synonym.bitkitcore.giftPay
 import kotlinx.collections.immutable.ImmutableList
 import kotlinx.collections.immutable.persistentListOf
 import kotlinx.collections.immutable.toImmutableList
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.async
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.currentCoroutineContext
@@ -41,6 +44,7 @@ import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
 import kotlinx.coroutines.withTimeoutOrNull
 import org.lightningdevkit.ldknode.Bolt11Invoice
 import org.lightningdevkit.ldknode.ChannelDetails
@@ -125,10 +129,48 @@ class BlocktankRepo @Inject constructor(
     }
 
     suspend fun getCjitEntry(channel: ChannelDetails): IcJitEntry? = withContext(bgDispatcher) {
-        return@withContext _blocktankState.value.cjitEntries.firstOrNull { order ->
-            order.channelSizeSat == channel.channelValueSats &&
-                order.lspNode.pubkey == channel.counterpartyNodeId
+        val fundingTxo = channel.fundingTxo ?: return@withContext null
+
+        fun List<IcJitEntry>.matching(): IcJitEntry? = firstOrNull { entry ->
+            val fundingTx = entry.channel?.fundingTx ?: return@firstOrNull false
+            fundingTx.id == fundingTxo.txid && fundingTx.vout == fundingTxo.vout.toULong()
+        }
+
+        val cached = _blocktankState.value.cjitEntries
+        cached.matching()?.let { return@withContext it }
+
+        // A ChannelReady can only be a CJIT if a live cached entry is still awaiting its channel; otherwise skip
+        // the server round-trip so a non-CJIT transfer confirmation isn't delayed by a slow Blocktank API.
+        val hasPendingCjit = cached.any {
+            it.channel == null && it.state != CJitStateEnum.EXPIRED && it.state != CJitStateEnum.FAILED
+        }
+        if (cached.isNotEmpty() && !hasPendingCjit) return@withContext null
+
+        // Match against the freshly fetched list so a concurrent refreshOrders() can't clobber state before we read.
+        return@withContext refreshCjitEntries().matching()
+    }
+
+    private suspend fun refreshCjitEntries(): List<IcJitEntry> {
+        repeat(CJIT_REFRESH_ATTEMPTS) { attempt ->
+            val entries = runCatching {
+                withTimeout(CJIT_REFRESH_TIMEOUT) {
+                    coreService.blocktank.cjitEntries(refresh = true)
+                }
+            }.mapCatching { entries ->
+                _blocktankState.update { it.copy(cjitEntries = entries.toImmutableList()) }
+                entries
+            }.getOrElse {
+                if (it is CancellationException && it !is TimeoutCancellationException) throw it
+                if (attempt == CJIT_REFRESH_ATTEMPTS - 1) {
+                    Logger.warn("Failed to refresh CJIT entries; using cached state", it, context = TAG)
+                }
+                null
+            }
+
+            entries?.let { return it }
+            delay(CJIT_REFRESH_RETRY_DELAY)
         }
+        return _blocktankState.value.cjitEntries
     }
 
     suspend fun refreshInfo() = withContext(bgDispatcher) {
@@ -553,6 +595,9 @@ class BlocktankRepo @Inject constructor(
         private const val PEER_CONNECTION_DELAY_MS = 2_000L
         private val TIMEOUT_GIFT_CODE = 30.seconds
         private val GIFT_PAYMENT_RECEIVE_TIMEOUT = 45.seconds
+        private const val CJIT_REFRESH_ATTEMPTS = 3
+        private val CJIT_REFRESH_TIMEOUT = 5.seconds
+        private val CJIT_REFRESH_RETRY_DELAY = 1.seconds
     }
 }
 

app/src/test/java/to/bitkit/ext/CoroutinesTest.kt

@@ -0,0 +1,35 @@
+package to.bitkit.ext
+
+import kotlinx.coroutines.CancellationException
+import org.junit.Test
+import to.bitkit.test.BaseUnitTest
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertTrue
+
+class CoroutinesTest : BaseUnitTest() {
+
+    @Test
+    fun `runSuspendCatching wraps a successful result`() = test {
+        val result = runSuspendCatching { 42 }
+
+        assertEquals(42, result.getOrNull())
+    }
+
+    @Test
+    fun `runSuspendCatching wraps a thrown exception as failure`() = test {
+        val error = IllegalStateException("boom")
+
+        val result = runSuspendCatching { throw error }
+
+        assertTrue(result.isFailure)
+        assertEquals(error, result.exceptionOrNull())
+    }
+
+    @Test
+    fun `runSuspendCatching re-throws CancellationException`() = test {
+        assertFailsWith<CancellationException> {
+            runSuspendCatching { throw CancellationException("cancelled") }
+        }
+    }
+}