From 1c2c43506d64809cbc1872acd4a7e6895a9dda7a Mon Sep 17 00:00:00 2001 From: Javier Acosta Date: Mon, 20 Jul 2026 00:55:17 -0300 Subject: [PATCH 1/4] chore: bump pallas to pick up the txmonitor server agent Points at the feat/txmonitor-server revision submitted upstream as txpipe/pallas#792; to be repinned to the mainline merge commit once that PR lands. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 291 +++++++++++++++++++++++++++++++++++++++++------------ Cargo.toml | 4 +- 2 files changed, 229 insertions(+), 66 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 51ba4e86..ceb28947 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1429,7 +1429,7 @@ dependencies = [ "opentelemetry", "opentelemetry-otlp", "opentelemetry_sdk", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "paste", "protoc-wkt", "rayon", @@ -1477,7 +1477,7 @@ dependencies = [ "itertools 0.14.0", "num-bigint", "num-rational", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "paste", "proptest", "rayon", @@ -1498,7 +1498,7 @@ dependencies = [ "hex", "itertools 0.14.0", "opentelemetry", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "proptest", "rayon", "regex", @@ -1520,7 +1520,7 @@ dependencies = [ "dolos-core", "dolos-testing", "fjall", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "tempfile", "thiserror 2.0.18", "tracing", @@ -1549,7 +1549,7 @@ dependencies = [ "num-bigint", "num-rational", "num-traits", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "rayon", "reqwest 0.12.28", "serde", @@ -1573,7 +1573,7 @@ dependencies = [ "dolos-testing", "hex", "http-body-util", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "serde", "serde_json", "tokio", @@ -1593,7 +1593,7 @@ dependencies = [ "futures-util", "hex", "itertools 0.14.0", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "redb", "redb-extras", "serde", @@ -1620,7 +1620,7 @@ dependencies = [ "futures-util", "hex", "itertools 0.14.0", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "rand 0.9.3", "tokio", "tokio-stream", @@ -1639,7 +1639,7 @@ dependencies = [ "itertools 0.14.0", "jsonrpsee", "opentelemetry", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "rand 0.9.3", "serde", "serde_json", @@ -3330,7 +3330,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3590,17 +3590,34 @@ version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "846ff0e3269c223e6ed6202eca1dc98399bc85f7db9be4e4574fab778cfa981b" dependencies = [ - "pallas-addresses", - "pallas-codec", - "pallas-configs", - "pallas-crypto", + "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-configs 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-network 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-traverse 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-txbuilder 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-utxorpc 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-validate 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "pallas" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-configs 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "pallas-hardano", - "pallas-network", - "pallas-primitives", - "pallas-traverse", - "pallas-txbuilder", - "pallas-utxorpc", - "pallas-validate", + "pallas-network 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-txbuilder 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-utxorpc 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-validate 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", ] [[package]] @@ -3614,8 +3631,23 @@ dependencies = [ "crc", "cryptoxide 0.4.4", "hex", - "pallas-codec", - "pallas-crypto", + "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "thiserror 2.0.18", +] + +[[package]] +name = "pallas-addresses" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "base58", + "bech32 0.11.1", + "crc", + "cryptoxide 0.4.4", + "hex", + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "thiserror 2.0.18", ] @@ -3631,6 +3663,17 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "pallas-codec" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "hex", + "minicbor 0.26.4", + "serde", + "thiserror 2.0.18", +] + [[package]] name = "pallas-configs" version = "1.1.1" @@ -3639,9 +3682,24 @@ checksum = "9413d9d84c67434857c08ec2d92c96452671c68fb1921c0c1999a7a699587186" dependencies = [ "base64 0.22.1", "num-rational", - "pallas-addresses", - "pallas-crypto", - "pallas-primitives", + "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "serde", + "serde_json", + "serde_with 3.21.0", +] + +[[package]] +name = "pallas-configs" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "base64 0.22.1", + "num-rational", + "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "serde", "serde_json", "serde_with 3.21.0", @@ -3655,7 +3713,20 @@ checksum = "5abe09448466b168f50eb0beaae4e38a16e49bdfe6b2fba5748ea17a025b15b1" dependencies = [ "cryptoxide 0.4.4", "hex", - "pallas-codec", + "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "rand_core 0.10.1", + "serde", + "thiserror 2.0.18", +] + +[[package]] +name = "pallas-crypto" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "cryptoxide 0.4.4", + "hex", + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "rand_core 0.10.1", "serde", "thiserror 2.0.18", @@ -3664,16 +3735,15 @@ dependencies = [ [[package]] name = "pallas-hardano" version = "1.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "adf818b49428697b0d0d5e680fecffa94020b141460dc6ebfe45295b43c8a6b1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" dependencies = [ "binary-layout", "hex", - "pallas-addresses", - "pallas-codec", - "pallas-crypto", - "pallas-network", - "pallas-traverse", + "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-network 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "serde", "serde_json", "serde_with 3.21.0", @@ -3691,8 +3761,25 @@ dependencies = [ "byteorder", "hex", "itertools 0.14.0", - "pallas-codec", - "pallas-crypto", + "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "rand 0.10.1", + "socket2 0.6.3", + "thiserror 2.0.18", + "tokio", + "tracing", +] + +[[package]] +name = "pallas-network" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "byteorder", + "hex", + "itertools 0.14.0", + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "rand 0.10.1", "socket2 0.6.3", "thiserror 2.0.18", @@ -3707,8 +3794,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "35c2b627b97f708bbb0252834e300d56b1ccea9c982c0e4189d1a414c1b53c18" dependencies = [ "hex", - "pallas-codec", - "pallas-crypto", + "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "serde", + "serde_json", +] + +[[package]] +name = "pallas-primitives" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "hex", + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "serde", "serde_json", ] @@ -3721,10 +3820,26 @@ checksum = "ba24ccce6561ab01d910aa8a89abf7f949f4009c0e36771e763f09d3efba1eed" dependencies = [ "hex", "itertools 0.14.0", - "pallas-addresses", - "pallas-codec", - "pallas-crypto", - "pallas-primitives", + "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "paste", + "serde", + "thiserror 2.0.18", +] + +[[package]] +name = "pallas-traverse" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "hex", + "itertools 0.14.0", + "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "paste", "serde", "thiserror 2.0.18", @@ -3737,11 +3852,27 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24573093e2c342a63cda545ef799c51248a9ccb07061a9de0f2126b694830697" dependencies = [ "hex", - "pallas-addresses", - "pallas-codec", - "pallas-crypto", - "pallas-primitives", - "pallas-traverse", + "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-traverse 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "serde", + "serde_json", + "thiserror 2.0.18", +] + +[[package]] +name = "pallas-txbuilder" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "hex", + "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "serde", "serde_json", "thiserror 2.0.18", @@ -3753,11 +3884,25 @@ version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bb8a5432ea38e559378d5872d012c5cf34df95e0ed7e5a224aa2645ce57f9675" dependencies = [ - "pallas-codec", - "pallas-crypto", - "pallas-primitives", - "pallas-traverse", - "pallas-validate", + "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-traverse 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-validate 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "prost-types 0.13.5", + "utxorpc-spec 0.19.0", +] + +[[package]] +name = "pallas-utxorpc" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" +dependencies = [ + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-validate 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "prost-types 0.13.5", "utxorpc-spec 0.19.0", ] @@ -3767,16 +3912,34 @@ name = "pallas-validate" version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dfaac47eb61c648c32cb98bd9fc74414bd173ee57b17c59412242731d0365964" +dependencies = [ + "chrono", + "hex", + "itertools 0.14.0", + "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-codec 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-primitives 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-traverse 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "serde", + "thiserror 2.0.18", + "tracing", +] + +[[package]] +name = "pallas-validate" +version = "1.1.1" +source = "git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c#79dbeeba3a5d3894d06f87c2e894c21a8ec2619c" dependencies = [ "amaru-uplc", "chrono", "hex", "itertools 0.14.0", - "pallas-addresses", - "pallas-codec", - "pallas-crypto", - "pallas-primitives", - "pallas-traverse", + "pallas-addresses 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-codec 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-crypto 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-primitives 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", + "pallas-traverse 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "serde", "thiserror 2.0.18", "tracing", @@ -4083,7 +4246,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck", - "itertools 0.14.0", + "itertools 0.12.1", "log", "multimap", "once_cell", @@ -4116,7 +4279,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.12.1", "proc-macro2", "quote", "syn", @@ -4129,7 +4292,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.12.1", "proc-macro2", "quote", "syn", @@ -4715,7 +4878,7 @@ dependencies = [ "security-framework 3.7.0", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6059,7 +6222,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2a581b7e2051be3ce376d896e3a05e5ac4f74f486f9de1848f3df2c308776d21" dependencies = [ "hex", - "pallas", + "pallas 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "serde", "thiserror 2.0.18", "trait-variant", @@ -6098,8 +6261,8 @@ dependencies = [ "cryptoxide 0.4.4", "ed25519-bip32", "hex", - "pallas-addresses", - "pallas-crypto", + "pallas-addresses 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pallas-crypto 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "reqwest 0.12.28", "serde", "serde_json", @@ -6977,7 +7140,7 @@ dependencies = [ "dolos-cardano", "dolos-core", "hex", - "pallas", + "pallas 1.1.1 (git+https://github.com/Javieracost/pallas.git?rev=79dbeeba3a5d3894d06f87c2e894c21a8ec2619c)", "postgres", "postgres-native-tls", "serde", diff --git a/Cargo.toml b/Cargo.toml index d01cf19b..b3eafadc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -207,8 +207,8 @@ readme = "README.md" authors = ["TxPipe "] [workspace.dependencies] -pallas = { version = "1.1.1", features = ["hardano", "phase2", "unstable"] } -# pallas = { git = "https://github.com/txpipe/pallas.git", rev = "a7b5a86", features = ["hardano", "phase2", "unstable"] } +# pallas = { version = "1.1.1", features = ["hardano", "phase2", "unstable"] } +pallas = { git = "https://github.com/Javieracost/pallas.git", rev = "79dbeeba3a5d3894d06f87c2e894c21a8ec2619c", features = ["hardano", "phase2", "unstable"] } # pallas = { path = "../pallas/pallas", features = ["hardano", "phase2", "unstable"] } thiserror = "2.0.12" From 92195aee2d911c8aee25adc0a8203c614a0b14d8 Mon Sep 17 00:00:00 2001 From: Javier Acosta Date: Sun, 19 Jul 2026 13:43:57 -0300 Subject: [PATCH 2/4] feat(serve): add local-tx-monitor server to the ouroboros N2C endpoint The ouroboros serve endpoint implemented only chainsync and statequery. Clients that open the local-tx-monitor protocol on the same connection - ogmios does so eagerly on every session - had the whole session killed by the multiplexer, since accepting the N2C handshake claims support for the full protocol set. Adds a txmonitor session handler backed by the domain mempool: - a snapshot is pending + inflight transactions, excluding Confirmed (already on-chain, a node mempool would not hold them), tagged with the current cursor slot - NextTx serves the era-wrapped payload; the wire era is the block era number minus one, same convention as the tx-submission peer client - HasTx matches on the tx hash alone, ignoring the era wrapper, so clients probing the same hash under multiple eras get a correct answer - GetSizes/GetMeasures report capacity as twice the max block body size from effective pparams, mirroring the cardano-node default - AwaitAcquire blocks until the snapshot fingerprint changes, waking on mempool events or tip changes; a closed mempool stream degrades to tip-driven wake-ups instead of busy-looping Includes two ToyDomain integration tests driving a real pallas NodeClient over a unix socket: snapshot queries against a seeded mempool, and await-acquire woken by a tip event. Co-Authored-By: Claude Fable 5 --- src/serve/o7s_unix/mod.rs | 14 +- src/serve/o7s_unix/txmonitor.rs | 453 ++++++++++++++++++++++++++++++++ 2 files changed, 465 insertions(+), 2 deletions(-) create mode 100644 src/serve/o7s_unix/txmonitor.rs diff --git a/src/serve/o7s_unix/mod.rs b/src/serve/o7s_unix/mod.rs index 39232f4f..66f5c3ea 100644 --- a/src/serve/o7s_unix/mod.rs +++ b/src/serve/o7s_unix/mod.rs @@ -8,6 +8,7 @@ use crate::prelude::*; mod chainsync; mod statequery; +mod txmonitor; mod utils; #[derive(Clone)] @@ -29,6 +30,7 @@ async fn handle_session( plexer, chainsync, statequery, + txmonitor, .. } = connection; @@ -44,13 +46,21 @@ async fn handle_session( cancel.clone(), )); - let result = tokio::try_join!(chainsync_task, statequery_task); + let txmonitor_task = tokio::spawn(txmonitor::handle_session( + domain.clone(), + txmonitor, + cancel.clone(), + )); + + let result = tokio::try_join!(chainsync_task, statequery_task, txmonitor_task); plexer.abort().await; - let (chainsync_res, statequery_res) = result.map_err(|e| ServeError::Internal(e.into()))?; + let (chainsync_res, statequery_res, txmonitor_res) = + result.map_err(|e| ServeError::Internal(e.into()))?; chainsync_res?; statequery_res?; + txmonitor_res?; Ok(()) } diff --git a/src/serve/o7s_unix/txmonitor.rs b/src/serve/o7s_unix/txmonitor.rs new file mode 100644 index 00000000..34421ca6 --- /dev/null +++ b/src/serve/o7s_unix/txmonitor.rs @@ -0,0 +1,453 @@ +use futures_util::StreamExt as _; +use std::collections::BTreeSet; + +use pallas::codec::utils::TagWrap; +use pallas::network::miniprotocols::txmonitor::{ + self, ClientQueryRequest, MempoolMeasures, MempoolSizeAndCapacity, SizeAndCapacity, +}; +use tracing::{debug, info, warn}; + +use dolos_cardano::load_effective_pparams; +use dolos_core::{MempoolStore as _, StateStore as _, TipSubscription as _}; + +use crate::prelude::*; + +// HACK: the tx era number differs from the block era number, we subtract 1 to +// make them match. +fn to_n2c_era(era: u16) -> u8 { + (era - 1) as u8 +} + +struct Snapshot { + slot: u64, + txs: Vec, + cursor: usize, +} + +impl Snapshot { + fn fingerprint(&self) -> (u64, BTreeSet) { + (self.slot, self.txs.iter().map(|tx| tx.hash).collect()) + } + + fn size_in_bytes(&self) -> u64 { + self.txs + .iter() + .map(|tx| tx.payload.cbor().len() as u64) + .sum() + } +} + +pub struct Session { + domain: D, + connection: txmonitor::Server, + snapshot: Option, +} + +impl Session { + fn take_snapshot(&self) -> Result { + let point = self + .domain + .state() + .read_cursor() + .map_err(Error::server)? + .unwrap_or(ChainPoint::Origin); + + let mempool = self.domain.mempool(); + + // confirmed txs are already on-chain, a node mempool wouldn't hold them + let txs = mempool + .peek_pending() + .into_iter() + .chain(mempool.peek_inflight()) + .filter(|tx| !matches!(tx.stage, MempoolTxStage::Confirmed)) + .collect(); + + Ok(Snapshot { + slot: point.slot(), + txs, + cursor: 0, + }) + } + + fn acquired(&self) -> &Snapshot { + self.snapshot + .as_ref() + .expect("txmonitor request handled without an acquired snapshot") + } + + async fn handle_acquire(&mut self) -> Result<(), Error> { + let snapshot = self.take_snapshot()?; + let slot = snapshot.slot; + + debug!(slot, txs = snapshot.txs.len(), "acquired mempool snapshot"); + + self.snapshot = Some(snapshot); + + self.connection + .send_acquired(slot) + .await + .map_err(Error::server)?; + + Ok(()) + } + + async fn handle_await_acquire(&mut self) -> Result<(), Error> { + let previous = self.acquired().fingerprint(); + + let mut mempool_updates = self.domain.mempool().subscribe(); + let mut mempool_open = true; + + let mut tip_updates = self + .domain + .watch_tip(None) + .map_err(|e| Error::server(e.to_string()))?; + + loop { + let snapshot = self.take_snapshot()?; + + if snapshot.fingerprint() != previous { + let slot = snapshot.slot; + + debug!(slot, txs = snapshot.txs.len(), "acquired changed snapshot"); + + self.snapshot = Some(snapshot); + + self.connection + .send_acquired(slot) + .await + .map_err(Error::server)?; + + return Ok(()); + } + + tokio::select! { + update = mempool_updates.next(), if mempool_open => { + match update { + Some(Err(e)) => return Err(Error::server(e)), + Some(Ok(_)) => (), + // a closed stream must not busy-loop the select; keep + // waking on tip changes only + None => mempool_open = false, + } + } + _ = tip_updates.next_tip() => (), + } + } + } + + async fn handle_next_tx(&mut self) -> Result<(), Error> { + let snapshot = self + .snapshot + .as_mut() + .expect("txmonitor request handled without an acquired snapshot"); + + let tx = snapshot.txs.get(snapshot.cursor).map(|tx| { + let EraCbor(era, cbor) = &tx.payload; + (to_n2c_era(*era), TagWrap::new(cbor.clone().into())) + }); + + if tx.is_some() { + snapshot.cursor += 1; + } + + self.connection + .send_next_tx(tx) + .await + .map_err(Error::server)?; + + Ok(()) + } + + async fn handle_has_tx(&mut self, id: txmonitor::TxId) -> Result<(), Error> { + // peers wrap the same hash in every plausible era and combine the + // answers, so we match on the hash alone and ignore the era tag + let (_era, hash) = &id; + + let has = self + .acquired() + .txs + .iter() + .any(|tx| tx.hash.as_slice() == hash.as_slice()); + + self.connection + .send_has_tx(has) + .await + .map_err(Error::server)?; + + Ok(()) + } + + fn capacity_in_bytes(&self) -> Result { + // cardano-node sizes its mempool at twice the max block body by + // default; report the same so clients get a familiar reference + let pparams = load_effective_pparams::(self.domain.state()) + .map_err(|e| Error::server(e.to_string()))?; + + Ok(pparams.max_block_body_size_or_default() * 2) + } + + async fn handle_get_sizes(&mut self) -> Result<(), Error> { + let capacity = self.capacity_in_bytes()?; + let snapshot = self.acquired(); + + let sizes = MempoolSizeAndCapacity { + capacity_in_bytes: capacity as u32, + size_in_bytes: snapshot.size_in_bytes() as u32, + number_of_txs: snapshot.txs.len() as u32, + }; + + self.connection + .send_size_and_capacity(sizes) + .await + .map_err(Error::server)?; + + Ok(()) + } + + async fn handle_get_measures(&mut self) -> Result<(), Error> { + let capacity = self.capacity_in_bytes()?; + let snapshot = self.acquired(); + + let measures = MempoolMeasures { + tx_count: snapshot.txs.len() as u32, + measures: vec![( + "transaction_bytes".to_string(), + SizeAndCapacity { + size: snapshot.size_in_bytes(), + capacity, + }, + )], + }; + + self.connection + .send_measures(measures) + .await + .map_err(Error::server)?; + + Ok(()) + } + + async fn process_requests(&mut self) -> Result<(), Error> { + loop { + if self + .connection + .recv_while_idle() + .await + .map_err(Error::server)? + .is_none() + { + break; + } + + self.handle_acquire().await?; + + loop { + match self + .connection + .recv_while_acquired() + .await + .map_err(Error::server)? + { + ClientQueryRequest::AwaitAcquire => { + self.handle_await_acquire().await?; + } + ClientQueryRequest::NextTx => { + self.handle_next_tx().await?; + } + ClientQueryRequest::HasTx(id) => { + self.handle_has_tx(id).await?; + } + ClientQueryRequest::GetSizes => { + self.handle_get_sizes().await?; + } + ClientQueryRequest::GetMeasures => { + self.handle_get_measures().await?; + } + ClientQueryRequest::Release => { + self.snapshot = None; + break; + } + } + } + } + + Ok(()) + } +} + +pub async fn handle_session( + domain: D, + connection: txmonitor::Server, + cancel: C, +) -> Result<(), ServeError> { + let mut session = Session { + domain, + connection, + snapshot: None, + }; + + info!("txmonitor session started"); + + tokio::select! { + result = session.process_requests() => { + if let Err(e) = result { + warn!(?e, "txmonitor session error"); + return Err(ServeError::Internal(e.into())); + } + info!("txmonitor client ended protocol"); + }, + _ = cancel.cancelled() => { + info!("txmonitor protocol was cancelled"); + } + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + use tokio_util::sync::CancellationToken; + + use dolos_testing::mempool::make_test_mempool_tx; + use dolos_testing::slot_to_chainpoint; + use dolos_testing::toy_domain::ToyDomain; + + use pallas::network::facades::{NodeClient, NodeServer}; + + use crate::serve::CancelTokenImpl; + + fn spawn_server( + domain: ToyDomain, + listener: tokio::net::UnixListener, + ) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let connection = NodeServer::accept(&listener, 0).await.unwrap(); + + let NodeServer { + plexer, txmonitor, .. + } = connection; + + let cancel = CancelTokenImpl(CancellationToken::new()); + + handle_session(domain, txmonitor, cancel).await.unwrap(); + + plexer.abort().await; + }) + } + + #[tokio::test] + async fn txmonitor_serves_snapshot_queries() { + let domain = ToyDomain::new(None, None); + + let seeded_hash = TxHash::from([0xab; 32]); + + domain + .mempool() + .receive(make_test_mempool_tx(seeded_hash)) + .unwrap(); + + let tempdir = tempfile::tempdir().unwrap(); + let socket = tempdir.path().join("node.socket"); + let listener = tokio::net::UnixListener::bind(&socket).unwrap(); + + let server = spawn_server(domain, listener); + + let mut client = NodeClient::connect(&socket, 0).await.unwrap(); + + client.monitor().acquire().await.unwrap(); + + let sizes = client.monitor().query_size_and_capacity().await.unwrap(); + assert_eq!(sizes.number_of_txs, 1); + assert_eq!(sizes.size_in_bytes, 1); + assert!(sizes.capacity_in_bytes > 0); + + let (era, body) = client.monitor().query_next_tx().await.unwrap().unwrap(); + assert_eq!(era, 6); + assert_eq!(body.0.len(), 1); + + let next = client.monitor().query_next_tx().await.unwrap(); + assert!(next.is_none()); + + let has = client + .monitor() + .query_has_tx((6, seeded_hash.to_vec().into())) + .await + .unwrap(); + assert!(has); + + // the era wrapper is not part of the identity check + let has = client + .monitor() + .query_has_tx((4, seeded_hash.to_vec().into())) + .await + .unwrap(); + assert!(has); + + let has = client + .monitor() + .query_has_tx((6, vec![0xcd; 32].into())) + .await + .unwrap(); + assert!(!has); + + let measures = client.monitor().query_measures().await.unwrap(); + assert_eq!(measures.tx_count, 1); + + client.monitor().release().await.unwrap(); + client.monitor().done().await.unwrap(); + + server.await.unwrap(); + } + + #[tokio::test] + async fn txmonitor_await_acquire_wakes_on_change() { + let domain = ToyDomain::new(None, None); + + let tempdir = tempfile::tempdir().unwrap(); + let socket = tempdir.path().join("node.socket"); + let listener = tokio::net::UnixListener::bind(&socket).unwrap(); + + let server = spawn_server(domain.clone(), listener); + + let mut client = NodeClient::connect(&socket, 0).await.unwrap(); + + client.monitor().acquire().await.unwrap(); + + let new_hash = TxHash::from([0x11; 32]); + + let trigger = tokio::spawn({ + let domain = domain.clone(); + + async move { + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + + domain + .mempool() + .receive(make_test_mempool_tx(new_hash)) + .unwrap(); + + // the toy mempool emits no events; a tip event provides the + // wake-up, mirroring a new block reaching the node + domain.notify_tip(TipEvent::Mark(slot_to_chainpoint(1))); + } + }); + + client.monitor().await_acquire().await.unwrap(); + + let has = client + .monitor() + .query_has_tx((6, new_hash.to_vec().into())) + .await + .unwrap(); + assert!(has); + + trigger.await.unwrap(); + + client.monitor().release().await.unwrap(); + client.monitor().done().await.unwrap(); + + server.await.unwrap(); + } +} From cda6c7c5ab38936372e7073beaa294206649a101 Mon Sep 17 00:00:00 2001 From: Javier Acosta Date: Sun, 19 Jul 2026 17:12:10 -0300 Subject: [PATCH 3/4] fix(serve): stop failing statequery volatile-tip acquires during sync Acquiring the volatile tip validated the cursor block against the archive store, which lags behind the cursor while the node is catching up, so a fully-functional node answered Failure(PointNotOnChain) to a request that cannot legitimately fail. On top of that, the request loop entered the acquired state even after sending a failure, so the client's spec-correct follow-up acquire was rejected as invalid for the current state and the whole statequery protocol died. Ogmios v7's health client hits both on every reconnect against a syncing mainnet node. Volatile-tip acquires now acquire the current cursor directly, and a failed acquire or re-acquire returns the loop to the idle state. Co-Authored-By: Claude Fable 5 --- src/serve/o7s_unix/statequery.rs | 138 +++++++++++++++++++++++++++---- 1 file changed, 124 insertions(+), 14 deletions(-) diff --git a/src/serve/o7s_unix/statequery.rs b/src/serve/o7s_unix/statequery.rs index d891cb27..0a6d9697 100644 --- a/src/serve/o7s_unix/statequery.rs +++ b/src/serve/o7s_unix/statequery.rs @@ -67,20 +67,33 @@ impl Session { Ok(()) } + /// Returns whether the acquire succeeded. On failure, the protocol is + /// back in the idle state and the client is expected to acquire again. async fn handle_acquire( &mut self, point: Option, - ) -> Result<(), Error> { + ) -> Result { debug!(?point, "handling acquire request"); - let chain_point = match point { - Some(p) => ChainPoint::from(p), - None => { - // None means acquire the latest point - self.tip_cursor()? - } + let Some(p) = point else { + // None means acquire the volatile tip, which always succeeds at + // the current cursor. The archive may still be catching up to the + // cursor while the node syncs, so it can't be used to validate + // the node's own tip. + let cursor = self + .domain + .state() + .read_cursor() + .map_err(Error::server)? + .unwrap_or(ChainPoint::Origin); + + self.acquired_point = Some(cursor); + self.send_acquired().await?; + return Ok(true); }; + let chain_point = ChainPoint::from(p); + let exists = match &chain_point { ChainPoint::Origin => true, ChainPoint::Specific(slot, hash) => { @@ -103,12 +116,12 @@ impl Session { if exists { self.acquired_point = Some(chain_point); self.send_acquired().await?; + Ok(true) } else { self.send_failure(localstate::AcquireFailure::PointNotOnChain) .await?; + Ok(false) } - - Ok(()) } /// Decode cardano-cli tagged query format. @@ -348,10 +361,12 @@ impl Session { Ok(()) } + /// Returns whether the re-acquire succeeded. On failure, the protocol is + /// back in the idle state, not the acquired state. async fn handle_reacquire( &mut self, point: Option, - ) -> Result<(), Error> { + ) -> Result { debug!(?point, "handling reacquire request"); self.handle_acquire(point).await } @@ -370,7 +385,10 @@ impl Session { .await .map_err(Error::server)? { - self.handle_acquire(req.0).await?; + if !self.handle_acquire(req.0).await? { + // failed acquire leaves the protocol idle + continue; + } } else { break; } @@ -386,9 +404,10 @@ impl Session { self.handle_query(query).await?; } localstate::ClientQueryRequest::ReAcquire(point) => { - self.handle_reacquire(point).await?; - // After reacquire, we stay in acquired state but with - // new point + if !self.handle_reacquire(point).await? { + // failed reacquire leaves the protocol idle + break; + } } localstate::ClientQueryRequest::Release => { self.handle_release().await?; @@ -431,3 +450,94 @@ pub async fn handle_session( Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + use tokio_util::sync::CancellationToken; + + use dolos_testing::toy_domain::ToyDomain; + + use pallas::network::facades::{NodeClient, NodeServer}; + use pallas::network::miniprotocols::localstate::ClientError; + use pallas::network::miniprotocols::Point; + + use crate::serve::CancelTokenImpl; + + fn spawn_server( + domain: ToyDomain, + listener: tokio::net::UnixListener, + ) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let connection = NodeServer::accept(&listener, 0).await.unwrap(); + + let NodeServer { + plexer, statequery, .. + } = connection; + + let cancel = CancelTokenImpl(CancellationToken::new()); + + handle_session(domain, statequery, cancel).await.unwrap(); + + plexer.abort().await; + }) + } + + #[tokio::test] + async fn statequery_acquires_volatile_tip() { + let domain = ToyDomain::new(None, None); + + let tempdir = tempfile::tempdir().unwrap(); + let socket = tempdir.path().join("node.socket"); + let listener = tokio::net::UnixListener::bind(&socket).unwrap(); + + let server = spawn_server(domain, listener); + + let mut client = NodeClient::connect(&socket, 0).await.unwrap(); + + client.statequery().acquire(None).await.unwrap(); + + client.statequery().send_release().await.unwrap(); + client.statequery().send_done().await.unwrap(); + + server.await.unwrap(); + } + + #[tokio::test] + async fn statequery_recovers_from_failed_acquire() { + let domain = ToyDomain::new(None, None); + + let tempdir = tempfile::tempdir().unwrap(); + let socket = tempdir.path().join("node.socket"); + let listener = tokio::net::UnixListener::bind(&socket).unwrap(); + + let server = spawn_server(domain, listener); + + let mut client = NodeClient::connect(&socket, 0).await.unwrap(); + + let bogus = Point::Specific(1, vec![0xab; 32]); + + let failure = client.statequery().acquire(Some(bogus.clone())).await; + assert!(matches!(failure, Err(ClientError::AcquirePointNotFound))); + + // the session must survive a failed acquire and accept a new one + client.statequery().acquire(None).await.unwrap(); + + // same for a failed re-acquire from the acquired state + client + .statequery() + .send_reacquire(Some(bogus)) + .await + .unwrap(); + let failure = client.statequery().recv_while_acquiring().await; + assert!(matches!(failure, Err(ClientError::AcquirePointNotFound))); + + client.statequery().acquire(None).await.unwrap(); + + client.statequery().send_release().await.unwrap(); + client.statequery().send_done().await.unwrap(); + + server.await.unwrap(); + } +} From c069b63c7576e402490c76c0c5048a8867174f0d Mon Sep 17 00:00:00 2001 From: Javier Acosta Date: Sun, 19 Jul 2026 17:40:34 -0300 Subject: [PATCH 4/4] fix(serve): serve historical chain-sync from the archive Four defects left archive-backed chain-sync unusable for any client that intersects below the WAL window (ogmios, kupo, any N2C syncer): - the initial rollback that opens every chain-sync exchange was sent to the first crawled block's point instead of the negotiated intersection, consuming that block in the process; archive-crawled points are slot-only and cannot encode into a wire point, so the send failed outright - that failure was silently discarded, so the server just stopped serving the protocol while the client waited forever for a reply - crawling from origin skipped the first block of the chain, since the archive batch loader always assumed the cursor block had already been delivered - a client awaiting at the tip died on the next chain re-org: an undo tip event returned without sending anything, leaving the client owed a reply and the server attempting to receive without agency The rollback now echoes the negotiated intersection point without consuming a block, chain-sync session errors are logged and propagated like the other mini-protocols, origin batches start at the first block in the archive, and the await loop skips over undo events until the mark or apply that follows them produces the owed reply. Co-Authored-By: Claude Fable 5 --- crates/core/src/crawl.rs | 11 +- src/serve/o7s_unix/chainsync.rs | 223 ++++++++++++++++++++++++++------ 2 files changed, 190 insertions(+), 44 deletions(-) diff --git a/crates/core/src/crawl.rs b/crates/core/src/crawl.rs index c62754e3..6ed45855 100644 --- a/crates/core/src/crawl.rs +++ b/crates/core/src/crawl.rs @@ -36,11 +36,18 @@ impl Batch { } fn from_archive(last_point: ChainPoint, domain: &D) -> Result { + // Origin is not a block, so there's nothing already seen to skip; for + // any concrete point, the block at the point itself was already + // delivered (or acknowledged via intersection) and must be skipped. + let seen = match &last_point { + ChainPoint::Origin => 0, + _ => 1, + }; + let page = domain .archive() .get_range(Some(last_point.slot()), None)? - // skip the last point, we already seen it - .skip(1) + .skip(seen) .take(LOAD_BATCH_SIZE) .map(|(slot, body)| (ChainPoint::Slot(slot), Arc::new(body))) .collect::>(); diff --git a/src/serve/o7s_unix/chainsync.rs b/src/serve/o7s_unix/chainsync.rs index a8f9a570..0c37d62c 100644 --- a/src/serve/o7s_unix/chainsync.rs +++ b/src/serve/o7s_unix/chainsync.rs @@ -4,14 +4,14 @@ use pallas::network::miniprotocols::{ chainsync::{BlockContent, ClientRequest, N2CServer, Tip}, Point, }; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; use crate::prelude::*; pub struct Session { domain: D, crawler: Option>, - is_new_intersection: bool, + pending_rollback: Option, connection: N2CServer, } @@ -41,7 +41,9 @@ impl Session { async fn send_intersect_found(&mut self, point: ChainPoint) -> Result<(), Error> { debug!("sending intersection found"); - self.is_new_intersection = true; + // chain-sync always starts by rolling the client back to the + // negotiated intersection before any forward event + self.pending_rollback = Some(point.clone()); let tip = self.prepare_tip()?; @@ -63,38 +65,29 @@ impl Session { debug!("waiting for tip change notification"); - let crawler = self.assume_crawler(); - - let event = crawler.next_tip().await; - - self.send_tip_event(event).await?; - - Ok(()) + // after an await-reply the client is owed exactly one chain-sync + // message; undo events have no wire counterpart (the mark that + // follows them becomes the rollback), so keep waiting through them + loop { + let event = self.assume_crawler().next_tip().await; + + match event { + TipEvent::Apply(_, block) => return self.send_forward(block).await, + TipEvent::Mark(point) => return self.send_rollback(point).await, + _ => continue, + } + } } - async fn send_forward(&mut self, point: ChainPoint, block: RawBlock) -> Result<(), Error> { + async fn send_forward(&mut self, block: RawBlock) -> Result<(), Error> { debug!("sending forward event"); let tip = self.prepare_tip()?; - let point = Point::try_from(point).map_err(|_| Error::custom("invalid point"))?; - - // Ouroboros chain-sync always starts by sending the intersection point as an - // initial rollback event. The `is_new_intersection`` flag allows us to track if - // we have already sent that initial rollback or not - if self.is_new_intersection { - self.connection - .send_roll_backward(point, tip) - .await - .map_err(Error::server)?; - - self.is_new_intersection = false; - } else { - self.connection - .send_roll_forward(BlockContent((*block).clone()), tip) - .await - .map_err(Error::server)?; - } + self.connection + .send_roll_forward(BlockContent((*block).clone()), tip) + .await + .map_err(Error::server)?; Ok(()) } @@ -112,24 +105,22 @@ impl Session { .map_err(Error::server) } - async fn send_tip_event(&mut self, log: TipEvent) -> Result<(), Error> { - match log { - TipEvent::Apply(p, b) => self.send_forward(p, b).await, - TipEvent::Mark(p) => self.send_rollback(p).await, - // we skip undo events and expect a Mark to come after - _ => Ok(()), - } - } - async fn handle_next_request(&mut self) -> Result<(), Error> { debug!("handling next request"); + // the rollback to the negotiated intersection is the reply to the + // first next-request; no block is consumed by it + if let Some(point) = self.pending_rollback.take() { + self.send_rollback(point).await?; + return Ok(()); + } + let crawler = self.assume_crawler(); let next = crawler.next_block(); - if let Some((point, block)) = next { - self.send_forward(point, block).await?; + if let Some((_, block)) = next { + self.send_forward(block).await?; } else { self.send_await_and_next().await?; } @@ -196,11 +187,15 @@ pub async fn handle_session( domain, connection, crawler: None, - is_new_intersection: false, + pending_rollback: None, }; tokio::select! { - _ = session.process_requests() => { + result = session.process_requests() => { + if let Err(e) = result { + warn!(?e, "chainsync session error"); + return Err(ServeError::Internal(e.into())); + } info!("client ended protocol"); }, _ = cancel.cancelled() => { @@ -210,3 +205,147 @@ pub async fn handle_session( Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + use tokio_util::sync::CancellationToken; + + use dolos_core::ChainPoint; + use dolos_testing::blocks::make_conway_block; + use dolos_testing::slot_to_chainpoint; + use dolos_testing::toy_domain::ToyDomain; + + use pallas::network::facades::{NodeClient, NodeServer}; + use pallas::network::miniprotocols::chainsync::NextResponse; + + use crate::serve::CancelTokenImpl; + + fn spawn_server( + domain: ToyDomain, + listener: tokio::net::UnixListener, + ) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let connection = NodeServer::accept(&listener, 0).await.unwrap(); + + let NodeServer { + plexer, chainsync, .. + } = connection; + + let cancel = CancelTokenImpl(CancellationToken::new()); + + handle_session(domain, chainsync, cancel).await.unwrap(); + + plexer.abort().await; + }) + } + + fn seed_archive_blocks( + domain: &ToyDomain, + slots: std::ops::Range, + ) -> Vec<(ChainPoint, Vec)> { + let writer = domain.archive().start_writer().unwrap(); + + let mut blocks = vec![]; + + for slot in slots { + let (point, block) = make_conway_block(slot); + writer.apply(&point, &block).unwrap(); + blocks.push((point, (*block).clone())); + } + + writer.commit().unwrap(); + + blocks + } + + #[tokio::test] + async fn chainsync_starts_with_rollback_to_intersection() { + let domain = ToyDomain::new(None, None); + + let tempdir = tempfile::tempdir().unwrap(); + let socket = tempdir.path().join("node.socket"); + let listener = tokio::net::UnixListener::bind(&socket).unwrap(); + + let server = spawn_server(domain.clone(), listener); + + let mut client = NodeClient::connect(&socket, 0).await.unwrap(); + + let (point, _) = client + .chainsync() + .find_intersect(vec![Point::Origin]) + .await + .unwrap(); + + assert_eq!(point, Some(Point::Origin)); + + // the reply to the first next-request is the rollback to the + // negotiated intersection + let next = client.chainsync().request_next().await.unwrap(); + assert!(matches!(next, NextResponse::RollBackward(Point::Origin, _))); + + // an empty chain then awaits at tip; undo events carry no chain-sync + // message of their own, so the reply must come from the mark that + // follows them + let next = client.chainsync().request_next().await.unwrap(); + assert!(matches!(next, NextResponse::Await)); + + let (undone_point, undone_block) = make_conway_block(9); + domain.notify_tip(dolos_core::TipEvent::Undo(undone_point, undone_block)); + domain.notify_tip(dolos_core::TipEvent::Mark(slot_to_chainpoint(7))); + + let next = client.chainsync().recv_while_must_reply().await.unwrap(); + assert!(matches!(next, NextResponse::RollBackward(_, _))); + + client.chainsync().send_done().await.unwrap(); + + server.await.unwrap(); + } + + #[tokio::test] + async fn chainsync_streams_archive_blocks_without_loss() { + let domain = ToyDomain::new(None, None); + + let blocks = seed_archive_blocks(&domain, 1..15); + + let tempdir = tempfile::tempdir().unwrap(); + let socket = tempdir.path().join("node.socket"); + let listener = tokio::net::UnixListener::bind(&socket).unwrap(); + + let server = spawn_server(domain.clone(), listener); + + let mut client = NodeClient::connect(&socket, 0).await.unwrap(); + + let intersect = Point::try_from(blocks[0].0.clone()).unwrap(); + + let (point, _) = client + .chainsync() + .find_intersect(vec![intersect.clone()]) + .await + .unwrap(); + + assert_eq!(point, Some(intersect.clone())); + + let next = client.chainsync().request_next().await.unwrap(); + match next { + NextResponse::RollBackward(p, _) => assert_eq!(p, intersect), + other => panic!("expected initial rollback, got {other:?}"), + } + + // every block after the intersection must arrive, in order, starting + // right after the intersection point and across the internal batch + // boundary + for (_, expected) in &blocks[1..13] { + let next = client.chainsync().request_next().await.unwrap(); + match next { + NextResponse::RollForward(content, _) => assert_eq!(&content.0, expected), + other => panic!("expected roll forward, got {other:?}"), + } + } + + client.chainsync().send_done().await.unwrap(); + + server.await.unwrap(); + } +}