tornadocontrib/web3-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add optional tokio-console and improve reconnect logic

Browse Source
This commit is contained in:
Bryan Stitt 2023-03-22 18:43:13 -07:00
parent 7b223efa4d
commit 2205a472fd
7 changed files with 452 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

381
Cargo.lock generated
View File

@ -737,6 +737,12 @@ dependencies = [
"thiserror", "thiserror",
] ]
[[package]]
name = "cassowary"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "df8670b8c7b9dae1793364eafadf7239c40d669904660c5960d74cfd80b46a53"
[[package]] [[package]]
name = "cc" name = "cc"
version = "1.0.79" version = "1.0.79"
@ -804,6 +810,15 @@ dependencies = [
"textwrap", "textwrap",
] ]
[[package]]
name = "clap_complete"
version = "3.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f7a2e0a962c45ce25afce14220bc24f9dade0a1787f185cecf96bfba7847cd8"
dependencies = [
"clap",
]
[[package]] [[package]]
name = "clap_derive" name = "clap_derive"
version = "3.2.18" version = "3.2.18"
@ -893,6 +908,34 @@ dependencies = [
"thiserror", "thiserror",
] ]
[[package]]
name = "color-eyre"
version = "0.5.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1f1885697ee8a177096d42f158922251a41973117f6d8a234cee94b9509157b7"
dependencies = [
"backtrace",
"color-spantrace",
"eyre",
"indenter",
"once_cell",
"owo-colors",
"tracing-error",
"url",
]
[[package]]
name = "color-spantrace"
version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6eee477a4a8a72f4addd4de416eb56d54bc307b284d6601bafdee1f4ea462d1"
dependencies = [
"once_cell",
"owo-colors",
"tracing-core",
"tracing-error",
]
[[package]] [[package]]
name = "combine" name = "combine"
version = "4.6.6" version = "4.6.6"
@ -943,6 +986,42 @@ dependencies = [
"windows-sys 0.42.0", "windows-sys 0.42.0",
] ]
[[package]]
name = "console-api"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e57ff02e8ad8e06ab9731d5dc72dc23bef9200778eae1a89d555d8c42e5d4a86"
dependencies = [
"prost",
"prost-types",
"tonic",
"tracing-core",
]
[[package]]
name = "console-subscriber"
version = "0.1.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22a3a81dfaf6b66bce5d159eddae701e3a002f194d378cbf7be5f053c281d9be"
dependencies = [
"console-api",
"crossbeam-channel",
"crossbeam-utils",
"futures",
"hdrhistogram",
"humantime",
"prost-types",
"serde",
"serde_json",
"thread_local",
"tokio",
"tokio-stream",
"tonic",
"tracing",
"tracing-core",
"tracing-subscriber 0.3.16",
]
[[package]] [[package]]
name = "const-oid" name = "const-oid"
version = "0.7.1" version = "0.7.1"
@ -1068,6 +1147,32 @@ dependencies = [
"cfg-if", "cfg-if",
] ]
[[package]]
name = "crossterm"
version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c0ebde6a9dd5e331cd6c6f48253254d117642c31653baa475e394657c59c1f7d"
dependencies = [
"bitflags",
"crossterm_winapi",
"futures-core",
"libc",
"mio 0.7.14",
"parking_lot 0.11.2",
"signal-hook",
"signal-hook-mio",
"winapi",
]
[[package]]
name = "crossterm_winapi"
version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3a6966607622438301997d3dac0d2f6e9a90c68bb6bc1785ea98456ab93c0507"
dependencies = [
"winapi",
]
[[package]] [[package]]
name = "crunchy" name = "crunchy"
version = "0.2.2" version = "0.2.2"
@ -1318,6 +1423,15 @@ dependencies = [
"subtle", "subtle",
] ]
[[package]]
name = "dirs"
version = "4.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ca3aa72a6f96ea37bbc5aa912f6788242832f75369bdfdadcb0e38423f100059"
dependencies = [
"dirs-sys",
]
[[package]] [[package]]
name = "dirs-next" name = "dirs-next"
version = "2.0.0" version = "2.0.0"
@ -1328,6 +1442,17 @@ dependencies = [
"dirs-sys-next", "dirs-sys-next",
] ]
[[package]]
name = "dirs-sys"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b1d1d91c932ef41c0f2663aa8b0ca0342d444d842c06914aa0a7e352d0bada6"
dependencies = [
"libc",
"redox_users",
"winapi",
]
[[package]] [[package]]
name = "dirs-sys-next" name = "dirs-sys-next"
version = "0.1.2" version = "0.1.2"
@ -2435,6 +2560,18 @@ dependencies = [
"tokio-rustls", "tokio-rustls",
] ]
[[package]]
name = "hyper-timeout"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bbb958482e8c7be4bc3cf272a766a2b0bf1a6755e7a6ae777f017a31d11b13b1"
dependencies = [
"hyper",
"pin-project-lite",
"tokio",
"tokio-io-timeout",
]
[[package]] [[package]]
name = "iana-time-zone" name = "iana-time-zone"
version = "0.1.53" version = "0.1.53"
@ -2828,6 +2965,19 @@ dependencies = [
"adler", "adler",
] ]
[[package]]
name = "mio"
version = "0.7.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8067b404fe97c70829f082dec8bcf4f71225d7eaea1d8645349cb76fa06205cc"
dependencies = [
"libc",
"log",
"miow",
"ntapi",
"winapi",
]
[[package]] [[package]]
name = "mio" name = "mio"
version = "0.8.6" version = "0.8.6"
@ -2840,6 +2990,15 @@ dependencies = [
"windows-sys 0.45.0", "windows-sys 0.45.0",
] ]
[[package]]
name = "miow"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9f1c5b025cda876f66ef43a113f91ebc9f4ccef34843000e0adf6ebbab84e21"
dependencies = [
"winapi",
]
[[package]] [[package]]
name = "moka" name = "moka"
version = "0.10.0" version = "0.10.0"
@ -2905,6 +3064,25 @@ version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e9e591e719385e6ebaeb5ce5d3887f7d5676fceca6411d1925ccc95745f3d6f7" checksum = "e9e591e719385e6ebaeb5ce5d3887f7d5676fceca6411d1925ccc95745f3d6f7"
[[package]]
name = "ntapi"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c28774a7fd2fbb4f0babd8237ce554b73af68021b5f695a3cebd6c59bac0980f"
dependencies = [
"winapi",
]
[[package]]
name = "nu-ansi-term"
version = "0.46.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84"
dependencies = [
"overload",
"winapi",
]
[[package]] [[package]]
name = "num" name = "num"
version = "0.4.0" version = "0.4.0"
@ -3137,6 +3315,18 @@ dependencies = [
"syn", "syn",
] ]
[[package]]
name = "overload"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
[[package]]
name = "owo-colors"
version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2386b4ebe91c2f7f51082d4cefa145d030e33a1842a96b12e4885cc3c01f7a55"
[[package]] [[package]]
name = "pagerduty-rs" name = "pagerduty-rs"
version = "0.1.6" version = "0.1.6"
@ -3589,6 +3779,38 @@ dependencies = [
"winapi", "winapi",
] ]
[[package]]
name = "prost"
version = "0.11.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e48e50df39172a3e7eb17e14642445da64996989bc212b583015435d39a58537"
dependencies = [
"bytes",
"prost-derive",
]
[[package]]
name = "prost-derive"
version = "0.11.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4ea9b0f8cbe5e15a8a042d030bd96668db28ecb567ec37d691971ff5731d2b1b"
dependencies = [
"anyhow",
"itertools",
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "prost-types"
version = "0.11.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "379119666929a1afd7a043aa6cf96fa67a6dce9af60c88095a4686dbce4c9c88"
dependencies = [
"prost",
]
[[package]] [[package]]
name = "ptr_meta" name = "ptr_meta"
version = "0.1.4" version = "0.1.4"
@ -4217,7 +4439,7 @@ dependencies = [
"regex", "regex",
"sea-schema", "sea-schema",
"tracing", "tracing",
"tracing-subscriber", "tracing-subscriber 0.3.16",
"url", "url",
] ]
@ -4248,7 +4470,7 @@ dependencies = [
"sea-orm-cli", "sea-orm-cli",
"sea-schema", "sea-schema",
"tracing", "tracing",
"tracing-subscriber", "tracing-subscriber 0.3.16",
] ]
[[package]] [[package]]
@ -4652,6 +4874,27 @@ dependencies = [
"lazy_static", "lazy_static",
] ]
[[package]]
name = "signal-hook"
version = "0.3.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "732768f1176d21d09e076c23a93123d40bba92d50c4058da34d45c8de8e682b9"
dependencies = [
"libc",
"signal-hook-registry",
]
[[package]]
name = "signal-hook-mio"
version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29ad2e15f37ec9a6cc544097b78a1ec90001e9f71b81338ca39f430adaca99af"
dependencies = [
"libc",
"mio 0.7.14",
"signal-hook",
]
[[package]] [[package]]
name = "signal-hook-registry" name = "signal-hook-registry"
version = "1.4.1" version = "1.4.1"
@ -5186,7 +5429,7 @@ dependencies = [
"bytes", "bytes",
"libc", "libc",
"memchr", "memchr",
"mio", "mio 0.8.6",
"num_cpus", "num_cpus",
"parking_lot 0.12.1", "parking_lot 0.12.1",
"pin-project-lite", "pin-project-lite",
@ -5197,6 +5440,45 @@ dependencies = [
"windows-sys 0.45.0", "windows-sys 0.45.0",
] ]
[[package]]
name = "tokio-console"
version = "0.1.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fce5f0a53cd350a00b3a37dcb85758eb3c333beeb334b40584f7747b1e01374e"
dependencies = [
"atty",
"clap",
"clap_complete",
"color-eyre",
"console-api",
"crossterm",
"dirs",
"futures",
"h2",
"hdrhistogram",
"humantime",
"once_cell",
"prost-types",
"regex",
"serde",
"tokio",
"toml 0.5.11",
"tonic",
"tracing",
"tracing-subscriber 0.3.16",
"tui",
]
[[package]]
name = "tokio-io-timeout"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "30b74022ada614a1b4834de765f9bb43877f910cc8ce4be40e89042c9223a8bf"
dependencies = [
"pin-project-lite",
"tokio",
]
[[package]] [[package]]
name = "tokio-macros" name = "tokio-macros"
version = "1.8.2" version = "1.8.2"
@ -5333,6 +5615,38 @@ dependencies = [
"toml_datetime 0.6.1", "toml_datetime 0.6.1",
] ]
[[package]]
name = "tonic"
version = "0.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f219fad3b929bef19b1f86fbc0358d35daed8f2cac972037ac0dc10bbb8d5fb"
dependencies = [
"async-stream",
"async-trait",
"axum",
"base64 0.13.1",
"bytes",
"futures-core",
"futures-util",
"h2",
"http",
"http-body",
"hyper",
"hyper-timeout",
"percent-encoding",
"pin-project",
"prost",
"prost-derive",
"tokio",
"tokio-stream",
"tokio-util",
"tower",
"tower-layer",
"tower-service",
"tracing",
"tracing-futures",
]
[[package]] [[package]]
name = "tower" name = "tower"
version = "0.4.13" version = "0.4.13"
@ -5341,9 +5655,13 @@ checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c"
dependencies = [ dependencies = [
"futures-core", "futures-core",
"futures-util", "futures-util",
"indexmap",
"pin-project", "pin-project",
"pin-project-lite", "pin-project-lite",
"rand",
"slab",
"tokio", "tokio",
"tokio-util",
"tower-layer", "tower-layer",
"tower-service", "tower-service",
"tracing", "tracing",
@ -5411,6 +5729,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24eb03ba0eab1fd845050058ce5e616558e8f8d8fca633e6b163fe25c797213a" checksum = "24eb03ba0eab1fd845050058ce5e616558e8f8d8fca633e6b163fe25c797213a"
dependencies = [ dependencies = [
"once_cell", "once_cell",
"valuable",
]
[[package]]
name = "tracing-error"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b4d7c0b83d4a500748fa5879461652b361edf5c9d51ede2a2ac03875ca185e24"
dependencies = [
"tracing",
"tracing-subscriber 0.2.25",
] ]
[[package]] [[package]]
@ -5423,6 +5752,28 @@ dependencies = [
"tracing", "tracing",
] ]
[[package]]
name = "tracing-log"
version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78ddad33d2d10b1ed7eb9d1f518a5674713876e97e5bb9b7345a7984fbb4f922"
dependencies = [
"lazy_static",
"log",
"tracing-core",
]
[[package]]
name = "tracing-subscriber"
version = "0.2.25"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0e0d2eaa99c3c2e41547cfa109e910a68ea03823cccad4a0525dcbc9b01e8c71"
dependencies = [
"sharded-slab",
"thread_local",
"tracing-core",
]
[[package]] [[package]]
name = "tracing-subscriber" name = "tracing-subscriber"
version = "0.3.16" version = "0.3.16"
@ -5430,12 +5781,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6176eae26dd70d0c919749377897b54a9276bd7061339665dd68777926b5a70" checksum = "a6176eae26dd70d0c919749377897b54a9276bd7061339665dd68777926b5a70"
dependencies = [ dependencies = [
"matchers", "matchers",
"nu-ansi-term",
"once_cell", "once_cell",
"regex", "regex",
"sharded-slab", "sharded-slab",
"smallvec",
"thread_local", "thread_local",
"tracing", "tracing",
"tracing-core", "tracing-core",
"tracing-log",
] ]
[[package]] [[package]]
@ -5450,6 +5804,19 @@ version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3528ecfd12c466c6f163363caf2d02a71161dd5e1cc6ae7b34207ea2d42d81ed" checksum = "3528ecfd12c466c6f163363caf2d02a71161dd5e1cc6ae7b34207ea2d42d81ed"
[[package]]
name = "tui"
version = "0.16.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39c8ce4e27049eed97cfa363a5048b09d995e209994634a0efc26a14ab6c0c23"
dependencies = [
"bitflags",
"cassowary",
"crossterm",
"unicode-segmentation",
"unicode-width",
]
[[package]] [[package]]
name = "tungstenite" name = "tungstenite"
version = "0.17.3" version = "0.17.3"
@ -5646,6 +6013,12 @@ dependencies = [
"serde", "serde",
] ]
[[package]]
name = "valuable"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d"
[[package]] [[package]]
name = "vcpkg" name = "vcpkg"
version = "0.2.15" version = "0.2.15"
@ -5798,6 +6171,7 @@ dependencies = [
"axum-client-ip", "axum-client-ip",
"axum-macros", "axum-macros",
"chrono", "chrono",
"console-subscriber",
"counter", "counter",
"deferred-rate-limiter", "deferred-rate-limiter",
"derive_more", "derive_more",
@ -5843,6 +6217,7 @@ dependencies = [
"thread-fast-rng", "thread-fast-rng",
"time 0.3.20", "time 0.3.20",
"tokio", "tokio",
"tokio-console",
"tokio-stream", "tokio-stream",
"toml 0.7.2", "toml 0.7.2",
"tower", "tower",

5
web3_proxy/Cargo.toml
View File

@ -7,8 +7,9 @@ default-run = "web3_proxy_cli"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[features] [features]
default = ["deadlock_detection"] default = ["deadlock_detection", "tokio-console"]
deadlock_detection = ["parking_lot/deadlock_detection"] deadlock_detection = ["parking_lot/deadlock_detection"]
tokio-console = ["dep:tokio-console", "dep:console-subscriber"]
# TODO: turn tokio-console on with a feature. console-subscriber = { version = "0.1.7" } # TODO: turn tokio-console on with a feature. console-subscriber = { version = "0.1.7" }
@ -31,6 +32,7 @@ axum = { version = "0.6.10", features = ["headers", "ws"] }
axum-client-ip = "0.4.0" axum-client-ip = "0.4.0"
axum-macros = "0.3.5" axum-macros = "0.3.5"
chrono = "0.4.23" chrono = "0.4.23"
console-subscriber = { version = "*", optional = true }
counter = "0.5.7" counter = "0.5.7"
derive_more = "0.99.17" derive_more = "0.99.17"
dotenv = "0.15.0" dotenv = "0.15.0"
@ -69,6 +71,7 @@ serde_prometheus = "0.2.1"
siwe = "0.5.0" siwe = "0.5.0"
time = "0.3.20" time = "0.3.20"
tokio = { version = "1.26.0", features = ["full"] } tokio = { version = "1.26.0", features = ["full"] }
tokio-console = { version = "*", optional = true }
tokio-stream = { version = "0.1.12", features = ["sync"] } tokio-stream = { version = "0.1.12", features = ["sync"] }
toml = "0.7.2" toml = "0.7.2"
tower = "0.4.13" tower = "0.4.13"

51
web3_proxy/src/bin/web3_proxy_cli/main.rs
View File

@ -118,6 +118,10 @@ fn main() -> anyhow::Result<()> {
// if RUST_LOG isn't set, configure a default // if RUST_LOG isn't set, configure a default
// TODO: is there a better way to do this? // TODO: is there a better way to do this?
#[cfg(tokio_console)]
console_subscriber::init();
#[cfg(not(tokio_console))]
let rust_log = match std::env::var("RUST_LOG") { let rust_log = match std::env::var("RUST_LOG") {
Ok(x) => x, Ok(x) => x,
Err(_) => match std::env::var("WEB3_PROXY_TRACE").map(|x| x == "true") { Err(_) => match std::env::var("WEB3_PROXY_TRACE").map(|x| x == "true") {
@ -190,35 +194,38 @@ fn main() -> anyhow::Result<()> {
(None, None) (None, None)
}; };
let logger = env_logger::builder().parse_filters(&rust_log).build(); #[cfg(not(tokio_console))]
{
let logger = env_logger::builder().parse_filters(&rust_log).build();
let max_level = logger.filter(); let max_level = logger.filter();
// connect to sentry for error reporting // connect to sentry for error reporting
// if no sentry, only log to stdout // if no sentry, only log to stdout
let _sentry_guard = if let Some(sentry_url) = cli_config.sentry_url.clone() { let _sentry_guard = if let Some(sentry_url) = cli_config.sentry_url.clone() {
let logger = sentry::integrations::log::SentryLogger::with_dest(logger); let logger = sentry::integrations::log::SentryLogger::with_dest(logger);
log::set_boxed_logger(Box::new(logger)).unwrap(); log::set_boxed_logger(Box::new(logger)).unwrap();
let guard = sentry::init(( let guard = sentry::init((
sentry_url, sentry_url,
sentry::ClientOptions { sentry::ClientOptions {
release: sentry::release_name!(), release: sentry::release_name!(),
// TODO: Set this a to lower value (from config) in production // TODO: Set this a to lower value (from config) in production
traces_sample_rate: 1.0, traces_sample_rate: 1.0,
..Default::default() ..Default::default()
}, },
)); ));
Some(guard) Some(guard)
} else { } else {
log::set_boxed_logger(Box::new(logger)).unwrap(); log::set_boxed_logger(Box::new(logger)).unwrap();
None None
}; };
log::set_max_level(max_level); log::set_max_level(max_level);
}
info!("{}", APP_USER_AGENT); info!("{}", APP_USER_AGENT);

2
web3_proxy/src/jsonrpc.rs
View File

@ -279,7 +279,7 @@ impl JsonRpcForwardedResponse {
} }
} }
} else { } else {
unimplemented!(); return Err(anyhow::anyhow!("unexpected ethers error!"));
} }
} }
} }

