hyperfile 0.10.4

//! impl request handler style IO process
use std::mem::ManuallyDrop;
use std::sync::Arc;
use bytes::Bytes;
use std::io::{Error, Result, ErrorKind};
#[cfg(feature = "wal")]
use log::{warn, info, debug};
use tokio::sync::{mpsc, oneshot, OwnedSemaphorePermit};
#[cfg(feature = "wal")]
use tokio::sync::OwnedMutexGuard;
use hyperfile_reactor::{Capacity, Channel, Task, TaskBuilder, TaskHandler};
use crate::{SegmentId, BlockIndex};
use crate::Cno;
#[cfg(feature = "wal")]
use crate::inode::OnDiskState;
use crate::buffer::{DataBlock, AlignedDataBlockWrapper, BatchDataBlockWrapper};
use super::hyper::Hyper;
use super::HyperTrait;

/// Wrapper around `TaskHandler` that bundles three priority
/// channels (highprio / cb / user) and exposes them as named send
/// methods. All three channels are unbounded.
///
/// Built once per Hyper task by [`build_channel_group`]; cheaply
/// cloneable thereafter (the inner handler is `Arc`-backed and the
/// `Channel` tokens are `Copy`).
///
/// ## Channel priorities
///
/// Lower number is served first.
///
/// - `highprio` (priority 0): requests put back after a contention
///   miss, when the range lock or the per-file permit was taken. They
///   were already accepted from the user, so they rank above new
///   requests and are not starved by them.
/// - `cb` (priority 1): internal callback re-routes — the later hops of
///   a multi-hop pipeline (write retrieve -> absorb, WAL write -> PUT
///   done -> cache update, a spawned block fetch handing its block back
///   to be cached). Above `user` so an in-progress hop finishes before a
///   new request starts one.
/// - `user` (priority 2): incoming user requests from
///   `HyperFileHandler::fh_*` / `HyperFileTokio::*`.
///
/// Note that a hop in flight can be holding the per-file permit:
/// `FileReqWrite::spawn_write_permit` carries one from `spawn_write`
/// until `absorb_write`, and only the handler task can run the hop that
/// releases it. Nothing on this task may therefore *wait* for that
/// permit — see `HyperFile::try_lock`. Ranking `cb` above `highprio`
/// looks like it would help and does not: it changes which of two
/// racing writes lands first, and under the `range-lock` feature that
/// was enough to make two of them retrieve the same block and collide
/// in `absorb_write`.
pub struct ChannelGroup<Ctx> {
    inner: TaskHandler<Ctx>,
    highprio: Channel<Ctx>,
    cb: Channel<Ctx>,
    user: Channel<Ctx>,
}

impl<Ctx> Clone for ChannelGroup<Ctx> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            highprio: self.highprio,
            cb: self.cb,
            user: self.user,
        }
    }
}

impl<Ctx> ChannelGroup<Ctx> {
    /// Send a context to the user channel. Returns
    /// `Err(BrokenPipe)` if the reactor task has terminated.
    pub fn send(&self, ctx: Ctx) -> Result<()> {
        self.inner.send(self.user, ctx).map_err(|_| {
            Error::new(ErrorKind::BrokenPipe, "reactor handler task died")
        })
    }

    /// Send a context to the callback channel.
    pub fn send_cb(&self, ctx: Ctx) -> Result<()> {
        self.inner.send(self.cb, ctx).map_err(|_| {
            Error::new(ErrorKind::BrokenPipe, "reactor handler task died")
        })
    }

    /// Send a context to the high-priority channel.
    pub fn send_highprio(&self, ctx: Ctx) -> Result<()> {
        self.inner.send(self.highprio, ctx).map_err(|_| {
            Error::new(ErrorKind::BrokenPipe, "reactor handler task died")
        })
    }
}

/// Build the standard 3-channel `TaskBuilder` used by every
/// hyperfile reactor task, returning the matching `ChannelGroup`
/// constructor.
///
/// Returns `(builder, |handler| -> ChannelGroup)`: the caller
/// passes the builder to `Reactor::spawn_async`, then feeds the
/// resulting `TaskHandler` through the closure to get a fully
/// wired `ChannelGroup`.
pub fn build_channel_group<Ctx>() -> (TaskBuilder<Ctx>, impl FnOnce(TaskHandler<Ctx>) -> ChannelGroup<Ctx>)
where
    Ctx: Send + 'static,
{
    let mut builder = TaskBuilder::<Ctx>::new();
    let highprio = builder.add_channel(0, Capacity::Unbounded);
    let cb       = builder.add_channel(1, Capacity::Unbounded);
    let user     = builder.add_channel(2, Capacity::Unbounded);
    (builder, move |inner| ChannelGroup { inner, highprio, cb, user })
}

pub type FileRespGetAttr = Result<libc::stat>;
pub type FileRespSetAttr = Result<libc::stat>;
pub type FileRespRead = Result<usize>;
pub type FileRespWrite = Result<usize>;
pub type FileRespWriteZero = Result<usize>;
pub type FileRespTrunc = Result<()>;
/// Whether the block existed, so whether the action ran. The action's
/// own return value travels on its own channel.
pub type FileRespWithBlock = Result<bool>;
pub type FileRespTiming = Result<TimingValue>;
pub type FileRespDirtyBlockCount = Result<usize>;
/// How many blocks a read-ahead installed.
pub type FileRespReadAhead = Result<usize>;
pub type FileRespSeek = Result<Option<usize>>;
pub type FileRespReadPlan = Result<Vec<Vec<crate::file::PlannedRead>>>;
pub type FileRespBlockPlacement = Result<Vec<Vec<Option<(crate::SegmentId, u64)>>>>;
/// How many of the requested blocks were resident.
pub type FileRespWithBlocks = Result<usize>;
/// How many of the requested blocks the closure was given bytes for.
pub type FileRespReadMany = Result<usize>;
/// The bytes read, truncated to what was actually available.
pub type FileRespReadOwned = Result<Bytes>;
pub type FileRespFlush = Result<SegmentId>;
pub type FileRespRelease = Result<SegmentId>;
pub type FileRespLastCno = Cno;

/// See `Hyper::begin_txn`.
#[cfg(feature = "wal")]
pub type FileRespBeginTxn = Result<()>;
/// See `Hyper::commit_txn`.
#[cfg(feature = "wal")]
pub type FileRespCommitTxn = Result<SegmentId>;
/// See `Hyper::abort_txn`.
#[cfg(feature = "wal")]
pub type FileRespAbortTxn = Result<()>;
/// See `Hyper::in_txn`.
#[cfg(feature = "wal")]
pub type FileRespInTxn = bool;

/// Response carrier for a request handed off to the reactor
/// handler.
///
/// Each variant holds the sender side of a channel the caller is
/// awaiting on. When the handler finishes normally it consumes the
/// variant (via `to_*()` accessors) and drives the sender directly.
///
/// The enum (rather than `union { ManuallyDrop<T> }`) matters for
/// crash safety: if the handler task panics before it reaches the
/// consumer call, this `FileResp` is dropped as part of the task
/// frame unwind, and the enum's automatic Drop drops the contained
/// sender. That closes the channel and lets the caller's
/// `rx.await` resolve with `Err(RecvError)` instead of hanging.
pub enum FileResp {
    GetAttr(oneshot::Sender<FileRespGetAttr>),
    SetAttr(oneshot::Sender<FileRespSetAttr>),
    Read(mpsc::Sender<FileRespRead>),
    Write(mpsc::Sender<FileRespWrite>),
    WriteZero(mpsc::Sender<FileRespWriteZero>),
    WriteAlignedBatch(mpsc::Sender<FileRespWrite>),
    WriteBatch(mpsc::Sender<FileRespWrite>),
    Trunc(oneshot::Sender<FileRespTrunc>),
    WithBlock(oneshot::Sender<FileRespWithBlock>),
    Timing(oneshot::Sender<FileRespTiming>),
    DirtyBlockCount(oneshot::Sender<FileRespDirtyBlockCount>),
    ReadAhead(oneshot::Sender<FileRespReadAhead>),
    Seek(oneshot::Sender<FileRespSeek>),
    ReadPlan(oneshot::Sender<FileRespReadPlan>),
    BlockPlacement(oneshot::Sender<FileRespBlockPlacement>),
    WithBlocks(oneshot::Sender<FileRespWithBlocks>),
    ReadMany(oneshot::Sender<FileRespReadMany>),
    /// No response: the requester was answered before requeuing.
    BlockAbsorb,
    ReadOwned(oneshot::Sender<FileRespReadOwned>),
    Flush(oneshot::Sender<FileRespFlush>),
    #[cfg(feature = "wal")]
    WalFlush,
    #[cfg(feature = "wal")]
    WalFlushDone,
    #[cfg(feature = "wal")]
    WalFlushRecovery,
    Release(oneshot::Sender<FileRespRelease>),
    LastCno(oneshot::Sender<FileRespLastCno>),
    #[cfg(feature = "wal")]
    BeginTxn(oneshot::Sender<FileRespBeginTxn>),
    #[cfg(feature = "wal")]
    CommitTxn(oneshot::Sender<FileRespCommitTxn>),
    #[cfg(feature = "wal")]
    AbortTxn(oneshot::Sender<FileRespAbortTxn>),
    #[cfg(feature = "wal")]
    InTxn(oneshot::Sender<FileRespInTxn>),
}

impl FileResp {
    pub fn to_getattr(self) -> oneshot::Sender<FileRespGetAttr> {
        match self {
            Self::GetAttr(tx) => tx,
            _ => panic!("FileResp::to_getattr called on wrong variant"),
        }
    }

    pub fn to_setattr(self) -> oneshot::Sender<FileRespSetAttr> {
        match self {
            Self::SetAttr(tx) => tx,
            _ => panic!("FileResp::to_setattr called on wrong variant"),
        }
    }

    pub fn to_read(self) -> mpsc::Sender<FileRespRead> {
        match self {
            Self::Read(tx) => tx,
            _ => panic!("FileResp::to_read called on wrong variant"),
        }
    }

    pub fn to_write(self) -> mpsc::Sender<FileRespWrite> {
        match self {
            Self::Write(tx) | Self::WriteAlignedBatch(tx) | Self::WriteBatch(tx) => tx,
            _ => panic!("FileResp::to_write called on wrong variant"),
        }
    }

    pub fn to_write_zero(self) -> mpsc::Sender<FileRespWriteZero> {
        match self {
            Self::WriteZero(tx) => tx,
            _ => panic!("FileResp::to_write_zero called on wrong variant"),
        }
    }

    /// Answer a read, whichever form it took.
    ///
    /// A borrowed read reports how many bytes it put in the caller's
    /// buffer. An owned read hands the buffer over instead, truncated to
    /// that same count, so `owned` must be the buffer the read filled.
    pub fn answer_read(self, n: usize, owned: Option<Vec<u8>>) {
        match self {
            Self::Read(tx) => { let _ = tx.try_send(Ok(n)); },
            Self::ReadOwned(tx) => {
                let mut buf = owned.unwrap_or_default();
                buf.truncate(n);
                let _ = tx.send(Ok(Bytes::from(buf)));
            },
            _ => panic!("FileResp::answer_read called on wrong variant"),
        }
    }

    /// Fail a read, whichever form it took.
    pub fn fail_read(self, e: Error) {
        match self {
            Self::Read(tx) => { let _ = tx.try_send(Err(e)); },
            Self::ReadOwned(tx) => { let _ = tx.send(Err(e)); },
            _ => panic!("FileResp::fail_read called on wrong variant"),
        }
    }

    pub fn to_read_owned(self) -> oneshot::Sender<FileRespReadOwned> {
        match self {
            Self::ReadOwned(tx) => tx,
            _ => panic!("FileResp::to_read_owned called on wrong variant"),
        }
    }

    pub fn to_read_many(self) -> oneshot::Sender<FileRespReadMany> {
        match self {
            Self::ReadMany(tx) => tx,
            _ => panic!("FileResp::to_read_many called on wrong variant"),
        }
    }

