From 5e7e928650e687206ddf321800c97a400f1230ba Mon Sep 17 00:00:00 2001 From: archipelago Date: Wed, 22 Jul 2026 17:30:47 -0400 Subject: [PATCH] =?UTF-8?q?feat(wallet):=20real-time=20tx=20push=20?= =?UTF-8?q?=E2=80=94=200-conf=20shows=20in=20seconds,=20not=20on=20poll?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backend streams LND /v1/transactions/subscribe (fires on mempool arrival and each confirmation) and nudges the /ws/db revision per event; the Home wallet card subscribes and refetches (debounced 800ms). An incoming broadcast now appears while the 30s poll is still asleep. Reconnects forever with capped backoff when LND is down/locked/absent. Co-Authored-By: Claude Fable 5 --- core/archipelago/src/api/rpc/lnd/mod.rs | 64 +++++++++++++++++++++++++ core/archipelago/src/api/rpc/mod.rs | 2 +- core/archipelago/src/server.rs | 6 +++ neode-ui/src/views/Home.vue | 21 +++++++- 4 files changed, 90 insertions(+), 3 deletions(-) diff --git a/core/archipelago/src/api/rpc/lnd/mod.rs b/core/archipelago/src/api/rpc/lnd/mod.rs index 4f2e4205..cb81d939 100644 --- a/core/archipelago/src/api/rpc/lnd/mod.rs +++ b/core/archipelago/src/api/rpc/lnd/mod.rs @@ -56,6 +56,70 @@ pub(crate) async fn read_lnd_admin_macaroon() -> Result> { } } +/// Real-time wallet push (user req 2026-07-22): the UI must reflect an +/// incoming on-chain transaction the moment the node sees it, not on the +/// next poll. Streams LND's `/v1/transactions/subscribe` — it fires on +/// 0-conf mempool arrival AND again on each confirmation — and nudges the +/// shared data-model revision on every event; /ws/db pushes that to every +/// connected client and the frontend refetches wallet balance/transactions. +/// Reconnects forever with capped backoff: LND restarting, wallet locked, or +/// LND not installed yet all just mean "try again shortly". +pub(crate) fn spawn_lnd_tx_watcher(state_manager: std::sync::Arc) { + tokio::spawn(async move { + let mut delay = std::time::Duration::from_secs(5); + loop { + match stream_lnd_transactions(&state_manager).await { + // Stream ended cleanly (LND shutdown) — resume fast. + Ok(()) => delay = std::time::Duration::from_secs(5), + Err(e) => { + tracing::debug!("lnd tx watcher: {e:#} — retrying in {delay:?}"); + } + } + tokio::time::sleep(delay).await; + delay = (delay * 2).min(std::time::Duration::from_secs(120)); + } + }); +} + +async fn stream_lnd_transactions(sm: &crate::state::StateManager) -> Result<()> { + let macaroon_hex = hex::encode(read_lnd_admin_macaroon().await?); + // Dedicated client: the shared lnd_client() carries a 15s total timeout, + // which would kill this deliberately long-lived stream. + let client = reqwest::Client::builder() + .no_proxy() + .connect_timeout(std::time::Duration::from_secs(10)) + .danger_accept_invalid_certs(true) + .build() + .context("Failed to create streaming HTTP client")?; + let mut resp = client + .get(format!("{LND_REST_BASE_URL}/v1/transactions/subscribe")) + .header("Grpc-Metadata-macaroon", &macaroon_hex) + .send() + .await + .context("subscribe request failed")?; + anyhow::ensure!( + resp.status().is_success(), + "transactions/subscribe returned {}", + resp.status() + ); + tracing::info!("lnd tx watcher: streaming wallet transaction events"); + while let Some(chunk) = resp.chunk().await? { + if chunk.is_empty() { + continue; + } + // Any streamed event = wallet activity. Revision bump is the push + // contract (same pattern as the mesh-peer bridge in server.rs) — the + // clients refetch, so we don't need to parse the event body. + let (data, _) = sm.get_snapshot().await; + sm.update_data(data).await; + tracing::debug!( + bytes = chunk.len(), + "lnd tx watcher: wallet tx event — nudged ws clients" + ); + } + Ok(()) +} + impl RpcHandler { /// Helper: create an authenticated LND REST client. /// Returns an HTTP client configured for LND's self-signed TLS and the diff --git a/core/archipelago/src/api/rpc/mod.rs b/core/archipelago/src/api/rpc/mod.rs index ffeb036b..fba353c3 100644 --- a/core/archipelago/src/api/rpc/mod.rs +++ b/core/archipelago/src/api/rpc/mod.rs @@ -15,7 +15,7 @@ mod fips; mod handshake; mod identity; mod interfaces; -pub(in crate::api) mod lnd; +pub(crate) mod lnd; mod marketplace; mod mesh; mod middleware; diff --git a/core/archipelago/src/server.rs b/core/archipelago/src/server.rs index 0a956eb0..a4e57433 100644 --- a/core/archipelago/src/server.rs +++ b/core/archipelago/src/server.rs @@ -101,6 +101,12 @@ impl Server { } state_manager.update_data(data.clone()).await; + // Real-time wallet push: stream LND transaction events → websocket + // revision nudges, so an incoming on-chain tx shows in the UI within + // seconds of hitting the mempool (works whenever LND is up; retries + // forever otherwise). User req 2026-07-22. + crate::api::rpc::lnd::spawn_lnd_tx_watcher(state_manager.clone()); + // Retry Tor address in background — Tor may not be ready at startup if data.server_info.tor_address.is_none() { let sm = state_manager.clone(); diff --git a/neode-ui/src/views/Home.vue b/neode-ui/src/views/Home.vue index 45fdf8de..c41229a4 100644 --- a/neode-ui/src/views/Home.vue +++ b/neode-ui/src/views/Home.vue @@ -300,6 +300,8 @@ import TransactionsModal from '@/components/TransactionsModal.vue' import WalletSettingsModal from '@/components/WalletSettingsModal.vue' import { useAppStore } from '../stores/app' import { useAppLauncherStore } from '@/stores/appLauncher' +import { wsClient } from '@/api/websocket' +import { useTxExplorer } from '@/composables/useTxExplorer' import { useLoginTransitionStore } from '../stores/loginTransition' import { useUIModeStore } from '@/stores/uiMode' import { useHomeStatusStore } from '@/stores/homeStatus' @@ -360,6 +362,8 @@ onBeforeUnmount(() => { if (typingInterval) clearInterval(typingInterval) if (systemStatsInterval) clearInterval(systemStatsInterval) if (walletRefreshInterval) clearInterval(walletRefreshInterval) + if (unsubscribeWs) { unsubscribeWs(); unsubscribeWs = null } + if (wsWalletDebounce) clearTimeout(wsWalletDebounce) }) watch(() => loginTransition.pendingWelcomeTyping, (pending) => { if (pending) showWelcomeBlock.value = true }) @@ -524,6 +528,15 @@ onMounted(async () => { // pending on-chain receive (or a fresh instant payment) only shows up // after a manual wallet action or a remount. walletRefreshInterval = setInterval(loadWeb5Status, 30000) + // Real-time wallet push (2026-07-22): the backend streams LND transaction + // events and nudges /ws/db the moment a tx hits the mempool. Any push = + // something changed → refetch the wallet (debounced; the RPC is cheap). + // This is what makes an incoming 0-conf tx appear in seconds instead of + // waiting out the 30s poll above. + unsubscribeWs = wsClient.subscribe(() => { + if (wsWalletDebounce) clearTimeout(wsWalletDebounce) + wsWalletDebounce = setTimeout(() => { void loadWeb5Status() }, 800) + }) }) // Wallet modals @@ -535,8 +548,10 @@ const walletConnected = ref(false); const walletOnchain = ref(0); const walletLi const walletArk = ref(0) const walletTransactions = ref([]) -// Overlay the explorer above the current page — never navigate away. -function openInMempool(txHash: string) { useAppLauncherStore().openSession('mempool', { path: `/tx/${txHash}` }) } +// Overlay the local Mempool app when it's running; otherwise route through +// the external-explorer consent flow (pruned nodes can't run Mempool). +const txExplorer = useTxExplorer() +function openInMempool(txHash: string) { txExplorer.openTx(txHash) } // wallet.ecash-history's shape (see handle_wallet_ecash_history in // api/rpc/wallet.rs) — distinct from the LND-shaped WalletTransaction used @@ -600,6 +615,8 @@ const systemUptimeDisplay = computed(() => { if (homeStatus.stats.uptimeSecs === let systemStatsInterval: ReturnType | null = null let walletRefreshInterval: ReturnType | null = null +let unsubscribeWs: (() => void) | null = null +let wsWalletDebounce: ReturnType | null = null async function loadSystemStats() { await homeStatus.refresh(packages.value)