Fix poll_expired incorrectly returning Pending when there are no outstanding requests.

README.md

@@ -40,7 +40,7 @@ rather than in a separate language such as .proto. This means there's no separat
 process, and no context switching between different languages.
 
 Some other features of tarpc:
-- Pluggable transport: any type implementing `Stream<Item = Request> + Sink<Response>` can be
+- Pluggable transport: any type impling `Stream<Item = Request> + Sink<Response>` can be
   used as a transport to connect the client and server.
 - `Send + 'static` optional: if the transport doesn't require it, neither does tarpc!
 - Cascading cancellation: dropping a request will send a cancellation message to the server.
@@ -55,7 +55,7 @@ Some other features of tarpc:
   [tracing](https://github.com/tokio-rs/tracing) primitives extended with
   [OpenTelemetry](https://opentelemetry.io/) traces. Using a compatible tracing subscriber like
   [Jaeger](https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-jaeger),
-  each RPC can be traced through the client, server, and other dependencies downstream of the
+  each RPC can be traced through the client, server, amd other dependencies downstream of the
   server. Even for applications not connected to a distributed tracing collector, the
   instrumentation can also be ingested by regular loggers like
   [env_logger](https://github.com/env-logger-rs/env_logger/).
@@ -67,7 +67,7 @@ Some other features of tarpc:
 Add to your `Cargo.toml` dependencies:
 
 ```toml
-tarpc = "0.29"
+tarpc = "0.27"
 ```
 
 The `tarpc::service` attribute expands to a collection of items that form an rpc service.
@@ -82,7 +82,7 @@ your `Cargo.toml`:
 ```toml
 anyhow = "1.0"
 futures = "0.3"
-tarpc = { version = "0.29", features = ["tokio1"] }
+tarpc = { version = "0.27", features = ["tokio1"] }
 tokio = { version = "1.0", features = ["macros"] }
 ```
 
@@ -100,7 +100,7 @@ use futures::{
 };
 use tarpc::{
     client, context,
-    server::{self, incoming::Incoming, Channel},
+    server::{self, incoming::Incoming},
 };
 
 // This is the service definition. It looks a lot like a trait definition.
@@ -128,7 +128,7 @@ impl World for HelloServer {
     type HelloFut = Ready<String>;
 
     fn hello(self, _: context::Context, name: String) -> Self::HelloFut {
-        future::ready(format!("Hello, {name}!"))
+        future::ready(format!("Hello, {}!", name))
     }
 }
 ```
@@ -155,7 +155,7 @@ async fn main() -> anyhow::Result<()> {
     // specifies a deadline and trace information which can be helpful in debugging requests.
     let hello = client.hello(context::current(), "Stim".to_string()).await?;
 
-    println!("{hello}");
+    println!("{}", hello);
 
     Ok(())
 }

RELEASES.md

@@ -1,49 +1,8 @@
-## 0.30.0 (2022-08-12)
-
-### Breaking Changes
-
-- Some types that impl Future are now annotated with `#[must_use]`. Code that previously created
-  these types but did not use them will now receive a warning. Code that disallows warnings will
-  receive a compilation error.
-
-### Fixes
-
-- Servers will more reliably clean up request state for requests with long deadlines when response
-processing is aborted without sending a response.
-
-### Other Changes
-
-- `TrackedRequest` now contains a response guard that can be used to ensure state cleanup for
-  aborted requests. (This was already handled automatically by `InFlightRequests`).
-- When the feature serde-transport is enabled, the crate tokio_serde is now re-exported.
-
-## 0.29.0 (2022-05-26)
-
-### Breaking Changes
-
-`Context.deadline` is now serialized as a Duration. This prevents clock skew from affecting deadline
-behavior. For more details see https://github.com/google/tarpc/pull/367 and its [related
-issue](https://github.com/google/tarpc/issues/366).
-
-## 0.28.0 (2022-04-06)
-
-### Breaking Changes
-
-- The minimum supported Rust version has increased to 1.58.0.
-- The version of opentelemetry depended on by tarpc has increased to 0.17.0.
-
-## 0.27.2 (2021-10-08)
-
-### Fixes
-
-Clients will now close their transport before dropping it. An attempt at a clean shutdown can help
-the server drop its connections more quickly.
-
 ## 0.27.1 (2021-09-22)
 
 ### Breaking Changes
 
-#### RPC error type is changing
+### RPC error type is changing
 
 RPC return types are changing from `Result<Response, io::Error>` to `Result<Response,
 tarpc::client::RpcError>`.

example-service/Cargo.toml

@@ -1,9 +1,8 @@
 [package]
 name = "tarpc-example-service"
-version = "0.12.0"
-rust-version = "1.56"
+version = "0.10.0"
 authors = ["Tim Kuehn <tikue@google.com>"]
-edition = "2021"
+edition = "2018"
 license = "MIT"
 documentation = "https://docs.rs/tarpc-example-service"
 homepage = "https://github.com/google/tarpc"
@@ -15,17 +14,17 @@ description = "An example server built on tarpc."
 
 [dependencies]
 anyhow = "1.0"
-clap = { version = "3.0.0-rc.9", features = ["derive"] }
+clap = "3.0.0-beta.2"
 log = "0.4"
 futures = "0.3"
-opentelemetry = { version = "0.17", features = ["rt-tokio"] }
-opentelemetry-jaeger = { version = "0.16", features = ["rt-tokio"] }
+opentelemetry = { version = "0.16", features = ["rt-tokio"] }
+opentelemetry-jaeger = { version = "0.15", features = ["rt-tokio"] }
 rand = "0.8"
-tarpc = { version = "0.30", path = "../tarpc", features = ["full"] }
+tarpc = { version = "0.27", path = "../tarpc", features = ["full"] }
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread"] }
 tracing = { version = "0.1" }
-tracing-opentelemetry = "0.17"
-tracing-subscriber = {version = "0.3", features = ["env-filter"]}
+tracing-opentelemetry = "0.15"
+tracing-subscriber = "0.2"
 
 [lib]
 name = "service"

example-service/src/client.rs

@@ -4,14 +4,14 @@
 // license that can be found in the LICENSE file or at
 // https://opensource.org/licenses/MIT.
 
-use clap::Parser;
+use clap::Clap;
 use service::{init_tracing, WorldClient};
 use std::{net::SocketAddr, time::Duration};
 use tarpc::{client, context, tokio_serde::formats::Json};
 use tokio::time::sleep;
 use tracing::Instrument;
 
-#[derive(Parser)]
+#[derive(Clap)]
 struct Flags {
     /// Sets the server address to connect to.
     #[clap(long)]

example-service/src/server.rs

@@ -4,7 +4,7 @@
 // license that can be found in the LICENSE file or at
 // https://opensource.org/licenses/MIT.
 
-use clap::Parser;
+use clap::Clap;
 use futures::{future, prelude::*};
 use rand::{
     distributions::{Distribution, Uniform},
@@ -22,7 +22,7 @@ use tarpc::{
 };
 use tokio::time;
 
-#[derive(Parser)]
+#[derive(Clap)]
 struct Flags {
     /// Sets the port number to listen on.
     #[clap(long)]
@@ -40,7 +40,7 @@ impl World for HelloServer {
         let sleep_time =
             Duration::from_millis(Uniform::new_inclusive(1, 10).sample(&mut thread_rng()));
         time::sleep(sleep_time).await;
-        format!("Hello, {name}! You are connected from {}", self.0)
+        format!("Hello, {}! You are connected from {}", name, self.0)
     }
 }
 

plugins/Cargo.toml

@@ -1,9 +1,8 @@
 [package]
 name = "tarpc-plugins"
 version = "0.12.0"
-rust-version = "1.56"
 authors = ["Adam Wright <adam.austin.wright@gmail.com>", "Tim Kuehn <timothy.j.kuehn@gmail.com>"]
-edition = "2021"
+edition = "2018"
 license = "MIT"
 documentation = "https://docs.rs/tarpc-plugins"
 homepage = "https://github.com/google/tarpc"

plugins/LICENSE

@@ -1,9 +0,0 @@
-The MIT License (MIT)
-
-Copyright 2016 Google Inc. All Rights Reserved.
-
-Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

plugins/src/lib.rs

@@ -83,7 +83,7 @@ impl Parse for Service {
                     ident_errors,
                     syn::Error::new(
                         rpc.ident.span(),
-                        format!("method name conflicts with generated fn `{ident}::serve`")
+                        format!("method name conflicts with generated fn `{}::serve`", ident)
                     )
                 );
             }
@@ -270,7 +270,7 @@ pub fn service(attr: TokenStream, input: TokenStream) -> TokenStream {
     let methods = rpcs.iter().map(|rpc| &rpc.ident).collect::<Vec<_>>();
     let request_names = methods
         .iter()
-        .map(|m| format!("{ident}.{m}"))
+        .map(|m| format!("{}.{}", ident, m))
         .collect::<Vec<_>>();
 
     ServiceGenerator {
@@ -306,7 +306,7 @@ pub fn service(attr: TokenStream, input: TokenStream) -> TokenStream {
             .collect::<Vec<_>>(),
         future_types: &camel_case_fn_names
             .iter()
-            .map(|name| parse_str(&format!("{name}Fut")).unwrap())
+            .map(|name| parse_str(&format!("{}Fut", name)).unwrap())
             .collect::<Vec<_>>(),
         derive_serialize: derive_serialize.as_ref(),
     }
@@ -409,7 +409,7 @@ fn verify_types_were_provided(
         if !provided.iter().any(|typedecl| typedecl.ident == expected) {
             let mut e = syn::Error::new(
                 span,
-                format!("not all trait items implemented, missing: `{expected}`"),
+                format!("not all trait items implemented, missing: `{}`", expected),
             );
             let fn_span = method.sig.fn_token.span();
             e.extend(syn::Error::new(
@@ -479,7 +479,7 @@ impl<'a> ServiceGenerator<'a> {
                     ),
                     output,
                 )| {
-                    let ty_doc = format!("The response future returned by [`{service_ident}::{ident}`].");
+                    let ty_doc = format!("The response future returned by [`{}::{}`].", service_ident, ident);
                     quote! {
                         #[doc = #ty_doc]
                         type #future_type: std::future::Future<Output = #output>;

tarpc/Cargo.toml

@@ -1,12 +1,8 @@
 [package]
 name = "tarpc"
-version = "0.30.0"
-rust-version = "1.58.0"
-authors = [
-    "Adam Wright <adam.austin.wright@gmail.com>",
-    "Tim Kuehn <timothy.j.kuehn@gmail.com>",
-]
-edition = "2021"
+version = "0.27.1"
+authors = ["Adam Wright <adam.austin.wright@gmail.com>", "Tim Kuehn <timothy.j.kuehn@gmail.com>"]
+edition = "2018"
 license = "MIT"
 documentation = "https://docs.rs/tarpc"
 homepage = "https://github.com/google/tarpc"
@@ -20,20 +16,13 @@ description = "An RPC framework for Rust with a focus on ease of use."
 default = []
 
 serde1 = ["tarpc-plugins/serde1", "serde", "serde/derive"]
-tokio1 = ["tokio/rt"]
+tokio1 = ["tokio/rt-multi-thread"]
 serde-transport = ["serde1", "tokio1", "tokio-serde", "tokio-util/codec"]
 serde-transport-json = ["tokio-serde/json"]
 serde-transport-bincode = ["tokio-serde/bincode"]
 tcp = ["tokio/net"]
 
-full = [
-    "serde1",
-    "tokio1",
-    "serde-transport",
-    "serde-transport-json",
-    "serde-transport-bincode",
-    "tcp",
-]
+full = ["serde1", "tokio1", "serde-transport", "serde-transport-json", "serde-transport-bincode", "tcp"]
 
 [badges]
 travis-ci = { repository = "google/tarpc" }
@@ -50,14 +39,11 @@ static_assertions = "1.1.0"
 tarpc-plugins = { path = "../plugins", version = "0.12" }
 thiserror = "1.0"
 tokio = { version = "1", features = ["time"] }
-tokio-util = { version = "0.7.3", features = ["time"] }
+tokio-util = { version = "0.6.3", features = ["time"] }
 tokio-serde = { optional = true, version = "0.8" }
-tracing = { version = "0.1", default-features = false, features = [
-    "attributes",
-    "log",
-] }
-tracing-opentelemetry = { version = "0.17.2", default-features = false }
-opentelemetry = { version = "0.17.0", default-features = false }
+tracing = { version = "0.1", default-features = false, features = ["attributes", "log"] }
+tracing-opentelemetry = { version = "0.15", default-features = false }
+opentelemetry = { version = "0.16", default-features = false }
 
 
 [dev-dependencies]
@@ -66,13 +52,11 @@ bincode = "1.3"
 bytes = { version = "1", features = ["serde"] }
 flate2 = "1.0"
 futures-test = "0.3"
-opentelemetry = { version = "0.17.0", default-features = false, features = [
-    "rt-tokio",
-] }
-opentelemetry-jaeger = { version = "0.16.0", features = ["rt-tokio"] }
+opentelemetry = { version = "0.16", default-features = false, features = ["rt-tokio"] }
+opentelemetry-jaeger = { version = "0.15", features = ["rt-tokio"] }
 pin-utils = "0.1.0-alpha"
 serde_bytes = "0.11"
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tracing-subscriber = "0.2"
 tokio = { version = "1", features = ["full", "test-util"] }
 tokio-serde = { version = "0.8", features = ["json", "bincode"] }
 trybuild = "1.0"

tarpc/LICENSE

@@ -1,9 +0,0 @@
-The MIT License (MIT)
-
-Copyright 2016 Google Inc. All Rights Reserved.
-
-Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

tarpc/examples/compression.rs

@@ -54,7 +54,7 @@ where
             if algorithm != CompressionAlgorithm::Deflate {
                 return Err(io::Error::new(
                     io::ErrorKind::InvalidData,
-                    format!("Compression algorithm {algorithm:?} not supported"),
+                    format!("Compression algorithm {:?} not supported", algorithm),
                 ));
             }
             let mut deflater = DeflateDecoder::new(payload.as_slice());
@@ -102,7 +102,7 @@ struct HelloServer;
 #[tarpc::server]
 impl World for HelloServer {
     async fn hello(self, _: context::Context, name: String) -> String {
-        format!("Hey, {name}!")
+        format!("Hey, {}!", name)
     }
 }
 

tarpc/examples/custom_transport.rs

@@ -1,9 +1,8 @@
-use tarpc::context::Context;
 use tarpc::serde_transport as transport;
 use tarpc::server::{BaseChannel, Channel};
-use tarpc::tokio_serde::formats::Bincode;
-use tarpc::tokio_util::codec::length_delimited::LengthDelimitedCodec;
+use tarpc::{context::Context, tokio_serde::formats::Bincode};
 use tokio::net::{UnixListener, UnixStream};
+use tokio_util::codec::length_delimited::LengthDelimitedCodec;
 
 #[tarpc::service]
 pub trait PingService {

tarpc/examples/pubsub.rs

@@ -52,9 +52,9 @@ use tarpc::{
     client, context,
     serde_transport::tcp,
     server::{self, Channel},
-    tokio_serde::formats::Json,
 };
 use tokio::net::ToSocketAddrs;
+use tokio_serde::formats::Json;
 use tracing::info;
 use tracing_subscriber::prelude::*;
 
@@ -129,6 +129,7 @@ impl Subscriber {
 
 #[derive(Debug)]
 struct Subscription {
+    subscriber: subscriber::SubscriberClient,
     topics: Vec<String>,
 }
 
@@ -209,6 +210,7 @@ impl Publisher {
             self.clients.lock().unwrap().insert(
                 subscriber_addr,
                 Subscription {
+                    subscriber: subscriber.clone(),
                     topics: topics.clone(),
                 },
             );
@@ -288,7 +290,7 @@ fn init_tracing(service_name: &str) -> anyhow::Result<()> {
         .install_batch(opentelemetry::runtime::Tokio)?;
 
     tracing_subscriber::registry()
-        .with(tracing_subscriber::filter::EnvFilter::from_default_env())
+        .with(tracing_subscriber::EnvFilter::from_default_env())
         .with(tracing_subscriber::fmt::layer())
         .with(tracing_opentelemetry::layer().with_tracer(tracer))
         .try_init()?;

tarpc/examples/readme.rs

@@ -29,7 +29,7 @@ impl World for HelloServer {
     type HelloFut = Ready<String>;
 
     fn hello(self, _: context::Context, name: String) -> Self::HelloFut {
-        future::ready(format!("Hello, {name}!"))
+        future::ready(format!("Hello, {}!", name))
     }
 }
 
@@ -49,7 +49,7 @@ async fn main() -> anyhow::Result<()> {
     // specifies a deadline and trace information which can be helpful in debugging requests.
     let hello = client.hello(context::current(), "Stim".to_string()).await?;
 
-    println!("{hello}");
+    println!("{}", hello);
 
     Ok(())
 }

tarpc/examples/tracing.rs

@@ -6,11 +6,12 @@
 
 use crate::{add::Add as AddService, double::Double as DoubleService};
 use futures::{future, prelude::*};
+use std::env;
 use tarpc::{
     client, context,
     server::{incoming::Incoming, BaseChannel},
-    tokio_serde::formats::Json,
 };
+use tokio_serde::formats::Json;
 use tracing_subscriber::prelude::*;
 
 pub mod add {
@@ -55,9 +56,9 @@ impl DoubleService for DoubleServer {
 }
 
 fn init_tracing(service_name: &str) -> anyhow::Result<()> {
+    env::set_var("OTEL_BSP_MAX_EXPORT_BATCH_SIZE", "12");
     let tracer = opentelemetry_jaeger::new_pipeline()
         .with_service_name(service_name)
-        .with_auto_split_batch(true)
         .with_max_packet_size(2usize.pow(13))
         .install_batch(opentelemetry::runtime::Tokio)?;
 

tarpc/src/cancellations.rs

@@ -1,49 +0,0 @@
-use futures::{prelude::*, task::*};
-use std::pin::Pin;
-use tokio::sync::mpsc;
-
-/// Sends request cancellation signals.
-#[derive(Debug, Clone)]
-pub struct RequestCancellation(mpsc::UnboundedSender<u64>);
-
-/// A stream of IDs of requests that have been canceled.
-#[derive(Debug)]
-pub struct CanceledRequests(mpsc::UnboundedReceiver<u64>);
-
-/// Returns a channel to send request cancellation messages.
-pub fn cancellations() -> (RequestCancellation, CanceledRequests) {
-    // Unbounded because messages are sent in the drop fn. This is fine, because it's still
-    // bounded by the number of in-flight requests.
-    let (tx, rx) = mpsc::unbounded_channel();
-    (RequestCancellation(tx), CanceledRequests(rx))
-}
-
-impl RequestCancellation {
-    /// Cancels the request with ID `request_id`.
-    ///
-    /// No validation is done of `request_id`. There is no way to know if the request id provided
-    /// corresponds to a request actually tracked by the backing channel. `RequestCancellation` is
-    /// a one-way communication channel.
-    ///
-    /// Once request data is cleaned up, a response will never be received by the client. This is
-    /// useful primarily when request processing ends prematurely for requests with long deadlines
-    /// which would otherwise continue to be tracked by the backing channel—a kind of leak.
-    pub fn cancel(&self, request_id: u64) {
-        let _ = self.0.send(request_id);
-    }
-}
-
-impl CanceledRequests {
-    /// Polls for a cancelled request.
-    pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<u64>> {
-        self.0.poll_recv(cx)
-    }
-}
-
-impl Stream for CanceledRequests {
-    type Item = u64;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<u64>> {
-        self.poll_recv(cx)
-    }
-}

tarpc/src/client.rs

@@ -8,10 +8,7 @@
 
 mod in_flight_requests;
 
-use crate::{
-    cancellations::{cancellations, CanceledRequests, RequestCancellation},
-    context, trace, ClientMessage, Request, Response, ServerError, Transport,
-};
+use crate::{context, trace, ClientMessage, Request, Response, ServerError, Transport};
 use futures::{prelude::*, ready, stream::Fuse, task::*};
 use in_flight_requests::{DeadlineExceededError, InFlightRequests};
 use pin_project::pin_project;
@@ -84,10 +81,14 @@ impl<C, D> fmt::Debug for NewClient<C, D> {
     }
 }
 
-const _CHECK_USIZE: () = assert!(
-    std::mem::size_of::<usize>() <= std::mem::size_of::<u64>(),
-    "usize is too big to fit in u64"
-);
+#[allow(dead_code)]
+#[allow(clippy::no_effect)]
+const CHECK_USIZE: () = {
+    if std::mem::size_of::<usize>() > std::mem::size_of::<u64>() {
+        // TODO: replace this with panic!() as soon as RFC 2345 gets stabilized
+        ["usize is too big to fit in u64"][42];
+    }
+};
 
 /// Handles communication from the client to request dispatch.
 #[derive(Debug)]
@@ -117,7 +118,6 @@ impl<Req, Resp> Channel<Req, Resp> {
         skip(self, ctx, request_name, request),
         fields(
             rpc.trace_id = tracing::field::Empty,
-            rpc.deadline = %humantime::format_rfc3339(ctx.deadline),
             otel.kind = "client",
             otel.name = request_name)
         )]
@@ -129,7 +129,7 @@ impl<Req, Resp> Channel<Req, Resp> {
     ) -> Result<Resp, RpcError> {
         let span = Span::current();
         ctx.trace_context = trace::Context::try_from(&span).unwrap_or_else(|_| {
-            tracing::trace!(
+            tracing::warn!(
                 "OpenTelemetry subscriber not installed; making unsampled child context."
             );
             ctx.trace_context.new_child()
@@ -172,8 +172,7 @@ struct ResponseGuard<'a, Resp> {
 
 /// An error that can occur in the processing of an RPC. This is not request-specific errors but
 /// rather cross-cutting errors that can always occur.
-#[derive(thiserror::Error, Clone, Debug, PartialEq, Eq, Hash)]
-#[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))]
+#[derive(thiserror::Error, Debug)]
 pub enum RpcError {
     /// The client disconnected from the server.
     #[error("the client disconnected from the server")]
@@ -258,7 +257,6 @@ where
 
 /// Handles the lifecycle of requests, writing requests to the wire, managing cancellations,
 /// and dispatching responses to the appropriate channel.
-#[must_use]
 #[pin_project]
 #[derive(Debug)]
 pub struct RequestDispatch<Req, Resp, C> {
@@ -395,7 +393,11 @@ where
         // Receiving Poll::Ready(None) when polling expired requests never indicates "Closed",
         // because there can temporarily be zero in-flight rquests. Therefore, there is no need to
         // track the status like is done with pending and cancelled requests.
-        if let Poll::Ready(Some(_)) = self.in_flight_requests().poll_expired(cx) {
+        if let Poll::Ready(Some(_)) = self
+            .in_flight_requests()
+            .poll_expired(cx)
+            .map_err(ChannelError::Timer)?
+        {
             // Expired requests are considered complete; there is no compelling reason to send a
             // cancellation message to the server, since it will have already exhausted its
             // allotted processing time.
@@ -520,7 +522,11 @@ where
             },
         });
         self.start_send(request)?;
-        tracing::info!("SendRequest");
+        let deadline = ctx.deadline;
+        tracing::info!(
+            tarpc.deadline = %humantime::format_rfc3339(deadline),
+            "SendRequest"
+        );
         drop(entered);
 
         self.in_flight_requests()
@@ -602,9 +608,49 @@ struct DispatchRequest<Req, Resp> {
     pub response_completion: oneshot::Sender<Result<Response<Resp>, DeadlineExceededError>>,
 }
 
+/// Sends request cancellation signals.
+#[derive(Debug, Clone)]
+struct RequestCancellation(mpsc::UnboundedSender<u64>);
+
+/// A stream of IDs of requests that have been canceled.
+#[derive(Debug)]
+struct CanceledRequests(mpsc::UnboundedReceiver<u64>);
+
+/// Returns a channel to send request cancellation messages.
+fn cancellations() -> (RequestCancellation, CanceledRequests) {
+    // Unbounded because messages are sent in the drop fn. This is fine, because it's still
+    // bounded by the number of in-flight requests.
+    let (tx, rx) = mpsc::unbounded_channel();
+    (RequestCancellation(tx), CanceledRequests(rx))
+}
+
+impl RequestCancellation {
+    /// Cancels the request with ID `request_id`.
+    fn cancel(&self, request_id: u64) {
+        let _ = self.0.send(request_id);
+    }
+}
+
+impl CanceledRequests {
+    fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<u64>> {
+        self.0.poll_recv(cx)
+    }
+}
+
+impl Stream for CanceledRequests {
+    type Item = u64;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<u64>> {
+        self.poll_recv(cx)
+    }
+}
+
 #[cfg(test)]
 mod tests {
-    use super::{cancellations, Channel, DispatchRequest, RequestDispatch, ResponseGuard};
+    use super::{
+        cancellations, CanceledRequests, Channel, DispatchRequest, RequestCancellation,
+        RequestDispatch, ResponseGuard,
+    };
     use crate::{
         client::{
             in_flight_requests::{DeadlineExceededError, InFlightRequests},
@@ -657,7 +703,7 @@ mod tests {
         });
         // resp's drop() is run, which should send a cancel message.
         let cx = &mut Context::from_waker(&noop_waker_ref());
-        assert_eq!(canceled_requests.poll_recv(cx), Poll::Ready(Some(3)));
+        assert_eq!(canceled_requests.0.poll_recv(cx), Poll::Ready(Some(3)));
     }
 
     #[tokio::test]
@@ -680,7 +726,7 @@ mod tests {
         .unwrap();
         drop(cancellation);
         let cx = &mut Context::from_waker(&noop_waker_ref());
-        assert_eq!(canceled_requests.poll_recv(cx), Poll::Ready(None));
+        assert_eq!(canceled_requests.0.poll_recv(cx), Poll::Ready(None));
     }
 
     #[tokio::test]
@@ -691,7 +737,6 @@ mod tests {
 
         let _resp = send_request(&mut channel, "hi", tx, &mut rx).await;
 
-        #[allow(unstable_name_collisions)]
         let req = dispatch.as_mut().poll_next_request(cx).ready();
         assert!(req.is_some());
 
@@ -722,7 +767,6 @@ mod tests {
         dispatch.await.unwrap();
     }
 
-    #[allow(unstable_name_collisions)]
     #[tokio::test]
     async fn stage_request_response_future_dropped_is_canceled_before_sending() {
         let (mut dispatch, mut channel, _server_channel) = set_up();
@@ -738,7 +782,6 @@ mod tests {
         assert!(dispatch.as_mut().poll_next_request(cx).ready().is_none());
     }
 
-    #[allow(unstable_name_collisions)]
     #[tokio::test]
     async fn stage_request_response_future_dropped_is_canceled_after_sending() {
         let (mut dispatch, mut channel, _server_channel) = set_up();
@@ -791,17 +834,18 @@ mod tests {
         let _ = tracing_subscriber::fmt().with_test_writer().try_init();
 
         let (to_dispatch, pending_requests) = mpsc::channel(1);
-        let (cancellation, canceled_requests) = cancellations();
+        let (cancel_tx, canceled_requests) = mpsc::unbounded_channel();
         let (client_channel, server_channel) = transport::channel::unbounded();
 
         let dispatch = RequestDispatch::<String, String, _> {
             transport: client_channel.fuse(),
             pending_requests: pending_requests,
-            canceled_requests,
+            canceled_requests: CanceledRequests(canceled_requests),
             in_flight_requests: InFlightRequests::default(),
             config: Config::default(),
         };
 
+        let cancellation = RequestCancellation(cancel_tx);
         let channel = Channel {
             to_dispatch,
             cancellation,
@@ -826,13 +870,13 @@ mod tests {
             request: request.to_string(),
             response_completion,
         };
-        let response_guard = ResponseGuard {
+        channel.to_dispatch.send(request).await.unwrap();
+
+        ResponseGuard {
             response,
             cancellation: &channel.cancellation,
             request_id,
-        };
-        channel.to_dispatch.send(request).await.unwrap();
-        response_guard
+        }
     }
 
     async fn send_response(

tarpc/src/client/in_flight_requests.rs

@@ -117,9 +117,12 @@ impl<Resp> InFlightRequests<Resp> {
 
     /// Yields a request that has expired, completing it with a TimedOut error.
     /// The caller should send cancellation messages for any yielded request ID.
-    pub fn poll_expired(&mut self, cx: &mut Context) -> Poll<Option<u64>> {
-        self.deadlines.poll_expired(cx).map(|expired| {
-            let request_id = expired?.into_inner();
+    pub fn poll_expired(
+        &mut self,
+        cx: &mut Context,
+    ) -> Poll<Option<Result<u64, tokio::time::error::Error>>> {
+        self.deadlines.poll_expired(cx).map_ok(|expired| {
+            let request_id = expired.into_inner();
             if let Some(request_data) = self.request_data.remove(&request_id) {
                 let _entered = request_data.span.enter();
                 tracing::error!("DeadlineExceeded");
@@ -128,7 +131,7 @@ impl<Resp> InFlightRequests<Resp> {
                     .response_completion
                     .send(Err(DeadlineExceededError));
             }
-            Some(request_id)
+            request_id
         })
     }
 }

tarpc/src/context.rs

@@ -28,8 +28,6 @@ pub struct Context {
     /// When the client expects the request to be complete by. The server should cancel the request
     /// if it is not complete by this time.
     #[cfg_attr(feature = "serde1", serde(default = "ten_seconds_from_now"))]
-    // Serialized as a Duration to prevent clock skew issues.
-    #[cfg_attr(feature = "serde1", serde(with = "absolute_to_relative_time"))]
     pub deadline: SystemTime,
     /// Uniquely identifies requests originating from the same source.
     /// When a service handles a request by making requests itself, those requests should
@@ -38,54 +36,6 @@ pub struct Context {
     pub trace_context: trace::Context,
 }
 
-#[cfg(feature = "serde1")]
-mod absolute_to_relative_time {
-    pub use serde::{Deserialize, Deserializer, Serialize, Serializer};
-    pub use std::time::{Duration, SystemTime};
-
-    pub fn serialize<S>(deadline: &SystemTime, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: Serializer,
-    {
-        let deadline = deadline
-            .duration_since(SystemTime::now())
-            .unwrap_or(Duration::ZERO);
-        deadline.serialize(serializer)
-    }
-
-    pub fn deserialize<'de, D>(deserializer: D) -> Result<SystemTime, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        let deadline = Duration::deserialize(deserializer)?;
-        Ok(SystemTime::now() + deadline)
-    }
-
-    #[cfg(test)]
-    #[derive(serde::Serialize, serde::Deserialize)]
-    struct AbsoluteToRelative(#[serde(with = "self")] SystemTime);
-
-    #[test]
-    fn test_serialize() {
-        let now = SystemTime::now();
-        let deadline = now + Duration::from_secs(10);
-        let serialized_deadline = bincode::serialize(&AbsoluteToRelative(deadline)).unwrap();
-        let deserialized_deadline: Duration = bincode::deserialize(&serialized_deadline).unwrap();
-        // TODO: how to avoid flakiness?
-        assert!(deserialized_deadline > Duration::from_secs(9));
-    }
-
-    #[test]
-    fn test_deserialize() {
-        let deadline = Duration::from_secs(10);
-        let serialized_deadline = bincode::serialize(&deadline).unwrap();
-        let AbsoluteToRelative(deserialized_deadline) =
-            bincode::deserialize(&serialized_deadline).unwrap();
-        // TODO: how to avoid flakiness?
-        assert!(deserialized_deadline > SystemTime::now() + Duration::from_secs(9));
-    }
-}
-
 assert_impl_all!(Context: Send, Sync);
 
 fn ten_seconds_from_now() -> SystemTime {

tarpc/src/lib.rs

@@ -27,7 +27,7 @@
 //! process, and no context switching between different languages.
 //!
 //! Some other features of tarpc:
-//! - Pluggable transport: any type implementing `Stream<Item = Request> + Sink<Response>` can be
+//! - Pluggable transport: any type impling `Stream<Item = Request> + Sink<Response>` can be
 //!   used as a transport to connect the client and server.
 //! - `Send + 'static` optional: if the transport doesn't require it, neither does tarpc!
 //! - Cascading cancellation: dropping a request will send a cancellation message to the server.
@@ -42,7 +42,7 @@
 //!   [tracing](https://github.com/tokio-rs/tracing) primitives extended with
 //!   [OpenTelemetry](https://opentelemetry.io/) traces. Using a compatible tracing subscriber like
 //!   [Jaeger](https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-jaeger),
-//!   each RPC can be traced through the client, server, and other dependencies downstream of the
+//!   each RPC can be traced through the client, server, amd other dependencies downstream of the
 //!   server. Even for applications not connected to a distributed tracing collector, the
 //!   instrumentation can also be ingested by regular loggers like
 //!   [env_logger](https://github.com/env-logger-rs/env_logger/).
@@ -54,7 +54,7 @@
 //! Add to your `Cargo.toml` dependencies:
 //!
 //! ```toml
-//! tarpc = "0.29"
+//! tarpc = "0.27"
 //! ```
 //!
 //! The `tarpc::service` attribute expands to a collection of items that form an rpc service.
@@ -69,7 +69,7 @@
 //! ```toml
 //! anyhow = "1.0"
 //! futures = "0.3"
-//! tarpc = { version = "0.29", features = ["tokio1"] }
+//! tarpc = { version = "0.27", features = ["tokio1"] }
 //! tokio = { version = "1.0", features = ["macros"] }
 //! ```
 //!
@@ -88,7 +88,7 @@
 //! };
 //! use tarpc::{
 //!     client, context,
-//!     server::{self, incoming::Incoming, Channel},
+//!     server::{self, incoming::Incoming},
 //! };
 //!
 //! // This is the service definition. It looks a lot like a trait definition.
@@ -132,7 +132,7 @@
 //!     type HelloFut = Ready<String>;
 //!
 //!     fn hello(self, _: context::Context, name: String) -> Self::HelloFut {
-//!         future::ready(format!("Hello, {name}!"))
+//!         future::ready(format!("Hello, {}!", name))
 //!     }
 //! }
 //! ```
@@ -168,7 +168,7 @@
 //! #     // an associated type representing the future output by the fn.
 //! #     type HelloFut = Ready<String>;
 //! #     fn hello(self, _: context::Context, name: String) -> Self::HelloFut {
-//! #         future::ready(format!("Hello, {name}!"))
+//! #         future::ready(format!("Hello, {}!", name))
 //! #     }
 //! # }
 //! # #[cfg(not(feature = "tokio1"))]
@@ -190,7 +190,7 @@
 //!     // specifies a deadline and trace information which can be helpful in debugging requests.
 //!     let hello = client.hello(context::current(), "Stim".to_string()).await?;
 //!
-//!     println!("{hello}");
+//!     println!("{}", hello);
 //!
 //!     Ok(())
 //! }
@@ -209,7 +209,7 @@
 pub use serde;
 
 #[cfg(feature = "serde-transport")]
-pub use {tokio_serde, tokio_util};
+pub use tokio_serde;
 
 #[cfg(feature = "serde-transport")]
 #[cfg_attr(docsrs, doc(cfg(feature = "serde-transport")))]
@@ -264,7 +264,7 @@ pub use tarpc_plugins::service;
 /// #[tarpc::server]
 /// impl World for HelloServer {
 ///     async fn hello(self, _: context::Context, name: String) -> String {
-///         format!("Hello, {name}! You are connected from {:?}.", self.0)
+///         format!("Hello, {}! You are connected from {:?}.", name, self.0)
 ///     }
 /// }
 /// ```
@@ -290,7 +290,7 @@ pub use tarpc_plugins::service;
 ///     fn hello(self, _: context::Context, name: String) -> Pin<Box<dyn Future<Output = String>
 ///     + Send>> {
 ///         Box::pin(async move {
-///             format!("Hello, {name}! You are connected from {:?}.", self.0)
+///             format!("Hello, {}! You are connected from {:?}.", name, self.0)
 ///         })
 ///     }
 /// }
@@ -300,7 +300,6 @@ pub use tarpc_plugins::service;
 /// `async`, meaning that this should not break existing code.
 pub use tarpc_plugins::server;
 
-pub(crate) mod cancellations;
 pub mod client;
 pub mod context;
 pub mod server;

tarpc/src/serde_transport.rs

@@ -149,7 +149,6 @@ pub mod tcp {
     }
 
     /// A connection Future that also exposes the length-delimited framing config.
-    #[must_use]
     #[pin_project]
     pub struct Connect<T, Item, SinkItem, CodecFn> {
         #[pin]

tarpc/src/server.rs

@@ -7,7 +7,6 @@
 //! Provides a server that concurrently handles many connections sending multiplexed requests.
 
 use crate::{
-    cancellations::{cancellations, CanceledRequests, RequestCancellation},
     context::{self, SpanExt},
     trace, ClientMessage, Request, Response, Transport,
 };
@@ -21,14 +20,7 @@ use futures::{
 };
 use in_flight_requests::{AlreadyExistsError, InFlightRequests};
 use pin_project::pin_project;
-use std::{
-    convert::TryFrom,
-    error::Error,
-    fmt,
-    marker::PhantomData,
-    mem::{self, ManuallyDrop},
-    pin::Pin,
-};
+use std::{convert::TryFrom, error::Error, fmt, marker::PhantomData, pin::Pin};
 use tracing::{info_span, instrument::Instrument, Span};
 
 mod in_flight_requests;
@@ -119,11 +111,6 @@ pub struct BaseChannel<Req, Resp, T> {
     /// Writes responses to the wire and reads requests off the wire.
     #[pin]
     transport: Fuse<T>,
-    /// In-flight requests that were dropped by the server before completion.
-    #[pin]
-    canceled_requests: CanceledRequests,
-    /// Notifies `canceled_requests` when a request is canceled.
-    request_cancellation: RequestCancellation,
     /// Holds data necessary to clean up in-flight requests.
     in_flight_requests: InFlightRequests,
     /// Types the request and response.
@@ -136,12 +123,9 @@ where
 {
     /// Creates a new channel backed by `transport` and configured with `config`.
     pub fn new(config: Config, transport: T) -> Self {
-        let (request_cancellation, canceled_requests) = cancellations();
         BaseChannel {
             config,
             transport: transport.fuse(),
-            canceled_requests,
-            request_cancellation,
             in_flight_requests: InFlightRequests::default(),
             ghost: PhantomData,
         }
@@ -166,24 +150,17 @@ where
         self.as_mut().project().in_flight_requests
     }
 
-    fn canceled_requests_pin_mut<'a>(
-        self: &'a mut Pin<&mut Self>,
-    ) -> Pin<&'a mut CanceledRequests> {
-        self.as_mut().project().canceled_requests
-    }
-
     fn transport_pin_mut<'a>(self: &'a mut Pin<&mut Self>) -> Pin<&'a mut Fuse<T>> {
         self.as_mut().project().transport
     }
 
     fn start_request(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         mut request: Request<Req>,
     ) -> Result<TrackedRequest<Req>, AlreadyExistsError> {
         let span = info_span!(
             "RPC",
             rpc.trace_id = %request.context.trace_id(),
-            rpc.deadline = %humantime::format_rfc3339(request.context.deadline),
             otel.kind = "server",
             otel.name = tracing::field::Empty,
         );
@@ -197,7 +174,7 @@ where
         });
         let entered = span.enter();
         tracing::info!("ReceiveRequest");
-        let start = self.in_flight_requests_mut().start_request(
+        let start = self.project().in_flight_requests.start_request(
             request.id,
             request.context.deadline,
             span.clone(),
@@ -206,13 +183,9 @@ where
             Ok(abort_registration) => {
                 drop(entered);
                 Ok(TrackedRequest {
+                    request,
                     abort_registration,
                     span,
-                    response_guard: ManuallyDrop::new(ResponseGuard {
-                        request_id: request.id,
-                        request_cancellation: self.request_cancellation.clone(),
-                    }),
-                    request,
                 })
             }
             Err(AlreadyExistsError) => {
@@ -239,8 +212,6 @@ pub struct TrackedRequest<Req> {
     pub abort_registration: AbortRegistration,
     /// A span representing the server processing of this request.
     pub span: Span,
-    /// An inert response guard. Becomes active in an InFlightRequest.
-    pub response_guard: ManuallyDrop<ResponseGuard>,
 }
 
 /// The server end of an open connection with a client, receiving requests from, and sending
@@ -259,15 +230,13 @@ pub struct TrackedRequest<Req> {
 ///    [`Sink::send`](futures::sink::SinkExt::send) - A user is free to manually read requests
 ///    from, and send responses into, a Channel in lieu of the previous methods. Channels stream
 ///    [`TrackedRequests`](TrackedRequest), which, in addition to the request itself, contains the
-///    server [`Span`], request lifetime [`AbortRegistration`], and an inert [`ResponseGuard`].
-///    Wrapping response logic in an [`Abortable`] future using the abort registration will ensure
-///    that the response does not execute longer than the request deadline. The `Channel` itself
-///    will clean up request state once either the deadline expires, or the response guard is
-///    dropped, or a response is sent.
-///
-/// Channels must be implemented using the decorator pattern: the only way to create a
-/// `TrackedRequest` is to get one from another `Channel`. Ultimately, all `TrackedRequests` are
-/// created by [`BaseChannel`].
+///    server [`Span`] and request lifetime [`AbortRegistration`]. Wrapping response
+///    logic in an [`Abortable`] future using the abort registration will ensure that the response
+///    does not execute longer than the request deadline. The `Channel` itself will clean up
+///    request state once either the deadline expires, or a cancellation message is received, or a
+///    response is sent. Because there is no guarantee that a cancellation message will ever be
+///    received for a request, or that requests come with reasonably short deadlines, services
+///    should strive to clean up Channel resources by sending a response for every request.
 pub trait Channel
 where
     Self: Transport<Response<<Self as Channel>::Resp>, TrackedRequest<<Self as Channel>::Req>>,
@@ -364,45 +333,20 @@ where
     type Item = Result<TrackedRequest<Req>, ChannelError<T::Error>>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
-        #[derive(Clone, Copy, Debug)]
+        #[derive(Debug)]
         enum ReceiverStatus {
             Ready,
             Pending,
             Closed,
         }
-
-        impl ReceiverStatus {
-            fn combine(self, other: Self) -> Self {
-                use ReceiverStatus::*;
-                match (self, other) {
-                    (Ready, _) | (_, Ready) => Ready,
-                    (Closed, Closed) => Closed,
-                    (Pending, Closed) | (Closed, Pending) | (Pending, Pending) => Pending,
-                }
-            }
-        }
-
         use ReceiverStatus::*;
 
         loop {
-            let cancellation_status = match self.canceled_requests_pin_mut().poll_recv(cx) {
-                Poll::Ready(Some(request_id)) => {
-                    if let Some(span) = self.in_flight_requests_mut().remove_request(request_id) {
-                        let _entered = span.enter();
-                        tracing::info!("ResponseCancelled");
-                    }
-                    Ready
-                }
-                // Pending cancellations don't block Channel closure, because all they do is ensure
-                // the Channel's internal state is cleaned up. But Channel closure also cleans up
-                // the Channel state, so there's no reason to wait on a cancellation before
-                // closing.
-                //
-                // Ready(None) can't happen, since `self` holds a Cancellation.
-                Poll::Pending | Poll::Ready(None) => Closed,
-            };
-
-            let expiration_status = match self.in_flight_requests_mut().poll_expired(cx) {
+            let expiration_status = match self
+                .in_flight_requests_mut()
+                .poll_expired(cx)
+                .map_err(ChannelError::Timer)?
+            {
                 // No need to send a response, since the client wouldn't be waiting for one
                 // anymore.
                 Poll::Ready(Some(_)) => Ready,
@@ -450,13 +394,10 @@ where
                 expiration_status,
                 request_status
             );
-            match cancellation_status
-                .combine(expiration_status)
-                .combine(request_status)
-            {
-                Ready => continue,
-                Closed => return Poll::Ready(None),
-                Pending => return Poll::Pending,
+            match (expiration_status, request_status) {
+                (Ready, _) | (_, Ready) => continue,
+                (Closed, Closed) => return Poll::Ready(None),
+                (Pending, Closed) | (Closed, Pending) | (Pending, Pending) => return Poll::Pending,
             }
         }
     }
@@ -478,7 +419,9 @@ where
 
     fn start_send(mut self: Pin<&mut Self>, response: Response<Resp>) -> Result<(), Self::Error> {
         if let Some(span) = self
-            .in_flight_requests_mut()
+            .as_mut()
+            .project()
+            .in_flight_requests
             .remove_request(response.request_id)
         {
             let _entered = span.enter();
@@ -555,11 +498,6 @@ impl<C> Requests<C>
 where
     C: Channel,
 {
-    /// Returns a reference to the inner channel over which messages are sent and received.
-    pub fn channel(&self) -> &C {
-        &self.channel
-    }
-
     /// Returns the inner channel over which messages are sent and received.
     pub fn channel_pin_mut<'a>(self: &'a mut Pin<&mut Self>) -> Pin<&'a mut C> {
         self.as_mut().project().channel
@@ -576,22 +514,12 @@ where
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<InFlightRequest<C::Req, C::Resp>, C::Error>>> {
-        self.channel_pin_mut().poll_next(cx).map_ok(
-            |TrackedRequest {
-                 request,
-                 abort_registration,
-                 span,
-                 response_guard,
-             }| {
-                InFlightRequest {
-                    request,
-                    abort_registration,
-                    span,
-                    response_guard: ManuallyDrop::into_inner(response_guard),
-                    response_tx: self.responses_tx.clone(),
-                }
-            },
-        )
+        self.channel_pin_mut()
+            .poll_next(cx)
+            .map_ok(|request| InFlightRequest {
+                request,
+                response_tx: self.responses_tx.clone(),
+            })
     }
 
     fn pump_write(
@@ -668,37 +596,17 @@ where
     }
 }
 
-/// A fail-safe to ensure requests are properly canceled if request processing is aborted before
-/// completing.
-#[derive(Debug)]
-pub struct ResponseGuard {
-    request_cancellation: RequestCancellation,
-    request_id: u64,
-}
-
-impl Drop for ResponseGuard {
-    fn drop(&mut self) {
-        self.request_cancellation.cancel(self.request_id);
-    }
-}
-
 /// A request produced by [Channel::requests].
-///
-/// If dropped without calling [`execute`](InFlightRequest::execute), a cancellation message will
-/// be sent to the Channel to clean up associated request state.
 #[derive(Debug)]
 pub struct InFlightRequest<Req, Res> {
-    request: Request<Req>,
-    abort_registration: AbortRegistration,
-    response_guard: ResponseGuard,
-    span: Span,
+    request: TrackedRequest<Req>,
     response_tx: mpsc::Sender<Response<Res>>,
 }
 
 impl<Req, Res> InFlightRequest<Req, Res> {
     /// Returns a reference to the request.
     pub fn get(&self) -> &Request<Req> {
-        &self.request
+        &self.request.request
     }
 
     /// Returns a [future](Future) that executes the request using the given [service
@@ -712,23 +620,22 @@ impl<Req, Res> InFlightRequest<Req, Res> {
     ///    message](ClientMessage::Cancel) for this request.
     /// 2. The request [deadline](crate::context::Context::deadline) is reached.
     /// 3. The service function completes.
-    ///
-    /// If the returned Future is dropped before completion, a cancellation message will be sent to
-    /// the Channel to clean up associated request state.
     pub async fn execute<S>(self, serve: S)
     where
         S: Serve<Req, Resp = Res>,
     {
         let Self {
             response_tx,
-            response_guard,
-            abort_registration,
-            span,
             request:
-                Request {
-                    context,
-                    message,
-                    id: request_id,
+                TrackedRequest {
+                    abort_registration,
+                    span,
+                    request:
+                        Request {
+                            context,
+                            message,
+                            id: request_id,
+                        },
                 },
         } = self;
         let method = serve.method(&message);
@@ -749,10 +656,6 @@ impl<Req, Res> InFlightRequest<Req, Res> {
         )
         .instrument(span)
         .await;
-        // Request processing has completed, meaning either the channel canceled the request or
-        // a request was sent back to the channel. Either way, the channel will clean up the
-        // request data, so the request does not need to be canceled.
-        mem::forget(response_guard);
     }
 }
 
@@ -1028,44 +931,6 @@ mod tests {
         assert_eq!(channel.in_flight_requests(), 0);
     }
 
-    #[tokio::test]
-    async fn in_flight_request_drop_cancels_request() {
-        let (mut requests, mut tx) = test_requests::<(), ()>();
-        tx.send(fake_request(())).await.unwrap();
-
-        let request = match requests.as_mut().poll_next(&mut noop_context()) {
-            Poll::Ready(Some(Ok(request))) => request,
-            result => panic!("Unexpected result: {:?}", result),
-        };
-        drop(request);
-
-        let poll = requests
-            .as_mut()
-            .channel_pin_mut()
-            .poll_next(&mut noop_context());
-        assert!(poll.is_pending());
-        let in_flight_requests = requests.channel().in_flight_requests();
-        assert_eq!(in_flight_requests, 0);
-    }
-
-    #[tokio::test]
-    async fn in_flight_requests_successful_execute_doesnt_cancel_request() {
-        let (mut requests, mut tx) = test_requests::<(), ()>();
-        tx.send(fake_request(())).await.unwrap();
-
-        let request = match requests.as_mut().poll_next(&mut noop_context()) {
-            Poll::Ready(Some(Ok(request))) => request,
-            result => panic!("Unexpected result: {:?}", result),
-        };
-        request.execute(|_, _| async {}).await;
-        assert!(requests
-            .as_mut()
-            .channel_pin_mut()
-            .canceled_requests
-            .poll_recv(&mut noop_context())
-            .is_pending());
-    }
-
     #[tokio::test]
     async fn requests_poll_next_response_returns_pending_when_buffer_full() {
         let (mut requests, _tx) = test_bounded_requests::<(), ()>(0);

tarpc/src/server/in_flight_requests.rs

@@ -94,14 +94,16 @@ impl InFlightRequests {
     }
 
     /// Yields a request that has expired, aborting any ongoing processing of that request.
-    pub fn poll_expired(&mut self, cx: &mut Context) -> Poll<Option<u64>> {
+    pub fn poll_expired(
+        &mut self,
+        cx: &mut Context,
+    ) -> Poll<Option<Result<u64, tokio::time::error::Error>>> {
         if self.deadlines.is_empty() {
             // TODO(https://github.com/tokio-rs/tokio/issues/4161)
             // This is a workaround for DelayQueue not always treating this case correctly.
             return Poll::Ready(None);
         }
-        self.deadlines.poll_expired(cx).map(|expired| {
-            let expired = expired?;
+        self.deadlines.poll_expired(cx).map_ok(|expired| {
             if let Some(RequestData {
                 abort_handle, span, ..
             }) = self.request_data.remove(expired.get_ref())
@@ -111,7 +113,7 @@ impl InFlightRequests {
                 abort_handle.abort();
                 tracing::error!("DeadlineExceeded");
             }
-            Some(expired.into_inner())
+            expired.into_inner()
         })
     }
 }
@@ -159,7 +161,7 @@ mod tests {
 
         assert_matches!(
             in_flight_requests.poll_expired(&mut noop_context()),
-            Poll::Ready(Some(_))
+            Poll::Ready(Some(Ok(_)))
         );
         assert_matches!(
             abortable_future.poll_unpin(&mut noop_context()),

tarpc/src/server/testing.rs

@@ -5,14 +5,13 @@
 // https://opensource.org/licenses/MIT.
 
 use crate::{
-    cancellations::{cancellations, CanceledRequests, RequestCancellation},
     context,
-    server::{Channel, Config, ResponseGuard, TrackedRequest},
+    server::{Channel, Config, TrackedRequest},
     Request, Response,
 };
 use futures::{task::*, Sink, Stream};
 use pin_project::pin_project;
-use std::{collections::VecDeque, io, mem::ManuallyDrop, pin::Pin, time::SystemTime};
+use std::{collections::VecDeque, io, pin::Pin, time::SystemTime};
 use tracing::Span;
 
 #[pin_project]
@@ -23,8 +22,6 @@ pub(crate) struct FakeChannel<In, Out> {
     pub sink: VecDeque<Out>,
     pub config: Config,
     pub in_flight_requests: super::in_flight_requests::InFlightRequests,
-    pub request_cancellation: RequestCancellation,
-    pub canceled_requests: CanceledRequests,
 }
 
 impl<In, Out> Stream for FakeChannel<In, Out>
@@ -89,7 +86,6 @@ where
 impl<Req, Resp> FakeChannel<io::Result<TrackedRequest<Req>>, Response<Resp>> {
     pub fn push_req(&mut self, id: u64, message: Req) {
         let (_, abort_registration) = futures::future::AbortHandle::new_pair();
-        let (request_cancellation, _) = cancellations();
         self.stream.push_back(Ok(TrackedRequest {
             request: Request {
                 context: context::Context {
@@ -101,24 +97,17 @@ impl<Req, Resp> FakeChannel<io::Result<TrackedRequest<Req>>, Response<Resp>> {
             },
             abort_registration,
             span: Span::none(),
-            response_guard: ManuallyDrop::new(ResponseGuard {
-                request_cancellation,
-                request_id: id,
-            }),
         }));
     }
 }
 
 impl FakeChannel<(), ()> {
     pub fn default<Req, Resp>() -> FakeChannel<io::Result<TrackedRequest<Req>>, Response<Resp>> {
-        let (request_cancellation, canceled_requests) = cancellations();
         FakeChannel {
             stream: Default::default(),
             sink: Default::default(),
             config: Default::default(),
             in_flight_requests: Default::default(),
-            request_cancellation,
-            canceled_requests,
         }
     }
 }

tarpc/src/server/tokio.rs

@@ -6,7 +6,6 @@ use std::pin::Pin;
 /// A future that drives the server by [spawning](tokio::spawn) a [`TokioChannelExecutor`](TokioChannelExecutor)
 /// for each new channel. Returned by
 /// [`Incoming::execute`](crate::server::incoming::Incoming::execute).
-#[must_use]
 #[pin_project]
 #[derive(Debug)]
 pub struct TokioServerExecutor<T, S> {
@@ -24,7 +23,6 @@ impl<T, S> TokioServerExecutor<T, S> {
 /// A future that drives the server by [spawning](tokio::spawn) each [response
 /// handler](super::InFlightRequest::execute) on tokio's default executor. Returned by
 /// [`Channel::execute`](crate::server::Channel::execute).
-#[must_use]
 #[pin_project]
 #[derive(Debug)]
 pub struct TokioChannelExecutor<T, S> {

tarpc/src/trace.rs

@@ -138,25 +138,25 @@ impl From<u64> for SpanId {
 
 impl From<opentelemetry::trace::TraceId> for TraceId {
     fn from(trace_id: opentelemetry::trace::TraceId) -> Self {
-        Self::from(u128::from_be_bytes(trace_id.to_bytes()))
+        Self::from(trace_id.to_u128())
     }
 }
 
 impl From<TraceId> for opentelemetry::trace::TraceId {
     fn from(trace_id: TraceId) -> Self {
-        Self::from_bytes(u128::from(trace_id).to_be_bytes())
+        Self::from_u128(trace_id.into())
     }
 }
 
 impl From<opentelemetry::trace::SpanId> for SpanId {
     fn from(span_id: opentelemetry::trace::SpanId) -> Self {
-        Self::from(u64::from_be_bytes(span_id.to_bytes()))
+        Self::from(span_id.to_u64())
     }
 }
 
 impl From<SpanId> for opentelemetry::trace::SpanId {
     fn from(span_id: SpanId) -> Self {
-        Self::from_bytes(u64::from(span_id).to_be_bytes())
+        Self::from_u64(span_id.0)
     }
 }
 

tarpc/src/transport/channel.rs

@@ -181,7 +181,7 @@ mod tests {
                     future::ready(request.parse::<u64>().map_err(|_| {
                         io::Error::new(
                             io::ErrorKind::InvalidInput,
-                            format!("{request:?} is not an int"),
+                            format!("{:?} is not an int", request),
                         )
                     }))
                 }),

tarpc/src/util.rs

@@ -38,34 +38,11 @@ where
     H: BuildHasher,
 {
     fn compact(&mut self, usage_ratio_threshold: f64) {
-        let usage_ratio_threshold = usage_ratio_threshold.clamp(f64::MIN_POSITIVE, 1.);
-        let cap = f64::max(1000., self.len() as f64 / usage_ratio_threshold);
-        self.shrink_to(cap as usize);
+        if self.capacity() > 1000 {
+            let usage_ratio = self.len() as f64 / self.capacity() as f64;
+            if usage_ratio < usage_ratio_threshold {
+                self.shrink_to_fit();
+            }
+        }
     }
 }
-
-#[test]
-fn test_compact() {
-    let mut map = HashMap::with_capacity(2048);
-    assert_eq!(map.capacity(), 3584);
-
-    // Make usage ratio 25%
-    for i in 0..896 {
-        map.insert(format!("k{i}"), "v");
-    }
-
-    map.compact(-1.0);
-    assert_eq!(map.capacity(), 3584);
-
-    map.compact(0.25);
-    assert_eq!(map.capacity(), 3584);
-
-    map.compact(0.50);
-    assert_eq!(map.capacity(), 1792);
-
-    map.compact(1.0);
-    assert_eq!(map.capacity(), 1792);
-
-    map.compact(2.0);
-    assert_eq!(map.capacity(), 1792);
-}

tarpc/tests/compile_fail.rs

@@ -2,8 +2,4 @@
 fn ui() {
     let t = trybuild::TestCases::new();
     t.compile_fail("tests/compile_fail/*.rs");
-    #[cfg(feature = "tokio1")]
-    t.compile_fail("tests/compile_fail/tokio/*.rs");
-    #[cfg(all(feature = "serde-transport", feature = "tcp"))]
-    t.compile_fail("tests/compile_fail/serde_transport/*.rs");
 }

tarpc/tests/compile_fail/must_use_request_dispatch.rs

@@ -1,15 +0,0 @@
-use tarpc::client;
-
-#[tarpc::service]
-trait World {
-    async fn hello(name: String) -> String;
-}
-
-fn main() {
-    let (client_transport, _) = tarpc::transport::channel::unbounded();
-
-    #[deny(unused_must_use)]
-    {
-        WorldClient::new(client::Config::default(), client_transport).dispatch;
-    }
-}

tarpc/tests/compile_fail/must_use_request_dispatch.stderr

@@ -1,11 +0,0 @@
-error: unused `RequestDispatch` that must be used
-  --> tests/compile_fail/must_use_request_dispatch.rs:13:9
-   |
-13 |         WorldClient::new(client::Config::default(), client_transport).dispatch;
-   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-   |
-note: the lint level is defined here
-  --> tests/compile_fail/must_use_request_dispatch.rs:11:12
-   |
-11 |     #[deny(unused_must_use)]
-   |            ^^^^^^^^^^^^^^^

tarpc/tests/compile_fail/serde_transport/must_use_tcp_connect.rs

@@ -1,9 +0,0 @@
-use tarpc::serde_transport;
-use tokio_serde::formats::Json;
-
-fn main() {
-    #[deny(unused_must_use)]
-    {
-        serde_transport::tcp::connect::<_, (), (), _, _>("0.0.0.0:0", Json::default);
-    }
-}

tarpc/tests/compile_fail/serde_transport/must_use_tcp_connect.stderr

@@ -1,11 +0,0 @@
-error: unused `Connect` that must be used
- --> tests/compile_fail/serde_transport/must_use_tcp_connect.rs:7:9
-  |
-7 |         serde_transport::tcp::connect::<_, (), (), _, _>("0.0.0.0:0", Json::default);
-  |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-  |
-note: the lint level is defined here
- --> tests/compile_fail/serde_transport/must_use_tcp_connect.rs:5:12
-  |
-5 |     #[deny(unused_must_use)]
-  |            ^^^^^^^^^^^^^^^

tarpc/tests/compile_fail/tarpc_server_missing_async.rs

@@ -7,8 +7,8 @@ struct HelloServer;
 
 #[tarpc::server]
 impl World for HelloServer {
-    fn hello(name: String) -> String {
-        format!("Hello, {name}!", name)
+    fn hello(name: String) ->  String {
+        format!("Hello, {}!", name)
     }
 }
 

tarpc/tests/compile_fail/tarpc_server_missing_async.stderr

@@ -7,5 +7,5 @@ error: not all trait items implemented, missing: `HelloFut`
 error: hint: `#[tarpc::server]` only rewrites async fns, and `fn hello` is not async
   --> $DIR/tarpc_server_missing_async.rs:10:5
    |
-10 |     fn hello(name: String) -> String {
+10 |     fn hello(name: String) ->  String {
    |     ^^

tarpc/tests/compile_fail/tokio/must_use_channel_executor.rs

@@ -1,29 +0,0 @@
-use tarpc::{
-    context,
-    server::{self, Channel},
-};
-
-#[tarpc::service]
-trait World {
-    async fn hello(name: String) -> String;
-}
-
-#[derive(Clone)]
-struct HelloServer;
-
-#[tarpc::server]
-impl World for HelloServer {
-    async fn hello(self, _: context::Context, name: String) -> String {
-        format!("Hello, {name}!")
-    }
-}
-
-fn main() {
-    let (_, server_transport) = tarpc::transport::channel::unbounded();
-    let server = server::BaseChannel::with_defaults(server_transport);
-
-    #[deny(unused_must_use)]
-    {
-        server.execute(HelloServer.serve());
-    }
-}

tarpc/tests/compile_fail/tokio/must_use_channel_executor.stderr

@@ -1,11 +0,0 @@
-error: unused `TokioChannelExecutor` that must be used
-  --> tests/compile_fail/tokio/must_use_channel_executor.rs:27:9
-   |
-27 |         server.execute(HelloServer.serve());
-   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-   |
-note: the lint level is defined here
-  --> tests/compile_fail/tokio/must_use_channel_executor.rs:25:12
-   |
-25 |     #[deny(unused_must_use)]
-   |            ^^^^^^^^^^^^^^^

tarpc/tests/compile_fail/tokio/must_use_server_executor.rs

@@ -1,30 +0,0 @@
-use futures::stream::once;
-use tarpc::{
-    context,
-    server::{self, incoming::Incoming},
-};
-
-#[tarpc::service]
-trait World {
-    async fn hello(name: String) -> String;
-}
-
-#[derive(Clone)]
-struct HelloServer;
-
-#[tarpc::server]
-impl World for HelloServer {
-    async fn hello(self, _: context::Context, name: String) -> String {
-        format!("Hello, {name}!")
-    }
-}
-
-fn main() {
-    let (_, server_transport) = tarpc::transport::channel::unbounded();
-    let server = once(async move { server::BaseChannel::with_defaults(server_transport) });
-
-    #[deny(unused_must_use)]
-    {
-        server.execute(HelloServer.serve());
-    }
-}

tarpc/tests/compile_fail/tokio/must_use_server_executor.stderr

@@ -1,11 +0,0 @@
-error: unused `TokioServerExecutor` that must be used
-  --> tests/compile_fail/tokio/must_use_server_executor.rs:28:9
-   |
-28 |         server.execute(HelloServer.serve());
-   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-   |
-note: the lint level is defined here
-  --> tests/compile_fail/tokio/must_use_server_executor.rs:26:12
-   |
-26 |     #[deny(unused_must_use)]
-   |            ^^^^^^^^^^^^^^^

tarpc/tests/service_functional.rs

@@ -31,7 +31,7 @@ impl Service for Server {
     type HeyFut = Ready<String>;
 
     fn hey(self, _: context::Context, name: String) -> Self::HeyFut {
-        ready(format!("Hey, {name}."))
+        ready(format!("Hey, {}.", name))
     }
 }