diff --git a/Cargo.lock b/Cargo.lock index 31a1d6cb5..29630c901 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -557,7 +557,7 @@ checksum = "f36834a5c0a2fa56e171bf256c34d70fca07d0c0031583edea1c4946b7889c9e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -790,7 +790,7 @@ dependencies = [ "proc-macro-error2", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -808,7 +808,7 @@ dependencies = [ "proc-macro2", "quote", "sha3", - "syn 2.0.116", + "syn 2.0.118", "syn-solidity", ] @@ -826,7 +826,7 @@ dependencies = [ "proc-macro2", "quote", "serde_json", - "syn 2.0.116", + "syn 2.0.118", "syn-solidity", ] @@ -956,7 +956,7 @@ dependencies = [ "darling 0.23.0", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -1145,7 +1145,7 @@ checksum = "e7e89fe77d1f0f4fe5b96dfc940923d88d17b6a773808124f21e764dfb063c6a" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -1275,7 +1275,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "62945a2f7e6de02a31fe400aa489f0e0f5b2502e69f95f853adb82a96c7a6b60" dependencies = [ "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -1313,7 +1313,7 @@ dependencies = [ "num-traits", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -1486,7 +1486,7 @@ checksum = "213888f660fddcca0d257e88e54ac05bca01885f258ccdf695bafd77031bb69d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -1607,7 +1607,7 @@ dependencies = [ "rustc-hash", "serde", "serde_derive", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -1624,7 +1624,7 @@ dependencies = [ "rustc-hash", "serde", "serde_derive", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -1717,7 +1717,7 @@ checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -1728,7 +1728,7 @@ checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -1784,7 +1784,7 @@ checksum = "ffdcb70bdbc4d478427380519163274ac86e52916e10f0a8889adf0f96d3fee7" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -2262,7 +2262,35 @@ checksum = "604fde5e028fea851ce1d8570bbdc034bec850d157f7569d10f347d06808c05c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", +] + +[[package]] +name = "axum-test" +version = "20.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a86bfe2ef15bee102ac34912f7f4542b0bb37dc464fa55461763999c4d625e7" +dependencies = [ + "anyhow", + "axum", + "bytes", + "bytesize", + "cookie", + "expect-json", + "http 1.4.0", + "http-body-util", + "hyper", + "hyper-util", + "mime", + "pretty_assertions", + "reserve-port", + "rust-multipart-rfc7578_2", + "serde", + "serde_json", + "serde_urlencoded", + "tokio", + "tower", + "url", ] [[package]] @@ -2636,7 +2664,7 @@ dependencies = [ "proc-macro2", "quote", "rustversion", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -2659,7 +2687,7 @@ dependencies = [ "proc-macro-crate", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -2725,6 +2753,12 @@ dependencies = [ "either", ] +[[package]] +name = "bytesize" +version = "2.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d7c8918969267b2932ffd5655509bbbea0833823058c378876953217f5fc50e" + [[package]] name = "c-kzg" version = "2.1.5" @@ -2831,6 +2865,17 @@ dependencies = [ "cpufeatures 0.2.17", ] +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core 0.10.0", +] + [[package]] name = "chacha20poly1305" version = "0.10.1" @@ -2838,7 +2883,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35" dependencies = [ "aead", - "chacha20", + "chacha20 0.9.1", "cipher", "poly1305", "zeroize", @@ -2846,9 +2891,9 @@ dependencies = [ [[package]] name = "chrono" -version = "0.4.43" +version = "0.4.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fac4744fb15ae8337dc853fee7fb3f4e48c0fbaa23d0afe49c447b4fab126118" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" dependencies = [ "iana-time-zone", "js-sys", @@ -2949,7 +2994,7 @@ dependencies = [ "heck 0.5.0", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3053,7 +3098,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] @@ -3187,6 +3232,16 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "cookie" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ddef33a339a91ea89fb53151bd0a4689cfce27055c291dfa69945475d22c747" +dependencies = [ + "time", + "version_check", +] + [[package]] name = "core-foundation" version = "0.10.1" @@ -3437,7 +3492,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3452,7 +3507,7 @@ dependencies = [ "proc-macro2", "quote", "scratch", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3486,7 +3541,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3500,7 +3555,7 @@ dependencies = [ "quote", "serde", "strsim", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3511,7 +3566,7 @@ checksum = "d38308df82d1080de0afee5d069fa14b0326a88c14f15c5ccda35b4a6c414c81" dependencies = [ "darling_core 0.21.3", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3522,7 +3577,7 @@ checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" dependencies = [ "darling_core 0.23.0", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3553,7 +3608,7 @@ checksum = "780eb241654bf097afb00fc5f054a09b687dad862e485fdcf8399bb056565370" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3607,7 +3662,7 @@ checksum = "ef941ded77d15ca19b40374869ac6000af1c9f2a4c0f3d4c70926287e6364a8f" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3629,10 +3684,16 @@ dependencies = [ "proc-macro2", "quote", "rustc_version 0.4.1", - "syn 2.0.116", + "syn 2.0.118", "unicode-xid", ] +[[package]] +name = "diff" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" + [[package]] name = "digest" version = "0.9.0" @@ -3715,7 +3776,7 @@ checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3811,7 +3872,7 @@ dependencies = [ "enum-ordinalize", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3845,6 +3906,15 @@ dependencies = [ "zeroize", ] +[[package]] +name = "email_address" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e079f19b08ca6239f47f8ba8509c11cf3ea30095831f7fed61441475edd8c449" +dependencies = [ + "serde", +] + [[package]] name = "embed-doc-image" version = "0.1.4" @@ -3899,7 +3969,7 @@ dependencies = [ "heck 0.5.0", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3919,7 +3989,7 @@ checksum = "8ca9601fb2d62598ee17836250842873a413586e5d7ed88b356e38ddbb0ec631" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3931,7 +4001,7 @@ dependencies = [ "once_cell", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -3958,7 +4028,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4003,6 +4073,35 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "expect-json" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "869f97f4abe8e78fc812a94ad6b721d72c4fb5532877c79610f2c238d7ccf6c4" +dependencies = [ + "chrono", + "email_address", + "expect-json-macros", + "num", + "regex", + "serde", + "serde_json", + "thiserror 2.0.18", + "typetag", + "uuid", +] + +[[package]] +name = "expect-json-macros" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0637949cd816934f3b7aab44ff98e7ec1fb903c379e07dcb9eac943ec33499e" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "eyre" version = "0.6.12" @@ -4239,7 +4338,7 @@ checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -4373,6 +4472,7 @@ dependencies = [ "cfg-if", "libc", "r-efi", + "rand_core 0.10.0", "wasip2", "wasip3", ] @@ -4410,7 +4510,7 @@ checksum = "53010ccb100b96a67bc32c0175f0ed1426b31b655d562898e57325f81c023ac0" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -4620,7 +4720,7 @@ dependencies = [ "proc-macro-error2", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -5130,7 +5230,7 @@ checksum = "a0eb5a3343abf848c0984fe4604b2b105da9539376e24fc0a3b0007411ae4fd9" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -5278,7 +5378,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5370,7 +5470,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" dependencies = [ "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -5474,7 +5574,7 @@ dependencies = [ "proc-macro-crate", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -5728,7 +5828,7 @@ checksum = "1b27834086c65ec3f9387b096d66e99f221cf081c2b738042aa252bcd41204e3" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -5910,7 +6010,7 @@ checksum = "82a2620949467f1c0f7c468958fea01ee6b5d7aff67c42b35ca589e430881482" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -6165,7 +6265,7 @@ dependencies = [ "proc-macro-crate", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -6581,7 +6681,7 @@ dependencies = [ "proc-macro-crate", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -6635,7 +6735,7 @@ dependencies = [ "regex", "regex-syntax", "structmeta", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -6731,7 +6831,7 @@ dependencies = [ "pest_meta", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -6807,7 +6907,7 @@ dependencies = [ "phf_shared", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -6836,7 +6936,7 @@ checksum = "6e918e4ff8c4549eb882f14b3a4bc8c8bc93de829416eacf579f1207a8fbf861" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -6987,6 +7087,16 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "pretty_assertions" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ae130e2f271fbc2ac3a40fb1d07180839cdbbe443c7a27e1e3c13c5cac0116d" +dependencies = [ + "diff", + "yansi", +] + [[package]] name = "prettyplease" version = "0.2.37" @@ -6994,7 +7104,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" dependencies = [ "proc-macro2", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -7060,7 +7170,7 @@ dependencies = [ "proc-macro-error-attr2", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -7117,7 +7227,7 @@ version = "0.13.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ - "heck 0.5.0", + "heck 0.4.1", "itertools 0.14.0", "log", "multimap", @@ -7127,7 +7237,7 @@ dependencies = [ "prost 0.13.5", "prost-types 0.13.5", "regex", - "syn 2.0.116", + "syn 2.0.118", "tempfile", ] @@ -7141,7 +7251,7 @@ dependencies = [ "itertools 0.14.0", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -7154,7 +7264,7 @@ dependencies = [ "itertools 0.14.0", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -7831,14 +7941,14 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.52.0", ] [[package]] name = "quote" -version = "1.0.44" +version = "1.0.46" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "21b2ebcf727b7760c461f091f9f0f539b77b8e87f2fd88131e7f1b433b3cece4" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" dependencies = [ "proc-macro2", ] @@ -7888,6 +7998,17 @@ dependencies = [ "serde", ] +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20 0.10.1", + "getrandom 0.4.1", + "rand_core 0.10.0", +] + [[package]] name = "rand_chacha" version = "0.3.1" @@ -8094,7 +8215,7 @@ checksum = "b7186006dcb21920990093f30e3dea63b7d6e977bf1256be20c3563a5db070da" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -8212,6 +8333,15 @@ dependencies = [ "web-sys", ] +[[package]] +name = "reserve-port" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71ea98a177596a4579881992bd2bd4af27772fc95d0e5f5668a8f9535eca6380" +dependencies = [ + "thiserror 2.0.18", +] + [[package]] name = "revm" version = "36.0.0" @@ -8654,7 +8784,7 @@ dependencies = [ "proc-macro2", "quote", "rust-embed-utils", - "syn 2.0.116", + "syn 2.0.118", "walkdir", ] @@ -8678,6 +8808,21 @@ dependencies = [ "ordered-multimap", ] +[[package]] +name = "rust-multipart-rfc7578_2" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00bdaa068902270ca7fa8619775e1838e23a63620abac0947ce0f715819b8cec" +dependencies = [ + "bytes", + "futures-core", + "futures-util", + "http 1.4.0", + "mime", + "rand 0.10.2", + "thiserror 2.0.18", +] + [[package]] name = "rustc-demangle" version = "0.1.27" @@ -8727,7 +8872,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8807,7 +8952,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs 1.0.6", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8979,7 +9124,7 @@ checksum = "1783eabc414609e28a5ba76aee5ddd52199f7107a0b24c2e9746a1ecc34a683d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -9217,7 +9362,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -9252,7 +9397,7 @@ checksum = "175ee3e80ae9982737ca543e96133087cbd9a485eecc3bc4de9c1a37b47ea59c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -9313,7 +9458,7 @@ dependencies = [ "darling 0.21.3", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -9358,7 +9503,7 @@ checksum = "94e153fc76e1c6a068703d6d29c508a0b15c061c4b7e43da59cc097bc342673c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -9694,7 +9839,7 @@ dependencies = [ "quote", "sqlx-core", "sqlx-macros-core", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -9717,7 +9862,7 @@ dependencies = [ "sqlx-mysql", "sqlx-postgres", "sqlx-sqlite", - "syn 2.0.116", + "syn 2.0.118", "tokio", "url", ] @@ -9834,7 +9979,7 @@ dependencies = [ "aes", "aes-gcm", "cbc", - "chacha20", + "chacha20 0.9.1", "cipher", "ctr", "poly1305", @@ -9898,7 +10043,7 @@ dependencies = [ "proc-macro2", "quote", "structmeta-derive", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -9909,7 +10054,7 @@ checksum = "152a0b65a590ff6c3da95cabe2353ee04e6167c896b28e3b14478c2636c922fc" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -9949,7 +10094,7 @@ dependencies = [ "heck 0.5.0", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -9971,9 +10116,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.116" +version = "2.0.118" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3df424c70518695237746f84cede799c9c58fcb37450d7b23716568cc8bc69cb" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" dependencies = [ "proc-macro2", "quote", @@ -9989,7 +10134,7 @@ dependencies = [ "paste", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -10009,7 +10154,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -10157,13 +10302,17 @@ checksum = "15af1a54071def0649b8e55c8d87afff9cd537565345e939264313aeb7031fc1" dependencies = [ "alloy", "axum", + "axum-test", "backon", + "eyre", "futures", "git-version", "humantime-serde", + "reserve-port", "secrecy", "serde", "sqlx", + "testcontainers-modules", "thiserror 2.0.18", "tokio", "tokio-util", @@ -10454,7 +10603,7 @@ dependencies = [ "getrandom 0.4.1", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -10615,7 +10764,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -10626,7 +10775,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -10750,14 +10899,14 @@ checksum = "2d2e76690929402faae40aebdda620a2c0e25dd6d3b9afe48867dfd95991f4bd" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] name = "tokio" -version = "1.49.0" +version = "1.50.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72a2903cd7736441aac9df9d7688bd0ce48edccaadf181c3b90be801e81d3d86" +checksum = "27ad5e34374e03cfffefc301becb44e9dc3c17584f414349ebe29ed26661822d" dependencies = [ "bytes", "libc", @@ -10778,7 +10927,7 @@ checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -11090,7 +11239,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -11250,6 +11399,30 @@ version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb" +[[package]] +name = "typetag" +version = "0.2.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c5a897b12c6c1151ad0b138b8db50252dc301f93bc3b027db05eec82aeed298c" +dependencies = [ + "erased-serde", + "inventory", + "once_cell", + "serde", + "typetag-impl", +] + +[[package]] +name = "typetag-impl" +version = "0.2.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf808357c6ed7e13ba0f3277ec8d8f21b2d501274895104263985330c726c1c5" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "ucd-trie" version = "0.1.7" @@ -11389,7 +11562,7 @@ dependencies = [ "indexmap 2.13.0", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -11404,7 +11577,7 @@ dependencies = [ "proc-macro2", "quote", "serde", - "syn 2.0.116", + "syn 2.0.118", "toml 0.9.12+spec-1.1.0", "uniffi_meta", ] @@ -11552,7 +11725,7 @@ checksum = "6d79d08d92ab8af4c5e8a6da20c47ae3f61a0f1dabc1997cdf2d082b757ca08b" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -11712,7 +11885,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", "wasm-bindgen-shared", ] @@ -11876,7 +12049,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] @@ -11947,7 +12120,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -11958,7 +12131,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -12340,7 +12513,7 @@ dependencies = [ "heck 0.5.0", "indexmap 2.13.0", "prettyplease", - "syn 2.0.116", + "syn 2.0.118", "wasm-metadata", "wit-bindgen-core", "wit-component", @@ -12356,7 +12529,7 @@ dependencies = [ "prettyplease", "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", "wit-bindgen-core", "wit-bindgen-rust", ] @@ -12589,16 +12762,18 @@ dependencies = [ name = "world-id-oprf-accountant" version = "0.1.0" dependencies = [ - "alloy-primitives", + "alloy", "ark-serialize 0.5.0", "axum", "backon", "config", "eyre", + "futures-util", "humantime-serde", "itertools 0.14.0", "metrics", "rustls", + "secrecy", "serde", "serde_json", "sqlx", @@ -12609,6 +12784,7 @@ dependencies = [ "thiserror 2.0.18", "tikv-jemallocator", "tokio", + "tokio-util", "tracing", "world-id-primitives", ] @@ -12945,6 +13121,12 @@ dependencies = [ "hashlink", ] +[[package]] +name = "yansi" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" + [[package]] name = "yoke" version = "0.8.1" @@ -12964,7 +13146,7 @@ checksum = "b659052874eb698efe5b9e8cf382204678a0086ebf46982b79d6ca3182927e5d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", "synstructure", ] @@ -12985,7 +13167,7 @@ checksum = "4122cd3169e94605190e77839c9a40d40ed048d305bfdc146e7df40ab0f3e517" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -13005,7 +13187,7 @@ checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", "synstructure", ] @@ -13026,7 +13208,7 @@ checksum = "85a5b4158499876c763cb03bc4e49185d3cccbabb15b33c627f7884f43db852e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] @@ -13059,7 +13241,7 @@ checksum = "eadce39539ca5cb3985590102671f2567e659fca9666581ad3411d59207951f3" dependencies = [ "proc-macro2", "quote", - "syn 2.0.116", + "syn 2.0.118", ] [[package]] diff --git a/services/oprf-accountant/Cargo.toml b/services/oprf-accountant/Cargo.toml index dc0486c77..465bc33d4 100644 --- a/services/oprf-accountant/Cargo.toml +++ b/services/oprf-accountant/Cargo.toml @@ -8,7 +8,7 @@ repository.workspace = true license.workspace = true [dependencies] -alloy-primitives.workspace = true +alloy = { workspace = true, features = ["full", "rpc", "rpc-client-ws"] } ark-babyjubjub = { workspace = true } ark-serde-compat = { workspace = true } ark-serialize = { workspace = true } @@ -16,14 +16,16 @@ axum = { workspace = true, features = ["json", "macros"] } backon = { workspace = true, features = ["std", "tokio-sleep"] } config.workspace = true eyre = { workspace = true } +futures-util = { workspace = true } humantime-serde = { workspace = true } itertools.workspace = true metrics = { workspace = true } rustls = { workspace = true } +secrecy = { workspace = true, features = ["serde"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sqlx = { workspace = true, features = ["postgres", "runtime-tokio-rustls"] } -taceo-nodes-common = { workspace = true, features = ["api", "postgres"] } +taceo-nodes-common = { workspace = true, features = ["api", "postgres", "web3"] } telemetry-batteries = { workspace = true } thiserror.workspace = true tokio = { workspace = true, features = [ @@ -34,8 +36,15 @@ tokio = { workspace = true, features = [ "time", "tokio-macros", ] } +tokio-util = { workspace = true } tracing = { workspace = true, features = ["release_max_level_debug"] } world-id-primitives = { workspace = true } +[dev-dependencies] +taceo-nodes-common = { workspace = true, features = [ + "postgres", + "test-utils", +] } + [target.'cfg(not(target_env = "msvc"))'.dependencies] tikv-jemallocator = { workspace = true } diff --git a/services/oprf-accountant/migrations/20260703123015_init_db.down.sql b/services/oprf-accountant/migrations/20260703123015_init_db.down.sql index 8b22b3335..683b2ef3f 100644 --- a/services/oprf-accountant/migrations/20260703123015_init_db.down.sql +++ b/services/oprf-accountant/migrations/20260703123015_init_db.down.sql @@ -1 +1,2 @@ +DROP TABLE IF EXISTS epoch_cursor; DROP TABLE IF EXISTS rp_signatures; diff --git a/services/oprf-accountant/migrations/20260703123015_init_db.up.sql b/services/oprf-accountant/migrations/20260703123015_init_db.up.sql index 6bffac68b..f26110b16 100644 --- a/services/oprf-accountant/migrations/20260703123015_init_db.up.sql +++ b/services/oprf-accountant/migrations/20260703123015_init_db.up.sql @@ -1,5 +1,5 @@ -- Stores RP (relying party) signatures over proof requests so they can be verified later. -create table if not exists rp_signatures ( +CREATE TABLE IF NOT EXISTS rp_signatures ( id BIGINT GENERATED ALWAYS AS identity PRIMARY KEY, rp_id BIGINT NOT NULL, epoch BIGINT NOT NULL, @@ -10,10 +10,21 @@ create table if not exists rp_signatures ( signed_created_at BIGINT NOT NULL, signed_expires_at BIGINT NOT NULL, -- RP ECDSA (secp256k1) signature over the message (alloy Signature, 65 bytes) - signature BYTEA NOT NULL, + signature BYTEA, -- Record insertion timestamp (bookkeeping) created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), -- A given nonce should be signed at most once per RP (replay protection) constraint uq_rp_signatures_rp_id_nonce unique (rp_id, nonce, epoch) -); \ No newline at end of file +); + +-- Singleton table holding the last epoch the accountant has fully processed (i.e. voted +-- on). `id` is pinned to `true` so there can only ever be one row. Seeded at -1 (no epoch +-- processed yet) so the first tick starts from epoch 0. +CREATE TABLE IF NOT EXISTS epoch_cursor ( + id BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (id), + epoch BIGINT NOT NULL +); + +INSERT INTO epoch_cursor (epoch) +VALUES (-1); \ No newline at end of file diff --git a/services/oprf-accountant/src/accountant_service.rs b/services/oprf-accountant/src/accountant_service.rs new file mode 100644 index 000000000..4454fb66a --- /dev/null +++ b/services/oprf-accountant/src/accountant_service.rs @@ -0,0 +1,936 @@ +//! Aggregates recorded RP (relying party) OPRF request counts per epoch and submits them as +//! billing votes to the `BillingContract`. +//! +//! [`OprfAccountantService::run`] drives this on a timer: each +//! [`OprfAccountantService::tick`] checks whether the most recently closed epoch's voting +//! window is open (see [`get_votes_for_current_vote_window`]), and if so, aggregates that +//! epoch's recorded [`BillableRpRequest`]s by RP id (via [`PostgresDb`]) and submits the +//! resulting counts as this node's vote. The epoch cursor persisted in [`PostgresDb`] tracks +//! progress so a restart resumes from where it left off instead of re-voting for +//! already-processed epochs. +//! +//! `timing_eras` is seeded once from `getEras()` when the service is constructed (passed to +//! [`OprfAccountantService::new`]), then kept up to date by [`OprfAccountantService::tick`], +//! which re-fetches `getEras()` on every tick and refreshes the shared `Arc>` if it +//! changed. Since a tick runs at least once per voting window, no `TimingUpdated` change is +//! ever missed. + +use std::{ + sync::{Arc, Mutex}, + time::Duration, +}; + +use alloy::{primitives::Address, providers::DynProvider, signers::local::PrivateKeySigner, sol}; +use eyre::Context; +use itertools::Itertools as _; +use tokio::time::MissedTickBehavior; +use tokio_util::sync::CancellationToken; +use tracing::instrument; + +use crate::{ + accountant_service::IBillingContract::IBillingContractInstance, api::BillableRpRequest, + postgres::PostgresDb, +}; + +// TODO replace with abi +sol! { + #[derive(Debug, Copy, PartialEq, Eq)] + struct TimingEra { + // the first epoch governed by this era's parameters. + uint32 startEpoch; + // the timestamp the era starts at: the end of epoch `startEpoch - 1` (genesis for era 0). + uint64 startTime; + // the epoch length in seconds. + uint64 epochLength; + // the voting window in seconds (<= epochLength). + uint64 votingWindow; + // the payment window in seconds. + uint64 paymentWindow; + } + + /// @notice A single (rpId, count) entry inside a node's billing vote. + #[derive(Debug, PartialEq, Eq)] + struct RpCount { + // the Relying Party the count is reported for. + uint64 rpId; + // the number of unique requests the node observed for the RP in the epoch. + uint64 count; + } + + /// @notice One chunk of an OPRF node's signed billing vote for a single epoch. + /// @dev The signed payload is the EIP-712 + /// `BillingVoteChunk(uint32 epoch,uint32 chunkIndex,bool isFinal,RpCount[] counts)` struct, + /// where `epoch` is supplied as the call argument shared by every chunk in the batch. + struct SignedVoteChunk { + // the zero-based chunk index for this node and epoch. Chunks must be submitted in order. + uint32 chunkIndex; + // true on the final chunk; the node counts as a voter only after this chunk is accepted. + bool isFinal; + // the per-RP counts reported by the node, strictly ascending globally across chunks, + // all counts non-zero. + RpCount[] counts; + // the node's EIP-712 signature over the chunk. The signer is recovered, not trusted from + // msg.sender, so any party may relay the chunks. + bytes signature; + } + + #[sol(rpc)] + interface IBillingContract { + // Emitted when the epoch timing parameters are updated. + event TimingUpdated( + uint64 epochLength, uint64 votingWindow, uint64 paymentWindow, uint32 eraStartEpoch, uint64 eraStartTime + ); + + // The full timing-era history, oldest first; the last entry is the current era. + function getEras() external view returns (TimingEra[] memory); + + // Submit one or more OPRF node billing vote chunks for a single epoch. + // + // Records votes only; does not finalize as a side effect (finalization is driven solely by + // {finalizeEpochs}), so a node's vote gas never carries another epoch's finalization cost. + // Quorum and pricing are read live at finalization (no per-epoch snapshot). Authenticates + // by recovered signer, not msg.sender. A node's chunks must be submitted in order and the + // node counts toward quorum only after its final chunk is accepted. + function submitBillingVotes(uint32 epoch, SignedVoteChunk[] calldata chunks) external; + } +} + +sol! { + /// EIP-712 typed-data payload for a single [`SignedVoteChunk`] (without its `signature` + /// field, which is what's being computed). Used only for signature hashing/recovery, not + /// as a Solidity call type. + struct BillingVoteChunk { + uint32 epoch; + uint32 chunkIndex; + bool isFinal; + RpCount[] counts; + } +} + +/// Outcome of checking whether we can vote right now for the most recently closed epoch. +#[derive(Debug, PartialEq)] +enum VoteWindowResult { + /// The most recently closed epoch's voting window either hasn't opened yet, or it opened + /// but `voting_window_offset` hasn't elapsed yet (giving OPRF nodes time to flush their + /// batched requests to us before we aggregate). + NotOpen, + /// The most recently closed epoch's voting window has already closed; we can no longer + /// vote for it. + AlreadyClosed, + /// The most recently closed epoch's voting window is open now. + Vote { + // the epoch to vote for (the last fully closed epoch). + epoch: u32, + // the per-RP counts for the epoch, ascending by `rpId` (as required by `submitBillingVotes`). + counts: Vec, + }, +} + +/// Aggregates recorded RP request counts per epoch and submits them as billing votes. +/// +/// Constructed via [`OprfAccountantService::new`]; driven by repeatedly calling +/// [`OprfAccountantService::tick`] (typically via [`OprfAccountantService::run`]). +#[derive(Clone)] +#[expect( + dead_code, + reason = "unused fields are used in the future implementation of vote submission" +)] +pub(crate) struct OprfAccountantService { + contract: IBillingContractInstance, + billing_contract: Address, + chain_id: u64, + signer: PrivateKeySigner, + db: PostgresDb, + timing_eras: Arc>>, + submit_interval: Duration, + voting_window_offset: Duration, + cancellation_token: CancellationToken, +} + +/// Construction parameters for [`OprfAccountantService::new`]. +pub(crate) struct OprfAccountantServiceArgs { + /// Provider used to call and sign transactions against the `BillingContract`. + pub(crate) provider: DynProvider, + /// Chain id `provider` is connected to. + pub(crate) chain_id: u64, + /// Address of the `BillingContract`. + pub(crate) billing_contract: Address, + /// Signer this node uses to sign its billing votes. + pub(crate) signer: PrivateKeySigner, + /// Database storing recorded RP requests and the epoch cursor. + pub(crate) db: PostgresDb, + /// Initial `TimingEra` history of the `BillingContract`. + /// Must have at least one entry (the `BillingContract` adds its first `TimingEra` on initialization). + pub(crate) timing_eras: Vec, + /// How often [`OprfAccountantService::run`] calls [`OprfAccountantService::tick`]. Must be + /// less than the current era's `epochLength` and `votingWindow` (see [`validate_timing`]). + pub(crate) submit_interval: Duration, + /// Extra delay after an epoch's voting window opens before we aggregate and vote for it, + /// giving OPRF nodes time to flush their batched requests to us. Must be less than the + /// current era's `votingWindow` (see [`validate_timing`]). + pub(crate) voting_window_offset: Duration, + /// Signals [`OprfAccountantService::run`] to stop. + pub(crate) cancellation_token: CancellationToken, +} + +impl OprfAccountantService { + /// Constructs the service: builds the `BillingContract` binding and validates + /// `submit_interval`/`voting_window_offset` against the current timing era. + /// + /// # Errors + /// Returns an error if the current era's timing parameters fail [`validate_timing`]. + /// + /// # Panics + /// Panics if `timing_eras` is empty. + pub(crate) async fn new( + OprfAccountantServiceArgs { + provider, + chain_id, + billing_contract, + signer, + db, + timing_eras, + submit_interval, + voting_window_offset, + cancellation_token, + }: OprfAccountantServiceArgs, + ) -> eyre::Result { + let contract = IBillingContract::new(billing_contract, provider); + validate_timing( + timing_eras.last().expect("at least one timing era"), + submit_interval, + voting_window_offset, + )?; + + Ok(Self { + contract, + billing_contract, + chain_id, + signer, + db, + timing_eras: Arc::new(Mutex::new(timing_eras)), + submit_interval, + voting_window_offset, + cancellation_token, + }) + } + + /// Persists a batch of [`BillableRpRequest`]s, bucketed into the epoch each request's + /// `expires_at` falls into. + /// + /// This reads `timing_eras` from the shared cache rather than fetching it fresh, so there's + /// a brief window — up to one `submit_interval` — after a `TimingUpdated` event where a + /// request landing right at the new era's boundary could be bucketed using the stale era. + /// This is fine: the cache is refreshed at least once per voting window (see + /// [`OprfAccountantService::tick`]), so it self-corrects on the very next tick, and + /// `TimingUpdated` (an owner-only, operational change) is expected to be rare. + /// + /// # Errors + /// Returns an error if the underlying database write fails. + #[instrument( + level = "trace", + skip_all, + name = "accountant_service::record_rp_request_batch" + )] + pub(crate) async fn record_rp_request_batch( + &self, + rp_requests: Vec, + ) -> eyre::Result<()> { + tracing::trace!( + num_requests = rp_requests.len(), + "recording RP request batch" + ); + let epochs = { + let timing_eras = self.timing_eras.lock().expect("not poisoned"); + rp_requests + .iter() + .map(|r| epoch_for_timestamp(&timing_eras, r.expires_at)) + .collect_vec() + }; + self.db + .store_request_batch(epochs, rp_requests) + .await + .context("while storing request batch") + } + + /// Runs the service until `cancellation_token` is cancelled, calling [`Self::tick`] every + /// `submit_interval`. + /// + /// A tick that returns an error is logged and retried on the next tick rather than stopping + /// the loop, since a transient failure (e.g. a dropped DB connection) shouldn't prevent + /// later epochs from being voted on. + pub(crate) async fn run(&self) { + tracing::info!("starting OprfAccountant worker"); + + let mut interval = tokio::time::interval(self.submit_interval); + // Burst would also be fine, but it is not needed here, Skip would be bad because it could skip epochs + interval.set_missed_tick_behavior(MissedTickBehavior::Delay); + + loop { + tokio::select! { + _ = interval.tick() => { + if let Err(err) = self.tick().await { + tracing::error!(error = ?err, "accountant tick failed; retrying next tick"); + } + } + _ = self.cancellation_token.cancelled() => { + tracing::info!("shutdown signal received, stopping account worker"); + break; + } + } + } + } + + /// Determines whether we can vote right now for the most recently closed epoch as of `now` + /// (see [`VoteWindowResult`]), aggregating that epoch's recorded counts if so. + /// + /// # Errors + /// Returns an error if the current era's timing parameters fail [`validate_timing`], or if + /// fetching the epoch's request counts from the database fails. + #[instrument( + level = "trace", + skip_all, + name = "accountant_service::get_votes_for_current_vote_window" + )] + async fn get_votes_for_current_vote_window(&self, now: u64) -> eyre::Result { + let eras = self.timing_eras.lock().expect("not poisoned").clone(); + validate_timing( + eras.last().expect("at least one timing era"), + self.submit_interval, + self.voting_window_offset, + )?; + + let current_epoch = epoch_for_timestamp(&eras, now); + if current_epoch == 0 { + // no epochs have closed yet, so we have nothing to vote for. + return Ok(VoteWindowResult::NotOpen); + } + + // the epoch we vote for is the last fully closed epoch (i.e. is strictly before the epoch `now` falls into). + let epoch = current_epoch - 1; + let (window_open, window_close) = voting_window(&eras, epoch); + let submit_after = window_open + self.voting_window_offset.as_secs(); + + if now < submit_after { + // not enough time has passed since the epoch closed for OPRF nodes' batched + // requests to have flushed to us yet; wait and pick this epoch back up next tick. + tracing::trace!( + epoch, + window_open, + window_close, + submit_after, + now, + "voting window not open yet" + ); + return Ok(VoteWindowResult::NotOpen); + } + + if now >= window_close { + tracing::warn!( + epoch, + window_open, + window_close, + submit_after, + now, + "voting window closed before we could vote for this epoch" + ); + return Ok(VoteWindowResult::AlreadyClosed); + } + + let counts = self.db.rp_counts_for_epoch(epoch).await?; + Ok(VoteWindowResult::Vote { epoch, counts }) + } + + /// Submits `counts` as this node's billing vote for `epoch`. A no-op if `counts` is empty. + /// + /// # Errors + /// Returns an error if submitting the vote fails. + async fn vote_for_epoch(&self, _epoch: u32, counts: Vec) -> eyre::Result<()> { + if counts.is_empty() { + return Ok(()); + } + + // TODO + + Ok(()) + } + + /// One iteration of the accounting loop: refreshes the local timing-era cache from the + /// contract (see the module docs), then checks whether the most recently closed epoch's + /// voting window is open (see [`Self::get_votes_for_current_vote_window`]) and, if so and we + /// haven't already voted for it, submits the vote and advances the epoch cursor. + /// + /// Logs a warning if the epoch cursor is more than one epoch behind the epoch we're about to + /// vote for, meaning one or more epochs were skipped (e.g. after downtime longer than a + /// voting window). + /// + /// # Errors + /// Returns an error if fetching the timing eras, checking the vote window, submitting the + /// vote, or updating the epoch cursor fails. + #[instrument(level = "info", skip_all, name = "accountant_service::tick")] + pub(crate) async fn tick(&self) -> eyre::Result<()> { + let now = now_unix_timestamp(); + + // Fetch the latest timing eras from the contract and update our local copy if it has changed. + // This ensures that we always have the most up-to-date timing information for determining voting windows. + // Current epoch is not affected by timing changes, new parameters affect later epochs only. + // Ticks should happen at least once per `voting_window` (which is guaranteed to be at most `epochLength`), + // so we should never miss a timing update. + let timing_eras = self + .contract + .getEras() + .call() + .await + .context("while fetching initial timing eras")?; + { + let mut current_timing_eras = self.timing_eras.lock().expect("not poisoned"); + if timing_eras != *current_timing_eras { + tracing::info!( + ?timing_eras, + ?current_timing_eras, + "timing eras changed; updating" + ); + *current_timing_eras = timing_eras; + } + } + + match self.get_votes_for_current_vote_window(now).await? { + VoteWindowResult::NotOpen => { + tracing::trace!("voting window not open yet"); + } + VoteWindowResult::AlreadyClosed => { + tracing::warn!("voting window already closed"); + } + VoteWindowResult::Vote { epoch, counts } => { + let epoch_cursor = self.db.get_epoch_cursor().await?; + if (epoch as i64).abs_diff(epoch_cursor) > 1 { + tracing::warn!( + epoch, + epoch_cursor, + "epoch cursor is more than one behind; some epochs may have been skipped" + ); + } + if epoch as i64 > epoch_cursor { + tracing::info!(epoch, ?counts, "submitting billing vote for epoch"); + self.vote_for_epoch(epoch, counts).await?; + tracing::info!(epoch, "successfully submitted billing vote for epoch"); + self.db.set_epoch_cursor(epoch).await?; + } else { + tracing::trace!(epoch, epoch_cursor, "already voted for this epoch"); + } + } + } + + Ok(()) + } +} + +/// Validates `tick_interval` and `voting_window_offset` against `era`'s timing parameters. +/// +/// # Errors +/// Returns an error if `era.votingWindow` is not at most `era.epochLength` (consecutive epochs' +/// voting windows could overlap), if `voting_window_offset` is not less than `era.votingWindow` +/// (votes would never be submitted), or if `tick_interval` is not less than `era.votingWindow` +/// (a tick can walk straight past an epoch's voting window). +fn validate_timing( + era: &TimingEra, + tick_interval: Duration, + voting_window_offset: Duration, +) -> eyre::Result<()> { + eyre::ensure!( + era.votingWindow <= era.epochLength, + "era votingWindow ({}s) must be at most the epochLength ({}s), or voting windows for \ + consecutive epochs could overlap", + era.votingWindow, + era.epochLength + ); + eyre::ensure!( + voting_window_offset.as_secs() < era.votingWindow, + "voting_window_offset ({voting_window_offset:?}) must be less than the current voting \ + window ({}s), or votes would never be submitted", + era.votingWindow + ); + eyre::ensure!( + tick_interval.as_secs() < era.votingWindow, + "tick_interval ({tick_interval:?}) must be less than the current voting window \ + ({}s), or a tick can walk straight past an epoch's voting window", + era.votingWindow + ); + Ok(()) +} + +/// Returns the era whose timespan `[startTime, nextEra.startTime)` contains `timestamp`: the +/// most recent era that had already started by `timestamp`, falling back to the oldest era for +/// timestamps that predate it. +/// +/// # Panics +/// Panics if `eras` is empty. +fn era_for_timestamp(eras: &[TimingEra], timestamp: u64) -> &TimingEra { + eras.iter() + .rev() + .find(|era| era.startTime <= timestamp) + .or_else(|| eras.first()) + .expect("at least one timing era") +} + +/// Returns the era governing `epoch`'s span: the era with the largest `startEpoch <= epoch`, +/// falling back to the oldest era for epochs older than any known era. +/// +/// # Panics +/// Panics if `eras` is empty. +fn era_for_epoch(eras: &[TimingEra], epoch: u32) -> &TimingEra { + eras.iter() + .rev() + .find(|era| era.startEpoch <= epoch) + .or_else(|| eras.first()) + .expect("at least one timing era") +} + +/// Returns the epoch a timestamp falls into, given the full timing-era history (oldest first, +/// as returned by [`IBillingContract::getEras`]): the epoch `e` such that +/// `epoch_start(e) <= timestamp < epoch_end(e)`, per the boundary formula documented on the +/// `BillingContract`'s `epochEnd` function. +/// +/// # Panics +/// Panics if `eras` is empty. +pub(crate) fn epoch_for_timestamp(eras: &[TimingEra], timestamp: u64) -> u32 { + let era = era_for_timestamp(eras, timestamp); + let elapsed_epochs = timestamp.saturating_sub(era.startTime) / era.epochLength; + era.startEpoch + .saturating_add(u32::try_from(elapsed_epochs).unwrap_or(u32::MAX)) +} + +/// Returns the timestamp at which `epoch` ends (and, per the contract's window semantics, at +/// which its voting window opens), using the era governing `epoch`'s span. +fn epoch_end(eras: &[TimingEra], epoch: u32) -> u64 { + let span_era = era_for_epoch(eras, epoch); + let epochs_since_era_start = u64::from(epoch - span_era.startEpoch) + 1; + span_era.startTime + epochs_since_era_start * span_era.epochLength +} + +/// Returns the `[open, close)` voting window for `epoch`. The window opens when the epoch ends +/// and uses the `votingWindow` of whichever era governs at that instant — which, for the last +/// epoch before a timing change, can differ from the era governing the epoch's span. +fn voting_window(eras: &[TimingEra], epoch: u32) -> (u64, u64) { + let open = epoch_end(eras, epoch); + let era = era_for_timestamp(eras, open); + (open, open + era.votingWindow) +} + +/// Returns the current unix timestamp in seconds. +fn now_unix_timestamp() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock before unix epoch") + .as_secs() +} + +#[cfg(test)] +mod tests { + use std::{num::NonZeroU32, time::Duration}; + + use alloy::{ + primitives::Address, + providers::{Provider as _, ProviderBuilder}, + signers::local::PrivateKeySigner, + transports::mock::Asserter, + }; + use secrecy::SecretString; + use taceo_nodes_common::{ + postgres::PostgresConfig, + test_utils::{next_test_schema, shared_postgres_testcontainer}, + }; + use tokio_util::sync::CancellationToken; + use world_id_primitives::rp::RpId; + + use super::{ + OprfAccountantService, RpCount, TimingEra, VoteWindowResult, epoch_end, + epoch_for_timestamp, era_for_epoch, era_for_timestamp, voting_window, + }; + use crate::{ + accountant_service::OprfAccountantServiceArgs, api::BillableRpRequest, postgres::PostgresDb, + }; + + /// The exact walkthrough from the `TimingEra` doc comment: era 0 = + /// `{startEpoch: 0, startTime: 1000, len: 100, vote: 80, pay: 200}`, and `setTiming(50, 40, + /// 100)` is called at t=1250 (inside epoch 2), appending era 1 = `{startEpoch: 3, startTime: + /// 1300, len: 50, vote: 40, pay: 100}`. + fn example_eras() -> Vec { + vec![ + TimingEra { + startEpoch: 0, + startTime: 1000, + epochLength: 100, + votingWindow: 80, + paymentWindow: 200, + }, + TimingEra { + startEpoch: 3, + startTime: 1300, + epochLength: 50, + votingWindow: 40, + paymentWindow: 100, + }, + ] + } + + #[test] + fn era_for_timestamp_returns_the_only_era_when_there_is_one() { + let eras = vec![example_eras()[0]]; + assert_eq!(era_for_timestamp(&eras, 1000).startEpoch, 0); + assert_eq!(era_for_timestamp(&eras, 50_000).startEpoch, 0); + } + + #[test] + fn era_for_timestamp_falls_back_to_the_oldest_era_before_it_started() { + let eras = vec![example_eras()[0]]; + // 500 predates the only era's startTime (1000). + assert_eq!(era_for_timestamp(&eras, 500).startEpoch, 0); + } + + #[test] + fn era_for_timestamp_picks_the_most_recently_started_era() { + let eras = example_eras(); + // strictly inside era 0's span + assert_eq!(era_for_timestamp(&eras, 1250).startEpoch, 0); + // exactly at era 1's start: era 1 governs from here on + assert_eq!(era_for_timestamp(&eras, 1300).startEpoch, 3); + // strictly inside era 1's span + assert_eq!(era_for_timestamp(&eras, 1350).startEpoch, 3); + } + + #[test] + fn era_for_epoch_returns_the_only_era_when_there_is_one() { + let eras = vec![example_eras()[0]]; + assert_eq!(era_for_epoch(&eras, 0).startEpoch, 0); + assert_eq!(era_for_epoch(&eras, 100).startEpoch, 0); + } + + #[test] + fn era_for_epoch_falls_back_to_the_oldest_era_for_epochs_older_than_any_known_era() { + let old_era = TimingEra { + startEpoch: 5, + ..example_eras()[0] + }; + let eras = vec![old_era]; + assert_eq!(era_for_epoch(&eras, 2).startEpoch, 5); + } + + #[test] + fn era_for_epoch_picks_the_era_with_the_largest_start_epoch_at_or_below_epoch() { + let eras = example_eras(); + assert_eq!(era_for_epoch(&eras, 0).startEpoch, 0); + assert_eq!(era_for_epoch(&eras, 2).startEpoch, 0); + assert_eq!(era_for_epoch(&eras, 3).startEpoch, 3); + assert_eq!(era_for_epoch(&eras, 10).startEpoch, 3); + } + + #[test] + fn epoch_for_timestamp_within_a_single_era() { + let eras = vec![example_eras()[0]]; + // epoch 0 spans [1000, 1100) + assert_eq!(epoch_for_timestamp(&eras, 1000), 0); + assert_eq!(epoch_for_timestamp(&eras, 1050), 0); + // epoch 1 spans [1100, 1200) + assert_eq!(epoch_for_timestamp(&eras, 1100), 1); + assert_eq!(epoch_for_timestamp(&eras, 1199), 1); + } + + #[test] + fn epoch_for_timestamp_matches_the_example_across_an_era_change() { + let eras = example_eras(); + // t=1250: "inside epoch 2", per the doc comment, while era 0's parameters still apply + // to the epoch's span even though era 1 has already been appended. + assert_eq!(epoch_for_timestamp(&eras, 1250), 2); + // epoch 3 spans [1300, 1350) under era 1's shorter length. + assert_eq!(epoch_for_timestamp(&eras, 1300), 3); + assert_eq!(epoch_for_timestamp(&eras, 1349), 3); + // epoch 4 spans [1350, 1400). + assert_eq!(epoch_for_timestamp(&eras, 1350), 4); + } + + #[test] + fn epoch_end_matches_the_example() { + let eras = example_eras(); + // epoch 1 [1100, 1200) — era-0 span + assert_eq!(epoch_end(&eras, 1), 1200); + // epoch 2 [1200, 1300) — era-0 span (epoch 2 keeps era 0's length) + assert_eq!(epoch_end(&eras, 2), 1300); + // epoch 3 [1300, 1350) — era-1 span + assert_eq!(epoch_end(&eras, 3), 1350); + } + + #[test] + fn voting_window_matches_the_example() { + let eras = example_eras(); + + // epoch 1: window [1200, 1280) — era-0 span, era-0 window. + assert_eq!(voting_window(&eras, 1), (1200, 1280)); + // epoch 2: window [1300, 1340) — era-0 span, but era-1 window, since the window opens + // exactly at the era boundary and is thus governed by the new era. + assert_eq!(voting_window(&eras, 2), (1300, 1340)); + // epoch 3: window [1350, 1390) — era-1 span, era-1 window. + assert_eq!(voting_window(&eras, 3), (1350, 1390)); + } + + /// Builds an [`OprfAccountantService`] backed by `eras`, a chain-less mock provider, and a + /// fresh Postgres schema in the shared testcontainer. + async fn setup_service(timing_eras: Vec) -> OprfAccountantService { + let connection_string = shared_postgres_testcontainer() + .await + .expect("shared postgres testcontainer starts"); + let mut db_config = PostgresConfig::with_default_values( + SecretString::from(connection_string.to_owned()), + next_test_schema(), + ); + db_config.max_connections = NonZeroU32::new(1).expect("non-zero"); + let db = PostgresDb::init(&db_config) + .await + .expect("postgres db initializes"); + + OprfAccountantService::new(OprfAccountantServiceArgs { + provider: ProviderBuilder::new() + .connect_mocked_client(Asserter::new()) + .erased(), + chain_id: 31337, + billing_contract: Address::ZERO, + signer: PrivateKeySigner::random(), + db, + timing_eras, + submit_interval: Duration::from_secs(1), + voting_window_offset: Duration::from_secs(5), + cancellation_token: CancellationToken::new(), + }) + .await + .expect("service constructs") + } + + fn request(rp_id: u64, nonce: u64, expires_at: u64) -> BillableRpRequest { + BillableRpRequest { + rp_id: RpId::new(rp_id), + nonce: ark_babyjubjub::Fq::from(nonce), + created_at: expires_at, + expires_at, + action: ark_babyjubjub::Fq::from(0u64), + signature: None, + } + } + + #[tokio::test] + async fn not_ready_before_first_epoch_closes() { + let service = setup_service(example_eras()).await; + + // still inside epoch 0's own span [1000, 1100); nothing has closed yet. + let result = service + .get_votes_for_current_vote_window(1050) + .await + .expect("computes vote window"); + + assert_eq!(result, VoteWindowResult::NotOpen); + } + + #[tokio::test] + async fn not_ready_while_waiting_for_flush_offset() { + let service = setup_service(example_eras()).await; + + // epoch 0's window opened at 1100, but submit_after (1105) hasn't passed yet. + let result = service + .get_votes_for_current_vote_window(1102) + .await + .expect("computes vote window"); + + assert_eq!(result, VoteWindowResult::NotOpen); + } + + #[tokio::test] + async fn already_closed_after_voting_window_passes() { + let service = setup_service(example_eras()).await; + + // epoch 0's window [1100, 1180) has closed by 1190, while `now` is still inside epoch + // 1's span [1100, 1200). + let result = service + .get_votes_for_current_vote_window(1190) + .await + .expect("computes vote window"); + + assert_eq!(result, VoteWindowResult::AlreadyClosed); + } + + #[tokio::test] + async fn votes_with_no_recorded_requests_are_empty() { + let service = setup_service(example_eras()).await; + + // past submit_after (1105), still inside the window (closes at 1180). + let result = service + .get_votes_for_current_vote_window(1110) + .await + .expect("computes vote window"); + + assert_eq!( + result, + VoteWindowResult::Vote { + epoch: 0, + counts: vec![] + } + ); + } + + #[tokio::test] + async fn votes_aggregate_recorded_requests_by_rp_id() { + let service = setup_service(example_eras()).await; + + // requests for both epoch 0 [1000, 1100) and epoch 1 [1100, 1200), to check that each + // epoch's vote only counts its own requests. + service + .record_rp_request_batch(vec![ + request(5, 1, 1010), + request(5, 2, 1020), + request(7, 3, 1030), + request(9, 4, 1150), + ]) + .await + .expect("requests are recorded"); + + let epoch0 = service + .get_votes_for_current_vote_window(1110) + .await + .expect("computes vote window"); + assert_eq!( + epoch0, + VoteWindowResult::Vote { + epoch: 0, + counts: vec![RpCount { rpId: 5, count: 2 }, RpCount { rpId: 7, count: 1 }], + } + ); + + // epoch 1's window is [1200, 1280); submit_after is 1205. + let epoch1 = service + .get_votes_for_current_vote_window(1210) + .await + .expect("computes vote window"); + assert_eq!( + epoch1, + VoteWindowResult::Vote { + epoch: 1, + counts: vec![RpCount { rpId: 9, count: 1 }], + } + ); + } + + #[tokio::test] + async fn nonce_can_be_reused_only_across_different_epochs() { + let service = setup_service(example_eras()).await; + + // rp 13's nonce 99 is used twice within epoch 0's span [1000, 1100) — the second is a + // replay and must not be double-counted — and once more in epoch 1's span + // [1100, 1200), where reusing the same (rp_id, nonce) pair is fine since the + // uniqueness constraint is scoped per epoch. + service + .record_rp_request_batch(vec![ + request(13, 99, 1010), + request(13, 99, 1020), + request(13, 99, 1150), + ]) + .await + .expect("requests are recorded"); + + let epoch0 = service + .get_votes_for_current_vote_window(1110) + .await + .expect("computes vote window"); + assert_eq!( + epoch0, + VoteWindowResult::Vote { + epoch: 0, + counts: vec![RpCount { rpId: 13, count: 1 }], + } + ); + + // epoch 1's window is [1200, 1280); submit_after is 1205. + let epoch1 = service + .get_votes_for_current_vote_window(1210) + .await + .expect("computes vote window"); + assert_eq!( + epoch1, + VoteWindowResult::Vote { + epoch: 1, + counts: vec![RpCount { rpId: 13, count: 1 }], + } + ); + } + + #[tokio::test] + async fn not_ready_before_submit_after_at_era_boundary() { + let service = setup_service(example_eras()).await; + + // epoch 2's window opens exactly at era 1's start (1300) with era 1's votingWindow (40), + // but submit_after (1300+5=1305) hasn't passed yet. + let result = service + .get_votes_for_current_vote_window(1301) + .await + .expect("computes vote window"); + + assert_eq!(result, VoteWindowResult::NotOpen); + } + + #[tokio::test] + async fn already_closed_for_epoch_governed_by_era_boundary_window() { + let service = setup_service(example_eras()).await; + + // epoch 2's window [1300, 1340) has closed by 1345, while `now` is still inside epoch + // 3's span [1300, 1350). + let result = service + .get_votes_for_current_vote_window(1345) + .await + .expect("computes vote window"); + + assert_eq!(result, VoteWindowResult::AlreadyClosed); + } + + #[tokio::test] + async fn votes_isolate_requests_across_era_boundary() { + let service = setup_service(example_eras()).await; + + // requests for both epoch 2's era-0 span [1200, 1300) and epoch 3's era-1 span + // [1300, 1350), to check that each epoch's vote only counts its own requests, even + // across the era boundary. + service + .record_rp_request_batch(vec![ + request(9, 1, 1250), + request(9, 2, 1260), + request(4, 3, 1270), + request(11, 4, 1310), + request(11, 5, 1320), + ]) + .await + .expect("requests are recorded"); + + // epoch 2 [1200, 1300) is still era-0 span, but its window [1300, 1340) is era-1's, + // since the window opens exactly at the era boundary. + let epoch2 = service + .get_votes_for_current_vote_window(1310) + .await + .expect("computes vote window"); + assert_eq!( + epoch2, + VoteWindowResult::Vote { + epoch: 2, + counts: vec![RpCount { rpId: 4, count: 1 }, RpCount { rpId: 9, count: 2 }], + } + ); + + // epoch 3 [1300, 1350) is fully governed by era 1; its window is [1350, 1390). + let epoch3 = service + .get_votes_for_current_vote_window(1360) + .await + .expect("computes vote window"); + assert_eq!( + epoch3, + VoteWindowResult::Vote { + epoch: 3, + counts: vec![RpCount { rpId: 11, count: 2 }], + } + ); + } +} diff --git a/services/oprf-accountant/src/api.rs b/services/oprf-accountant/src/api.rs index 3fbad2e3b..c3f14a5a6 100644 --- a/services/oprf-accountant/src/api.rs +++ b/services/oprf-accountant/src/api.rs @@ -1,8 +1,10 @@ +use alloy::signers::Signature; use axum::{Json, Router, extract::State, http::StatusCode, response::IntoResponse, routing::post}; use serde::{Deserialize, Serialize}; +use tracing::instrument; use world_id_primitives::{oprf::NullifierOprfRequestAuthV1, rp::RpId}; -use crate::{AppState, postgres::PostgresDb}; +use crate::{AppState, accountant_service::OprfAccountantService}; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct BillableRpRequest { @@ -13,8 +15,7 @@ pub struct BillableRpRequest { pub expires_at: u64, #[serde(with = "ark_serde_compat::field")] pub action: ark_babyjubjub::Fq, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub signature: Option, + pub signature: Option, } impl From<&NullifierOprfRequestAuthV1> for BillableRpRequest { @@ -39,13 +40,21 @@ impl From<&NullifierOprfRequestAuthV1> for BillableRpRequest { } } +#[instrument(level = "info", skip_all)] async fn post_request( - State(db): State, + State(accountant): State, Json(rp_requests): Json>, ) -> impl IntoResponse { - // TODO get the epochs for the RpVote from the contract - let _ = db.store_request_batch(rp_requests).await; - StatusCode::OK + match accountant.record_rp_request_batch(rp_requests).await { + Ok(()) => { + tracing::trace!("Successfully recorded RP request batch"); + StatusCode::OK + } + Err(err) => { + tracing::error!(?err, "Failed to record RP request batch: {err}"); + StatusCode::INTERNAL_SERVER_ERROR + } + } } pub(crate) fn routes() -> Router { diff --git a/services/oprf-accountant/src/config.rs b/services/oprf-accountant/src/config.rs index 126d24137..c59dcdc55 100644 --- a/services/oprf-accountant/src/config.rs +++ b/services/oprf-accountant/src/config.rs @@ -1,7 +1,11 @@ //! Configuration types and environment parsing for the OPRF accountant. +use std::time::Duration; + +use alloy::primitives::Address; +use secrecy::SecretString; use serde::Deserialize; -use taceo_nodes_common::postgres::PostgresConfig; +use taceo_nodes_common::{postgres::PostgresConfig, web3}; /// The configuration for the OPRF accountant service. /// @@ -12,4 +16,40 @@ pub struct OprfAccountantConfig { /// The postgres config #[serde(rename = "postgres")] pub postgres_config: PostgresConfig, + + /// The address of the `BillingContract` smart contract + pub billing_contract: Address, + + /// The interval in which to check if we can submit votes to the `BillingContract`. + /// + /// Must less than the `votingWindow` the contract is configured with, + /// otherwise the accountant may miss an epoch's voting window entirely. + #[serde( + default = "OprfAccountantConfig::default_submit_interval", + with = "humantime_serde" + )] + pub tick_interval: Duration, + + /// A additional offset to be added to the start of the voting window. + /// + /// This is used to ensure that the OPRF nodes have enough time to send their requests + /// to the accountant before the voting window starts. + /// The offset should be at least be 2x the flush interval of the `AccountantBatcher`. + pub voting_window_offset: Duration, + + /// The blockchain RPC config + #[serde(rename = "rpc")] + pub rpc_provider_config: web3::HttpRpcProviderConfig, + + // TODO split into 1 for the EIP-712 signature and 1 for the transaction signing. + /// The private key used to sign billing vote chunks and to submit `submitBillingVotes` + /// transactions. + pub wallet_private_key: SecretString, +} + +impl OprfAccountantConfig { + /// Default submit interval + fn default_submit_interval() -> Duration { + Duration::from_secs(30) + } } diff --git a/services/oprf-accountant/src/lib.rs b/services/oprf-accountant/src/lib.rs index 4690f3a1f..726d9070a 100644 --- a/services/oprf-accountant/src/lib.rs +++ b/services/oprf-accountant/src/lib.rs @@ -21,24 +21,74 @@ //! //! It provides an Axum based HTTP server. +use alloy::{providers::Provider, signers::local::PrivateKeySigner}; use axum::{Router, extract::FromRef}; +use secrecy::ExposeSecret as _; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; -use crate::{config::OprfAccountantConfig, postgres::PostgresDb}; +use crate::{ + accountant_service::{IBillingContract, OprfAccountantService, OprfAccountantServiceArgs}, + config::OprfAccountantConfig, + postgres::PostgresDb, +}; +pub mod accountant_service; pub mod api; pub mod config; pub mod metrics; pub mod postgres; #[derive(Clone)] -struct AppState(PostgresDb); +struct AppState { + accountant: OprfAccountantService, +} -impl FromRef for PostgresDb { +impl FromRef for OprfAccountantService { fn from_ref(input: &AppState) -> Self { - input.0.clone() + input.accountant.clone() } } -pub async fn start(_config: &OprfAccountantConfig, db: PostgresDb) -> Router { - Router::new().merge(api::routes()).with_state(AppState(db)) +pub async fn start( + config: &OprfAccountantConfig, + db: PostgresDb, + cancellation_token: CancellationToken, +) -> eyre::Result<(Router, JoinHandle<()>)> { + let provider = + taceo_nodes_common::web3::HttpRpcProviderBuilder::with_config(&config.rpc_provider_config) + .build()? + .inner(); + let signer: PrivateKeySigner = config.wallet_private_key.expose_secret().parse()?; + + let contract = IBillingContract::new(config.billing_contract, provider.clone()); + let timing_eras = contract.getEras().call().await?; + let chain_id = provider.get_chain_id().await?; + + let accountant = OprfAccountantService::new(OprfAccountantServiceArgs { + provider, + chain_id, + billing_contract: config.billing_contract, + signer, + db: db.clone(), + timing_eras, + submit_interval: config.tick_interval, + voting_window_offset: config.voting_window_offset, + cancellation_token: cancellation_token.clone(), + }) + .await?; + + let accountant_task = tokio::spawn({ + let accountant = accountant.clone(); + async move { + let _guard = cancellation_token.drop_guard_ref(); + accountant.run().await + } + }); + + let app_state = AppState { accountant }; + + let router = Router::new().merge(api::routes()).with_state(app_state); + + Ok((router, accountant_task)) } diff --git a/services/oprf-accountant/src/main.rs b/services/oprf-accountant/src/main.rs index 090daeb48..6c54a29e7 100644 --- a/services/oprf-accountant/src/main.rs +++ b/services/oprf-accountant/src/main.rs @@ -12,7 +12,7 @@ use tikv_jemallocator::Jemalloc; #[global_allocator] static GLOBAL: Jemalloc = Jemalloc; -use std::{net::SocketAddr, process::ExitCode}; +use std::{net::SocketAddr, process::ExitCode, time::Duration}; use config::{Config, Environment}; use eyre::Context; @@ -24,6 +24,12 @@ struct FullOprfAccountantConfig { /// The bind addr of the AXUM server #[serde(default = "default_bind_addr")] pub bind_addr: SocketAddr, + + /// Max wait time the service waits for its workers during shutdown. + #[serde(default = "default_max_wait_shutdown")] + #[serde(with = "humantime_serde")] + pub max_wait_time_shutdown: Duration, + /// The OPRF accountant service config #[serde(rename = "service")] pub service_config: OprfAccountantConfig, @@ -59,6 +65,10 @@ fn default_bind_addr() -> SocketAddr { "0.0.0.0:4322".parse().expect("valid SocketAddr") } +fn default_max_wait_shutdown() -> Duration { + Duration::from_secs(10) +} + async fn run(config: FullOprfAccountantConfig) -> eyre::Result<()> { tracing::info!("{}", taceo_nodes_common::version_info!()); tracing::info!("starting oprf-accountant with config: {config:#?}"); @@ -69,22 +79,61 @@ async fn run(config: FullOprfAccountantConfig) -> eyre::Result<()> { let db = PostgresDb::init(&config.service_config.postgres_config).await?; // Clone the values we need afterwards let bind_addr = config.bind_addr; + let max_wait_time_shutdown = config.max_wait_time_shutdown; - let accountant_router = world_id_oprf_accountant::start(&config.service_config, db).await; + let (accountant_router, accountant_task) = + world_id_oprf_accountant::start(&config.service_config, db, cancellation_token.clone()) + .await?; let router = Router::new() .merge(taceo_nodes_common::api::routes( taceo_nodes_common::version_info!(), )) .merge(accountant_router); - tracing::info!("starting axum server on {bind_addr}",); - let listener = tokio::net::TcpListener::bind(bind_addr).await?; - axum::serve(listener, router) - .with_graceful_shutdown(async move { cancellation_token.cancelled().await }) - .await - .context("while serving axum")?; - tracing::info!("axum server shutdown"); - Ok(()) + + let server = tokio::spawn({ + let cancellation_token = cancellation_token.clone(); + async move { + // we cancel the token if this task closes for some reason + let _drop_guard = cancellation_token.drop_guard_ref(); + tracing::info!("starting axum server on to {bind_addr}"); + let tcp_listener = tokio::net::TcpListener::bind(bind_addr) + .await + .context("while binding tcp-listener")?; + let axum_result = axum::serve(tcp_listener, router) + .with_graceful_shutdown({ + let cancellation_token = cancellation_token.clone(); + async move { cancellation_token.cancelled().await } + }) + .await + .context("while running axum"); + tracing::info!("axum server shutdown"); + axum_result + } + }); + + tracing::info!("everything started successfully - now waiting for shutdown..."); + cancellation_token.cancelled().await; + + tracing::info!("waiting for shutdown of services (max wait time {max_wait_time_shutdown:?}).."); + + match tokio::time::timeout(max_wait_time_shutdown, async move { + let (axum_result, accountant_result) = tokio::join!(server, accountant_task); + axum_result??; + accountant_result?; + eyre::Ok(()) + }) + .await + { + Ok(Ok(_)) => { + tracing::info!("successfully finished shutdown in time"); + Ok(()) + } + Ok(Err(err)) => Err(err), + Err(_) => { + eyre::bail!("could not finish shutdown in time"); + } + } } fn main() -> ExitCode { diff --git a/services/oprf-accountant/src/postgres.rs b/services/oprf-accountant/src/postgres.rs index 274014928..bfa371f6d 100644 --- a/services/oprf-accountant/src/postgres.rs +++ b/services/oprf-accountant/src/postgres.rs @@ -6,7 +6,7 @@ use sqlx::PgPool; use taceo_nodes_common::postgres::{CreateSchema, PostgresConfig}; use tracing::instrument; -use crate::api::BillableRpRequest; +use crate::{accountant_service::RpCount, api::BillableRpRequest}; type Result = std::result::Result; @@ -52,16 +52,18 @@ impl PostgresDb { } pub(crate) async fn store_request_batch( - self, + &self, + epochs: Vec, rp_requests: Vec, ) -> Result<()> { + if epochs.len() != rp_requests.len() { + return Err(eyre::eyre!("epochs and rp_requests must have the same length").into()); + } let rp_ids = rp_requests .iter() .map(|r| r.rp_id.into_inner() as i64) .collect_vec(); - // let epochs: Vec = rp_requests.iter().map(|r| r.epoch).collect(); - // TODO get epochs - let epochs: Vec = Vec::new(); + let epochs = epochs.iter().map(|e| *e as i64).collect_vec(); let nonces = rp_requests .iter() .map(|r| to_db_ark_serialize_uncompressed(&r.nonce)) @@ -111,6 +113,65 @@ impl PostgresDb { Ok(()) } + /// Returns the last epoch the accountant submitted votes for, if any. + /// + /// Starts at -1 if no epoch has been processed yet. + pub(crate) async fn get_epoch_cursor(&self) -> Result { + let query = || async { + Ok( + sqlx::query_scalar::<_, i64>("SELECT epoch FROM epoch_cursor WHERE id") + .fetch_one(&self.pool) + .await?, + ) + }; + self.with_retry("get-epoch-cursor", query).await + } + + /// Records `epoch` as the last epoch the accountant voted for. + pub(crate) async fn set_epoch_cursor(&self, epoch: u32) -> Result<()> { + let epoch = i64::from(epoch); + let query = || async { + Ok(sqlx::query( + " + INSERT INTO epoch_cursor (id, epoch) VALUES (TRUE, $1) + ON CONFLICT (id) DO UPDATE SET epoch = EXCLUDED.epoch + ", + ) + .bind(epoch) + .execute(&self.pool) + .await?) + }; + self.with_retry("set-epoch-cursor", query).await?; + Ok(()) + } + + /// Returns the number of requests observed per RP for `epoch`, ascending by `rp_id` (as + /// required by `submitBillingVotes`). + pub(crate) async fn rp_counts_for_epoch(&self, epoch: u32) -> Result> { + let epoch = i64::from(epoch); + let query = || async { + Ok(sqlx::query_as::<_, (i64, i64)>( + " + SELECT rp_id, COUNT(*) FROM rp_signatures + WHERE epoch = $1 + GROUP BY rp_id + ORDER BY rp_id + ", + ) + .bind(epoch) + .fetch_all(&self.pool) + .await?) + }; + let rows = self.with_retry("rp-counts-for-epoch", query).await?; + Ok(rows + .into_iter() + .map(|(rp_id, count)| RpCount { + rpId: rp_id as u64, + count: count as u64, + }) + .collect()) + } + async fn with_retry(&self, op_name: &str, f: F) -> Result where F: Fn() -> Fut,