From 2ff7c20d00a48785e3ab25a53b69ca3cffe422c7 Mon Sep 17 00:00:00 2001 From: Sergey Zhuravlev Date: Mon, 31 Aug 2026 15:57:55 +0200 Subject: [PATCH 1/3] feat(truapi): add localStorage.subscribe and worker pending operations RFC 0027. Two additions so a background worker can finish a multi-step task and coordinate with the app through storage: - localStorage.subscribe(key) streams a key's value on every change, within the product's own namespace. - worker.beginOperation() / worker.endOperation(id) declare a pending operation; the host keeps the worker running while any operation is open. endOperation is idempotent. Adds wire ids 198-205, the canonical trait surface, v01/versioned plumbing, dispatcher and bridge codegen, host-cli and native/wasm runtime wiring, and the iOS/Android/TS host bindings. Claude-Session: https://claude.ai/code/session_01BedPRKxjb918nanVEz1gpM --- .../kotlin/io/parity/truapi/TrUAPIHost.kt | 27 +++ ...torage-subscriptions-pending-operations.md | 189 ++++++++++++++++++ docs/rfcs/_index.md | 1 + .../Sources/TrUAPIHost/TrUAPIHost.swift | 33 +++ .../Sources/TrUAPIHost/truapi.swift | 141 +++++++++++++ .../Sources/TrUAPIHost/truapi_server.swift | 169 ++++++++++++++++ .../include/truapi_serverFFI.h | 47 +++++ js/packages/truapi-host/src/test-support.ts | 9 + .../src/web/create-worker-host-runtime.ts | 33 +++ .../src/web/worker-provider.test.ts | 44 +++- .../truapi-codegen/tests/golden/dispatcher.rs | 117 ++++++++++- .../tests/golden/host-callbacks-adapter.ts | 27 +++ .../tests/golden/host-callbacks.ts | 37 ++++ .../tests/golden/wasm_bridge.rs | 63 ++++++ .../truapi-codegen/tests/golden/wire_table.rs | 34 +++- .../tests/golden/worker-callbacks.ts | 26 ++- rust/crates/truapi-host-cli/src/platform.rs | 50 ++++- rust/crates/truapi-platform/src/lib.rs | 49 ++++- .../truapi-server/src/generated/dispatcher.rs | 125 +++++++++++- .../truapi-server/src/generated/wire_table.rs | 34 +++- rust/crates/truapi-server/src/native.rs | 142 ++++++++++++- rust/crates/truapi-server/src/runtime.rs | 88 +++++++- rust/crates/truapi-server/src/test_support.rs | 43 +++- .../src/wasm/generated_bridge.rs | 63 ++++++ rust/crates/truapi-server/tests/common/mod.rs | 28 ++- rust/crates/truapi/src/api.rs | 4 + rust/crates/truapi/src/api/local_storage.rs | 32 ++- rust/crates/truapi/src/api/worker.rs | 54 +++++ rust/crates/truapi/src/lib.rs | 21 +- rust/crates/truapi/src/v01.rs | 2 + rust/crates/truapi/src/v01/local_storage.rs | 15 ++ rust/crates/truapi/src/v01/worker.rs | 42 ++++ rust/crates/truapi/src/versioned.rs | 1 + .../truapi/src/versioned/local_storage.rs | 2 + rust/crates/truapi/src/versioned/worker.rs | 12 ++ 35 files changed, 1764 insertions(+), 40 deletions(-) create mode 100644 docs/rfcs/0027-storage-subscriptions-pending-operations.md create mode 100644 rust/crates/truapi/src/api/worker.rs create mode 100644 rust/crates/truapi/src/v01/worker.rs create mode 100644 rust/crates/truapi/src/versioned/worker.rs 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 b0e89c929..69509806e 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 @@ -354,6 +354,19 @@ interface HostBridge { @Throws(HostRejection::class) fun supportedChains(): HostChainSet = HostChainSet(network = "", chains = emptyList()) + /** + * Begin a pending operation for a product's worker, returning a + * host-assigned id. The host keeps the product's worker alive while it holds + * at least one open operation. [label] is a log/UI hint, empty when the + * product gave none. Default: no worker keep-alive. + */ + @Throws(HostRejection::class) + suspend fun beginOperation(productId: String, label: String): UInt = 0u + + /** End a pending operation. Idempotent: an unknown or already-ended id succeeds. */ + @Throws(HostRejection::class) + suspend fun endOperation(productId: String, id: UInt) {} + /** Product-scoped key-value storage for the Rust core. */ val storage: HostStorage @@ -495,6 +508,12 @@ private class HostCallbackAdapter(private val bridge: HostBridge) : HostCallback override fun localStorageClear(key: String) = withStorageException { bridge.storage.clear(key) } + + override suspend fun beginOperation(productId: String, label: String): UInt = + withHostRejection { bridge.beginOperation(productId, label) } + + override suspend fun endOperation(productId: String, id: UInt) = + withHostRejection { bridge.endOperation(productId, id) } } // A host that throws an exception type its callback does not declare crosses @@ -944,6 +963,14 @@ class TrUAPIProductExecution internal constructor( inner.notifyLocaleChanged(locale) } + /** + * Push a host storage change for [key] to active TrUAPI storage + * subscriptions. A null [value] represents a cleared or absent key. + */ + fun notifyStorageChanged(key: String, value: ByteArray?) { + inner.notifyStorageChanged(key, value) + } + /** Push a preimage lookup update to active subscriptions for [key]. */ fun notifyPreimageChanged(key: ByteArray, value: ByteArray?) { inner.notifyPreimageChanged(key, value) diff --git a/docs/rfcs/0027-storage-subscriptions-pending-operations.md b/docs/rfcs/0027-storage-subscriptions-pending-operations.md new file mode 100644 index 000000000..ff65146d2 --- /dev/null +++ b/docs/rfcs/0027-storage-subscriptions-pending-operations.md @@ -0,0 +1,189 @@ +--- +title: "Product storage subscriptions and worker pending operations" +owner: "Sergey Zhuravlev" +--- + +# RFC 0027: Product storage subscriptions and worker pending operations + +| | | +| --------------- | ---------------------------------------- | +| **RFC Number** | 27 | +| **Start Date** | 2026-08-25 | +| **Description** | Two small TrUAPI additions so a background worker can finish a multi-step task and coordinate with the app through storage. | +| **Authors** | Sergey Zhuravlev | + +## Summary + +Two additions to TrUAPI: + +- `localStorage.subscribe(key)` streams a key's value on every change, within the product's own namespace. +- `worker.beginOperation()` / `worker.endOperation(id)` declare a pending operation; the host keeps the worker running while any operation is open. + +Both come from one flow: a funding operation, part of a safety-net release, runs in a worker, submits a transaction, and needs to finish and report progress even after the user leaves the app. + +## Motivation + +The product runs a funding operation, one part of a safety-net release. It builds a transaction, submits it, waits for it to be included, and records the result. Some of those steps hit a backend, so one operation can run for tens of seconds with polling in between. It runs in a worker so it continues after the user leaves the product's screen. Two things make that unsafe today. + +The worker dies when the user leaves. The host disposes a worker once its on-screen surface is gone, and the in-flight submission is aborted with it (the worker's own `dispose` is a no-op that defers to the main thread, `worker-runtime.ts:182`). Being killed between submitting a funding transaction and confirming it is the worst place to stop, and the product has no way to tell the host it is mid-operation. + +The UI and the worker can't see each other's progress. The on-screen product and the worker are separate runtimes over one storage namespace, but a write in one stays invisible to the other until it re-reads. So a progress view the worker feeds, or a worker that should react to what the user just did on screen, can only re-read on a timer. For a value that changes a few times a minute, that polling is both late and wasteful. + +## Detailed design + +### localStorage.subscribe + +A subscription method on the `LocalStorage` trait, next to `read` (12), `write` (14), `clear` (16): + +```rust +/// Subscribe to changes of one key in the product's own storage namespace. +/// +/// Emits the current value immediately, then one item per later change. +#[wire(start_id = 198)] // exact id assigned at implementation, free range above 197 +async fn subscribe( + &self, + cx: &CallContext, + request: HostLocalStorageSubscribeRequest, // { key: String } +) -> Subscription; +``` + +```rust +pub struct HostLocalStorageChangeItem { + /// Value after the change. `Some` on write, `None` after clear. + pub value: Option>, +} +``` + +The wire side is nothing new. `theme.subscribe` and `chat.list_subscribe` already return `Subscription`, and the TS client already exposes them as RxJS observables. + +The host emits the changes. On web and old JS hosts the app and worker are separate WASM instances with separate `RuntimeServices`, so the core alone can't carry a change from one to the other. The host can: it sits above both instances and owns the store, so it sees every write to the namespace whoever made it. The core just forwards the host's stream to the subscriber. + +This adds one method to the host `ProductStorage` trait, the same shape as the existing `ChatPlatform::subscribe_chat_rooms` (`rust/crates/truapi-platform/src/lib.rs`): + +```rust +/// Emit a product-scoped key's current value, then each later change, +/// from any of the product's runtimes. A write that doesn't change the +/// bytes emits nothing (see below). +fn subscribe_storage( + &self, + product: &ProductContext, + key: String, +) -> BoxStream<'static, Result>; +``` + +The core passes the calling product's context to `subscribe_storage`, the same scoping `read` and `write` already use, so a product only ever sees its own keys. The first item is the current value, so there's no read-then-subscribe gap. After that, a write emits `Some(value)` and a clear emits `None`. This works the same on web and native, because the source is the host, not a shared core instance. + +### Byte-identical writes do nothing + +If `write` gets the same bytes the key already holds, the host skips the store write and the change event. Same for `clear` on an absent key. The host does this because it holds the current bytes to compare. So a runtime that rewrites unchanged state on a timer costs nothing and wakes no one. + +### Pending operations + +The worker begins a pending operation while it has work in flight and ends it when done. The host keeps the worker alive while any operation is open. + +A `begin`/`end` pair on a new `Worker` trait: + +```rust +/// Begin a pending operation. The worker is kept alive while it has at least +/// one open operation. Returns an id for `end_operation`. +/// +/// Worker execution kind only. +#[wire(request_id = 202)] // exact id assigned at implementation +async fn begin_operation( + &self, + cx: &CallContext, + request: HostBeginOperationRequest, // { label: Option } for host UI/logs +) -> Result>; + +/// End a pending operation. Idempotent: an unknown or already-ended id +/// returns `Ok`. +#[wire(request_id = 204)] // exact id assigned at implementation +async fn end_operation( + &self, + cx: &CallContext, + request: HostEndOperationRequest, // { id: OperationId } +) -> Result<(), CallError>; +``` + +```rust +/// Opaque host-assigned operation identifier, unique per product. Mirrors +/// `NotificationId`, which is a `u32` type alias. +pub type OperationId = u32; + +pub struct HostBeginOperationResponse { + /// Pass this to `end_operation`. + pub id: OperationId, +} + +/// Domain error for the operation methods. +pub enum HostOperationError { + /// The product already holds the host's per-product cap of open + /// operations. `end` never returns this; it is idempotent and always + /// succeeds. + TooManyOpen, +} +``` + +Why operations and not a timer. A timer makes the worker guess the duration, and a short guess kills the transaction mid-flight. An operation ties liveness to the work itself: alive while something is open, gone when it closes. The id is session-scoped, not something the worker persists. If the host dies, the worker and the id die with it, and the leftover operation record is reconciled on the next launch (until a reaper exists, see open questions). So losing the id costs nothing. + +The host owns the operations and the lifecycle. `begin_operation` and `end_operation` are thin: the core forwards each to a host platform trait, scoped to the calling product. The host stores the product's open operations and keeps its worker alive while any stand. There's no separate keep-alive signal, because the operation existing is the signal. + +```rust +/// Host store for a product's pending operations. The host keeps the +/// product's worker alive while it holds at least one open operation. +/// Optional: a host that omits it answers `begin_operation` `Unsupported`. +#[async_trait] +pub trait ProductOperations: Send + Sync { + /// Record a pending operation for this product. Returns its id. + async fn begin_operation( + &self, + product: &ProductContext, + label: Option, + ) -> Result; + + /// Remove a pending operation. Idempotent: an unknown or already-ended + /// id returns `Ok`. + async fn end_operation( + &self, + product: &ProductContext, + id: OperationId, + ) -> Result<(), GenericError>; +} +``` + +Ref-counted and product-scoped. Two tasks each begin an operation, and the worker stays alive until both end. The count belongs to the product, like its storage, so any open operation holds the product's worker whichever runtime opened it. + +Best-effort is the ceiling. On iOS and Android the OS can kill a backgrounded worker whatever the host does. An open operation lets the host ask for what background time the platform allows (a background task assertion on iOS, a foreground service or WorkManager on Android), but the worker still has to resume from saved state after a kill. Operations lower the odds of a mid-flight teardown. They don't remove it. + +Kept generic on purpose. An operation is opaque: an optional label for logs and a future host UI, no funding or deposit typing. It's a plain liveness signal the host can build on later, not a funding session (see future directions). A `status` field and a `list_operations` read belong to that later UI, not v1. v1 is begin and end. + +## Drawbacks + +Both features cost host work. Each of the three hosts (web worker, iOS, Android) implements `subscribe_storage` plus the identical-write skip, and `ProductOperations` with the worker lifecycle tied to it. The operations side is the awkward one, since keeping a process alive is an OS concern with no core-only answer. `subscribe_storage` is cheaper and reuses the `subscribe_chat_rooms` shape a host has likely written already. + +An open operation keeps a WASM instance resident, which costs battery. Best-effort teardown is the only guardrail in v1: a worker that never ends an operation pins itself, and with one product that's acceptable. The reaper that reclaims a stuck operation is deferred (see open questions). + +## Security and privacy + +The subscription's only new risk is scope leakage, and the core blocks it the same way `read` and `write` do: it passes the calling `ProductContext` to `subscribe_storage`, so there's no way to name another product's key. `begin_operation` and `end_operation` are gated to the `Worker` kind, like the Chat modality, so an app or widget can't call them, and an id from one worker means nothing to another. + +Neither feature moves new data across a boundary. The subscription carries values the product already owns. The operation `label`, if a host shows it, is product text and should be bounded and screened like any other. + +## Testing + +The subscription tests against a fake `ProductStorage`: subscribe, check the initial value, write and check the item, write identical bytes and check nothing fires, clear and check `None`, and check product A never sees product B's writes. The operation flow tests against a fake `ProductOperations`: `begin_operation` and `end_operation` reach the host scoped to the calling product, ending an unknown id is `Ok`, and product A can't end product B's operation. Whether an open operation actually keeps the worker resident is host behavior and needs a real device. + +## Compatibility + +All three wire methods are additive and break nothing (`localStorage.subscribe` at 198, `worker.beginOperation` at 202, `worker.endOperation` at 204). On the host side, `subscribe_storage` lands on the required `ProductStorage` trait and `ProductOperations` is a required capability too, so every host implements both. The byte-identical skip is in the core, not the host, so every host inherits it. Target is v0.2 / latest. + +## Unresolved questions + +- Reaping stuck operations. What reclaims an operation a worker opened and never ended? A time cap, a count cap, or the user cancelling it through a future UI. Deferred for v1 since one product isn't critical, but it has to exist before this is load-bearing for many products. +- Rapid distinct writes. Identical writes already drop. For a burst of different values on one key, does the host emit each or only the latest? Emitting each is the literal contract; coalescing saves a progress-bar consumer work. Either way, state it in the trait doc. + +## Future directions + +Operations grow into a general liveness rule. The worker stays alive while `can_execute` holds, and pending operations are one term: `can_execute = has pending operations || has chats || has pocket cards || ...`. This RFC ships the first term. + +Around that, a host UI listing active operations, with `status` on each and a `list_operations` read to feed it, and a user cancel that ends the operation and releases the worker. A prefix subscription instead of a single storage key, if products want to watch a set at once. All are out of scope here and none needs a wire break later. diff --git a/docs/rfcs/_index.md b/docs/rfcs/_index.md index daa061f95..1e31e769d 100644 --- a/docs/rfcs/_index.md +++ b/docs/rfcs/_index.md @@ -26,3 +26,4 @@ created: 2026-03-13 | 0022 | [Account key derivations](0022-account-derivations.md) | draft | Valentin Sergeev | — | | 0023 | [sr25519 VRF signing for product accounts](0023-account-sign-vrf.md) | draft | Valentin Sergeev | — | | 0026 | [Host chain discovery and name resolution](0026-supported-chains.md) | draft | Valentin Fernandez | [#354](https://github.com/paritytech/host-rust-core/pull/354) | +| 0027 | [Product storage subscriptions and worker pending operations](0027-storage-subscriptions-pending-operations.md) | draft | Sergey Zhuravlev | — | diff --git a/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift b/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift index 43456bfab..66f2af10a 100644 --- a/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift +++ b/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift @@ -333,6 +333,16 @@ public protocol HostBridge: AnyObject, Sendable { /// per chain role. Invoked on the dispatcher thread; must return promptly. func supportedChains() throws -> HostChainSet + /// Begin a pending operation for a product's worker, returning a + /// host-assigned id. The host keeps the product's worker alive while it + /// holds at least one open operation. `label` is a log/UI hint, empty when + /// the product gave none. + func beginOperation(productId: String, label: String) async throws -> UInt32 + + /// End a pending operation. Idempotent: an unknown or already-ended id + /// succeeds. + func endOperation(productId: String, id: UInt32) async throws + /// Scoped key-value storage for the Rust core. var storage: HostStorageBackend { get } @@ -404,6 +414,10 @@ public extension HostBridge { func supportedChains() throws -> HostChainSet { HostChainSet(network: "", chains: []) } func devicePermissionStatus(request: HostDevicePermissionRequest) async throws -> NativeDevicePermissionStatus { .notApplicable } + /// Default: no worker keep-alive. Override to run background operations to + /// completion after the product's surface goes away. + func beginOperation(productId: String, label: String) async throws -> UInt32 { 0 } + func endOperation(productId: String, id: UInt32) async throws {} } /// Adapter that bridges the public `ChatHostBridge` to the generated UniFFI @@ -602,6 +616,18 @@ private final class HostCallbackAdapter: HostCallbacks, @unchecked Sendable { } } + func beginOperation(productId: String, label: String) async throws -> UInt32 { + try await withHostRejection { + try await bridge.beginOperation(productId: productId, label: label) + } + } + + func endOperation(productId: String, id: UInt32) async throws { + try await withHostRejection { + try await bridge.endOperation(productId: productId, id: id) + } + } + private func withHostRejection(_ operation: () throws -> T) throws -> T { do { return try operation() @@ -819,6 +845,7 @@ public protocol TrUAPIProductExecutionProtocol: AnyObject, Sendable { ) throws func notifyThemeChanged(theme: HostThemeSubscribeItem) func notifyLocaleChanged(locale: HostLocaleSubscribeItem) + func notifyStorageChanged(key: String, value: Data?) func notifyPreimageChanged(key: Data, value: Data?) func notifyChainResponse(connectionId: UInt32, json: String) func notifyChainClosed(connectionId: UInt32) @@ -898,6 +925,12 @@ public final class TrUAPIProductExecution: TrUAPIProductExecutionProtocol, @unch inner.notifyLocaleChanged(locale: locale) } + /// Push a host storage change for `key` to active TrUAPI storage + /// subscriptions. `value == nil` represents a cleared or absent key. + public func notifyStorageChanged(key: String, value: Data?) { + inner.notifyStorageChanged(key: key, value: value) + } + public func notifyPreimageChanged(key: Data, value: Data?) { inner.notifyPreimageChanged(key: key, value: value) } diff --git a/ios/truapi-host/Sources/TrUAPIHost/truapi.swift b/ios/truapi-host/Sources/TrUAPIHost/truapi.swift index 899bb6904..3b39c9a68 100644 --- a/ios/truapi-host/Sources/TrUAPIHost/truapi.swift +++ b/ios/truapi-host/Sources/TrUAPIHost/truapi.swift @@ -1955,6 +1955,65 @@ public func FfiConverterTypeHostChatActionSubscribeItem_lower(_ value: HostChatA } +/** + * A change to a subscribed storage key, pushed to the subscriber. + */ +public struct HostLocalStorageChangeItem: Equatable, Hashable { + /** + * Value after the change. `Some` on write, `None` after clear. + */ + public var value: Data? + + // Default memberwise initializers are never public by default, so we + // declare one manually. + public init( + /** + * Value after the change. `Some` on write, `None` after clear. + */value: Data?) { + self.value = value + } + + + + +} + +#if compiler(>=6) +extension HostLocalStorageChangeItem: Sendable {} +#endif + +#if swift(>=5.8) +@_documentation(visibility: private) +#endif +public struct FfiConverterTypeHostLocalStorageChangeItem: FfiConverterRustBuffer { + public static func read(from buf: inout (data: Data, offset: Data.Index)) throws -> HostLocalStorageChangeItem { + return + try HostLocalStorageChangeItem( + value: FfiConverterOptionData.read(from: &buf) + ) + } + + public static func write(_ value: HostLocalStorageChangeItem, into buf: inout [UInt8]) { + FfiConverterOptionData.write(value.value, into: &buf) + } +} + + +#if swift(>=5.8) +@_documentation(visibility: private) +#endif +public func FfiConverterTypeHostLocalStorageChangeItem_lift(_ buf: RustBuffer) throws -> HostLocalStorageChangeItem { + return try FfiConverterTypeHostLocalStorageChangeItem.lift(buf) +} + +#if swift(>=5.8) +@_documentation(visibility: private) +#endif +public func FfiConverterTypeHostLocalStorageChangeItem_lower(_ value: HostLocalStorageChangeItem) -> RustBuffer { + return FfiConverterTypeHostLocalStorageChangeItem.lower(value) +} + + /** * Locale the host currently presents its interface in, pushed to subscribers. */ @@ -5675,6 +5734,88 @@ public func FfiConverterTypeHostPlatform_lower(_ value: HostPlatform) -> RustBuf +/** + * Pending-operation error. + */ + +public enum HostWorkerOperationError: Equatable, Hashable { + + /** + * The product is already at the host's per-product limit of open + * operations. + */ + case tooManyOpen + /** + * Catch-all host failure. + */ + case unknown( + /** + * Human-readable failure reason. + */reason: String + ) + + + + + +} + +#if compiler(>=6) +extension HostWorkerOperationError: Sendable {} +#endif + +#if swift(>=5.8) +@_documentation(visibility: private) +#endif +public struct FfiConverterTypeHostWorkerOperationError: FfiConverterRustBuffer { + typealias SwiftType = HostWorkerOperationError + + public static func read(from buf: inout (data: Data, offset: Data.Index)) throws -> HostWorkerOperationError { + let variant: Int32 = try readInt(&buf) + switch variant { + + case 1: return .tooManyOpen + + case 2: return .unknown(reason: try FfiConverterString.read(from: &buf) + ) + + default: throw UniffiInternalError.unexpectedEnumCase + } + } + + public static func write(_ value: HostWorkerOperationError, into buf: inout [UInt8]) { + switch value { + + + case .tooManyOpen: + writeInt(&buf, Int32(1)) + + + case let .unknown(reason): + writeInt(&buf, Int32(2)) + FfiConverterString.write(reason, into: &buf) + + } + } +} + + +#if swift(>=5.8) +@_documentation(visibility: private) +#endif +public func FfiConverterTypeHostWorkerOperationError_lift(_ buf: RustBuffer) throws -> HostWorkerOperationError { + return try FfiConverterTypeHostWorkerOperationError.lift(buf) +} + +#if swift(>=5.8) +@_documentation(visibility: private) +#endif +public func FfiConverterTypeHostWorkerOperationError_lower(_ value: HostWorkerOperationError) -> RustBuffer { + return FfiConverterTypeHostWorkerOperationError.lower(value) +} + + + /** * Layout and styling modifiers applied to custom renderer components. */ diff --git a/ios/truapi-host/Sources/TrUAPIHost/truapi_server.swift b/ios/truapi-host/Sources/TrUAPIHost/truapi_server.swift index a54cc1a3e..a22e7e6b5 100644 --- a/ios/truapi-host/Sources/TrUAPIHost/truapi_server.swift +++ b/ios/truapi-host/Sources/TrUAPIHost/truapi_server.swift @@ -796,6 +796,19 @@ public protocol HostCallbacks: AnyObject, Sendable { */ func localStorageClear(key: String) throws + /** + * Record a pending operation for a product's worker, returning a + * host-assigned id. The host keeps the product's worker alive while it + * holds at least one open operation. + */ + func beginOperation(productId: String, label: String) async throws -> UInt32 + + /** + * End a pending operation. Idempotent: an unknown or already-ended id + * succeeds. + */ + func endOperation(productId: String, id: UInt32) async throws + } /** * Callback surface that iOS and Android implement. @@ -1233,6 +1246,47 @@ open func localStorageClear(key: String)throws {try rustCallWithError(FfiConve } } + /** + * Record a pending operation for a product's worker, returning a + * host-assigned id. The host keeps the product's worker alive while it + * holds at least one open operation. + */ +open func beginOperation(productId: String, label: String)async throws -> UInt32 { + return + try await uniffiRustCallAsync( + rustFutureFunc: { + uniffi_truapi_server_fn_method_hostcallbacks_begin_operation( + self.uniffiCloneHandle(),FfiConverterString.lower(productId),FfiConverterString.lower(label) + ) + }, + pollFunc: ffi_truapi_server_rust_future_poll_u32, + completeFunc: ffi_truapi_server_rust_future_complete_u32, + freeFunc: ffi_truapi_server_rust_future_free_u32, + liftFunc: FfiConverterUInt32.lift, + errorHandler: FfiConverterTypeHostRejection_lift + ) +} + + /** + * End a pending operation. Idempotent: an unknown or already-ended id + * succeeds. + */ +open func endOperation(productId: String, id: UInt32)async throws { + return + try await uniffiRustCallAsync( + rustFutureFunc: { + uniffi_truapi_server_fn_method_hostcallbacks_end_operation( + self.uniffiCloneHandle(),FfiConverterString.lower(productId),FfiConverterUInt32.lower(id) + ) + }, + pollFunc: ffi_truapi_server_rust_future_poll_void, + completeFunc: ffi_truapi_server_rust_future_complete_void, + freeFunc: ffi_truapi_server_rust_future_free_void, + liftFunc: { $0 }, + errorHandler: FfiConverterTypeHostRejection_lift + ) +} + } @@ -1977,6 +2031,94 @@ fileprivate struct UniffiCallbackInterfaceHostCallbacks { writeReturn: writeReturn, lowerError: FfiConverterTypeHostStorageError_lower ) + }, + beginOperation: { ( + uniffiHandle: UInt64, + productId: RustBuffer, + label: RustBuffer, + uniffiFutureCallback: @escaping UniffiForeignFutureCompleteU32, + uniffiCallbackData: UInt64, + uniffiOutDroppedCallback: UnsafeMutablePointer + ) in + let makeCall = { + () async throws -> UInt32 in + guard let uniffiObj = try? FfiConverterTypeHostCallbacks.handleMap.get(handle: uniffiHandle) else { + throw UniffiInternalError.unexpectedStaleHandle + } + return try await uniffiObj.beginOperation( + productId: try FfiConverterString.lift(productId), + label: try FfiConverterString.lift(label) + ) + } + + let uniffiHandleSuccess = { (returnValue: UInt32) in + uniffiFutureCallback( + uniffiCallbackData, + UniffiForeignFutureResultU32( + returnValue: FfiConverterUInt32.lower(returnValue), + callStatus: RustCallStatus() + ) + ) + } + let uniffiHandleError = { (statusCode, errorBuf) in + uniffiFutureCallback( + uniffiCallbackData, + UniffiForeignFutureResultU32( + returnValue: 0, + callStatus: RustCallStatus(code: statusCode, errorBuf: errorBuf) + ) + ) + } + uniffiTraitInterfaceCallAsyncWithError( + makeCall: makeCall, + handleSuccess: uniffiHandleSuccess, + handleError: uniffiHandleError, + lowerError: FfiConverterTypeHostRejection_lower, + droppedCallback: uniffiOutDroppedCallback + ) + }, + endOperation: { ( + uniffiHandle: UInt64, + productId: RustBuffer, + id: UInt32, + uniffiFutureCallback: @escaping UniffiForeignFutureCompleteVoid, + uniffiCallbackData: UInt64, + uniffiOutDroppedCallback: UnsafeMutablePointer + ) in + let makeCall = { + () async throws -> () in + guard let uniffiObj = try? FfiConverterTypeHostCallbacks.handleMap.get(handle: uniffiHandle) else { + throw UniffiInternalError.unexpectedStaleHandle + } + return try await uniffiObj.endOperation( + productId: try FfiConverterString.lift(productId), + id: try FfiConverterUInt32.lift(id) + ) + } + + let uniffiHandleSuccess = { (returnValue: ()) in + uniffiFutureCallback( + uniffiCallbackData, + UniffiForeignFutureResultVoid( + callStatus: RustCallStatus() + ) + ) + } + let uniffiHandleError = { (statusCode, errorBuf) in + uniffiFutureCallback( + uniffiCallbackData, + UniffiForeignFutureResultVoid( + callStatus: RustCallStatus(code: statusCode, errorBuf: errorBuf) + ) + ) + } + uniffiTraitInterfaceCallAsyncWithError( + makeCall: makeCall, + handleSuccess: uniffiHandleSuccess, + handleError: uniffiHandleError, + lowerError: FfiConverterTypeHostRejection_lower, + droppedCallback: uniffiOutDroppedCallback + ) } ) @@ -2594,6 +2736,11 @@ public protocol NativeProductExecutionProtocol: AnyObject, Sendable { */ func notifyPreimageChanged(key: Data, value: Data?) + /** + * Push a host storage change to this execution's subscriptions for `key`. + */ + func notifyStorageChanged(key: String, value: Data?) + /** * Push a host theme replacement to this execution's subscriptions. */ @@ -2788,6 +2935,19 @@ open func notifyPreimageChanged(key: Data, value: Data?) {try! rustCall() { FfiConverterOptionData.lower(value),uniffiCallStatus ) } +} + + /** + * Push a host storage change to this execution's subscriptions for `key`. + */ +open func notifyStorageChanged(key: String, value: Data?) {try! rustCall() { + uniffiCallStatus in + uniffi_truapi_server_fn_method_nativeproductexecution_notify_storage_changed( + self.uniffiCloneHandle(), + FfiConverterString.lower(key), + FfiConverterOptionData.lower(value),uniffiCallStatus + ) +} } /** @@ -5944,6 +6104,12 @@ private let initializationResult: InitializationResult = { if (uniffi_truapi_server_checksum_method_hostcallbacks_local_storage_clear() != 64902) { return InitializationResult.apiChecksumMismatch } + if (uniffi_truapi_server_checksum_method_hostcallbacks_begin_operation() != 49135) { + return InitializationResult.apiChecksumMismatch + } + if (uniffi_truapi_server_checksum_method_hostcallbacks_end_operation() != 25056) { + return InitializationResult.apiChecksumMismatch + } if (uniffi_truapi_server_checksum_method_nativechatcallbacks_create_room() != 15676) { return InitializationResult.apiChecksumMismatch } @@ -5974,6 +6140,9 @@ private let initializationResult: InitializationResult = { if (uniffi_truapi_server_checksum_method_nativeproductexecution_notify_preimage_changed() != 21769) { return InitializationResult.apiChecksumMismatch } + if (uniffi_truapi_server_checksum_method_nativeproductexecution_notify_storage_changed() != 47215) { + return InitializationResult.apiChecksumMismatch + } if (uniffi_truapi_server_checksum_method_nativeproductexecution_notify_theme_changed() != 3284) { return InitializationResult.apiChecksumMismatch } diff --git a/ios/truapi-host/Sources/truapi_serverFFI/include/truapi_serverFFI.h b/ios/truapi-host/Sources/truapi_serverFFI/include/truapi_serverFFI.h index 6595a1494..0382b3754 100644 --- a/ios/truapi-host/Sources/truapi_serverFFI/include/truapi_serverFFI.h +++ b/ios/truapi-host/Sources/truapi_serverFFI/include/truapi_serverFFI.h @@ -416,6 +416,18 @@ typedef void (*UniffiCallbackInterfaceHostCallbacksMethod22)(uint64_t, RustBuffe RustCallStatus *_Nonnull uniffiCallStatus ); +#endif +#ifndef UNIFFI_FFIDEF_CALLBACK_INTERFACE_HOST_CALLBACKS_METHOD23 +#define UNIFFI_FFIDEF_CALLBACK_INTERFACE_HOST_CALLBACKS_METHOD23 +typedef void (*UniffiCallbackInterfaceHostCallbacksMethod23)(uint64_t, RustBuffer, RustBuffer, UniffiForeignFutureCompleteU32 _Nonnull, uint64_t, UniffiForeignFutureDroppedCallbackStruct* _Nonnull + ); + +#endif +#ifndef UNIFFI_FFIDEF_CALLBACK_INTERFACE_HOST_CALLBACKS_METHOD24 +#define UNIFFI_FFIDEF_CALLBACK_INTERFACE_HOST_CALLBACKS_METHOD24 +typedef void (*UniffiCallbackInterfaceHostCallbacksMethod24)(uint64_t, RustBuffer, uint32_t, UniffiForeignFutureCompleteVoid _Nonnull, uint64_t, UniffiForeignFutureDroppedCallbackStruct* _Nonnull + ); + #endif #ifndef UNIFFI_FFIDEF_CALLBACK_INTERFACE_NATIVE_CHAT_CALLBACKS_METHOD0 #define UNIFFI_FFIDEF_CALLBACK_INTERFACE_NATIVE_CHAT_CALLBACKS_METHOD0 @@ -484,6 +496,8 @@ typedef struct UniffiVTableCallbackInterfaceHostCallbacks { UniffiCallbackInterfaceHostCallbacksMethod20 _Nonnull localStorageRead; UniffiCallbackInterfaceHostCallbacksMethod21 _Nonnull localStorageWrite; UniffiCallbackInterfaceHostCallbacksMethod22 _Nonnull localStorageClear; + UniffiCallbackInterfaceHostCallbacksMethod23 _Nonnull beginOperation; + UniffiCallbackInterfaceHostCallbacksMethod24 _Nonnull endOperation; } UniffiVTableCallbackInterfaceHostCallbacks; #endif @@ -629,6 +643,16 @@ void uniffi_truapi_server_fn_method_hostcallbacks_local_storage_write(uint64_t p void uniffi_truapi_server_fn_method_hostcallbacks_local_storage_clear(uint64_t ptr, RustBuffer key, RustCallStatus *_Nonnull out_status ); #endif +#ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_METHOD_HOSTCALLBACKS_BEGIN_OPERATION +#define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_METHOD_HOSTCALLBACKS_BEGIN_OPERATION +uint64_t uniffi_truapi_server_fn_method_hostcallbacks_begin_operation(uint64_t ptr, RustBuffer product_id, RustBuffer label +); +#endif +#ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_METHOD_HOSTCALLBACKS_END_OPERATION +#define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_METHOD_HOSTCALLBACKS_END_OPERATION +uint64_t uniffi_truapi_server_fn_method_hostcallbacks_end_operation(uint64_t ptr, RustBuffer product_id, uint32_t id +); +#endif #ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_CLONE_NATIVECHATCALLBACKS #define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_CLONE_NATIVECHATCALLBACKS uint64_t uniffi_truapi_server_fn_clone_nativechatcallbacks(uint64_t handle, RustCallStatus *_Nonnull out_status @@ -704,6 +728,11 @@ void uniffi_truapi_server_fn_method_nativeproductexecution_notify_locale_changed void uniffi_truapi_server_fn_method_nativeproductexecution_notify_preimage_changed(uint64_t ptr, RustBuffer key, RustBuffer value, RustCallStatus *_Nonnull out_status ); #endif +#ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_METHOD_NATIVEPRODUCTEXECUTION_NOTIFY_STORAGE_CHANGED +#define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_METHOD_NATIVEPRODUCTEXECUTION_NOTIFY_STORAGE_CHANGED +void uniffi_truapi_server_fn_method_nativeproductexecution_notify_storage_changed(uint64_t ptr, RustBuffer key, RustBuffer value, RustCallStatus *_Nonnull out_status +); +#endif #ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_METHOD_NATIVEPRODUCTEXECUTION_NOTIFY_THEME_CHANGED #define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_FN_METHOD_NATIVEPRODUCTEXECUTION_NOTIFY_THEME_CHANGED void uniffi_truapi_server_fn_method_nativeproductexecution_notify_theme_changed(uint64_t ptr, RustBuffer theme, RustCallStatus *_Nonnull out_status @@ -1267,6 +1296,18 @@ uint16_t uniffi_truapi_server_checksum_method_hostcallbacks_local_storage_write( #define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_HOSTCALLBACKS_LOCAL_STORAGE_CLEAR uint16_t uniffi_truapi_server_checksum_method_hostcallbacks_local_storage_clear(void +); +#endif +#ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_HOSTCALLBACKS_BEGIN_OPERATION +#define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_HOSTCALLBACKS_BEGIN_OPERATION +uint16_t uniffi_truapi_server_checksum_method_hostcallbacks_begin_operation(void + +); +#endif +#ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_HOSTCALLBACKS_END_OPERATION +#define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_HOSTCALLBACKS_END_OPERATION +uint16_t uniffi_truapi_server_checksum_method_hostcallbacks_end_operation(void + ); #endif #ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_NATIVECHATCALLBACKS_CREATE_ROOM @@ -1327,6 +1368,12 @@ uint16_t uniffi_truapi_server_checksum_method_nativeproductexecution_notify_loca #define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_NATIVEPRODUCTEXECUTION_NOTIFY_PREIMAGE_CHANGED uint16_t uniffi_truapi_server_checksum_method_nativeproductexecution_notify_preimage_changed(void +); +#endif +#ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_NATIVEPRODUCTEXECUTION_NOTIFY_STORAGE_CHANGED +#define UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_NATIVEPRODUCTEXECUTION_NOTIFY_STORAGE_CHANGED +uint16_t uniffi_truapi_server_checksum_method_nativeproductexecution_notify_storage_changed(void + ); #endif #ifndef UNIFFI_FFIDEF_UNIFFI_TRUAPI_SERVER_CHECKSUM_METHOD_NATIVEPRODUCTEXECUTION_NOTIFY_THEME_CHANGED diff --git a/js/packages/truapi-host/src/test-support.ts b/js/packages/truapi-host/src/test-support.ts index d2ca60740..1514f1fc5 100644 --- a/js/packages/truapi-host/src/test-support.ts +++ b/js/packages/truapi-host/src/test-support.ts @@ -29,6 +29,11 @@ export function makeHostCallbacks( read: async () => undefined, write: async () => {}, clear: async () => {}, + async *subscribeStorage() {}, + }, + productOperations: { + beginOperation: async () => ({ id: 1 }), + endOperation: async () => {}, }, coreStorage: { readCoreStorage: async () => undefined, @@ -65,6 +70,10 @@ export function makeHostCallbacks( ...defaults.productStorage, ...overrides.productStorage, }, + productOperations: { + ...defaults.productOperations, + ...overrides.productOperations, + }, coreStorage: { ...defaults.coreStorage, ...overrides.coreStorage, diff --git a/js/packages/truapi-host/src/web/create-worker-host-runtime.ts b/js/packages/truapi-host/src/web/create-worker-host-runtime.ts index 451f0e541..c3bc69f32 100644 --- a/js/packages/truapi-host/src/web/create-worker-host-runtime.ts +++ b/js/packages/truapi-host/src/web/create-worker-host-runtime.ts @@ -118,6 +118,17 @@ interface RuntimeState { } >; subscriptionDisposers: Map void>; + /** + * Open worker pending operations (`worker.beginOperation`). While this is + * above zero the worker is kept alive: a `dispose()` is deferred until the + * last operation ends. Worker-global, not per-core, because a + * `callbackRequest` carries no core id and "keep the worker alive" is + * worker-scoped. ponytail: no cap on how long an operation may hold the + * worker; add a timeout ceiling here if a stuck operation becomes a problem. + */ + operationCount: number; + /** A dispose() arrived while operations were open; run it once they drain. */ + disposePending: boolean; chainConnections: Map; pendingDisconnects: Map< number, @@ -372,6 +383,18 @@ function handleCallbackRequest( .then(() => fn(...msg.args)) .then( (value) => { + // Keep the worker alive across an open pending operation: count begins + // and ends only on success, so a rejected begin never leaves a stuck + // count. When the last operation ends and a dispose is pending, run it. + if (msg.name === "beginOperation") { + state.operationCount += 1; + } else if (msg.name === "endOperation" && state.operationCount > 0) { + state.operationCount -= 1; + if (state.operationCount === 0 && state.disposePending) { + state.disposePending = false; + teardown(state, new Error("runtime disposed"), false); + } + } state.worker.postMessage({ kind: "callbackResponse", requestId: msg.requestId, @@ -781,6 +804,8 @@ export function createWebWorkerPairingHostRuntime( cores: new Map(), pendingCores: new Map(), subscriptionDisposers: new Map(), + operationCount: 0, + disposePending: false, chainConnections: new Map(), pendingDisconnects: new Map(), pendingSessionActivations: new Map(), @@ -1220,6 +1245,14 @@ function buildRuntime(state: RuntimeState): WorkerPairingHostRuntime { }, dispose(): void { devGlobalTargets.delete(runtime); + // Defer a clean dispose while the worker holds an open operation, so a + // background task (e.g. a funding transaction) runs to completion. The + // last endOperation runs the deferred teardown. Fault teardown is never + // deferred. + if (state.operationCount > 0) { + state.disposePending = true; + return; + } teardown(state, new Error("runtime disposed"), false); }, }; diff --git a/js/packages/truapi-host/src/web/worker-provider.test.ts b/js/packages/truapi-host/src/web/worker-provider.test.ts index bf8e6cabf..3e7379a24 100644 --- a/js/packages/truapi-host/src/web/worker-provider.test.ts +++ b/js/packages/truapi-host/src/web/worker-provider.test.ts @@ -10,7 +10,11 @@ import { bytesToHex } from "@parity/truapi/scale"; import type { GenericError, Result, ThemeVariant } from "@parity/truapi"; import { createWasmRawCallbacks } from "../generated/host-callbacks-adapter.js"; -import { AuthState, CoreStorageKey } from "../generated/host-callbacks.js"; +import { + AuthState, + CoreStorageKey, + ProductContext, +} from "../generated/host-callbacks.js"; import type { AuthState as AuthStateValue, PreimageHost, @@ -959,6 +963,44 @@ describe("createWebWorkerPairingHostRuntime", () => { provider.dispose(); }); + it("keeps the worker alive until pending operations end", async () => { + const worker = new FakeWorker(); + const provider = await readyProvider(worker, { + runtimeConfig: runtimeConfig({ executionKind: "Worker" }), + }); + + const product = ProductContext.enc({ + productId: "dotli.dot", + executionKind: "Worker", + }); + + // Open a pending operation. + worker.emit({ + kind: "callbackRequest", + requestId: 1, + name: "beginOperation", + args: [product, "funding"], + }); + await settle(); + + // Disposing while the operation is open defers the worker teardown. + provider.dispose(); + await settle(); + expect(worker.terminated).toBe(false); + + // Ending the operation runs the deferred teardown, which terminates the + // worker on a zero-delay timer. + worker.emit({ + kind: "callbackRequest", + requestId: 2, + name: "endOperation", + args: [product, 1], + }); + await settle(); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(worker.terminated).toBe(true); + }); + it("routes payload-carrying subscriptions by name", async () => { const worker = new FakeWorker(); const keys: Uint8Array[] = []; diff --git a/rust/crates/truapi-codegen/tests/golden/dispatcher.rs b/rust/crates/truapi-codegen/tests/golden/dispatcher.rs index 7b059b295..2129afbb7 100644 --- a/rust/crates/truapi-codegen/tests/golden/dispatcher.rs +++ b/rust/crates/truapi-codegen/tests/golden/dispatcher.rs @@ -29,6 +29,7 @@ use truapi::api::{ StatementStore, System, Theme, + Worker, }; use truapi::versioned::{self, Versioned}; use truapi_platform::ProductExecutionKind; @@ -63,7 +64,8 @@ where register_signing(dispatcher, host.clone()); register_statement_store(dispatcher, host.clone()); register_system(dispatcher, host.clone()); - register_theme(dispatcher, host); + register_theme(dispatcher, host.clone()); + register_worker(dispatcher, host); } /// Start the host-initiated `chat_custom_message_render` subscription. @@ -1626,7 +1628,7 @@ where }); } { - let host = host; + let host = host.clone(); dispatcher.on_request(wire_table::LOCAL_STORAGE_CLEAR, move |request_id: String, bytes: Vec| { let host = host.clone(); Box::pin(async move { @@ -1664,6 +1666,21 @@ where }) }); } + { + let host = host; + dispatcher.on_subscription(wire_table::LOCAL_STORAGE_SUBSCRIBE, move |request_id: String, bytes: Vec| { + let host = host.clone(); + Box::pin(async move { + let request: versioned::local_storage::HostLocalStorageSubscribeRequest = match Decode::decode(&mut &bytes[..]) { + Ok(request) => request, + Err(_) => return Err(Vec::new()), + }; + let cx = CallContext::with_request_id(request_id.clone()); + let stream = host.subscribe(&cx, request).await; + Ok(subscription_stream::(stream)) + }) + }); + } } fn register_locale

