From 93d7a4f3d9418f4cead8724ac88a07f8d5eb0f51 Mon Sep 17 00:00:00 2001 From: karczuRF Date: Mon, 10 Aug 2026 14:36:35 +0200 Subject: [PATCH 1/5] feat(lore-0167): add prices.usd_rate and snapshot the peg rates from oracle_prices close_usd is not a stored fact, it is a cached product: every enrichment tier computes close * , and the rate is a function of (quote asset, timestamp) only. Today we look it up, multiply it into hundreds of millions of rows, and discard it. This writes it down. The urgency is retention. oracle_prices is pruned at INTERVAL 13 MONTH, so the earliest depeg-aware readings age out permanently around 2026-10/11. A view cannot avoid this by joining oracle_prices directly - the published series would MUTATE as rows age out, a bucket reading 0.9993 silently reverting to a $1 fallback - which is why views.sql forbids that join. Hence a forever-retained snapshot. Keyed on natural identity, never asset_id: 0139 is confirmed genuine collisions, measured today at 3,281 ids serving 6,568 identities. The table is deliberately absent from cleanup-worker's opt-in RETENTION list, with a tripwire test - the protection is an ABSENCE, exactly the invariant a later reader breaks with one tidy line. populate_usd_rate_from_oracle copies peg observations incrementally by per-identity watermark; oracle-worker calls it after write_oracle. The call is NON-FATAL by design: oracle_prices is the source of truth and is already written, the copy is derived and self-heals on the next run, and failing here would stop oracle polling itself - trading a durable gap for a live outage. The 0139 guard is load-bearing, not decoration. oracle_prices is asset_id-keyed and usd_rate is identity-keyed, so this copy is the one place the two key spaces meet; an unchecked translation would file one asset's readings under another's identity in a table meant to be trusted forever. Refused in both directions, and the test asserts the refusal writes nothing. XLM is polled but deliberately not snapshotted - 0154 owns the pivot methods - recorded as an explicit scope boundary WITH the counter-argument that the 13-month clock applies to XLM's history identically. Tests: 2 CH ITs (copy/watermark/no-duplicate-re-run, and the 0139 refusal), a shape IT asserting the sorting key excludes asset_id, and the retention tripwire. Workspace lib suite green on the 26.3.10.60 pin; no new clippy warnings. Statement-count guard in lib.rs updated 29 -> 30. --- ..._usd-rate-table-and-peg-rate-population.md | 107 ++++++++++- packages/cleanup-worker/src/lib.rs | 37 ++++ packages/oracle-worker/src/lib.rs | 63 ++++++- packages/prices-clickhouse/schema/init.sql | 88 +++++++++ packages/prices-clickhouse/src/lib.rs | 10 +- .../prices-clickhouse/tests/usd_rate_it.rs | 105 +++++++++++ packages/prices-ingest-core/src/error.rs | 7 + packages/prices-ingest-core/src/writer.rs | 163 ++++++++++++++++ .../tests/usd_rate_population_it.rs | 178 ++++++++++++++++++ 9 files changed, 743 insertions(+), 15 deletions(-) create mode 100644 packages/prices-clickhouse/tests/usd_rate_it.rs create mode 100644 packages/prices-ingest-core/tests/usd_rate_population_it.rs diff --git a/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md b/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md index 5961ad0c..7b81dd1f 100644 --- a/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md +++ b/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md @@ -50,6 +50,40 @@ history: and the key is NATURAL IDENTITY not quote_asset_id - 0139 is confirmed genuine collisions, measured 2026-08-10 at 3,281 asset_ids across 6,568 identities, which a quote_asset_id-keyed table would have inherited. + - date: 2026-08-10 + status: active + who: okarcz + note: > + IMPLEMENTED (schema + population). prices.usd_rate added to init.sql with + 0154's exact shape; populate_usd_rate_from_oracle on OhlcvWriter copies + peg observations from oracle_prices, incremental by per-identity + watermark; oracle-worker calls it after write_oracle. + Design decision worth flagging - the snapshot is NON-FATAL in the worker. + oracle_prices is the source of truth and is already written by that point; + the copy is derived and watermarked, so a failed pass self-heals on the + next run. Failing the worker would stop oracle POLLING itself, trading a + durable gap for a live outage. + The 0139 guard is the load-bearing piece: oracle_prices is keyed on + asset_id and usd_rate on natural identity, so this copy is the ONE place + the two key spaces meet. With 3,281 ids serving 6,568 identities on prod, + an unchecked translation would file one asset's readings under another's + identity in a table meant to be trusted forever. The write is refused in + both directions (identity -> exactly one id; that id -> exactly one + identity) and the IT asserts the refusal writes NOTHING. + XLM is polled but deliberately NOT snapshotted - 0154 owns the pivot + methods. Recorded as an explicit scope boundary on peg_identities(), + WITH the counter-argument: the 13-month expiry applies to XLM's history + identically, so if 0154 has not started before 202509 ages out this should + be reconsidered rather than deferred by default. + Tests: 2 new CH ITs on the 26.3.10.60 pin (copy + watermark + no-duplicate + re-run; and the 0139 refusal), 1 shape IT, 1 retention tripwire in + cleanup-worker. Whole workspace lib suite green; no new clippy warnings. + Two self-inflicted bugs found and fixed while building, both recorded + because they were invisible in review: watermark_before was Default 0 with + a .min() that pinned it there forever (now Option), and the two ITs + shared the real `prices` database and truncated each other's fixtures + under cargo's parallel runner - both failed in ways that looked like + product bugs until serialised. --- # `prices.usd_rate` + peg-asset rate population @@ -189,17 +223,72 @@ either. Record it rather than mitigate it. ## Acceptance Criteria -- [ ] `prices.usd_rate` exists with 0154's exact shape, keyed on natural identity. -- [ ] Absent from `cleanup-worker`'s `RETENTION` list, with a comment stating that - the omission is deliberate. -- [ ] Peg-asset observations populated from `oracle_prices`, incremental, with +- [x] `prices.usd_rate` exists with 0154's exact shape, keyed on natural identity. + Asserted in `usd_rate_it.rs` on the exact column list **and order**, the + `ReplacingMergeTree(version)` engine, and — the load-bearing one — that the + sorting key does **not** contain `asset_id`. +- [x] Absent from `cleanup-worker`'s `RETENTION` list, with a comment stating that + the omission is deliberate. Plus a **tripwire test** in `cleanup-worker` + itself: the protection is an *absence*, which is exactly the invariant a + later reader breaks with one tidy-looking line. +- [x] Peg-asset observations populated from `oracle_prices`, incremental, with `method`/`hops` set; re-runnable without duplication (RMT `version`). -- [ ] The ASOF + staleness resolution rule implemented and documented in the - table's schema comment, not only in this task. -- [ ] Pre-oracle buckets have **no** row — verified, not assumed. + `populate_usd_rate_from_oracle`, watermarked per identity; the IT copies, + re-runs (no duplication), then appends only a newly-arrived reading. +- [~] The ASOF + staleness resolution rule implemented and documented in the + table's schema comment, not only in this task. **Documented in full at the + table** (`init.sql`), including why it is not an average and the + composes-across-grains argument. *Implementation* is necessarily a + **consumer** concern — nothing reads `usd_rate` yet, and [[0168]] is its + first reader. Not claimed as done here. +- [~] Pre-oracle buckets have **no** row — verified, not assumed. True **by + construction**: the population only ever copies rows that exist in + `oracle_prices`, so a bucket with no oracle reading gets no row, and there + is no synthetic-`peg`-row path to write one. Not yet asserted against real + pre-2025-09 data — that check belongs with the prod backfill below. - [ ] Reproduces today's `close_usd` for the oracle and peg tiers on a sample - window (0154 constraint 5), on CH **26.3.10.60**. -- [ ] `0154` and `0151` updated to record that the table moved here. + window (0154 constraint 5), on CH **26.3.10.60**. **Requires prod data — + operator-run, see §Prod backfill.** This is the gate before anything + *reads* the table for pricing. +- [x] `0154` and `0151` updated to record that the table moved here. **Already + done in PR #182** when 0167 was authored — re-checked 2026-08-10, both + reference 0167; no further edit needed. + +## Prod backfill — operator-run, not automatic + +The worker only snapshots **forward** from its watermark, so on first run against +prod it copies the whole surviving `oracle_prices` window in one pass — which is +the point, given the 13-month clock. Nothing is deployed by merging this; the +population runs when `oracle-worker` next executes against prod, or on demand. + +⏳ **Do this before ~2026-10**, or `202509` is gone. + +**Gate before any consumer trusts it** (0154 constraint 5 — reproduce today's +`close_usd` from the stored rate): + +```sql +-- Every USDC-quoted candle should satisfy close_usd ~= close * rate-at-or-before. +SELECT round(100 * countIf(abs(toFloat64(p.close_usd) - toFloat64(p.close) * toFloat64(r.usd_rate)) > 1e-6) + / count(), 4) AS pct_mismatch, + count() AS sampled +FROM prices.price_ohlcv_1d AS p FINAL +INNER JOIN prices.assets AS q FINAL ON q.asset_id = p.quote_asset_id +ASOF LEFT JOIN prices.usd_rate AS r + ON r.asset_code = q.asset_code AND r.issuer_address = q.issuer_address + AND r.timestamp <= p.timestamp +WHERE q.asset_code = 'USDC' AND p.close_usd > 0 + AND p.timestamp >= now() - INTERVAL 30 DAY; +``` + +⚠️ Expect a **small non-zero** mismatch, not exactly 0: enrichment baked +`close_usd` at the rate current *then*, and the peg tier used a flat `$1` where +the oracle tier did not. A large mismatch means the rate table disagrees with +the tier that produced the candle — a bug either way, which is the point of the +check. + +⚠️ **Do not gate on USDT until [[0172]] is understood.** Its candles close at +~0.14 against USDC on prod, so any reconciliation through USDT will fail for a +reason that has nothing to do with this table. ## Out of scope diff --git a/packages/cleanup-worker/src/lib.rs b/packages/cleanup-worker/src/lib.rs index d6a07cd5..acd4584a 100644 --- a/packages/cleanup-worker/src/lib.rs +++ b/packages/cleanup-worker/src/lib.rs @@ -18,6 +18,15 @@ use serde::Deserialize; /// `(table, retention interval SQL)` per §3.6. Only these tables are pruned; /// `price_ohlcv_{1h,4h,1d,1w,1M}` are retained forever. +/// +/// ⚠️ **`prices.usd_rate` is deliberately ABSENT and must stay that way** +/// (task 0167). This list is opt-in, so an unlisted table is retained forever — +/// which is the entire point of that table. It exists precisely *because* +/// `oracle_prices` expires at 13 months and takes the earliest depeg-aware +/// history with it; `usd_rate` is the forever-retained snapshot that escapes +/// that. Adding it here would silently re-create the problem it was built to +/// solve, and the loss would be unrecoverable rather than merely wrong. +/// See the block comment above `CREATE TABLE prices.usd_rate` in `init.sql`. pub const RETENTION: &[(&str, &str)] = &[ ("price_ohlcv_1m", "INTERVAL 7 DAY"), ("price_ohlcv_15m", "INTERVAL 30 DAY"), @@ -69,3 +78,31 @@ pub async fn run_cleanup(client: &Client) -> Result Ok(stats) } + +#[cfg(test)] +mod usd_rate_retention_tests { + use super::RETENTION; + + /// Task 0167. `prices.usd_rate` exists precisely BECAUSE `oracle_prices` is + /// pruned at 13 months and takes the earliest depeg-aware history with it — + /// unrecoverably, since the readings cannot be re-derived after the fact. + /// + /// `RETENTION` is an opt-in allowlist, so the protection is an ABSENCE, and + /// an absence is exactly the invariant a future reader breaks by adding one + /// tidy-looking line. This test is that line's tripwire. + #[test] + fn usd_rate_is_never_pruned() { + let listed: Vec<&str> = RETENTION.iter().map(|(t, _)| *t).collect(); + assert!( + !listed.contains(&"usd_rate"), + "usd_rate must never be pruned — it is the forever-retained snapshot \ + of oracle_prices, which IS pruned. Adding it here re-creates the \ + exact unrecoverable data loss the table was built to escape. \ + Listed: {listed:?}" + ); + assert!( + listed.contains(&"oracle_prices"), + "sanity: oracle_prices must still be pruned, or this test proves nothing" + ); + } +} diff --git a/packages/oracle-worker/src/lib.rs b/packages/oracle-worker/src/lib.rs index 87a98f99..bac4fa88 100644 --- a/packages/oracle-worker/src/lib.rs +++ b/packages/oracle-worker/src/lib.rs @@ -12,7 +12,9 @@ //! `oracle_prices.price_usd Decimal(38,14)`. use base64::Engine; -use prices_ingest_core::{AssetRegistry, OhlcvWriter, OracleSample, reflector_key_to_identity}; +use prices_ingest_core::{ + AssetIdentity, AssetRegistry, OhlcvWriter, OracleSample, reflector_key_to_identity, +}; use stellar_xdr::{ ContractId, Hash, HostFunction, Int128Parts, InvokeContractArgs, InvokeHostFunctionOp, Limits, Memo, MuxedAccount, Operation, OperationBody, Preconditions, ReadXdr, ScAddress, ScMap, @@ -32,6 +34,37 @@ pub const DEFAULT_SOROBAN_RPC: &str = "https://mainnet.sorobanrpc.com"; /// can grow independently of the mapping. pub const TRACKED_SYMBOLS: &[&str] = &["XLM", "USDC", "USDT"]; +/// The identities whose oracle readings are snapshotted into `prices.usd_rate` +/// (task 0167). Deliberately a **subset** of [`TRACKED_SYMBOLS`]. +/// +/// ⚠️ **XLM is polled but NOT snapshotted here, and that is a scope boundary, +/// not an oversight.** XLM is the reference asset the *pivot* tier prices +/// everything else through, and task 0154 owns the `'pivot'` / `'pivot2'` +/// methods and the transitivity rules that go with them. Writing XLM rows here +/// would pre-empt those decisions in a table 0154 then has to live with. +/// +/// ⏳ **But the 13-month expiry argument applies to XLM's history identically**, +/// and that argument is the whole reason 0167 was pulled forward. If 0154 has +/// not started before `202509` ages out (~2026-10/11), snapshotting XLM as +/// `method = 'oracle'`, `hops = 0` — which is what it factually is, no pivot +/// involved — should be reconsidered on its own merits rather than deferred by +/// default. Raised explicitly so the omission is a decision, not an accident. +pub fn peg_identities() -> Vec { + // Built rather than declared const: AssetIdentity::Credit holds Strings. + // Sourced from the same consts the enrichment peg tier and views.sql use, + // so the three cannot drift apart. + vec![ + AssetIdentity::Credit { + code: "USDC".to_string(), + issuer: prices_clickhouse::USDC_ISSUER.to_string(), + }, + AssetIdentity::Credit { + code: "USDT".to_string(), + issuer: prices_clickhouse::USDT_ISSUER.to_string(), + }, + ] +} + #[derive(Debug, thiserror::Error)] pub enum OracleError { #[error(transparent)] @@ -202,6 +235,9 @@ pub struct OracleStats { pub queried: usize, pub written: usize, pub skipped: usize, + /// Peg identities whose rates were snapshotted into `prices.usd_rate` + /// (task 0167). Zero is not an error — see [`run_oracle`]. + pub rates_snapshotted: usize, } /// Poll Reflector for each tracked symbol and write the prices to @@ -261,10 +297,35 @@ pub async fn run_oracle( let written = samples.len(); writer.write_oracle(&samples).await?; + + // Snapshot the peg rates into the forever-retained prices.usd_rate + // (task 0167). This runs AFTER write_oracle so the rows just polled are + // included in the same pass rather than waiting for the next one. + // + // Deliberately NON-FATAL. oracle_prices is the source of truth and has + // already been written by this point; the snapshot is a derived copy that + // is incremental by watermark, so a failed pass costs nothing but is + // retried in full on the next run. Failing the whole worker here would stop + // oracle polling itself — trading a durable, self-healing gap for a live + // outage. The 0139 guard inside the copy is the most likely reason to land + // here, and it is a data condition an operator must resolve, not something + // a retry fixes. + let rates_snapshotted = match writer + .populate_usd_rate_from_oracle(&peg_identities()) + .await + { + Ok(stats) => stats.identities, + Err(err) => { + tracing::error!(error = %err, "usd_rate snapshot failed; oracle_prices is unaffected"); + 0 + } + }; + Ok(OracleStats { queried: TRACKED_SYMBOLS.len(), written, skipped, + rates_snapshotted, }) } diff --git a/packages/prices-clickhouse/schema/init.sql b/packages/prices-clickhouse/schema/init.sql index 45cbafe0..1efddabb 100644 --- a/packages/prices-clickhouse/schema/init.sql +++ b/packages/prices-clickhouse/schema/init.sql @@ -209,6 +209,94 @@ PARTITION BY toYYYYMM(timestamp) ORDER BY (asset_id, oracle_name, timestamp) SETTINGS index_granularity = 8192; +---------------------------------------------------------------------- +-- prices.usd_rate — the USD rate of an asset, as a first-class value. +-- Task 0167; shape specified by 0154, scoped by the 0151 decision. +-- +-- ## Why this table exists +-- close_usd is not a stored fact, it is a CACHED PRODUCT. All three enrichment +-- tiers compute the same shape (ch_enrich.rs): +-- close_usd = close * +-- The rate is a function of (quote asset, timestamp) ONLY — never of the candle +-- being priced. So today we look the rate up, multiply it into hundreds of +-- millions of rows, and DISCARD it. This table writes it down instead: a handful +-- of assets per bucket rather than one product per candle. +-- +-- ⏳ The urgent reason is retention. oracle_prices is pruned at INTERVAL 13 +-- MONTH (cleanup-worker/src/lib.rs), so the earliest depeg-aware history ages +-- out permanently. A view CANNOT solve this by joining oracle_prices directly: +-- the published series would MUTATE as rows age out (a bucket reading 0.9993 +-- silently reverting to 1.0000), which is why views.sql forbids that join. The +-- rate must be snapshotted into a forever-retained table. +-- +-- ## ⚠️ Key is NATURAL IDENTITY, never asset_id +-- Task 0139 is confirmed as genuine asset_id collisions between unrelated +-- assets — measured 2026-08-10 at 3,281 asset_ids serving 6,568 identities +-- (asset_id 4194 is both STW and ARBRIDGE). A rate keyed on asset_id would be +-- ambiguous for exactly those ids and would bake a non-unique key into new +-- infrastructure. Natural identity sidesteps 0139 whichever way its fix lands. +-- +-- ## ⚠️ Deliberately ABSENT from cleanup-worker's RETENTION list +-- That list is OPT-IN: a table not named there is retained forever, which is +-- exactly what this table needs. Do NOT "helpfully" add it — adding it would +-- re-create the very expiry problem the table exists to escape. +-- +-- ## Resolution rule — ASOF at-or-before, bounded by staleness. NEVER averaged. +-- Rows are OBSERVATIONS at the source's own cadence, not bucket aggregates. A +-- consumer needing the rate at time T takes the newest row with timestamp <= T, +-- and refuses it past a staleness window (unbounded forward-fill would present a +-- three-day-old reading as current). For a bucket-grained consumer such as +-- price_usd_series, T is the BUCKET'S END — i.e. the bucket's closing rate. +-- +-- Three reasons this is not an average or a vwap: +-- 1. It is the rule the codebase already uses — the oracle tier is an +-- ASOF LEFT JOIN ... ON o.timestamp <= p.timestamp with a staleness floor +-- (ch_enrich.rs), and the XLM pivot forward-fills the same way. +-- 2. It COMPOSES ACROSS ALL SIX GRANULARITIES FOR FREE. A daily close is the +-- ASOF at day-end, which IS the last hourly close. Averages do not compose +-- (the mean of hourly means is not the daily mean unless counts match), so +-- an averaging rule would need six definitions plus a consistency proof. +-- 3. price_usd_series means "one USD CLOSE per bucket" — a bucket average +-- would be a different statistic wearing the same column name. +-- vwap is impossible regardless: oracle observations carry no volume. +-- +-- Accepted cost: a close is more exposed to a single outlier reading at a bucket +-- boundary than an average is. Negligible for peg assets (~0.1% band), and task +-- 0154 ASOFs at the CANDLE's timestamp so the boundary case does not arise there. +-- +-- ## Columns +-- method 'oracle' — a measured reading (hops = 0) +-- 'peg' — the $1 assumption (hops = 0) +-- 'pivot' — via XLM (hops = 1) } owned by 0154, +-- 'pivot2' — via another rated asset (2) } not written here +-- ⚠️ ABSENCE IS THE SIGNAL for pre-oracle history. Deep history (before the +-- oracle window, ~2025-09) gets NO ROW, and the consumer's own peg fallback +-- applies. Do NOT write synthetic method='peg' rows at $1 to "fill" it — that +-- makes a fallback indistinguishable from a measurement, which is precisely +-- the close_usd = 0 mistake (one value meaning several things) in a new place. +-- +-- version — the write time, so a later correction of the same +-- (identity, timestamp) wins. Re-running the population is therefore +-- idempotent: identical content, and RMT collapses on the ORDER BY key. +---------------------------------------------------------------------- + +CREATE TABLE IF NOT EXISTS prices.usd_rate ( + asset_kind LowCardinality(String), + asset_code String, + issuer_address String, + contract_address String, + timestamp DateTime CODEC(DoubleDelta), + usd_rate Decimal(38, 14), + method LowCardinality(String), + reference_asset String DEFAULT '', + hops UInt8 DEFAULT 0, + version UInt64 +) +ENGINE = ReplacingMergeTree(version) +PARTITION BY toYYYYMM(timestamp) +ORDER BY (asset_kind, asset_code, issuer_address, contract_address, timestamp) +SETTINGS index_granularity = 8192; + ---------------------------------------------------------------------- -- Backfill bookkeeping. ---------------------------------------------------------------------- diff --git a/packages/prices-clickhouse/src/lib.rs b/packages/prices-clickhouse/src/lib.rs index 09e4366f..33b5c5de 100644 --- a/packages/prices-clickhouse/src/lib.rs +++ b/packages/prices-clickhouse/src/lib.rs @@ -212,18 +212,18 @@ mod tests { #[test] fn init_sql_parses_into_statements() { - // 1 CREATE DATABASE + 18 CREATE TABLE (assets, asset_metadata, _1m, + // 1 CREATE DATABASE + 19 CREATE TABLE (assets, asset_metadata, _1m, // _15m, _1h, _4h, _1d, _1w, _1M, current_prices, asset_supply, - // oracle_prices, backfill_sdex_ledgers, backfill_progress, + // oracle_prices, usd_rate, backfill_sdex_ledgers, backfill_progress, // discovery_state, unresolved_pools, pool_registry, ingest_cursor) + 7 // close_usd ALTERs (one per OHLCV grain) + 1 assets.sac_address ALTER // (task 0061) + 2 backfill_progress ALTERs (earliest_data_available - // [0073 half → 0053] + newest_data_available [0053]) = 29 statements. + // [0073 half → 0053] + newest_data_available [0053]) = 30 statements. // (+discovery_state task 0054, +asset_supply task 0039, +unresolved_pools // + pool_registry task 0053, +asset_metadata task 0067, +ingest_cursor - // task 0064.) + // task 0064, +usd_rate task 0167.) let stmts = split_statements(INIT_SQL); - assert_eq!(stmts.len(), 29, "got {}", stmts.len()); + assert_eq!(stmts.len(), 30, "got {}", stmts.len()); } #[test] diff --git a/packages/prices-clickhouse/tests/usd_rate_it.rs b/packages/prices-clickhouse/tests/usd_rate_it.rs new file mode 100644 index 00000000..65ecb8b2 --- /dev/null +++ b/packages/prices-clickhouse/tests/usd_rate_it.rs @@ -0,0 +1,105 @@ +//! Live-ClickHouse integration test for `prices.usd_rate` (task 0167). +//! +//! docker compose up -d clickhouse +//! cargo test -p prices-clickhouse --test usd_rate_it -- --ignored + +use clickhouse::Client; + +fn ch_url() -> String { + std::env::var("CLICKHOUSE_URL").unwrap_or_else(|_| "http://localhost:8123".to_string()) +} + +fn rewrite(sql: &str, db: &str) -> String { + sql.replace("prices.", &format!("{db}.")) + .replace("IF NOT EXISTS prices", &format!("IF NOT EXISTS {db}")) +} + +async fn setup(db: &str) -> Client { + let client = Client::default().with_url(ch_url()); + client + .query(&format!("DROP DATABASE IF EXISTS {db}")) + .execute() + .await + .unwrap(); + client + .query(&format!("CREATE DATABASE {db}")) + .execute() + .await + .unwrap(); + prices_clickhouse::apply_sql(&client, &rewrite(prices_clickhouse::INIT_SQL, db)) + .await + .unwrap(); + client +} + +#[tokio::test] +#[ignore = "requires a local ClickHouse (cargo test -- --ignored)"] +async fn usd_rate_has_the_0154_shape_keyed_on_natural_identity() { + let db = "it_usd_rate_shape"; + let client = setup(db).await; + + let cols: Vec<(String, String)> = client + .query( + "SELECT name, type FROM system.columns \ + WHERE database = ? AND table = 'usd_rate' ORDER BY position", + ) + .bind(db) + .fetch_all::<(String, String)>() + .await + .unwrap(); + let names: Vec<&str> = cols.iter().map(|(n, _)| n.as_str()).collect(); + assert_eq!( + names, + vec![ + "asset_kind", + "asset_code", + "issuer_address", + "contract_address", + "timestamp", + "usd_rate", + "method", + "reference_asset", + "hops", + "version", + ], + "0154's exact column set, in order" + ); + + // ⚠️ The key must be natural identity, NOT asset_id — task 0139 is confirmed + // asset_id collisions, so an asset_id key would be non-unique by construction. + let (engine, sorting): (String, String) = client + .query( + "SELECT engine_full, sorting_key FROM system.tables \ + WHERE database = ? AND name = 'usd_rate'", + ) + .bind(db) + .fetch_one::<(String, String)>() + .await + .unwrap(); + assert!( + engine.contains("ReplacingMergeTree(version)"), + "must dedupe on version, got {engine}" + ); + assert!( + !sorting.contains("asset_id"), + "usd_rate must NOT be keyed on asset_id (task 0139), got {sorting}" + ); + for col in [ + "asset_kind", + "asset_code", + "issuer_address", + "contract_address", + "timestamp", + ] { + assert!( + sorting.contains(col), + "sorting key missing {col}: {sorting}" + ); + } + + client + .query(&format!("DROP DATABASE {db}")) + .execute() + .await + .unwrap(); +} diff --git a/packages/prices-ingest-core/src/error.rs b/packages/prices-ingest-core/src/error.rs index 8a177c14..5b6dfb0f 100644 --- a/packages/prices-ingest-core/src/error.rs +++ b/packages/prices-ingest-core/src/error.rs @@ -19,4 +19,11 @@ pub enum IngestError { // consumer of the writer (live Lambda + SDEX backfill) is leak-safe. #[error("clickhouse: {}", crate::safe_log::redact_clickhouse(.0))] Clickhouse(#[from] clickhouse::error::Error), + + /// A data precondition made a write unsafe, so it was refused rather than + /// performed. Distinct from a ClickHouse failure: nothing went wrong at the + /// transport level, we declined to write. Carries only identifiers and + /// counts, never row values, so it stays leak-safe like the variant above. + #[error("precondition failed: {0}")] + Precondition(String), } diff --git a/packages/prices-ingest-core/src/writer.rs b/packages/prices-ingest-core/src/writer.rs index 66387e1d..60390450 100644 --- a/packages/prices-ingest-core/src/writer.rs +++ b/packages/prices-ingest-core/src/writer.rs @@ -353,6 +353,169 @@ impl OhlcvWriter { insert.end().await?; Ok(()) } + + /// Snapshot peg-asset USD rates from `oracle_prices` into `prices.usd_rate` + /// (task 0167). + /// + /// **Why this copy exists at all.** `oracle_prices` is pruned at + /// `INTERVAL 13 MONTH`, so the earliest depeg-aware readings age out + /// permanently. A consumer cannot simply join `oracle_prices` instead: the + /// published series would *mutate* as rows age out — a bucket reading + /// `0.9993` silently reverting to a `$1` fallback later — which is why + /// `views.sql` forbids that join. So the rate is snapshotted into a + /// forever-retained table, the same way every other USD number is baked + /// rather than joined. + /// + /// **Incremental by watermark, per identity.** Only observations newer than + /// the newest already stored for that identity are copied, so a re-run is + /// cheap. It is also idempotent: `usd_rate` is a `ReplacingMergeTree` keyed + /// on `(natural identity, timestamp)`, and `version` is the write time, so a + /// repeat writes identical content and a genuine later correction of the + /// same timestamp wins. + /// + /// ⚠️ **The 0139 guard is load-bearing, not defensive decoration.** + /// `oracle_prices` is keyed on `asset_id`, `usd_rate` is keyed on natural + /// identity, so this is the one place the two key spaces meet — and task + /// 0139 is confirmed as genuine `asset_id` collisions between unrelated + /// assets (measured 2026-08-10: 3,281 ids serving 6,568 identities; + /// `asset_id 4194` is both `STW` and `ARBRIDGE`). Translating without + /// checking would file one asset's oracle readings under another asset's + /// identity — silently, and in a table built to be trusted forever. If the + /// mapping is not 1:1 in **both** directions we refuse the write and say so. + pub async fn populate_usd_rate_from_oracle( + &self, + pegs: &[AssetIdentity], + ) -> Result { + let mut stats = UsdRateStats::default(); + + for identity in pegs { + let (kind, code, issuer, contract) = identity_columns(identity); + + // --- 0139 guard, both directions ----------------------------- + // Forward: this identity must resolve to exactly one asset_id. + let ids: Vec = self + .client + .query( + "SELECT asset_id FROM prices.assets FINAL \ + WHERE asset_code = ? AND issuer_address = ? AND contract_address = ?", + ) + .bind(code) + .bind(issuer) + .bind(contract) + .fetch_all::() + .await?; + let [asset_id] = ids[..] else { + return Err(IngestError::Precondition(format!( + "peg identity {code} resolves to {} asset_ids (expected exactly 1); \ + refusing to write usd_rate — see task 0139", + ids.len() + ))); + }; + // Reverse: that asset_id must not be shared with another identity, + // or its oracle rows are not unambiguously this asset's. + let sharers: u64 = self + .client + .query("SELECT count() FROM prices.assets FINAL WHERE asset_id = ?") + .bind(asset_id) + .fetch_one::() + .await?; + if sharers != 1 { + return Err(IngestError::Precondition(format!( + "peg identity {code} maps to asset_id {asset_id}, which serves {sharers} \ + identities; refusing to write usd_rate — see task 0139" + ))); + } + + // --- watermark ------------------------------------------------ + let before: u32 = self + .client + .query( + "SELECT toUInt32(ifNull(max(timestamp), toDateTime(0))) \ + FROM prices.usd_rate \ + WHERE asset_kind = ? AND asset_code = ? AND issuer_address = ? \ + AND contract_address = ? AND method = 'oracle'", + ) + .bind(kind) + .bind(code) + .bind(issuer) + .bind(contract) + .fetch_one::() + .await?; + + // --- copy forward --------------------------------------------- + // `price_usd > 0` drops unusable readings rather than storing a + // zero that a consumer cannot distinguish from a real rate — the + // close_usd = 0 mistake, which this table exists partly to avoid. + self.client + .query( + "INSERT INTO prices.usd_rate \ + (asset_kind, asset_code, issuer_address, contract_address, \ + timestamp, usd_rate, method, reference_asset, hops, version) \ + SELECT ?, ?, ?, ?, o.timestamp, o.price_usd, 'oracle', '', 0, \ + toUInt64(now()) \ + FROM prices.oracle_prices AS o FINAL \ + WHERE o.asset_id = ? AND o.price_usd > 0 AND o.timestamp > toDateTime(?)", + ) + .bind(kind) + .bind(code) + .bind(issuer) + .bind(contract) + .bind(asset_id) + .bind(before) + .execute() + .await?; + + let after: u32 = self + .client + .query( + "SELECT toUInt32(ifNull(max(timestamp), toDateTime(0))) \ + FROM prices.usd_rate \ + WHERE asset_kind = ? AND asset_code = ? AND issuer_address = ? \ + AND contract_address = ? AND method = 'oracle'", + ) + .bind(kind) + .bind(code) + .bind(issuer) + .bind(contract) + .fetch_one::() + .await?; + + stats.identities += 1; + // `min` over a Default-initialised 0 would pin this at 0 forever, + // so track the first value explicitly. + stats.watermark_before = Some(match stats.watermark_before { + Some(w) => w.min(before), + None => before, + }); + stats.watermark_after = stats.watermark_after.max(after); + } + Ok(stats) + } +} + +/// Split an [`AssetIdentity`] into the four natural-identity columns +/// `usd_rate` (and the read-surface views) are keyed on. The `''` padding for +/// the inapplicable columns is the same convention `views.sql` documents, so a +/// rate row joins a series row without a translation step. +fn identity_columns(id: &AssetIdentity) -> (&str, &str, &str, &str) { + match id { + AssetIdentity::Native => ("native", "XLM", "", ""), + AssetIdentity::Credit { code, issuer } => ("credit", code, issuer, ""), + AssetIdentity::Contract(addr) => ("contract", "", "", addr), + } +} + +/// Outcome of a [`OhlcvWriter::populate_usd_rate_from_oracle`] pass. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct UsdRateStats { + /// Peg identities processed (each one guarded and copied independently). + pub identities: usize, + /// Oldest per-identity watermark before the pass. `None` when no identity + /// was processed — distinct from `Some(0)`, which means "processed, and + /// nothing had been stored yet". + pub watermark_before: Option, + /// Newest per-identity watermark after it. + pub watermark_after: u32, } #[derive(Debug, Serialize, clickhouse::Row)] diff --git a/packages/prices-ingest-core/tests/usd_rate_population_it.rs b/packages/prices-ingest-core/tests/usd_rate_population_it.rs new file mode 100644 index 00000000..abb0b1d6 --- /dev/null +++ b/packages/prices-ingest-core/tests/usd_rate_population_it.rs @@ -0,0 +1,178 @@ +//! Task 0167 — `populate_usd_rate_from_oracle` against a live ClickHouse. +//! +//! docker compose up -d clickhouse +//! cargo test -p prices-ingest-core --test usd_rate_population_it -- --ignored +//! +//! Uses the real `prices` schema rewritten onto a scratch database. The writer +//! hardcodes `prices.*` table names, so the scratch db is selected on the +//! client rather than by rewriting the writer's SQL. + +use clickhouse::Client; +use prices_ingest_core::{AssetIdentity, OhlcvWriter}; + +fn ch_url() -> String { + std::env::var("CLICKHOUSE_URL").unwrap_or_else(|_| "http://localhost:8123".to_string()) +} + +fn usdc() -> AssetIdentity { + AssetIdentity::Credit { + code: "USDC".to_string(), + issuer: prices_clickhouse::USDC_ISSUER.to_string(), + } +} + +/// The writer's SQL is hardcoded against `prices.*`, so tests cannot each own a +/// scratch database — they share the real one and reset it. That makes them +/// mutually destructive under cargo's default parallel runner (the first +/// version of this file truncated one test's fixture out from under the +/// other's, and BOTH failed in ways that looked like product bugs). This lock +/// serialises them; it is test-harness plumbing, not a statement about the +/// writer, which is safe to call concurrently against distinct identities. +static DB_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +async fn fresh_prices_schema() -> Client { + let admin = Client::default().with_url(ch_url()); + prices_clickhouse::apply_sql(&admin, prices_clickhouse::INIT_SQL) + .await + .unwrap(); + for t in ["usd_rate", "oracle_prices", "assets"] { + admin + .query(&format!("TRUNCATE TABLE IF EXISTS prices.{t}")) + .execute() + .await + .unwrap(); + } + admin +} + +async fn seed_usdc(client: &Client, asset_id: u32) { + client + .query(&format!( + "INSERT INTO prices.assets \ + (asset_id, asset_code, asset_type, issuer_address, contract_address, sac_address) \ + VALUES ({asset_id},'USDC','classic','{}','','')", + prices_clickhouse::USDC_ISSUER + )) + .execute() + .await + .unwrap(); +} + +async fn rate_rows(client: &Client) -> Vec<(u32, f64, String, u8)> { + client + .query( + "SELECT toUInt32(timestamp), toFloat64(usd_rate), method, hops \ + FROM prices.usd_rate FINAL WHERE asset_code = 'USDC' ORDER BY timestamp", + ) + .fetch_all::<(u32, f64, String, u8)>() + .await + .unwrap() +} + +#[tokio::test] +#[ignore = "requires a local ClickHouse (cargo test -- --ignored)"] +async fn copies_oracle_readings_then_resumes_from_the_watermark() { + let _guard = DB_LOCK.lock().await; + let client = fresh_prices_schema().await; + seed_usdc(&client, 3).await; + let writer = OhlcvWriter::new(client.clone()); + + // Two readings, deliberately NOT $1 — the whole point is a depeg-aware rate. + client + .query( + "INSERT INTO prices.oracle_prices (timestamp, asset_id, oracle_name, price_usd, raw_data) \ + VALUES (1750000000, 3, 'reflector', 0.9993, ''), \ + (1750003600, 3, 'reflector', 1.0004, '')", + ) + .execute() + .await + .unwrap(); + + let first = writer + .populate_usd_rate_from_oracle(&[usdc()]) + .await + .unwrap(); + assert_eq!(first.identities, 1); + assert_eq!(first.watermark_before, Some(0), "nothing stored yet"); + assert_eq!(first.watermark_after, 1750003600); + + let rows = rate_rows(&client).await; + assert_eq!(rows.len(), 2, "both readings copied"); + assert!((rows[0].1 - 0.9993).abs() < 1e-9, "the real rate, not $1"); + assert_eq!(rows[0].2, "oracle", "method"); + assert_eq!(rows[0].3, 0, "hops = 0 for a measured reading"); + + // Re-run with no new readings: idempotent, no duplicates. + let second = writer + .populate_usd_rate_from_oracle(&[usdc()]) + .await + .unwrap(); + assert_eq!(second.watermark_before, Some(1750003600), "watermark held"); + assert_eq!( + rate_rows(&client).await.len(), + 2, + "re-running must not duplicate" + ); + + // A new reading arrives; only it is copied. + client + .query( + "INSERT INTO prices.oracle_prices (timestamp, asset_id, oracle_name, price_usd, raw_data) \ + VALUES (1750007200, 3, 'reflector', 0.9987, '')", + ) + .execute() + .await + .unwrap(); + let third = writer + .populate_usd_rate_from_oracle(&[usdc()]) + .await + .unwrap(); + assert_eq!(third.watermark_after, 1750007200); + assert_eq!(rate_rows(&client).await.len(), 3, "incremental append"); +} + +/// ⚠️ The 0139 guard. `oracle_prices` is keyed on `asset_id` and `usd_rate` on +/// natural identity, so this copy is the one place the two key spaces meet. +/// With 3,281 ids serving 6,568 identities on prod, translating without +/// checking would file one asset's readings under another's identity — in a +/// table built to be trusted forever. The write must be REFUSED, not attempted. +#[tokio::test] +#[ignore = "requires a local ClickHouse (cargo test -- --ignored)"] +async fn refuses_to_write_when_the_peg_asset_id_is_shared() { + let _guard = DB_LOCK.lock().await; + let client = fresh_prices_schema().await; + seed_usdc(&client, 3).await; + // A second, unrelated identity squatting on the same surrogate id. + client + .query( + "INSERT INTO prices.assets \ + (asset_id, asset_code, asset_type, issuer_address, contract_address, sac_address) \ + VALUES (3,'ARBRIDGE','classic','GARB','','')", + ) + .execute() + .await + .unwrap(); + client + .query( + "INSERT INTO prices.oracle_prices (timestamp, asset_id, oracle_name, price_usd, raw_data) \ + VALUES (1750000000, 3, 'reflector', 0.9993, '')", + ) + .execute() + .await + .unwrap(); + + let writer = OhlcvWriter::new(client.clone()); + let err = writer + .populate_usd_rate_from_oracle(&[usdc()]) + .await + .expect_err("a shared asset_id must refuse the write"); + let msg = err.to_string(); + assert!(msg.contains("0139"), "error must name the cause: {msg}"); + + assert_eq!( + rate_rows(&client).await.len(), + 0, + "refusing means writing NOTHING — a partial write is the failure mode \ + this guard exists to prevent" + ); +} From b34b9358d2f7509830f82c6993aedfdbd50048f0 Mon Sep 17 00:00:00 2001 From: karczuRF Date: Mon, 10 Aug 2026 15:41:17 +0200 Subject: [PATCH 2/5] =?UTF-8?q?fix(lore-0167):=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20partial-write=20guard,=20gap-filling=20copy,=20meth?= =?UTF-8?q?od=20in=20key?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Eight review findings, all verified against the code before accepting. Two serious. The 0139 guard ran INSIDE the write loop, so a failure on a later identity left earlier ones already written - a partial write, which is the exact failure mode the guard exists to prevent. My test could not catch it because it used a single identity. Guards now run as a pre-pass over every identity before any write, and the error names every offender. The resume watermark was max(timestamp) with a strict > filter, which silently skips any reading landing BELOW the frontier. Not hypothetical: write_oracle is also called from sdex-backfill/ingest.rs and prices-ledger-processor/reconcile.rs, which decode oracle readings from HISTORICAL ledgers. Once the scheduled worker advanced the watermark, a backdated reading would never be snapshotted and would then expire from oracle_prices at 13 months - precisely the permanent loss this table exists to prevent. Replaced with a gap-filling LEFT ANTI JOIN on (timestamp, value); both tables are small so the cost is nil. Anti-joining on the value also makes an upstream correction re-copy and win on version, which the strict > had made unreachable despite the doc claiming otherwise. method added to the sorting key: without it a 'pivot' row from 0154 at the same (identity, timestamp) as a measured 'oracle' reading would silently replace it under RMT, later-write-wins rather than better-evidence-wins. Fixed while the table is empty; changing a sorting key later means a rebuild. rates_snapshotted was computed then dropped by the Lambda entrypoint - with the deliberate non-fatal error path, a permanently broken snapshot would report success ~288 times a day with no counter moving. Now logged and returned. Also: stats count rows written rather than identities attempted; newest is per-identity so one stalled peg is visible instead of hidden behind a max(); and oracle_name is a parameter, not a hardcoded 'reflector', because the enrichment tier reads it from config and the snapshotted rate must be the rate that priced the candles or 0154 constraint 5 compares two different things. Two new ITs cover the highs. Workspace lib suite + all CH ITs green on 26.3.10.60. --- ..._usd-rate-table-and-peg-rate-population.md | 39 ++++ packages/oracle-worker/src/lib.rs | 30 ++- packages/oracle-worker/src/main.rs | 2 + packages/prices-clickhouse/schema/init.sql | 17 +- packages/prices-ingest-core/src/writer.rs | 192 +++++++++++------- .../tests/usd_rate_population_it.rs | 137 ++++++++++++- 6 files changed, 324 insertions(+), 93 deletions(-) diff --git a/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md b/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md index 7b81dd1f..94bd73db 100644 --- a/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md +++ b/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md @@ -84,6 +84,45 @@ history: shared the real `prices` database and truncated each other's fixtures under cargo's parallel runner - both failed in ways that looked like product bugs until serialised. + - date: 2026-08-10 + status: active + who: okarcz + note: > + REVIEW FIXES (PR #191). Eight findings, all accepted after verifying each + against the code rather than on assertion. Two were serious. + HIGH 1 - the 0139 guard ran INSIDE the write loop, so a failure on a later + identity left earlier ones already written. That is a partial write, the + exact failure mode the guard exists to prevent, and my test could not + catch it because it used a single identity. Guards now run as a pre-pass + over every identity before any write, and the error names every offender. + New test asserts a collision on USDC writes nothing for the clean USDT. + HIGH 2 - the resume watermark was max(timestamp) with a strict > filter, + which silently skips any reading that lands BELOW the frontier. Not + hypothetical: write_oracle is also called from sdex-backfill/ingest.rs and + prices-ledger-processor/reconcile.rs, which decode oracle readings from + HISTORICAL ledgers. Once the 5-minute worker advanced the watermark, any + backdated reading would never be snapshotted and would then expire from + oracle_prices at 13 months - precisely the permanent loss this table + exists to prevent. Replaced with a gap-filling LEFT ANTI JOIN on + (timestamp, value); both tables are small so the cost is nil. Anti-joining + on the value also makes an upstream correction re-copy and win on version, + which the strict > had made unreachable despite the doc claiming it. + MEDIUM - `method` added to the sorting key. Without it a 'pivot' row from + 0154 at the same (identity, timestamp) as a measured 'oracle' reading + would silently REPLACE it under RMT, with the later write winning rather + than the better evidence. Fixed while the table is still empty; changing a + sorting key later means a rebuild. + MEDIUM - OracleStats.rates_snapshotted was computed and then dropped by + the Lambda entrypoint. Combined with the deliberate non-fatal error path, + a permanently broken snapshot would report success on ~288 invocations a + day with no counter moving. Now logged and returned, per-identity. + LOW - stats counted identities ATTEMPTED not rows written (now + rows_inserted); watermark_after was a max() across identities that would + hide one stalled peg (now per-identity newest); and oracle_name is a + PARAMETER rather than a hardcoded 'reflector', because the enrichment tier + reads it from config and the rate we snapshot must be the rate that priced + the candles or 0154 constraint 5 compares two different things. + Full workspace lib suite + all CH ITs green on the 26.3.10.60 pin. --- # `prices.usd_rate` + peg-asset rate population diff --git a/packages/oracle-worker/src/lib.rs b/packages/oracle-worker/src/lib.rs index bac4fa88..072c3fc2 100644 --- a/packages/oracle-worker/src/lib.rs +++ b/packages/oracle-worker/src/lib.rs @@ -34,6 +34,12 @@ pub const DEFAULT_SOROBAN_RPC: &str = "https://mainnet.sorobanrpc.com"; /// can grow independently of the mapping. pub const TRACKED_SYMBOLS: &[&str] = &["XLM", "USDC", "USDT"]; +/// The `oracle_prices.oracle_name` this worker writes, and the one the task-0167 +/// snapshot copies. Must match the enrichment tier's `ORACLE_NAME` (default +/// `reflector`) — the rate we snapshot has to be the rate that priced the +/// candles, or 0154's constraint-5 reconciliation compares two different things. +pub const ORACLE_NAME: &str = "reflector"; + /// The identities whose oracle readings are snapshotted into `prices.usd_rate` /// (task 0167). Deliberately a **subset** of [`TRACKED_SYMBOLS`]. /// @@ -235,9 +241,11 @@ pub struct OracleStats { pub queried: usize, pub written: usize, pub skipped: usize, - /// Peg identities whose rates were snapshotted into `prices.usd_rate` - /// (task 0167). Zero is not an error — see [`run_oracle`]. - pub rates_snapshotted: usize, + /// Rows written into `prices.usd_rate` by this pass (task 0167). Zero is + /// normal on a steady-state run — the snapshot only copies observations it + /// does not already hold. It is NOT normal for it to be zero forever while + /// `written` keeps climbing; see [`run_oracle`] on why that needs a signal. + pub rates_snapshotted: u64, } /// Poll Reflector for each tracked symbol and write the prices to @@ -271,7 +279,7 @@ pub async fn run_oracle( // a backstop for the 2106 u32 ceiling, not the unit conversion. timestamp: (pd.timestamp / 1000).min(u32::MAX as u64) as u32, asset_id, - oracle_name: "reflector".to_string(), + oracle_name: ORACLE_NAME.to_string(), price_usd: pd.price, raw_data: format!("{{\"symbol\":\"{symbol}\"}}"), }); @@ -311,10 +319,20 @@ pub async fn run_oracle( // here, and it is a data condition an operator must resolve, not something // a retry fixes. let rates_snapshotted = match writer - .populate_usd_rate_from_oracle(&peg_identities()) + .populate_usd_rate_from_oracle(&peg_identities(), ORACLE_NAME) .await { - Ok(stats) => stats.identities, + Ok(stats) => { + // Logged per-identity: a `max()` across identities would report a + // healthy frontier while one peg sat stalled at zero. + tracing::info!( + identities = stats.identities, + rows = stats.rows_inserted, + newest = ?stats.newest, + "usd_rate snapshot" + ); + stats.rows_inserted + } Err(err) => { tracing::error!(error = %err, "usd_rate snapshot failed; oracle_prices is unaffected"); 0 diff --git a/packages/oracle-worker/src/main.rs b/packages/oracle-worker/src/main.rs index 57f0046e..03118ad0 100644 --- a/packages/oracle-worker/src/main.rs +++ b/packages/oracle-worker/src/main.rs @@ -45,12 +45,14 @@ async fn main() -> Result<(), lambda_runtime::Error> { queried = stats.queried, written = stats.written, skipped = stats.skipped, + rates_snapshotted = stats.rates_snapshotted, "oracle-worker run complete" ); Ok::(serde_json::json!({ "queried": stats.queried, "written": stats.written, "skipped": stats.skipped, + "rates_snapshotted": stats.rates_snapshotted, })) } })) diff --git a/packages/prices-clickhouse/schema/init.sql b/packages/prices-clickhouse/schema/init.sql index 1efddabb..b39f174d 100644 --- a/packages/prices-clickhouse/schema/init.sql +++ b/packages/prices-clickhouse/schema/init.sql @@ -275,9 +275,18 @@ SETTINGS index_granularity = 8192; -- makes a fallback indistinguishable from a measurement, which is precisely -- the close_usd = 0 mistake (one value meaning several things) in a new place. -- --- version — the write time, so a later correction of the same --- (identity, timestamp) wins. Re-running the population is therefore --- idempotent: identical content, and RMT collapses on the ORDER BY key. +-- ⚠️ `method` IS PART OF THE SORTING KEY, deliberately. RMT dedups on the +-- sorting key, so without it a 'pivot' row written by 0154 at the same +-- (identity, timestamp) as a measured 'oracle' reading would silently REPLACE +-- it — and the winner would be whichever was written later, not whichever is +-- better evidence. With `method` in the key the two coexist and the consumer +-- chooses. Fixed while the table was still empty; changing a sorting key +-- afterwards means a rebuild. +-- +-- version — the write time. A re-run writes nothing (the copy skips +-- observations already stored with the same value), and a genuine upstream +-- CORRECTION at an already-stored timestamp differs in value, so it is +-- re-copied and its higher version wins. ---------------------------------------------------------------------- CREATE TABLE IF NOT EXISTS prices.usd_rate ( @@ -294,7 +303,7 @@ CREATE TABLE IF NOT EXISTS prices.usd_rate ( ) ENGINE = ReplacingMergeTree(version) PARTITION BY toYYYYMM(timestamp) -ORDER BY (asset_kind, asset_code, issuer_address, contract_address, timestamp) +ORDER BY (asset_kind, asset_code, issuer_address, contract_address, timestamp, method) SETTINGS index_granularity = 8192; ---------------------------------------------------------------------- diff --git a/packages/prices-ingest-core/src/writer.rs b/packages/prices-ingest-core/src/writer.rs index 60390450..2b137554 100644 --- a/packages/prices-ingest-core/src/writer.rs +++ b/packages/prices-ingest-core/src/writer.rs @@ -357,42 +357,51 @@ impl OhlcvWriter { /// Snapshot peg-asset USD rates from `oracle_prices` into `prices.usd_rate` /// (task 0167). /// - /// **Why this copy exists at all.** `oracle_prices` is pruned at - /// `INTERVAL 13 MONTH`, so the earliest depeg-aware readings age out - /// permanently. A consumer cannot simply join `oracle_prices` instead: the - /// published series would *mutate* as rows age out — a bucket reading - /// `0.9993` silently reverting to a `$1` fallback later — which is why - /// `views.sql` forbids that join. So the rate is snapshotted into a - /// forever-retained table, the same way every other USD number is baked - /// rather than joined. + /// **Why the copy exists.** `oracle_prices` is pruned at `INTERVAL 13 + /// MONTH`, so the earliest depeg-aware readings age out permanently. A + /// consumer cannot join `oracle_prices` instead: the published series would + /// *mutate* as rows age out — a bucket reading `0.9993` silently reverting + /// to a `$1` fallback — which is why `views.sql` forbids that join. /// - /// **Incremental by watermark, per identity.** Only observations newer than - /// the newest already stored for that identity are copied, so a re-run is - /// cheap. It is also idempotent: `usd_rate` is a `ReplacingMergeTree` keyed - /// on `(natural identity, timestamp)`, and `version` is the write time, so a - /// repeat writes identical content and a genuine later correction of the - /// same timestamp wins. + /// **Gap-filling, not watermarked.** An earlier version resumed from + /// `max(timestamp)` and copied only rows newer than that. That was wrong: + /// `write_oracle` is *also* called by `sdex-backfill` and the ledger + /// processor's reconcile path, which write readings decoded from + /// **historical** ledgers — i.e. BELOW the frontier. Once the scheduled + /// worker had advanced the watermark to now, any backdated reading would + /// never be snapshotted, and would then age out of `oracle_prices` at 13 + /// months: precisely the permanent loss this table exists to prevent. The + /// copy therefore anti-joins on `(timestamp, value)` and fills any gap + /// wherever it sits. Both tables are small (a few assets), so this is cheap. /// - /// ⚠️ **The 0139 guard is load-bearing, not defensive decoration.** - /// `oracle_prices` is keyed on `asset_id`, `usd_rate` is keyed on natural - /// identity, so this is the one place the two key spaces meet — and task - /// 0139 is confirmed as genuine `asset_id` collisions between unrelated - /// assets (measured 2026-08-10: 3,281 ids serving 6,568 identities; - /// `asset_id 4194` is both `STW` and `ARBRIDGE`). Translating without - /// checking would file one asset's oracle readings under another asset's - /// identity — silently, and in a table built to be trusted forever. If the - /// mapping is not 1:1 in **both** directions we refuse the write and say so. + /// Anti-joining on the **value** as well as the timestamp also means a + /// genuine upstream correction at an already-stored timestamp differs, gets + /// re-copied, and wins on the higher `version` — while an unchanged reading + /// matches and is skipped, keeping re-runs free. + /// + /// ⚠️ **The 0139 guard runs as a PRE-PASS over every identity before any + /// write.** `oracle_prices` is keyed on `asset_id` and `usd_rate` on natural + /// identity, so this is the one place the two key spaces meet, and 0139 is + /// confirmed as genuine collisions (3,281 ids serving 6,568 identities; + /// `asset_id 4194` is both `STW` and `ARBRIDGE`). Translating unchecked + /// would file one asset's readings under another's identity, permanently. + /// Guarding per-identity *inside* the write loop was the first shape and it + /// was wrong twice over: a failure on the first identity silently skipped + /// every later one, and a failure on a later identity left the earlier ones + /// already written — a partial write, which is the exact failure mode the + /// guard exists to prevent. All identities are checked first; if any fails, + /// nothing is written and the error names every offender. pub async fn populate_usd_rate_from_oracle( &self, pegs: &[AssetIdentity], + oracle_name: &str, ) -> Result { - let mut stats = UsdRateStats::default(); + // ---- pre-pass: resolve + guard EVERY identity before writing ---- + let mut resolved: Vec<(&AssetIdentity, u32)> = Vec::with_capacity(pegs.len()); + let mut problems: Vec = Vec::new(); for identity in pegs { - let (kind, code, issuer, contract) = identity_columns(identity); - - // --- 0139 guard, both directions ----------------------------- - // Forward: this identity must resolve to exactly one asset_id. + let (_, code, issuer, contract) = identity_columns(identity); let ids: Vec = self .client .query( @@ -405,14 +414,12 @@ impl OhlcvWriter { .fetch_all::() .await?; let [asset_id] = ids[..] else { - return Err(IngestError::Precondition(format!( - "peg identity {code} resolves to {} asset_ids (expected exactly 1); \ - refusing to write usd_rate — see task 0139", + problems.push(format!( + "{code} resolves to {} asset_ids (expected 1)", ids.len() - ))); + )); + continue; }; - // Reverse: that asset_id must not be shared with another identity, - // or its oracle rows are not unambiguously this asset's. let sharers: u64 = self .client .query("SELECT count() FROM prices.assets FINAL WHERE asset_id = ?") @@ -420,32 +427,38 @@ impl OhlcvWriter { .fetch_one::() .await?; if sharers != 1 { - return Err(IngestError::Precondition(format!( - "peg identity {code} maps to asset_id {asset_id}, which serves {sharers} \ - identities; refusing to write usd_rate — see task 0139" - ))); + problems.push(format!( + "{code} maps to asset_id {asset_id}, shared by {sharers} identities" + )); + continue; } + resolved.push((identity, asset_id)); + } - // --- watermark ------------------------------------------------ - let before: u32 = self - .client - .query( - "SELECT toUInt32(ifNull(max(timestamp), toDateTime(0))) \ - FROM prices.usd_rate \ - WHERE asset_kind = ? AND asset_code = ? AND issuer_address = ? \ - AND contract_address = ? AND method = 'oracle'", - ) - .bind(kind) - .bind(code) - .bind(issuer) - .bind(contract) - .fetch_one::() - .await?; + if !problems.is_empty() { + return Err(IngestError::Precondition(format!( + "refusing to write usd_rate — ambiguous asset_id mapping (task 0139): {}", + problems.join("; ") + ))); + } + + // ---- write: every identity is known-safe by here ---------------- + let mut stats = UsdRateStats::default(); + for (identity, asset_id) in resolved { + let (kind, code, issuer, contract) = identity_columns(identity); - // --- copy forward --------------------------------------------- - // `price_usd > 0` drops unusable readings rather than storing a - // zero that a consumer cannot distinguish from a real rate — the - // close_usd = 0 mistake, which this table exists partly to avoid. + let before: u64 = self.count_rates(kind, code, issuer, contract).await?; + + // `price_usd > 0` drops unusable readings rather than storing a zero + // a consumer cannot tell from a real rate — the close_usd = 0 + // mistake, which this table partly exists to avoid. + // + // `oracle_name` is a parameter, not the literal 'reflector': the + // enrichment tier reads it from config (ORACLE_NAME), and the rate + // we snapshot must be the rate that priced the candles, or the + // 0154-constraint-5 reconciliation compares two different things. + // It also keeps the RMT winner deterministic if a second oracle is + // ever added, instead of leaving it to insert order. self.client .query( "INSERT INTO prices.usd_rate \ @@ -454,18 +467,29 @@ impl OhlcvWriter { SELECT ?, ?, ?, ?, o.timestamp, o.price_usd, 'oracle', '', 0, \ toUInt64(now()) \ FROM prices.oracle_prices AS o FINAL \ - WHERE o.asset_id = ? AND o.price_usd > 0 AND o.timestamp > toDateTime(?)", + LEFT ANTI JOIN \ + ( \ + SELECT timestamp, usd_rate FROM prices.usd_rate FINAL \ + WHERE asset_kind = ? AND asset_code = ? AND issuer_address = ? \ + AND contract_address = ? AND method = 'oracle' \ + ) AS r ON o.timestamp = r.timestamp AND o.price_usd = r.usd_rate \ + WHERE o.asset_id = ? AND o.price_usd > 0 AND o.oracle_name = ?", ) .bind(kind) .bind(code) .bind(issuer) .bind(contract) + .bind(kind) + .bind(code) + .bind(issuer) + .bind(contract) .bind(asset_id) - .bind(before) + .bind(oracle_name) .execute() .await?; - let after: u32 = self + let after_rows: u64 = self.count_rates(kind, code, issuer, contract).await?; + let newest: u32 = self .client .query( "SELECT toUInt32(ifNull(max(timestamp), toDateTime(0))) \ @@ -481,16 +505,33 @@ impl OhlcvWriter { .await?; stats.identities += 1; - // `min` over a Default-initialised 0 would pin this at 0 forever, - // so track the first value explicitly. - stats.watermark_before = Some(match stats.watermark_before { - Some(w) => w.min(before), - None => before, - }); - stats.watermark_after = stats.watermark_after.max(after); + stats.rows_inserted += after_rows.saturating_sub(before); + stats.newest.push((code.to_string(), newest)); } Ok(stats) } + + async fn count_rates( + &self, + kind: &str, + code: &str, + issuer: &str, + contract: &str, + ) -> Result { + Ok(self + .client + .query( + "SELECT count() FROM prices.usd_rate FINAL \ + WHERE asset_kind = ? AND asset_code = ? AND issuer_address = ? \ + AND contract_address = ? AND method = 'oracle'", + ) + .bind(kind) + .bind(code) + .bind(issuer) + .bind(contract) + .fetch_one::() + .await?) + } } /// Split an [`AssetIdentity`] into the four natural-identity columns @@ -508,14 +549,17 @@ fn identity_columns(id: &AssetIdentity) -> (&str, &str, &str, &str) { /// Outcome of a [`OhlcvWriter::populate_usd_rate_from_oracle`] pass. #[derive(Debug, Default, Clone, PartialEq, Eq)] pub struct UsdRateStats { - /// Peg identities processed (each one guarded and copied independently). + /// Identities processed. All guards passed, or the call would have errored + /// without writing anything. pub identities: usize, - /// Oldest per-identity watermark before the pass. `None` when no identity - /// was processed — distinct from `Some(0)`, which means "processed, and - /// nothing had been stored yet". - pub watermark_before: Option, - /// Newest per-identity watermark after it. - pub watermark_after: u32, + /// Rows actually written. This is the number that says whether the pass did + /// anything — `identities` counts what was *attempted* and reads the same on + /// a run that copied thousands of rows and one that copied none. + pub rows_inserted: u64, + /// Newest snapshotted observation PER IDENTITY, `(asset_code, timestamp)`. + /// Deliberately not a `max()` across identities: that would report a healthy + /// frontier while one peg was silently stalled at zero. + pub newest: Vec<(String, u32)>, } #[derive(Debug, Serialize, clickhouse::Row)] diff --git a/packages/prices-ingest-core/tests/usd_rate_population_it.rs b/packages/prices-ingest-core/tests/usd_rate_population_it.rs index abb0b1d6..e279bfa5 100644 --- a/packages/prices-ingest-core/tests/usd_rate_population_it.rs +++ b/packages/prices-ingest-core/tests/usd_rate_population_it.rs @@ -71,7 +71,7 @@ async fn rate_rows(client: &Client) -> Vec<(u32, f64, String, u8)> { #[tokio::test] #[ignore = "requires a local ClickHouse (cargo test -- --ignored)"] -async fn copies_oracle_readings_then_resumes_from_the_watermark() { +async fn copies_oracle_readings_and_re_runs_without_duplicating() { let _guard = DB_LOCK.lock().await; let client = fresh_prices_schema().await; seed_usdc(&client, 3).await; @@ -89,12 +89,12 @@ async fn copies_oracle_readings_then_resumes_from_the_watermark() { .unwrap(); let first = writer - .populate_usd_rate_from_oracle(&[usdc()]) + .populate_usd_rate_from_oracle(&[usdc()], "reflector") .await .unwrap(); assert_eq!(first.identities, 1); - assert_eq!(first.watermark_before, Some(0), "nothing stored yet"); - assert_eq!(first.watermark_after, 1750003600); + assert_eq!(first.rows_inserted, 2, "both readings copied"); + assert_eq!(first.newest, vec![("USDC".to_string(), 1750003600)]); let rows = rate_rows(&client).await; assert_eq!(rows.len(), 2, "both readings copied"); @@ -104,10 +104,10 @@ async fn copies_oracle_readings_then_resumes_from_the_watermark() { // Re-run with no new readings: idempotent, no duplicates. let second = writer - .populate_usd_rate_from_oracle(&[usdc()]) + .populate_usd_rate_from_oracle(&[usdc()], "reflector") .await .unwrap(); - assert_eq!(second.watermark_before, Some(1750003600), "watermark held"); + assert_eq!(second.rows_inserted, 0, "a no-op re-run writes nothing"); assert_eq!( rate_rows(&client).await.len(), 2, @@ -124,10 +124,11 @@ async fn copies_oracle_readings_then_resumes_from_the_watermark() { .await .unwrap(); let third = writer - .populate_usd_rate_from_oracle(&[usdc()]) + .populate_usd_rate_from_oracle(&[usdc()], "reflector") .await .unwrap(); - assert_eq!(third.watermark_after, 1750007200); + assert_eq!(third.rows_inserted, 1, "only the new reading"); + assert_eq!(third.newest, vec![("USDC".to_string(), 1750007200)]); assert_eq!(rate_rows(&client).await.len(), 3, "incremental append"); } @@ -163,7 +164,7 @@ async fn refuses_to_write_when_the_peg_asset_id_is_shared() { let writer = OhlcvWriter::new(client.clone()); let err = writer - .populate_usd_rate_from_oracle(&[usdc()]) + .populate_usd_rate_from_oracle(&[usdc()], "reflector") .await .expect_err("a shared asset_id must refuse the write"); let msg = err.to_string(); @@ -176,3 +177,121 @@ async fn refuses_to_write_when_the_peg_asset_id_is_shared() { this guard exists to prevent" ); } + +fn usdt() -> AssetIdentity { + AssetIdentity::Credit { + code: "USDT".to_string(), + issuer: prices_clickhouse::USDT_ISSUER.to_string(), + } +} + +/// Review finding 2. `write_oracle` is also called by `sdex-backfill` and the +/// ledger processor's reconcile path, which decode oracle readings from +/// **historical** ledgers — i.e. with timestamps BELOW the current frontier. +/// A `max(timestamp)` watermark would skip those forever, and they would then +/// age out of `oracle_prices` at 13 months: the exact permanent loss this table +/// exists to prevent. The copy must fill gaps wherever they sit. +#[tokio::test] +#[ignore = "requires a local ClickHouse (cargo test -- --ignored)"] +async fn snapshots_a_backdated_reading_that_lands_below_the_frontier() { + let _guard = DB_LOCK.lock().await; + let client = fresh_prices_schema().await; + seed_usdc(&client, 3).await; + let writer = OhlcvWriter::new(client.clone()); + + client + .query( + "INSERT INTO prices.oracle_prices (timestamp, asset_id, oracle_name, price_usd, raw_data) \ + VALUES (1750003600, 3, 'reflector', 1.0004, '')", + ) + .execute() + .await + .unwrap(); + writer + .populate_usd_rate_from_oracle(&[usdc()], "reflector") + .await + .unwrap(); + + // A backfill now writes an OLDER reading — below the frontier just set. + client + .query( + "INSERT INTO prices.oracle_prices (timestamp, asset_id, oracle_name, price_usd, raw_data) \ + VALUES (1740000000, 3, 'reflector', 0.9981, '')", + ) + .execute() + .await + .unwrap(); + + let stats = writer + .populate_usd_rate_from_oracle(&[usdc()], "reflector") + .await + .unwrap(); + assert_eq!( + stats.rows_inserted, 1, + "a backdated reading must still be snapshotted — a max() watermark \ + would skip it, and it then expires from oracle_prices for good" + ); + let rows = rate_rows(&client).await; + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].0, 1740000000, "the older row is present"); +} + +/// Review finding 1. Guarding per-identity *inside* the write loop meant a +/// failure on a later identity left earlier identities already written — a +/// partial write, which is the failure mode the guard exists to prevent. The +/// original test only used one identity, so it could not catch this. +#[tokio::test] +#[ignore = "requires a local ClickHouse (cargo test -- --ignored)"] +async fn a_collision_on_one_peg_writes_nothing_for_any_peg() { + let _guard = DB_LOCK.lock().await; + let client = fresh_prices_schema().await; + seed_usdc(&client, 3).await; + // USDT is clean... + client + .query(&format!( + "INSERT INTO prices.assets \ + (asset_id, asset_code, asset_type, issuer_address, contract_address, sac_address) \ + VALUES (111,'USDT','classic','{}','','')", + prices_clickhouse::USDT_ISSUER + )) + .execute() + .await + .unwrap(); + // ...but USDC's surrogate id is shared, and USDC is processed FIRST. + client + .query( + "INSERT INTO prices.assets \ + (asset_id, asset_code, asset_type, issuer_address, contract_address, sac_address) \ + VALUES (3,'ARBRIDGE','classic','GARB','','')", + ) + .execute() + .await + .unwrap(); + client + .query( + "INSERT INTO prices.oracle_prices (timestamp, asset_id, oracle_name, price_usd, raw_data) \ + VALUES (1750000000, 3, 'reflector', 0.9993, ''), \ + (1750000000, 111, 'reflector', 0.9997, '')", + ) + .execute() + .await + .unwrap(); + + let writer = OhlcvWriter::new(client.clone()); + let err = writer + .populate_usd_rate_from_oracle(&[usdc(), usdt()], "reflector") + .await + .expect_err("one bad peg must fail the whole pass"); + assert!(err.to_string().contains("0139"), "{err}"); + + let total: u64 = client + .query("SELECT count() FROM prices.usd_rate") + .fetch_one::() + .await + .unwrap(); + assert_eq!( + total, 0, + "USDT is clean but must NOT be written — a guard that fails after a \ + partial write is worse than no guard" + ); +} From 680da97f8629a0d34feca3c5283733a2f1b47e8a Mon Sep 17 00:00:00 2001 From: karczuRF Date: Mon, 10 Aug 2026 16:51:47 +0200 Subject: [PATCH 3/5] fix(lore-0167): don't snapshot task-0086's junk 1970 timestamps Found by measuring prod, not by review. Sizing usd_rate meant querying oracle_prices, and min(timestamp) came back 1970-01-21 - which is 0086, an open confirmed bug where the oracle worker intermittently writes the real epoch divided by ~1000, with a CORRECT price and a junk timestamp. The population copied o.timestamp verbatim, filtered only on price_usd > 0, so those rows would have been snapshotted. That is strictly worse here than upstream: oracle_prices sheds them at 13 months, usd_rate is retained FOREVER, so a known defect would have become permanent history in a table whose entire selling point is being trustworthy. Adds ORACLE_EPOCH_FLOOR (2020-01-01) to the copy predicate, with a test asserting a 0086-shaped row is skipped while the good reading beside it is copied. The floor cannot exclude real data - no oracle we poll existed before Soroban - and it does NOT fix 0086, which still pollutes oracle_prices and every other reader of it. --- ..._usd-rate-table-and-peg-rate-population.md | 23 +++++++++ packages/prices-ingest-core/src/writer.rs | 21 +++++++- .../tests/usd_rate_population_it.rs | 51 +++++++++++++++++++ 3 files changed, 94 insertions(+), 1 deletion(-) diff --git a/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md b/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md index 94bd73db..e9574a6f 100644 --- a/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md +++ b/lore/1-tasks/active/0167_FEATURE_usd-rate-table-and-peg-rate-population.md @@ -123,6 +123,29 @@ history: reads it from config and the rate we snapshot must be the rate that priced the candles or 0154 constraint 5 compares two different things. Full workspace lib suite + all CH ITs green on the 26.3.10.60 pin. + - date: 2026-08-10 + status: active + who: okarcz + note: > + 0086 GUARD ADDED, found by measuring prod rather than by review. Sizing the + table meant querying oracle_prices, and min(timestamp) came back + 1970-01-21 - which is [[0086]], an open confirmed bug where the oracle + worker intermittently writes the real epoch divided by ~1000, with a + CORRECT price and a junk timestamp. + My population copied o.timestamp verbatim, filtered only on price_usd > 0, + so those rows would have been snapshotted. That is strictly worse here + than upstream: oracle_prices sheds them at 13 months, usd_rate is retained + FOREVER, so a known defect would have become permanent history in a table + whose entire selling point is that it is trustworthy. + Added ORACLE_EPOCH_FLOOR (2020-01-01) to the copy predicate, with a test. + The floor cannot exclude real data - no oracle we poll existed before + Soroban - and it does NOT fix 0086, which still pollutes oracle_prices and + every other reader. + Prod shape while measuring: oracle_prices holds 452,596 rows of which + 90,722 are peg-asset rows. The gap is expected - write_oracle is also fed + by the event-decoded path, which carries the whole Reflector symbol set + (BTC/ETH/XRP/...), not just the three we poll. 90,722 is what the first + snapshot will copy, minus the 0086 rows. --- # `prices.usd_rate` + peg-asset rate population diff --git a/packages/prices-ingest-core/src/writer.rs b/packages/prices-ingest-core/src/writer.rs index 2b137554..25377ddf 100644 --- a/packages/prices-ingest-core/src/writer.rs +++ b/packages/prices-ingest-core/src/writer.rs @@ -22,6 +22,12 @@ use crate::error::IngestError; use crate::registry_io::PoolRegistryRow; use crate::soroban::Registries; +/// Earliest plausible oracle observation, as an epoch second. Anything at or +/// below this is a task-0086 unit bug (real epoch divided by ~1000, landing in +/// 1970-01), not a real reading — no oracle we poll existed before Soroban. +/// Used to keep that junk out of the forever-retained `prices.usd_rate`. +pub const ORACLE_EPOCH_FLOOR: u32 = 1_577_836_800; // 2020-01-01T00:00:00Z + /// Convert a `Decimal` to the `i128` mantissa ClickHouse expects for a /// `Decimal(38, 14)` column. Saturates rather than panicking: AMM /// amounts/prices are i128-derived and can exceed the 38-digit budget, and an @@ -379,6 +385,17 @@ impl OhlcvWriter { /// re-copied, and wins on the higher `version` — while an unchanged reading /// matches and is skipped, keeping re-runs free. /// + /// ⚠️ **Junk timestamps are filtered out, and this is not defensive + /// decoration either.** Task 0086 is an open, confirmed bug: the oracle + /// worker intermittently writes `oracle_prices` rows whose `timestamp` is + /// the real epoch divided by ~1000, landing them in `1970-01` with a + /// *correct* `price_usd`. Copying those verbatim would be worse here than + /// upstream — `oracle_prices` sheds them after 13 months, whereas + /// `usd_rate` is retained FOREVER, so a known upstream defect would become + /// permanent. [`ORACLE_EPOCH_FLOOR`] drops them. It does not fix 0086, and + /// is not a reason to leave 0086 open: the same junk still pollutes + /// `oracle_prices` and anything else reading it. + /// /// ⚠️ **The 0139 guard runs as a PRE-PASS over every identity before any /// write.** `oracle_prices` is keyed on `asset_id` and `usd_rate` on natural /// identity, so this is the one place the two key spaces meet, and 0139 is @@ -473,7 +490,8 @@ impl OhlcvWriter { WHERE asset_kind = ? AND asset_code = ? AND issuer_address = ? \ AND contract_address = ? AND method = 'oracle' \ ) AS r ON o.timestamp = r.timestamp AND o.price_usd = r.usd_rate \ - WHERE o.asset_id = ? AND o.price_usd > 0 AND o.oracle_name = ?", + WHERE o.asset_id = ? AND o.price_usd > 0 AND o.oracle_name = ? \ + AND o.timestamp > toDateTime(?)", ) .bind(kind) .bind(code) @@ -485,6 +503,7 @@ impl OhlcvWriter { .bind(contract) .bind(asset_id) .bind(oracle_name) + .bind(ORACLE_EPOCH_FLOOR) .execute() .await?; diff --git a/packages/prices-ingest-core/tests/usd_rate_population_it.rs b/packages/prices-ingest-core/tests/usd_rate_population_it.rs index e279bfa5..2a903be7 100644 --- a/packages/prices-ingest-core/tests/usd_rate_population_it.rs +++ b/packages/prices-ingest-core/tests/usd_rate_population_it.rs @@ -295,3 +295,54 @@ async fn a_collision_on_one_peg_writes_nothing_for_any_peg() { partial write is worse than no guard" ); } + +/// Task 0086 is an open bug: the oracle worker intermittently writes +/// `oracle_prices` rows whose timestamp is the real epoch divided by ~1000, +/// landing in 1970-01 with a *correct* price. Found in prod while sizing 0167 — +/// `min(timestamp)` on `oracle_prices` reads `1970-01-21`. +/// +/// Copying those into `usd_rate` would be worse than leaving them upstream: +/// `oracle_prices` sheds them at 13 months, `usd_rate` is retained forever, so +/// a known upstream defect would become permanent history. +#[tokio::test] +#[ignore = "requires a local ClickHouse (cargo test -- --ignored)"] +async fn does_not_snapshot_the_0086_junk_1970_timestamps() { + let _guard = DB_LOCK.lock().await; + let client = fresh_prices_schema().await; + seed_usdc(&client, 3).await; + let writer = OhlcvWriter::new(client.clone()); + + // One good reading and one 0086-shaped row: correct price, epoch/1000. + client + .query( + "INSERT INTO prices.oracle_prices (timestamp, asset_id, oracle_name, price_usd, raw_data) \ + VALUES (1750000000, 3, 'reflector', 0.9993, ''), \ + ( 1750000, 3, 'reflector', 0.9991, '')", + ) + .execute() + .await + .unwrap(); + + let stats = writer + .populate_usd_rate_from_oracle(&[usdc()], "reflector") + .await + .unwrap(); + assert_eq!(stats.rows_inserted, 1, "only the good reading is copied"); + + let rows = rate_rows(&client).await; + assert_eq!(rows.len(), 1); + assert_eq!( + rows[0].0, 1750000000, + "the 1970 row must not be snapshotted" + ); + + let junk: u64 = client + .query("SELECT count() FROM prices.usd_rate WHERE timestamp < toDateTime('2020-01-01')") + .fetch_one::() + .await + .unwrap(); + assert_eq!( + junk, 0, + "no pre-2020 rows may reach the forever-retained table" + ); +} From 088c40ab2b9f4c698962131a1c43b45c91a7b947 Mon Sep 17 00:00:00 2001 From: karczuRF Date: Mon, 10 Aug 2026 17:03:22 +0200 Subject: [PATCH 4/5] docs(lore-0167): document prices.usd_rate in the database-schema overview Adds section 3.4a alongside oracle_prices, plus the ERD entity (both the main diagram and Appendix A), the storage-engines summary row, and the retention section. The retention entry is the one that matters most: it states that usd_rate is NEVER pruned and WHY - the cleanup list is opt-in, and this table exists precisely because oracle_prices expires and takes the earliest depeg-aware history with it, unrecoverably. Someone tidying the retention list is the realistic way that gets undone, so the reason sits next to the rule rather than only in the task. The section also records the things that are easy to get wrong later: the key is natural identity not asset_id (0139 is confirmed collisions, 3,281 ids over 6,568 identities); method is in the sorting key so a pivot estimate cannot silently replace a measured reading under RMT; absence is the signal for pre-oracle history, so no synthetic $1 rows; and the measured size (~2.3 MiB per year at worst) so nobody re-litigates forever-retention on cost grounds. clickhouse-prod-schema.sql is deliberately untouched - it documents BE's schema, not ours. --- .../database-schema-overview.md | 135 ++++++++++++++++-- 1 file changed, 125 insertions(+), 10 deletions(-) diff --git a/docs/database-schema/database-schema-overview.md b/docs/database-schema/database-schema-overview.md index 25cdc3f7..7551ba8e 100644 --- a/docs/database-schema/database-schema-overview.md +++ b/docs/database-schema/database-schema-overview.md @@ -143,16 +143,16 @@ are pushed to the Hetzner cluster via separate post-backfill tools. ## 2. Database Tech Stack -| Component | Technology | -| ---------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Database engine | **ClickHouse** on BE's shared Hetzner cluster (separate `prices` database, ADR 0007) | -| Storage engines | `ReplacingMergeTree(version)` for OHLCV and `unresolved_pools`; `ReplacingMergeTree(updated_at)` for `current_prices` / `assets` / `asset_metadata` / `backfill_progress` / `pool_registry` / `discovery_state`; `ReplacingMergeTree(fetched_at)` for `asset_supply`; `ReplacingMergeTree(ledger)` for `ingest_cursor` (highest-ledger-wins, §3.12); bare `ReplacingMergeTree` for `oracle_prices` / `backfill_sdex_ledgers` | -| Rollups | Chain of CH materialised views: `price_ohlcv_1m → _15m → _1h → _4h → _1d → _1w → _1M` (replaces the OHLCV Rollup Lambda) | -| Partitioning | `PARTITION BY toYYYYMM(timestamp)` on every OHLCV/oracle table; cleanup via `ALTER TABLE … DROP PARTITION` | -| Database client (Rust) | [`clickhouse`](https://crates.io/crates/clickhouse) — async, native protocol over HTTPS-mTLS | -| Schema tooling | Plain SQL DDL applied by the prices-api schema applier on first deploy; prices-api owns `prices.*` migrations unilaterally (ADR 0007 §3.7) | -| Hosting | BE-managed Hetzner box behind Caddy:443; cross-cloud (AWS → Hetzner) hop, ~80–130 ms RTT mitigated by warm connection reuse and batched per-ledger writes | -| Credentials | AWS Secrets Manager — per-env client `{cert,key,ca}` as a single JSON bundle secret per identity (one secret per identity per env, named by `MTLS_SECRET_NAME`; ADR 0007 / task 0063) | +| Component | Technology | +| ---------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Database engine | **ClickHouse** on BE's shared Hetzner cluster (separate `prices` database, ADR 0007) | +| Storage engines | `ReplacingMergeTree(version)` for OHLCV and `unresolved_pools`; `ReplacingMergeTree(updated_at)` for `current_prices` / `assets` / `asset_metadata` / `backfill_progress` / `pool_registry` / `discovery_state`; `ReplacingMergeTree(fetched_at)` for `asset_supply`; `ReplacingMergeTree(ledger)` for `ingest_cursor` (highest-ledger-wins, §3.12); bare `ReplacingMergeTree` for `oracle_prices` / `backfill_sdex_ledgers`; `ReplacingMergeTree(version)` for `usd_rate` | +| Rollups | Chain of CH materialised views: `price_ohlcv_1m → _15m → _1h → _4h → _1d → _1w → _1M` (replaces the OHLCV Rollup Lambda) | +| Partitioning | `PARTITION BY toYYYYMM(timestamp)` on every OHLCV/oracle table; cleanup via `ALTER TABLE … DROP PARTITION` | +| Database client (Rust) | [`clickhouse`](https://crates.io/crates/clickhouse) — async, native protocol over HTTPS-mTLS | +| Schema tooling | Plain SQL DDL applied by the prices-api schema applier on first deploy; prices-api owns `prices.*` migrations unilaterally (ADR 0007 §3.7) | +| Hosting | BE-managed Hetzner box behind Caddy:443; cross-cloud (AWS → Hetzner) hop, ~80–130 ms RTT mitigated by warm connection reuse and batched per-ledger writes | +| Credentials | AWS Secrets Manager — per-env client `{cert,key,ca}` as a single JSON bundle secret per identity (one secret per identity per env, named by `MTLS_SECRET_NAME`; ADR 0007 / task 0063) | **Why ClickHouse on a BE-shared cluster (ADR 0007):** @@ -200,6 +200,7 @@ erDiagram assets ||--o{ current_prices : "asset_id (logical)" assets ||--o{ price_ohlcv_1m : "asset_id (logical)" assets ||--o{ oracle_prices : "asset_id (logical)" + oracle_prices ||--o{ usd_rate : "snapshot (asset_id resolved to natural identity)" price_ohlcv_1m ||--o{ price_ohlcv_15m : "MV: 1m → 15m" price_ohlcv_15m ||--o{ price_ohlcv_1h : "MV: 15m → 1h" price_ohlcv_1h ||--o{ price_ohlcv_4h : "MV: 1h → 4h" @@ -266,6 +267,22 @@ erDiagram ORDER_BY sort_key "asset_id, oracle_name, timestamp" } + usd_rate { + LowCardinality_S asset_kind "native|credit|contract" + String asset_code + String issuer_address + String contract_address + DateTime timestamp "DoubleDelta codec" + Decimal_38_14 usd_rate + LowCardinality_S method "oracle|peg|pivot|pivot2" + String reference_asset "'' for oracle/peg" + UInt8 hops "0 oracle/peg, 1 XLM pivot, 2 second hop" + UInt64 version + ENGINE engine "ReplacingMergeTree(version)" + PARTITION_BY partition "toYYYYMM(timestamp)" + ORDER_BY sort_key "asset_kind, asset_code, issuer_address, contract_address, timestamp, method" + } + backfill_progress { LowCardinality_S task_name PK "sdex_archive | soroban_amm" UInt64 start_ledger @@ -829,6 +846,95 @@ SETTINGS index_granularity = 8192; `GET /oracles/{asset_identifier}` for cross-reference. It does **not** feed the `price_usd` field in any other endpoint. +### 3.4a `prices.usd_rate` — USD rate per asset, as a first-class value (task 0167) + +`close_usd` is not a stored fact — it is a **cached product**. Every enrichment +tier computes the same shape (`ch_enrich.rs`): + +``` +close_usd = close × +``` + +The rate is a function of `(quote asset, timestamp)` **only**, never of the +candle being priced. Today it is looked up, multiplied into hundreds of millions +of rows, and then **discarded** — never written down anywhere. This table stores +it: a handful of assets per bucket instead of one product per candle. + +⏳ **The urgency is retention.** `oracle_prices` is pruned at 13 months (§3.4), +so the earliest depeg-aware readings age out permanently. A view cannot avoid +this by joining `oracle_prices` directly — the published series would **mutate** +as rows age out (a bucket reading `0.9993` silently reverting to a `$1` fallback +later), which is why `views.sql` forbids that join. Hence a forever-retained +snapshot. + +```sql +CREATE TABLE prices.usd_rate ( + asset_kind LowCardinality(String), -- natural identity, NOT asset_id + asset_code String, + issuer_address String, + contract_address String, + timestamp DateTime CODEC(DoubleDelta), + usd_rate Decimal(38, 14), + method LowCardinality(String), -- 'oracle'|'peg'|'pivot'|'pivot2' + reference_asset String DEFAULT '', -- what it pivoted through + hops UInt8 DEFAULT 0, -- 0 oracle/peg, 1 XLM pivot, 2 hop + version UInt64 +) +ENGINE = ReplacingMergeTree(version) +PARTITION BY toYYYYMM(timestamp) +ORDER BY (asset_kind, asset_code, issuer_address, contract_address, timestamp, method) +SETTINGS index_granularity = 8192; +``` + +⚠️ **Keyed on natural identity, never `asset_id`.** Task 0139 is confirmed as +genuine `asset_id` collisions between unrelated assets — measured 2026-08-10 at +**3,281 ids serving 6,568 identities** (`asset_id 4194` is both `STW` and +`ARBRIDGE`). An `asset_id` key would be non-unique by construction. It is also +why the population step guards the `asset_id` → identity translation in **both** +directions and refuses to write when ambiguous: `oracle_prices` is +`asset_id`-keyed and this table is not, so the copy is the one place the two key +spaces meet. + +⚠️ **`method` is part of the sorting key, deliberately.** `ReplacingMergeTree` +dedups on the sorting key, so without it a `'pivot'` estimate written at the same +`(identity, timestamp)` as a measured `'oracle'` reading would silently +**replace** it — and the winner would be whichever was written later, not +whichever is better evidence. + +**Resolution rule — ASOF at-or-before, bounded by staleness. Never averaged.** +Rows are _observations_, not bucket aggregates. A consumer needing the rate at +time `T` takes the newest row with `timestamp <= T`, refusing it past a staleness +window. For a bucket-grained consumer such as `price_usd_series`, `T` is the +**bucket's end**. This is the rule the enrichment path already uses, and it +composes across all six granularities for free — a daily close is the ASOF at +day-end, which _is_ the last hourly close. Averages do not compose. vwap is +impossible regardless: oracle observations carry no volume. + +⚠️ **Absence is the signal.** Pre-oracle history (before ~2025-09) gets **no +row**, and the consumer's own peg fallback applies. Synthetic `method = 'peg'` +rows at `$1` are deliberately **not** written — that would make a fallback +indistinguishable from a measurement, which is the `close_usd = 0` mistake in a +new place. + +**Population.** Written by the **Oracle Fetcher** Lambda immediately after it +writes `oracle_prices`, copying peg-asset observations (USDC/USDT) as +`method = 'oracle'`, `hops = 0`. Gap-filling rather than watermarked — an +anti-join on `(timestamp, value)` — because `write_oracle` is also called by the +SDEX backfill and the ledger processor's reconcile path, which write readings +decoded from **historical** ledgers, i.e. below any frontier. Task 0086's junk +`1970-01` timestamps are filtered out: `oracle_prices` sheds them at 13 months, +this table would keep them forever. + +**Size.** Measured on the 26.3.10.60 pin: ~11.5 bytes/row compressed at worst +(high-entropy values), so two peg assets at a 5-minute cadence cost **~2.3 MiB +per year** — roughly 0.01% of the estate's annual growth. Retaining it forever is +effectively free. If task 0154 later writes pivot rates for thousands of assets, +revisit at that granularity choice. + +**Consumers.** None yet — task 0168 is the first, replacing the hardcoded `$1` +peg fallback in `price_usd_series` with the measured rate. Task 0154's second +pivot tier is the next. + ### 3.5 `prices.backfill_progress` — Backfill Progress Tracking One-row-per-stream tracking table powering `GET /backfill/status`. The backfill @@ -1257,6 +1363,14 @@ Coarse-grained data (1h, 4h, 1d, 1w, 1M) → keep forever Oracle table: prices.oracle_prices → DROP PARTITION for months > 13 months old +⚠️ prices.usd_rate → NEVER pruned. Retained forever, deliberately. + The retention list in cleanup-worker is OPT-IN: a table not named there is + kept indefinitely. usd_rate exists precisely BECAUSE oracle_prices expires + and takes the earliest depeg-aware history with it — unrecoverably, since + those readings cannot be re-derived after the fact. Adding usd_rate to the + pruning list would re-create the exact data loss it was built to escape. + Guarded by a test in cleanup-worker (task 0167). + Implementation: ALTER TABLE prices.price_ohlcv_1m DROP PARTITION '' ALTER TABLE prices.price_ohlcv_15m DROP PARTITION '' @@ -2083,6 +2197,7 @@ erDiagram assets ||--o{ current_prices : "asset_id (logical)" assets ||--o{ price_ohlcv_1m : "asset_id (logical)" assets ||--o{ oracle_prices : "asset_id (logical)" + oracle_prices ||--o{ usd_rate : "snapshot (asset_id resolved to natural identity)" assets ||--o| asset_metadata : "asset_id (logical, 1:1 enrichment)" assets ||--o| asset_supply : "asset_id (logical, 1:1 supply)" price_ohlcv_1m ||--o{ price_ohlcv_15m : "MV: 1m → 15m" From 0e5d1d54d66a4ece66a67641d46dcda544e293b8 Mon Sep 17 00:00:00 2001 From: karczuRF Date: Mon, 10 Aug 2026 17:21:33 +0200 Subject: [PATCH 5/5] docs(lore-0168): record the enrichment peg tier as a known adjacent gap Measured live oracle readings on prod: USDC 1.00066784838102 (+0.067%), USDT 0.99930223861292 (-0.070%). Three properties worth having on the record. The ~0.1% figure is real and ordinary, not a depeg event. The two deviate in OPPOSITE directions, so the spread between them is ~0.137% and a flat $1 does not mostly cancel - anything comparing a USDC-denominated value against a USDT-denominated one carries the full 0.14%. And it is a persistent bias rather than noise: five consecutive 5-minute readings held sign and magnitude to four decimal places, so it does not average out across candles the way jitter would. That is a stronger argument for 0168 than a depeg would be, because it is permanent and invisible rather than rare and dramatic. The gap: 0168 fixes the VIEW's peg fallback, while the enrichment peg tier bakes the same flat $1 into close_usd itself for every USDC/USDT-quoted candle the oracle tier did not reach - all deep history plus anything outside the staleness bound. Shipping 0168 leaves the view and the candles disagreeing by that margin. Not folded in: pointing the peg tier at usd_rate is a write-path change to the enrichment hot loop that 0111 owns, and correcting history means re-enrichment rather than a view swap. Also flagged not to queue it ahead of 0172 on magnitude - 0.07% on close_usd is plausibly fine, 0172 is a ~7x error on 102 live pools. --- ...epeg-aware-peg-rate-in-price-usd-series.md | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/lore/1-tasks/backlog/0168_FEATURE_publish-depeg-aware-peg-rate-in-price-usd-series.md b/lore/1-tasks/backlog/0168_FEATURE_publish-depeg-aware-peg-rate-in-price-usd-series.md index 8aa52a60..d73cb88b 100644 --- a/lore/1-tasks/backlog/0168_FEATURE_publish-depeg-aware-peg-rate-in-price-usd-series.md +++ b/lore/1-tasks/backlog/0168_FEATURE_publish-depeg-aware-peg-rate-in-price-usd-series.md @@ -103,9 +103,58 @@ Fold these into 0165 **before it merges**, or this task turns into a rewrite: - [ ] Visible in [[0150]] if that materialises the view, or the fix is lost at materialisation time. +## ⚠️ Known adjacent gap this task does NOT close — the enrichment peg tier + +**Measured on prod 2026-08-10**, from live `oracle_prices` readings: + +| | rate | off par | +|---|---|---| +| USDC | 1.00066784838102 | **+0.067%** | +| USDT | 0.99930223861292 | **−0.070%** | + +Three properties, each of which strengthens the case for this task: + +1. **The ~0.1% figure is real**, ~0.07% per asset. Not hypothetical, not a + depeg event — this is an ordinary Sunday afternoon. +2. **The two deviate in OPPOSITE directions**, so the spread *between* them is + **~0.137%**. A flat `$1` is therefore not a small uniform offset that mostly + cancels; anything comparing a USDC-denominated value against a + USDT-denominated one carries the whole 0.14%. +3. **It is a persistent bias, not noise.** Five consecutive 5-minute readings + held the same sign and magnitude to four decimal places. Jitter around par + would average out across many candles; a stable offset does not — it is + present on *every* row, always in the same direction. That is a stronger + argument than a depeg would be: a depeg is rare and visible, this is + permanent and invisible. + +**The gap:** this task fixes the *view's* peg fallback. The enrichment **peg +tier** bakes the same flat `$1` into `close_usd` itself — + +> a USDC- or USDT-quoted candle gets `close_usd = close × $1`, exact and +> oracle-free, back to SDEX genesis (`ch_enrich.rs`) + +— so every USDC-quoted candle's `close_usd` is ~0.067% **low** and every +USDT-quoted one ~0.070% **high**, wherever the oracle tier did not win. That is +all deep history before the oracle window (~2025-09) plus anything outside the +staleness bound. **Shipping this task leaves that untouched**, and a reader +comparing the view against the candles will find them disagreeing by that margin. + +**Why it is not folded in here.** Pointing the peg tier at [[0167]]'s +`prices.usd_rate` is the obvious fix and becomes possible once that table +exists — but it is a *write-path* change to the enrichment hot loop, which +[[0111]] is already the open performance task for, and correcting history means +re-enrichment rather than a view swap. Different risk class, different task. + +⚠️ **Do not queue it ahead of [[0172]] on magnitude alone.** 0.07% on `close_usd` +is plausibly acceptable for TVL; 0172 is USDT candles reading ~0.14 against USDC, +a ~7× error on 102 live pools. Fix the order-of-magnitude problem first. + ## Out of scope - Building the rate table or populating it — [[0167]]. +- **Correcting `close_usd` itself** (the enrichment peg tier above). Noted + deliberately rather than filed, 2026-08-10 — file it when 0111 makes the + enrichment write path safe to touch, or when someone needs better than 0.07%. - `current_price_usd` / `current_prices`, which is suspected to carry the same base-only assumption as 0165 but is a refreshable-MV rebuild and must not ride along on a view swap.