From c482e20c0270083eabbace6d21018075f13af0d5 Mon Sep 17 00:00:00 2001 From: Krishna Vishal Date: Sun, 25 Oct 2026 18:08:44 +0530 Subject: [PATCH] fix(gateways): stop Kafka Produce sends at the request deadline A send that outlived its request landed later, so a client retry stored the batch twice; failed sends that wrote nothing answer 6, not -1. --- Cargo.lock | 1 + gateways/kafka/Cargo.toml | 3 + gateways/kafka/README.md | 16 +- gateways/kafka/docs/TEST_SUITE.md | 2 +- gateways/kafka/src/bridge/error.rs | 57 +++++++ .../kafka/src/bridge/iggy_bridge/produce.rs | 23 ++- .../kafka/src/protocol/handlers/produce.rs | 20 ++- .../kafka/tests/produce_real_bridge_tests.rs | 148 +++++++++++++++++- 8 files changed, 254 insertions(+), 16 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1ac6d1461d..681a4aac4b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7289,6 +7289,7 @@ dependencies = [ "futures", "iggy", "iggy_binary_protocol", + "iggy_common", "iggy_kafka_wire", "kafka-protocol", "libc", diff --git a/gateways/kafka/Cargo.toml b/gateways/kafka/Cargo.toml index 4b90af8f9d..8f342104b7 100644 --- a/gateways/kafka/Cargo.toml +++ b/gateways/kafka/Cargo.toml @@ -39,6 +39,9 @@ base64 = { workspace = true } bytes = { workspace = true } futures = { workspace = true } iggy = { workspace = true } +# `with_binary_deadline`, which hands a Produce request's deadline to the SDK. The prelude does not +# export it. +iggy_common = { workspace = true } iggy_kafka_wire = { workspace = true } # The codec features stay on for the encoder, which the batch tests use to build fixtures. kafka-protocol = { workspace = true, features = ["gzip", "lz4", "snappy", "zstd"] } diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index c32b75d3af..e04a8573f8 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -477,7 +477,7 @@ One partition, one Iggy send. Each partition answers for itself. | `acks` | `0`, `1`, `-1` write the same. Other values: 21, before any per-partition check. | | `acks=0` | Writes, answers nothing. Any failed partition closes the connection. | | Topics | Creates none. Missing: 3. Bad name: 17. | -| `timeout_ms` | Honored, max 20 s. Past it: 7. | +| `timeout_ms` | Honored, max 20 s. The Iggy send stops by then too. Past it: 7. A send that never went out: 6. | | Compression | gzip, snappy, lz4. zstd from v7, else 76. | | Producer id | Picks the producer's Iggy session. See [Concurrency ceiling](#concurrency-ceiling). | | Producer epoch, sequence | Ignored, so a retry writes twice. | @@ -495,8 +495,8 @@ One partition, one Iggy send. Each partition answers for itself. | 10 | Record, send or partition too large, even alone | Java splits multi-record batches. Else fails. | | 87 | Record the gateway cannot map. Reason in `error_message` from v8. | Fails. | | 35 | Transactional or control batch | Fails. | -| 6 | Request budget ran out. Nothing written. | Retries. | -| 7 | Deadline passed: in the send, waiting for decode memory, or behind the producer's own earlier send. Or connection lost mid-send. May be written. | Retries. Can duplicate. | +| 6 | Request budget ran out. Or the send did not reach Iggy by the deadline, or Iggy refused it: no primary, partition busy or recreated, client shut down. Nothing written. | Retries. | +| 7 | Deadline passed: after the send went out, waiting for decode memory, or behind the producer's own earlier send. Or connection lost mid-send. Possibly written. | Retries. Can duplicate. | Partition cap: `max_frame_size` decompressed bytes, `max_frame_size / 64` record slots, 3 headers per slot. Past it: 10. @@ -610,8 +610,10 @@ Iggy becomes unreachable later. `send_records` uses the Produce deadline instead: - One request of a producer runs at a time, on that producer's client. -- A send past its deadline keeps the producer's turn until it ends. That producer's requests answer - 7 meanwhile. Other producers send on. +- The SDK stops retrying the send 50 ms before the deadline. So it sends nothing after the request + answered, and the client's retry is the only copy of a send that never went out. +- A send still running past its deadline keeps the producer's turn until it ends. That producer's + requests answer 7 meanwhile. Other producers send on. - After 45 s the gateway gives up on the send. It then closes that connection before the turn frees, so the send cannot land after a later one. The SDK ends its write within 30 s more, unless Iggy stopped reading: the producer then waits until that connection breaks. @@ -821,6 +823,10 @@ instead of -1, because the Java `AdminClient` fails a partition on any non-retri incarnation of the partition, as after a delete and recreate; the retry probes again and answers 3 if the topic is gone. ListOffsets reads do not get `RequestTooOld` today, so that arm is defensive. +Produce also answers `HistoryUnavailable`, `LifecycleBusy` and `ClientShutdown` with 6 instead of +-1 (`to_produce_error_code`). The SDK returns them only for a send that wrote nothing. A send that +possibly landed ends as `OperationUncertain`, which answers 7. + ### Server limits the gateway inherits These are Iggy server limits, not gateway settings. A Kafka client cannot act on any of them, so diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md index 3e23962cb7..f73e4bb97d 100644 --- a/gateways/kafka/docs/TEST_SUITE.md +++ b/gateways/kafka/docs/TEST_SUITE.md @@ -84,7 +84,7 @@ file under `tests/` anymore. | [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases, and the bridge dialing a TLS-only server over TLS | No (needs the `iggy-server` binary - see Prerequisites) | | [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, lookups by time and `MAX_TIMESTAMP`, the `-1` answers, code 3, several partitions looked up by time in one request, and a stalled lookup leaving the later partitions their answers | No (needs the `iggy-server` binary - see Prerequisites) | | [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) | -| [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition | No (needs the `iggy-server` binary - see Prerequisites) | +| [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition. Through a relay that cuts Iggy off, a send that misses its deadline answers 6, and the client's retry stores the batch once | No (needs the `iggy-server` binary - see Prerequisites) | | [`create_partitions_real_bridge_tests.rs`](../tests/create_partitions_real_bridge_tests.rs) | CreatePartitions (key 37) through the whole handler against a real, spawned `iggy-server`: a grow appends indices and shows in Metadata's listing, the grow-only rule (37), a missing topic (3), `validate_only`, duplicate names (42), the manual assignment checks (39), the per-request and topic caps, the `manage_topics` gate (29), a duplicate answered before authorization (42, then 29 for the rest), two overlapping grows to one count (one 0, one 37), a fixed partition topic (44, `validate_only` too), and v0/v1 requests | No (needs the `iggy-server` binary - see Prerequisites) | | [`offsets_real_bridge_tests.rs`](../tests/offsets_real_bridge_tests.rs) | OffsetCommit (key 8) and OffsetFetch (key 9) through the whole handler against a real, spawned `iggy-server`: offsets stored as sent, the Iggy external group key, a negative commit as a delete, -1 for nothing committed, a null topic list, and a gateway restart ([#3542](https://github.com/apache/iggy/issues/3542) acceptance) | No (needs the `iggy-server` binary - see Prerequisites) | | [`offset_groups_real_bridge_tests.rs`](../tests/offset_groups_real_bridge_tests.rs) | ListGroups (key 16), DescribeGroups (key 15), DeleteGroups (key 42) and OffsetDelete (key 47) through the whole handler against a real, spawned `iggy-server`. A group with offsets and no members is `Empty`, and a live group is listed once. Also the filters, a commit after the last scan, a new gateway, a native Iggy group, and only the live groups once a scan failed with Iggy down. DeleteGroups refuses a group with members, and after they leave deletes it on every topic. Through a relay that holds an Iggy call past its deadline, a commit and a group deletion each keep the other out until the call ends, and past the SDK's 30 s deadline until a replay proves what the call did. A delete that an earlier delete in doubt kept from Iggy lets go of its group. A deletion that ends after its request answered 14 answers 0 to the retry. OffsetDelete keeps the offsets of a topic that a member subscribes to, deletes the others, and finds a group that another gateway committed for since the last scan | No (needs the `iggy-server` binary - see Prerequisites) | diff --git a/gateways/kafka/src/bridge/error.rs b/gateways/kafka/src/bridge/error.rs index d348098aff..403a9e071c 100644 --- a/gateways/kafka/src/bridge/error.rs +++ b/gateways/kafka/src/bridge/error.rs @@ -196,6 +196,22 @@ impl BridgeError { } } + /// [`Self::to_kafka_error_code`] for Produce. The SDK returns `HistoryUnavailable`, + /// `LifecycleBusy` and `ClientShutdown` only for a send that wrote nothing, because a send + /// that may have landed ends as `OperationUncertain`. So these answer 6, which clients retry, + /// not -1, which loses the records. + #[must_use] + pub const fn to_produce_error_code(&self) -> i16 { + match self { + Self::Iggy( + IggyError::HistoryUnavailable + | IggyError::LifecycleBusy + | IggyError::ClientShutdown, + ) => ERROR_NOT_LEADER_OR_FOLLOWER, + _ => self.to_kafka_error_code(), + } + } + /// True when the same call can succeed later: Kafka clients retry its code, 6 or 7. A wrong /// password, a missing permission or a bad configuration is not. #[must_use] @@ -421,6 +437,47 @@ mod tests { } } + #[test] + fn given_a_send_that_wrote_nothing_when_mapped_for_produce_should_answer_retriable() { + let wrote_nothing = [ + IggyError::HistoryUnavailable, + IggyError::LifecycleBusy, + IggyError::ClientShutdown, + ]; + for error in wrote_nothing { + let err = BridgeError::Iggy(error); + assert_eq!( + err.to_kafka_error_code(), + ERROR_UNKNOWN_SERVER_ERROR, + "{err}" + ); + assert_eq!( + err.to_produce_error_code(), + ERROR_NOT_LEADER_OR_FOLLOWER, + "{err}" + ); + } + } + + #[test] + fn given_a_send_that_may_have_landed_when_mapped_for_produce_should_keep_the_shared_code() { + let may_have_landed = [ + ( + BridgeError::Iggy(IggyError::OperationUncertain(1)), + ERROR_REQUEST_TIMED_OUT, + ), + (BridgeError::Timeout, ERROR_REQUEST_TIMED_OUT), + // A retry after the client was evicted could write it twice. + ( + BridgeError::Iggy(IggyError::RequestTooOld), + ERROR_UNKNOWN_SERVER_ERROR, + ), + ]; + for (err, code) in may_have_landed { + assert_eq!(err.to_produce_error_code(), code, "{err}"); + } + } + #[test] fn given_any_other_failure_when_mapped_for_list_offsets_should_keep_the_shared_code() { let others = [ diff --git a/gateways/kafka/src/bridge/iggy_bridge/produce.rs b/gateways/kafka/src/bridge/iggy_bridge/produce.rs index d5dc03266f..1966924652 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/produce.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/produce.rs @@ -27,6 +27,7 @@ use std::time::Duration; use futures::future::join_all; use iggy::prelude::{IggyClient, IggyError, IggyMessage, MessageClient, Partitioning}; +use iggy_common::{binary_operation_deadline, with_binary_deadline}; use tokio::runtime::Handle; use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot}; use tokio::time::{Instant, MissedTickBehavior, interval, sleep_until, timeout, timeout_at}; @@ -39,6 +40,10 @@ use crate::bridge::error::BridgeError; /// Frame bytes besides the messages: request header, batch header and ids. const SEND_FRAME_ROOM: u64 = 4 * 1024; +/// How much sooner than its request the SDK stops a send. The SDK's own answer then tells whether +/// the send went out, before the request gives up on it with 7. +const SDK_STOP_ROOM: Duration = Duration::from_millis(50); + /// Sessions that connect and sign in at once. A sign-in costs Iggy a password hash on a shard /// thread, so a burst of new producers must not start them all together. const SESSION_OPENS: usize = 4; @@ -374,6 +379,9 @@ impl IggyBridge { /// `Ok(None)`: committed, but the server named no offset (a duplicate request, or a journal /// entry that is gone). Do not retry it. /// + /// The SDK stops the send by `deadline`, so no attempt of it starts after the request + /// answered, and a client retry is the only copy of a send that never went out. + /// /// A send still running at `deadline` keeps `turn` and `kept`, and hands back `None` for /// them. So the producer's later requests answer 7 meanwhile and cannot overtake it, and what /// `kept` reserves for `messages` stays reserved. Other producers send on. @@ -386,7 +394,8 @@ impl IggyBridge { /// [`BridgeError::Timeout`] if `deadline` passes. The send may have landed. /// [`BridgeError::SendLost`] if the connection broke during the send. It may have landed. /// [`BridgeError::Iggy`] if the session cannot connect, if the stream, topic or partition is - /// missing, or if the send is too large. + /// missing, or if the send is too large. Also when the SDK stopped the send at `deadline`: + /// `OperationUncertain` if it went out, so it may have landed, and its refusal if it did not. pub async fn send_records( &self, turn: ProduceTurn, @@ -408,7 +417,13 @@ impl IggyBridge { return (Err(BridgeError::Timeout), Some((turn, kept))); } let (stream_id, topic_id) = (target.stream_id.clone(), target.topic_id.clone()); - let recovery_deadline = (Instant::now() + SLOT_LIMIT).into_std(); + // With less than the room left, the SDK keeps the whole deadline, and the send answers 7. + // Never past the SDK's own limit, which `SLOT_LIMIT` counts on. + let sdk_deadline = deadline + .checked_sub(SDK_STOP_ROOM) + .filter(|stop| *stop > Instant::now()) + .unwrap_or(deadline) + .min(binary_operation_deadline()); let send = async move { // Owns the messages, so a send cut off at the limit drops them. let send = { @@ -421,13 +436,15 @@ impl IggyBridge { .await { Err(IggyError::LaneBlocked(operation)) => { - resolve_blocker(&client, operation, recovery_deadline).await?; + resolve_blocker(&client, operation, sdk_deadline.into_std()) + .await?; } result => return result, } } } }; + let send = with_binary_deadline(sdk_deadline, send); within_slot_limit(&session, &client, kept, send).await }; // The SDK cannot cancel a send. One that outlives this call keeps the turn in its task, diff --git a/gateways/kafka/src/protocol/handlers/produce.rs b/gateways/kafka/src/protocol/handlers/produce.rs index d006c9b4ae..4b09aa8ff0 100644 --- a/gateways/kafka/src/protocol/handlers/produce.rs +++ b/gateways/kafka/src/protocol/handlers/produce.rs @@ -517,10 +517,11 @@ fn request_timeout(timeout_ms: i32) -> Duration { .min(REQUEST_DEADLINE) } -/// Sends one partition's messages, or answers 7 once the deadline has passed. Hands `held` back, +/// Sends one partition's messages, and gives up on them at the deadline. Hands `held` back, /// unless the send still runs and keeps it. /// -/// 7 means nothing was written, or the send may still land and a retry duplicate it. +/// 7 means nothing was written, or the send may still land and a retry duplicate it. 6 means that +/// it never went out. async fn send_by( bridge: &IggyBridge, (turn, admission): Held, @@ -535,22 +536,31 @@ async fn send_by( .await; let sent = match sent { Ok(base_offset) => Ok(base_offset), - Err(BridgeError::Timeout) => { + // The SDK retries a send until the deadline, and then ends it with one of these two: not + // sent, or in doubt. + Err( + error @ (BridgeError::Timeout + | BridgeError::Iggy( + IggyError::TransientNotAccepted | IggyError::OperationUncertain(_), + )), + ) => { // The client picks `timeout_ms`, so its expiry is no signal to the operator. if deadline.ceiling { tracing::warn!( + %error, kafka_topic, partition_id, "Produce ran past the gateway deadline" ); } else { tracing::debug!( + %error, kafka_topic, partition_id, "Produce ran past the client deadline" ); } - Err(ERROR_REQUEST_TIMED_OUT) + Err(error.to_produce_error_code()) } Err(error) => Err(refusal_code(kafka_topic, &error)), }; @@ -559,7 +569,7 @@ async fn send_by( /// The Kafka code for a bridge refusal, logged at a level that matches who can fix it. fn refusal_code(kafka_topic: &str, error: &BridgeError) -> i16 { - let code = error.to_kafka_error_code(); + let code = error.to_produce_error_code(); if error.is_bridge_login_rejected() { tracing::error!(%error, kafka_topic, "Iggy rejected the bridge's credentials"); } else if matches!(code, ERROR_UNKNOWN_SERVER_ERROR) { diff --git a/gateways/kafka/tests/produce_real_bridge_tests.rs b/gateways/kafka/tests/produce_real_bridge_tests.rs index 3a70c18256..8b09ae2f60 100644 --- a/gateways/kafka/tests/produce_real_bridge_tests.rs +++ b/gateways/kafka/tests/produce_real_bridge_tests.rs @@ -21,6 +21,7 @@ //! the same way in both directions cannot pass. use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use bytes::{Bytes, BytesMut}; @@ -35,9 +36,11 @@ use kafka_protocol::records::{ RecordBatchEncoder, RecordEncodeOptions, TimestampType, }; use serial_test::serial; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::watch; use tokio_util::sync::CancellationToken; -use iggy_gateway_kafka::bridge::{IggyBridge, Producer}; +use iggy_gateway_kafka::bridge::{IggyBridge, IggyBridgeConfig, Producer}; use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig}; use iggy_gateway_kafka::protocol::api::{ API_KEY_INIT_PRODUCER_ID, API_KEY_PRODUCE, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER, @@ -68,7 +71,12 @@ const ERROR_INVALID_RECORD: i16 = 87; /// A live gateway with the topic already provisioned, since Produce creates nothing. async fn gateway_with_topic(server: &TestServer, partitions: u32) -> GatewayState { - let bridge = IggyBridge::connect(server.test_config()) + gateway_on(server.test_config(), partitions).await +} + +/// [`gateway_with_topic`], with a bridge that dials as `config` says. +async fn gateway_on(config: IggyBridgeConfig, partitions: u32) -> GatewayState { + let bridge = IggyBridge::connect(config) .await .expect("bridge should connect to a ready server"); bridge @@ -664,6 +672,142 @@ async fn given_a_stopped_server_when_handled_should_answer_retriable() { ); } +/// The SDK stops a send that cannot reach Iggy by its request's deadline. Otherwise the send lands +/// once Iggy is back, after the client got its answer, and the client's retry writes the batch a +/// second time. +#[tokio::test] +#[serial] +async fn given_iggy_out_of_reach_past_the_deadline_when_the_client_retries_should_store_the_batch_once() + { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let relay = OutageRelay::start(&server).await; + let mut config = server.test_config(); + config.address.clone_from(&relay.address); + let state = gateway_on(config, 1).await; + // Each request here comes on a new connection, so only a producer id keeps one Iggy session. + let id = issued_producer_id(&state).await; + let first = [(0, idempotent_batch(id, b"first"))]; + assert_eq!( + produce(&state, 3, 1, TOPIC, &first).await, + vec![(0, ERROR_NONE, 0)], + "the producer's Iggy session is open" + ); + + relay.go_down().await; + let retried = [(0, idempotent_batch(id, b"retried"))]; + let body = encode_timed_request(3, 1, 1_000, TOPIC, &retried); + let response = handle_request_bounded(&state, API_KEY_PRODUCE, 3, body) + .await + .expect_response("acks=1 is answered"); + assert_eq!( + partition_results(3, response), + vec![(0, ERROR_NOT_LEADER_OR_FOLLOWER, -1)], + "the send never went out, so the answer tells the client that a retry is safe" + ); + + relay.come_up(); + tokio::time::sleep(SDK_RECONNECT).await; + assert_eq!( + produce(&state, 3, 1, TOPIC, &retried).await, + vec![(0, ERROR_NONE, 1)], + "the client's retry lands right after the first batch" + ); + let payloads: Vec = stored(&server, 0, 10) + .await + .into_iter() + .map(|message| message.payload) + .collect(); + assert_eq!( + payloads, + [&b"first"[..], b"retried"], + "one copy of each batch" + ); +} + +/// Longer than the SDK takes to connect again and send what it still holds. +const SDK_RECONNECT: Duration = Duration::from_secs(3); + +/// A TCP relay to a server that can go down. Down, it cuts its connections, and holds the ones that +/// open until it is up again. So a client behind it cannot send, and does not learn why. +struct OutageRelay { + address: String, + up: watch::Sender, + connections: Arc, +} + +#[derive(Default)] +struct Connections { + accepted: AtomicUsize, + open: AtomicUsize, +} + +impl OutageRelay { + async fn start(server: &TestServer) -> Self { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind the relay"); + let address = listener.local_addr().expect("relay address").to_string(); + let upstream = server.test_config().address; + let (up, _) = watch::channel(true); + let connections = Arc::new(Connections::default()); + let (switch, counts) = (up.clone(), Arc::clone(&connections)); + tokio::spawn(async move { + while let Ok((inbound, _)) = listener.accept().await { + counts.accepted.fetch_add(1, Ordering::SeqCst); + counts.open.fetch_add(1, Ordering::SeqCst); + let (upstream, up, counts) = + (upstream.clone(), switch.subscribe(), Arc::clone(&counts)); + tokio::spawn(async move { + relay_while_up(inbound, &upstream, up).await; + counts.open.fetch_sub(1, Ordering::SeqCst); + }); + } + }); + Self { + address, + up, + connections, + } + } + + /// Goes down, and returns once each client it cut off dialed again. A client that has not + /// seen the cut yet can still write a request into its old connection, and then cannot know + /// whether the request arrived. + async fn go_down(&self) { + let cut = self.connections.open.load(Ordering::SeqCst); + let redialed = self.connections.accepted.load(Ordering::SeqCst) + cut; + self.up.send_replace(false); + let reached = async { + while self.connections.accepted.load(Ordering::SeqCst) < redialed { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }; + tokio::time::timeout(STEP, reached) + .await + .expect("every client that the relay cut off dials again"); + } + + fn come_up(&self) { + self.up.send_replace(true); + } +} + +/// Copies both ways between a client and the server while the relay is up. A connection that +/// opens while it is down waits, and one that is open ends when it goes down. +async fn relay_while_up(mut inbound: TcpStream, upstream: &str, mut up: watch::Receiver) { + if up.wait_for(|up| *up).await.is_err() { + return; + } + let Ok(mut outbound) = TcpStream::connect(upstream).await else { + return; + }; + tokio::select! { + _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound) => {} + _ = up.wait_for(|up| !*up) => {} + } +} + /// Long enough for any one step against a local server. const STEP: Duration = Duration::from_secs(10);