Cancellation token - Request context

[caenguidanos]

Hi!

In some languages like Go and C# we have the concept on request cancellation. I expect/try to do the same in Rust ecosystem.

For one hand, in Go, we have:

func main() {
    e := echo.New()

    var db *sql.DB

    e.GET("/product", func(c echo.Context) error {
        requestContext := c.Request().Context()

        var products []Product

        row := db.QueryRowContext(requestContext, "SELECT * FROM products")

	if row.Err() != nil {
		// ...
	}
    })

    e.Logger.Fatal(e.Start(":8080"))
}

The http library exposes a RequestContext that we can pass to the database driver. If the user aborts the request in some way before the query is completed then cancel the query.

The same is happening with C#:

public sealed class ProductController : Controller
{
    private IProductRepository _productRepository;

    public ProductController(IProductRepository productRepository)
    {
        _productRepository = productRepository;
    }

    [HttpGet("/")]
    public async Task<IEnumerable<Product>> Get(CancellationToken cancellationToken)
    {
        List<Product> products = await productRepository.ToListAsync(cancellationToken);

        return products;
    }
}

I see that tokio_postgres has some integration here but i don't find any reference in the actix world.

Someone has any idea about how to implement and integrate this is Rust-Actix-TokioPostgres?

[marcsantiago]

^ Bump, I've searched all over stack overflow, the web, and discord. This question has come up many times without a meaningful answer or solution.

"the server can't really know when the client cancels a request unless it also closes the connection, in which case the actix web server will not poll the handler any more" -Discord

Which we know be inherently false given other language authors do just that. It incredibly important to provide mechanisms to propagate client cancellations. These cancellations take many forms e.g. the connection was disconnected, the client set a request level timeout that was exceeded by the server, the client simply closed the connection prematurely.

Further examples include N users leaving pages after the page has requested resources from the server.

The inability to propagate these cancellations puts the viability of actix_web in production use cases at risk both from a resource perspective but also from a cost perspective. Every cancelled request that is not propagated cost money either represented as computing resources or downstream network I/O.

[caenguidanos]

I have a workaround.

Actix will drop the memory of the initialized variables inside the handlers when a HttpRequest in aborted. Then, we can use the Drop trait for implementing this:

mod repositories {
  use std::rc::Rc;
  
  use crate::academy;
  
  pub struct QuizRelationPostgresRepository {
      db: Rc<deadpool_postgres::Object>,
      cancel_tokens: Vec<tokio_postgres::CancelToken>,
  }
  
  impl Drop for QuizRelationPostgresRepository {
      fn drop(&mut self) {
          let cloned_cancel_tokens = self.cancel_tokens.clone();
  
          tokio::spawn(async {
              for cancel_token in cloned_cancel_tokens {
                  if let Err(error) = cancel_token.cancel_query(tokio_postgres::NoTls).await {
                      println!("{error}")
                  }
              }
          });
      }
  }
  
  impl QuizRelationPostgresRepository {
      pub fn new(db: Rc<deadpool_postgres::Object>) -> Self {
          Self { db, cancel_tokens: vec![] }
      }
  
      async fn refresh_quiz_relation_with_names(&mut self) -> Result<(), academy::domain::Error> {
          let db = &self.db;
          self.cancel_tokens.push(db.cancel_token());
  
          let stmt = db
              .prepare_cached("REFRESH MATERIALIZED VIEW quiz_relation_with_names")
              .await
              .map_err(|error| academy::domain::Error::Persistence(error.to_string()))?;
  
          db.execute(&stmt, &[])
              .await
              .map_err(|error| academy::domain::Error::Persistence(error.to_string()))?;
  
          Ok(())
      }
  }
}

It's like a transient strategy for the repository dependency because you need to create one each time on each handler, but works fine.

[marcsantiago]

@caenguidanos I'm not 100% sure that the hack above works generally speaking.

struct RequestContext {
    cancellation_token: tokio_util::sync::CancellationToken,
}

impl RequestContext {
    fn new() -> Self {
        RequestContext {
            cancellation_token: CancellationToken::new(),
        }
    }
}

impl Drop for RequestContext {
    fn drop(&mut self) {
        // called only after 10 seconds even if the client killed the request
        println!("drop for RequestContext called");
        self.cancellation_token.cancel();
    }
}

#[get("/can_cancel_hack")]
async fn client_can_cancel_drop_hack(req: HttpRequest) -> Result<HttpResponse> {
    let ctx = RequestContext::new();
    let resp = slow_reply_with_ctx(ctx);
    Ok(resp.await?)
}

async fn slow_reply_with_ctx(ctx: RequestContext) -> Result<HttpResponse> {
    println!("Going to sleep");
    select! {
        _ = ctx.cancellation_token.cancelled() => {
            println!("cancellation received"); // is never printed if cancelled...
            return Ok(HttpResponse::GatewayTimeout().finish());
        },
           _ =  tokio::time::sleep(Duration::from_secs(10)) =>{
            // pretend like this made calls to other workers
           Ok(HttpResponse::Ok().body("Hey sorry I took so long I was sleeping!"))
        },
    }
}

In the example above, even if the client terminates the requests... e.g. hit the cancel button on postman, the Drop is called only after the end of the slow_reply_with_ctx's scope.. So that means it waits the full 10 seconds in this example. The RequestContext has no idea of that the scope is limited to client's cancellation of the request. Hence, actix-web needs to expose a token such that N users can act on it.

[caenguidanos]

Yeah, you're right.

[marcsantiago]

According to their diagram (also I love the diagrams, all of them!)
https://actix.rs/docs/conn_lifecycle#accept-loop-in-more-detail

I don't want to turn this into a language comparison thread. Go is only used as an example because it's very tangible.

It seems like what's missing is the propagation of connection errors or disconnects. The LifeCycle is already polling. Languages like Go poll the connections at the contained within the "req: HttpRequest" object is a reference to the Context tied to all those events. This works for HTTP1 and HTTP2. It trivializes determining when a client cancels a request for the engineer implementing any server side APIs or resources.

Go does this at the Transport layer.
https://cs.opensource.google/go/go/+/refs/tags/go1.20.4:src/net/http/transport.go
https://www.prakharsrivastav.com/posts/golang-context-and-cancellation/

Given how Rust handles memory, it may not be easy to hook into a cancellation reference and bubble it up the same way you might in other languages. Maybe it's possible then to allow N users to optional pass in a handler, a OneShot Channel to listen to, something that is part of the app_data context, or some additional trait "Is_Cancelled" that can be provided as that app_data or data context.

[marcsantiago]

I looks like there is a crate that actix could add as a and guard it as a feature
https://docs.rs/io-context/latest/io_context/

and a generally good article on Cancellation issues within the Rust ecosystem
https://greptime.com/blogs/2023-01-12-hidden-control-flow