(dispatcher: &mut Dispatcher, host: Arc

) @@ -2700,3 +2717,99 @@ where }); } } + +fn register_worker

(dispatcher: &mut Dispatcher, host: Arc

) +where + P: Worker + Send + Sync + 'static, +{ + { + let execution_allowed = dispatcher.allows_execution(ProductExecutionKind::Worker); + let host = host.clone(); + dispatcher.on_request(wire_table::WORKER_BEGIN_OPERATION, move |request_id: String, bytes: Vec| { + let host = host.clone(); + Box::pin(async move { + let request: versioned::worker::HostWorkerBeginOperationRequest = match Decode::decode(&mut &bytes[..]) { + Ok(request) => request, + Err(err) => { + let error: truapi::CallError = + truapi::CallError::MalformedFrame { reason: err.to_string() }; + return Ok(encode_versioned_err_payload( + error, + ::LATEST, + )); + } + }; + let target_version = request.version(); + let cx = CallContext::with_request_id(request_id.clone()); + if !execution_allowed { + let error: truapi::CallError = + truapi::CallError::Denied; + return Ok(encode_versioned_err_payload(error, target_version)); + } + let response: versioned::worker::HostWorkerBeginOperationResponse = match host.begin_operation(&cx, request).await { + Ok(value) => value, + Err(err) => { + return Ok(encode_versioned_err_payload( + downgrade_call_error(err, target_version), + target_version, + )); + } + }; + // Downgraded to the caller's version: a handler answers in + // latest terms, and a peer that asked in an older version + // cannot decode a newer variant. + Ok(encode_versioned_ok_payload( + ::from_latest( + truapi::versioned::IntoLatest::into_latest(response), + target_version, + ), + )) + }) + }); + } + { + let execution_allowed = dispatcher.allows_execution(ProductExecutionKind::Worker); + let host = host; + dispatcher.on_request(wire_table::WORKER_END_OPERATION, move |request_id: String, bytes: Vec| { + let host = host.clone(); + Box::pin(async move { + let request: versioned::worker::HostWorkerEndOperationRequest = match Decode::decode(&mut &bytes[..]) { + Ok(request) => request, + Err(err) => { + let error: truapi::CallError = + truapi::CallError::MalformedFrame { reason: err.to_string() }; + return Ok(encode_versioned_err_payload( + error, + ::LATEST, + )); + } + }; + let target_version = request.version(); + let cx = CallContext::with_request_id(request_id.clone()); + if !execution_allowed { + let error: truapi::CallError = + truapi::CallError::Denied; + return Ok(encode_versioned_err_payload(error, target_version)); + } + let response: versioned::worker::HostWorkerEndOperationResponse = match host.end_operation(&cx, request).await { + Ok(value) => value, + Err(err) => { + return Ok(encode_versioned_err_payload( + downgrade_call_error(err, target_version), + target_version, + )); + } + }; + // Downgraded to the caller's version: a handler answers in + // latest terms, and a peer that asked in an older version + // cannot decode a newer variant. + Ok(encode_versioned_ok_payload( + ::from_latest( + truapi::versioned::IntoLatest::into_latest(response), + target_version, + ), + )) + }) + }); + } +} diff --git a/rust/crates/truapi-codegen/tests/golden/host-callbacks-adapter.ts b/rust/crates/truapi-codegen/tests/golden/host-callbacks-adapter.ts index be3d7e146..a3f25365f 100644 --- a/rust/crates/truapi-codegen/tests/golden/host-callbacks-adapter.ts +++ b/rust/crates/truapi-codegen/tests/golden/host-callbacks-adapter.ts @@ -17,10 +17,12 @@ import { HostDevicePermissionResponse, HostFeatureSupportedRequest, HostFeatureSupportedResponse, + HostLocalStorageChangeItem, HostLocaleSubscribeItem, HostPushNotificationRequest, HostPushNotificationResponse, HostThemeSubscribeItem, + HostWorkerBeginOperationResponse, RemotePermissionRequest, RemotePermissionResponse, } from "@parity/truapi"; @@ -83,9 +85,16 @@ export interface RawCallbacks { sendItem: (item?: Uint8Array) => void, sendError: (error: GenericError) => void, ): (() => void) | void; + beginOperation(product: Uint8Array, label: string): Promise; + endOperation(product: Uint8Array, id: number): Promise; read(key: string): Promise; write(key: string, value: Uint8Array): Promise; clear(key: string): Promise; + subscribeStorage( + key: Uint8Array, + sendItem: (item?: Uint8Array) => void, + sendError: (error: GenericError) => void, + ): (() => void) | void; subscribeTheme( sendItem: (item?: Uint8Array) => void, sendError: (error: GenericError) => void, @@ -194,10 +203,28 @@ export function createWasmRawCallbacks( sendItem, sendError, ), + beginOperation: async (product, label) => + HostWorkerBeginOperationResponse.enc( + await callbacks.productOperations.beginOperation( + ProductContext.dec(product), + label, + ), + ), + endOperation: async (product, id) => + await callbacks.productOperations.endOperation( + ProductContext.dec(product), + id, + ), read: async (key) => await callbacks.productStorage.read(key), write: async (key, value) => await callbacks.productStorage.write(key, value), clear: async (key) => await callbacks.productStorage.clear(key), + subscribeStorage: (key, sendItem, sendError) => + driveResultStream( + callbacks.productStorage.subscribeStorage(key), + (item) => sendItem(HostLocalStorageChangeItem.enc(item)), + sendError, + ), subscribeTheme: (sendItem, sendError) => driveResultStream( callbacks.theme.subscribeTheme(), diff --git a/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts b/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts index b9ccb6ee7..a4f588d47 100644 --- a/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts +++ b/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts @@ -36,10 +36,12 @@ import type { HostDevicePermissionResponse, HostFeatureSupportedRequest, HostFeatureSupportedResponse, + HostLocalStorageChangeItem, HostLocaleSubscribeItem, HostPushNotificationRequest, HostPushNotificationResponse, HostThemeSubscribeItem, + HostWorkerBeginOperationResponse, NotificationId, RemotePermissionResponse, Result, @@ -1276,6 +1278,29 @@ export interface PreimageHost { ): AsyncIterable>; } +/** + * Host store for a product's pending operations. The host keeps the product's + * worker runtime alive while it holds at least one open operation. Worker + * products reach these through the `Worker` protocol trait, which is gated to + * the Worker execution kind, so non-worker products never call them. + */ +export interface ProductOperations { + /** + * Record a pending operation for this product. Returns its id. `label` is + * a host log and UI hint, empty when the product gave none. + */ + beginOperation( + product: ProductContext, + label: string, + ): Promise; + + /** + * Remove a pending operation. Idempotent: an unknown or already-ended id + * returns `Ok`. + */ + endOperation(product: ProductContext, id: number): Promise; +} + /** * Product-scoped key-value storage. * @@ -1298,6 +1323,16 @@ export interface ProductStorage { * Clear a value at a key. */ clear(key: string): Promise; + + /** + * Emit the current value of a key, then each later change to it, from any + * of the product's runtimes. A write that leaves the stored bytes + * unchanged emits nothing. `key` is the UTF-8 bytes of the namespaced + * storage key, the same key passed to `Self::read` as a `String`. + */ + subscribeStorage( + key: Uint8Array, + ): AsyncIterable>; } /** @@ -1339,6 +1374,7 @@ export interface HostCallbacks { theme: ThemeHost; locale: LocaleHost; preimage: PreimageHost; + productOperations: ProductOperations; chat?: ChatPlatform; permissionStatus?: PermissionStatusHost; } @@ -1356,6 +1392,7 @@ export interface RequiredHostCallbacks { theme: Required; locale: Required; preimage: Required; + productOperations: Required; chat?: Required; permissionStatus?: Required; } diff --git a/rust/crates/truapi-codegen/tests/golden/wasm_bridge.rs b/rust/crates/truapi-codegen/tests/golden/wasm_bridge.rs index d1b23cd61..65e9fbcd1 100644 --- a/rust/crates/truapi-codegen/tests/golden/wasm_bridge.rs +++ b/rust/crates/truapi-codegen/tests/golden/wasm_bridge.rs @@ -45,9 +45,12 @@ pub(super) struct JsBridge { pub(super) device_permission: Function, pub(super) remote_permission: Function, pub(super) lookup_preimage: Function, + pub(super) begin_operation: Function, + pub(super) end_operation: Function, pub(super) read: Function, pub(super) write: Function, pub(super) clear: Function, + pub(super) subscribe_storage: Function, pub(super) subscribe_theme: Function, pub(super) confirm_user_action: Function, pub(super) chat_present: bool, @@ -81,9 +84,12 @@ impl JsBridge { device_permission: get_function(callbacks, "devicePermission")?, remote_permission: get_function(callbacks, "remotePermission")?, lookup_preimage: get_function(callbacks, "lookupPreimage")?, + begin_operation: get_function(callbacks, "beginOperation")?, + end_operation: get_function(callbacks, "endOperation")?, read: get_function(callbacks, "read")?, write: get_function(callbacks, "write")?, clear: get_function(callbacks, "clear")?, + subscribe_storage: get_function(callbacks, "subscribeStorage")?, subscribe_theme: get_function(callbacks, "subscribeTheme")?, confirm_user_action: get_function(callbacks, "confirmUserAction")?, chat_present: get_optional_function(callbacks, "createChatRoom")?.is_some() @@ -389,6 +395,46 @@ impl truapi_platform::PreimageHost for WasmPlatform { } } +#[truapi_platform::async_trait] +impl truapi_platform::ProductOperations for WasmPlatform { + async fn begin_operation( + &self, + product: &truapi_platform::ProductContext, + label: String, + ) -> Result { + let bytes = invoke_bytes_return( + &self.bridge.begin_operation, + vec![ + Uint8Array::from(product.encode().as_slice()).into(), + JsValue::from_str(&label), + ], + ) + .await + .map_err(|reason| v01::HostWorkerOperationError::Unknown { reason })?; + decode_bytes::( + bytes, + "beginOperation response did not decode", + ) + .map_err(|reason| v01::HostWorkerOperationError::Unknown { reason }) + } + + async fn end_operation( + &self, + product: &truapi_platform::ProductContext, + id: u32, + ) -> Result<(), v01::HostWorkerOperationError> { + invoke_unit( + &self.bridge.end_operation, + vec![ + Uint8Array::from(product.encode().as_slice()).into(), + JsValue::from_f64(f64::from(id)), + ], + ) + .await + .map_err(|reason| v01::HostWorkerOperationError::Unknown { reason }) + } +} + #[truapi_platform::async_trait] impl truapi_platform::ProductStorage for WasmPlatform { async fn read(&self, key: String) -> Result>, v01::HostLocalStorageReadError> { @@ -422,6 +468,17 @@ impl truapi_platform::ProductStorage for WasmPlatform { .await .map_err(|reason| v01::HostLocalStorageReadError::Unknown { reason }) } + + fn subscribe_storage( + &self, + key: Vec, + ) -> BoxStream<'static, Result> { + invoke_js_subscription( + &self.bridge.subscribe_storage, + Some(key), + parse_host_local_storage_change_item_item, + ) + } } impl truapi_platform::ThemeHost for WasmPlatform { @@ -457,6 +514,12 @@ fn parse_host_chat_list_subscribe_item_item( decode_js_item::(value, "HostChatListSubscribeItem") } +fn parse_host_local_storage_change_item_item( + value: JsValue, +) -> Result { + decode_js_item::(value, "HostLocalStorageChangeItem") +} + fn parse_host_locale_subscribe_item_item( value: JsValue, ) -> Result { diff --git a/rust/crates/truapi-codegen/tests/golden/wire_table.rs b/rust/crates/truapi-codegen/tests/golden/wire_table.rs index 5d59d7f93..4fbce8324 100644 --- a/rust/crates/truapi-codegen/tests/golden/wire_table.rs +++ b/rust/crates/truapi-codegen/tests/golden/wire_table.rs @@ -50,7 +50,7 @@ pub enum WireKind { /// `TRUAPI_WIRE_SCHEMA_HASH`. A host stamps it on each debug envelope so /// the debugger refuses to decode a frame whose contract differs from /// its own, even when the coarse handshake codec version is unchanged. -pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "0449982638d57658"; +pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "e141a8b071605ab4"; /// Wire discriminants for `system_handshake`. pub const SYSTEM_HANDSHAKE: RequestFrameIds = RequestFrameIds { @@ -516,6 +516,26 @@ pub const LOCALE_SUBSCRIBE: SubscriptionFrameIds = SubscriptionFrameIds { receive_id: 197, }; +/// Wire discriminants for `local_storage_subscribe`. +pub const LOCAL_STORAGE_SUBSCRIBE: SubscriptionFrameIds = SubscriptionFrameIds { + start_id: 198, + stop_id: 199, + interrupt_id: 200, + receive_id: 201, +}; + +/// Wire discriminants for `worker_begin_operation`. +pub const WORKER_BEGIN_OPERATION: RequestFrameIds = RequestFrameIds { + request_id: 202, + response_id: 203, +}; + +/// Wire discriminants for `worker_end_operation`. +pub const WORKER_END_OPERATION: RequestFrameIds = RequestFrameIds { + request_id: 204, + response_id: 205, +}; + /// The full wire table. Ordering is part of the wire protocol; /// only ever append. Removed methods leave their slot empty. pub const WIRE_TABLE: &[WireEntry] = &[ @@ -807,4 +827,16 @@ pub const WIRE_TABLE: &[WireEntry] = &[ method: "locale_subscribe", kind: WireKind::Subscription(LOCALE_SUBSCRIBE), }, + WireEntry { + method: "local_storage_subscribe", + kind: WireKind::Subscription(LOCAL_STORAGE_SUBSCRIBE), + }, + WireEntry { + method: "worker_begin_operation", + kind: WireKind::Request(WORKER_BEGIN_OPERATION), + }, + WireEntry { + method: "worker_end_operation", + kind: WireKind::Request(WORKER_END_OPERATION), + }, ]; diff --git a/rust/crates/truapi-codegen/tests/golden/worker-callbacks.ts b/rust/crates/truapi-codegen/tests/golden/worker-callbacks.ts index 4fb2eb95b..8edccf8cc 100644 --- a/rust/crates/truapi-codegen/tests/golden/worker-callbacks.ts +++ b/rust/crates/truapi-codegen/tests/golden/worker-callbacks.ts @@ -26,6 +26,8 @@ export const CALLBACK_NAMES = [ "devicePermissionStatus", "devicePermission", "remotePermission", + "beginOperation", + "endOperation", "read", "write", "clear", @@ -37,6 +39,7 @@ export const SUBSCRIPTION_NAMES = [ "subscribeChatRooms", "subscribeLocale", "lookupPreimage", + "subscribeStorage", "subscribeTheme", ] as const; export type SubscriptionName = (typeof SUBSCRIPTION_NAMES)[number]; @@ -71,6 +74,8 @@ function rawCallbacks( | "cancelNotification" | "devicePermission" | "remotePermission" + | "beginOperation" + | "endOperation" | "read" | "write" | "clear" @@ -120,6 +125,14 @@ function rawCallbacks( bridge.callbackRequest("remotePermission", [request]) as ReturnType< Required["remotePermission"] >, + beginOperation: (product, label) => + bridge.callbackRequest("beginOperation", [product, label]) as ReturnType< + Required["beginOperation"] + >, + endOperation: (product, id) => + bridge.callbackRequest("endOperation", [product, id]) as ReturnType< + Required["endOperation"] + >, read: (key) => bridge.callbackRequest("read", [key]) as ReturnType< Required["read"] @@ -142,13 +155,18 @@ function rawCallbacks( function subscriptionRawCallbacks( bridge: WorkerCallbackBridge, ): Required< - Pick + Pick< + RawCallbacks, + "subscribeLocale" | "lookupPreimage" | "subscribeStorage" | "subscribeTheme" + > > { return { subscribeLocale: (sendItem, sendError) => bridge.startSubscription("subscribeLocale", null, sendItem, sendError), lookupPreimage: (key, sendItem, sendError) => bridge.startSubscription("lookupPreimage", key, sendItem, sendError), + subscribeStorage: (key, sendItem, sendError) => + bridge.startSubscription("subscribeStorage", key, sendItem, sendError), subscribeTheme: (sendItem, sendError) => bridge.startSubscription("subscribeTheme", null, sendItem, sendError), }; @@ -251,6 +269,12 @@ export function startRawSubscription( return undefined; } return callbacks.lookupPreimage(payload, sendItem, sendError); + case "subscribeStorage": + if (payload === null) { + console.warn(`[truapi worker] ${name} requires payload`); + return undefined; + } + return callbacks.subscribeStorage(payload, sendItem, sendError); case "subscribeTheme": return callbacks.subscribeTheme(sendItem, sendError); } diff --git a/rust/crates/truapi-host-cli/src/platform.rs b/rust/crates/truapi-host-cli/src/platform.rs index d349d0c12..2c21ac37d 100644 --- a/rust/crates/truapi-host-cli/src/platform.rs +++ b/rust/crates/truapi-host-cli/src/platform.rs @@ -24,14 +24,15 @@ use truapi::latest as api; use truapi_platform::{ AuthState, ChainProvider, CoreStorage, CoreStorageKey, DevicePermissionStatus, Features, JsonRpcConnection, LocaleHost, Navigation, Notifications, PermissionStatusHost, Permissions, - PreimageHost, ProductStorage, ProductStorageKey, SessionUiInfo, ThemeHost, UserConfirmation, - UserConfirmationReview, + PreimageHost, ProductContext, ProductOperations, ProductStorage, ProductStorageKey, + SessionUiInfo, ThemeHost, UserConfirmation, UserConfirmationReview, }; use crate::chain::WsChainProvider; use crate::terminal_ui::{SystemEvent, UiHandle}; static NEXT_STORAGE_TEMP_ID: AtomicU32 = AtomicU32::new(0); +static NEXT_OPERATION_ID: AtomicU32 = AtomicU32::new(1); /// How the host answers confirmation prompts (the web/iOS "sign?" modals). #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -490,6 +491,51 @@ impl ProductStorage for CliPlatform { self.persist_product_storage(scoped.product_id(), values) .map_err(|reason| api::HostLocalStorageReadError::Unknown { reason }) } + + fn subscribe_storage( + &self, + key: Vec, + ) -> BoxStream<'static, Result> { + // ponytail: emits the current value once; the CLI host does not push + // later changes. Wire a per-key broadcast off write/clear if e2e needs + // cross-context storage sync against the CLI signing host. + let key = String::from_utf8_lossy(&key).into_owned(); + let value = ProductStorageKey::decode(&key).ok().and_then(|scoped| { + self.product_storage + .lock() + .expect("product storage mutex poisoned") + .get(scoped.product_id()) + .and_then(|values| values.get(scoped.key())) + .cloned() + }); + Box::pin(stream::once(async move { + Ok(api::HostLocalStorageChangeItem { value }) + })) + } +} + +#[async_trait] +impl ProductOperations for CliPlatform { + async fn begin_operation( + &self, + _product: &ProductContext, + _label: String, + ) -> Result { + // ponytail: the headless CLI has no worker to keep alive; hand back a + // unique id so a product can pair begin/end. Add refcounting if the CLI + // ever backgrounds worker executions. + Ok(api::HostWorkerBeginOperationResponse { + id: NEXT_OPERATION_ID.fetch_add(1, Ordering::Relaxed), + }) + } + + async fn end_operation( + &self, + _product: &ProductContext, + _id: u32, + ) -> Result<(), api::HostWorkerOperationError> { + Ok(()) + } } #[async_trait] diff --git a/rust/crates/truapi-platform/src/lib.rs b/rust/crates/truapi-platform/src/lib.rs index 983d42f50..cde066253 100644 --- a/rust/crates/truapi-platform/src/lib.rs +++ b/rust/crates/truapi-platform/src/lib.rs @@ -36,13 +36,14 @@ use truapi::latest::{ HostChatListSubscribeItem, HostChatPostMessageError, HostChatPostMessageRequest, HostChatPostMessageResponse, HostChatRegisterBotError, HostChatRegisterBotRequest, HostChatRegisterBotResponse, HostDevicePermissionRequest, HostDevicePermissionResponse, - HostFeatureSupportedRequest, HostFeatureSupportedResponse, HostLocalStorageReadError, - HostLocaleSubscribeItem, HostNavigateToError, HostPlatform, HostPushNotificationRequest, - HostPushNotificationResponse, HostSignPayloadRequest, HostSignPayloadWithLegacyAccountRequest, - HostSignRawRequest, HostSignRawWithLegacyAccountRequest, HostThemeSubscribeItem, - LegacyAccountTxPayload, NotificationId, ProductAccountId, ProductAccountTxPayload, - ProductProofContext, RemotePermission, RemotePermissionRequest, RemotePermissionResponse, - RingLocation, + HostFeatureSupportedRequest, HostFeatureSupportedResponse, HostLocalStorageChangeItem, + HostLocalStorageReadError, HostLocaleSubscribeItem, HostNavigateToError, HostPlatform, + HostPushNotificationRequest, HostPushNotificationResponse, HostSignPayloadRequest, + HostSignPayloadWithLegacyAccountRequest, HostSignRawRequest, + HostSignRawWithLegacyAccountRequest, HostThemeSubscribeItem, HostWorkerBeginOperationResponse, + HostWorkerOperationError, LegacyAccountTxPayload, NotificationId, ProductAccountId, + ProductAccountTxPayload, ProductProofContext, RemotePermission, RemotePermissionRequest, + RemotePermissionResponse, RingLocation, }; use truapi::v01::HostAccountSignVrfRequest; use url::{Host, Url}; @@ -946,6 +947,15 @@ pub trait ProductStorage: Send + Sync { /// Clear a value at a key. async fn clear(&self, key: String) -> Result<(), HostLocalStorageReadError>; + + /// Emit the current value of a key, then each later change to it, from any + /// of the product's runtimes. A write that leaves the stored bytes + /// unchanged emits nothing. `key` is the UTF-8 bytes of the namespaced + /// storage key, the same key passed to [`Self::read`] as a `String`. + fn subscribe_storage( + &self, + key: Vec, + ) -> BoxStream<'static, Result>; } /// Open URLs in the system browser. Input is already trimmed, categorized, @@ -2844,6 +2854,29 @@ pub trait PermissionStatusHost: Send + Sync { ) -> Result; } +/// Host store for a product's pending operations. The host keeps the product's +/// worker runtime alive while it holds at least one open operation. Worker +/// products reach these through the `Worker` protocol trait, which is gated to +/// the Worker execution kind, so non-worker products never call them. +#[async_trait] +pub trait ProductOperations: Send + Sync { + /// Record a pending operation for this product. Returns its id. `label` is + /// a host log and UI hint, empty when the product gave none. + async fn begin_operation( + &self, + product: &ProductContext, + label: String, + ) -> Result; + + /// Remove a pending operation. Idempotent: an unknown or already-ended id + /// returns `Ok`. + async fn end_operation( + &self, + product: &ProductContext, + id: u32, + ) -> Result<(), HostWorkerOperationError>; +} + /// Combined platform interface. A host must provide every capability trait /// listed here. Members marked optional may be omitted; the core answers their /// product calls with `Unsupported`. See [`OptionalPlatform`]. @@ -2860,6 +2893,7 @@ pub trait Platform: + ThemeHost + LocaleHost + PreimageHost + + ProductOperations { } @@ -2876,6 +2910,7 @@ impl Platform for T where + ThemeHost + LocaleHost + PreimageHost + + ProductOperations { } diff --git a/rust/crates/truapi-server/src/generated/dispatcher.rs b/rust/crates/truapi-server/src/generated/dispatcher.rs index e23c58ce0..13d17a341 100644 --- a/rust/crates/truapi-server/src/generated/dispatcher.rs +++ b/rust/crates/truapi-server/src/generated/dispatcher.rs @@ -14,7 +14,7 @@ use parity_scale_codec::Decode; use truapi::CallContext; use truapi::api::{ Account, Chain, Chat, CoinPayment, Entropy, LocalStorage, Locale, Notifications, Payment, - Permissions, Preimage, ResourceAllocation, Signing, StatementStore, System, Theme, + Permissions, Preimage, ResourceAllocation, Signing, StatementStore, System, Theme, Worker, }; use truapi::versioned::{self, Versioned}; use truapi_platform::ProductExecutionKind; @@ -49,7 +49,8 @@ where register_signing(dispatcher, host.clone()); register_statement_store(dispatcher, host.clone()); register_system(dispatcher, host.clone()); - register_theme(dispatcher, host); + register_theme(dispatcher, host.clone()); + register_worker(dispatcher, host); } /// Start the host-initiated `chat_custom_message_render` subscription. @@ -1641,7 +1642,7 @@ where }); } { - let host = host; + let host = host.clone(); dispatcher.on_request(wire_table::LOCAL_STORAGE_CLEAR, move |request_id: String, bytes: Vec| { let host = host.clone(); Box::pin(async move { @@ -1679,6 +1680,28 @@ where }) }); } + { + let host = host; + dispatcher.on_subscription( + wire_table::LOCAL_STORAGE_SUBSCRIBE, + move |request_id: String, bytes: Vec| { + let host = host.clone(); + Box::pin(async move { + let request: versioned::local_storage::HostLocalStorageSubscribeRequest = + match Decode::decode(&mut &bytes[..]) { + Ok(request) => request, + Err(_) => return Err(Vec::new()), + }; + let cx = CallContext::with_request_id(request_id.clone()); + let stream = host.subscribe(&cx, request).await; + Ok(subscription_stream::< + versioned::local_storage::HostLocalStorageChangeItem, + _, + >(stream)) + }) + }, + ); + } } fn register_locale