    pub fn to_with_blocks(self) -> oneshot::Sender<FileRespWithBlocks> {
        match self {
            Self::WithBlocks(tx) => tx,
            _ => panic!("FileResp::to_with_blocks called on wrong variant"),
        }
    }

    pub fn to_seek(self) -> oneshot::Sender<FileRespSeek> {
        match self {
            Self::Seek(tx) => tx,
            _ => panic!("FileResp::to_seek called on wrong variant"),
        }
    }

    pub fn to_read_plan(self) -> oneshot::Sender<FileRespReadPlan> {
        match self {
            Self::ReadPlan(tx) => tx,
            _ => panic!("FileResp::to_read_plan called on wrong variant"),
        }
    }

    pub fn to_block_placement(self) -> oneshot::Sender<FileRespBlockPlacement> {
        match self {
            Self::BlockPlacement(tx) => tx,
            _ => panic!("FileResp::to_block_placement called on wrong variant"),
        }
    }

    pub fn to_read_ahead(self) -> oneshot::Sender<FileRespReadAhead> {
        match self {
            Self::ReadAhead(tx) => tx,
            _ => panic!("FileResp::to_read_ahead called on wrong variant"),
        }
    }

    pub fn to_dirty_block_count(self) -> oneshot::Sender<FileRespDirtyBlockCount> {
        match self {
            Self::DirtyBlockCount(tx) => tx,
            _ => panic!("FileResp::to_dirty_block_count called on wrong variant"),
        }
    }

    pub fn to_timing(self) -> oneshot::Sender<FileRespTiming> {
        match self {
            Self::Timing(tx) => tx,
            _ => panic!("FileResp::to_timing called on wrong variant"),
        }
    }

    /// Answer a block action. Split from `to_with_block` because the
    /// spawning path answers from a task rather than the arm, and has no
    /// use for the sender itself.
    pub fn answer_with_block(self, res: FileRespWithBlock) {
        match self {
            Self::WithBlock(tx) => { let _ = tx.send(res); },
            _ => panic!("FileResp::answer_with_block called on wrong variant"),
        }
    }

    pub fn fail_with_block(self, e: Error) {
        match self {
            Self::WithBlock(tx) => { let _ = tx.send(Err(e)); },
            _ => panic!("FileResp::fail_with_block called on wrong variant"),
        }
    }

    pub fn to_with_block(self) -> oneshot::Sender<FileRespWithBlock> {
        match self {
            Self::WithBlock(tx) => tx,
            _ => panic!("FileResp::to_with_block called on wrong variant"),
        }
    }

    pub fn to_trunc(self) -> oneshot::Sender<FileRespTrunc> {
        match self {
            Self::Trunc(tx) => tx,
            _ => panic!("FileResp::to_trunc called on wrong variant"),
        }
    }

    pub fn to_flush(self) -> oneshot::Sender<FileRespFlush> {
        match self {
            Self::Flush(tx) => tx,
            _ => panic!("FileResp::to_flush called on wrong variant"),
        }
    }

    #[cfg(feature = "wal")]
    pub fn to_wal_flush(self) {
        match self {
            Self::WalFlush => {}
            _ => panic!("FileResp::to_wal_flush called on wrong variant"),
        }
    }

    #[cfg(feature = "wal")]
    pub fn to_wal_flush_done(self) {
        match self {
            Self::WalFlushDone => {}
            _ => panic!("FileResp::to_wal_flush_done called on wrong variant"),
        }
    }

    #[cfg(feature = "wal")]
    pub fn to_wal_flush_recovery(self) {
        match self {
            Self::WalFlushRecovery => {}
            _ => panic!("FileResp::to_wal_flush_recovery called on wrong variant"),
        }
    }

    pub fn to_release(self) -> oneshot::Sender<FileRespRelease> {
        match self {
            Self::Release(tx) => tx,
            _ => panic!("FileResp::to_release called on wrong variant"),
        }
    }

    #[cfg(feature = "wal")]
    pub fn to_begin_txn(self) -> oneshot::Sender<FileRespBeginTxn> {
        match self {
            Self::BeginTxn(tx) => tx,
            _ => panic!("FileResp::to_begin_txn called on wrong variant"),
        }
    }

    #[cfg(feature = "wal")]
    pub fn to_commit_txn(self) -> oneshot::Sender<FileRespCommitTxn> {
        match self {
            Self::CommitTxn(tx) => tx,
            _ => panic!("FileResp::to_commit_txn called on wrong variant"),
        }
    }

    #[cfg(feature = "wal")]
    pub fn to_abort_txn(self) -> oneshot::Sender<FileRespAbortTxn> {
        match self {
            Self::AbortTxn(tx) => tx,
            _ => panic!("FileResp::to_abort_txn called on wrong variant"),
        }
    }

    #[cfg(feature = "wal")]
    pub fn to_in_txn(self) -> oneshot::Sender<FileRespInTxn> {
        match self {
            Self::InTxn(tx) => tx,
            _ => panic!("FileResp::to_in_txn called on wrong variant"),
        }
    }

    pub fn to_last_cno(self) -> oneshot::Sender<FileRespLastCno> {
        match self {
            Self::LastCno(tx) => tx,
            _ => panic!("FileResp::to_last_cno called on wrong variant"),
        }
    }

    pub fn clone_write_resp(&self) -> mpsc::Sender<FileRespWrite> {
        match self {
            Self::Write(tx) | Self::WriteAlignedBatch(tx) | Self::WriteBatch(tx) => tx.clone(),
            _ => panic!("FileResp::clone_write_resp called on wrong variant"),
        }
    }

    pub fn clone_write_zero_resp(&self) -> mpsc::Sender<FileRespWriteZero> {
        match self {
            Self::WriteZero(tx) => tx.clone(),
            _ => panic!("FileResp::clone_write_zero_resp called on wrong variant"),
        }
    }
}

// define request params
pub struct FileReqRead<'a> {
    /// Set when the reactor, not the caller, owns the destination.
    ///
    /// `fh_read` points `buf` at the caller's memory, valid only while
    /// the caller stays parked on the response. `fh_read_owned` instead
    /// allocates here and points `buf` into it, so the destination
    /// belongs to the request and travels with it — through the
    /// flush-wait and range-lock requeues, into the task that fills it,
    /// and safely past a caller that has gone away. A `Vec` does not
    /// move its allocation when the handle moves, so `buf` stays valid.
    ///
    /// Which of the two it is decides what the response carries: a byte
    /// count for a borrowed read, the buffer itself for an owned one.
    pub owned: Option<Vec<u8>>,
    pub buf: &'a mut [u8],
    pub offset: usize,
    pub fh: ChannelGroup<FileContext<'a>>,
}

pub struct FileReqWrite<'a> {
    pub buf: &'a [u8],
    pub offset: usize,
    pub fetched: Vec<DataBlock>,
    /// Set when a block this write needed could not be read. The absorb
    /// fails the write rather than applying it over a block that was
    /// never filled, which would replace its contents with zeroes and
    /// persist that at the next flush.
    pub fetch_err: Option<Error>,
    pub spawn_write_permit: Option<OwnedSemaphorePermit>, // hold owned permit for spawn_write
    pub fh: ChannelGroup<FileContext<'a>>,
    /// Keeps `buf`'s referent alive when the caller does not.
    ///
    /// `fh_write` points `buf` at the caller's memory, which is only
    /// valid while the caller stays parked on the response.
    /// `fh_write_owned` instead puts the bytes here and points `buf`
    /// into them, so the referent belongs to the request and travels
    /// with it — through the absorb and requeue stages, and safely past
    /// a caller that has gone away. `Bytes` does not move its
    /// allocation when the handle moves, so `buf` stays valid.
    pub owned: Option<Bytes>,
}

pub struct FileReqWriteZero<'a> {
    pub offset: usize,
    pub len: usize,
    pub fetched: Vec<DataBlock>,
    /// See `FileReqWrite::fetch_err`.
    pub fetch_err: Option<Error>,
    pub spawn_write_permit: Option<OwnedSemaphorePermit>, // hold owned permit for spawn_write
    pub fh: ChannelGroup<FileContext<'a>>,
}

pub struct FileReqWriteAlignedBatch<'a> {
    pub data_blocks: Vec<AlignedDataBlockWrapper>,
    /// Needed to put the request back after a contention miss.
    pub fh: ChannelGroup<FileContext<'a>>,
}

pub struct FileReqWriteBatch<'a> {
    pub data_blocks: Vec<BatchDataBlockWrapper>,
    /// Needed to put the request back after a contention miss.
    pub fh: ChannelGroup<FileContext<'a>>,
}

pub struct FileReqTrunc<'a> {
    pub offset: usize,
    /// Needed to put the request back after a contention miss.
    pub fh: ChannelGroup<FileContext<'a>>,
}

/// A caller-supplied action to run against a borrowed data block.
///
/// The reactor owns the `Hyper`, so a block borrow cannot be handed
/// back across the channel: the caller would receive it only after
/// the response arrives, by which point the reactor is free to
/// process the next request — from this handle or any clone of it —
/// and a write, flush, truncate or eviction could invalidate the
/// borrow. Views into the local-disk cache also have their backing
/// space punched on eviction.
///
/// So the action travels to the block instead of the block travelling
/// to the caller, and the borrow never leaves the reactor task. The
/// closure returns nothing: `fh_with_block*` wraps the caller's
/// closure in one that sends the result down a dedicated channel,
/// which keeps this type free of the caller's return type and so
/// storable in `FileReqBody`.
///
/// The boxes are `'static` in the type but not in fact: the closure
/// may borrow the caller's locals, and `fh_with_block*` erases that
/// lifetime to get it here. What keeps that sound is
/// [`BlockActionGate`], not the type.
pub enum BlockAction {
    Ref(Box<dyn FnOnce(&[u8]) + Send>),
    Mut(Box<dyn FnOnce(&mut [u8]) + Send>),
}

/// Decides whether a borrowed closure may still run.
///
/// The closure may hold references into the caller's frame, so it must
/// not run after the caller has gone. Awaiting the response covers the
/// ordinary path, but an awaiting future can be dropped, and a
/// cancelled caller would free what the closure still points at.
///
/// `true` means the caller is gone. The caller sets it when its guard
/// drops; the dispatcher checks it while holding the lock and skips the
/// closure if set. Holding the lock across the check and the call is
/// what closes the race.
///
/// The lock is deliberately taken *after* the block has been fetched,
/// so the critical section is one closure body with no I/O in it. A
/// cancelling caller therefore blocks for microseconds at worst. This
/// matters: the reactor's object-store I/O can be driven by the
/// caller's runtime, so blocking the caller across an await would
/// deadlock.
pub type BlockActionGate = Arc<std::sync::Mutex<bool>>;

/// Run once per index in a batch: `Some` for a block resident in the
/// cache, `None` for one that is not.
pub type BatchBlockAction = Box<dyn FnMut(BlockIndex, Option<&[u8]>) + Send>;

/// Which counters a timing request is about.
#[derive(Clone, Copy, Debug)]
pub enum TimingOp {
    Read,
    ReadReset,
    Flush,
    FlushReset,
}

pub struct FileReqTiming {
    pub op: TimingOp,
}

pub struct FileReqDirtyBlockCount {}

pub struct FileReqWithBlocks {
    pub indices: Vec<BlockIndex>,
    pub action: BatchBlockAction,
    /// Guards `action` against running after the caller is gone, as for
    /// a single block. See [`BlockActionGate`].
    pub gate: BlockActionGate,
}

pub struct FileReqReadMany<'a> {
    pub indices: Vec<BlockIndex>,
    pub action: BatchBlockAction,
    /// Guards `action`, as for a single block. See [`BlockActionGate`].
    pub gate: BlockActionGate,
    /// Needed so the spawned fetch can hand what it read back to be
    /// cached.
    pub fh: ChannelGroup<FileContext<'a>>,
}

