diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2db2aa193..cfe0fd459 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -169,6 +169,12 @@ jobs: - name: Check committed iOS bindings are current run: ./ios/truapi-host/scripts/sync-bindings.sh --check + # The provider ships its own UniFFI surface and its own committed + # bindings; `rebuild.sh` overwrites them in place, so only this check + # can fail on drift. + - name: Check committed TrUAPIProvider bindings are current + run: make provider-swift-check + ios-changes: name: iOS change filter runs-on: ubuntu-latest @@ -197,7 +203,7 @@ jobs: exit 0 fi if git diff --name-only HEAD^1 HEAD \ - | grep -qE '^(ios/|Package\.swift$|Makefile$|rust/crates/truapi-server/src/native)'; then + | grep -qE '^(ios/|Package\.swift$|Makefile$|rust/crates/truapi-server/src/native|rust/crates/truapi-provider/)'; then echo "ios=true" >> "$GITHUB_OUTPUT" else echo "ios=false" >> "$GITHUB_OUTPUT" diff --git a/Makefile b/Makefile index fcbb82491..cd8a8caed 100644 --- a/Makefile +++ b/Makefile @@ -93,6 +93,7 @@ PROVIDER_CDYLIB := $(UNIFFI_CDYLIB_DIR)/libtruapi_provider.so endif UNIFFI_SWIFT_TMP := target/uniffi-swift-out +PROVIDER_SWIFT_TMP := target/uniffi-provider-swift-check uniffi: ## Generate Swift bindings from the truapi-server cdylib into target/uniffi-swift-out (consumed by ios/truapi-host/scripts/rebuild.sh). $(CARGO) build -p truapi-server --profile codegen --features ws-bridge @@ -215,6 +216,23 @@ android-publish-local: uniffi-kotlin ## Generate Kotlin bindings, then publish t PROVIDER_KOTLIN_OUT := android/truapi-provider/src/main/kotlin/generated PROVIDER_JNILIBS := android/truapi-provider/src/main/jniLibs +provider-swift: ## Generate the TrUAPIProvider Swift bindings into target/uniffi-provider-swift-out (no Xcode, no iOS targets). + $(CARGO) build -p truapi-provider --profile codegen --no-default-features --features uniffi + rm -rf $(PROVIDER_SWIFT_TMP) + mkdir -p $(PROVIDER_SWIFT_TMP) + $(CARGO) run -p uniffi-bindgen-cli -- generate \ + --library $(PROVIDER_CDYLIB) \ + --language swift \ + --out-dir $(PROVIDER_SWIFT_TMP) + +provider-swift-check: provider-swift ## Fail if the committed TrUAPIProvider bindings are stale. + @diff -u ios/truapi-provider/Sources/TrUAPIProvider/truapi_provider.swift \ + $(PROVIDER_SWIFT_TMP)/truapi_provider.swift \ + && diff -u ios/truapi-provider/Sources/truapi_providerFFI/include/truapi_providerFFI.h \ + $(PROVIDER_SWIFT_TMP)/truapi_providerFFI.h \ + && echo "Committed TrUAPIProvider bindings are current." \ + || { echo "Committed TrUAPIProvider bindings are stale: run 'make provider-ios'."; exit 1; } + provider-ios: ## Build the TrUAPIProvider Swift bindings + xcframework (adds --sim-only via SIM_ONLY=1). bash ios/truapi-provider/scripts/rebuild.sh $(if $(SIM_ONLY),--sim-only,) diff --git a/android/truapi-host/src/main/kotlin/io/parity/truapi/TrUAPIHost.kt b/android/truapi-host/src/main/kotlin/io/parity/truapi/TrUAPIHost.kt index e1aa4726c..9b8b11d33 100644 --- a/android/truapi-host/src/main/kotlin/io/parity/truapi/TrUAPIHost.kt +++ b/android/truapi-host/src/main/kotlin/io/parity/truapi/TrUAPIHost.kt @@ -26,7 +26,10 @@ package io.parity.truapi +import kotlinx.coroutines.CancellationException import uniffi.truapi.HostDevicePermissionRequest +import uniffi.truapi.HostLocalStorageReadError +import uniffi.truapi.HostNavigateToError import uniffi.truapi.HostFeatureSupportedRequest import uniffi.truapi.HostPushNotificationRequest import uniffi.truapi.RemotePermission @@ -330,74 +333,134 @@ interface HostBridge { val coreStorage: HostCoreStorage } +// A host that throws an exception type its callback does not declare crosses +// the FFI as an unexpected callback error. The Rust core converts those rather +// than aborting, but the reason it receives is then a raw JVM description, so +// each adapter funnels host throws into the declared type here. + +// Bounded: this reaches the product as the rejection reason, and a host +// message can carry a whole failed statement. +private fun hostRejectionReason(error: Throwable): String = + (error.message ?: error.toString()).take(HOST_REJECTION_REASON_MAX_CHARS) + +private const val HOST_REJECTION_REASON_MAX_CHARS = 256 + +private inline fun withHostRejection(operation: () -> T): T = + try { + operation() + } catch (rejection: HostRejection) { + throw rejection + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Throwable) { + throw HostRejection.Rejected(hostRejectionReason(error)).apply { initCause(error) } + } + +private inline fun withNavigateRejection(operation: () -> T): T = + try { + operation() + } catch (rejection: HostNavigateRejection) { + throw rejection + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Throwable) { + throw HostNavigateRejection.Navigate( + HostNavigateToError.Unknown(hostRejectionReason(error)), + ).apply { initCause(error) } + } + +private inline fun withStorageException(operation: () -> T): T = + try { + operation() + } catch (storage: HostStorageException) { + throw storage + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Throwable) { + throw HostStorageException.Storage( + HostLocalStorageReadError.Unknown(hostRejectionReason(error)), + ).apply { initCause(error) } + } + /** * Adapter from the public [HostBridge] surface to the generated UniFFI * [HostCallbacks] interface. Keeps the public API stable even if uniffi-bindgen * renames generated symbols. */ private class HostCallbackAdapter(private val bridge: HostBridge) : HostCallbacks { - override fun onCoreLog(marker: String, detail: String) = - bridge.onCoreLog(marker, detail) + // The core declares this and `authStateChanged` infallible, so uniffi has + // no error type to convert a throw into and panics -- which aborts under + // `panic = "abort"`. Neither may let a host exception reach the FFI. + override fun onCoreLog(marker: String, detail: String) { + runCatching { bridge.onCoreLog(marker, detail) } + } override suspend fun navigateTo(url: String) = - bridge.navigateTo(url) + withNavigateRejection { bridge.navigateTo(url) } override suspend fun pushNotification(request: HostPushNotificationRequest): UInt = - bridge.pushNotification(request) + withHostRejection { bridge.pushNotification(request) } override fun cancelNotification(id: UInt) = - bridge.cancelNotification(id) + withHostRejection { bridge.cancelNotification(id) } override suspend fun devicePermission(request: HostDevicePermissionRequest): Boolean = - bridge.devicePermission(request) + withHostRejection { bridge.devicePermission(request) } override suspend fun remotePermission(request: RemotePermission): Boolean = - bridge.remotePermission(request) - - override fun authStateChanged(state: AuthState) = - bridge.authStateChanged(state) + withHostRejection { bridge.remotePermission(request) } + + override fun authStateChanged(state: AuthState) { + try { + bridge.authStateChanged(state) + } catch (error: Throwable) { + runCatching { + bridge.onCoreLog("host.auth_state_changed.threw", error.stackTraceToString()) + } + } + } override fun coreStorageRead(key: ByteArray): ByteArray? = - bridge.coreStorage.read(key) + withHostRejection { bridge.coreStorage.read(key) } override fun coreStorageWrite(key: ByteArray, value: ByteArray) = - bridge.coreStorage.write(key, value) + withHostRejection { bridge.coreStorage.write(key, value) } override fun coreStorageClear(key: ByteArray) = - bridge.coreStorage.clear(key) + withHostRejection { bridge.coreStorage.clear(key) } override fun chainConnect(genesisHash: ByteArray): UInt? = - bridge.chainConnect(genesisHash) + withHostRejection { bridge.chainConnect(genesisHash) } override fun chainSend(connectionId: UInt, request: String) = - bridge.chainSend(connectionId, request) + withHostRejection { bridge.chainSend(connectionId, request) } override fun chainClose(connectionId: UInt) = - bridge.chainClose(connectionId) + withHostRejection { bridge.chainClose(connectionId) } override suspend fun confirmUserAction(review: UserConfirmationReview): Boolean = - bridge.confirmUserAction(review) + withHostRejection { bridge.confirmUserAction(review) } override suspend fun lookupPreimage(key: ByteArray): ByteArray? = - bridge.lookupPreimage(key) + withHostRejection { bridge.lookupPreimage(key) } override fun currentTheme(): HostThemeSubscribeItem = - bridge.currentTheme() + withHostRejection { bridge.currentTheme() } override suspend fun featureSupported(request: HostFeatureSupportedRequest): Boolean = - bridge.featureSupported(request) + withHostRejection { bridge.featureSupported(request) } override fun supportedChains(): HostChainSet = - bridge.supportedChains() + withHostRejection { bridge.supportedChains() } override fun localStorageRead(key: String): ByteArray? = - bridge.storage.read(key) + withStorageException { bridge.storage.read(key) } override fun localStorageWrite(key: String, value: ByteArray) = - bridge.storage.write(key, value) + withStorageException { bridge.storage.write(key, value) } override fun localStorageClear(key: String) = - bridge.storage.clear(key) + withStorageException { bridge.storage.clear(key) } } /** diff --git a/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift b/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift index 40a4a6b2d..873d90f1c 100644 --- a/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift +++ b/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift @@ -530,7 +530,7 @@ private final class ChatCallbackAdapter: NativeChatCallbacks, @unchecked Sendabl } catch let error as HostRejection { throw error } catch { - throw HostRejection.Rejected(reason: error.localizedDescription) + throw HostRejection.Rejected(reason: hostRejectionReason(error)) } } } @@ -673,7 +673,7 @@ private final class HostCallbackAdapter: HostCallbacks, @unchecked Sendable { } catch let error as HostRejection { throw error } catch { - throw HostRejection.Rejected(reason: error.localizedDescription) + throw HostRejection.Rejected(reason: hostRejectionReason(error)) } } @@ -683,7 +683,7 @@ private final class HostCallbackAdapter: HostCallbacks, @unchecked Sendable { } catch let error as HostRejection { throw error } catch { - throw HostRejection.Rejected(reason: error.localizedDescription) + throw HostRejection.Rejected(reason: hostRejectionReason(error)) } } @@ -693,7 +693,7 @@ private final class HostCallbackAdapter: HostCallbacks, @unchecked Sendable { } catch let error as HostNavigateRejection { throw error } catch { - throw HostNavigateRejection.Navigate(.unknown(reason: error.localizedDescription)) + throw HostNavigateRejection.Navigate(.unknown(reason: hostRejectionReason(error))) } } @@ -703,7 +703,7 @@ private final class HostCallbackAdapter: HostCallbacks, @unchecked Sendable { } catch let error as HostNavigateRejection { throw error } catch { - throw HostNavigateRejection.Navigate(.unknown(reason: error.localizedDescription)) + throw HostNavigateRejection.Navigate(.unknown(reason: hostRejectionReason(error))) } } @@ -713,7 +713,7 @@ private final class HostCallbackAdapter: HostCallbacks, @unchecked Sendable { } catch let error as HostStorageError { throw error } catch { - throw HostStorageError.Storage(.unknown(reason: error.localizedDescription)) + throw HostStorageError.Storage(.unknown(reason: hostRejectionReason(error))) } } } @@ -1100,6 +1100,33 @@ public final class TrUAPIHostCore: TrUAPIHostCoreProtocol { } } +/// Reason text for an error a host threw from a callback. +/// +/// A value that is not a `LocalizedError` has no author-written description, +/// and `localizedDescription` renders it as "The operation couldn't be +/// completed. (Module.Type error 1.)" — which names the host's module and says +/// nothing about the failure. A plain value's `String(describing:)` prints its +/// stored properties, so only the type name crosses to the product. +private func hostRejectionReason(_ error: Error) -> String { + let reason: String + if let described = (error as? LocalizedError)?.errorDescription { + reason = described + } else if type(of: error) is NSError.Type { + // Foundation writes these, and the text describes the failure rather + // than the host's internals. + reason = error.localizedDescription + } else { + // A plain value's `String(describing:)` prints its stored properties, + // so only the type name crosses to the product. + reason = String(describing: type(of: error)) + } + return String(reason.prefix(hostRejectionReasonMaxCharacters)) +} + +/// Bounded: the reason reaches the product, and a host message can carry a +/// whole failed statement. +private let hostRejectionReasonMaxCharacters = 256 + private func customRendererStream( _ subscribe: (CustomRendererStreamObserver) throws -> NativeCustomRendererSubscription ) throws -> AsyncThrowingStream { diff --git a/ios/truapi-provider/README.md b/ios/truapi-provider/README.md index 64863c32f..7ac037f57 100644 --- a/ios/truapi-provider/README.md +++ b/ios/truapi-provider/README.md @@ -47,9 +47,9 @@ No Rust toolchain is needed: the xcframework carries the compiled crate, and the Everything is generated from [`ffi.rs`](../../rust/crates/truapi-provider/src/ffi.rs): - `ChainProvider` — construct **one per process** and share it. Every connection runs on the single embedded light client, so they share sync, peers, and warm state while keeping their own request queue and response stream. `connect(genesisHash:listener:)` resolves the network from the bundled catalog (relay wiring and statement-store placement included), so the 32-byte genesis hash is the only argument. -- `ChainMessageListener` — the host implements it; `onMessage(message:)` receives each JSON-RPC response and notification, `onClosed()` fires once the stream ends. +- `ChainMessageListener` — the host implements it; `onMessage(message:)` receives each JSON-RPC response and notification, `onClosed()` fires once the stream ends. Both may throw: a listener that throws stops the pump for that connection rather than being called again for every response, and an error it does not declare is reported as `.listener(reason:)` instead of aborting the process. - `ChainConnection` — `send(request:)` queues a request, `disconnect()` tears the connection down. -- `ChainProviderError` — `.connect(reason:)` when the genesis is outside the catalog or the transport fails, `.badGenesis` when the hash is not 32 bytes. +- `ChainProviderError` — `.connect(reason:)` when the genesis is outside the catalog or the transport fails, `.badGenesis` when the hash is not 32 bytes, `.listener(reason:)` when the host's listener failed in a way it did not declare. ## Architecture @@ -74,12 +74,12 @@ import Foundation import TrUAPIProvider final class Responses: ChainMessageListener, @unchecked Sendable { - func onMessage(message: String) { + func onMessage(message: String) throws { // A JSON-RPC response or subscription notification, verbatim from smoldot. DispatchQueue.main.async { /* decode and render */ } } - func onClosed() { + func onClosed() throws { DispatchQueue.main.async { /* the stream ended: drop the connection */ } } } diff --git a/ios/truapi-provider/Sources/TrUAPIProvider/truapi_provider.swift b/ios/truapi-provider/Sources/TrUAPIProvider/truapi_provider.swift index 0db866345..76fa8aea8 100644 --- a/ios/truapi-provider/Sources/TrUAPIProvider/truapi_provider.swift +++ b/ios/truapi-provider/Sources/TrUAPIProvider/truapi_provider.swift @@ -701,12 +701,12 @@ public protocol ChainMessageListener: AnyObject, Sendable { /** * Called for each JSON-RPC response or notification string. */ - func onMessage(message: String) + func onMessage(message: String) throws /** * Called once the connection's response stream ends. */ - func onClosed() + func onClosed() throws } /** @@ -769,7 +769,7 @@ open class ChainMessageListenerImpl: ChainMessageListener, @unchecked Sendable { /** * Called for each JSON-RPC response or notification string. */ -open func onMessage(message: String) {try! rustCall() { +open func onMessage(message: String)throws {try rustCallWithError(FfiConverterTypeChainProviderError_lift) { uniffiCallStatus in uniffi_truapi_provider_fn_method_chainmessagelistener_on_message( self.uniffiCloneHandle(), @@ -781,7 +781,7 @@ open func onMessage(message: String) {try! rustCall() { /** * Called once the connection's response stream ends. */ -open func onClosed() {try! rustCall() { +open func onClosed()throws {try rustCallWithError(FfiConverterTypeChainProviderError_lift) { uniffiCallStatus in uniffi_truapi_provider_fn_method_chainmessagelistener_on_closed( self.uniffiCloneHandle(),uniffiCallStatus @@ -828,17 +828,18 @@ fileprivate struct UniffiCallbackInterfaceChainMessageListener { guard let uniffiObj = try? FfiConverterTypeChainMessageListener.handleMap.get(handle: uniffiHandle) else { throw UniffiInternalError.unexpectedStaleHandle } - return uniffiObj.onMessage( + return try uniffiObj.onMessage( message: try FfiConverterString.lift(message) ) } let writeReturn = { () } - uniffiTraitInterfaceCall( + uniffiTraitInterfaceCallWithError( callStatus: uniffiCallStatus, makeCall: makeCall, - writeReturn: writeReturn + writeReturn: writeReturn, + lowerError: FfiConverterTypeChainProviderError_lower ) }, onClosed: { ( @@ -851,16 +852,17 @@ fileprivate struct UniffiCallbackInterfaceChainMessageListener { guard let uniffiObj = try? FfiConverterTypeChainMessageListener.handleMap.get(handle: uniffiHandle) else { throw UniffiInternalError.unexpectedStaleHandle } - return uniffiObj.onClosed( + return try uniffiObj.onClosed( ) } let writeReturn = { () } - uniffiTraitInterfaceCall( + uniffiTraitInterfaceCallWithError( callStatus: uniffiCallStatus, makeCall: makeCall, - writeReturn: writeReturn + writeReturn: writeReturn, + lowerError: FfiConverterTypeChainProviderError_lower ) } ) @@ -1108,6 +1110,14 @@ enum ChainProviderError: Swift.Error, Equatable, Hashable, Foundation.LocalizedE * The genesis hash was not exactly 32 bytes. */ case BadGenesis + /** + * The host's listener failed in a way it did not declare. + */ + case Listener( + /** + * Human-readable failure reason. + */reason: String + ) @@ -1141,6 +1151,9 @@ public struct FfiConverterTypeChainProviderError: FfiConverterRustBuffer { reason: try FfiConverterString.read(from: &buf) ) case 2: return .BadGenesis + case 3: return .Listener( + reason: try FfiConverterString.read(from: &buf) + ) default: throw UniffiInternalError.unexpectedEnumCase } @@ -1161,6 +1174,11 @@ public struct FfiConverterTypeChainProviderError: FfiConverterRustBuffer { case .BadGenesis: writeInt(&buf, Int32(2)) + + case let .Listener(reason): + writeInt(&buf, Int32(3)) + FfiConverterString.write(reason, into: &buf) + } } } @@ -1201,10 +1219,10 @@ private let initializationResult: InitializationResult = { if (uniffi_truapi_provider_checksum_method_chainconnection_send() != 52883) { return InitializationResult.apiChecksumMismatch } - if (uniffi_truapi_provider_checksum_method_chainmessagelistener_on_message() != 65048) { + if (uniffi_truapi_provider_checksum_method_chainmessagelistener_on_message() != 2156) { return InitializationResult.apiChecksumMismatch } - if (uniffi_truapi_provider_checksum_method_chainmessagelistener_on_closed() != 39748) { + if (uniffi_truapi_provider_checksum_method_chainmessagelistener_on_closed() != 1058) { return InitializationResult.apiChecksumMismatch } if (uniffi_truapi_provider_checksum_method_chainprovider_connect() != 16550) { diff --git a/rust/crates/truapi-provider/src/ffi.rs b/rust/crates/truapi-provider/src/ffi.rs index 99d81473d..8ebb0724f 100644 --- a/rust/crates/truapi-provider/src/ffi.rs +++ b/rust/crates/truapi-provider/src/ffi.rs @@ -10,9 +10,10 @@ //! background thread pumps the response stream and invokes it until the //! connection closes. -use std::sync::Arc; +use std::sync::{Arc, Weak}; use futures::executor::block_on; +use futures::stream::BoxStream; use futures::stream::StreamExt; use truapi_platform::{ChainProvider as _, JsonRpcConnection}; @@ -30,6 +31,25 @@ pub enum ChainProviderError { /// The genesis hash was not exactly 32 bytes. #[error("genesis hash must be 32 bytes")] BadGenesis, + /// The host's listener failed in a way it did not declare. + #[error("{reason}")] + Listener { + /// Human-readable failure reason. + reason: String, + }, +} + +impl From for ChainProviderError { + fn from(err: uniffi::UnexpectedUniFFICallbackError) -> Self { + // Without this the generic converter panics, and `panic = "abort"` on + // the shipping profile turns a listener's exception into a process + // abort rather than an ended stream. + tracing::warn!( + reason = %err.reason, + "chain listener threw an undeclared error; reporting it as a rejection" + ); + ChainProviderError::Listener { reason: err.reason } + } } /// Sink for a connection's inbound JSON-RPC responses and notifications, @@ -37,9 +57,9 @@ pub enum ChainProviderError { #[uniffi::export(with_foreign)] pub trait ChainMessageListener: Send + Sync { /// Called for each JSON-RPC response or notification string. - fn on_message(&self, message: String); + fn on_message(&self, message: String) -> Result<(), ChainProviderError>; /// Called once the connection's response stream ends. - fn on_closed(&self); + fn on_closed(&self) -> Result<(), ChainProviderError>; } /// Embedded-smoldot chain provider. Construct one per process and share it; @@ -76,20 +96,45 @@ impl ChainProvider { })?; let connection: Arc = Arc::from(connection); - let mut responses = connection.responses(); - std::thread::spawn(move || { - block_on(async move { - while let Some(message) = responses.next().await { - listener.on_message(message); - } - listener.on_closed(); - }); - }); + let responses = connection.responses(); + // Weak on purpose: a strong handle here would keep the connection alive + // forever. `Drop` is what calls `close()`, `close()` is what ends the + // response stream, and the pump parks on that stream -- so holding a + // strong reference means the drop that would release it can never run. + let pumped = Arc::downgrade(&connection); + std::thread::spawn(move || block_on(pump_responses(responses, pumped, listener))); Ok(Arc::new(ChainConnection { inner: connection })) } } +/// Deliver a connection's responses to its listener until the stream ends or +/// the listener fails. +/// +/// A failing listener closes the connection: the response stream is take-once, +/// so it cannot be pumped again, and leaving the handle open would queue every +/// later send against a receiver that is gone. `on_closed` fires either way, so +/// a host awaiting teardown is not left waiting -- though it carries no +/// argument, so it cannot say which of the two happened. +async fn pump_responses( + mut responses: BoxStream<'static, String>, + connection: Weak, + listener: Arc, +) { + while let Some(message) = responses.next().await { + if let Err(error) = listener.on_message(message) { + tracing::warn!(%error, "chain listener failed; closing the connection"); + if let Some(connection) = connection.upgrade() { + connection.close(); + } + break; + } + } + if let Err(error) = listener.on_closed() { + tracing::warn!(%error, "chain listener failed on close"); + } +} + /// A live JSON-RPC connection: a raw string pipe to one chain. #[derive(uniffi::Object)] pub struct ChainConnection { @@ -98,6 +143,7 @@ pub struct ChainConnection { #[cfg(all(test, feature = "networks"))] mod tests { + use core::future::Future; use std::sync::Mutex; use std::sync::mpsc::{Receiver, Sender, channel}; use std::time::Duration; @@ -160,15 +206,125 @@ mod tests { } } + /// Fails on the message at `fail_at`, recording what it was asked to do. + struct FailingListener { + fail_at: usize, + delivered: Mutex>, + closed: Mutex, + } + + impl ChainMessageListener for FailingListener { + fn on_message(&self, message: String) -> Result<(), ChainProviderError> { + let mut delivered = self.delivered.lock().expect("not poisoned"); + delivered.push(message); + if delivered.len() == self.fail_at { + return Err(ChainProviderError::Listener { + reason: "cannot decode".to_string(), + }); + } + Ok(()) + } + + fn on_closed(&self) -> Result<(), ChainProviderError> { + *self.closed.lock().expect("not poisoned") = true; + Ok(()) + } + } + + /// Records whether the pump closed it. + struct ClosingConnection { + closed: Mutex, + } + + impl JsonRpcConnection for ClosingConnection { + fn send(&self, _request: String) {} + + fn responses(&self) -> BoxStream<'static, String> { + Box::pin(futures::stream::empty()) + } + + fn close(&self) { + *self.closed.lock().expect("not poisoned") = true; + } + } + + #[test] + fn the_pump_does_not_keep_its_connection_alive() { + // The pump parks on a stream that only ends once the connection is + // dropped and closed. A strong handle here would make that drop + // unreachable, leaking the chain and the thread that waits on it. + let connection: Arc = Arc::new(ClosingConnection { + closed: Mutex::new(false), + }); + let weak = Arc::downgrade(&connection); + let listener = Arc::new(FailingListener { + fail_at: usize::MAX, + delivered: Mutex::new(Vec::new()), + closed: Mutex::new(false), + }); + + // Parked on a stream that never yields, which is where the pump spends + // its life. Polled first so the future is actually running: an async + // body executes nothing until then, so dropping before the first poll + // would prove nothing. + let mut pump = Box::pin(pump_responses( + Box::pin(futures::stream::pending()), + weak.clone(), + listener, + )); + let mut cx = core::task::Context::from_waker(futures::task::noop_waker_ref()); + assert!(pump.as_mut().poll(&mut cx).is_pending()); + + drop(connection); + assert!( + weak.upgrade().is_none(), + "a parked pump must not hold its connection alive" + ); + } + + #[test] + fn a_failing_listener_ends_the_connection_rather_than_going_deaf() { + let listener = Arc::new(FailingListener { + fail_at: 2, + delivered: Mutex::new(Vec::new()), + closed: Mutex::new(false), + }); + let connection = Arc::new(ClosingConnection { + closed: Mutex::new(false), + }); + let responses = futures::stream::iter(["first", "second", "third"].map(str::to_string)); + + block_on(pump_responses( + Box::pin(responses), + Arc::downgrade(&(connection.clone() as Arc)), + listener.clone(), + )); + + // Pumping stops at the failure rather than calling the listener again + // for every remaining response. + assert_eq!( + listener.delivered.lock().expect("not poisoned").as_slice(), + &["first".to_string(), "second".to_string()] + ); + // The host is told the stream ended; without this it waits forever on a + // teardown that never arrives. + assert!(*listener.closed.lock().expect("not poisoned")); + // And the connection is closed, so later sends are refused at the + // source instead of queueing against a receiver that is gone. + assert!(*connection.closed.lock().expect("not poisoned")); + } + impl ChainMessageListener for Collector { - fn on_message(&self, message: String) { + fn on_message(&self, message: String) -> Result<(), ChainProviderError> { let _ = self.messages.send(message); + Ok(()) } - fn on_closed(&self) { + fn on_closed(&self) -> Result<(), ChainProviderError> { if let Some(closed) = self.closed.lock().expect("not poisoned").take() { let _ = closed.send(()); } + Ok(()) } } diff --git a/rust/crates/truapi-server/src/host_logic/extrinsic.rs b/rust/crates/truapi-server/src/host_logic/extrinsic.rs index 93911157f..1437e1cb7 100644 --- a/rust/crates/truapi-server/src/host_logic/extrinsic.rs +++ b/rust/crates/truapi-server/src/host_logic/extrinsic.rs @@ -310,6 +310,12 @@ fn type_is_empty(type_id: R::TypeId, types: &R) -> bool { } /// Check that `bytes` traverse `type_id` and leave nothing over. +/// +/// `decode_with_visitor` applies no depth limit, and here the recursion depth +/// is set by product-supplied bytes against a type graph that lives in chain +/// metadata rather than in any Rust definition. That is safe only while the +/// extension types stay flat, which they are today; a recursive extension type +/// would need the bound that guards decoding elsewhere. fn traverse_exactly( bytes: &[u8], type_id: R::TypeId, diff --git a/rust/crates/truapi-server/src/native.rs b/rust/crates/truapi-server/src/native.rs index 1da59c979..702c9b4a7 100644 --- a/rust/crates/truapi-server/src/native.rs +++ b/rust/crates/truapi-server/src/native.rs @@ -60,6 +60,16 @@ pub enum HostStorageError { Storage(v01::HostLocalStorageReadError), } +impl From for HostStorageError { + fn from(err: uniffi::UnexpectedUniFFICallbackError) -> Self { + tracing::warn!( + reason = %err.reason, + "host callback threw an undeclared error; reporting it as a rejection" + ); + HostStorageError::Storage(v01::HostLocalStorageReadError::Unknown { reason: err.reason }) + } +} + impl From for v01::HostLocalStorageReadError { fn from(err: HostStorageError) -> Self { let HostStorageError::Storage(err) = err; @@ -93,6 +103,16 @@ impl From for v01::GenericError { } } +impl From for HostRejection { + fn from(err: uniffi::UnexpectedUniFFICallbackError) -> Self { + tracing::warn!( + reason = %err.reason, + "host callback threw an undeclared error; reporting it as a rejection" + ); + HostRejection::Rejected { reason: err.reason } + } +} + impl From for HostRejection { fn from(err: v01::GenericError) -> Self { HostRejection::Rejected { reason: err.reason } @@ -114,6 +134,16 @@ pub enum HostNavigateRejection { Navigate(v01::HostNavigateToError), } +impl From for HostNavigateRejection { + fn from(err: uniffi::UnexpectedUniFFICallbackError) -> Self { + tracing::warn!( + reason = %err.reason, + "host callback threw an undeclared error; reporting it as a rejection" + ); + HostNavigateRejection::Navigate(v01::HostNavigateToError::Unknown { reason: err.reason }) + } +} + /// Native-friendly SSO deeplink scheme. #[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Enum)] pub enum NativePairingDeeplinkScheme { @@ -1927,6 +1957,47 @@ mod tests { } } + #[test] + fn an_unexpected_foreign_error_converts_instead_of_panicking() { + // A host that throws an exception its trait does not declare lands in + // `try_convert_unexpected_callback_error`. Without a `From` impl the + // generic converter panics, and `panic = "abort"` turns that into a + // process abort on the shipping build. + let reason = "android.database.sqlite.SQLiteFullException"; + let rejection = >:: + try_convert_unexpected_callback_error( + uniffi::UnexpectedUniFFICallbackError::new(reason), + ) + .expect("an unexpected foreign error must convert"); + let HostRejection::Rejected { reason: converted } = rejection; + assert_eq!(converted, reason); + + let storage = >:: + try_convert_unexpected_callback_error( + uniffi::UnexpectedUniFFICallbackError::new(reason), + ) + .expect("an unexpected foreign error must convert"); + assert_eq!( + v01::HostLocalStorageReadError::from(storage), + v01::HostLocalStorageReadError::Unknown { + reason: reason.to_string(), + } + ); + + let navigate = >:: + try_convert_unexpected_callback_error( + uniffi::UnexpectedUniFFICallbackError::new(reason), + ) + .expect("an unexpected foreign error must convert"); + let HostNavigateRejection::Navigate(navigate) = navigate; + assert_eq!( + navigate, + v01::HostNavigateToError::Unknown { + reason: reason.to_string(), + } + ); + } + /// The renewal account is derived from `product_id`, and a product /// connection derives its own from the normalized form, so an unnormalized /// id here renews an account no product uses while the real one lapses. diff --git a/rust/crates/truapi-server/src/subscription.rs b/rust/crates/truapi-server/src/subscription.rs index 10bcdaab8..750415379 100644 --- a/rust/crates/truapi-server/src/subscription.rs +++ b/rust/crates/truapi-server/src/subscription.rs @@ -16,7 +16,7 @@ use futures::channel::mpsc; use futures::future::{BoxFuture, Either, select}; use futures::stream::BoxStream; use futures::{Stream, StreamExt}; -use parity_scale_codec::{Decode, Encode}; +use parity_scale_codec::{Decode, DecodeLimit, Encode}; use crate::frame::{IdFactory, Payload, ProtocolMessage}; use crate::generated::wire_table::SubscriptionFrameIds; @@ -462,6 +462,13 @@ impl HostInitiatedSubscription { } } +/// Nesting a product-supplied subscription item may reach before it is refused. +/// +/// Recursive payloads such as a custom renderer tree would otherwise decode +/// until the thread's stack is exhausted, which aborts the process rather than +/// failing the call. Far above any nesting the protocol's own types need. +const MAX_SUBSCRIPTION_DECODE_DEPTH: u32 = 64; + impl Stream for HostInitiatedSubscription where Item: Decode + Unpin, @@ -472,9 +479,18 @@ where match Pin::new(&mut self.receiver).poll_next(cx) { Poll::Ready(Some(bytes)) => { let mut input = &bytes[..]; - match Item::decode(&mut input) { + match Item::decode_with_depth_limit(MAX_SUBSCRIPTION_DECODE_DEPTH, &mut input) { Ok(item) if input.is_empty() => Poll::Ready(Some(item)), Ok(_) | Err(_) => { + // The peer sees a bare stop frame and the host sees a + // completion, both identical to a clean teardown, so + // this is the only record that the item was refused. + // The codec's own error chains to kilobytes, so it is + // deliberately not included. + tracing::warn!( + request_id = %self.request_id, + "refused a host subscription item: undecodable or nested past the limit" + ); self.stop(); Poll::Ready(None) } @@ -599,6 +615,91 @@ mod tests { assert!(frames[1].payload.value.is_empty()); } + #[test] + fn a_deeply_nested_host_item_is_refused_rather_than_exhausting_the_stack() { + // A recursive product-supplied payload decodes until the thread's stack + // is gone, and a stack overflow aborts the process rather than failing + // the call -- no `catch_unwind` and no `panic = "abort"` handling + // applies to it. The depth bound turns that into an ordinary refusal + // that ends this subscription alone. + let transport_typed = Arc::new(RecordingTransport::new()); + let transport: Arc = transport_typed.clone(); + let manager = HostInitiatedSubscriptionManager::new(); + let mut nested = manager.start::(host_ids(), vec![], transport.clone()); + let mut healthy = manager.start::(host_ids(), vec![], transport); + + // One `Deeper` byte per level, terminated by `Leaf`. + let mut bomb = vec![0x01; (MAX_SUBSCRIPTION_DECODE_DEPTH as usize) * 4]; + bomb.push(0x00); + manager.handle_message(ProtocolMessage { + request_id: "h:1".into(), + payload: Payload { + id: 55, + value: bomb, + }, + }); + assert_eq!(futures::executor::block_on(nested.next()), None); + + // A payload inside the bound still arrives, on its own subscription. + manager.handle_message(ProtocolMessage { + request_id: "h:2".into(), + payload: Payload { + id: 55, + value: NestedItem::Leaf.encode(), + }, + }); + assert_eq!( + futures::executor::block_on(healthy.next()), + Some(NestedItem::Leaf) + ); + } + + #[test] + fn the_depth_bound_lands_where_the_real_render_item_nests() { + // The fixture above recurses through `Box`, which uses a different + // `Decode` impl than the `Vec` the production type recurses + // through. Pin the boundary on the type actually decoded here. + fn nested(depth: u32) -> truapi::versioned::chat::ProductChatCustomMessageRenderItem { + let mut node = truapi::v01::CustomRendererNode::Nil; + for _ in 0..depth { + node = truapi::v01::CustomRendererNode::Box { + modifiers: Vec::new(), + props: truapi::v01::BoxProps { + content_alignment: None, + }, + children: vec![node], + }; + } + truapi::versioned::chat::ProductChatCustomMessageRenderItem::V1(node) + } + + let decode = |depth: u32| { + let bytes = nested(depth).encode(); + let mut input = &bytes[..]; + truapi::versioned::chat::ProductChatCustomMessageRenderItem::decode_with_depth_limit( + MAX_SUBSCRIPTION_DECODE_DEPTH, + &mut input, + ) + .is_ok() + }; + + assert!( + decode(MAX_SUBSCRIPTION_DECODE_DEPTH), + "the limit must be usable" + ); + assert!( + !decode(MAX_SUBSCRIPTION_DECODE_DEPTH + 1), + "one past the limit must be refused" + ); + } + + /// Stands in for the recursive protocol payloads a product can supply. + #[derive(Debug, PartialEq, Eq, Encode, Decode)] + enum NestedItem { + Leaf, + Deeper(Box), + } + #[test] fn malformed_host_item_ends_only_its_render_instance() { let transport_typed = Arc::new(RecordingTransport::new());