diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 5c077193..0200eed8 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -74,7 +74,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -85,7 +85,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -239,6 +239,16 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[package]] +name = "assert-json-diff" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e4f2b81832e72834d7518d8487a0396a28cc408186a2e8854c0f98011faf12" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "async-trait" version = "0.1.92" @@ -250,6 +260,12 @@ dependencies = [ "syn 3.0.4", ] +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "autocfg" version = "1.5.1" @@ -359,6 +375,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + [[package]] name = "cfg_eval" version = "0.1.2" @@ -623,6 +645,24 @@ version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" +[[package]] +name = "deadpool" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b" +dependencies = [ + "deadpool-runtime", + "lazy_static", + "num_cpus", + "tokio", +] + +[[package]] +name = "deadpool-runtime" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b" + [[package]] name = "defmt" version = "1.1.1" @@ -707,6 +747,17 @@ dependencies = [ "crypto-common 0.2.2", ] +[[package]] +name = "displaydoc" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "downcast-rs" version = "1.2.1" @@ -864,7 +915,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -913,6 +964,30 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -920,6 +995,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -939,6 +1015,12 @@ dependencies = [ "futures-util", ] +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + [[package]] name = "futures-macro" version = "0.3.34" @@ -968,10 +1050,13 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", + "futures-io", "futures-macro", "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -1019,9 +1104,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 6.0.0", "rand_core 0.10.1", + "wasm-bindgen", ] [[package]] @@ -1035,6 +1122,25 @@ dependencies = [ "subtle", ] +[[package]] +name = "h2" +version = "0.4.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef8e5e5a340588f4452631496976cf8636d4a7ecf600239fdc27615d2530bc16" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap 2.14.0", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "hash32" version = "0.3.1" @@ -1081,6 +1187,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" + [[package]] name = "hex" version = "0.4.3" @@ -1115,12 +1227,41 @@ dependencies = [ "itoa", ] +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + [[package]] name = "httparse" version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hybrid-array" version = "0.4.14" @@ -1130,6 +1271,67 @@ dependencies = [ "typenum", ] +[[package]] +name = "hyper" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "h2", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-rustls" +version = "0.27.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "tokio", + "tokio-rustls", + "tower-service", + "webpki-roots", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "base64", + "bytes", + "futures-channel", + "futures-util", + "http", + "http-body", + "hyper", + "ipnet", + "libc", + "percent-encoding", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", +] + [[package]] name = "iana-time-zone" version = "0.1.65" @@ -1154,12 +1356,116 @@ dependencies = [ "cc", ] +[[package]] +name = "icu_collections" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0" + +[[package]] +name = "icu_properties" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148" +dependencies = [ + "displaydoc", + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa" + +[[package]] +name = "icu_provider" +version = "2.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d27bbb9d3abbefac45d55f647c9de1d44aafcd1186eb91879afef17c396c3e73" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + [[package]] name = "ident_case" version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + [[package]] name = "indexmap" version = "1.9.3" @@ -1189,6 +1495,12 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e04e2fd2b8188ea827b32ef11de88377086d690286ab35747ef7f9bf3ccb590" +[[package]] +name = "ipnet" +version = "2.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a756c3fac73139e83f14c2d742155dd2b78d3ee56597b419a0579b7bdd6dd78" + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -1313,6 +1625,12 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "litemap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" + [[package]] name = "lock_api" version = "0.4.14" @@ -1328,6 +1646,12 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "macro-string" version = "0.1.4" @@ -1353,7 +1677,7 @@ checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" dependencies = [ "libc", "wasi", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1401,6 +1725,16 @@ dependencies = [ "autocfg", ] +[[package]] +name = "num_cpus" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" +dependencies = [ + "hermit-abi", + "libc", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -1552,6 +1886,15 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "potential_utf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" +dependencies = [ + "zerovec", +] + [[package]] name = "powerfmt" version = "0.2.0" @@ -1630,6 +1973,62 @@ dependencies = [ "thiserror 1.0.69", ] +[[package]] +name = "quinn" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror 2.0.20", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83" +dependencies = [ + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand 0.10.2", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror 2.0.20", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + [[package]] name = "quote" version = "1.0.47" @@ -1727,6 +2126,15 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core 0.10.1", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1786,30 +2194,129 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" [[package]] -name = "rfc6979" -version = "0.4.0" +name = "reqwest" +version = "0.12.28" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ - "hmac", - "subtle", -] - -[[package]] -name = "rustc_version" -version = "0.4.1" + "base64", + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "webpki-roots", +] + +[[package]] +name = "rfc6979" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2" +dependencies = [ + "hmac", + "subtle", +] + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + +[[package]] +name = "rustc_version" +version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" dependencies = [ "semver", ] +[[package]] +name = "rustls" +version = "0.23.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "web-time", + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + [[package]] name = "schemars" version = "0.8.22" @@ -1913,6 +2420,18 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "serde_with" version = "3.22.0" @@ -2033,7 +2552,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -2279,6 +2798,7 @@ dependencies = [ "opentelemetry-stdout", "opentelemetry_sdk", "prometheus", + "reqwest", "serde", "serde_json", "soroban-sdk", @@ -2290,6 +2810,7 @@ dependencies = [ "tracing", "tracing-futures", "tracing-opentelemetry", + "wiremock", ] [[package]] @@ -2380,6 +2901,26 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] + +[[package]] +name = "synstructure" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "thiserror" version = "1.0.69" @@ -2459,6 +3000,16 @@ dependencies = [ "time-core", ] +[[package]] +name = "tinystr" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643" +dependencies = [ + "displaydoc", + "zerovec", +] + [[package]] name = "tinyvec" version = "1.12.0" @@ -2488,7 +3039,7 @@ dependencies = [ "signal-hook-registry", "socket2", "tokio-macros", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -2502,6 +3053,16 @@ dependencies = [ "syn 3.0.4", ] +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-tungstenite" version = "0.30.0" @@ -2514,6 +3075,65 @@ dependencies = [ "tungstenite", ] +[[package]] +name = "tokio-util" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", +] + +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags 2.13.1", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + [[package]] name = "tracing" version = "0.1.44" @@ -2594,6 +3214,12 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "tungstenite" version = "0.30.0" @@ -2622,6 +3248,30 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + [[package]] name = "utf8parse" version = "0.2.2" @@ -2651,6 +3301,15 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -2679,6 +3338,16 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.77" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b7777d5cc23d0e91404e53ce2d5e8ec7acae3026b16233dba62cd3246457950" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.127" @@ -2748,6 +3417,16 @@ dependencies = [ "indexmap-nostd", ] +[[package]] +name = "web-sys" +version = "0.3.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c435338968042f4f59a557f690a253676d47ce13ceb55d70100e7facf6620a30" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "web-time" version = "1.1.0" @@ -2758,6 +3437,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-roots" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "windows-core" version = "0.62.2" @@ -2817,6 +3505,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -2826,12 +3523,128 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "wiremock" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08db1edfb05d9b3c1542e521aea074442088292f00b5f28e435c714a98f85031" +dependencies = [ + "assert-json-diff", + "base64", + "deadpool", + "futures", + "http", + "http-body-util", + "hyper", + "hyper-util", + "log", + "once_cell", + "regex", + "serde", + "serde_json", + "tokio", + "url", +] + [[package]] name = "wit-bindgen" version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + +[[package]] +name = "yoke" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + [[package]] name = "zerocopy" version = "0.8.56" @@ -2852,6 +3665,27 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + [[package]] name = "zeroize" version = "1.9.0" @@ -2872,6 +3706,39 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "zerotrie" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0464e17806c1d976d5cba29399c7f08e516e279e2ba493f63123b5fca67dd8" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "zmij" version = "1.0.23" diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 5f15f0b4..a44b90a9 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -24,7 +24,9 @@ opentelemetry = "0.32.0" opentelemetry_sdk = "0.32.1" tracing-opentelemetry = "0.33.0" opentelemetry-stdout = "0.32.0" +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } [dev-dependencies] soroban-sdk = { version = "=27.0.6", features = ["testutils"] } stellar-insights = { path = "../contracts/stellar_insights", features = ["testutils"] } +wiremock = "0.6" diff --git a/backend/src/ingestion/fetch.rs b/backend/src/ingestion/fetch.rs new file mode 100644 index 00000000..7aceedde --- /dev/null +++ b/backend/src/ingestion/fetch.rs @@ -0,0 +1,360 @@ +//! Sources of ledger data for the ingestion pipeline: a real Horizon-backed +//! wrapper, and a deterministic in-memory fake used to drive the pipeline in +//! tests (including simulated reorgs) without a live network. + +use async_trait::async_trait; +use serde::Deserialize; +use std::collections::BTreeMap; + +use crate::network::NetworkClient; +use crate::snapshot::generator::RawSnapshotRow; + +use super::{fnv1a_hex, IngestionError}; + +/// One fetched ledger: its position in the chain, its hash linkage, and the +/// derived rows this pipeline computed from it. +#[derive(Clone, Debug, PartialEq)] +pub struct FetchedLedger { + pub sequence: u64, + pub hash: String, + pub prev_hash: String, + pub rows: Vec, +} + +/// A source of ledger data, fetchable one sequence at a time. +/// +/// `fetch_ledger` returning `Ok(None)` means the ledger has not closed yet — +/// distinct from an error, since "not yet available" is the pipeline's +/// normal steady state once it has caught up to the chain tip. +#[async_trait] +pub trait LedgerSource: Send + Sync { + async fn fetch_ledger(&self, sequence: u64) -> Result, IngestionError>; +} + +/// Fetches ledgers from a Horizon-compatible REST API. +/// +/// Derives one coarse, ledger-level [`RawSnapshotRow`] per ledger from its +/// transaction success/failure counts. This is intentionally a proxy, not +/// the full per-corridor asset-pair breakdown a real analytics engine would +/// compute from individual payment operations — that computation doesn't +/// exist in this codebase yet and is out of scope here; this wrapper's job +/// is to prove the checkpointing and reorg-handling machinery around it +/// works, with a real (if coarse) row flowing through it end to end. +pub struct HorizonLedgerSource { + client: reqwest::Client, + base_url: String, + network_label: String, +} + +impl HorizonLedgerSource { + pub fn new(network: &NetworkClient) -> Self { + Self { + client: reqwest::Client::new(), + base_url: network.horizon_url.clone(), + network_label: network.network.to_string(), + } + } + + #[cfg(test)] + fn with_base_url(base_url: String, network_label: String) -> Self { + Self { + client: reqwest::Client::new(), + base_url, + network_label, + } + } +} + +#[derive(Debug, Deserialize)] +struct HorizonLedgerResponse { + sequence: u64, + hash: String, + prev_hash: String, + successful_transaction_count: u64, + failed_transaction_count: u64, +} + +#[async_trait] +impl LedgerSource for HorizonLedgerSource { + async fn fetch_ledger(&self, sequence: u64) -> Result, IngestionError> { + let url = format!("{}/ledgers/{sequence}", self.base_url.trim_end_matches('/')); + + let response = + self.client + .get(&url) + .send() + .await + .map_err(|error| IngestionError::Fetch { + sequence, + message: error.to_string(), + })?; + + if response.status() == reqwest::StatusCode::NOT_FOUND { + return Ok(None); + } + + if !response.status().is_success() { + return Err(IngestionError::Fetch { + sequence, + message: format!("Horizon returned HTTP {}", response.status()), + }); + } + + let body: HorizonLedgerResponse = + response + .json() + .await + .map_err(|error| IngestionError::Fetch { + sequence, + message: format!("failed to parse Horizon ledger response: {error}"), + })?; + + let total = body.successful_transaction_count + body.failed_transaction_count; + let reliability = if total == 0 { + 1.0 + } else { + body.successful_transaction_count as f64 / total as f64 + }; + + let row = RawSnapshotRow { + ledger_sequence: body.sequence, + corridor: "__ledger_aggregate__".to_string(), + source: self.network_label.clone(), + reliability, + volume: body.successful_transaction_count as f64, + latency_ms: 0.0, + }; + + Ok(Some(FetchedLedger { + sequence: body.sequence, + hash: body.hash, + prev_hash: body.prev_hash, + rows: vec![row], + })) + } +} + +#[derive(Clone, Debug)] +struct ChainLedger { + hash: String, + prev_hash: String, + rows: Vec, +} + +/// Deterministic, in-memory chain of ledgers for tests. +/// +/// Ledgers are appended one at a time on top of the current tip, with each +/// ledger's hash derived from its own tag plus its predecessor's hash — so +/// [`FakeLedgerSource::fork_from`] naturally produces a chain whose hash +/// linkage genuinely diverges from the original at and after the fork +/// point, exactly like a real reorg, rather than a canned response the +/// pipeline can't meaningfully detect anything from. +#[derive(Default)] +pub struct FakeLedgerSource { + ledgers: tokio::sync::Mutex>, +} + +const GENESIS_HASH: &str = "genesis"; + +impl FakeLedgerSource { + pub fn new() -> Self { + Self::default() + } + + fn hash_of(sequence: u64, prev_hash: &str, tag: &str) -> String { + fnv1a_hex(&format!("{sequence}:{prev_hash}:{tag}")) + } + + /// Appends one ledger on top of the current tip and returns its + /// sequence number. `tag` only needs to be distinct per fork branch — + /// its value has no meaning beyond feeding the hash. + pub async fn append(&self, tag: &str, rows: Vec) -> u64 { + let mut guard = self.ledgers.lock().await; + let (tip_sequence, tip_hash) = guard + .iter() + .next_back() + .map(|(seq, ledger)| (*seq, ledger.hash.clone())) + .unwrap_or((0, GENESIS_HASH.to_string())); + + let sequence = tip_sequence + 1; + let hash = Self::hash_of(sequence, &tip_hash, tag); + guard.insert( + sequence, + ChainLedger { + hash, + prev_hash: tip_hash, + rows, + }, + ); + sequence + } + + /// Simulates a reorg: discards every ledger at or after `from_sequence` + /// and replaces them with a new fork built from `replacement`, chained + /// on top of whatever ledger remains at `from_sequence - 1`. + pub async fn fork_from( + &self, + from_sequence: u64, + replacement: Vec<(&str, Vec)>, + ) { + let mut guard = self.ledgers.lock().await; + guard.retain(|seq, _| *seq < from_sequence); + + let mut tip_hash = guard + .get(&(from_sequence - 1)) + .map(|ledger| ledger.hash.clone()) + .unwrap_or_else(|| GENESIS_HASH.to_string()); + + for (index, (tag, rows)) in replacement.into_iter().enumerate() { + let sequence = from_sequence + index as u64; + let hash = Self::hash_of(sequence, &tip_hash, tag); + guard.insert( + sequence, + ChainLedger { + hash: hash.clone(), + prev_hash: tip_hash, + rows, + }, + ); + tip_hash = hash; + } + } +} + +#[async_trait] +impl LedgerSource for FakeLedgerSource { + async fn fetch_ledger(&self, sequence: u64) -> Result, IngestionError> { + Ok(self + .ledgers + .lock() + .await + .get(&sequence) + .map(|ledger| FetchedLedger { + sequence, + hash: ledger.hash.clone(), + prev_hash: ledger.prev_hash.clone(), + rows: ledger.rows.clone(), + })) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + fn row(corridor: &str) -> RawSnapshotRow { + RawSnapshotRow { + ledger_sequence: 0, + corridor: corridor.to_string(), + source: "testnet".to_string(), + reliability: 1.0, + volume: 1.0, + latency_ms: 1.0, + } + } + + #[tokio::test] + async fn fake_source_chains_appended_ledgers() { + let source = FakeLedgerSource::new(); + let seq1 = source.append("a", vec![row("eur/usd")]).await; + let seq2 = source.append("a", vec![row("eur/usd")]).await; + + assert_eq!((seq1, seq2), (1, 2)); + + let ledger1 = source.fetch_ledger(1).await.unwrap().unwrap(); + let ledger2 = source.fetch_ledger(2).await.unwrap().unwrap(); + + assert_eq!(ledger1.prev_hash, GENESIS_HASH); + assert_eq!(ledger2.prev_hash, ledger1.hash); + } + + #[tokio::test] + async fn fetching_beyond_the_tip_returns_none() { + let source = FakeLedgerSource::new(); + source.append("a", vec![row("eur/usd")]).await; + + assert_eq!(source.fetch_ledger(2).await.unwrap(), None); + } + + #[tokio::test] + async fn fork_from_changes_hash_linkage_from_the_fork_point() { + let source = FakeLedgerSource::new(); + source.append("a", vec![row("eur/usd")]).await; // 1 + let before = source.append("a", vec![row("eur/usd")]).await; // 2 + source.append("a", vec![row("eur/usd")]).await; // 3 + + let ledger1_before = source.fetch_ledger(1).await.unwrap().unwrap(); + + source + .fork_from( + 2, + vec![("b", vec![row("gbp/usd")]), ("b", vec![row("gbp/usd")])], + ) + .await; + + let ledger1_after = source.fetch_ledger(1).await.unwrap().unwrap(); + let ledger2_after = source.fetch_ledger(2).await.unwrap().unwrap(); + let ledger3_after = source.fetch_ledger(3).await.unwrap().unwrap(); + + // Ledgers before the fork point are untouched. + assert_eq!(ledger1_before, ledger1_after); + assert_eq!(before, 2); + // The forked ledgers carry the new fork's rows and a hash chain + // that no longer matches the original branch. + assert_eq!(ledger2_after.rows[0].corridor, "gbp/usd"); + assert_eq!(ledger3_after.prev_hash, ledger2_after.hash); + } + + #[tokio::test] + async fn horizon_source_parses_a_real_response_shape() { + let mock_server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/ledgers/100")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "sequence": 100, + "hash": "deadbeef", + "prev_hash": "cafebabe", + "successful_transaction_count": 9, + "failed_transaction_count": 1 + }))) + .mount(&mock_server) + .await; + + let source = HorizonLedgerSource::with_base_url(mock_server.uri(), "testnet".to_string()); + let fetched = source.fetch_ledger(100).await.unwrap().unwrap(); + + assert_eq!(fetched.sequence, 100); + assert_eq!(fetched.hash, "deadbeef"); + assert_eq!(fetched.prev_hash, "cafebabe"); + assert_eq!(fetched.rows.len(), 1); + assert!((fetched.rows[0].reliability - 0.9).abs() < f64::EPSILON); + } + + #[tokio::test] + async fn horizon_source_treats_404_as_not_yet_closed() { + let mock_server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/ledgers/999")) + .respond_with(ResponseTemplate::new(404)) + .mount(&mock_server) + .await; + + let source = HorizonLedgerSource::with_base_url(mock_server.uri(), "testnet".to_string()); + assert_eq!(source.fetch_ledger(999).await.unwrap(), None); + } + + #[tokio::test] + async fn horizon_source_surfaces_server_errors() { + let mock_server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/ledgers/1")) + .respond_with(ResponseTemplate::new(500)) + .mount(&mock_server) + .await; + + let source = HorizonLedgerSource::with_base_url(mock_server.uri(), "testnet".to_string()); + assert!(source.fetch_ledger(1).await.is_err()); + } +} diff --git a/backend/src/ingestion/mod.rs b/backend/src/ingestion/mod.rs new file mode 100644 index 00000000..6c70e179 --- /dev/null +++ b/backend/src/ingestion/mod.rs @@ -0,0 +1,421 @@ +//! Ledger ingestion: pulls ledger data from Horizon/RPC and writes derived +//! corridor/anchor/network rows into a database in a way that survives +//! being re-run over the same range twice, and survives a reorg. +//! +//! The checkpoint carries `(ledger_sequence, ledger_hash)`, not sequence +//! alone, so a reorg is detectable directly — the next fetched ledger's +//! `prev_hash` stops matching the checkpoint's hash — rather than inferred +//! indirectly from a gap or a downstream inconsistency. Every derived write +//! is a full replace keyed on `(ledger_sequence, entity_id)` +//! ([`upsert::DerivedStore`]), so re-ingesting a range, or rolling one back +//! after a reorg and re-ingesting it, converges on the same stored state +//! instead of accumulating duplicates or leaving orphaned rows from an +//! abandoned fork. + +pub mod fetch; +pub mod reconcile; +pub mod upsert; +pub mod watermark; + +use std::time::Duration; + +pub use fetch::{FakeLedgerSource, FetchedLedger, HorizonLedgerSource, LedgerSource}; +pub use reconcile::{DriftAlert, DriftAlertSink, InMemoryDriftAlertSink, ReconciliationCheck}; +pub use upsert::{DerivedStore, EntityId, InMemoryDerivedStore}; +pub use watermark::{InMemoryWatermarkStore, Watermark, WatermarkStore}; + +#[derive(Debug, thiserror::Error)] +pub enum IngestionError { + #[error("failed to fetch ledger {sequence}: {message}")] + Fetch { sequence: u64, message: String }, + #[error("watermark store failure: {0}")] + Watermark(String), + #[error("derived store failure: {0}")] + DerivedStore(String), + #[error("alert sink failure: {0}")] + Alert(String), + #[error("reorg rollback target ledger {sequence} is unavailable")] + ReorgRollbackTargetUnavailable { sequence: u64 }, +} + +/// Minimal, dependency-free FNV-1a hash, used where a deterministic, +/// cross-call-stable digest is needed but no cryptographic property is +/// required. `std`'s `DefaultHasher` is explicitly documented as not +/// suitable for this: its algorithm is unspecified and may change between +/// compilations, which would make [`FakeLedgerSource`]'s hash chain and +/// the reconciliation checksum unreliable to reason about. +pub(crate) fn fnv1a_u64(input: &str) -> u64 { + const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325; + const FNV_PRIME: u64 = 0x100000001b3; + + let mut hash = FNV_OFFSET_BASIS; + for byte in input.as_bytes() { + hash ^= u64::from(*byte); + hash = hash.wrapping_mul(FNV_PRIME); + } + hash +} + +pub(crate) fn fnv1a_hex(input: &str) -> String { + format!("{:016x}", fnv1a_u64(input)) +} + +/// The result of one [`IngestionPipeline::run_once`] call. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum IngestOutcome { + /// The next ledger has not closed yet; nothing to do this tick. + Idle, + /// One ledger was fetched and its rows upserted. + Ingested { sequence: u64 }, + /// A reorg was detected and handled: the watermark was rolled back to + /// `rolled_back_to` (or ingestion reset to start from scratch, see + /// [`IngestionPipeline::start_sequence`]) and every derived row from + /// `resumed_from` onward was invalidated. The caller's next `run_once` + /// resumes forward ingestion from `resumed_from`. + ReorgHandled { + rolled_back_to: u64, + resumed_from: u64, + }, +} + +/// Orchestrates checkpointed, reorg-aware ledger ingestion. +pub struct IngestionPipeline { + source: L, + watermark: W, + derived: D, + /// First ledger sequence this pipeline ever ingests. + start_sequence: u64, + /// How many ledgers back to roll on a detected reorg, on the assumption + /// that a ledger this far behind the previous tip is now stable. This + /// is a real assumption, not a proof — a reorg deeper than this will be + /// only partially corrected — but it matches how block-explorer-style + /// indexers commonly bound reorg handling in practice. + confirmation_depth: u64, +} + +impl IngestionPipeline +where + L: LedgerSource, + W: WatermarkStore, + D: DerivedStore, +{ + pub fn new( + source: L, + watermark: W, + derived: D, + start_sequence: u64, + confirmation_depth: u64, + ) -> Self { + Self { + source, + watermark, + derived, + start_sequence: start_sequence.max(1), + confirmation_depth: confirmation_depth.max(1), + } + } + + /// The ledger source this pipeline reads from. + pub fn source(&self) -> &L { + &self.source + } + + /// The checkpoint store, so a caller can inspect ingestion progress or + /// share it with a [`reconcile::ReconciliationCheck`] over the same + /// pipeline. + pub fn watermark_store(&self) -> &W { + &self.watermark + } + + /// The derived-row store, so a caller can actually read what has been + /// ingested (e.g. to serve it from an API). + pub fn derived_store(&self) -> &D { + &self.derived + } + + /// Advances ingestion by at most one ledger. + pub async fn run_once(&self) -> Result { + let current = self.watermark.load().await?; + let next_sequence = match ¤t { + Some(watermark) => watermark.sequence + 1, + None => self.start_sequence, + }; + + let Some(fetched) = self.source.fetch_ledger(next_sequence).await? else { + return Ok(IngestOutcome::Idle); + }; + + if let Some(watermark) = ¤t { + if fetched.prev_hash != watermark.hash { + return self.handle_reorg(watermark.clone()).await; + } + } + + self.derived.upsert(fetched.sequence, fetched.rows).await?; + self.watermark + .save(Watermark { + sequence: fetched.sequence, + hash: fetched.hash, + }) + .await?; + + Ok(IngestOutcome::Ingested { + sequence: fetched.sequence, + }) + } + + async fn handle_reorg(&self, stale: Watermark) -> Result { + let floor = self.start_sequence.saturating_sub(1); + let rollback_target = stale + .sequence + .saturating_sub(self.confirmation_depth) + .max(floor); + + if rollback_target <= floor { + self.derived.invalidate_from(self.start_sequence).await?; + self.watermark.clear().await?; + return Ok(IngestOutcome::ReorgHandled { + rolled_back_to: floor, + resumed_from: self.start_sequence, + }); + } + + let anchor = self.source.fetch_ledger(rollback_target).await?.ok_or( + IngestionError::ReorgRollbackTargetUnavailable { + sequence: rollback_target, + }, + )?; + + self.derived.invalidate_from(rollback_target + 1).await?; + self.watermark + .save(Watermark { + sequence: anchor.sequence, + hash: anchor.hash, + }) + .await?; + + Ok(IngestOutcome::ReorgHandled { + rolled_back_to: rollback_target, + resumed_from: rollback_target + 1, + }) + } + + /// Runs [`Self::run_once`] on a fixed interval, forever. A failed tick + /// is logged and retried next interval rather than ending the loop — + /// unlike a `?`-propagating `loop`, one transient fetch failure does + /// not permanently kill background ingestion. + pub async fn run_forever(&self, poll_interval: Duration) { + let mut ticker = tokio::time::interval(poll_interval); + loop { + ticker.tick().await; + match self.run_once().await { + Ok(IngestOutcome::Idle) => {} + Ok(outcome) => tracing::debug!(?outcome, "ingestion tick"), + Err(error) => { + tracing::warn!(error = %error, "ingestion tick failed; retrying next tick"); + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::snapshot::generator::RawSnapshotRow; + + fn row(sequence: u64, corridor: &str) -> RawSnapshotRow { + RawSnapshotRow { + ledger_sequence: sequence, + corridor: corridor.to_string(), + source: "testnet".to_string(), + reliability: 1.0, + volume: 1.0, + latency_ms: 1.0, + } + } + + fn pipeline( + source: FakeLedgerSource, + ) -> IngestionPipeline { + IngestionPipeline::new( + source, + InMemoryWatermarkStore::new(), + InMemoryDerivedStore::new(), + 1, + 3, + ) + } + + #[tokio::test] + async fn ingests_ledgers_one_at_a_time_and_advances_the_watermark() { + let source = FakeLedgerSource::new(); + source.append("a", vec![row(1, "eur/usd")]).await; + source.append("a", vec![row(2, "eur/usd")]).await; + + let pipeline = pipeline(source); + + assert_eq!( + pipeline.run_once().await.unwrap(), + IngestOutcome::Ingested { sequence: 1 } + ); + assert_eq!( + pipeline.run_once().await.unwrap(), + IngestOutcome::Ingested { sequence: 2 } + ); + assert_eq!(pipeline.run_once().await.unwrap(), IngestOutcome::Idle); + + let rows = pipeline.derived_store().rows_in_range(1, 2).await.unwrap(); + assert_eq!(rows.len(), 2); + } + + #[tokio::test] + async fn re_running_over_an_already_processed_range_is_byte_identical() { + let source = FakeLedgerSource::new(); + source.append("a", vec![row(1, "eur/usd")]).await; + source.append("a", vec![row(2, "eur/usd")]).await; + + let pipeline = pipeline(source); + pipeline.run_once().await.unwrap(); + pipeline.run_once().await.unwrap(); + let first_pass = pipeline.derived_store().rows_in_range(1, 2).await.unwrap(); + + // Force a full re-ingest of the same range without going through a + // reorg: reset the watermark and re-run from the start. + pipeline.watermark_store().clear().await.unwrap(); + pipeline.run_once().await.unwrap(); + pipeline.run_once().await.unwrap(); + let second_pass = pipeline.derived_store().rows_in_range(1, 2).await.unwrap(); + + assert_eq!( + first_pass, second_pass, + "aggregate rows must be byte-identical on re-run" + ); + } + + #[tokio::test] + async fn a_reorg_rolls_back_by_the_confirmation_depth_and_drops_orphaned_rows() { + let source = FakeLedgerSource::new(); + for seq in 1..=5u64 { + source.append("a", vec![row(seq, "eur/usd")]).await; + } + + let pipeline = pipeline(source); + for _ in 1..=5 { + pipeline.run_once().await.unwrap(); + } + assert_eq!( + pipeline + .derived_store() + .rows_in_range(1, 5) + .await + .unwrap() + .len(), + 5 + ); + + // Ledgers 4 and 5 get reorged out onto a new fork that also extends + // one ledger past the old tip — a reorg is only observable once a + // new ledger with different ancestry actually arrives, so the + // replacement must reach ledger 6 for the pipeline to notice + // anything happened. The new fork carries a different corridor so a + // leftover orphan from the old fork is unmistakable. + pipeline + .source() + .fork_from( + 4, + vec![ + ("b", vec![row(4, "gbp/usd")]), + ("b", vec![row(5, "gbp/usd")]), + ("b", vec![row(6, "gbp/usd")]), + ], + ) + .await; + + let outcome = pipeline.run_once().await.unwrap(); + assert_eq!( + outcome, + IngestOutcome::ReorgHandled { + rolled_back_to: 2, + resumed_from: 3 + } + ); + + // Walk forward again to fully re-ingest the corrected chain. + loop { + match pipeline.run_once().await.unwrap() { + IngestOutcome::Idle => break, + _ => continue, + } + } + + let final_rows = pipeline.derived_store().rows_in_range(1, 6).await.unwrap(); + let corridors: std::collections::BTreeSet<_> = + final_rows.iter().map(|r| r.corridor.as_str()).collect(); + + assert_eq!(final_rows.len(), 6, "no duplicate or orphaned rows"); + assert!( + corridors.contains("gbp/usd"), + "new fork's rows must be present" + ); + assert!( + !final_rows + .iter() + .any(|r| r.ledger_sequence >= 4 && r.corridor == "eur/usd"), + "old fork's rows at reorged sequences must not survive" + ); + } + + #[tokio::test] + async fn a_reorg_deeper_than_the_start_sequence_resets_ingestion_entirely() { + let source = FakeLedgerSource::new(); + source.append("a", vec![row(1, "eur/usd")]).await; + source.append("a", vec![row(2, "eur/usd")]).await; + + let pipeline = IngestionPipeline::new( + source, + InMemoryWatermarkStore::new(), + InMemoryDerivedStore::new(), + 1, + 10, // confirmation depth far exceeds the chain ingested so far + ); + pipeline.run_once().await.unwrap(); + pipeline.run_once().await.unwrap(); + + // Extend one ledger past the old tip (2), so the reorg is actually + // observable — see the comment in the test above. + pipeline + .source() + .fork_from( + 2, + vec![ + ("b", vec![row(2, "gbp/usd")]), + ("b", vec![row(3, "gbp/usd")]), + ], + ) + .await; + + let outcome = pipeline.run_once().await.unwrap(); + assert_eq!( + outcome, + IngestOutcome::ReorgHandled { + rolled_back_to: 0, + resumed_from: 1 + } + ); + assert_eq!(pipeline.watermark_store().load().await.unwrap(), None); + assert!(pipeline + .derived_store() + .rows_in_range(0, 100) + .await + .unwrap() + .is_empty()); + } + + #[test] + fn fnv1a_is_deterministic_and_sensitive_to_input() { + assert_eq!(fnv1a_u64("a"), fnv1a_u64("a")); + assert_ne!(fnv1a_u64("a"), fnv1a_u64("b")); + assert_eq!(fnv1a_hex("a").len(), 16); + } +} diff --git a/backend/src/ingestion/reconcile.rs b/backend/src/ingestion/reconcile.rs new file mode 100644 index 00000000..9dbb1a09 --- /dev/null +++ b/backend/src/ingestion/reconcile.rs @@ -0,0 +1,303 @@ +//! Scheduled drift detection: recomputes a checksum directly from raw +//! ledger data and compares it against what the pipeline actually has +//! stored, so a bug in the ingestion path itself — not just an outage — +//! gets caught. +//! +//! Watermark-based checkpointing proves the pipeline *ran*; it says nothing +//! about whether what it wrote still matches what the source of truth would +//! produce today. This is the independent check for that. + +use std::time::Duration; + +use async_trait::async_trait; + +use crate::snapshot::generator::RawSnapshotRow; + +use super::fetch::LedgerSource; +use super::upsert::DerivedStore; +use super::watermark::WatermarkStore; +use super::{fnv1a_u64, IngestionError}; + +/// A detected mismatch between raw and derived data over one window. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct DriftAlert { + pub window_start: u64, + pub window_end: u64, + pub raw_checksum: u64, + pub derived_checksum: u64, +} + +#[async_trait] +pub trait DriftAlertSink: Send + Sync { + async fn emit(&self, alert: DriftAlert) -> Result<(), IngestionError>; +} + +/// Reference [`DriftAlertSink`] that just records what it was given. +#[derive(Debug, Default)] +pub struct InMemoryDriftAlertSink { + alerts: tokio::sync::Mutex>, +} + +impl InMemoryDriftAlertSink { + pub fn new() -> Self { + Self::default() + } + + pub async fn alerts(&self) -> Vec { + self.alerts.lock().await.clone() + } +} + +#[async_trait] +impl DriftAlertSink for InMemoryDriftAlertSink { + async fn emit(&self, alert: DriftAlert) -> Result<(), IngestionError> { + self.alerts.lock().await.push(alert); + Ok(()) + } +} + +/// Order-independent checksum over a set of rows. +/// +/// Order-independence matters because the two sides being compared — +/// re-fetched raw ledgers and stored derived rows — have no guaranteed +/// common iteration order; a checksum sensitive to order would flag drift +/// on every run regardless of whether the data actually matched. Wrapping +/// addition (rather than XOR) is deliberate: XOR-combining would let two +/// occurrences of the same row cancel out, silently hiding exactly the +/// double-counting bug this check exists to catch. +fn checksum(rows: &[RawSnapshotRow]) -> u64 { + rows.iter().fold(0u64, |acc, row| { + let canonical = format!( + "{}|{}|{}|{}|{}|{}", + row.ledger_sequence, + row.source, + row.corridor, + row.reliability, + row.volume, + row.latency_ms + ); + acc.wrapping_add(fnv1a_u64(&canonical)) + }) +} + +/// Periodically checks a trailing window of already-ingested ledgers for +/// drift between the derived store and a fresh recomputation from raw data. +pub struct ReconciliationCheck { + source: L, + watermark: W, + derived: D, + alerts: A, + window: u64, +} + +impl ReconciliationCheck +where + L: LedgerSource, + W: WatermarkStore, + D: DerivedStore, + A: DriftAlertSink, +{ + /// `window` is how many trailing ledgers each check covers. + pub fn new(source: L, watermark: W, derived: D, alerts: A, window: u64) -> Self { + Self { + source, + watermark, + derived, + alerts, + window: window.max(1), + } + } + + /// Checks the trailing window ending at the current watermark. Returns + /// `Ok(None)` when nothing has been ingested yet, or when the window + /// matched; returns the alert (already emitted) on drift. + pub async fn run_once(&self) -> Result, IngestionError> { + let Some(watermark) = self.watermark.load().await? else { + return Ok(None); + }; + + let end = watermark.sequence; + let start = end.saturating_sub(self.window - 1).max(1); + + let mut raw_rows = Vec::new(); + for sequence in start..=end { + if let Some(ledger) = self.source.fetch_ledger(sequence).await? { + raw_rows.extend(ledger.rows); + } + } + + let stored_rows = self.derived.rows_in_range(start, end).await?; + + let raw_checksum = checksum(&raw_rows); + let derived_checksum = checksum(&stored_rows); + + if raw_checksum == derived_checksum { + return Ok(None); + } + + let alert = DriftAlert { + window_start: start, + window_end: end, + raw_checksum, + derived_checksum, + }; + self.alerts.emit(alert.clone()).await?; + Ok(Some(alert)) + } + + /// Runs [`Self::run_once`] on a fixed interval, forever. A failed check + /// is logged and retried next tick rather than ending the loop — a + /// transient fetch failure checking last week's data is not a reason to + /// stop checking tomorrow's. + pub async fn run_forever(&self, interval: Duration) { + let mut ticker = tokio::time::interval(interval); + loop { + ticker.tick().await; + if let Err(error) = self.run_once().await { + tracing::warn!(error = %error, "reconciliation check failed; retrying next tick"); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ingestion::fetch::FakeLedgerSource; + use crate::ingestion::upsert::InMemoryDerivedStore; + use crate::ingestion::watermark::{InMemoryWatermarkStore, Watermark}; + + fn row(sequence: u64, corridor: &str, volume: f64) -> RawSnapshotRow { + RawSnapshotRow { + ledger_sequence: sequence, + corridor: corridor.to_string(), + source: "testnet".to_string(), + reliability: 1.0, + volume, + latency_ms: 5.0, + } + } + + async fn setup( + window: u64, + ) -> ReconciliationCheck< + FakeLedgerSource, + InMemoryWatermarkStore, + InMemoryDerivedStore, + InMemoryDriftAlertSink, + > { + let source = FakeLedgerSource::new(); + for seq in 1..=5u64 { + source + .append("a", vec![row(seq, "eur/usd", seq as f64)]) + .await; + } + + let watermark = InMemoryWatermarkStore::new(); + watermark + .save(Watermark { + sequence: 5, + hash: "irrelevant-for-this-check".to_string(), + }) + .await + .unwrap(); + + let derived = InMemoryDerivedStore::new(); + for seq in 1..=5u64 { + derived + .upsert(seq, vec![row(seq, "eur/usd", seq as f64)]) + .await + .unwrap(); + } + + ReconciliationCheck::new( + source, + watermark, + derived, + InMemoryDriftAlertSink::new(), + window, + ) + } + + #[tokio::test] + async fn matching_raw_and_derived_data_raises_no_alert() { + let check = setup(5).await; + assert_eq!(check.run_once().await.unwrap(), None); + assert!(check.alerts.alerts().await.is_empty()); + } + + #[tokio::test] + async fn no_watermark_means_nothing_to_check_yet() { + let source = FakeLedgerSource::new(); + let check = ReconciliationCheck::new( + source, + InMemoryWatermarkStore::new(), + InMemoryDerivedStore::new(), + InMemoryDriftAlertSink::new(), + 5, + ); + + assert_eq!(check.run_once().await.unwrap(), None); + } + + #[tokio::test] + async fn a_stale_derived_row_is_detected_and_alerted() { + let check = setup(5).await; + // Simulate the derived store having drifted from raw ledger data — + // e.g. a bug in the ingestion write path corrupted one row. + check + .derived + .upsert(3, vec![row(3, "eur/usd", 999.0)]) + .await + .unwrap(); + + let alert = check + .run_once() + .await + .unwrap() + .expect("drift must be detected"); + assert_eq!(alert.window_start, 1); + assert_eq!(alert.window_end, 5); + assert_ne!(alert.raw_checksum, alert.derived_checksum); + assert_eq!(check.alerts.alerts().await, vec![alert]); + } + + #[tokio::test] + async fn a_missing_derived_row_is_detected() { + let check = setup(5).await; + check.derived.invalidate_from(5).await.unwrap(); + + assert!(check.run_once().await.unwrap().is_some()); + } + + #[test] + fn checksum_is_order_independent() { + let a = vec![row(1, "eur/usd", 1.0), row(2, "btc/usd", 2.0)]; + let mut b = a.clone(); + b.reverse(); + assert_eq!(checksum(&a), checksum(&b)); + } + + #[test] + fn checksum_changes_when_a_row_is_duplicated() { + // A checksum built from XOR would let a duplicated row cancel out + // and silently pass, hiding exactly the double-counting bug this + // check exists to catch; wrapping addition must not have that + // problem. + let single = vec![row(1, "eur/usd", 1.0)]; + let mut doubled = single.clone(); + doubled.push(row(1, "eur/usd", 1.0)); + + assert_ne!( + checksum(&single), + checksum(&doubled), + "a duplicated row must change the checksum, not cancel out" + ); + } + + #[tokio::test] + async fn window_is_clamped_to_at_least_one() { + let check = setup(0).await; + assert_eq!(check.window, 1); + } +} diff --git a/backend/src/ingestion/upsert.rs b/backend/src/ingestion/upsert.rs new file mode 100644 index 00000000..e5ab6fad --- /dev/null +++ b/backend/src/ingestion/upsert.rs @@ -0,0 +1,212 @@ +//! Idempotent, ledger-keyed storage for derived rows. +//! +//! Every write is a full replace of the row set for one `(ledger_sequence, +//! entity_id)` pair, never a blind insert. Re-ingesting the same ledger with +//! the same inputs therefore always converges on the same stored value +//! instead of accumulating duplicates, and rolling a reorg back is just +//! deleting every key at or after the rollback point. + +use std::collections::BTreeMap; + +use async_trait::async_trait; + +use crate::snapshot::generator::RawSnapshotRow; + +use super::IngestionError; + +/// Identifies one derived row within a ledger: which network it came from, +/// and which corridor it describes. Two rows with the same `(ledger_sequence, +/// EntityId)` are the same logical fact and the second write replaces the +/// first — they never coexist. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct EntityId { + pub source: String, + pub corridor: String, +} + +impl EntityId { + fn of(row: &RawSnapshotRow) -> Self { + Self { + source: row.source.clone(), + corridor: row.corridor.clone(), + } + } +} + +/// Idempotent, ledger-sequence-keyed storage for [`RawSnapshotRow`]s. +#[async_trait] +pub trait DerivedStore: Send + Sync { + /// Replaces the stored rows for `sequence` with `rows`. Any entity + /// previously stored for this sequence but absent from `rows` is + /// dropped — the write is a full replace of that ledger's rows, not a + /// merge, so a corridor that no longer applies (e.g. after a reorg + /// replaced it) cannot linger as a stale orphan. + async fn upsert(&self, sequence: u64, rows: Vec) -> Result<(), IngestionError>; + + /// Deletes every stored row with `ledger_sequence >= from_sequence`. + /// Used to unwind the abandoned side of a reorg before re-ingesting the + /// canonical chain over the same range. + async fn invalidate_from(&self, from_sequence: u64) -> Result<(), IngestionError>; + + /// Returns every stored row with `start <= ledger_sequence <= end`, + /// recomputed fresh from what is actually stored — never from an + /// incrementally-maintained running total, so a rolled-back reorg can + /// never leave a stale aggregate behind. + async fn rows_in_range( + &self, + start: u64, + end: u64, + ) -> Result, IngestionError>; +} + +/// Reference [`DerivedStore`] backed by process memory. +#[derive(Debug, Default)] +pub struct InMemoryDerivedStore { + rows: tokio::sync::Mutex>, +} + +impl InMemoryDerivedStore { + pub fn new() -> Self { + Self::default() + } +} + +#[async_trait] +impl DerivedStore for InMemoryDerivedStore { + async fn upsert(&self, sequence: u64, rows: Vec) -> Result<(), IngestionError> { + let mut guard = self.rows.lock().await; + + guard.retain(|(seq, _), _| *seq != sequence); + for row in rows { + guard.insert((sequence, EntityId::of(&row)), row); + } + + Ok(()) + } + + async fn invalidate_from(&self, from_sequence: u64) -> Result<(), IngestionError> { + self.rows + .lock() + .await + .retain(|(seq, _), _| *seq < from_sequence); + Ok(()) + } + + async fn rows_in_range( + &self, + start: u64, + end: u64, + ) -> Result, IngestionError> { + Ok(self + .rows + .lock() + .await + .iter() + .filter(|((seq, _), _)| *seq >= start && *seq <= end) + .map(|(_, row)| row.clone()) + .collect()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn row(sequence: u64, corridor: &str, volume: f64) -> RawSnapshotRow { + RawSnapshotRow { + ledger_sequence: sequence, + corridor: corridor.to_string(), + source: "testnet".to_string(), + reliability: 1.0, + volume, + latency_ms: 10.0, + } + } + + #[tokio::test] + async fn re_upserting_the_same_ledger_does_not_duplicate_rows() { + let store = InMemoryDerivedStore::new(); + + store + .upsert(10, vec![row(10, "eur/usd", 5.0)]) + .await + .unwrap(); + store + .upsert(10, vec![row(10, "eur/usd", 5.0)]) + .await + .unwrap(); + + let rows = store.rows_in_range(10, 10).await.unwrap(); + assert_eq!(rows.len(), 1, "re-upsert must replace, not append"); + assert_eq!(rows[0].volume, 5.0); + } + + #[tokio::test] + async fn re_upserting_with_a_different_value_replaces_it() { + let store = InMemoryDerivedStore::new(); + + store + .upsert(10, vec![row(10, "eur/usd", 5.0)]) + .await + .unwrap(); + store + .upsert(10, vec![row(10, "eur/usd", 9.0)]) + .await + .unwrap(); + + let rows = store.rows_in_range(10, 10).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].volume, 9.0); + } + + #[tokio::test] + async fn upsert_drops_entities_no_longer_present_for_that_ledger() { + let store = InMemoryDerivedStore::new(); + + store + .upsert(10, vec![row(10, "eur/usd", 5.0), row(10, "btc/usd", 1.0)]) + .await + .unwrap(); + // Second upsert for the same sequence only carries eur/usd — this is + // exactly what a reorg replacement looks like when the new fork's + // ledger doesn't reproduce every corridor the old fork had. + store + .upsert(10, vec![row(10, "eur/usd", 5.0)]) + .await + .unwrap(); + + let rows = store.rows_in_range(10, 10).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].corridor, "eur/usd"); + } + + #[tokio::test] + async fn invalidate_from_removes_only_the_targeted_range() { + let store = InMemoryDerivedStore::new(); + store.upsert(1, vec![row(1, "eur/usd", 1.0)]).await.unwrap(); + store.upsert(2, vec![row(2, "eur/usd", 2.0)]).await.unwrap(); + store.upsert(3, vec![row(3, "eur/usd", 3.0)]).await.unwrap(); + + store.invalidate_from(2).await.unwrap(); + + let remaining = store.rows_in_range(0, 100).await.unwrap(); + let sequences: Vec = remaining.iter().map(|r| r.ledger_sequence).collect(); + assert_eq!(sequences, vec![1]); + } + + #[tokio::test] + async fn rows_in_range_is_bounded_on_both_ends() { + let store = InMemoryDerivedStore::new(); + for seq in 1..=5 { + store + .upsert(seq, vec![row(seq, "eur/usd", seq as f64)]) + .await + .unwrap(); + } + + let rows = store.rows_in_range(2, 4).await.unwrap(); + let mut sequences: Vec = rows.iter().map(|r| r.ledger_sequence).collect(); + sequences.sort_unstable(); + assert_eq!(sequences, vec![2, 3, 4]); + } +} diff --git a/backend/src/ingestion/watermark.rs b/backend/src/ingestion/watermark.rs new file mode 100644 index 00000000..be563740 --- /dev/null +++ b/backend/src/ingestion/watermark.rs @@ -0,0 +1,138 @@ +//! The ingestion checkpoint: the last ledger successfully ingested, keyed on +//! both its sequence number *and* its hash. +//! +//! A checkpoint keyed on sequence alone can only detect a reorg indirectly, +//! as a gap or a downstream inconsistency. Carrying the hash alongside the +//! sequence makes a reorg directly observable: the next fetched ledger's +//! `prev_hash` either matches this hash (linear continuation) or it doesn't +//! (the chain moved out from under us). + +use async_trait::async_trait; + +use super::IngestionError; + +/// The last ledger this pipeline has durably committed. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Watermark { + pub sequence: u64, + pub hash: String, +} + +/// Durable storage for the ingestion checkpoint. +/// +/// Implementors must make `save` and `load` consistent with a single logical +/// checkpoint: a `load` that races a `save` may return either the old or the +/// new value, but never a value that was never saved. +#[async_trait] +pub trait WatermarkStore: Send + Sync { + async fn load(&self) -> Result, IngestionError>; + async fn save(&self, watermark: Watermark) -> Result<(), IngestionError>; + /// Clears the checkpoint entirely, so the next `load` returns `None`. + /// + /// Used when a reorg's confirmation depth reaches back past the + /// pipeline's configured start ledger — there is no stable ledger left + /// to roll back to, so ingestion must restart from scratch. + async fn clear(&self) -> Result<(), IngestionError>; +} + +/// Reference [`WatermarkStore`] backed by process memory. +/// +/// Not durable across restarts — a production deployment plugs in a real +/// persistent store behind the same trait, mirroring how +/// [`InMemoryAnalyticsSink`](crate::event_indexer::InMemoryAnalyticsSink) and +/// the reconciliation module's in-memory stores stand in for production +/// adapters elsewhere in this crate. +#[derive(Debug, Default)] +pub struct InMemoryWatermarkStore { + state: tokio::sync::Mutex>, +} + +impl InMemoryWatermarkStore { + pub fn new() -> Self { + Self::default() + } +} + +#[async_trait] +impl WatermarkStore for InMemoryWatermarkStore { + async fn load(&self) -> Result, IngestionError> { + Ok(self.state.lock().await.clone()) + } + + async fn save(&self, watermark: Watermark) -> Result<(), IngestionError> { + *self.state.lock().await = Some(watermark); + Ok(()) + } + + async fn clear(&self) -> Result<(), IngestionError> { + *self.state.lock().await = None; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn starts_empty() { + let store = InMemoryWatermarkStore::new(); + assert_eq!(store.load().await.unwrap(), None); + } + + #[tokio::test] + async fn round_trips_a_saved_watermark() { + let store = InMemoryWatermarkStore::new(); + let watermark = Watermark { + sequence: 42, + hash: "abc123".to_string(), + }; + + store.save(watermark.clone()).await.unwrap(); + + assert_eq!(store.load().await.unwrap(), Some(watermark)); + } + + #[tokio::test] + async fn a_later_save_overwrites_the_earlier_one() { + let store = InMemoryWatermarkStore::new(); + store + .save(Watermark { + sequence: 1, + hash: "a".to_string(), + }) + .await + .unwrap(); + store + .save(Watermark { + sequence: 2, + hash: "b".to_string(), + }) + .await + .unwrap(); + + assert_eq!( + store.load().await.unwrap(), + Some(Watermark { + sequence: 2, + hash: "b".to_string() + }) + ); + } + + #[tokio::test] + async fn clear_resets_to_empty() { + let store = InMemoryWatermarkStore::new(); + store + .save(Watermark { + sequence: 7, + hash: "x".to_string(), + }) + .await + .unwrap(); + + store.clear().await.unwrap(); + + assert_eq!(store.load().await.unwrap(), None); + } +} diff --git a/backend/src/lib.rs b/backend/src/lib.rs index e689d1cb..e0a3f479 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -8,6 +8,7 @@ pub mod contract_ops; pub mod distributed_lock; pub mod event_indexer; +pub mod ingestion; pub mod network; pub mod observability; pub mod realtime; diff --git a/backend/tests/ingestion_reorg_test.rs b/backend/tests/ingestion_reorg_test.rs new file mode 100644 index 00000000..71f24be3 --- /dev/null +++ b/backend/tests/ingestion_reorg_test.rs @@ -0,0 +1,138 @@ +//! Replays a synthetic reorg — two conflicting ledgers at the same height — +//! through the full ingestion pipeline (fetch -> checkpoint -> upsert) and +//! asserts the final state matches only the canonical chain, with no +//! manual intervention. This is the acceptance test for issue #319. + +use stellar_insights_backend::ingestion::{ + DerivedStore, FakeLedgerSource, InMemoryDerivedStore, InMemoryWatermarkStore, IngestOutcome, + IngestionPipeline, WatermarkStore, +}; +use stellar_insights_backend::snapshot::generator::RawSnapshotRow; + +fn row(sequence: u64, corridor: &str, volume: f64) -> RawSnapshotRow { + RawSnapshotRow { + ledger_sequence: sequence, + corridor: corridor.to_string(), + source: "testnet".to_string(), + reliability: 1.0, + volume, + latency_ms: 12.0, + } +} + +async fn run_to_idle( + pipeline: &IngestionPipeline, +) { + loop { + if pipeline.run_once().await.expect("run_once must not error") == IngestOutcome::Idle { + break; + } + } +} + +#[tokio::test] +async fn a_reorg_fixture_replays_to_only_the_canonical_chain_with_no_manual_intervention() { + let source = FakeLedgerSource::new(); + // Canonical-looking chain, fork A: ledgers 1..=6. + for seq in 1..=6u64 { + source + .append("fork-a", vec![row(seq, "eur/usd", seq as f64 * 10.0)]) + .await; + } + + let pipeline = IngestionPipeline::new( + source, + InMemoryWatermarkStore::new(), + InMemoryDerivedStore::new(), + 1, + 3, + ); + + run_to_idle(&pipeline).await; + assert_eq!( + pipeline + .derived_store() + .rows_in_range(1, 6) + .await + .unwrap() + .len(), + 6 + ); + + // A reorg replaces ledgers 5 and 6 with a competing fork B, which also + // extends one ledger past the old tip (7) — the only way the pipeline + // can discover a reorg is by observing a new ledger whose ancestry + // doesn't match what it already committed. + pipeline + .source() + .fork_from( + 5, + vec![ + ("fork-b", vec![row(5, "gbp/usd", 500.0)]), + ("fork-b", vec![row(6, "gbp/usd", 600.0)]), + ("fork-b", vec![row(7, "gbp/usd", 700.0)]), + ], + ) + .await; + + // No manual intervention: just keep calling run_once, exactly as a + // production poll loop would. + run_to_idle(&pipeline).await; + + let final_rows = pipeline.derived_store().rows_in_range(1, 7).await.unwrap(); + let mut by_sequence: Vec<(u64, &str)> = final_rows + .iter() + .map(|row| (row.ledger_sequence, row.corridor.as_str())) + .collect(); + by_sequence.sort_unstable(); + + assert_eq!( + by_sequence, + vec![ + (1, "eur/usd"), + (2, "eur/usd"), + (3, "eur/usd"), + (4, "eur/usd"), + (5, "gbp/usd"), + (6, "gbp/usd"), + (7, "gbp/usd"), + ], + "final state must reflect only the canonical (fork-b) chain, with no \ + duplicate or orphaned rows from the abandoned fork-a ledgers" + ); + + // The watermark itself must also point at the canonical chain's tip. + let watermark = pipeline + .watermark_store() + .load() + .await + .unwrap() + .expect("watermark must be set"); + assert_eq!(watermark.sequence, 7); +} + +#[tokio::test] +async fn re_running_ingestion_over_an_already_processed_range_is_byte_identical() { + let source = FakeLedgerSource::new(); + for seq in 1..=4u64 { + source + .append("fork-a", vec![row(seq, "eur/usd", seq as f64)]) + .await; + } + + let pipeline = IngestionPipeline::new( + source, + InMemoryWatermarkStore::new(), + InMemoryDerivedStore::new(), + 1, + 3, + ); + run_to_idle(&pipeline).await; + let first_pass = pipeline.derived_store().rows_in_range(1, 4).await.unwrap(); + + pipeline.watermark_store().clear().await.unwrap(); + run_to_idle(&pipeline).await; + let second_pass = pipeline.derived_store().rows_in_range(1, 4).await.unwrap(); + + assert_eq!(first_pass, second_pass); +}