patx/skirr

--- src/lib.rs.before
+++ src/lib.rs
@@ -19,13 +19,24 @@
 type Link = Rc<RefCell<Connection>>;
 type Request = Rc<RefCell<RequestState>>;
 
-fn ready(py: Python<'_>, value: Py<PyAny>, state: Option<&State>) -> PyResult<Py<PyAny>> {
-    let yield_first = state.is_some_and(|s| {
-        let n = s.sends.get() + 1;
-        s.sends.set(if n >= 128 { 0 } else { n });
-        n >= 128
-    });
-    completed::value(py, Some(value), yield_first, state)
+// Fairness belongs to the producing request. Unrelated short requests must
+// not spend each other's quota and delay cleanup after an already-written body.
+fn cooperate(request: &Request) -> bool {
+    let mut r = request.borrow_mut();
+    r.operations += 1;
+    if r.operations < 128 {
+        return false;
+    }
+    r.operations = 0;
+    true
+}
+fn ready(
+    py: Python<'_>,
+    value: Py<PyAny>,
+    request: &Request,
+    state: Option<&State>,
+) -> PyResult<Py<PyAny>> {
+    completed::value(py, Some(value), cooperate(request), state)
 }
 fn message(py: Python<'_>, kind: &str, body: Option<(Py<PyBytes>, bool)>) -> PyResult<Py<PyAny>> {
     let d = PyDict::new(py);
@@ -156,6 +167,7 @@
     head: bool,
     minor: u8,
     task_done: bool,
+    operations: usize,
     trailers: bool,
     body_finished: bool,
     trailers_started: bool,
@@ -173,7 +185,7 @@
     connections: RefCell<HashMap<u64, Link>>,
     tasks: RefCell<HashMap<u64, Py<PyAny>>>,
     next: Cell<u64>,
-    sends: Cell<usize>,
+    eager_left: Cell<usize>,
     stopping: Cell<bool>,
     date: RefCell<Vec<u8>>,
     date_at: Cell<u64>,
@@ -570,6 +582,7 @@
             head: h.method == "HEAD",
             minor: h.minor,
             task_done: false,
+            operations: 0,
             trailers: false,
             body_finished: false,
             trailers_started: false,
@@ -598,7 +611,9 @@
                 .bind(py)
                 .call_method0(pyo3::intern!(py, "get_task_factory"))?
                 .is_none()
+                && self.eager_left.get() > 0
             {
+                self.eager_left.set(self.eager_left.get() - 1);
                 self.task_type
                     .bind(py)
                     .call((coro,), Some(self.task_options.bind(py)))?
@@ -1000,10 +1015,7 @@
         } else {
             return Err(PyValueError::new_err("Unsupported ASGI message"));
         }
-        let n = s.sends.get() + 1;
-        s.sends.set(n);
-        if n >= 128 {
-            s.sends.set(0);
+        if cooperate(&self.request) {
             return completed::value(py, None, true, Some(&s));
         }
         Ok(s.done.clone_ref(py))
@@ -1019,13 +1031,28 @@
 impl Receiver {
     fn __call__(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
         let Some(s) = self.state.upgrade() else {
-            return ready(py, message(py, "http.disconnect", None)?, None);
+            return ready(
+                py,
+                message(py, "http.disconnect", None)?,
+                &self.request,
+                None,
+            );
         };
         let Some(c) = self.request.borrow().connection.upgrade() else {
-            return ready(py, message(py, "http.disconnect", None)?, Some(&s));
+            return ready(
+                py,
+                message(py, "http.disconnect", None)?,
+                &self.request,
+                Some(&s),
+            );
         };
         if c.borrow().dead || self.request.borrow().complete || self.request.borrow().task_done {
-            return ready(py, message(py, "http.disconnect", None)?, Some(&s));
+            return ready(
+                py,
+                message(py, "http.disconnect", None)?,
+                &self.request,
+                Some(&s),
+            );
         }
         s.pump(py, &c)?;
         if !self.request.borrow().body.is_empty()
@@ -1033,7 +1060,7 @@
         {
             let msg = take_message(py, &self.request)?;
             s.pump(py, &c)?;
-            return ready(py, msg, Some(&s));
+            return ready(py, msg, &self.request, Some(&s));
         }
         let old = self.request.borrow_mut().receive.take();
         if let Some(f) = old
@@ -1072,6 +1099,9 @@
 #[pymethods]
 impl Done {
     fn __call__(&self, py: Python<'_>, task: &Bound<'_, PyAny>) -> PyResult<()> {
+        if let Some(s) = self.state.upgrade() {
+            s.eager_left.set(0);
+        }
         self.finish(py, task, true)
     }
 }
@@ -1151,6 +1181,7 @@
 impl Resume {
     fn __call__(&self, py: Python<'_>) -> PyResult<()> {
         if let (Some(s), Some(c)) = (self.state.upgrade(), self.connection.upgrade()) {
+            s.eager_left.set(0);
             c.borrow_mut().resume_scheduled = false;
             s.pump(py, &c)?;
         }
@@ -1218,7 +1249,7 @@
                 connections: RefCell::new(HashMap::new()),
                 tasks: RefCell::new(HashMap::new()),
                 next: Cell::new(0),
-                sends: Cell::new(0),
+                eager_left: Cell::new(0),
                 stopping: Cell::new(false),
                 date: RefCell::new(Vec::new()),
                 date_at: Cell::new(0),
@@ -1236,7 +1267,7 @@
     }
     fn poll(&self, py: Python<'_>) -> PyResult<()> {
         let s = &self.state;
-        s.sends.set(0);
+        s.eager_left.set(0);
         for (token, mask) in s.poller.events()? {
             if token == 0 {
                 if s.stopping.get() {