From f1b90d77d744b841d72972c8ace381528e635c35 Mon Sep 17 00:00:00 2001 From: jonaswre Date: Wed, 5 Aug 2026 07:19:29 +0000 Subject: [PATCH 1/2] fix(blobs): let the pusher see a refused push `execute_push_sink` opened a bidirectional stream and immediately stopped the receive half -- "we are not going to need this!" -- then returned `Ok(Stats::default())` however the push went. That half is the only channel a refusal travels on: the provider resets it with the reason code (`handle_read_request_result` -> `writer.reset(e.code())`) when the event mask, an interceptor or a rate limit turns the push down. Throwing it away turned every refused push into a reported success, so a node upgrading to an enforced `EventMask::DEFAULT` keeps reporting that pushes worked while the receiver drops every byte. Keep the stream and read it at the end: cleanly closed means accepted, reset means refused, and the code says why. A provider from before this existed drops its half without writing, which arrives as a clean end of stream and reads as acceptance -- the same answer it gave before -- so neither side needs the other to be new. Pushes are now acknowledged, which is the point but is also a timing change: `execute_push` returns once the blob has landed rather than once the last byte reached the socket. The test asserted store contents only, because the refusal was unobservable; it now asserts both the refusal and the contents, and needed no polling window once the push was acknowledged. Verified by neutralising just the acknowledgement: the deny push goes back to `Ok(Stats { .. })` and the test fails on it. Co-Authored-By: Claude Opus 5 --- protocols/krikos-blobs/src/api/remote.rs | 77 ++++++++++++++++++++++-- protocols/krikos-blobs/src/tests.rs | 63 +++++++------------ 2 files changed, 96 insertions(+), 44 deletions(-) diff --git a/protocols/krikos-blobs/src/api/remote.rs b/protocols/krikos-blobs/src/api/remote.rs index b6e4e824341..a6573647d0e 100644 --- a/protocols/krikos-blobs/src/api/remote.rs +++ b/protocols/krikos-blobs/src/api/remote.rs @@ -37,8 +37,8 @@ use crate::{ hashseq::{HashSeq, HashSeqIter}, limits::PROGRESS_QUEUE_CAPACITY, protocol::{ - ChunkRangesSeq, GetManyRequest, GetRequest, MAX_MESSAGE_SIZE, ObserveItem, ObserveRequest, - PushRequest, Request, RequestType, + ChunkRangesSeq, ERR_LIMIT, ERR_PERMISSION, GetManyRequest, GetRequest, MAX_MESSAGE_SIZE, + ObserveItem, ObserveRequest, PushRequest, Request, RequestType, }, provider::events::{ClientResult, ProgressError}, store::KRIKOS_BLOCK_SIZE, @@ -607,8 +607,11 @@ impl Remote { payload_bytes_sent: 0, sender: progress, }; - // we are not going to need this! - recv.stop(0u32.into()).anyerr()?; + // The receive half is kept open, and read at the end. It carries no data, but a + // provider that refuses the push resets it with the reason code + // (`handle_read_request_result` -> `writer.reset(e.code())`), and that reset is + // the only signal a refusal produces. Stopping this stream here -- before the + // request had even been written -- is what made a refused push report success. // write the request. Unlike for reading, we can just serialize it sync using postcard. let request = write_push_request(request, &mut send).await?; let mut request_ranges = request.ranges.iter_infinite(); @@ -623,6 +626,7 @@ impl Remote { if request.ranges.is_blob() { // we are done send.finish().anyerr()?; + push_accepted(&mut recv).await?; return Ok(Default::default()); } let hash_seq = self.store().get_bytes(root).await?; @@ -638,6 +642,7 @@ impl Remote { } } send.finish().anyerr()?; + push_accepted(&mut recv).await?; Ok(Default::default()) } @@ -817,6 +822,70 @@ impl Remote { } } +/// Wait for the provider's verdict on a push we have finished sending. +/// +/// The push protocol carries no response payload, so the provider's half of the stream +/// is used purely as an acknowledgement: it is closed cleanly when the push was accepted +/// and handled, and reset with a [`ProgressError`] code when it was refused — by the +/// event mask, by an interceptor, or by a rate limit. +/// +/// This is why the caller must not stop that stream. Waiting on it also makes a push +/// synchronous with respect to the receiver: `execute_push` now returns once the blob +/// has actually landed, rather than once the last byte has been handed to the socket. +/// +/// A provider from before this acknowledgement existed simply drops its half without +/// writing, which arrives as a clean end of stream and is read as acceptance — the same +/// answer such a provider gave before. +async fn push_accepted(recv: &mut krikos::endpoint::RecvStream) -> Result<()> { + // Scoped, not a module import: `RecvStreamExt` also carries a + // `read_length_prefixed` that would become ambiguous with the one the observe path + // above resolves today. + use krikos::endpoint::ReadError; + + use crate::util::RecvStreamExt; + + let err = match recv.expect_eof().await { + Ok(()) => return Ok(()), + Err(err) => err, + }; + let reset = err + .get_ref() + .and_then(|e| e.downcast_ref::()) + .and_then(|e| match e { + ReadError::Reset(code) => Some(code), + _ => None, + }); + let Some(code) = reset else { + // Not a verdict at all: the connection died, or the stream produced data no + // push provider is supposed to send. + return Err(e!(PushError::Io, err).into()); + }; + let reason = if *code == ERR_PERMISSION { + "the receiver does not accept pushes from us".to_string() + } else if *code == ERR_LIMIT { + "the receiver is rate limiting us".to_string() + } else { + format!("the receiver rejected the push, code {code}") + }; + Err(e!(PushError::Refused, AnyError::from_string(reason)).into()) +} + +/// Why a push did not complete. +#[allow(missing_docs)] +#[non_exhaustive] +#[stack_error(derive, add_meta)] +pub enum PushError { + /// The receiver refused the push. It has stored nothing. + #[error("push refused by the receiver")] + Refused { source: AnyError }, + /// The acknowledgement could not be read. + #[error("failed to read the push acknowledgement")] + Io { + #[error(std_err)] + source: io::Error, + }, +} + /// Failures for a get operation #[allow(missing_docs)] #[non_exhaustive] diff --git a/protocols/krikos-blobs/src/tests.rs b/protocols/krikos-blobs/src/tests.rs index 1b99a1322f2..abb4544ec18 100644 --- a/protocols/krikos-blobs/src/tests.rs +++ b/protocols/krikos-blobs/src/tests.rs @@ -1,4 +1,4 @@ -use std::{collections::HashSet, io, ops::Range, path::PathBuf, time::Duration}; +use std::{collections::HashSet, io, ops::Range, path::PathBuf}; use bao_tree::ChunkRanges; use bytes::Bytes; @@ -454,13 +454,11 @@ async fn two_nodes_push_blobs_mem() -> TestResult<()> { /// writes to the local store, so an unauthorized peer must not be able to place blobs of /// its choosing into ours. /// -/// The pusher cannot observe the refusal: `execute_push_sink` stops its receive stream -/// and returns `Stats::default()` unconditionally, and the provider handles each stream -/// in a detached task, so a rejection never surfaces as a connection- or call-level -/// error. The property under test is therefore the one that actually matters — the blob -/// must not land in the receiving store. `allow` is a control: pushing the same blob to -/// a node that permits pushes proves the push path works in this setup, so a passing -/// "denied" assertion cannot be an artifact of a push that never happened. +/// Two things are asserted, because either alone can pass vacuously: the push must fail +/// for the pusher, and the blob must not be in the receiving store. `allow` is the +/// control — pushing the same blob to a node that permits pushes proves the push path +/// works in this setup, so a failing "denied" push cannot be an artifact of a push that +/// never happened. async fn push_is_rejected_when_disabled( pusher: Router, pusher_store: &Store, @@ -479,7 +477,7 @@ async fn push_is_rejected_when_disabled( // called `finish()`, but dropping the `Connection` at that point tears down the // still-in-flight stream and the receiver never finishes importing. let mut conns = Vec::new(); - for (target, must_succeed) in [(&deny, false), (&allow, true)] { + for (label, target, should_succeed) in [("deny", &deny, false), ("allow", &allow, true)] { let conn = pusher .endpoint() .connect(target.endpoint().addr(), crate::ALPN) @@ -488,46 +486,31 @@ async fn push_is_rejected_when_disabled( .remote() .execute_push_sink(conn.clone(), request.clone(), Drain) .await; - // Whether the refusal reaches the pusher at all is a race, which is exactly why - // the property under test is store contents rather than this result: the - // provider resets the stream, and `execute_push_sink` only sees that if the - // reset lands before it has finished writing. Both outcomes are correct for the - // denying node. The permitting node must succeed, or the control below — the - // thing that proves a passing deny assertion is not vacuous — proves nothing. - if must_succeed { - res?; - } + assert_eq!( + res.is_ok(), + should_succeed, + "{label}: push result was {res:?}" + ); conns.push(conn); } - // The permissive node completing its push bounds the wait for the denying one: - // both were pushed over loopback, the denying one first. + // No polling window is needed for either store any more: a push is acknowledged, so + // by the time the permitting node's push returned `Ok` it had finished importing, + // and by the time the denying node's push returned `Err` it had already decided. allow_count_rx.changed().await?; assert_eq!( allow_store.get_bytes(hash).await?, test_data(size), "control: a node that permits pushes must receive the blob" ); - // Nothing on the denying node is observable from here — a disabled request type - // emits no event, `execute_push_sink` does not wait for the receiver, and the - // provider imports on a detached task. Sampling `has` once could therefore run - // before an accepted push had landed and pass on a broken mask, so keep sampling - // for a while: an accepted push over loopback lands well inside this window. - let deadline = tokio::time::Instant::now() + Duration::from_secs(1); - loop { - // `status`, not `has`: `has` is only true for a *complete* blob, so a rejection - // that arrived after some chunks had already been imported would leave - // attacker-chosen bytes in the store and still satisfy the assertion. - assert_eq!( - deny_store.blobs().status(hash).await?, - BlobStatus::NotFound, - "a push rejected by the event mask must not write to the receiving store" - ); - if tokio::time::Instant::now() >= deadline { - break; - } - tokio::time::sleep(Duration::from_millis(50)).await; - } + // `status`, not `has`: `has` is only true for a *complete* blob, so a rejection that + // arrived after some chunks had already been imported would leave attacker-chosen + // bytes in the store and still satisfy the assertion. + assert_eq!( + deny_store.blobs().status(hash).await?, + BlobStatus::NotFound, + "a push rejected by the event mask must not write to the receiving store" + ); tokio::try_join!(pusher.shutdown(), deny.shutdown(), allow.shutdown())?; Ok(()) From a0b061743f7f7bb63267157af7391015c903a569 Mon Sep 17 00:00:00 2001 From: jonaswre Date: Wed, 5 Aug 2026 07:19:41 +0000 Subject: [PATCH 2/2] fix(relay): drop an undecodable frame instead of the client's session A frame the server could not decode returned an error from `handle_frame`, which ended `run_inner` and tore the connection down. The client is relaying for every peer it is talking to, so one unparseable frame cost all of that rather than one datagram -- and it is the exact inverse of what the send side already does, where an unencodable packet is dropped precisely because disconnecting over it would be out of proportion. `RecvError` already separates the two cases. A `Proto` error means this frame is unusable; the connection underneath is fine, so drop it, count it and carry on. A `StreamError` means there is nothing left to keep, so it stays fatal. Nothing bounds how many a client may send, deliberately: the read side is already rate limited, so a client sending malformed frames gets no more of this server than one sending valid ones. `recv_frames_invalid` is what makes it visible -- sustained growth there is a version mismatch or a bug on the client side. The new test writes the frame straight onto the byte sink, because a conforming client cannot produce one -- `Conn::start_send` refuses to encode an empty datagram, which is why only a buggy or hostile peer sends it -- and then checks that a Ping queued behind it is still answered. Verified by neutralising the split: the Ping is never answered because the session is already gone. Co-Authored-By: Claude Opus 5 --- docs/testing/determinism-audit.md | 2 +- krikos-relay/src/server/client.rs | 96 ++++++++++++++++++++- krikos-relay/src/server/metrics.rs | 9 ++ scripts/determinism-boundaries.semantic.txt | 1 + scripts/determinism-boundaries.txt | 23 ++--- 5 files changed, 118 insertions(+), 13 deletions(-) diff --git a/docs/testing/determinism-audit.md b/docs/testing/determinism-audit.md index e3c3dd0f542..ec0fb642667 100644 --- a/docs/testing/determinism-audit.md +++ b/docs/testing/determinism-audit.md @@ -403,7 +403,7 @@ and the simulation guide. | `RemoteMap`, `RemoteStateActor`, net-report, protocol router, and relay actor `JoinSet::spawn` calls | **Architectural problem** | Replace direct Tokio `JoinSet` ownership in simulator-supported paths with an environment-owned task group abstraction. | | `krikos-relay/src/client/tls.rs:200`, `quic.rs:124`, `server.rs:772,810,860`, `server/client.rs:146`, `server/http_server/listener.rs:323,350`, `server/http_server/upgrade.rs:113`, `server/resolver.rs:79` | **Architectural problem** | Add relay runtime/task capabilities before claiming deterministic relay coverage. | | DNS/HTTP server task sets and `krikos-dns-server/src/http/transport.rs` | **Acceptable nondeterminism** in the real-server backend | Listener and connection work is now owned by supervisors, bounded before spawn, and cancelled/drained on shutdown. Runtime/I/O injection remains required before full DNS-server simulation. | -| Direct spawns in `#[cfg(test)]` regions, `tests/`, `examples/`, and `krikos/bench` | **Acceptable nondeterminism** | Do not route through production environment unless the scenario runner reuses that code. Includes `krikos-relay/src/server/client.rs::tests::undeliverable_packet_does_not_end_the_session`, which drives one client actor over a `tokio::io::duplex` pair: the spawn owns only the actor under test, the test joins it before returning, and no simulator backend executes this module. | +| Direct spawns in `#[cfg(test)]` regions, `tests/`, `examples/`, and `krikos/bench` | **Acceptable nondeterminism** | Do not route through production environment unless the scenario runner reuses that code. Includes `krikos-relay/src/server/client.rs::tests::undeliverable_packet_does_not_end_the_session` and `::tests::undecodable_frame_does_not_end_the_session`, which each drive one client actor over a `tokio::io::duplex` pair: the spawn owns only the actor under test, the test joins it before returning, and no simulator backend executes this module. | | `krikos-relay/src/main.rs:627` certificate file `spawn_blocking` | **Acceptable nondeterminism** | Keep in production binary; exclude from in-process simulation. | | DNS-server bounded per-IP token buckets | **Behavioral randomness** with explicit transition input | The LRU has a validated hard capacity; token transitions receive `Instant` explicitly. Only the HTTP middleware samples the production clock. There is no GC thread. | diff --git a/krikos-relay/src/server/client.rs b/krikos-relay/src/server/client.rs index 084c376383d..33ba11499a9 100644 --- a/krikos-relay/src/server/client.rs +++ b/krikos-relay/src/server/client.rs @@ -542,7 +542,23 @@ where ) -> Result<(), HandleFrameError> { trace!(?maybe_frame, "handle incoming frame"); let frame = match maybe_frame { - Some(frame) => frame?, + Some(Ok(frame)) => frame, + // A frame we cannot decode costs this client one datagram, not its session. + // It is relaying for every peer it is talking to, so tearing the connection + // down over one unparsable frame is far out of proportion — and it is the + // mirror of `send_packet`, which already drops rather than disconnects when + // the *encoder* refuses. Only a broken stream is fatal, because then there + // is nothing left to keep. + // + // Nothing bounds how many of these a client may send, deliberately: the + // read side is already rate limited, so a client sending malformed frames + // gets no more of this server than one sending valid ones. + Some(Err(err @ RelayRecvError::Proto { .. })) => { + self.metrics.recv_frames_invalid.inc(); + warn!("dropping undecodable frame: {err:#}"); + return Ok(()); + } + Some(Err(err)) => return Err(err.into()), None => return Err(e!(HandleFrameError::StreamTerminated)), }; @@ -885,6 +901,84 @@ mod tests { Ok(()) } + /// A frame the server cannot decode must not end the sending client's session. + /// + /// The client is relaying for every peer it is talking to, so disconnecting it over + /// one unparsable frame loses all of that instead of one datagram. This is the + /// mirror of `undeliverable_packet_does_not_end_the_session`, which covers the same + /// property on the encode side. + /// + /// The frame is written straight onto the wire, because a conforming client cannot + /// produce one: `Conn::start_send` refuses to encode an empty datagram, which is + /// exactly why only a buggy or hostile peer sends it. + #[tokio::test] + #[traced_test] + async fn undecodable_frame_does_not_end_the_session() -> Result { + let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(23); + let (send_queue_s, send_queue_r) = mpsc::channel(10); + let (message_s, message_r) = mpsc::channel(10); + let endpoint_id = SecretKey::from_bytes(&rng.random()).public(); + let dst = SecretKey::from_bytes(&rng.random()).public(); + let (io, io_rw) = tokio::io::duplex(1024); + let mut io_rw = Conn::test(io_rw, Default::default()); + let stream = RelayedStream::test(io); + let clients = Clients::default(); + let metrics = Arc::new(Metrics::default()); + let actor = Actor { + stream, + timeout: Duration::from_secs(1), + packet_send_queue: send_queue_r, + message_send_queue: message_r, + guard: Some(OnDisconnectGuard::empty(endpoint_id)), + endpoint_id, + registered: false, + clients: clients.clone(), + client_counter: ClientCounter::new(clients.runtime().wall_clock()), + ping_tracker: PingTracker::default(), + metrics: metrics.clone(), + clock: clients.runtime().clock(), + }; + let done = CancellationToken::new(); + let io_done = done.clone(); + let handle = tokio::task::spawn(async move { actor.run(io_done).await }); + + // A datagram frame with empty contents: well formed enough to reach the parser, + // rejected by it because the relay could not forward it on. + let undecodable = ClientToRelayMsg::Datagrams { + dst_endpoint_id: dst, + datagrams: Datagrams::from(&[][..]), + } + .write_to(Vec::new()); + // Straight onto the byte sink, under `Conn`'s own encoder. + io_rw + .conn + .send(bytes::Bytes::from(undecodable)) + .await + .anyerr()?; + + // Queued behind it: if the session survived, this still gets answered. + let payload = [3u8; 8]; + io_rw.send(ClientToRelayMsg::Ping(payload)).await?; + let frame = recv_frame(FrameType::Pong, &mut io_rw).await.anyerr()?; + assert_eq!(frame, RelayToClientMsg::Pong(payload)); + + assert!( + !handle.is_finished(), + "an undecodable frame must not end the session" + ); + assert_eq!( + metrics.recv_frames_invalid.get(), + 1, + "the undecodable frame should be counted" + ); + + done.cancel(); + drop(send_queue_s); + drop(message_s); + handle.await.std_context("join")?; + Ok(()) + } + #[tokio::test(start_paused = true)] #[traced_test] async fn client_actor_survives_production_keepalive_horizon_when_polled() -> Result { diff --git a/krikos-relay/src/server/metrics.rs b/krikos-relay/src/server/metrics.rs index 3c6371cc0d4..2e73c74aa7d 100644 --- a/krikos-relay/src/server/metrics.rs +++ b/krikos-relay/src/server/metrics.rs @@ -46,6 +46,15 @@ pub struct Metrics { #[metrics(help = "Number of 'send' packets dropped.")] pub send_packets_dropped: Counter, + /// Frames received from a client that could not be decoded, and were dropped. + /// + /// The client stays connected. A frame this server cannot parse is one datagram lost, + /// not a broken connection, and the sending client is usually relaying for several + /// peers at once. Sustained growth here means a client is sending frames this server + /// does not accept — a version mismatch or a bug on its side. + #[metrics(help = "Number of frames received from a client that could not be decoded.")] + pub recv_frames_invalid: Counter, + /// Intended to count sent frames other than datagrams, but currently unused (never incremented). #[metrics(help = "Number of packets sent that were not 'send' packets")] pub other_packets_sent: Counter, diff --git a/scripts/determinism-boundaries.semantic.txt b/scripts/determinism-boundaries.semantic.txt index 6a8f16b8556..ed91ba2c981 100644 --- a/scripts/determinism-boundaries.semantic.txt +++ b/scripts/determinism-boundaries.semantic.txt @@ -541,6 +541,7 @@ spawn-task krikos-relay/src/quic.rs tests::test_qad_client_closes_unresponsive_f spawn-task krikos-relay/src/quic.rs tests::test_qad_connect_delayed tokio::spawn 1 spawn-task krikos-relay/src/server/client.rs tests::client_actor_survives_production_keepalive_horizon_when_polled tokio::task::spawn 1 spawn-task krikos-relay/src/server/client.rs tests::test_client_actor_basic tokio::task::spawn 1 +spawn-task krikos-relay/src/server/client.rs tests::undecodable_frame_does_not_end_the_session tokio::task::spawn 1 spawn-task krikos-relay/src/server/client.rs tests::undeliverable_packet_does_not_end_the_session tokio::task::spawn 1 spawn-task krikos-relay/src/server/http_server/listener.rs impl::spawn tokio::task::spawn 1 spawn-task krikos-relay/src/server/http_server/tests.rs test_server_basic tokio::spawn 1 diff --git a/scripts/determinism-boundaries.txt b/scripts/determinism-boundaries.txt index 8bad716dd8e..37022db73f8 100644 --- a/scripts/determinism-boundaries.txt +++ b/scripts/determinism-boundaries.txt @@ -49,12 +49,12 @@ clock-timer krikos-relay/src/quic.rs:678 let start = Instant::now(); clock-timer krikos-relay/src/server/admission.rs:107 last_refill: Instant::now(), clock-timer krikos-relay/src/server/admission.rs:112 let now = Instant::now(); clock-timer krikos-relay/src/server/admission.rs:150 tokio::time::advance(std::time::Duration::from_millis(500)).await; -clock-timer krikos-relay/src/server/client.rs:1097 let recv_frame = tokio::time::timeout(Duration::from_millis(500), stream.next()) -clock-timer krikos-relay/src/server/client.rs:1108 let res = tokio::time::timeout(Duration::from_millis(100), stream.next()).await; -clock-timer krikos-relay/src/server/client.rs:1114 tokio::time::sleep(Duration::from_secs(1)).await; -clock-timer krikos-relay/src/server/client.rs:1117 let recv_frame = tokio::time::timeout(Duration::from_millis(500), stream.next()) -clock-timer krikos-relay/src/server/client.rs:917 tokio::time::advance(Duration::from_secs(21)).await; -clock-timer krikos-relay/src/server/client.rs:924 tokio::time::advance(Duration::from_secs(6)).await; +clock-timer krikos-relay/src/server/client.rs:1011 tokio::time::advance(Duration::from_secs(21)).await; +clock-timer krikos-relay/src/server/client.rs:1018 tokio::time::advance(Duration::from_secs(6)).await; +clock-timer krikos-relay/src/server/client.rs:1191 let recv_frame = tokio::time::timeout(Duration::from_millis(500), stream.next()) +clock-timer krikos-relay/src/server/client.rs:1202 let res = tokio::time::timeout(Duration::from_millis(100), stream.next()).await; +clock-timer krikos-relay/src/server/client.rs:1208 tokio::time::sleep(Duration::from_secs(1)).await; +clock-timer krikos-relay/src/server/client.rs:1211 let recv_frame = tokio::time::timeout(Duration::from_millis(500), stream.next()) clock-timer krikos-relay/src/server/clients.rs:619 tokio::time::timeout(Duration::from_secs(1), async move { clock-timer krikos-relay/src/server/clients.rs:624 tokio::time::sleep(Duration::from_millis(100)).await; clock-timer krikos-relay/src/server/clients.rs:696 tokio::time::timeout(Duration::from_secs(1), { @@ -841,9 +841,10 @@ spawn-task krikos-relay/src/quic.rs:30 use tokio::{sync::Semaphore, task::JoinSe spawn-task krikos-relay/src/quic.rs:619 let task = AbortOnDropHandle::new(tokio::spawn({ spawn-task krikos-relay/src/quic.rs:679 let server_task = tokio::spawn( spawn-task krikos-relay/src/server.rs:51 task::{JoinError, JoinSet}, -spawn-task krikos-relay/src/server/client.rs:739 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); -spawn-task krikos-relay/src/server/client.rs:841 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); -spawn-task krikos-relay/src/server/client.rs:915 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); +spawn-task krikos-relay/src/server/client.rs:1009 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); +spawn-task krikos-relay/src/server/client.rs:755 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); +spawn-task krikos-relay/src/server/client.rs:857 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); +spawn-task krikos-relay/src/server/client.rs:943 let handle = tokio::task::spawn(async move { actor.run(io_done).await }); spawn-task krikos-relay/src/server/http_server/listener.rs:323 let task = tokio::task::spawn( spawn-task krikos-relay/src/server/http_server/listener.rs:327 let mut set = tokio::task::JoinSet::new(); spawn-task krikos-relay/src/server/http_server/tests.rs:305 let handler_task = tokio::spawn(async move { @@ -1003,8 +1004,8 @@ unordered-collection krikos-dns/src/endpoint_info.rs:45 collections::{BTreeSet, unordered-collection krikos-dns/src/endpoint_info.rs:78 fn dedup(items: &mut Vec) -> HashSet { unordered-collection krikos-dns/src/endpoint_info.rs:80 let mut seen = HashSet::new(); unordered-collection krikos-relay/src/server/client.rs:4 collections::HashSet, -unordered-collection krikos-relay/src/server/client.rs:620 clients: HashSet, -unordered-collection krikos-relay/src/server/client.rs:629 clients: HashSet::new(), +unordered-collection krikos-relay/src/server/client.rs:636 clients: HashSet, +unordered-collection krikos-relay/src/server/client.rs:645 clients: HashSet::new(), unordered-collection krikos-relay/src/server/clients.rs:12 use dashmap::DashMap; unordered-collection krikos-relay/src/server/clients.rs:180 clients: DashMap::new(), unordered-collection krikos-relay/src/server/clients.rs:181 sent_to: DashMap::new(),