hyperfile 0.5.0

use std::io::Result;
use aws_sdk_s3::Client;
use hyperfile_reactor::Reactor;
use tokio::sync::oneshot;
use crate::BlockIndex;
use crate::file::handler::{ChannelGroup, build_channel_group, BlockAction};
use crate::config::{HyperFileMetaConfig, HyperFileRuntimeConfig};
use crate::buffer::{AlignedDataBlockWrapper, BatchDataBlockWrapper};
use crate::staging::{s3::S3Staging, StagingIntercept};
use super::hyper::Hyper;
use super::flags::FileFlags;
use super::mode::FileMode;
use super::handler::FileContext;

/// Reactor-mode handle to a hyperfile-backed file.
///
/// Wraps a [`hyperfile_reactor::Reactor`]-spawned task that
/// owns the underlying [`Hyper`] and serializes operations
/// through a request/response channel. Cloning is cheap (just
/// duplicates channel senders); use [`Self::clone`] to share a
/// handle across tasks. The reactor task winds down when every
/// `HyperFileHandler` clone (and any sibling `HyperFileTokio`)
/// has been dropped.
///
/// # Security: hyperfile does not enforce POSIX permissions
///
/// `fh_chmod` / `fh_chown` / `fh_setattr` record the
/// `mode` / `uid` / `gid` you give them, and `fh_getattr` reads
/// them back, but the read/write/truncate paths **do not check
/// the bits**. There is no `PermissionDenied` / `EACCES` path.
/// Permission enforcement is the caller's responsibility (FUSE
/// kernel checks, app-level IAM, etc.); see `docs/posix.md`'s
/// "Permissions and ownership" section.
#[derive(Clone)]
pub struct HyperFileHandler<'a> {
    inner: ChannelGroup<FileContext<'a>>,
}

