Compare commits

..

1 Commits

Author SHA1 Message Date
Tommy Verrall 55a0f80d73 Feat: implement supply chain attack mitigation
- Add yarn resolutions for vulnerable packages (chalk, strip-ansi, color-convert, etc.)
- Add .npmrc and .nvmrc security configurations
2025-09-10 18:51:38 +02:00
279 changed files with 10194 additions and 8743 deletions
+5 -1
View File
@@ -30,11 +30,13 @@ jobs:
release_date: ${{ fromJSON(steps.create-release.outputs.assets)[0].published_at }}
client_hash: ${{ steps.binary-hashes.outputs.client_hash }}
nymvisor_hash: ${{ steps.binary-hashes.outputs.nymvisor_hash }}
nymnode_hash: ${{ steps.binary-hashes.outputs.nymnode_hash }}
socks5_hash: ${{ steps.binary-hashes.outputs.socks5_hash }}
netreq_hash: ${{ steps.binary-hashes.outputs.netreq_hash }}
cli_hash: ${{ steps.binary-hashes.outputs.cli_hash }}
client_version: ${{ steps.binary-versions.outputs.client_version }}
nymvisor_version: ${{ steps.binary-versions.outputs.nymvisor_version }}
nymnode_version: ${{ steps.binary-versions.outputs.nymnode_version }}
socks5_version: ${{ steps.binary-versions.outputs.socks5_version }}
netreq_version: ${{ steps.binary-versions.outputs.netreq_version }}
cli_version: ${{ steps.binary-versions.outputs.cli_version }}
@@ -54,7 +56,7 @@ jobs:
- name: Install Rust stable
uses: actions-rs/toolchain@v1
with:
toolchain: 1.88.0
toolchain: 1.86.0
override: true
- name: Build all binaries
@@ -74,6 +76,7 @@ jobs:
target/release/nym-network-requester
target/release/nym-cli
target/release/nymvisor
target/release/nym-node
retention-days: 30
- id: create-release
@@ -88,6 +91,7 @@ jobs:
target/release/nym-network-requester
target/release/nym-cli
target/release/nymvisor
target/release/nym-node
push-release-data-client:
if: ${{ (startsWith(github.ref, 'refs/tags/nym-binaries-') && github.event_name == 'release') || github.event_name == 'workflow_dispatch' }}
+58
View File
@@ -0,0 +1,58 @@
# Security and sensitive files
.env*
*.key
*.pem
*.p12
*.pfx
secrets/
private/
config/secrets/
# Development files
node_modules/
.npm/
.npmrc
.nvmrc
*.log
*.tmp
.DS_Store
Thumbs.db
# Build artifacts
dist/
build/
target/
*.tgz
*.tar.gz
# IDE files
.vscode/
.idea/
*.swp
*.swo
*~
# Test files
test/
tests/
__tests__/
*.test.js
*.test.ts
*.spec.js
*.spec.ts
# Documentation
docs/
*.md
!README.md
# CI/CD files
.github/
.gitlab-ci.yml
.travis.yml
.circleci/
azure-pipelines.yml
# Scripts
scripts/
!scripts/security-check.sh
+21
View File
@@ -0,0 +1,21 @@
audit-level=moderate
fund=false
update-notifier=false
ignore-scripts=false
strict-ssl=true
registry=https://registry.npmjs.org/
audit=true
package-lock=true
package-lock-only=false
save-exact=false
# use npm ci for production builds (faster and more secure)
# this will be enforced in CI/CD scripts
# prevent installation of optional dependencies that might contain vulnerabilities
optional=false
audit=true
update-notifier=false
save-exact=false
+1
View File
@@ -0,0 +1 @@
20.18.0
-42
View File
@@ -4,48 +4,6 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
## [Unreleased]
## [2025.16-halloumi] (2025-09-16)
- Backport metadata endpoint ([#6010])
- bugfix: make sure tables are removed in correct order to not trigger FK constraint issue ([#5987])
- chore: move authenticator into gateway crate ([#5982])
- Fix the ns api ci workflow ([#5981])
- Remove freshness check on testrun submit ([#5977])
- Update sysinfo to the latest ([#5976])
- bugfix: manually calculate per node work on rewarded set changes ([#5972])
- fixing the ci for ns agent ([#5965])
- Feature/testing utils ([#5963])
- bugfix: fix ci-build for linux (and use updated runner) ([#5958])
- chore: updated refs to cheddar rev of nym repo ([#5955])
- http api client adjustment ([#5953])
- chore: fix rust 1.89 clippy issues ([#5944])
- Wireguard metadata client library ([#5943])
- chore: remove unused import ([#5942])
- feat: introduce additional checks when attempting to send to bounded channels ([#5941])
- Move credential verifier in peer controller ([#5938])
- change PK/FK on expiration date signatures tables ([#5934])
- Wireguard private metadata ([#5915])
[#6010]: https://github.com/nymtech/nym/pull/6010
[#5987]: https://github.com/nymtech/nym/pull/5987
[#5982]: https://github.com/nymtech/nym/pull/5982
[#5981]: https://github.com/nymtech/nym/pull/5981
[#5977]: https://github.com/nymtech/nym/pull/5977
[#5976]: https://github.com/nymtech/nym/pull/5976
[#5972]: https://github.com/nymtech/nym/pull/5972
[#5965]: https://github.com/nymtech/nym/pull/5965
[#5963]: https://github.com/nymtech/nym/pull/5963
[#5958]: https://github.com/nymtech/nym/pull/5958
[#5955]: https://github.com/nymtech/nym/pull/5955
[#5953]: https://github.com/nymtech/nym/pull/5953
[#5944]: https://github.com/nymtech/nym/pull/5944
[#5943]: https://github.com/nymtech/nym/pull/5943
[#5942]: https://github.com/nymtech/nym/pull/5942
[#5941]: https://github.com/nymtech/nym/pull/5941
[#5938]: https://github.com/nymtech/nym/pull/5938
[#5934]: https://github.com/nymtech/nym/pull/5934
[#5915]: https://github.com/nymtech/nym/pull/5915
## [2025.15-gruyere] (2025-08-20)
- Migrate strum to 0.27.2 ([#5960])
Generated
+82 -136
View File
@@ -11,7 +11,7 @@ dependencies = [
"macroific",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -408,7 +408,7 @@ dependencies = [
"rustc-hash",
"serde",
"serde_derive",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -490,7 +490,7 @@ checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -501,7 +501,7 @@ checksum = "e539d3fca749fcee5236ab05e93a52867dd549cc157c8cb7f99595f3cedffdb5"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -709,7 +709,7 @@ checksum = "57d123550fa8d071b7255cb0cc04dc302baa6c8c4a79f55701552684d8399bce"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -1340,7 +1340,7 @@ dependencies = [
"heck 0.5.0",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -1591,7 +1591,7 @@ checksum = "a782b93fae93e57ca8ad3e9e994e784583f5933aeaaa5c80a545c4b437be2047"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -1615,7 +1615,7 @@ checksum = "e01c9214319017f6ebd8e299036e1f717fa9bb6724e758f7d6fb2477599d1a29"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -1859,7 +1859,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13b588ba4ac1a99f7f2964d24b3d896ddc6bf847ee3855dbd4366f058cfcd331"
dependencies = [
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -1963,7 +1963,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -2115,7 +2115,7 @@ dependencies = [
"proc-macro2",
"quote",
"strsim",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -2126,7 +2126,7 @@ checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead"
dependencies = [
"darling_core",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -2179,7 +2179,7 @@ dependencies = [
"macroific",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -2222,7 +2222,7 @@ checksum = "30542c1ad912e0e3d22a1935c290e12e8a29d704a420177a31faad4a601a0800"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -2251,7 +2251,7 @@ checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
"unicode-xid",
]
@@ -2263,7 +2263,7 @@ checksum = "bda628edc44c4bb645fbe0f758797143e4e07926f7ebf4e9bdfbd3d2ce621df3"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
"unicode-xid",
]
@@ -2332,7 +2332,7 @@ checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -2385,7 +2385,7 @@ version = "0.1.0"
dependencies = [
"cosmwasm-std",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -2547,7 +2547,7 @@ dependencies = [
"heck 0.5.0",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -2677,7 +2677,7 @@ dependencies = [
"macroific",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -2907,7 +2907,7 @@ checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -3784,7 +3784,7 @@ checksum = "0ab604ee7085efba6efc65e4ebca0e9533e3aff6cb501d7d77b211e3a781c6d5"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -3853,7 +3853,7 @@ dependencies = [
"macroific",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -3955,9 +3955,9 @@ checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02"
[[package]]
name = "inventory"
version = "0.3.21"
version = "0.3.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bc61209c082fbeb19919bee74b176221b27223e27b65d781eb91af24eb1fb46e"
checksum = "ab08d7cd2c5897f2c949e5383ea7c7db03fb19130ffcfbf7eda795137ae3cb83"
dependencies = [
"rustversion",
]
@@ -4103,7 +4103,7 @@ checksum = "03343451ff899767262ec32146f6d559dd759fdadf42ff0e227c7c48f72594b4"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -4375,7 +4375,7 @@ dependencies = [
"proc-macro2",
"quote",
"sealed",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -4387,7 +4387,7 @@ dependencies = [
"proc-macro2",
"quote",
"sealed",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -4400,7 +4400,7 @@ dependencies = [
"macroific_core",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -4428,7 +4428,7 @@ checksum = "ac84fd3f360fcc43dc5f5d186f02a94192761a080e8bc58621ad4d12296a58cf"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -4883,7 +4883,7 @@ dependencies = [
[[package]]
name = "nym-api"
version = "1.1.65"
version = "1.1.64"
dependencies = [
"anyhow",
"async-trait",
@@ -4921,7 +4921,6 @@ dependencies = [
"nym-ecash-signer-check",
"nym-ecash-time",
"nym-gateway-client",
"nym-http-api-client",
"nym-http-api-common",
"nym-mixnet-contract-common",
"nym-node-requests",
@@ -5095,7 +5094,7 @@ dependencies = [
[[package]]
name = "nym-cli"
version = "1.1.62"
version = "1.1.61"
dependencies = [
"anyhow",
"base64 0.22.1",
@@ -5153,7 +5152,6 @@ dependencies = [
"nym-crypto",
"nym-ecash-contract-common",
"nym-ecash-time",
"nym-http-api-client",
"nym-id",
"nym-mixnet-contract-common",
"nym-multisig-contract-common",
@@ -5178,7 +5176,7 @@ dependencies = [
[[package]]
name = "nym-client"
version = "1.1.62"
version = "1.1.61"
dependencies = [
"bs58",
"clap",
@@ -5242,7 +5240,6 @@ dependencies = [
"nym-http-api-client",
"nym-id",
"nym-mixnet-client",
"nym-mixnet-contract-common",
"nym-network-defaults",
"nym-nonexhaustive-delayqueue",
"nym-pemstore",
@@ -5454,7 +5451,7 @@ dependencies = [
[[package]]
name = "nym-credential-proxy"
version = "0.2.0"
version = "0.1.8"
dependencies = [
"anyhow",
"axum 0.7.9",
@@ -5633,7 +5630,6 @@ dependencies = [
"nym-crypto",
"nym-ecash-contract-common",
"nym-ecash-time",
"nym-http-api-client",
"nym-network-defaults",
"nym-serde-helpers",
"nym-validator-client",
@@ -5735,7 +5731,6 @@ version = "0.1.0"
dependencies = [
"futures",
"nym-ecash-signer-check-types",
"nym-http-api-client",
"nym-network-defaults",
"nym-validator-client",
"semver 1.0.26",
@@ -5993,23 +5988,17 @@ dependencies = [
"async-trait",
"bincode",
"bytes",
"cfg-if",
"encoding_rs",
"hickory-resolver",
"http 1.3.1",
"inventory",
"itertools 0.14.0",
"mime",
"nym-bin-common",
"nym-http-api-client-macro",
"nym-http-api-common",
"nym-network-defaults",
"once_cell",
"reqwest 0.12.22",
"serde",
"serde_json",
"serde_plain",
"serde_yaml",
"thiserror 2.0.12",
"tokio",
"tracing",
@@ -6017,19 +6006,6 @@ dependencies = [
"wasmtimer",
]
[[package]]
name = "nym-http-api-client-macro"
version = "0.1.0"
dependencies = [
"nym-http-api-client",
"proc-macro-crate",
"proc-macro2",
"quote",
"reqwest 0.12.22",
"syn 2.0.106",
"uuid",
]
[[package]]
name = "nym-http-api-common"
version = "0.1.0"
@@ -6293,8 +6269,6 @@ dependencies = [
"nym-client-core",
"nym-crypto",
"nym-gateway-requests",
"nym-http-api-client",
"nym-mixnet-contract-common",
"nym-network-defaults",
"nym-sdk",
"nym-sphinx",
@@ -6316,7 +6290,7 @@ dependencies = [
[[package]]
name = "nym-network-requester"
version = "1.1.63"
version = "1.1.62"
dependencies = [
"addr",
"anyhow",
@@ -6366,7 +6340,7 @@ dependencies = [
[[package]]
name = "nym-node"
version = "1.18.0"
version = "1.17.0"
dependencies = [
"anyhow",
"arc-swap",
@@ -6436,7 +6410,6 @@ dependencies = [
"thiserror 2.0.12",
"time",
"tokio",
"tokio-stream",
"tokio-util",
"toml 0.8.23",
"tower-http 0.5.2",
@@ -6763,7 +6736,6 @@ dependencies = [
"nym-credentials-interface",
"nym-crypto",
"nym-gateway-requests",
"nym-http-api-client",
"nym-network-defaults",
"nym-ordered-buffer",
"nym-service-providers-common",
@@ -6853,7 +6825,7 @@ dependencies = [
[[package]]
name = "nym-socks5-client"
version = "1.1.62"
version = "1.1.61"
dependencies = [
"bs58",
"clap",
@@ -7185,11 +7157,9 @@ dependencies = [
name = "nym-task"
version = "0.1.0"
dependencies = [
"anyhow",
"cfg-if",
"futures",
"log",
"nym-test-utils",
"thiserror 2.0.12",
"tokio",
"tokio-util",
@@ -7377,7 +7347,6 @@ dependencies = [
"nym-credentials-interface",
"nym-crypto",
"nym-ecash-time",
"nym-http-api-client",
"nym-network-defaults",
"nym-pemstore",
"nym-serde-helpers",
@@ -7407,9 +7376,7 @@ dependencies = [
"bytes",
"futures",
"humantime",
"nym-api-requests",
"nym-crypto",
"nym-http-api-client",
"nym-task",
"nym-validator-client",
"rand 0.8.5",
@@ -7611,7 +7578,7 @@ dependencies = [
[[package]]
name = "nymvisor"
version = "0.1.27"
version = "0.1.26"
dependencies = [
"anyhow",
"bytes",
@@ -8049,7 +8016,7 @@ dependencies = [
"pest_meta",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -8112,7 +8079,7 @@ dependencies = [
"phf_shared",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -8141,7 +8108,7 @@ checksum = "6e918e4ff8c4549eb882f14b3a4bc8c8bc93de829416eacf579f1207a8fbf861"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -8349,15 +8316,6 @@ dependencies = [
"elliptic-curve",
]
[[package]]
name = "proc-macro-crate"
version = "3.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "edce586971a4dfaa28950c6f18ed55e0406c1ab88bbce2c6f6293a7aaba73d35"
dependencies = [
"toml_edit",
]
[[package]]
name = "proc-macro-error-attr2"
version = "2.0.0"
@@ -8377,7 +8335,7 @@ dependencies = [
"proc-macro-error-attr2",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -8433,7 +8391,7 @@ dependencies = [
"itertools 0.14.0",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -8698,7 +8656,7 @@ checksum = "1165225c21bff1f3bbce98f5a1f889949bc902d3575308cc7b0de30b4f6d27c7"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -8948,7 +8906,7 @@ dependencies = [
"proc-macro2",
"quote",
"rust-embed-utils",
"syn 2.0.106",
"syn 2.0.104",
"walkdir",
]
@@ -9252,7 +9210,7 @@ dependencies = [
"proc-macro2",
"quote",
"serde_derive_internals 0.29.1",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9284,7 +9242,7 @@ checksum = "1783eabc414609e28a5ba76aee5ddd52199f7107a0b24c2e9746a1ecc34a683d"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9305,7 +9263,7 @@ checksum = "22f968c5ea23d555e670b449c1c5e7b2fc399fdaec1d304a17cd48e288abc107"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9445,7 +9403,7 @@ checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9456,7 +9414,7 @@ checksum = "e578a843d40b4189a4d66bba51d7684f57da5bd7c304c64e14bd63efbef49509"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9467,7 +9425,7 @@ checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9529,7 +9487,7 @@ checksum = "aafbefbe175fa9bf03ca83ef89beecff7d2a95aaacd5732325b90ac8c3bd7b90"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9542,15 +9500,6 @@ dependencies = [
"serde",
]
[[package]]
name = "serde_plain"
version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ce1fc6db65a611022b23a0dec6975d63fb80a302cb3388835ff02c097258d50"
dependencies = [
"serde",
]
[[package]]
name = "serde_repr"
version = "0.1.20"
@@ -9559,7 +9508,7 @@ checksum = "175ee3e80ae9982737ca543e96133087cbd9a485eecc3bc4de9c1a37b47ea59c"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9612,7 +9561,7 @@ dependencies = [
"darling",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9958,7 +9907,7 @@ dependencies = [
"quote",
"sqlx-core",
"sqlx-macros-core",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -9981,7 +9930,7 @@ dependencies = [
"sqlx-mysql",
"sqlx-postgres",
"sqlx-sqlite",
"syn 2.0.106",
"syn 2.0.104",
"tokio",
"url",
]
@@ -10190,7 +10139,7 @@ dependencies = [
"heck 0.5.0",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -10246,9 +10195,9 @@ dependencies = [
[[package]]
name = "syn"
version = "2.0.106"
version = "2.0.104"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ede7c438028d4436d71104916910f5bb611972c5cfd7f89b8300a8186e6fada6"
checksum = "17b6f705963418cdb9927482fa304bc562ece2fdd4f616084c50b7023b435a40"
dependencies = [
"proc-macro2",
"quote",
@@ -10278,7 +10227,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -10475,7 +10424,7 @@ dependencies = [
"proc-macro2",
"quote",
"regex",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -10499,7 +10448,6 @@ dependencies = [
"nym-crypto",
"nym-ecash-contract-common",
"nym-group-contract-common",
"nym-http-api-client",
"nym-mixnet-contract-common",
"nym-multisig-contract-common",
"nym-pemstore",
@@ -10555,7 +10503,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -10566,7 +10514,7 @@ checksum = "7f7cf42b4507d8ea322120659672cf1b9dbb93f8f2d4ecfd6e51350ff5b17a1d"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -10698,7 +10646,7 @@ checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -11019,7 +10967,7 @@ checksum = "81383ab64e72a7a8b8e13130c49e3dab29def6d0c7d76a03087b3cf71c5c6903"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -11189,7 +11137,7 @@ checksum = "0e9d8656589772eeec2cf7a8264d9cda40fb28b9bc53118ceb9e8c07f8f38730"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
"termcolor",
]
@@ -11216,7 +11164,7 @@ dependencies = [
"proc-macro2",
"quote",
"serde_derive_internals 0.28.0",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -11406,7 +11354,7 @@ dependencies = [
"indexmap 2.10.0",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -11421,7 +11369,7 @@ dependencies = [
"proc-macro2",
"quote",
"serde",
"syn 2.0.106",
"syn 2.0.104",
"toml 0.5.11",
"uniffi_meta",
]
@@ -11548,7 +11496,7 @@ dependencies = [
"proc-macro2",
"quote",
"regex",
"syn 2.0.106",
"syn 2.0.104",
"uuid",
]
@@ -11587,7 +11535,7 @@ checksum = "268d76aaebb80eba79240b805972e52d7d410d4bcc52321b951318b0f440cd60"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -11598,7 +11546,7 @@ checksum = "382673bda1d05c85b4550d32fd4192ccd4cffe9a908543a0795d1e7682b36246"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
"utoipauto-core",
]
@@ -11622,7 +11570,6 @@ dependencies = [
"clap",
"comfy-table",
"nym-bin-common",
"nym-http-api-client",
"nym-network-defaults",
"nym-validator-client",
"serde",
@@ -11776,7 +11723,7 @@ dependencies = [
"log",
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
"wasm-bindgen-shared",
]
@@ -11811,7 +11758,7 @@ checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
"wasm-bindgen-backend",
"wasm-bindgen-shared",
]
@@ -11846,7 +11793,7 @@ checksum = "17d5042cc5fa009658f9a7333ef24291b1291a25b6382dd68862a7f3b969f69b"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -11862,7 +11809,6 @@ dependencies = [
"nym-credential-storage",
"nym-crypto",
"nym-gateway-client",
"nym-http-api-client",
"nym-sphinx",
"nym-sphinx-acknowledgements",
"nym-statistics-common",
@@ -12118,7 +12064,7 @@ checksum = "a47fddd13af08290e67f4acabf4b459f647552718f683a7b415d290ac744a836"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -12129,7 +12075,7 @@ checksum = "bd9211b69f8dcdfa817bfd14bf1c97c9188afa36f4750130fcdf3f400eca9fa8"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -12553,7 +12499,7 @@ checksum = "38da3c9736e16c5d3c8c597a9aaa5d1fa565d0532ae05e27c24aa62fb32c0ab6"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
"synstructure",
]
@@ -12574,7 +12520,7 @@ checksum = "9ecf5b4cc5364572d7f4c329661bcc82724222973f2cab6f050a4e5c22f75181"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -12594,7 +12540,7 @@ checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
"synstructure",
]
@@ -12615,7 +12561,7 @@ checksum = "ce36e65b0d2999d2aafac989fb249189a141aee1f53c612c1f37d72631959f69"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
@@ -12648,7 +12594,7 @@ checksum = "5b96237efa0c878c64bd89c436f661be4e46b2f3eff1ebb976f7ef2321d2f58f"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"syn 2.0.104",
]
[[package]]
+1 -3
View File
@@ -58,7 +58,7 @@ members = [
"common/gateway-requests",
"common/gateway-stats-storage",
"common/gateway-storage",
"common/http-api-client", "common/http-api-client-macro",
"common/http-api-client",
"common/http-api-common",
"common/inclusion-probability",
"common/ip-packet-requests",
@@ -275,7 +275,6 @@ hyper = "1.6.0"
hyper-util = "0.1"
indicatif = "0.18.0"
inquire = "0.6.2"
inventory = "0.3.21"
ip_network = "0.4.1"
ipnetwork = "0.20"
itertools = "0.14.0"
@@ -323,7 +322,6 @@ serde_json_path = "0.7.2"
serde_repr = "0.1"
serde_with = "3.9.0"
serde_yaml = "0.9.25"
serde_plain = "1.0.2"
sha2 = "0.10.9"
si-scale = "0.2.3"
snow = "0.9.6"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-client"
version = "1.1.62"
version = "1.1.61"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
description = "Implementation of the Nym Client"
edition = "2021"
+10 -14
View File
@@ -11,7 +11,7 @@ use nym_client_core::client::base_client::{
BaseClientBuilder, ClientInput, ClientOutput, ClientState,
};
use nym_sphinx::params::PacketType;
use nym_task::ShutdownManager;
use nym_task::TaskHandle;
use nym_validator_client::QueryHttpRpcNyxdClient;
use std::error::Error;
use std::path::PathBuf;
@@ -29,8 +29,6 @@ pub struct SocketClient {
/// Optional path to a .json file containing standalone network details.
custom_mixnet: Option<PathBuf>,
shutdown_manager: ShutdownManager,
}
impl SocketClient {
@@ -42,7 +40,6 @@ impl SocketClient {
SocketClient {
config,
custom_mixnet,
shutdown_manager: Default::default(),
}
}
@@ -52,7 +49,7 @@ impl SocketClient {
client_output: ClientOutput,
client_state: ClientState,
self_address: &Recipient,
shutdown_token: nym_task::ShutdownToken,
task_client: nym_task::TaskClient,
packet_type: PacketType,
) {
info!("Starting websocket listener...");
@@ -80,24 +77,24 @@ impl SocketClient {
shared_lane_queue_lengths,
reply_controller_sender,
Some(packet_type),
shutdown_token.clone(),
task_client.fork("websocket_handler"),
);
websocket::Listener::new(
config.socket.host,
config.socket.listening_port,
shutdown_token.child_token(),
task_client.with_suffix("websocket_listener"),
)
.start(websocket_handler);
}
/// blocking version of `start_socket` method. Will run forever (or until SIGINT is sent)
pub async fn run_socket_forever(self) -> Result<(), Box<dyn Error + Send + Sync>> {
let mut shutdown = self.start_socket().await?;
let shutdown = self.start_socket().await?;
shutdown.run_until_shutdown().await;
let res = shutdown.wait_for_shutdown().await;
log::info!("Stopping nym-client");
Ok(())
res
}
async fn initialise_storage(&self) -> Result<OnDiskPersistent, ClientError> {
@@ -122,7 +119,6 @@ impl SocketClient {
let mut base_client =
BaseClientBuilder::new(self.config().base(), storage, dkg_query_client)
.with_shutdown(self.shutdown_manager.shutdown_tracker_owned())
.with_user_agent(user_agent);
if let Some(custom_mixnet) = &self.custom_mixnet {
@@ -132,7 +128,7 @@ impl SocketClient {
Ok(base_client)
}
pub async fn start_socket(self) -> Result<ShutdownManager, ClientError> {
pub async fn start_socket(self) -> Result<TaskHandle, ClientError> {
if !self.config.socket.socket_type.is_websocket() {
return Err(ClientError::InvalidSocketMode);
}
@@ -151,13 +147,13 @@ impl SocketClient {
client_output,
client_state,
&self_address,
self.shutdown_manager.child_shutdown_token(),
started_client.task_handle.get_handle(),
packet_type,
);
info!("Client startup finished!");
info!("The address of this client is: {self_address}");
Ok(self.shutdown_manager)
Ok(started_client.task_handle)
}
}
+27 -21
View File
@@ -19,7 +19,7 @@ use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::connections::{
ConnectionCommand, ConnectionCommandSender, ConnectionId, LaneQueueLengths, TransmissionLane,
};
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use std::time::Duration;
use tokio::net::TcpStream;
use tokio::time::Instant;
@@ -44,7 +44,7 @@ pub(crate) struct HandlerBuilder {
lane_queue_lengths: LaneQueueLengths,
reply_controller_sender: ReplyControllerSender,
packet_type: Option<PacketType>,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl HandlerBuilder {
@@ -57,7 +57,7 @@ impl HandlerBuilder {
lane_queue_lengths: LaneQueueLengths,
reply_controller_sender: ReplyControllerSender,
packet_type: Option<PacketType>,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self {
Self {
msg_input,
@@ -67,13 +67,14 @@ impl HandlerBuilder {
lane_queue_lengths,
reply_controller_sender,
packet_type,
shutdown_token,
task_client,
}
}
// TODO: make sure we only ever have one active handler
pub fn create_active_handler(&self) -> Handler {
let shutdown_token = self.shutdown_token.clone();
let mut task_client = self.task_client.fork("active_handler");
task_client.disarm();
Handler {
msg_input: self.msg_input.clone(),
client_connection_tx: self.client_connection_tx.clone(),
@@ -84,7 +85,7 @@ impl HandlerBuilder {
lane_queue_lengths: self.lane_queue_lengths.clone(),
reply_controller_sender: self.reply_controller_sender.clone(),
packet_type: self.packet_type,
shutdown_token,
task_client,
}
}
}
@@ -99,14 +100,19 @@ pub(crate) struct Handler {
lane_queue_lengths: LaneQueueLengths,
reply_controller_sender: ReplyControllerSender,
packet_type: Option<PacketType>,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl Drop for Handler {
fn drop(&mut self) {
let _ = self
if let Err(err) = self
.buffer_requester
.unbounded_send(ReceivedBufferMessage::ReceiverDisconnect);
.unbounded_send(ReceivedBufferMessage::ReceiverDisconnect)
{
if !self.task_client.is_shutdown_poll() {
error!("failed to disconnect the receiver from the buffer: {err}");
}
}
}
}
@@ -136,7 +142,7 @@ impl Handler {
{
Ok(length) => length,
Err(err) => {
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!(
"Failed to get reply queue length for connection {connection_id}: {err}"
);
@@ -186,7 +192,7 @@ impl Handler {
// the ack control is now responsible for chunking, etc.
let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type);
if let Err(err) = self.msg_input.send(input_msg).await {
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send message to the input buffer: {err}");
}
}
@@ -219,7 +225,7 @@ impl Handler {
let input_msg =
InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
if let Err(err) = self.msg_input.send(input_msg).await {
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send anonymous message to the input buffer: {err}");
}
}
@@ -247,7 +253,7 @@ impl Handler {
let input_msg = InputMessage::new_reply(recipient_tag, message, lane, self.packet_type);
if let Err(err) = self.msg_input.send(input_msg).await {
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send reply message to the input buffer: {err}");
}
}
@@ -269,7 +275,7 @@ impl Handler {
.client_connection_tx
.unbounded_send(ConnectionCommand::Close(connection_id))
{
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send close connection command: {err}");
}
}
@@ -388,14 +394,11 @@ impl Handler {
}
async fn listen_for_requests(&mut self, mut msg_receiver: ReconstructedMessagesReceiver) {
let shutdown_token = self.shutdown_token.clone();
let mut task_client = self.task_client.fork("select");
task_client.disarm();
loop {
while !task_client.is_shutdown() {
tokio::select! {
_ = shutdown_token.cancelled() => {
log::trace!("Websocket handler: Received shutdown");
break;
}
// we can either get a client request from the websocket
socket_msg = self.next_websocket_request() => {
if socket_msg.is_none() {
@@ -433,6 +436,9 @@ impl Handler {
break;
}
}
_ = task_client.recv() => {
log::trace!("Websocket handler: Received shutdown");
}
}
}
log::debug!("Websocket handler: Exiting");
@@ -458,7 +464,7 @@ impl Handler {
reconstructed_sender,
))
{
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!("failed to announce the receiver to the buffer: {err}");
}
}
+7 -7
View File
@@ -3,7 +3,7 @@
use super::handler::HandlerBuilder;
use log::*;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use std::net::IpAddr;
use std::{net::SocketAddr, process, sync::Arc};
use tokio::io::AsyncWriteExt;
@@ -23,15 +23,15 @@ impl State {
pub(crate) struct Listener {
address: SocketAddr,
state: State,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl Listener {
pub(crate) fn new(host: IpAddr, port: u16, shutdown_token: ShutdownToken) -> Self {
pub(crate) fn new(host: IpAddr, port: u16, task_client: TaskClient) -> Self {
Listener {
address: SocketAddr::new(host, port),
state: State::AwaitingConnection,
shutdown_token,
task_client,
}
}
@@ -46,11 +46,11 @@ impl Listener {
let notify = Arc::new(Notify::new());
while !self.shutdown_token.is_cancelled() {
while !self.task_client.is_shutdown() {
tokio::select! {
// When the handler finishes we check if shutdown is signalled
_ = notify.notified() => {
if self.shutdown_token.is_cancelled() {
if self.task_client.is_shutdown() {
log::trace!("Websocket listener: detected shutdown after connection closed");
break;
}
@@ -59,7 +59,7 @@ impl Listener {
}
// ... but when there is no connected client at the time of shutdown being
// signalled, we handle it here.
_ = self.shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
if !self.state.is_connected() {
log::trace!("Not connected: shutting down");
break;
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-socks5-client"
version = "1.1.62"
version = "1.1.61"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
description = "A SOCKS5 localhost proxy that converts incoming messages to Sphinx and sends them to a Nym address"
edition = "2021"
+1 -1
View File
@@ -13,7 +13,7 @@ use nym_credentials_interface::{
};
use nym_ecash_time::Date;
use nym_validator_client::coconut::all_ecash_api_clients;
use nym_validator_client::nym_api::{EpochId, NymApiClientExt};
use nym_validator_client::nym_api::EpochId;
use nym_validator_client::nyxd::contract_traits::DkgQueryClient;
use nym_validator_client::EcashApiClient;
use rand::prelude::SliceRandom;
-1
View File
@@ -53,7 +53,6 @@ nym-client-core-config-types = { path = "./config-types", features = [
nym-client-core-surb-storage = { path = "./surb-storage" }
nym-client-core-gateways-storage = { path = "./gateways-storage" }
nym-ecash-time = { path = "../ecash-time" }
nym-mixnet-contract-common = { path = "../cosmwasm-smart-contracts/mixnet-contract" }
[target."cfg(not(target_arch = \"wasm32\"))".dependencies]
nym-mixnet-client = { path = "../client-libs/mixnet-client", default-features = false }
+77 -148
View File
@@ -27,13 +27,13 @@ use crate::client::topology_control::nym_api_provider::NymApiTopologyProvider;
use crate::client::topology_control::{
TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
};
use crate::config;
use crate::config::{Config, DebugConfig};
use crate::error::ClientCoreError;
use crate::init::{
setup_gateway,
types::{GatewaySetup, InitialisationResult},
};
use crate::{config, spawn_future};
use futures::channel::mpsc;
use nym_bandwidth_controller::BandwidthController;
use nym_client_core_config_types::{ForgetMe, RememberMe};
@@ -48,15 +48,16 @@ use nym_gateway_client::{
use nym_sphinx::acknowledgements::AckKey;
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::addressing::nodes::NodeIdentity;
use nym_sphinx::params::PacketType;
use nym_sphinx::receiver::{ReconstructedMessage, SphinxMessageReceiver};
use nym_statistics_common::clients::ClientStatsSender;
use nym_statistics_common::generate_client_stats_id;
use nym_task::connections::{ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths};
use nym_task::ShutdownTracker;
use nym_task::{TaskClient, TaskHandle};
use nym_topology::provider_trait::TopologyProvider;
use nym_topology::HardcodedTopologyProvider;
use nym_validator_client::nym_api::NymApiClientExt;
use nym_validator_client::{nyxd::contract_traits::DkgQueryClient, UserAgent};
use nym_validator_client::{nyxd::contract_traits::DkgQueryClient, NymApiClient, UserAgent};
use rand::prelude::SliceRandom;
use rand::rngs::OsRng;
use rand::thread_rng;
@@ -94,6 +95,7 @@ impl ClientInput {
}
}
#[derive(Clone)]
pub struct ClientOutput {
pub received_buffer_request_sender: ReceivedBufferRequestSender,
}
@@ -193,7 +195,7 @@ pub struct BaseClientBuilder<C, S: MixnetClientStorage> {
wait_for_gateway: bool,
custom_topology_provider: Option<Box<dyn TopologyProvider + Send + Sync>>,
custom_gateway_transceiver: Option<Box<dyn GatewayTransceiver + Send>>,
shutdown: Option<ShutdownTracker>,
shutdown: Option<TaskClient>,
user_agent: Option<UserAgent>,
setup_method: GatewaySetup,
@@ -279,7 +281,7 @@ where
}
#[must_use]
pub fn with_shutdown(mut self, shutdown: ShutdownTracker) -> Self {
pub fn with_shutdown(mut self, shutdown: TaskClient) -> Self {
self.shutdown = Some(shutdown);
self
}
@@ -323,11 +325,11 @@ where
topology_accessor: TopologyAccessor,
mix_tx: BatchMixMessageSender,
stats_tx: ClientStatsSender,
shutdown_tracker: &ShutdownTracker,
task_client: TaskClient,
) {
info!("Starting loop cover traffic stream...");
let mut stream = LoopCoverTrafficStream::new(
let stream = LoopCoverTrafficStream::new(
ack_key,
debug_config.acknowledgements.average_ack_delay,
mix_tx,
@@ -336,9 +338,10 @@ where
debug_config.traffic,
debug_config.cover_traffic,
stats_tx,
task_client,
);
shutdown_tracker
.try_spawn_named_with_shutdown(async move { stream.run().await }, "CoverTrafficStream");
stream.start();
}
#[allow(clippy::too_many_arguments)]
@@ -354,12 +357,13 @@ where
reply_controller_receiver: ReplyControllerReceiver,
lane_queue_lengths: LaneQueueLengths,
client_connection_rx: ConnectionCommandReceiver,
task_client: TaskClient,
packet_type: PacketType,
stats_tx: ClientStatsSender,
shutdown_tracker: &ShutdownTracker,
) {
info!("Starting real traffic stream...");
let real_messages_controller = RealMessagesController::new(
RealMessagesController::new(
controller_config,
key_rotation_config,
ack_receiver,
@@ -372,63 +376,9 @@ where
lane_queue_lengths,
client_connection_rx,
stats_tx,
shutdown_tracker.clone_shutdown_token(),
);
// break out all the subtasks
let (mut out_queue_control, mut reply_controller, ack_controller) =
real_messages_controller.into_tasks();
let (
mut ack_listener,
mut input_listener,
mut retransmission_listener,
mut sent_notification_listener,
mut ack_action_controller,
) = ack_controller.into_tasks();
shutdown_tracker.try_spawn_named(
async move { out_queue_control.run().await },
"RealMessagesController::OutQueueControl",
);
let shutdown_token = shutdown_tracker.clone_shutdown_token();
shutdown_tracker.try_spawn_named(
async move { reply_controller.run(shutdown_token).await },
"RealMessagesController::ReplyController",
);
let shutdown_token = shutdown_tracker.clone_shutdown_token();
shutdown_tracker.try_spawn_named(
async move { ack_listener.run(shutdown_token).await },
"AcknowledgementController::AcknowledgementListener",
);
let shutdown_token = shutdown_tracker.clone_shutdown_token();
shutdown_tracker.try_spawn_named(
async move { input_listener.run(shutdown_token).await },
"AcknowledgementController::InputMessageListener",
);
let shutdown_token = shutdown_tracker.clone_shutdown_token();
shutdown_tracker.try_spawn_named(
async move { retransmission_listener.run(shutdown_token).await },
"AcknowledgementController::RetransmissionRequestListener",
);
shutdown_tracker.try_spawn_named_with_shutdown(
async move {
sent_notification_listener.run().await;
},
"AcknowledgementController::SentNotificationListener",
);
let shutdown_token = shutdown_tracker.clone_shutdown_token();
shutdown_tracker.try_spawn_named(
async move { ack_action_controller.run(shutdown_token).await },
"AcknowledgementController::ActionController",
);
// .start(packet_type);
task_client,
)
.start(packet_type);
}
// buffer controlling all messages fetched from provider
@@ -439,29 +389,21 @@ where
mixnet_receiver: MixnetMessageReceiver,
reply_key_storage: SentReplyKeys,
reply_controller_sender: ReplyControllerSender,
shutdown: TaskClient,
metrics_reporter: ClientStatsSender,
shutdown_tracker: &ShutdownTracker,
) {
info!("Starting received messages buffer controller...");
let controller = ReceivedMessagesBufferController::<SphinxMessageReceiver>::new(
local_encryption_keypair,
query_receiver,
mixnet_receiver,
reply_key_storage,
reply_controller_sender,
metrics_reporter,
shutdown_tracker.clone_shutdown_token(),
);
let (mut msg_receiver, mut req_receiver) = controller.into_tasks();
shutdown_tracker.try_spawn_named(
async move { msg_receiver.run().await },
"ReceivedMessagesBufferController::FragmentedMessageReceiver",
);
shutdown_tracker.try_spawn_named(
async move { req_receiver.run().await },
"ReceivedMessagesBufferController::RequestReceiver",
);
let controller: ReceivedMessagesBufferController<SphinxMessageReceiver> =
ReceivedMessagesBufferController::new(
local_encryption_keypair,
query_receiver,
mixnet_receiver,
reply_key_storage,
reply_controller_sender,
metrics_reporter,
shutdown,
);
controller.start()
}
#[allow(clippy::too_many_arguments)]
@@ -473,7 +415,7 @@ where
packet_router: PacketRouter,
stats_reporter: ClientStatsSender,
#[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
shutdown_tracker: &ShutdownTracker,
shutdown: TaskClient,
) -> Result<GatewayClient<C, S::CredentialStore>, ClientCoreError>
where
<S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
@@ -492,7 +434,7 @@ where
packet_router,
bandwidth_controller,
stats_reporter,
shutdown_tracker.clone_shutdown_token(),
shutdown,
)
} else {
let cfg = GatewayConfig::new(
@@ -517,7 +459,7 @@ where
stats_reporter,
#[cfg(unix)]
connection_fd_callback,
shutdown_tracker.clone_shutdown_token(),
shutdown,
)
};
@@ -580,7 +522,7 @@ where
packet_router: PacketRouter,
stats_reporter: ClientStatsSender,
#[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
shutdown_tracker: &ShutdownTracker,
mut shutdown: TaskClient,
) -> Result<Box<dyn GatewayTransceiver + Send>, ClientCoreError>
where
<S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
@@ -597,6 +539,7 @@ where
Err(ClientCoreError::CustomGatewaySelectionExpected)
} else {
// and make sure to invalidate the task client, so we wouldn't cause premature shutdown
shutdown.disarm();
custom_gateway_transceiver.set_packet_router(packet_router)?;
Ok(custom_gateway_transceiver)
};
@@ -612,7 +555,7 @@ where
stats_reporter,
#[cfg(unix)]
connection_fd_callback,
shutdown_tracker,
shutdown,
)
.await?;
@@ -623,7 +566,7 @@ where
custom_provider: Option<Box<dyn TopologyProvider + Send + Sync>>,
config_topology: config::Topology,
nym_api_urls: Vec<Url>,
nym_api_client: nym_http_api_client::Client,
nym_api_client: NymApiClient,
) -> Box<dyn TopologyProvider + Send + Sync> {
// if no custom provider was ... provided ..., create one using nym-api
custom_provider.unwrap_or_else(|| {
@@ -643,20 +586,22 @@ where
topology_accessor: TopologyAccessor,
local_gateway: NodeIdentity,
wait_for_gateway: bool,
shutdown_tracker: &ShutdownTracker,
mut task_client: TaskClient,
) -> Result<(), ClientCoreError> {
let topology_refresher_config =
TopologyRefresherConfig::new(topology_config.topology_refresh_rate);
if topology_config.disable_refreshing {
// if we're not spawning the refresher, don't cause shutdown immediately
info!("The background topology refresher is not going to be started");
info!("The background topology refesher is not going to be started");
task_client.disarm();
}
let mut topology_refresher = TopologyRefresher::new(
topology_refresher_config,
topology_accessor,
topology_provider,
task_client,
);
// before returning, block entire runtime to refresh the current network view so that any
// components depending on topology would see a non-empty view
@@ -701,10 +646,7 @@ where
// don't spawn the refresher if we don't want to be refreshing the topology.
// only use the initial values obtained
info!("Starting topology refresher...");
shutdown_tracker.try_spawn_named_with_shutdown(
async move { topology_refresher.run().await },
"TopologyRefresher",
);
topology_refresher.start();
}
Ok(())
@@ -715,7 +657,7 @@ where
user_agent: Option<UserAgent>,
client_stats_id: String,
input_sender: Sender<InputMessage>,
shutdown_tracker: &ShutdownTracker,
task_client: TaskClient,
) -> ClientStatsSender {
info!("Starting statistics control...");
StatisticsControl::create_and_start(
@@ -725,23 +667,18 @@ where
.unwrap_or("unknown".to_string()),
client_stats_id,
input_sender.clone(),
shutdown_tracker,
task_client,
)
}
fn start_mix_traffic_controller(
gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
shutdown_tracker: &ShutdownTracker,
shutdown: TaskClient,
) -> (BatchMixMessageSender, ClientRequestSender) {
info!("Starting mix traffic controller...");
let (mut mix_traffic_controller, mix_tx, client_tx) =
MixTrafficController::new(gateway_transceiver, shutdown_tracker.clone_shutdown_token());
shutdown_tracker.try_spawn_named(
async move { mix_traffic_controller.run().await },
"MixTrafficController",
);
let (mix_traffic_controller, mix_tx, client_tx) =
MixTrafficController::new(gateway_transceiver, shutdown);
mix_traffic_controller.start();
(mix_tx, client_tx)
}
@@ -749,7 +686,7 @@ where
async fn setup_persistent_reply_storage(
backend: S::ReplyStore,
key_rotation_config: KeyRotationConfig,
shutdown_tracker: &ShutdownTracker,
shutdown: TaskClient,
) -> Result<CombinedReplyStorage, ClientCoreError>
where
<S::ReplyStore as ReplyStorageBackend>::StorageError: Sync + Send,
@@ -774,14 +711,13 @@ where
})?;
let store_clone = mem_store.clone();
let shutdown_token = shutdown_tracker.clone_shutdown_token();
shutdown_tracker.try_spawn_named(
spawn_future!(
async move {
persistent_storage
.flush_on_shutdown(store_clone, shutdown_token)
.flush_on_shutdown(store_clone, shutdown)
.await
},
"PersistentReplyStorage::flush_on_shutdown",
"PersistentReplyStorage::flush_on_shutdown"
);
Ok(mem_store)
@@ -813,29 +749,21 @@ where
setup_gateway(setup_method, key_store, details_store).await
}
fn construct_nym_api_client(
config: &Config,
user_agent: Option<UserAgent>,
) -> Result<nym_http_api_client::Client, ClientCoreError> {
fn construct_nym_api_client(config: &Config, user_agent: Option<UserAgent>) -> NymApiClient {
let mut nym_api_urls = config.get_nym_api_endpoints();
nym_api_urls.shuffle(&mut thread_rng());
let mut builder = nym_http_api_client::Client::builder(nym_api_urls[0].clone())
.map_err(ClientCoreError::from)?;
if let Some(user_agent) = user_agent {
builder = builder.with_user_agent(user_agent);
NymApiClient::new_with_user_agent(nym_api_urls[0].clone(), user_agent)
} else {
NymApiClient::new(nym_api_urls[0].clone())
}
builder = builder.with_bincode();
builder.build().map_err(ClientCoreError::from)
}
async fn determine_key_rotation_state(
client: &nym_http_api_client::Client,
client: &NymApiClient,
) -> Result<KeyRotationConfig, ClientCoreError> {
Ok(client.get_key_rotation_info().await?.into())
Ok(client.nym_api.get_key_rotation_info().await?.into())
}
pub async fn start_base(mut self) -> Result<BaseClient, ClientCoreError>
@@ -880,12 +808,12 @@ where
let shared_topology_accessor =
TopologyAccessor::new(self.config.debug.topology.ignore_egress_epoch_role);
// Create a shutdown tracker for this client - either as a child of provided tracker
// or get one from the registry
let shutdown_tracker = match self.shutdown {
Some(parent_tracker) => parent_tracker.child_tracker(),
None => nym_task::get_sdk_shutdown_tracker()?,
};
// Shutdown notifier for signalling tasks to stop
let shutdown = self
.shutdown
.map(Into::<TaskHandle>::into)
.unwrap_or_default()
.name_if_unnamed("BaseNymClient");
// channels responsible for dealing with reply-related fun
let (reply_controller_sender, reply_controller_receiver) =
@@ -902,7 +830,7 @@ where
.dkg_query_client
.map(|client| BandwidthController::new(credential_store, client));
let nym_api_client = Self::construct_nym_api_client(&self.config, self.user_agent.clone())?;
let nym_api_client = Self::construct_nym_api_client(&self.config, self.user_agent.clone());
let key_rotation_config = Self::determine_key_rotation_state(&nym_api_client).await?;
let topology_provider = Self::setup_topology_provider(
@@ -917,7 +845,7 @@ where
self.user_agent.clone(),
generate_client_stats_id(*self_address.identity()),
input_sender.clone(),
&shutdown_tracker.child_tracker(),
shutdown.fork("statistics_control"),
);
// needs to be started as the first thing to block if required waiting for the gateway
@@ -927,14 +855,14 @@ where
shared_topology_accessor.clone(),
self_address.gateway(),
self.wait_for_gateway,
&shutdown_tracker.child_tracker(),
shutdown.fork("topology_refresher"),
)
.await?;
let gateway_packet_router = PacketRouter::new(
ack_sender,
mixnet_messages_sender,
shutdown_tracker.clone_shutdown_token(),
shutdown.get_handle().named("gateway-packet-router"),
);
let gateway_transceiver = Self::setup_gateway_transceiver(
@@ -947,7 +875,7 @@ where
stats_reporter.clone(),
#[cfg(unix)]
self.connection_fd_callback,
&shutdown_tracker.child_tracker(),
shutdown.fork("gateway_transceiver"),
)
.await?;
let gateway_ws_fd = gateway_transceiver.ws_fd();
@@ -955,7 +883,7 @@ where
let reply_storage = Self::setup_persistent_reply_storage(
reply_storage_backend,
key_rotation_config,
&shutdown_tracker.child_tracker(),
shutdown.fork("persistent_reply_storage"),
)
.await?;
@@ -965,8 +893,8 @@ where
mixnet_messages_receiver,
reply_storage.key_storage(),
reply_controller_sender.clone(),
shutdown.fork("received_messages_buffer"),
stats_reporter.clone(),
&shutdown_tracker.child_tracker(),
);
// The message_sender is the transmitter for any component generating sphinx packets
@@ -976,7 +904,7 @@ where
let (message_sender, client_request_sender) = Self::start_mix_traffic_controller(
gateway_transceiver,
&shutdown_tracker.child_tracker(),
shutdown.fork("mix_traffic_controller"),
);
// Channels that the websocket listener can use to signal downstream to the real traffic
@@ -1005,8 +933,9 @@ where
reply_controller_receiver,
shared_lane_queue_lengths.clone(),
client_connection_rx,
shutdown.fork("real_traffic_controller"),
self.config.debug.traffic.packet_type,
stats_reporter.clone(),
&shutdown_tracker.child_tracker(),
);
if !self
@@ -1022,7 +951,7 @@ where
shared_topology_accessor.clone(),
message_sender,
stats_reporter.clone(),
&shutdown_tracker.child_tracker(),
shutdown.fork("cover_traffic_stream"),
);
}
@@ -1050,7 +979,7 @@ where
gateway_connection: GatewayConnection { gateway_ws_fd },
},
stats_reporter,
shutdown_handle: shutdown_tracker, // The primary tracker for this client
task_handle: shutdown,
client_request_sender,
forget_me: self.config.debug.forget_me,
remember_me: self.config.debug.remember_me,
@@ -1066,7 +995,7 @@ pub struct BaseClient {
pub client_state: ClientState,
pub stats_reporter: ClientStatsSender,
pub client_request_sender: ClientRequestSender,
pub shutdown_handle: ShutdownTracker,
pub task_handle: TaskHandle,
pub forget_me: ForgetMe,
pub remember_me: RememberMe,
}
@@ -3,7 +3,7 @@
use crate::client::mix_traffic::BatchMixMessageSender;
use crate::client::topology_control::TopologyAccessor;
use crate::config;
use crate::{config, spawn_future};
use futures::task::{Context, Poll};
use futures::{Future, Stream, StreamExt};
use nym_sphinx::acknowledgements::AckKey;
@@ -12,6 +12,7 @@ use nym_sphinx::cover::generate_loop_cover_packet;
use nym_sphinx::params::{PacketSize, PacketType};
use nym_sphinx::utils::sample_poisson_duration;
use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender};
use nym_task::TaskClient;
use rand::{rngs::OsRng, CryptoRng, Rng};
use std::pin::Pin;
use std::sync::Arc;
@@ -68,6 +69,8 @@ where
packet_type: PacketType,
stats_tx: ClientStatsSender,
task_client: TaskClient,
}
impl<R> Stream for LoopCoverTrafficStream<R>
@@ -114,6 +117,7 @@ impl LoopCoverTrafficStream<OsRng> {
traffic_config: config::Traffic,
cover_config: config::CoverTraffic,
stats_tx: ClientStatsSender,
task_client: TaskClient,
) -> Self {
let rng = OsRng;
@@ -133,6 +137,7 @@ impl LoopCoverTrafficStream<OsRng> {
use_legacy_sphinx_format: traffic_config.use_legacy_sphinx_format,
packet_type: traffic_config.packet_type,
stats_tx,
task_client,
}
}
@@ -230,13 +235,12 @@ impl LoopCoverTrafficStream<OsRng> {
tokio::task::yield_now().await;
}
// it's fine if cover traffic stream task gets killed whilst processing next message
#[allow(clippy::panic)]
pub async fn run(&mut self) {
pub fn start(mut self) {
if self.cover_traffic.disable_loop_cover_traffic_stream {
// we should have never got here in the first place - the task should have never been created to begin with
// so panic and review the code that lead to this branch
panic!("attempted to run LoopCoverTrafficStream while config explicitly disabled it.")
panic!("attempted to start LoopCoverTrafficStream while config explicitly disabled it.")
}
// we should set initial delay only when we actually start the stream
@@ -246,11 +250,32 @@ impl LoopCoverTrafficStream<OsRng> {
);
self.set_next_delay(sampled);
while self.next().await.is_some() {
self.on_new_message().await;
}
let mut shutdown = self.task_client.fork("select");
// this should never get triggered
error!("cover traffic stream has been exhausted!")
spawn_future!(
async move {
debug!("Started LoopCoverTrafficStream with graceful shutdown support");
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
tracing::trace!("LoopCoverTrafficStream: Received shutdown");
}
next = self.next() => {
if next.is_some() {
self.on_new_message().await;
} else {
tracing::trace!("LoopCoverTrafficStream: Stopping since channel closed");
break;
}
}
}
}
shutdown.recv_timeout().await;
tracing::debug!("LoopCoverTrafficStream: Exiting");
},
"LoopCoverTrafficStream"
)
}
}
@@ -2,9 +2,11 @@
// SPDX-License-Identifier: Apache-2.0
use crate::client::mix_traffic::transceiver::GatewayTransceiver;
use crate::error::ClientCoreError;
use crate::spawn_future;
use nym_gateway_requests::ClientRequest;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use tracing::*;
use transceiver::ErasedGatewayError;
@@ -32,13 +34,13 @@ pub struct MixTrafficController {
// in long run `gateway_client` will be moved away from `MixTrafficController` anyway.
consecutive_gateway_failure_count: usize,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl MixTrafficController {
pub fn new<T>(
gateway_transceiver: T,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> (
MixTrafficController,
BatchMixMessageSender,
@@ -58,7 +60,7 @@ impl MixTrafficController {
mix_rx: message_receiver,
client_rx: client_receiver,
consecutive_gateway_failure_count: 0,
shutdown_token,
task_client,
},
message_sender,
client_sender,
@@ -67,7 +69,7 @@ impl MixTrafficController {
pub fn new_dynamic(
gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> (
MixTrafficController,
BatchMixMessageSender,
@@ -82,7 +84,7 @@ impl MixTrafficController {
mix_rx: message_receiver,
client_rx: client_receiver,
consecutive_gateway_failure_count: 0,
shutdown_token,
task_client,
},
message_sender,
client_sender,
@@ -105,7 +107,7 @@ impl MixTrafficController {
tokio::select! {
biased;
_ = self.shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
trace!("received shutdown while handling messages");
Ok(())
}
@@ -125,7 +127,7 @@ impl MixTrafficController {
async fn on_client_request(&mut self, client_request: ClientRequest) {
tokio::select! {
biased;
_ = self.shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
trace!("received shutdown while handling client request");
}
result = self.gateway_transceiver.send_client_request(client_request) => {
@@ -136,44 +138,52 @@ impl MixTrafficController {
}
}
pub async fn run(&mut self) {
debug!("Started MixTrafficController with graceful shutdown support");
loop {
tokio::select! {
biased;
_ = self.shutdown_token.cancelled() => {
trace!("MixTrafficController: Received shutdown");
break;
}
mix_packets = self.mix_rx.recv() => match mix_packets {
Some(mix_packets) => {
if let Err(err) = self.on_messages(mix_packets).await {
error!("Failed to send sphinx packet(s) to the gateway: {err}");
if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT {
// Disconnect from the gateway. If we should try to re-connect
// is handled at a higher layer.
error!("Failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead");
// Do we need to handle the embedded mixnet client case
// separately?
pub fn start(mut self) {
spawn_future!(
async move {
debug!("Started MixTrafficController with graceful shutdown support");
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = self.task_client.recv() => {
tracing::trace!("MixTrafficController: Received shutdown");
break;
}
mix_packets = self.mix_rx.recv() => match mix_packets {
Some(mix_packets) => {
if let Err(err) = self.on_messages(mix_packets).await {
error!("Failed to send sphinx packet(s) to the gateway: {err}");
if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT {
// Disconnect from the gateway. If we should try to re-connect
// is handled at a higher layer.
error!("Failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead");
// Do we need to handle the embedded mixnet client case
// separately?
self.task_client.send_we_stopped(Box::new(ClientCoreError::GatewayFailedToForwardMessages));
break;
}
}
},
None => {
tracing::trace!("MixTrafficController: Stopping since channel closed");
break;
}
}
},
None => {
trace!("MixTrafficController: Stopping since channel closed");
break;
},
client_request = self.client_rx.recv() => match client_request {
Some(client_request) => {
self.on_client_request(client_request).await;
},
None => {
tracing::trace!("MixTrafficController, client request channel closed");
break
}
},
}
},
client_request = self.client_rx.recv() => match client_request {
Some(client_request) => {
self.on_client_request(client_request).await;
},
None => {
trace!("MixTrafficController, client request channel closed");
break}
},
}
}
debug!("MixTrafficController: Exiting");
}
self.task_client.recv_timeout().await;
tracing::debug!("MixTrafficController: Exiting");
},
"MixTrafficController"
);
}
}
@@ -10,17 +10,18 @@ use nym_sphinx::{
acknowledgements::{identifier::recover_identifier, AckKey},
chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID},
};
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use std::sync::Arc;
use tracing::*;
/// Module responsible for listening for any data resembling acknowledgements from the network
/// and firing actions to remove them from the 'Pending' state.
pub(crate) struct AcknowledgementListener {
pub(super) struct AcknowledgementListener {
ack_key: Arc<AckKey>,
ack_receiver: AcknowledgementReceiver,
action_sender: AckActionSender,
stats_tx: ClientStatsSender,
task_client: TaskClient,
}
impl AcknowledgementListener {
@@ -29,12 +30,14 @@ impl AcknowledgementListener {
ack_receiver: AcknowledgementReceiver,
action_sender: AckActionSender,
stats_tx: ClientStatsSender,
task_client: TaskClient,
) -> Self {
AcknowledgementListener {
ack_key,
ack_receiver,
action_sender,
stats_tx,
task_client,
}
}
@@ -65,9 +68,14 @@ impl AcknowledgementListener {
trace!("Received {frag_id} from the mix network");
self.stats_tx
.report(PacketStatisticsEvent::RealAckReceived(ack_content.len()).into());
let _ = self
if let Err(err) = self
.action_sender
.unbounded_send(Action::new_remove(frag_id));
.unbounded_send(Action::new_remove(frag_id))
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to send remove action to action controller: {err}");
}
}
}
async fn handle_ack_receiver_item(&mut self, item: Vec<Vec<u8>>) {
@@ -77,16 +85,11 @@ impl AcknowledgementListener {
}
}
pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
pub(super) async fn run(&mut self) {
debug!("Started AcknowledgementListener with graceful shutdown support");
loop {
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
tracing::trace!("AcknowledgementListener: Received shutdown");
break;
}
acks = self.ack_receiver.next() => match acks {
Some(acks) => self.handle_ack_receiver_item(acks).await,
None => {
@@ -94,9 +97,12 @@ impl AcknowledgementListener {
break;
}
},
_ = self.task_client.recv() => {
tracing::trace!("AcknowledgementListener: Received shutdown");
}
}
}
self.task_client.recv_timeout().await;
tracing::debug!("AcknowledgementListener: Exiting");
}
}
@@ -8,7 +8,7 @@ use futures::StreamExt;
use nym_nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue, QueueKey};
use nym_sphinx::chunking::fragment::FragmentIdentifier;
use nym_sphinx::Delay as SphinxDelay;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
@@ -82,7 +82,7 @@ impl Config {
}
}
pub(crate) struct ActionController {
pub(super) struct ActionController {
/// Configurable parameters of the `ActionController`
config: Config,
@@ -102,6 +102,8 @@ pub(crate) struct ActionController {
/// Channel for notifying `RetransmissionRequestListener` about expired acknowledgements.
retransmission_sender: RetransmissionRequestSender,
task_client: TaskClient,
}
impl ActionController {
@@ -109,6 +111,7 @@ impl ActionController {
config: Config,
retransmission_sender: RetransmissionRequestSender,
incoming_actions: AckActionReceiver,
task_client: TaskClient,
) -> Self {
ActionController {
config,
@@ -116,6 +119,7 @@ impl ActionController {
pending_acks_timers: NonExhaustiveDelayQueue::new(),
incoming_actions,
retransmission_sender,
task_client,
}
}
@@ -222,9 +226,14 @@ impl ActionController {
// downgrading an arc and then upgrading vs cloning is difference of 30ns vs 15ns
// so it's literally a NO difference while it might prevent us from unnecessarily
// resending data (in maybe 1 in 1 million cases, but it's something)
let _ = self
if let Err(err) = self
.retransmission_sender
.unbounded_send(Arc::downgrade(pending_ack_data));
.unbounded_send(Arc::downgrade(pending_ack_data))
{
if !self.task_client.is_shutdown_poll() {
tracing::error!("Failed to send pending ack for retransmission: {err}");
}
}
} else {
// this shouldn't cause any issues but shouldn't have happened to begin with!
error!("An already removed pending ack has expired")
@@ -242,16 +251,11 @@ impl ActionController {
}
}
pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
pub(super) async fn run(&mut self) {
debug!("Started ActionController with graceful shutdown support");
loop {
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
tracing::trace!("ActionController: Received shutdown");
break;
}
action = self.incoming_actions.next() => match action {
Some(action) => self.process_action(action),
None => {
@@ -268,8 +272,13 @@ impl ActionController {
break;
}
},
_ = self.task_client.recv() => {
tracing::trace!("ActionController: Received shutdown");
break;
}
}
}
self.task_client.recv_timeout().await;
tracing::debug!("ActionController: Exiting");
}
}
@@ -10,20 +10,21 @@ use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::params::PacketType;
use nym_task::connections::TransmissionLane;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use rand::{CryptoRng, Rng};
use tracing::*;
/// Module responsible for dealing with the received messages: splitting them, creating acknowledgements,
/// putting everything into sphinx packets, etc.
/// It also makes an initial sending attempt for said messages.
pub(crate) struct InputMessageListener<R>
pub(super) struct InputMessageListener<R>
where
R: CryptoRng + Rng,
{
input_receiver: InputMessageReceiver,
message_handler: MessageHandler<R>,
reply_controller_sender: ReplyControllerSender,
task_client: TaskClient,
}
impl<R> InputMessageListener<R>
@@ -37,11 +38,13 @@ where
input_receiver: InputMessageReceiver,
message_handler: MessageHandler<R>,
reply_controller_sender: ReplyControllerSender,
task_client: TaskClient,
) -> Self {
InputMessageListener {
input_receiver,
message_handler,
reply_controller_sender,
task_client,
}
}
@@ -65,9 +68,14 @@ where
max_retransmissions: Option<u32>,
) {
// offload reply handling to the dedicated task
let _ =
if let Err(err) =
self.reply_controller_sender
.send_reply(recipient_tag, data, lane, max_retransmissions);
.send_reply(recipient_tag, data, lane, max_retransmissions)
{
if !self.task_client.is_shutdown_poll() {
error!("failed to send a reply - {err}");
}
}
}
async fn handle_plain_message(
@@ -213,13 +221,13 @@ where
};
}
pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
pub(super) async fn run(&mut self) {
debug!("Started InputMessageListener with graceful shutdown support");
loop {
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
tracing::trace!("InputMessageListener: Received shutdown");
break;
}
@@ -235,6 +243,7 @@ where
}
}
self.task_client.recv_timeout().await;
tracing::debug!("InputMessageListener: Exiting");
}
}
@@ -10,6 +10,7 @@ use self::{
use crate::client::inbound_messages::InputMessageReceiver;
use crate::client::real_messages_control::message_handler::MessageHandler;
use crate::client::replies::reply_controller::ReplyControllerSender;
use crate::spawn_future;
use action_controller::AckActionReceiver;
use futures::channel::mpsc;
use nym_gateway_client::AcknowledgementReceiver;
@@ -22,11 +23,13 @@ use nym_sphinx::{
Delay as SphinxDelay,
};
use nym_statistics_common::clients::ClientStatsSender;
use nym_task::TaskClient;
use rand::{CryptoRng, Rng};
use std::{
sync::{Arc, Weak},
time::Duration,
};
use tracing::*;
pub(crate) use action_controller::{AckActionSender, Action};
@@ -187,9 +190,6 @@ pub(super) struct Config {
/// Predefined packet size used for the encapsulated messages.
packet_size: PacketSize,
/// Type of packets used for retransmissions
packet_type: PacketType,
}
impl Config {
@@ -197,14 +197,12 @@ impl Config {
maximum_retransmissions: Option<u32>,
ack_wait_addition: Duration,
ack_wait_multiplier: f64,
packet_type: PacketType,
) -> Self {
Config {
maximum_retransmissions,
ack_wait_addition,
ack_wait_multiplier,
packet_size: Default::default(),
packet_type,
}
}
@@ -214,7 +212,7 @@ impl Config {
}
}
pub(crate) struct AcknowledgementController<R>
pub(super) struct AcknowledgementController<R>
where
R: CryptoRng + Rng,
{
@@ -236,6 +234,7 @@ where
message_handler: MessageHandler<R>,
reply_controller_sender: ReplyControllerSender,
stats_tx: ClientStatsSender,
task_client: TaskClient,
) -> Self {
let (retransmission_tx, retransmission_rx) = mpsc::unbounded();
@@ -245,6 +244,7 @@ where
action_config,
retransmission_tx,
connectors.ack_action_receiver,
task_client.fork("action_controller"),
);
// will listen for any acks coming from the network
@@ -253,6 +253,7 @@ where
connectors.ack_receiver,
connectors.ack_action_sender.clone(),
stats_tx,
task_client.fork("acknowledgement_listener"),
);
// will listen for any new messages from the client
@@ -260,6 +261,7 @@ where
connectors.input_receiver,
message_handler.clone(),
reply_controller_sender.clone(),
task_client.fork("input_message_listener"),
);
// will listen for any ack timeouts and trigger retransmission
@@ -269,13 +271,16 @@ where
message_handler,
retransmission_rx,
reply_controller_sender,
config.packet_type,
task_client.fork("retransmission_request_listener"),
);
// will listen for events indicating the packet was sent through the network so that
// the retransmission timer should be started.
let sent_notification_listener =
SentNotificationListener::new(connectors.sent_notifier, connectors.ack_action_sender);
let sent_notification_listener = SentNotificationListener::new(
connectors.sent_notifier,
connectors.ack_action_sender,
task_client.with_suffix("sent_notification_listener"),
);
AcknowledgementController {
acknowledgement_listener,
@@ -286,21 +291,51 @@ where
}
}
pub(crate) fn into_tasks(
self,
) -> (
AcknowledgementListener,
InputMessageListener<R>,
RetransmissionRequestListener<R>,
SentNotificationListener,
ActionController,
) {
(
self.acknowledgement_listener,
self.input_message_listener,
self.retransmission_request_listener,
self.sent_notification_listener,
self.action_controller,
)
pub(super) fn start(self, packet_type: PacketType) {
let mut acknowledgement_listener = self.acknowledgement_listener;
let mut input_message_listener = self.input_message_listener;
let mut retransmission_request_listener = self.retransmission_request_listener;
let mut sent_notification_listener = self.sent_notification_listener;
let mut action_controller = self.action_controller;
spawn_future!(
async move {
acknowledgement_listener.run().await;
debug!("The acknowledgement listener has finished execution!");
},
"AcknowledgementController::AcknowledgementListener"
);
spawn_future!(
async move {
input_message_listener.run().await;
debug!("The input listener has finished execution!");
},
"AcknowledgementController::InputMessageListener"
);
spawn_future!(
async move {
retransmission_request_listener.run(packet_type).await;
debug!("The retransmission request listener has finished execution!");
},
"AcknowledgementController::RetransmissionRequestListener"
);
spawn_future!(
async move {
sent_notification_listener.run().await;
debug!("The sent notification listener has finished execution!");
},
"AcknowledgementController::SentNotificationListener"
);
spawn_future!(
async move {
action_controller.run().await;
debug!("The controller has finished execution!");
},
"AcknowledgementController::ActionController"
);
}
}
@@ -13,19 +13,19 @@ use futures::StreamExt;
use nym_sphinx::chunking::fragment::Fragment;
use nym_sphinx::preparer::PreparedFragment;
use nym_sphinx::{addressing::clients::Recipient, params::PacketType};
use nym_task::{connections::TransmissionLane, ShutdownToken};
use nym_task::{connections::TransmissionLane, TaskClient};
use rand::{CryptoRng, Rng};
use std::sync::{Arc, Weak};
use tracing::*;
// responsible for packet retransmission upon fired timer
pub(crate) struct RetransmissionRequestListener<R> {
pub(super) struct RetransmissionRequestListener<R> {
maximum_retransmissions: Option<u32>,
action_sender: AckActionSender,
message_handler: MessageHandler<R>,
request_receiver: RetransmissionRequestReceiver,
reply_controller_sender: ReplyControllerSender,
packet_type: PacketType,
task_client: TaskClient,
}
impl<R> RetransmissionRequestListener<R>
@@ -38,7 +38,7 @@ where
message_handler: MessageHandler<R>,
request_receiver: RetransmissionRequestReceiver,
reply_controller_sender: ReplyControllerSender,
packet_type: PacketType,
task_client: TaskClient,
) -> Self {
RetransmissionRequestListener {
maximum_retransmissions,
@@ -46,7 +46,7 @@ where
message_handler,
request_receiver,
reply_controller_sender,
packet_type,
task_client,
}
}
@@ -67,6 +67,7 @@ where
async fn on_retransmission_request(
&mut self,
weak_timed_out_ack: Weak<PendingAcknowledgement>,
packet_type: PacketType,
) {
let timed_out_ack = match weak_timed_out_ack.upgrade() {
Some(timed_out_ack) => timed_out_ack,
@@ -96,18 +97,22 @@ where
} => {
// if this is retransmission for reply, offload it to the dedicated task
// that deals with all the surbs
let _ = self.reply_controller_sender.send_retransmission_data(
if let Err(err) = self.reply_controller_sender.send_retransmission_data(
*recipient_tag,
weak_timed_out_ack,
*extra_surb_request,
);
) {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send retransmission data to the reply controller: {err}");
}
}
return;
}
PacketDestination::KnownRecipient(recipient) => {
self.prepare_normal_retransmission_chunk(
**recipient,
timed_out_ack.message_chunk.clone(),
self.packet_type,
packet_type,
)
.await
}
@@ -148,9 +153,14 @@ where
// is sent to the `OutQueueControl` and has gone through its internal queue
// with the additional poisson delay.
// And since Actions are executed in order `UpdateTimer` will HAVE TO be executed before `StartTimer`
let _ = self
if let Err(err) = self
.action_sender
.unbounded_send(Action::new_update_pending_ack(frag_id, new_delay));
.unbounded_send(Action::new_update_pending_ack(frag_id, new_delay))
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to send update pending ack action to the controller: {err}");
}
}
// send to `OutQueueControl` to eventually send to the mix network
self.message_handler
@@ -164,18 +174,18 @@ where
.await
}
pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
pub(super) async fn run(&mut self, packet_type: PacketType) {
debug!("Started RetransmissionRequestListener with graceful shutdown support");
loop {
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
tracing::trace!("RetransmissionRequestListener: Received shutdown");
break;
}
timed_out_ack = self.request_receiver.next() => match timed_out_ack {
Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack).await,
Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack, packet_type).await,
None => {
tracing::trace!("RetransmissionRequestListener: Stopping since channel closed");
break;
@@ -184,6 +194,7 @@ where
}
}
self.task_client.recv_timeout().await;
tracing::debug!("RetransmissionRequestListener: Exiting");
}
}
@@ -5,25 +5,29 @@ use super::action_controller::{AckActionSender, Action};
use super::SentPacketNotificationReceiver;
use futures::StreamExt;
use nym_sphinx::chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID};
use nym_task::TaskClient;
use tracing::*;
/// Module responsible for starting up retransmission timers.
/// It is required because when we send our packet to the `real traffic stream` controlled
/// by a poisson timer, there's no guarantee the message will be sent immediately, so we might
/// accidentally fire retransmission way quicker than we should have.
pub(crate) struct SentNotificationListener {
pub(super) struct SentNotificationListener {
sent_notifier: SentPacketNotificationReceiver,
action_sender: AckActionSender,
task_client: TaskClient,
}
impl SentNotificationListener {
pub(super) fn new(
sent_notifier: SentPacketNotificationReceiver,
action_sender: AckActionSender,
task_client: TaskClient,
) -> Self {
SentNotificationListener {
sent_notifier,
action_sender,
task_client,
}
}
@@ -32,18 +36,37 @@ impl SentNotificationListener {
trace!("sent off a cover message - no need to start retransmission timer!");
return;
}
let _ = self
if let Err(err) = self
.action_sender
.unbounded_send(Action::new_start_timer(frag_id));
.unbounded_send(Action::new_start_timer(frag_id))
{
if !self.task_client.is_shutdown_poll() {
error!("Failed to send start timer action to action controller: {err}");
}
}
}
pub(crate) async fn run(&mut self) {
pub(super) async fn run(&mut self) {
debug!("Started SentNotificationListener with graceful shutdown support");
while let Some(frag_id) = self.sent_notifier.next().await {
self.on_sent_message(frag_id).await;
while !self.task_client.is_shutdown() {
tokio::select! {
frag_id = self.sent_notifier.next() => match frag_id {
Some(frag_id) => {
self.on_sent_message(frag_id).await;
}
None => {
tracing::trace!("SentNotificationListener: Stopping since channel closed");
break;
}
},
_ = self.task_client.recv() => {
tracing::trace!("SentNotificationListener: Received shutdown");
break;
}
}
}
assert!(self.task_client.is_shutdown_poll());
tracing::debug!("SentNotificationListener: Exiting");
}
}
@@ -20,7 +20,7 @@ use nym_sphinx::params::{PacketSize, PacketType};
use nym_sphinx::preparer::{MessagePreparer, PreparedFragment};
use nym_sphinx::Delay;
use nym_task::connections::TransmissionLane;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use nym_topology::{NymRouteProvider, NymTopologyError};
use rand::{CryptoRng, Rng};
use std::collections::HashMap;
@@ -189,7 +189,7 @@ pub(crate) struct MessageHandler<R> {
topology_access: TopologyAccessor,
reply_key_storage: SentReplyKeys,
tag_storage: UsedSenderTags,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl<R> MessageHandler<R>
@@ -205,7 +205,7 @@ where
topology_access: TopologyAccessor,
reply_key_storage: SentReplyKeys,
tag_storage: UsedSenderTags,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self
where
R: Copy,
@@ -228,7 +228,7 @@ where
topology_access,
reply_key_storage,
tag_storage,
shutdown_token,
task_client,
}
}
@@ -712,7 +712,7 @@ where
.action_sender
.unbounded_send(Action::UpdatePendingAck(id, new_delay))
{
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send update action to the controller: {err}");
}
}
@@ -723,7 +723,7 @@ where
.action_sender
.unbounded_send(Action::new_insert(pending_acks))
{
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!("Failed to send insert action to the controller: {err}");
}
}
@@ -737,7 +737,7 @@ where
) {
tokio::select! {
biased;
_ = self.shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
trace!("received shutdown while attempting to forward mixnet messages");
}
sending_res = self.real_message_sender.send((messages, transmission_lane)) => {
@@ -14,21 +14,26 @@ use crate::client::replies::reply_controller::{
ReplyController, ReplyControllerReceiver, ReplyControllerSender,
};
use crate::client::replies::reply_storage::CombinedReplyStorage;
use crate::client::{
inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors,
topology_control::TopologyAccessor,
};
use crate::config;
use crate::{
client::{
inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors,
topology_control::TopologyAccessor,
},
spawn_future,
};
use futures::channel::mpsc;
use nym_gateway_client::AcknowledgementReceiver;
use nym_sphinx::acknowledgements::AckKey;
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::params::PacketType;
use nym_statistics_common::clients::ClientStatsSender;
use nym_task::connections::{ConnectionCommandReceiver, LaneQueueLengths};
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use rand::{rngs::OsRng, CryptoRng, Rng};
use std::sync::Arc;
use tracing::*;
use crate::client::replies::reply_controller::key_rotation_helpers::KeyRotationConfig;
pub(crate) use acknowledgement_control::{AckActionSender, Action};
@@ -64,7 +69,6 @@ impl<'a> From<&'a Config> for acknowledgement_control::Config {
cfg.traffic.maximum_number_of_retransmissions,
cfg.acks.ack_wait_addition,
cfg.acks.ack_wait_multiplier,
cfg.traffic.packet_type,
)
.with_custom_packet_size(cfg.traffic.primary_packet_size)
}
@@ -142,7 +146,7 @@ impl RealMessagesController<OsRng> {
lane_queue_lengths: LaneQueueLengths,
client_connection_rx: ConnectionCommandReceiver,
stats_tx: ClientStatsSender,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self {
let rng = OsRng;
@@ -174,7 +178,7 @@ impl RealMessagesController<OsRng> {
topology_access.clone(),
reply_storage.key_storage(),
reply_storage.tags_storage(),
shutdown_token.clone(),
task_client.fork("message_handler"),
);
let ack_control = AcknowledgementController::new(
@@ -184,6 +188,7 @@ impl RealMessagesController<OsRng> {
message_handler.clone(),
reply_controller_sender,
stats_tx.clone(),
task_client.fork("ack_control"),
);
let reply_control = ReplyController::new(
@@ -191,6 +196,7 @@ impl RealMessagesController<OsRng> {
message_handler,
reply_storage,
reply_controller_receiver,
task_client.fork("reply_controller"),
);
let out_queue_control = OutQueueControl::new(
@@ -203,7 +209,7 @@ impl RealMessagesController<OsRng> {
lane_queue_lengths,
client_connection_rx,
stats_tx,
shutdown_token.clone(),
task_client.with_suffix("out_queue_control"),
);
RealMessagesController {
@@ -213,13 +219,26 @@ impl RealMessagesController<OsRng> {
}
}
pub fn into_tasks(
self,
) -> (
OutQueueControl<OsRng>,
ReplyController<OsRng>,
AcknowledgementController<OsRng>,
) {
(self.out_queue_control, self.reply_control, self.ack_control)
pub fn start(self, packet_type: PacketType) {
let mut out_queue_control = self.out_queue_control;
let ack_control = self.ack_control;
let mut reply_control = self.reply_control;
spawn_future!(
async move {
out_queue_control.run().await;
debug!("The out queue controller has finished execution!");
},
"RealMessagesController::OutQueueControl)"
);
spawn_future!(
async move {
reply_control.run().await;
debug!("The reply controller has finished execution!");
},
"RealMessagesController::ReplyController"
);
ack_control.start(packet_type);
}
}
@@ -21,7 +21,7 @@ use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, C
use nym_task::connections::{
ConnectionCommand, ConnectionCommandReceiver, ConnectionId, LaneQueueLengths, TransmissionLane,
};
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use rand::{CryptoRng, Rng};
use std::pin::Pin;
use std::sync::Arc;
@@ -119,7 +119,7 @@ where
/// Channel used for sending metrics events (specifically `PacketStatistics` events) to the metrics tracker.
stats_tx: ClientStatsSender,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
#[derive(Debug)]
@@ -179,7 +179,7 @@ where
lane_queue_lengths: LaneQueueLengths,
client_connection_rx: ConnectionCommandReceiver,
stats_tx: ClientStatsSender,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self {
OutQueueControl {
config,
@@ -194,7 +194,7 @@ where
client_connection_rx,
lane_queue_lengths,
stats_tx,
shutdown_token,
task_client,
}
}
@@ -282,7 +282,7 @@ where
let sending_res = tokio::select! {
biased;
_ = self.shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
trace!("received shutdown signal while attempting to send mix message");
return
}
@@ -293,7 +293,7 @@ where
match sending_res {
Err(_) => {
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
tracing::error!(
"failed to send mixnet packet due to closed channel (outside of shutdown!)"
);
@@ -536,7 +536,9 @@ where
}
#[cfg(not(target_arch = "wasm32"))]
fn log_status(&self) {
fn log_status(&self, shutdown: &mut TaskClient) {
use crate::error::ClientCoreStatusMessage;
let packets = self.transmission_buffer.total_size();
let lanes = self.transmission_buffer.lanes();
let mult = self.sending_delay_controller.current_multiplier();
@@ -565,33 +567,32 @@ where
tracing::debug!("{status_str}");
}
// leave the code commented in case somebody wanted to restore this logic with a different channel
// // Send status message to whoever is listening (possibly UI)
// if mult == self.sending_delay_controller.max_multiplier() {
// shutdown.send_status_msg(Box::new(ClientCoreStatusMessage::GatewayIsVerySlow));
// } else if mult > self.sending_delay_controller.min_multiplier() {
// shutdown.send_status_msg(Box::new(ClientCoreStatusMessage::GatewayIsSlow));
// }
// Send status message to whoever is listening (possibly UI)
if mult == self.sending_delay_controller.max_multiplier() {
shutdown.send_status_msg(Box::new(ClientCoreStatusMessage::GatewayIsVerySlow));
} else if mult > self.sending_delay_controller.min_multiplier() {
shutdown.send_status_msg(Box::new(ClientCoreStatusMessage::GatewayIsSlow));
}
}
pub(crate) async fn run(&mut self) {
pub(super) async fn run(&mut self) {
debug!("Started OutQueueControl with graceful shutdown support");
// avoid borrow on self
let shutdown_token = self.shutdown_token.clone();
let mut shutdown = self.task_client.fork("select");
#[cfg(not(target_arch = "wasm32"))]
{
let mut status_timer = tokio::time::interval(Duration::from_secs(5));
loop {
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
_ = shutdown.recv() => {
tracing::trace!("OutQueueControl: Received shutdown");
break;
}
_ = status_timer.tick() => {
self.log_status();
self.log_status(&mut shutdown);
}
next_message = self.next() => if let Some(next_message) = next_message {
self.on_message(next_message).await;
@@ -601,16 +602,16 @@ where
}
}
}
shutdown.recv_timeout().await;
}
#[cfg(target_arch = "wasm32")]
{
loop {
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
_ = shutdown.recv() => {
tracing::trace!("OutQueueControl: Received shutdown");
break;
}
next_message = self.next() => if let Some(next_message) = next_message {
self.on_message(next_message).await;
@@ -83,13 +83,11 @@ impl SendingDelayController {
self.current_multiplier
}
#[allow(dead_code)]
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn min_multiplier(&self) -> u32 {
self.lower_bound
}
#[allow(dead_code)]
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn max_multiplier(&self) -> u32 {
self.upper_bound
@@ -5,6 +5,7 @@ use crate::client::helpers::get_time_now;
use crate::client::replies::{
reply_controller::ReplyControllerSender, reply_storage::SentReplyKeys,
};
use crate::spawn_future;
use futures::channel::mpsc;
use futures::lock::Mutex;
use futures::StreamExt;
@@ -19,7 +20,7 @@ use nym_sphinx::message::{NymMessage, PlainMessage};
use nym_sphinx::params::ReplySurbKeyDigestAlgorithm;
use nym_sphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMessage};
use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender};
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
@@ -171,7 +172,7 @@ struct ReceivedMessagesBuffer<R: MessageReceiver> {
inner: Arc<Mutex<ReceivedMessagesBufferInner<R>>>,
reply_key_storage: SentReplyKeys,
reply_controller_sender: ReplyControllerSender,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
@@ -180,7 +181,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
reply_key_storage: SentReplyKeys,
reply_controller_sender: ReplyControllerSender,
stats_tx: ClientStatsSender,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self {
ReceivedMessagesBuffer {
inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner {
@@ -194,7 +195,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
})),
reply_key_storage,
reply_controller_sender,
shutdown_token,
task_client,
}
}
@@ -315,7 +316,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
reply_surbs,
from_surb_request,
) {
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!("{err}");
}
}
@@ -338,7 +339,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
.reply_controller_sender
.send_additional_surbs_request(*recipient, amount)
{
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
error!("{err}");
}
}
@@ -465,22 +466,22 @@ pub enum ReceivedBufferMessage {
ReceiverDisconnect,
}
pub(crate) struct RequestReceiver<R: MessageReceiver> {
struct RequestReceiver<R: MessageReceiver> {
received_buffer: ReceivedMessagesBuffer<R>,
query_receiver: ReceivedBufferRequestReceiver,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl<R: MessageReceiver> RequestReceiver<R> {
fn new(
received_buffer: ReceivedMessagesBuffer<R>,
query_receiver: ReceivedBufferRequestReceiver,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self {
RequestReceiver {
received_buffer,
query_receiver,
shutdown_token,
task_client,
}
}
@@ -495,70 +496,66 @@ impl<R: MessageReceiver> RequestReceiver<R> {
}
}
pub(crate) async fn run(&mut self) {
async fn run(&mut self) {
debug!("Started RequestReceiver with graceful shutdown support");
loop {
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = self.shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
tracing::trace!("RequestReceiver: Received shutdown");
break;
}
request = self.query_receiver.next() => {
if let Some(message) = request {
self.handle_message(message).await
} else {
tracing::trace!("RequestReceiver: Stopping since channel closed");
self.shutdown_token.cancelled().await;
break;
}
},
}
}
self.task_client.recv().await;
tracing::debug!("RequestReceiver: Exiting");
}
}
pub(crate) struct FragmentedMessageReceiver<R: MessageReceiver> {
struct FragmentedMessageReceiver<R: MessageReceiver> {
received_buffer: ReceivedMessagesBuffer<R>,
mixnet_packet_receiver: MixnetMessageReceiver,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl<R: MessageReceiver> FragmentedMessageReceiver<R> {
fn new(
received_buffer: ReceivedMessagesBuffer<R>,
mixnet_packet_receiver: MixnetMessageReceiver,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self {
FragmentedMessageReceiver {
received_buffer,
mixnet_packet_receiver,
shutdown_token,
task_client,
}
}
pub(crate) async fn run(&mut self) -> Result<(), MessageRecoveryError> {
async fn run(&mut self) -> Result<(), MessageRecoveryError> {
debug!("Started FragmentedMessageReceiver with graceful shutdown support");
loop {
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = self.shutdown_token.cancelled() => {
tracing::trace!("FragmentedMessageReceiver: Received shutdown");
break;
}
new_messages = self.mixnet_packet_receiver.next() => {
if let Some(new_messages) = new_messages {
self.received_buffer.handle_new_received(new_messages).await?;
} else {
tracing::trace!("FragmentedMessageReceiver: Stopping since channel closed");
self.shutdown_token.cancelled().await;
break;
}
},
_ = self.task_client.recv_with_delay() => {
tracing::trace!("FragmentedMessageReceiver: Received shutdown");
}
}
}
self.task_client.recv_timeout().await;
tracing::debug!("FragmentedMessageReceiver: Exiting");
Ok(())
}
@@ -577,31 +574,48 @@ impl<R: MessageReceiver + Clone + Send + 'static> ReceivedMessagesBufferControll
reply_key_storage: SentReplyKeys,
reply_controller_sender: ReplyControllerSender,
metrics_reporter: ClientStatsSender,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self {
let received_buffer = ReceivedMessagesBuffer::new(
local_encryption_keypair,
reply_key_storage,
reply_controller_sender,
metrics_reporter,
shutdown_token.clone(),
task_client.fork("received_messages_buffer"),
);
ReceivedMessagesBufferController {
fragmented_message_receiver: FragmentedMessageReceiver::new(
received_buffer.clone(),
mixnet_packet_receiver,
shutdown_token.clone(),
task_client.fork("fragmented_message_receiver"),
),
request_receiver: RequestReceiver::new(
received_buffer,
query_receiver,
shutdown_token.clone(),
task_client.with_suffix("request_receiver"),
),
}
}
pub(crate) fn into_tasks(self) -> (FragmentedMessageReceiver<R>, RequestReceiver<R>) {
(self.fragmented_message_receiver, self.request_receiver)
pub fn start(self) {
let mut fragmented_message_receiver = self.fragmented_message_receiver;
let mut request_receiver = self.request_receiver;
spawn_future!(
async move {
match fragmented_message_receiver.run().await {
Ok(_) => {}
Err(e) => error!("{e}"),
}
},
"ReceivedMessagesBufferController::FragmentedMessageReceiver"
);
spawn_future!(
async move {
request_receiver.run().await;
},
"ReceivedMessagesBufferController::RequestReceiver"
);
}
}
@@ -7,7 +7,7 @@ use crate::client::replies::reply_controller::key_rotation_helpers::KeyRotationC
use crate::client::replies::reply_storage::CombinedReplyStorage;
use crate::config;
use futures::StreamExt;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use rand::rngs::OsRng;
use rand::{CryptoRng, Rng};
use std::time::Duration;
@@ -60,6 +60,9 @@ pub struct ReplyController<R> {
receiver_controller: ReceiverReplyController<R>,
request_receiver: ReplyControllerReceiver,
// Listen for shutdown signals
task_client: TaskClient,
}
impl ReplyController<OsRng> {
@@ -68,6 +71,7 @@ impl ReplyController<OsRng> {
message_handler: MessageHandler<OsRng>,
full_reply_storage: CombinedReplyStorage,
request_receiver: ReplyControllerReceiver,
task_client: TaskClient,
) -> Self {
ReplyController {
config,
@@ -82,6 +86,7 @@ impl ReplyController<OsRng> {
message_handler,
),
request_receiver,
task_client,
}
}
}
@@ -143,21 +148,22 @@ where
self.sender_controller.inspect_and_clear_stale_data(now)
}
pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
pub(crate) async fn run(&mut self) {
debug!("Started ReplyController with graceful shutdown support");
let mut shutdown = self.task_client.fork("reply-controller");
let polling_rate = Duration::from_secs(5);
let mut stale_inspection = new_interval_stream(polling_rate);
let polling_rate = self.config.key_rotation.epoch_duration / 8;
let mut invalidation_inspection = new_interval_stream(polling_rate);
loop {
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
_ = shutdown.recv() => {
tracing::trace!("ReplyController: Received shutdown");
break;
},
req = self.request_receiver.next() => match req {
Some(req) => self.handle_request(req).await,
@@ -175,6 +181,7 @@ where
}
}
}
assert!(shutdown.is_shutdown_poll());
tracing::debug!("ReplyController: Exiting");
}
}
@@ -16,17 +16,21 @@
#![warn(clippy::todo)]
#![warn(clippy::dbg_macro)]
use crate::client::inbound_messages::{InputMessage, InputMessageSender};
use futures::StreamExt;
use nym_client_core_config_types::StatsReporting;
use nym_sphinx::addressing::Recipient;
use nym_statistics_common::clients::{
ClientStatsController, ClientStatsReceiver, ClientStatsSender,
};
use nym_task::{connections::TransmissionLane, ShutdownToken, ShutdownTracker};
use nym_task::{connections::TransmissionLane, TaskClient};
use std::time::Duration;
/// Time interval between reporting statistics locally (logging/shutdown_token)
use crate::{
client::inbound_messages::{InputMessage, InputMessageSender},
spawn_future,
};
/// Time interval between reporting statistics locally (logging/task_client)
const LOCAL_REPORT_INTERVAL: Duration = Duration::from_secs(2);
/// Interval for taking snapshots of the statistics
const SNAPSHOT_INTERVAL: Duration = Duration::from_millis(500);
@@ -47,6 +51,9 @@ pub(crate) struct StatisticsControl {
/// Config for stats reporting (enabled, address, interval)
reporting_config: StatsReporting,
/// Task client for listening for shutdown
task_client: TaskClient,
}
impl StatisticsControl {
@@ -55,20 +62,24 @@ impl StatisticsControl {
client_type: String,
client_stats_id: String,
report_tx: InputMessageSender,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> (Self, ClientStatsSender) {
let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel();
let stats = ClientStatsController::new(client_stats_id, client_type);
let mut task_client_stats_sender = task_client.fork("stats_sender");
task_client_stats_sender.disarm();
(
StatisticsControl {
stats,
stats_rx,
report_tx,
reporting_config,
task_client,
},
ClientStatsSender::new(Some(stats_tx), shutdown_token),
ClientStatsSender::new(Some(stats_tx), task_client_stats_sender),
)
}
@@ -88,8 +99,7 @@ impl StatisticsControl {
}
}
// manually control the shutdown mechanism as we don't want to get interrupted mid-snapshot
pub async fn run(&mut self, shutdown_token: ShutdownToken) {
async fn run(&mut self) {
tracing::debug!("Started StatisticsControl with graceful shutdown support");
#[cfg(not(target_arch = "wasm32"))]
@@ -119,10 +129,10 @@ impl StatisticsControl {
let mut snapshot_interval =
gloo_timers::future::IntervalStream::new(SNAPSHOT_INTERVAL.as_millis() as u32);
loop {
while !self.task_client.is_shutdown() {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
tracing::trace!("StatisticsControl: Received shutdown");
break;
},
@@ -147,34 +157,37 @@ impl StatisticsControl {
}
_ = local_report_interval.next() => {
self.stats.local_report();
self.stats.local_report(&mut self.task_client);
}
}
}
tracing::debug!("StatisticsControl: Exiting");
}
pub(crate) fn start(mut self) {
spawn_future!(
async move {
self.run().await;
},
"StatisticsControl"
)
}
pub(crate) fn create_and_start(
reporting_config: StatsReporting,
client_type: String,
client_stats_id: String,
report_tx: InputMessageSender,
shutdown_tracker: &ShutdownTracker,
task_client: TaskClient,
) -> ClientStatsSender {
let (mut controller, sender) = Self::create(
let (controller, sender) = Self::create(
reporting_config,
client_type,
client_stats_id,
report_tx,
shutdown_tracker.child_shutdown_token(),
);
let shutdown_token = shutdown_tracker.clone_shutdown_token();
shutdown_tracker.try_spawn_named(
async move {
controller.run(shutdown_token).await;
},
"StatisticsControl",
task_client,
);
controller.start();
sender
}
}
@@ -1,9 +1,11 @@
// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::spawn_future;
pub(crate) use accessor::{TopologyAccessor, TopologyReadPermit};
use futures::StreamExt;
use nym_sphinx::addressing::nodes::NodeIdentity;
use nym_task::TaskClient;
use nym_topology::NymTopologyError;
use std::time::Duration;
use tracing::*;
@@ -39,6 +41,8 @@ pub struct TopologyRefresher {
refresh_rate: Duration,
consecutive_failure_count: usize,
task_client: TaskClient,
}
impl TopologyRefresher {
@@ -46,12 +50,14 @@ impl TopologyRefresher {
cfg: TopologyRefresherConfig,
topology_accessor: TopologyAccessor,
topology_provider: Box<dyn TopologyProvider + Send + Sync>,
task_client: TaskClient,
) -> Self {
TopologyRefresher {
topology_provider,
topology_accessor,
refresh_rate: cfg.refresh_rate,
consecutive_failure_count: 0,
task_client,
}
}
@@ -138,30 +144,40 @@ impl TopologyRefresher {
}
}
// it's perfectly fine if task is interrupted mid-refresh
// there's no data to persist or send over
pub async fn run(&mut self) {
debug!("Started TopologyRefresher with graceful shutdown support");
pub fn start(mut self) {
spawn_future!(
async move {
debug!("Started TopologyRefresher with graceful shutdown support");
#[cfg(not(target_arch = "wasm32"))]
let mut interval =
tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(self.refresh_rate));
#[cfg(not(target_arch = "wasm32"))]
let mut interval = tokio_stream::wrappers::IntervalStream::new(
tokio::time::interval(self.refresh_rate),
);
#[cfg(target_arch = "wasm32")]
let mut interval =
gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
#[cfg(target_arch = "wasm32")]
let mut interval =
gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
// We already have an initial topology, so no need to refresh it immediately.
// My understanding is that js setInterval does not fire immediately, so it's not
// needed there.
#[cfg(not(target_arch = "wasm32"))]
interval.next().await;
// We already have an initial topology, so no need to refresh it immediately.
// My understanding is that js setInterval does not fire immediately, so it's not
// needed there.
#[cfg(not(target_arch = "wasm32"))]
interval.next().await;
while interval.next().await.is_some() {
self.try_refresh().await;
}
// this should never get triggered
error!("topology refresher interval has been exhausted!")
while !self.task_client.is_shutdown() {
tokio::select! {
_ = interval.next() => {
self.try_refresh().await;
},
_ = self.task_client.recv() => {
tracing::trace!("TopologyRefresher: Received shutdown");
},
}
}
self.task_client.recv_timeout().await;
tracing::debug!("TopologyRefresher: Exiting");
},
"TopologyRefresher"
)
}
}
@@ -2,10 +2,8 @@
// SPDX-License-Identifier: Apache-2.0
use async_trait::async_trait;
use nym_mixnet_contract_common::EpochRewardedSet;
use nym_topology::provider_trait::{ToTopologyMetadata, TopologyProvider};
use nym_topology::NymTopology;
use nym_validator_client::nym_api::NymApiClientExt;
use rand::prelude::SliceRandom;
use rand::thread_rng;
use std::cmp::min;
@@ -41,43 +39,30 @@ impl Config {
pub struct NymApiTopologyProvider {
config: Config,
validator_client: nym_http_api_client::Client,
validator_client: nym_validator_client::client::NymApiClient,
nym_api_urls: Vec<Url>,
currently_used_api: usize,
use_bincode: bool,
}
impl NymApiTopologyProvider {
pub fn new(
config: impl Into<Config>,
mut nym_api_urls: Vec<Url>,
validator_client: nym_http_api_client::Client,
mut validator_client: nym_validator_client::client::NymApiClient,
) -> Self {
nym_api_urls.shuffle(&mut thread_rng());
let mut provider = NymApiTopologyProvider {
validator_client.change_nym_api(nym_api_urls[0].clone());
NymApiTopologyProvider {
config: config.into(),
validator_client,
nym_api_urls,
currently_used_api: 0,
use_bincode: true,
};
// Set all API URLs - the client will try them in order with automatic failover
provider.validator_client.change_base_urls(
provider
.nym_api_urls
.iter()
.map(|u| u.clone().into())
.collect(),
);
provider
}
}
pub fn disable_bincode(&mut self) {
self.use_bincode = false;
// Note: The unified client doesn't support toggling bincode after creation.
// This would require recreating the client without bincode.
// For now, we'll track the preference but it won't take effect.
warn!("Disabling bincode on existing client is not currently supported");
self.validator_client.use_bincode = false;
}
fn use_next_nym_api(&mut self) {
@@ -87,19 +72,8 @@ impl NymApiTopologyProvider {
}
self.currently_used_api = (self.currently_used_api + 1) % self.nym_api_urls.len();
// Provide all URLs starting from the next one in rotation order
// This enables automatic failover to other endpoints
let rotated_urls: Vec<_> = self
.nym_api_urls
.iter()
.cycle()
.skip(self.currently_used_api)
.take(self.nym_api_urls.len())
.map(|u| u.clone().into())
.collect();
self.validator_client.change_base_urls(rotated_urls)
self.validator_client
.change_nym_api(self.nym_api_urls[self.currently_used_api].clone())
}
async fn get_current_compatible_topology(&mut self) -> Option<NymTopology> {
@@ -125,13 +99,8 @@ impl NymApiTopologyProvider {
.filter(|n| n.performance.round_to_integer() >= self.config.min_node_performance())
.collect::<Vec<_>>();
let epoch_rewarded_set: EpochRewardedSet = rewarded_set.into();
NymTopology::new(
metadata.to_topology_metadata(),
epoch_rewarded_set,
Vec::new(),
)
.with_skimmed_nodes(&nodes_filtered)
NymTopology::new(metadata.to_topology_metadata(), rewarded_set, Vec::new())
.with_skimmed_nodes(&nodes_filtered)
} else {
// if we're not using extended topology, we're only getting active set mixnodes and gateways
@@ -179,13 +148,8 @@ impl NymApiTopologyProvider {
}
}
let epoch_rewarded_set: EpochRewardedSet = rewarded_set.into();
NymTopology::new(
metadata.to_topology_metadata(),
epoch_rewarded_set,
Vec::new(),
)
.with_skimmed_nodes(&nodes)
NymTopology::new(metadata.to_topology_metadata(), rewarded_set, Vec::new())
.with_skimmed_nodes(&nodes)
};
if !topology.is_minimally_routable() {
+4 -13
View File
@@ -4,7 +4,6 @@
use crate::client::mix_traffic::transceiver::ErasedGatewayError;
use nym_crypto::asymmetric::ed25519::Ed25519RecoveryError;
use nym_gateway_client::error::GatewayClientError;
use nym_task::RegistryAccessError;
use nym_topology::node::RoutingNodeError;
use nym_topology::{NodeId, NymTopologyError};
use nym_validator_client::nym_api::error::NymAPIError;
@@ -57,7 +56,10 @@ pub enum ClientCoreError {
ListOfNymApisIsEmpty,
#[error("failed to resolve a query to nym API: {source}")]
NymApiQueryFailure { source: Box<NymAPIError> },
NymApiQueryFailure {
#[from]
source: NymAPIError,
},
#[error(
"the current network topology seem to be insufficient to route any packets through:\n\t{0}"
@@ -243,9 +245,6 @@ pub enum ClientCoreError {
#[error("failed to select valid gateway due to incomputable latency")]
GatewaySelectionFailure { source: WeightedError },
#[error("Could not access task registry, {0}")]
RegistryAccess(#[from] RegistryAccessError),
}
impl From<tungstenite::Error> for ClientCoreError {
@@ -256,14 +255,6 @@ impl From<tungstenite::Error> for ClientCoreError {
}
}
impl From<NymAPIError> for ClientCoreError {
fn from(err: NymAPIError) -> ClientCoreError {
ClientCoreError::NymApiQueryFailure {
source: Box::new(err),
}
}
}
/// Set of messages that the client can send to listeners via the task manager
#[derive(Debug)]
pub enum ClientCoreStatusMessage {
+8 -63
View File
@@ -7,8 +7,7 @@ use futures::{SinkExt, StreamExt};
use nym_crypto::asymmetric::ed25519;
use nym_gateway_client::GatewayClient;
use nym_topology::node::RoutingNode;
use nym_validator_client::client::{IdentityKeyRef, NymApiClientExt};
use nym_validator_client::nym_nodes::SkimmedNodesWithMetadata;
use nym_validator_client::client::IdentityKeyRef;
use nym_validator_client::UserAgent;
use rand::{seq::SliceRandom, Rng};
#[cfg(unix)]
@@ -84,48 +83,6 @@ struct GatewayWithLatency<'a, G: ConnectableGateway> {
latency: Duration,
}
// Helper to collect all pages of entry nodes - replicates NymApiClient's convenience method
async fn get_all_basic_entry_nodes_with_metadata(
client: &nym_http_api_client::Client,
use_bincode: bool,
) -> Result<SkimmedNodesWithMetadata, ClientCoreError> {
// Get first page to obtain metadata
let mut page = 0;
let res = client
.get_basic_entry_assigned_nodes_v2(false, Some(page), None, use_bincode)
.await?;
let mut nodes = res.nodes.data;
let metadata = res.metadata;
if res.nodes.pagination.total == nodes.len() {
return Ok(SkimmedNodesWithMetadata::new(nodes, metadata));
}
page += 1;
// Collect remaining pages
loop {
let mut res = client
.get_basic_entry_assigned_nodes_v2(false, Some(page), None, use_bincode)
.await?;
if !metadata.consistency_check(&res.metadata) {
return Err(ClientCoreError::ValidatorClientError(
nym_validator_client::ValidatorClientError::InconsistentPagedMetadata,
));
}
nodes.append(&mut res.nodes.data);
if nodes.len() < res.nodes.pagination.total {
page += 1
} else {
break;
}
}
Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
}
impl<'a, G: ConnectableGateway> GatewayWithLatency<'a, G> {
fn new(gateway: &'a G, latency: Duration) -> Self {
GatewayWithLatency { gateway, latency }
@@ -142,28 +99,16 @@ pub async fn gateways_for_init<R: Rng>(
let nym_api = nym_apis
.choose(rng)
.ok_or(ClientCoreError::ListOfNymApisIsEmpty)?;
// Use the unified HTTP client directly with optional user agent
let mut builder = nym_http_api_client::Client::builder(nym_api.clone())
.map_err(|e| {
ClientCoreError::ValidatorClientError(nym_validator_client::ValidatorClientError::from(
e,
))
})?
.with_bincode(); // Use bincode for better performance
if let Some(user_agent) = user_agent {
builder = builder.with_user_agent(user_agent);
}
let client = builder.build().map_err(|e| {
ClientCoreError::ValidatorClientError(nym_validator_client::ValidatorClientError::from(e))
})?;
let client = if let Some(user_agent) = user_agent {
nym_validator_client::client::NymApiClient::new_with_user_agent(nym_api.clone(), user_agent)
} else {
nym_validator_client::client::NymApiClient::new(nym_api.clone())
};
tracing::debug!("Fetching list of gateways from: {nym_api}");
// Use our helper to handle pagination
let gateways = get_all_basic_entry_nodes_with_metadata(&client, true)
let gateways = client
.get_all_basic_entry_assigned_nodes_with_metadata()
.await?
.nodes;
info!("nym api reports {} gateways", gateways.len());
+35 -3
View File
@@ -17,9 +17,7 @@ pub use nym_topology::{
HardcodedTopologyProvider, NymRouteProvider, NymTopology, NymTopologyError, TopologyProvider,
};
#[deprecated(note = "use spawn_future from nym_task crate instead")]
#[cfg(target_arch = "wasm32")]
#[track_caller]
pub fn spawn_future<F>(future: F)
where
F: Future<Output = ()> + 'static,
@@ -27,7 +25,9 @@ where
wasm_bindgen_futures::spawn_local(future);
}
#[deprecated(note = "use spawn_future from nym_task crate instead")]
// TODO: expose similar API to the rest of the codebase,
// perhaps with some simple trait for a task to define its name
#[cfg(not(target_arch = "wasm32"))]
#[track_caller]
pub fn spawn_future<F>(future: F)
@@ -37,3 +37,35 @@ where
{
tokio::spawn(future);
}
#[cfg(not(target_arch = "wasm32"))]
#[track_caller]
pub fn spawn_named_future<F>(future: F, name: &str)
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
cfg_if::cfg_if! {if #[cfg(tokio_unstable)] {
#[allow(clippy::expect_used)]
tokio::task::Builder::new().name(name).spawn(future).expect("failed to spawn future");
} else {
let _ = name;
tracing::debug!(r#"the underlying binary hasn't been built with `RUSTFLAGS="--cfg tokio_unstable"` - the future naming won't do anything"#);
spawn_future(future);
}}
}
#[macro_export]
macro_rules! spawn_future {
($future:expr) => {{
$crate::spawn_future($future)
}};
($future:expr, $name:expr) => {{
cfg_if::cfg_if! {if #[cfg(not(target_arch = "wasm32"))] {
$crate::spawn_named_future($future, $name)
} else {
let _ = $name;
$crate::spawn_future($future)
}}
}};
}
+2 -2
View File
@@ -40,7 +40,7 @@ where
pub async fn flush_on_shutdown(
mut self,
mem_state: CombinedReplyStorage,
shutdown: nym_task::ShutdownToken,
mut shutdown: nym_task::TaskClient,
) {
use tracing::{debug, error, info};
@@ -50,7 +50,7 @@ where
return;
}
shutdown.cancelled().await;
shutdown.recv().await;
info!("PersistentReplyStorage is flushing all reply-related data to underlying storage");
if let Err(err) = self.backend.flush_surb_storage(&mem_state).await {
@@ -12,7 +12,7 @@ use crate::socket_state::{ws_fd, PartiallyDelegatedHandle, SocketState};
use crate::traits::GatewayPacketRouter;
use crate::{cleanup_socket_message, try_decrypt_binary_message};
use futures::{SinkExt, StreamExt};
use nym_bandwidth_controller::BandwidthController;
use nym_bandwidth_controller::{BandwidthController, BandwidthStatusMessage};
use nym_credential_storage::ephemeral_storage::EphemeralStorage as EphemeralCredentialStorage;
use nym_credential_storage::storage::Storage as CredentialStorage;
use nym_credentials::CredentialSpendingData;
@@ -27,7 +27,7 @@ use nym_gateway_requests::{
use nym_sphinx::forwarding::packet::MixPacket;
use nym_statistics_common::clients::connection::ConnectionStatsEvent;
use nym_statistics_common::clients::ClientStatsSender;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use nym_validator_client::nyxd::contract_traits::DkgQueryClient;
use rand::rngs::OsRng;
use std::sync::Arc;
@@ -109,7 +109,7 @@ pub struct GatewayClient<C, St = EphemeralCredentialStorage> {
connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
/// Listen to shutdown messages and send notifications back to the task manager
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl<C, St> GatewayClient<C, St> {
@@ -124,7 +124,7 @@ impl<C, St> GatewayClient<C, St> {
bandwidth_controller: Option<BandwidthController<C, St>>,
stats_reporter: ClientStatsSender,
#[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self {
GatewayClient {
cfg,
@@ -141,7 +141,7 @@ impl<C, St> GatewayClient<C, St> {
negotiated_protocol: None,
#[cfg(unix)]
connection_fd_callback,
shutdown_token,
task_client,
}
}
@@ -293,7 +293,7 @@ impl<C, St> GatewayClient<C, St> {
loop {
tokio::select! {
_ = self.shutdown_token.cancelled() => {
_ = self.task_client.recv() => {
log::trace!("GatewayClient control response: Received shutdown");
log::debug!("GatewayClient control response: Exiting");
break Err(GatewayClientError::ConnectionClosedGatewayShutdown);
@@ -514,7 +514,7 @@ impl<C, St> GatewayClient<C, St> {
self.cfg.bandwidth.require_tickets,
derive_aes256_gcm_siv_key,
#[cfg(not(target_arch = "wasm32"))]
self.shutdown_token.clone(),
self.task_client.clone(),
)
.await
.map_err(GatewayClientError::RegistrationFailure),
@@ -631,6 +631,9 @@ impl<C, St> GatewayClient<C, St> {
self.negotiated_protocol = protocol_version;
log::debug!("authenticated: {status}, bandwidth remaining: {bandwidth_remaining}");
self.task_client.send_status_msg(Box::new(
BandwidthStatusMessage::RemainingBandwidth(bandwidth_remaining),
));
Ok(())
}
ServerResponse::Error { message } => Err(GatewayClientError::GatewayError(message)),
@@ -1066,7 +1069,7 @@ impl<C, St> GatewayClient<C, St> {
.expect("no shared key present even though we're authenticated!"),
),
self.bandwidth.clone(),
self.shutdown_token.clone(),
self.task_client.clone(),
)
}
_ => unreachable!(),
@@ -1140,8 +1143,8 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
// perfectly fine here, because it's not meant to be used
let (ack_tx, _) = mpsc::unbounded();
let (mix_tx, _) = mpsc::unbounded();
let shutdown_token = ShutdownToken::default();
let packet_router = PacketRouter::new(ack_tx, mix_tx, shutdown_token.clone());
let task_client = TaskClient::dummy();
let packet_router = PacketRouter::new(ack_tx, mix_tx, task_client.clone());
GatewayClient {
cfg: GatewayClientConfig::default().with_disabled_credentials_mode(true),
@@ -1154,11 +1157,11 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
connection: SocketState::NotConnected,
packet_router,
bandwidth_controller: None,
stats_reporter: ClientStatsSender::new(None, shutdown_token.clone()),
stats_reporter: ClientStatsSender::new(None, task_client.clone()),
negotiated_protocol: None,
#[cfg(unix)]
connection_fd_callback,
shutdown_token,
task_client,
}
}
@@ -1167,7 +1170,7 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
packet_router: PacketRouter,
bandwidth_controller: Option<BandwidthController<C, St>>,
stats_reporter: ClientStatsSender,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> GatewayClient<C, St> {
// invariants that can't be broken
// (unless somebody decided to expose some field that wasn't meant to be exposed)
@@ -1190,7 +1193,7 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
negotiated_protocol: self.negotiated_protocol,
#[cfg(unix)]
connection_fd_callback: self.connection_fd_callback,
shutdown_token,
task_client,
}
}
}
@@ -7,7 +7,7 @@
use crate::error::GatewayClientError;
use crate::GatewayPacketRouter;
use futures::channel::mpsc;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
pub type MixnetMessageSender = mpsc::UnboundedSender<Vec<Vec<u8>>>;
pub type MixnetMessageReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
@@ -19,14 +19,14 @@ pub type AcknowledgementReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
pub struct PacketRouter {
ack_sender: AcknowledgementSender,
mixnet_message_sender: MixnetMessageSender,
shutdown: ShutdownToken,
shutdown: TaskClient,
}
impl PacketRouter {
pub fn new(
ack_sender: AcknowledgementSender,
mixnet_message_sender: MixnetMessageSender,
shutdown: ShutdownToken,
shutdown: TaskClient,
) -> Self {
PacketRouter {
ack_sender,
@@ -42,7 +42,7 @@ impl PacketRouter {
if let Err(err) = self.mixnet_message_sender.unbounded_send(received_messages) {
// check if the failure is due to the shutdown being in progress and thus the receiver channel
// having already been dropped
if self.shutdown.is_cancelled() {
if self.shutdown.is_shutdown_poll() || self.shutdown.is_dummy() {
// This should ideally not happen, but it's ok
tracing::warn!("Failed to send mixnet messages due to receiver task shutdown");
return Err(GatewayClientError::ShutdownInProgress);
@@ -58,7 +58,7 @@ impl PacketRouter {
if let Err(err) = self.ack_sender.unbounded_send(received_acks) {
// check if the failure is due to the shutdown being in progress and thus the receiver channel
// having already been dropped
if self.shutdown.is_cancelled() {
if self.shutdown.is_shutdown_poll() || self.shutdown.is_dummy() {
// This should ideally not happen, but it's ok
tracing::warn!("Failed to send acks due to receiver task shutdown");
return Err(GatewayClientError::ShutdownInProgress);
@@ -69,6 +69,10 @@ impl PacketRouter {
}
Ok(())
}
pub fn disarm(&mut self) {
self.shutdown.disarm();
}
}
impl GatewayPacketRouter for PacketRouter {
@@ -11,7 +11,7 @@ use futures::stream::{SplitSink, SplitStream};
use futures::{SinkExt, StreamExt};
use nym_gateway_requests::shared_key::SharedGatewayKey;
use nym_gateway_requests::{SensitiveServerResponse, ServerResponse, SimpleGatewayRequestsError};
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use si_scale::helpers::bibytes2;
use std::os::raw::c_int as RawFd;
use std::sync::Arc;
@@ -87,13 +87,13 @@ impl PartiallyDelegatedRouter {
}
}
async fn run(mut self, mut split_stream: SplitStream<WsConn>, shutdown_token: ShutdownToken) {
async fn run(mut self, mut split_stream: SplitStream<WsConn>, mut task_client: TaskClient) {
let mut chunked_stream = (&mut split_stream).ready_chunks(8);
let ret: Result<_, GatewayClientError> = loop {
tokio::select! {
biased;
// received system-wide shutdown
_ = shutdown_token.cancelled() => {
_ = task_client.recv() => {
log::trace!("GatewayClient listener: Received shutdown");
log::debug!("GatewayClient listener: Exiting");
return;
@@ -118,7 +118,11 @@ impl PartiallyDelegatedRouter {
let return_res = match ret {
Err(err) => self.stream_return.send(Err(err)),
Ok(_) => self.stream_return.send(Ok(split_stream)),
Ok(_) => {
self.packet_router.disarm();
task_client.disarm();
self.stream_return.send(Ok(split_stream))
}
};
if return_res.is_err() {
@@ -262,8 +266,8 @@ impl PartiallyDelegatedRouter {
Ok(plaintexts)
}
fn spawn(self, split_stream: SplitStream<WsConn>, shutdown_token: ShutdownToken) {
let fut = async move { self.run(split_stream, shutdown_token).await };
fn spawn(self, split_stream: SplitStream<WsConn>, task_client: TaskClient) {
let fut = async move { self.run(split_stream, task_client).await };
#[cfg(target_arch = "wasm32")]
wasm_bindgen_futures::spawn_local(fut);
@@ -279,7 +283,7 @@ impl PartiallyDelegatedHandle {
packet_router: PacketRouter,
shared_key: Arc<SharedGatewayKey>,
client_bandwidth: ClientBandwidth,
shutdown: ShutdownToken,
shutdown: TaskClient,
) -> Self {
// when called for, it NEEDS TO yield back the stream so that we could merge it and
// read control request responses.
+145 -16
View File
@@ -5,8 +5,8 @@ use crate::nyxd::{self, NyxdClient};
use crate::signing::direct_wallet::DirectSecp256k1HdWallet;
use crate::signing::signer::{NoSigner, OfflineSigner};
use crate::{
DirectSigningReqwestRpcValidatorClient, QueryReqwestRpcValidatorClient, ReqwestRpcClient,
ValidatorClientError,
nym_api, DirectSigningReqwestRpcValidatorClient, QueryReqwestRpcValidatorClient,
ReqwestRpcClient, ValidatorClientError,
};
use nym_api_requests::ecash::models::{
AggregatedCoinIndicesSignatureResponse, AggregatedExpirationDateSignatureResponse,
@@ -20,9 +20,11 @@ use nym_api_requests::ecash::{
PartialExpirationDateSignatureResponse, VerificationKeyResponse,
};
use nym_api_requests::models::{
ApiHealthResponse, GatewayCoreStatusResponse, HistoricalPerformanceResponse,
MixnodeCoreStatusResponse, NymNodeDescription,
ApiHealthResponse, GatewayBondAnnotated, GatewayCoreStatusResponse,
HistoricalPerformanceResponse, MixnodeCoreStatusResponse, MixnodeStatusResponse,
NymNodeDescription, RewardEstimationResponse, StakeSaturationResponse,
};
use nym_api_requests::models::{LegacyDescribedGateway, MixNodeBondAnnotated};
use nym_api_requests::nym_nodes::{
NodesByAddressesResponse, SemiSkimmedNodesWithMetadata, SkimmedNode, SkimmedNodesWithMetadata,
};
@@ -151,7 +153,7 @@ impl Config {
pub struct Client<C, S = NoSigner> {
// ideally they would have been read-only, but unfortunately rust doesn't have such features
// #[deprecated(note = "please use `nym_api_client` instead")]
pub nym_api: nym_http_api_client::Client,
pub nym_api: nym_api::Client,
// pub nym_api_client: NymApiClient,
pub nyxd: NyxdClient<C, S>,
}
@@ -212,7 +214,7 @@ impl Client<ReqwestRpcClient> {
impl<C> Client<C> {
pub fn new_with_rpc_client(config: Config, rpc_client: C) -> Self {
let nym_api_client = nym_http_api_client::Client::new(config.api_url.clone(), None);
let nym_api_client = nym_api::Client::new(config.api_url.clone(), None);
Client {
nym_api: nym_api_client,
@@ -226,7 +228,7 @@ impl<C, S> Client<C, S> {
where
S: OfflineSigner,
{
let nym_api_client = nym_http_api_client::Client::new(config.api_url.clone(), None);
let nym_api_client = nym_api::Client::new(config.api_url.clone(), None);
Client {
nym_api: nym_api_client,
@@ -247,6 +249,65 @@ impl<C, S> Client<C, S> {
self.nym_api.change_base_urls(vec![new_endpoint.into()])
}
#[deprecated]
pub async fn get_cached_mixnodes(&self) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
Ok(self.nym_api.get_mixnodes().await?)
}
#[deprecated]
pub async fn get_cached_mixnodes_detailed(
&self,
) -> Result<Vec<MixNodeBondAnnotated>, ValidatorClientError> {
Ok(self.nym_api.get_mixnodes_detailed().await?)
}
#[deprecated]
pub async fn get_cached_mixnodes_detailed_unfiltered(
&self,
) -> Result<Vec<MixNodeBondAnnotated>, ValidatorClientError> {
Ok(self.nym_api.get_mixnodes_detailed_unfiltered().await?)
}
#[deprecated]
pub async fn get_cached_rewarded_mixnodes(
&self,
) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
Ok(self.nym_api.get_rewarded_mixnodes().await?)
}
#[deprecated]
pub async fn get_cached_rewarded_mixnodes_detailed(
&self,
) -> Result<Vec<MixNodeBondAnnotated>, ValidatorClientError> {
Ok(self.nym_api.get_rewarded_mixnodes_detailed().await?)
}
#[deprecated]
pub async fn get_cached_active_mixnodes(
&self,
) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
Ok(self.nym_api.get_active_mixnodes().await?)
}
#[deprecated]
pub async fn get_cached_active_mixnodes_detailed(
&self,
) -> Result<Vec<MixNodeBondAnnotated>, ValidatorClientError> {
Ok(self.nym_api.get_active_mixnodes_detailed().await?)
}
#[deprecated]
pub async fn get_cached_gateways(&self) -> Result<Vec<GatewayBond>, ValidatorClientError> {
Ok(self.nym_api.get_gateways().await?)
}
#[deprecated]
pub async fn get_cached_gateways_detailed_unfiltered(
&self,
) -> Result<Vec<GatewayBondAnnotated>, ValidatorClientError> {
Ok(self.nym_api.get_gateways_detailed_unfiltered().await?)
}
pub async fn get_full_node_performance_history(
&self,
node_id: NodeId,
@@ -324,25 +385,38 @@ impl<C, S> Client<C, S> {
}
}
/// DEPRECATED: Use nym_http_api_client::Client with from_network() or with_bincode() instead
#[deprecated(
since = "1.2.0",
note = "Use nym_http_api_client::Client::from_network() or ClientBuilder::with_bincode() instead"
)]
#[derive(Clone)]
pub struct NymApiClient {
pub use_bincode: bool,
pub nym_api: nym_http_api_client::Client,
pub nym_api: nym_api::Client,
// TODO: perhaps if we really need it at some (currently I don't see any reasons for it)
// we could re-implement the communication with the REST API on port 1317
}
impl From<nym_api::Client> for NymApiClient {
fn from(nym_api: nym_api::Client) -> Self {
NymApiClient {
use_bincode: false,
nym_api,
}
}
}
// we have to allow the use of deprecated method here as they're calling the deprecated trait methods
#[allow(deprecated)]
impl NymApiClient {
pub fn new(api_url: Url) -> Self {
let nym_api = nym_api::Client::new(api_url, None);
NymApiClient {
use_bincode: true,
nym_api,
}
}
#[cfg(not(target_arch = "wasm32"))]
pub fn new_with_timeout(api_url: Url, timeout: std::time::Duration) -> Self {
let nym_api = nym_http_api_client::Client::new(api_url, Some(timeout));
let nym_api = nym_api::Client::new(api_url, Some(timeout));
NymApiClient {
use_bincode: true,
@@ -357,10 +431,10 @@ impl NymApiClient {
}
pub fn new_with_user_agent(api_url: Url, user_agent: impl Into<UserAgent>) -> Self {
let nym_api = nym_http_api_client::Client::builder(api_url)
let nym_api = nym_api::Client::builder::<_, ValidatorClientError>(api_url)
.expect("invalid api url")
.with_user_agent(user_agent.into())
.build()
.build::<ValidatorClientError>()
.expect("failed to build nym api client");
NymApiClient {
@@ -497,6 +571,37 @@ impl NymApiClient {
Ok(self.nym_api.health().await?)
}
#[deprecated]
pub async fn get_cached_active_mixnodes(
&self,
) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
Ok(self.nym_api.get_active_mixnodes().await?)
}
#[deprecated]
pub async fn get_cached_rewarded_mixnodes(
&self,
) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
Ok(self.nym_api.get_rewarded_mixnodes().await?)
}
#[deprecated]
pub async fn get_cached_mixnodes(&self) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
Ok(self.nym_api.get_mixnodes().await?)
}
#[deprecated]
pub async fn get_cached_gateways(&self) -> Result<Vec<GatewayBond>, ValidatorClientError> {
Ok(self.nym_api.get_gateways().await?)
}
#[deprecated]
pub async fn get_cached_described_gateways(
&self,
) -> Result<Vec<LegacyDescribedGateway>, ValidatorClientError> {
Ok(self.nym_api.get_gateways_described().await?)
}
pub async fn get_all_described_nodes(
&self,
) -> Result<Vec<NymNodeDescription>, ValidatorClientError> {
@@ -563,6 +668,30 @@ impl NymApiClient {
.await?)
}
#[deprecated]
pub async fn get_mixnode_status(
&self,
mix_id: NodeId,
) -> Result<MixnodeStatusResponse, ValidatorClientError> {
Ok(self.nym_api.get_mixnode_status(mix_id).await?)
}
#[deprecated]
pub async fn get_mixnode_reward_estimation(
&self,
mix_id: NodeId,
) -> Result<RewardEstimationResponse, ValidatorClientError> {
Ok(self.nym_api.get_mixnode_reward_estimation(mix_id).await?)
}
#[deprecated]
pub async fn get_mixnode_stake_saturation(
&self,
mix_id: NodeId,
) -> Result<StakeSaturationResponse, ValidatorClientError> {
Ok(self.nym_api.get_mixnode_stake_saturation(mix_id).await?)
}
pub async fn blind_sign(
&self,
request_body: &BlindSignRequestBody,
@@ -3,6 +3,7 @@
use crate::nyxd::contract_traits::{DkgQueryClient, PagedDkgQueryClient};
use crate::nyxd::error::NyxdError;
use crate::NymApiClient;
use nym_coconut_dkg_common::types::{EpochId, NodeIndex};
use nym_coconut_dkg_common::verification_key::ContractVKShare;
use nym_compact_ecash::error::CompactEcashError;
@@ -14,7 +15,7 @@ use url::Url;
// TODO: it really doesn't feel like this should live in this crate.
#[derive(Clone)]
pub struct EcashApiClient {
pub api_client: nym_http_api_client::Client,
pub api_client: NymApiClient,
pub verification_key: VerificationKeyAuth,
pub node_id: NodeIndex,
pub cosmos_address: cosmrs::AccountId,
@@ -24,15 +25,10 @@ impl Display for EcashApiClient {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
"[id: {}] {} @ ({})",
"[id: {}] {} @ {}",
self.node_id,
self.cosmos_address,
self.api_client
.base_urls()
.iter()
.map(|url| url.to_string())
.collect::<Vec<String>>()
.join(", ")
self.api_client.api_url()
)
}
}
@@ -64,9 +60,6 @@ pub enum EcashApiError {
source: CompactEcashError,
},
#[error("failed to create API client: {0}")]
ClientError(String),
#[error("the provided account address is malformed: {source}")]
MalformedAccountAddress {
#[from]
@@ -96,13 +89,8 @@ impl TryFrom<ContractVKShare> for EcashApiClient {
// In non-client applications this resolver can cause warning logs about H2 connection
// failure. This indicates that the long lived https connection was closed by the remote
// peer and the resolver will have to reconnect. It should not impact actual functionality
let api_client = nym_http_api_client::Client::builder(url_address)
.map_err(|e| EcashApiError::ClientError(e.to_string()))?
.build()
.map_err(|e| EcashApiError::ClientError(e.to_string()))?;
Ok(EcashApiClient {
api_client,
api_client: NymApiClient::new(url_address),
verification_key: VerificationKeyAuth::try_from_bs58(&share.share)?,
node_id: share.node_index,
cosmos_address: share.owner.as_str().parse()?,
@@ -1,8 +1,7 @@
use crate::nym_api::NymApiClientExt;
use crate::nyxd::contract_traits::MixnetQueryClient;
use crate::nyxd::error::NyxdError;
use crate::nyxd::Config as ClientConfig;
use crate::{QueryHttpRpcNyxdClient, ValidatorClientError};
use crate::{NymApiClient, QueryHttpRpcNyxdClient, ValidatorClientError};
use colored::Colorize;
use core::fmt;
use itertools::Itertools;
@@ -88,17 +87,8 @@ fn setup_connection_tests<H: BuildHasher + 'static>(
}
});
let api_connection_test_clients = api_urls.filter_map(|(network, url)| {
match nym_http_api_client::Client::builder(url.clone()).and_then(|b| b.build()) {
Ok(client) => Some(ClientForConnectionTest::Api(network, url, client)),
Err(err) => {
eprintln!(
"Failed to create API client for {}: {err}",
network.network_name
);
None
}
}
let api_connection_test_clients = api_urls.map(|(network, url)| {
ClientForConnectionTest::Api(network, url.clone(), NymApiClient::new(url))
});
nyxd_connection_test_clients.chain(api_connection_test_clients)
@@ -170,7 +160,7 @@ async fn test_nyxd_connection(
async fn test_nym_api_connection(
network: NymNetworkDetails,
url: &Url,
client: &nym_http_api_client::Client,
client: &NymApiClient,
) -> ConnectionResult {
let result = match timeout(
Duration::from_secs(CONNECTION_TEST_TIMEOUT_SEC),
@@ -196,7 +186,7 @@ async fn test_nym_api_connection(
enum ClientForConnectionTest {
Nyxd(NymNetworkDetails, Url, Box<QueryHttpRpcNyxdClient>),
Api(NymNetworkDetails, Url, nym_http_api_client::Client),
Api(NymNetworkDetails, Url, NymApiClient),
}
impl ClientForConnectionTest {
@@ -9,7 +9,8 @@ use thiserror::Error;
pub enum ValidatorClientError {
#[error("nym api request failed: {source}")]
NymAPIError {
source: Box<nym_api::error::NymAPIError>,
#[from]
source: nym_api::error::NymAPIError,
},
#[error("Tendermint RPC request failure: {0}")]
@@ -27,11 +28,3 @@ pub enum ValidatorClientError {
#[error("No validator API url has been provided")]
NoAPIUrlAvailable,
}
impl From<nym_api::error::NymAPIError> for ValidatorClientError {
fn from(source: nym_api::error::NymAPIError) -> Self {
ValidatorClientError::NymAPIError {
source: Box::new(source),
}
}
}
@@ -14,6 +14,7 @@ pub mod signing;
pub use crate::error::ValidatorClientError;
pub use crate::rpc::reqwest::ReqwestRpcClient;
pub use crate::signing::direct_wallet::DirectSecp256k1HdWallet;
pub use client::NymApiClient;
pub use client::{Client, Config, EcashApiClient};
pub use nym_api_requests::*;
pub use nym_http_api_client::UserAgent;
@@ -1,6 +1,7 @@
// Copyright 2022-2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_api_requests::models::RequestError;
use nym_http_api_client::HttpClientError;
pub type NymAPIError = HttpClientError;
pub type NymAPIError = HttpClientError<RequestError>;
@@ -3,7 +3,6 @@
use crate::nym_api::error::NymAPIError;
use crate::nym_api::routes::{ecash, CORE_STATUS_COUNT, SINCE_ARG};
use crate::nym_nodes::SkimmedNodesWithMetadata;
use async_trait::async_trait;
use nym_api_requests::ecash::models::{
AggregatedCoinIndicesSignatureResponse, AggregatedExpirationDateSignatureResponse,
@@ -16,8 +15,9 @@ use nym_api_requests::ecash::models::{
use nym_api_requests::ecash::VerificationKeyResponse;
use nym_api_requests::models::{
AnnotationResponse, ApiHealthResponse, BinaryBuildInformationOwned, ChainBlocksStatusResponse,
ChainStatusResponse, KeyRotationInfoResponse, NodePerformanceResponse, NodeRefreshBody,
NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse, SignerInformationResponse,
ChainStatusResponse, KeyRotationInfoResponse, LegacyDescribedMixNode, NodePerformanceResponse,
NodeRefreshBody, NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse,
SignerInformationResponse,
};
use nym_api_requests::nym_nodes::{
NodesByAddressesRequestBody, NodesByAddressesResponse, PaginatedCachedNodesResponseV1,
@@ -31,22 +31,26 @@ pub use nym_api_requests::{
VerifyEcashCredentialBody,
},
models::{
GatewayCoreStatusResponse, GatewayStatusReportResponse, GatewayUptimeHistoryResponse,
MixnodeCoreStatusResponse, MixnodeStatusReportResponse, MixnodeStatusResponse,
MixnodeUptimeHistoryResponse, StakeSaturationResponse, UptimeResponse,
ComputeRewardEstParam, GatewayBondAnnotated, GatewayCoreStatusResponse,
GatewayStatusReportResponse, GatewayUptimeHistoryResponse, LegacyDescribedGateway,
MixNodeBondAnnotated, MixnodeCoreStatusResponse, MixnodeStatusReportResponse,
MixnodeStatusResponse, MixnodeUptimeHistoryResponse, RewardEstimationResponse,
StakeSaturationResponse, UptimeResponse,
},
nym_nodes::{CachedNodesResponse, SemiSkimmedNode, SemiSkimmedNodesWithMetadata, SkimmedNode},
nym_nodes::{CachedNodesResponse, SemiSkimmedNode, SkimmedNode},
NymNetworkDetailsResponse,
};
use nym_contracts_common::IdentityKey;
use nym_http_api_client::{ApiClient, NO_PARAMS};
use nym_mixnet_contract_common::{IdentityKeyRef, NodeId, NymNodeDetails};
use nym_mixnet_contract_common::mixnode::MixNodeDetails;
use nym_mixnet_contract_common::{GatewayBond, IdentityKeyRef, NodeId, NymNodeDetails};
use std::net::IpAddr;
use time::format_description::BorrowedFormatItem;
use time::Date;
use tracing::instrument;
use crate::ValidatorClientError;
pub use nym_coconut_dkg_common::types::EpochId;
pub use nym_http_api_client::Client;
pub mod error;
pub mod routes;
@@ -58,9 +62,6 @@ pub fn rfc_3339_date() -> Vec<BorrowedFormatItem<'static>> {
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait NymApiClientExt: ApiClient {
/// Get the current API URL being used by the client
fn api_url(&self) -> &url::Url;
async fn health(&self) -> Result<ApiHealthResponse, NymAPIError> {
self.get_json(
&[
@@ -86,6 +87,104 @@ pub trait NymApiClientExt: ApiClient {
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_mixnodes(&self) -> Result<Vec<MixNodeDetails>, NymAPIError> {
self.get_json(&[routes::V1_API_VERSION, routes::MIXNODES], NO_PARAMS)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_mixnodes_detailed(&self) -> Result<Vec<MixNodeBondAnnotated>, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS,
routes::MIXNODES,
routes::DETAILED,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_gateways_detailed(&self) -> Result<Vec<GatewayBondAnnotated>, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS,
routes::GATEWAYS,
routes::DETAILED,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_gateways_detailed_unfiltered(
&self,
) -> Result<Vec<GatewayBondAnnotated>, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS,
routes::GATEWAYS,
routes::DETAILED_UNFILTERED,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_mixnodes_detailed_unfiltered(
&self,
) -> Result<Vec<MixNodeBondAnnotated>, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS,
routes::MIXNODES,
routes::DETAILED_UNFILTERED,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_gateways(&self) -> Result<Vec<GatewayBond>, NymAPIError> {
self.get_json(&[routes::V1_API_VERSION, routes::GATEWAYS], NO_PARAMS)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_gateways_described(&self) -> Result<Vec<LegacyDescribedGateway>, NymAPIError> {
self.get_json(
&[routes::V1_API_VERSION, routes::GATEWAYS, routes::DESCRIBED],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_mixnodes_described(&self) -> Result<Vec<LegacyDescribedMixNode>, NymAPIError> {
self.get_json(
&[routes::V1_API_VERSION, routes::MIXNODES, routes::DESCRIBED],
NO_PARAMS,
)
.await
}
#[tracing::instrument(level = "debug", skip_all)]
async fn get_node_performance_history(
&self,
@@ -142,156 +241,6 @@ pub trait NymApiClientExt: ApiClient {
.await
}
async fn get_current_rewarded_set(&self) -> Result<RewardedSetResponse, NymAPIError> {
self.get_rewarded_set().await
}
async fn get_all_basic_nodes_with_metadata(
&self,
) -> Result<SkimmedNodesWithMetadata, NymAPIError> {
// unroll first loop iteration in order to obtain the metadata
let mut page = 0;
let res = self
.get_basic_nodes_v2(false, Some(page), None, true)
.await?;
let mut nodes = res.nodes.data;
let metadata = res.metadata;
if res.nodes.pagination.total == nodes.len() {
return Ok(SkimmedNodesWithMetadata::new(nodes, metadata));
}
page += 1;
loop {
let mut res = self
.get_basic_nodes_v2(false, Some(page), None, true)
.await?;
if !metadata.consistency_check(&res.metadata) {
// Create a custom error for inconsistent metadata
return Err(NymAPIError::InternalResponseInconsistency {
url: self.api_url().clone(),
details: "Inconsistent paged metadata".to_string(),
});
}
nodes.append(&mut res.nodes.data);
if nodes.len() >= res.nodes.pagination.total {
break;
} else {
page += 1
}
}
Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
}
async fn get_all_basic_active_mixing_assigned_nodes_with_metadata(
&self,
) -> Result<SkimmedNodesWithMetadata, NymAPIError> {
// Get all mixing nodes that are in the active/rewarded set
let mut page = 0;
let res = self
.get_basic_active_mixing_assigned_nodes_v2(false, Some(page), None, false)
.await?;
let metadata = res.metadata;
let mut nodes = res.nodes.data;
if res.nodes.pagination.total == nodes.len() {
return Ok(SkimmedNodesWithMetadata::new(nodes, metadata));
}
page += 1;
loop {
let res = self
.get_basic_active_mixing_assigned_nodes_v2(false, Some(page), None, false)
.await?;
if !metadata.consistency_check(&res.metadata) {
return Err(NymAPIError::InternalResponseInconsistency {
url: self.api_url().clone(),
details: "Inconsistent paged metadata".to_string(),
});
}
nodes.append(&mut res.nodes.data.clone());
// Check if we've got all nodes
if nodes.len() >= res.nodes.pagination.total {
break;
} else {
page += 1;
}
}
Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
}
async fn get_all_basic_entry_assigned_nodes_with_metadata(
&self,
) -> Result<SkimmedNodesWithMetadata, NymAPIError> {
// Get all nodes that can act as entry gateways
let mut page = 0;
let res = self
.get_basic_entry_assigned_nodes_v2(false, Some(page), None, false)
.await?;
let metadata = res.metadata;
let mut nodes = res.nodes.data;
if res.nodes.pagination.total == nodes.len() {
return Ok(SkimmedNodesWithMetadata::new(nodes, metadata));
}
page += 1;
loop {
let res = self
.get_basic_entry_assigned_nodes_v2(false, Some(page), None, false)
.await?;
if !metadata.consistency_check(&res.metadata) {
return Err(NymAPIError::InternalResponseInconsistency {
url: self.api_url().clone(),
details: "Inconsistent paged metadata".to_string(),
});
}
nodes.append(&mut res.nodes.data.clone());
// Check if we've got all nodes
if nodes.len() >= res.nodes.pagination.total {
break;
} else {
page += 1;
}
}
Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
}
async fn get_all_described_nodes(&self) -> Result<Vec<NymNodeDescription>, NymAPIError> {
// TODO: deal with paging in macro or some helper function or something, because it's the same pattern everywhere
let mut page = 0;
let mut descriptions = Vec::new();
loop {
let mut res = self.get_nodes_described(Some(page), None).await?;
descriptions.append(&mut res.data);
if descriptions.len() < res.pagination.total {
page += 1
} else {
break;
}
}
Ok(descriptions)
}
#[tracing::instrument(level = "debug", skip_all)]
async fn get_nym_nodes(
&self,
@@ -319,25 +268,6 @@ pub trait NymApiClientExt: ApiClient {
.await
}
async fn get_all_bonded_nym_nodes(&self) -> Result<Vec<NymNodeDetails>, ValidatorClientError> {
// TODO: deal with paging in macro or some helper function or something, because it's the same pattern everywhere
let mut page = 0;
let mut bonds = Vec::new();
loop {
let mut res = self.get_nym_nodes(Some(page), None).await?;
bonds.append(&mut res.data);
if bonds.len() < res.pagination.total {
page += 1
} else {
break;
}
}
Ok(bonds)
}
#[deprecated]
#[tracing::instrument(level = "debug", skip_all)]
async fn get_basic_mixnodes(&self) -> Result<CachedNodesResponse<SkimmedNode>, NymAPIError> {
@@ -747,6 +677,42 @@ pub trait NymApiClientExt: ApiClient {
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_active_mixnodes(&self) -> Result<Vec<MixNodeDetails>, NymAPIError> {
self.get_json(
&[routes::V1_API_VERSION, routes::MIXNODES, routes::ACTIVE],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_active_mixnodes_detailed(&self) -> Result<Vec<MixNodeBondAnnotated>, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS,
routes::MIXNODES,
routes::ACTIVE,
routes::DETAILED,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_rewarded_mixnodes(&self) -> Result<Vec<MixNodeDetails>, NymAPIError> {
self.get_json(
&[routes::V1_API_VERSION, routes::MIXNODES, routes::REWARDED],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_mixnode_report(
@@ -823,6 +789,24 @@ pub trait NymApiClientExt: ApiClient {
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_rewarded_mixnodes_detailed(
&self,
) -> Result<Vec<MixNodeBondAnnotated>, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS,
routes::MIXNODES,
routes::REWARDED,
routes::DETAILED,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_gateway_core_status_count(
@@ -890,6 +874,104 @@ pub trait NymApiClientExt: ApiClient {
}
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_mixnode_status(
&self,
mix_id: NodeId,
) -> Result<MixnodeStatusResponse, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS_ROUTES,
routes::MIXNODE,
&mix_id.to_string(),
routes::STATUS,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_mixnode_reward_estimation(
&self,
mix_id: NodeId,
) -> Result<RewardEstimationResponse, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS_ROUTES,
routes::MIXNODE,
&mix_id.to_string(),
routes::REWARD_ESTIMATION,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn compute_mixnode_reward_estimation(
&self,
mix_id: NodeId,
request_body: &ComputeRewardEstParam,
) -> Result<RewardEstimationResponse, NymAPIError> {
self.post_json(
&[
routes::V1_API_VERSION,
routes::STATUS_ROUTES,
routes::MIXNODE,
&mix_id.to_string(),
routes::COMPUTE_REWARD_ESTIMATION,
],
NO_PARAMS,
request_body,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_mixnode_stake_saturation(
&self,
mix_id: NodeId,
) -> Result<StakeSaturationResponse, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS_ROUTES,
routes::MIXNODE,
&mix_id.to_string(),
routes::STAKE_SATURATION,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[allow(deprecated)]
#[instrument(level = "debug", skip(self))]
async fn get_mixnode_inclusion_probability(
&self,
mix_id: NodeId,
) -> Result<nym_api_requests::models::InclusionProbabilityResponse, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::STATUS_ROUTES,
routes::MIXNODE,
&mix_id.to_string(),
routes::INCLUSION_CHANCE,
],
NO_PARAMS,
)
.await
}
#[instrument(level = "debug", skip(self))]
async fn get_current_node_performance(
&self,
@@ -938,6 +1020,34 @@ pub trait NymApiClientExt: ApiClient {
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_mixnodes_blacklisted(&self) -> Result<Vec<NodeId>, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::MIXNODES,
routes::BLACKLISTED,
],
NO_PARAMS,
)
.await
}
#[deprecated]
#[instrument(level = "debug", skip(self))]
async fn get_gateways_blacklisted(&self) -> Result<Vec<IdentityKey>, NymAPIError> {
self.get_json(
&[
routes::V1_API_VERSION,
routes::GATEWAYS,
routes::BLACKLISTED,
],
NO_PARAMS,
)
.await
}
#[instrument(level = "debug", skip(self, request_body))]
async fn blind_sign(
&self,
@@ -1261,49 +1371,8 @@ pub trait NymApiClientExt: ApiClient {
)
.await
}
/// Method to change the base API URLs being used by the client
fn change_base_urls(&mut self, urls: Vec<url::Url>);
/// Retrieve expanded information for all bonded nodes on the network
async fn get_all_expanded_nodes(&self) -> Result<SemiSkimmedNodesWithMetadata, NymAPIError> {
// Unroll the first iteration to get the metadata
let mut page = 0;
let res = self.get_expanded_nodes(false, Some(page), None).await?;
let mut nodes = res.nodes.data;
let metadata = res.metadata;
if res.nodes.pagination.total == nodes.len() {
return Ok(SemiSkimmedNodesWithMetadata::new(nodes, metadata));
}
page += 1;
loop {
let mut res = self.get_expanded_nodes(false, Some(page), None).await?;
nodes.append(&mut res.nodes.data);
if nodes.len() < res.nodes.pagination.total {
page += 1
} else {
break;
}
}
Ok(SemiSkimmedNodesWithMetadata::new(nodes, metadata))
}
}
// Client is already nym_http_api_client::Client (re-exported above), so just one impl needed
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl NymApiClientExt for nym_http_api_client::Client {
fn api_url(&self) -> &url::Url {
self.current_url().as_ref()
}
fn change_base_urls(&mut self, urls: Vec<url::Url>) {
self.change_base_urls(urls.into_iter().map(|u| u.into()).collect());
}
}
impl NymApiClientExt for Client {}
-1
View File
@@ -38,7 +38,6 @@ cosmrs = { workspace = true }
cosmwasm-std = { workspace = true }
nym-validator-client = { path = "../client-libs/validator-client" }
nym-http-api-client = { path = "../http-api-client" }
nym-bin-common = { path = "../../common/bin-common", features = ["output_format"] }
nym-crypto = { path = "../../common/crypto", features = ["asymmetric"] }
nym-network-defaults = { path = "../network-defaults" }
+1 -1
View File
@@ -2,12 +2,12 @@
// SPDX-License-Identifier: Apache-2.0
use crate::context::errors::ContextError;
pub use nym_http_api_client::Client as NymApiClient;
use nym_network_defaults::{
setup_env,
var_names::{MIXNET_CONTRACT_ADDRESS, NYM_API, NYXD, VESTING_CONTRACT_ADDRESS},
NymNetworkDetails,
};
pub use nym_validator_client::nym_api::Client as NymApiClient;
use nym_validator_client::nyxd::{self, AccountId, NyxdClient};
use nym_validator_client::{
DirectSigningHttpRpcNyxdClient, DirectSigningHttpRpcValidatorClient, QueryHttpRpcNyxdClient,
@@ -1,3 +1,3 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type Role = "entry_gateway" | "layer1" | "layer2" | "layer3" | "exit_gateway" | "standby";
export type Role = "EntryGateway" | "Layer1" | "Layer2" | "Layer3" | "ExitGateway" | "Standby";
@@ -23,7 +23,6 @@ use serde_repr::{Deserialize_repr, Serialize_repr};
/// Full details associated with given mixnode.
#[cw_serde]
#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
pub struct MixNodeDetails {
/// Basic bond information of this mixnode, such as owner address, original pledge, etc.
pub bond_information: MixNodeBond,
@@ -696,7 +695,6 @@ impl From<LegacyMixLayer> for u8 {
Copy,
)]
#[schemars(crate = "::cosmwasm_schema::schemars")]
#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
pub struct PendingMixNodeChanges {
pub pledge_change: Option<EpochEventId>,
+1 -12
View File
@@ -3,7 +3,7 @@
use nym_ecash_signer_check::SignerCheckError;
use nym_validator_client::coconut::EcashApiError;
use nym_validator_client::nym_api::{error::NymAPIError, EpochId};
use nym_validator_client::nym_api::EpochId;
use nym_validator_client::nyxd::error::NyxdError;
use std::io;
use std::net::SocketAddr;
@@ -71,9 +71,6 @@ pub enum CredentialProxyError {
source: EcashApiError,
},
#[error("Nym API request failed: {source}")]
NymApiFailure { source: Box<NymAPIError> },
#[error("Compact ecash internal error: {0}")]
CompactEcashInternalError(#[from] nym_compact_ecash::error::CompactEcashError),
@@ -157,14 +154,6 @@ pub enum CredentialProxyError {
},
}
impl From<NymAPIError> for CredentialProxyError {
fn from(source: NymAPIError) -> Self {
CredentialProxyError::NymApiFailure {
source: Box::new(source),
}
}
}
impl CredentialProxyError {
pub fn database_inconsistency<S: Into<String>>(reason: S) -> CredentialProxyError {
CredentialProxyError::DatabaseInconsistency {
@@ -23,7 +23,6 @@ use nym_credentials::{
AggregatedCoinIndicesSignatures, AggregatedExpirationDateSignatures, EpochVerificationKey,
};
use nym_ecash_contract_common::deposit::DepositId;
use nym_validator_client::client::NymApiClientExt;
use nym_validator_client::coconut::EcashApiError;
use nym_validator_client::nym_api::EpochId;
use nym_validator_client::nyxd::contract_traits::dkg_query_client::Epoch;
@@ -11,7 +11,6 @@ use nym_credential_proxy_requests::api::v1::ticketbook::models::{
TicketbookWalletSharesResponse, WalletShare, WebhookTicketbookWalletShares,
WebhookTicketbookWalletSharesRequest,
};
use nym_validator_client::client::NymApiClientExt;
use nym_validator_client::ecash::BlindSignRequestBody;
use std::collections::HashMap;
use std::sync::Arc;
@@ -14,7 +14,7 @@ use nym_api_requests::ecash::models::{BatchRedeemTicketsBody, VerifyEcashTicketB
use nym_credentials_interface::Bandwidth;
use nym_credentials_interface::{ClientTicket, TicketType};
use nym_validator_client::coconut::EcashApiError;
use nym_validator_client::nym_api::{EpochId, NymApiClientExt};
use nym_validator_client::nym_api::EpochId;
use nym_validator_client::nyxd::contract_traits::{
EcashSigningClient, MultisigQueryClient, MultisigSigningClient, PagedMultisigQueryClient,
};
@@ -126,7 +126,7 @@ pub struct CredentialHandlerConfig {
pub maximum_time_between_redemption: Duration,
}
pub struct CredentialHandler {
pub(crate) struct CredentialHandler {
config: CredentialHandlerConfig,
multisig_threshold: f32,
ticket_receiver: UnboundedReceiver<ClientTicket>,
@@ -354,7 +354,7 @@ impl CredentialHandler {
Err(err) => {
error!("failed to send ticket {ticket_id} for verification to ecash signer '{client}': {err}. if we don't reach quorum, we'll retry later");
Err(EcashTicketError::ApiFailure(EcashApiError::NymApi {
source: nym_validator_client::ValidatorClientError::from(err),
source: err,
}))
}
}
@@ -907,7 +907,7 @@ impl CredentialHandler {
Ok(())
}
pub async fn run(mut self, shutdown: nym_task::ShutdownToken) {
async fn run(mut self, mut shutdown: nym_task::TaskClient) {
info!("Starting Ecash CredentialSender");
// attempt to clear any pending operations
@@ -919,12 +919,11 @@ impl CredentialHandler {
let start = Instant::now() + self.config.pending_poller;
let mut resolver_interval = interval_at(start, self.config.pending_poller);
loop {
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.cancelled() => {
_ = shutdown.recv() => {
trace!("client_handling::credentialSender : received shutdown");
break
},
Some(ticket) = self.ticket_receiver.next() => {
let (queued_up, _) = self.ticket_receiver.size_hint();
@@ -947,4 +946,8 @@ impl CredentialHandler {
}
}
}
pub(crate) fn start(self, shutdown: nym_task::TaskClient) {
tokio::spawn(async move { self.run(shutdown).await });
}
}
@@ -82,8 +82,9 @@ impl EcashManager {
credential_handler_cfg: CredentialHandlerConfig,
nyxd_client: DirectSigningHttpRpcNyxdClient,
pk_bytes: [u8; 32],
shutdown: nym_task::TaskClient,
storage: GatewayStorage,
) -> Result<(Self, CredentialHandler), Error> {
) -> Result<Self, Error> {
let shared_state = SharedState::new(nyxd_client, Box::new(storage)).await?;
let (cred_sender, cred_receiver) = mpsc::unbounded();
@@ -91,16 +92,14 @@ impl EcashManager {
let cs =
CredentialHandler::new(credential_handler_cfg, cred_receiver, shared_state.clone())
.await?;
cs.start(shutdown);
Ok((
EcashManager {
shared_state,
pk_bytes,
pay_infos: Default::default(),
cred_sender,
},
cs,
))
Ok(EcashManager {
shared_state,
pk_bytes,
pay_infos: Default::default(),
cred_sender,
})
}
pub async fn verify_pay_info(&self, pay_info: NymPayInfo) -> Result<usize, EcashTicketError> {
-1
View File
@@ -22,7 +22,6 @@ nym-ecash-time = { path = "../ecash-time", features = ["expiration"] }
nym-credentials-interface = { path = "../credentials-interface" }
nym-crypto = { path = "../crypto" }
nym-api-requests = { path = "../../nym-api/nym-api-requests" }
nym-http-api-client = { path = "../http-api-client" }
nym-validator-client = { path = "../client-libs/validator-client", default-features = false }
nym-ecash-contract-common = { path = "../cosmwasm-smart-contracts/ecash-contract" }
nym-network-defaults = { path = "../network-defaults" }
@@ -15,7 +15,7 @@ use nym_credentials_interface::{
use nym_crypto::asymmetric::ed25519;
use nym_ecash_contract_common::deposit::DepositId;
use nym_ecash_time::{ecash_default_expiration_date, ecash_today, EcashTime};
use nym_validator_client::nym_api::{EpochId, NymApiClientExt};
use nym_validator_client::nym_api::EpochId;
use serde::{Deserialize, Serialize};
use time::Date;
@@ -116,7 +116,7 @@ impl IssuanceTicketBook {
pub async fn obtain_blinded_credential(
&self,
client: &nym_http_api_client::Client,
client: &nym_validator_client::client::NymApiClient,
request_body: &BlindSignRequestBody,
) -> Result<BlindedSignature, Error> {
let server_response = client.blind_sign(request_body).await?;
@@ -179,7 +179,7 @@ impl IssuanceTicketBook {
// ideally this would have been generic over credential type, but we really don't need secp256k1 keys for bandwidth vouchers
pub async fn obtain_partial_ticketbook_credential(
&self,
client: &nym_http_api_client::Client,
client: &nym_validator_client::client::NymApiClient,
signer_index: u64,
validator_vk: &VerificationKeyAuth,
signing_data: CredentialSigningData,
-1
View File
@@ -10,7 +10,6 @@ use nym_credentials_interface::{
VerificationKeyAuth, WalletSignatures,
};
use nym_validator_client::client::EcashApiClient;
use nym_validator_client::nym_api::NymApiClientExt;
// so we wouldn't break all the existing imports
pub use nym_ecash_time::{cred_exp_date, ecash_date_offset, ecash_today, EcashTime};
+1 -10
View File
@@ -4,7 +4,7 @@
use crate::ecash::bandwidth::issued::CURRENT_SERIALIZATION_REVISION;
use nym_credentials_interface::CompactEcashError;
use nym_crypto::asymmetric::x25519::KeyRecoveryError;
use nym_validator_client::{nym_api::error::NymAPIError, ValidatorClientError};
use nym_validator_client::ValidatorClientError;
use thiserror::Error;
#[derive(Debug, Error)]
@@ -37,9 +37,6 @@ pub enum Error {
#[error("Ran into a validator client error - {0}")]
ValidatorClientError(#[from] ValidatorClientError),
#[error("Nym API request failed - {0}")]
NymAPIError(Box<NymAPIError>),
#[error("Bandwidth operation overflowed. {0}")]
BandwidthOverflow(String),
@@ -64,9 +61,3 @@ pub enum Error {
#[error("failed to create a secp256k1 signature")]
Secp256k1SignFailure,
}
impl From<NymAPIError> for Error {
fn from(e: NymAPIError) -> Self {
Error::NymAPIError(Box::new(e))
}
}
-1
View File
@@ -22,7 +22,6 @@ url = { workspace = true }
nym-validator-client = { path = "../client-libs/validator-client" }
nym-network-defaults = { path = "../network-defaults" }
nym-ecash-signer-check-types = { path = "../ecash-signer-check-types" }
nym-http-api-client = { path = "../http-api-client" }
[lints]
workspace = true
+25 -31
View File
@@ -1,14 +1,15 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::{LocalChainStatus, SignerCheckError, SigningStatus, TypedSignerResult};
use crate::{LocalChainStatus, SigningStatus, TypedSignerResult};
use nym_ecash_signer_check_types::dealer_information::RawDealerInformation;
use nym_ecash_signer_check_types::status::{SignerStatus, SignerTestResult};
use nym_validator_client::client::NymApiClientExt;
use nym_validator_client::models::BinaryBuildInformationOwned;
use nym_validator_client::nym_api::NymApiClientExt;
use nym_validator_client::nyxd::contract_traits::dkg_query_client::{
ContractVKShare, DealerDetails,
};
use nym_validator_client::NymApiClient;
use std::time::Duration;
use tracing::{error, warn};
use url::Url;
@@ -31,38 +32,37 @@ pub(crate) mod signing_status {
}
struct ClientUnderTest {
api_client: nym_http_api_client::Client,
api_client: NymApiClient,
build_info: Option<BinaryBuildInformationOwned>,
}
impl ClientUnderTest {
pub(crate) fn new(api_url: &Url) -> Result<Self, SignerCheckError> {
// The builder should not fail with a valid URL that's already parsed
// If it does fail, it's an internal error that we can't recover from
let api_client = nym_http_api_client::Client::builder(api_url.clone())?.build()?;
Ok(ClientUnderTest {
api_client,
pub(crate) fn new(api_url: &Url) -> Self {
ClientUnderTest {
api_client: NymApiClient::new(api_url.clone()),
build_info: None,
})
}
}
pub(crate) async fn try_retrieve_build_information(&mut self) -> bool {
match tokio::time::timeout(Duration::from_secs(5), self.api_client.build_information())
.await
match tokio::time::timeout(
Duration::from_secs(5),
self.api_client.nym_api.build_information(),
)
.await
{
Ok(Ok(build_information)) => {
self.build_info = Some(build_information);
true
}
Ok(Err(err)) => {
warn!("{}: failed to retrieve build information: {err}. the signer is most likely down", self.api_client.current_url());
warn!("{}: failed to retrieve build information: {err}. the signer is most likely down", self.api_client.api_url());
false
}
Err(_timeout) => {
warn!(
"{}: timed out while attempting to retrieve build information",
self.api_client.current_url()
self.api_client.api_url()
);
false
}
@@ -77,7 +77,7 @@ impl ClientUnderTest {
.inspect_err(|err| {
error!(
"ecash signer '{}' reports invalid version {}: {err}",
self.api_client.current_url(),
self.api_client.api_url(),
build_info.build_version
)
})
@@ -121,14 +121,14 @@ impl ClientUnderTest {
// check if it supports the current query
if self.supports_chain_status_query() {
return match self.api_client.get_chain_blocks_status().await {
return match self.api_client.nym_api.get_chain_blocks_status().await {
Ok(status) => LocalChainStatus::Reachable {
response: Box::new(status),
},
Err(err) => {
warn!(
"{}: failed to retrieve local chain status: {err}",
self.api_client.current_url()
self.api_client.api_url()
);
LocalChainStatus::Unreachable
}
@@ -136,14 +136,14 @@ impl ClientUnderTest {
}
// fallback to the legacy query
match self.api_client.get_chain_status().await {
match self.api_client.nym_api.get_chain_status().await {
Ok(status) => LocalChainStatus::ReachableLegacy {
response: Box::new(status),
},
Err(err) => {
warn!(
"{}: failed to retrieve [legacy] local chain status: {err}",
self.api_client.current_url()
self.api_client.api_url()
);
LocalChainStatus::Unreachable
}
@@ -158,14 +158,14 @@ impl ClientUnderTest {
// check if it supports the current query
if self.supports_signing_status_query() {
return match self.api_client.get_signer_status().await {
return match self.api_client.nym_api.get_signer_status().await {
Ok(response) => SigningStatus::Reachable {
response: Box::new(response),
},
Err(err) => {
warn!(
"{}: failed to retrieve signer chain status: {err}",
self.api_client.current_url()
self.api_client.api_url()
);
SigningStatus::Unreachable
}
@@ -173,14 +173,14 @@ impl ClientUnderTest {
}
// fallback to the legacy query
match self.api_client.get_signer_information().await {
match self.api_client.nym_api.get_signer_information().await {
Ok(status) => SigningStatus::ReachableLegacy {
response: Box::new(status),
},
Err(err) => {
warn!(
"{}: failed to retrieve [legacy] signer chain status: {err}",
self.api_client.current_url()
self.api_client.api_url()
);
// NOTE: this might equally mean the signing is disabled
SigningStatus::Unreachable
@@ -201,13 +201,7 @@ pub(crate) async fn check_client(
return SignerStatus::ProvidedInvalidDetails.with_details(dealer_information, dkg_epoch);
};
let mut client = match ClientUnderTest::new(&parsed_information.announce_address) {
Ok(client) => client,
Err(err) => {
error!("failed to create client instance: {err}");
return SignerStatus::Unreachable.with_details(dealer_information, dkg_epoch);
}
};
let mut client = ClientUnderTest::new(&parsed_information.announce_address);
// 8. check basic connection status - can you retrieve build information?
if !client.try_retrieve_build_information().await {
-12
View File
@@ -1,7 +1,6 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_http_api_client::HttpClientError;
use nym_validator_client::nyxd::error::NyxdError;
use thiserror::Error;
@@ -12,17 +11,6 @@ pub enum SignerCheckError {
#[error("failed to query the DKG contract: {source}")]
DKGContractQueryFailure { source: NyxdError },
#[error("failed to build client: {source}")]
HttpClient { source: Box<HttpClientError> },
}
impl From<HttpClientError> for SignerCheckError {
fn from(e: HttpClientError) -> Self {
SignerCheckError::HttpClient {
source: Box::new(e),
}
}
}
impl SignerCheckError {
@@ -14,7 +14,7 @@ use std::task::{Context, Poll};
use tungstenite::{Error as WsError, Message as WsMessage};
#[cfg(not(target_arch = "wasm32"))]
use nym_task::ShutdownToken;
use nym_task::TaskClient;
pub(crate) type WsItem = Result<WsMessage, WsError>;
@@ -52,7 +52,7 @@ pub fn client_handshake<'a, S, R>(
gateway_pubkey: ed25519::PublicKey,
expects_credential_usage: bool,
derive_aes256_gcm_siv_key: bool,
#[cfg(not(target_arch = "wasm32"))] shutdown_token: ShutdownToken,
#[cfg(not(target_arch = "wasm32"))] shutdown: TaskClient,
) -> GatewayHandshake<'a>
where
S: Stream<Item = WsItem> + Sink<WsMessage> + Unpin + Send + 'a,
@@ -64,7 +64,7 @@ where
identity,
Some(gateway_pubkey),
#[cfg(not(target_arch = "wasm32"))]
shutdown_token,
shutdown,
)
.with_credential_usage(expects_credential_usage)
.with_aes256_gcm_siv_key(derive_aes256_gcm_siv_key);
@@ -80,13 +80,13 @@ pub fn gateway_handshake<'a, S, R>(
ws_stream: &'a mut S,
identity: &'a ed25519::KeyPair,
received_init_payload: Vec<u8>,
shutdown_token: ShutdownToken,
shutdown: TaskClient,
) -> GatewayHandshake<'a>
where
S: Stream<Item = WsItem> + Sink<WsMessage> + Unpin + Send + 'a,
R: CryptoRng + RngCore + Send,
{
let state = State::new(rng, ws_stream, identity, None, shutdown_token);
let state = State::new(rng, ws_stream, identity, None, shutdown);
GatewayHandshake {
handshake_future: Box::pin(state.perform_gateway_handshake(received_init_payload)),
}
@@ -149,7 +149,7 @@ mod tests {
*gateway_keys.public_key(),
false,
true,
ShutdownToken::default(),
TaskClient::dummy(),
);
let client_fut = handshake_client.spawn_timeboxed();
@@ -176,7 +176,7 @@ mod tests {
gateway_ws,
gateway_keys,
init_msg,
ShutdownToken::default(),
TaskClient::dummy(),
);
let gateway_fut = handshake_gateway.spawn_timeboxed();
@@ -24,7 +24,7 @@ use tracing::log::*;
use tungstenite::Message as WsMessage;
#[cfg(not(target_arch = "wasm32"))]
use nym_task::ShutdownToken;
use nym_task::TaskClient;
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::timeout;
@@ -63,7 +63,7 @@ pub(crate) struct State<'a, S, R> {
// channel to receive shutdown signal
#[cfg(not(target_arch = "wasm32"))]
shutdown_token: ShutdownToken,
shutdown: TaskClient,
}
impl<'a, S, R> State<'a, S, R> {
@@ -72,7 +72,7 @@ impl<'a, S, R> State<'a, S, R> {
ws_stream: &'a mut S,
identity: &'a ed25519::KeyPair,
remote_pubkey: Option<ed25519::PublicKey>,
#[cfg(not(target_arch = "wasm32"))] shutdown_token: ShutdownToken,
#[cfg(not(target_arch = "wasm32"))] shutdown: TaskClient,
) -> Self
where
R: CryptoRng + RngCore,
@@ -89,7 +89,7 @@ impl<'a, S, R> State<'a, S, R> {
expects_credential_usage: false,
derive_aes256_gcm_siv_key: false,
#[cfg(not(target_arch = "wasm32"))]
shutdown_token,
shutdown,
}
}
@@ -306,7 +306,7 @@ impl<'a, S, R> State<'a, S, R> {
loop {
tokio::select! {
biased;
_ = self.shutdown_token.cancelled() => return Err(HandshakeError::ReceivedShutdown),
_ = self.shutdown.recv() => return Err(HandshakeError::ReceivedShutdown),
msg = self.ws_stream.next() => {
let Some(ret) = Self::on_wg_msg(msg)? else {
continue;
-31
View File
@@ -1,31 +0,0 @@
[package]
name = "nym-http-api-client-macro"
version = "0.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
edition.workspace = true
license.workspace = true
rust-version.workspace = true
readme.workspace = true
[lib]
proc-macro = true
[dependencies]
proc-macro2 = "1"
syn = { workspace = true, features = ["full"] }
quote = "1.0.40"
proc-macro-crate = "3"
uuid = { version = "1.0", features = ["v4"] }
[dev-dependencies]
nym-http-api-client = { path = "../http-api-client" }
reqwest = { workspace = true }
[features]
debug-inventory = []
[lints]
workspace = true
-13
View File
@@ -1,13 +0,0 @@
use std::env;
fn main() {
// Enable debug output during build
if env::var("CARGO_FEATURE_DEBUG_INVENTORY").is_ok() || env::var("DEBUG_HTTP_INVENTORY").is_ok()
{
println!("cargo:warning=HTTP Client Inventory Debug Enabled");
println!("cargo:rustc-cfg=debug_inventory");
}
// Force rebuild when this environment variable changes
println!("cargo:rerun-if-env-changed=DEBUG_HTTP_INVENTORY");
}
-388
View File
@@ -1,388 +0,0 @@
//! Proc-macros for configuring HTTP clients globally via the `inventory` crate.
//!
//! This crate provides macros that allow any crate in the workspace to contribute
//! configuration modifications to `reqwest::ClientBuilder` instances through a
//! compile-time registry pattern.
//!
//! # Overview
//!
//! The macros work by:
//! 1. Collecting configuration functions from across all crates at compile time
//! 2. Sorting them by priority (lower numbers run first)
//! 3. Applying them sequentially to build HTTP clients with consistent settings
//!
//! # Examples
//!
//! ## Basic Usage with `client_defaults!`
//!
//! ```ignore
//! use nym_http_api_client_macro::client_defaults;
//!
//! // Register default configurations with priority
//! client_defaults!(
//! priority = 10; // Optional, defaults to 0
//! timeout = std::time::Duration::from_secs(30),
//! gzip = true,
//! user_agent = "MyApp/1.0"
//! );
//! ```
//!
//! ## Using `client_cfg!` for one-off configurations
//!
//! ```ignore
//! use nym_http_api_client_macro::client_cfg;
//!
//! let configure = client_cfg!(
//! timeout = std::time::Duration::from_secs(60),
//! default_headers {
//! "X-Custom-Header" => "value",
//! "Authorization" => "auth_token"
//! }
//! );
//!
//! let builder = reqwest::ClientBuilder::new();
//! let configured = configure(builder);
//! ```
//!
//! # DSL Reference
//!
//! The macro DSL supports several patterns:
//! - `key = value` - Calls `builder.key(value)`
//! - `key(arg1, arg2)` - Calls `builder.key(arg1, arg2)`
//! - `flag` - Calls `builder.flag()` with no arguments
//! - `default_headers { "name" => "value", ... }` - Sets default headers
//!
//! # Priority System
//!
//! Configurations are applied in priority order (lower numbers first):
//! - Negative priorities: Early configuration (e.g., -100 for base settings)
//! - Zero (default): Standard configuration
//! - Positive priorities: Late configuration (e.g., 100 for overrides)
use proc_macro::TokenStream;
use proc_macro2::{Span, TokenStream as TokenStream2};
use proc_macro_crate::{crate_name, FoundCrate};
use quote::{format_ident, quote};
use syn::{
braced,
parse::{Parse, ParseStream},
parse_macro_input,
punctuated::Punctuated,
token, Expr, Ident, LitInt, Result, Token,
};
// ------------------ core crate path resolution ------------------
fn core_path() -> TokenStream2 {
match crate_name("nym-http-api-client") {
Ok(FoundCrate::Itself) => quote!(crate),
Ok(FoundCrate::Name(name)) => {
let ident = Ident::new(&name, Span::call_site());
quote!( ::#ident )
}
Err(_) => {
// Fallback if the crate is not found by name (unlikely if deps set up correctly)
quote!(::nym_http_api_client)
}
}
}
// ------------------ DSL parsing ------------------
struct Items(Punctuated<Item, Token![,]>);
impl Parse for Items {
fn parse(input: ParseStream<'_>) -> Result<Self> {
Ok(Self(Punctuated::parse_terminated(input)?))
}
}
enum Item {
Assign {
key: Ident,
_eq: Token![=],
value: Expr,
}, // foo = EXPR
Call {
key: Ident,
args: Punctuated<Expr, Token![,]>,
_p: token::Paren,
}, // foo(a,b)
DefaultHeaders {
_key: Ident,
map: HeaderMapInit,
}, // default_headers { ... }
Flag {
key: Ident,
}, // foo
}
impl Parse for Item {
fn parse(input: ParseStream<'_>) -> Result<Self> {
let key: Ident = input.parse()?;
if input.peek(Token![=]) {
let _eq: Token![=] = input.parse()?;
let value: Expr = input.parse()?;
return Ok(Self::Assign { key, _eq, value });
}
if input.peek(token::Paren) {
let content;
let _p = syn::parenthesized!(content in input);
let args = Punctuated::<Expr, Token![,]>::parse_terminated(&content)?;
return Ok(Self::Call { key, args, _p });
}
if input.peek(token::Brace) && key == format_ident!("default_headers") {
let map = input.parse::<HeaderMapInit>()?;
return Ok(Self::DefaultHeaders { _key: key, map });
}
Ok(Self::Flag { key })
}
}
struct HeaderPair {
k: Expr,
_arrow: Token![=>],
v: Expr,
}
impl Parse for HeaderPair {
fn parse(input: ParseStream<'_>) -> Result<Self> {
Ok(Self {
k: input.parse()?,
_arrow: input.parse()?,
v: input.parse()?,
})
}
}
struct HeaderMapInit {
_brace: token::Brace,
pairs: Punctuated<HeaderPair, Token![,]>,
}
impl Parse for HeaderMapInit {
fn parse(input: ParseStream<'_>) -> Result<Self> {
let content;
let _brace = braced!(content in input);
let pairs = Punctuated::<HeaderPair, Token![,]>::parse_terminated(&content)?;
Ok(Self { _brace, pairs })
}
}
// Generate statements that mutate a builder named `b` using the resolved core path.
fn to_stmts(items: Items, core: &TokenStream2) -> TokenStream2 {
let mut stmts = Vec::new();
for it in items.0 {
match it {
Item::Assign { key, value, .. } => {
let m = key;
stmts.push(quote! { b = b.#m(#value); });
}
Item::Call { key, args, .. } => {
let m = key;
let args = args.iter();
stmts.push(quote! { b = b.#m( #( #args ),* ); });
}
Item::DefaultHeaders { map, .. } => {
let (ks, vs): (Vec<_>, Vec<_>) = map.pairs.into_iter().map(|p| (p.k, p.v)).unzip();
stmts.push(quote! {
let mut __cm = #core::reqwest::header::HeaderMap::new();
#(
{
use #core::reqwest::header::{HeaderName, HeaderValue};
let __k = HeaderName::try_from(#ks)
.unwrap_or_else(|e| panic!("Invalid header name: {}", e));
let __v = HeaderValue::try_from(#vs)
.unwrap_or_else(|e| panic!("Invalid header value: {}", e));
__cm.insert(__k, __v);
}
)*
b = b.default_headers(__cm);
});
}
Item::Flag { key } => {
let m = key;
stmts.push(quote! { b = b.#m(); });
}
}
}
quote! { #(#stmts)* }
}
// ------------------ client_cfg! ------------------
/// Creates a closure that configures a `ReqwestClientBuilder`.
///
/// This macro generates a closure that can be used to configure a single
/// `reqwest::ClientBuilder` instance without affecting global defaults.
///
/// # Example
///
/// ```ignore
/// use nym_http_api_client_macro::client_cfg;
///
/// let config = client_cfg!(
/// timeout = std::time::Duration::from_secs(30),
/// gzip = true
/// );
/// let client = config(reqwest::ClientBuilder::new()).build().unwrap();
/// ```
#[proc_macro]
pub fn client_cfg(input: TokenStream) -> TokenStream {
let items = parse_macro_input!(input as Items);
let core = core_path();
let body = to_stmts(items, &core);
let out = quote! {
|mut b: #core::ReqwestClientBuilder| { #body b }
};
out.into()
}
// ------------------ client_defaults! with optional priority header ------------------
struct MaybePrioritized {
priority: i32,
items: Items,
}
impl Parse for MaybePrioritized {
fn parse(input: ParseStream<'_>) -> Result<Self> {
// Optional header: `priority = <int> ;`
let fork = input.fork();
let mut priority = 0i32;
if fork.peek(Ident) && fork.parse::<Ident>()? == "priority" && fork.peek(Token![=]) {
// commit
let _ = input.parse::<Ident>()?; // priority
let _ = input.parse::<Token![=]>()?;
let lit: LitInt = input.parse()?;
priority = lit.base10_parse()?;
let _ = input.parse::<Token![;]>()?;
}
let items = input.parse::<Items>()?;
Ok(Self { priority, items })
}
}
/// Registers global default configurations for HTTP clients.
///
/// This macro submits a configuration record to the global registry that will
/// be applied to all HTTP clients created with `default_builder()`.
///
/// # Parameters
///
/// - `priority` (optional): Integer priority for ordering (lower runs first, default: 0)
/// - Configuration items: Any valid `reqwest::ClientBuilder` method calls
///
/// # Example
///
/// ```ignore
/// use nym_http_api_client_macro::client_defaults;
///
/// client_defaults!(
/// priority = -50; // Run early in the configuration chain
/// connect_timeout = std::time::Duration::from_secs(10),
/// pool_max_idle_per_host = 32,
/// default_headers {
/// "User-Agent" => "MyApp/1.0",
/// "Accept" => "application/json"
/// }
/// );
/// ```
#[proc_macro]
pub fn client_defaults(input: TokenStream) -> TokenStream {
let MaybePrioritized { priority, items } = parse_macro_input!(input as MaybePrioritized);
let core = core_path();
// Generate a description of what this config does (before consuming items)
let config_description = if cfg!(feature = "debug-inventory") {
let descriptions = items
.0
.iter()
.map(|item| match item {
Item::Assign { key, value, .. } => {
format!("{}={:?}", quote!(#key), quote!(#value).to_string())
}
Item::Call { key, args, .. } => {
let args_str = args
.iter()
.map(|a| quote!(#a).to_string())
.collect::<Vec<_>>()
.join(", ");
format!("{}({})", quote!(#key), args_str)
}
Item::Flag { key } => {
format!("{}()", quote!(#key))
}
Item::DefaultHeaders { .. } => "default_headers{{...}}".to_string(),
})
.collect::<Vec<_>>()
.join(", ");
quote! {
pub const __CONFIG_DESC: &str = #descriptions;
}
} else {
quote! {}
};
// Now consume items to generate the body
let body = to_stmts(items, &core);
// Generate a unique identifier for this submission
let submission_id = format!("__client_defaults_{}", uuid::Uuid::new_v4().simple());
let submission_ident = syn::Ident::new(&submission_id, proc_macro2::Span::call_site());
// Debug output at compile time if enabled
if std::env::var("DEBUG_HTTP_INVENTORY").is_ok() {
eprintln!(
"cargo:warning=[HTTP-INVENTORY] Registering config with priority={} from {}",
priority,
std::env::var("CARGO_PKG_NAME").unwrap_or_else(|_| "unknown".to_string())
);
}
// Add debug_print_inventory call if the feature is enabled
let debug_call = if cfg!(feature = "debug-inventory") {
quote! {
#config_description
// Ensure the debug function gets called when config is applied
pub fn __cfg_with_debug(
b: #core::ReqwestClientBuilder
) -> #core::ReqwestClientBuilder {
eprintln!("[HTTP-INVENTORY] Applying: {} (priority={})", __CONFIG_DESC, #priority);
__cfg(b)
}
}
} else {
quote! {}
};
// Use the debug wrapper if feature is enabled
let apply_fn = if cfg!(feature = "debug-inventory") {
quote! { __cfg_with_debug }
} else {
quote! { __cfg }
};
let out = quote! {
#[allow(non_snake_case)]
mod #submission_ident {
use super::*;
#[allow(unused)]
pub fn __cfg(
mut b: #core::ReqwestClientBuilder
) -> #core::ReqwestClientBuilder {
#body
b
}
#debug_call
#core::inventory::submit! {
#core::registry::ConfigRecord {
priority: #priority,
apply: #apply_fn,
}
}
}
};
out.into()
}
@@ -1,64 +0,0 @@
use nym_http_api_client_macro::{client_cfg, client_defaults};
use std::time::Duration;
#[test]
fn test_client_cfg_basic() {
// Test that the macro compiles with basic configuration
let _config = client_cfg!(timeout = Duration::from_secs(30), gzip = true);
}
#[test]
fn test_client_cfg_with_headers() {
// Test that the macro compiles with default headers
let _config = client_cfg!(
timeout = Duration::from_secs(30),
default_headers {
"User-Agent" => "TestApp/1.0",
"Accept" => "application/json"
}
);
}
#[test]
fn test_client_cfg_with_method_calls() {
// Test that the macro compiles with method calls
let _config = client_cfg!(
pool_max_idle_per_host = 32,
tcp_nodelay = true,
danger_accept_invalid_certs = true
);
}
#[test]
fn test_client_defaults_with_priority() {
// Test that client_defaults macro compiles with priority
client_defaults!(
priority = -100;
gzip = true,
deflate = true
);
}
#[test]
fn test_client_defaults_without_priority() {
// Test that client_defaults macro compiles without priority (defaults to 0)
client_defaults!(brotli = true, zstd = true);
}
#[test]
fn test_empty_client_cfg() {
// Test that empty configuration compiles
let _config = client_cfg!();
}
// Integration test to verify the closure actually works
#[test]
fn test_client_cfg_closure_application() {
let config = client_cfg!(gzip = true);
// Apply the configuration to a new builder
let builder = reqwest::ClientBuilder::new();
let _configured_builder = config(builder);
// Note: We can't easily test the internal state of the builder,
// but we verify it compiles and runs without panic
}
-8
View File
@@ -13,25 +13,19 @@ license.workspace = true
[features]
default=["tunneling"]
tunneling=[]
network-defaults = ["dep:nym-network-defaults"]
debug-inventory = ["nym-http-api-client-macro/debug-inventory"]
[dependencies]
async-trait = { workspace = true }
bincode = { workspace = true }
cfg-if = { workspace = true}
reqwest = { workspace = true, features = ["json", "gzip", "deflate", "brotli", "zstd", "rustls-tls"] }
http.workspace = true
url = { workspace = true }
once_cell = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
serde_yaml = { workspace = true}
serde_plain = { workspace = true }
thiserror = { workspace = true }
tracing = { workspace = true }
itertools = { workspace = true }
inventory = { workspace = true }
# used for decoding text responses (they were already implicitly included)
bytes = { workspace = true }
@@ -40,8 +34,6 @@ mime = { workspace = true }
nym-http-api-common = { path = "../http-api-common", default-features = false }
nym-bin-common = { path = "../bin-common" }
nym-network-defaults = { path = "../network-defaults", optional = true }
nym-http-api-client-macro = { path = "../http-api-client-macro" }
[target."cfg(not(target_arch = \"wasm32\"))".dependencies]
hickory-resolver = { workspace = true, features = ["https-ring", "tls-ring", "webpki-roots"] }
@@ -1,26 +0,0 @@
use nym_http_api_client::registry;
fn main() {
println!("Debugging HTTP Client Inventory");
println!("================================");
// Print all registered configurations
registry::debug_print_inventory();
// Also print the count
println!(
"\nTotal registered configs: {}",
registry::registered_config_count()
);
// Show the detailed breakdown
println!("\nDetailed configuration list:");
for (i, (priority, ptr)) in registry::inspect_registered_configs().iter().enumerate() {
println!(
" Config #{}: priority={}, function=0x{:x}",
i + 1,
priority,
ptr
);
}
}
@@ -1,99 +0,0 @@
use nym_http_api_client::registry;
use nym_http_api_client::{inventory, ReqwestClientBuilder};
use nym_http_api_client_macro::client_defaults;
use std::time::{Duration, Instant};
#[tokio::main]
async fn main() {
println!("Testing HTTP Client Timeout Configuration");
println!("==========================================\n");
client_defaults!(timeout = std::time::Duration::from_secs(300),);
// Build a client using the registry (should have 300s timeout)
let client = registry::build_client().expect("Failed to build client");
println!("Testing timeout behavior...");
println!("The inventory should have set timeout to 300 seconds");
// Test 1: Try a request to a slow endpoint that delays for 5 seconds
// This should succeed since timeout is 300s
println!("\nTest 1: Request with 5 second delay (should succeed)");
let start = Instant::now();
match client.get("https://httpbin.org/delay/5").send().await {
Ok(_) => {
let elapsed = start.elapsed();
println!("✓ Request succeeded after {:?}", elapsed);
}
Err(e) => {
let elapsed = start.elapsed();
if e.is_timeout() {
println!(
"✗ Request timed out after {:?} - timeout might be shorter than expected!",
elapsed
);
} else {
println!("✗ Request failed after {:?}: {}", elapsed, e);
}
}
}
// Test 2: Try to inspect the client's actual configuration
println!("\nTest 2: Client debug information");
println!("Client Debug: {:?}", client);
// Test 3: Create a client with explicit short timeout to compare behavior
println!("\nTest 3: Control test with 2 second timeout");
let short_timeout_client = reqwest::Client::builder()
.timeout(Duration::from_secs(2))
.build()
.expect("Failed to build short timeout client");
let start = Instant::now();
match short_timeout_client
.get("https://httpbin.org/delay/5")
.send()
.await
{
Ok(_) => {
let elapsed = start.elapsed();
println!(
"✗ Request succeeded after {:?} - timeout not working!",
elapsed
);
}
Err(e) => {
let elapsed = start.elapsed();
if e.is_timeout() {
println!("✓ Request timed out as expected after {:?}", elapsed);
} else {
println!("? Request failed after {:?}: {}", elapsed, e);
}
}
}
// Test 4: Create a client through the registry and verify timeout on a hanging connection
println!("\nTest 4: Testing with a connection that hangs");
println!("Making request to an endpoint that will hang...");
let start = Instant::now();
// This IP is reserved for documentation and will hang
match client.get("http://192.0.2.1:81").send().await {
Ok(_) => {
println!("✗ Request succeeded - unexpected!");
}
Err(e) => {
let elapsed = start.elapsed();
if e.is_timeout() {
println!("✓ Request timed out after {:?}", elapsed);
if elapsed < Duration::from_secs(290) {
println!(" Note: Timeout occurred faster than 300s, might be connection timeout not total timeout");
}
} else if e.is_connect() {
println!("✓ Connection failed after {:?} (connect timeout)", elapsed);
} else {
println!("? Request failed after {:?}: {}", elapsed, e);
}
}
}
}
+3 -7
View File
@@ -54,14 +54,10 @@ impl Front {
#[derive(Debug, Default, PartialEq, Clone)]
#[cfg(feature = "tunneling")]
/// Policy for when to use domain fronting for HTTP requests.
pub enum FrontPolicy {
/// Always use domain fronting for all requests.
Always,
/// Only use domain fronting when retrying failed requests.
OnRetry,
#[default]
/// Never use domain fronting.
Off,
}
@@ -100,14 +96,14 @@ mod tests {
// Some(vec!["https://cdn77.com"]),
// ).unwrap(); // cdn77
let client = ClientBuilder::new(url1)
let client = ClientBuilder::new::<_, &str>(url1)
.expect("bad url")
.with_fronting(FrontPolicy::Always)
.build()
.build::<&str>()
.expect("failed to build client");
let response = client
.send_request::<_, (), &str, &str>(
.send_request::<_, (), &str, &str, &str>(
reqwest::Method::GET,
&["api", "v1", "network", "details"],
NO_PARAMS,
File diff suppressed because it is too large Load Diff
-101
View File
@@ -1,101 +0,0 @@
//! Global registry for HTTP client configurations.
//!
//! This module provides a compile-time registry system that allows any crate
//! in the workspace to contribute configuration modifications to HTTP clients.
use crate::ReqwestClientBuilder;
/// A configuration record that modifies a `ReqwestClientBuilder`.
///
/// Records are collected at compile-time via the `inventory` crate and
/// applied in priority order when building HTTP clients.
pub struct ConfigRecord {
/// Lower numbers run earlier.
pub priority: i32,
/// A function that takes a builder and returns a mutated builder.
pub apply: fn(ReqwestClientBuilder) -> ReqwestClientBuilder,
}
inventory::collect!(ConfigRecord);
/// Returns the default builder with all registered configurations applied.
pub fn default_builder() -> ReqwestClientBuilder {
let mut b = ReqwestClientBuilder::new();
let mut records: Vec<&'static ConfigRecord> =
inventory::iter::<ConfigRecord>.into_iter().collect();
records.sort_by_key(|r| r.priority); // lower runs first
#[cfg(feature = "debug-inventory")]
{
eprintln!(
"[HTTP-INVENTORY] Building client with {} registered configurations",
records.len()
);
}
for r in records {
b = (r.apply)(b);
}
#[cfg(feature = "debug-inventory")]
{
eprintln!("[HTTP-INVENTORY] Final builder state (Debug):");
eprintln!("{:#?}", b);
eprintln!(
"[HTTP-INVENTORY] Note: reqwest::ClientBuilder doesn't expose all internal state"
);
eprintln!("[HTTP-INVENTORY] Building test client to verify configuration...");
// Try to build a client to see if it works
match b.try_clone().unwrap().build() {
Ok(client) => {
eprintln!("[HTTP-INVENTORY] ✓ Client built successfully");
eprintln!("[HTTP-INVENTORY] Client debug info: {:#?}", client);
}
Err(e) => {
eprintln!("[HTTP-INVENTORY] ✗ Failed to build client: {}", e);
}
}
}
b
}
/// Builds a client using the default builder with all registered configurations.
pub fn build_client() -> reqwest::Result<reqwest::Client> {
default_builder().build()
}
/// Debug function to inspect registered configurations.
/// Returns a vector of (priority, function_pointer) tuples for debugging.
pub fn inspect_registered_configs() -> Vec<(i32, usize)> {
let mut configs: Vec<(i32, usize)> = inventory::iter::<ConfigRecord>
.into_iter()
.map(|record| (record.priority, record.apply as usize))
.collect();
configs.sort_by_key(|(priority, _)| *priority);
configs
}
/// Print all registered configurations to stderr for debugging.
/// This shows the priority and function pointer address of each registered config.
pub fn debug_print_inventory() {
eprintln!("[HTTP-INVENTORY] Registered configurations:");
let configs = inspect_registered_configs();
if configs.is_empty() {
eprintln!(" (none)");
} else {
for (i, (priority, ptr)) in configs.iter().enumerate() {
eprintln!(
" [{:2}] Priority: {:4}, Function: 0x{:016x}",
i, priority, ptr
);
}
eprintln!(" Total: {} configurations", configs.len());
}
}
/// Returns the count of registered configuration records.
pub fn registered_config_count() -> usize {
inventory::iter::<ConfigRecord>.into_iter().count()
}
+9 -6
View File
@@ -95,10 +95,10 @@ async fn api_client_retry() -> Result<(), Box<dyn std::error::Error>> {
"http://example.com/".parse()?,
])
.with_retries(3)
.build()?;
.build::<HttpClientError>()?;
let req = client.create_get_request(&["/"], NO_PARAMS).unwrap();
let resp = client.send(req).await?;
let req = client.create_get_request(&["/"], NO_PARAMS);
let resp = client.send::<HttpClientError>(req).await?;
assert_eq!(resp.status(), 200);
@@ -111,7 +111,10 @@ async fn api_client_retry() -> Result<(), Box<dyn std::error::Error>> {
#[test]
fn host_updating() {
let url = Url::new("http://example.com", None).unwrap();
let mut client = ClientBuilder::new(url).unwrap().build().unwrap();
let mut client = ClientBuilder::new::<_, &str>(url)
.unwrap()
.build::<&str>()
.unwrap();
// check that the url is set correctly
let current_url = client.current_url();
@@ -168,10 +171,10 @@ fn host_updating() {
#[cfg(feature = "tunneling")]
fn fronted_host_updating() {
let url = Url::new("http://example.com", Some(vec!["http://front1.com"])).unwrap();
let mut client = ClientBuilder::new(url)
let mut client = ClientBuilder::new::<_, &str>(url)
.unwrap()
.with_fronting(crate::fronted::FrontPolicy::Always)
.build()
.build::<&str>()
.unwrap();
// check that the url is set correctly
@@ -1,70 +0,0 @@
use nym_http_api_client::registry;
// Create separate modules to avoid name conflicts
mod config_early {
use nym_http_api_client_macro::client_defaults;
client_defaults!(
priority = -200;
tcp_nodelay = true
);
}
mod config_late {
use nym_http_api_client_macro::client_defaults;
client_defaults!(
priority = 100;
pool_idle_timeout = std::time::Duration::from_secs(90)
);
}
#[test]
fn test_registry_collects_configs() {
// Verify that configurations are being registered
let count = registry::registered_config_count();
// Should have at least the ones we registered above plus the default from lib.rs
assert!(
count >= 3,
"Expected at least 3 registered configs, got {}",
count
);
}
#[test]
fn test_default_builder_applies_configs() {
// Test that default_builder returns a configured builder
let _builder = registry::default_builder();
// The builder should have all configurations applied
// We can't easily inspect the internals, but we verify it doesn't panic
}
#[test]
fn test_build_client_works() {
// Test that we can successfully build a client with all configurations
let result = registry::build_client();
assert!(result.is_ok(), "Failed to build client: {:?}", result.err());
}
#[cfg(debug_assertions)]
#[test]
fn test_inspect_configs() {
// In debug mode, test that we can inspect registered configurations
let configs = registry::inspect_registered_configs();
// Verify configs are sorted by priority
for window in configs.windows(2) {
assert!(window[0].0 <= window[1].0, "Configs not sorted by priority");
}
// Verify we have configs at different priority levels
let priorities: Vec<i32> = configs.iter().map(|(p, _)| *p).collect();
assert!(
priorities.iter().any(|&p| p < 0),
"Expected negative priority configs"
);
assert!(
priorities.iter().any(|&p| p >= 0),
"Expected non-negative priority configs"
);
}
+2 -9
View File
@@ -55,7 +55,6 @@ pub struct ApiUrl {
pub front_hosts: Option<Vec<String>>,
}
#[derive(Copy, Clone)]
pub struct ApiUrlConst<'a> {
pub url: &'a str,
pub front_hosts: Option<&'a [&'a str]>,
@@ -189,14 +188,8 @@ impl NymNetworkDetails {
),
},
nym_vpn_api_url: parse_optional_str(mainnet::NYM_VPN_API),
nym_api_urls: Some(mainnet::NYM_APIS.iter().copied().map(Into::into).collect()),
nym_vpn_api_urls: Some(
mainnet::NYM_VPN_APIS
.iter()
.copied()
.map(Into::into)
.collect(),
),
nym_api_urls: None,
nym_vpn_api_urls: None,
}
}
+7 -8
View File
@@ -9,7 +9,7 @@ use futures::StreamExt;
use nym_crypto::asymmetric::x25519;
use nym_sphinx::acknowledgements::AckKey;
use nym_sphinx::receiver::{MessageReceiver, SphinxMessageReceiver};
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use serde::de::DeserializeOwned;
use std::sync::Arc;
@@ -24,7 +24,7 @@ pub struct SimpleMessageReceiver<T, R: MessageReceiver = SphinxMessageReceiver>
acks_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
received_sender: ReceivedSender<T>,
shutdown: ShutdownToken,
shutdown: TaskClient,
}
impl<T> SimpleMessageReceiver<T, SphinxMessageReceiver> {
@@ -34,7 +34,7 @@ impl<T> SimpleMessageReceiver<T, SphinxMessageReceiver> {
mixnet_message_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
acks_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
received_sender: ReceivedSender<T>,
shutdown: ShutdownToken,
shutdown: TaskClient,
) -> Self {
Self::new(
local_encryption_keypair,
@@ -54,7 +54,7 @@ impl<T, R: MessageReceiver> SimpleMessageReceiver<T, R> {
mixnet_message_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
acks_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
received_sender: ReceivedSender<T>,
shutdown: ShutdownToken,
shutdown: TaskClient,
) -> Self {
SimpleMessageReceiver {
message_processor: TestPacketProcessor::new(local_encryption_keypair, ack_key),
@@ -91,12 +91,11 @@ impl<T, R: MessageReceiver> SimpleMessageReceiver<T, R> {
where
T: DeserializeOwned,
{
loop {
while !self.shutdown.is_shutdown() {
tokio::select! {
biased;
_ = self.shutdown.cancelled() => {
log_info!("SimpleMessageReceiver: received shutdown");
break
_ = self.shutdown.recv() => {
log_info!("SimpleMessageReceiver: received shutdown")
}
mixnet_messages = self.mixnet_message_receiver.next() => {
let Some(mixnet_messages) = mixnet_messages else {
+44 -32
View File
@@ -23,7 +23,9 @@ use nym_client_core::init::types::GatewaySetup;
use nym_credential_storage::storage::Storage as CredentialStorage;
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::params::PacketType;
use nym_task::{ShutdownManager, ShutdownTracker};
use nym_task::{TaskClient, TaskHandle, TaskStatus};
use anyhow::anyhow;
use nym_validator_client::UserAgent;
use std::error::Error;
use std::path::PathBuf;
@@ -44,7 +46,7 @@ pub enum Socks5ControlMessage {
pub struct StartedSocks5Client {
/// Handle for managing graceful shutdown of this client. If dropped, the client will be stopped.
pub shutdown_handle: ShutdownManager,
pub shutdown_handle: TaskHandle,
/// Address of the started client
pub address: Recipient,
@@ -63,8 +65,6 @@ pub struct NymClient<S> {
/// Optional path to a .json file containing standalone network details.
custom_mixnet: Option<PathBuf>,
shutdown_manager: ShutdownManager,
}
impl<S> NymClient<S>
@@ -92,7 +92,6 @@ where
setup_method: GatewaySetup::MustLoad { gateway_id: None },
user_agent,
custom_mixnet,
shutdown_manager: Default::default(),
}
}
@@ -109,7 +108,7 @@ where
client_output: ClientOutput,
client_status: ClientState,
self_address: Recipient,
shutdown: ShutdownTracker,
shutdown: TaskClient,
packet_type: PacketType,
) {
info!("Starting socks5 listener...");
@@ -149,39 +148,51 @@ where
socks5_config.send_anonymously,
socks5_config.socks5_debug,
),
shutdown,
shutdown.clone(),
packet_type,
);
nym_task::spawn_future(async move {
sphinx_socks
.serve(
input_sender,
received_buffer_request_sender,
connection_command_sender,
)
.await
});
nym_task::spawn_with_report_error(
async move {
sphinx_socks
.serve(
input_sender,
received_buffer_request_sender,
connection_command_sender,
)
.await
},
shutdown,
);
}
/// blocking version of `start` method. Will run forever (or until SIGINT is sent)
pub async fn run_forever(self) -> Result<(), Box<dyn Error + Send + Sync>> {
let mut started = self.start().await?;
let started = self.start().await?;
started.shutdown_handle.run_until_shutdown().await;
let res = started.shutdown_handle.wait_for_shutdown().await;
log::info!("Stopping nym-socks5-client");
Ok(())
res
}
// Variant of `run_forever` that listens for remote control messages
pub async fn run_and_listen(
self,
mut receiver: Socks5ControlMessageReceiver,
sender: nym_task::StatusSender,
) -> Result<(), Box<dyn Error + Send + Sync>> {
// Start the main task
let started = self.start().await?;
let mut task_manager = started.shutdown_handle;
let mut shutdown = started
.shutdown_handle
.try_into_task_manager()
.ok_or(anyhow!(
"attempted to use `run_and_listen` without owning shutdown handle"
))?;
let mut shutdown_signals = task_manager.detach_shutdown_signals();
// Listen to status messages from task, that we forward back to the caller
shutdown
.start_status_listener(sender, TaskStatus::Ready)
.await;
let res = tokio::select! {
biased;
@@ -196,20 +207,22 @@ where
}
}
Ok(())
},
_ = shutdown_signals.wait_for_signal() => {
log::info!("Received shutdown signal");
}
Some(msg) = shutdown.wait_for_error() => {
log::info!("Task error: {msg:?}");
Err(msg)
}
_ = tokio::signal::ctrl_c() => {
log::info!("Received SIGINT");
Ok(())
},
};
if !task_manager.is_cancelled() {
log::info!("Sending shutdown");
task_manager.send_cancellation();
}
log::info!("Sending shutdown");
shutdown.signal_shutdown().ok();
log::info!("Waiting for tasks to finish... (Press ctrl-c to force)");
task_manager.perform_shutdown().await;
shutdown.wait_for_shutdown().await;
log::info!("Stopping nym-socks5-client");
res
@@ -225,7 +238,6 @@ where
let mut base_builder =
BaseClientBuilder::new(self.config.base(), self.storage, dkg_query_client)
.with_shutdown(self.shutdown_manager.shutdown_tracker_owned())
.with_gateway_setup(self.setup_method)
.with_user_agent(self.user_agent);
@@ -249,7 +261,7 @@ where
client_output,
client_state,
self_address,
self.shutdown_manager.shutdown_tracker_owned(),
started_client.task_handle.get_handle(),
packet_type,
);
@@ -257,7 +269,7 @@ where
info!("The address of this client is: {self_address}");
Ok(StartedSocks5Client {
shutdown_handle: self.shutdown_manager,
shutdown_handle: started_client.task_handle,
address: self_address,
})
}
@@ -21,7 +21,7 @@ use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::params::PacketSize;
use nym_sphinx::params::PacketType;
use nym_task::connections::{LaneQueueLengths, TransmissionLane};
use nym_task::ShutdownTracker;
use nym_task::TaskClient;
use pin_project::pin_project;
use rand::RngCore;
use std::io;
@@ -185,7 +185,7 @@ pub(crate) struct SocksClient {
self_address: Recipient,
started_proxy: bool,
lane_queue_lengths: LaneQueueLengths,
shutdown_listener: ShutdownTracker,
shutdown_listener: TaskClient,
packet_type: Option<PacketType>,
}
@@ -214,9 +214,12 @@ impl SocksClient {
controller_sender: ControllerSender,
self_address: &Recipient,
lane_queue_lengths: LaneQueueLengths,
shutdown_listener: ShutdownTracker,
mut shutdown_listener: TaskClient,
packet_type: Option<PacketType>,
) -> Self {
// If this task fails and exits, we don't want to send shutdown signal
shutdown_listener.disarm();
let connection_id = Self::generate_random();
SocksClient {
@@ -291,6 +294,7 @@ impl SocksClient {
.shutdown()
.await
.map_err(|source| SocksProxyError::SocketShutdownFailure { source })?;
self.shutdown_listener.disarm();
Ok(())
}
@@ -13,13 +13,13 @@ use nym_service_providers_common::interface::{ControlResponse, ResponseContent};
use nym_socks5_proxy_helpers::connection_controller::{ControllerCommand, ControllerSender};
use nym_socks5_requests::{Socks5ProviderResponse, Socks5Response, Socks5ResponseContent};
use nym_sphinx::receiver::ReconstructedMessage;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
pub(crate) struct MixnetResponseListener {
buffer_requester: ReceivedBufferRequestSender,
mix_response_receiver: ReconstructedMessagesReceiver,
controller_sender: ControllerSender,
shutdown: ShutdownToken,
shutdown: TaskClient,
}
impl Drop for MixnetResponseListener {
@@ -28,7 +28,7 @@ impl Drop for MixnetResponseListener {
.buffer_requester
.unbounded_send(ReceivedBufferMessage::ReceiverDisconnect)
{
if self.shutdown.is_cancelled() {
if self.shutdown.is_shutdown_poll() {
log::debug!("The buffer request failed: {err}");
} else {
log::error!("The buffer request failed: {err}");
@@ -41,7 +41,7 @@ impl MixnetResponseListener {
pub(crate) fn new(
buffer_requester: ReceivedBufferRequestSender,
controller_sender: ControllerSender,
shutdown: ShutdownToken,
shutdown: TaskClient,
) -> Self {
let (mix_response_sender, mix_response_receiver) = mpsc::unbounded();
buffer_requester
@@ -130,18 +130,13 @@ impl MixnetResponseListener {
}
pub(crate) async fn run(&mut self) {
loop {
while !self.shutdown.is_shutdown() {
tokio::select! {
biased;
_ = self.shutdown.cancelled() => {
log::trace!("MixnetResponseListener: Received shutdown");
break;
}
received_responses = self.mix_response_receiver.next() => {
if let Some(received_responses) = received_responses {
for reconstructed_message in received_responses {
if let Err(err) = self.on_message(reconstructed_message) {
debug!("message handling error: {err}")
self.shutdown.send_status_msg(Box::new(err));
}
}
} else {
@@ -149,8 +144,12 @@ impl MixnetResponseListener {
break;
}
},
_ = self.shutdown.recv() => {
log::trace!("MixnetResponseListener: Received shutdown");
}
}
}
self.shutdown.recv_timeout().await;
log::debug!("MixnetResponseListener: Exiting");
}
}
+24 -31
View File
@@ -12,7 +12,7 @@ use nym_socks5_proxy_helpers::connection_controller::Controller;
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::params::PacketType;
use nym_task::connections::{ConnectionCommandSender, LaneQueueLengths};
use nym_task::ShutdownTracker;
use nym_task::TaskClient;
use std::net::SocketAddr;
use tap::TapFallible;
use tokio::net::TcpListener;
@@ -25,7 +25,7 @@ pub struct NymSocksServer {
self_address: Recipient,
client_config: client::Config,
lane_queue_lengths: LaneQueueLengths,
shutdown: ShutdownTracker,
shutdown: TaskClient,
packet_type: PacketType,
}
@@ -39,7 +39,7 @@ impl NymSocksServer {
self_address: Recipient,
lane_queue_lengths: LaneQueueLengths,
client_config: client::Config,
shutdown: ShutdownTracker,
shutdown: TaskClient,
packet_type: PacketType,
) -> Self {
info!("Listening on {bind_address}");
@@ -72,7 +72,7 @@ impl NymSocksServer {
let (mut active_streams_controller, controller_sender) = Controller::new(
client_connection_tx,
//BroadcastActiveConnections::Off,
self.shutdown.clone_shutdown_token(),
self.shutdown.clone(),
);
tokio::spawn(async move {
active_streams_controller.run().await;
@@ -82,30 +82,20 @@ impl NymSocksServer {
let mut mixnet_response_listener = MixnetResponseListener::new(
buffer_requester,
controller_sender.clone(),
self.shutdown.clone_shutdown_token(),
);
self.shutdown.try_spawn_named(
async move {
mixnet_response_listener.run().await;
},
"Socks5MixnetListener",
self.shutdown.clone(),
);
tokio::spawn(async move {
mixnet_response_listener.run().await;
});
// TODO:, if required, there should be another task here responsible for control requests.
// it should get `input_sender` to send actual requests into the mixnet
// and some channel that connects it from `MixnetResponseListener` to receive
// any control responses
let shutdown = self.shutdown.clone_shutdown_token();
loop {
tokio::select! {
biased;
_ = shutdown.cancelled() => {
log::trace!("NymSocksServer: Received shutdown");
log::debug!("NymSocksServer: Exiting");
return Ok(());
}
Ok((stream, remote)) = listener.accept() => {
Ok((stream, _remote)) = listener.accept() => {
let mut client = SocksClient::new(
self.client_config,
stream,
@@ -119,20 +109,23 @@ impl NymSocksServer {
Some(self.packet_type)
);
self.shutdown.try_spawn_named(
async move {
if let Err(err) = client.run().await {
error!("Error! {err}");
if client.send_error(err).await.is_err() {
warn!("Failed to send error code");
};
if client.shutdown().await.is_err() {
warn!("Failed to shutdown TcpStream");
};
tokio::spawn(async move {
if let Err(err) = client.run().await {
error!("Error! {err}");
if client.send_error(err).await.is_err() {
warn!("Failed to send error code");
};
}, &format!("Socks5Client::{remote}")
);
if client.shutdown().await.is_err() {
warn!("Failed to shutdown TcpStream");
};
}
});
},
_ = self.shutdown.recv() => {
log::trace!("NymSocksServer: Received shutdown");
log::debug!("NymSocksServer: Exiting");
return Ok(());
}
}
}
}
@@ -7,7 +7,7 @@ use log::*;
use nym_ordered_buffer::{OrderedMessageBuffer, ReadContiguousData};
use nym_socks5_requests::{ConnectionId, SocketData};
use nym_task::connections::{ConnectionCommand, ConnectionCommandSender};
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use std::collections::{HashMap, HashSet};
/// A generic message produced after reading from a socket/connection.
@@ -101,13 +101,13 @@ pub struct Controller {
// un-order messages. Note we don't ever expect to have more than 1-2 messages per connection here
pending_messages: HashMap<ConnectionId, Vec<SocketData>>,
shutdown: ShutdownToken,
shutdown: TaskClient,
}
impl Controller {
pub fn new(
client_connection_tx: ConnectionCommandSender,
shutdown: ShutdownToken,
shutdown: TaskClient,
) -> (Self, ControllerSender) {
let (sender, receiver) = mpsc::unbounded();
(
@@ -155,7 +155,7 @@ impl Controller {
.client_connection_tx
.unbounded_send(ConnectionCommand::Close(conn_id))
{
if self.shutdown.is_cancelled() {
if self.shutdown.is_shutdown_poll() {
log::debug!("Failed to send: {err}");
} else {
log::error!("Failed to send: {err}");
@@ -230,6 +230,7 @@ impl Controller {
},
}
}
self.shutdown.recv_timeout().await;
log::debug!("SOCKS5 Controller: Exiting");
}
}
@@ -11,7 +11,7 @@ use log::*;
use nym_socks5_requests::{ConnectionId, SocketData};
use nym_task::connections::LaneQueueLengths;
use nym_task::connections::TransmissionLane;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use std::sync::Arc;
use std::time::Duration;
use tokio::select;
@@ -81,7 +81,7 @@ pub(super) async fn run_inbound<F, S>(
available_plaintext_per_mix_packet: usize,
shutdown_notify: Arc<Notify>,
lane_queue_lengths: Option<LaneQueueLengths>,
shutdown_listener: ShutdownToken,
mut shutdown_listener: TaskClient,
) -> OwnedReadHalf
where
F: Fn(SocketData) -> S + Send + 'static,
@@ -129,7 +129,7 @@ where
message_sender.send_empty_close().await;
break;
}
_ = shutdown_listener.cancelled() => {
_ = shutdown_listener.recv() => {
log::trace!("ProxyRunner inbound: Received shutdown");
break;
}
@@ -171,5 +171,6 @@ where
trace!("{connection_id} - inbound closed");
shutdown_notify.notify_one();
shutdown_listener.disarm();
reader
}
@@ -5,7 +5,7 @@ use crate::connection_controller::ConnectionReceiver;
use crate::ordered_sender::OrderedMessageSender;
use nym_socks5_requests::{ConnectionId, SocketData};
use nym_task::connections::LaneQueueLengths;
use nym_task::ShutdownTracker;
use nym_task::TaskClient;
use std::fmt::Debug;
use std::{sync::Arc, time::Duration};
use tokio::{net::TcpStream, sync::Notify};
@@ -57,8 +57,7 @@ pub struct ProxyRunner<S> {
available_plaintext_per_mix_packet: usize,
// Listens to shutdown commands from higher up
// and spawn new tracked tasks
shutdown_tracker: ShutdownTracker,
shutdown_listener: TaskClient,
}
impl<S> ProxyRunner<S>
@@ -75,7 +74,7 @@ where
available_plaintext_per_mix_packet: usize,
connection_id: ConnectionId,
lane_queue_lengths: Option<LaneQueueLengths>,
shutdown_tracker: ShutdownTracker,
shutdown_listener: TaskClient,
) -> Self {
ProxyRunner {
mix_receiver: Some(mix_receiver),
@@ -86,7 +85,7 @@ where
connection_id,
lane_queue_lengths,
available_plaintext_per_mix_packet,
shutdown_tracker,
shutdown_listener,
}
}
@@ -114,7 +113,7 @@ where
self.available_plaintext_per_mix_packet,
Arc::clone(&shutdown_notify),
self.lane_queue_lengths.clone(),
self.shutdown_tracker.clone_shutdown_token(),
self.shutdown_listener.clone(),
);
let outbound_future = outbound::run_outbound(
@@ -124,26 +123,14 @@ where
self.mix_receiver.take().unwrap(),
self.connection_id,
shutdown_notify,
self.shutdown_tracker.clone_shutdown_token(),
self.shutdown_listener.clone(),
);
// TODO: this shouldn't really have to spawn tasks inside "library" code, but
// if we used join directly, stuff would have been executed on the same thread
// (it's not bad, but an unnecessary slowdown)
let handle_inbound = self.shutdown_tracker.try_spawn_named(
inbound_future,
&format!(
"Socks5Inbound::{}::{}",
self.remote_source_address, self.connection_id
),
);
let handle_outbound = self.shutdown_tracker.try_spawn_named(
outbound_future,
&format!(
"Socks5Outbound::{}::{}",
self.remote_source_address, self.connection_id
),
);
let handle_inbound = tokio::spawn(inbound_future);
let handle_outbound = tokio::spawn(outbound_future);
let (inbound_result, outbound_result) =
futures::future::join(handle_inbound, handle_outbound).await;
@@ -161,6 +148,7 @@ where
}
pub fn into_inner(mut self) -> (TcpStream, ConnectionReceiver) {
self.shutdown_listener.disarm();
(
self.socket.take().unwrap(),
self.mix_receiver.take().unwrap(),
@@ -7,7 +7,7 @@ use futures::FutureExt;
use futures::StreamExt;
use log::*;
use nym_socks5_requests::ConnectionId;
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use std::{sync::Arc, time::Duration};
use tokio::io::AsyncWriteExt;
use tokio::select;
@@ -51,7 +51,7 @@ pub(super) async fn run_outbound(
mut mix_receiver: ConnectionReceiver,
connection_id: ConnectionId,
shutdown_notify: Arc<Notify>,
shutdown_listener: ShutdownToken,
mut shutdown_listener: TaskClient,
) -> (OwnedWriteHalf, ConnectionReceiver) {
let shutdown_future = shutdown_notify.notified().then(|_| sleep(SHUTDOWN_TIMEOUT));
tokio::pin!(shutdown_future);
@@ -60,11 +60,6 @@ pub(super) async fn run_outbound(
loop {
select! {
biased;
_ = shutdown_listener.cancelled() => {
log::trace!("ProxyRunner outbound: Received shutdown");
break;
}
connection_message = mix_receiver.next() => {
if let Some(connection_message) = connection_message {
if deal_with_message(connection_message, &mut writer, &local_destination_address, &remote_source_address, connection_id).await {
@@ -85,11 +80,16 @@ pub(super) async fn run_outbound(
debug!("closing outbound proxy after inbound was closed {SHUTDOWN_TIMEOUT:?} ago");
break;
}
_ = shutdown_listener.recv() => {
log::trace!("ProxyRunner outbound: Received shutdown");
break;
}
}
}
trace!("{connection_id} - outbound closed");
shutdown_notify.notify_one();
shutdown_listener.disarm();
(writer, mix_receiver)
}
+7 -7
View File
@@ -3,7 +3,7 @@
use crate::report::client::{ClientStatsReport, OsInformation};
use nym_task::ShutdownToken;
use nym_task::TaskClient;
use time::{OffsetDateTime, Time};
use tokio::sync::mpsc::UnboundedSender;
@@ -25,18 +25,18 @@ pub type ClientStatsReceiver = tokio::sync::mpsc::UnboundedReceiver<ClientStatsE
#[derive(Clone)]
pub struct ClientStatsSender {
stats_tx: Option<UnboundedSender<ClientStatsEvents>>,
shutdown_token: ShutdownToken,
task_client: TaskClient,
}
impl ClientStatsSender {
/// Create a new statistics Sender
pub fn new(
stats_tx: Option<UnboundedSender<ClientStatsEvents>>,
shutdown_token: ShutdownToken,
task_client: TaskClient,
) -> Self {
ClientStatsSender {
stats_tx,
shutdown_token,
task_client,
}
}
@@ -44,7 +44,7 @@ impl ClientStatsSender {
pub fn report(&self, event: ClientStatsEvents) {
if let Some(tx) = &self.stats_tx {
if let Err(err) = tx.send(event) {
if !self.shutdown_token.is_cancelled() {
if !self.task_client.is_shutdown_poll() {
log::error!("Failed to send stats event: {err}");
}
}
@@ -137,8 +137,8 @@ impl ClientStatsController {
self.packet_stats.snapshot();
}
pub fn local_report(&mut self) {
self.packet_stats.local_report();
pub fn local_report(&mut self, task_client: &mut TaskClient) {
self.packet_stats.local_report(task_client);
self.gateway_conn_stats.local_report();
self.nym_api_stats.local_report();
}
@@ -449,16 +449,15 @@ impl PacketStatisticsControl {
self.stats.clone()
}
pub(crate) fn local_report(&mut self) {
let _rates = self.report_rates();
pub(crate) fn local_report(&mut self, task_client: &mut nym_task::TaskClient) {
let rates = self.report_rates();
self.check_for_notable_events();
self.report_counters();
// leave the code commented in case somebody wanted to restore this logic with a different channel
// // Report our current bandwidth used to e.g a GUI client
// if let Some(rates) = rates {
// task_client.send_status_msg(Box::new(MixnetBandwidthStatisticsEvent::new(rates)));
// }
// Report our current bandwidth used to e.g a GUI client
if let Some(rates) = rates {
task_client.send_status_msg(Box::new(MixnetBandwidthStatisticsEvent::new(rates)));
}
}
// Add the current stats to the history, and remove old ones.
-8
View File
@@ -30,13 +30,5 @@ workspace = true
workspace = true
features = ["tokio"]
[features]
tokio-tracing = ["tokio/tracing"]
[dev-dependencies]
anyhow = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread", "net", "signal", "test-util", "macros"] }
nym-test-utils = { path = "../test-utils" }
[lints]
workspace = true
+414
View File
@@ -0,0 +1,414 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::{TaskClient, TaskManager};
use futures::stream::FuturesUnordered;
use futures::StreamExt;
use std::future::Future;
use std::mem;
use std::ops::Deref;
use std::pin::Pin;
use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::sleep;
use tokio_util::sync::{CancellationToken, DropGuard};
use tokio_util::task::TaskTracker;
use tracing::{debug, info, trace};
#[cfg(unix)]
use tokio::signal::unix::{signal, SignalKind};
pub const DEFAULT_MAX_SHUTDOWN_DURATION: Duration = Duration::from_secs(5);
pub fn token_name(name: &Option<String>) -> String {
name.clone().unwrap_or_else(|| "unknown".to_string())
}
// a wrapper around tokio's CancellationToken that adds optional `name` information to more easily
// track down sources of shutdown
#[derive(Debug, Default)]
pub struct ShutdownToken {
name: Option<String>,
inner: CancellationToken,
}
impl Clone for ShutdownToken {
fn clone(&self) -> Self {
// make sure to not accidentally overflow the stack if we keep cloning the handle
let name = if let Some(name) = &self.name {
if name != Self::OVERFLOW_NAME && name.len() < Self::MAX_NAME_LENGTH {
Some(format!("{name}-child"))
} else {
Some(Self::OVERFLOW_NAME.to_string())
}
} else {
None
};
ShutdownToken {
name,
inner: self.inner.clone(),
}
}
}
impl Deref for ShutdownToken {
type Target = CancellationToken;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl ShutdownToken {
const MAX_NAME_LENGTH: usize = 128;
const OVERFLOW_NAME: &'static str = "reached maximum ShutdownToken children name depth";
pub fn new(name: impl Into<String>) -> Self {
ShutdownToken {
name: Some(name.into()),
inner: CancellationToken::new(),
}
}
pub fn ephemeral() -> Self {
ShutdownToken::new("ephemeral-token")
}
// Creates a ShutdownToken which will get cancelled whenever the current token gets cancelled.
// Unlike a cloned/forked ShutdownToken, cancelling a child token does not cancel the parent token.
#[must_use]
pub fn child_token<S: Into<String>>(&self, child_suffix: S) -> Self {
let suffix = child_suffix.into();
let child_name = if let Some(base) = &self.name {
format!("{base}-{suffix}")
} else {
format!("unknown-{suffix}")
};
ShutdownToken {
name: Some(child_name),
inner: self.inner.child_token(),
}
}
// Creates a clone of the ShutdownToken which will get cancelled whenever the current token gets cancelled, and vice versa.
#[must_use]
pub fn clone_with_suffix<S: Into<String>>(&self, child_suffix: S) -> Self {
let mut child = self.clone();
let suffix = child_suffix.into();
let child_name = if let Some(base) = &self.name {
format!("{base}-{suffix}")
} else {
format!("unknown-{suffix}")
};
child.name = Some(child_name);
child
}
// exposed method with the old name for easier migration
// it will eventually be removed so please try to use `.clone_with_suffix` instead
#[must_use]
#[deprecated(note = "use .clone_with_suffix instead")]
pub fn fork<S: Into<String>>(&self, child_suffix: S) -> Self {
self.clone_with_suffix(child_suffix)
}
// exposed method with the old name for easier migration
// it will eventually be removed so please try to use `.clone().named(name)` instead
#[must_use]
#[deprecated(note = "use .clone().named(name) instead")]
pub fn fork_named<S: Into<String>>(&self, name: S) -> Self {
self.clone().named(name)
}
#[must_use]
pub fn named<S: Into<String>>(mut self, name: S) -> Self {
self.name = Some(name.into());
self
}
#[must_use]
pub fn add_suffix<S: Into<String>>(self, suffix: S) -> Self {
let suffix = suffix.into();
let name = if let Some(base) = &self.name {
format!("{base}-{suffix}")
} else {
format!("unknown-{suffix}")
};
self.named(name)
}
// Returned guard will cancel this token (and all its children) on drop unless disarmed.
pub fn drop_guard(self) -> ShutdownDropGuard {
ShutdownDropGuard {
name: self.name,
inner: self.inner.drop_guard(),
}
}
pub fn name(&self) -> String {
token_name(&self.name)
}
pub async fn run_until_cancelled<F>(&self, fut: F) -> Option<F::Output>
where
F: Future,
{
let res = self.inner.run_until_cancelled(fut).await;
trace!("'{}' got cancelled", self.name());
res
}
}
pub struct ShutdownDropGuard {
name: Option<String>,
inner: DropGuard,
}
impl Deref for ShutdownDropGuard {
type Target = DropGuard;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl ShutdownDropGuard {
pub fn disarm(self) -> ShutdownToken {
ShutdownToken {
name: self.name,
inner: self.inner.disarm(),
}
}
pub fn name(&self) -> String {
token_name(&self.name)
}
}
#[derive(Default)]
pub struct ShutdownSignals(JoinSet<()>);
impl ShutdownSignals {
pub async fn wait_for_signal(&mut self) {
self.0.join_next().await;
}
}
pub struct ShutdownManager {
pub root_token: ShutdownToken,
legacy_task_manager: Option<TaskManager>,
shutdown_signals: ShutdownSignals,
// the reason I'm not using a `JoinSet` is because it forces us to use futures with the same `::Output` type
tracker: TaskTracker,
max_shutdown_duration: Duration,
}
impl Deref for ShutdownManager {
type Target = TaskTracker;
fn deref(&self) -> &Self::Target {
&self.tracker
}
}
impl ShutdownManager {
pub fn new(root_token_name: impl Into<String>) -> Self {
let manager = ShutdownManager {
root_token: ShutdownToken::new(root_token_name),
legacy_task_manager: None,
shutdown_signals: Default::default(),
tracker: Default::default(),
max_shutdown_duration: Duration::from_secs(10),
};
// we need to add an explicit watcher for the cancellation token being cancelled
// so that we could cancel all legacy tasks
let cancel_watcher = manager.root_token.clone();
manager.with_shutdown(async move { cancel_watcher.cancelled().await })
}
pub fn empty_mock() -> Self {
ShutdownManager {
root_token: ShutdownToken::ephemeral(),
legacy_task_manager: None,
shutdown_signals: Default::default(),
tracker: Default::default(),
max_shutdown_duration: Default::default(),
}
}
pub fn with_legacy_task_manager(mut self) -> Self {
let mut legacy_manager =
TaskManager::default().named(format!("{}-legacy", self.root_token.name()));
let mut legacy_error_rx = legacy_manager.task_return_error_rx();
let mut legacy_drop_rx = legacy_manager.task_drop_rx();
self.legacy_task_manager = Some(legacy_manager);
// add a task that listens for legacy task clients being dropped to trigger cancellation
self.with_shutdown(async move {
tokio::select! {
_ = legacy_error_rx.recv() => (),
_ = legacy_drop_rx.recv() => (),
}
info!("received legacy shutdown signal");
})
}
#[cfg(not(target_arch = "wasm32"))]
pub fn with_default_shutdown_signals(self) -> std::io::Result<Self> {
cfg_if::cfg_if! {
if #[cfg(unix)] {
self.with_interrupt_signal()
.with_terminate_signal()?
.with_quit_signal()
} else {
Ok(self.with_interrupt_signal())
}
}
}
#[must_use]
#[track_caller]
pub fn with_shutdown<F>(mut self, shutdown: F) -> Self
where
F: Future<Output = ()>,
F: Send + 'static,
{
let shutdown_token = self.root_token.clone();
self.shutdown_signals.0.spawn(async move {
shutdown.await;
info!("sending cancellation after receiving shutdown signal");
shutdown_token.cancel();
});
self
}
#[cfg(unix)]
#[track_caller]
pub fn with_shutdown_signal(self, signal_kind: SignalKind) -> std::io::Result<Self> {
let mut sig = signal(signal_kind)?;
Ok(self.with_shutdown(async move {
sig.recv().await;
}))
}
#[cfg(not(target_arch = "wasm32"))]
#[track_caller]
pub fn with_interrupt_signal(self) -> Self {
self.with_shutdown(async move {
let _ = tokio::signal::ctrl_c().await;
})
}
#[cfg(unix)]
#[track_caller]
pub fn with_terminate_signal(self) -> std::io::Result<Self> {
self.with_shutdown_signal(SignalKind::terminate())
}
#[cfg(unix)]
#[track_caller]
pub fn with_quit_signal(self) -> std::io::Result<Self> {
self.with_shutdown_signal(SignalKind::quit())
}
#[must_use]
pub fn with_shutdown_duration(mut self, duration: Duration) -> Self {
self.max_shutdown_duration = duration;
self
}
pub fn child_token<S: Into<String>>(&self, child_suffix: S) -> ShutdownToken {
self.root_token.child_token(child_suffix)
}
pub fn clone_token<S: Into<String>>(&self, child_suffix: S) -> ShutdownToken {
self.root_token.clone_with_suffix(child_suffix)
}
#[must_use]
pub fn subscribe_legacy<S: Into<String>>(&self, child_suffix: S) -> TaskClient {
// alternatively we could have set self.legacy_task_manager = Some(TaskManager::default());
// on demand if it wasn't unavailable, but then we'd have to use mutable reference
#[allow(clippy::expect_used)]
self.legacy_task_manager
.as_ref()
.expect("did not enable legacy shutdown support")
.subscribe_named(child_suffix)
}
async fn finish_shutdown(mut self) {
let mut wait_futures = FuturesUnordered::<Pin<Box<dyn Future<Output = ()>>>>::new();
// force shutdown via ctrl-c
wait_futures.push(Box::pin(async move {
#[cfg(not(target_arch = "wasm32"))]
let interrupt_future = tokio::signal::ctrl_c();
#[cfg(target_arch = "wasm32")]
let interrupt_future = futures::future::pending::<()>();
let _ = interrupt_future.await;
info!("received interrupt - forcing shutdown");
}));
// timeout
wait_futures.push(Box::pin(async move {
sleep(self.max_shutdown_duration).await;
info!("timeout reached, forcing shutdown");
}));
// graceful
wait_futures.push(Box::pin(async move {
self.tracker.wait().await;
debug!("migrated tasks successfully shutdown");
if let Some(legacy) = self.legacy_task_manager.as_mut() {
legacy.wait_for_graceful_shutdown().await;
debug!("legacy tasks successfully shutdown");
}
info!("all registered tasks successfully shutdown")
}));
wait_futures.next().await;
}
pub fn detach_shutdown_signals(&mut self) -> ShutdownSignals {
mem::take(&mut self.shutdown_signals)
}
pub fn replace_shutdown_signals(&mut self, signals: ShutdownSignals) {
self.shutdown_signals = signals;
}
// cancellation safe
pub async fn wait_for_shutdown_signal(&mut self) {
self.shutdown_signals.0.join_next().await;
}
pub async fn perform_shutdown(mut self) {
if let Some(legacy_manager) = self.legacy_task_manager.as_mut() {
info!("attempting to shutdown legacy tasks");
let _ = legacy_manager.signal_shutdown();
}
info!("waiting for tasks to finish... (press ctrl-c to force)");
self.finish_shutdown().await;
}
pub async fn run_until_shutdown(mut self) {
self.wait_for_shutdown_signal().await;
self.perform_shutdown().await;
}
}
-744
View File
@@ -1,744 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::cancellation::tracker::{Cancelled, ShutdownTracker};
use crate::spawn::JoinHandle;
use crate::ShutdownToken;
use futures::stream::FuturesUnordered;
use futures::StreamExt;
use log::error;
use std::future::Future;
use std::mem;
use std::pin::Pin;
use std::time::Duration;
use tracing::info;
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::sleep;
#[cfg(target_arch = "wasm32")]
use wasmtimer::tokio::sleep;
#[cfg(unix)]
use tokio::signal::unix::{signal, SignalKind};
use tokio::task::JoinSet;
/// A top level structure responsible for controlling process shutdown by listening to
/// the underlying registered signals and issuing cancellation to tasks derived from its root cancellation token.
#[allow(deprecated)]
pub struct ShutdownManager {
/// Optional reference to the legacy [TaskManager](crate::TaskManager) to allow easier
/// transition to the new system.
pub(crate) legacy_task_manager: Option<crate::TaskManager>,
/// Registered [ShutdownSignals](ShutdownSignals) that will trigger process shutdown if detected.
pub(crate) shutdown_signals: ShutdownSignals,
/// Combined [TaskTracker](tokio_util::task::TaskTracker) and [ShutdownToken](ShutdownToken)
/// for spawning and tracking tasks associated with this ShutdownManager.
pub(crate) tracker: ShutdownTracker,
/// The maximum shutdown duration when tracked tasks could gracefully exit
/// before forcing the shutdown.
pub(crate) max_shutdown_duration: Duration,
}
/// Wrapper behind futures that upon completion will trigger binary shutdown.
#[derive(Default)]
pub struct ShutdownSignals(JoinSet<()>);
impl ShutdownSignals {
/// Wait for any of the registered signals to be ready
pub async fn wait_for_signal(&mut self) {
self.0.join_next().await;
}
}
// note: default implementation will ONLY listen for SIGINT and will ignore SIGTERM and SIGQUIT
// this is due to result type when registering the signal
#[cfg(not(target_arch = "wasm32"))]
impl Default for ShutdownManager {
fn default() -> Self {
ShutdownManager::new_without_signals()
.with_interrupt_signal()
.with_cancel_on_panic()
}
}
#[cfg(not(target_arch = "wasm32"))]
impl ShutdownManager {
/// Create new instance of ShutdownManager with the most sensible defaults, so that:
/// - shutdown will be triggered upon either SIGINT, SIGTERM (unix only) or SIGQUIT (unix only) being sent
/// - shutdown will be triggered upon any task panicking
pub fn build_new_default() -> std::io::Result<Self> {
Ok(ShutdownManager::new_without_signals()
.with_default_shutdown_signals()?
.with_cancel_on_panic())
}
/// Register a new shutdown signal that upon completion will trigger system shutdown.
#[must_use]
#[track_caller]
pub fn with_shutdown<F>(mut self, shutdown: F) -> Self
where
F: Future<Output = ()>,
F: Send + 'static,
{
let shutdown_token = self.tracker.clone_shutdown_token();
self.shutdown_signals.0.spawn(async move {
shutdown.await;
info!("sending cancellation after receiving shutdown signal");
shutdown_token.cancel();
});
self
}
/// Include support for the legacy [TaskManager](TaskManager) to this instance of the ShutdownManager.
/// This will allow issuing [TaskClient](TaskClient) for tasks that still require them.
#[allow(deprecated)]
pub fn with_legacy_task_manager(mut self) -> Self {
let mut legacy_manager = crate::TaskManager::default().named("legacy-task-manager");
let mut legacy_error_rx = legacy_manager.task_return_error_rx();
let mut legacy_drop_rx = legacy_manager.task_drop_rx();
self.legacy_task_manager = Some(legacy_manager);
// add a task that listens for legacy task clients being dropped to trigger cancellation
self.with_shutdown(async move {
tokio::select! {
_ = legacy_error_rx.recv() => (),
_ = legacy_drop_rx.recv() => (),
}
info!("received legacy shutdown signal");
})
}
/// Add the specified signal to the currently registered shutdown signals that will trigger
/// cancellation of all registered tasks.
#[cfg(unix)]
#[track_caller]
pub fn with_shutdown_signal(self, signal_kind: SignalKind) -> std::io::Result<Self> {
let mut sig = signal(signal_kind)?;
Ok(self.with_shutdown(async move {
sig.recv().await;
}))
}
/// Add the SIGTERM signal to the currently registered shutdown signals that will trigger
/// cancellation of all registered tasks.
#[cfg(unix)]
#[track_caller]
pub fn with_terminate_signal(self) -> std::io::Result<Self> {
self.with_shutdown_signal(SignalKind::terminate())
}
/// Add the SIGQUIT signal to the currently registered shutdown signals that will trigger
/// cancellation of all registered tasks.
#[cfg(unix)]
#[track_caller]
pub fn with_quit_signal(self) -> std::io::Result<Self> {
self.with_shutdown_signal(SignalKind::quit())
}
/// Add default signals to the set of the currently registered shutdown signals that will trigger
/// cancellation of all registered tasks.
/// This includes SIGINT, SIGTERM and SIGQUIT for unix-based platforms and SIGINT for other targets (such as windows)/
pub fn with_default_shutdown_signals(self) -> std::io::Result<Self> {
cfg_if::cfg_if! {
if #[cfg(unix)] {
self.with_interrupt_signal()
.with_terminate_signal()?
.with_quit_signal()
} else {
Ok(self.with_interrupt_signal())
}
}
}
/// Add the SIGINT (ctrl-c) signal to the currently registered shutdown signals that will trigger
/// cancellation of all registered tasks.
#[track_caller]
pub fn with_interrupt_signal(self) -> Self {
self.with_shutdown(async move {
let _ = tokio::signal::ctrl_c().await;
})
}
/// Spawn the provided future on the current Tokio runtime, and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn<F>(&self, task: F) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
self.tracker.spawn(task)
}
/// Spawn the provided future on the current Tokio runtime,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
/// Furthermore, attach a name to the spawned task to more easily track it within a [tokio console](https://github.com/tokio-rs/console)
///
/// Note that is no different from [spawn](Self::spawn) if the underlying binary
/// has not been built with `RUSTFLAGS="--cfg tokio_unstable"` and `--features="tokio-tracing"`
#[track_caller]
pub fn try_spawn_named<F>(&self, task: F, name: &str) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
self.tracker.try_spawn_named(task, name)
}
/// Spawn the provided future on the provided Tokio runtime,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_on<F>(&self, task: F, handle: &tokio::runtime::Handle) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
self.tracker.spawn_on(task, handle)
}
/// Spawn the provided future on the current [LocalSet](tokio::task::LocalSet),
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_local<F>(&self, task: F) -> JoinHandle<F::Output>
where
F: Future + 'static,
F::Output: 'static,
{
self.tracker.spawn_local(task)
}
/// Spawn the provided blocking task on the current Tokio runtime,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_blocking<F, T>(&self, task: F) -> JoinHandle<T>
where
F: FnOnce() -> T,
F: Send + 'static,
T: Send + 'static,
{
self.tracker.spawn_blocking(task)
}
/// Spawn the provided blocking task on the provided Tokio runtime,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_blocking_on<F, T>(&self, task: F, handle: &tokio::runtime::Handle) -> JoinHandle<T>
where
F: FnOnce() -> T,
F: Send + 'static,
T: Send + 'static,
{
self.tracker.spawn_blocking_on(task, handle)
}
/// Spawn the provided future on the current Tokio runtime
/// that will get cancelled once a global shutdown signal is detected,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
///
/// Note that to fully use the naming feature, such as tracking within a [tokio console](https://github.com/tokio-rs/console),
/// the underlying binary has to be built with `RUSTFLAGS="--cfg tokio_unstable"` and `--features="tokio-tracing"`
#[track_caller]
pub fn try_spawn_named_with_shutdown<F>(
&self,
task: F,
name: &str,
) -> JoinHandle<Result<F::Output, Cancelled>>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
self.tracker.try_spawn_named_with_shutdown(task, name)
}
/// Spawn the provided future on the current Tokio runtime
/// that will get cancelled once a global shutdown signal is detected,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_with_shutdown<F>(&self, task: F) -> JoinHandle<Result<F::Output, Cancelled>>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
self.tracker.spawn_with_shutdown(task)
}
}
#[cfg(target_arch = "wasm32")]
impl ShutdownManager {
/// Run the provided future on the current thread, and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn<F>(&self, task: F) -> JoinHandle<F::Output>
where
F: Future + 'static,
{
self.tracker.spawn(task)
}
/// Run the provided future on the current thread, and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
/// It has exactly the same behaviour as [spawn](Self::spawn) and it only exists to provide
/// the same interface as non-wasm32 targets.
#[track_caller]
pub fn try_spawn_named<F>(&self, task: F, name: &str) -> JoinHandle<F::Output>
where
F: Future + 'static,
{
self.tracker.try_spawn_named(task, name)
}
/// Run the provided future on the current thread
/// that will get cancelled once a global shutdown signal is detected,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
/// It has exactly the same behaviour as [spawn_with_shutdown](Self::spawn_with_shutdown) and it only exists to provide
/// the same interface as non-wasm32 targets.
#[track_caller]
pub fn try_spawn_named_with_shutdown<F>(
&self,
task: F,
name: &str,
) -> JoinHandle<Result<F::Output, Cancelled>>
where
F: Future<Output = ()> + Send + 'static,
{
self.tracker.try_spawn_named_with_shutdown(task, name)
}
/// Run the provided future on the current thread
/// that will get cancelled once a global shutdown signal is detected,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_with_shutdown<F>(&self, task: F) -> JoinHandle<Result<F::Output, Cancelled>>
where
F: Future<Output = ()> + Send + 'static,
{
self.tracker.spawn_with_shutdown(task)
}
}
impl ShutdownManager {
/// Create new instance of ShutdownManager without any external shutdown signals registered,
/// meaning it will only attempt to wait for all tasks spawned on its tracker to gracefully finish execution.
pub fn new_without_signals() -> Self {
Self::new_from_external_shutdown_token(ShutdownToken::new())
}
/// Create new instance of the ShutdownManager using an external shutdown token.
///
/// Note: it will not listen to any external shutdown signals!
/// You might want further customise it with [shutdown signals](Self::with_shutdown)
/// (or just use [the default set](Self::with_default_shutdown_signals).
/// Similarly, you might want to include [cancellation on panic](Self::with_cancel_on_panic)
/// to make sure everything gets cancelled if one of the tasks panics.
pub fn new_from_external_shutdown_token(shutdown_token: ShutdownToken) -> Self {
let manager = ShutdownManager {
legacy_task_manager: None,
shutdown_signals: Default::default(),
tracker: ShutdownTracker::new_from_external_shutdown_token(shutdown_token),
max_shutdown_duration: Duration::from_secs(10),
};
// we need to add an explicit watcher for the cancellation token being cancelled
// so that we could cancel all legacy tasks
cfg_if::cfg_if! {if #[cfg(not(target_arch = "wasm32"))] {
let cancel_watcher = manager.tracker.clone_shutdown_token();
manager.with_shutdown(async move { cancel_watcher.cancelled().await })
} else {
manager
}}
}
/// Create an empty testing mock of the ShutdownManager with no signals registered.
pub fn empty_mock() -> Self {
ShutdownManager {
legacy_task_manager: None,
shutdown_signals: Default::default(),
tracker: Default::default(),
max_shutdown_duration: Default::default(),
}
}
/// Add additional panic hook such that upon triggering, the root [ShutdownToken](ShutdownToken) gets cancelled.
/// Note: an unfortunate limitation of this is that graceful shutdown will no longer be possible
/// since that task that has panicked will not exit and thus all shutdowns will have to be either forced
/// or will have to time out.
#[must_use]
pub fn with_cancel_on_panic(self) -> Self {
let current_hook = std::panic::take_hook();
let shutdown_token = self.clone_shutdown_token();
std::panic::set_hook(Box::new(move |panic_info| {
// 1. call existing hook
current_hook(panic_info);
let location = panic_info
.location()
.map(|l| l.to_string())
.unwrap_or_else(|| "<unknown>".to_string());
let payload = if let Some(payload) = panic_info.payload().downcast_ref::<&str>() {
payload
} else {
""
};
// 2. issue cancellation
error!("panicked at {location}: {payload}. issuing global cancellation");
shutdown_token.cancel();
}));
self
}
/// Change the maximum shutdown duration when tracked tasks could gracefully exit
/// before forcing the shutdown.
#[must_use]
pub fn with_shutdown_duration(mut self, duration: Duration) -> Self {
self.max_shutdown_duration = duration;
self
}
/// Returns true if the root [ShutdownToken](ShutdownToken) has been cancelled.
pub fn is_cancelled(&self) -> bool {
self.tracker.root_cancellation_token.is_cancelled()
}
/// Get a reference to the used [ShutdownTracker](ShutdownTracker)
pub fn shutdown_tracker(&self) -> &ShutdownTracker {
&self.tracker
}
/// Get a cloned instance of the used [ShutdownTracker](ShutdownTracker)
pub fn shutdown_tracker_owned(&self) -> ShutdownTracker {
self.tracker.clone()
}
/// Waits until the underlying [TaskTracker](tokio_util::task::TaskTracker) is both closed and empty.
///
/// If the underlying [TaskTracker](tokio_util::task::TaskTracker) is already closed and empty when this method is called, then it
/// returns immediately.
pub async fn wait_for_tracker(&self) {
self.tracker.wait_for_tracker().await;
}
/// Close the underlying [TaskTracker](tokio_util::task::TaskTracker).
///
/// This allows [`wait_for_tracker`] futures to complete. It does not prevent you from spawning new tasks.
///
/// Returns `true` if this closed the underlying [TaskTracker](tokio_util::task::TaskTracker), or `false` if it was already closed.
///
/// [`wait_for_tracker`]: ShutdownTracker::wait_for_tracker
pub fn close_tracker(&self) -> bool {
self.tracker.close_tracker()
}
/// Reopen the underlying [TaskTracker](tokio_util::task::TaskTracker).
///
/// This prevents [`wait_for_tracker`] futures from completing even if the underlying [TaskTracker](tokio_util::task::TaskTracker) is empty.
///
/// Returns `true` if this reopened the underlying [TaskTracker](tokio_util::task::TaskTracker), or `false` if it was already open.
///
/// [`wait_for_tracker`]: ShutdownTracker::wait_for_tracker
pub fn reopen_tracker(&self) -> bool {
self.tracker.reopen_tracker()
}
/// Returns `true` if the underlying [TaskTracker](tokio_util::task::TaskTracker) is [closed](Self::close_tracker).
pub fn is_tracker_closed(&self) -> bool {
self.tracker.is_tracker_closed()
}
/// Returns the number of tasks tracked by the underlying [TaskTracker](tokio_util::task::TaskTracker).
pub fn tracked_tasks(&self) -> usize {
self.tracker.tracked_tasks()
}
/// Returns `true` if there are no tasks in the underlying [TaskTracker](tokio_util::task::TaskTracker).
pub fn is_tracker_empty(&self) -> bool {
self.tracker.is_tracker_empty()
}
/// Obtain a [ShutdownToken](crate::cancellation::ShutdownToken) that is a child of the root token
pub fn child_shutdown_token(&self) -> ShutdownToken {
self.tracker.root_cancellation_token.child_token()
}
/// Obtain a [ShutdownToken](crate::cancellation::ShutdownToken) on the same hierarchical structure as the root token
pub fn clone_shutdown_token(&self) -> ShutdownToken {
self.tracker.root_cancellation_token.clone()
}
/// Attempt to create a handle to a legacy [TaskClient] to support tasks that hasn't migrated
/// from the legacy [TaskManager].
/// Note. To use this method [ShutdownManager] must be built with `.with_legacy_task_manager()`
#[must_use]
#[deprecated]
#[allow(deprecated)]
pub fn subscribe_legacy<S: Into<String>>(&self, child_suffix: S) -> crate::TaskClient {
// alternatively we could have set self.legacy_task_manager = Some(TaskManager::default());
// on demand if it wasn't unavailable, but then we'd have to use mutable reference
#[allow(clippy::expect_used)]
self.legacy_task_manager
.as_ref()
.expect("did not enable legacy shutdown support")
.subscribe_named(child_suffix)
}
/// Finalise the shutdown procedure by waiting until either:
/// - all tracked tasks have terminated
/// - timeout has been reached
/// - shutdown has been forced (by sending SIGINT)
async fn finish_shutdown(&mut self) {
let mut wait_futures = FuturesUnordered::<Pin<Box<dyn Future<Output = ()> + Send>>>::new();
// force shutdown via ctrl-c
wait_futures.push(Box::pin(async move {
#[cfg(not(target_arch = "wasm32"))]
let interrupt_future = tokio::signal::ctrl_c();
#[cfg(target_arch = "wasm32")]
let interrupt_future = futures::future::pending::<()>();
let _ = interrupt_future.await;
info!("received interrupt - forcing shutdown");
}));
// timeout
let max_shutdown = self.max_shutdown_duration;
wait_futures.push(Box::pin(async move {
sleep(max_shutdown).await;
info!("timeout reached - forcing shutdown");
}));
// graceful
let tracker = self.tracker.clone();
wait_futures.push(Box::pin(async move {
tracker.wait_for_tracker().await;
info!("all tracked tasks successfully shutdown");
if let Some(legacy) = self.legacy_task_manager.as_mut() {
legacy.wait_for_graceful_shutdown().await;
info!("all legacy tasks successfully shutdown");
}
info!("all registered tasks successfully shutdown")
}));
wait_futures.next().await;
}
/// Remove the current set of [ShutdownSignals] from this instance of
/// [ShutdownManager] replacing it with an empty set.
///
/// This is potentially useful if one wishes to start listening for the signals
/// before the whole process has been fully set up.
pub fn detach_shutdown_signals(&mut self) -> ShutdownSignals {
mem::take(&mut self.shutdown_signals)
}
/// Replace the current set of [ShutdownSignals] used for determining
/// whether the underlying process should be stopped.
pub fn replace_shutdown_signals(&mut self, signals: ShutdownSignals) {
self.shutdown_signals = signals;
}
/// Send cancellation signal to all registered tasks by cancelling the root token
/// and sending shutdown signal, if applicable, on the legacy [TaskManager]
pub fn send_cancellation(&self) {
if let Some(legacy_manager) = self.legacy_task_manager.as_ref() {
info!("attempting to shutdown legacy tasks");
let _ = legacy_manager.signal_shutdown();
}
self.tracker.root_cancellation_token.cancel();
}
/// Wait until receiving one of the registered shutdown signals
/// this method is cancellation safe
pub async fn wait_for_shutdown_signal(&mut self) {
#[cfg(not(target_arch = "wasm32"))]
self.shutdown_signals.0.join_next().await;
#[cfg(target_arch = "wasm32")]
self.tracker.root_cancellation_token.cancelled().await;
}
/// Perform system shutdown by sending relevant signals and waiting until either:
/// - all tracked tasks have terminated
/// - timeout has been reached
/// - shutdown has been forced (by sending SIGINT)
pub async fn perform_shutdown(&mut self) {
self.send_cancellation();
info!("waiting for tasks to finish... (press ctrl-c to force)");
self.finish_shutdown().await;
}
/// Wait until a shutdown signal has been received and trigger system shutdown.
pub async fn run_until_shutdown(&mut self) {
self.close_tracker();
self.wait_for_shutdown_signal().await;
self.perform_shutdown().await;
}
}
#[cfg(test)]
mod tests {
use super::*;
use nym_test_utils::traits::{ElapsedExt, Timeboxed};
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
#[tokio::test]
async fn shutdown_with_no_tracked_tasks_and_signals() -> anyhow::Result<()> {
let mut manager = ShutdownManager::new_without_signals();
let res = manager.run_until_shutdown().timeboxed().await;
assert!(res.has_elapsed());
let mut manager = ShutdownManager::new_without_signals();
let shutdown = manager.clone_shutdown_token();
shutdown.cancel();
let res = manager.run_until_shutdown().timeboxed().await;
assert!(!res.has_elapsed());
Ok(())
}
#[tokio::test]
async fn shutdown_signal() -> anyhow::Result<()> {
let timeout_shutdown = sleep(Duration::from_millis(100));
let mut manager = ShutdownManager::new_without_signals().with_shutdown(timeout_shutdown);
// execution finishes after the sleep gets finishes
let res = manager
.run_until_shutdown()
.execute_with_deadline(Duration::from_millis(200))
.await;
assert!(!res.has_elapsed());
Ok(())
}
#[tokio::test]
async fn panic_hook() -> anyhow::Result<()> {
let mut manager = ShutdownManager::new_without_signals().with_cancel_on_panic();
manager.spawn_with_shutdown(async move {
sleep(Duration::from_millis(10000)).await;
});
manager.spawn_with_shutdown(async move {
sleep(Duration::from_millis(10)).await;
panic!("panicking");
});
// execution finishes after the panic gets triggered
let res = manager
.run_until_shutdown()
.execute_with_deadline(Duration::from_millis(200))
.await;
assert!(!res.has_elapsed());
Ok(())
}
#[tokio::test]
async fn task_cancellation() -> anyhow::Result<()> {
let timeout_shutdown = sleep(Duration::from_millis(100));
let mut manager = ShutdownManager::new_without_signals().with_shutdown(timeout_shutdown);
let cancelled1 = Arc::new(AtomicBool::new(false));
let cancelled1_clone = cancelled1.clone();
let cancelled2 = Arc::new(AtomicBool::new(false));
let cancelled2_clone = cancelled2.clone();
let shutdown = manager.clone_shutdown_token();
manager.spawn(async move {
shutdown.cancelled().await;
cancelled1_clone.store(true, std::sync::atomic::Ordering::Relaxed);
});
let shutdown = manager.clone_shutdown_token();
manager.spawn(async move {
shutdown.cancelled().await;
cancelled2_clone.store(true, std::sync::atomic::Ordering::Relaxed);
});
let res = manager
.run_until_shutdown()
.execute_with_deadline(Duration::from_millis(200))
.await;
assert!(!res.has_elapsed());
assert!(cancelled1.load(std::sync::atomic::Ordering::Relaxed));
assert!(cancelled2.load(std::sync::atomic::Ordering::Relaxed));
Ok(())
}
#[tokio::test]
async fn cancellation_within_task() -> anyhow::Result<()> {
let mut manager = ShutdownManager::new_without_signals();
let cancelled1 = Arc::new(AtomicBool::new(false));
let cancelled1_clone = cancelled1.clone();
let shutdown = manager.clone_shutdown_token();
manager.spawn(async move {
shutdown.cancelled().await;
cancelled1_clone.store(true, std::sync::atomic::Ordering::Relaxed);
});
let shutdown = manager.clone_shutdown_token();
manager.spawn(async move {
sleep(Duration::from_millis(10)).await;
shutdown.cancel();
});
let res = manager
.run_until_shutdown()
.execute_with_deadline(Duration::from_millis(200))
.await;
assert!(!res.has_elapsed());
assert!(cancelled1.load(std::sync::atomic::Ordering::Relaxed));
Ok(())
}
#[tokio::test]
async fn shutdown_timeout() -> anyhow::Result<()> {
let timeout_shutdown = sleep(Duration::from_millis(50));
let mut manager = ShutdownManager::new_without_signals()
.with_shutdown(timeout_shutdown)
.with_shutdown_duration(Duration::from_millis(1000));
// ignore shutdown signals
manager.spawn(async move {
sleep(Duration::from_millis(1000)).await;
});
let res = manager
.run_until_shutdown()
.execute_with_deadline(Duration::from_millis(200))
.await;
assert!(res.has_elapsed());
let timeout_shutdown = sleep(Duration::from_millis(50));
let mut manager = ShutdownManager::new_without_signals()
.with_shutdown(timeout_shutdown)
.with_shutdown_duration(Duration::from_millis(100));
// ignore shutdown signals
manager.spawn(async move {
sleep(Duration::from_millis(1000)).await;
});
let res = manager
.run_until_shutdown()
.execute_with_deadline(Duration::from_millis(200))
.await;
assert!(!res.has_elapsed());
Ok(())
}
}
-54
View File
@@ -1,54 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
//! A [CancellationToken](tokio_util::sync::CancellationToken)-backed shutdown mechanism for Nym binaries.
//!
//! It allows creation of a centralised manager for keeping track of all signals that are meant
//! to trigger exit of all associated tasks and sending cancellation to the aforementioned futures.
//!
//! # Default usage
//!
//! ```no_run
//! use std::time::Duration;
//! use tokio::time::sleep;
//! use nym_task::{ShutdownManager, ShutdownToken};
//!
//! async fn my_task() {
//! loop {
//! sleep(Duration::from_secs(5)).await
//! // do some periodic work that can be easily interrupted
//! }
//! }
//!
//! async fn important_work_that_cant_be_interrupted() {}
//!
//! async fn my_managed_task(shutdown_token: ShutdownToken) {
//! tokio::select! {
//! _ = shutdown_token.cancelled() => {}
//! _ = important_work_that_cant_be_interrupted() => {}
//! }
//! }
//! #[tokio::main]
//! async fn main() {
//! let mut shutdown_manager = ShutdownManager::build_new_default().expect("failed to register default shutdown signals");
//!
//! let shutdown_token = shutdown_manager.child_shutdown_token();
//! shutdown_manager.try_spawn_named(async move { my_managed_task(shutdown_token).await }, "important-managed-task");
//! shutdown_manager.try_spawn_named_with_shutdown(my_task(), "another-task");
//!
//! // wait for shutdown signal
//! shutdown_manager.run_until_shutdown().await;
//! }
//! ```
use std::time::Duration;
pub mod manager;
pub mod token;
pub mod tracker;
pub use manager::ShutdownManager;
pub use token::{ShutdownDropGuard, ShutdownToken};
pub use tracker::ShutdownTracker;
pub const DEFAULT_MAX_SHUTDOWN_DURATION: Duration = Duration::from_secs(5);
-170
View File
@@ -1,170 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::event::SentStatus;
use std::future::Future;
use tokio_util::sync::{
CancellationToken, DropGuard, WaitForCancellationFuture, WaitForCancellationFutureOwned,
};
use tracing::warn;
/// A wrapped [CancellationToken](tokio_util::sync::CancellationToken) that is used for
/// signalling and listening for cancellation requests.
// We don't use CancellationToken in case we wanted to include additional fields/methods
// down the line.
#[derive(Debug, Clone, Default)]
pub struct ShutdownToken {
inner: CancellationToken,
}
impl From<CancellationToken> for ShutdownToken {
fn from(inner: CancellationToken) -> Self {
ShutdownToken { inner }
}
}
impl ShutdownToken {
/// A drop in no-op replacement for `send_status_msg` for easier migration from [TaskClient](crate::TaskClient).
#[deprecated]
#[track_caller]
pub fn send_status_msg(&self, status: SentStatus) {
let caller = std::panic::Location::caller();
warn!("{caller} attempted to send {status} - there are no more listeners of those");
}
/// Creates a new ShutdownToken in the non-cancelled state.
pub fn new() -> Self {
ShutdownToken {
inner: CancellationToken::new(),
}
}
/// Creates a new ShutdownToken given a tokio `CancellationToken`.
pub fn new_from_tokio_token(cancellation_token: CancellationToken) -> Self {
ShutdownToken {
inner: cancellation_token,
}
}
/// Gets reference to the underlying [CancellationToken](tokio_util::sync::CancellationToken).
pub fn inner(&self) -> &CancellationToken {
&self.inner
}
/// Get an owned [CancellationToken](tokio_util::sync::CancellationToken) for public API use.
/// This is useful when you need to expose cancellation to SDK users without
/// exposing the internal ShutdownToken type.
pub fn to_cancellation_token(&self) -> CancellationToken {
self.inner.clone()
}
/// Creates a `ShutdownToken` which will get cancelled whenever the
/// current token gets cancelled. Unlike a cloned `ShutdownToken`,
/// cancelling a child token does not cancel the parent token.
///
/// If the current token is already cancelled, the child token will get
/// returned in cancelled state.
pub fn child_token(&self) -> ShutdownToken {
ShutdownToken {
inner: self.inner.child_token(),
}
}
/// Cancel the underlying [CancellationToken](tokio_util::sync::CancellationToken) and all child tokens which had been
/// derived from it.
///
/// This will wake up all tasks which are waiting for cancellation.
pub fn cancel(&self) {
self.inner.cancel();
}
/// Returns `true` if the underlying [CancellationToken](tokio_util::sync::CancellationToken) is cancelled.
pub fn is_cancelled(&self) -> bool {
self.inner.is_cancelled()
}
/// Returns a `Future` that gets fulfilled when cancellation is requested.
///
/// The future will complete immediately if the token is already cancelled
/// when this method is called.
///
/// # Cancel safety
///
/// This method is cancel safe.
pub fn cancelled(&self) -> WaitForCancellationFuture<'_> {
self.inner.cancelled()
}
/// Returns a `Future` that gets fulfilled when cancellation is requested.
///
/// The future will complete immediately if the token is already cancelled
/// when this method is called.
///
/// The function takes self by value and returns a future that owns the
/// token.
///
/// # Cancel safety
///
/// This method is cancel safe.
pub fn cancelled_owned(self) -> WaitForCancellationFutureOwned {
self.inner.cancelled_owned()
}
/// Creates a `ShutdownDropGuard` for this token.
///
/// Returned guard will cancel this token (and all its children) on drop
/// unless disarmed.
pub fn drop_guard(self) -> ShutdownDropGuard {
ShutdownDropGuard {
inner: self.inner.drop_guard(),
}
}
/// Runs a future to completion and returns its result wrapped inside an `Option`
/// unless the `ShutdownToken` is cancelled. In that case the function returns
/// `None` and the future gets dropped.
///
/// # Cancel safety
///
/// This method is only cancel safe if `fut` is cancel safe.
pub async fn run_until_cancelled<F>(&self, fut: F) -> Option<F::Output>
where
F: Future,
{
self.inner.run_until_cancelled(fut).await
}
/// Runs a future to completion and returns its result wrapped inside an `Option`
/// unless the `ShutdownToken` is cancelled. In that case the function returns
/// `None` and the future gets dropped.
///
/// The function takes self by value and returns a future that owns the token.
///
/// # Cancel safety
///
/// This method is only cancel safe if `fut` is cancel safe.
pub async fn run_until_cancelled_owned<F>(self, fut: F) -> Option<F::Output>
where
F: Future,
{
self.inner.run_until_cancelled_owned(fut).await
}
}
/// A wrapper for [DropGuard](tokio_util::sync::DropGuard) that wraps around a cancellation token
/// which automatically cancels it on drop.
/// It is created using `drop_guard` method on the `ShutdownToken`.
pub struct ShutdownDropGuard {
inner: DropGuard,
}
impl ShutdownDropGuard {
/// Returns stored [ShutdownToken](ShutdownToken) and removes this drop guard instance
/// (i.e. it will no longer cancel token). Other guards for this token
/// are not affected.
pub fn disarm(self) -> ShutdownToken {
ShutdownToken {
inner: self.inner.disarm(),
}
}
}
-353
View File
@@ -1,353 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::cancellation::token::ShutdownToken;
use crate::spawn::{spawn_named_future, JoinHandle};
use crate::spawn_future;
use std::future::Future;
use thiserror::Error;
use tokio_util::task::TaskTracker;
use tracing::{debug, trace};
#[derive(Debug, Error)]
#[error("task got cancelled")]
pub struct Cancelled;
/// Extracted [TaskTracker](tokio_util::task::TaskTracker) and [ShutdownToken](ShutdownToken) to more easily allow tracking nested tasks
/// without having to pass whole [ShutdownManager](ShutdownManager) around.
#[derive(Clone, Default, Debug)]
pub struct ShutdownTracker {
/// The root [ShutdownToken](ShutdownToken) that will trigger all derived tasks
/// to receive cancellation signal.
pub(crate) root_cancellation_token: ShutdownToken,
// Note: the reason we're not using a `JoinSet` is
// because it forces us to use futures with the same `::Output` type,
// which is not really a desirable property in this instance.
/// Tracker used for keeping track of all registered tasks
/// so that they could be stopped gracefully before ending the process.
pub(crate) tracker: TaskTracker,
}
#[cfg(not(target_arch = "wasm32"))]
impl ShutdownTracker {
/// Spawn the provided future on the current Tokio runtime, and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn<F>(&self, task: F) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
let tracked = self.tracker.track_future(task);
spawn_future(tracked)
}
/// Spawn the provided future on the current Tokio runtime,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
/// Furthermore, attach a name to the spawned task to more easily track it within a [tokio console](https://github.com/tokio-rs/console)
///
/// Note that is no different from [spawn](Self::spawn) if the underlying binary
/// has not been built with `RUSTFLAGS="--cfg tokio_unstable"` and `--features="tokio-tracing"`
#[track_caller]
pub fn try_spawn_named<F>(&self, task: F, name: &str) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
trace!("attempting to spawn task {name}");
let tracked = self.tracker.track_future(task);
spawn_named_future(tracked, name)
}
/// Spawn the provided future on the provided Tokio runtime,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_on<F>(&self, task: F, handle: &tokio::runtime::Handle) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
self.tracker.spawn_on(task, handle)
}
/// Spawn the provided future on the current [LocalSet](tokio::task::LocalSet),
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_local<F>(&self, task: F) -> JoinHandle<F::Output>
where
F: Future + 'static,
F::Output: 'static,
{
self.tracker.spawn_local(task)
}
/// Spawn the provided blocking task on the current Tokio runtime,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_blocking<F, T>(&self, task: F) -> JoinHandle<T>
where
F: FnOnce() -> T,
F: Send + 'static,
T: Send + 'static,
{
self.tracker.spawn_blocking(task)
}
/// Spawn the provided blocking task on the provided Tokio runtime,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_blocking_on<F, T>(&self, task: F, handle: &tokio::runtime::Handle) -> JoinHandle<T>
where
F: FnOnce() -> T,
F: Send + 'static,
T: Send + 'static,
{
self.tracker.spawn_blocking_on(task, handle)
}
/// Spawn the provided future on the current Tokio runtime
/// that will get cancelled once a global shutdown signal is detected,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
///
/// Note that to fully use the naming feature, such as tracking within a [tokio console](https://github.com/tokio-rs/console),
/// the underlying binary has to be built with `RUSTFLAGS="--cfg tokio_unstable"` and `--features="tokio-tracing"`
#[track_caller]
pub fn try_spawn_named_with_shutdown<F>(
&self,
task: F,
name: &str,
) -> JoinHandle<Result<F::Output, Cancelled>>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
trace!("attempting to spawn task {name} (with top-level cancellation)");
let caller = std::panic::Location::caller();
let shutdown_token = self.clone_shutdown_token();
let name_owned = name.to_string();
let tracked = self.tracker.track_future(async move {
match shutdown_token.run_until_cancelled_owned(task).await {
Some(result) => {
debug!("{name_owned} @ {caller}: task has finished execution");
Ok(result)
}
None => {
trace!("{name_owned} @ {caller}: shutdown signal received, shutting down");
Err(Cancelled)
}
}
});
spawn_named_future(tracked, name)
}
/// Spawn the provided future on the current Tokio runtime
/// that will get cancelled once a global shutdown signal is detected,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_with_shutdown<F>(&self, task: F) -> JoinHandle<Result<F::Output, Cancelled>>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
let caller = std::panic::Location::caller();
let shutdown_token = self.clone_shutdown_token();
self.tracker.spawn(async move {
match shutdown_token.run_until_cancelled_owned(task).await {
Some(result) => {
debug!("{caller}: task has finished execution");
Ok(result)
}
None => {
trace!("{caller}: shutdown signal received, shutting down");
Err(Cancelled)
}
}
})
}
}
#[cfg(target_arch = "wasm32")]
impl ShutdownTracker {
/// Run the provided future on the current thread, and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn<F>(&self, task: F) -> JoinHandle<F::Output>
where
F: Future + 'static,
{
let tracked = self.tracker.track_future(task);
spawn_future(tracked)
}
/// Run the provided future on the current thread, and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
/// It has exactly the same behaviour as [spawn](Self::spawn) and it only exists to provide
/// the same interface as non-wasm32 targets.
#[track_caller]
pub fn try_spawn_named<F>(&self, task: F, name: &str) -> JoinHandle<F::Output>
where
F: Future + 'static,
{
let tracked = self.tracker.track_future(task);
spawn_named_future(tracked, name)
}
/// Run the provided future on the current thread
/// that will get cancelled once a global shutdown signal is detected,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
/// It has exactly the same behaviour as [spawn_with_shutdown](Self::spawn_with_shutdown) and it only exists to provide
/// the same interface as non-wasm32 targets.
#[track_caller]
pub fn try_spawn_named_with_shutdown<F>(
&self,
task: F,
name: &str,
) -> JoinHandle<Result<F::Output, Cancelled>>
where
F: Future<Output = ()> + 'static,
{
let caller = std::panic::Location::caller();
let shutdown_token = self.clone_shutdown_token();
let tracked = self.tracker.track_future(async move {
match shutdown_token.run_until_cancelled_owned(task).await {
Some(result) => {
debug!("{caller}: task has finished execution");
Ok(result)
}
None => {
trace!("{caller}: shutdown signal received, shutting down");
Err(Cancelled)
}
}
});
spawn_named_future(tracked, name)
}
/// Run the provided future on the current thread
/// that will get cancelled once a global shutdown signal is detected,
/// and track it in the underlying [TaskTracker](tokio_util::task::TaskTracker).
#[track_caller]
pub fn spawn_with_shutdown<F>(&self, task: F) -> JoinHandle<Result<F::Output, Cancelled>>
where
F: Future<Output = ()> + 'static,
{
let caller = std::panic::Location::caller();
let shutdown_token = self.clone_shutdown_token();
let tracked = self.tracker.track_future(async move {
match shutdown_token.run_until_cancelled_owned(task).await {
Some(result) => {
debug!("{caller}: task has finished execution");
Ok(result)
}
None => {
trace!("{caller}: shutdown signal received, shutting down");
Err(Cancelled)
}
}
});
spawn_future(tracked)
}
}
impl ShutdownTracker {
/// Create new instance of the ShutdownTracker using an external shutdown token.
/// This could be useful in situations where shutdown is being managed by an external entity
/// that is not [ShutdownManager](ShutdownManager), but interface requires providing a ShutdownTracker,
/// such as client-core tasks
pub fn new_from_external_shutdown_token(shutdown_token: ShutdownToken) -> Self {
ShutdownTracker {
root_cancellation_token: shutdown_token,
tracker: Default::default(),
}
}
/// Waits until the underlying [TaskTracker](tokio_util::task::TaskTracker) is both closed and empty.
///
/// If the underlying [TaskTracker](tokio_util::task::TaskTracker) is already closed and empty when this method is called, then it
/// returns immediately.
pub async fn wait_for_tracker(&self) {
self.tracker.wait().await;
}
/// Close the underlying [TaskTracker](tokio_util::task::TaskTracker).
///
/// This allows [`wait_for_tracker`] futures to complete. It does not prevent you from spawning new tasks.
///
/// Returns `true` if this closed the underlying [TaskTracker](tokio_util::task::TaskTracker), or `false` if it was already closed.
///
/// [`wait_for_tracker`]: Self::wait_for_tracker
pub fn close_tracker(&self) -> bool {
self.tracker.close()
}
/// Reopen the underlying [TaskTracker](tokio_util::task::TaskTracker).
///
/// This prevents [`wait_for_tracker`] futures from completing even if the underlying [TaskTracker](tokio_util::task::TaskTracker) is empty.
///
/// Returns `true` if this reopened the underlying [TaskTracker](tokio_util::task::TaskTracker), or `false` if it was already open.
///
/// [`wait_for_tracker`]: Self::wait_for_tracker
pub fn reopen_tracker(&self) -> bool {
self.tracker.reopen()
}
/// Returns `true` if the underlying [TaskTracker](tokio_util::task::TaskTracker) is [closed](Self::close_tracker).
pub fn is_tracker_closed(&self) -> bool {
self.tracker.is_closed()
}
/// Returns the number of tasks tracked by the underlying [TaskTracker](tokio_util::task::TaskTracker).
pub fn tracked_tasks(&self) -> usize {
self.tracker.len()
}
/// Returns `true` if there are no tasks in the underlying [TaskTracker](tokio_util::task::TaskTracker).
pub fn is_tracker_empty(&self) -> bool {
self.tracker.is_empty()
}
/// Obtain a [ShutdownToken](crate::cancellation::ShutdownToken) that is a child of the root token
pub fn child_shutdown_token(&self) -> ShutdownToken {
self.root_cancellation_token.child_token()
}
/// Obtain a [ShutdownToken](crate::cancellation::ShutdownToken) on the same hierarchical structure as the root token
pub fn clone_shutdown_token(&self) -> ShutdownToken {
self.root_cancellation_token.clone()
}
/// Create a child ShutdownTracker that inherits cancellation from this tracker
/// but has its own TaskTracker for managing sub-tasks.
///
/// This enables hierarchical task management where:
/// - Parent cancellation flows to all children
/// - Each level tracks its own tasks independently
/// - Components can wait for their specific sub-tasks to complete
pub fn child_tracker(&self) -> ShutdownTracker {
// Child token inherits cancellation from parent
let child_token = self.root_cancellation_token.child_token();
// New TaskTracker for this level's tasks
let child_task_tracker = TaskTracker::new();
ShutdownTracker {
root_cancellation_token: child_token,
tracker: child_task_tracker,
}
}
/// Convenience method to perform a complete shutdown sequence.
/// This method:
/// 1. Signals cancellation to all tasks
/// 2. Closes the tracker to prevent new tasks
/// 3. Waits for all existing tasks to complete
pub async fn shutdown(self) {
// Signal cancellation to all tasks
self.root_cancellation_token.cancel();
// Close the tracker to prevent new tasks from being spawned
self.tracker.close();
// Wait for all existing tasks to complete
self.tracker.wait().await;
}
}
+20 -17
View File
@@ -2,9 +2,12 @@
// SPDX-License-Identifier: Apache-2.0
use futures::channel::mpsc;
use std::collections::HashMap;
use std::{
collections::HashMap,
time::{Duration, Instant},
};
// const LANE_CONSIDERED_CLEAR: usize = 10;
const LANE_CONSIDERED_CLEAR: usize = 10;
pub type ConnectionId = u64;
@@ -80,21 +83,21 @@ impl LaneQueueLengths {
}
}
// pub async fn wait_until_clear(&self, lane: &TransmissionLane, timeout: Option<Duration>) {
// let total_time_waited = Instant::now();
// loop {
// let lane_length = self.get(lane).unwrap_or_default();
// if lane_length < LANE_CONSIDERED_CLEAR {
// break;
// }
// if timeout.is_some_and(|timeout| total_time_waited.elapsed() > timeout) {
// log::warn!("Timeout reached while waiting for queue to clear");
// break;
// }
// log::trace!("Waiting for queue to clear ({lane_length} items left)");
// tokio::time::sleep(Duration::from_millis(100)).await;
// }
// }
pub async fn wait_until_clear(&self, lane: &TransmissionLane, timeout: Option<Duration>) {
let total_time_waited = Instant::now();
loop {
let lane_length = self.get(lane).unwrap_or_default();
if lane_length < LANE_CONSIDERED_CLEAR {
break;
}
if timeout.is_some_and(|timeout| total_time_waited.elapsed() > timeout) {
log::warn!("Timeout reached while waiting for queue to clear");
break;
}
log::trace!("Waiting for queue to clear ({lane_length} items left)");
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
}
impl Default for LaneQueueLengths {
+3 -13
View File
@@ -5,25 +5,15 @@ pub mod cancellation;
pub mod connections;
pub mod event;
pub mod manager;
pub(crate) mod runtime_registry;
#[cfg(not(target_arch = "wasm32"))]
pub mod signal;
pub mod spawn;
pub use cancellation::{ShutdownDropGuard, ShutdownManager, ShutdownToken, ShutdownTracker};
pub use cancellation::{ShutdownDropGuard, ShutdownManager, ShutdownToken};
pub use event::{StatusReceiver, StatusSender, TaskStatus, TaskStatusEvent};
#[allow(deprecated)]
pub use manager::{TaskClient, TaskManager};
pub use spawn::spawn_future;
pub use manager::{TaskClient, TaskHandle, TaskManager};
pub use spawn::{spawn, spawn_with_report_error};
pub use tokio_util::task::TaskTracker;
#[cfg(not(target_arch = "wasm32"))]
pub use signal::{wait_for_signal, wait_for_signal_and_error};
pub use crate::runtime_registry::RegistryAccessError;
/// Get or create a ShutdownTracker for SDK use.
/// This provides automatic task management without requiring manual setup.
pub fn get_sdk_shutdown_tracker() -> Result<ShutdownTracker, RegistryAccessError> {
Ok(runtime_registry::RuntimeRegistry::get_or_create_sdk()?.shutdown_tracker_owned())
}
+1 -20
View File
@@ -44,7 +44,6 @@ enum TaskError {
/// Listens to status and error messages from tasks, as well as notifying them to gracefully
/// shutdown. Keeps track of if task stop unexpectedly, such as in a panic.
#[deprecated(note = "use ShutdownManager instead")]
#[derive(Debug)]
pub struct TaskManager {
// optional name assigned to the task manager that all subscribed task clients will inherit
@@ -73,7 +72,6 @@ pub struct TaskManager {
task_status_rx: Option<StatusReceiver>,
}
#[allow(deprecated)]
impl Default for TaskManager {
fn default() -> Self {
let (notify_tx, notify_rx) = watch::channel(());
@@ -97,8 +95,6 @@ impl Default for TaskManager {
}
}
#[allow(deprecated)]
#[allow(clippy::expect_used)]
impl TaskManager {
pub fn new(shutdown_timer_secs: u64) -> Self {
Self {
@@ -172,7 +168,7 @@ impl TaskManager {
if let Some(mut task_status_rx) = self.task_status_rx.take() {
log::info!("Starting status message listener");
crate::spawn::spawn_future(async move {
crate::spawn::spawn(async move {
loop {
if let Some(msg) = task_status_rx.next().await {
log::trace!("Got msg: {msg}");
@@ -190,14 +186,12 @@ impl TaskManager {
}
// used for compatibility with the ShutdownManager
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn task_return_error_rx(&mut self) -> ErrorReceiver {
self.task_return_error_rx
.take()
.expect("unable to get error channel: attempt to wait twice?")
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn task_drop_rx(&mut self) -> ErrorReceiver {
self.task_drop_rx
.take()
@@ -265,7 +259,6 @@ impl TaskManager {
/// Listen for shutdown notifications, and can send error and status messages back to the
/// `TaskManager`
#[derive(Debug)]
#[deprecated(note = "use ShutdownToken instead")]
pub struct TaskClient {
// optional name assigned to the shutdown handle
name: Option<String>,
@@ -293,7 +286,6 @@ pub struct TaskClient {
mode: ClientOperatingMode,
}
#[allow(deprecated)]
impl Clone for TaskClient {
fn clone(&self) -> Self {
// make sure to not accidentally overflow the stack if we keep cloning the handle
@@ -321,7 +313,6 @@ impl Clone for TaskClient {
}
}
#[allow(deprecated)]
impl TaskClient {
const MAX_NAME_LENGTH: usize = 128;
const OVERFLOW_NAME: &'static str = "reached maximum TaskClient children name depth";
@@ -442,8 +433,6 @@ impl TaskClient {
.await
}
// legacy code
#[allow(clippy::panic)]
pub async fn recv_timeout(&mut self) {
if self.mode.is_dummy() {
return pending().await;
@@ -516,7 +505,6 @@ impl TaskClient {
}
}
#[allow(deprecated)]
impl Drop for TaskClient {
fn drop(&mut self) {
if !self.mode.should_signal_on_drop() {
@@ -584,8 +572,6 @@ impl ClientOperatingMode {
}
}
#[deprecated]
#[allow(deprecated)]
#[derive(Debug)]
pub enum TaskHandle {
/// Full [`TaskManager`] that was created by the underlying task.
@@ -595,28 +581,24 @@ pub enum TaskHandle {
External(TaskClient),
}
#[allow(deprecated)]
impl From<TaskManager> for TaskHandle {
fn from(value: TaskManager) -> Self {
TaskHandle::Internal(value)
}
}
#[allow(deprecated)]
impl From<TaskClient> for TaskHandle {
fn from(value: TaskClient) -> Self {
TaskHandle::External(value)
}
}
#[allow(deprecated)]
impl Default for TaskHandle {
fn default() -> Self {
TaskHandle::Internal(TaskManager::default())
}
}
#[allow(deprecated)]
impl TaskHandle {
#[must_use]
pub fn name_if_unnamed<S: Into<String>>(self, name: S) -> Self {
@@ -684,7 +666,6 @@ mod tests {
use super::*;
#[tokio::test]
#[allow(deprecated)]
async fn signal_shutdown() {
let shutdown = TaskManager::default();
let mut listener = shutdown.subscribe();
-96
View File
@@ -1,96 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use thiserror::Error;
use crate::ShutdownManager;
use std::sync::RwLock;
use std::sync::{Arc, LazyLock};
/// Global registry that manages ShutdownManagers transparently.
/// This allows SDK components to get automatic task management without
/// exposing the complexity to end users.
pub(crate) struct RuntimeRegistry {
// For SDK clients: auto-created manager without signal handling
sdk_manager: RwLock<Option<Arc<ShutdownManager>>>,
}
#[derive(Debug, Error)]
pub enum RegistryAccessError {
#[error("the runtime registry is poisoned")]
Poisoned,
}
impl RuntimeRegistry {
/// Get or create a ShutdownManager for SDK use.
/// This manager doesn't listen to OS signals, making it suitable for library use.
pub(crate) fn get_or_create_sdk() -> Result<Arc<ShutdownManager>, RegistryAccessError> {
let guard = REGISTRY
.sdk_manager
.read()
.map_err(|_| RegistryAccessError::Poisoned)?;
if let Some(manager) = guard.as_ref() {
return Ok(manager.clone());
}
drop(guard);
let mut guard = REGISTRY
.sdk_manager
.write()
.map_err(|_| RegistryAccessError::Poisoned)?;
Ok(guard
.get_or_insert_with(|| Arc::new(ShutdownManager::new_without_signals()))
.clone())
}
/// Check if an SDK manager has been created.
/// Useful for testing and debugging.
#[allow(dead_code)]
pub(crate) fn has_sdk_manager() -> Result<bool, RegistryAccessError> {
Ok(REGISTRY
.sdk_manager
.read()
.map_err(|_| RegistryAccessError::Poisoned)?
.is_some())
}
/// Clear the SDK manager.
/// This is primarily for testing to ensure isolation between tests.
#[cfg(test)]
pub(crate) async fn clear() -> Result<(), RegistryAccessError> {
*REGISTRY
.sdk_manager
.write()
.map_err(|_| RegistryAccessError::Poisoned)? = None;
Ok(())
}
}
/// Global instance of the runtime registry.
/// Uses LazyLock for on-demand initialization.
static REGISTRY: LazyLock<RuntimeRegistry> = LazyLock::new(|| RuntimeRegistry {
sdk_manager: RwLock::new(None),
});
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_get_or_create_sdk() {
// Clear any existing manager
let _ = RuntimeRegistry::clear().await;
assert!(!RuntimeRegistry::has_sdk_manager().unwrap());
let manager1 = RuntimeRegistry::get_or_create_sdk().unwrap();
assert!(RuntimeRegistry::has_sdk_manager().unwrap());
let manager2 = RuntimeRegistry::get_or_create_sdk().unwrap();
// Should return the same instance
assert!(Arc::ptr_eq(&manager1, &manager2));
let _ = RuntimeRegistry::clear().await;
assert!(!RuntimeRegistry::has_sdk_manager().unwrap());
}
}
+3 -7
View File
@@ -1,7 +1,6 @@
use crate::manager::SentError;
use crate::{manager::SentError, TaskManager};
#[cfg(unix)]
#[allow(clippy::expect_used)]
pub async fn wait_for_signal() {
use tokio::signal::unix::{signal, SignalKind};
let mut sigterm = signal(SignalKind::terminate()).expect("Failed to setup SIGTERM channel");
@@ -29,10 +28,8 @@ pub async fn wait_for_signal() {
}
}
#[allow(deprecated)]
#[cfg(unix)]
#[allow(clippy::expect_used)]
pub async fn wait_for_signal_and_error(shutdown: &mut crate::TaskManager) -> Result<(), SentError> {
pub async fn wait_for_signal_and_error(shutdown: &mut TaskManager) -> Result<(), SentError> {
use tokio::signal::unix::{signal, SignalKind};
let mut sigterm = signal(SignalKind::terminate()).expect("Failed to setup SIGTERM channel");
@@ -58,9 +55,8 @@ pub async fn wait_for_signal_and_error(shutdown: &mut crate::TaskManager) -> Res
}
}
#[allow(deprecated)]
#[cfg(not(unix))]
pub async fn wait_for_signal_and_error(shutdown: &mut crate::TaskManager) -> Result<(), SentError> {
pub async fn wait_for_signal_and_error(shutdown: &mut TaskManager) -> Result<(), SentError> {
tokio::select! {
_ = tokio::signal::ctrl_c() => {
log::info!("Received SIGINT");
+16 -60
View File
@@ -1,79 +1,35 @@
use crate::TaskClient;
use std::future::Future;
#[cfg(not(target_arch = "wasm32"))]
pub type JoinHandle<F> = tokio::task::JoinHandle<F>;
// no JoinHandle equivalent in wasm
#[cfg(target_arch = "wasm32")]
#[derive(Clone, Copy)]
pub struct FakeJoinHandle<F> {
_p: std::marker::PhantomData<F>,
}
#[cfg(target_arch = "wasm32")]
pub type JoinHandle<F> = FakeJoinHandle<F>;
#[cfg(target_arch = "wasm32")]
#[track_caller]
pub fn spawn_future<F>(future: F) -> JoinHandle<F::Output>
pub fn spawn<F>(future: F)
where
F: Future + 'static,
F: Future<Output = ()> + 'static,
{
wasm_bindgen_futures::spawn_local(async move {
// make sure the future outputs `()`
future.await;
});
FakeJoinHandle {
_p: std::marker::PhantomData,
}
wasm_bindgen_futures::spawn_local(future);
}
// Note: prefer spawning tasks directly on the ShutdownManager
#[cfg(not(target_arch = "wasm32"))]
#[track_caller]
pub fn spawn_future<F>(future: F) -> JoinHandle<F::Output>
pub fn spawn<F>(future: F)
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
tokio::spawn(future)
tokio::spawn(future);
}
// Note: prefer spawning tasks directly on the ShutdownManager
#[cfg(not(target_arch = "wasm32"))]
#[track_caller]
pub fn spawn_named_future<F>(future: F, name: &str) -> JoinHandle<F::Output>
pub fn spawn_with_report_error<F, T, E>(future: F, mut shutdown: TaskClient)
where
F: Future + Send + 'static,
F::Output: Send + 'static,
F: Future<Output = Result<T, E>> + Send + 'static,
T: 'static,
E: std::error::Error + Send + Sync + 'static,
{
cfg_if::cfg_if! {if #[cfg(all(tokio_unstable, feature="tokio-tracing"))] {
#[allow(clippy::expect_used)]
tokio::task::Builder::new().name(name).spawn(future).expect("failed to spawn future")
} else {
let _ = name;
tracing::debug!(r#"the underlying binary hasn't been built with `RUSTFLAGS="--cfg tokio_unstable"` - the future naming won't do anything"#);
spawn_future(future)
}}
}
#[cfg(target_arch = "wasm32")]
#[track_caller]
pub fn spawn_named_future<F>(future: F, name: &str) -> JoinHandle<F::Output>
where
F: Future + 'static,
{
// not supported in wasm
let _ = name;
spawn_future(future)
}
#[macro_export]
macro_rules! spawn_future {
($future:expr) => {{
$crate::spawn_future($future)
}};
($future:expr, $name:expr) => {{
$crate::spawn_named_future($future, $name)
}};
let future_that_sends = async move {
if let Err(err) = future.await {
shutdown.send_we_stopped(Box::new(err));
}
};
spawn(future_that_sends);
}

Some files were not shown because too many files have changed in this diff Show More