Restore Kotlin build errors and expand tests

payjoin-ffi/kotlin/CONTRIBUTING.md

@@ -25,5 +25,5 @@ Protocol `close` is renamed to `closeSession` only in `[bindings.kotlin.rename]`
 
 With nix, `nix develop .#kotlin` provides the pinned MSRV Rust toolchain, JDK 21, and
 `BITCOIND_EXE` (via `nixpkgs`, with `BITCOIND_SKIP_DOWNLOAD=1`), and is what CI uses. Without
-nix, use a local JDK 21+ and the Gradle wrapper; `corepc-node` will download `bitcoind` on
-first test run.
+nix, install Python 3 for binding generation and use a local JDK 21+ and the Gradle wrapper;
+`corepc-node` will download `bitcoind` on first test run.

payjoin-ffi/kotlin/README.md

@@ -22,7 +22,7 @@ Or `bash ./contrib/test.sh` from this directory (uses `Cargo-recent.lock`).
 `bitcoind` and `BITCOIND_SKIP_DOWNLOAD=1` is set, so no download happens. Outside the nix shell,
 Bitcoin Core is downloaded by corepc-node (`29_0`) on first run instead.
 
-Without nix: Rust (see repo `rust-toolchain.toml` / MSRV 1.85), JDK 21+, and the Gradle wrapper in this directory. Network access is required the first time bitcoind is fetched.
+Without nix: Rust (see repo `rust-toolchain.toml` / MSRV 1.85), JDK 21+, Python 3, and the Gradle wrapper in this directory. Python parses Cargo's native library output during generation. Network access is required the first time bitcoind is fetched.
 
 ## Stability
 

payjoin-ffi/kotlin/scripts/generate_bindings.sh