6
web3_proxy/src/rpcs/many.rs
View File

@ -881,7 +881,7 @@ impl Web3Rpcs {
request.id.clone(), request.id.clone(),
) { ) {
Ok(response) => { Ok(response) => {
if let Some(error) = &response.error.as_ref() { if let Some(error) = response.error.as_ref() {
// trace!(?response, "rpc error"); // trace!(?response, "rpc error");
if let Some(request_metadata) = request_metadata { if let Some(request_metadata) = request_metadata {
@ -956,8 +956,10 @@ impl Web3Rpcs {
// TODO: emit a stat. if a server is getting skipped a lot, something is not right // TODO: emit a stat. if a server is getting skipped a lot, something is not right
// TODO: if we get a TrySendError, reconnect. wait why do we see a trysenderror on a dual provider? shouldn't it be using reqwest
debug!( debug!(
"Backend server error on {}! Retrying on another. err={:?}", "Backend server error on {}! Retrying on another. err={:#?}",
rpc, err rpc, err
); );

71
web3_proxy/src/rpcs/one.rs
View File

@ -9,8 +9,8 @@ use crate::rpcs::request::RequestRevertHandler;
use anyhow::{anyhow, Context}; use anyhow::{anyhow, Context};
use ethers::prelude::{Bytes, Middleware, ProviderError, TxHash, H256, U64}; use ethers::prelude::{Bytes, Middleware, ProviderError, TxHash, H256, U64};
use ethers::types::{Address, Transaction, U256}; use ethers::types::{Address, Transaction, U256};
use futures::future::try_join_all;
use futures::StreamExt; use futures::StreamExt;
use futures::future::try_join_all;
use futures::stream::FuturesUnordered; use futures::stream::FuturesUnordered;
use log::{debug, error, info, trace, warn, Level}; use log::{debug, error, info, trace, warn, Level};
use migration::sea_orm::DatabaseConnection; use migration::sea_orm::DatabaseConnection;
@ -702,12 +702,14 @@ impl Web3Rpc {
RequestRevertHandler::ErrorLevel RequestRevertHandler::ErrorLevel
}; };
let mut delay_start = false;
// this does loop. just only when reconnect is enabled // this does loop. just only when reconnect is enabled
#[allow(clippy::never_loop)] #[allow(clippy::never_loop)]
loop { loop {
debug!("subscription loop started"); debug!("subscription loop started");
let mut futures = FuturesUnordered::new(); let mut futures = vec![];
let http_interval_receiver = http_interval_sender.as_ref().map(|x| x.subscribe()); let http_interval_receiver = http_interval_sender.as_ref().map(|x| x.subscribe());
@ -723,7 +725,7 @@ impl Web3Rpc {
block_sender.as_ref(), block_sender.as_ref(),
chain_id, chain_id,
authorization.db_conn.as_ref(), authorization.db_conn.as_ref(),
false, delay_start,
) )
.await?; .await?;
@ -865,48 +867,41 @@ impl Web3Rpc {
futures.push(flatten_handle(tokio::spawn(f))); futures.push(flatten_handle(tokio::spawn(f)));
} }
while let Some(x) = futures.next().await { match try_join_all(futures).await {
match x { Ok(_) => {
Ok(_) => { // future exited without error
// future exited without error // TODO: think about this more. we never set it to false. this can't be right
// TODO: think about this more. we never set it to false. this can't be right break;
info!("future on {} exited successfully", self) }
Err(err) => {
let disconnect_sender = self.disconnect_watch.as_ref().unwrap();
if self.reconnect.load(atomic::Ordering::Acquire) {
warn!("{} connection ended. reconnecting. err={:?}", self, err);
// TODO: i'm not sure if this is necessary, but telling everything to disconnect seems like a better idea than relying on timeouts and dropped futures.
disconnect_sender.send_replace(true);
disconnect_sender.send_replace(false);
// we call retrying_connect here with initial_delay=true. above, initial_delay=false
delay_start = true;
continue;
} }
Err(err) => {
if self.reconnect.load(atomic::Ordering::Acquire) {
warn!("{} connection ended. reconnecting. err={:?}", self, err);
let disconnect_sender = self.disconnect_watch.as_ref().unwrap(); // reconnect is not enabled.
if *disconnect_receiver.borrow() {
info!("{} is disconnecting", self);
break;
} else {
error!("{} subscription exited. err={:?}", self, err);
// TODO: i'm not sure if this is necessary, but telling everything to disconnect seems like a better idea than relying on timeouts and dropped futures. disconnect_sender.send_replace(true);
disconnect_sender.send_replace(true);
disconnect_sender.send_replace(false);
// we call retrying_connect here with initial_delay=true. above, initial_delay=false break;
self.retrying_connect(
block_sender.as_ref(),
chain_id,
authorization.db_conn.as_ref(),
true,
)
.await?;
continue;
}
// reconnect is not enabled.
if *disconnect_receiver.borrow() {
info!("{} is disconnecting", self);
break;
} else {
error!("{} subscription exited. err={:?}", self, err);
break;
}
} }
} }
} }
break;
} }
info!("all subscriptions on {} completed", self); info!("all subscriptions on {} completed", self);

3
web3_proxy/src/rpcs/request.rs
View File

@ -164,8 +164,9 @@ impl OpenRequestHandle {
}; };
let mut logged = false; let mut logged = false;
while provider.is_none() { while provider.is_none() || provider.as_ref().map(|x| !x.ready()).unwrap() {
// trace!("waiting on provider: locking..."); // trace!("waiting on provider: locking...");
// TODO: i dont like this. subscribing to a channel could be better
sleep(Duration::from_millis(100)).await; sleep(Duration::from_millis(100)).await;
if !logged { if !logged {