Fix Windows wheel build and inherited flow-control, typing, and profiling CI failures

src/bindings/loop_api/asyncio_cache.rs

@@ -39,6 +39,7 @@ cached_function!(
 );
 cached_function!(future_done_stop_fn, "rsloop._loop", "future_done_stop");
 
+#[cfg_attr(feature = "profile", hotpath::measure)]
 #[pyfunction]
 fn loop_reference_gone(_reference: &Bound<'_, PyAny>) {}
 
@@ -49,6 +50,7 @@ struct LoopBoundCallback {
 }
 #[pymethods]
 impl LoopBoundCallback {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopBoundCallback"))]
     fn __call__(&self, py: Python<'_>, arg: &Bound<'_, PyAny>) -> PyResult<()> {
         use pyo3::types::PyWeakrefMethods;
         if let Some(loop_obj) = self.loop_ref.bind(py).upgrade() {
@@ -58,6 +60,7 @@ impl LoopBoundCallback {
     }
 }
 
+#[cfg_attr(feature = "profile", hotpath::measure)]
 pub(super) fn loop_bound_callback(
     py: Python<'_>,
     core: &crate::engine::LoopCore,
@@ -108,6 +111,7 @@ pub(super) fn loop_bound_callback(
     Ok(callbacks[index].clone_ref(py))
 }
 
+#[cfg_attr(feature = "profile", hotpath::measure)]
 pub(super) fn ensure_future<'py>(
     py: Python<'py>,
     future: &Py<PyAny>,

src/bindings/loop_api/socket_operation.rs

@@ -370,6 +370,10 @@ fn arm_wait(py: Python<'_>, operation: Py<SocketOperation>) -> PyResult<()> {
 /// when the service future is cancelled before its first poll.
 struct RequestCleanup(Arc<WakeState>);
 impl Drop for RequestCleanup {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<RequestCleanup as Drop>")
+    )]
     fn drop(&mut self) {
         let pending = self
             .0

src/engine/callbacks.rs

@@ -53,6 +53,10 @@ struct CallbackResult<'py> {
     _python: std::marker::PhantomData<Python<'py>>,
 }
 impl Drop for CallbackResult<'_> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<CallbackResult as Drop>")
+    )]
     fn drop(&mut self) {
         // SAFETY: this owns a nonnull result reference; the Python lifetime
         // prevents moving its destruction outside the attached invocation.
@@ -248,13 +252,15 @@ impl<const SOURCE_WORDS: usize> ReadyCallback<SOURCE_WORDS> {
         self.invoke_with(py, |ready| ready.invoke_direct(py))
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ReadyCallback"))]
     /// Invoke a scheduled callback and release its result after context exit.
     pub fn invoke_discard(&self, py: Python<'_>) -> PyResult<()> {
         let result = self.invoke_with(py, |ready| ready.invoke_direct_raw(py))?;
         drop(result);
         Ok(())
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ReadyCallback"))]
     fn invoke_with<R>(&self, py: Python<'_>, call: impl Fn(&Self) -> PyResult<R>) -> PyResult<R> {
         if !self.context_needs_run {
             return call(self);
@@ -291,6 +297,7 @@ impl<const SOURCE_WORDS: usize> ReadyCallback<SOURCE_WORDS> {
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ReadyCallback"))]
     fn invoke_direct_raw<'py>(&self, py: Python<'py>) -> PyResult<CallbackResult<'py>> {
         // SAFETY: the callback and stored arguments remain owned by self. The
         // inline arrays and tuple are valid for the synchronous call, which

src/engine/loop_core.rs

@@ -104,6 +104,7 @@ struct WaitForWake<'a> {
     wake: &'a LoopWake,
 }
 impl<'a> WaitForWake<'a> {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "WaitForWake"))]
     fn new(wake: &'a LoopWake) -> Self {
         Self { wake }
     }
@@ -338,6 +339,10 @@ pub struct LoopState {
 }
 
 impl crate::state_snapshot::SnapshotFlags for LoopState {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<LoopState as crate :: state_snapshot :: SnapshotFlags>")
+    )]
     fn snapshot_flags(&self) -> u8 {
         u8::from(self.running) | (u8::from(self.stopping) << 1) | (u8::from(self.closed) << 2)
     }
@@ -491,6 +496,7 @@ impl LoopCore {
         self.send_remote_command(LoopCommand::ScheduleReadyHandle(handle))
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))]
     #[inline]
     pub(crate) fn current_turn(&self) -> u64 {
         self.turn_number.load(Ordering::Relaxed)

src/state_snapshot.rs

@@ -24,25 +24,29 @@ pub struct SnapshotGuard<'a, T: SnapshotFlags> {
 }
 
 impl<T: SnapshotFlags> SnapshotMutex<T> {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     pub fn new(value: T) -> Self {
         Self {
             flags: AtomicU8::new(value.snapshot_flags()),
             inner: Mutex::new(value),
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     #[inline]
     pub fn flags(&self) -> u8 {
         self.flags.load(Ordering::Acquire)
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     pub fn lock(&self) -> LockResult<SnapshotGuard<'_, T>> {
         self.inner
             .lock()
             .map(|guard| self.wrap(guard))
             .map_err(|error| PoisonError::new(self.wrap(error.into_inner())))
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     fn wrap<'a>(&'a self, guard: MutexGuard<'a, T>) -> SnapshotGuard<'a, T> {
         SnapshotGuard {
             owner: self,
@@ -51,6 +55,7 @@ impl<T: SnapshotFlags> SnapshotMutex<T> {
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     pub fn wait_timeout<'a>(
         &'a self,
         cv: &Condvar,
@@ -74,6 +79,7 @@ impl<T: SnapshotFlags> SnapshotMutex<T> {
 }
 
 impl<T: SnapshotFlags> SnapshotGuard<'_, T> {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotGuard"))]
     fn publish(&mut self) {
         if self.dirty {
             let flags = self
@@ -91,17 +97,29 @@ impl<T: SnapshotFlags> SnapshotGuard<'_, T> {
 
 impl<T: SnapshotFlags> Deref for SnapshotGuard<'_, T> {
     type Target = T;
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<SnapshotGuard as Deref>")
+    )]
     fn deref(&self) -> &T {
         self.guard.as_ref().expect("live snapshot guard")
     }
 }
 impl<T: SnapshotFlags> DerefMut for SnapshotGuard<'_, T> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<SnapshotGuard as DerefMut>")
+    )]
     fn deref_mut(&mut self) -> &mut T {
         self.dirty = true;
         self.guard.as_mut().expect("live snapshot guard")
     }
 }
 impl<T: SnapshotFlags> Drop for SnapshotGuard<'_, T> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<SnapshotGuard as Drop>")
+    )]
     fn drop(&mut self) {
         self.publish();
     }
@@ -117,6 +135,57 @@ mod tests {
         }
     }
 
+    #[test]
+    fn timed_wait_publishes_flags_and_returns_a_live_guard() {
+        let state = SnapshotMutex::new(State(0));
+        let cv = Condvar::new();
+        let mut guard = state.lock().unwrap();
+        guard.0 = 3;
+
+        let (mut guard, timeout) = state.wait_timeout(&cv, guard, Duration::ZERO).unwrap();
+        assert!(timeout.timed_out());
+        assert_eq!(state.flags(), 3);
+        assert_eq!(guard.0, 3);
+
+        guard.0 = 5;
+        drop(guard);
+        assert_eq!(state.flags(), 5);
+        assert_eq!(state.lock().unwrap().0, 5);
+    }
+
+    #[test]
+    fn notified_wait_observes_other_thread_mutation_and_republishes() {
+        let state = SnapshotMutex::new(State(0));
+        let cv = Condvar::new();
+        let mut guard = state.lock().unwrap();
+        guard.0 = 3;
+
+        std::thread::scope(|scope| {
+            let notifier = scope.spawn(|| {
+                let mut other = state.lock().unwrap();
+                assert_eq!(state.flags(), 3);
+                assert_eq!(other.0, 3);
+                other.0 = 5;
+                drop(other);
+                cv.notify_one();
+            });
+            // Condvar waits may wake spuriously. Check the protected predicate.
+            while guard.0 != 5 {
+                let (next, timeout) = state
+                    .wait_timeout(&cv, guard, Duration::from_secs(5))
+                    .unwrap();
+                guard = next;
+                assert!(guard.0 == 5 || !timeout.timed_out());
+            }
+            notifier.join().unwrap();
+            assert_eq!(state.flags(), 5);
+            guard.0 = 7;
+            drop(guard);
+            assert_eq!(state.flags(), 7);
+            assert_eq!(state.lock().unwrap().0, 7);
+        });
+    }
+
     #[test]
     fn mutation_and_unwind_publish_flags_before_unlocking() {
         let state = SnapshotMutex::new(State(0));

src/transport/stream/buffers.rs

@@ -42,6 +42,7 @@ pub(super) struct OwnedReadBuffer {
 }
 
 impl OwnedReadBuffer {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "OwnedReadBuffer"))]
     pub(super) fn is_pooled(&self) -> bool {
         self.pool.is_some()
     }
