//! Completion handles for I/O registered on one worker.
//!
//! Adapters register requests inside their async bodies, then await an
//! [`Operation`] containing its waiter ID and weak worker identity. The driver
//! retains both pending requests and completed results in one slab. A poll on
//! another thread forwards observation through a completion channel while the
//! request stays on its original worker.
//!
//! An operation may be polled or dropped on any thread after registration.
//! Dropping it releases its observer. The driver decides whether to cancel the
//! request or finish retained work such as writes and syncs.

use super::{
    mailbox::{Cancel, Forward},
    registration::{Key, Observation, Registration},
    request::{Request, RequestOutput, SyncRequest},
    runtime::Local,
    waiter::{Observer, WaiterId},
};
use crate::Error;
use commonware_utils::channel::oneshot;
use std::{sync::Arc, task::Waker};

/// Handle to an I/O request registered on a worker.
pub(crate) type Operation = Registration<WaiterId>;

impl Operation {
    /// Register on the current worker and return a handle to its result.
    ///
    /// Call this inside the adapter's async body so creating an I/O future
    /// remains lazy. The request is transferred before this handle is awaited.
    /// A closing worker returns a handle that resolves to [`Error::Closed`].
    pub fn register(request: Request) -> Self {
        let owner = Local::current().expect("io_uring I/O requires a current worker");
        let mut local = owner.borrow_mut();
        if local.closing {
            // Request owners may run destructors that reenter the worker.
            drop(local);
            drop(request);
            return Self::closed();
        }
        let mailbox = Arc::downgrade(&local.mailbox);

        // The driver takes ownership even when the SQ is full. Polling installs
        // a waker only while the request is still pending.
        let Local {
            driver,
            deferred,
            now,
            ..
        } = &mut *local;
        let id = driver
            .as_mut()
            .unwrap()
            .admit(request, Observer::Local(None), *now, deferred);
        Self::new(mailbox, id)
    }
}

impl Key for WaiterId {
    type Output = RequestOutput;

    fn observe(self, local: &mut Local, waker: &Waker) -> Observation<RequestOutput> {
        local.driver.as_mut().unwrap().observe(self, waker)
    }

    fn refresh(self, local: &mut Local, waker: Waker) -> Option<Waker> {
        local.driver.as_mut().unwrap().set_waker(self, waker)
    }

    fn forward(self, sender: oneshot::Sender<Result<RequestOutput, Error>>) -> Forward {
        Forward::Waiter(self, sender)
    }

    fn cancel(self) -> Cancel {
        Cancel::Waiter(self)
    }
}

/// Transfer a sync to its worker and return its independent completion receiver.
///
/// Dropping the receiver leaves the sync running. A closing worker rejects the
/// request and publishes [`Error::Closed`] through the receiver.
pub fn start_sync(request: SyncRequest) -> oneshot::Receiver<Result<(), Error>> {
    let owner = Local::current().expect("io_uring sync requires a current worker");
    let (sender, receiver) = oneshot::channel();
    let mut local = owner.borrow_mut();
    if local.closing {
        // Release the request and publish closure outside the local borrow.
        drop(local);
        drop(request);
        let _ = sender.send(Err(Error::Closed));
    } else {
        // Registration needs no task waker. The worker publishes to this
        // channel when the sync finishes.
        let Local {
            driver,
            deferred,
            now,
            ..
        } = &mut *local;
        driver.as_mut().unwrap().admit(
            Request::Sync(request),
            Observer::DetachedSync(sender),
            *now,
            deferred,
        );
    }
    receiver
}

#[cfg(test)]
pub mod tests {
    use super::*;
    use crate::{
        Blob as _, Clock as _, IoBufMut, IoBufs, ReadOptions, Runner as _, Storage as _,
        WriteOptions,
        iouring::{
            Config, RingConfig, Runner,
            request::{RecvRequest, SendRequest},
            sleep::Sleep,
        },
        storage::{hold::Hold, iouring::Shared},
        utils::{extract_panic_message, reschedule},
    };
    use commonware_utils::sync::Mutex;
    use futures::{FutureExt as _, future::pending, poll};
    use std::{
        cell::RefCell,
        fs::{self, OpenOptions},
        future::Future,
        io::Write as _,
        mem,
        os::{fd::OwnedFd, unix::net::UnixStream},
        panic::{AssertUnwindSafe, catch_unwind},
        pin::pin,
        ptr,
        sync::{
            atomic::{AtomicUsize, Ordering},
            mpsc,
        },
        task::{Context, Poll, RawWaker, RawWakerVTable},
        thread,
        time::{Duration, Instant},
    };

    /// Return the number of unfinished logical requests admitted to the current worker.
    pub fn admitted() -> usize {
        let owner = Local::current().expect("io_uring I/O requires a current worker");
        owner.borrow().driver.as_ref().unwrap().len()
    }

