deps: use Tower 0.4 from git instead of 0.3.1.

This addresses at least three pain points:

- we were affected by bugs that were already fixed in git, but not in
  the released crate;
- we can use service combinators to transform requests and responses;
- we can use the hedge middleware.

The version in git is still marked as 0.3.1 but these changes will be
part of tower 0.4: tower-rs/tower#431

Cargo.toml

@@ -19,3 +19,6 @@ panic = "abort"
 
 [profile.release]
 panic = "abort"
+
+[patch.crates-io]
+tower = { git = "https://github.com/tower-rs/tower", rev = "ad348d8" }

tower-batch/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2018"
 
 [dependencies]
 tokio = { version = "0.2.22", features = ["time", "sync", "stream", "tracing"] }
-tower = "0.3"
+tower = { version = "0.3", features = ["util", "buffer"] }
 futures-core = "0.3.5"
 pin-project = "0.4.23"
 tracing = "0.1.19"

zebra-consensus/Cargo.toml

@@ -18,7 +18,7 @@ futures-util = "0.3.5"
 metrics = "0.12"
 thiserror = "1.0.20"
 tokio = { version = "0.2.22", features = ["time", "sync", "stream", "tracing"] }
-tower = "0.3"
+tower = { version = "0.3", features = ["timeout", "util", "buffer"] }
 tower-util = "0.3"
 tracing = "0.1.19"
 tracing-futures = "0.2.4"

zebra-consensus/src/primitives/redjubjub.rs

@@ -15,10 +15,9 @@ use once_cell::sync::Lazy;
 use rand::thread_rng;
 use redjubjub::{batch, *};
 use tokio::sync::broadcast::{channel, RecvError, Sender};
-use tower::Service;
+use tower::{util::ServiceFn, Service};
 use tower_batch::{Batch, BatchControl};
 use tower_fallback::Fallback;
-use tower_util::ServiceFn;
 
 /// Global batch verification context for RedJubjub signatures.
 ///

zebra-network/Cargo.toml

@@ -24,8 +24,7 @@ thiserror = "1"
 futures = "0.3"
 tokio = { version = "0.2.22", features = ["net", "time", "stream", "tracing", "macros"] }
 tokio-util = { version = "0.2", features = ["codec"] }
-tower = "0.3"
-tower-load = "0.3"
+tower = { version = "0.3", features = ["retry", "discover", "load", "load-shed", "timeout", "util", "buffer"] }
 
 metrics = "0.12"
 tracing = "0.1"

zebra-network/src/peer_set/initialize.rs

@@ -19,13 +19,9 @@ use tokio::{
     sync::broadcast,
 };
 use tower::{
-    buffer::Buffer,
-    discover::{Change, ServiceStream},
-    layer::Layer,
-    util::BoxService,
-    Service, ServiceExt,
+    buffer::Buffer, discover::Change, layer::Layer, load::peak_ewma::PeakEwmaDiscover,
+    util::BoxService, Service, ServiceExt,
 };
-use tower_load::{peak_ewma::PeakEwmaDiscover, NoInstrument};
 
 use crate::{
     constants, peer, timestamp_collector::TimestampCollector, AddressBook, BoxError, Config,
@@ -106,14 +102,12 @@ where
     // Connect the rx end to a PeerSet, wrapping new peers in load instruments.
     let peer_set = PeerSet::new(
         PeakEwmaDiscover::new(
-            ServiceStream::new(
-                // ServiceStream interprets an error as stream termination,
-                // so discard any errored connections...
-                peerset_rx.filter(|result| future::ready(result.is_ok())),
-            ),
+            // Discover interprets an error as stream termination,
+            // so discard any errored connections...
+            peerset_rx.filter(|result| future::ready(result.is_ok())),
             constants::EWMA_DEFAULT_RTT,
             constants::EWMA_DECAY_TIME,
-            NoInstrument,
+            tower::load::CompleteOnResponse::default(),
         ),
         demand_tx.clone(),
         handle_rx,

zebra-network/src/peer_set/set.rs

@@ -20,9 +20,9 @@ use tokio::sync::{broadcast, oneshot::error::TryRecvError};
 use tokio::task::JoinHandle;
 use tower::{
     discover::{Change, Discover},
+    load::Load,
     Service,
 };
-use tower_load::Load;
 
 use crate::{
     protocol::{
@@ -131,7 +131,10 @@ where
     fn poll_discover(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>> {
         use futures::ready;
         loop {
-            match ready!(Pin::new(&mut self.discover).poll_discover(cx)).map_err(Into::into)? {
+            match ready!(Pin::new(&mut self.discover).poll_discover(cx))
+                .ok_or("discovery stream closed")?
+                .map_err(Into::into)?
+            {
                 Change::Remove(key) => {
                     trace!(?key, "got Change::Remove from Discover");
                     self.remove(&key);

zebra-state/Cargo.toml

@@ -19,7 +19,7 @@ sled = "0.34.4"
 
 futures = "0.3.5"
 metrics = "0.12"
-tower = "0.3.1"
+tower = { version = "0.3.1", features = ["buffer", "util"] }
 tracing = "0.1"
 tracing-error = "0.1.2"
 thiserror = "1.0.20"

zebra-test/Cargo.toml

@@ -10,7 +10,7 @@ edition = "2018"
 [dependencies]
 hex = "0.4.2"
 lazy_static = "1.4.0"
-tower = "0.3.1"
+tower = { version = "0.3.1", features = ["util"] }
 futures = "0.3.5"
 color-eyre = "0.5.3"
 tracing = "0.1.19"