perf: parallel inbox reads and deliveries, kept final reads, background listening

dsm_client/android/app/src/main/java/com/dsm/wallet/service/BleBackgroundService.kt

@@ -99,13 +99,16 @@ class BleBackgroundService : Service() {
         // Create notification channel (required for Android O+)
         createNotificationChannel()
 
-        // Start foreground with notification + explicit service type (required API 34+)
+        // Start foreground with notification + explicit service type (required API 34+).
+        // connectedDevice only: it has no daily limit, and the NFC permission the
+        // manifest declares satisfies its prerequisite. dataSync is capped at six
+        // hours a day from Android 15, and a service that keeps listening for its
+        // contacts in the background would reach the cap.
         if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.UPSIDE_DOWN_CAKE) {
             startForeground(
                 NOTIFICATION_ID,
                 createNotification(),
                 ServiceInfo.FOREGROUND_SERVICE_TYPE_CONNECTED_DEVICE
-                    or ServiceInfo.FOREGROUND_SERVICE_TYPE_DATA_SYNC
             )
         } else {
             startForeground(NOTIFICATION_ID, createNotification())
@@ -128,6 +131,13 @@ class BleBackgroundService : Service() {
         return START_STICKY
     }
 
+    // A foreground service type with a time limit reached it (Android 15). The
+    // service must stop within seconds or the system throws.
+    override fun onTimeout(startId: Int, fgsType: Int) {
+        Log.w(TAG, "foreground service type $fgsType reached its time limit; stopping")
+        stopSelf()
+    }
+
     override fun onDestroy() {
         Log.i(TAG, "BLE background service destroyed")
         try {
@@ -239,8 +249,8 @@ class BleBackgroundService : Service() {
         )
 
         return NotificationCompat.Builder(this, CHANNEL_ID)
-            .setContentTitle("DSM Offline Mode Active")
-            .setContentText("Ready for Bluetooth transfers")
+            .setContentTitle("DSM is ready to receive")
+            .setContentText("Listening for transfers from your contacts")
             .setSmallIcon(android.R.drawable.stat_sys_data_bluetooth) // Use system Bluetooth icon
             .setContentIntent(pendingIntent)
             .setOngoing(true) // Cannot be dismissed by user

dsm_client/deterministic_state_machine/dsm_sdk/Cargo.toml

@@ -94,7 +94,7 @@ ff = "0.14.0"
 
 # Networking
 rustls = { version = "0.23.37", features = ["ring"] }
-reqwest = { version = "0.12.28", default-features = false, features = ["rustls-tls", "json"] }
+reqwest = { version = "0.12.28", default-features = false, features = ["rustls-tls", "json", "http2"] }
 rcgen = "0.14.8"
 tokio-rustls = "0.26.4"
 url = "2.4.0"

dsm_client/deterministic_state_machine/dsm_sdk/src/economic_fixtures.rs

@@ -205,6 +205,8 @@ impl TestIdentity {
 /// Any identity this process held before is replaced: this is a new device.
 pub fn create_identity(seed: u8) -> TestIdentity {
     crate::reset_sdk_context_for_testing();
+    #[cfg(test)]
+    crate::sdk::final_reads::forget_everything();
     let mnemonic = test_mnemonic(seed);
     crate::sdk::recovery_sdk::RecoverySDK::derive_and_cache_key(&mnemonic)
         .expect("unlock the mnemonic");

dsm_client/deterministic_state_machine/dsm_sdk/src/handlers/node_e2e_tests.rs

@@ -1325,8 +1325,11 @@ async fn a_key_held_by_an_exercise_its_own_bytes_refute_is_skipped_on_those_byte
     );
 
     // The walk at the next generation, by a verifier that has read nothing
-    // yet — this one keeps the held key's final reads — with every node's
-    // request log cleared: what the walk reads, the nodes are asked.
+    // yet — this one keeps the held key's final reads, and so does this
+    // process for every later verifier, so what it kept is forgotten — with
+    // every node's request log cleared: what the walk reads, the nodes are
+    // asked.
+    crate::sdk::final_reads::forget_everything();
     let walking = VerifierContext::new(&set, Some(own), parents.as_ref()).expect("a verifier");
     for node in &p.nodes.nodes {
         node.forget_requests();
@@ -1373,6 +1376,33 @@ async fn a_key_held_by_an_exercise_its_own_bytes_refute_is_skipped_on_those_byte
     assert!(cells_read.contains(&k0), "the held key was read");
     assert!(cells_read.contains(&k1), "the next key was read");
 
+    // A later verifier in this process starts from the held key's kept
+    // final reads: the walk comes out the same, the held key is not asked
+    // for again, and the open next key is, since it may have been written.
+    for node in &p.nodes.nodes {
+        node.forget_requests();
+    }
+    let later = VerifierContext::new(&set, Some(own), parents.as_ref()).expect("a verifier");
+    let again = later
+        .verifier()
+        .walk_parent(&chains, &m.vault_id, &r1, 0, WALK_BUDGET)
+        .expect("the walk again");
+    assert_eq!(again.outcome, walked.outcome);
+    let asked_again: Vec<String> = p
+        .nodes
+        .nodes
+        .iter()
+        .flat_map(|node| node.requests())
+        .collect();
+    assert!(
+        !asked_again.contains(&k0),
+        "the held key was asked for again: {asked_again:?}"
+    );
+    assert!(
+        asked_again.contains(&k1),
+        "the open next key was read again"
+    );
+
     // The next trade takes the next key, and the vault prices it at the
     // reserves B's first trade left.
     let q2 = realized_trade(&p, &m, 10).await;

dsm_client/deterministic_state_machine/dsm_sdk/src/handlers/storage_routes.rs

@@ -2173,14 +2173,37 @@ impl AppRouterImpl {
         // Routes read to full coverage, routes read partially, and routes no
         // member answered for. Only the first kind says "nothing more there".
         let (mut routes_read, mut routes_partial, mut routes_unread) = (0usize, 0usize, 0usize);
-        for tagged in tagged_addresses {
-            let retrieved = b0x_sdk.retrieve_resuming(&tagged.address).await;
+        // Every route is read at once, each by a reader of its own: a sync
+        // costs the slowest route's reads, not the sum of every route's. What
+        // each read is then processed route by route, in the order it always
+        // was; a route's processing never depends on another route's read.
+        let mut readers = Vec::with_capacity(tagged_addresses.len());
+        for _ in &tagged_addresses {
+            match crate::sdk::b0x_sdk::B0xSDK::new(
+                device_id_b32.to_string(),
+                self.core_sdk.clone(),
+                storage_endpoints.to_vec(),
+            ) {
+                Ok(reader) => readers.push(reader),
+                Err(e) => return Err(report.stop(format!("b0x init failed: {e}"))),
+            }
+        }
+        let retrievals = futures::future::join_all(
+            readers
+                .iter_mut()
+                .zip(&tagged_addresses)
+                .map(|(reader, tagged)| reader.retrieve_resuming(&tagged.address)),
+        )
+        .await;
+        for ((tagged, mut route_sdk), retrieved) in
+            tagged_addresses.into_iter().zip(readers).zip(retrievals)
+        {
             // Replies ride the same spool as forward transfers, as distinct
             // payloads; they are drained whether or not this route yielded
             // transfers, since a reply alone can release a pending gate.
-            self.process_countersign_deltas(&mut b0x_sdk, &tagged.address, &mut report)
+            self.process_countersign_deltas(&mut route_sdk, &tagged.address, &mut report)
                 .await;
-            self.process_finality_checkpoints(&mut b0x_sdk, &tagged.address, &mut report)
+            self.process_finality_checkpoints(&mut route_sdk, &tagged.address, &mut report)
                 .await;
             // §5.2: the route fixes the tip an item was composed on — the
             // inbox address hashes the relationship's tip (storage node spec
@@ -2189,7 +2212,7 @@ impl AppRouterImpl {
             // of an object the canonical apply has decided is consumed; any
             // other is left where it is.
             let stale = tagged.freshness == RouteFreshness::PreviousTip;
-            for artifact in b0x_sdk.take_evidence_artifacts() {
+            for artifact in route_sdk.take_evidence_artifacts() {
                 let copy = if stale {
                     crate::handlers::recipient_dispatch::classify_stale_receipt_copy(
                         &artifact.evidence.full_receipt_bytes,
@@ -2212,7 +2235,7 @@ impl AppRouterImpl {
                     &mut report,
                 );
             }
-            for (method, body) in b0x_sdk.take_cert_resync_messages() {
+            for (method, body) in route_sdk.take_cert_resync_messages() {
                 let outcome = if method == crate::storage::client_db::CERT_RESYNC_REQUEST_METHOD {
                     self.handle_cert_resync_request(&body, storage_endpoints.to_vec())
                         .await

dsm_client/deterministic_state_machine/dsm_sdk/src/jni/unified_protobuf_bridge.rs

@@ -601,28 +601,33 @@ pub extern "system" fn Java_com_dsm_wallet_bridge_UnifiedNativeApi_getTransportH
 
 /// App-backgrounded lifecycle transition. Rust performs the ENTIRE decision:
 /// it stops the inbox poller unless a §16.6 settlement step is still owed (a
-/// sender-side pending gate, or a countersigned reply not yet delivered), and
-/// returns the single directive the platform layer must obey.
+/// sender-side pending gate, or a countersigned reply not yet delivered) or a
+/// contact can send to this device, and returns the single directive the
+/// platform layer must obey.
 ///
 /// Returns TRUE when the host MUST keep its foreground service alive: killing
 /// the service kills the poller with it, stranding money in flight until the
-/// user happens to reopen the app. The caller performs no protocol reasoning of
-/// its own — it relays this directive and nothing else.
+/// user happens to reopen the app, and leaving what a contact sends unread
+/// until then. The caller performs no protocol reasoning of its own — it
+/// relays this directive and nothing else.
 #[no_mangle]
 pub extern "system" fn Java_com_dsm_wallet_bridge_UnifiedNativeApi_onAppBackgrounded(
     _env: jni::sys::JNIEnv,
     _clazz: jni::sys::jclass,
 ) -> jni::sys::jboolean {
-    // Only a readable "nothing owed" lets the host go: the poller was stopped.
-    // Outstanding work, an unreadable settlement state, or a panic keep it
-    // alive.
+    // Only a readable "nothing owed, nobody to hear from" lets the host go:
+    // the poller was stopped. Outstanding work, a contact, an unreadable
+    // store, or a panic keep it alive.
     let stopped = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
         crate::logging::init_android_device_logging();
         crate::sdk::inbox_poller::stop_poller_for_lifecycle()
     }));
     match stopped {
-        Ok(Ok(true)) => 0,
-        Ok(Ok(false)) => 1,
+        Ok(Ok(crate::sdk::inbox_poller::Backgrounded::Stopped)) => 0,
+        Ok(Ok(
+            crate::sdk::inbox_poller::Backgrounded::Settling
+            | crate::sdk::inbox_poller::Backgrounded::Listening,
+        )) => 1,
         Ok(Err(e)) => {
             log::error!("onAppBackgrounded: settlement state unreadable, keeping alive: {e}");
             1