@@ -235,6 +236,7 @@ impl OwnedWriteBuffer {
         self.bytes.extend_from_slice(data);
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "OwnedWriteBuffer"))]
     /// Materialize Python-backed batches in the transport pool so their
     /// allocation is recycled when the staged write finishes.
     pub(super) fn extend_from_slice_pooled(&mut self, data: &[u8], pool: &Arc<WriteBufferPool>) {

src/transport/stream/connect.rs

@@ -29,6 +29,7 @@ struct ConnectDone {
 }
 #[pymethods]
 impl ConnectDone {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ConnectDone"))]
     fn __call__(&self, _future: &Bound<'_, PyAny>) {
         self.wake.done.store(true, Ordering::Release);
         self.wake.waker.wake();
@@ -39,13 +40,15 @@ impl ConnectDone {
         }
         Ok(())
     }
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ConnectDone"))]
     fn __clear__(&mut self) {
         self.socket = None;
         self.wake.done.store(true, Ordering::Release);
         self.wake.waker.wake();
     }
 }
 
+#[cfg_attr(feature = "profile", hotpath::measure)]
 /// Keep the original socket alive and wake the native wait on cancellation.
 pub(crate) fn connect_wait_state(
     py: Python<'_>,

src/transport/stream/core_events.rs

@@ -166,8 +166,12 @@ impl StreamTransportCore {
                 ));
             }
             let (guard, _) = self