    /// Arbitrary waker callbacks that reenter the current worker's local state.
    ///
    /// RawWaker exposes clone and drop as well as wake, so every callback can
    /// check for an outstanding worker borrow or inject a panic.
    #[derive(Default)]
    pub struct Reentrant {
        /// Number of cloned wakers, including clones that panic.
        clones: AtomicUsize,
        /// Number of consuming and borrowed wakes.
        pub wakes: AtomicUsize,
        /// Number of waker destructors invoked.
        pub drops: AtomicUsize,
        /// Callback that should panic once, or zero when no panic is armed.
        pub panic_callback: AtomicUsize,
        /// Optional action run while cloning, outside any worker borrow.
        on_clone: Option<fn()>,
    }

    /// The one vtable of every waker [`Reentrant`] makes.
    ///
    /// A static has a single address, so `will_wake` finds every clone of one
    /// waker equal.
    static REENTRANT_VTABLE: RawWakerVTable = RawWakerVTable::new(
        Reentrant::clone,
        Reentrant::wake,
        Reentrant::wake_by_ref,
        Reentrant::drop,
    );

    impl Reentrant {
        /// Value of `panic_callback` that panics the next clone.
        pub const CLONE: usize = 1;

        /// Value of `panic_callback` that panics the next wake, consuming or
        /// borrowed.
        const WAKE: usize = 2;

        /// Value of `panic_callback` that panics the next destructor.
        const DROP: usize = 3;

        /// Panic if `callback` is the one armed, disarming it first.
        fn panic_once(&self, callback: usize) {
            // Disarm before unwinding so cleanup can use the remaining wakers.
            if self
                .panic_callback
                .compare_exchange(callback, 0, Ordering::Relaxed, Ordering::Relaxed)
                .is_ok()
            {
                panic!("waker callback panic {callback}");
            }
        }

        /// Fail if the current worker's local state is borrowed.
        fn check_local() {
            if let Some(local) = Local::current() {
                // A callback under an outstanding Local borrow fails here.
                let _borrow = local.borrow_mut();
            }
        }

        /// The vtable's clone: checks for a borrow, counts the clone, panics if
        /// armed to, then runs `on_clone`.
        unsafe fn clone(data: *const ()) -> RawWaker {
            // SAFETY: Each raw waker owns one Arc reference to Reentrant.
            // ManuallyDrop borrows that reference without consuming it.
            let owner = mem::ManuallyDrop::new(unsafe { Arc::from_raw(data.cast::<Self>()) });
            Self::check_local();
            owner.clones.fetch_add(1, Ordering::Relaxed);
            owner.panic_once(Self::CLONE);
            if let Some(on_clone) = owner.on_clone {
                on_clone();
            }
            RawWaker::new(Arc::into_raw(Arc::clone(&owner)).cast(), &REENTRANT_VTABLE)
        }

        /// The vtable's consuming wake: checks for a borrow, counts the wake,
        /// and panics if armed to.
        unsafe fn wake(data: *const ()) {
            // SAFETY: Consuming wake takes exactly the Arc reference owned by
            // this raw waker. No reference is reconstructed a second time.
            let owner = unsafe { Arc::from_raw(data.cast::<Self>()) };
            Self::check_local();
            owner.wakes.fetch_add(1, Ordering::Relaxed);
            owner.panic_once(Self::WAKE);
        }

        /// The vtable's borrowed wake, as [`Self::wake`] without consuming the
        /// waker's reference.
        unsafe fn wake_by_ref(data: *const ()) {
            // SAFETY: The waker retains its Arc reference after this callback,
            // so the temporary reconstructed Arc must not decrement it.
            let owner = mem::ManuallyDrop::new(unsafe { Arc::from_raw(data.cast::<Self>()) });
            Self::check_local();
            owner.wakes.fetch_add(1, Ordering::Relaxed);
            owner.panic_once(Self::WAKE);
        }

        /// The vtable's destructor: checks for a borrow, counts the drop, and
        /// panics if armed to.
        unsafe fn drop(data: *const ()) {
            // SAFETY: Raw-waker destruction consumes its one owned Arc reference.
            let owner = unsafe { Arc::from_raw(data.cast::<Self>()) };
            Self::check_local();
            owner.drops.fetch_add(1, Ordering::Relaxed);
            owner.panic_once(Self::DROP);
        }

        /// A waker that holds one reference to this fixture.
        pub fn waker(self: &Arc<Self>) -> Waker {
            let raw = RawWaker::new(Arc::into_raw(self.clone()).cast(), &REENTRANT_VTABLE);
            // SAFETY: The vtable consistently owns or borrows one Arc reference
            // and Reentrant contains only thread-safe atomic state.
            unsafe { Waker::from_raw(raw) }
        }
    }