/// Which of the two `lseek` extension queries to answer.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SeekWhence {
    /// `SEEK_DATA`.
    Data,
    /// `SEEK_HOLE`.
    Hole,
}

/// Where to start looking, and what to look for.
pub struct FileReqSeek {
    pub offset: usize,
    pub whence: SeekWhence,
}

/// Ranges to answer a read-plan query about.
///
/// Batch-shaped even for one range, so the single-range and many-range
/// entry points cannot drift apart.
pub struct FileReqReadPlan {
    pub ranges: Vec<(u64, u64)>,
}

/// Block ranges to answer a placement query about.
pub struct FileReqBlockPlacement {
    pub ranges: Vec<(BlockIndex, usize)>,
}

pub struct FileReqReadAhead<'a> {
    pub offset: usize,
    pub len: usize,
    pub fh: ChannelGroup<FileContext<'a>>,
}

/// Hands a block loaded by a spawned task back for caching.
///
/// A spawned read-only block fetch owns the block it filled, and
/// putting it in the cache needs `&mut self`, which only a handler arm
/// has. So the task requeues this, the same way the write path's
/// retrieve requeues its fetched blocks for `absorb_write` to insert.
///
/// Caching is what makes the block path cheaper than the byte path —
/// repeated access to the same blocks costs one request instead of one
/// each — so skipping this would give away the reason to use it.
pub struct FileReqBlockAbsorb {
    pub blk_idx: BlockIndex,
    pub block: DataBlock,
}

/// A counter snapshot, or an acknowledgement for a reset.
///
/// Snapshots are owned values rather than references to the live
/// counters, which cannot leave the reactor task.
#[derive(Clone, Copy, Debug)]
pub enum TimingValue {
    Read(super::ReadTimingSnapshot),
    Flush(super::FlushTimingSnapshot),
    Reset,
}

pub struct FileReqWithBlock<'a> {
    pub blk_idx: BlockIndex,
    /// Materialize a zero-filled block if the index has no data.
    /// Ignored by [`BlockAction::Ref`], which cannot create.
    pub create: bool,
    pub action: BlockAction,
    /// Guards `action` against running after the caller is gone. See
    /// [`BlockActionGate`].
    pub gate: BlockActionGate,
    /// Needed to requeue: the read-only path hands the block back for
    /// caching, and defers to a retry while a flush is in progress.
    pub fh: ChannelGroup<FileContext<'a>>,
}

pub struct FileReqGetAttr {}

pub struct FileReqSetAttr<'a> {
    pub stat: libc::stat,
    /// Carried so the arm can re-queue this request when a flush is in flight;
    /// see the `SetAttr` arm of `Task::handle`.
    pub fh: ChannelGroup<FileContext<'a>>,
}

pub struct FileReqFlush<'a> {
    pub fh: ChannelGroup<FileContext<'a>>,
}

#[cfg(feature = "wal")]
pub struct FileReqWalFlush<'a> {
    pub fh: ChannelGroup<FileContext<'a>>,
}

#[cfg(feature = "wal")]
pub struct FileReqWalFlushDone {
    pub lock: OwnedMutexGuard<()>,
    pub segid: SegmentId,
    pub od_state: OnDiskState,
    pub bmap_cache_limit: usize,
    /// Fingerprint of the inode that was published, which the session now builds on.
    pub base: u64,
}

#[cfg(feature = "wal")]
pub struct FileReqWalFlushRecovery {
    pub lock: OwnedMutexGuard<()>,
}

pub struct FileReqRelease<'a> {
    pub fh: ChannelGroup<FileContext<'a>>,
}

pub struct FileReqLastCno {}

/// See `Hyper::begin_txn`. Nothing to carry: the interval lives in the handler
/// task, because that is where the file does.
#[cfg(feature = "wal")]
pub struct FileReqBeginTxn {}

/// See `Hyper::commit_txn`. Carries the channel so the arm can re-queue when a
/// flush is in flight — committing publishes, and publishing needs the lock.
#[cfg(feature = "wal")]
pub struct FileReqCommitTxn<'a> {
    pub fh: ChannelGroup<FileContext<'a>>,
}

/// See `Hyper::abort_txn`. Carries the channel for the same reason: the rollback
/// reads back what is persisted, which must not race a publish.
#[cfg(feature = "wal")]
pub struct FileReqAbortTxn<'a> {
    pub fh: ChannelGroup<FileContext<'a>>,
}

/// See `Hyper::in_txn`.
#[cfg(feature = "wal")]
pub struct FileReqInTxn {}

pub enum FileReqOp {
    GetAttr,
    SetAttr,
    Read,
    Write,
    WriteAbsorb,
    WriteAbsorbBh,
    WriteZero,
    WriteZeroAbsorb,
    WriteZeroAbsorbBh,
    WriteAlignedBatch,
    WriteBatch,
    Trunc,
    WithBlock,
    Timing,
    DirtyBlockCount,
    ReadAhead,
    Seek,
    ReadPlan,
    BlockPlacement,
    WithBlocks,
    ReadMany,
    BlockAbsorb,
    Flush,
    FlushData,
    #[cfg(feature = "wal")]
    WalFlush,
    #[cfg(feature = "wal")]
    WalFlushDone,
    #[cfg(feature = "wal")]
    WalFlushRecovery,
    Release,
    LastCno,
    #[cfg(feature = "wal")]
    BeginTxn,
    #[cfg(feature = "wal")]
    CommitTxn,
    #[cfg(feature = "wal")]
    AbortTxn,
    #[cfg(feature = "wal")]
    InTxn,
    #[cfg(feature = "wal")]
    WriteWal,
    #[cfg(feature = "wal")]
    WriteZeroWal,
}

#[repr(C)]
pub union FileReqBody<'a> {
    getattr: ManuallyDrop<FileReqGetAttr>,
    setattr: ManuallyDrop<FileReqSetAttr<'a>>,
    read: ManuallyDrop<FileReqRead<'a>>,
    write: ManuallyDrop<FileReqWrite<'a>>,
    write_zero: ManuallyDrop<FileReqWriteZero<'a>>,
    write_aligned_batch: ManuallyDrop<FileReqWriteAlignedBatch<'a>>,
    write_batch: ManuallyDrop<FileReqWriteBatch<'a>>,
    trunc: ManuallyDrop<FileReqTrunc<'a>>,
    with_block: ManuallyDrop<FileReqWithBlock<'a>>,
    timing: ManuallyDrop<FileReqTiming>,
    dirty_block_count: ManuallyDrop<FileReqDirtyBlockCount>,
    read_ahead: ManuallyDrop<FileReqReadAhead<'a>>,
    seek: ManuallyDrop<FileReqSeek>,
    read_plan: ManuallyDrop<FileReqReadPlan>,
    block_placement: ManuallyDrop<FileReqBlockPlacement>,
    with_blocks: ManuallyDrop<FileReqWithBlocks>,
    read_many: ManuallyDrop<FileReqReadMany<'a>>,
    block_absorb: ManuallyDrop<FileReqBlockAbsorb>,
    #[cfg(feature = "wal")]
    wal_flush: ManuallyDrop<FileReqWalFlush<'a>>,
    #[cfg(feature = "wal")]
    wal_flush_done: ManuallyDrop<FileReqWalFlushDone>,
    #[cfg(feature = "wal")]
    wal_flush_recovery: ManuallyDrop<FileReqWalFlushRecovery>,
    flush: ManuallyDrop<FileReqFlush<'a>>,
    release: ManuallyDrop<FileReqRelease<'a>>,
    last_cno: ManuallyDrop<FileReqLastCno>,
    #[cfg(feature = "wal")]
    begin_txn: ManuallyDrop<FileReqBeginTxn>,
    #[cfg(feature = "wal")]
    commit_txn: ManuallyDrop<FileReqCommitTxn<'a>>,
    #[cfg(feature = "wal")]
    abort_txn: ManuallyDrop<FileReqAbortTxn<'a>>,
    #[cfg(feature = "wal")]
    in_txn: ManuallyDrop<FileReqInTxn>,
}

pub struct FileReq<'a> {
    op: FileReqOp,
    body: FileReqBody<'a>,
}

pub struct FileContext<'a> {
    req: Option<FileReq<'a>>,
    resp: Option<FileResp>,
}

impl<'a> FileContext<'a> {
    pub fn take(mut self) -> (FileReq<'a>, FileResp) {
        (
            self.req.take().unwrap(),
            self.resp.take().unwrap()
        )
    }