-                .state_cv
-                .wait_timeout(state, deadline.saturating_duration_since(now))
+                .state
+                .wait_timeout(
+                    &self.state_cv,
+                    state,
+                    deadline.saturating_duration_since(now),
+                )
                 .expect("poisoned transport state");
             state = guard;
         }

src/transport/stream/core_write.rs

@@ -409,6 +409,10 @@ impl StreamTransportCore {
         self.try_write_bytes_with_status(bytes, self.write_backpressure_active())
     }
 
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "StreamTransportCore")
+    )]
     /// Learn whether this connection emits batches. The first isolated write
     /// stays direct; a repeated write stages the tail and predicts batching on
     /// the next turn, so recurring header/body pairs still coalesce together.
@@ -425,6 +429,10 @@ impl StreamTransportCore {
         }
     }
 
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "StreamTransportCore")
+    )]
     pub(super) fn try_write_bytes_with_status(
         self: &Arc<Self>,
         bytes: &Bound<'_, PyBytes>,

src/transport/stream/fast.rs

@@ -1209,6 +1209,10 @@ impl ExactReadAccumulator {
         buffer.consume(amount);
     }
 
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "ExactReadAccumulator")
+    )]
     fn fill_from_slice(&mut self, data: &[u8]) -> usize {
         let amount = exact_fill_amount(data.len(), self.filled, self.expected);
         if amount == 0 {
@@ -1721,6 +1725,10 @@ impl PyFastStreamReader {
         self.maybe_resume_transport(py)
     }
 
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "PyFastStreamReader")
+    )]
     /// Complete a waiting exact read directly from incoming bytes. Reconcile
     /// cancellation first, restoring its partial data before any new
     /// consumption.

src/transport/stream/mod.rs

@@ -150,6 +150,12 @@ struct StreamTransportState {
 }
 
 impl crate::state_snapshot::SnapshotFlags for StreamTransportState {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(
+            impl_type = "<StreamTransportState as crate :: state_snapshot :: SnapshotFlags>"
+        )
+    )]
     fn snapshot_flags(&self) -> u8 {
         u8::from(self.closing)
             | (u8::from(self.writable) << 1)

src/transport/stream/py_transport.rs

@@ -28,23 +28,39 @@ struct InlineWriteSegments<'py> {
     len: usize,
 }
 impl<'py> InlineWriteSegments<'py> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "InlineWriteSegments")
+    )]
     fn new() -> Self {
         Self {
             items: std::array::from_fn(|_| std::mem::MaybeUninit::uninit()),
             len: 0,
         }
     }
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "InlineWriteSegments")
+    )]
     fn push(&mut self, value: Bound<'py, PyBytes>) {
         self.items[self.len].write(value);
         self.len += 1;
     }
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "InlineWriteSegments")
+    )]
     fn as_slice(&self) -> &[Bound<'py, PyBytes>] {
         // SAFETY: push initializes each prefix element before increasing len;
         // MaybeUninit<T> has T's layout, and this immutable borrow retains it.
         unsafe { std::slice::from_raw_parts(self.items.as_ptr().cast(), self.len) }
     }
 }
 impl Drop for InlineWriteSegments<'_> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<InlineWriteSegments as Drop>")
+    )]
     fn drop(&mut self) {
         for item in &mut self.items[..self.len] {
             // SAFETY: exactly the initialized prefix contains owned values.