@@ -38,7 +38,7 @@ run_cargo() {
 
 # compiler-artifact filenames include target-dir, CARGO_BUILD_TARGET, and profile.
 NATIVE_LIB="$(
-    run_cargo build --message-format=json --profile "$PAYJOIN_FFI_PROFILE" -p payjoin-ffi |
+    run_cargo build --message-format=json-render-diagnostics --profile "$PAYJOIN_FFI_PROFILE" -p payjoin-ffi |
         python3 -c '
 import json, os, sys
 

payjoin-ffi/kotlin/src/test/kotlin/org/payjoindevkit/ControlledPersisters.kt

@@ -0,0 +1,86 @@
+package org.payjoindevkit
+
+import java.util.concurrent.CopyOnWriteArrayList
+import kotlin.test.assertFalse
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.supervisorScope
+
+internal class ControlledPersister : JsonReceiverSessionPersister, JsonSenderSessionPersister {
+    val events = CopyOnWriteArrayList<String>()
+    val calls = CopyOnWriteArrayList<String>()
+    @Volatile var failure: String? = null
+    @Volatile var closed = false
+
+    private fun before(operation: String) {
+        calls.add(operation)
+        if (failure == operation) throw ForeignException.InternalException("storage $operation failed")
+    }
+
+    override fun save(event: String) {
+        before("save")
+        events.add(event)
+    }
+
+    override fun load(): List<String> {
+        before("load")
+        return events.toList()
+    }
+
+    override fun closeSession() {
+        before("close")
+        closed = true
+    }
+}
+
+internal class ControlledAsyncPersister : JsonReceiverSessionPersisterAsync, JsonSenderSessionPersisterAsync {
+    val events = CopyOnWriteArrayList<String>()
+    val calls = CopyOnWriteArrayList<String>()
+    @Volatile var failure: String? = null
+    @Volatile var closed = false
+    @Volatile private var gatedOperation: String? = null
+    @Volatile private var entered = CompletableDeferred<Unit>()
+    @Volatile private var release = CompletableDeferred<Unit>()
+
+    private suspend fun before(operation: String) {
+        calls.add(operation)
+        if (gatedOperation == operation) {
+            entered.complete(Unit)
+            release.await()
+        }
+        if (failure == operation) throw ForeignException.InternalException("storage $operation failed")
+    }
+
+    override suspend fun save(event: String) {
+        before("save")
+        events.add(event)
+    }
+
+    override suspend fun load(): List<String> {
+        before("load")
+        return events.toList()
+    }
+
+    override suspend fun closeSession() {
+        before("close")
+        closed = true
+    }
+
+    // Wait until Rust enters the callback, then prove the FFI call remains pending
+    // before releasing it. Callers bound the whole operation with withTimeout.
+    suspend fun <T> gated(operation: String, block: suspend () -> T): T = supervisorScope {
+        entered = CompletableDeferred()
+        release = CompletableDeferred()
+        gatedOperation = operation
+        val call = async { block() }
+        try {
+            entered.await()
+            assertFalse(call.isCompleted, "FFI call returned before $operation callback resumed")
+            release.complete(Unit)
+            call.await()
+        } finally {
+            release.complete(Unit)
+            gatedOperation = null
+        }
+    }
+}

payjoin-ffi/kotlin/src/test/kotlin/org/payjoindevkit/IntegrationTests.kt

@@ -3,8 +3,13 @@ package org.payjoindevkit
 import java.util.HexFormat
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertIs
+import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
 import kotlin.test.fail
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonNull
 import kotlinx.serialization.json.boolean
@@ -27,7 +32,54 @@ import kotlinx.serialization.json.jsonPrimitive
  */
 class IntegrationTests {
     @Test
-    fun v2ToV2Payjoin() {
+    fun receiverPendingFallbackCloseSync() = runScenario { proposal, persister ->
+        proposal.cancel().use { transition ->
+            assertNotNull(transition.save(persister)).use { pending ->
+                assertTrue(pending.fallbackTx().isNotEmpty())
+                replayReceiverEventLog(persister).use {
+                    assertIs<ReceiveSession.ReceiverPendingFallback>(it.state())
+                }
+                assertFalse(persister.closed)
+                pending.closeSession().use { it.save(persister) }
+                assertTrue(persister.closed)
+                replayReceiverEventLog(persister).use { assertIs<ReceiveSession.Closed>(it.state()) }
+            }
+        }
+    }
+
+    @Test
+    fun receiverPendingFallbackCloseAsync() = runScenario { proposal, persister ->
+        runBlocking {
+            withTimeout(15_000) {
+                // Use the same persisted proposal history through the async interface.
+                val asyncPersister = ControlledAsyncPersister()
+                asyncPersister.events.addAll(persister.load())
+                proposal.cancel().use { transition ->
+                    assertNotNull(transition.saveAsync(asyncPersister)).use { pending ->
+                        assertTrue(pending.fallbackTx().isNotEmpty())
+                        replayReceiverEventLogAsync(asyncPersister).use {
+                            assertIs<ReceiveSession.ReceiverPendingFallback>(it.state())
+                        }
+                        assertFalse(asyncPersister.closed)
+                        pending.closeSession().use { transition ->
+                            asyncPersister.gated("close") { transition.saveAsync(asyncPersister) }
+                        }
+                        assertTrue(asyncPersister.closed)
+                        replayReceiverEventLogAsync(asyncPersister).use {
+                            assertIs<ReceiveSession.Closed>(it.state())
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    @Test
+    fun v2ToV2Payjoin() = runScenario()
+
+    private fun runScenario(
+        cancelReceiver: ((PayjoinProposal, InMemoryReceiverPersister) -> Unit)? = null,
+    ) {
         initTracing()
         TestServices.initialize().use { services ->
             services.waitForServicesReady()
@@ -45,6 +97,7 @@ class IntegrationTests {
                                     ohttpKeys,
                                     senderRpc,
                                     receiverRpc,
+                                    cancelReceiver,
                                 )
                             }
                         }
@@ -61,6 +114,7 @@ class IntegrationTests {
         ohttpKeys: OhttpKeys,
         senderRpc: RpcClient,
         receiverRpc: RpcClient,
+        cancelReceiver: ((PayjoinProposal, InMemoryReceiverPersister) -> Unit)?,
     ) {
         val receiverAddress = Json.parseToJsonElement(rpc(receiverRpc, "getnewaddress")).jsonPrimitive.content
         val senderOutpoints = listOutpoints(senderRpc)
@@ -95,6 +149,10 @@ class IntegrationTests {
                         // Inside the Receiver:
                         val payjoinProposal = waitForReceiverProposal(session, recvPersister, http, relay, receiverRpc)
                         payjoinProposal.use {
+                            if (cancelReceiver != null) {
+                                cancelReceiver(payjoinProposal, recvPersister)
+                                return
+                            }
                             payjoinProposal.createPostRequest(relay).useDisposable { posted ->
                                 val body = http.post(posted.request)
                                 payjoinProposal.processResponse(body, posted.clientResponse).use { transition ->