feat(wallet): real-time tx push — 0-conf shows in seconds, not on poll
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 <noreply@anthropic.com>
This commit is contained in:
parent
088b3e255a
commit
5e7e928650
@ -56,6 +56,70 @@ pub(crate) async fn read_lnd_admin_macaroon() -> Result<Vec<u8>> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// 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<crate::state::StateManager>) {
|
||||||
|
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 {
|
impl RpcHandler {
|
||||||
/// Helper: create an authenticated LND REST client.
|
/// Helper: create an authenticated LND REST client.
|
||||||
/// Returns an HTTP client configured for LND's self-signed TLS and the
|
/// Returns an HTTP client configured for LND's self-signed TLS and the
|
||||||
|
|||||||
@ -15,7 +15,7 @@ mod fips;
|
|||||||
mod handshake;
|
mod handshake;
|
||||||
mod identity;
|
mod identity;
|
||||||
mod interfaces;
|
mod interfaces;
|
||||||
pub(in crate::api) mod lnd;
|
pub(crate) mod lnd;
|
||||||
mod marketplace;
|
mod marketplace;
|
||||||
mod mesh;
|
mod mesh;
|
||||||
mod middleware;
|
mod middleware;
|
||||||
|
|||||||
@ -101,6 +101,12 @@ impl Server {
|
|||||||
}
|
}
|
||||||
state_manager.update_data(data.clone()).await;
|
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
|
// Retry Tor address in background — Tor may not be ready at startup
|
||||||
if data.server_info.tor_address.is_none() {
|
if data.server_info.tor_address.is_none() {
|
||||||
let sm = state_manager.clone();
|
let sm = state_manager.clone();
|
||||||
|
|||||||
@ -300,6 +300,8 @@ import TransactionsModal from '@/components/TransactionsModal.vue'
|
|||||||
import WalletSettingsModal from '@/components/WalletSettingsModal.vue'
|
import WalletSettingsModal from '@/components/WalletSettingsModal.vue'
|
||||||
import { useAppStore } from '../stores/app'
|
import { useAppStore } from '../stores/app'
|
||||||
import { useAppLauncherStore } from '@/stores/appLauncher'
|
import { useAppLauncherStore } from '@/stores/appLauncher'
|
||||||
|
import { wsClient } from '@/api/websocket'
|
||||||
|
import { useTxExplorer } from '@/composables/useTxExplorer'
|
||||||
import { useLoginTransitionStore } from '../stores/loginTransition'
|
import { useLoginTransitionStore } from '../stores/loginTransition'
|
||||||
import { useUIModeStore } from '@/stores/uiMode'
|
import { useUIModeStore } from '@/stores/uiMode'
|
||||||
import { useHomeStatusStore } from '@/stores/homeStatus'
|
import { useHomeStatusStore } from '@/stores/homeStatus'
|
||||||
@ -360,6 +362,8 @@ onBeforeUnmount(() => {
|
|||||||
if (typingInterval) clearInterval(typingInterval)
|
if (typingInterval) clearInterval(typingInterval)
|
||||||
if (systemStatsInterval) clearInterval(systemStatsInterval)
|
if (systemStatsInterval) clearInterval(systemStatsInterval)
|
||||||
if (walletRefreshInterval) clearInterval(walletRefreshInterval)
|
if (walletRefreshInterval) clearInterval(walletRefreshInterval)
|
||||||
|
if (unsubscribeWs) { unsubscribeWs(); unsubscribeWs = null }
|
||||||
|
if (wsWalletDebounce) clearTimeout(wsWalletDebounce)
|
||||||
})
|
})
|
||||||
|
|
||||||
watch(() => loginTransition.pendingWelcomeTyping, (pending) => { if (pending) showWelcomeBlock.value = true })
|
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
|
// pending on-chain receive (or a fresh instant payment) only shows up
|
||||||
// after a manual wallet action or a remount.
|
// after a manual wallet action or a remount.
|
||||||
walletRefreshInterval = setInterval(loadWeb5Status, 30000)
|
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
|
// Wallet modals
|
||||||
@ -535,8 +548,10 @@ const walletConnected = ref(false); const walletOnchain = ref(0); const walletLi
|
|||||||
const walletArk = ref(0)
|
const walletArk = ref(0)
|
||||||
const walletTransactions = ref<WalletTransaction[]>([])
|
const walletTransactions = ref<WalletTransaction[]>([])
|
||||||
|
|
||||||
// Overlay the explorer above the current page — never navigate away.
|
// Overlay the local Mempool app when it's running; otherwise route through
|
||||||
function openInMempool(txHash: string) { useAppLauncherStore().openSession('mempool', { path: `/tx/${txHash}` }) }
|
// 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
|
// wallet.ecash-history's shape (see handle_wallet_ecash_history in
|
||||||
// api/rpc/wallet.rs) — distinct from the LND-shaped WalletTransaction used
|
// 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<typeof setInterval> | null = null
|
let systemStatsInterval: ReturnType<typeof setInterval> | null = null
|
||||||
let walletRefreshInterval: ReturnType<typeof setInterval> | null = null
|
let walletRefreshInterval: ReturnType<typeof setInterval> | null = null
|
||||||
|
let unsubscribeWs: (() => void) | null = null
|
||||||
|
let wsWalletDebounce: ReturnType<typeof setTimeout> | null = null
|
||||||
|
|
||||||
async function loadSystemStats() {
|
async function loadSystemStats() {
|
||||||
await homeStatus.refresh(packages.value)
|
await homeStatus.refresh(packages.value)
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user