(dispatcher: &mut Dispatcher, host: Arc

) @@ -2734,3 +2757,99 @@ where ); } } + +fn register_worker

(dispatcher: &mut Dispatcher, host: Arc

) +where + P: Worker + Send + Sync + 'static, +{ + { + let execution_allowed = dispatcher.allows_execution(ProductExecutionKind::Worker); + let host = host.clone(); + dispatcher.on_request(wire_table::WORKER_BEGIN_OPERATION, move |request_id: String, bytes: Vec| { + let host = host.clone(); + Box::pin(async move { + let request: versioned::worker::HostWorkerBeginOperationRequest = match Decode::decode(&mut &bytes[..]) { + Ok(request) => request, + Err(err) => { + let error: truapi::CallError = + truapi::CallError::MalformedFrame { reason: err.to_string() }; + return Ok(encode_versioned_err_payload( + error, + ::LATEST, + )); + } + }; + let target_version = request.version(); + let cx = CallContext::with_request_id(request_id.clone()); + if !execution_allowed { + let error: truapi::CallError = + truapi::CallError::Denied; + return Ok(encode_versioned_err_payload(error, target_version)); + } + let response: versioned::worker::HostWorkerBeginOperationResponse = match host.begin_operation(&cx, request).await { + Ok(value) => value, + Err(err) => { + return Ok(encode_versioned_err_payload( + downgrade_call_error(err, target_version), + target_version, + )); + } + }; + // Downgraded to the caller's version: a handler answers in + // latest terms, and a peer that asked in an older version + // cannot decode a newer variant. + Ok(encode_versioned_ok_payload( + ::from_latest( + truapi::versioned::IntoLatest::into_latest(response), + target_version, + ), + )) + }) + }); + } + { + let execution_allowed = dispatcher.allows_execution(ProductExecutionKind::Worker); + let host = host; + dispatcher.on_request(wire_table::WORKER_END_OPERATION, move |request_id: String, bytes: Vec| { + let host = host.clone(); + Box::pin(async move { + let request: versioned::worker::HostWorkerEndOperationRequest = match Decode::decode(&mut &bytes[..]) { + Ok(request) => request, + Err(err) => { + let error: truapi::CallError = + truapi::CallError::MalformedFrame { reason: err.to_string() }; + return Ok(encode_versioned_err_payload( + error, + ::LATEST, + )); + } + }; + let target_version = request.version(); + let cx = CallContext::with_request_id(request_id.clone()); + if !execution_allowed { + let error: truapi::CallError = + truapi::CallError::Denied; + return Ok(encode_versioned_err_payload(error, target_version)); + } + let response: versioned::worker::HostWorkerEndOperationResponse = match host.end_operation(&cx, request).await { + Ok(value) => value, + Err(err) => { + return Ok(encode_versioned_err_payload( + downgrade_call_error(err, target_version), + target_version, + )); + } + }; + // Downgraded to the caller's version: a handler answers in + // latest terms, and a peer that asked in an older version + // cannot decode a newer variant. + Ok(encode_versioned_ok_payload( + ::from_latest( + truapi::versioned::IntoLatest::into_latest(response), + target_version, + ), + )) + }) + }); + } +} diff --git a/rust/crates/truapi-server/src/generated/wire_table.rs b/rust/crates/truapi-server/src/generated/wire_table.rs index 5d59d7f93..4fbce8324 100644 --- a/rust/crates/truapi-server/src/generated/wire_table.rs +++ b/rust/crates/truapi-server/src/generated/wire_table.rs @@ -50,7 +50,7 @@ pub enum WireKind { /// `TRUAPI_WIRE_SCHEMA_HASH`. A host stamps it on each debug envelope so /// the debugger refuses to decode a frame whose contract differs from /// its own, even when the coarse handshake codec version is unchanged. -pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "0449982638d57658"; +pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "e141a8b071605ab4"; /// Wire discriminants for `system_handshake`. pub const SYSTEM_HANDSHAKE: RequestFrameIds = RequestFrameIds { @@ -516,6 +516,26 @@ pub const LOCALE_SUBSCRIBE: SubscriptionFrameIds = SubscriptionFrameIds { receive_id: 197, }; +/// Wire discriminants for `local_storage_subscribe`. +pub const LOCAL_STORAGE_SUBSCRIBE: SubscriptionFrameIds = SubscriptionFrameIds { + start_id: 198, + stop_id: 199, + interrupt_id: 200, + receive_id: 201, +}; + +/// Wire discriminants for `worker_begin_operation`. +pub const WORKER_BEGIN_OPERATION: RequestFrameIds = RequestFrameIds { + request_id: 202, + response_id: 203, +}; + +/// Wire discriminants for `worker_end_operation`. +pub const WORKER_END_OPERATION: RequestFrameIds = RequestFrameIds { + request_id: 204, + response_id: 205, +}; + /// The full wire table. Ordering is part of the wire protocol; /// only ever append. Removed methods leave their slot empty. pub const WIRE_TABLE: &[WireEntry] = &[ @@ -807,4 +827,16 @@ pub const WIRE_TABLE: &[WireEntry] = &[ method: "locale_subscribe", kind: WireKind::Subscription(LOCALE_SUBSCRIBE), }, + WireEntry { + method: "local_storage_subscribe", + kind: WireKind::Subscription(LOCAL_STORAGE_SUBSCRIBE), + }, + WireEntry { + method: "worker_begin_operation", + kind: WireKind::Request(WORKER_BEGIN_OPERATION), + }, + WireEntry { + method: "worker_end_operation", + kind: WireKind::Request(WORKER_END_OPERATION), + }, ]; diff --git a/rust/crates/truapi-server/src/native.rs b/rust/crates/truapi-server/src/native.rs index 415469e97..3c48d3de1 100644 --- a/rust/crates/truapi-server/src/native.rs +++ b/rust/crates/truapi-server/src/native.rs @@ -24,7 +24,7 @@ use truapi_platform::{ AuthPresenter, AuthState, ChainProvider, CoreAdmin, CoreStorage, CoreStorageKey, Features, HostInfo, JsonRpcConnection, LocaleHost, Navigation, Notifications, PermissionAuthorizationRequest, PermissionAuthorizationStatus, Permissions, PlatformInfo, - PreimageHost, ProductContext, ProductExecutionKind, ProductStorage, + PreimageHost, ProductContext, ProductExecutionKind, ProductOperations, ProductStorage, RuntimeConfigValidationError, SigningHostConfig, ThemeHost, UserConfirmation, UserConfirmationReview, async_trait, normalize_product_identifier, }; @@ -508,6 +508,18 @@ pub trait HostCallbacks: Send + Sync { fn local_storage_write(&self, key: String, value: Vec) -> Result<(), HostStorageError>; /// Clear a value from the host's scoped key-value store. fn local_storage_clear(&self, key: String) -> Result<(), HostStorageError>; + + /// Record a pending operation for a product's worker, returning a + /// host-assigned id. The host keeps the product's worker alive while it + /// holds at least one open operation. + async fn begin_operation( + &self, + product_id: String, + label: String, + ) -> Result; + /// End a pending operation. Idempotent: an unknown or already-ended id + /// succeeds. + async fn end_operation(&self, product_id: String, id: u32) -> Result<(), HostRejection>; } /// Native Chat storage and UI adapter. Hosts that support the Chat modality @@ -1044,6 +1056,11 @@ impl NativeProductExecution { self.events.notify_preimage_changed(&key, value); } + /// Push a host storage change to this execution's subscriptions for `key`. + pub fn notify_storage_changed(&self, key: String, value: Option>) { + self.events.notify_storage_changed(&key, value); + } + /// Notify this execution's chain adapter of one JSON-RPC response. pub fn notify_chain_response(&self, connection_id: u32, json: String) { self.shared_events @@ -1217,6 +1234,7 @@ struct NativeEventBus { locale_changes: Mutex>>>, preimage_changes: Mutex>, + storage_changes: Mutex>, chain_responses: Mutex>>, chat_room_changes: Mutex>>, } @@ -1226,6 +1244,11 @@ struct PreimageSubscription { tx: mpsc::UnboundedSender>, v01::GenericError>>, } +struct StorageSubscription { + key: String, + tx: mpsc::UnboundedSender>, +} + impl NativeEventBus { fn subscribe_theme( &self, @@ -1289,6 +1312,31 @@ impl NativeEventBus { }); } + fn subscribe_storage_changes( + &self, + key: String, + ) -> mpsc::UnboundedReceiver> { + let (tx, rx) = mpsc::unbounded(); + self.storage_changes + .lock() + .expect("native storage subscribers mutex poisoned") + .push(StorageSubscription { key, tx }); + rx + } + + fn notify_storage_changed(&self, key: &str, value: Option>) { + let item = v01::HostLocalStorageChangeItem { value }; + self.storage_changes + .lock() + .expect("native storage subscribers mutex poisoned") + .retain(|sub| { + if sub.key != key { + return true; + } + sub.tx.unbounded_send(Ok(item.clone())).is_ok() + }); + } + fn register_chain(&self, connection_id: u32) -> mpsc::UnboundedReceiver { let (tx, rx) = mpsc::unbounded(); self.chain_responses @@ -1486,6 +1534,60 @@ impl ProductStorage for CallbackPlatform { async fn clear(&self, key: String) -> Result<(), v01::HostLocalStorageReadError> { self.callbacks.local_storage_clear(key).map_err(Into::into) } + + fn subscribe_storage( + &self, + key: Vec, + ) -> BoxStream<'static, Result> { + // Register the change receiver first so no event between the read and + // the subscription is lost, then read the current value lazily. The + // host pushes later changes through `notify_storage_changed`. + let key = String::from_utf8_lossy(&key).into_owned(); + let rx = self.events.subscribe_storage_changes(key.clone()); + let callbacks = self.callbacks.clone(); + let current = async move { + callbacks + .local_storage_read(key) + .map(|value| v01::HostLocalStorageChangeItem { value }) + .map_err(|error| { + let error: v01::HostLocalStorageReadError = error.into(); + v01::GenericError { + reason: error.to_string(), + } + }) + }; + stream::once(current).chain(rx).boxed() + } +} + +#[async_trait] +impl ProductOperations for CallbackPlatform { + async fn begin_operation( + &self, + product: &ProductContext, + label: String, + ) -> Result { + self.callbacks + .begin_operation(product.product_id.clone(), label) + .await + .map(|id| v01::HostWorkerBeginOperationResponse { id }) + .map_err(|error| v01::HostWorkerOperationError::Unknown { + reason: error.to_string(), + }) + } + + async fn end_operation( + &self, + product: &ProductContext, + id: u32, + ) -> Result<(), v01::HostWorkerOperationError> { + self.callbacks + .end_operation(product.product_id.clone(), id) + .await + .map_err(|error| v01::HostWorkerOperationError::Unknown { + reason: error.to_string(), + }) + } } #[async_trait] @@ -2057,6 +2159,16 @@ mod tests { fn local_storage_clear(&self, _key: String) -> Result<(), HostStorageError> { Ok(()) } + async fn begin_operation( + &self, + _product_id: String, + _label: String, + ) -> Result { + Ok(1) + } + async fn end_operation(&self, _product_id: String, _id: u32) -> Result<(), HostRejection> { + Ok(()) + } } impl NativeChatCallbacks for EventCallbacks { @@ -3145,6 +3257,20 @@ mod tests { fn local_storage_clear(&self, _key: String) -> Result<(), HostStorageError> { Ok(()) } + async fn begin_operation( + &self, + _product_id: String, + _label: String, + ) -> Result { + Ok(1) + } + async fn end_operation( + &self, + _product_id: String, + _id: u32, + ) -> Result<(), HostRejection> { + Ok(()) + } } let execution = native_product_execution(Arc::new(Noop), "dotli.dot"); @@ -3299,6 +3425,20 @@ mod tests { fn local_storage_clear(&self, _key: String) -> Result<(), HostStorageError> { Ok(()) } + async fn begin_operation( + &self, + _product_id: String, + _label: String, + ) -> Result { + Ok(1) + } + async fn end_operation( + &self, + _product_id: String, + _id: u32, + ) -> Result<(), HostRejection> { + Ok(()) + } } let (release_tx, release_rx) = tokio::sync::mpsc::channel::<()>(1); diff --git a/rust/crates/truapi-server/src/runtime.rs b/rust/crates/truapi-server/src/runtime.rs index c083b24a8..cd841e69e 100644 --- a/rust/crates/truapi-server/src/runtime.rs +++ b/rust/crates/truapi-server/src/runtime.rs @@ -87,7 +87,7 @@ use parity_scale_codec::Encode; use tracing::{debug, instrument, warn}; use truapi::api::{ Account, Chain, Chat, CoinPayment, Entropy, LocalStorage, Locale, Notifications, Payment, - Permissions, Preimage, ResourceAllocation, Signing, System, Theme, + Permissions, Preimage, ResourceAllocation, Signing, System, Theme, Worker, }; use truapi::versioned::account::{ HostAccountConnectionStatusSubscribeItem, HostAccountCreateProofError, @@ -146,9 +146,10 @@ use truapi::versioned::entropy::{ HostDeriveEntropyError, HostDeriveEntropyRequest, HostDeriveEntropyResponse, }; use truapi::versioned::local_storage::{ - HostLocalStorageClearError, HostLocalStorageClearRequest, HostLocalStorageClearResponse, - HostLocalStorageReadError, HostLocalStorageReadRequest, HostLocalStorageReadResponse, - HostLocalStorageWriteError, HostLocalStorageWriteRequest, HostLocalStorageWriteResponse, + HostLocalStorageChangeItem, HostLocalStorageClearError, HostLocalStorageClearRequest, + HostLocalStorageClearResponse, HostLocalStorageReadError, HostLocalStorageReadRequest, + HostLocalStorageReadResponse, HostLocalStorageSubscribeRequest, HostLocalStorageWriteError, + HostLocalStorageWriteRequest, HostLocalStorageWriteResponse, }; use truapi::versioned::locale::HostLocaleSubscribeItem; use truapi::versioned::notifications::{ @@ -191,6 +192,11 @@ use truapi::versioned::system::{ HostNavigateToResponse, }; use truapi::versioned::theme::HostThemeSubscribeItem; +use truapi::versioned::worker::{ + HostWorkerBeginOperationError, HostWorkerBeginOperationRequest, + HostWorkerBeginOperationResponse, HostWorkerEndOperationError, HostWorkerEndOperationRequest, + HostWorkerEndOperationResponse, +}; use truapi::{CallContext, CallError, CancellationReason, Subscription}; use truapi::{latest, v01}; use truapi_platform::Platform; @@ -994,8 +1000,18 @@ impl LocalStorage for ProductRuntimeHost { ) -> Result> { let HostLocalStorageWriteRequest::V1(v01::HostLocalStorageWriteRequest { key, value }) = request; + let storage_key = self.product_storage_key(key); + // Skip the store write when the value is byte-identical to what the key + // already holds, so a subscriber sees only real changes and no + // redundant persistence happens. A failed pre-read falls through to the + // write rather than blocking it. + if let Ok(Some(current)) = self.platform.read(storage_key.clone()).await + && current == value + { + return Ok(HostLocalStorageWriteResponse::V1); + } self.platform - .write(self.product_storage_key(key), value) + .write(storage_key, value) .await .map(|()| HostLocalStorageWriteResponse::V1) .map_err(|err| CallError::Domain(HostLocalStorageWriteError::V1(err))) @@ -1014,6 +1030,68 @@ impl LocalStorage for ProductRuntimeHost { .map(|()| HostLocalStorageClearResponse::V1) .map_err(|err| CallError::Domain(HostLocalStorageClearError::V1(err))) } + + #[instrument(skip_all, fields(runtime.method = "local_storage.subscribe"))] + async fn subscribe( + &self, + _cx: &CallContext, + request: HostLocalStorageSubscribeRequest, + ) -> Subscription { + let HostLocalStorageSubscribeRequest::V1(v01::HostLocalStorageSubscribeRequest { key }) = + request; + Subscription::new(Box::pin( + self.platform + .subscribe_storage(self.product_storage_key(key).into_bytes()) + .filter_map(|item| async { + match item { + Ok(item) => Some(HostLocalStorageChangeItem::V1(item)), + Err(error) => { + warn!( + reason = %error.reason, + "local storage subscription platform stream failed" + ); + None + } + } + }), + )) + } +} + +// --------------------------------------------------------------------------- +// Worker +// --------------------------------------------------------------------------- + +#[truapi_platform::async_trait] +impl Worker for ProductRuntimeHost { + #[instrument(skip_all, fields(runtime.method = "worker.begin_operation"))] + async fn begin_operation( + &self, + _cx: &CallContext, + request: HostWorkerBeginOperationRequest, + ) -> Result> { + let HostWorkerBeginOperationRequest::V1(v01::HostWorkerBeginOperationRequest { label }) = + request; + self.platform + .begin_operation(&self.product, label.unwrap_or_default()) + .await + .map(HostWorkerBeginOperationResponse::V1) + .map_err(|error| CallError::Domain(HostWorkerBeginOperationError::V1(error))) + } + + #[instrument(skip_all, fields(runtime.method = "worker.end_operation"))] + async fn end_operation( + &self, + _cx: &CallContext, + request: HostWorkerEndOperationRequest, + ) -> Result> { + let HostWorkerEndOperationRequest::V1(v01::HostWorkerEndOperationRequest { id }) = request; + self.platform + .end_operation(&self.product, id) + .await + .map(|()| HostWorkerEndOperationResponse::V1) + .map_err(|error| CallError::Domain(HostWorkerEndOperationError::V1(error))) + } } // --------------------------------------------------------------------------- diff --git a/rust/crates/truapi-server/src/test_support.rs b/rust/crates/truapi-server/src/test_support.rs index 516e143aa..6f964fa1e 100644 --- a/rust/crates/truapi-server/src/test_support.rs +++ b/rust/crates/truapi-server/src/test_support.rs @@ -29,9 +29,10 @@ use truapi_platform::{ CoreStorage as PlatformCoreStorage, CoreStorageKey, Features as PlatformFeatures, HostInfo, JsonRpcConnection, LocaleHost, Navigation as PlatformNavigation, Notifications as PlatformNotifications, PairingHostConfig, Permissions as PlatformPermissions, - PlatformInfo, PreimageHost, ProductContext, ProductStorage as PlatformProductStorage, - ProductSubtreeReview, ResourceAllocationReview, SignVrfReview, StatementStoreProductSignReview, - ThemeHost, UserConfirmation, UserConfirmationReview, + PlatformInfo, PreimageHost, ProductContext, ProductOperations as PlatformProductOperations, + ProductStorage as PlatformProductStorage, ProductSubtreeReview, ResourceAllocationReview, + SignVrfReview, StatementStoreProductSignReview, ThemeHost, UserConfirmation, + UserConfirmationReview, }; use x25519_dalek::{PublicKey as X25519PublicKey, StaticSecret as X25519SecretKey}; @@ -793,6 +794,42 @@ impl PlatformProductStorage for StubPlatform { .remove(&key); Ok(()) } + + fn subscribe_storage( + &self, + key: Vec, + ) -> BoxStream<'static, Result> { + let key = String::from_utf8_lossy(&key).into_owned(); + let value = self + .local_storage + .lock() + .expect("local storage mutex poisoned") + .get(&key) + .cloned(); + Box::pin(stream::once(async move { + Ok(v01::HostLocalStorageChangeItem { value }) + })) + } +} + +#[truapi_platform::async_trait] +impl PlatformProductOperations for StubPlatform { + async fn begin_operation( + &self, + _product: &ProductContext, + _label: String, + ) -> Result { + // The stub has no worker lifecycle to keep alive; a fixed id suffices. + Ok(v01::HostWorkerBeginOperationResponse { id: 1 }) + } + + async fn end_operation( + &self, + _product: &ProductContext, + _id: u32, + ) -> Result<(), v01::HostWorkerOperationError> { + Ok(()) + } } #[truapi_platform::async_trait] diff --git a/rust/crates/truapi-server/src/wasm/generated_bridge.rs b/rust/crates/truapi-server/src/wasm/generated_bridge.rs index d1b23cd61..65e9fbcd1 100644 --- a/rust/crates/truapi-server/src/wasm/generated_bridge.rs +++ b/rust/crates/truapi-server/src/wasm/generated_bridge.rs @@ -45,9 +45,12 @@ pub(super) struct JsBridge { pub(super) device_permission: Function, pub(super) remote_permission: Function, pub(super) lookup_preimage: Function, + pub(super) begin_operation: Function, + pub(super) end_operation: Function, pub(super) read: Function, pub(super) write: Function, pub(super) clear: Function, + pub(super) subscribe_storage: Function, pub(super) subscribe_theme: Function, pub(super) confirm_user_action: Function, pub(super) chat_present: bool, @@ -81,9 +84,12 @@ impl JsBridge { device_permission: get_function(callbacks, "devicePermission")?, remote_permission: get_function(callbacks, "remotePermission")?, lookup_preimage: get_function(callbacks, "lookupPreimage")?, + begin_operation: get_function(callbacks, "beginOperation")?, + end_operation: get_function(callbacks, "endOperation")?, read: get_function(callbacks, "read")?, write: get_function(callbacks, "write")?, clear: get_function(callbacks, "clear")?, + subscribe_storage: get_function(callbacks, "subscribeStorage")?, subscribe_theme: get_function(callbacks, "subscribeTheme")?, confirm_user_action: get_function(callbacks, "confirmUserAction")?, chat_present: get_optional_function(callbacks, "createChatRoom")?.is_some() @@ -389,6 +395,46 @@ impl truapi_platform::PreimageHost for WasmPlatform { } } +#[truapi_platform::async_trait] +impl truapi_platform::ProductOperations for WasmPlatform { + async fn begin_operation( + &self, + product: &truapi_platform::ProductContext, + label: String, + ) -> Result { + let bytes = invoke_bytes_return( + &self.bridge.begin_operation, + vec![ + Uint8Array::from(product.encode().as_slice()).into(), + JsValue::from_str(&label), + ], + ) + .await + .map_err(|reason| v01::HostWorkerOperationError::Unknown { reason })?; + decode_bytes::( + bytes, + "beginOperation response did not decode", + ) + .map_err(|reason| v01::HostWorkerOperationError::Unknown { reason }) + } + + async fn end_operation( + &self, + product: &truapi_platform::ProductContext, + id: u32, + ) -> Result<(), v01::HostWorkerOperationError> { + invoke_unit( + &self.bridge.end_operation, + vec![ + Uint8Array::from(product.encode().as_slice()).into(), + JsValue::from_f64(f64::from(id)), + ], + ) + .await + .map_err(|reason| v01::HostWorkerOperationError::Unknown { reason }) + } +} + #[truapi_platform::async_trait] impl truapi_platform::ProductStorage for WasmPlatform { async fn read(&self, key: String) -> Result>, v01::HostLocalStorageReadError> { @@ -422,6 +468,17 @@ impl truapi_platform::ProductStorage for WasmPlatform { .await .map_err(|reason| v01::HostLocalStorageReadError::Unknown { reason }) } + + fn subscribe_storage( + &self, + key: Vec, + ) -> BoxStream<'static, Result> { + invoke_js_subscription( + &self.bridge.subscribe_storage, + Some(key), + parse_host_local_storage_change_item_item, + ) + } } impl truapi_platform::ThemeHost for WasmPlatform { @@ -457,6 +514,12 @@ fn parse_host_chat_list_subscribe_item_item( decode_js_item::(value, "HostChatListSubscribeItem") } +fn parse_host_local_storage_change_item_item( + value: JsValue, +) -> Result { + decode_js_item::(value, "HostLocalStorageChangeItem") +} + fn parse_host_locale_subscribe_item_item( value: JsValue, ) -> Result { diff --git a/rust/crates/truapi-server/tests/common/mod.rs b/rust/crates/truapi-server/tests/common/mod.rs index 387155156..2ecae7774 100644 --- a/rust/crates/truapi-server/tests/common/mod.rs +++ b/rust/crates/truapi-server/tests/common/mod.rs @@ -8,8 +8,8 @@ use truapi::v01; use truapi_platform::{ AuthPresenter, ChainProvider, CoreStorage, CoreStorageKey, Features, HostInfo, JsonRpcConnection, LocaleHost, Navigation, Notifications, PairingHostConfig, Permissions, - PlatformInfo, PreimageHost, ProductContext, ProductStorage, ThemeHost, UserConfirmation, - UserConfirmationReview, + PlatformInfo, PreimageHost, ProductContext, ProductOperations, ProductStorage, ThemeHost, + UserConfirmation, UserConfirmationReview, }; use truapi_server::frame::ProtocolMessage; use truapi_server::transport::Transport; @@ -86,6 +86,30 @@ impl ProductStorage for WireShapePlatform { async fn clear(&self, _key: String) -> Result<(), v01::HostLocalStorageReadError> { Ok(()) } + fn subscribe_storage( + &self, + _key: Vec, + ) -> BoxStream<'static, Result> { + Box::pin(stream::empty()) + } +} + +#[truapi_platform::async_trait] +impl ProductOperations for WireShapePlatform { + async fn begin_operation( + &self, + _product: &ProductContext, + _label: String, + ) -> Result { + Ok(v01::HostWorkerBeginOperationResponse { id: 1 }) + } + async fn end_operation( + &self, + _product: &ProductContext, + _id: u32, + ) -> Result<(), v01::HostWorkerOperationError> { + Ok(()) + } } #[truapi_platform::async_trait] diff --git a/rust/crates/truapi/src/api.rs b/rust/crates/truapi/src/api.rs index a4a7bb03f..8cd3f3f02 100644 --- a/rust/crates/truapi/src/api.rs +++ b/rust/crates/truapi/src/api.rs @@ -16,6 +16,7 @@ pub mod signing; pub mod statement_store; pub mod system; pub mod theme; +pub mod worker; pub use account::Account; pub use chain::Chain; @@ -33,6 +34,7 @@ pub use signing::Signing; pub use statement_store::StatementStore; pub use system::System; pub use theme::Theme; +pub use worker::Worker; /// The unified TrUAPI contract. pub trait TrUApi: @@ -52,6 +54,7 @@ pub trait TrUApi: + StatementStore + System + Theme + + Worker + Send + Sync { @@ -74,6 +77,7 @@ impl TrUApi for T where + StatementStore + System + Theme + + Worker + Send + Sync { diff --git a/rust/crates/truapi/src/api/local_storage.rs b/rust/crates/truapi/src/api/local_storage.rs index 5c2057858..2c882bd66 100644 --- a/rust/crates/truapi/src/api/local_storage.rs +++ b/rust/crates/truapi/src/api/local_storage.rs @@ -1,12 +1,13 @@ //! Unified [`LocalStorage`] trait. use crate::versioned::local_storage::{ - HostLocalStorageClearError, HostLocalStorageClearRequest, HostLocalStorageClearResponse, - HostLocalStorageReadError, HostLocalStorageReadRequest, HostLocalStorageReadResponse, - HostLocalStorageWriteError, HostLocalStorageWriteRequest, HostLocalStorageWriteResponse, + HostLocalStorageChangeItem, HostLocalStorageClearError, HostLocalStorageClearRequest, + HostLocalStorageClearResponse, HostLocalStorageReadError, HostLocalStorageReadRequest, + HostLocalStorageReadResponse, HostLocalStorageSubscribeRequest, HostLocalStorageWriteError, + HostLocalStorageWriteRequest, HostLocalStorageWriteResponse, }; use crate::wire; -use crate::{CallContext, CallError}; +use crate::{CallContext, CallError, Subscription}; /// Local key/value storage scoped to the calling product. #[crate::async_trait] @@ -55,4 +56,27 @@ pub trait LocalStorage: Send + Sync { cx: &CallContext, request: HostLocalStorageClearRequest, ) -> Result>; + + /// Subscribe to changes of one key in the product's own storage namespace. + /// + /// Emits the current value immediately, then one item per later write or + /// clear of the key by any of the product's runtimes. A write that leaves + /// the stored bytes unchanged emits nothing. + /// + /// ```ts + /// import { firstValueFrom, from } from "rxjs"; + /// + /// const item = await firstValueFrom( + /// from(truapi.localStorage.subscribe({ request: { key: "test-key" } })), + /// ); + /// console.log("storage change received:", item); + /// ``` + #[wire(start_id = 198)] + async fn subscribe( + &self, + _cx: &CallContext, + _request: HostLocalStorageSubscribeRequest, + ) -> Subscription { + Subscription::empty() + } } diff --git a/rust/crates/truapi/src/api/worker.rs b/rust/crates/truapi/src/api/worker.rs new file mode 100644 index 000000000..15274866f --- /dev/null +++ b/rust/crates/truapi/src/api/worker.rs @@ -0,0 +1,54 @@ +//! Unified [`Worker`] trait. + +use crate::versioned::worker::{ + HostWorkerBeginOperationError, HostWorkerBeginOperationRequest, + HostWorkerBeginOperationResponse, HostWorkerEndOperationError, HostWorkerEndOperationRequest, + HostWorkerEndOperationResponse, +}; +use crate::wire; +use crate::{CallContext, CallError}; + +/// Worker background-operation APIs. +/// +/// A worker holds itself alive by keeping an operation open. The host keeps +/// the product's worker running while it has at least one open operation. +#[crate::service(required_execution = Worker)] +#[crate::async_trait] +pub trait Worker: Send + Sync { + /// Begin a pending operation. The worker is kept alive while it has at + /// least one open operation. Returns an id for `end_operation`. + /// + /// ```ts + /// const result = await truapi.worker.beginOperation({ label: "funding" }); + /// assert(result.isOk(), "beginOperation failed:", result); + /// console.log("operation started:", result.value.id); + /// await truapi.worker.endOperation({ id: result.value.id }); + /// ``` + #[wire(request_id = 202)] + async fn begin_operation( + &self, + _cx: &CallContext, + _request: HostWorkerBeginOperationRequest, + ) -> Result> { + Err(CallError::unavailable()) + } + + /// End a pending operation. Idempotent: an unknown or already-ended id + /// returns success. + /// + /// ```ts + /// const begun = await truapi.worker.beginOperation({}); + /// assert(begun.isOk(), "beginOperation failed:", begun); + /// const result = await truapi.worker.endOperation({ id: begun.value.id }); + /// assert(result.isOk(), "endOperation failed:", result); + /// console.log("operation ended"); + /// ``` + #[wire(request_id = 204)] + async fn end_operation( + &self, + _cx: &CallContext, + _request: HostWorkerEndOperationRequest, + ) -> Result> { + Err(CallError::unavailable()) + } +} diff --git a/rust/crates/truapi/src/lib.rs b/rust/crates/truapi/src/lib.rs index 06d3557f4..e7e3066d1 100644 --- a/rust/crates/truapi/src/lib.rs +++ b/rust/crates/truapi/src/lib.rs @@ -55,13 +55,14 @@ pub mod latest { ChatActionLayout, ChatActions, ChatBotRegistrationStatus, ChatCustomMessage, ChatFile, ChatMedia, ChatMessageContent, ChatReaction, ChatRichText, ChatRoomRegistrationStatus, ContextualAlias, DerivationIndex, GenericError, HostPlatform, HostSignPayloadData, - NotificationId, OperationStartedResult, ProductAccountId, ProductProofContext, RawPayload, - RegisteredRingVrfKey, RemotePermission, RemoteStatementStoreCreateProofError, - RemoteStatementStoreCreateProofRequest, RemoteStatementStoreCreateProofResponse, - RemoteStatementStoreSubscribeItem, RemoteStatementStoreSubscribeRequest, RingLocation, - RingVrfKeyDisclosure, RingVrfPublicKey, RuntimeApi, RuntimeSpec, RuntimeType, - SignedStatement, Statement, StatementProof, StorageQueryItem, StorageQueryType, - StorageResultItem, ThemeName, ThemeVariant, TxPayloadExtension, + HostWorkerOperationError, NotificationId, OperationId, OperationStartedResult, + ProductAccountId, ProductProofContext, RawPayload, RegisteredRingVrfKey, RemotePermission, + RemoteStatementStoreCreateProofError, RemoteStatementStoreCreateProofRequest, + RemoteStatementStoreCreateProofResponse, RemoteStatementStoreSubscribeItem, + RemoteStatementStoreSubscribeRequest, RingLocation, RingVrfKeyDisclosure, RingVrfPublicKey, + RuntimeApi, RuntimeSpec, RuntimeType, SignedStatement, Statement, StatementProof, + StorageQueryItem, StorageQueryType, StorageResultItem, ThemeName, ThemeVariant, + TxPayloadExtension, }; /// Latest payload type of a versioned envelope. @@ -130,6 +131,9 @@ pub mod latest { /// Product context bound to the current host runtime. pub type HostGetProductContextResponse = LatestOf; + /// Storage key change pushed to a subscriber. + pub type HostLocalStorageChangeItem = + LatestOf; /// Local storage operation error. pub type HostLocalStorageReadError = LatestOf; @@ -167,6 +171,9 @@ pub mod latest { LatestOf; /// Current host theme pushed to subscribers. pub type HostThemeSubscribeItem = LatestOf; + /// Result of beginning a worker pending operation. + pub type HostWorkerBeginOperationResponse = + LatestOf; /// Transaction creation payload for a legacy account. pub type LegacyAccountTxPayload = LatestOf; diff --git a/rust/crates/truapi/src/v01.rs b/rust/crates/truapi/src/v01.rs index afb43cf7c..0cd859d48 100644 --- a/rust/crates/truapi/src/v01.rs +++ b/rust/crates/truapi/src/v01.rs @@ -18,6 +18,7 @@ mod statement_store; mod system; mod theme; mod transaction; +mod worker; pub use account::*; pub use chain::*; @@ -37,3 +38,4 @@ pub use statement_store::*; pub use system::*; pub use theme::*; pub use transaction::*; +pub use worker::*; diff --git a/rust/crates/truapi/src/v01/local_storage.rs b/rust/crates/truapi/src/v01/local_storage.rs index 3bd56307b..d3adf7997 100644 --- a/rust/crates/truapi/src/v01/local_storage.rs +++ b/rust/crates/truapi/src/v01/local_storage.rs @@ -45,3 +45,18 @@ pub struct HostLocalStorageClearRequest { /// Storage key to clear. pub key: String, } + +/// Request to subscribe to changes of one local storage key. +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] +pub struct HostLocalStorageSubscribeRequest { + /// Storage key to observe. + pub key: String, +} + +/// A change to a subscribed storage key, pushed to the subscriber. +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct HostLocalStorageChangeItem { + /// Value after the change. `Some` on write, `None` after clear. + pub value: Option>, +} diff --git a/rust/crates/truapi/src/v01/worker.rs b/rust/crates/truapi/src/v01/worker.rs new file mode 100644 index 000000000..f02b4db51 --- /dev/null +++ b/rust/crates/truapi/src/v01/worker.rs @@ -0,0 +1,42 @@ +use derive_more::Display; +use parity_scale_codec::{Decode, Encode}; + +/// Opaque host-assigned pending-operation identifier, unique per product. +pub type OperationId = u32; + +/// Request to begin a pending operation. +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] +pub struct HostWorkerBeginOperationRequest { + /// Optional label for host logs and UI. + pub label: Option, +} + +/// Response carrying the id of a newly begun operation. +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] +pub struct HostWorkerBeginOperationResponse { + /// Id to pass to `end_operation`. + pub id: OperationId, +} + +/// Request to end a pending operation. +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)] +pub struct HostWorkerEndOperationRequest { + /// Id returned by `begin_operation`. + pub id: OperationId, +} + +/// Pending-operation error. +#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode, Display)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] +pub enum HostWorkerOperationError { + /// The product is already at the host's per-product limit of open + /// operations. + #[display("too many open operations")] + TooManyOpen, + /// Catch-all host failure. + #[display("{reason}")] + Unknown { + /// Human-readable failure reason. + reason: String, + }, +} diff --git a/rust/crates/truapi/src/versioned.rs b/rust/crates/truapi/src/versioned.rs index 4d5e37c26..938b63fd3 100644 --- a/rust/crates/truapi/src/versioned.rs +++ b/rust/crates/truapi/src/versioned.rs @@ -46,6 +46,7 @@ pub mod signing; pub mod statement_store; pub mod system; pub mod theme; +pub mod worker; #[cfg(test)] mod tests { diff --git a/rust/crates/truapi/src/versioned/local_storage.rs b/rust/crates/truapi/src/versioned/local_storage.rs index 708eb4c7b..9760cb43d 100644 --- a/rust/crates/truapi/src/versioned/local_storage.rs +++ b/rust/crates/truapi/src/versioned/local_storage.rs @@ -12,4 +12,6 @@ truapi_macros::versioned_type! { pub enum HostLocalStorageClearRequest { V1 => v01::HostLocalStorageClearRequest } pub enum HostLocalStorageClearResponse { V1 } pub enum HostLocalStorageClearError { V1 => v01::HostLocalStorageReadError } + pub enum HostLocalStorageSubscribeRequest { V1 => v01::HostLocalStorageSubscribeRequest } + pub enum HostLocalStorageChangeItem { V1 => v01::HostLocalStorageChangeItem } } diff --git a/rust/crates/truapi/src/versioned/worker.rs b/rust/crates/truapi/src/versioned/worker.rs new file mode 100644 index 000000000..9e2045a25 --- /dev/null +++ b/rust/crates/truapi/src/versioned/worker.rs @@ -0,0 +1,12 @@ +//! Versioned wrappers for [`Worker`](crate::api::Worker) methods. + +use crate::v01; + +truapi_macros::versioned_type! { + pub enum HostWorkerBeginOperationRequest { V1 => v01::HostWorkerBeginOperationRequest } + pub enum HostWorkerBeginOperationResponse { V1 => v01::HostWorkerBeginOperationResponse } + pub enum HostWorkerBeginOperationError { V1 => v01::HostWorkerOperationError } + pub enum HostWorkerEndOperationRequest { V1 => v01::HostWorkerEndOperationRequest } + pub enum HostWorkerEndOperationResponse { V1 } + pub enum HostWorkerEndOperationError { V1 => v01::HostWorkerOperationError } +} From d82dcf63f31e632e8e11a196c20fa570cdc0fc3b Mon Sep 17 00:00:00 2001 From: Sergey Zhuravlev Date: Fri, 4 Sep 2026 12:20:00 +0200 Subject: [PATCH 2/3] fix(truapi-host): key worker pending operations by id; test RFC 0027 paths The web host runtime tracks open operations as a set of host-assigned ids instead of a counter, so ending an unknown or already-ended id releases no other operation's hold. The begin response arrives as SCALE bytes from the raw callback, so the runtime decodes it to read the id. Add core tests against the stub platform for storage subscriptions (current value first, write, byte-identical rewrite skipped before the platform, clear, product scoping) and for begin/end reaching the platform under the calling product. The stub now records writes, pushes storage changes to subscribers, and records operations. Shorten RFC 0027 to the design itself: no wire ids, no justification prose, and every stated behavior matches the code (core skips identical writes, ProductOperations is required, nothing persists operations). Claude-Session: https://claude.ai/code/session_01RToiDVG81smFNdDAo53WuF --- ...torage-subscriptions-pending-operations.md | 131 ++++---------- .../src/web/create-worker-host-runtime.ts | 33 ++-- .../src/web/worker-provider.test.ts | 53 ++++++ rust/crates/truapi-server/src/runtime.rs | 165 ++++++++++++++++++ rust/crates/truapi-server/src/test_support.rs | 80 +++++++-- 5 files changed, 343 insertions(+), 119 deletions(-) diff --git a/docs/rfcs/0027-storage-subscriptions-pending-operations.md b/docs/rfcs/0027-storage-subscriptions-pending-operations.md index ff65146d2..a3dcba056 100644 --- a/docs/rfcs/0027-storage-subscriptions-pending-operations.md +++ b/docs/rfcs/0027-storage-subscriptions-pending-operations.md @@ -1,6 +1,7 @@ --- title: "Product storage subscriptions and worker pending operations" owner: "Sergey Zhuravlev" +status: draft --- # RFC 0027: Product storage subscriptions and worker pending operations @@ -9,61 +10,44 @@ owner: "Sergey Zhuravlev" | --------------- | ---------------------------------------- | | **RFC Number** | 27 | | **Start Date** | 2026-08-25 | -| **Description** | Two small TrUAPI additions so a background worker can finish a multi-step task and coordinate with the app through storage. | +| **Description** | Two TrUAPI additions so a background worker can finish a multi-step task and coordinate with the app through storage. | | **Authors** | Sergey Zhuravlev | ## Summary -Two additions to TrUAPI: - - `localStorage.subscribe(key)` streams a key's value on every change, within the product's own namespace. -- `worker.beginOperation()` / `worker.endOperation(id)` declare a pending operation; the host keeps the worker running while any operation is open. - -Both come from one flow: a funding operation, part of a safety-net release, runs in a worker, submits a transaction, and needs to finish and report progress even after the user leaves the app. +- `worker.beginOperation()` / `worker.endOperation(id)` declare a pending operation. The host keeps the worker running while any operation is open. ## Motivation -The product runs a funding operation, one part of a safety-net release. It builds a transaction, submits it, waits for it to be included, and records the result. Some of those steps hit a backend, so one operation can run for tens of seconds with polling in between. It runs in a worker so it continues after the user leaves the product's screen. Two things make that unsafe today. +A funding operation, part of a safety-net release, builds a transaction, submits it, waits for inclusion and records the result. Steps hit a backend, so one run takes tens of seconds with polling in between. It runs in a worker so it outlives the product's screen. Two things make that unsafe today. -The worker dies when the user leaves. The host disposes a worker once its on-screen surface is gone, and the in-flight submission is aborted with it (the worker's own `dispose` is a no-op that defers to the main thread, `worker-runtime.ts:182`). Being killed between submitting a funding transaction and confirming it is the worst place to stop, and the product has no way to tell the host it is mid-operation. +The host disposes a worker once its on-screen surface is gone (the worker's own `dispose` in `worker-runtime.ts` is a no-op that defers to the main thread), and the in-flight submission dies with it. The product has no way to say it is mid-operation. -The UI and the worker can't see each other's progress. The on-screen product and the worker are separate runtimes over one storage namespace, but a write in one stays invisible to the other until it re-reads. So a progress view the worker feeds, or a worker that should react to what the user just did on screen, can only re-read on a timer. For a value that changes a few times a minute, that polling is both late and wasteful. +The screen and the worker are separate runtimes over one storage namespace, and a write in one is invisible to the other until it re-reads. A progress view fed by the worker can only poll. -## Detailed design +## Detailed Design ### localStorage.subscribe -A subscription method on the `LocalStorage` trait, next to `read` (12), `write` (14), `clear` (16): +On the `LocalStorage` trait next to `read`, `write` and `clear`: ```rust -/// Subscribe to changes of one key in the product's own storage namespace. -/// -/// Emits the current value immediately, then one item per later change. -#[wire(start_id = 198)] // exact id assigned at implementation, free range above 197 async fn subscribe( &self, cx: &CallContext, request: HostLocalStorageSubscribeRequest, // { key: String } ) -> Subscription; -``` -```rust pub struct HostLocalStorageChangeItem { - /// Value after the change. `Some` on write, `None` after clear. + /// `Some` on write, `None` after clear. pub value: Option>, } ``` -The wire side is nothing new. `theme.subscribe` and `chat.list_subscribe` already return `Subscription`, and the TS client already exposes them as RxJS observables. - -The host emits the changes. On web and old JS hosts the app and worker are separate WASM instances with separate `RuntimeServices`, so the core alone can't carry a change from one to the other. The host can: it sits above both instances and owns the store, so it sees every write to the namespace whoever made it. The core just forwards the host's stream to the subscriber. - -This adds one method to the host `ProductStorage` trait, the same shape as the existing `ChatPlatform::subscribe_chat_rooms` (`rust/crates/truapi-platform/src/lib.rs`): +The host owns the store both runtimes write to, so the host emits the changes and the core forwards its stream to the subscriber. This adds one method to the required host `ProductStorage` trait: ```rust -/// Emit a product-scoped key's current value, then each later change, -/// from any of the product's runtimes. A write that doesn't change the -/// bytes emits nothing (see below). fn subscribe_storage( &self, product: &ProductContext, @@ -71,119 +55,76 @@ fn subscribe_storage( ) -> BoxStream<'static, Result>; ``` -The core passes the calling product's context to `subscribe_storage`, the same scoping `read` and `write` already use, so a product only ever sees its own keys. The first item is the current value, so there's no read-then-subscribe gap. After that, a write emits `Some(value)` and a clear emits `None`. This works the same on web and native, because the source is the host, not a shared core instance. +The core passes the calling product's context, the same scoping `read` and `write` use, so a product only sees its own keys. The first item is the current value, so there is no read-then-subscribe gap. After that a write emits `Some(value)` and a clear emits `None`. A burst of distinct values emits one item per value; nothing coalesces. -### Byte-identical writes do nothing - -If `write` gets the same bytes the key already holds, the host skips the store write and the change event. Same for `clear` on an absent key. The host does this because it holds the current bytes to compare. So a runtime that rewrites unchanged state on a timer costs nothing and wakes no one. +If `write` gets the bytes the key already holds, the core skips the store write, so the host never sees it and nothing is emitted. `clear` always reaches the host and always emits `None`, even on an absent key. ### Pending operations -The worker begins a pending operation while it has work in flight and ends it when done. The host keeps the worker alive while any operation is open. - -A `begin`/`end` pair on a new `Worker` trait: +A `begin`/`end` pair on a new `Worker` trait, gated to the Worker execution kind: ```rust -/// Begin a pending operation. The worker is kept alive while it has at least -/// one open operation. Returns an id for `end_operation`. -/// -/// Worker execution kind only. -#[wire(request_id = 202)] // exact id assigned at implementation async fn begin_operation( &self, cx: &CallContext, - request: HostBeginOperationRequest, // { label: Option } for host UI/logs -) -> Result>; + request: HostWorkerBeginOperationRequest, // { label: Option } for host logs and UI +) -> Result>; -/// End a pending operation. Idempotent: an unknown or already-ended id -/// returns `Ok`. -#[wire(request_id = 204)] // exact id assigned at implementation +/// Idempotent: an unknown or already-ended id returns `Ok`. async fn end_operation( &self, cx: &CallContext, - request: HostEndOperationRequest, // { id: OperationId } -) -> Result<(), CallError>; -``` + request: HostWorkerEndOperationRequest, // { id: OperationId } +) -> Result>; -```rust -/// Opaque host-assigned operation identifier, unique per product. Mirrors -/// `NotificationId`, which is a `u32` type alias. +/// Host-assigned, unique per product. Like `NotificationId`. pub type OperationId = u32; -pub struct HostBeginOperationResponse { - /// Pass this to `end_operation`. +pub struct HostWorkerBeginOperationResponse { pub id: OperationId, } -/// Domain error for the operation methods. -pub enum HostOperationError { - /// The product already holds the host's per-product cap of open - /// operations. `end` never returns this; it is idempotent and always - /// succeeds. +pub enum HostWorkerOperationError { + /// The product is at the host's per-product cap of open operations. + /// `end` never returns this. TooManyOpen, + Unknown { reason: String }, } ``` -Why operations and not a timer. A timer makes the worker guess the duration, and a short guess kills the transaction mid-flight. An operation ties liveness to the work itself: alive while something is open, gone when it closes. The id is session-scoped, not something the worker persists. If the host dies, the worker and the id die with it, and the leftover operation record is reconciled on the next launch (until a reaper exists, see open questions). So losing the id costs nothing. +The id is session-scoped and never persisted. No host stores operations, so if the host dies the worker, the id and the operation die together and the next launch starts with none open. -The host owns the operations and the lifecycle. `begin_operation` and `end_operation` are thin: the core forwards each to a host platform trait, scoped to the calling product. The host stores the product's open operations and keeps its worker alive while any stand. There's no separate keep-alive signal, because the operation existing is the signal. +The host owns operations and the lifecycle. The core forwards each call to a host trait scoped to the calling product, and the operation existing is the keep-alive signal. Every host implements the trait; one with no worker lifecycle, such as the headless CLI, hands out ids and tracks nothing. ```rust -/// Host store for a product's pending operations. The host keeps the -/// product's worker alive while it holds at least one open operation. -/// Optional: a host that omits it answers `begin_operation` `Unsupported`. #[async_trait] pub trait ProductOperations: Send + Sync { - /// Record a pending operation for this product. Returns its id. + /// `label` is empty when the product gave none. async fn begin_operation( &self, product: &ProductContext, - label: Option, - ) -> Result; + label: String, + ) -> Result; - /// Remove a pending operation. Idempotent: an unknown or already-ended - /// id returns `Ok`. + /// Idempotent. async fn end_operation( &self, product: &ProductContext, id: OperationId, - ) -> Result<(), GenericError>; + ) -> Result<(), HostWorkerOperationError>; } ``` -Ref-counted and product-scoped. Two tasks each begin an operation, and the worker stays alive until both end. The count belongs to the product, like its storage, so any open operation holds the product's worker whichever runtime opened it. - -Best-effort is the ceiling. On iOS and Android the OS can kill a backgrounded worker whatever the host does. An open operation lets the host ask for what background time the platform allows (a background task assertion on iOS, a foreground service or WorkManager on Android), but the worker still has to resume from saved state after a kill. Operations lower the odds of a mid-flight teardown. They don't remove it. +Operations are keyed by id and scoped to the product. Two tasks each begin one and the worker stays alive until both end. Ending an id twice, or one that was never begun, releases nothing. -Kept generic on purpose. An operation is opaque: an optional label for logs and a future host UI, no funding or deposit typing. It's a plain liveness signal the host can build on later, not a funding session (see future directions). A `status` field and a `list_operations` read belong to that later UI, not v1. v1 is begin and end. +Keep-alive is best-effort. iOS and Android can kill a backgrounded worker. An open operation lets the host request the background time the platform allows (a background task assertion on iOS, a foreground service or WorkManager on Android), and the worker resumes from saved state after a kill. -## Drawbacks - -Both features cost host work. Each of the three hosts (web worker, iOS, Android) implements `subscribe_storage` plus the identical-write skip, and `ProductOperations` with the worker lifecycle tied to it. The operations side is the awkward one, since keeping a process alive is an OS concern with no core-only answer. `subscribe_storage` is cheaper and reuses the `subscribe_chat_rooms` shape a host has likely written already. - -An open operation keeps a WASM instance resident, which costs battery. Best-effort teardown is the only guardrail in v1: a worker that never ends an operation pins itself, and with one product that's acceptable. The reaper that reclaims a stuck operation is deferred (see open questions). - -## Security and privacy - -The subscription's only new risk is scope leakage, and the core blocks it the same way `read` and `write` do: it passes the calling `ProductContext` to `subscribe_storage`, so there's no way to name another product's key. `begin_operation` and `end_operation` are gated to the `Worker` kind, like the Chat modality, so an app or widget can't call them, and an id from one worker means nothing to another. - -Neither feature moves new data across a boundary. The subscription carries values the product already owns. The operation `label`, if a host shows it, is product text and should be bounded and screened like any other. - -## Testing - -The subscription tests against a fake `ProductStorage`: subscribe, check the initial value, write and check the item, write identical bytes and check nothing fires, clear and check `None`, and check product A never sees product B's writes. The operation flow tests against a fake `ProductOperations`: `begin_operation` and `end_operation` reach the host scoped to the calling product, ending an unknown id is `Ok`, and product A can't end product B's operation. Whether an open operation actually keeps the worker resident is host behavior and needs a real device. +An operation is opaque: an optional label, no funding or deposit typing. ## Compatibility -All three wire methods are additive and break nothing (`localStorage.subscribe` at 198, `worker.beginOperation` at 202, `worker.endOperation` at 204). On the host side, `subscribe_storage` lands on the required `ProductStorage` trait and `ProductOperations` is a required capability too, so every host implements both. The byte-identical skip is in the core, not the host, so every host inherits it. Target is v0.2 / latest. - -## Unresolved questions - -- Reaping stuck operations. What reclaims an operation a worker opened and never ended? A time cap, a count cap, or the user cancelling it through a future UI. Deferred for v1 since one product isn't critical, but it has to exist before this is load-bearing for many products. -- Rapid distinct writes. Identical writes already drop. For a burst of different values on one key, does the host emit each or only the latest? Emitting each is the literal contract; coalescing saves a progress-bar consumer work. Either way, state it in the trait doc. +All three wire methods are additive. `subscribe_storage` and `ProductOperations` are required host capabilities. Target is v0.2 / latest. ## Future directions -Operations grow into a general liveness rule. The worker stays alive while `can_execute` holds, and pending operations are one term: `can_execute = has pending operations || has chats || has pocket cards || ...`. This RFC ships the first term. - -Around that, a host UI listing active operations, with `status` on each and a `list_operations` read to feed it, and a user cancel that ends the operation and releases the worker. A prefix subscription instead of a single storage key, if products want to watch a set at once. All are out of scope here and none needs a wire break later. +Operations become one term of a general liveness rule: `can_execute = has pending operations || has chats || has pocket cards || ...`. On top of that, a host UI listing active operations with `status`, a `list_operations` read to feed it, and user cancel. A prefix subscription if products want to watch a set of keys. diff --git a/js/packages/truapi-host/src/web/create-worker-host-runtime.ts b/js/packages/truapi-host/src/web/create-worker-host-runtime.ts index c3bc69f32..ef28c466e 100644 --- a/js/packages/truapi-host/src/web/create-worker-host-runtime.ts +++ b/js/packages/truapi-host/src/web/create-worker-host-runtime.ts @@ -17,6 +17,7 @@ import type { import { CustomRendererNode as CustomRendererNodeCodec, HostChatActionSubscribeItem as HostChatActionSubscribeItemCodec, + HostWorkerBeginOperationResponse as HostWorkerBeginOperationResponseCodec, } from "@parity/truapi"; import { PermissionAuthorizationRequest as PermissionAuthorizationRequestCodec } from "../generated/host-callbacks.js"; import { createWasmRawCallbacks } from "../generated/host-callbacks-adapter.js"; @@ -119,14 +120,15 @@ interface RuntimeState { >; subscriptionDisposers: Map void>; /** - * Open worker pending operations (`worker.beginOperation`). While this is - * above zero the worker is kept alive: a `dispose()` is deferred until the - * last operation ends. Worker-global, not per-core, because a + * Ids of open worker pending operations (`worker.beginOperation`). While + * this is non-empty the worker is kept alive: a `dispose()` is deferred + * until the last operation ends. Keyed by id so ending an unknown or + * already-ended id releases nothing. Worker-global, not per-core, because a * `callbackRequest` carries no core id and "keep the worker alive" is * worker-scoped. ponytail: no cap on how long an operation may hold the * worker; add a timeout ceiling here if a stuck operation becomes a problem. */ - operationCount: number; + openOperations: Set; /** A dispose() arrived while operations were open; run it once they drain. */ disposePending: boolean; chainConnections: Map; @@ -383,14 +385,19 @@ function handleCallbackRequest( .then(() => fn(...msg.args)) .then( (value) => { - // Keep the worker alive across an open pending operation: count begins - // and ends only on success, so a rejected begin never leaves a stuck - // count. When the last operation ends and a dispose is pending, run it. + // Keep the worker alive across an open pending operation: track ids + // only on success, so a rejected begin never leaves a stuck hold. + // When the last operation ends and a dispose is pending, run it. if (msg.name === "beginOperation") { - state.operationCount += 1; - } else if (msg.name === "endOperation" && state.operationCount > 0) { - state.operationCount -= 1; - if (state.operationCount === 0 && state.disposePending) { + state.openOperations.add( + HostWorkerBeginOperationResponseCodec.dec(value as Uint8Array).id, + ); + } else if (msg.name === "endOperation") { + const id = msg.args[1]; + if (typeof id === "number") { + state.openOperations.delete(id); + } + if (state.openOperations.size === 0 && state.disposePending) { state.disposePending = false; teardown(state, new Error("runtime disposed"), false); } @@ -804,7 +811,7 @@ export function createWebWorkerPairingHostRuntime( cores: new Map(), pendingCores: new Map(), subscriptionDisposers: new Map(), - operationCount: 0, + openOperations: new Set(), disposePending: false, chainConnections: new Map(), pendingDisconnects: new Map(), @@ -1249,7 +1256,7 @@ function buildRuntime(state: RuntimeState): WorkerPairingHostRuntime { // background task (e.g. a funding transaction) runs to completion. The // last endOperation runs the deferred teardown. Fault teardown is never // deferred. - if (state.operationCount > 0) { + if (state.openOperations.size > 0) { state.disposePending = true; return; } diff --git a/js/packages/truapi-host/src/web/worker-provider.test.ts b/js/packages/truapi-host/src/web/worker-provider.test.ts index 3e7379a24..84fa222dc 100644 --- a/js/packages/truapi-host/src/web/worker-provider.test.ts +++ b/js/packages/truapi-host/src/web/worker-provider.test.ts @@ -1001,6 +1001,59 @@ describe("createWebWorkerPairingHostRuntime", () => { expect(worker.terminated).toBe(true); }); + it("ending an unknown or already-ended operation releases no other hold", async () => { + const worker = new FakeWorker(); + let nextId = 1; + const providerPromise = createProviderFromRuntime( + asWorker(worker), + makeHostCallbacks({ + productOperations: { beginOperation: async () => ({ id: nextId++ }) }, + }), + { runtimeConfig: runtimeConfig({ executionKind: "Worker" }) }, + ); + worker.emit({ kind: "loaded" }); + worker.emit({ kind: "ready" }); + const provider = await finishProviderReady(worker, providerPromise); + + const product = ProductContext.enc({ + productId: "dotli.dot", + executionKind: "Worker", + }); + const begin = (requestId: number) => + worker.emit({ + kind: "callbackRequest", + requestId, + name: "beginOperation", + args: [product, ""], + }); + const end = (requestId: number, id: number) => + worker.emit({ + kind: "callbackRequest", + requestId, + name: "endOperation", + args: [product, id], + }); + + begin(1); + begin(2); + await settle(); + provider.dispose(); + + // Operation 1 ends twice and an unknown id ends once. Operation 2 still + // holds the worker. + end(3, 1); + end(4, 1); + end(5, 99); + await settle(); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(worker.terminated).toBe(false); + + end(6, 2); + await settle(); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(worker.terminated).toBe(true); + }); + it("routes payload-carrying subscriptions by name", async () => { const worker = new FakeWorker(); const keys: Uint8Array[] = []; diff --git a/rust/crates/truapi-server/src/runtime.rs b/rust/crates/truapi-server/src/runtime.rs index cd841e69e..faf50e414 100644 --- a/rust/crates/truapi-server/src/runtime.rs +++ b/rust/crates/truapi-server/src/runtime.rs @@ -4930,6 +4930,171 @@ mod tests { ); } + fn storage_item(value: Option<&[u8]>) -> HostLocalStorageChangeItem { + HostLocalStorageChangeItem::V1(v01::HostLocalStorageChangeItem { + value: value.map(<[u8]>::to_vec), + }) + } + + fn subscribe_storage_key( + host: &ProductRuntimeHost, + key: &str, + ) -> Subscription { + futures::executor::block_on(LocalStorage::subscribe( + host, + &CallContext::default(), + HostLocalStorageSubscribeRequest::V1(v01::HostLocalStorageSubscribeRequest { + key: key.to_string(), + }), + )) + } + + fn write_storage_key(host: &ProductRuntimeHost, key: &str, value: &[u8]) { + futures::executor::block_on(host.write( + &CallContext::default(), + HostLocalStorageWriteRequest::V1(v01::HostLocalStorageWriteRequest { + key: key.to_string(), + value: value.to_vec(), + }), + )) + .expect("storage write"); + } + + #[test] + fn local_storage_subscribe_sees_writes_and_clears_but_not_identical_rewrites() { + let platform = stub_platform(); + let host = ProductRuntimeHost::new( + platform.clone(), + runtime_config("myapp.dot"), + test_spawner(), + ); + let mut subscription = subscribe_storage_key(&host, "progress"); + let next = |subscription: &mut Subscription| { + futures::executor::block_on(subscription.next()).expect("storage item") + }; + + assert_eq!( + next(&mut subscription), + storage_item(None), + "the first item is the current value" + ); + + write_storage_key(&host, "progress", b"1"); + assert_eq!(next(&mut subscription), storage_item(Some(b"1"))); + + write_storage_key(&host, "progress", b"1"); + assert!( + futures::FutureExt::now_or_never(subscription.next()).is_none(), + "a byte-identical rewrite emits nothing" + ); + assert_eq!( + platform + .local_storage_writes + .lock() + .expect("local storage writes mutex poisoned") + .len(), + 1, + "the core skips the identical write before it reaches the platform" + ); + + futures::executor::block_on(host.clear( + &CallContext::default(), + HostLocalStorageClearRequest::V1(v01::HostLocalStorageClearRequest { + key: "progress".to_string(), + }), + )) + .expect("storage clear"); + assert_eq!(next(&mut subscription), storage_item(None)); + } + + #[test] + fn local_storage_subscribe_is_scoped_to_the_calling_product() { + let platform = stub_platform(); + let mine = ProductRuntimeHost::new( + platform.clone(), + runtime_config("myapp.dot"), + test_spawner(), + ); + let other = ProductRuntimeHost::new(platform, runtime_config("other.dot"), test_spawner()); + write_storage_key(&mine, "shared", b"mine"); + + let mut mine_items = subscribe_storage_key(&mine, "shared"); + let mut other_items = subscribe_storage_key(&other, "shared"); + assert_eq!( + futures::executor::block_on(mine_items.next()), + Some(storage_item(Some(b"mine"))) + ); + assert_eq!( + futures::executor::block_on(other_items.next()), + Some(storage_item(None)), + "the same key name in another product is a different key" + ); + + write_storage_key(&mine, "shared", b"again"); + assert_eq!( + futures::executor::block_on(mine_items.next()), + Some(storage_item(Some(b"again"))) + ); + assert!( + futures::FutureExt::now_or_never(other_items.next()).is_none(), + "another product's write never reaches this subscriber" + ); + } + + #[test] + fn worker_operations_reach_the_platform_scoped_to_the_calling_product() { + let platform = stub_platform(); + let host = ProductRuntimeHost::new( + platform.clone(), + runtime_config("myapp.dot"), + test_spawner(), + ); + let cx = CallContext::default(); + let begin = |label: Option<&str>| { + let HostWorkerBeginOperationResponse::V1(response) = + futures::executor::block_on(host.begin_operation( + &cx, + HostWorkerBeginOperationRequest::V1(v01::HostWorkerBeginOperationRequest { + label: label.map(str::to_string), + }), + )) + .expect("begin operation"); + response.id + }; + let end = |id: u32| { + futures::executor::block_on(host.end_operation( + &cx, + HostWorkerEndOperationRequest::V1(v01::HostWorkerEndOperationRequest { id }), + )) + .expect("end operation") + }; + + assert_eq!(begin(Some("funding")), 1); + assert_eq!(begin(None), 2); + assert_eq!( + *platform + .begun_operations + .lock() + .expect("begun operations mutex poisoned"), + vec![ + ("myapp.dot".to_string(), "funding".to_string()), + ("myapp.dot".to_string(), String::new()), + ], + "begin reaches the platform under the calling product; a missing label is empty" + ); + + end(1); + end(99); + assert_eq!( + *platform + .ended_operations + .lock() + .expect("ended operations mutex poisoned"), + vec![("myapp.dot".to_string(), 1), ("myapp.dot".to_string(), 99)], + "end reaches the platform under the calling product, unknown ids included" + ); + } + #[test] fn theme_subscribe_maps_platform_values() { let host = ProductRuntimeHost::new_compat(stub_platform(), test_spawner()); diff --git a/rust/crates/truapi-server/src/test_support.rs b/rust/crates/truapi-server/src/test_support.rs index 6f964fa1e..8b82cd25b 100644 --- a/rust/crates/truapi-server/src/test_support.rs +++ b/rust/crates/truapi-server/src/test_support.rs @@ -18,7 +18,7 @@ use crate::subscription::Spawner; use crate::subscription::thread_per_subscription_spawner; use futures::Stream; -use futures::stream::{self, BoxStream}; +use futures::stream::{self, BoxStream, StreamExt}; use parity_scale_codec::{Decode, Encode}; use schnorrkel::{ExpansionMode, MiniSecretKey}; use truapi::v01; @@ -158,10 +158,45 @@ pub(crate) struct StubPlatform { /// forged value to exercise the in-core integrity check. pub(crate) preimage_lookup_value: Option>, pub(crate) local_storage: Arc>>>, + /// Every product storage write that reached the platform, in order, so a + /// test can see which writes the core skipped. + pub(crate) local_storage_writes: Arc>>, + /// Open `subscribe_storage` streams by namespaced key; `write` and + /// `clear` push each change to the matching ones. + pub(crate) storage_subscribers: Arc>>, + /// Every `begin_operation` as `(product_id, label)`, in order. The + /// returned id is the call's 1-based position. + pub(crate) begun_operations: Arc>>, + /// Every `end_operation` as `(product_id, id)`, in order. + pub(crate) ended_operations: Arc>>, /// When set, product/core storage reads fail with this reason. pub(crate) local_storage_error: Option<&'static str>, } +/// One product storage write as the platform saw it: namespaced key and bytes. +pub(crate) type StorageWrite = (String, Vec); + +/// Sender side of one stubbed storage subscription. +pub(crate) type StorageChangeSender = futures::channel::mpsc::UnboundedSender< + Result, +>; + +impl StubPlatform { + fn push_storage_change(&self, key: &str, value: Option>) { + self.storage_subscribers + .lock() + .expect("storage subscribers mutex poisoned") + .retain(|(subscribed, tx)| { + subscribed != key + || tx + .unbounded_send(Ok(v01::HostLocalStorageChangeItem { + value: value.clone(), + })) + .is_ok() + }); + } +} + /// Scripted peer behavior for the recording connection's SSO exchange. #[derive(Clone)] pub(crate) enum SsoResponseScript { @@ -776,10 +811,15 @@ impl PlatformProductStorage for StubPlatform { reason: reason.to_string(), }); } + self.local_storage_writes + .lock() + .expect("local storage writes mutex poisoned") + .push((key.clone(), value.clone())); self.local_storage .lock() .expect("local storage mutex poisoned") - .insert(key, value); + .insert(key.clone(), value.clone()); + self.push_storage_change(&key, Some(value)); Ok(()) } async fn clear(&self, key: String) -> Result<(), v01::HostLocalStorageReadError> { @@ -792,6 +832,7 @@ impl PlatformProductStorage for StubPlatform { .lock() .expect("local storage mutex poisoned") .remove(&key); + self.push_storage_change(&key, None); Ok(()) } @@ -800,15 +841,20 @@ impl PlatformProductStorage for StubPlatform { key: Vec, ) -> BoxStream<'static, Result> { let key = String::from_utf8_lossy(&key).into_owned(); + let (tx, rx) = futures::channel::mpsc::unbounded(); + self.storage_subscribers + .lock() + .expect("storage subscribers mutex poisoned") + .push((key.clone(), tx)); let value = self .local_storage .lock() .expect("local storage mutex poisoned") .get(&key) .cloned(); - Box::pin(stream::once(async move { - Ok(v01::HostLocalStorageChangeItem { value }) - })) + Box::pin( + stream::once(async move { Ok(v01::HostLocalStorageChangeItem { value }) }).chain(rx), + ) } } @@ -816,18 +862,30 @@ impl PlatformProductStorage for StubPlatform { impl PlatformProductOperations for StubPlatform { async fn begin_operation( &self, - _product: &ProductContext, - _label: String, + product: &ProductContext, + label: String, ) -> Result { - // The stub has no worker lifecycle to keep alive; a fixed id suffices. - Ok(v01::HostWorkerBeginOperationResponse { id: 1 }) + // The stub has no worker lifecycle to keep alive; it only records the + // call and hands back its position as the id. + let mut begun = self + .begun_operations + .lock() + .expect("begun operations mutex poisoned"); + begun.push((product.product_id.clone(), label)); + Ok(v01::HostWorkerBeginOperationResponse { + id: begun.len() as u32, + }) } async fn end_operation( &self, - _product: &ProductContext, - _id: u32, + product: &ProductContext, + id: u32, ) -> Result<(), v01::HostWorkerOperationError> { + self.ended_operations + .lock() + .expect("ended operations mutex poisoned") + .push((product.product_id.clone(), id)); Ok(()) } } From ea1a58f8289304f33ac871cdb8485c297dbb236a Mon Sep 17 00:00:00 2001 From: Sergey Zhuravlev Date: Fri, 4 Sep 2026 18:58:41 +0200 Subject: [PATCH 3/3] test(playground): pin the chat diagnosis guard to the Chat service The guard compared CHAT_DIAGNOSIS_METHODS against every service that requires Worker execution. The Worker service (begin_operation and end_operation) is also Worker-only, so the Chat-specific diagnosis no longer matched. Filter by service name instead. --- playground/tests/unit/chat-diagnosis.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/playground/tests/unit/chat-diagnosis.test.ts b/playground/tests/unit/chat-diagnosis.test.ts index 866ca004d..b8a2e4800 100644 --- a/playground/tests/unit/chat-diagnosis.test.ts +++ b/playground/tests/unit/chat-diagnosis.test.ts @@ -7,7 +7,7 @@ describe("ChatDiagnosis", () => { // Expectation comes from codegen, so a missing method fails here. test("covers every generated Chat method", () => { const generated = servicesForExecution(generatedServices, "Worker") - .filter((service) => service.requiredExecution === "Worker") + .filter((service) => service.name === "Chat") .flatMap((service) => service.methods.map((method) => `${service.name}/${method.name}`), );