diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml new file mode 100644 index 00000000..499026e9 --- /dev/null +++ b/.github/workflows/rust.yml @@ -0,0 +1,42 @@ +name: Rust CI + +on: + pull_request: + paths: + - "packages/**" + - "Cargo.toml" + - "Cargo.lock" + - ".github/workflows/rust.yml" + push: + branches: [develop, master] + paths: + - "packages/**" + - "Cargo.toml" + - "Cargo.lock" + - ".github/workflows/rust.yml" + +env: + CARGO_TERM_COLOR: always + +jobs: + check: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: dtolnay/rust-toolchain@stable + + - uses: Swatinem/rust-cache@v2 + + - name: cargo fmt + run: cargo fmt --all -- --check + + - name: cargo check + run: cargo check --workspace + + - name: cargo test + run: cargo test --workspace + + # Scoped to new crates — sdex-backfill has pre-existing clippy issues. + - name: cargo clippy + run: cargo clippy -p extractors-core -p phoenix-extractor -p ledger-processor -- -D warnings diff --git a/Cargo.lock b/Cargo.lock index 4ea214aa..d62e9bb8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -446,6 +446,13 @@ version = "1.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40404c3f5f511ec4da6fe866ddf6a717c309fdbb69fbbad7b0f3edab8f2e835f" +[[package]] +name = "extractors-core" +version = "0.1.0" +dependencies = [ + "thiserror 2.0.18", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -871,6 +878,15 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +[[package]] +name = "ledger-processor" +version = "0.1.0" +dependencies = [ + "extractors-core", + "phoenix-extractor", + "thiserror 2.0.18", +] + [[package]] name = "libc" version = "0.2.186" @@ -989,6 +1005,14 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "phoenix-extractor" +version = "0.1.0" +dependencies = [ + "extractors-core", + "thiserror 2.0.18", +] + [[package]] name = "pin-project-lite" version = "0.2.17" diff --git a/Cargo.toml b/Cargo.toml index 531e5905..0cd475b9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,11 @@ [workspace] resolver = "3" -members = ["packages/sdex-backfill"] +members = [ + "packages/sdex-backfill", + "packages/extractors-core", + "packages/phoenix-extractor", + "packages/ledger-processor", +] [workspace.dependencies] stellar-xdr = { version = "=26.0.0", features = ["curr"] } diff --git a/lore/1-tasks/active/0034_FEATURE_consumer-multi-xyk-wasm-tolerance.md b/lore/1-tasks/active/0034_FEATURE_consumer-multi-xyk-wasm-tolerance.md index 5b6a1423..e852190e 100644 --- a/lore/1-tasks/active/0034_FEATURE_consumer-multi-xyk-wasm-tolerance.md +++ b/lore/1-tasks/active/0034_FEATURE_consumer-multi-xyk-wasm-tolerance.md @@ -80,9 +80,85 @@ and add a runtime warning if an unrecognized hash appears. ## Acceptance Criteria -- [ ] Consumer's Phoenix venue lookup does not silently drop pools +- [x] Consumer's Phoenix venue lookup does not silently drop pools whose WASM hash differs from the most common XYK build. -- [ ] Classifier documented as `pool_type + event_count`, with a unit +- [x] Classifier documented as `pool_type + event_count`, with a unit test covering both XYK pool variants by config-fixture. - [ ] PHO/USDC swaps (pool `CD5XNKK3...IAA`) verified end-to-end - through the consumer in a staging run. + through the consumer in a staging run. (deferred — requires live environment) + +## Implementation Notes + +### Crates created + +This task also delivered the 0037 skeleton as a prerequisite since no +consumer code existed. Three new workspace members under `packages/`: + +| Crate | Path | Role | +|-------|------|------| +| `extractors-core` | `packages/extractors-core` | `SwapExtractor` trait, `SorobanEventRow`, `TaggedValue`, `TradeRow`, `Venue` enum — transcribed from 0018 Appendix A | +| `phoenix-extractor` | `packages/phoenix-extractor` | `PhoenixPoolRegistry` (contract_id → pool_type lookup) + `PhoenixXykExtractor` (8-event grouping decoder) | +| `ledger-processor` | `packages/ledger-processor` | lib + stub binary; `dispatch()` routes by venue, then `(pool_type, event_count)` for Phoenix | + +### Classifier design + +`PhoenixPoolRegistry` keys lookup by **contract_id** and stores +`pool_type: u32` from the factory's `query_config()`. WASM hash is +stored as `Option<[u8; 32]>` metadata but is **never consulted for +extractor selection**. Routing logic in `dispatch_phoenix()`: + +- `pool_type == 0` AND `rows.len() >= 8` → `PhoenixXykExtractor` +- `pool_type != 0` AND `rows.len() >= 6` → stable path (stub, no + mainnet stable pools exist yet per 0032) + +This survives future Phoenix XYK rebuilds without code changes. + +### Tests (14 total) + +**phoenix-extractor (8 tests):** +- Registry fixture construction + lookup for both WASM variants +- Proof that different WASM hashes both resolve as XYK via pool_type +- XYK extractor: 8-event group decode, PHO/USDC alt-WASM pool, + insufficient rows rejection, unordered field tolerance + +**ledger-processor (6 tests):** +- Dispatch routes XLM/USDC (common WASM) correctly +- Dispatch routes PHO/USDC (alt WASM) identically +- Explicit proof that dispatch uses pool_type, not WASM hash +- Stable pool (pool_type != 0) returns error (intentionally unimplemented) +- Unknown venue skipped, empty rows return empty + +### CI + +Added `.github/workflows/rust.yml` — runs `cargo fmt`, `cargo check`, +`cargo test`, `cargo clippy` on PRs touching `packages/` or `Cargo.*`. + +## Design Decisions + +### From Plan + +1. **`pool_type + event_count` classifier**: per 0032 S-note §"So what?" + recommendation. WASM hash stored but never used for routing. + +2. **Per-venue extractor trait**: `SwapExtractor` with + `extract(&[SorobanEventRow]) -> ExtractResult` per 0018 Appendix A. + +### Emerged + +3. **Absorbed 0037 skeleton into this task**: no consumer code existed, + so the 0037 crate layout was a prerequisite. Built the minimum + skeleton (3 crates) needed for 0034's classifier to compile and test. + +4. **Field-name-based extraction over positional**: the XYK extractor + matches fields by `topic[1]` string name rather than relying on + emission order. This tolerates reordered events within a group + (tested explicitly). + +5. **`TaggedValue` enum for CH-level data**: models BE's tagged-JSON + encoding (`type` + `value`) from `R-be-storage-format.md` rather + than raw XDR `ScVal`. This is what the consumer actually reads from + ClickHouse. + +6. **Scoped clippy in CI**: runs clippy only on the three new crates, + not workspace-wide, because `sdex-backfill` has pre-existing clippy + issues unrelated to this task. diff --git a/packages/extractors-core/Cargo.toml b/packages/extractors-core/Cargo.toml new file mode 100644 index 00000000..009345ff --- /dev/null +++ b/packages/extractors-core/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "extractors-core" +version = "0.1.0" +edition = "2024" +description = "Per-venue swap extractor trait and shared types for the Tranche 1 Ledger Processor" + +[dependencies] +thiserror = { workspace = true } diff --git a/packages/extractors-core/src/lib.rs b/packages/extractors-core/src/lib.rs new file mode 100644 index 00000000..0c1edb2a --- /dev/null +++ b/packages/extractors-core/src/lib.rs @@ -0,0 +1,89 @@ +use std::collections::HashMap; + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum Venue { + Soroswap, + Aquarius, + Phoenix, +} + +#[derive(Debug, Clone)] +pub struct SorobanEventRow { + pub contract_id: String, + pub transaction_id: String, + pub ledger_sequence: u64, + pub event_index: u32, + pub topics: Vec, + pub data: TaggedValue, +} + +#[derive(Debug, Clone, PartialEq)] +pub enum TaggedValue { + Symbol(String), + String(String), + Address(String), + I128(i128), + Map(Vec<(TaggedValue, TaggedValue)>), + Vec(Vec), + Null, +} + +impl TaggedValue { + pub fn as_str(&self) -> Option<&str> { + match self { + TaggedValue::Symbol(s) | TaggedValue::String(s) | TaggedValue::Address(s) => Some(s), + _ => None, + } + } + + pub fn as_i128(&self) -> Option { + match self { + TaggedValue::I128(v) => Some(*v), + _ => None, + } + } + + pub fn as_address(&self) -> Option<&str> { + match self { + TaggedValue::Address(s) => Some(s), + _ => None, + } + } +} + +#[derive(Debug, Clone)] +pub struct TradeRow { + pub venue: Venue, + pub contract_id: String, + pub transaction_id: String, + pub ledger_sequence: u64, + pub first_event_index: u32, + pub token_in: String, + pub token_out: String, + pub amount_in: i128, + pub amount_out: i128, + pub fee: Option, + pub trader: Option, +} + +#[derive(Debug)] +pub struct ExtractResult { + pub trades: Vec, + pub rows_consumed: usize, +} + +pub trait SwapExtractor { + fn extract(&self, rows: &[SorobanEventRow]) -> Result; +} + +#[derive(Debug, thiserror::Error)] +pub enum ExtractError { + #[error("not enough rows: need {expected}, got {actual}")] + InsufficientRows { expected: usize, actual: usize }, + #[error("missing field in event group: {0}")] + MissingField(String), + #[error("unexpected topic shape in row at event_index {0}")] + UnexpectedTopicShape(u32), +} + +pub type VenueRegistry = HashMap; diff --git a/packages/ledger-processor/Cargo.toml b/packages/ledger-processor/Cargo.toml new file mode 100644 index 00000000..c60c5309 --- /dev/null +++ b/packages/ledger-processor/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "ledger-processor" +version = "0.1.0" +edition = "2024" +description = "Tranche 1 Ledger Processor — dispatches Soroban swap events to per-venue extractors" + +[lib] +name = "ledger_processor" +path = "src/lib.rs" + +[[bin]] +name = "ledger-processor" +path = "src/main.rs" + +[dependencies] +extractors-core = { path = "../extractors-core" } +phoenix-extractor = { path = "../phoenix-extractor" } +thiserror = { workspace = true } + +[dev-dependencies] +phoenix-extractor = { path = "../phoenix-extractor", features = ["test-fixtures"] } diff --git a/packages/ledger-processor/src/dispatch.rs b/packages/ledger-processor/src/dispatch.rs new file mode 100644 index 00000000..8f08fe90 --- /dev/null +++ b/packages/ledger-processor/src/dispatch.rs @@ -0,0 +1,185 @@ +use extractors_core::{ + ExtractError, SorobanEventRow, SwapExtractor, TradeRow, Venue, VenueRegistry, +}; +use phoenix_extractor::{ + PHOENIX_STABLE_EVENT_COUNT, PHOENIX_XYK_EVENT_COUNT, POOL_TYPE_XYK, PhoenixPoolRegistry, + PhoenixXykExtractor, +}; + +#[derive(Debug, thiserror::Error)] +pub enum DispatchError { + #[error("extract error: {0}")] + Extract(#[from] ExtractError), + #[error("pool {contract_id} not found in Phoenix registry")] + UnknownPool { contract_id: String }, + #[error( + "unknown Phoenix pool_type {pool_type} with {event_count} events for pool {contract_id}" + )] + UnknownPhoenixShape { + contract_id: String, + pool_type: u32, + event_count: usize, + }, + #[error("venue {venue:?} extractor not yet implemented for pool {contract_id}")] + VenueNotImplemented { venue: Venue, contract_id: String }, +} + +/// Route a batch of contiguous Soroban event rows for a single +/// (transaction_id, contract_id) group to the correct extractor. +/// +/// Phoenix routing uses (pool_type, event_count) — never WASM hash. +pub fn dispatch_phoenix( + rows: &[SorobanEventRow], + registry: &PhoenixPoolRegistry, +) -> Result, DispatchError> { + if rows.is_empty() { + return Ok(vec![]); + } + + let contract_id = &rows[0].contract_id; + let pool = registry + .lookup(contract_id) + .ok_or_else(|| DispatchError::UnknownPool { + contract_id: contract_id.clone(), + })?; + + match (pool.pool_type, rows.len()) { + (POOL_TYPE_XYK, n) if n >= PHOENIX_XYK_EVENT_COUNT => { + let result = PhoenixXykExtractor.extract(rows)?; + Ok(result.trades) + } + (pool_type, n) if pool_type != POOL_TYPE_XYK && n >= PHOENIX_STABLE_EVENT_COUNT => { + // Stable extractor not yet implemented — no stable pools exist on mainnet. + Err(DispatchError::UnknownPhoenixShape { + contract_id: contract_id.clone(), + pool_type, + event_count: n, + }) + } + (pool_type, event_count) => Err(DispatchError::UnknownPhoenixShape { + contract_id: contract_id.clone(), + pool_type, + event_count, + }), + } +} + +/// Top-level dispatcher: routes events by venue, then by pool shape for Phoenix. +pub fn dispatch( + rows: &[SorobanEventRow], + venue_registry: &VenueRegistry, + phoenix_registry: &PhoenixPoolRegistry, +) -> Result, DispatchError> { + if rows.is_empty() { + return Ok(vec![]); + } + + let contract_id = &rows[0].contract_id; + let venue = venue_registry.get(contract_id); + + match venue { + Some(Venue::Phoenix) => dispatch_phoenix(rows, phoenix_registry), + Some(Venue::Soroswap) => Err(DispatchError::VenueNotImplemented { + venue: Venue::Soroswap, + contract_id: contract_id.clone(), + }), + Some(Venue::Aquarius) => Err(DispatchError::VenueNotImplemented { + venue: Venue::Aquarius, + contract_id: contract_id.clone(), + }), + None => Ok(vec![]), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use phoenix_extractor::test_fixtures::*; + + fn phoenix_registry_both_wasm_variants() -> PhoenixPoolRegistry { + PhoenixPoolRegistry::from_fixture(&[(XLM_USDC_POOL, 0), (PHO_USDC_POOL, 0)]) + } + + fn venue_registry_phoenix(pools: &[&str]) -> VenueRegistry { + pools + .iter() + .map(|p| (p.to_string(), Venue::Phoenix)) + .collect() + } + + #[test] + fn dispatch_routes_xlm_usdc_xyk_pool() { + let rows = make_phoenix_xyk_events(XLM_USDC_POOL, 5); + let phoenix_reg = phoenix_registry_both_wasm_variants(); + let venue_reg = venue_registry_phoenix(&[XLM_USDC_POOL, PHO_USDC_POOL]); + + let trades = dispatch(&rows, &venue_reg, &phoenix_reg).unwrap(); + assert_eq!(trades.len(), 1); + assert_eq!(trades[0].contract_id, XLM_USDC_POOL); + assert_eq!(trades[0].venue, Venue::Phoenix); + } + + #[test] + fn dispatch_routes_pho_usdc_alt_wasm_pool_identically() { + let rows = make_phoenix_xyk_events(PHO_USDC_POOL, 5); + let phoenix_reg = phoenix_registry_both_wasm_variants(); + let venue_reg = venue_registry_phoenix(&[XLM_USDC_POOL, PHO_USDC_POOL]); + + let trades = dispatch(&rows, &venue_reg, &phoenix_reg).unwrap(); + assert_eq!(trades.len(), 1); + assert_eq!(trades[0].contract_id, PHO_USDC_POOL); + assert_eq!(trades[0].amount_in, 11659417676); + assert_eq!(trades[0].amount_out, 1857322909); + } + + #[test] + fn dispatch_phoenix_uses_pool_type_not_wasm_hash() { + let mut reg = PhoenixPoolRegistry::new(); + reg.register_with_wasm(XLM_USDC_POOL.to_string(), 0, common_xyk_wasm_hash()); + reg.register_with_wasm(PHO_USDC_POOL.to_string(), 0, alt_xyk_wasm_hash()); + + let pool_a = reg.lookup(XLM_USDC_POOL).unwrap(); + let pool_b = reg.lookup(PHO_USDC_POOL).unwrap(); + assert_ne!(pool_a.wasm_hash, pool_b.wasm_hash); + + for pool in [XLM_USDC_POOL, PHO_USDC_POOL] { + let rows = make_phoenix_xyk_events(pool, 5); + let trades = dispatch_phoenix(&rows, ®).unwrap(); + assert_eq!(trades.len(), 1, "pool {pool} should produce 1 trade"); + } + } + + #[test] + fn dispatch_stable_pool_returns_error_unimplemented() { + let rows = make_phoenix_xyk_events(XLM_USDC_POOL, 0); + let mut reg = PhoenixPoolRegistry::new(); + reg.register(XLM_USDC_POOL.to_string(), 1); + + let result = dispatch_phoenix(&rows, ®); + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("unknown Phoenix pool_type 1"), + "expected stable-pool error, got: {err}" + ); + } + + #[test] + fn dispatch_skips_unknown_venue() { + let rows = make_phoenix_xyk_events("CUNKNOWN_POOL_NOT_IN_REGISTRY", 5); + let phoenix_reg = phoenix_registry_both_wasm_variants(); + let venue_reg = venue_registry_phoenix(&[XLM_USDC_POOL]); + + let trades = dispatch(&rows, &venue_reg, &phoenix_reg).unwrap(); + assert!(trades.is_empty()); + } + + #[test] + fn dispatch_empty_rows_returns_empty() { + let phoenix_reg = phoenix_registry_both_wasm_variants(); + let venue_reg = venue_registry_phoenix(&[XLM_USDC_POOL]); + + let trades = dispatch(&[], &venue_reg, &phoenix_reg).unwrap(); + assert!(trades.is_empty()); + } +} diff --git a/packages/ledger-processor/src/lib.rs b/packages/ledger-processor/src/lib.rs new file mode 100644 index 00000000..bcdf76dd --- /dev/null +++ b/packages/ledger-processor/src/lib.rs @@ -0,0 +1 @@ +pub mod dispatch; diff --git a/packages/ledger-processor/src/main.rs b/packages/ledger-processor/src/main.rs new file mode 100644 index 00000000..65d5b66a --- /dev/null +++ b/packages/ledger-processor/src/main.rs @@ -0,0 +1,3 @@ +fn main() { + println!("ledger-processor stub"); +} diff --git a/packages/phoenix-extractor/Cargo.toml b/packages/phoenix-extractor/Cargo.toml new file mode 100644 index 00000000..f0729d74 --- /dev/null +++ b/packages/phoenix-extractor/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "phoenix-extractor" +version = "0.1.0" +edition = "2024" +description = "Phoenix AMM swap extractor — XYK (8-event) and stable (6-event) grouping decoders" + +[features] +test-fixtures = [] + +[dependencies] +extractors-core = { path = "../extractors-core" } +thiserror = { workspace = true } diff --git a/packages/phoenix-extractor/src/lib.rs b/packages/phoenix-extractor/src/lib.rs new file mode 100644 index 00000000..03fa1a61 --- /dev/null +++ b/packages/phoenix-extractor/src/lib.rs @@ -0,0 +1,13 @@ +mod registry; +mod xyk; + +pub use registry::{PhoenixPool, PhoenixPoolRegistry}; +pub use xyk::PhoenixXykExtractor; + +pub const PHOENIX_XYK_EVENT_COUNT: usize = 8; +pub const PHOENIX_STABLE_EVENT_COUNT: usize = 6; + +pub const POOL_TYPE_XYK: u32 = 0; + +#[cfg(any(test, feature = "test-fixtures"))] +pub mod test_fixtures; diff --git a/packages/phoenix-extractor/src/registry.rs b/packages/phoenix-extractor/src/registry.rs new file mode 100644 index 00000000..90496d6a --- /dev/null +++ b/packages/phoenix-extractor/src/registry.rs @@ -0,0 +1,125 @@ +use std::collections::HashMap; + +/// A Phoenix pool as registered from the factory. +/// +/// Keyed by `contract_id` (C-strkey), NOT by WASM hash. +/// Two distinct XYK WASM builds exist in production (167ab414…506c and +/// 13b158655e…f2ca) — both report pool_type == 0 and emit identical +/// 8-event swap groupings. Keying off WASM hash would silently drop +/// the PHO/USDC pool. +#[derive(Debug, Clone)] +pub struct PhoenixPool { + pub contract_id: String, + pub pool_type: u32, + pub wasm_hash: Option<[u8; 32]>, +} + +#[derive(Debug, Default)] +pub struct PhoenixPoolRegistry { + pools: HashMap, +} + +impl PhoenixPoolRegistry { + pub fn new() -> Self { + Self::default() + } + + pub fn register(&mut self, contract_id: String, pool_type: u32) { + self.pools.insert( + contract_id.clone(), + PhoenixPool { + contract_id, + pool_type, + wasm_hash: None, + }, + ); + } + + pub fn register_with_wasm(&mut self, contract_id: String, pool_type: u32, wasm_hash: [u8; 32]) { + self.pools.insert( + contract_id.clone(), + PhoenixPool { + contract_id, + pool_type, + wasm_hash: Some(wasm_hash), + }, + ); + } + + pub fn lookup(&self, contract_id: &str) -> Option<&PhoenixPool> { + self.pools.get(contract_id) + } + + pub fn contains(&self, contract_id: &str) -> bool { + self.pools.contains_key(contract_id) + } + + pub fn pool_count(&self) -> usize { + self.pools.len() + } + + /// Build a registry from a fixture list of (contract_id, pool_type) pairs. + pub fn from_fixture(entries: &[(&str, u32)]) -> Self { + let mut reg = Self::new(); + for &(contract_id, pool_type) in entries { + reg.register(contract_id.to_string(), pool_type); + } + reg + } + + /// Build a registry from a fixture list that includes WASM hashes. + pub fn from_fixture_with_wasm(entries: &[(&str, u32, [u8; 32])]) -> Self { + let mut reg = Self::new(); + for (contract_id, pool_type, wasm_hash) in entries { + reg.register_with_wasm(contract_id.to_string(), *pool_type, *wasm_hash); + } + reg + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_fixtures::*; + + #[test] + fn registry_from_fixture_returns_correct_pool_type() { + let reg = PhoenixPoolRegistry::from_fixture(&[(XLM_USDC_POOL, 0), (PHO_USDC_POOL, 0)]); + + let xlm = reg.lookup(XLM_USDC_POOL).expect("XLM/USDC pool"); + assert_eq!(xlm.pool_type, 0); + assert_eq!(xlm.wasm_hash, None); + + let pho = reg.lookup(PHO_USDC_POOL).expect("PHO/USDC pool"); + assert_eq!(pho.pool_type, 0); + } + + #[test] + fn registry_with_different_wasm_hashes_both_resolve_as_xyk() { + let reg = PhoenixPoolRegistry::from_fixture_with_wasm(&[ + (XLM_USDC_POOL, 0, common_xyk_wasm_hash()), + (PHO_USDC_POOL, 0, alt_xyk_wasm_hash()), + ]); + + let xlm = reg.lookup(XLM_USDC_POOL).unwrap(); + let pho = reg.lookup(PHO_USDC_POOL).unwrap(); + + // Different WASM hashes… + assert_ne!(xlm.wasm_hash, pho.wasm_hash); + // …but both are XYK (pool_type 0) + assert_eq!(xlm.pool_type, 0); + assert_eq!(pho.pool_type, 0); + } + + #[test] + fn lookup_unknown_pool_returns_none() { + let reg = PhoenixPoolRegistry::from_fixture(&[(XLM_USDC_POOL, 0)]); + assert!(reg.lookup("CNOTAPOOL").is_none()); + } + + #[test] + fn pool_count_reflects_registered_entries() { + let reg = PhoenixPoolRegistry::from_fixture(&[(XLM_USDC_POOL, 0), (PHO_USDC_POOL, 0)]); + assert_eq!(reg.pool_count(), 2); + } +} diff --git a/packages/phoenix-extractor/src/test_fixtures.rs b/packages/phoenix-extractor/src/test_fixtures.rs new file mode 100644 index 00000000..714a253e --- /dev/null +++ b/packages/phoenix-extractor/src/test_fixtures.rs @@ -0,0 +1,55 @@ +use extractors_core::{SorobanEventRow, TaggedValue}; + +pub const XLM_USDC_POOL: &str = "CBHCRSVX3ZZ7EGTSYMKPEFGZNWRVCSESQR3UABET4MIW52N4EVU6BIZX"; +pub const PHO_USDC_POOL: &str = "CD5XNKK3B6BEF2N7ULNHHGAMOKZ7P6456BFNIHRF4WNTEDKBRWAE7IAA"; +pub const TRADER: &str = "GDCRZPZYBZ24RHRO3WBPJGFDL7NDFKUQBS3ZDB6YGBJB3TGKMFYBQ3LD"; +pub const XLM_SAC: &str = "CAS3J7GYLGXMF6TDJBBYYSE3HQ6BBSMLNUQ34T6TZMYMW2EVH34XOWMA"; +pub const USDC_SAC: &str = "CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75"; +pub const TX_HASH: &str = "559498bdf567340c0780b80f2bfa07bcc58713fc328e659ef72461849a326aa8"; + +pub fn make_phoenix_xyk_events(pool: &str, base_index: u32) -> Vec { + let fields: &[(&str, TaggedValue)] = &[ + ("sender", TaggedValue::Address(TRADER.into())), + ("sell_token", TaggedValue::Address(XLM_SAC.into())), + ("offer_amount", TaggedValue::I128(11659417676)), + ("actual received amount", TaggedValue::I128(11659417676)), + ("buy_token", TaggedValue::Address(USDC_SAC.into())), + ("return_amount", TaggedValue::I128(1857322909)), + ("spread_amount", TaggedValue::I128(503808)), + ("referral_fee_amount", TaggedValue::I128(0)), + ]; + + fields + .iter() + .enumerate() + .map(|(i, (name, data))| SorobanEventRow { + contract_id: pool.to_string(), + transaction_id: TX_HASH.to_string(), + ledger_sequence: 62460522, + event_index: base_index + i as u32, + topics: vec![ + TaggedValue::String("swap".into()), + TaggedValue::String((*name).into()), + ], + data: data.clone(), + }) + .collect() +} + +pub fn common_xyk_wasm_hash() -> [u8; 32] { + let mut h = [0u8; 32]; + h[0] = 0x16; + h[1] = 0x7a; + h[2] = 0xb4; + h[3] = 0x14; + h +} + +pub fn alt_xyk_wasm_hash() -> [u8; 32] { + let mut h = [0u8; 32]; + h[0] = 0x13; + h[1] = 0xb1; + h[2] = 0x58; + h[3] = 0x65; + h +} diff --git a/packages/phoenix-extractor/src/xyk.rs b/packages/phoenix-extractor/src/xyk.rs new file mode 100644 index 00000000..2f67e54f --- /dev/null +++ b/packages/phoenix-extractor/src/xyk.rs @@ -0,0 +1,138 @@ +use extractors_core::{ + ExtractError, ExtractResult, SorobanEventRow, SwapExtractor, TradeRow, Venue, +}; + +use crate::PHOENIX_XYK_EVENT_COUNT; + +/// Extracts a single Phoenix XYK swap from 8 contiguous Soroban event rows. +/// +/// Each row has topics = [String("swap"), String("")] and data = . +/// The 8 fields in emission order: +/// sender, sell_token, offer_amount, actual received amount, +/// buy_token, return_amount, spread_amount, referral_fee_amount +pub struct PhoenixXykExtractor; + +impl SwapExtractor for PhoenixXykExtractor { + fn extract(&self, rows: &[SorobanEventRow]) -> Result { + if rows.len() < PHOENIX_XYK_EVENT_COUNT { + return Err(ExtractError::InsufficientRows { + expected: PHOENIX_XYK_EVENT_COUNT, + actual: rows.len(), + }); + } + + let group = &rows[..PHOENIX_XYK_EVENT_COUNT]; + + let mut sender = None; + let mut sell_token = None; + let mut offer_amount = None; + let mut buy_token = None; + let mut return_amount = None; + + for row in group { + let topic0 = row + .topics + .first() + .and_then(|t| t.as_str()) + .ok_or(ExtractError::UnexpectedTopicShape(row.event_index))?; + if topic0 != "swap" { + return Err(ExtractError::UnexpectedTopicShape(row.event_index)); + } + + let field_name = row + .topics + .get(1) + .and_then(|t| t.as_str()) + .ok_or(ExtractError::UnexpectedTopicShape(row.event_index))?; + + match field_name { + "sender" => sender = row.data.as_address().map(|s| s.to_string()), + "sell_token" => sell_token = row.data.as_address().map(|s| s.to_string()), + "offer_amount" => offer_amount = row.data.as_i128(), + "buy_token" => buy_token = row.data.as_address().map(|s| s.to_string()), + "return_amount" => return_amount = row.data.as_i128(), + "actual received amount" | "spread_amount" | "referral_fee_amount" => {} + _ => {} + } + } + + let first = &group[0]; + + let trade = TradeRow { + venue: Venue::Phoenix, + contract_id: first.contract_id.clone(), + transaction_id: first.transaction_id.clone(), + ledger_sequence: first.ledger_sequence, + first_event_index: first.event_index, + token_in: sell_token.ok_or_else(|| ExtractError::MissingField("sell_token".into()))?, + token_out: buy_token.ok_or_else(|| ExtractError::MissingField("buy_token".into()))?, + amount_in: offer_amount + .ok_or_else(|| ExtractError::MissingField("offer_amount".into()))?, + amount_out: return_amount + .ok_or_else(|| ExtractError::MissingField("return_amount".into()))?, + fee: None, + trader: sender, + }; + + Ok(ExtractResult { + trades: vec![trade], + rows_consumed: PHOENIX_XYK_EVENT_COUNT, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_fixtures::*; + + #[test] + fn xyk_extractor_decodes_8_event_group() { + let rows = make_phoenix_xyk_events(XLM_USDC_POOL, 5); + let result = PhoenixXykExtractor.extract(&rows).unwrap(); + + assert_eq!(result.rows_consumed, 8); + assert_eq!(result.trades.len(), 1); + + let trade = &result.trades[0]; + assert_eq!(trade.venue, Venue::Phoenix); + assert_eq!(trade.contract_id, XLM_USDC_POOL); + assert_eq!(trade.token_in, XLM_SAC); + assert_eq!(trade.token_out, USDC_SAC); + assert_eq!(trade.amount_in, 11659417676); + assert_eq!(trade.amount_out, 1857322909); + assert_eq!(trade.trader.as_deref(), Some(TRADER)); + assert_eq!(trade.fee, None); + } + + #[test] + fn xyk_extractor_works_for_pho_usdc_pool_with_alt_wasm() { + let rows = make_phoenix_xyk_events(PHO_USDC_POOL, 0); + let result = PhoenixXykExtractor.extract(&rows).unwrap(); + + assert_eq!(result.trades.len(), 1); + let trade = &result.trades[0]; + assert_eq!(trade.contract_id, PHO_USDC_POOL); + assert_eq!(trade.venue, Venue::Phoenix); + assert_eq!(trade.amount_in, 11659417676); + assert_eq!(trade.amount_out, 1857322909); + } + + #[test] + fn xyk_extractor_rejects_fewer_than_8_rows() { + let rows = make_phoenix_xyk_events(XLM_USDC_POOL, 0); + let result = PhoenixXykExtractor.extract(&rows[..5]); + assert!(result.is_err()); + } + + #[test] + fn xyk_extractor_tolerates_unordered_fields() { + let mut rows = make_phoenix_xyk_events(XLM_USDC_POOL, 0); + rows.swap(1, 4); + let result = PhoenixXykExtractor.extract(&rows).unwrap(); + + let trade = &result.trades[0]; + assert_eq!(trade.token_in, XLM_SAC); + assert_eq!(trade.token_out, USDC_SAC); + } +} diff --git a/packages/sdex-backfill/src/bucket.rs b/packages/sdex-backfill/src/bucket.rs index 2504ba4b..83fde541 100644 --- a/packages/sdex-backfill/src/bucket.rs +++ b/packages/sdex-backfill/src/bucket.rs @@ -92,10 +92,14 @@ impl CandleAccumulator { } pub fn flush_all(&mut self) -> Vec { - let mut flushed: Vec = self.candles.drain().map(|(_, mut c)| { - finalise_vwap(&mut c); - c - }).collect(); + let mut flushed: Vec = self + .candles + .drain() + .map(|(_, mut c)| { + finalise_vwap(&mut c); + c + }) + .collect(); flushed.sort_by_key(|c| (c.minute_start, c.asset_id, c.quote_asset_id)); flushed } diff --git a/packages/sdex-backfill/src/canonical.rs b/packages/sdex-backfill/src/canonical.rs index dd18899a..bfe80d27 100644 --- a/packages/sdex-backfill/src/canonical.rs +++ b/packages/sdex-backfill/src/canonical.rs @@ -79,7 +79,10 @@ impl AssetRegistry { next_id = next_id.max(id + 1); by_identity.insert(identity, id); } - Self { by_identity, next_id } + Self { + by_identity, + next_id, + } } pub fn get_or_assign(&mut self, identity: &AssetIdentity) -> u32 { diff --git a/packages/sdex-backfill/src/filter.rs b/packages/sdex-backfill/src/filter.rs index 37cbb55f..59196db5 100644 --- a/packages/sdex-backfill/src/filter.rs +++ b/packages/sdex-backfill/src/filter.rs @@ -46,13 +46,9 @@ pub fn extract_trades(lcm: &LedgerCloseMeta) -> Vec { }; for (claim_idx, claim) in claims.iter().enumerate() { - if let Some(trade) = claim_to_raw_trade( - claim, - sequence, - closed_at, - op_idx as u16, - claim_idx as u16, - ) { + if let Some(trade) = + claim_to_raw_trade(claim, sequence, closed_at, op_idx as u16, claim_idx as u16) + { trades.push(trade); } } @@ -65,9 +61,7 @@ pub fn extract_trades(lcm: &LedgerCloseMeta) -> Vec { fn extract_claims(tr: &OperationResultTr) -> &[ClaimAtom] { use OperationResultTr::*; match tr { - ManageSellOffer(stellar_xdr::curr::ManageSellOfferResult::Success(s)) => { - &s.offers_claimed - } + ManageSellOffer(stellar_xdr::curr::ManageSellOfferResult::Success(s)) => &s.offers_claimed, ManageBuyOffer(stellar_xdr::curr::ManageBuyOfferResult::Success(s)) => &s.offers_claimed, CreatePassiveSellOffer(stellar_xdr::curr::ManageSellOfferResult::Success(s)) => { &s.offers_claimed diff --git a/packages/sdex-backfill/src/ingest.rs b/packages/sdex-backfill/src/ingest.rs index 3eeb04fa..95856d0f 100644 --- a/packages/sdex-backfill/src/ingest.rs +++ b/packages/sdex-backfill/src/ingest.rs @@ -32,7 +32,10 @@ pub async fn index_partition( registry: &mut AssetRegistry, ) -> Result { let (first, last) = partition.clamped(range_start, range_end); - info!(partition = partition.start, first, last, "partition indexing started"); + info!( + partition = partition.start, + first, last, "partition indexing started" + ); let wall_start = Instant::now(); let mut stats = PartitionStats::default(); @@ -105,15 +108,9 @@ pub async fn index_partition( fn ledger_minute(lcm: &stellar_xdr::curr::LedgerCloseMeta) -> u32 { let closed_at = match lcm { - stellar_xdr::curr::LedgerCloseMeta::V0(v) => { - v.ledger_header.header.scp_value.close_time.0 - } - stellar_xdr::curr::LedgerCloseMeta::V1(v) => { - v.ledger_header.header.scp_value.close_time.0 - } - stellar_xdr::curr::LedgerCloseMeta::V2(v) => { - v.ledger_header.header.scp_value.close_time.0 - } + stellar_xdr::curr::LedgerCloseMeta::V0(v) => v.ledger_header.header.scp_value.close_time.0, + stellar_xdr::curr::LedgerCloseMeta::V1(v) => v.ledger_header.header.scp_value.close_time.0, + stellar_xdr::curr::LedgerCloseMeta::V2(v) => v.ledger_header.header.scp_value.close_time.0, }; ((closed_at as u32) / 60) * 60 } diff --git a/packages/sdex-backfill/src/obs.rs b/packages/sdex-backfill/src/obs.rs index ec75df85..fbde4d0b 100644 --- a/packages/sdex-backfill/src/obs.rs +++ b/packages/sdex-backfill/src/obs.rs @@ -2,11 +2,9 @@ use tracing_subscriber::{EnvFilter, fmt, prelude::*}; pub fn init(verbose: bool) { let filter = if verbose { - EnvFilter::try_from_default_env() - .unwrap_or_else(|_| EnvFilter::new("sdex_backfill=info")) + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("sdex_backfill=info")) } else { - EnvFilter::try_from_default_env() - .unwrap_or_else(|_| EnvFilter::new("sdex_backfill=warn")) + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("sdex_backfill=warn")) }; tracing_subscriber::registry() diff --git a/packages/sdex-backfill/src/run.rs b/packages/sdex-backfill/src/run.rs index c806bb04..192254bd 100644 --- a/packages/sdex-backfill/src/run.rs +++ b/packages/sdex-backfill/src/run.rs @@ -24,7 +24,9 @@ pub async fn execute( tokio::fs::create_dir_all(temp_dir).await?; preflight_aws().await; - sink.preflight().await.unwrap_or_else(|e| panic!("pre-flight: sink unreachable: {e}")); + sink.preflight() + .await + .unwrap_or_else(|e| panic!("pre-flight: sink unreachable: {e}")); info!("pre-flight: all checks passed"); let partitions = partitions_for_range(start, end); @@ -41,7 +43,8 @@ pub async fn execute( .collect(); info!( - start, end, + start, + end, total_partitions = partitions.len(), already_done = partitions.len() - todo.len(), to_process = todo.len(), @@ -65,7 +68,10 @@ pub async fn execute( SyncOutcome::Complete ); if !current_complete { - warn!(partition = todo[0].start, "first partition S3 incomplete — will skip"); + warn!( + partition = todo[0].start, + "first partition S3 incomplete — will skip" + ); partitions_skipped_s3 += 1; } @@ -74,9 +80,9 @@ pub async fn execute( if let Some(next) = todo.get(i + 1) { let next = (*next).clone(); let temp = temp_dir.to_path_buf(); - Some(tokio::spawn(async move { - sync_partition(&next, &temp).await - })) + Some(tokio::spawn( + async move { sync_partition(&next, &temp).await }, + )) } else { None }; @@ -99,7 +105,10 @@ pub async fn execute( totals.candles_written += stats.candles_written; totals.total_bytes += stats.total_bytes; } else { - info!(partition = partition.start, "skipping S3-incomplete partition"); + info!( + partition = partition.start, + "skipping S3-incomplete partition" + ); } if !keep_partitions { diff --git a/packages/sdex-backfill/src/sink.rs b/packages/sdex-backfill/src/sink.rs index 8e444281..b06759bc 100644 --- a/packages/sdex-backfill/src/sink.rs +++ b/packages/sdex-backfill/src/sink.rs @@ -25,14 +25,15 @@ impl Sink { } pub async fn preflight(&self) -> Result<(), BackfillError> { - self.client - .query("SELECT 1") - .execute() - .await?; + self.client.query("SELECT 1").execute().await?; Ok(()) } - pub async fn load_completed(&self, start: u32, end: u32) -> Result, BackfillError> { + pub async fn load_completed( + &self, + start: u32, + end: u32, + ) -> Result, BackfillError> { let rows = self .client .query( @@ -46,7 +47,8 @@ impl Sink { let set: HashSet = rows.into_iter().collect(); info!( - start, end, + start, + end, completed = set.len(), "loaded completed ledgers from backfill_sdex_ledgers" ); @@ -56,9 +58,7 @@ impl Sink { pub async fn load_assets(&self) -> Result, BackfillError> { let rows = self .client - .query( - "SELECT asset_id, asset_code, issuer_address FROM prices.assets", - ) + .query("SELECT asset_id, asset_code, issuer_address FROM prices.assets") .fetch_all::() .await?; @@ -77,7 +77,10 @@ impl Sink { }) .collect(); - info!(existing_assets = assets.len(), "loaded asset registry from ClickHouse"); + info!( + existing_assets = assets.len(), + "loaded asset registry from ClickHouse" + ); Ok(assets) } @@ -86,9 +89,7 @@ impl Sink { return Ok(()); } - let mut insert = self - .client - .insert("prices.price_ohlcv_1m")?; + let mut insert = self.client.insert("prices.price_ohlcv_1m")?; for candle in candles { insert @@ -119,9 +120,7 @@ impl Sink { for (identity, &id) in registry.assets() { let (asset_code, asset_type, issuer_address) = match identity { AssetIdentity::Native => ("XLM".to_string(), "classic", String::new()), - AssetIdentity::Credit { code, issuer } => { - (code.clone(), "classic", issuer.clone()) - } + AssetIdentity::Credit { code, issuer } => (code.clone(), "classic", issuer.clone()), }; insert diff --git a/packages/sdex-backfill/src/sync.rs b/packages/sdex-backfill/src/sync.rs index 030721d5..d84cf506 100644 --- a/packages/sdex-backfill/src/sync.rs +++ b/packages/sdex-backfill/src/sync.rs @@ -32,7 +32,8 @@ pub async fn sync_partition( if let Some((file_count, total_bytes)) = local_partition_complete(&local).await? { info!( partition = partition.start, - file_count, total_bytes, + file_count, + total_bytes, "partition local folder already complete — skipping aws s3 sync" ); return Ok(SyncOutcome::Complete); @@ -47,7 +48,8 @@ pub async fn sync_partition( info!( partition = partition.start, sync_duration_ms = duration.as_millis(), - file_count, total_bytes, + file_count, + total_bytes, "partition sync complete" ); return Ok(SyncOutcome::Complete); @@ -78,7 +80,10 @@ pub async fn sync_partition( run_sync_once(partition, &local).await?; let (file_count_retry, _) = dir_stats(&local).await?; if file_count_retry == PARTITION_SIZE as usize { - info!(partition = partition.start, "partition sync complete after retry"); + info!( + partition = partition.start, + "partition sync complete after retry" + ); return Ok(SyncOutcome::Complete); } @@ -113,10 +118,7 @@ async fn local_partition_complete(dir: &Path) -> Result, Ba } } -async fn run_sync_with_retry( - partition: &Partition, - local: &Path, -) -> Result<(), BackfillError> { +async fn run_sync_with_retry(partition: &Partition, local: &Path) -> Result<(), BackfillError> { let mut delay = RETRY_BASE_DELAY; for attempt in 1..=RETRY_ATTEMPTS { match run_sync_once(partition, local).await {