feat(data): ask our own NEAR node for view calls while it is at the tip

FastNEAR kept answering 429 after the pacer shipped: its limit is per
address, and every host behind the office IP shares it. gargravarr tracks
the pool's and the token's shards, so `pool_rpc_urls` now also serves
view calls first — pool state, order book, quotes, supply, balances — but
only while its newest block is under two minutes old and it is not
syncing, re-checked every 30 s. Behind the tip it would answer `final`
from its own final block: on 2026-10-06, 11,536 blocks behind, it gave
qtc.omft.near's supply as of three hours earlier, which would have
replaced "unreadable" with a figure that looked current and was not. Any
call it cannot answer falls back to the public endpoint.

Refs #39

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015pkwBsR54NRmZyU4QVhvfj
This commit is contained in:
2026-10-06 18:17:11 +03:00
co-authored by Claude Opus 5.5
parent 26dfdc956c
commit 1d25689e3d
4 changed files with 99 additions and 17 deletions
+3 -1
View File
@@ -150,7 +150,9 @@ rpc_urls = ["https://near.internal"]
# gargravarr tracks the shards of dclv2.ref-labs.near and qtc.omft.near (8 and
# 4), where the pool's whitelisted accounts sign their direct calls
# (blackbeard/observer#39). Plain HTTP on the hanzalova mesh, admitted to bob
# alone by a firewalld rich rule, as frankie's RPC is to the proxy.
# alone by a firewalld rich rule, as frankie's RPC is to the proxy. It also
# answers view calls first while it is at the tip (pool state, quotes, supply,
# balances), taking them off FastNEAR's rate limit.
pool_rpc_urls = ["http://gargravarr.hanzalova.internal:24403"]
# The QTC bridge to NEAR, read from this chain's side (blackbeard/observer#39).
+5
View File
@@ -277,6 +277,11 @@ pub struct IntentsConfig {
/// signed on those shards that call a pool directly — the whitelisted
/// accounts trading or moving liquidity without intents.near — are found
/// there and queued for `refdcl` like the relay's. Empty turns the scan off.
///
/// The same node also answers view calls first — pool state, the order
/// book, quotes, supply and balances — while it is at the tip, so they stop
/// depending on a public endpoint's rate limit. Behind the tip, or for a
/// contract on a shard it does not track, they go to `view_urls`.
#[serde(default)]
pub pool_rpc_urls: Vec<String>,
}
+3 -1
View File
@@ -95,7 +95,9 @@ pub fn spawn(chain: Arc<ChainRuntime>, store: Store) {
config.archive_urls.clone(),
config.view_urls.clone(),
NEAR_TIMEOUT,
) {
)
.and_then(|n| n.with_own_views(config.pool_rpc_urls.clone(), NEAR_TIMEOUT))
{
Ok(n) => n,
Err(e) => {
tracing::error!(chain = %chain.id(), error = %e, "intents tracking not started");
+88 -15
View File
@@ -119,12 +119,23 @@ pub struct NearClient {
/// Every view endpoint is one this fleet runs; when not, view calls are
/// paced through [`PUBLIC_VIEW_GATE`].
views_are_own: bool,
/// A node of ours to ask view calls first, while it is at the tip. See
/// [`NearClient::with_own_views`].
own_views: Option<RpcClient>,
/// When the own node was last judged fresh or stale, and which.
own_fresh: std::sync::Arc<tokio::sync::Mutex<Option<(tokio::time::Instant, bool)>>>,
/// Every live endpoint is one this fleet runs. See [`is_own_endpoint`].
live_is_own: bool,
/// Every archive endpoint is.
archive_is_own: bool,
}
/// How old the own view node's newest block may be for it to answer.
const OWN_VIEW_MAX_LAG: std::time::Duration = std::time::Duration::from_secs(120);
/// How often the own view node's freshness is re-checked.
const OWN_VIEW_RECHECK: std::time::Duration = std::time::Duration::from_secs(30);
/// The least time between two calls to a public view endpoint, process-wide.
///
/// FastNEAR answers about two requests a second sustained and 429s after
@@ -203,12 +214,64 @@ impl NearClient {
live_is_own: own(&live),
archive_is_own: own(&archive),
views_are_own: own(&views),
own_views: None,
own_fresh: Default::default(),
live: RpcClient::with_endpoints(live, timeout)?,
archive: RpcClient::with_endpoints(archive, timeout)?,
views: RpcClient::with_endpoints(views, timeout)?,
})
}
/// Ask `urls` — a node of ours — for view calls first, falling back to the
/// view endpoints (blackbeard/observer#39).
///
/// Only while it is at the tip: a node still catching up answers a `final`
/// view from *its* final block, hours old, with nothing in the reply to say
/// so. On 2026-10-06 gargravarr, 11,536 blocks behind, gave `qtc.omft.near`'s
/// supply as it stood three hours earlier — a figure that would have
/// replaced "unreadable" with one that looked current and was not. A call
/// it cannot answer (a contract on a shard it does not track) falls back
/// too, so nothing is answered from the wrong place.
pub fn with_own_views(
mut self,
urls: Vec<String>,
timeout: Duration,
) -> Result<Self, DataError> {
if !urls.is_empty() {
self.own_views = Some(RpcClient::with_endpoints(urls, timeout)?);
}
Ok(self)
}
/// Whether the own view node's newest block is recent: within
/// [`OWN_VIEW_MAX_LAG`] of now and not syncing. Re-checked at most every
/// [`OWN_VIEW_RECHECK`].
async fn own_views_fresh(&self, own: &RpcClient) -> bool {
let mut cached = self.own_fresh.lock().await;
if let Some((at, fresh)) = *cached {
if at.elapsed() < OWN_VIEW_RECHECK {
return fresh;
}
}
let fresh = match own.call("status", json!([])).await {
Ok(status) => {
let sync = &status["sync_info"];
let recent = sync["latest_block_time"]
.as_str()
.and_then(|t| DateTime::parse_from_rfc3339(t).ok())
.is_some_and(|t| {
(Utc::now() - t.with_timezone(&Utc))
.to_std()
.is_ok_and(|lag| lag < OWN_VIEW_MAX_LAG)
});
recent && sync["syncing"].as_bool() == Some(false)
}
Err(_) => false,
};
*cached = Some((tokio::time::Instant::now(), fresh));
fresh
}
/// Whether a block read at `height` is served by endpoints we run: the
/// live ones at or above `live_from` (the oldest block they hold, as
/// `heights` reports), the archive below it — the same split `chunk`
@@ -393,22 +456,32 @@ impl NearClient {
args: Value,
) -> Result<Value, DataError> {
use base64::Engine as _;
if !self.views_are_own {
pace_public_view().await;
let params = json!({
"request_type": "call_function",
"finality": "final",
"account_id": contract,
"method_name": method,
"args_base64": base64::engine::general_purpose::STANDARD.encode(args.to_string()),
});
let mut answered = None;
if let Some(own) = &self.own_views {
if self.own_views_fresh(own).await {
match own.call("query", params.clone()).await {
Ok(r) => answered = Some(r),
Err(e) => tracing::debug!(%contract, %method, error = %e,
"own view node could not answer; asking the public one"),
}
}
}
let result = self
.views
.call(
"query",
json!({
"request_type": "call_function",
"finality": "final",
"account_id": contract,
"method_name": method,
"args_base64": base64::engine::general_purpose::STANDARD.encode(args.to_string()),
}),
)
.await?;
let result = match answered {
Some(r) => r,
None => {
if !self.views_are_own {
pace_public_view().await;
}
self.views.call("query", params).await?
}
};
let bytes: Vec<u8> = serde_json::from_value(result["result"].clone())?;
Ok(serde_json::from_slice(&bytes)?)
}