blackbeard/observer
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ws): push each swap as it settles on NEAR Intents

Browse Source 
`ServerMessage::IntentsSwap` carries the trade exactly as /intents/swaps
returns it — one conversion, `intents::to_swap`, serves both, so the route
and the push cannot describe the same trade differently.

Only trades read within LIVE_WITHIN_BLOCKS (~a minute) of NEAR's final tip
are pushed. The backfill never broadcasts, and the threshold is what keeps an
upward catch-up after a restart from announcing an hour of history as news —
the confusion the wormhole pulse shipped once already.

The market panel re-reads its figures on each push rather than adjusting a
copy: a VWAP over the latest N trades drops its oldest as it gains one.

Verified by replaying the trade at NEAR 218,261,895 with the threshold opened
locally and catching the frame on /v1/ws.

Refs #27

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015pkwBsR54NRmZyU4QVhvfj
This commit is contained in:
grenade
2026-10-03 18:35:32 +03:00
co-authored by Claude Opus 5.5
parent 0d5ebd19ee
commit 100d508588
7 changed files with 162 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/blackbeard-api/src/intents.rs
+89
View File
@@ -51,6 +51,14 @@ const BACKFILL_PACE: Duration = Duration::from_secs(1);
/// every ~0.6 s; checking twice a block is plenty.
const TIP_WAIT: Duration = Duration::from_secs(1);
/// How close to NEAR's final tip a trade must be read for it to be pushed live.
///
/// About a minute of NEAR blocks. The backfill never broadcasts; this is what
/// keeps a catch-up after a restart — which walks *upward*, like the live path
/// — from announcing an hour of trades as though each had just happened. The
/// wormhole pulse shipped exactly that confusion once, by a different road.
const LIVE_WITHIN_BLOCKS: u64 = 100;
/// Pause after a failure before trying again.
const RETRY: Duration = Duration::from_secs(15);
@@ -170,6 +178,9 @@ async fn scan(chain: Arc<ChainRuntime>, store: Store, near: NearClient, config:
}
tracing::info!(chain = %id, near_height = height, trades = rows.len(),
"intents trades recorded");
if !descending && tip.saturating_sub(height) <= LIVE_WITHIN_BLOCKS {
announce(&chain, &store, &config.asset, rows).await;
}
}
reached = Some(height);
}
@@ -348,3 +359,81 @@ async fn listing(chain: Arc<ChainRuntime>, store: Store, near: NearClient, confi
tokio::time::sleep(LISTING_INTERVAL).await;
}
}
/// Push freshly settled trades to connected browsers.
async fn announce(chain: &Arc<ChainRuntime>, store: &Store, asset: &str, rows: Vec<IntentTrade>) {
let mut names: Vec<String> = vec![asset.to_owned()];
names.extend(rows.iter().map(|r| r.counter_token.clone()));
let tokens = token_meta(store, names).await;
let asset_decimals = tokens.get(asset).map(|t| t.1);
for row in rows {
chain.broadcast(
&blackbeard_entities::ServerMessage::IntentsSwap {
chain: chain.id(),
asset: asset.to_owned(),
swap: to_swap(row, &tokens, asset_decimals),
},
None,
);
}
}
/// Cached `(symbol, decimals)` for every token named that has them.
///
/// A token with no cached metadata is simply absent, and its trades go out
/// without a price — the same answer the routes give.
pub async fn token_meta(
store: &Store,
names: Vec<String>,
) -> std::collections::HashMap<String, (String, u8)> {
let mut out = std::collections::HashMap::new();
for name in names {
if out.contains_key(&name) {
continue;
}
match store.near_token(&name).await {
Ok(Some(meta)) => {
out.insert(name, meta);
}
Ok(None) => {}
Err(e) => tracing::warn!(token = %name, error = %e, "token metadata unreadable"),
}
}
out
}
/// A stored trade as the wire carries it. One conversion for the route and
/// the push, so the two cannot describe the same trade differently.
pub fn to_swap(
r: IntentTrade,
tokens: &std::collections::HashMap<String, (String, u8)>,
asset_decimals: Option<u8>,
) -> blackbeard_entities::IntentsSwap {
let counter = tokens.get(&r.counter_token);
let price = match (
asset_decimals,
counter,
r.asset_amount.parse::<u128>(),
r.counter_amount.parse::<u128>(),
) {
(Some(ad), Some((_, cd)), Ok(a), Ok(c)) => Some(core::price(a, ad, c, *cd)),
_ => None,
};
blackbeard_entities::IntentsSwap {
near_height: r.near_height,
trade_index: r.trade_index,
at: r.at,
tx_hash: r.tx_hash,
receipt_id: r.receipt_id,
counter_symbol: counter.map(|c| c.0.clone()),
counter_token: r.counter_token,
asset_amount: BigUintDec(r.asset_amount),
counter_amount: BigUintDec(r.counter_amount),
price,
side: r.side,
taker: r.taker,
referral: r.referral,
accounts: r.accounts,
intent_hashes: r.intent_hashes,
}
}
crates/blackbeard-api/src/routes.rs
+1 -31
View File
@@ -1809,37 +1809,7 @@ async fn intents_swaps(
.map(|r| format!("{}:{}", r.near_height, r.trade_index));
let swaps = rows
.into_iter()
.map(|r| {
let counter = tokens.get(&r.counter_token);
let price = match (
asset_decimals,
counter,
r.asset_amount.parse::<u128>(),
r.counter_amount.parse::<u128>(),
) {
(Some(ad), Some((_, cd)), Ok(a), Ok(c)) => {
Some(blackbeard_core::intents::price(a, ad, c, *cd))
}
_ => None,
};
blackbeard_entities::IntentsSwap {
near_height: r.near_height,
trade_index: r.trade_index,
at: r.at,
tx_hash: r.tx_hash,
receipt_id: r.receipt_id,
counter_symbol: counter.map(|c| c.0.clone()),
counter_token: r.counter_token,
asset_amount: BigUintDec(r.asset_amount),
counter_amount: BigUintDec(r.counter_amount),
price,
side: r.side,
taker: r.taker,
referral: r.referral,
accounts: r.accounts,
intent_hashes: r.intent_hashes,
}
})
.map(|r| crate::intents::to_swap(r, &tokens, asset_decimals))
.collect();
Ok(Json(blackbeard_entities::IntentsSwaps {
crates/blackbeard-entities/src/intents.rs
+1 -1
View File
@@ -18,7 +18,7 @@ use ts_rs::TS;
use crate::{BigUintDec, ChainId};
/// One settled trade of the asset against one counter token.
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[ts(export, export_to = "IntentsSwap.ts")]
pub struct IntentsSwap {
/// NEAR block it settled in.
crates/blackbeard-entities/src/ws.rs
+17 -1
View File
@@ -14,7 +14,9 @@
use serde::{Deserialize, Serialize};
use ts_rs::TS;
use crate::{BigUintDec, ChainId, ChainInfo, ChainSummary, LeaderboardRow, RecentBlock};
use crate::{
BigUintDec, ChainId, ChainInfo, ChainSummary, IntentsSwap, LeaderboardRow, RecentBlock,
};
/// How far back a leaderboard looks.
///
@@ -251,6 +253,20 @@ pub enum ServerMessage {
/// showing.
notes: Vec<WormholeNote>,
},
/// A swap of the chain's token just settled on NEAR Intents.
///
/// Sent only for a trade read close to NEAR's tip — never by the backfill,
/// and never by a long catch-up after a restart, both of which would
/// announce history as though it had just happened. A page that needs
/// what it missed asks `/intents/swaps`.
IntentsSwap {
/// Which chain's token.
chain: ChainId,
/// The tracked asset, as a multi-token id.
asset: String,
/// The trade, exactly as `/intents/swaps` would return it.
swap: IntentsSwap,
},
/// A chain's reachability changed — the node went away, or a chain that was
/// awaiting launch has started producing blocks.
ChainStatus {
web/src/api/generated/ServerMessage.ts
+14 -1
View File
@@ -3,6 +3,7 @@ import type { BigUintDec } from "./BigUintDec";
import type { ChainId } from "./ChainId";
import type { ChainInfo } from "./ChainInfo";
import type { ChainSummary } from "./ChainSummary";
import type { IntentsSwap } from "./IntentsSwap";
import type { LeaderboardRow } from "./LeaderboardRow";
import type { RecentBlock } from "./RecentBlock";
import type { Window } from "./Window";
@@ -113,7 +114,19 @@ amount_out: BigUintDec,
* reads. Anything rendering these has to say how many it is not
* showing.
*/
notes: Array<WormholeNote>, } | { "type": "chain_status",
notes: Array<WormholeNote>, } | { "type": "intents_swap",
/**
* Which chain's token.
*/
chain: ChainId,
/**
* The tracked asset, as a multi-token id.
*/
asset: string,
/**
* The trade, exactly as `/intents/swaps` would return it.
*/
swap: IntentsSwap, } | { "type": "chain_status",
/**
* Which chain.
*/
web/src/api/socket.ts
+27
View File
@@ -15,6 +15,7 @@
import type { ChainInfo } from './generated/ChainInfo'
import type { ChainSummary } from './generated/ChainSummary'
import type { ClientMessage } from './generated/ClientMessage'
import type { IntentsSwap } from './generated/IntentsSwap'
import type { LeaderboardRow } from './generated/LeaderboardRow'
import type { RecentBlock } from './generated/RecentBlock'
import type { ServerMessage } from './generated/ServerMessage'
@@ -127,6 +128,8 @@ export class Observer {
private listeners = new Set<() => void>()
/** Pulse subscribers, kept apart from `listeners` — see `WormholePulse`. */
private pulseListeners = new Set<(pulse: WormholePulse) => void>()
/** Swap subscribers, kept apart for the same reason: a swap is an event. */
private swapListeners = new Set<(swap: IntentsSwap) => void>()
private socket: WebSocket | null = null
private reconnectDelay = RECONNECT_MIN_MS
private reconnectTimer: number | null = null
@@ -147,6 +150,18 @@ export class Observer {
}
}
/**
* Hear every swap of the subscribed chain's token as it settles on NEAR
* Intents. Only trades read near NEAR's tip are pushed; anything older is
* a query, not news.
*/
onIntentsSwap = (listener: (swap: IntentsSwap) => void): (() => void) => {
this.swapListeners.add(listener)
return () => {
this.swapListeners.delete(listener)
}
}
/** `useSyncExternalStore` subscribe. */
subscribe = (listener: () => void): (() => void) => {
this.listeners.add(listener)
@@ -305,6 +320,18 @@ export class Observer {
break
}
case 'intents_swap': {
if (message.chain !== this.state.chain) return
for (const listener of this.swapListeners) {
try {
listener(message.swap)
} catch (e) {
console.warn('[observer] intents swap subscriber threw', e)
}
}
break
}
case 'chain_status':
this.patch({
chains: this.state.chains.map((c) => (c.id === message.chain ? message.info : c)),
web/src/components/IntentsPanel.tsx
+13 -2
View File
@@ -22,6 +22,7 @@ import { useEffect, useState } from 'react'
import type { IntentsPrice } from '../api/generated/IntentsPrice'
import { RequestFailed, fetchIntentsPrice } from '../api/rest'
import { ago, height as fmtHeight, price as fmtPrice, tokens } from '../lib/format'
import { useObserverInstance } from '../lib/observer-context'
import { Field } from './Field'
/** A multi-token id short enough for a label: `nep141:zec.omft.near` → `zec.omft.near`. */
@@ -42,10 +43,17 @@ export function IntentsPanel({
const [data, setData] = useState<IntentsPrice | null>(null)
const [absent, setAbsent] = useState(false)
const [error, setError] = useState<string | null>(null)
// Bumped by every pushed swap. The figures are re-read rather than adjusted
// here: a VWAP over the latest N trades loses its oldest trade as it gains a
// new one, and arithmetic on a copy drifts from the server's answer while
// looking authoritative — the reason the wormhole stage re-reads its stock.
const [settled, setSettled] = useState(0)
const observer = useObserverInstance()
useEffect(() => observer.onIntentsSwap(() => setSettled((n) => n + 1)), [observer])
useEffect(() => {
const controller = new AbortController()
setData(null)
setAbsent(false)
setError(null)
fetchIntentsPrice(chain, controller.signal)
@@ -56,7 +64,10 @@ export function IntentsPanel({
else setError(e instanceof RequestFailed ? e.message : 'Could not reach the observer.')
})
return () => controller.abort()
}, [chain])
}, [chain, settled])
// A different chain's figures must never stand in for this one's while it loads.
useEffect(() => setData(null), [chain])
if (absent) return null