    /// Limit the driver to one in-flight request.
    fn runner() -> Runner {
        Runner::new(Config::default().with_ring_config(RingConfig {
            size: 1,
            ..Default::default()
        }))
    }

    /// Connected sockets whose local endpoint can be shared by requests.
    ///
    /// Keeping the peer open and silent blocks receives while allowing sends.
    fn socket() -> (Arc<OwnedFd>, UnixStream) {
        let (local, peer) = UnixStream::pair().unwrap();
        local.set_nonblocking(true).unwrap();
        (Arc::new(local.into()), peer)
    }

    /// Poll elsewhere while the owning worker cannot service the handoff.
    pub fn poll_on_foreign_thread<F: Future + Send + Unpin + 'static>(
        mut future: F,
        waker: Waker,
    ) -> F {
        thread::spawn(move || {
            assert!(
                future
                    .poll_unpin(&mut Context::from_waker(&waker))
                    .is_pending()
            );
            future
        })
        .join()
        .unwrap()
    }

    /// Register a one-byte send on the current worker.
    fn send(fd: Arc<OwnedFd>) -> Operation {
        Operation::register(Request::Send(SendRequest {
            fd,
            write: IoBufs::from(vec![1]).into(),
            deadline: None,
        }))
    }

    /// Register a one-byte receive, optionally bounded by a deadline.
    fn recv(fd: Arc<OwnedFd>, deadline: Option<Instant>) -> Operation {
        Operation::register(Request::Recv(RecvRequest {
            fd,
            buf: IoBufMut::from([0]),
            offset: 0,
            len: 1,
            exact: true,
            deadline,
        }))
    }

    /// Service the worker until every logical request retires.
    async fn drained() {
        let deadline = Instant::now() + Duration::from_secs(10);
        while !Local::current()
            .unwrap()
            .borrow()
            .driver
            .as_ref()
            .unwrap()
            .is_empty()
        {
            assert!(Instant::now() < deadline, "I/O did not retire");
            reschedule().await;
        }
    }

    #[test]
    fn test_start_sync_returns_while_ring_is_full() {
        let callbacks = Arc::new(Reentrant::default());

        runner().start(|context| async move {
            let (blob, _) = context.open("observer_sync", b"file").await.unwrap();
            blob.write_at(0, b"dirty", WriteOptions::default())
                .await
                .unwrap();
            let (fd, _peer) = socket();
            let mut blocker = recv(fd, None);
            assert!(poll!(&mut blocker).is_pending());
            reschedule().await;

            // The only in-flight slot is occupied, but sync accepts ownership now.
            let mut sync = pin!(blob.start_sync());
            let waker = callbacks.waker();
            let Poll::Ready(handle) = sync.poll_unpin(&mut Context::from_waker(&waker)) else {
                panic!("sync must return its completion handle on the first poll");
            };
            assert_eq!(callbacks.clones.load(Ordering::Relaxed), 0);
            assert_eq!(callbacks.wakes.load(Ordering::Relaxed), 0);

            drop(blocker);
            handle.await.unwrap();
        });
    }

    #[test]
    fn test_expired_registration_does_not_consume_ready_data() {
        let callbacks = Arc::new(Reentrant::default());
        runner().start(|context| async move {
            let deadline = Instant::now();
            context.sleep(Duration::from_millis(1)).await;
            assert!(Local::current().unwrap().borrow().now >= deadline);

            let (fd, mut peer) = socket();
            peer.write_all(b"x").unwrap();
            let mut operation = recv(fd.clone(), Some(deadline));
            let waker = callbacks.waker();
            assert!(matches!(
                operation.poll_unpin(&mut Context::from_waker(&waker)),
                Poll::Ready(Ok(RequestOutput::Recv(Err((_, Error::Timeout)))))
            ));
            assert_eq!(callbacks.clones.load(Ordering::Relaxed), 0);
            assert!(
                Local::current()
                    .unwrap()
                    .borrow()
                    .driver
                    .as_ref()
                    .unwrap()
                    .is_empty()
            );

            let RequestOutput::Recv(Ok((buffer, len))) =
                recv(fd, Some(Instant::now() + Duration::from_secs(1)))
                    .await
                    .unwrap()
            else {
                panic!("expired receive consumed ready data");
            };
            assert_eq!(len, 1);
            assert_eq!(buffer.as_ref(), b"x");
        });
    }

    #[test]
    fn test_worker_closure_rejects_registration_and_polling() {
        let callbacks = Arc::new(Reentrant::default());
        runner().start(|context| async move {
            let (blob, _) = context.open("observer_closed", b"file").await.unwrap();

            // Dirty the open so start_sync submits a request to the closed worker.
            blob.write_at(0, b"x", WriteOptions::default())
                .await
                .unwrap();
            let (fd, _peer) = socket();
            let mut operation = recv(fd.clone(), None);
            let waker = callbacks.waker();
            let mut cx = Context::from_waker(&waker);

            Local::current().unwrap().borrow_mut().closing = true;
            assert!(matches!(
                operation.poll_unpin(&mut cx),
                Poll::Ready(Err(Error::Closed))
            ));
            assert_eq!(callbacks.clones.load(Ordering::Relaxed), 0);

            // New operations and detached syncs must also reject closure.
            let mut rejected = recv(fd, None);
            assert!(matches!(
                rejected.poll_unpin(&mut cx),
                Poll::Ready(Err(Error::Closed))
            ));
            let handle = blob.start_sync().await;
            assert!(matches!(handle.await, Err(Error::Closed)));
            assert!(
                Local::current()
                    .unwrap()
                    .borrow()
                    .driver
                    .as_ref()
                    .unwrap()
                    .is_empty()
            );
        });
    }

    #[test]
    fn test_observer_clone_can_register_and_cancel_other_work() {
        for in_flight in [false, true] {
            let callbacks = Arc::new(Reentrant {
                on_clone: Some(|| {
                    let mut sleep = Sleep::new(Duration::from_secs(60));
                    let (fd, _peer) = socket();
                    let mut operation = recv(fd, None);
                    let mut cx = Context::from_waker(Waker::noop());
                    assert!(sleep.poll_unpin(&mut cx).is_pending());
                    assert!(operation.poll_unpin(&mut cx).is_pending());
                    drop(operation);
                    drop(sleep);
                }),
                ..Default::default()
            });

            runner().start(|_| async {
                let (fd, mut peer) = socket();
                let mut operation = recv(fd, Some(Instant::now() + Duration::from_secs(60)));
                let mut sleep = Sleep::new(Duration::from_secs(60));
                assert!(poll!(&mut operation).is_pending());
                assert!(poll!(&mut sleep).is_pending());
                if in_flight {
                    reschedule().await;
                }

                // Each replacement may register and cancel other identities.
                // The current registrations still need their own completion.
                let waker = callbacks.waker();
                let mut cx = Context::from_waker(&waker);
                assert!(operation.poll_unpin(&mut cx).is_pending());
                assert!(sleep.poll_unpin(&mut cx).is_pending());
                assert_eq!(callbacks.clones.load(Ordering::Relaxed), 2);

                drop(sleep);
                peer.write_all(b"x").unwrap();
                let RequestOutput::Recv(Ok((buffer, len))) = operation.await.unwrap() else {
                    panic!("receive failed after observer replacement");
                };
                assert_eq!(len, 1);
                assert_eq!(buffer.as_ref(), b"x");
            });
        }
    }

    #[test]
    fn test_pending_observers_retain_equivalent_wakers() {
        let callbacks = Arc::new(Reentrant::default());

        runner().start(|_| async {
            let (fd, _peer) = socket();
            let mut first = recv(fd.clone(), None);
            let mut queued = recv(fd, None);
            let mut sleep = Sleep::new(Duration::from_secs(60));
            let waker = callbacks.waker();
            let mut cx = Context::from_waker(&waker);
            assert!(first.poll_unpin(&mut cx).is_pending());
            reschedule().await;
            assert!(queued.poll_unpin(&mut cx).is_pending());
            assert!(sleep.poll_unpin(&mut cx).is_pending());

            // Equivalent wakers need no replacement for an in-flight request,
            // a queued request, or a registered sleep.
            let clones = callbacks.clones.load(Ordering::Relaxed);
            assert!(first.poll_unpin(&mut cx).is_pending());
            assert!(queued.poll_unpin(&mut cx).is_pending());
            assert!(sleep.poll_unpin(&mut cx).is_pending());
            assert_eq!(callbacks.clones.load(Ordering::Relaxed), clones);
        });
    }

    #[test]
    fn test_observer_clone_wake_and_drop_run_outside_local_borrows() {
        let callbacks = Arc::new(Reentrant::default());

        runner().start(|_| async {
            let (fd, _peer) = socket();
            let mut operation = send(fd);
            let waker = callbacks.waker();
            let mut cx = Context::from_waker(&waker);
            assert!(operation.poll_unpin(&mut cx).is_pending());

            // Replace the observer with a noop waker, then restore it. The
            // old observer must be dropped after the worker borrow ends.
            assert!(
                operation
                    .poll_unpin(&mut Context::from_waker(Waker::noop()))
                    .is_pending()
            );
            assert!(operation.poll_unpin(&mut cx).is_pending());

            while callbacks.wakes.load(Ordering::Relaxed) == 0 {
                reschedule().await;
            }

            // A different waker would need cloning if the request were pending.
            // Consuming the completed result must skip that callback entirely.
            let ready = Arc::new(Reentrant::default());
            let ready_waker = ready.waker();
            assert!(matches!(
                operation.poll_unpin(&mut Context::from_waker(&ready_waker)),
                Poll::Ready(Ok(RequestOutput::Send(Ok(()))))
            ));
            assert_eq!(ready.clones.load(Ordering::Relaxed), 0);
            drop(waker);
        });

        // Two installations, one completion wake, and destruction of the
        // displaced clone and the original waker.
        assert_eq!(callbacks.clones.load(Ordering::Relaxed), 2);
        assert_eq!(callbacks.wakes.load(Ordering::Relaxed), 1);
        assert_eq!(callbacks.drops.load(Ordering::Relaxed), 2);
        assert_eq!(Arc::strong_count(&callbacks), 1);
    }

    #[test]
    fn test_observer_clone_panic_preserves_queued_and_in_flight_cancellation() {
        for queued in [false, true] {
            let callbacks = Arc::new(Reentrant::default());
            let (fd, _peer) = socket();

            runner().start(|_| async {
                let mut blocker = queued.then(|| recv(fd.clone(), None));
                if let Some(blocker) = &mut blocker {
                    assert!(poll!(blocker).is_pending());
                }

                let mut operation = recv(fd.clone(), None);
                let waker = callbacks.waker();
                let mut cx = Context::from_waker(&waker);
                assert!(poll!(&mut operation).is_pending());
                reschedule().await;

                let registration = operation.key();
                callbacks
                    .panic_callback
                    .store(Reentrant::CLONE, Ordering::Relaxed);
                let panic = catch_unwind(AssertUnwindSafe(|| {
                    let _ = operation.poll_unpin(&mut cx);
                }))
                .expect_err("observer clone must panic");
                assert_eq!(extract_panic_message(&*panic), "waker callback panic 1");
                assert!(
                    operation.key() == registration,
                    "clone panic lost cancellation identity"
                );

                // A clone panic must retain the identity needed to cancel
                // both queued and in-flight requests.
                drop(operation);
                drop(blocker);
                assert!(matches!(
                    send(fd.clone()).await,
                    Ok(RequestOutput::Send(Ok(())))
                ));
            });
            assert_eq!(Arc::strong_count(&fd), 1);
            assert_eq!(Arc::strong_count(&callbacks), 1);
        }
    }

    #[test]
    fn test_observer_wake_and_drop_panics_finish_worker_retirement() {
        // These callbacks run in worker service. Their panics must leave
        // through Runner::start only after all worker cleanup has finished.
        for callback in [Reentrant::WAKE, Reentrant::DROP] {
            let callbacks = Arc::new(Reentrant::default());
            let (fd, _peer) = socket();

            // Keep Local alive to inspect cleanup after its scope is gone.
            let mut retained_local = None;
            let panic = catch_unwind(AssertUnwindSafe(|| {
                runner().start(|_| async {
                    retained_local = Local::current();
                    let mut operation = if callback == Reentrant::WAKE {
                        send(fd.clone())
                    } else {
                        recv(fd.clone(), None)
                    };
                    let waker = callbacks.waker();
                    let mut cx = Context::from_waker(&waker);
                    assert!(operation.poll_unpin(&mut cx).is_pending());
                    callbacks.panic_callback.store(callback, Ordering::Relaxed);
                    if callback == Reentrant::DROP {
                        // Replacing an observer queues its destructor. The
                        // receive stays pending until failure cleanup cancels it.
                        assert!(
                            operation
                                .poll_unpin(&mut Context::from_waker(Waker::noop()))
                                .is_pending()
                        );
                    }
                    pending::<()>().await;
                });
            }))
            .expect_err("deferred observer callback must fail the runner");
            assert_eq!(
                extract_panic_message(&*panic),
                format!("waker callback panic {callback}")
            );
            let local = retained_local.unwrap();
            let local = local.borrow();
            assert!(local.driver.is_none());
            assert!(Local::current().is_none());
            assert_eq!(Arc::strong_count(&fd), 1);
            assert_eq!(Arc::strong_count(&callbacks), 1);
        }
    }

    #[test]
    fn test_queued_operation_completes_without_further_polls() {
        let callbacks = Arc::new(Reentrant::default());

        runner().start(|_| async {
            let (fd, _peer) = socket();
            let mut blocker = recv(fd.clone(), None);
            assert!(poll!(&mut blocker).is_pending());
            reschedule().await;

            let mut queued = send(fd.clone());
            let waker = callbacks.waker();
            assert!(
                queued
                    .poll_unpin(&mut Context::from_waker(&waker))
                    .is_pending()
            );
            drop(blocker);

            // Let a later request drive the worker. The queued future receives
            // no polls, and its unconsumed result must not occupy the sole
            // in-flight slot needed by this later request.
            assert!(matches!(send(fd).await, Ok(RequestOutput::Send(Ok(())))));
            assert_eq!(callbacks.clones.load(Ordering::Relaxed), 1);
            assert_eq!(callbacks.wakes.load(Ordering::Relaxed), 1);
            assert!(matches!(
                queued.poll_unpin(&mut Context::from_waker(&waker)),
                Poll::Ready(Ok(RequestOutput::Send(Ok(()))))
            ));
            assert_eq!(callbacks.clones.load(Ordering::Relaxed), 1);
        });
    }

    #[test]
    fn test_queued_deadline_and_timer_progress_with_full_ring() {
        runner().start(|context| async move {
            let (fd, _peer) = socket();
            let mut first = recv(fd.clone(), None);
            assert!(poll!(&mut first).is_pending());
            let deadline = Instant::now() + Duration::from_millis(10);
            let waiting = recv(fd, Some(deadline));
            assert!(matches!(
                waiting.await,
                Ok(RequestOutput::Recv(Err((_, Error::Timeout))))
            ));

            // The untimed receive still occupies the only in-flight slot.
            // Sleeper expiry must also progress independently of that slot.
            context.sleep(Duration::from_millis(1)).await;
            drop(first);
        });
    }

    #[test]
    fn test_escaped_operation_observes_worker_closure() {
        // Return the handle without awaiting it, then poll it after cleanup.
        let (operation,) = runner().start(|_| async {
            let (fd, _peer) = socket();
            let mut operation = recv(fd, None);
            assert!(poll!(&mut operation).is_pending());
            (operation,)
        });
        assert!(matches!(
            futures::executor::block_on(operation),
            Err(Error::Closed)
        ));
    }

    #[test]
    fn test_foreign_gate_waiter_observes_origin_closure() {
        let directory =
            std::env::temp_dir().join(format!("commonware_operation_gate_{}", std::process::id()));
        let hold = Hold::acquire(&directory).unwrap();
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(true)
            .open(directory.join("blob"))
            .unwrap();
        let shared = Shared::detached(file, hold);

        let (operation, release) = runner().start(|_| async {
            let gate = shared.durability.clone();
            let mailbox = Local::current().unwrap().borrow().mailbox.clone();
            let (acquired, ready) = mpsc::channel();
            let release = thread::spawn(move || {
                let permit = futures::executor::block_on(gate.lock());
                acquired.send(()).unwrap();
                let deadline = Instant::now() + Duration::from_secs(10);
                while mailbox.is_open() {
                    assert!(Instant::now() < deadline, "origin mailbox did not close");
                    thread::yield_now();
                }
                drop(permit);
            });
            ready.recv_timeout(Duration::from_secs(10)).unwrap();

            let mut operation = Operation::register(Request::Sync(SyncRequest::new(shared)));
            assert!(poll!(&mut operation).is_pending());
            let operation = poll_on_foreign_thread(operation, Waker::noop().clone());
            (operation, release)
        });

        release.join().unwrap();
        assert!(matches!(
            futures::executor::block_on(operation),
            Err(Error::Closed)
        ));
        fs::remove_dir_all(directory).unwrap();
    }

    #[test]
    fn test_operation_can_move_between_threads() {
        for (in_flight, completed) in [(false, false), (true, false), (true, true)] {
            for return_to_owner in [false, true] {
                runner().start(|_| async move {
                    let (fd, mut peer) = socket();
                    let mut operation = recv(fd, None);
                    assert!(poll!(&mut operation).is_pending());
                    if in_flight {
                        reschedule().await;
                    }
                    if completed {
                        peer.write_all(b"x").unwrap();
                        drained().await;
                    }

                    let operation = poll_on_foreign_thread(operation, Waker::noop().clone());

                    if !completed {
                        peer.write_all(b"x").unwrap();
                    }
                    let output = if return_to_owner {
                        operation.await.unwrap()
                    } else {
                        let (sender, receiver) = oneshot::channel();
                        let thread = thread::spawn(move || {
                            let output = futures::executor::block_on(operation);
                            assert!(sender.send(output).is_ok());
                        });
                        let output = receiver.await.unwrap().unwrap();
                        thread.join().unwrap();
                        output
                    };
                    let RequestOutput::Recv(Ok((buffer, len))) = output else {
                        panic!("receive failed after moving between threads");
                    };
                    assert_eq!(len, 1);
                    assert_eq!(buffer.as_ref(), b"x");
                });
            }
        }
    }

    #[test]
    fn test_local_drop_overtakes_forwarding() {
        for in_flight in [false, true] {
            runner().start(|_| async move {
                let (fd, _peer) = socket();
                let mut operation = recv(fd.clone(), None);
                assert!(poll!(&mut operation).is_pending());
                if in_flight {
                    reschedule().await;
                }
                let old = operation.key();
                let operation = poll_on_foreign_thread(operation, Waker::noop().clone());

                // Local cancellation runs before the queued forwarding message.
                // A new request can reuse a queued operation's released slot.
                drop(operation);
                let replacement = send(fd);
                if !in_flight {
                    let current = replacement.key();
                    assert_eq!(old.0.index, current.0.index);
                    assert_ne!(old.0.generation, current.0.generation);
                }
                assert!(matches!(replacement.await, Ok(RequestOutput::Send(Ok(())))));
                drained().await;
            });
        }
    }

    #[test]
    fn test_forwarded_observer_callbacks_and_shutdown() {
        for (promoted, completed) in [(false, false), (true, false), (true, true)] {
            let callbacks = Arc::new(Reentrant::default());
            let (operation,) = runner().start(|_| async {
                let (fd, mut peer) = socket();
                let mut operation = recv(fd, None);
                let waker = callbacks.waker();
                assert!(
                    operation
                        .poll_unpin(&mut Context::from_waker(&waker))
                        .is_pending()
                );
                let operation = poll_on_foreign_thread(operation, callbacks.waker());
                if promoted {
                    reschedule().await;
                    assert!(callbacks.drops.load(Ordering::Relaxed) >= 2);
                }
                if completed {
                    peer.write_all(b"x").unwrap();
                    drained().await;
                }
                (operation,)
            });

            // Once the owner processes the handoff, completion or closure wakes the receiver.
            // Before that, the queued sender can independently notify it when dropped.
            if promoted {
                assert!(callbacks.wakes.load(Ordering::Relaxed) > 0);
            }
            let output = futures::executor::block_on(operation);
            if completed {
                let Ok(RequestOutput::Recv(Ok((buffer, len)))) = output else {
                    panic!("forwarded completion lost during shutdown");
                };
                assert_eq!(len, 1);
                assert_eq!(buffer.as_ref(), b"x");
            } else {
                assert!(matches!(output, Err(Error::Closed)));
            }
            assert_eq!(Arc::strong_count(&callbacks), 1);
        }
    }

    #[test]
    fn test_forwarded_receive_cancelled_after_promotion() {
        runner().start(|_| async {
            let (fd, _peer) = socket();
            let mut operation = recv(fd, None);
            assert!(poll!(&mut operation).is_pending());
            let operation = poll_on_foreign_thread(operation, Waker::noop().clone());
            reschedule().await;
            thread::spawn(move || drop(operation)).join().unwrap();
            drained().await;
        });
    }

    #[test]
    fn test_forwarded_write_finishes_after_drop() {
        for promoted in [false, true] {
            runner().start(|context| async move {
                let (blob, _) = context.open("forwarded_write", b"file").await.unwrap();
                let blob = Arc::new(blob);
                let (fd, _peer) = socket();
                let mut blocker = recv(fd, None);
                assert!(poll!(&mut blocker).is_pending());
                reschedule().await;

                let writer = blob.clone();
                let mut write = async move {
                    writer
                        .write_at(0, b"retained", WriteOptions::default())
                        .await
                }
                .boxed();
                assert!(poll!(&mut write).is_pending());
                let write = poll_on_foreign_thread(write, Waker::noop().clone());
                if promoted {
                    reschedule().await;
                }
                drop(write);
                drop(blocker);
                drained().await;

                let bytes = blob.read_at(0, 8, ReadOptions::default()).await.unwrap();
                assert_eq!(bytes.coalesce(), b"retained");
                context.remove("forwarded_write", None).await.unwrap();
            });
        }
    }

    #[test]
    fn test_forwarded_observer_clone_panic_keeps_cancellation() {
        runner().start(|_| async {
            let (fd, _peer) = socket();
            let mut operation = recv(fd, None);
            assert!(poll!(&mut operation).is_pending());
            let callbacks = Arc::new(Reentrant::default());
            callbacks
                .panic_callback
                .store(Reentrant::CLONE, Ordering::Relaxed);
            let operation = thread::spawn(move || {
                let waker = callbacks.waker();
                let panic = catch_unwind(AssertUnwindSafe(|| {
                    operation.poll_unpin(&mut Context::from_waker(&waker))
                }))
                .expect_err("receiver waker clone must panic");
                assert_eq!(extract_panic_message(&*panic), "waker callback panic 1");
                operation
            })
            .join()
            .unwrap();
            drop(operation);
            drained().await;
        });
    }

    #[test]
    fn test_foreign_drop_releases_registration() {
        for (in_flight, poll_first) in [(false, false), (false, true), (true, false), (true, true)]
        {
            runner().start(|_| async {
                let (fd, _peer) = socket();
                let mut operation = recv(fd, None);
                assert!(poll!(&mut operation).is_pending());

                // A queued receive retires locally. A staged receive must wait
                // for its operation CQE after cancellation.
                if in_flight {
                    reschedule().await;
                }

                thread::spawn(move || {
                    if poll_first {
                        assert!(
                            operation
                                .poll_unpin(&mut Context::from_waker(Waker::noop()))
                                .is_pending()
                        );
                    }
                    // Destruction must release the original registration whether
                    // or not its observer has moved to another thread.
                    drop(operation);
                })
                .join()
                .unwrap();

                // The owner must process the foreign message and finish any
                // kernel cancellation before the registration can retire.
                drained().await;
            });
        }
    }

    thread_local! {
        /// Work run by the next clone of an [`admitting_waker`] on this thread.
        static ON_CLONE: RefCell<Option<Box<dyn FnOnce()>>> = const { RefCell::new(None) };
    }

    /// The vtable of [`admitting_waker`].
    ///
    /// Clone runs the armed work, while wake and drop do nothing, so no
    /// callback borrows the worker.
    static ADMITTING_VTABLE: RawWakerVTable = RawWakerVTable::new(
        admitting_clone,
        admitting_noop,
        admitting_noop,
        admitting_noop,
    );

    /// Run the armed work, then return another admitting waker.
    unsafe fn admitting_clone(_: *const ()) -> RawWaker {
        if let Some(work) = ON_CLONE.with(|slot| slot.borrow_mut().take()) {
            work();
        }
        RawWaker::new(ptr::null(), &ADMITTING_VTABLE)
    }

    /// Wake and drop of an admitting waker.
    const unsafe fn admitting_noop(_: *const ()) {}

    /// A waker whose next clone runs `work` outside any worker borrow.
    fn admitting_waker(work: impl FnOnce() + 'static) -> Waker {
        ON_CLONE.with(|slot| *slot.borrow_mut() = Some(Box::new(work)));
        // SAFETY: The vtable never dereferences the null data pointer.
        unsafe { Waker::from_raw(RawWaker::new(ptr::null(), &ADMITTING_VTABLE)) }
    }

    /// A sync waits in the driver for its open's permit. The permit is released
    /// to it, and before the worker services again the sync's future is polled
    /// with a new waker. Cloning that waker admits another sync that must wait
    /// for the same permit, and that admission polls the notified permit waits.
    /// The open is clean, so acquiring the permit completes the observed sync
    /// while its poll refreshes the waker.
    #[test]
    fn test_observer_clone_admitting_a_waiting_sync_keeps_the_observed_result() {
        let directory = std::env::temp_dir().join(format!(
            "commonware_operation_refresh_{}",
            std::process::id()
        ));
        let hold = Hold::acquire(&directory).unwrap();
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(true)
            .open(directory.join("blob"))
            .unwrap();
        let shared = Shared::detached(file, hold);

        let outcome = runner().start(|_| async move {
            // Hold the open's permit so the sync waits for it in the driver.
            let permit = shared.durability.clone().try_lock_owned().unwrap();
            let mut sync = Operation::register(Request::Sync(SyncRequest::new(shared.clone())));
            assert!(poll!(&mut sync).is_pending());

            // Release the permit to the waiting sync. The worker has not
            // serviced since, so the driver still lists the sync as waiting.
            drop(permit);

            // Poll with a new waker whose clone admits a second sync of the
            // same open, which finds the permit granted and waits as well.
            let admitted = Arc::new(Mutex::new(None));
            let waker = admitting_waker({
                let admitted = admitted.clone();
                let shared = shared.clone();
                move || {
                    let second = Operation::register(Request::Sync(SyncRequest::new(shared)));
                    *admitted.lock() = Some(second);
                }
            });
            let poll = catch_unwind(AssertUnwindSafe(|| {
                sync.poll_unpin(&mut Context::from_waker(&waker))
            }));

            // Report the poll's outcome, then let both syncs retire.
            let outcome = match poll {
                Ok(Poll::Pending) => "pending".to_string(),
                Ok(Poll::Ready(Ok(RequestOutput::Sync(Ok(()))))) => "ready".to_string(),
                Ok(Poll::Ready(other)) => format!("unexpected result: {other:?}"),
                Err(panic) => format!("panic: {}", extract_panic_message(&*panic)),
            };
            drop(sync);
            drop(admitted.lock().take());
            drained().await;
            outcome
        });
        fs::remove_dir_all(directory).unwrap();

        // The poll must return the result completed during its clone.
        assert_eq!(outcome, "ready");
    }
}