    pub fn new_getattr() -> (Self, oneshot::Receiver<FileRespGetAttr>) {
        let (tx, rx) = oneshot::channel::<FileRespGetAttr>();
        let req = FileReq {
            op: FileReqOp::GetAttr,
            body: FileReqBody { getattr: ManuallyDrop::new(FileReqGetAttr {}), },
        };
        let resp = FileResp::GetAttr(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    pub fn new_setattr(fh: ChannelGroup<FileContext<'a>>, stat: libc::stat) -> (Self, oneshot::Receiver<FileRespSetAttr>) {
        let (tx, rx) = oneshot::channel::<FileRespSetAttr>();
        let req = FileReq {
            op: FileReqOp::SetAttr,
            body: FileReqBody { setattr: ManuallyDrop::new(FileReqSetAttr { stat: stat, fh, }), },
        };
        let resp = FileResp::SetAttr(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    // return both tx and rx, for mpsc channel we need to keep tx until we receved response
    pub fn new_read(buf: &'a mut [u8], offset: usize, fh: ChannelGroup<FileContext<'a>>) -> (Self, mpsc::Sender<FileRespRead>, mpsc::Receiver<FileRespRead>) {
        let (tx, rx) = mpsc::channel::<FileRespRead>(1);
        let req = FileReq {
            op: FileReqOp::Read,
            body: FileReqBody { read: ManuallyDrop::new(FileReqRead { buf: buf, offset: offset, fh: fh, owned: None }), },
        };
        let resp = FileResp::Read(tx.clone());
        (Self { req: Some(req), resp: Some(resp), }, tx, rx)
    }

    // return both tx and rx, for mpsc channel we need to keep tx until we receved response
    pub fn new_write(buf: &'a [u8], offset: usize, fh: ChannelGroup<FileContext<'a>>) -> (Self, mpsc::Sender<FileRespWrite>, mpsc::Receiver<FileRespWrite>) {
        let (tx, rx) = mpsc::channel::<FileRespWrite>(1);
        let req = FileReq {
            op: FileReqOp::Write,
            body: FileReqBody { write: ManuallyDrop::new(FileReqWrite { buf: buf, offset: offset, spawn_write_permit: None, fh: fh, fetched: Vec::new(), fetch_err: None, owned: None, }), },
        };
        let resp = FileResp::Write(tx.clone());
        (Self { req: Some(req), resp: Some(resp), }, tx, rx)
    }

    #[cfg(feature = "wal")]
    pub fn write_wal(req: FileReqWrite<'a>, resp: FileResp) -> Self {
        let new_req = FileReq {
            op: FileReqOp::WriteWal,
            body: FileReqBody { write: ManuallyDrop::new(req), },
        };
        Self { req: Some(new_req), resp: Some(resp), }
    }

    // convert context from op Write to op WriteAbsorb
    pub fn write_absorb(req: FileReqWrite<'a>, resp: FileResp) -> Self {
        let new_req = FileReq {
            op: FileReqOp::WriteAbsorb,
            body: FileReqBody { write: ManuallyDrop::new(req), },
        };
        Self { req: Some(new_req), resp: Some(resp), }
    }

    pub fn write_absorb_bh(req: FileReqWrite<'a>, resp: FileResp) -> Self {
        let new_req = FileReq {
            op: FileReqOp::WriteAbsorbBh,
            body: FileReqBody { write: ManuallyDrop::new(req), },
        };
        Self { req: Some(new_req), resp: Some(resp), }
    }

    // return both tx and rx, for mpsc channel we need to keep tx until we receved response
    pub fn new_write_zero(offset: usize, len: usize, fh: ChannelGroup<FileContext<'a>>) -> (Self, mpsc::Sender<FileRespWrite>, mpsc::Receiver<FileRespWrite>) {
        let (tx, rx) = mpsc::channel::<FileRespWriteZero>(1);
        let req = FileReq {
            op: FileReqOp::WriteZero,
            body: FileReqBody { write_zero: ManuallyDrop::new(FileReqWriteZero { offset: offset, len: len, spawn_write_permit: None, fh: fh, fetched: Vec::new(), fetch_err: None, }), },
        };
        let resp = FileResp::WriteZero(tx.clone());
        (Self { req: Some(req), resp: Some(resp), }, tx, rx)
    }

    #[cfg(feature = "wal")]
    pub fn write_zero_wal(req: FileReqWriteZero<'a>, resp: FileResp) -> Self {
        let new_req = FileReq {
            op: FileReqOp::WriteZeroWal,
            body: FileReqBody { write_zero: ManuallyDrop::new(req), },
        };
        Self { req: Some(new_req), resp: Some(resp), }
    }

    // convert context from op Write to op WriteAbsorb
    pub fn write_zero_absorb(req: FileReqWriteZero<'a>, resp: FileResp) -> Self {
        let new_req = FileReq {
            op: FileReqOp::WriteZeroAbsorb,
            body: FileReqBody { write_zero: ManuallyDrop::new(req), },
        };
        Self { req: Some(new_req), resp: Some(resp), }
    }

    pub fn write_zero_absorb_bh(req: FileReqWriteZero<'a>, resp: FileResp) -> Self {
        let new_req = FileReq {
            op: FileReqOp::WriteZeroAbsorbBh,
            body: FileReqBody { write_zero: ManuallyDrop::new(req), },
        };
        Self { req: Some(new_req), resp: Some(resp), }
    }

    pub fn new_write_aligned_batch(v: Vec<AlignedDataBlockWrapper>, fh: ChannelGroup<FileContext<'a>>) -> (Self, mpsc::Receiver<FileRespWrite>) {
        let (tx, rx) = mpsc::channel::<FileRespWrite>(1);
        let new_req = FileReq {
            op: FileReqOp::WriteAlignedBatch,
            body: FileReqBody { write_aligned_batch: ManuallyDrop::new(FileReqWriteAlignedBatch { data_blocks: v, fh }), },
        };
        let resp = FileResp::WriteAlignedBatch(tx);
        (Self { req: Some(new_req), resp: Some(resp), }, rx)
    }

    pub fn new_write_batch(v: Vec<BatchDataBlockWrapper>, fh: ChannelGroup<FileContext<'a>>) -> (Self, mpsc::Receiver<FileRespWrite>) {
        let (tx, rx) = mpsc::channel::<FileRespWrite>(1);
        let new_req = FileReq {
            op: FileReqOp::WriteBatch,
            body: FileReqBody { write_batch: ManuallyDrop::new(FileReqWriteBatch { data_blocks: v, fh }), },
        };
        let resp = FileResp::WriteBatch(tx);
        (Self { req: Some(new_req), resp: Some(resp), }, rx)
    }

    pub fn new_trunc(offset: usize, fh: ChannelGroup<FileContext<'a>>) -> (Self, oneshot::Receiver<FileRespTrunc>) {
        let (tx, rx) = oneshot::channel::<FileRespTrunc>();
        let req = FileReq {
            op: FileReqOp::Trunc,
            body: FileReqBody { trunc: ManuallyDrop::new(FileReqTrunc { offset: offset, fh }), },
        };
        let resp = FileResp::Trunc(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    /// A write whose bytes the request owns. See
    /// [`FileReqWrite::owned`].
    pub fn new_write_owned(buf: Bytes, offset: usize, fh: ChannelGroup<FileContext<'a>>)
        -> (Self, mpsc::Sender<FileRespWrite>, mpsc::Receiver<FileRespWrite>)
    {
        let (tx, rx) = mpsc::channel::<FileRespWrite>(1);
        // SAFETY: points into the `Bytes` stored alongside it. `Bytes`
        // keeps its allocation put when the handle moves, and the
        // handle travels with the request, so the referent outlives
        // every stage that reads it — including after the caller is
        // gone.
        let slice: &'a [u8] = unsafe {
            std::slice::from_raw_parts(buf.as_ptr(), buf.len())
        };
        let req = FileReq {
            op: FileReqOp::Write,
            body: FileReqBody { write: ManuallyDrop::new(FileReqWrite {
                buf: slice, offset, spawn_write_permit: None, fh,
                fetched: Vec::new(), fetch_err: None, owned: Some(buf),
            }), },
        };
        let resp = FileResp::Write(tx.clone());
        (Self { req: Some(req), resp: Some(resp), }, tx, rx)
    }

    /// A read whose destination the reactor owns. Shares
    /// [`FileReqOp::Read`] with the borrowed form, so it takes the same
    /// spawning path; see [`FileReqRead::owned`].
    pub fn new_read_owned(offset: usize, len: usize, fh: ChannelGroup<FileContext<'a>>)
        -> (Self, oneshot::Receiver<FileRespReadOwned>)
    {
        let (tx, rx) = oneshot::channel::<FileRespReadOwned>();
        let mut owned = vec![0u8; len];
        // SAFETY: points into the `Vec` stored alongside it. A `Vec`
        // keeps its allocation put when the handle moves, and the handle
        // travels with the request, so the referent outlives every
        // stage that writes to it. The `Vec` is not touched again until
        // the reads have finished and it is moved out to be returned.
        let slice: &'a mut [u8] = unsafe {
            std::slice::from_raw_parts_mut(owned.as_mut_ptr(), owned.len())
        };
        let req = FileReq {
            op: FileReqOp::Read,
            body: FileReqBody { read: ManuallyDrop::new(FileReqRead {
                buf: slice, offset, fh, owned: Some(owned),
            }), },
        };
        let resp = FileResp::ReadOwned(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    /// Rebuild a truncate so it can be retried after a contention miss.
    pub fn reform_trunc(offset: usize, fh: ChannelGroup<FileContext<'a>>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::Trunc,
            body: FileReqBody { trunc: ManuallyDrop::new(FileReqTrunc { offset, fh }) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    /// Rebuild an aligned batch write so it can be retried. The blocks
    /// travel back with it, so a contention miss costs no data.
    pub fn reform_write_aligned_batch(data_blocks: Vec<AlignedDataBlockWrapper>,
        fh: ChannelGroup<FileContext<'a>>, resp: FileResp) -> Self
    {
        let req = FileReq {
            op: FileReqOp::WriteAlignedBatch,
            body: FileReqBody { write_aligned_batch: ManuallyDrop::new(FileReqWriteAlignedBatch { data_blocks, fh }) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    /// Rebuild a batch write so it can be retried.
    pub fn reform_write_batch(data_blocks: Vec<BatchDataBlockWrapper>,
        fh: ChannelGroup<FileContext<'a>>, resp: FileResp) -> Self
    {
        let req = FileReq {
            op: FileReqOp::WriteBatch,
            body: FileReqBody { write_batch: ManuallyDrop::new(FileReqWriteBatch { data_blocks, fh }) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    /// Rebuild a block-action request so it can be retried, as the byte
    /// paths do when a flush is in progress.
    pub fn reform_with_block(blk_idx: BlockIndex, create: bool, action: BlockAction,
        gate: BlockActionGate, fh: ChannelGroup<FileContext<'a>>, resp: FileResp) -> Self
    {
        let req = FileReq {
            op: FileReqOp::WithBlock,
            body: FileReqBody {
                with_block: ManuallyDrop::new(FileReqWithBlock { blk_idx, create, action, gate, fh }),
            },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    /// Requeue a loaded block for caching. Carries no response: the
    /// caller has already been answered by the task that filled it.
    pub fn block_absorb(blk_idx: BlockIndex, block: DataBlock) -> Self {
        let req = FileReq {
            op: FileReqOp::BlockAbsorb,
            body: FileReqBody { block_absorb: ManuallyDrop::new(FileReqBlockAbsorb { blk_idx, block }), },
        };
        Self { req: Some(req), resp: Some(FileResp::BlockAbsorb) }
    }

    pub fn new_read_many(indices: Vec<BlockIndex>, action: BatchBlockAction, gate: BlockActionGate,
        fh: ChannelGroup<FileContext<'a>>) -> (Self, oneshot::Receiver<FileRespReadMany>) {
        let (tx, rx) = oneshot::channel::<FileRespReadMany>();
        let req = FileReq {
            op: FileReqOp::ReadMany,
            body: FileReqBody { read_many: ManuallyDrop::new(FileReqReadMany {
                indices, action, gate, fh }), },
        };
        let resp = FileResp::ReadMany(tx);
        (Self { req: Some(req), resp: Some(resp) }, rx)
    }

    pub fn new_with_blocks(indices: Vec<BlockIndex>, action: BatchBlockAction, gate: BlockActionGate) -> (Self, oneshot::Receiver<FileRespWithBlocks>) {
        let (tx, rx) = oneshot::channel::<FileRespWithBlocks>();
        let req = FileReq {
            op: FileReqOp::WithBlocks,
            body: FileReqBody { with_blocks: ManuallyDrop::new(FileReqWithBlocks { indices, action, gate }), },
        };
        let resp = FileResp::WithBlocks(tx);
        (Self { req: Some(req), resp: Some(resp) }, rx)
    }

    pub fn new_read_ahead(offset: usize, len: usize, fh: ChannelGroup<FileContext<'a>>) -> (Self, oneshot::Receiver<FileRespReadAhead>) {
        let (tx, rx) = oneshot::channel::<FileRespReadAhead>();
        let req = FileReq {
            op: FileReqOp::ReadAhead,
            body: FileReqBody { read_ahead: ManuallyDrop::new(FileReqReadAhead { offset, len, fh }), },
        };
        let resp = FileResp::ReadAhead(tx);
        (Self { req: Some(req), resp: Some(resp) }, rx)
    }

    pub fn new_seek(offset: usize, whence: SeekWhence) -> (Self, oneshot::Receiver<FileRespSeek>) {
        let (tx, rx) = oneshot::channel::<FileRespSeek>();
        let req = FileReq {
            op: FileReqOp::Seek,
            body: FileReqBody { seek: ManuallyDrop::new(FileReqSeek { offset, whence }), },
        };
        let resp = FileResp::Seek(tx);
        (Self { req: Some(req), resp: Some(resp) }, rx)
    }

    pub fn new_read_plan(ranges: Vec<(u64, u64)>) -> (Self, oneshot::Receiver<FileRespReadPlan>) {
        let (tx, rx) = oneshot::channel::<FileRespReadPlan>();
        let req = FileReq {
            op: FileReqOp::ReadPlan,
            body: FileReqBody { read_plan: ManuallyDrop::new(FileReqReadPlan { ranges }), },
        };
        let resp = FileResp::ReadPlan(tx);
        (Self { req: Some(req), resp: Some(resp) }, rx)
    }

    pub fn new_block_placement(ranges: Vec<(BlockIndex, usize)>) -> (Self, oneshot::Receiver<FileRespBlockPlacement>) {
        let (tx, rx) = oneshot::channel::<FileRespBlockPlacement>();
        let req = FileReq {
            op: FileReqOp::BlockPlacement,
            body: FileReqBody { block_placement: ManuallyDrop::new(FileReqBlockPlacement { ranges }), },
        };
        let resp = FileResp::BlockPlacement(tx);
        (Self { req: Some(req), resp: Some(resp) }, rx)
    }

    pub fn new_dirty_block_count() -> (Self, oneshot::Receiver<FileRespDirtyBlockCount>) {
        let (tx, rx) = oneshot::channel::<FileRespDirtyBlockCount>();
        let req = FileReq {
            op: FileReqOp::DirtyBlockCount,
            body: FileReqBody { dirty_block_count: ManuallyDrop::new(FileReqDirtyBlockCount {}), },
        };
        let resp = FileResp::DirtyBlockCount(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    pub fn new_timing(op: TimingOp) -> (Self, oneshot::Receiver<FileRespTiming>) {
        let (tx, rx) = oneshot::channel::<FileRespTiming>();
        let req = FileReq {
            op: FileReqOp::Timing,
            body: FileReqBody { timing: ManuallyDrop::new(FileReqTiming { op }), },
        };
        let resp = FileResp::Timing(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    pub fn new_with_block(blk_idx: BlockIndex, create: bool, action: BlockAction,
        gate: BlockActionGate, fh: ChannelGroup<FileContext<'a>>)
        -> (Self, oneshot::Receiver<FileRespWithBlock>)
    {
        let (tx, rx) = oneshot::channel::<FileRespWithBlock>();
        let req = FileReq {
            op: FileReqOp::WithBlock,
            body: FileReqBody {
                with_block: ManuallyDrop::new(FileReqWithBlock { blk_idx, create, action, gate, fh }),
            },
        };
        let resp = FileResp::WithBlock(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    pub fn new_flush(fh: ChannelGroup<FileContext<'a>>) -> (Self, oneshot::Receiver<FileRespFlush>) {
        let (tx, rx) = oneshot::channel::<FileRespFlush>();
        let req = FileReq {
            op: FileReqOp::Flush,
            body: FileReqBody { flush: ManuallyDrop::new(FileReqFlush { fh, }), },
        };
        let resp = FileResp::Flush(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    /// Like `new_flush` but with `fdatasync` semantics — see
    /// `Hyper::fs_fdatasync`. Reuses the `FileReqFlush` body and
    /// `FileResp::Flush` response shape; the only difference is
    /// the `op` discriminator routes to the
    /// `FileReqOp::FlushData` arm of `Task::handle`.
    pub fn new_flush_data(fh: ChannelGroup<FileContext<'a>>) -> (Self, oneshot::Receiver<FileRespFlush>) {
        let (tx, rx) = oneshot::channel::<FileRespFlush>();
        let req = FileReq {
            op: FileReqOp::FlushData,
            body: FileReqBody { flush: ManuallyDrop::new(FileReqFlush { fh, }), },
        };
        let resp = FileResp::Flush(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    // only used by internal driven process when wal enabled
    #[cfg(feature = "wal")]
    pub fn new_wal_flush(fh: ChannelGroup<FileContext<'a>>) -> Self {
        let req = FileReq {
            op: FileReqOp::WalFlush,
            body: FileReqBody { wal_flush: ManuallyDrop::new(FileReqWalFlush { fh, }), },
        };
        let resp = FileResp::WalFlush;
        Self { req: Some(req), resp: Some(resp) }
    }

    #[cfg(feature = "wal")]
    pub fn new_wal_flush_done(lock: OwnedMutexGuard<()>, segid: SegmentId, od_state: OnDiskState, bmap_cache_limit: usize, base: u64) -> Self {
        let req = FileReq {
            op: FileReqOp::WalFlushDone,
            body: FileReqBody { wal_flush_done: ManuallyDrop::new(FileReqWalFlushDone { lock, segid, od_state, bmap_cache_limit, base, }), },
        };
        let resp = FileResp::WalFlushDone;
        Self { req: Some(req), resp: Some(resp) }
    }

    #[cfg(feature = "wal")]
    pub fn new_wal_flush_recovery(lock: OwnedMutexGuard<()>) -> Self {
        let req = FileReq {
            op: FileReqOp::WalFlushRecovery,
            body: FileReqBody { wal_flush_recovery: ManuallyDrop::new(FileReqWalFlushRecovery { lock, }), },
        };
        let resp = FileResp::WalFlushRecovery;
        Self { req: Some(req), resp: Some(resp) }
    }

    pub fn new_release(fh: ChannelGroup<FileContext<'a>>) -> (Self, oneshot::Receiver<FileRespRelease>) {
        let (tx, rx) = oneshot::channel::<FileRespRelease>();
        let req = FileReq {
            op: FileReqOp::Release,
            body: FileReqBody { release: ManuallyDrop::new(FileReqRelease { fh, }), },
        };
        let resp = FileResp::Release(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    /// Open an atomic interval on the handler side. See `Hyper::begin_txn` for
    /// what it does and does not promise — no isolation, no implicit undo.
    ///
    /// The interval belongs to the handler task, not the caller: every write
    /// sent after this and before a commit falls inside it, whoever sent it.
    #[cfg(feature = "wal")]
    pub fn new_begin_txn() -> (Self, oneshot::Receiver<FileRespBeginTxn>) {
        let (tx, rx) = oneshot::channel::<FileRespBeginTxn>();
        let req = FileReq {
            op: FileReqOp::BeginTxn,
            body: FileReqBody { begin_txn: ManuallyDrop::new(FileReqBeginTxn {}), },
        };
        let resp = FileResp::BeginTxn(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    /// See `Hyper::commit_txn`.
    #[cfg(feature = "wal")]
    pub fn new_commit_txn(fh: ChannelGroup<FileContext<'a>>) -> (Self, oneshot::Receiver<FileRespCommitTxn>) {
        let (tx, rx) = oneshot::channel::<FileRespCommitTxn>();
        let req = FileReq {
            op: FileReqOp::CommitTxn,
            body: FileReqBody { commit_txn: ManuallyDrop::new(FileReqCommitTxn { fh, }), },
        };
        let resp = FileResp::CommitTxn(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    /// See `Hyper::abort_txn`.
    #[cfg(feature = "wal")]
    pub fn new_abort_txn(fh: ChannelGroup<FileContext<'a>>) -> (Self, oneshot::Receiver<FileRespAbortTxn>) {
        let (tx, rx) = oneshot::channel::<FileRespAbortTxn>();
        let req = FileReq {
            op: FileReqOp::AbortTxn,
            body: FileReqBody { abort_txn: ManuallyDrop::new(FileReqAbortTxn { fh, }), },
        };
        let resp = FileResp::AbortTxn(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    /// See `Hyper::in_txn`. Worth asking after an error: a commit that reported
    /// one leaves the interval open, and the caller cannot tell from its own
    /// side because the state is not there.
    #[cfg(feature = "wal")]
    pub fn new_in_txn() -> (Self, oneshot::Receiver<FileRespInTxn>) {
        let (tx, rx) = oneshot::channel::<FileRespInTxn>();
        let req = FileReq {
            op: FileReqOp::InTxn,
            body: FileReqBody { in_txn: ManuallyDrop::new(FileReqInTxn {}), },
        };
        let resp = FileResp::InTxn(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    #[cfg(feature = "wal")]
    pub fn reform_commit_txn(req: FileReqCommitTxn<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::CommitTxn,
            body: FileReqBody { commit_txn: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    #[cfg(feature = "wal")]
    pub fn reform_abort_txn(req: FileReqAbortTxn<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::AbortTxn,
            body: FileReqBody { abort_txn: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    pub fn new_last_cno() -> (Self, oneshot::Receiver<FileRespLastCno>) {
        let (tx, rx) = oneshot::channel::<FileRespLastCno>();
        let req = FileReq {
            op: FileReqOp::LastCno,
            body: FileReqBody { last_cno: ManuallyDrop::new(FileReqLastCno {}), },
        };
        let resp = FileResp::LastCno(tx);
        (Self { req: Some(req), resp: Some(resp), }, rx)
    }

    pub fn reform_read(req: FileReqRead<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::Read,
            body: FileReqBody { read: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    pub fn reform_write(req: FileReqWrite<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::Write,
            body: FileReqBody { write: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    pub fn reform_write_zero(req: FileReqWriteZero<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::WriteZero,
            body: FileReqBody { write_zero: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    pub fn reform_release(req: FileReqRelease<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::Release,
            body: FileReqBody { release: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    pub fn reform_setattr(req: FileReqSetAttr<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::SetAttr,
            body: FileReqBody { setattr: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    pub fn reform_flush(req: FileReqFlush<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::Flush,
            body: FileReqBody { flush: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    pub fn reform_flush_data(req: FileReqFlush<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::FlushData,
            body: FileReqBody { flush: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }

    #[cfg(feature = "wal")]
    pub fn reform_wal_flush(req: FileReqWalFlush<'a>, resp: FileResp) -> Self {
        let req = FileReq {
            op: FileReqOp::WalFlush,
            body: FileReqBody { wal_flush: ManuallyDrop::new(req) },
        };
        Self { req: Some(req), resp: Some(resp) }
    }
}

impl<'a: 'static> Task<FileContext<'a>> for Hyper<'a>
{
    // main loop
    async fn handle(&mut self, ctx: FileContext<'a>) {
        let (req, resp) = ctx.take();
        match req.op {
            FileReqOp::GetAttr => {
                let md = unsafe { req.body.getattr };
                let _ = ManuallyDrop::into_inner(md);
                let stat = self.inner.stat();
                let resp = resp.to_getattr();
                let _ = resp.send(Ok(stat));
            },
            FileReqOp::SetAttr => {
                let md = unsafe { req.body.setattr };
                let req = ManuallyDrop::into_inner(md);
                let stat = req.stat;
                let res = self.inner.update_stat(&stat).await;
                match res {
                    // Persisting the change needs the flush lock, and under a
                    // WAL that lock may be travelling with a publish already in
                    // flight — its guard comes back as a callback to this task.
                    // Waiting for it here would stop this task from ever taking
                    // that callback, so `update_stat` reports the conflict and
                    // the request goes back on the queue. Returning to the loop
                    // is the whole point: that is when the callback gets taken
                    // and the lock comes back.
                    Err(ref e) if super::is_requeued(e) => {
                        let fh = req.fh.clone();
                        let ctx = FileContext::reform_setattr(req, resp);
                        let _ = fh.send_highprio(ctx);
                    },
                    _ => { let _ = resp.to_setattr().send(res); },
                }
            },
            FileReqOp::Read => {
                let md = unsafe { req.body.read };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                let range = req.offset as u64..(req.offset + req.buf.len()) as u64;
                // `spawn_read` answers its own errors, so the response
                // is handed straight over. An owned read replies on a
                // oneshot, which cannot be cloned for a second owner the
                // way the borrowed form's mpsc sender was.
                let res = self.inner.spawn_read(req, resp).await;
                match res {
                    Ok(_) => {}
                    Err(ref e) => {
                        if !super::is_requeued(e) {
                            #[cfg(feature = "range-lock")]
                            self.inner.range_lock.try_unlock(range);
                        }
                    },
                }
            },
            FileReqOp::Write => {
                let md = unsafe { req.body.write };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                let range = req.offset as u64..(req.offset + req.buf.len()) as u64;
                // prepare error response handler
                let _resp_write = resp.to_write();
                let resp_write = _resp_write.clone();
                let resp = FileResp::Write(_resp_write);
                let res = self.inner.spawn_write(req, resp).await;
                match res {
                    Ok(bytes) => {
                        let _ = resp_write.try_send(Ok(bytes));
                    },
                    Err(ref e) => {
                        if !super::is_requeued(e) {
                            #[cfg(feature = "range-lock")]
                            self.inner.range_lock.try_unlock(range);
                            let _ = resp_write.try_send(res);
                        }
                    },
                }
            },
            #[cfg(feature = "wal")]
            FileReqOp::WriteWal => {
                let md = unsafe { req.body.write };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                let range = req.offset as u64..(req.offset + req.buf.len()) as u64;
                // prepare error response handler
                let _resp_write = resp.to_write();
                let resp_write = _resp_write.clone();
                let resp = FileResp::Write(_resp_write);
                let res = self.inner.spawn_write_wal(req, resp).await;
                match res {
                    Ok(_) => {},
                    Err(ref e) => {
                        if !super::is_requeued(e) {
                            #[cfg(feature = "range-lock")]
                            self.inner.range_lock.try_unlock(range);
                            let _ = resp_write.try_send(res);
                        }
                    },
                }
            },
            FileReqOp::WriteAbsorb => {
                let md = unsafe { req.body.write };
                let mut req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                let range = req.offset as u64..(req.offset + req.buf.len()) as u64;
                let _resp_write = resp.to_write();
                let resp_write = _resp_write.clone();
                let resp = FileResp::Write(_resp_write);
                let mut fetched = Vec::new();
                fetched.append(&mut req.fetched);
                let res = self.inner.absorb_write(req, resp, fetched).await;
                match res {
                    Ok(bytes) => {
                        let _ = resp_write.try_send(Ok(bytes));
                    },
                    Err(ref e) => {
                        if !super::is_requeued(e) {
                            #[cfg(feature = "range-lock")]
                            self.inner.range_lock.try_unlock(range);
                            let _ = resp_write.try_send(res);
                        }
                    },
                }
            },
            FileReqOp::WriteAbsorbBh => {
                let md = unsafe { req.body.write };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                let range = req.offset as u64..(req.offset + req.buf.len()) as u64;
                let _resp_write = resp.to_write();
                let resp_write = _resp_write.clone();
                let resp = FileResp::Write(_resp_write);
                let res = self.inner.absorb_write_bh(req, resp).await;
                match res {
                    // Handed on to fetch blocks a flush took; the retry
                    // answers the caller. The range lock and the permit
                    // stay with the request, which is why nothing is
                    // released here.
                    Err(ref e) if super::is_requeued(e) => {},
                    // A real failure ends the write here, so the range it
                    // held has to be given back. The success path releases
                    // it inside `absorb_write_bh`.
                    Err(_) => {
                        #[cfg(feature = "range-lock")]
                        self.inner.range_lock.try_unlock(range);
                        let _ = resp_write.try_send(res);
                    },
                    Ok(_) => { let _ = resp_write.try_send(res); },
                }
            },
            FileReqOp::WriteZero => {
                let md = unsafe { req.body.write_zero };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                let range = req.offset as u64..(req.offset + req.len) as u64;
                // prepare error response handler
                let _resp_write = resp.to_write_zero();
                let resp_write = _resp_write.clone();
                let resp = FileResp::WriteZero(_resp_write);
                let res = self.inner.spawn_write_zero(req, resp).await;
                match res {
                    Ok(bytes) => {
                        let _ = resp_write.try_send(Ok(bytes));
                    },
                    Err(ref e) => {
                        if !super::is_requeued(e) {
                            #[cfg(feature = "range-lock")]
                            self.inner.range_lock.try_unlock(range);
                            let _ = resp_write.try_send(res);
                        }
                    },
                }
            },
            #[cfg(feature = "wal")]
            FileReqOp::WriteZeroWal => {
                let md = unsafe { req.body.write_zero };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                let range = req.offset as u64..(req.offset + req.len) as u64;
                // prepare error response handler
                let _resp_write = resp.to_write_zero();
                let resp_write = _resp_write.clone();
                let resp = FileResp::WriteZero(_resp_write);
                let res = self.inner.spawn_write_zero_wal(req, resp).await;
                match res {
                    Ok(_) => {},
                    Err(ref e) => {
                        if !super::is_requeued(e) {
                            #[cfg(feature = "range-lock")]
                            self.inner.range_lock.try_unlock(range);
                            let _ = resp_write.try_send(res);
                        }
                    },
                }
            },
            FileReqOp::WriteZeroAbsorb => {
                let md = unsafe { req.body.write_zero };
                let mut req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                let range = req.offset as u64..(req.offset + req.len) as u64;
                let _resp_write = resp.to_write_zero();
                let resp_write = _resp_write.clone();
                let resp = FileResp::WriteZero(_resp_write);
                let mut fetched = Vec::new();
                fetched.append(&mut req.fetched);
                let res = self.inner.absorb_write_zero(req, resp, fetched).await;
                match res {
                    Ok(bytes) => {
                        let _ = resp_write.try_send(Ok(bytes));
                    },
                    Err(ref e) => {
                        if !super::is_requeued(e) {
                            #[cfg(feature = "range-lock")]
                            self.inner.range_lock.try_unlock(range);
                            let _ = resp_write.try_send(res);
                        }
                    },
                }
            },
            FileReqOp::WriteZeroAbsorbBh => {
                let md = unsafe { req.body.write_zero };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                let range = req.offset as u64..(req.offset + req.len) as u64;
                let _resp_write = resp.to_write_zero();
                let resp_write = _resp_write.clone();
                let resp = FileResp::WriteZero(_resp_write);
                let res = self.inner.absorb_write_zero_bh(req, resp).await;
                match res {
                    // See the WriteAbsorbBh arm: handed on for a refetch.
                    Err(ref e) if super::is_requeued(e) => {},
                    Err(_) => {
                        #[cfg(feature = "range-lock")]
                        self.inner.range_lock.try_unlock(range);
                        let _ = resp_write.try_send(res);
                    },
                    Ok(_) => { let _ = resp_write.try_send(res); },
                }
            },
            FileReqOp::WriteAlignedBatch => {
                let md = unsafe { req.body.write_aligned_batch };
                let req = ManuallyDrop::into_inner(md);
                // Take the permit before the blocks are consumed, so a
                // contention miss can put them back. Waiting here would
                // block the one task that can release the permit.
                let Ok(permit) = self.inner.try_lock() else {
                    let fh = req.fh.clone();
                    let ctx = FileContext::reform_write_aligned_batch(req.data_blocks, req.fh, resp);
                    let _ = fh.send_highprio(ctx);
                    return;
                };
                let res = self.inner.write_aligned_batch_locked(req.data_blocks, permit).await;
                let _ = resp.to_write().try_send(res);
            },
            FileReqOp::WriteBatch => {
                let md = unsafe { req.body.write_batch };
                let req = ManuallyDrop::into_inner(md);
                let Ok(permit) = self.inner.try_lock() else {
                    let fh = req.fh.clone();
                    let ctx = FileContext::reform_write_batch(req.data_blocks, req.fh, resp);
                    let _ = fh.send_highprio(ctx);
                    return;
                };
                let res = self.inner.write_batch_locked(req.data_blocks, permit).await;
                let _ = resp.to_write().try_send(res);
            },
            FileReqOp::Trunc => {
                let md = unsafe { req.body.trunc };
                let req = ManuallyDrop::into_inner(md);
                let offset = req.offset;
                let Ok(permit) = self.inner.try_lock() else {
                    let fh = req.fh.clone();
                    let ctx = FileContext::reform_trunc(offset, req.fh, resp);
                    let _ = fh.send_highprio(ctx);
                    return;
                };
                let res = self.inner.truncate_locked(offset, permit).await;
                let _ = resp.to_trunc().send(res);
            },
            FileReqOp::BlockAbsorb => {
                let md = unsafe { req.body.block_absorb };
                let r = ManuallyDrop::into_inner(md);
                self.inner.absorb_block(r.blk_idx, r.block);
            },
            FileReqOp::ReadMany => {
                let md = unsafe { req.body.read_many };
                let r = ManuallyDrop::into_inner(md);
                let _resp_rm = resp.to_read_many();
                let resp = FileResp::ReadMany(_resp_rm);
                if let Err(e) = self.inner.spawn_read_many(r, resp).await {
                    log::warn!("read many failed to start: {:?}", e);
                }
            },
            FileReqOp::WithBlocks => {
                let md = unsafe { req.body.with_blocks };
                let mut r = ManuallyDrop::into_inner(md);
                // Purely in-memory: every index is answered from the cache
                // or reported absent. Nothing here awaits, so one message
                // serves the whole batch without holding up the queue for
                // an object request. A caller that wants the blocks there
                // first warms them with `fh_read_ahead`.
                let mut resident = 0usize;
                {
                    let closed = *r.gate.lock().unwrap();
                    if !closed {
                        for blk_idx in r.indices.iter().copied() {
                            match self.inner.cached_block(blk_idx) {
                                Some(block) => {
                                    (r.action)(blk_idx, Some(block.as_slice()));
                                    block.unlock();
                                    resident += 1;
                                },
                                None => (r.action)(blk_idx, None),
                            }
                        }
                    }
                }
                let _ = resp.to_with_blocks().send(Ok(resident));
            },
            FileReqOp::Seek => {
                let md = unsafe { req.body.seek };
                let r = ManuallyDrop::into_inner(md);
                // A map walk, so it is answered from here: the map does not
                // leave this task.
                let res = match r.whence {
                    SeekWhence::Data => self.inner.seek_data(r.offset).await,
                    SeekWhence::Hole => self.inner.seek_hole(r.offset).await,
                };
                let _ = resp.to_seek().send(res);
            },
            FileReqOp::ReadPlan => {
                let md = unsafe { req.body.read_plan };
                let r = ManuallyDrop::into_inner(md);
                // Metadata only, so it is answered from here rather than
                // spawned: the walk needs the map, which does not leave
                // this task.
                let res = self.inner.read_plan_many(&r.ranges).await;
                let _ = resp.to_read_plan().send(res);
            },
            FileReqOp::BlockPlacement => {
                let md = unsafe { req.body.block_placement };
                let r = ManuallyDrop::into_inner(md);
                let res = self.inner.block_placement_many(&r.ranges).await;
                let _ = resp.to_block_placement().send(res);
            },
            FileReqOp::ReadAhead => {
                let md = unsafe { req.body.read_ahead };
                let r = ManuallyDrop::into_inner(md);
                let _resp_ra = resp.to_read_ahead();
                let resp = FileResp::ReadAhead(_resp_ra);
                // The response comes from the spawned fetch, so a failure
                // to even plan is the only thing answered here.
                if let Err(e) = self.inner.spawn_read_ahead(r, resp).await {
                    log::warn!("read ahead failed to start: {:?}", e);
                }
            },
            FileReqOp::DirtyBlockCount => {
                let md = unsafe { req.body.dirty_block_count };
                let _ = ManuallyDrop::into_inner(md);
                let n = self.inner.dirty_block_count();
                let _ = resp.to_dirty_block_count().send(Ok(n));
            },
            FileReqOp::Timing => {
                let md = unsafe { req.body.timing };
                let req = ManuallyDrop::into_inner(md);
                let v = match req.op {
                    TimingOp::Read => TimingValue::Read(self.inner.read_timing().snapshot()),
                    TimingOp::ReadReset => {
                        self.inner.read_timing_reset();
                        TimingValue::Reset
                    },
                    TimingOp::Flush => TimingValue::Flush(self.inner.flush_timing().snapshot()),
                    TimingOp::FlushReset => {
                        self.inner.flush_timing_reset();
                        TimingValue::Reset
                    },
                };
                let _ = resp.to_timing().send(Ok(v));
            },
            FileReqOp::WithBlock => {
                let md = unsafe { req.body.with_block };
                let req = ManuallyDrop::into_inner(md);
                let FileReqWithBlock { blk_idx, create, action, gate, fh } = req;
                // One handle travels in the rebuilt request, one is used
                // to send it.
                let ctx_fh = fh.clone();
                match action {
                    // Read-only: fetched off this task when it has to be
                    // fetched, so concurrent readers overlap.
                    // `spawn_block_ref` answers, including on error.
                    BlockAction::Ref(f) => {
                        let _ = self.inner.spawn_block_ref(blk_idx, f, gate, resp, fh).await;
                    },
                    // Mutable: dirties the block, installs a bmap
                    // placeholder and joins the next flush, so it stays
                    // on this task where those are serialized with
                    // everything else. The borrow lives only as long as
                    // the closure call.
                    BlockAction::Mut(f) => {
                        match self.inner.block_mut(blk_idx, create).await {
                            Ok(Some(mut block)) => {
                                let cancelled = gate.lock().unwrap_or_else(|e| e.into_inner());
                                if !*cancelled {
                                    f(block.as_mut_slice());
                                }
                                resp.answer_with_block(Ok(true));
                            },
                            Ok(None) => resp.answer_with_block(Ok(false)),
                            // The permit is held by a pipeline only this
                            // task can advance, so put the request back
                            // rather than wait or fail. `cb` outranks
                            // `highprio`, so that pipeline goes first.
                            Err(e) if super::is_requeued(&e) => {
                                let ctx = FileContext::reform_with_block(
                                    blk_idx, create, BlockAction::Mut(f), gate, fh, resp);
                                let _ = ctx_fh.send_highprio(ctx);
                            },
                            Err(e) => resp.answer_with_block(Err(e)),
                        }
                    },
                }
            },
            // single flush interface for extenral
            #[cfg(not(feature = "wal"))]
            FileReqOp::Flush => {
                let md = unsafe { req.body.flush };
                let _req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                if self.inner.range_lock.is_locked() {
                    // Hold off new range locks while we wait for the
                    // in-flight ones to drain, otherwise arriving reads
                    // and writes keep the map non-empty for good.
                    self.inner.state.set_flush_pending();
                    let fh = _req.fh.clone();
                    let ctx = FileContext::reform_flush(_req, resp);
                    if fh.send_cb(ctx).is_err() {
                        self.inner.state.clear_flush_pending();
                    }
                    return;
                }
                #[cfg(feature = "range-lock")]
                self.inner.state.clear_flush_pending();
                let res = self.inner.flush().await;
                let _ = resp.to_flush().send(res);
            },
            // POSIX-`fdatasync` flavoured flush. Under non-WAL,
            // `inner.flush_data` short-circuits when only attrs
            // are dirty.
            #[cfg(not(feature = "wal"))]
            FileReqOp::FlushData => {
                let md = unsafe { req.body.flush };
                let _req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                if self.inner.range_lock.is_locked() {
                    // See the note on the flush arm above.
                    self.inner.state.set_flush_pending();
                    let fh = _req.fh.clone();
                    let ctx = FileContext::reform_flush_data(_req, resp);
                    if fh.send_cb(ctx).is_err() {
                        self.inner.state.clear_flush_pending();
                    }
                    return;
                }
                #[cfg(feature = "range-lock")]
                self.inner.state.clear_flush_pending();
                let res = self.inner.flush_data().await;
                let _ = resp.to_flush().send(res);
            },
            #[cfg(feature = "wal")]
            FileReqOp::Flush => {
                let md = unsafe { req.body.flush };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                if self.inner.range_lock.is_locked() {
                    // See the note on the non-WAL flush arm: without this
                    // the drain never completes under steady traffic.
                    self.inner.state.set_flush_pending();
                    let fh = req.fh.clone();
                    let ctx = FileContext::reform_flush(req, resp);
                    // move flush op to cb queue
                    // so that flush op can run immediately after all inflight write op finished
                    if fh.send_cb(ctx).is_err() {
                        self.inner.state.clear_flush_pending();
                    }
                    return;
                }
                #[cfg(feature = "range-lock")]
                self.inner.state.clear_flush_pending();
                let res = if self.inner.wal.is_none() {
                    let res = self.inner.flush().await;
                    if res.is_err() {
                        warn!("kick flush failed {:?}", res);
                    }
                    res
                } else {
                    let res = self.inner.kick_wal_protected_flush_reactor(req.fh.clone()).await;
                    if res.is_err() {
                        debug!("kick wal flush failed {:?}, requeue this request", res);
                        let fh = req.fh.clone();
                        let ctx = FileContext::reform_flush(req, resp);
                        // move flush op to cb queue
                        // so that flush op can run immediately after all inflight write op finished
                        let _ = fh.send_cb(ctx);
                        return;
                    }
                    res
                };
                let _ = resp.to_flush().send(res);
            },
            // POSIX-`fdatasync` flavoured flush, WAL on. Same
            // routing as `Flush` except we short-circuit when
            // only attrs are dirty: the WAL kick path doesn't
            // itself check, so we'd otherwise pay an unnecessary
            // segment build for a metadata-only change.
            #[cfg(feature = "wal")]
            FileReqOp::FlushData => {
                let md = unsafe { req.body.flush };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                if self.inner.range_lock.is_locked() {
                    let fh = req.fh.clone();
                    let ctx = FileContext::reform_flush_data(req, resp);
                    let _ = fh.send_cb(ctx);
                    return;
                }
                // Short-circuit: only attr is dirty (or nothing
                // is dirty) — fdatasync is allowed to skip.
                if self.inner.dirty_block_count() == 0 && !self.inner.is_bmap_dirty() {
                    let cno = self.inner.in_memory_last_ondisk_cno();
                    let _ = resp.to_flush().send(Ok(SegmentId::new_from_cno(cno)));
                    return;
                }
                // Otherwise fdatasync collapses to fsync — we
                // need to write a segment for the dirty data /
                // bmap, which carries the inode along anyway.
                let res = if self.inner.wal.is_none() {
                    self.inner.flush().await
                } else {
                    let res = self.inner.kick_wal_protected_flush_reactor(req.fh.clone()).await;
                    if res.is_err() {
                        debug!("kick wal flush_data failed {:?}, requeue this request", res);
                        let fh = req.fh.clone();
                        let ctx = FileContext::reform_flush_data(req, resp);
                        let _ = fh.send_cb(ctx);
                        return;
                    }
                    res
                };
                let _ = resp.to_flush().send(res);
            },
            #[cfg(feature = "wal")]
            FileReqOp::WalFlush => {
                let md = unsafe { req.body.wal_flush };
                let req = ManuallyDrop::into_inner(md);
                #[cfg(feature = "range-lock")]
                if self.inner.range_lock.is_locked() {
                    let fh = req.fh.clone();
                    let ctx = FileContext::reform_wal_flush(req, resp);
                    // move flush op to cb queue
                    // so that flush op can run immediately after all inflight write op finished
                    let _ = fh.send_cb(ctx);
                    return;
                }
                let res = self.inner.kick_wal_protected_flush_reactor(req.fh).await;
                if res.is_err() {
                    debug!("kick wal flush failed {:?}, ignore this flush request", res);
                }
                let _ = resp.to_wal_flush();
            },
            #[cfg(feature = "wal")]
            FileReqOp::WalFlushDone => {
                let md = unsafe { req.body.wal_flush_done };
                let req = ManuallyDrop::into_inner(md);
                let lock = req.lock;
                let segid = req.segid;
                let od_state = req.od_state;
                let bmap_cache_limit = req.bmap_cache_limit;
                self.inner.wal_flush_done(lock, segid, od_state, bmap_cache_limit, req.base).await;
                info!("wal flush done, segid: {}", segid);
                let _ = resp.to_wal_flush_done();
            },
            #[cfg(feature = "wal")]
            FileReqOp::WalFlushRecovery => {
                let md = unsafe { req.body.wal_flush_recovery };
                let req = ManuallyDrop::into_inner(md);
                let lock = req.lock;
                // Nothing to answer: this arm was queued by a detached upload
                // that failed, so its caller was told long ago that the data
                // is durable — which it is, the wal holds it. Recovery having
                // failed means the file is now read-only, which
                // `wal_flush_recovery` has already recorded. Panicking here
                // would take the whole process with it, including reads that
                // are still being served correctly.
                if let Err(e) = self.inner.wal_flush_recovery(lock).await {
                    log::warn!("wal flush recovery failed, file is read-only: {}", e);
                }
                let _ = resp.to_wal_flush_recovery();
            },
            FileReqOp::Release => {
                let md = unsafe { req.body.release };
                let req = ManuallyDrop::into_inner(md);
                let res = self.inner.release().await;
                match res {
                    Ok(_) => {
                        let _ = resp.to_release().send(res);
                    },
                    Err(ref e) => {
                        if !super::is_requeued(e) {
                            let _ = resp.to_release().send(res);
                        } else {
                            let fh = req.fh.clone();
                            let ctx = FileContext::reform_release(req, resp);
                            let _ = fh.send_highprio(ctx);
                        }
                    },
                }
            },
            FileReqOp::LastCno => {
                let md = unsafe { req.body.last_cno };
                let _ = ManuallyDrop::into_inner(md);
                let res = self.inner.last_cno();
                let _ = resp.to_last_cno().send(res);
            },
            #[cfg(feature = "wal")]
            FileReqOp::BeginTxn => {
                let md = unsafe { req.body.begin_txn };
                let _ = ManuallyDrop::into_inner(md);
                let res = self.inner.begin_txn().await;
                let _ = resp.to_begin_txn().send(res);
            },
            #[cfg(feature = "wal")]
            FileReqOp::CommitTxn => {
                let md = unsafe { req.body.commit_txn };
                let req = ManuallyDrop::into_inner(md);
                let res = self.inner.commit_txn().await;
                match res {
                    // Committing publishes, and the flush lock may be travelling
                    // with a publish already in flight — see the `SetAttr` arm.
                    // Re-queue rather than wait, or the callback carrying the
                    // lock back can never be taken.
                    Err(ref e) if super::is_requeued(e) => {
                        let fh = req.fh.clone();
                        let ctx = FileContext::reform_commit_txn(req, resp);
                        let _ = fh.send_highprio(ctx);
                    },
                    _ => { let _ = resp.to_commit_txn().send(res); },
                }
            },
            #[cfg(feature = "wal")]
            FileReqOp::AbortTxn => {
                let md = unsafe { req.body.abort_txn };
                let req = ManuallyDrop::into_inner(md);
                let res = self.inner.abort_txn().await;
                match res {
                    Err(ref e) if super::is_requeued(e) => {
                        let fh = req.fh.clone();
                        let ctx = FileContext::reform_abort_txn(req, resp);
                        let _ = fh.send_highprio(ctx);
                    },
                    _ => { let _ = resp.to_abort_txn().send(res); },
                }
            },
            #[cfg(feature = "wal")]
            FileReqOp::InTxn => {
                let md = unsafe { req.body.in_txn };
                let _ = ManuallyDrop::into_inner(md);
                let res = self.inner.in_txn();
                let _ = resp.to_in_txn().send(res);
            },
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::mem::ManuallyDrop;
    use tokio::task::LocalSet;
    use hyperfile_reactor::Reactor;

    // Minimal Task impl to obtain a real ChannelGroup for tests.
    struct DummyTask;
    impl Task<FileContext<'static>> for DummyTask {
        async fn handle(&mut self, _ctx: FileContext<'static>) {}
    }

    async fn make_handler() -> ChannelGroup<FileContext<'static>> {
        // Tests construct FileReq structs that carry an `fh:
        // ChannelGroup` field, but never actually send through it.
        // Spawn a throwaway reactor + dummy task to obtain a live
        // ChannelGroup, then leak the reactor so the senders inside
        // the ChannelGroup stay valid for the lifetime of the test.
        let reactor = Reactor::<FileContext<'static>, DummyTask>::new_current()
            .expect("reactor");
        let (builder, finish) = build_channel_group();
        let handler = reactor.spawn_async(DummyTask, builder).await
            .expect("spawn");
        std::mem::forget(reactor);
        finish(handler)
    }

    // ==================== oneshot patterns ====================

    #[test]
    fn getattr_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let (ctx, rx) = FileContext::new_getattr();
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::GetAttr));
            let _body = ManuallyDrop::into_inner(unsafe { req.body.getattr });
            let stat: libc::stat = unsafe { std::mem::zeroed() };
            resp.to_getattr().send(Ok(stat)).unwrap();
            assert!(rx.await.unwrap().is_ok());
        });
    }

    #[test]
    fn setattr_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let stat_in: libc::stat = unsafe { std::mem::zeroed() };
            let fh = make_handler().await;
            let (ctx, rx) = FileContext::new_setattr(fh, stat_in);
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::SetAttr));
            let body = ManuallyDrop::into_inner(unsafe { req.body.setattr });
            assert_eq!(body.stat.st_size, 0);
            let stat_out: libc::stat = unsafe { std::mem::zeroed() };
            resp.to_setattr().send(Ok(stat_out)).unwrap();
            assert!(rx.await.unwrap().is_ok());
        });
    }

    #[test]
    fn trunc_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, rx) = FileContext::new_trunc(4096, fh);
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::Trunc));
            let body = ManuallyDrop::into_inner(unsafe { req.body.trunc });
            assert_eq!(body.offset, 4096);
            resp.to_trunc().send(Ok(())).unwrap();
            assert!(rx.await.unwrap().is_ok());
        });
    }

    #[test]
    fn last_cno_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let (ctx, rx) = FileContext::new_last_cno();
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::LastCno));
            let _body = ManuallyDrop::into_inner(unsafe { req.body.last_cno });
            resp.to_last_cno().send(42).unwrap();
            assert_eq!(rx.await.unwrap(), 42);
        });
    }

    // ==================== mpsc patterns ====================

    #[test]
    fn read_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let mut buf = vec![0u8; 100];
            let buf_ref = unsafe { std::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len()) };
            let (ctx, _tx, mut rx) = FileContext::new_read(buf_ref, 0, fh);
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::Read));
            let body = ManuallyDrop::into_inner(unsafe { req.body.read });
            assert_eq!(body.buf.len(), 100);
            assert_eq!(body.offset, 0);
            resp.to_read().send(Ok(100)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 100);
        });
    }

    #[test]
    fn write_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let data = [0xABu8; 200];
            let data_ref = unsafe { std::slice::from_raw_parts(data.as_ptr(), data.len()) };
            let (ctx, _tx, mut rx) = FileContext::new_write(data_ref, 512, fh);
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::Write));
            let body = ManuallyDrop::into_inner(unsafe { req.body.write });
            assert_eq!(body.buf.len(), 200);
            assert_eq!(body.buf[0], 0xAB);
            assert_eq!(body.offset, 512);
            resp.to_write().send(Ok(200)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 200);
        });
    }

    #[test]
    fn write_zero_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, _tx, mut rx) = FileContext::new_write_zero(1024, 4096, fh);
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::WriteZero));
            let body = ManuallyDrop::into_inner(unsafe { req.body.write_zero });
            assert_eq!(body.offset, 1024);
            assert_eq!(body.len, 4096);
            resp.to_write_zero().send(Ok(4096)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 4096);
        });
    }

    // ==================== flush / release (oneshot + TaskHandler) ====================

    #[test]
    fn flush_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, rx) = FileContext::new_flush(fh);
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::Flush));
            let _body = ManuallyDrop::into_inner(unsafe { req.body.flush });
            resp.to_flush().send(Ok(SegmentId::new(7))).unwrap();
            assert_eq!(rx.await.unwrap().unwrap(), SegmentId::new(7));
        });
    }

    #[test]
    fn release_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, rx) = FileContext::new_release(fh);
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::Release));
            let _body = ManuallyDrop::into_inner(unsafe { req.body.release });
            resp.to_release().send(Ok(SegmentId::new(99))).unwrap();
            assert_eq!(rx.await.unwrap().unwrap(), SegmentId::new(99));
        });
    }

    // ==================== batch patterns ====================

    #[test]
    fn write_aligned_batch_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let blocks = vec![
                AlignedDataBlockWrapper::new(0, 4096, false),
                AlignedDataBlockWrapper::new(1, 4096, true),
            ];
            let fh = make_handler().await;
            let (ctx, mut rx) = FileContext::new_write_aligned_batch(blocks, fh);
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::WriteAlignedBatch));
            let body = ManuallyDrop::into_inner(unsafe { req.body.write_aligned_batch });
            assert_eq!(body.data_blocks.len(), 2);
            assert!(!body.data_blocks[0].is_zero());
            assert!(body.data_blocks[1].is_zero());
            resp.to_write().send(Ok(8192)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 8192);
        });
    }

    #[test]
    fn write_batch_construct_take_respond() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let blocks = vec![BatchDataBlockWrapper::new(0, 4096, false)];
            let fh = make_handler().await;
            let (ctx, mut rx) = FileContext::new_write_batch(blocks, fh);
            let (req, resp) = ctx.take();
            assert!(matches!(req.op, FileReqOp::WriteBatch));
            let body = ManuallyDrop::into_inner(unsafe { req.body.write_batch });
            assert_eq!(body.data_blocks.len(), 1);
            resp.to_write().send(Ok(4096)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 4096);
        });
    }

    // ==================== clone_write_resp / clone_write_zero_resp ====================

    #[test]
    fn clone_write_resp_sends_on_cloned_channel() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            static BUF: [u8; 10] = [0u8; 10];
            let (ctx, _tx, mut rx) = FileContext::new_write(&BUF, 0, fh);
            let (_req, resp) = ctx.take();
            let cloned = resp.clone_write_resp();
            cloned.send(Ok(10)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 10);
        });
    }

    #[test]
    fn clone_write_zero_resp_sends_on_cloned_channel() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, _tx, mut rx) = FileContext::new_write_zero(0, 100, fh);
            let (_req, resp) = ctx.take();
            let cloned = resp.clone_write_zero_resp();
            cloned.send(Ok(100)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 100);
        });
    }

    // ==================== reform_* constructors ====================

    #[test]
    fn reform_read_preserves_data() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let mut buf = vec![0u8; 50];
            let buf_ref = unsafe { std::slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len()) };
            let (ctx, _tx, mut rx) = FileContext::new_read(buf_ref, 10, fh);
            let (req, resp) = ctx.take();
            let body = ManuallyDrop::into_inner(unsafe { req.body.read });
            let ctx2 = FileContext::reform_read(body, resp);
            let (req2, resp2) = ctx2.take();
            assert!(matches!(req2.op, FileReqOp::Read));
            let body2 = ManuallyDrop::into_inner(unsafe { req2.body.read });
            assert_eq!(body2.offset, 10);
            assert_eq!(body2.buf.len(), 50);
            resp2.to_read().send(Ok(50)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 50);
        });
    }

    #[test]
    fn reform_write_preserves_data() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let data = [1u8; 30];
            let data_ref = unsafe { std::slice::from_raw_parts(data.as_ptr(), data.len()) };
            let (ctx, _tx, mut rx) = FileContext::new_write(data_ref, 20, fh);
            let (req, resp) = ctx.take();
            let body = ManuallyDrop::into_inner(unsafe { req.body.write });
            let ctx2 = FileContext::reform_write(body, resp);
            let (req2, resp2) = ctx2.take();
            assert!(matches!(req2.op, FileReqOp::Write));
            let body2 = ManuallyDrop::into_inner(unsafe { req2.body.write });
            assert_eq!(body2.offset, 20);
            assert_eq!(body2.buf[0], 1);
            resp2.to_write().send(Ok(30)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 30);
        });
    }

    #[test]
    fn reform_write_zero_preserves_data() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, _tx, mut rx) = FileContext::new_write_zero(100, 200, fh);
            let (req, resp) = ctx.take();
            let body = ManuallyDrop::into_inner(unsafe { req.body.write_zero });
            let ctx2 = FileContext::reform_write_zero(body, resp);
            let (req2, resp2) = ctx2.take();
            assert!(matches!(req2.op, FileReqOp::WriteZero));
            let body2 = ManuallyDrop::into_inner(unsafe { req2.body.write_zero });
            assert_eq!(body2.offset, 100);
            assert_eq!(body2.len, 200);
            resp2.to_write_zero().send(Ok(200)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 200);
        });
    }

    #[test]
    fn reform_flush_preserves_channel() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, rx) = FileContext::new_flush(fh);
            let (req, resp) = ctx.take();
            let body = ManuallyDrop::into_inner(unsafe { req.body.flush });
            let ctx2 = FileContext::reform_flush(body, resp);
            let (req2, resp2) = ctx2.take();
            assert!(matches!(req2.op, FileReqOp::Flush));
            let _body2 = ManuallyDrop::into_inner(unsafe { req2.body.flush });
            resp2.to_flush().send(Ok(SegmentId::new(5))).unwrap();
            assert_eq!(rx.await.unwrap().unwrap(), SegmentId::new(5));
        });
    }

    #[test]
    fn reform_release_preserves_channel() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, rx) = FileContext::new_release(fh);
            let (req, resp) = ctx.take();
            let body = ManuallyDrop::into_inner(unsafe { req.body.release });
            let ctx2 = FileContext::reform_release(body, resp);
            let (req2, resp2) = ctx2.take();
            assert!(matches!(req2.op, FileReqOp::Release));
            let _body2 = ManuallyDrop::into_inner(unsafe { req2.body.release });
            resp2.to_release().send(Ok(SegmentId::new(11))).unwrap();
            assert_eq!(rx.await.unwrap().unwrap(), SegmentId::new(11));
        });
    }

    // ==================== absorb_* constructors ====================

    #[test]
    fn write_absorb_sets_correct_op() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            static BUF: [u8; 10] = [0u8; 10];
            let (ctx, _tx, mut rx) = FileContext::new_write(&BUF, 0, fh);
            let (req, resp) = ctx.take();
            let body = ManuallyDrop::into_inner(unsafe { req.body.write });
            let ctx2 = FileContext::write_absorb(body, resp);
            let (req2, resp2) = ctx2.take();
            assert!(matches!(req2.op, FileReqOp::WriteAbsorb));
            let _body2 = ManuallyDrop::into_inner(unsafe { req2.body.write });
            resp2.to_write().send(Ok(10)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 10);
        });
    }

    #[test]
    fn write_absorb_bh_sets_correct_op() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            static BUF: [u8; 10] = [0u8; 10];
            let (ctx, _tx, _rx) = FileContext::new_write(&BUF, 0, fh);
            let (req, resp) = ctx.take();
            let body = ManuallyDrop::into_inner(unsafe { req.body.write });
            let ctx2 = FileContext::write_absorb_bh(body, resp);
            let (req2, _resp2) = ctx2.take();
            assert!(matches!(req2.op, FileReqOp::WriteAbsorbBh));
        });
    }

    #[test]
    fn write_zero_absorb_sets_correct_op() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, _tx, mut rx) = FileContext::new_write_zero(0, 100, fh);
            let (req, resp) = ctx.take();
            let body = ManuallyDrop::into_inner(unsafe { req.body.write_zero });
            let ctx2 = FileContext::write_zero_absorb(body, resp);
            let (req2, resp2) = ctx2.take();
            assert!(matches!(req2.op, FileReqOp::WriteZeroAbsorb));
            let _body2 = ManuallyDrop::into_inner(unsafe { req2.body.write_zero });
            resp2.to_write_zero().send(Ok(100)).await.unwrap();
            assert_eq!(rx.recv().await.unwrap().unwrap(), 100);
        });
    }

    #[test]
    fn write_zero_absorb_bh_sets_correct_op() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            let (ctx, _tx, _rx) = FileContext::new_write_zero(0, 100, fh);
            let (req, resp) = ctx.take();
            let body = ManuallyDrop::into_inner(unsafe { req.body.write_zero });
            let ctx2 = FileContext::write_zero_absorb_bh(body, resp);
            let (req2, _resp2) = ctx2.take();
            assert!(matches!(req2.op, FileReqOp::WriteZeroAbsorbBh));
        });
    }

    // ==================== error path ====================

    #[test]
    fn getattr_error_response() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let (ctx, rx) = FileContext::new_getattr();
            let (_req, resp) = ctx.take();
            resp.to_getattr().send(Err(std::io::Error::new(ErrorKind::NotFound, "not found"))).unwrap();
            let result = rx.await.unwrap();
            assert!(result.is_err());
            assert_eq!(result.unwrap_err().kind(), ErrorKind::NotFound);
        });
    }

    #[test]
    fn write_error_response() {
        let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
        let local = LocalSet::new();
        local.block_on(&rt, async {
            let fh = make_handler().await;
            static BUF: [u8; 1] = [0u8; 1];
            let (ctx, _tx, mut rx) = FileContext::new_write(&BUF, 0, fh);
            let (_req, resp) = ctx.take();
            resp.to_write().send(Err(std::io::Error::new(ErrorKind::Other, "fail"))).await.unwrap();
            let result = rx.recv().await.unwrap();
            assert!(result.is_err());
        });
    }
}