diff --git a/.changeset/funding-provider-battery.md b/.changeset/funding-provider-battery.md
new file mode 100644
index 0000000000..79bbc6b6d7
--- /dev/null
+++ b/.changeset/funding-provider-battery.md
@@ -0,0 +1,5 @@
+---
+"@parity/truapi": patch
+---
+
+`make e2e-funding-cli` also runs a provider worker: the scripted funding host hands `provide` sessions to the product that asked, completes the top-ups and payment requests it starts, and the battery resumes an inbound session after a host restart.
diff --git a/.changeset/funding-provider.md b/.changeset/funding-provider.md
new file mode 100644
index 0000000000..2e0350cf04
--- /dev/null
+++ b/.changeset/funding-provider.md
@@ -0,0 +1,5 @@
+---
+"@parity/truapi": major
+---
+
+New `FundingProvider` trait for a provider's worker: `serveSubscribe` receives the sessions assigned to it and cancel requests, `report` stores its progress on the session, and `presentFrame` shows one of its screens in a host frame. Sessions end as `Delivered` or `Released` from the claims of the top-ups, or the completion of the payment request, the provider names. Hosts assign a session with `select_funding_provider`, and must implement `present_provider_frame` on `FundingPlatform` (natively `NativeFundingCallbacks`).
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7f3646a2ea..a999387160 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -145,6 +145,56 @@ jobs:
cat "$RUNNER_TEMP/headless-smoke.log"
grep -q headless-install-ok "$RUNNER_TEMP/headless-smoke.log"
+ funding-battery:
+ name: Funding battery (CLI host)
+ needs: [changes, codegen]
+ if: needs.changes.outputs.funding_battery == 'true'
+ timeout-minutes: 30
+ runs-on: ubuntu-latest
+ env:
+ TRUAPI_HOST_NO_UPDATE: "1"
+ # A fixed development signer, so the host signs in without provisioning
+ # an account on chain. Funding needs a session, not a funded account.
+ HOST_CLI_SIGNER_MNEMONIC: "bottom drive obey lake curtain smoke basket hold race lonely fit walk"
+ BATTERY_PHASE_TIMEOUT: "600"
+ steps:
+ - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+ with:
+ persist-credentials: false
+
+ - uses: dtolnay/rust-toolchain@5b842231ba77f5c045dba54ac5560fed2db780e2 # stable
+ with:
+ toolchain: stable
+
+ - uses: ./.github/actions/rust-cache
+
+ - uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
+ with:
+ node-version: 24
+
+ - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2.2.0
+ with:
+ bun-version: "1.4.2"
+
+ - name: Download codegen output
+ uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+ with:
+ name: codegen-output
+
+ - name: Isolate the host state
+ run: echo "TRUAPI_HOST_BASE_PATH=$RUNNER_TEMP/funding-host-state" >> "$GITHUB_ENV"
+
+ - name: Run the funding battery
+ run: make e2e-funding-cli
+
+ - name: Upload host logs and transcripts
+ if: failure()
+ uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+ with:
+ name: funding-battery-logs
+ path: target/battery
+ retention-days: 14
+
wasm-provider:
name: truapi-provider (wasm browser tests)
timeout-minutes: 10
@@ -284,6 +334,7 @@ jobs:
sdk_provider_kotlin: ${{ steps.filter.outputs.sdk_provider_kotlin }}
host_android: ${{ steps.filter.outputs.host_android }}
cli_package: ${{ steps.filter.outputs.cli_package }}
+ funding_battery: ${{ steps.filter.outputs.funding_battery }}
workflows: ${{ steps.filter.outputs.workflows }}
needs_changeset: ${{ steps.filter.outputs.needs_changeset }}
adds_changeset: ${{ steps.filter.outputs.adds_changeset }}
@@ -322,6 +373,7 @@ jobs:
echo "sdk_provider_kotlin=true"
echo "host_android=true"
echo "cli_package=true"
+ echo "funding_battery=true"
echo "workflows=true"
} >> "$GITHUB_OUTPUT"
exit 0
@@ -357,6 +409,10 @@ jobs:
# source, and bundling it is a step no other job performs. A release
# is the wrong place to find out one of them moved.
gate cli_package '^(rust/crates/(truapi-host-cli|truapi|truapi-codegen|truapi-macros)/|js/container/|js/packages/truapi/(src/|package\.json$|tsconfig[^/]*\.json$)|Makefile$|nightly-toolchain$|Cargo\.toml$|Cargo\.lock$|package\.json$|package-lock\.json$|scripts/(codegen\.sh|build-cli-runner\.ts|cli-runner-package\.test\.ts)$|\.github/workflows/(ci|cli-package|release-cli)\.yml$)'
+ # The funding battery drives the core's funding and provider surface
+ # through the CLI host and the generated client, so it reads every
+ # crate the host is built from, the client, and the scripts it runs.
+ gate funding_battery '^(rust/crates/(truapi|truapi-host-cli|truapi-codegen|truapi-macros|truapi-provider|truapi-verifiable)/|js/packages/truapi/(src/|package\.json$|tsconfig[^/]*\.json$)|js/container/|playground/(package\.json|yarn\.lock)$|scripts/(battery|codegen)\.sh$|Makefile$|nightly-toolchain$|Cargo\.toml$|Cargo\.lock$|package\.json$|package-lock\.json$|\.github/workflows/ci\.yml$)'
# Include generators, dependency pins and package build configuration:
# these can change a published artifact without touching its sources.
@@ -1311,6 +1367,7 @@ jobs:
[
rust,
headless-install,
+ funding-battery,
wasm-provider,
licenses,
codegen,
@@ -1368,7 +1425,7 @@ jobs:
env:
NEEDS: ${{ toJSON(needs) }}
REQUIRED: >-
- rust headless-install wasm-provider licenses codegen ios-bindings changes ios-swift
+ rust headless-install funding-battery wasm-provider licenses codegen ios-bindings changes ios-swift
android-bindings provider-android-bindings host-android-bindings
host-android-detekt
cli-package ts-client
diff --git a/README.md b/README.md
index c6ffd30494..45749e9ba5 100644
--- a/README.md
+++ b/README.md
@@ -306,7 +306,7 @@ make e2e-signing-cli # same direct signing-host phase
make e2e-pairing-cli # same paired pairing-host phase
make e2e-chat-cli # chat content screening against a chat signing-host
make e2e-pocket-cli # Pocket protocol check against a Pocket signing-host
-make e2e-funding-cli # funding requests and statuses against a scripted funding host
+make e2e-funding-cli # funding requests and a provider worker against a scripted funding host
```
The Pocket phase runs its product as a Worker execution, the only execution
diff --git a/explorer/diagnosis-reports/funding/signing-host-cli.md b/explorer/diagnosis-reports/funding/signing-host-cli.md
index 549760394a..87b29f02a3 100644
--- a/explorer/diagnosis-reports/funding/signing-host-cli.md
+++ b/explorer/diagnosis-reports/funding/signing-host-cli.md
@@ -7,3 +7,11 @@
| `Funding/request_failed` | ✅ | |
| `Funding/request_dismissed` | ✅ | |
| `Funding/status_subscribe_unknown` | ✅ | |
+| `Funding/provider_assigned` | ✅ | |
+| `Funding/provider_present_frame` | ✅ | |
+| `Funding/provider_reports_forward_only` | ✅ | |
+| `Funding/provider_credits_through_top_up` | ✅ | |
+| `Funding/provider_out_released` | ✅ | |
+| `Funding/provider_cancel` | ✅ | |
+| `Funding/provider_resumes_after_restart` | ✅ | |
+| `Funding/provider_in_delivered` | ✅ | |
diff --git a/rust/crates/truapi-codegen/src/rust.rs b/rust/crates/truapi-codegen/src/rust.rs
index 8ede859db6..eeb942a6b3 100644
--- a/rust/crates/truapi-codegen/src/rust.rs
+++ b/rust/crates/truapi-codegen/src/rust.rs
@@ -62,6 +62,7 @@ const TRAIT_MODULE_MAP: &[(&str, &str)] = &[
("Contacts", "contacts"),
("Entropy", "entropy"),
("Funding", "funding"),
+ ("FundingProvider", "funding_provider"),
("JsonRpc", "jsonrpc"),
("LocalStorage", "local_storage"),
("Payment", "payment"),
diff --git a/rust/crates/truapi-codegen/tests/golden/dispatcher.rs b/rust/crates/truapi-codegen/tests/golden/dispatcher.rs
index 05e0c82801..3ba6ceddf3 100644
--- a/rust/crates/truapi-codegen/tests/golden/dispatcher.rs
+++ b/rust/crates/truapi-codegen/tests/golden/dispatcher.rs
@@ -20,6 +20,7 @@ use truapi::api::{
Contacts,
Entropy,
Funding,
+ FundingProvider,
LocalStorage,
Locale,
Notifications,
@@ -58,6 +59,7 @@ where
register_contacts(dispatcher, host.clone());
register_entropy(dispatcher, host.clone());
register_funding(dispatcher, host.clone());
+ register_funding_provider(dispatcher, host.clone());
register_local_storage(dispatcher, host.clone());
register_locale(dispatcher, host.clone());
register_notifications(dispatcher, host.clone());
@@ -1401,6 +1403,117 @@ where
}
}
+fn register_funding_provider(dispatcher: &mut Dispatcher, host: Arc
)
+where
+ P: FundingProvider + Send + Sync + 'static,
+{
+ {
+ let execution_allowed = dispatcher.allows_execution(ProductExecutionKind::Worker);
+ let host = host.clone();
+ dispatcher.on_subscription(wire_table::FUNDING_PROVIDER_SERVE_SUBSCRIBE, move |request_id: String, bytes: Vec| {
+ let host = host.clone();
+ Box::pin(async move {
+ let request: versioned::funding_provider::HostFundingServeSubscribeRequest = match DecodeAll::decode_all(&mut &bytes[..]) {
+ Ok(request) => request,
+ Err(err) => {
+ let error: truapi::CallError =
+ truapi::CallError::MalformedFrame { reason: err.to_string() };
+ return Err(subscription_interrupt(error));
+ }
+ };
+ let target_version = request.version();
+ let cx = CallContext::with_request_id(request_id);
+ if !execution_allowed {
+ let error: truapi::CallError = truapi::CallError::Denied;
+ return Err(subscription_interrupt(error));
+ }
+ let stream = host.serve_subscribe(&cx, request).await;
+ let stream = futures::StreamExt::map(
+ stream,
+ move |item: Result>| {
+ item.map(|item| {
+ ::from_latest(
+ truapi::versioned::IntoLatest::into_latest(item),
+ target_version,
+ )
+ })
+ .map_err(|error| downgrade_call_error(error, target_version))
+ },
+ );
+ Ok(subscription_stream(stream))
+ })
+ });
+ }
+ {
+ let execution_allowed = dispatcher.allows_execution(ProductExecutionKind::Worker);
+ let host = host.clone();
+ dispatcher.on_request(wire_table::FUNDING_PROVIDER_REPORT, move |request_id: String, bytes: Vec, cancel: truapi::CancellationToken| {
+ let host = host.clone();
+ Box::pin(async move {
+ let request: versioned::funding_provider::HostFundingReportRequest = match DecodeAll::decode_all(&mut &bytes[..]) {
+ Ok(request) => request,
+ Err(err) => {
+ let error: truapi::CallError =
+ truapi::CallError::MalformedFrame { reason: err.to_string() };
+ let result: Result> = Err(error);
+ return result.encode();
+ }
+ };
+ let target_version = request.version();
+ let cx = CallContext::with_parts(request_id, cancel);
+ if !execution_allowed {
+ let error: truapi::CallError = truapi::CallError::Denied;
+ let result: Result> = Err(error);
+ return result.encode();
+ }
+ let result: Result> =
+ match host.report(&cx, request).await {
+ Ok(response) => Ok(::from_latest(
+ truapi::versioned::IntoLatest::into_latest(response),
+ target_version,
+ )),
+ Err(err) => Err(downgrade_call_error(err, target_version)),
+ };
+ result.encode()
+ })
+ });
+ }
+ {
+ let execution_allowed = dispatcher.allows_execution(ProductExecutionKind::Worker);
+ let host = host;
+ dispatcher.on_request(wire_table::FUNDING_PROVIDER_PRESENT_FRAME, move |request_id: String, bytes: Vec, cancel: truapi::CancellationToken| {
+ let host = host.clone();
+ Box::pin(async move {
+ let request: versioned::funding_provider::HostFundingPresentFrameRequest = match DecodeAll::decode_all(&mut &bytes[..]) {
+ Ok(request) => request,
+ Err(err) => {
+ let error: truapi::CallError =
+ truapi::CallError::MalformedFrame { reason: err.to_string() };
+ let result: Result> = Err(error);
+ return result.encode();
+ }
+ };
+ let target_version = request.version();
+ let cx = CallContext::with_parts(request_id, cancel);
+ if !execution_allowed {
+ let error: truapi::CallError = truapi::CallError::Denied;
+ let result: Result> = Err(error);
+ return result.encode();
+ }
+ let result: Result> =
+ match host.present_frame(&cx, request).await {
+ Ok(response) => Ok(::from_latest(
+ truapi::versioned::IntoLatest::into_latest(response),
+ target_version,
+ )),
+ Err(err) => Err(downgrade_call_error(err, target_version)),
+ };
+ result.encode()
+ })
+ });
+ }
+}
+
fn register_local_storage(dispatcher: &mut Dispatcher, host: Arc
)
where
P: LocalStorage + Send + Sync + 'static,
diff --git a/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts b/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts
index e9d62b8ef5..bf72b4ee56 100644
--- a/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts
+++ b/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts
@@ -26,6 +26,7 @@ import {
} from "@parity/truapi";
import type {
+ FundingFrameOutcome,
GenericError,
HostChatCreateRoomRequest,
HostChatCreateRoomResponse,
@@ -1461,6 +1462,16 @@ export interface FundingPlatform {
session: FundingPresentation,
): Promise;
+ /**
+ * Show `provider`'s screen at `route` for session `intent`, such as its
+ * KYC or card entry, in a frame the host owns, and answer once it closes.
+ */
+ presentProviderFrame(
+ provider: ProductContext,
+ intent: string,
+ route: string,
+ ): Promise;
+
/**
* Observe a session's status change, for host UI such as the in-flight
* pill.
diff --git a/rust/crates/truapi-codegen/tests/golden/wire_table.rs b/rust/crates/truapi-codegen/tests/golden/wire_table.rs
index 96976ddf4b..4ab1922e11 100644
--- a/rust/crates/truapi-codegen/tests/golden/wire_table.rs
+++ b/rust/crates/truapi-codegen/tests/golden/wire_table.rs
@@ -42,7 +42,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 = "8558b0cd19000411";
+pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "04b772501af2b00d";
/// Wire discriminants for `system_handshake`.
pub const SYSTEM_HANDSHAKE: MethodIds = MethodIds {
@@ -560,6 +560,24 @@ pub const FUNDING_STATUS_SUBSCRIBE: MethodIds = MethodIds {
method_id: 1,
};
+/// Wire discriminants for `funding_provider_serve_subscribe`.
+pub const FUNDING_PROVIDER_SERVE_SUBSCRIBE: MethodIds = MethodIds {
+ trait_id: 22,
+ method_id: 0,
+};
+
+/// Wire discriminants for `funding_provider_report`.
+pub const FUNDING_PROVIDER_REPORT: MethodIds = MethodIds {
+ trait_id: 22,
+ method_id: 1,
+};
+
+/// Wire discriminants for `funding_provider_present_frame`.
+pub const FUNDING_PROVIDER_PRESENT_FRAME: MethodIds = MethodIds {
+ trait_id: 22,
+ method_id: 2,
+};
+
/// The full wire table. Trait ids and per-trait method ordering are
/// part of the wire protocol; only ever append within a trait.
/// Removed methods leave their slot empty.
@@ -908,4 +926,16 @@ pub const WIRE_TABLE: &[WireEntry] = &[
method: "funding_status_subscribe",
kind: WireKind::Subscription(FUNDING_STATUS_SUBSCRIBE),
},
+ WireEntry {
+ method: "funding_provider_serve_subscribe",
+ kind: WireKind::Subscription(FUNDING_PROVIDER_SERVE_SUBSCRIBE),
+ },
+ WireEntry {
+ method: "funding_provider_report",
+ kind: WireKind::Request(FUNDING_PROVIDER_REPORT),
+ },
+ WireEntry {
+ method: "funding_provider_present_frame",
+ kind: WireKind::Request(FUNDING_PROVIDER_PRESENT_FRAME),
+ },
];
diff --git a/rust/crates/truapi-host-cli/README.md b/rust/crates/truapi-host-cli/README.md
index d54ff85b5c..b8192506b1 100644
--- a/rust/crates/truapi-host-cli/README.md
+++ b/rust/crates/truapi-host-cli/README.md
@@ -774,10 +774,19 @@ Scripts under `js/scripts/` include:
The funding phase builds the host with `--features test-host` and installs
a scripted funding overlay: each request the product makes is answered with
the next outcome from `TRUAPI_FUNDING_OUTCOMES`
- (`deliver:900,release:500,fail,dismiss`), and a started session is settled
- through the core's test hook with no chain behind it. Every overlay request
- and session change is recorded in `TRUAPI_FUNDING_LOG`, which the cases read.
- The live on-ramp on Paseo Next is `truapi-host funding-check`.
+ (`deliver:900,release:500,fail,dismiss,provide,provide,provide-cancel`). A
+ `deliver`, `release` or `fail` session is settled through the core's test
+ hook with no chain behind it. A `provide` session is handed to the product
+ that asked, which runs as a Worker and serves it as the provider: it reports
+ progress, starts top-ups and payment requests, which scripted engines
+ complete in full and record in `TRUAPI_FUNDING_LEDGER`, and the core settles
+ the session from them. `provide-cancel` also cancels it. The phase runs the
+ host twice on the same storage, so the provider resumes an inbound session
+ after a restart. Every overlay request and session change is recorded in
+ `TRUAPI_FUNDING_LOG`, which the cases read. CI runs this phase on every
+ change to the core, the CLI or the client, with a fixed development signer
+ in `HOST_CLI_SIGNER_MNEMONIC` so the host needs no account provisioned on
+ chain.
Contacts are served on every phase, from `TRUAPI_CONTACTS`
(`alice=0x<32-byte account>;bob=0x…`) or, unset, from a two-name development
diff --git a/rust/crates/truapi-host-cli/js/funding-e2e.ts b/rust/crates/truapi-host-cli/js/funding-e2e.ts
index adc5c5ce00..07d1cd3a87 100644
--- a/rust/crates/truapi-host-cli/js/funding-e2e.ts
+++ b/rust/crates/truapi-host-cli/js/funding-e2e.ts
@@ -18,19 +18,20 @@ import type { DiagnosisRow } from "./diagnosis.ts";
// The cases make their requests in the order `scripts/battery.sh` scripts the
// host's outcomes: deliver:900, release:500, fail, dismiss.
-const WAIT_MS = 15_000;
+export const WAIT_MS = 15_000;
-interface TranscriptLine {
+export interface TranscriptLine {
kind: string;
intent?: string;
direction?: string;
amount?: string | null;
outcome?: string;
tag?: string;
+ route?: string;
}
/** What the host recorded. */
-function transcript(path: string): TranscriptLine[] {
+export function transcript(path: string): TranscriptLine[] {
if (!existsSync(path)) {
return [];
}
@@ -49,7 +50,7 @@ function transcript(path: string): TranscriptLine[] {
/** Every item a stream emits until it completes, or a rejection once
* `WAIT_MS` passes or the stream fails. */
-function untilComplete- (
+export function untilComplete
- (
stream: ObservableLike
- ,
what: string,
): Promise {
@@ -103,7 +104,7 @@ function interruption
- (
}
/** Wait until the host's transcript holds a line matching `predicate`. */
-async function waitForTranscript(
+export async function waitForTranscript(
path: string,
predicate: (line: TranscriptLine) => boolean,
what: string,
@@ -119,7 +120,7 @@ async function waitForTranscript(
}
}
-const stringify = (value: unknown) =>
+export const stringify = (value: unknown) =>
JSON.stringify(value, (_, inner) =>
typeof inner === "bigint" ? inner.toString() : inner,
);
diff --git a/rust/crates/truapi-host-cli/js/funding-provider-e2e.ts b/rust/crates/truapi-host-cli/js/funding-provider-e2e.ts
new file mode 100644
index 0000000000..34d8c2e980
--- /dev/null
+++ b/rust/crates/truapi-host-cli/js/funding-provider-e2e.ts
@@ -0,0 +1,338 @@
+// End-to-end check of a provider worker serving funding sessions, against a
+// host built with `--features test-host` whose scripted overlay answers
+// `provide` or `provide-cancel`: it hands the session to the product that
+// asked, so this script is both the product requesting funds and the
+// provider's worker serving the request.
+//
+// The run is split across a host restart. `runProviderStart` takes an
+// inbound session as far as its top-up, and `runProviderResume`, in a new host
+// process on the same storage, checks the session is handed over again and
+// lets the core deliver it from the claim. Top-ups and payments go to the
+// host's scripted engines, which complete them in full.
+import type {
+ ObservableLike,
+ TrUApiClient,
+} from "../../../../js/packages/truapi/src/index.ts";
+import type {
+ FundingUpdate,
+ HostFundingServeSubscribeItem,
+ HostFundingStatusSubscribeItem,
+} from "../../../../js/packages/truapi/src/generated/types.ts";
+import type { DiagnosisRow } from "./diagnosis.ts";
+import {
+ WAIT_MS,
+ stringify,
+ untilComplete,
+ waitForTranscript,
+} from "./funding-e2e.ts";
+
+/** What the first run leaves for the run after the restart. */
+export interface ProviderState {
+ intent: string;
+ topUpId: string;
+ amount: string;
+}
+
+type HexString = `0x${string}`;
+
+const IN_AMOUNT = 1_000n;
+const OUT_AMOUNT = 500n;
+const CANCEL_AMOUNT = 300n;
+
+function randomId(): HexString {
+ const bytes = crypto.getRandomValues(new Uint8Array(32));
+ return `0x${Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join("")}`;
+}
+
+/** Every item the provider's serve stream emits, kept as they arrive. */
+class Served {
+ private readonly items: HostFundingServeSubscribeItem[] = [];
+ private failure: unknown;
+ private readonly subscription: { unsubscribe(): void };
+
+ constructor(stream: ObservableLike) {
+ this.subscription = stream.subscribe({
+ next: (item) => {
+ this.items.push(item);
+ },
+ error: (reason: unknown) => {
+ this.failure = reason;
+ },
+ complete: () => {},
+ });
+ }
+
+ /** The first item matching `predicate`, waiting up to `WAIT_MS`. */
+ async next(
+ predicate: (item: HostFundingServeSubscribeItem) => boolean,
+ what: string,
+ ): Promise {
+ const deadline = Date.now() + WAIT_MS;
+ for (;;) {
+ const hit = this.items.find(predicate);
+ if (hit) return hit;
+ if (this.failure !== undefined) {
+ throw new Error(`serve stream failed: ${stringify(this.failure)}`);
+ }
+ if (Date.now() >= deadline) {
+ throw new Error(`never served ${what}`);
+ }
+ await new Promise((resolve) => setTimeout(resolve, 50));
+ }
+ }
+
+ close() {
+ this.subscription.unsubscribe();
+ }
+}
+
+const assignedTo = (intent: string) => (item: HostFundingServeSubscribeItem) =>
+ item.tag === "Assigned" && item.value.session.intent === intent;
+
+function rowFactory(rows: DiagnosisRow[]) {
+ return async (methodName: string, check: () => Promise) => {
+ const startedAt = performance.now();
+ const row = (status: DiagnosisRow["status"], output: string) =>
+ rows.push({
+ id: `Funding/${methodName}`,
+ serviceName: "Funding",
+ methodName,
+ status,
+ output,
+ durationMs: Math.round(performance.now() - startedAt),
+ });
+ try {
+ row("pass", await check());
+ } catch (error) {
+ // One stuck case is recorded and the rest still run.
+ row("fail", error instanceof Error ? error.message : String(error));
+ }
+ };
+}
+
+async function report(
+ client: TrUApiClient,
+ intent: string,
+ update: FundingUpdate,
+): Promise {
+ const reported = await client.fundingProvider.report({ intent, update });
+ if (reported.isErr()) {
+ throw new Error(
+ `report ${update.tag} refused: ${stringify(reported.error)}`,
+ );
+ }
+}
+
+async function requestFunding(
+ client: TrUApiClient,
+ direction: "In" | "Out",
+ amount: bigint,
+): Promise {
+ const requested = await client.funding.request({ direction, amount });
+ if (requested.isErr()) {
+ throw new Error(`funding.request failed: ${stringify(requested.error)}`);
+ }
+ return requested.value.intent;
+}
+
+async function lastStatus(
+ client: TrUApiClient,
+ intent: string,
+): Promise {
+ const items = await untilComplete(
+ client.funding.statusSubscribe({ request: { intent } }),
+ `session ${intent}`,
+ );
+ return items.at(-1);
+}
+
+/** The first run: assignment, a provider screen, forward-only reports, the
+ * outbound and cancelled sessions, and an inbound one up to its top-up. */
+export async function runProviderStart(
+ client: TrUApiClient,
+ fundingLogPath: string,
+): Promise<{ rows: DiagnosisRow[]; state?: ProviderState }> {
+ const rows: DiagnosisRow[] = [];
+ const check = rowFactory(rows);
+ const served = new Served(client.fundingProvider.serveSubscribe());
+ let inbound = "";
+ let state: ProviderState | undefined;
+
+ await check("provider_assigned", async () => {
+ inbound = await requestFunding(client, "In", IN_AMOUNT);
+ const item = await served.next(
+ assignedTo(inbound),
+ `the session ${inbound}`,
+ );
+ if (item.tag !== "Assigned") throw new Error(stringify(item));
+ const { direction, amount, lastUpdate } = item.value.session;
+ if (
+ direction !== "In" ||
+ amount !== IN_AMOUNT ||
+ lastUpdate !== undefined
+ ) {
+ throw new Error(`unexpected assignment ${stringify(item)}`);
+ }
+ return "the provider's worker was handed the session it was chosen for";
+ });
+
+ await check("provider_present_frame", async () => {
+ const framed = await client.fundingProvider.presentFrame({
+ intent: inbound,
+ route: "/kyc",
+ });
+ if (framed.isErr() || framed.value.outcome !== "Closed") {
+ throw new Error(`presentFrame: ${stringify(framed)}`);
+ }
+ await waitForTranscript(
+ fundingLogPath,
+ (line) =>
+ line.kind === "frame" &&
+ line.intent === inbound &&
+ line.route === "/kyc",
+ "the provider frame",
+ );
+ return "the host showed the provider's screen and it closed";
+ });
+
+ await check("provider_reports_forward_only", async () => {
+ await report(client, inbound, { tag: "AwaitingPayment" });
+ await report(client, inbound, {
+ tag: "PaymentReceived",
+ value: { finalized: true },
+ });
+ await report(client, inbound, { tag: "Converting" });
+ const backwards = await client.fundingProvider.report({
+ intent: inbound,
+ update: { tag: "AwaitingPayment" },
+ });
+ const outOfOrder =
+ backwards.isErr() && stringify(backwards.error).includes("OutOfOrder");
+ if (!outOfOrder) {
+ throw new Error(
+ `a backwards report was not refused: ${stringify(backwards)}`,
+ );
+ }
+ return "updates moved forward and a backwards one was refused";
+ });
+
+ await check("provider_credits_through_top_up", async () => {
+ const topUpId = randomId();
+ const toppedUp = await client.payment.topUp({
+ amount: IN_AMOUNT,
+ source: {
+ tag: "ProductAccount",
+ value: { derivationIndex: { tag: "Index", value: 0 } },
+ },
+ id: topUpId,
+ });
+ if (toppedUp.isErr()) {
+ throw new Error(`payment.topUp failed: ${stringify(toppedUp.error)}`);
+ }
+ await report(client, inbound, {
+ tag: "Crediting",
+ value: { topUpId, amount: IN_AMOUNT },
+ });
+ state = { intent: inbound, topUpId, amount: IN_AMOUNT.toString() };
+ return "the top-up was started and named in a Crediting report";
+ });
+
+ await check("provider_out_released", async () => {
+ const outbound = await requestFunding(client, "Out", OUT_AMOUNT);
+ await served.next(assignedTo(outbound), `the session ${outbound}`);
+ const paymentId = randomId();
+ const paid = await client.payment.request({
+ amount: OUT_AMOUNT,
+ destination: `0x${"11".repeat(32)}`,
+ id: paymentId,
+ });
+ if (paid.isErr()) {
+ throw new Error(`payment.request failed: ${stringify(paid.error)}`);
+ }
+ await report(client, outbound, {
+ tag: "Collecting",
+ value: { paymentId, amount: OUT_AMOUNT },
+ });
+ const last = await lastStatus(client, outbound);
+ if (last?.tag !== "Released" || last.value.debited !== OUT_AMOUNT) {
+ throw new Error(`ended as ${stringify(last)}`);
+ }
+ return "the core released the session once the named payment completed";
+ });
+
+ await check("provider_cancel", async () => {
+ const cancelled = await requestFunding(client, "In", CANCEL_AMOUNT);
+ await served.next(assignedTo(cancelled), `the session ${cancelled}`);
+ await served.next(
+ (item) => item.tag === "Cancel" && item.value.intent === cancelled,
+ `a cancel of ${cancelled}`,
+ );
+ await report(client, cancelled, {
+ tag: "Failed",
+ value: { reason: { tag: "Cancelled" } },
+ });
+ const last = await lastStatus(client, cancelled);
+ if (last?.tag !== "Failed" || last.value.reason.tag !== "Cancelled") {
+ throw new Error(`ended as ${stringify(last)}`);
+ }
+ return "the user's cancel reached the provider, which ended the session";
+ });
+
+ served.close();
+ return { rows, state };
+}
+
+/** The run after the restart: the session is handed over again where it
+ * stopped, and the core delivers it from what its top-up claimed. */
+export async function runProviderResume(
+ client: TrUApiClient,
+ fundingLogPath: string,
+ state: ProviderState | undefined,
+): Promise {
+ const rows: DiagnosisRow[] = [];
+ const check = rowFactory(rows);
+ if (!state) {
+ await check("provider_resumes_after_restart", async () => {
+ throw new Error("the first run left no session to resume");
+ });
+ return rows;
+ }
+ const served = new Served(client.fundingProvider.serveSubscribe());
+
+ await check("provider_resumes_after_restart", async () => {
+ const item = await served.next(
+ assignedTo(state.intent),
+ `the session ${state.intent}`,
+ );
+ const last =
+ item.tag === "Assigned" ? item.value.session.lastUpdate : undefined;
+ if (last?.tag !== "Crediting" || last.value.topUpId !== state.topUpId) {
+ throw new Error(`replayed ${stringify(item)}`);
+ }
+ return "the restarted worker was handed the session with its last update";
+ });
+
+ await check("provider_in_delivered", async () => {
+ await report(client, state.intent, { tag: "Delivered" });
+ const last = await lastStatus(client, state.intent);
+ if (
+ last?.tag !== "Delivered" ||
+ last.value.credited !== BigInt(state.amount)
+ ) {
+ throw new Error(`ended as ${stringify(last)}`);
+ }
+ await waitForTranscript(
+ fundingLogPath,
+ (line) =>
+ line.kind === "status" &&
+ line.intent === state.intent &&
+ line.tag === "Delivered" &&
+ line.amount === state.amount,
+ `Delivered reaching the host for ${state.intent}`,
+ );
+ return "the core delivered the session from the claim of the named top-up";
+ });
+
+ served.close();
+ return rows;
+}
diff --git a/rust/crates/truapi-host-cli/js/scripts/funding-battery.ts b/rust/crates/truapi-host-cli/js/scripts/funding-battery.ts
index d911321d99..91171f3ded 100644
--- a/rust/crates/truapi-host-cli/js/scripts/funding-battery.ts
+++ b/rust/crates/truapi-host-cli/js/scripts/funding-battery.ts
@@ -7,10 +7,23 @@
// which builds the CLI with `--features test-host` and starts a signing host
// with a scripted funding overlay, so a product's requests are shown, started
// or dismissed, and settled with no chain behind them.
-import { mkdirSync, writeFileSync } from "node:fs";
+//
+// The battery runs this script twice, as a Worker, around a host restart.
+// `TRUAPI_FUNDING_STAGE=start` runs the product's cases and the provider's up
+// to an inbound top-up, and keeps its rows in `TRUAPI_FUNDING_STATE`;
+// `TRUAPI_FUNDING_STAGE=resume` finishes that session in the new process and
+// writes the report with every row. Without a stage it runs the product's
+// cases alone.
+import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { dirname } from "node:path";
import { fileURLToPath } from "node:url";
+import type { DiagnosisRow } from "../diagnosis.ts";
import { runFundingE2e } from "../funding-e2e.ts";
+import {
+ type ProviderState,
+ runProviderResume,
+ runProviderStart,
+} from "../funding-provider-e2e.ts";
import {
cliModalityDiagnosisReportMetadata,
renderDiagnosisReport,
@@ -39,36 +52,89 @@ if (
);
}
-const rows = await runFundingE2e(truapi, process.env.TRUAPI_FUNDING_LOG);
-for (const row of rows) {
- const mark = { pass: "✅", fail: "❌", skipped: "⏭️" }[row.status];
- console.log(`${mark} ${row.id} (${row.durationMs}ms) ${row.output}`);
-}
+const fundingLog = process.env.TRUAPI_FUNDING_LOG;
+const stage = process.env.TRUAPI_FUNDING_STAGE;
+const statePath = process.env.TRUAPI_FUNDING_STATE;
-// Committed, so a rerun overwrites it and the diff shows what changed. The
-// explorer's Funding matrix reads it from the `funding/` directory it lands in.
-mkdirSync(dirname(REPORT_PATH), { recursive: true });
-writeFileSync(REPORT_PATH, renderDiagnosisReport(report.title, rows));
-console.log(`funding battery: report saved to ${REPORT_PATH}`);
+const print = (rows: DiagnosisRow[]) => {
+ for (const row of rows) {
+ const mark = { pass: "✅", fail: "❌", skipped: "⏭️" }[row.status];
+ console.log(`${mark} ${row.id} (${row.durationMs}ms) ${row.output}`);
+ }
+};
-const skipped = rows.filter((row) => row.status === "skipped");
-if (skipped.length > 0) {
- // A skip here means the run could not read what the host observed, which is
- // half of what these cases assert.
- throw new Error(
- `funding battery skipped ${skipped.length} case(s): ${skipped
- .map((row) => row.output)
- .join("; ")}`,
- );
+interface SavedRun {
+ rows: DiagnosisRow[];
+ provider?: ProviderState;
}
-const failures = rows.filter((row) => row.status === "fail");
-if (failures.length > 0) {
- throw new Error(
- `funding battery failed: ${failures.length} of ${rows.length} cases\n${failures
- .map((row) => `${row.id}: ${row.output}`)
- .join("\n")}`,
- );
+/** The rows the report holds: the saved first run plus the resumed one, or
+ * the product's cases alone without a stage. */
+async function rowsToReport(): Promise {
+ if (stage !== "resume") {
+ const rows = await runFundingE2e(truapi, fundingLog);
+ print(rows);
+ return rows;
+ }
+ if (!statePath || !fundingLog) {
+ throw new Error(
+ "the resume stage needs TRUAPI_FUNDING_STATE and TRUAPI_FUNDING_LOG",
+ );
+ }
+ const saved: SavedRun = existsSync(statePath)
+ ? JSON.parse(readFileSync(statePath, "utf8"))
+ : { rows: [] };
+ const resumed = await runProviderResume(truapi, fundingLog, saved.provider);
+ print(resumed);
+ return [...saved.rows, ...resumed];
}
-console.log(`funding battery: ${rows.length} cases passed`);
+if (stage === "start") {
+ if (!statePath || !fundingLog) {
+ throw new Error(
+ "the start stage needs TRUAPI_FUNDING_STATE and TRUAPI_FUNDING_LOG",
+ );
+ }
+ const consumer = await runFundingE2e(truapi, fundingLog);
+ const provider = await runProviderStart(truapi, fundingLog);
+ const saved: SavedRun = {
+ rows: [...consumer, ...provider.rows],
+ provider: provider.state,
+ };
+ print(saved.rows);
+ writeFileSync(statePath, JSON.stringify(saved));
+ console.log(`funding battery: first run kept in ${statePath}`);
+} else {
+ finish(await rowsToReport());
+}
+
+/** Write the report, then fail on any skipped or failed case. */
+function finish(rows: DiagnosisRow[]) {
+ // Committed, so a rerun overwrites it and the diff shows what changed. The
+ // explorer's Funding matrix reads it from the `funding/` directory it lands in.
+ mkdirSync(dirname(REPORT_PATH), { recursive: true });
+ writeFileSync(REPORT_PATH, renderDiagnosisReport(report.title, rows));
+ console.log(`funding battery: report saved to ${REPORT_PATH}`);
+
+ const skipped = rows.filter((row) => row.status === "skipped");
+ if (skipped.length > 0) {
+ // A skip here means the run could not read what the host observed, which is
+ // half of what these cases assert.
+ throw new Error(
+ `funding battery skipped ${skipped.length} case(s): ${skipped
+ .map((row) => row.output)
+ .join("; ")}`,
+ );
+ }
+
+ const failures = rows.filter((row) => row.status === "fail");
+ if (failures.length > 0) {
+ throw new Error(
+ `funding battery failed: ${failures.length} of ${rows.length} cases\n${failures
+ .map((row) => `${row.id}: ${row.output}`)
+ .join("\n")}`,
+ );
+ }
+
+ console.log(`funding battery: ${rows.length} cases passed`);
+}
diff --git a/rust/crates/truapi-host-cli/src/funding_host.rs b/rust/crates/truapi-host-cli/src/funding_host.rs
index 88cf03b175..57c978f9b2 100644
--- a/rust/crates/truapi-host-cli/src/funding_host.rs
+++ b/rust/crates/truapi-host-cli/src/funding_host.rs
@@ -8,6 +8,12 @@
//! There is no chain behind it: this exists so a battery can drive the
//! product's side of funding headlessly.
//!
+//! `provide` and `provide-cancel` instead hand the session to the product that
+//! asked, so a provider worker can serve its own request; `provide-cancel`
+//! then cancels it, as a user would. Top-ups and payment requests that worker
+//! starts are accepted and complete in full, recorded in the ledger named by
+//! `TRUAPI_FUNDING_LEDGER` so their statuses outlive a restart.
+//!
//! Every overlay request and every session change the core reports is
//! appended to the transcript named by `TRUAPI_FUNDING_LOG`, one JSON object
//! per line, so a battery can assert what the host saw rather than only what
@@ -20,13 +26,19 @@ use std::path::PathBuf;
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::time::Duration;
+use futures::stream::{self, BoxStream, StreamExt};
use truapi::SigningHostRuntime;
use truapi::host_logic::funding::FundingStage;
use truapi::latest::{
- FundingDirection, FundingFailure, GenericError, HostFundingStatusSubscribeItem,
+ FundingDirection, FundingFailure, FundingFrameOutcome, GenericError,
+ HostFundingStatusSubscribeItem, HostPaymentError, HostPaymentRequest,
+ HostPaymentStatusSubscribeError, HostPaymentStatusSubscribeItem, HostPaymentTopUpError,
+ HostPaymentTopUpRequest, HostPaymentTopUpStatusSubscribeError,
+ HostPaymentTopUpStatusSubscribeItem,
};
use truapi::platform::{
- FundingPlatform, FundingPresentOutcome, FundingPresentation, ProductContext, async_trait,
+ FundingPlatform, FundingPresentOutcome, FundingPresentation, PaymentPlatform, ProductContext,
+ TopUpPlatform, async_trait,
};
/// How long a started session stays in flight before it is settled, so a
@@ -44,6 +56,10 @@ enum Outcome {
Fail,
/// The user closes the overlay.
Dismiss,
+ /// The user picks the product that asked as the provider.
+ Provide,
+ /// As `Provide`, then the user cancels.
+ ProvideThenCancel,
}
/// Parse `deliver:900,release:500,fail,dismiss`.
@@ -55,6 +71,8 @@ fn parse_outcomes(spec: &str) -> VecDeque {
Some(("release", amount)) => amount.parse().ok().map(Outcome::Release),
None if entry == "fail" => Some(Outcome::Fail),
None if entry == "dismiss" => Some(Outcome::Dismiss),
+ None if entry == "provide" => Some(Outcome::Provide),
+ None if entry == "provide-cancel" => Some(Outcome::ProvideThenCancel),
_ => None,
})
.collect()
@@ -88,6 +106,11 @@ impl CliFundingHost {
});
let _ = host.runtime.set(Arc::downgrade(runtime));
runtime.set_funding_platform(host);
+ if let Some(ledger) = std::env::var_os("TRUAPI_FUNDING_LEDGER").map(PathBuf::from) {
+ let payments = Arc::new(CliPaymentHost { ledger });
+ runtime.set_top_up_platform(payments.clone());
+ runtime.set_payment_platform(payments);
+ }
}
/// Append one line to the transcript, in a single write so concurrent
@@ -106,6 +129,38 @@ impl CliFundingHost {
}
}
+ /// Hand session `intent` to the product that asked for it, and cancel it
+ /// shortly after when `cancel` is set.
+ async fn provide(
+ &self,
+ runtime: Option>,
+ product: Option<&ProductContext>,
+ intent: String,
+ cancel: bool,
+ ) -> Result {
+ let (Some(runtime), Some(product)) =
+ (runtime.and_then(|runtime| runtime.upgrade()), product)
+ else {
+ return Ok(FundingPresentOutcome::Dismissed);
+ };
+ runtime
+ .select_funding_provider(&intent, &product.product_id)
+ .await?;
+ if cancel {
+ let runtime = Arc::downgrade(&runtime);
+ tokio::spawn(async move {
+ tokio::time::sleep(SETTLE_AFTER).await;
+ let Some(runtime) = runtime.upgrade() else {
+ return;
+ };
+ if let Err(error) = runtime.cancel_funding(&intent).await {
+ tracing::warn!(%intent, reason = %error.reason, "cancelling a scripted funding session failed");
+ }
+ });
+ }
+ Ok(FundingPresentOutcome::Started)
+ }
+
fn next_outcome(&self) -> Outcome {
self.outcomes
.lock()
@@ -138,7 +193,7 @@ fn settled_stage(outcome: Outcome) -> Option {
},
settled_at_ms,
}),
- Outcome::Dismiss => None,
+ Outcome::Dismiss | Outcome::Provide | Outcome::ProvideThenCancel => None,
}
}
@@ -161,11 +216,21 @@ impl FundingPlatform for CliFundingHost {
"product": product.map(|product| product.product_id.clone()),
"outcome": format!("{outcome:?}"),
}));
+ let runtime = self.runtime.get().cloned();
+ let intent = session.intent;
+ if matches!(outcome, Outcome::Provide | Outcome::ProvideThenCancel) {
+ return self
+ .provide(
+ runtime,
+ product,
+ intent,
+ outcome == Outcome::ProvideThenCancel,
+ )
+ .await;
+ }
let Some(stage) = settled_stage(outcome) else {
return Ok(FundingPresentOutcome::Dismissed);
};
- let runtime = self.runtime.get().cloned();
- let intent = session.intent;
tokio::spawn(async move {
tokio::time::sleep(SETTLE_AFTER).await;
let Some(runtime) = runtime.and_then(|runtime| runtime.upgrade()) else {
@@ -184,6 +249,21 @@ impl FundingPlatform for CliFundingHost {
Ok(FundingPresentOutcome::Started)
}
+ async fn present_provider_frame(
+ &self,
+ provider: &ProductContext,
+ intent: String,
+ route: String,
+ ) -> Result {
+ self.record(serde_json::json!({
+ "kind": "frame",
+ "intent": intent,
+ "provider": provider.product_id,
+ "route": route,
+ }));
+ Ok(FundingFrameOutcome::Closed)
+ }
+
fn funding_session_changed(&self, intent: String, status: HostFundingStatusSubscribeItem) {
let (tag, amount) = match status {
HostFundingStatusSubscribeItem::AwaitingDeposit { .. } => ("AwaitingDeposit", None),
@@ -201,6 +281,116 @@ impl FundingPlatform for CliFundingHost {
}
}
+/// Scripted top-up and payment engines: every top-up claims in full and every
+/// payment completes, once started. What was started is kept in a ledger file,
+/// one JSON object per line, so a restarted host still answers for it.
+struct CliPaymentHost {
+ ledger: PathBuf,
+}
+
+impl CliPaymentHost {
+ /// Whether `kind` with `id` was started by `product`.
+ fn started(&self, kind: &str, product: &ProductContext, id: &[u8; 32]) -> bool {
+ let id = hex::encode(id);
+ std::fs::read_to_string(&self.ledger)
+ .unwrap_or_default()
+ .lines()
+ .filter_map(|line| serde_json::from_str::(line).ok())
+ .any(|line| {
+ line["kind"] == kind
+ && line["product"] == product.product_id.as_str()
+ && line["id"] == id.as_str()
+ })
+ }
+
+ /// Record a start; `false` when the id was already used.
+ fn start(&self, kind: &str, product: &ProductContext, id: &[u8; 32], amount: u128) -> bool {
+ if self.started(kind, product, id) {
+ return false;
+ }
+ let line = serde_json::json!({
+ "kind": kind,
+ "product": product.product_id,
+ "id": hex::encode(id),
+ "amount": amount.to_string(),
+ });
+ let appended = OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&self.ledger)
+ .and_then(|mut file| file.write_all(format!("{line}\n").as_bytes()));
+ if let Err(error) = appended {
+ tracing::warn!(path = ?self.ledger, %error, "funding ledger could not be appended to");
+ }
+ true
+ }
+}
+
+#[async_trait]
+impl TopUpPlatform for CliPaymentHost {
+ async fn top_up(
+ &self,
+ product: &ProductContext,
+ request: HostPaymentTopUpRequest,
+ ) -> Result<(), HostPaymentTopUpError> {
+ if self.start("top_up", product, &request.id, request.amount) {
+ Ok(())
+ } else {
+ Err(HostPaymentTopUpError::AlreadyExists)
+ }
+ }
+
+ fn subscribe_top_up_status(
+ &self,
+ product: &ProductContext,
+ id: [u8; 32],
+ ) -> BoxStream<
+ 'static,
+ Result,
+ > {
+ if !self.started("top_up", product, &id) {
+ return stream::iter([Err(HostPaymentTopUpStatusSubscribeError::NotFound)]).boxed();
+ }
+ stream::iter([
+ Ok(HostPaymentTopUpStatusSubscribeItem::Detecting),
+ Ok(HostPaymentTopUpStatusSubscribeItem::Claiming),
+ Ok(HostPaymentTopUpStatusSubscribeItem::Claimed { finalized: true }),
+ ])
+ .boxed()
+ }
+}
+
+#[async_trait]
+impl PaymentPlatform for CliPaymentHost {
+ async fn request_payment(
+ &self,
+ product: &ProductContext,
+ request: HostPaymentRequest,
+ ) -> Result<(), HostPaymentError> {
+ if self.start("payment", product, &request.id, request.amount) {
+ Ok(())
+ } else {
+ Err(HostPaymentError::AlreadyExists)
+ }
+ }
+
+ fn subscribe_payment_status(
+ &self,
+ product: &ProductContext,
+ id: [u8; 32],
+ ) -> BoxStream<'static, Result>
+ {
+ if !self.started("payment", product, &id) {
+ return stream::iter([Err(HostPaymentStatusSubscribeError::PaymentNotFound)]).boxed();
+ }
+ stream::iter([
+ Ok(HostPaymentStatusSubscribeItem::Processing),
+ Ok(HostPaymentStatusSubscribeItem::Completed),
+ ])
+ .boxed()
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -210,12 +400,16 @@ mod tests {
#[test]
fn outcomes_are_read_in_order_and_unknown_entries_dropped() {
assert_eq!(
- parse_outcomes("deliver:900, release:500,fail,bogus,dismiss,deliver:x"),
+ parse_outcomes(
+ "deliver:900, release:500,fail,bogus,dismiss,deliver:x,provide,provide-cancel"
+ ),
VecDeque::from([
Outcome::Deliver(900),
Outcome::Release(500),
Outcome::Fail,
Outcome::Dismiss,
+ Outcome::Provide,
+ Outcome::ProvideThenCancel,
])
);
}
diff --git a/rust/crates/truapi/RUNTIME.md b/rust/crates/truapi/RUNTIME.md
index 908c55f596..f63ade8ad0 100644
--- a/rust/crates/truapi/RUNTIME.md
+++ b/rust/crates/truapi/RUNTIME.md
@@ -406,14 +406,27 @@ AutoSigning without approval. Legacy-account signing still asks the user.
Native hosts reach this through `NativeTrUApiHostRuntime`:
`set_funding_callbacks` (the overlay), `set_top_up_callbacks` with
`notify_top_up_status` (the top-up engine), `open_funding`,
- `funding_session`, `funding_sessions`, `cancel_funding` and
- `acknowledge_funding_session`. Amounts cross the FFI as decimal strings.
+ `funding_session`, `funding_sessions`, `select_funding_provider`,
+ `cancel_funding` and `acknowledge_funding_session`. Amounts cross the FFI as decimal strings.
`funding_sessions` lists sessions in flight first, then ended ones, each
newest first. An ended session is handed to the host through
`funding_session_changed` each time funding resumes until the host calls
`acknowledge_funding_session`, so its history writes every outcome once;
the core keeps the 50 newest recorded sessions and every unrecorded one
within the 200 newest ended.
+ `select_funding_provider` hands a session to the provider the user chose.
+ That provider's worker runs it through the `FundingProvider` trait:
+ `serveSubscribe` replays its sessions in flight and then streams new ones
+ and cancel requests, `report` stores each update on the session (only from
+ the assigned provider, only forward), and `presentFrame` asks the host to
+ show one of its screens through `present_provider_frame`. The core holds the
+ provider's worker while a session is assigned and open. It ends a session as
+ `Delivered` or `Released` itself, from the claims of the top-ups the
+ provider named (`Crediting`, then `Delivered`) or the completion of the
+ payment request it named (`Collecting`); once funds move, the provider can
+ no longer fail the session and its deadline no longer applies. A cancel ends
+ an unassigned session, asks the provider of an assigned one, and is refused
+ once the provider has seen the user's payment.
- `PaymentPlatform`: pay from the user's balance to an account once the user
approves, and stream each payment's status by its caller-chosen id.
Installed with `set_payment_platform`; native hosts use
diff --git a/rust/crates/truapi/src/api.rs b/rust/crates/truapi/src/api.rs
index 614f2c109d..a2cabf9b26 100644
--- a/rust/crates/truapi/src/api.rs
+++ b/rust/crates/truapi/src/api.rs
@@ -7,6 +7,7 @@ pub mod coin_payment;
pub mod contacts;
pub mod entropy;
pub mod funding;
+pub mod funding_provider;
pub mod local_storage;
pub mod locale;
pub mod notifications;
@@ -29,6 +30,7 @@ pub use coin_payment::CoinPayment;
pub use contacts::Contacts;
pub use entropy::Entropy;
pub use funding::Funding;
+pub use funding_provider::FundingProvider;
pub use local_storage::LocalStorage;
pub use locale::Locale;
pub use notifications::Notifications;
@@ -53,6 +55,7 @@ pub trait TrUApi:
+ Contacts
+ Entropy
+ Funding
+ + FundingProvider
+ LocalStorage
+ Locale
+ Notifications
@@ -80,6 +83,7 @@ impl TrUApi for T where
+ Contacts
+ Entropy
+ Funding
+ + FundingProvider
+ LocalStorage
+ Locale
+ Notifications
diff --git a/rust/crates/truapi/src/api/funding_provider.rs b/rust/crates/truapi/src/api/funding_provider.rs
new file mode 100644
index 0000000000..303387e0e8
--- /dev/null
+++ b/rust/crates/truapi/src/api/funding_provider.rs
@@ -0,0 +1,83 @@
+//! Unified [`FundingProvider`] trait.
+
+use crate::versioned::funding_provider::{
+ HostFundingPresentFrameError, HostFundingPresentFrameRequest, HostFundingPresentFrameResponse,
+ HostFundingReportError, HostFundingReportRequest, HostFundingReportResponse,
+ HostFundingServeSubscribeError, HostFundingServeSubscribeItem,
+ HostFundingServeSubscribeRequest,
+};
+use crate::{CallContext, CallError, Subscription};
+use crate::{wire, wire_trait};
+
+/// Serve the funding sessions the user chose this provider for.
+///
+/// The provider's worker runs each session it is assigned and reports its
+/// progress. The host stores every report, and decides `Delivered` and
+/// `Released` itself from the top-ups and payments the provider names.
+#[wire_trait(id = 22)]
+#[crate::service(required_execution = Worker)]
+#[crate::async_trait]
+pub trait FundingProvider: Send + Sync {
+ /// Receive the sessions assigned to the calling provider, and requests to
+ /// cancel them.
+ ///
+ /// Emits every session still in flight first, so a restarted worker picks
+ /// up where it stopped, then each new assignment and cancel request.
+ ///
+ /// ```ts
+ /// import { firstValueFrom, from } from "rxjs";
+ ///
+ /// const item = await firstValueFrom(
+ /// from(truapi.fundingProvider.serveSubscribe()),
+ /// );
+ /// console.log("funding provider item:", item);
+ /// ```
+ #[wire(id = 0)]
+ async fn serve_subscribe(
+ &self,
+ _cx: &CallContext,
+ _request: HostFundingServeSubscribeRequest,
+ ) -> Subscription>
+ {
+ Subscription::interrupted(CallError::unavailable())
+ }
+
+ /// Report progress on a session assigned to the calling provider.
+ ///
+ /// The host stores the update before answering.
+ ///
+ /// ```ts
+ /// const result = await truapi.fundingProvider.report({
+ /// intent: "fs_example",
+ /// update: { tag: "AwaitingPayment", value: undefined },
+ /// });
+ /// console.log("funding report:", result);
+ /// ```
+ #[wire(id = 1)]
+ async fn report(
+ &self,
+ _cx: &CallContext,
+ _request: HostFundingReportRequest,
+ ) -> Result> {
+ Err(CallError::unavailable())
+ }
+
+ /// Show one of the provider's screens, such as KYC or card entry, in a
+ /// frame the host owns. Answers once it closes.
+ ///
+ /// ```ts
+ /// const result = await truapi.fundingProvider.presentFrame({
+ /// intent: "fs_example",
+ /// route: "/kyc",
+ /// });
+ /// console.log("provider frame:", result);
+ /// ```
+ #[wire(id = 2)]
+ async fn present_frame(
+ &self,
+ _cx: &CallContext,
+ _request: HostFundingPresentFrameRequest,
+ ) -> Result> {
+ Err(CallError::unavailable())
+ }
+}
diff --git a/rust/crates/truapi/src/host_core.rs b/rust/crates/truapi/src/host_core.rs
index 59cb762b14..60f940a74b 100644
--- a/rust/crates/truapi/src/host_core.rs
+++ b/rust/crates/truapi/src/host_core.rs
@@ -804,8 +804,9 @@ impl SigningHostRuntime {
})
}
- /// Cancel open funding session `intent`. Returns whether it was still
- /// open.
+ /// Cancel funding session `intent` at the user's request: it ends if no
+ /// provider serves it, otherwise its provider is asked to stop. Returns
+ /// whether the cancel was taken.
#[instrument(skip_all, fields(runtime.method = "signing_host_runtime.cancel_funding"))]
pub async fn cancel_funding(&self, intent: &str) -> Result {
self.services
@@ -816,6 +817,22 @@ impl SigningHostRuntime {
})
}
+ /// Hand open funding session `intent` to the provider the user chose,
+ /// by product id. Returns whether it was open and not yet assigned.
+ #[instrument(skip_all, fields(runtime.method = "signing_host_runtime.select_funding_provider"))]
+ pub async fn select_funding_provider(
+ &self,
+ intent: &str,
+ provider_id: &str,
+ ) -> Result {
+ self.services
+ .select_funding_provider(intent, provider_id)
+ .await
+ .map_err(|err| v01::GenericError {
+ reason: err.to_string(),
+ })
+ }
+
/// One funding session as the core holds it, for the host's own status
/// and history views.
pub fn funding_session(
diff --git a/rust/crates/truapi/src/host_logic/funding.rs b/rust/crates/truapi/src/host_logic/funding.rs
index e351fa3906..ac3dabf614 100644
--- a/rust/crates/truapi/src/host_logic/funding.rs
+++ b/rust/crates/truapi/src/host_logic/funding.rs
@@ -9,7 +9,10 @@
use parity_scale_codec::{Decode, Encode};
use tracing::warn;
-use truapi::latest::{FundingDirection, FundingFailure, HostFundingStatusSubscribeItem};
+use truapi::latest::{
+ FundingAssignment, FundingDirection, FundingFailure, FundingUpdate,
+ HostFundingStatusSubscribeItem,
+};
use crate::platform::{CoreStorage, CoreStorageKey};
@@ -21,6 +24,9 @@ const SETTLED_HISTORY_LIMIT: usize = 50;
/// How far back, counting every ended session newest first, those the host
/// has not recorded yet are kept.
const UNACKNOWLEDGED_LIMIT: usize = 200;
+/// Most top-ups a provider may start for one session, as getcash claims in
+/// up to three attempts.
+const TOP_UP_LIMIT: usize = 3;
/// What the core knows about one session, independent of any host surface.
#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
@@ -48,6 +54,60 @@ pub struct FundingSession {
/// Whether the host has recorded the session's outcome in its own
/// history. An ended session is handed to the host until it has.
pub acknowledged: bool,
+ /// Product id of the provider the user chose. `None` until they choose.
+ pub provider_id: Option,
+ /// Whether the user asked to cancel and the provider has not answered.
+ pub cancel_requested: bool,
+ /// Every update the provider reported, oldest first.
+ pub updates: Vec,
+}
+
+/// One update a provider reported, and when the core stored it.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+#[cfg_attr(
+ all(feature = "runtime", not(target_arch = "wasm32")),
+ derive(uniffi::Record)
+)]
+pub struct FundingUpdateRecord {
+ /// The update.
+ pub update: FundingUpdate,
+ /// When it was stored, in Unix milliseconds.
+ pub at_ms: u64,
+}
+
+/// Why a provider's report was refused.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ReportRefusal {
+ /// The session is not open or not assigned to the reporting provider.
+ NotFound,
+ /// The update does not follow the last one, or does not fit the
+ /// session's direction.
+ OutOfOrder,
+ /// The top-up or payment request it names is already named by another
+ /// session of the same provider.
+ DuplicateId,
+}
+
+/// What a cancel did.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CancelOutcome {
+ /// No provider serves the session, so it ended as cancelled.
+ Ended,
+ /// The provider serving it was asked to stop.
+ Requested,
+ /// Too late: the session ended, or the user's payment already reached
+ /// the provider.
+ Refused,
+}
+
+/// What the core follows to decide a session's outcome once funds move.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Settlement {
+ /// The top-ups an inbound session credits through, with the amount each
+ /// asks for.
+ TopUps(Vec<([u8; 32], u128)>),
+ /// The payment an outbound session collects through, with its amount.
+ Payment([u8; 32], u128),
}
/// Stage of a session, as the core persists it.
@@ -100,9 +160,175 @@ impl FundingSession {
opened_at_ms: now_ms,
deadline_ms: now_ms.saturating_add(SESSION_WINDOW_MS),
acknowledged: false,
+ provider_id: None,
+ cancel_requested: false,
+ updates: Vec::new(),
+ }
+ }
+
+ /// Assign the session to the provider the user chose. Returns whether it
+ /// was open and not yet assigned.
+ pub fn assign(&mut self, provider_id: &str) -> bool {
+ if self.is_terminal() || self.provider_id.is_some() {
+ return false;
+ }
+ self.provider_id = Some(provider_id.to_string());
+ true
+ }
+
+ /// The session as the provider serving it receives it.
+ pub fn assignment(&self) -> FundingAssignment {
+ FundingAssignment {
+ intent: self.intent.clone(),
+ direction: self.direction,
+ amount: self.amount,
+ expires_at: self.deadline_ms,
+ last_update: self.last_update().cloned(),
+ }
+ }
+
+ /// The provider's most recent update.
+ pub fn last_update(&self) -> Option<&FundingUpdate> {
+ self.updates.last().map(|record| &record.update)
+ }
+
+ /// Store `update` from `provider_id`. A `Failed` update ends the session.
+ pub fn report(
+ &mut self,
+ provider_id: &str,
+ update: FundingUpdate,
+ now_ms: u64,
+ ) -> Result<(), ReportRefusal> {
+ if self.is_terminal() || self.provider_id.as_deref() != Some(provider_id) {
+ return Err(ReportRefusal::NotFound);
+ }
+ if !self.follows(&update) {
+ return Err(ReportRefusal::OutOfOrder);
+ }
+ if let FundingUpdate::Failed { reason } = &update {
+ self.stage = FundingStage::Failed {
+ reason: reason.clone(),
+ settled_at_ms: now_ms,
+ };
+ }
+ self.updates.push(FundingUpdateRecord {
+ update,
+ at_ms: now_ms,
+ });
+ Ok(())
+ }
+
+ /// Whether `update` may come next. Updates only move forward, a session
+ /// may credit through a few top-ups, and once funds move the provider
+ /// can no longer fail it: the top-ups or payment decide.
+ fn follows(&self, update: &FundingUpdate) -> bool {
+ let fits_direction = match update {
+ FundingUpdate::Collecting { .. } => self.direction == FundingDirection::Out,
+ FundingUpdate::Failed { .. } => true,
+ _ => self.direction == FundingDirection::In,
+ };
+ let last = self.last_update().map(update_rank);
+ let top_ups = self.top_ups();
+ fits_direction
+ && match update {
+ FundingUpdate::Failed { .. } => !self.funds_moving(),
+ FundingUpdate::Crediting { top_up_id, .. } => {
+ last <= Some(update_rank(update))
+ && top_ups.len() < TOP_UP_LIMIT
+ && top_ups.iter().all(|(id, _)| id != top_up_id)
+ }
+ FundingUpdate::Delivered => last < Some(update_rank(update)) && !top_ups.is_empty(),
+ _ => last < Some(update_rank(update)),
+ }
+ }
+
+ /// The top-up or payment request id `update` names, if any.
+ pub fn named_id(update: &FundingUpdate) -> Option<[u8; 32]> {
+ match update {
+ FundingUpdate::Crediting { top_up_id, .. } => Some(*top_up_id),
+ FundingUpdate::Collecting { payment_id, .. } => Some(*payment_id),
+ _ => None,
+ }
+ }
+
+ /// Whether any of this session's updates names `id`.
+ pub fn names(&self, id: &[u8; 32]) -> bool {
+ self.updates
+ .iter()
+ .any(|record| Self::named_id(&record.update).as_ref() == Some(id))
+ }
+
+ /// The top-ups the provider started, in order, with the amount each asks
+ /// for.
+ fn top_ups(&self) -> Vec<([u8; 32], u128)> {
+ self.updates
+ .iter()
+ .filter_map(|record| match record.update {
+ FundingUpdate::Crediting { top_up_id, amount } => Some((top_up_id, amount)),
+ _ => None,
+ })
+ .collect()
+ }
+
+ /// Whether funds are moving: the provider started a top-up or a payment
+ /// request. From then on the session neither expires nor cancels.
+ pub fn funds_moving(&self) -> bool {
+ self.updates.iter().any(|record| {
+ matches!(
+ record.update,
+ FundingUpdate::Crediting { .. } | FundingUpdate::Collecting { .. }
+ )
+ })
+ }
+
+ /// What the core must follow to end the session, once the provider has
+ /// handed the outcome over.
+ pub fn settlement_due(&self) -> Option {
+ if self.is_terminal() {
+ return None;
+ }
+ match self.last_update()? {
+ FundingUpdate::Delivered => Some(Settlement::TopUps(self.top_ups())),
+ FundingUpdate::Collecting { payment_id, amount } => {
+ Some(Settlement::Payment(*payment_id, *amount))
+ }
+ _ => None,
}
}
+ /// End the session with what its top-ups claimed or its payment moved.
+ /// Nothing moved fails it. Returns whether it changed.
+ pub fn settle(&mut self, moved: u128, now_ms: u64) -> bool {
+ if self.is_terminal() {
+ return false;
+ }
+ self.stage = match (moved, self.direction) {
+ (0, FundingDirection::In) => FundingStage::Failed {
+ reason: FundingFailure::Other {
+ code: "not_claimed".to_string(),
+ message: "Nothing reached your balance".to_string(),
+ },
+ settled_at_ms: now_ms,
+ },
+ (0, FundingDirection::Out) => FundingStage::Failed {
+ reason: FundingFailure::Other {
+ code: "payment_failed".to_string(),
+ message: "Your payment did not go through".to_string(),
+ },
+ settled_at_ms: now_ms,
+ },
+ (credited, FundingDirection::In) => FundingStage::Delivered {
+ credited,
+ settled_at_ms: now_ms,
+ },
+ (debited, FundingDirection::Out) => FundingStage::Released {
+ debited,
+ settled_at_ms: now_ms,
+ },
+ };
+ true
+ }
+
/// Whether the host still has to hear of the session when funding
/// resumes: it is in flight, or it ended and the host has not recorded
/// the outcome.
@@ -110,9 +336,28 @@ impl FundingSession {
!self.is_terminal() || !self.acknowledged
}
- /// End an open session as cancelled. Returns whether it changed.
- pub fn cancel(&mut self, now_ms: u64) -> bool {
- self.fail(FundingFailure::Cancelled, now_ms)
+ /// Cancel at the user's request: end the session if no provider serves
+ /// it, otherwise ask the provider to stop, which it answers by reporting.
+ /// Refused once the user's payment reached the provider, as getcash
+ /// refuses once anything arrived.
+ pub fn cancel(&mut self, now_ms: u64) -> CancelOutcome {
+ if self.is_terminal() {
+ return CancelOutcome::Refused;
+ }
+ if self.provider_id.is_none() {
+ self.fail(FundingFailure::Cancelled, now_ms);
+ return CancelOutcome::Ended;
+ }
+ let paid = self.funds_moving()
+ || self
+ .updates
+ .iter()
+ .any(|record| matches!(record.update, FundingUpdate::PaymentReceived { .. }));
+ if paid {
+ return CancelOutcome::Refused;
+ }
+ self.cancel_requested = true;
+ CancelOutcome::Requested
}
/// Whether the session has ended.
@@ -135,7 +380,7 @@ impl FundingSession {
match (&self.stage, self.direction) {
(FundingStage::Open, FundingDirection::In) => {
HostFundingStatusSubscribeItem::AwaitingDeposit {
- expires_at: Some(self.deadline_ms),
+ expires_at: (!self.funds_moving()).then_some(self.deadline_ms),
}
}
(FundingStage::Open, FundingDirection::Out) => {
@@ -172,7 +417,23 @@ impl FundingSession {
/// Expire the session if it is still open at its deadline. Returns whether
/// it expired.
pub fn expire_if_due(&mut self, now_ms: u64) -> bool {
- now_ms >= self.deadline_ms && self.fail(FundingFailure::Expired, now_ms)
+ now_ms >= self.deadline_ms
+ && !self.funds_moving()
+ && self.fail(FundingFailure::Expired, now_ms)
+ }
+}
+
+/// Order of an update within a session: a later update ranks higher, and a
+/// top-up and a payment request rank where funds start moving.
+fn update_rank(update: &FundingUpdate) -> u8 {
+ match update {
+ FundingUpdate::AwaitingPayment => 0,
+ FundingUpdate::PaymentReceived { finalized: false } => 1,
+ FundingUpdate::PaymentReceived { finalized: true } => 2,
+ FundingUpdate::Converting => 3,
+ FundingUpdate::Crediting { .. } | FundingUpdate::Collecting { .. } => 4,
+ FundingUpdate::Delivered => 5,
+ FundingUpdate::Failed { .. } => 6,
}
}
@@ -185,6 +446,12 @@ pub enum FundingSessionError {
/// Failure detail.
reason: String,
},
+ /// The chosen provider is not a valid product id.
+ #[display("invalid funding provider: {reason}")]
+ InvalidProvider {
+ /// Why the id was refused.
+ reason: String,
+ },
}
/// The sessions worth keeping: every open one, then the most recently ended
@@ -265,6 +532,7 @@ mod tests {
use crate::test_support::stub_platform;
const NOW: u64 = 1_700_000_000_000;
+ const PROVIDER: &str = "ramp.dot";
fn session(direction: FundingDirection) -> FundingSession {
FundingSession::new(
@@ -375,21 +643,170 @@ mod tests {
);
}
- // Cancel ends an open session once; an ended one keeps its outcome.
+ fn served(direction: FundingDirection) -> FundingSession {
+ let mut session = session(direction);
+ assert!(session.assign(PROVIDER));
+ session
+ }
+
+ fn reported(session: &mut FundingSession, updates: &[FundingUpdate]) {
+ for update in updates {
+ session
+ .report(PROVIDER, update.clone(), NOW)
+ .unwrap_or_else(|refusal| panic!("{update:?} refused: {refusal:?}"));
+ }
+ }
+
+ // With no provider to ask, a cancel ends the session itself; once it
+ // ended, its outcome stands.
#[test]
- fn a_session_is_cancelled_once() {
+ fn an_unserved_session_is_cancelled_once() {
let mut open = session(FundingDirection::In);
let mut ended = expired("fs_2", NOW);
assert_eq!(
(open.cancel(NOW + 1), open.wire_item(), ended.cancel(NOW + 1)),
(
- true,
+ CancelOutcome::Ended,
HostFundingStatusSubscribeItem::Failed {
reason: FundingFailure::Cancelled,
moved: 0,
},
- false
+ CancelOutcome::Refused
+ )
+ );
+ }
+
+ // The provider holds the user's money once it has seen the payment, so
+ // only it can stop the session before that, and nobody can after.
+ #[test]
+ fn a_served_session_asks_its_provider_until_the_payment_arrives() {
+ let mut before = served(FundingDirection::In);
+ let mut after = served(FundingDirection::In);
+ reported(&mut after, &[FundingUpdate::PaymentReceived { finalized: false }]);
+
+ assert_eq!(
+ (before.cancel(NOW), before.cancel_requested, before.is_terminal(), after.cancel(NOW)),
+ (CancelOutcome::Requested, true, false, CancelOutcome::Refused)
+ );
+ }
+
+ #[test]
+ fn a_session_is_assigned_once() {
+ let mut session = served(FundingDirection::In);
+
+ assert!(!session.assign("other.dot"));
+ assert_eq!(session.provider_id.as_deref(), Some(PROVIDER));
+ }
+
+ // Reports come only from the assigned provider and only move forward, so
+ // the stored timeline is one the host can draw in order.
+ #[test]
+ fn reports_come_from_the_provider_and_only_move_forward() {
+ let mut session = served(FundingDirection::In);
+ reported(&mut session, &[FundingUpdate::Converting]);
+
+ assert_eq!(
+ (
+ session.report("other.dot", FundingUpdate::Converting, NOW),
+ session.report(PROVIDER, FundingUpdate::AwaitingPayment, NOW),
+ session.report(PROVIDER, FundingUpdate::Converting, NOW),
+ session.report(PROVIDER, FundingUpdate::Collecting { payment_id: [1; 32], amount: 5 }, NOW),
+ session.report(PROVIDER, FundingUpdate::Delivered, NOW),
+ ),
+ (
+ Err(ReportRefusal::NotFound),
+ Err(ReportRefusal::OutOfOrder),
+ Err(ReportRefusal::OutOfOrder),
+ Err(ReportRefusal::OutOfOrder),
+ Err(ReportRefusal::OutOfOrder),
+ )
+ );
+ }
+
+ // A partial claim is retried with a new top-up, as getcash does in up to
+ // three attempts; the host follows all of them once the provider is done.
+ #[test]
+ fn an_inbound_session_settles_from_every_top_up_it_names() {
+ let mut session = served(FundingDirection::In);
+ reported(
+ &mut session,
+ &[
+ FundingUpdate::Crediting { top_up_id: [1; 32], amount: 70 },
+ FundingUpdate::Crediting { top_up_id: [2; 32], amount: 30 },
+ ],
+ );
+ let repeated = session.report(PROVIDER, FundingUpdate::Crediting { top_up_id: [1; 32], amount: 30 }, NOW);
+ let due_before_done = session.settlement_due();
+ reported(&mut session, &[FundingUpdate::Delivered]);
+
+ assert_eq!(
+ (repeated, due_before_done, session.settlement_due()),
+ (
+ Err(ReportRefusal::OutOfOrder),
+ None,
+ Some(Settlement::TopUps(vec![([1; 32], 70), ([2; 32], 30)]))
+ )
+ );
+ }
+
+ // Once a top-up or payment started, the money is in motion: the provider
+ // cannot fail the session and the deadline does not end it.
+ #[test]
+ fn moving_funds_are_settled_by_the_host_not_the_provider_or_the_clock() {
+ let mut session = served(FundingDirection::Out);
+ reported(&mut session, &[FundingUpdate::Collecting { payment_id: [3; 32], amount: 40 }]);
+
+ let failed = session.report(PROVIDER, FundingUpdate::Failed { reason: FundingFailure::ProviderTimeout }, NOW);
+ let expired = session.expire_if_due(NOW + SESSION_WINDOW_MS);
+
+ assert_eq!(
+ (failed, expired, session.settlement_due()),
+ (Err(ReportRefusal::OutOfOrder), false, Some(Settlement::Payment([3; 32], 40)))
+ );
+ }
+
+ #[test]
+ fn settling_records_what_moved_and_nothing_fails_the_session() {
+ let mut credited = served(FundingDirection::In);
+ let mut debited = served(FundingDirection::Out);
+ let mut unclaimed = served(FundingDirection::In);
+
+ credited.settle(90, NOW);
+ debited.settle(40, NOW);
+ unclaimed.settle(0, NOW);
+
+ assert_eq!(
+ (credited.wire_item(), debited.wire_item(), unclaimed.wire_item()),
+ (
+ HostFundingStatusSubscribeItem::Delivered { credited: 90 },
+ HostFundingStatusSubscribeItem::Released { debited: 40 },
+ HostFundingStatusSubscribeItem::Failed {
+ reason: FundingFailure::Other {
+ code: "not_claimed".to_string(),
+ message: "Nothing reached your balance".to_string(),
+ },
+ moved: 0,
+ },
+ )
+ );
+ }
+
+ #[test]
+ fn a_provider_failure_ends_the_session_and_is_kept() {
+ let mut session = served(FundingDirection::In);
+ let failure = FundingUpdate::Failed { reason: FundingFailure::VerificationRefused };
+
+ reported(&mut session, &[FundingUpdate::AwaitingPayment, failure.clone()]);
+
+ assert_eq!(
+ (session.wire_item(), session.last_update()),
+ (
+ HostFundingStatusSubscribeItem::Failed {
+ reason: FundingFailure::VerificationRefused,
+ moved: 0,
+ },
+ Some(&failure)
)
);
}
diff --git a/rust/crates/truapi/src/lib.rs b/rust/crates/truapi/src/lib.rs
index bfa8d34695..fe33d86d62 100644
--- a/rust/crates/truapi/src/lib.rs
+++ b/rust/crates/truapi/src/lib.rs
@@ -98,13 +98,14 @@ pub mod latest {
ChatMessageContent, ChatReaction, ChatRichText, ChatRoom, ChatRoomParticipation,
ChatRoomRegistrationStatus, ColorToken, ColumnProps, ContactHandle, ContactPickOutcome,
ContentAlignment, ContextualAlias, DerivationIndex, Dimensions, Effect, EffectProps,
- FundingDirection, FundingFailure, GenericError, HorizontalAlignment,
- HostAccountCreateProofRequest, HostAccountGetAliasRequest,
- HostAccountListRingVrfKeysRequest, HostAccountRegisterRingVrfKeyRequest,
- HostAccountRingVrfSignRequest, HostAccountSignVrfError, HostAccountSignVrfRequest,
- HostPlatform, HostSignPayloadData, HostWorkerOperationError, ImageFit, ImageProps,
- ImageSource, Modifier, OperationStartedResult, PaymentTopUpSource, PocketCard,
- ProductAccountId, ProductProofContext, RawPayload, RegisteredRingVrfKey, RemotePermission,
+ FundingAssignment, FundingDirection, FundingFailure, FundingFrameOutcome, FundingUpdate,
+ GenericError, HorizontalAlignment, HostAccountCreateProofRequest,
+ HostAccountGetAliasRequest, HostAccountListRingVrfKeysRequest,
+ HostAccountRegisterRingVrfKeyRequest, HostAccountRingVrfSignRequest,
+ HostAccountSignVrfError, HostAccountSignVrfRequest, HostPlatform, HostSignPayloadData,
+ HostWorkerOperationError, ImageFit, ImageProps, ImageSource, Modifier,
+ OperationStartedResult, PaymentTopUpSource, PocketCard, ProductAccountId,
+ ProductProofContext, RawPayload, RegisteredRingVrfKey, RemotePermission,
RemoteStatementStoreCreateProofError, RemoteStatementStoreCreateProofRequest,
RemoteStatementStoreCreateProofResponse, RemoteStatementStoreSubscribeItem,
RemoteStatementStoreSubscribeRequest, RenderContext, RendererNode, RingLocation,
@@ -213,6 +214,26 @@ pub mod latest {
/// Failure watching a funding session.
pub type HostFundingStatusSubscribeError =
LatestOf;
+ /// A provider's assigned sessions and cancel requests.
+ pub type HostFundingServeSubscribeItem =
+ LatestOf;
+ /// Failure serving funding sessions.
+ pub type HostFundingServeSubscribeError =
+ LatestOf;
+ /// A provider's progress report.
+ pub type HostFundingReportRequest =
+ LatestOf;
+ /// Failure reporting progress.
+ pub type HostFundingReportError = LatestOf;
+ /// Request to show a provider screen.
+ pub type HostFundingPresentFrameRequest =
+ LatestOf;
+ /// How a provider screen closed.
+ pub type HostFundingPresentFrameResponse =
+ LatestOf;
+ /// Failure showing a provider screen.
+ pub type HostFundingPresentFrameError =
+ LatestOf;
/// Payment request.
pub type HostPaymentRequest = LatestOf;
/// Payment request failure.
diff --git a/rust/crates/truapi/src/native/callbacks.rs b/rust/crates/truapi/src/native/callbacks.rs
index ec1f418216..5d74a4d983 100644
--- a/rust/crates/truapi/src/native/callbacks.rs
+++ b/rust/crates/truapi/src/native/callbacks.rs
@@ -327,6 +327,15 @@ pub trait NativeFundingCallbacks: Send + Sync {
amount: Option,
) -> Result;
+ /// Show provider `provider_id`'s screen at `route` for session `intent`
+ /// in a frame the host owns, and report how it closed.
+ async fn present_provider_frame(
+ &self,
+ provider_id: String,
+ intent: String,
+ route: String,
+ ) -> Result;
+
/// A session's status changed, for host UI such as the in-flight pill.
fn funding_session_changed(&self, intent: String, status: v01::HostFundingStatusSubscribeItem);
}
diff --git a/rust/crates/truapi/src/native/platform.rs b/rust/crates/truapi/src/native/platform.rs
index 962d79187d..4cf2362529 100644
--- a/rust/crates/truapi/src/native/platform.rs
+++ b/rust/crates/truapi/src/native/platform.rs
@@ -658,6 +658,18 @@ impl crate::platform::FundingPlatform for FundingCallbackPlatform {
.map_err(v01::GenericError::from)
}
+ async fn present_provider_frame(
+ &self,
+ provider: &ProductContext,
+ intent: String,
+ route: String,
+ ) -> Result {
+ self.funding
+ .present_provider_frame(provider.product_id.clone(), intent, route)
+ .await
+ .map_err(v01::GenericError::from)
+ }
+
fn funding_session_changed(&self, intent: String, status: v01::HostFundingStatusSubscribeItem) {
self.funding.funding_session_changed(intent, status);
}
diff --git a/rust/crates/truapi/src/native/runtime.rs b/rust/crates/truapi/src/native/runtime.rs
index 13a7e77fd4..7c74eda383 100644
--- a/rust/crates/truapi/src/native/runtime.rs
+++ b/rust/crates/truapi/src/native/runtime.rs
@@ -329,10 +329,24 @@ impl NativeTrUApiHostRuntime {
Ok(self.runtime.acknowledge_funding_session(&intent).await?)
}
- /// Cancel open funding session `intent`.
+ /// Cancel funding session `intent` at the user's request: it ends if no
+ /// provider serves it, otherwise its provider is asked to stop.
pub async fn cancel_funding(&self, intent: String) -> Result {
Ok(self.runtime.cancel_funding(&intent).await?)
}
+
+ /// Hand open funding session `intent` to the provider the user chose, by
+ /// product id.
+ pub async fn select_funding_provider(
+ &self,
+ intent: String,
+ provider_id: String,
+ ) -> Result {
+ Ok(self
+ .runtime
+ .select_funding_provider(&intent, &provider_id)
+ .await?)
+ }
}
#[uniffi::export]
diff --git a/rust/crates/truapi/src/platform.rs b/rust/crates/truapi/src/platform.rs
index 291130960f..d0b7cdef89 100644
--- a/rust/crates/truapi/src/platform.rs
+++ b/rust/crates/truapi/src/platform.rs
@@ -3347,6 +3347,15 @@ pub trait FundingPlatform: Send + Sync {
session: FundingPresentation,
) -> Result;
+ /// Show `provider`'s screen at `route` for session `intent`, such as its
+ /// KYC or card entry, in a frame the host owns, and answer once it closes.
+ async fn present_provider_frame(
+ &self,
+ provider: &ProductContext,
+ intent: String,
+ route: String,
+ ) -> Result;
+
/// Observe a session's status change, for host UI such as the in-flight
/// pill.
fn funding_session_changed(&self, intent: String, status: HostFundingStatusSubscribeItem) {
diff --git a/rust/crates/truapi/src/runtime.rs b/rust/crates/truapi/src/runtime.rs
index e6021beb48..76251246e7 100644
--- a/rust/crates/truapi/src/runtime.rs
+++ b/rust/crates/truapi/src/runtime.rs
@@ -91,7 +91,7 @@ pub use signing_host::StatementRenewalTarget;
#[cfg(not(target_arch = "wasm32"))]
pub use signing_host::TrackedStatementRenewalTarget;
use tracing::{instrument, warn};
-use truapi::api::{Chat, Contacts, Funding, Pocket, Renderer};
+use truapi::api::{Chat, Contacts, Funding, FundingProvider, Pocket, Renderer};
use truapi::versioned::account::{HostAccountGetError, HostAccountSignVrfError};
use truapi::versioned::chat::{
HostChatActionSubscribeError, HostChatActionSubscribeItem, HostChatActionSubscribeRequest,
@@ -107,6 +107,12 @@ use truapi::versioned::funding::{
HostFundingError, HostFundingRequest, HostFundingResponse, HostFundingStatusSubscribeError,
HostFundingStatusSubscribeItem, HostFundingStatusSubscribeRequest,
};
+use truapi::versioned::funding_provider::{
+ HostFundingPresentFrameError, HostFundingPresentFrameRequest, HostFundingPresentFrameResponse,
+ HostFundingReportError, HostFundingReportRequest, HostFundingReportResponse,
+ HostFundingServeSubscribeError, HostFundingServeSubscribeItem,
+ HostFundingServeSubscribeRequest,
+};
use truapi::versioned::pocket::{
HostPocketListSubscribeError, HostPocketListSubscribeItem, HostPocketListSubscribeRequest,
HostPocketRemoveCardError, HostPocketRemoveCardRequest, HostPocketRemoveCardResponse,
@@ -128,6 +134,7 @@ use crate::host_internal::sso_messages::RingVrfError;
use crate::host_logic::product_account::{
derivation_index_bytes, derive_product_public_key, public_key_from_address,
};
+use crate::host_logic::funding::ReportRefusal;
use crate::host_logic::session::SessionInfo;
#[cfg(test)]
use crate::host_logic::session::SessionState;
@@ -1267,6 +1274,18 @@ impl ProductRuntimeHost {
self.pocket_platform.clone().ok_or(CallError::Unsupported)
}
+ /// Funding provider access policy for this connection: a provider serves
+ /// sessions only from a Worker execution with an active session, and only
+ /// where the host has a funding surface.
+ fn funding_provider_access(&self) -> Result, CallError> {
+ if self.product.execution_kind != crate::platform::ProductExecutionKind::Worker
+ || self.authority.session_state().current().is_none()
+ {
+ return Err(CallError::Denied);
+ }
+ self.services.funding().platform().ok_or(CallError::Unsupported)
+ }
+
/// Replace the contact handles a call declares with the accounts they
/// name, before the call is shown to the user or signed.
///
@@ -1564,6 +1583,98 @@ impl Renderer for ProductRuntimeHost {
}
}
+#[truapi::async_trait]
+impl FundingProvider for ProductRuntimeHost {
+ #[instrument(skip_all, fields(runtime.method = "funding_provider.serve_subscribe"))]
+ async fn serve_subscribe(
+ &self,
+ _cx: &CallContext,
+ _request: HostFundingServeSubscribeRequest,
+ ) -> Subscription>
+ {
+ if let Err(error) = self.funding_provider_access() {
+ return Subscription::interrupted(error);
+ }
+ let registry = self.services.funding();
+ registry.bind(&self.services);
+ let loaded = registry
+ .commit(self.platform.as_ref(), current_unix_millis(), |_| {
+ ((), Vec::new())
+ })
+ .await;
+ if let Err(error) = loaded {
+ return Subscription::interrupted(CallError::HostFailure {
+ reason: error.to_string(),
+ });
+ }
+ registry.keep_expiring(&self.services);
+ Subscription::new(Box::pin(
+ registry
+ .serve(&self.product.product_id)
+ .map(|item| Ok(HostFundingServeSubscribeItem::V1(item))),
+ ))
+ }
+
+ #[instrument(skip_all, fields(runtime.method = "funding_provider.report"))]
+ async fn report(
+ &self,
+ _cx: &CallContext,
+ request: HostFundingReportRequest,
+ ) -> Result> {
+ let HostFundingReportRequest::V1(request) = request;
+ self.funding_provider_access()?;
+ let registry = self.services.funding();
+ registry.bind(&self.services);
+ let reported = registry
+ .report(
+ self.platform.as_ref(),
+ current_unix_millis(),
+ &self.product.product_id,
+ &request.intent,
+ request.update,
+ )
+ .await
+ .map_err(|error| CallError::HostFailure {
+ reason: error.to_string(),
+ })?;
+ let domain = |error| CallError::Domain(HostFundingReportError::V1(error));
+ match reported {
+ Ok(()) => Ok(HostFundingReportResponse::V1),
+ Err(ReportRefusal::NotFound) => Err(domain(v01::HostFundingReportError::NotFound)),
+ Err(ReportRefusal::OutOfOrder) => Err(domain(v01::HostFundingReportError::OutOfOrder)),
+ Err(ReportRefusal::DuplicateId) => Err(domain(v01::HostFundingReportError::DuplicateId)),
+ }
+ }
+
+ #[instrument(skip_all, fields(runtime.method = "funding_provider.present_frame"))]
+ async fn present_frame(
+ &self,
+ _cx: &CallContext,
+ request: HostFundingPresentFrameRequest,
+ ) -> Result> {
+ let HostFundingPresentFrameRequest::V1(request) = request;
+ let platform = self.funding_provider_access()?;
+ if !self
+ .services
+ .funding()
+ .is_serving(&self.product.product_id, &request.intent)
+ {
+ return Err(CallError::Domain(HostFundingPresentFrameError::V1(
+ v01::HostFundingPresentFrameError::NotFound,
+ )));
+ }
+ let outcome = platform
+ .present_provider_frame(&self.product, request.intent, request.route)
+ .await
+ .map_err(|error| CallError::HostFailure {
+ reason: error.reason,
+ })?;
+ Ok(HostFundingPresentFrameResponse::V1(
+ v01::HostFundingPresentFrameResponse { outcome },
+ ))
+ }
+}
+
#[truapi::async_trait]
impl Funding for ProductRuntimeHost {
#[instrument(skip_all, fields(runtime.method = "funding.request"))]
diff --git a/rust/crates/truapi/src/runtime/funding.rs b/rust/crates/truapi/src/runtime/funding.rs
index c09fe8892f..fd9ded08d4 100644
--- a/rust/crates/truapi/src/runtime/funding.rs
+++ b/rust/crates/truapi/src/runtime/funding.rs
@@ -9,7 +9,7 @@
//! writes so persisted snapshots land in order, persists before anyone hears of
//! the change, and then notifies subscribers and the host.
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, OnceLock, Weak};
@@ -18,15 +18,21 @@ use core::time::Duration;
use futures::channel::mpsc;
use futures::lock::Mutex as AsyncMutex;
use futures::stream::{self, BoxStream, StreamExt};
-use truapi::latest::{FundingDirection, GenericError, HostFundingStatusSubscribeItem};
+use truapi::latest::{
+ FundingDirection, FundingUpdate, GenericError, HostFundingServeSubscribeItem,
+ HostFundingStatusSubscribeItem, HostPaymentStatusSubscribeError,
+ HostPaymentStatusSubscribeItem, HostPaymentTopUpStatusSubscribeError,
+ HostPaymentTopUpStatusSubscribeItem,
+};
use super::services::RuntimeServices;
use crate::host_logic::funding::{
- FundingSession, FundingSessionError, load_sessions, retained, store_sessions,
+ CancelOutcome, FundingSession, FundingSessionError, ReportRefusal, Settlement, load_sessions,
+ retained, store_sessions,
};
use crate::platform::{
CoreStorage, FundingPlatform, FundingPresentOutcome, FundingPresentation, Platform,
- ProductContext,
+ ProductContext, ProductExecutionKind,
};
use crate::unix_time::current_unix_millis;
@@ -34,6 +40,7 @@ use crate::unix_time::current_unix_millis;
const SWEEP_RETRY: Duration = Duration::from_secs(30);
type Subscribers = HashMap>>;
+type Servers = HashMap>>;
/// Host-global funding sessions, their subscribers and the host surface.
#[derive(Default)]
@@ -46,6 +53,16 @@ pub struct FundingRegistry {
/// Whether a task is waiting on the next deadline.
sweeping: AtomicBool,
platform: OnceLock>,
+ /// Each provider's open `serve_subscribe` streams, by product id.
+ servers: Mutex,
+ /// Sessions holding a reference on their provider's worker, with that
+ /// provider.
+ holding: Mutex>,
+ /// Sessions whose top-ups or payment are being followed to an outcome.
+ following: Mutex>,
+ /// The services the registry belongs to, for worker references and
+ /// following top-ups and payments.
+ services: OnceLock>,
}
impl FundingRegistry {
@@ -60,6 +77,113 @@ impl FundingRegistry {
self.platform.get().cloned()
}
+ /// Attach the registry to the services it belongs to. Set-once.
+ pub fn bind(&self, services: &Arc) {
+ let _ = self.services.set(Arc::downgrade(services));
+ }
+
+ /// Assign open session `intent` to the provider the user chose, and hand
+ /// it to that provider. Returns whether it was open and unassigned.
+ pub async fn select_provider(
+ &self,
+ storage: &(impl CoreStorage + ?Sized),
+ now_ms: u64,
+ intent: &str,
+ provider_id: &str,
+ ) -> Result {
+ let intent = intent.to_string();
+ let provider = provider_id.to_string();
+ let assigned = self
+ .commit(storage, now_ms, move |sessions| {
+ let assigned = sessions
+ .get_mut(&intent)
+ .and_then(|session| session.assign(&provider).then(|| session.assignment()));
+ let changed = assigned.iter().map(|session| session.intent.clone()).collect();
+ (assigned, changed)
+ })
+ .await?;
+ let Some(session) = assigned else {
+ return Ok(false);
+ };
+ self.serve_item(provider_id, HostFundingServeSubscribeItem::Assigned { session });
+ Ok(true)
+ }
+
+ /// The sessions assigned to `provider_id` and requests to cancel them:
+ /// every one still in flight first, then each later one. An assignment
+ /// may arrive more than once.
+ pub fn serve(&self, provider_id: &str) -> BoxStream<'static, HostFundingServeSubscribeItem> {
+ let mut servers = self.lock_servers();
+ let mut assigned: Vec<_> = self
+ .lock_sessions()
+ .values()
+ .filter(|session| {
+ !session.is_terminal() && session.provider_id.as_deref() == Some(provider_id)
+ })
+ .cloned()
+ .collect();
+ assigned.sort_by_key(|session| session.opened_at_ms);
+ let replay: Vec<_> = assigned
+ .into_iter()
+ .flat_map(|session| {
+ let cancel = session
+ .cancel_requested
+ .then(|| HostFundingServeSubscribeItem::Cancel {
+ intent: session.intent.clone(),
+ });
+ std::iter::once(HostFundingServeSubscribeItem::Assigned {
+ session: session.assignment(),
+ })
+ .chain(cancel)
+ })
+ .collect();
+ let (sender, receiver) = mpsc::unbounded();
+ servers
+ .entry(provider_id.to_string())
+ .or_default()
+ .push(sender);
+ stream::iter(replay).chain(receiver).boxed()
+ }
+
+ /// Store `update` from `provider_id` on session `intent`.
+ pub async fn report(
+ &self,
+ storage: &(impl CoreStorage + ?Sized),
+ now_ms: u64,
+ provider_id: &str,
+ intent: &str,
+ update: FundingUpdate,
+ ) -> Result, FundingSessionError> {
+ let intent = intent.to_string();
+ let provider = provider_id.to_string();
+ self.commit(storage, now_ms, move |sessions| {
+ // Ids are the provider's own, so one named by another of its
+ // sessions would settle two sessions from a single claim.
+ let duplicate = FundingSession::named_id(&update).is_some_and(|id| {
+ sessions.values().any(|other| {
+ other.intent != intent
+ && other.provider_id.as_deref() == Some(provider.as_str())
+ && other.names(&id)
+ })
+ });
+ let reported = match sessions.get_mut(&intent) {
+ Some(_) if duplicate => Err(ReportRefusal::DuplicateId),
+ Some(session) => session.report(&provider, update, now_ms),
+ None => Err(ReportRefusal::NotFound),
+ };
+ let changed = if reported.is_ok() { vec![intent] } else { Vec::new() };
+ (reported, changed)
+ })
+ .await
+ }
+
+ /// Whether open session `intent` is assigned to `provider_id`.
+ pub fn is_serving(&self, provider_id: &str, intent: &str) -> bool {
+ self.get(intent).is_some_and(|session| {
+ !session.is_terminal() && session.provider_id.as_deref() == Some(provider_id)
+ })
+ }
+
/// Snapshot one session.
pub fn get(&self, intent: &str) -> Option {
self.lock_sessions().get(intent).cloned()
@@ -99,21 +223,35 @@ impl FundingRegistry {
.await
}
- /// Cancel open session `intent`. Returns whether it was still open.
+ /// Cancel session `intent` at the user's request: end it if no provider
+ /// serves it, otherwise ask the provider to stop.
pub async fn cancel(
&self,
storage: &(impl CoreStorage + ?Sized),
now_ms: u64,
intent: &str,
- ) -> Result {
- let intent = intent.to_string();
- self.commit(storage, now_ms, move |sessions| {
- let cancelled = sessions
- .get_mut(&intent)
- .is_some_and(|session| session.cancel(now_ms));
- (cancelled, if cancelled { vec![intent] } else { Vec::new() })
- })
- .await
+ ) -> Result {
+ let target = intent.to_string();
+ let (outcome, provider) = self
+ .commit(storage, now_ms, move |sessions| {
+ let Some(session) = sessions.get_mut(&target) else {
+ return ((CancelOutcome::Refused, None), Vec::new());
+ };
+ let outcome = session.cancel(now_ms);
+ let provider = session.provider_id.clone();
+ let changed = if outcome == CancelOutcome::Refused { Vec::new() } else { vec![target] };
+ ((outcome, provider), changed)
+ })
+ .await?;
+ if let (CancelOutcome::Requested, Some(provider)) = (outcome, provider) {
+ self.serve_item(
+ &provider,
+ HostFundingServeSubscribeItem::Cancel {
+ intent: intent.to_string(),
+ },
+ );
+ }
+ Ok(outcome)
}
/// Watch one session, receiving its current stage immediately. A terminal
@@ -182,9 +320,68 @@ impl FundingRegistry {
self.fan_out(&session);
}
}
+ drop(loaded);
+ self.follow_sessions();
Ok(result)
}
+ /// Bring worker references and followers in line with the sessions:
+ /// every assigned open session holds its provider's worker, and every
+ /// session whose outcome is due has its top-ups or payment followed.
+ fn follow_sessions(&self) {
+ let Some(services) = self.services.get().and_then(Weak::upgrade) else {
+ return;
+ };
+ let sessions = self.lock_sessions().clone();
+ let (acquire, release) = {
+ let mut holding = self.lock_holding();
+ let mut acquire = Vec::new();
+ for session in sessions.values().filter(|session| !session.is_terminal()) {
+ if let Some(provider) = &session.provider_id
+ && !holding.contains_key(&session.intent)
+ {
+ holding.insert(session.intent.clone(), provider.clone());
+ acquire.push(provider.clone());
+ }
+ }
+ let ended: Vec = holding
+ .keys()
+ .filter(|intent| sessions.get(*intent).is_none_or(FundingSession::is_terminal))
+ .cloned()
+ .collect();
+ let release: Vec = ended
+ .iter()
+ .filter_map(|intent| holding.remove(intent))
+ .collect();
+ (acquire, release)
+ };
+ for provider in &acquire {
+ services.worker_ledger.acquire(provider);
+ }
+ for provider in &release {
+ services.worker_ledger.release(provider);
+ }
+ for session in sessions.values() {
+ let (Some(settlement), Some(provider)) =
+ (session.settlement_due(), session.provider_id.clone())
+ else {
+ continue;
+ };
+ if self.lock_following().insert(session.intent.clone()) {
+ services
+ .clone()
+ .follow_settlement(session.intent.clone(), provider, settlement);
+ }
+ }
+ }
+
+ /// Send `item` to every open stream of `provider_id`.
+ fn serve_item(&self, provider_id: &str, item: HostFundingServeSubscribeItem) {
+ if let Some(senders) = self.lock_servers().get_mut(provider_id) {
+ senders.retain(|sender| sender.unbounded_send(item.clone()).is_ok());
+ }
+ }
+
/// Keep one task waiting on the earliest open deadline while any session
/// is open, so a session expires on time whether or not anyone asks. The
/// task ends once no session is open or the registry is dropped.
@@ -267,6 +464,20 @@ impl FundingRegistry {
.lock()
.expect("funding subscribers mutex poisoned")
}
+
+ fn lock_servers(&self) -> std::sync::MutexGuard<'_, Servers> {
+ self.servers.lock().expect("funding servers mutex poisoned")
+ }
+
+ fn lock_holding(&self) -> std::sync::MutexGuard<'_, HashMap> {
+ self.holding.lock().expect("funding holding mutex poisoned")
+ }
+
+ fn lock_following(&self) -> std::sync::MutexGuard<'_, HashSet> {
+ self.following
+ .lock()
+ .expect("funding following mutex poisoned")
+ }
}
/// Why a session could not be opened.
@@ -290,6 +501,7 @@ impl RuntimeServices {
/// Load persisted sessions and arm expiry for them, so sessions from
/// before a restart end on time and the host hears of them.
pub fn resume_funding(self: &Arc) {
+ self.funding().bind(self);
let services = self.clone();
(self.spawner)(Box::pin(async move {
let registry = services.funding();
@@ -346,13 +558,150 @@ impl RuntimeServices {
.await
}
- /// Cancel open session `intent`. Returns whether it was still open.
+ /// Cancel session `intent` at the user's request. Returns whether the
+ /// cancel was taken: the session ended, or its provider was asked to
+ /// stop.
pub async fn cancel_funding(&self, intent: &str) -> Result {
- self.funding()
+ let outcome = self
+ .funding()
.cancel(self.platform.as_ref(), current_unix_millis(), intent)
+ .await?;
+ Ok(outcome != CancelOutcome::Refused)
+ }
+
+ /// Hand open session `intent` to the provider the user chose. Returns
+ /// whether it was open and not yet assigned.
+ pub async fn select_funding_provider(
+ self: &Arc,
+ intent: &str,
+ provider_id: &str,
+ ) -> Result {
+ let provider = ProductContext::new_with_execution(
+ provider_id.to_string(),
+ ProductExecutionKind::Worker,
+ )
+ .map_err(|error| FundingSessionError::InvalidProvider {
+ reason: error.to_string(),
+ })?;
+ let registry = self.funding();
+ registry.bind(self);
+ registry
+ .select_provider(
+ self.platform.as_ref(),
+ current_unix_millis(),
+ intent,
+ &provider.product_id,
+ )
.await
}
+ /// Follow a session's top-ups or payment to the end, then settle it with
+ /// what moved. A stream that ends without an outcome leaves the session
+ /// to be followed again on the next change to the sessions.
+ fn follow_settlement(self: Arc, intent: String, provider_id: String, settlement: Settlement) {
+ let spawner = self.spawner.clone();
+ spawner(Box::pin(async move {
+ let registry = self.funding();
+ let provider =
+ match ProductContext::new_with_execution(provider_id, ProductExecutionKind::Worker) {
+ Ok(provider) => provider,
+ Err(error) => {
+ tracing::warn!(%error, "a funding session names an invalid provider");
+ registry.lock_following().remove(&intent);
+ return;
+ }
+ };
+ let moved = match settlement {
+ Settlement::TopUps(top_ups) => self.claimed(&provider, top_ups).await,
+ Settlement::Payment(id, amount) => self.paid(&provider, id, amount).await,
+ };
+ if let Some(moved) = moved {
+ let settling = intent.clone();
+ let settled = registry
+ .commit(self.platform.as_ref(), current_unix_millis(), move |sessions| {
+ let now_ms = current_unix_millis();
+ let settled = sessions
+ .get_mut(&settling)
+ .is_some_and(|session| session.settle(moved, now_ms));
+ ((), if settled { vec![settling] } else { Vec::new() })
+ })
+ .await;
+ if let Err(error) = settled {
+ tracing::warn!(%error, "settling a funding session failed");
+ }
+ }
+ registry.lock_following().remove(&intent);
+ }));
+ }
+
+ /// What `top_ups` credited: the full amount of each one claimed, what a
+ /// partial claim reports, nothing for one not claimed or unknown to the
+ /// host. `None` if a status could not be read to its end.
+ async fn claimed(&self, provider: &ProductContext, top_ups: Vec<([u8; 32], u128)>) -> Option {
+ let Some(platform) = self.top_up_platform() else {
+ tracing::warn!("no top-up platform to follow a funding session's claims");
+ return None;
+ };
+ let mut credited: u128 = 0;
+ for (id, amount) in top_ups {
+ let mut claimed = None;
+ let mut statuses = platform.subscribe_top_up_status(provider, id);
+ while let Some(status) = statuses.next().await {
+ match status {
+ Ok(HostPaymentTopUpStatusSubscribeItem::Claimed { finalized }) => {
+ claimed = Some(amount);
+ if finalized {
+ break;
+ }
+ }
+ Ok(HostPaymentTopUpStatusSubscribeItem::ClaimedPartially { actual_claimed }) => {
+ claimed = Some(actual_claimed);
+ break;
+ }
+ Ok(HostPaymentTopUpStatusSubscribeItem::NotClaimed)
+ | Err(HostPaymentTopUpStatusSubscribeError::NotFound) => {
+ claimed = Some(0);
+ break;
+ }
+ Ok(_) => {}
+ Err(HostPaymentTopUpStatusSubscribeError::Unknown { reason }) => {
+ tracing::warn!(%reason, "reading a top-up's status failed");
+ return None;
+ }
+ }
+ }
+ credited = credited.saturating_add(claimed?);
+ }
+ Some(credited)
+ }
+
+ /// What payment `id` moved: its amount once completed, what a partial
+ /// claim reports, nothing once failed or unknown to the host. `None` if
+ /// its status could not be read to its end.
+ async fn paid(&self, provider: &ProductContext, id: [u8; 32], amount: u128) -> Option {
+ let Some(platform) = self.payment_platform() else {
+ tracing::warn!("no payment platform to follow a funding session's payment");
+ return None;
+ };
+ let mut statuses = platform.subscribe_payment_status(provider, id);
+ while let Some(status) = statuses.next().await {
+ match status {
+ Ok(HostPaymentStatusSubscribeItem::Completed) => return Some(amount),
+ Ok(HostPaymentStatusSubscribeItem::PartiallyClaimed { actual_claimed }) => {
+ return Some(actual_claimed);
+ }
+ Ok(HostPaymentStatusSubscribeItem::Failed { .. })
+ | Err(HostPaymentStatusSubscribeError::PaymentNotFound) => return Some(0),
+ Ok(HostPaymentStatusSubscribeItem::Processing) => {}
+ Err(HostPaymentStatusSubscribeError::Unknown { reason }) => {
+ tracing::warn!(%reason, "reading a payment's status failed");
+ return None;
+ }
+ }
+ }
+ None
+ }
+
/// Open a session and show the host's funding overlay for it: the one
/// path a product's `request` and the host's own Balance card both take.
///
@@ -365,6 +714,7 @@ impl RuntimeServices {
amount: Option,
) -> Result {
let registry = self.funding();
+ registry.bind(self);
let platform = registry.platform().ok_or(OpenFundingError::Unsupported)?;
let now_ms = current_unix_millis();
let session = FundingSession::new(
@@ -434,7 +784,7 @@ mod tests {
use futures::executor::block_on;
use truapi::latest::FundingFailure;
- use crate::host_logic::funding::FundingStage;
+ use crate::host_logic::funding::{FundingStage, ReportRefusal};
use crate::test_support::stub_platform;
const NOW: u64 = 1_700_000_000_000;
@@ -549,6 +899,44 @@ mod tests {
);
}
+ // A provider's top-up and payment ids are its own, so naming one in two
+ // sessions would settle both from a single claim and show the money twice
+ // in the user's history. Another provider's ids are a different namespace.
+ #[test]
+ fn a_provider_cannot_name_one_top_up_in_two_sessions() {
+ let storage = stub_platform();
+ let registry = FundingRegistry::default();
+ let served = |intent: &str, provider: &str| {
+ let mut session = session(intent, NOW);
+ assert!(session.assign(provider));
+ session
+ };
+ for session in [
+ served("fs_a", "ramp.dot"),
+ served("fs_b", "ramp.dot"),
+ served("fs_c", "other.dot"),
+ ] {
+ insert(®istry, storage.as_ref(), session);
+ }
+ let crediting = FundingUpdate::Crediting {
+ top_up_id: [9; 32],
+ amount: 100,
+ };
+ let report = |provider: &str, intent: &str| {
+ block_on(registry.report(storage.as_ref(), NOW, provider, intent, crediting.clone()))
+ .expect("stored")
+ };
+
+ assert_eq!(
+ (
+ report("ramp.dot", "fs_a"),
+ report("ramp.dot", "fs_b"),
+ report("other.dot", "fs_c"),
+ ),
+ (Ok(()), Err(ReportRefusal::DuplicateId), Ok(()))
+ );
+ }
+
#[test]
fn a_failed_save_changes_nothing_in_memory() {
let storage = stub_platform();
diff --git a/rust/crates/truapi/src/runtime/tests.rs b/rust/crates/truapi/src/runtime/tests.rs
index 0cf130837a..8d5fb3da98 100644
--- a/rust/crates/truapi/src/runtime/tests.rs
+++ b/rust/crates/truapi/src/runtime/tests.rs
@@ -2266,6 +2266,15 @@ impl crate::platform::FundingPlatform for RecordingFundingPlatform {
Ok(self.outcome)
}
+ async fn present_provider_frame(
+ &self,
+ _provider: &ProductContext,
+ _intent: String,
+ _route: String,
+ ) -> Result {
+ Ok(truapi::latest::FundingFrameOutcome::Closed)
+ }
+
fn funding_session_changed(
&self,
intent: String,
@@ -2566,6 +2575,168 @@ fn a_test_host_settles_a_funding_session_once() {
);
}
+fn funding_services_over(platform: Arc) -> Arc {
+ let (host_config, _) = runtime_config("funding.dot");
+ RuntimeServices::new(
+ platform,
+ host_config.host.host_info.clone(),
+ host_config.people_chain_genesis_hash,
+ host_config.bulletin_chain_genesis_hash,
+ host_config.asset_hub_chain_genesis_hash,
+ test_spawner(),
+ )
+}
+
+/// A provider's worker connection, signed in.
+fn provider_worker(services: &Arc, provider_id: &str) -> ProductRuntimeHost {
+ let (host_config, _) = runtime_config(provider_id);
+ let product = ProductContext::new_with_execution(
+ provider_id.to_string(),
+ crate::platform::ProductExecutionKind::Worker,
+ )
+ .expect("test provider context is valid");
+ let pairing_host = PairingHost::new(services.clone(), host_config);
+ let host = ProductRuntimeHost::from_services(
+ services.clone(),
+ crate::host_core::ConnectionAdapters::from_services(services),
+ pairing_host,
+ product,
+ );
+ install_pairing_session(&host, session_info());
+ host
+}
+
+fn first_served(
+ host: &ProductRuntimeHost,
+) -> Option>> {
+ futures::executor::block_on(
+ futures::executor::block_on(truapi::api::FundingProvider::serve_subscribe(
+ host,
+ &CallContext::default(),
+ truapi::versioned::funding_provider::HostFundingServeSubscribeRequest::V1,
+ ))
+ .next(),
+ )
+ .map(|item| item.map(|truapi::versioned::funding_provider::HostFundingServeSubscribeItem::V1(item)| item))
+}
+
+fn report_funding(
+ host: &ProductRuntimeHost,
+ intent: &str,
+ update: v01::FundingUpdate,
+) -> Result<(), CallError> {
+ futures::executor::block_on(truapi::api::FundingProvider::report(
+ host,
+ &CallContext::default(),
+ truapi::versioned::funding_provider::HostFundingReportRequest::V1(
+ v01::HostFundingReportRequest {
+ intent: intent.to_string(),
+ update,
+ },
+ ),
+ ))
+ .map(|_| ())
+}
+
+fn wait_for_stage(
+ services: &RuntimeServices,
+ intent: &str,
+) -> Option {
+ for _ in 0..200 {
+ let session = services.funding().get(intent)?;
+ if session.is_terminal() {
+ return Some(session.wire_item());
+ }
+ std::thread::sleep(Duration::from_millis(10));
+ }
+ None
+}
+
+// The provider's worker takes a session from assignment to delivered, and a
+// restart in between loses nothing: the restarted worker is handed the
+// session again with its last update, and the host still decides delivery
+// from the claim of the top-up the provider named.
+#[test]
+fn a_provider_worker_delivers_a_session_across_a_restart() {
+ let storage = stub_platform();
+ let start = |storage: &Arc| {
+ let services = funding_services_over(storage.clone());
+ assert!(services.funding().install_platform(RecordingFundingPlatform::answering(
+ crate::platform::FundingPresentOutcome::Started,
+ )));
+ assert!(services.install_top_up_platform(Arc::new(RecordingTopUpPlatform::default())));
+ services.resume_funding();
+ services
+ };
+ let before = start(&storage);
+ let intent = futures::executor::block_on(before.open_funding(
+ None,
+ v01::FundingDirection::In,
+ Some(1_000),
+ ))
+ .expect("opened")
+ .intent;
+ assert!(futures::executor::block_on(before.select_funding_provider(&intent, "ramp.dot")).expect("selected"));
+ let worker = provider_worker(&before, "ramp.dot");
+ let assigned = first_served(&worker);
+ let crediting = v01::FundingUpdate::Crediting { top_up_id: [7; 32], amount: 1_000 };
+ report_funding(&worker, &intent, v01::FundingUpdate::Converting).expect("converting");
+ report_funding(&worker, &intent, crediting.clone()).expect("crediting");
+
+ let after = start(&storage);
+ let restarted = provider_worker(&after, "ramp.dot");
+ let replayed = first_served(&restarted);
+ report_funding(&restarted, &intent, v01::FundingUpdate::Delivered).expect("delivered");
+
+ let assignment = |last_update| {
+ Some(Ok(v01::HostFundingServeSubscribeItem::Assigned {
+ session: v01::FundingAssignment {
+ intent: intent.clone(),
+ direction: v01::FundingDirection::In,
+ amount: Some(1_000),
+ expires_at: after.funding().get(&intent).expect("kept").deadline_ms,
+ last_update,
+ },
+ }))
+ };
+ assert_eq!(
+ (assigned, replayed, wait_for_stage(&after, &intent)),
+ (
+ assignment(None),
+ assignment(Some(crediting)),
+ Some(v01::HostFundingStatusSubscribeItem::Delivered { credited: 1_000 }),
+ )
+ );
+}
+
+// Only the assigned provider's worker may serve or report on a session.
+#[test]
+fn only_the_assigned_provider_worker_reports() {
+ let services = funding_services();
+ assert!(services.funding().install_platform(RecordingFundingPlatform::answering(
+ crate::platform::FundingPresentOutcome::Started,
+ )));
+ let intent = futures::executor::block_on(services.open_funding(None, v01::FundingDirection::In, None))
+ .expect("opened")
+ .intent;
+ assert!(futures::executor::block_on(services.select_funding_provider(&intent, "ramp.dot")).expect("selected"));
+ let app = funding_host(&services, "ramp.dot", true);
+ let other = provider_worker(&services, "other.dot");
+
+ assert!(matches!(
+ (
+ report_funding(&app, &intent, v01::FundingUpdate::AwaitingPayment),
+ report_funding(&other, &intent, v01::FundingUpdate::AwaitingPayment),
+ ),
+ (
+ Err(CallError::Denied),
+ Err(CallError::Domain(truapi::versioned::funding_provider::HostFundingReportError::V1(
+ v01::HostFundingReportError::NotFound
+ ))),
+ )
+ ));
+}
+
#[derive(Default)]
struct RecordingTopUpPlatform {
started: Mutex>,
diff --git a/rust/crates/truapi/src/v01.rs b/rust/crates/truapi/src/v01.rs
index c3ddd88b55..9756a6eb6b 100644
--- a/rust/crates/truapi/src/v01.rs
+++ b/rust/crates/truapi/src/v01.rs
@@ -8,6 +8,7 @@ mod common;
mod contacts;
mod entropy;
mod funding;
+mod funding_provider;
mod local_storage;
mod locale;
mod notifications;
@@ -32,6 +33,7 @@ pub use common::*;
pub use contacts::*;
pub use entropy::*;
pub use funding::*;
+pub use funding_provider::*;
pub use local_storage::*;
pub use locale::*;
pub use notifications::*;
diff --git a/rust/crates/truapi/src/v01/funding_provider.rs b/rust/crates/truapi/src/v01/funding_provider.rs
new file mode 100644
index 0000000000..5d41a19084
--- /dev/null
+++ b/rust/crates/truapi/src/v01/funding_provider.rs
@@ -0,0 +1,164 @@
+use parity_scale_codec::{Decode, Encode};
+
+use super::funding::{FundingDirection, FundingFailure};
+
+/// A funding session as the provider serving it receives it.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct FundingAssignment {
+ /// Session id, used in every report about it.
+ pub intent: String,
+ /// Which way value moves.
+ pub direction: FundingDirection,
+ /// Amount sought, in the user's payment balance units. `None` when the
+ /// user left it to the provider's screens.
+ pub amount: Option,
+ /// When the session expires unless funds are already moving, in Unix
+ /// milliseconds.
+ pub expires_at: u64,
+ /// The last update the provider reported, so a restarted worker resumes
+ /// where it stopped.
+ pub last_update: Option,
+}
+
+/// Progress a provider reports for a session it serves.
+///
+/// Updates only move forward. Once funds are moving the host decides the
+/// outcome: an inbound session is delivered from the claims of the top-ups it
+/// names, an outbound one is released once the user's payment completes.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+#[cfg_attr(
+ all(feature = "runtime", not(target_arch = "wasm32")),
+ derive(uniffi::Enum)
+)]
+pub enum FundingUpdate {
+ /// Inbound: waiting for the user to pay the provider.
+ AwaitingPayment,
+ /// Inbound: the provider has seen the user's payment.
+ PaymentReceived {
+ /// Whether the payment can no longer be reversed.
+ finalized: bool,
+ },
+ /// Inbound: the provider is converting the payment to the user's balance
+ /// asset.
+ Converting,
+ /// Inbound: the provider started a top-up with `payment.topUp` to credit
+ /// the user. A partial claim may be followed by further top-ups.
+ Crediting {
+ /// Id the top-up was started with.
+ top_up_id: [u8; 32],
+ /// Amount the top-up asks to credit.
+ amount: u128,
+ },
+ /// Inbound: the provider started every top-up it will. The session is
+ /// delivered with what they claimed.
+ Delivered,
+ /// Outbound: the provider asked the user to pay it with
+ /// `payment.request`. The session is released once that payment
+ /// completes.
+ Collecting {
+ /// Id the payment request was made with.
+ payment_id: [u8; 32],
+ /// Amount the payment asks for.
+ amount: u128,
+ },
+ /// The session ended without success, before any funds moved.
+ Failed {
+ /// Why it ended.
+ reason: FundingFailure,
+ },
+}
+
+/// Item of [`crate::api::FundingProvider::serve_subscribe`].
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub enum HostFundingServeSubscribeItem {
+ /// A session the user chose this provider for, or one still in flight
+ /// when the stream opened.
+ Assigned {
+ /// The session.
+ session: FundingAssignment,
+ },
+ /// The user asked to cancel a session. The provider reports `Failed`
+ /// with `Cancelled` if it can still stop.
+ Cancel {
+ /// Session to cancel.
+ intent: String,
+ },
+}
+
+/// Error from [`crate::api::FundingProvider::serve_subscribe`].
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub enum HostFundingServeSubscribeError {
+ /// Catch-all.
+ Unknown {
+ /// Human-readable failure reason.
+ reason: String,
+ },
+}
+
+/// Request to report progress on a session.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct HostFundingReportRequest {
+ /// Session the update is about.
+ pub intent: String,
+ /// The update.
+ pub update: FundingUpdate,
+}
+
+/// Error from [`crate::api::FundingProvider::report`].
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub enum HostFundingReportError {
+ /// No open session with this id is assigned to the caller.
+ NotFound,
+ /// The update does not follow the session's last one, or does not fit
+ /// its direction.
+ OutOfOrder,
+ /// The top-up or payment request it names is already named by another of
+ /// the caller's sessions.
+ DuplicateId,
+ /// Catch-all.
+ Unknown {
+ /// Human-readable failure reason.
+ reason: String,
+ },
+}
+
+/// Request to show one of the provider's screens for a session.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct HostFundingPresentFrameRequest {
+ /// Session the screen belongs to.
+ pub intent: String,
+ /// Route of the provider's App to show, such as its KYC or card entry.
+ pub route: String,
+}
+
+/// How a provider screen closed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Encode, Decode)]
+#[cfg_attr(
+ all(feature = "runtime", not(target_arch = "wasm32")),
+ derive(uniffi::Enum)
+)]
+pub enum FundingFrameOutcome {
+ /// The provider's screen finished and closed itself.
+ Closed,
+ /// The user closed it.
+ Dismissed,
+}
+
+/// Response of [`crate::api::FundingProvider::present_frame`].
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct HostFundingPresentFrameResponse {
+ /// How the screen closed.
+ pub outcome: FundingFrameOutcome,
+}
+
+/// Error from [`crate::api::FundingProvider::present_frame`].
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub enum HostFundingPresentFrameError {
+ /// No open session with this id is assigned to the caller.
+ NotFound,
+ /// Catch-all.
+ Unknown {
+ /// Human-readable failure reason.
+ reason: String,
+ },
+}
diff --git a/rust/crates/truapi/src/versioned.rs b/rust/crates/truapi/src/versioned.rs
index c24d74b70f..212f0b5113 100644
--- a/rust/crates/truapi/src/versioned.rs
+++ b/rust/crates/truapi/src/versioned.rs
@@ -41,6 +41,7 @@ pub mod coin_payment;
pub mod contacts;
pub mod entropy;
pub mod funding;
+pub mod funding_provider;
pub mod local_storage;
pub mod locale;
pub mod notifications;
diff --git a/rust/crates/truapi/src/versioned/funding_provider.rs b/rust/crates/truapi/src/versioned/funding_provider.rs
new file mode 100644
index 0000000000..70f34f8e02
--- /dev/null
+++ b/rust/crates/truapi/src/versioned/funding_provider.rs
@@ -0,0 +1,15 @@
+//! Versioned wrappers for [`FundingProvider`](crate::api::FundingProvider) methods.
+
+use crate::v01;
+
+truapi_macros::versioned_type! {
+ pub enum HostFundingServeSubscribeRequest { V1 }
+ pub enum HostFundingServeSubscribeItem { V1 => v01::HostFundingServeSubscribeItem }
+ pub enum HostFundingServeSubscribeError { V1 => v01::HostFundingServeSubscribeError }
+ pub enum HostFundingReportRequest { V1 => v01::HostFundingReportRequest }
+ pub enum HostFundingReportResponse { V1 }
+ pub enum HostFundingReportError { V1 => v01::HostFundingReportError }
+ pub enum HostFundingPresentFrameRequest { V1 => v01::HostFundingPresentFrameRequest }
+ pub enum HostFundingPresentFrameResponse { V1 => v01::HostFundingPresentFrameResponse }
+ pub enum HostFundingPresentFrameError { V1 => v01::HostFundingPresentFrameError }
+}
diff --git a/scripts/battery.sh b/scripts/battery.sh
index 7afcfb8c32..c2164b6e05 100755
--- a/scripts/battery.sh
+++ b/scripts/battery.sh
@@ -255,20 +255,37 @@ pocket_phase() {
funding_phase() {
local log="$LOG_DIR/funding-host-cli.log"
- echo "battery: funding phase (host transcript $LOG_DIR/funding-host-transcript.jsonl)"
+ echo "battery: funding phase (host transcripts $LOG_DIR/funding-host-transcript*.jsonl)"
# Overlay requests and session changes as the host saw them. The cases read
# it so a pass cannot rest on the product's word alone.
export TRUAPI_FUNDING_LOG="$ROOT/$LOG_DIR/funding-host-transcript.jsonl"
- # One outcome per request the cases make, in order; see funding-e2e.ts.
- export TRUAPI_FUNDING_OUTCOMES="deliver:900,release:500,fail,dismiss"
+ # One outcome per request the cases make, in order; see funding-e2e.ts and
+ # funding-provider-e2e.ts.
+ export TRUAPI_FUNDING_OUTCOMES="deliver:900,release:500,fail,dismiss,provide,provide,provide-cancel"
+ # The scripted top-up and payment engines' record, kept across the restart.
+ export TRUAPI_FUNDING_LEDGER="$ROOT/$LOG_DIR/funding-host-ledger.jsonl"
+ export TRUAPI_FUNDING_STATE="$ROOT/$LOG_DIR/funding-battery-state.json"
+ rm -f "$TRUAPI_FUNDING_LOG" "$TRUAPI_FUNDING_LEDGER" "$TRUAPI_FUNDING_STATE"
+ # Run as a Worker, so the script can also serve the sessions it asks for as
+ # the provider; then once more on the same storage, as a restarted host.
+ local rc=0
+ funding_run start "$log" || rc=$?
+ [ "$rc" = 0 ] || return "$rc"
+ export TRUAPI_FUNDING_LOG="$ROOT/$LOG_DIR/funding-host-transcript-resumed.jsonl"
rm -f "$TRUAPI_FUNDING_LOG"
- "$HOST" signing-host \
+ funding_run resume "$LOG_DIR/funding-host-cli-resumed.log"
+}
+
+funding_run() {
+ local stage="$1" log="$2"
+ TRUAPI_FUNDING_STAGE="$stage" "$HOST" signing-host \
--product-id "$PRODUCT_ID" \
+ --execution-kind worker \
--script "$FUNDING_SCRIPT" \
--auto-accept \
${HOST_ARGS[@]+"${HOST_ARGS[@]}"} > >(tee "$log") 2>&1 &
local host_pid=$! rc=0
- start_watchdog "$host_pid" "funding phase"
+ start_watchdog "$host_pid" "funding phase ($stage)"
wait "$host_pid" || rc=$?
stop_watchdog
return "$rc"