impl<'a: 'static> HyperFileHandler<'a> {
    /// Wrap a pre-constructed `Hyper` with a handler task. Useful when
    /// the caller needs full `HyperFileConfig` control (e.g. WAL
    /// configuration) that isn't surfaced by the other `fh_*`
    /// constructors.
    pub async fn fh_from_hyper(reactor: &Reactor<FileContext<'a>, Hyper<'a>>, hyper: Hyper<'a>) -> Result<Self>
    {
        let (builder, finish) = build_channel_group();
        let handler = reactor.spawn_async(hyper, builder).await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor thread is gone"))?;
        Ok(Self { inner: finish(handler) })
    }

    pub async fn fh_create(reactor: &Reactor<FileContext<'a>, Hyper<'a>>, client: &Client, uri: &str, flags: FileFlags, mode: FileMode) -> Result<Self>
    {
        let hyper = Hyper::fs_create(client, uri, flags, mode).await?;
        let (builder, finish) = build_channel_group();
        let handler = reactor.spawn_async(hyper, builder).await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor thread is gone"))?;
        Ok(Self { inner: finish(handler) })
    }

    pub async fn fh_create_with(reactor: &Reactor<FileContext<'a>, Hyper<'a>>, client: &Client, uri: &str, flags: FileFlags, mode: FileMode, interceptor: impl StagingIntercept<S3Staging> + 'static) -> Result<Self>
    {
        let hyper = Hyper::fs_create_with_interceptor(client, uri, flags, mode, interceptor).await?;
        let (builder, finish) = build_channel_group();
        let handler = reactor.spawn_async(hyper, builder).await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor thread is gone"))?;
        Ok(Self { inner: finish(handler) })
    }

    pub async fn fh_create_opt(reactor: &Reactor<FileContext<'a>, Hyper<'a>>, client: &Client, uri: &str, flags: FileFlags, mode: FileMode,
            meta_config: &HyperFileMetaConfig, runtime_config: &HyperFileRuntimeConfig) -> Result<Self>
    {
        let hyper = Hyper::fs_create_opt(client, uri, flags, mode, meta_config, runtime_config).await?;
        let (builder, finish) = build_channel_group();
        let handler = reactor.spawn_async(hyper, builder).await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor thread is gone"))?;
        Ok(Self { inner: finish(handler) })
    }

    pub async fn fh_create_opt_with_interceptor(reactor: &Reactor<FileContext<'a>, Hyper<'a>>,
            client: &Client, uri: &str, flags: FileFlags, mode: FileMode,
            meta_config: &HyperFileMetaConfig, runtime_config: &HyperFileRuntimeConfig,
            interceptor: impl StagingIntercept<S3Staging> + 'static) -> Result<Self>
    {
        let hyper = Hyper::fs_create_opt_with_interceptor(client, uri, flags, mode, meta_config, runtime_config, interceptor).await?;
        let (builder, finish) = build_channel_group();
        let handler = reactor.spawn_async(hyper, builder).await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor thread is gone"))?;
        Ok(Self { inner: finish(handler) })
    }

    pub async fn fh_open(reactor: &Reactor<FileContext<'a>, Hyper<'a>>, client: &Client, uri: &str, flags: FileFlags) -> Result<Self>
    {
        let hyper = Hyper::fs_open(client, uri, flags).await?;
        let (builder, finish) = build_channel_group();
        let handler = reactor.spawn_async(hyper, builder).await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor thread is gone"))?;
        Ok(Self { inner: finish(handler) })
    }

    pub async fn fh_open_opt(reactor: &Reactor<FileContext<'a>, Hyper<'a>>, client: &Client, uri: &str, flags: FileFlags,
            runtime_config: &HyperFileRuntimeConfig) -> Result<Self>
    {
        let hyper = Hyper::fs_open_opt(client, uri, flags, runtime_config).await?;
        let (builder, finish) = build_channel_group();
        let handler = reactor.spawn_async(hyper, builder).await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor thread is gone"))?;
        Ok(Self { inner: finish(handler) })
    }

    pub async fn fh_open_or_create_with_default_opt(reactor: &Reactor<FileContext<'a>, Hyper<'a>>, client: &Client, uri: &str, flags: FileFlags, mode: FileMode) -> Result<Self>
    {
        let hyper = Hyper::fs_open_or_create_with_default_opt(client, uri, flags, mode).await?;
        let (builder, finish) = build_channel_group();
        let handler = reactor.spawn_async(hyper, builder).await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor thread is gone"))?;
        Ok(Self { inner: finish(handler) })
    }

    pub async fn fh_unlink(client: &Client, uri: &str) -> Result<()>
    {
        Hyper::fs_unlink(client, uri).await
    }

    /// Rename the file at `src_uri` to `dst_uri`.
    ///
    /// **Not yet implemented.** Delegates to
    /// [`Hyper::fs_rename`], which currently returns
    /// [`std::io::ErrorKind::Unsupported`]. See that method's
    /// docs (and `docs/posix.md`) for the design issues blocking
    /// implementation.
    pub async fn fh_rename(client: &Client, src_uri: &str, dst_uri: &str) -> Result<()>
    {
        Hyper::fs_rename(client, src_uri, dst_uri).await
    }

    pub async fn fh_release(&mut self) -> Result<u64>
    {
        let (ctx, rx) = FileContext::new_release(self.inner.clone());
        self.inner.send(ctx)?;
        rx.await.map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?
    }

    pub async fn fh_read(&mut self, off: usize, buf: &mut [u8]) -> Result<usize>
    {
        let b = unsafe {
            std::slice::from_raw_parts_mut(buf.as_ptr() as *mut u8, buf.len())
        };
        let (ctx, tx, mut rx) = FileContext::new_read(b, off, self.inner.clone());
        self.inner.send(ctx)?;
        let res = rx.recv().await.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?;
        drop(tx);
        let _ = buf;
        res
    }

    pub async fn fh_write(&mut self, off: usize, buf: &[u8]) -> Result<usize>
    {
        let b = unsafe {
            std::slice::from_raw_parts(buf.as_ptr() as *const u8, buf.len())
        };
        let (ctx, tx, mut rx) = FileContext::new_write(b, off, self.inner.clone());
        self.inner.send(ctx)?;
        let res = rx.recv().await.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?;
        drop(tx);
        let _ = buf;
        res
    }

    pub async fn fh_write_zero(&mut self, off: usize, len: usize) -> Result<usize>
    {
        let (ctx, tx, mut rx) = FileContext::new_write_zero(off, len, self.inner.clone());
        self.inner.send(ctx)?;
        let res = rx.recv().await.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?;
        drop(tx);
        res
    }

    pub async fn fh_write_aligned_batch(&mut self, blocks: Vec<AlignedDataBlockWrapper>) -> Result<usize>
    {
        let (ctx, mut rx) = FileContext::new_write_aligned_batch(blocks);
        self.inner.send(ctx)?;
        rx.recv().await.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?
    }

    pub async fn fh_write_batch(&mut self, blocks: Vec<BatchDataBlockWrapper>) -> Result<usize>
    {
        let (ctx, mut rx) = FileContext::new_write_batch(blocks);
        self.inner.send(ctx)?;
        rx.recv().await.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?
    }

    pub async fn fh_flush(&mut self) -> Result<u64>
    {
        let (ctx, rx) = FileContext::new_flush(self.inner.clone());
        self.inner.send(ctx)?;
        rx.await.map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?
    }

    /// POSIX-`fdatasync` flavoured flush. See
    /// `Hyper::fs_fdatasync` for semantics. Skips the segment
    /// write entirely when only attrs are dirty.
    pub async fn fh_fdatasync(&mut self) -> Result<u64>
    {
        let (ctx, rx) = FileContext::new_flush_data(self.inner.clone());
        self.inner.send(ctx)?;
        rx.await.map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?
    }

    pub async fn fh_truncate(&mut self, offset: usize) -> Result<()>
    {
        let (ctx, rx) = FileContext::new_trunc(offset);
        self.inner.send(ctx)?;
        rx.await.map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?
    }

    /// Run `f` against block `idx` for reading, and return what it
    /// produced. `Ok(None)` when the block is not backed by data, in
    /// which case `f` is not called.
    ///
    /// # Why a closure and not a guard
    ///
    /// The direct API hands out a borrow — `Hyper::fs_block` returns a
    /// `BlockRef` — but the reactor surface cannot. The `Hyper` lives
    /// inside the reactor task, so a borrow would reach the caller
    /// only once the response arrives, and from that moment the
    /// reactor is free to serve the next request from this handle or
    /// any clone of it. A write, flush, truncate or cache eviction
    /// would then invalidate a borrow the caller still holds; blocks
    /// in the local-disk cache additionally have their backing space
    /// punched when evicted. There is no lifetime that expresses
    /// "valid until the next message", which is why `fh_read` is
    /// sound only for as long as its caller is parked awaiting the
    /// response.
    ///
    /// So the action travels to the block. `f` runs inside the reactor
    /// task while it holds the real borrow, and only owned values
    /// cross the channel — hence the `Send + 'static` bounds. Move
    /// what `f` needs into it and return what the caller needs out.
    ///
    /// The slice is always `data_block_size` bytes.
    pub async fn fh_with_block<R, F>(&mut self, idx: BlockIndex, f: F) -> Result<Option<R>>
    where
        F: FnOnce(&[u8]) -> R + Send + 'static,
        R: Send + 'static,
    {
        let (rtx, rrx) = oneshot::channel::<R>();
        let action = BlockAction::Ref(Box::new(move |buf: &[u8]| {
            let _ = rtx.send(f(buf));
        }));
        self.dispatch_with_block(idx, false, action, rrx).await
    }

    /// Run `f` against block `idx` for modification, and return what
    /// it produced. `Ok(None)` when the block is not backed by data
    /// and `create` is false, in which case `f` is not called.
    ///
    /// Writes through the slice land in the buffer the next flush will
    /// write out; no write-back call is needed. `create` materializes
    /// a zero-filled block for an index with no data. `i_size` is not
    /// changed. See [`Hyper::fs_block_mut`] for the full semantics and
    /// [`Self::fh_with_block`] for why this takes a closure rather
    /// than returning a guard.
    ///
    /// [`Hyper::fs_block_mut`]: crate::file::hyper::Hyper::fs_block_mut
    pub async fn fh_with_block_mut<R, F>(&mut self, idx: BlockIndex, create: bool, f: F) -> Result<Option<R>>
    where
        F: FnOnce(&mut [u8]) -> R + Send + 'static,
        R: Send + 'static,
    {
        let (rtx, rrx) = oneshot::channel::<R>();
        let action = BlockAction::Mut(Box::new(move |buf: &mut [u8]| {
            let _ = rtx.send(f(buf));
        }));
        self.dispatch_with_block(idx, create, action, rrx).await
    }

    /// Shared tail of the two `fh_with_block*` entry points: send the
    /// action, learn whether it ran, and if it did collect its result.
    async fn dispatch_with_block<R>(
        &mut self,
        idx: BlockIndex,
        create: bool,
        action: BlockAction,
        rrx: oneshot::Receiver<R>,
    ) -> Result<Option<R>>
    {
        let (ctx, rx) = FileContext::new_with_block(idx, create, action);
        self.inner.send(ctx)?;
        let ran = rx.await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))??;
        if !ran {
            return Ok(None);
        }
        // The action ran to completion before the reactor answered, so
        // its value is already queued. A closure that panicked would
        // have taken the reactor task with it, surfacing as the send
        // error above rather than here.
        let value = rrx.await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe,
                "block action did not report a result"))?;
        Ok(Some(value))
    }

    pub async fn fh_getattr(&self) -> Result<libc::stat>
    {
        let (ctx, rx) = FileContext::new_getattr();
        self.inner.send(ctx)?;
        rx.await.map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?
    }

    pub async fn fh_setattr(&self, stat: libc::stat) -> Result<libc::stat>
    {
        let (ctx, rx) = FileContext::new_setattr(stat);
        self.inner.send(ctx)?;
        rx.await.map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))?
    }

    pub async fn fh_last_cno(&self) -> Result<u64>
    {
        let (ctx, rx) = FileContext::new_last_cno();
        self.inner.send(ctx)?;
        rx.await.map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "reactor handler task died"))
    }
}