From 1ef970c7e23bf2524f5367388de0e8e6dc93b94e Mon Sep 17 00:00:00 2001 From: lorban Date: Tue, 9 Jun 2026 12:50:54 +0200 Subject: [PATCH] Restore initial channel from configured extranonce Allow the translator to reuse a known upstream extranonce prefix when it starts. The restore_channel config entry or RESTORE_CHANNEL environment variable is parsed into bytes and stored in runtime configuration. Use that value for the first OpenExtendedMiningChannel request by building a restore-channel user identity from the RESTORE-CHANNEL marker and the configured prefix. When the pool returns OpenExtendedMiningChannelSuccess, validate that the returned extranonce prefix matches the configured value and then clear restore state. Update channel bookkeeping, bridge restore handling, and tests for the configured extranonce flow. --- Cargo.lock | 271 +++++++++---------- src/config.rs | 54 +++- src/jd_client/job_declarator/mod.rs | 21 +- src/jd_client/mining_downstream/mod.rs | 300 ++++++++++++++++++++-- src/jd_client/mining_upstream/mod.rs | 1 + src/jd_client/mining_upstream/upstream.rs | 211 ++++++++++++++- src/lib.rs | 9 +- src/router/mod.rs | 11 +- src/translator/mod.rs | 46 +++- src/translator/proxy/bridge.rs | 182 ++++++++++++- src/translator/upstream/upstream.rs | 236 +++++++++++++---- tests/library_init.rs | 4 +- 12 files changed, 1075 insertions(+), 271 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 07580b9b..01d79648 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -305,7 +305,7 @@ dependencies = [ [[package]] name = "binary_sv2" version = "5.0.1" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "buffer_sv2 3.0.1", "derive_codec_sv2 1.1.2", @@ -313,9 +313,9 @@ dependencies = [ [[package]] name = "bitcoin" -version = "0.32.8" +version = "0.32.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e499f9fc0407f50fe98af744ab44fa67d409f76b6772e1689ec8485eb0c0f66" +checksum = "9cf93e61f2dbc3e3c41234ca26a65e2c0b0975c52e0f069ab9893ebbede584d3" dependencies = [ "base58ck", "base64 0.21.7", @@ -332,9 +332,9 @@ dependencies = [ [[package]] name = "bitcoin-capnp-types" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e20759e30b46af17a13f2e34c9e090c3672938b5a0c22358cba971d1a8f5d492" +checksum = "a58af19b421a85566a1d46617f40b18eea77d565a0271bdff2deabf0d27c075c" dependencies = [ "capnp", "capnpc", @@ -367,8 +367,8 @@ dependencies = [ [[package]] name = "bitcoin_core_sv2" -version = "0.1.1" -source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ecc9eb8ddabe24e4999b671107a60e993c8d788f" +version = "0.3.0" +source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ccdef6d6574c54a9bd2627709b38703773292765" dependencies = [ "async-channel", "bitcoin-capnp-types", @@ -469,7 +469,7 @@ dependencies = [ [[package]] name = "buffer_sv2" version = "3.0.1" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "aes-gcm", ] @@ -500,18 +500,18 @@ checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" [[package]] name = "capnp" -version = "0.21.7" +version = "0.25.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e92edec8974fcd7ece90bb021db782abe14a61c10c817f197f700fef7430eb8" +checksum = "63da65e5e9ffc3b8f993d4ad222a548152549351a643f6b850a7773cb6ff2809" dependencies = [ "embedded-io", ] [[package]] name = "capnp-futures" -version = "0.21.0" +version = "0.25.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d04478adeb234836f886ec554a0d96e3af3a939ba7b3962af5addddf7ab71231" +checksum = "73b69dfddccc57844f9a90f9d72b44b97c326914851ea94fb7da40ef9cad6e8d" dependencies = [ "capnp", "futures-channel", @@ -520,9 +520,9 @@ dependencies = [ [[package]] name = "capnp-rpc" -version = "0.21.0" +version = "0.25.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85e9c19ef52ff1b9c9822fb21bfa68a72bc58711676295ff06eb88e64c7877f7" +checksum = "e3c74c337e87f75f3174ffb3513738ab0acc631c04f64cc33b867c60f84771da" dependencies = [ "capnp", "capnp-futures", @@ -531,18 +531,18 @@ dependencies = [ [[package]] name = "capnpc" -version = "0.21.4" +version = "0.25.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6da96dcb0a0e0c526daf42bac55e1550f18ad973df9ef9ba75204f332c80ad16" +checksum = "fca02be865c8c5a78bfc24b9819006ab6b59bef238467203928e26459557af93" dependencies = [ "capnp", ] [[package]] name = "cc" -version = "1.2.61" +version = "1.2.62" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d16d90359e986641506914ba71350897565610e87ce0ad9e6f28569db3dd5c6d" +checksum = "a1dce859f0832a7d088c4f1119888ab94ef4b5d6795d1ce05afb7fe159d79f98" dependencies = [ "find-msvc-tools", "shlex", @@ -586,14 +586,14 @@ dependencies = [ [[package]] name = "channels_sv2" -version = "5.0.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +version = "6.0.0" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", "bitcoin", - "mining_sv2 9.0.0", + "mining_sv2 10.0.0", "primitive-types", - "template_distribution_sv2 5.0.0", + "template_distribution_sv2 5.1.0", "tracing", ] @@ -665,7 +665,7 @@ dependencies = [ [[package]] name = "codec_sv2" version = "5.0.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", "buffer_sv2 3.0.1", @@ -692,8 +692,8 @@ dependencies = [ [[package]] name = "common_messages_sv2" -version = "7.1.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +version = "7.2.0" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", ] @@ -972,9 +972,9 @@ dependencies = [ [[package]] name = "dashmap" -version = "6.1.0" +version = "6.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf" +checksum = "e6361d5c062261c78a176addb82d4c821ae42bed6089de0e12603cd25de2059c" dependencies = [ "cfg-if", "crossbeam-utils", @@ -1040,7 +1040,7 @@ dependencies = [ [[package]] name = "derive_codec_sv2" version = "1.1.2" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" [[package]] name = "derive_codec_sv2" @@ -1184,9 +1184,9 @@ checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" [[package]] name = "embedded-io" -version = "0.6.1" +version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" +checksum = "9eb1aa714776b75c7e67e1da744b81a129b3ff919c8712b5e1b32252c1f07cc7" [[package]] name = "encode_unicode" @@ -1228,7 +1228,7 @@ checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" [[package]] name = "extensions_sv2" version = "0.1.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", ] @@ -1247,13 +1247,12 @@ checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" [[package]] name = "filetime" -version = "0.2.27" +version = "0.2.29" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f98844151eee8917efc50bd9e8318cb963ae8b297431495d3f758616ea5c57db" +checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" dependencies = [ "cfg-if", "libc", - "libredox", ] [[package]] @@ -1333,7 +1332,7 @@ dependencies = [ [[package]] name = "framing_sv2" version = "6.0.1" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", "buffer_sv2 3.0.1", @@ -1496,9 +1495,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.13" +version = "0.4.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54" +checksum = "171fefbc92fe4a4de27e0698d6a5b392d6a0e333506bc49133760b3bcf948733" dependencies = [ "atomic-waker", "bytes", @@ -1515,17 +1514,17 @@ dependencies = [ [[package]] name = "handlers_sv2" -version = "0.3.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +version = "0.4.0" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", - "common_messages_sv2 7.1.0", + "common_messages_sv2 7.2.0", "extensions_sv2", "framing_sv2 6.0.1", - "job_declaration_sv2 7.0.0", - "mining_sv2 9.0.0", + "job_declaration_sv2 7.1.0", + "mining_sv2 10.0.0", "parsers_sv2", - "template_distribution_sv2 5.0.0", + "template_distribution_sv2 5.1.0", "trait-variant", ] @@ -1550,9 +1549,9 @@ dependencies = [ [[package]] name = "hashbrown" -version = "0.17.0" +version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" [[package]] name = "hashlink" @@ -1601,9 +1600,12 @@ dependencies = [ [[package]] name = "hex-conservative" -version = "1.0.1" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "366fa3443ac84474447710ec17bb00b05dfbd096137817981e86f992f21a2793" +checksum = "a7289f6b628ce69fb1a371d0fdcf8ff38cd93ec00e3010eb055d1e044998c8d1" +dependencies = [ + "arrayvec", +] [[package]] name = "hex_lit" @@ -1886,7 +1888,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", - "hashbrown 0.17.0", + "hashbrown 0.17.1", "serde", "serde_core", ] @@ -1915,8 +1917,8 @@ dependencies = [ [[package]] name = "integration_tests_sv2" -version = "0.1.1" -source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ecc9eb8ddabe24e4999b671107a60e993c8d788f" +version = "0.3.0" +source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ccdef6d6574c54a9bd2627709b38703773292765" dependencies = [ "async-channel", "clap", @@ -1944,16 +1946,6 @@ version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" -[[package]] -name = "iri-string" -version = "0.7.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "25e659a4bb38e810ebc252e53b5814ff908a8c58c2a9ce2fae1bbec24cbf4e20" -dependencies = [ - "memchr", - "serde", -] - [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -1968,8 +1960,8 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "jd_client_sv2" -version = "0.1.4" -source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ecc9eb8ddabe24e4999b671107a60e993c8d788f" +version = "0.3.0" +source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ccdef6d6574c54a9bd2627709b38703773292765" dependencies = [ "async-channel", "bitcoin_core_sv2", @@ -1985,8 +1977,8 @@ dependencies = [ [[package]] name = "jd_server_sv2" -version = "0.1.2" -source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ecc9eb8ddabe24e4999b671107a60e993c8d788f" +version = "0.3.0" +source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ccdef6d6574c54a9bd2627709b38703773292765" dependencies = [ "async-channel", "async-trait", @@ -2030,17 +2022,17 @@ dependencies = [ [[package]] name = "job_declaration_sv2" -version = "7.0.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +version = "7.1.0" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", ] [[package]] name = "js-sys" -version = "0.3.97" +version = "0.3.98" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1840c94c045fbcf8ba2812c95db44499f7c64910a912551aaaa541decebcacf" +checksum = "67df7112613f8bfd9150013a0314e196f4800d3201ae742489d999db2f979f08" dependencies = [ "cfg-if", "futures-util", @@ -2123,10 +2115,7 @@ version = "0.1.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e02f3bb43d335493c96bf3fd3a321600bf6bd07ed34bc64118e9293bdffea46c" dependencies = [ - "bitflags", "libc", - "plain", - "redox_syscall 0.7.4", ] [[package]] @@ -2217,8 +2206,8 @@ dependencies = [ [[package]] name = "mining_sv2" -version = "9.0.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +version = "10.0.0" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", ] @@ -2231,7 +2220,7 @@ checksum = "867b1f11e0545ad5ebbddd8a9f18756d9657a6758bf10d96cf15ddd0b726b650" dependencies = [ "bech32", "bitcoin", - "hex-conservative 1.0.1", + "hex-conservative 1.1.0", ] [[package]] @@ -2307,7 +2296,7 @@ dependencies = [ [[package]] name = "noise_sv2" version = "1.4.2" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "aes-gcm", "chacha20poly1305", @@ -2394,15 +2383,14 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "openssl" -version = "0.10.78" +version = "0.10.80" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f38c4372413cdaaf3cc79dd92d29d7d9f5ab09b51b10dded508fb90bb70b9222" +checksum = "a45fa2aa886c42762255da344f0a0d313e254066c46aad76f300c3d3da62d967" dependencies = [ "bitflags", "cfg-if", "foreign-types", "libc", - "once_cell", "openssl-macros", "openssl-sys", ] @@ -2426,9 +2414,9 @@ checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" [[package]] name = "openssl-sys" -version = "0.9.114" +version = "0.9.116" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "13ce1245cd07fcc4cfdb438f7507b0c7e4f3849a69fd84d52374c66d83741bb6" +checksum = "f28a22dc7140cda5f096e5e7724a6962ca81a7f8bfd2979f9b18c11af56318c4" dependencies = [ "cc", "libc", @@ -2498,23 +2486,23 @@ checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" dependencies = [ "cfg-if", "libc", - "redox_syscall 0.5.18", + "redox_syscall", "smallvec", "windows-link", ] [[package]] name = "parsers_sv2" -version = "0.3.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +version = "0.4.0" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", - "common_messages_sv2 7.1.0", + "common_messages_sv2 7.2.0", "extensions_sv2", "framing_sv2 6.0.1", - "job_declaration_sv2 7.0.0", - "mining_sv2 9.0.0", - "template_distribution_sv2 5.0.0", + "job_declaration_sv2 7.1.0", + "mining_sv2 10.0.0", + "template_distribution_sv2 5.1.0", ] [[package]] @@ -2603,12 +2591,6 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" -[[package]] -name = "plain" -version = "0.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" - [[package]] name = "poly1305" version = "0.8.0" @@ -2634,8 +2616,8 @@ dependencies = [ [[package]] name = "pool_sv2" -version = "0.2.2" -source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ecc9eb8ddabe24e4999b671107a60e993c8d788f" +version = "0.4.0" +source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ccdef6d6574c54a9bd2627709b38703773292765" dependencies = [ "async-channel", "bitcoin_core_sv2", @@ -2919,15 +2901,6 @@ dependencies = [ "bitflags", ] -[[package]] -name = "redox_syscall" -version = "0.7.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f450ad9c3b1da563fb6948a8e0fb0fb9269711c9c73d9ea1de5058c79c8d643a" -dependencies = [ - "bitflags", -] - [[package]] name = "redox_users" version = "0.5.2" @@ -3505,9 +3478,9 @@ checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214" [[package]] name = "siphasher" -version = "1.0.2" +version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b2aa850e253778c88a04c3d7323b043aeda9d3e30d5971937c1855769763678e" +checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" [[package]] name = "slab" @@ -3555,8 +3528,8 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" [[package]] name = "stratum-apps" -version = "0.4.0" -source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ecc9eb8ddabe24e4999b671107a60e993c8d788f" +version = "0.5.0" +source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ccdef6d6574c54a9bd2627709b38703773292765" dependencies = [ "async-channel", "axum", @@ -3592,36 +3565,36 @@ dependencies = [ [[package]] name = "stratum-core" -version = "0.3.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +version = "0.4.0" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", "bitcoin", "buffer_sv2 3.0.1", "channels_sv2", "codec_sv2 5.0.0", - "common_messages_sv2 7.1.0", + "common_messages_sv2 7.2.0", "extensions_sv2", "framing_sv2 6.0.1", "handlers_sv2", - "job_declaration_sv2 7.0.0", - "mining_sv2 9.0.0", + "job_declaration_sv2 7.1.0", + "mining_sv2 10.0.0", "noise_sv2 1.4.2", "parsers_sv2", "stratum_translation", "sv1_api 4.0.0", - "template_distribution_sv2 5.0.0", + "template_distribution_sv2 5.1.0", ] [[package]] name = "stratum_translation" -version = "0.2.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +version = "0.3.0" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", "bitcoin", "channels_sv2", - "mining_sv2 9.0.0", + "mining_sv2 10.0.0", "sv1_api 4.0.0", "tracing", ] @@ -3655,7 +3628,7 @@ dependencies = [ [[package]] name = "sv1_api" version = "4.0.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", "bitcoin_hashes 0.3.2", @@ -3778,8 +3751,8 @@ dependencies = [ [[package]] name = "template_distribution_sv2" -version = "5.0.0" -source = "git+https://github.com/stratum-mining/stratum?branch=main#f14226ae146c9dcde1e7191ccb8fe185e1751d3c" +version = "5.1.0" +source = "git+https://github.com/stratum-mining/stratum?branch=main#7af1b737065ce238b8e39d846edac05c0abe143f" dependencies = [ "binary_sv2 5.0.1", ] @@ -3900,9 +3873,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.52.1" +version = "1.52.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" dependencies = [ "bytes", "libc", @@ -4014,7 +3987,7 @@ dependencies = [ "indexmap", "toml_datetime 1.1.1+spec-1.1.0", "toml_parser", - "winnow 1.0.2", + "winnow 1.0.3", ] [[package]] @@ -4023,7 +3996,7 @@ version = "1.1.2+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526" dependencies = [ - "winnow 1.0.2", + "winnow 1.0.3", ] [[package]] @@ -4050,20 +4023,20 @@ dependencies = [ [[package]] name = "tower-http" -version = "0.6.8" +version = "0.6.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8" +checksum = "68d6fdd9f81c2819c9a8b0e0cd91660e7746a8e6ea2ba7c6b2b057985f6bcb51" dependencies = [ "bitflags", "bytes", "futures-util", "http", "http-body", - "iri-string", "pin-project-lite", "tower", "tower-layer", "tower-service", + "url", ] [[package]] @@ -4166,8 +4139,8 @@ dependencies = [ [[package]] name = "translator_sv2" -version = "0.2.3" -source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ecc9eb8ddabe24e4999b671107a60e993c8d788f" +version = "0.3.0" +source = "git+https://github.com/stratum-mining/sv2-apps.git?branch=main#ccdef6d6574c54a9bd2627709b38703773292765" dependencies = [ "async-channel", "clap", @@ -4291,9 +4264,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" [[package]] name = "utoipa" -version = "5.4.0" +version = "5.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2fcc29c80c21c31608227e0912b2d7fddba57ad76b606890627ba8ee7964e993" +checksum = "8bde15df68e80b16c7d16b9616e80770ad158988daa56a27dccd1e55558b0160" dependencies = [ "indexmap", "serde", @@ -4303,9 +4276,9 @@ dependencies = [ [[package]] name = "utoipa-gen" -version = "5.4.0" +version = "5.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6d79d08d92ab8af4c5e8a6da20c47ae3f61a0f1dabc1997cdf2d082b757ca08b" +checksum = "6ba0b99ee52df3028635d93840c797102da61f8a7bb3cf751032455895b52ef8" dependencies = [ "proc-macro2", "quote", @@ -4394,9 +4367,9 @@ dependencies = [ [[package]] name = "wasm-bindgen" -version = "0.2.120" +version = "0.2.121" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df52b6d9b87e0c74c9edfa1eb2d9bf85e5d63515474513aa50fa181b3c4f5db1" +checksum = "49ace1d07c165b0864824eee619580c4689389afa9dc9ed3a4c75040d82e6790" dependencies = [ "cfg-if", "once_cell", @@ -4407,9 +4380,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-futures" -version = "0.4.70" +version = "0.4.71" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af934872acec734c2d80e6617bbb5ff4f12b052dd8e6332b0817bce889516084" +checksum = "96492d0d3ffba25305a7dc88720d250b1401d7edca02cc3bcd50633b424673b8" dependencies = [ "js-sys", "wasm-bindgen", @@ -4417,9 +4390,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.120" +version = "0.2.121" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78b1041f495fb322e64aca85f5756b2172e35cd459376e67f2a6c9dffcedb103" +checksum = "8e68e6f4afd367a562002c05637acb8578ff2dea1943df76afb9e83d177c8578" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -4427,9 +4400,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.120" +version = "0.2.121" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dcd0ff20416988a18ac686d4d4d0f6aae9ebf08a389ff5d29012b05af2a1b41" +checksum = "d95a9ec35c64b2a7cb35d3fead40c4238d0940c86d107136999567a4703259f2" dependencies = [ "bumpalo", "proc-macro2", @@ -4440,9 +4413,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.120" +version = "0.2.121" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49757b3c82ebf16c57d69365a142940b384176c24df52a087fb748e2085359ea" +checksum = "c4e0100b01e9f0d03189a92b96772a1fb998639d981193d7dbab487302513441" dependencies = [ "unicode-ident", ] @@ -4483,9 +4456,9 @@ dependencies = [ [[package]] name = "web-sys" -version = "0.3.97" +version = "0.3.98" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2eadbac71025cd7b0834f20d1fe8472e8495821b4e9801eb0a60bd1f19827602" +checksum = "4b572dff8bcf38bad0fa19729c89bb5748b2b9b1d8be70cf90df697e3a8f32aa" dependencies = [ "js-sys", "wasm-bindgen", @@ -4820,9 +4793,9 @@ dependencies = [ [[package]] name = "winnow" -version = "1.0.2" +version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2ee1708bef14716a11bae175f579062d4554d95be2c6829f518df847b7b3fdd0" +checksum = "0592e1c9d151f854e6fd382574c3a0855250e1d9b2f99d9281c6e6391af352f1" dependencies = [ "memchr", ] @@ -5002,9 +4975,9 @@ dependencies = [ [[package]] name = "zerofrom" -version = "0.1.7" +version = "0.1.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "69faa1f2a1ea75661980b013019ed6687ed0e83d069bc1114e2cc74c6c04c4df" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" dependencies = [ "zerofrom-derive", ] diff --git a/src/config.rs b/src/config.rs index ea12b192..b7d45f12 100644 --- a/src/config.rs +++ b/src/config.rs @@ -95,6 +95,7 @@ struct ConfigFile { tp_address: Option, api_base_url: Option, pool_addresses: Option>, + restore_channel: Option, interval: Option, delay: Option, downstream_hashrate: Option, @@ -130,6 +131,7 @@ impl ConfigFile { tp_address: None, api_base_url: None, pool_addresses: None, + restore_channel: None, interval: None, delay: None, downstream_hashrate: None, @@ -166,6 +168,7 @@ pub struct Configuration { tp_address: Option, api_base_url: Option, pool_addresses: Vec, + restore_channel: Option>, interval: u64, delay: u64, downstream_hashrate: f32, @@ -228,6 +231,7 @@ and make that test pass." tp_address: Option, api_base_url: Option, pool_addresses: Vec, + restore_channel: Option>, interval: u64, delay: u64, downstream_hashrate: f32, @@ -275,6 +279,7 @@ and make that test pass." tp_address, api_base_url, pool_addresses, + restore_channel, interval, delay, downstream_hashrate, @@ -361,10 +366,11 @@ and make that test pass." None, None, Vec::new(), + None, 120_000, 0, DEFAULT_SV1_HASHPOWER, - crate::translator::downstream::diff_management::NON_LOCAL_DOWNSTREAM_MIN_DIFFICULTY, + crate::translator::NON_LOCAL_DOWNSTREAM_MIN_DIFFICULTY, "info".to_string(), "off".to_string(), false, @@ -432,6 +438,10 @@ and make that test pass." } } + pub fn restore_channel() -> Option> { + Self::cfg().restore_channel.clone() + } + pub async fn pool_address() -> Option> { match fetch_pool_urls().await { Ok(addresses) => Some(addresses), @@ -661,6 +671,10 @@ and make that test pass." "DMND_CLIENT_POOL_ADDRESSES", ], ); + let restore_channel = config + .restore_channel + .or_else(|| std::env::var("RESTORE_CHANNEL").ok()) + .map(Self::parse_restore_channel); let miner_name = args .miner_name @@ -749,9 +763,7 @@ and make that test pass." .ok() .and_then(|s| parse_downstream_min_difficulty(&s).ok()) }) - .unwrap_or( - crate::translator::downstream::diff_management::NON_LOCAL_DOWNSTREAM_MIN_DIFFICULTY, - ); + .unwrap_or(crate::translator::NON_LOCAL_DOWNSTREAM_MIN_DIFFICULTY); println!("Using downstream minimum difficulty: {downstream_min_difficulty}"); let listening_addr = args.listening_addr.or(config.listening_addr).or_else(|| { @@ -873,6 +885,7 @@ and make that test pass." tp_address, api_base_url, pool_addresses, + restore_channel, interval, delay, downstream_hashrate, @@ -903,6 +916,39 @@ and make that test pass." api_tx_token, ) } + + fn parse_restore_channel(value: String) -> Vec { + let trimmed = value.trim(); + if let Some(hex) = trimmed + .strip_prefix("0x") + .or_else(|| trimmed.strip_prefix("hex:")) + { + return Self::parse_restore_channel_hex(hex).unwrap_or_else(|error| panic!("{error}")); + } + + value.into_bytes() + } + + fn parse_restore_channel_hex(hex: &str) -> Result, String> { + if hex.len() % 2 != 0 { + return Err( + "Invalid restore_channel hex value: expected an even number of hex digits" + .to_string(), + ); + } + + hex.as_bytes() + .chunks(2) + .map(|pair| { + let pair = std::str::from_utf8(pair).map_err(|_| { + "Invalid restore_channel hex value: expected ASCII hex digits".to_string() + })?; + u8::from_str_radix(pair, 16).map_err(|_| { + format!("Invalid restore_channel hex byte `{pair}`: expected 00 through ff") + }) + }) + .collect() + } } /// Parses a hashrate string (e.g., "10T", "2.5P", "500E") into an f32 value in h/s. diff --git a/src/jd_client/job_declarator/mod.rs b/src/jd_client/job_declarator/mod.rs index b74c2afc..a20a5fcd 100644 --- a/src/jd_client/job_declarator/mod.rs +++ b/src/jd_client/job_declarator/mod.rs @@ -57,7 +57,6 @@ pub struct JobDeclarator { sender: TSender>>, allocated_tokens: Vec>, req_ids: Id, - min_extranonce_size: u16, // (Sent DeclareMiningJob, is future, template id, merkle path) last_declare_mining_jobs_sent: HashMap>, pub last_set_new_prev_hash: Option>, @@ -100,8 +99,6 @@ impl JobDeclarator { info!("JD CONNECTED"); } - let min_extranonce_size = crate::MIN_EXTRANONCE_SIZE; - let task_manager = TaskManager::initialize(); let abortable = task_manager .safe_lock(|t| t.get_aborter()) @@ -111,7 +108,6 @@ impl JobDeclarator { sender, allocated_tokens: vec![], req_ids: Id::new(), - min_extranonce_size, last_declare_mining_jobs_sent: HashMap::with_capacity(2), last_set_new_prev_hash: None, future_jobs: HashMap::with_hasher(BuildNoHashHasher::default()), @@ -241,8 +237,8 @@ impl JobDeclarator { } super::IS_CUSTOM_JOB_SET.store(false, std::sync::atomic::Ordering::Release); // now as u64 unix time - let (id, _, sender) = self_mutex - .safe_lock(|s| (s.req_ids.next(), s.min_extranonce_size, s.sender.clone())) + let (id, sender) = self_mutex + .safe_lock(|s| (s.req_ids.next(), s.sender.clone())) .map_err(|_| Error::JobDeclaratorMutexCorrupted)?; let template_transactions = tx_list_.to_vec(); @@ -433,7 +429,7 @@ impl JobDeclarator { } }; if sender.send(sv2_frame.into()).await.is_err() { - error!("Job declarator failed to send message"); + error!("Job declarator failed to send message AAAAA"); ProxyState::update_jd_state(JdState::Down); break; }; @@ -447,7 +443,7 @@ impl JobDeclarator { } } }); - TaskManager::add_allocate_tokens(task_manager, main_task.into()) + TaskManager::add_main_task(task_manager, main_task.into()) .await .map_err(|_| Error::JobDeclaratorTaskManagerFailed) } @@ -477,14 +473,15 @@ impl JobDeclarator { && s.last_set_new_prev_hash != Some(set_new_prev_hash.clone()) //it means that a new prev_hash is arrived while the previous hasn't exited the loop yet { - s.set_new_prev_hash_counter -= 1; + s.set_new_prev_hash_counter = s.set_new_prev_hash_counter.saturating_sub(1); Some(None) } else { s.future_jobs .remove(&id) .map(|(job, merkle_path, template, pool_outs)| { - s.future_jobs = HashMap::with_hasher(BuildNoHashHasher::default()); - s.set_new_prev_hash_counter -= 1; + s.future_jobs.clear(); + s.set_new_prev_hash_counter = + s.set_new_prev_hash_counter.saturating_sub(1); Some((job, s.up.clone(), merkle_path, template, pool_outs)) }) } @@ -555,7 +552,7 @@ impl JobDeclarator { .expect("Infallible operation"); if sender.send(frame.into()).await.is_err() { - error!("Job declarator failed to send message"); + error!("Job declarator failed to send message BBBB"); ProxyState::update_jd_state(JdState::Down); } } diff --git a/src/jd_client/mining_downstream/mod.rs b/src/jd_client/mining_downstream/mod.rs index 4b27852f..3e4c4f07 100644 --- a/src/jd_client/mining_downstream/mod.rs +++ b/src/jd_client/mining_downstream/mod.rs @@ -5,7 +5,10 @@ use crate::{ }; use tokio::time::{timeout, Duration}; -use super::{job_declarator::JobDeclarator, mining_upstream::Upstream as UpstreamMiningNode}; +use super::{ + job_declarator::JobDeclarator, + mining_upstream::{Upstream as UpstreamMiningNode, UpstreamChannelFactory}, +}; use crate::jd_client::error::Error as JdClientError; use roles_logic_sv2::{ channel_logic::channel_factory::{OnNewShare, PoolChannelFactory, Share}, @@ -23,11 +26,11 @@ use tokio::{ sync::mpsc::{Receiver as TReceiver, Sender as TSender}, task, }; -use tracing::{debug, error, warn}; +use tracing::{debug, error, info, warn}; use codec_sv2::{StandardEitherFrame, StandardSv2Frame}; -use bitcoin::{consensus::Decodable, TxOut}; +use bitcoin::{consensus::Decodable, hex::DisplayHex, TxOut}; pub type Message = MiningDeviceMessages<'static>; pub type StdFrame = StandardSv2Frame; @@ -60,20 +63,130 @@ pub enum DownstreamMiningNodeStatus { } impl DownstreamMiningNodeStatus { - fn set_channel(&mut self, channel: PoolChannelFactory) -> bool { + fn set_channel(&mut self, channel: UpstreamChannelFactory) -> bool { + let UpstreamChannelFactory { + factory, + channel_id, + target, + extranonce_prefix, + extranonce_size, + } = channel; + let incoming_channel_ids = factory.get_extended_channels_ids(); match self { DownstreamMiningNodeStatus::Paired(up) => { - let self_ = Self::ChannelOpened((channel, up.clone())); + debug!( + "JD downstream installing channel factory from Paired state: incoming channel ids {:?}", + incoming_channel_ids + ); + let self_ = Self::ChannelOpened((factory, up.clone())); + let _ = std::mem::replace(self, self_); + true + } + DownstreamMiningNodeStatus::ChannelOpened((old, _up)) => { + let existing_channel_ids = old.get_extended_channels_ids(); + info!( + "JD downstream restoring channel on existing factory: existing channel ids {:?}, incoming channel ids {:?}", + existing_channel_ids, + incoming_channel_ids, + ); + + if old.get_extranonce_len() != factory.get_extranonce_len() { + error!( + "JD downstream restored channel extranonce layout changed: existing len {} incoming len {}", + old.get_extranonce_len(), + factory.get_extranonce_len(), + ); + return false; + } + + for existing_channel_id in existing_channel_ids { + old.close_channel(existing_channel_id); + } + + let extranonce: Extranonce = match extranonce_prefix.try_into() { + Ok(extranonce) => extranonce, + Err(_) => { + error!( + "JD downstream failed to restore channel {}: invalid extranonce prefix", + channel_id + ); + return false; + } + }; + if old + .replicate_upstream_extended_channel_only_jd( + target.clone(), + extranonce, + channel_id, + extranonce_size, + ) + .is_none() + { + error!( + "JD downstream failed to restore channel {} on existing factory", + channel_id + ); + return false; + } + + let mut upstream_target = target.clone().into(); + old.set_target(&mut upstream_target); + old.update_target_for_channel(channel_id, target.into()); + debug!( + "JD downstream restored channel on existing factory: channel_id={} channel ids {:?}", + channel_id, + old.get_extended_channels_ids(), + ); + true + } + DownstreamMiningNodeStatus::SoloMinerPaired() => { + debug!( + "JD downstream installing solo channel factory from SoloMinerPaired state: incoming channel ids {:?}", + incoming_channel_ids + ); + let self_ = Self::SoloMinerChannelOpend(factory); let _ = std::mem::replace(self, self_); true } - DownstreamMiningNodeStatus::ChannelOpened(_) => false, + DownstreamMiningNodeStatus::SoloMinerChannelOpend(old) => { + warn!( + "JD downstream refused solo channel factory replacement: existing channel ids {:?}, incoming channel ids {:?}", + old.get_extended_channels_ids(), + incoming_channel_ids + ); + false + } + } + } + + fn set_solo_channel(&mut self, channel: PoolChannelFactory) -> bool { + let incoming_channel_ids = channel.get_extended_channels_ids(); + match self { DownstreamMiningNodeStatus::SoloMinerPaired() => { + debug!( + "JD downstream installing solo channel factory from SoloMinerPaired state: incoming channel ids {:?}", + incoming_channel_ids + ); let self_ = Self::SoloMinerChannelOpend(channel); let _ = std::mem::replace(self, self_); true } - DownstreamMiningNodeStatus::SoloMinerChannelOpend(_) => false, + DownstreamMiningNodeStatus::SoloMinerChannelOpend(old) => { + warn!( + "JD downstream refused solo channel factory replacement: existing channel ids {:?}, incoming channel ids {:?}", + old.get_extended_channels_ids(), + incoming_channel_ids + ); + false + } + DownstreamMiningNodeStatus::Paired(_) + | DownstreamMiningNodeStatus::ChannelOpened(_) => { + warn!( + "JD downstream refused solo channel factory install in pooled mode: incoming channel ids {:?}", + incoming_channel_ids + ); + false + } } } @@ -205,14 +318,31 @@ impl DownstreamMiningNode { return; } }; - if let Ok(factory) = UpstreamMiningNode::take_channel_factory(upstream).await { - if self_mutex - .safe_lock(|s| { - s.status.set_channel(factory); - }) - .is_err() - { - error!("Jd can not get channel factory"); + loop { + let factory = + match UpstreamMiningNode::take_channel_factory(upstream.clone()).await { + Ok(factory) => factory, + Err(e) => { + error!("Jd can not get channel factory: {e}"); + ProxyState::update_downstream_state( + DownstreamType::JdClientMiningDownstream, + ); + return; + } + }; + + match self_mutex.safe_lock(|s| s.status.set_channel(factory)) { + Ok(true) => debug!("Jd mining downstream channel factory refreshed"), + Ok(false) => { + warn!("Jd mining downstream channel factory refresh was ignored") + } + Err(_) => { + error!("Jd can not set channel factory"); + ProxyState::update_downstream_state( + DownstreamType::JdClientMiningDownstream, + ); + return; + } } } } @@ -323,7 +453,46 @@ impl DownstreamMiningNode { Ok(SendTo::Respond(message)) => Self::send(&self_mutex, message).await?, Ok(SendTo::None(None)) => (), Ok(m) => unreachable!("Unexpected message type: {:?}", m), - Err(Error::ShareDoNotMatchAnyJob) => warn!("Error: ShareDoNotMatchAnyJob"), + Err(Error::ShareDoNotMatchAnyJob) => { + let last_template_id = self_mutex.safe_lock(|s| s.last_template_id).ok(); + match incoming.as_ref() { + Some(Mining::SubmitSharesExtended(share)) => { + let factory_extranonce_prefix = self_mutex + .safe_lock(|s| { + s.status + .get_channel() + .ok() + .and_then(|channel| { + channel.get_extranonce_prefix(share.channel_id) + }) + }) + .ok() + .flatten() + .map(|prefix| prefix.as_hex().to_string()) + .unwrap_or_else(|| "".to_string()); + warn!( + "JD downstream local share validation failed: ShareDoNotMatchAnyJob channel_id={} job_id={} sequence_number={} nonce={} ntime={} share_extranonce={} factory_extranonce_prefix={} last_template_id={:?}", + share.channel_id, + share.job_id, + share.sequence_number, + share.nonce, + share.ntime, + share.extranonce.to_vec().as_hex(), + factory_extranonce_prefix, + last_template_id, + ) + } + Some(message) => warn!( + "JD downstream local share validation failed: ShareDoNotMatchAnyJob incoming={:?} last_template_id={:?}", + message, + last_template_id, + ), + None => warn!( + "JD downstream local share validation failed: ShareDoNotMatchAnyJob incoming=None last_template_id={:?}", + last_template_id, + ), + } + } Err(e) => return Err(JdClientError::RolesSv2Logic(e)), } Ok(()) @@ -536,7 +705,7 @@ impl error!("Signature + extranonce lens exceed 32 bytes"); })?; - self.status.set_channel(channel_factory); + self.status.set_solo_channel(channel_factory); let request_id = m.request_id; let hash_rate = m.nominal_hash_rate; @@ -584,12 +753,32 @@ impl &mut self, m: SubmitSharesExtended, ) -> Result, Error> { - match self + debug!( + "JD downstream validating SubmitSharesExtended with local PoolChannelFactory: channel_id={} job_id={} sequence_number={} nonce={} ntime={} share_extranonce={} last_template_id={}", + m.channel_id, + m.job_id, + m.sequence_number, + m.nonce, + m.ntime, + m.extranonce.to_vec().as_hex(), + self.last_template_id, + ); + let channel = self .status .get_channel() - .map_err(|_| Error::NotFoundChannelId)? - .on_submit_shares_extended(m.clone())? - { + .map_err(|_| Error::NotFoundChannelId)?; + let factory_extranonce_prefix = channel + .get_extranonce_prefix(m.channel_id) + .map(|prefix| prefix.as_hex().to_string()) + .unwrap_or_else(|| "".to_string()); + debug!( + "JD downstream validating SubmitSharesExtended against local PoolChannelFactory channel ids {:?}; factory_extranonce_prefix={}", + channel.get_extended_channels_ids(), + factory_extranonce_prefix, + ); + let data = channel.get_additional_coinbase_script_data(m.channel_id, m.job_id); + debug!("get_additional_coinbase_script_data result: {:?}", data); + match channel.on_submit_shares_extended(m.clone())? { OnNewShare::SendErrorDownstream(s) => { error!("Share do not meet downstream target"); Ok(SendTo::Respond(Mining::SubmitSharesError(s))) @@ -680,3 +869,72 @@ impl } } impl IsMiningDownstream for DownstreamMiningNode {} + +#[cfg(test)] +mod tests { + use super::*; + use roles_logic_sv2::{ + channel_logic::channel_factory::ExtendedChannelKind, + job_creator::JobsCreators, + mining_sv2::{ExtendedExtranonce, Extranonce, Target}, + }; + + fn upstream_channel_factory(channel_id: u32) -> UpstreamChannelFactory { + let prefix_len = 16; + let extranonce_size = 16; + let ids = Arc::new(Mutex::new(roles_logic_sv2::utils::GroupId::new())); + let extranonces = ExtendedExtranonce::new( + 0..prefix_len, + prefix_len..prefix_len, + prefix_len..prefix_len + extranonce_size as usize, + ); + let creator = JobsCreators::new((prefix_len + extranonce_size as usize) as u8); + let target = Target::new(u128::MAX, u128::MAX); + let channel_kind = ExtendedChannelKind::ProxyJd { + upstream_target: target, + }; + let mut channel_factory = + PoolChannelFactory::new(ids, extranonces, creator, 1.0, channel_kind, vec![], vec![]) + .unwrap(); + let target: binary_sv2::U256<'static> = vec![0xff; 32].try_into().unwrap(); + let extranonce_prefix = vec![channel_id as u8; prefix_len]; + let extranonce: Extranonce = extranonce_prefix.clone().try_into().unwrap(); + channel_factory + .replicate_upstream_extended_channel_only_jd( + target.clone(), + extranonce, + channel_id, + extranonce_size, + ) + .unwrap(); + UpstreamChannelFactory { + factory: channel_factory, + channel_id, + target, + extranonce_prefix, + extranonce_size, + } + } + + #[tokio::test] + async fn set_channel_replaces_existing_pooled_factory() { + let (sender, _receiver) = tokio::sync::mpsc::channel(1); + let upstream = UpstreamMiningNode::new(crate::MIN_EXTRANONCE_SIZE, sender) + .await + .unwrap(); + let mut status = DownstreamMiningNodeStatus::Paired(upstream.clone()); + + assert!(status.set_channel(upstream_channel_factory(4))); + { + let channel = status.get_channel().unwrap(); + assert_eq!(channel.get_extended_channels_ids(), vec![4]); + } + + assert!(status.set_channel(upstream_channel_factory(1))); + { + let channel = status.get_channel().unwrap(); + assert_eq!(channel.get_extended_channels_ids(), vec![1]); + } + assert!(Arc::ptr_eq(&status.get_upstream().unwrap(), &upstream)); + } +} diff --git a/src/jd_client/mining_upstream/mod.rs b/src/jd_client/mining_upstream/mod.rs index e0f36fc2..49260db0 100644 --- a/src/jd_client/mining_upstream/mod.rs +++ b/src/jd_client/mining_upstream/mod.rs @@ -1,3 +1,4 @@ pub mod upstream; pub use upstream::Upstream; +pub use upstream::UpstreamChannelFactory; mod task_manager; diff --git a/src/jd_client/mining_upstream/upstream.rs b/src/jd_client/mining_upstream/upstream.rs index bd000f9d..dda73cf4 100644 --- a/src/jd_client/mining_upstream/upstream.rs +++ b/src/jd_client/mining_upstream/upstream.rs @@ -27,6 +27,15 @@ use std::collections::VecDeque; use super::task_manager::TaskManager; +#[derive(Debug)] +pub struct UpstreamChannelFactory { + pub factory: PoolChannelFactory, + pub channel_id: u32, + pub target: U256<'static>, + pub extranonce_prefix: Vec, + pub extranonce_size: u16, +} + #[derive(Debug)] struct CircularBuffer { buffer: VecDeque<(u64, u32)>, @@ -106,7 +115,7 @@ pub struct Upstream { /// Sends messages to the SV2 Upstream role pub sender: TSender>, pub downstream: Option>>, - channel_factory: Option, + channel_factory: Option, template_to_job_id: TemplateToJobId, req_ids: Id, } @@ -276,7 +285,7 @@ impl Upstream { pub async fn take_channel_factory( self_: Arc>, - ) -> Result { + ) -> Result { while self_ .safe_lock(|s| s.channel_factory.is_none()) .map_err(|_| Error::JdClientUpstreamMutexCorrupted)? @@ -414,22 +423,30 @@ impl ParseUpstreamMiningMessages"); - self.channel_id = Some(m.channel_id); + self.channel_id = Some(channel_id); channel_factory .replicate_upstream_extended_channel_only_jd( - m.target.into_static(), + target.clone(), extranonce, - m.channel_id, - m.extranonce_size, + channel_id, + extranonce_size, ) .expect("Impossible to open downstream channel"); - self.channel_factory = Some(channel_factory); + self.channel_factory = Some(UpstreamChannelFactory { + factory: channel_factory, + channel_id, + target, + extranonce_prefix, + extranonce_size, + }); let downstream = match self.downstream.as_ref() { Some(downstream) => downstream.clone(), @@ -574,6 +591,10 @@ impl ParseUpstreamMiningMessages Result, RolesLogicError> { if let Some(factory) = self.channel_factory.as_mut() { - factory.update_target_for_channel(m.channel_id, m.maximum_target.clone().into()); - factory.set_target(&mut m.maximum_target.clone().into()); + factory + .factory + .update_target_for_channel(m.channel_id, m.maximum_target.clone().into()); + factory + .factory + .set_target(&mut m.maximum_target.clone().into()); + if m.channel_id == factory.channel_id { + factory.target = m.maximum_target.clone().into_static(); + } } if let Some(downstream) = &self.downstream { downstream @@ -632,6 +660,16 @@ impl ParseUpstreamMiningMessages Result, RolesLogicError> { + if let Some(channel_id) = self.channel_id.take() { + info!( + "JD mining upstream pool reconnected; marking channel_id={} not ready until channel reopen succeeds", + channel_id + ); + } else { + info!("JD mining upstream pool reconnected before an upstream channel was ready"); + } + self.channel_factory = None; + if let Some(downstream) = &self.downstream { Ok(SendTo::RelaySameMessageToRemote(downstream.clone())) } else { @@ -639,3 +677,150 @@ impl ParseUpstreamMiningMessages U256<'static> { + vec![255_u8; 32] + .try_into() + .expect("test u256 must contain 32 bytes") + } + + fn reconnect_message() -> roles_logic_sv2::mining_sv2::Reconnect<'static> { + roles_logic_sv2::mining_sv2::Reconnect { + new_host: String::new() + .try_into() + .expect("empty reconnect host must fit in Str0255"), + new_port: 0, + } + } + + fn open_success(channel_id: u32) -> OpenExtendedMiningChannelSuccess<'static> { + OpenExtendedMiningChannelSuccess { + request_id: 0, + channel_id, + target: u256_max(), + extranonce_size: 16, + extranonce_prefix: vec![channel_id as u8; 16] + .try_into() + .expect("test extranonce prefix must fit in B032"), + } + } + + fn declare_mining_job() -> DeclareMiningJob<'static> { + DeclareMiningJob { + request_id: 0, + mining_job_token: Vec::new().try_into().unwrap(), + version: 0, + coinbase_prefix: Vec::new().try_into().unwrap(), + coinbase_suffix: Vec::new().try_into().unwrap(), + tx_list: Vec::>::new().into(), + excess_data: Vec::new().try_into().unwrap(), + } + } + + fn set_new_prev_hash() -> roles_logic_sv2::template_distribution_sv2::SetNewPrevHash<'static> { + roles_logic_sv2::template_distribution_sv2::SetNewPrevHash { + template_id: 1, + prev_hash: u256_max(), + header_timestamp: 0, + n_bits: 0, + target: u256_max(), + } + } + + #[tokio::test] + async fn reconnect_marks_channel_not_ready_until_open_success() { + let (sender, _receiver) = mpsc::channel(1); + let upstream = Upstream::new(16, sender).await.unwrap(); + + upstream + .safe_lock(|u| u.handle_open_extended_mining_channel_success(open_success(7))) + .unwrap() + .unwrap_err(); + assert_eq!(upstream.safe_lock(|u| u.channel_id).unwrap(), Some(7)); + + upstream + .safe_lock(|u| u.handle_reconnect(reconnect_message())) + .unwrap() + .unwrap_err(); + assert_eq!(upstream.safe_lock(|u| u.channel_id).unwrap(), None); + assert!(upstream.safe_lock(|u| u.channel_factory.is_none()).unwrap()); + + upstream + .safe_lock(|u| u.handle_open_extended_mining_channel_success(open_success(42))) + .unwrap() + .unwrap_err(); + assert_eq!(upstream.safe_lock(|u| u.channel_id).unwrap(), Some(42)); + assert_eq!( + upstream + .safe_lock(|u| u.channel_factory.as_ref().map(|factory| factory.channel_id)) + .unwrap(), + Some(42) + ); + } + + #[tokio::test] + async fn set_custom_jobs_waits_for_reopened_channel_after_reconnect() { + let (sender, mut receiver) = mpsc::channel(1); + let upstream = Upstream::new(16, sender).await.unwrap(); + + upstream + .safe_lock(|u| u.handle_open_extended_mining_channel_success(open_success(7))) + .unwrap() + .unwrap_err(); + upstream + .safe_lock(|u| u.handle_reconnect(reconnect_message())) + .unwrap() + .unwrap_err(); + + let upstream_for_job = upstream.clone(); + let set_custom_task = tokio::spawn(async move { + Upstream::set_custom_jobs( + &upstream_for_job, + declare_mining_job(), + set_new_prev_hash(), + Vec::>::new().into(), + Vec::new().try_into().unwrap(), + 0, + Vec::new().try_into().unwrap(), + 0, + 0, + Vec::new(), + 0, + 1, + ) + .await + }); + + assert!( + timeout(Duration::from_millis(50), receiver.recv()) + .await + .is_err(), + "set_custom_jobs must not send while the upstream channel is not ready" + ); + + upstream + .safe_lock(|u| u.handle_open_extended_mining_channel_success(open_success(42))) + .unwrap() + .unwrap_err(); + + let message = timeout(Duration::from_millis(250), receiver.recv()) + .await + .unwrap() + .unwrap(); + match message { + Mining::SetCustomMiningJob(job) => { + assert_eq!(job.channel_id, 42); + } + other => panic!("expected SetCustomMiningJob, got {other:?}"), + } + + set_custom_task.await.unwrap().unwrap(); + } +} diff --git a/src/lib.rs b/src/lib.rs index 41a7ef42..28b3bcb0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17,6 +17,7 @@ use crate::shared::utils::AbortOnDrop; use key_utils::Secp256k1PublicKey; use lazy_static::lazy_static; use proxy_state::{PoolState, ProxyState, TpState, TranslatorState}; +use roles_logic_sv2::utils::Mutex; use std::sync::{ atomic::{AtomicBool, Ordering}, OnceLock, @@ -69,10 +70,10 @@ pub(crate) struct DownstreamConnection { pub(crate) type DownstreamHandoffSender = tokio::sync::mpsc::Sender; lazy_static! { - static ref TP_ADDRESS: roles_logic_sv2::utils::Mutex> = - roles_logic_sv2::utils::Mutex::new(Configuration::tp_address()); - static ref ACTIVE_POOL_ADDRESS: roles_logic_sv2::utils::Mutex> = - roles_logic_sv2::utils::Mutex::new(None); // Connected pool address + static ref TP_ADDRESS: Mutex> = + Mutex::new(Configuration::tp_address()); + static ref ACTIVE_POOL_ADDRESS: Mutex> = + Mutex::new(None); // Connected pool address } lazy_static! { diff --git a/src/router/mod.rs b/src/router/mod.rs index 032c5a72..266f2b91 100644 --- a/src/router/mod.rs +++ b/src/router/mod.rs @@ -384,7 +384,7 @@ impl PoolLatency { let initiator = Initiator::from_raw_k(authority_public_key.into_bytes()) // Safe expect Key is a constant and must be right .expect("Unable to create initialtor"); - let (mut receiver, mut sender, _, _) = + let (mut jds_receiver, mut jds_sender, _, _) = match Connection::new(stream, HandshakeRole::Initiator(initiator)).await { Ok(connection) => connection, Err(e) => { @@ -393,7 +393,8 @@ impl PoolLatency { } }; if let Err(e) = - SetupConnectionHandler::setup(&mut receiver, &mut sender, address).await + SetupConnectionHandler::setup(&mut jds_receiver, &mut jds_sender, address) + .await { error!("Failed to setup connection: {:?}", e); return Err(()); @@ -401,9 +402,11 @@ impl PoolLatency { self.open_sv2_jd_connection = Some(open_sv2_jd_connection_timer.elapsed()); - let (sender, mut _receiver) = tokio::sync::mpsc::channel(10); + let (mining_sender, mut _mining_receiver) = tokio::sync::mpsc::channel(10); let upstream = - match crate::jd_client::mining_upstream::Upstream::new(0, sender).await { + match crate::jd_client::mining_upstream::Upstream::new(0, mining_sender) + .await + { Ok(upstream) => upstream, Err(e) => { error!("Failed to create upstream: {:?}", e); diff --git a/src/translator/mod.rs b/src/translator/mod.rs index d70c0088..34883b16 100644 --- a/src/translator/mod.rs +++ b/src/translator/mod.rs @@ -9,7 +9,7 @@ use bitcoin::Address; use error::Error; use roles_logic_sv2::{parsers::Mining, utils::Mutex}; -use tracing::error; +use tracing::{error, info}; use std::sync::Arc; use tokio::sync::mpsc::channel; @@ -25,6 +25,7 @@ use tokio::sync::mpsc::{Receiver as TReceiver, Sender as TSender}; use self::upstream::diff_management::UpstreamDifficultyConfig; mod task_manager; +pub(crate) use downstream::diff_management::NON_LOCAL_DOWNSTREAM_MIN_DIFFICULTY; use task_manager::TaskManager; pub async fn start( @@ -35,7 +36,7 @@ pub async fn start( Option
, )>, stats_sender: crate::api::stats::StatsSender, - signature: String, + _signature: String, ) -> Result> { let task_manager = TaskManager::initialize(pool_connection.clone()); let abortable = task_manager @@ -104,7 +105,6 @@ pub async fn start( target.clone(), diff_config.clone(), send_to_up, - signature, ) .await?; @@ -124,7 +124,8 @@ pub async fn start( let target = target.clone(); let task_manager = task_manager.clone(); tokio::task::spawn(async move { - let (extended_extranonce, up_id) = match rx_sv2_extranonce.recv().await { + let result = rx_sv2_extranonce.recv().await; + let (extended_extranonce, up_id) = match result { Some((extended_extranonce, up_id)) => (extended_extranonce, up_id), None => { error!("Failed to receive from rx_sv2_extranonce"); @@ -134,15 +135,17 @@ pub async fn start( }; loop { - let target: [u8; 32] = match target.safe_lock(|t| t.clone()) { - Ok(target) => target.try_into().expect("Internal error: this operation cannot fail because Vec can always be converted into [u8; 32]"), + let target_bytes: [u8; 32] = match target.safe_lock(|t| t.clone()) { + Ok(target) => target.try_into().expect( + "Internal error: this operation cannot fail because Vec can always be converted into [u8; 32]", + ), Err(e) => { error!("{}", Error::TargetError(roles_logic_sv2::Error::PoisonLock(e.to_string()))); - break + return; } }; - if target != [0; 32] { + if target_bytes != [0; 32] { break; }; tokio::task::yield_now().await; @@ -153,7 +156,7 @@ pub async fn start( tx_sv2_submit_shares_ext, tx_sv1_notify.clone(), extended_extranonce, - target, + target.clone(), up_id, ) { Ok(b) => b, @@ -181,7 +184,7 @@ pub async fn start( let downstream_aborter = match downstream::Downstream::accept_connections( tx_sv1_bridge, tx_sv1_notify, - b, + b.clone(), diff_config, downstreams, stats_sender, @@ -210,6 +213,29 @@ pub async fn start( { error!("{}", Error::TranslatorTaskManagerFailed); } + while let Some((extended_extranonce, up_id)) = rx_sv2_extranonce.recv().await { + let restore_result = b.safe_lock(|bridge| { + bridge.restore_upstream_channel(extended_extranonce, up_id) + }); + match restore_result { + Ok(Ok(())) => { + info!("Translator bridge restored upstream channel id {up_id}"); + } + Ok(Err(e)) => { + error!("Failed to restore translator bridge upstream channel: {e}"); + ProxyState::update_translator_state(TranslatorState::Down); + return; + } + Err(e) => { + error!("Translator bridge mutex poisoned while restoring upstream channel: {e}"); + ProxyState::update_translator_state(TranslatorState::Down); + return; + } + } + } + + error!("Failed to receive from rx_sv2_extranonce"); + ProxyState::update_translator_state(TranslatorState::Down); }) }; TaskManager::add_startup_task(task_manager.clone(), startup_task.into()) diff --git a/src/translator/proxy/bridge.rs b/src/translator/proxy/bridge.rs index f7cfef78..f1ccefa1 100644 --- a/src/translator/proxy/bridge.rs +++ b/src/translator/proxy/bridge.rs @@ -40,6 +40,23 @@ lazy_static! { static ref SUBMIT_FAIL_COUNTER: AtomicU32 = AtomicU32::new(0); } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct UpstreamExtranonceLayout { + len: usize, + range0_len: usize, + range2_len: usize, +} + +impl UpstreamExtranonceLayout { + fn from_extranonce(extranonce: &ExtendedExtranonce) -> Self { + Self { + len: extranonce.get_len(), + range0_len: extranonce.get_range0_len(), + range2_len: extranonce.get_range2_len(), + } + } +} + /// Bridge between the SV2 `Upstream` and SV1 `Downstream` responsible for the following messaging /// translation: /// 1. SV1 `mining.submit` -> SV2 `SubmitSharesExtended` @@ -67,6 +84,8 @@ pub struct Bridge { future_jobs: Vec>, last_p_hash: Option>, target: Arc>>, + upstream_channel_id: u32, + upstream_extranonce_layout: UpstreamExtranonceLayout, } impl Bridge { @@ -103,10 +122,15 @@ impl Bridge { channel_id: u32, ) -> Result>, Error<'static>> { info!("Creating new bridge for channel_id {}:", channel_id); + let upstream_extranonce_layout = UpstreamExtranonceLayout::from_extranonce(&extranonces); let ids = Arc::new(Mutex::new(GroupId::new())); - let upstream_target: [u8; 32] = target.safe_lock(|t| { - t.clone().try_into().expect("Internal error: this operation can not fail because Vec can always be converted into [u8; 32]") -}).map_err(|e| Error::TargetError(RolesLogicError::PoisonLock(e.to_string())))?; + let upstream_target: [u8; 32] = target + .safe_lock(|t| { + t.clone().try_into().expect( + "Internal error: this operation can not fail because Vec can always be converted into [u8; 32]", + ) + }) + .map_err(|e| Error::TargetError(RolesLogicError::PoisonLock(e.to_string())))?; let upstream_target: Target = upstream_target.into(); Ok(Arc::new(Mutex::new(Self { tx_sv2_submit_shares_ext, @@ -124,9 +148,37 @@ impl Bridge { future_jobs: vec![], last_p_hash: None, target, + upstream_channel_id: channel_id, + upstream_extranonce_layout, }))) } + pub fn restore_upstream_channel( + &mut self, + extranonces: ExtendedExtranonce, + channel_id: u32, + ) -> ProxyResult<'static, ()> { + let restored_layout = UpstreamExtranonceLayout::from_extranonce(&extranonces); + if self.upstream_extranonce_layout != restored_layout { + return Err(Error::InvalidExtranonce(format!( + "restored upstream extranonce layout changed for channel {channel_id}: expected len/range0/range2 {}/{}/{} got {}/{}/{}", + self.upstream_extranonce_layout.len, + self.upstream_extranonce_layout.range0_len, + self.upstream_extranonce_layout.range2_len, + restored_layout.len, + restored_layout.range0_len, + restored_layout.range2_len, + ))); + } + + info!( + "Restored bridge upstream channel id from {} to {}", + self.upstream_channel_id, channel_id + ); + self.upstream_channel_id = channel_id; + Ok(()) + } + #[allow(clippy::result_large_err)] pub fn on_new_sv1_connection( &mut self, @@ -296,8 +348,20 @@ impl Bridge { &share_id, &channel_id, &job_id ); } - let (tx_sv2_submit_shares_ext, target_mutex) = self_ - .safe_lock(|s| (s.tx_sv2_submit_shares_ext.clone(), s.target.clone())) + let ( + tx_sv2_submit_shares_ext, + target_mutex, + upstream_channel_id, + upstream_extranonce_prefix_len, + ) = self_ + .safe_lock(|s| { + ( + s.tx_sv2_submit_shares_ext.clone(), + s.target.clone(), + s.upstream_channel_id, + s.upstream_extranonce_layout.range0_len, + ) + }) .map_err(|_| Error::BridgeMutexPoisoned)?; let upstream_target: [u8; 32] = target_mutex .safe_lock(|t| t.clone()) @@ -309,6 +373,13 @@ impl Bridge { dbg_target.reverse(); debug!("Pool target: {:?}", dbg_target.as_hex()); let mut upstream_target: Target = upstream_target.into(); + debug!( + "Bridge validating SV1 submit before translation: downstream_channel_id={} sv1_job_id={} share_id={} current_upstream_channel_id={}", + channel_id, + job_id, + share_id, + upstream_channel_id, + ); let translated_share = match self_ .safe_lock(|s| { let job_id = share.job_id.parse::().expect("Invalid job_id"); @@ -366,15 +437,50 @@ impl Bridge { } }; - let res = self_ + debug!( + "Bridge validating translated SubmitSharesExtended locally: channel_id={} job_id={} sequence_number={} nonce={} ntime={} upstream_channel_id={}", + translated_share.channel_id, + translated_share.job_id, + translated_share.sequence_number, + translated_share.nonce, + translated_share.ntime, + upstream_channel_id, + ); + let share_extranonce_hex = translated_share.extranonce.to_vec().as_hex().to_string(); + let (res, factory_extranonce_prefix) = self_ .safe_lock(|s| { // Ordering::Relaxed is safe here because we only need simple counter updates. // No need for strict ordering since it just tracks failures. SUBMIT_FAIL_COUNTER.store(0, Ordering::Relaxed); - s.channel_factory - .on_submit_shares_extended(translated_share.clone()) + let factory_extranonce_prefix = s + .channel_factory + .extranonce_from_downstream_extranonce( + translated_share.extranonce.to_vec().try_into().expect( + "Internal error: share extranonce bytes must fit in Extranonce", + ), + ) + .and_then(|full_extranonce| { + full_extranonce + .to_vec() + .get(..upstream_extranonce_prefix_len) + .map(|prefix| prefix.to_vec()) + }); + let result = s + .channel_factory + .on_submit_shares_extended(translated_share.clone()); + (result, factory_extranonce_prefix) }) .map_err(|_| Error::BridgeMutexPoisoned)?; + let factory_extranonce_prefix_hex = factory_extranonce_prefix + .map(|prefix| prefix.as_hex().to_string()) + .unwrap_or_else(|| "".to_string()); + debug!( + "Bridge checked SubmitSharesExtended extranonce locally: channel_id={} job_id={} share_extranonce={} factory_extranonce_prefix={}", + translated_share.channel_id, + translated_share.job_id, + share_extranonce_hex, + factory_extranonce_prefix_hex, + ); match res { Ok(OnNewShare::SendErrorDownstream(e)) => { @@ -398,11 +504,21 @@ impl Bridge { &share_id, &channel_id, &job_id ); } - let upstream_share = match s { + let mut upstream_share = match s { Share::Extended(share) => share, // We are in an extended channel shares are extended Share::Standard(_) => unreachable!(), }; + debug!( + "Bridge forwarding accepted SubmitSharesExtended upstream: local_channel_id={} restored_upstream_channel_id={} job_id={} sequence_number={} share_extranonce={} factory_extranonce_prefix={}", + upstream_share.channel_id, + upstream_channel_id, + upstream_share.job_id, + upstream_share.sequence_number, + upstream_share.extranonce.to_vec().as_hex(), + factory_extranonce_prefix_hex, + ); + upstream_share.channel_id = upstream_channel_id; if let Ok(allowed_by_rate_limit) = allow_submit_share() { if !allowed_by_rate_limit { @@ -646,6 +762,12 @@ impl Bridge { // convert to non segwit jobs so we dont have to depend if miner's support segwit or not self_ .safe_lock(|s| { + debug!( + "Bridge registering NewExtendedMiningJob in local factory: channel_id={} job_id={} is_future={}", + sv2_new_extended_mining_job.channel_id, + sv2_new_extended_mining_job.job_id, + sv2_new_extended_mining_job.is_future(), + ); s.channel_factory .on_new_extended_mining_job(sv2_new_extended_mining_job.as_static().clone()) }) @@ -897,6 +1019,48 @@ mod test { .unwrap() } + #[test] + fn restoring_upstream_channel_updates_bridge_without_reopening_downstreams() { + let extranonces = ExtendedExtranonce::new(0..16, 16..18, 18..32); + let bridge = test_utils::create_bridge(extranonces).unwrap(); + + bridge + .safe_lock(|bridge| { + let first_downstream = bridge.on_new_sv1_connection(1_000_000_000_000.0).unwrap(); + assert_eq!(bridge.upstream_channel_id, 1); + + bridge + .restore_upstream_channel(ExtendedExtranonce::new(0..16, 16..18, 18..32), 42) + .unwrap(); + + assert_eq!(bridge.upstream_channel_id, 42); + assert!(bridge + .channel_factory + .update_target_for_channel(first_downstream.channel_id, [255; 32].into()) + .is_some()); + + let second_downstream = bridge.on_new_sv1_connection(2_000_000_000_000.0).unwrap(); + assert_ne!(first_downstream.channel_id, second_downstream.channel_id); + }) + .unwrap(); + } + + #[test] + fn restoring_upstream_channel_rejects_changed_extranonce_layout() { + let extranonces = ExtendedExtranonce::new(0..16, 16..18, 18..32); + let bridge = test_utils::create_bridge(extranonces).unwrap(); + + bridge + .safe_lock(|bridge| { + let err = bridge + .restore_upstream_channel(ExtendedExtranonce::new(0..16, 16..19, 19..32), 42) + .unwrap_err(); + assert!(matches!(err, Error::InvalidExtranonce(_))); + assert_eq!(bridge.upstream_channel_id, 1); + }) + .unwrap(); + } + #[test] fn test_version_bits_insert() { let extranonces = ExtendedExtranonce::new(0..6, 6..8, 8..16); diff --git a/src/translator/upstream/upstream.rs b/src/translator/upstream/upstream.rs index 0d002621..10716f6c 100644 --- a/src/translator/upstream/upstream.rs +++ b/src/translator/upstream/upstream.rs @@ -32,7 +32,7 @@ use tokio::{ }, task, }; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; use super::task_manager::TaskManager; use crate::{ @@ -43,6 +43,8 @@ use crate::{ use bitcoin::BlockHash; pub static IS_NEW_JOB_HANDLED: AtomicBool = AtomicBool::new(true); +pub(super) const RESTORE_CHANNEL_USER_ID_PREFIX: &[u8] = b"RESTORE-CHANNEL"; + /// Represents the currently active `prevhash` of the mining job being worked on OR being submitted /// from the Downstream role. #[derive(Debug, Clone)] @@ -65,6 +67,7 @@ pub struct Upstream { last_job_id: Option, /// Bytes used as implicit first part of `extranonce`. extranonce_prefix: Option>, + restore_channel: Option>, /// Sends SV2 `SetNewPrevHash` messages to be translated (along with SV2 `NewExtendedMiningJob` /// messages) into SV1 `mining.notify` messages. Received and translated by the `Bridge`. tx_sv2_set_new_prev_hash: tokio::sync::mpsc::Sender>, @@ -90,7 +93,6 @@ pub struct Upstream { // than the configured percentage pub(super) difficulty_config: Arc>, pub sender: TSender>, - signature: String, sent_up: u32, rejected: u32, toa: Vec, @@ -119,22 +121,22 @@ impl Upstream { target: Arc>>, difficulty_config: Arc>, sender: TSender>, - signature: String, ) -> ProxyResult<'static, Arc>> { + let restore_channel = crate::Configuration::restore_channel(); Ok(Arc::new(Mutex::new(Self { - extranonce_prefix: None, + extranonce_prefix: restore_channel.clone(), tx_sv2_set_new_prev_hash, tx_sv2_new_ext_mining_job, channel_id: None, job_id: None, last_job_id: None, min_extranonce_size, + restore_channel, upstream_extranonce1_size: crate::UPSTREAM_EXTRANONCE1_SIZE, tx_sv2_extranonce, target, difficulty_config, sender, - signature, sent_up: 0, rejected: 0, toa: Vec::new(), @@ -183,26 +185,15 @@ impl Upstream { /// Setups the connection with the SV2 Upstream role (most typically a SV2 Pool). async fn connect(self_: Arc>) -> ProxyResult<'static, ()> { - let sender = self_ - .safe_lock(|s| s.sender.clone()) - .map_err(|_e| Error::TranslatorUpstreamMutexPoisoned)?; - - // Send open channel request - let nominal_hash_rate = self_ - .safe_lock(|u| { - u.difficulty_config - .safe_lock(|c| c.channel_nominal_hashrate) - .map_err(|_e| Error::TranslatorDiffConfigMutexPoisoned) + let (sender, open_channel) = self_ + .safe_lock(|s| { + let sender = s.sender.clone(); + let restore_channel = s.restore_channel.clone(); + let open_channel = s.next_open_channel_message(restore_channel)?; + Ok::<_, RolesLogicError>((sender, open_channel)) }) - .map_err(|_e| Error::TranslatorUpstreamMutexPoisoned)??; - let user_identity = "ABC".to_string().try_into().expect("Internal error: this operation can not fail because the string ABC can always be converted into Inner"); - let open_channel = Mining::OpenExtendedMiningChannel(OpenExtendedMiningChannel { - request_id: 0, // TODO - user_identity, // TODO - nominal_hash_rate, - max_target: u256_max(), - min_extranonce_size: crate::MIN_EXTRANONCE2_SIZE, - }); + .map_err(|_e| Error::TranslatorUpstreamMutexPoisoned)? + .map_err(Error::RolesSv2Logic)?; if sender.send(open_channel).await.is_err() { error!("Failed to send message"); @@ -211,6 +202,42 @@ impl Upstream { Ok(()) } + fn next_open_channel_message( + &mut self, + extranonce_prefix: Option>, + ) -> Result, RolesLogicError> { + self.extranonce_prefix = extranonce_prefix.clone(); + let nominal_hash_rate = self + .difficulty_config + .safe_lock(|c| c.channel_nominal_hashrate) + .map_err(|e| RolesLogicError::PoisonLock(e.to_string()))?; + let user_identity = match extranonce_prefix.as_ref() { + Some(inner) => { + let mut user_identity = + Vec::with_capacity(RESTORE_CHANNEL_USER_ID_PREFIX.len() + inner.len()); + user_identity.extend_from_slice(RESTORE_CHANNEL_USER_ID_PREFIX); + user_identity.extend_from_slice(inner); + info!( + "Opening upstream extended channel restore with extranonce size {:?}", + inner + ); + user_identity + } + None => b"ABC".to_vec(), + } + .try_into() + .expect("Internal error: restore channel user identity length must fit in B0255"); + Ok(Mining::OpenExtendedMiningChannel( + OpenExtendedMiningChannel { + request_id: 0, + user_identity, + nominal_hash_rate, + max_target: u256_max(), + min_extranonce_size: crate::MIN_EXTRANONCE2_SIZE, + }, + )) + } + fn handle_token_updates( self_: Arc>, mut rx_update_token: TReceiver, @@ -534,12 +561,10 @@ impl Upstream { } }; - let (channel_id, signature) = match self_.safe_lock(|s| { - s.channel_id - .ok_or(RolesLogicError::NotFoundChannelId) - .map(|channel_id| (channel_id, s.signature.clone())) - }) { - Ok(Ok(values)) => values, + let channel_id = match self_ + .safe_lock(|s| s.channel_id.ok_or(RolesLogicError::NotFoundChannelId)) + { + Ok(Ok(channel_id)) => channel_id, Ok(Err(e)) => { error!("{}", Error::RolesSv2Logic(e)); let _ = result_tx.send(Err( @@ -576,11 +601,21 @@ impl Upstream { } }; + let original_channel_id = share.channel_id; + let original_sequence_number = share.sequence_number; + debug!( + "Translator upstream rewriting SubmitSharesExtended for current pool channel: original_channel_id={} current_channel_id={} job_id={} original_sequence_number={} assigned_sequence_number={}", + original_channel_id, + channel_id, + share.job_id, + original_sequence_number, + sequence_number, + ); share.channel_id = channel_id; share.sequence_number = sequence_number; - let mut extranonce = signature.as_bytes().to_vec(); - extranonce.extend_from_slice(&share.extranonce.to_vec()); - share.extranonce = extranonce.try_into().unwrap(); + //let mut extranonce = signature.as_bytes().to_vec(); + //extranonce.extend_from_slice(&share.extranonce.to_vec()); + //share.extranonce = extranonce.try_into().unwrap(); let message = roles_logic_sv2::parsers::Mining::SubmitSharesExtended(share); @@ -721,16 +756,23 @@ impl ParseUpstreamMiningMessages Result, RolesLogicError> { info!( "Handling OpenExtendedMiningChannelSuccess message from Pool for Channel Id: {}", m.channel_id ); - let mut prefix = m.extranonce_prefix.to_vec(); - prefix.extend_from_slice(self.signature.as_bytes()); - m.extranonce_prefix = prefix.try_into().unwrap(); - m.extranonce_size -= self.signature.len() as u16; + let raw_extranonce_prefix = m.extranonce_prefix.to_vec(); + if let Some(inner) = self.extranonce_prefix.as_ref() { + if raw_extranonce_prefix.as_slice() != inner.as_slice() { + error!( + "Restore channel extranonce prefix mismatch for Channel Id {}: expected {:?}, received {:?}. Clearing restore state so the next reconnect can open a fresh channel.", + m.channel_id, inner, raw_extranonce_prefix + ); + panic!() + } + self.restore_channel = None; + } let tproxy_e1_len = proxy_extranonce1_len(m.extranonce_size as usize, self.min_extranonce_size.into()) as u16; @@ -754,7 +796,6 @@ impl ParseUpstreamMiningMessages Arc> { + test_upstream_with_sender(mpsc::channel(1).0, None).await + } + + async fn test_upstream_with_sender( + sender: TSender>, + restore_channel: Option>, + ) -> Arc> { let (tx_sv2_set_new_prev_hash, _rx_sv2_set_new_prev_hash) = mpsc::channel(1); let (tx_sv2_new_ext_mining_job, _rx_sv2_new_ext_mining_job) = mpsc::channel(1); let (tx_sv2_extranonce, _rx_sv2_extranonce) = mpsc::channel(1); - let (sender, _receiver) = mpsc::channel(1); - Upstream::new( + let upstream = Upstream::new( tx_sv2_set_new_prev_hash, tx_sv2_new_ext_mining_job, crate::MIN_EXTRANONCE_SIZE - 1, @@ -961,10 +1008,112 @@ mod tests { UpstreamDifficultyConfig::new(crate::CHANNEL_DIFF_UPDTATE_INTERVAL, 0.0).0, )), sender, - "sig".to_string(), ) .await - .unwrap() + .unwrap(); + upstream + .safe_lock(|u| { + u.extranonce_prefix = restore_channel.clone(); + u.restore_channel = restore_channel; + }) + .unwrap(); + upstream + } + + fn open_success( + channel_id: u32, + extranonce_prefix: Vec, + extranonce_size: u16, + ) -> roles_logic_sv2::mining_sv2::OpenExtendedMiningChannelSuccess<'static> { + roles_logic_sv2::mining_sv2::OpenExtendedMiningChannelSuccess { + request_id: 0, + channel_id, + target: u256_max(), + extranonce_size, + extranonce_prefix: extranonce_prefix + .try_into() + .expect("test extranonce prefix must fit in B032"), + } + } + + #[tokio::test] + async fn first_connect_sends_normal_open_channel_identity() { + let (sender, mut receiver) = mpsc::channel(1); + let upstream = test_upstream_with_sender(sender, None).await; + + Upstream::connect(upstream).await.unwrap(); + + let message = timeout(Duration::from_millis(250), receiver.recv()) + .await + .unwrap() + .unwrap(); + match message { + Mining::OpenExtendedMiningChannel(open) => { + assert_eq!(open.user_identity.to_vec(), b"ABC"); + assert_eq!(open.request_id, 0); + assert_eq!(open.min_extranonce_size, crate::MIN_EXTRANONCE2_SIZE); + } + other => panic!("expected OpenExtendedMiningChannel, got {other:?}"), + } + } + + #[tokio::test] + async fn reconnect_sends_restore_identity_from_stored_prefix() { + let (sender, mut receiver) = mpsc::channel(1); + let restore_channel = vec![1; 16]; + let upstream = test_upstream_with_sender(sender, Some(restore_channel.clone())).await; + let difficulty_config = upstream.safe_lock(|u| u.difficulty_config.clone()).unwrap(); + difficulty_config + .safe_lock(|c| c.channel_nominal_hashrate = 42.0) + .unwrap(); + + Upstream::connect(upstream).await.unwrap(); + + let message = timeout(Duration::from_millis(250), receiver.recv()) + .await + .unwrap() + .unwrap(); + match message { + Mining::OpenExtendedMiningChannel(open) => { + let mut expected_user_identity = RESTORE_CHANNEL_USER_ID_PREFIX.to_vec(); + expected_user_identity.extend_from_slice(&restore_channel); + assert_eq!(open.user_identity.to_vec(), expected_user_identity); + assert_eq!(open.request_id, 0); + assert_eq!(open.nominal_hash_rate, 42.0); + assert_eq!(open.min_extranonce_size, crate::MIN_EXTRANONCE2_SIZE); + } + other => panic!("expected OpenExtendedMiningChannel, got {other:?}"), + } + } + + #[tokio::test] + async fn open_success_without_restore_keeps_restore_state_empty() { + let upstream = test_upstream_with_sender(mpsc::channel(1).0, None).await; + let pool_prefix = vec![10; 16]; + let success = open_success(21, pool_prefix, 16); + + upstream + .safe_lock(|u| u.handle_open_extended_mining_channel_success(success)) + .unwrap() + .unwrap(); + + assert!(upstream.safe_lock(|u| u.restore_channel.is_none()).unwrap()); + } + + #[tokio::test] + async fn restored_success_updates_channel_id_and_clears_restore_state() { + let previous_prefix = vec![31; 16]; + let upstream = + test_upstream_with_sender(mpsc::channel(1).0, Some(previous_prefix.clone())).await; + let success = open_success(55, previous_prefix.clone(), 16); + + upstream + .safe_lock(|u| u.handle_open_extended_mining_channel_success(success)) + .unwrap() + .unwrap(); + + assert_eq!(upstream.safe_lock(|u| u.channel_id).unwrap(), Some(55)); + assert!(upstream.safe_lock(|u| u.restore_channel.is_none()).unwrap()); } #[tokio::test] @@ -1084,7 +1233,6 @@ mod tests { Arc::new(Mutex::new(vec![0; 32])), difficulty_config.clone(), sender, - "sig".to_string(), ) .await .unwrap(); diff --git a/tests/library_init.rs b/tests/library_init.rs index ef7b7600..6ae82e03 100644 --- a/tests/library_init.rs +++ b/tests/library_init.rs @@ -125,10 +125,12 @@ async fn library_init_sv2_setup_connection() { Some("test_token".to_string()), Some(tp_sniffer_addr.to_string()), None, - vec![mock_pool_mining_addr.to_string()], + vec![proxy_target.to_string()], + None, 120_000, 0, 100_000_000_000_000.0, + 131_072.0, "info".to_string(), "off".to_string(), false,