hyperfile 0.11.2

pub mod file;
pub mod flags;
pub mod mode;
#[cfg(feature = "reactor")]
pub mod handler;
#[cfg(feature = "reactor")]
pub mod reactor;
pub mod hyper;
pub mod fs;
#[cfg(feature = "reactor")]
pub mod fh;
#[cfg(feature = "reactor")]
pub mod tokio_wrapper;
pub mod block;
pub mod read_plan;
pub use read_plan::{ReadPolicy, ReadPlan, PlannedGet, GetTarget};
#[cfg(feature = "range-lock")]
pub mod lock;
mod state;

use std::io::{Error, ErrorKind, Result};
use std::time::{Instant, Duration};
use std::sync::atomic::{AtomicU64, Ordering};
use std::collections::BTreeMap;
#[cfg(feature = "wal")]
use std::sync::Weak;
#[cfg(feature = "wal")]
use std::pin::Pin;
use log::{info, debug, warn};
use tokio::sync::{OwnedSemaphorePermit, OwnedMutexGuard};
use btree_ondisk::{bmap::BMap, BlockLoader, NodeValue, NodeCache};
use btree_ondisk::btree::BtreeNodeDirty;
#[cfg(all(feature = "wal", feature = "reactor"))]
use crate::file::handler::ChannelGroup;
use crate::Cno;
#[cfg(all(feature = "wal", feature = "reactor"))]
use crate::file::handler::FileContext;
use crate::*;
use crate::buffer::DataBlock;
use crate::{BlockIndex, BlockPtr};
use crate::{SegmentId, SegmentOffset};
use crate::config::{HyperFileConfig, HyperFileMetaConfig};
use crate::ondisk::{BMapRawType, InodeRaw, SEGMENT_FLAG_PARTIAL};
use crate::inode::{Inode, OnDiskState, FlushInodeFlag};
use crate::staging::Staging;

/// Per-phase cumulative nanosecond counters for the flush path.
///
/// Instrumented from inside `flush_process`. Each phase's elapsed
/// time is added to the matching counter on every flush; the
/// `flush_count` counter is incremented once per completed flush.
///
/// Read externally via `HyperFile::flush_timing()`; reset via
/// `HyperFile::flush_timing_reset()`. Both accessors are
/// `#[doc(hidden)]` — the instrumentation is intended for
/// benchmarks and regression profiling, not part of the public
/// behavioral contract.
///
/// The WAL flush path is not yet instrumented; running this
/// against a WAL-enabled file will return zero counters.
#[derive(Default, Debug)]
pub struct FlushTiming {
    pub pre_build_ns: AtomicU64,
    pub build_segment_ns: AtomicU64,
    pub segment_done_ns: AtomicU64,
    pub flush_inode_ns: AtomicU64,
    pub cleanup_ns: AtomicU64,
    pub flush_count: AtomicU64,
    /// Under the stream format: data bytes sent ahead of a flush.
    pub stream_sent_bytes: AtomicU64,
    /// Of those, the bytes that died before their object counted: the block was
    /// written again in flight, or its object failed. Sent again later.
    pub stream_dead_bytes: AtomicU64,
    /// Index-node bytes written into objects sent ahead.
    pub stream_meta_bytes: AtomicU64,
    /// Under the stream format: data bytes the flushes themselves wrote — what
    /// was not sent ahead, or was and changed.
    pub stream_flush_bytes: AtomicU64,
}

/// Snapshot of `FlushTiming` values at a single point in time.
#[derive(Clone, Copy, Debug, Default)]
pub struct FlushTimingSnapshot {
    pub pre_build_ns: u64,
    pub build_segment_ns: u64,
    pub segment_done_ns: u64,
    pub flush_inode_ns: u64,
    pub cleanup_ns: u64,
    pub flush_count: u64,
    pub stream_sent_bytes: u64,
    pub stream_dead_bytes: u64,
    pub stream_meta_bytes: u64,
    pub stream_flush_bytes: u64,
}

impl FlushTiming {
    pub fn snapshot(&self) -> FlushTimingSnapshot {
        FlushTimingSnapshot {
            pre_build_ns: self.pre_build_ns.load(Ordering::Relaxed),
            build_segment_ns: self.build_segment_ns.load(Ordering::Relaxed),
            segment_done_ns: self.segment_done_ns.load(Ordering::Relaxed),
            flush_inode_ns: self.flush_inode_ns.load(Ordering::Relaxed),
            cleanup_ns: self.cleanup_ns.load(Ordering::Relaxed),
            flush_count: self.flush_count.load(Ordering::Relaxed),
            stream_sent_bytes: self.stream_sent_bytes.load(Ordering::Relaxed),
            stream_dead_bytes: self.stream_dead_bytes.load(Ordering::Relaxed),
            stream_meta_bytes: self.stream_meta_bytes.load(Ordering::Relaxed),
            stream_flush_bytes: self.stream_flush_bytes.load(Ordering::Relaxed),
        }
    }

    pub fn reset(&self) {
        self.pre_build_ns.store(0, Ordering::Relaxed);
        self.build_segment_ns.store(0, Ordering::Relaxed);
        self.segment_done_ns.store(0, Ordering::Relaxed);
        self.flush_inode_ns.store(0, Ordering::Relaxed);
        self.cleanup_ns.store(0, Ordering::Relaxed);
        self.flush_count.store(0, Ordering::Relaxed);
        self.stream_sent_bytes.store(0, Ordering::Relaxed);
        self.stream_dead_bytes.store(0, Ordering::Relaxed);
        self.stream_meta_bytes.store(0, Ordering::Relaxed);
        self.stream_flush_bytes.store(0, Ordering::Relaxed);
    }
}

/// One entry of a read plan: what reading part of a range would cost.
///
/// See [`HyperFile::read_plan`](crate::file::file::HyperFile::read_plan).
/// Entries account for the whole range asked about, in the order the
/// requests would be made, so a range that needs nothing fetched still
/// appears — as [`Self::Local`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PlannedRead {
    /// One object request, covering `[off, off + len)` of the file, served
    /// from `len` bytes at `at` in segment `segid`.
    ///
    /// May cover many consecutive blocks: a request is extended while the
    /// next block's location is adjacent in the same object and the
    /// request stays under `read_get_max_bytes`.
    Get {
        off: u64,
        len: u64,
        /// Which object this request hits — the checkpoint, and which of its
        /// pieces when it was streamed as several.
        ///
        /// The whole identity, not just the checkpoint: a streamed checkpoint has
        /// several places for a run of blocks to break, and two requests that look
        /// adjacent by `at` alone are two requests when their objects differ.
        segid: crate::SegmentId,
        at: u64,
    },
    /// `[off, off + len)` needs no object request: a hole, a block in the
    /// data cache, or — with `wal` — a block in a segment still pinned in
    /// memory.
    Local { off: u64, len: u64 },
}

impl PlannedRead {
    /// The file range this entry covers.
    pub fn range(&self) -> (u64, u64) {
        match self {
            Self::Get { off, len, .. } => (*off, *len),
            Self::Local { off, len } => (*off, *len),
        }
    }

    /// Whether this entry is an object request.
    pub fn is_get(&self) -> bool {
        matches!(self, Self::Get { .. })
    }
}

/// Read-side counters, the counterpart of [`FlushTiming`].
///
/// The read path's cost is dominated by object-store round trips, and
/// there was previously no way to count them: a caller can time a read
/// but cannot tell one request for a coalesced range from many, or a
/// cache hit from a fetch. That matters most for callers that assert on
/// round-trip counts rather than on wall time, since wall time varies
/// with the network.
///
/// Requests are split by what they fetch, because the three answer
/// different questions:
///
/// * data — how well block reads coalesce into ranged requests;
/// * meta — how many index nodes a lookup had to descend through;
/// * inode — one per cold open, so mostly a constant.
///
/// Counters are shared by every clone of a file's staging handle, so
/// they cover one file. They are never reset implicitly; call
/// [`Self::reset`] to bracket a measurement.
#[derive(Default, Debug)]
pub struct ReadTiming {
    /// Object requests issued to fetch data blocks. One coalesced
    /// range counts once, which is the point.
    pub data_gets: AtomicU64,
    /// Bytes returned by those requests, including any read past what
    /// the caller asked for as a result of coalescing.
    pub data_bytes: AtomicU64,
    /// Of `data_gets`, those issued by read-ahead rather than by a read.
    ///
    /// A subset, not a separate total: a read-ahead's request is counted in
    /// `data_gets` as well, so a read's own requests are
    /// `data_gets - read_ahead_gets`.
    ///
    /// Without this the two cannot be told apart, since the counter is
    /// incremented where the request is made and staging does not know who
    /// asked. A consumer comparing two files found one costing twice the
    /// requests of the other with both laid out identically, and could not
    /// establish whether they were measuring the layout or how much
    /// read-ahead each file attracted — the alternative being to turn
    /// read-ahead off, which measures a different system.
    pub read_ahead_gets: AtomicU64,
    /// Bytes returned by those requests. A subset of `data_bytes`, for the
    /// same reason.
    pub read_ahead_bytes: AtomicU64,
    /// Object requests issued to fetch meta (index) blocks.
    pub meta_gets: AtomicU64,
    /// Bytes returned by those requests.
    pub meta_bytes: AtomicU64,
    /// Object requests issued to fetch the inode.
    pub inode_gets: AtomicU64,
    /// Block reads served from the data cache, without reaching
    /// staging. Together with `data_gets` this separates "the cache
    /// worked" from "the read coalesced well".
    pub cache_hits: AtomicU64,
    /// Nanoseconds spent awaiting staging reads, so that a share of
    /// wall time can be attributed to the object store rather than
    /// inferred.
    pub staging_ns: AtomicU64,
    /// Block reads answered from a segment that is still being written
    /// out, which only happens with `wal`. These cost no object request
    /// and no wait for the flush, which is the point of doing them —
    /// counting them separately is the only way to tell that the WAL read
    /// path is working rather than merely correct.
    pub inflight_reads: AtomicU64,
}

/// Snapshot of [`ReadTiming`] values at a single point in time.
#[derive(Clone, Copy, Debug, Default)]
pub struct ReadTimingSnapshot {
    pub data_gets: u64,
    pub data_bytes: u64,
    /// Of `data_gets` / `data_bytes`, what read-ahead issued. See
    /// [`ReadTiming::read_ahead_gets`].
    pub read_ahead_gets: u64,
    pub read_ahead_bytes: u64,
    pub meta_gets: u64,
    pub meta_bytes: u64,
    pub inode_gets: u64,
    pub cache_hits: u64,
    pub staging_ns: u64,
    pub inflight_reads: u64,
}

impl ReadTimingSnapshot {
    /// Every object request, whatever it fetched.
    pub fn total_gets(&self) -> u64 {
        self.data_gets + self.meta_gets + self.inode_gets
    }

    /// Every byte fetched. Excludes the inode, whose size is not
    /// tracked separately.
    pub fn total_bytes(&self) -> u64 {
        self.data_bytes + self.meta_bytes
    }
}

impl ReadTiming {
    pub fn snapshot(&self) -> ReadTimingSnapshot {
        ReadTimingSnapshot {
            data_gets: self.data_gets.load(Ordering::Relaxed),
            data_bytes: self.data_bytes.load(Ordering::Relaxed),
            read_ahead_gets: self.read_ahead_gets.load(Ordering::Relaxed),
            read_ahead_bytes: self.read_ahead_bytes.load(Ordering::Relaxed),
            meta_gets: self.meta_gets.load(Ordering::Relaxed),
            meta_bytes: self.meta_bytes.load(Ordering::Relaxed),
            inode_gets: self.inode_gets.load(Ordering::Relaxed),
            cache_hits: self.cache_hits.load(Ordering::Relaxed),
            staging_ns: self.staging_ns.load(Ordering::Relaxed),
            inflight_reads: self.inflight_reads.load(Ordering::Relaxed),
        }
    }

    pub fn reset(&self) {
        self.data_gets.store(0, Ordering::Relaxed);
        self.read_ahead_gets.store(0, Ordering::Relaxed);
        self.read_ahead_bytes.store(0, Ordering::Relaxed);
        self.data_bytes.store(0, Ordering::Relaxed);
        self.meta_gets.store(0, Ordering::Relaxed);
        self.meta_bytes.store(0, Ordering::Relaxed);
        self.inode_gets.store(0, Ordering::Relaxed);
        self.cache_hits.store(0, Ordering::Relaxed);
        self.staging_ns.store(0, Ordering::Relaxed);
        self.inflight_reads.store(0, Ordering::Relaxed);
    }

    #[inline]
    pub(crate) fn add_data_get(&self, bytes: usize, elapsed_ns: u64) {
        self.data_gets.fetch_add(1, Ordering::Relaxed);
        self.data_bytes.fetch_add(bytes as u64, Ordering::Relaxed);
        self.staging_ns.fetch_add(elapsed_ns, Ordering::Relaxed);
    }

    /// Attribute a request already counted in `data_gets` to read-ahead.
    ///
    /// Called by the read-ahead path itself rather than by staging, which
    /// has no way to know: it sees a ranged load and nothing about its
    /// purpose. Called once per successful load, so a fetch that failed
    /// after staging counted it is not attributed here — the two counters
    /// disagree by that request, which is the honest answer.
    pub(crate) fn add_read_ahead_get(&self, bytes: usize) {
        self.read_ahead_gets.fetch_add(1, Ordering::Relaxed);
        self.read_ahead_bytes.fetch_add(bytes as u64, Ordering::Relaxed);
    }

    #[inline]
    pub(crate) fn add_meta_get(&self, bytes: usize, elapsed_ns: u64) {
        self.meta_gets.fetch_add(1, Ordering::Relaxed);
        self.meta_bytes.fetch_add(bytes as u64, Ordering::Relaxed);
        self.staging_ns.fetch_add(elapsed_ns, Ordering::Relaxed);
    }

    #[inline]
    pub(crate) fn add_inode_get(&self, elapsed_ns: u64) {
        self.inode_gets.fetch_add(1, Ordering::Relaxed);
        self.staging_ns.fetch_add(elapsed_ns, Ordering::Relaxed);
    }

    #[inline]
    pub(crate) fn add_cache_hit(&self) {
        self.cache_hits.fetch_add(1, Ordering::Relaxed);
    }

    /// Only the reactor's read path serves a block out of a flush's pinned buffer, so
    /// only there is there anything to count.
    #[cfg(all(feature = "wal", feature = "reactor"))]
    #[inline]
    pub(crate) fn add_inflight_read(&self) {
        self.inflight_reads.fetch_add(1, Ordering::Relaxed);
    }
}

/// An operation the reactor's handler put back in its queue, to be answered when it
/// runs again — a flush in progress, a lock it has to wait for, a refetch. Not a
/// failure: the handler answers the caller later, not now.
///
/// Told apart by this payload and not by the error's kind. The kind is
/// `ResourceBusy`, which is also what a store answering 429 or 5xx is: a check of
/// the kind took such a failure for a request already requeued, and answered no
/// one.
#[cfg(feature = "reactor")]
#[derive(Debug)]
pub(crate) struct Requeued(pub(crate) &'static str);

#[cfg(feature = "reactor")]
impl std::fmt::Display for Requeued {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "requeued: {}", self.0)
    }
}

#[cfg(feature = "reactor")]
impl std::error::Error for Requeued {}

/// The error an operation returns when it has put its request back in the queue.
#[cfg(feature = "reactor")]
pub(crate) fn requeued(why: &'static str) -> Error {
    Error::new(ErrorKind::ResourceBusy, Requeued(why))
}

/// Whether `e` says the request was put back in the queue (see [`Requeued`]).
#[cfg(feature = "reactor")]
pub(crate) fn is_requeued(e: &Error) -> bool {
    e.get_ref().is_some_and(|i| i.is::<Requeued>())
}

pub struct DirtyDataBlocks<'a> {
    pub inner: Option<BTreeMap<BlockIndex, &'a DataBlock>>,
    // if we need to owned (clone) the data
    pub owned: Option<BTreeMap<BlockIndex, DataBlock>>,
}

impl<'a> DirtyDataBlocks<'a> {
    pub fn len(&self) -> usize {
        if let Some(inner) = &self.inner {
            return inner.len();
        } else if let Some(owned) = &self.owned {
            return owned.len();
        }
        // if both inner/owned are None, return 0
        return 0;
    }

    pub fn data(&'a self) -> BTreeMap<BlockIndex, &'a DataBlock> {
        if let Some(inner) = &self.inner {
            return inner.clone();
        } else if let Some(owned) = &self.owned {
            let inner: BTreeMap<BlockIndex, &'a DataBlock> = owned
                                .iter()
                                .map(|(idx, blk)| (*idx, blk))
                                .collect();
            return inner;
        }
        panic!("invalid DirtyDataBlocks");
    }
}

/// A flush that failed after its segment was written or while it was being written.
///
/// A segment that reached storage is a complete checkpoint -- every segment carries
/// its own inode -- and while nothing points at it, the next open takes it up: a
/// crash in that window loses nothing. This says so to a caller whose flush failed
/// with its data still pending -- an explicit flush on the reactor, or release, which
/// roll nothing back -- so that it knows the data is not gone.
///
/// Where the operation is rolled back instead, the rollback abandons whatever the
/// session's checkpoint numbers may have left, and takes this off the error once it
/// no longer holds (see `HyperFile::rollback_after`).
///
/// "Landed" means the segment write succeeded and only the inode write failed. A
/// failed segment write is not proof of absence -- a timeout can follow a stored
/// object -- so that case is carried too, as possibly on storage. (A resend that
/// finds its own first attempt is not a failure: the write's mark says so.)
#[derive(Debug)]
pub struct PublishIncomplete {
    /// The checkpoint that is, or may be, on storage with no inode naming it.
    pub segid: SegmentId,
    /// Whether the segment is known to be there.
    pub landed: bool,
    source: Error,
}

impl PublishIncomplete {
    pub(crate) fn landed(source: Error, segid: SegmentId) -> Error {
        Error::new(source.kind(), Self { segid, landed: true, source })
    }

    pub(crate) fn maybe(source: Error, segid: SegmentId) -> Error {
        Error::new(source.kind(), Self { segid, landed: false, source })
    }

    /// The checkpoint `e` says is, or may be, on storage unpublished.
    pub fn of(e: &Error) -> Option<SegmentId> {
        e.get_ref()?.downcast_ref::<Self>().map(|p| p.segid)
    }

    /// `e` without this wrapping, for when what it says no longer holds.
    pub(crate) fn into_source(e: Error) -> Error {
        if e.get_ref().is_some_and(|i| i.is::<Self>()) {
            if let Ok(p) = e.into_inner().unwrap().downcast::<Self>() {
                return p.source;
            }
            unreachable!("checked to be a PublishIncomplete");
        }
        e
    }

    /// Carry what an earlier attempt left on storage onto a later attempt's error,
    /// unless that error already says something about it.
    fn carry(e: Error, earlier: Option<(SegmentId, bool)>) -> Error {
        match earlier {
            Some((segid, landed)) if Self::of(&e).is_none() =>
                Error::new(e.kind(), Self { segid, landed, source: e }),
            _ => e,
        }
    }
}

impl std::fmt::Display for PublishIncomplete {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let is = if self.landed { "is" } else { "may be" };
        write!(f, "{}; checkpoint {} {} on storage with no inode pointing at it, and the \
                   next open takes it up unless this file publishes again first",
               self.source, self.segid, is)
    }
}

impl std::error::Error for PublishIncomplete {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(&self.source)
    }
}

pub trait HyperTrait<T: Staging<L> + segment::SegmentReadWrite + Send + Clone + 'static, L: BlockLoader<BlockPtr> + Clone, C: NodeCache<BlockPtr> + Clone, V: Copy + Default + std::fmt::Display + NodeValue + 'static> {
    // block ptr
    fn blk_ptr_encode(&self, segid: SegmentId, offset: SegmentOffset, seq: usize) -> BlockPtr;
    /// Settle what was sent ahead of a flush under the stream format, closing the
    /// object being filled with `close`; see `stream_ahead`. Nothing by default.
    fn stream_settle(&mut self, close: bool) -> impl Future<Output = Result<()>> {
        let _ = close;
        async { Ok(()) }
    }
    fn blk_ptr_decode(&self, blk_ptr: &BlockPtr) -> (SegmentId, SegmentOffset);
    fn blk_ptr_decode_display(&self, blk_ptr: &BlockPtr) -> String;
    // data cache
    fn clear_data_blocks_cache(&mut self);
    fn set_data_blocks_cache_unlimited(&mut self);
    fn restore_data_blocks_cache_limit(&mut self);
    // dirty data
    fn get_data_blocks_dirty(&self) -> DirtyDataBlocks<'_>;
    fn clear_data_blocks_dirty(&mut self);
    /// Move just these blocks out of the dirty tier. See `DataCache::demote_dirty`.
    fn demote_data_blocks_dirty(&mut self, indexes: &[BlockIndex]);
    /// The checkpoint being accumulated out of partial segments, if any.
    fn pending_parts_get(&self) -> Option<crate::file::file::PendingParts>;
    /// Blocks written by reference, in index order. See [`crate::block_source`].
    fn refs_snapshot(&self) -> Vec<BlockIndex>;
    /// Where referenced blocks are read from.
    fn block_source_get(&self) -> Option<std::sync::Arc<dyn crate::block_source::BlockSource>>;
    /// Forget the references: the checkpoint holding them has been published.
    fn refs_clear(&mut self);
    /// Forget these references: a partial segment holding their bytes has landed,
    /// so their pointers now resolve to storage.
    fn refs_remove(&mut self, blocks: &[BlockIndex]);
    /// Whether the checkpoint `segid`, built from the current state, changes no
    /// content: see [`crate::ondisk::InodeRaw::FLAG_COMPACTED`]. Only if nothing
    /// has changed content since the last publish, every dirty block was
    /// relocated, none is a reference, and `segid` directly follows the last
    /// published checkpoint — so that the one before it is the content.
    fn compaction_flag(&self, segid: SegmentId) -> bool;
    /// A partial segment is about to take the dirty blocks out of view: record
    /// now whether they were all relocated, so the checkpoint they belong to is
    /// judged on all of them.
    fn fold_partial_into_compaction(&self);
    /// `raw` has just been published: keep its compacted flag for the
    /// attribute-only writes that follow, and start the bookkeeping afresh.
    fn checkpoint_published(&self, raw: &InodeRaw);
    /// The inode to write when only attributes are published.
    fn attr_only_raw(&self, b: BMapRawType) -> InodeRaw;
    /// Record that a partial landed, so the next one takes the following index.
    fn pending_parts_advance(&mut self, written: SegmentId);
    /// Forget the accumulated checkpoint. Its objects become unreferenced.
    fn pending_parts_clear(&mut self);
    /// Hold `segid` and the parts below `next_part` for the checkpoint being
    /// written, keeping whether anything was demoted under it already. Shared
    /// access: a stream reserves its parts while its blocks are borrowed.
    fn pending_parts_reserve(&self, segid: SegmentId, next_part: u16);
    // lock
    fn lock(&self) -> impl Future<Output = OwnedSemaphorePermit>;
    fn unlock(&self, permit: OwnedSemaphorePermit);
    fn flush_lock(&self) -> impl Future<Output = OwnedMutexGuard<()>>;
    fn flush_unlock(&self, lock: OwnedMutexGuard<()>);
    // bmap
    fn bmap_as_slice(&self) -> &[u8];
    fn bmap_get_block_loader(&self) -> L;
    fn bmap_get_node_cache(&self) -> C;
    fn bmap_dirty(&self) -> bool;
    fn bmap_lookup_dirty(&self) -> Vec<BtreeNodeDirty<'_, BlockIndex, V, BlockPtr>>;
    fn bmap_assign_meta_node(&self, blk_ptr: BlockPtr, node: BtreeNodeDirty<'_, BlockIndex, V, BlockPtr>) -> impl Future<Output = Result<()>>;
    fn bmap_assign_data_node(&self, blk_idx: &BlockIndex, blk_ptr: BlockPtr) -> impl Future<Output = Result<()>>;
    fn bmap_clear_dirty(&mut self);
    fn bmap_update<'a>(&mut self, bmap: BMap<'a, BlockIndex, V, BlockPtr, L, C>);
    fn bmap_insert_dummy_value(bmap: &mut BMap<'_, BlockIndex, V, BlockPtr, L, C>, blk_idx: &BlockIndex) -> impl Future<Output = Result<Option<V>>>;
    fn bmap_set_cache_unlimited(&self) -> usize;
    fn bmap_set_cache_limit(&self, liimt: usize);
    // inode
    fn inode(&self) -> &Inode;
    fn inode_mut(&self) -> &mut Inode;
    // others
    fn staging(&self) -> &T;
    fn config(&self) -> &HyperFileConfig;
    fn set_last_flush(&mut self);
    fn sleep(dur: Duration) -> impl Future<Output = ()>;
    fn flush_timing(&self) -> &FlushTiming;

    /// `Err` once publishing has failed unrecoverably. See
    /// `State::publish_failed`.
    fn check_writable(&self) -> Result<()>;

    /// The write-ahead log, if one is configured.
    #[cfg(feature = "wal")]
    fn wal_mut(&mut self) -> Option<&mut Box<dyn crate::wal::WalReadWrite + Send>>;

    /// Whether an transaction is open. See `HyperFile::begin_txn`.
    #[cfg(feature = "wal")]
    fn in_txn_trait(&self) -> bool;

    /// Whether this flush may be satisfied by the log without publishing. See
    /// `HyperFileWalConfig::publish_every`.
    #[cfg(feature = "wal")]
    fn wal_may_defer_publish(&self) -> bool;

    /// Record that a flush was satisfied by the log.
    #[cfg(feature = "wal")]
    fn wal_count_deferred_barrier(&mut self);

    // wal
    #[cfg(feature = "wal")]
    fn wal_set_mem_segment(&self, cno: Cno, mem_segdata: Weak<Pin<Box<Vec<u8>>>>) -> impl Future<Output = ()>;
    #[cfg(feature = "wal")]
    fn wal_clear_mem_segment(&self, cno: Cno) -> impl Future<Output = ()>;
    /// Fire-and-forget delete of the WAL objects for the given
    /// segid. Spawned so the caller (typically the flush path)
    /// doesn't block on the round trip. If the delete fails or
    /// the spawned task is cancelled, the WAL entries remain on
    /// storage but recovery filters them out by last_ondisk_cno
    /// so correctness is unaffected.
    #[cfg(feature = "wal")]
    fn wal_spawn_delete_segment(&self, segid: SegmentId);

    /// Delete every log prefix this publish supersedes, and reset the deferred
    /// count.
    ///
    /// One publish can stand for several groups: each deferred flush sealed its
    /// own, and all of them are contained in the segment just written. Deleting
    /// only the newest would leave the rest behind, so the log would grow in
    /// proportion to how often publishing is deferred — which is the cost that
    /// deferral is trying to avoid paying elsewhere.
    #[cfg(feature = "wal")]
    fn wal_drop_superseded_prefixes(&mut self, published: SegmentId) {
        let deferred = self.wal_deferred_count();
        for back in 1..=(deferred + 1) {
            self.wal_spawn_delete_segment(SegmentId::new_from_cno(published.as_cno().saturating_sub(back as u64)));
        }
        self.wal_reset_deferred_count();
    }

    /// Flushes satisfied by the log since the last publish.
    #[cfg(feature = "wal")]
    fn wal_deferred_count(&self) -> usize;

    #[cfg(feature = "wal")]
    fn wal_reset_deferred_count(&mut self);

    // provided method
    fn bmap_get_raw(&self) -> BMapRawType {
        let mut b: BMapRawType = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        b.copy_from_slice(self.bmap_as_slice());
        b
    }

    // recover inode from segment
    fn recover_partial_flush(&mut self, cno: Cno, od_state: &Option<OnDiskState>) -> impl Future<Output = Result<()>> {async move {
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let _ = self.staging().load_inode_from_segment(&mut raw_inode.as_mut_u8_slice(), SegmentId::new_from_cno(cno)).await?;
        let od_state = self.staging().flush_inode(raw_inode.as_u8_slice(), od_state, FlushInodeFlag::Update).await?;
        self.inode_mut().clear_attr_dirty();
        self.inode_mut().set_ondisk_state(od_state);
        self.inode_mut().set_base(&raw_inode);
        let last_cno = self.inode().get_last_cno();
        self.inode_mut().set_last_ondisk_cno(last_cno);
        Ok(())
    }}

    /// try recover partial flush (inode ahead segment)
    fn try_recover_partial_flush(&mut self) -> impl Future<Output = Result<(InodeRaw, Option<OnDiskState>)>> {async {
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let mut inode_state;
        loop {
            debug!("try_recover_partial_flush - loop entry");
            match self.staging().load_inode(&mut raw_inode.as_mut_u8_slice()).await {
                Ok(od_state) => { inode_state = od_state; },
                Err(e) => { return Err(e); },
            }
            debug!("try_recover_partial_flush - inode ondisk state: {:?}", inode_state);

            // test next segid
            let next_segid = SegmentId::new_from_cno(raw_inode.i_last_seq + 1);
            match self.staging().load_segment_timestamp(next_segid).await {
                Ok((server_time, segment_lm)) => {
                    if server_time - segment_lm > DEFAULT_PARTIAL_FLUSH_TIMEOUT as i64 {
                        self.recover_partial_flush(next_segid.as_cno(), &inode_state).await?;
                    } else {
                        // sleep for a while for next check
                        Self::sleep(Duration::from_secs(DEFAULT_PARTIAL_FLUSH_CHECK_INTERVAL_SECS)).await;
                    }
                    continue;
                },
                Err(e) => {
                    if e.kind() != ErrorKind::NotFound { return Err(e); }
                    // if we have inode but last segment not exists, let's break
                    break;
                },
            }
        }

        Ok((raw_inode, inode_state))
    }}

    /// Take up checkpoints that reached storage after the persisted inode and that
    /// no inode points at yet -- what a crash between a segment and its inode
    /// leaves -- and step past numbers taken by anything else.
    ///
    /// Every segment carries the inode of the checkpoint it completes, and a
    /// streamed checkpoint's part 0 is written last, so a consistency-point object
    /// on storage is a whole commit whose pointer update was lost. Starting from
    /// the persisted inode's counter, each next number is looked at:
    ///
    /// - a commit (the bare name unparted, part 0 parted) that passes
    ///   [`Self::commit_at`] is taken up, and its own inode becomes the base for
    ///   the next number;
    /// - anything else there -- bytes that are not a segment, partials with no
    ///   part 0, a name this format never writes -- is stepped past without being
    ///   read further, because a taken name fails the next create-only write on it;
    /// - a free number ends the walk.
    ///
    /// A writable open then points the inode at what it took up, once, with the
    /// conditional write on the inode it started from. It first waits out
    /// `DEFAULT_PARTIAL_FLUSH_TIMEOUT` on a commit that fresh, because a
    /// writer still running may be about to point the inode at it itself, and
    /// doing that first would fail that writer's own conditional write. A read-only
    /// open writes nothing and waits for nothing: it takes the same commits up in
    /// memory, so it sees what the next writer will publish.
    ///
    /// Returns the inode the session starts from and the on-disk state it holds.
    fn roll_forward(staging: &T, loader: L, meta: &HyperFileMetaConfig, persisted: InodeRaw,
            state: Option<OnDiskState>, writable: bool)
        -> impl Future<Output = Result<(InodeRaw, Option<OnDiskState>)>> {async move {
        let parted = meta.block_ptr_format.is_parted();
        let mut writable = writable;
        let mut persisted = persisted;
        let mut state = state;
        let mut failed_writes = 0usize;
        'restart: loop {
            let mut chain = persisted;
            let mut taken_up = 0usize;
            loop {
                let next = chain.i_last_seq + 1;
                if next > crate::meta_format::BLOCK_PTR_MAX_CHECKPOINT {
                    return Err(Error::new(ErrorKind::InvalidData, format!(
                        "checkpoint numbers up to {} are all taken on storage; a block \
                         pointer cannot name the next one", next - 1)));
                }
                let segid = SegmentId::new_from_cno(next);
                let (bare, summary, first) = futures::try_join!(
                    staging.segment_exists(segid.checkpoint()),
                    staging.segment_exists(segid.at_part(segment::Segment::SUMMARY_PART)),
                    staging.segment_exists(segid.at_part(segment::Segment::FIRST_DATA_PART)),
                )?;
                if !(bare || summary || first) {
                    break;
                }
                let own_name = if parted { summary } else { bare };
                let commit = if own_name {
                    Self::commit_at(staging, loader.clone(), meta, &chain, segid).await?
                } else {
                    None
                };
                let Some(inode) = commit else {
                    warn!("checkpoint {} is taken on storage by something that is not a \
                          commit to take up; stepping past it", segid);
                    chain.i_last_seq = next;
                    continue;
                };
                if writable {
                    let (server, modified) = staging.load_segment_timestamp(segid).await?;
                    if server - modified <= DEFAULT_PARTIAL_FLUSH_TIMEOUT as i64 {
                        debug!("checkpoint {} is only {} s old; giving its writer time to \
                               publish it", segid, server - modified);
                        Self::sleep(Duration::from_secs(DEFAULT_PARTIAL_FLUSH_CHECK_INTERVAL_SECS)).await;
                        (persisted, state) = Self::load_persisted(staging, meta).await?;
                        continue 'restart;
                    }
                }
                info!("taking up checkpoint {}, which reached storage without an inode \
                      pointing at it", segid);
                chain = inode;
                taken_up += 1;
            }
            if taken_up == 0 || !writable {
                return Ok((chain, state));
            }
            match staging.flush_inode(chain.as_u8_slice(), &state, FlushInodeFlag::Update).await {
                Ok(s) => return Ok((chain, s)),
                Err(e) => {
                    failed_writes += 1;
                    warn!("could not point the inode at checkpoint {}: {}", chain.i_last_cno, e);
                    // Whatever happened -- a conflict, or a write that landed and
                    // lost its answer -- the next step is decided from storage.
                    // Past the retries this session carries the commits in memory
                    // instead, and its first publish points the inode at them.
                    if failed_writes >= crate::DEFAULT_FLUSH_RETRIES {
                        writable = false;
                    }
                    (persisted, state) = Self::load_persisted(staging, meta).await?;
                }
            }
        }
    }}

    /// The persisted inode, refused if its counters are impossible or its format is
    /// not the one this open started with.
    fn load_persisted(staging: &T, meta: &HyperFileMetaConfig)
        -> impl Future<Output = Result<(InodeRaw, Option<OnDiskState>)>> {async move {
        let mut raw: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let state = staging.load_inode(&mut raw.as_mut_u8_slice()).await?;
        raw.check_counters()?;
        if raw.i_meta_config != meta.as_u32() {
            return Err(Error::new(ErrorKind::InvalidData, format!(
                "the inode's format changed from {:#010x} to {:#010x} during open",
                meta.as_u32(), raw.i_meta_config)));
        }
        Ok((raw, state))
    }}

    /// The inode of the commit at `segid`, if the object there is one to take up.
    ///
    /// Only the segment header is read, and nothing about it is believed until it
    /// has been checked: that it is a segment header at all; that it names this
    /// checkpoint and is not a partial; and that the inode it carries has this
    /// container's format, counters this checkpoint could have, and a map root that
    /// decodes -- which is where bytes that merely begin like a header fail. `base`
    /// is the inode the walk has reached, which the candidate must have been able to
    /// follow.
    ///
    /// Which inode the candidate was built on is checked where the segment records
    /// it (`SegmentHeader::s_base`). Segments from before that was recorded are
    /// judged by their counters alone, which is exact only for a single writer.
    fn commit_at(staging: &T, loader: L, meta: &HyperFileMetaConfig, base: &InodeRaw, segid: SegmentId)
        -> impl Future<Output = Result<Option<InodeRaw>>> {async move {
        let name = if meta.block_ptr_format.is_parted() {
            segid.at_part(segment::Segment::SUMMARY_PART)
        } else {
            segid.checkpoint()
        };
        let hdr = match staging.load_segment_header(name, meta.block_ptr_format.is_stream()).await {
            Ok(h) => h,
            // Shorter than a header, not one, empty, or gone since it was seen: not
            // a commit.
            Err(e) if matches!(e.kind(), ErrorKind::InvalidData | ErrorKind::InvalidInput
                | ErrorKind::NotFound) => return Ok(None),
            Err(e) => return Err(e),
        };
        let inode = hdr.s_inode;
        let cno = segid.as_cno();
        let fits = hdr.has_magic()
            && hdr.s_cno == cno
            // Built on the state the walk has reached, where the segment says. Two
            // writers can each have a checkpoint past the same inode; only the one
            // built on it is its successor, and taking up the other would put a
            // stale view over a commit already published. Zero is a segment from
            // before this was recorded, left to the counters.
            && (hdr.s_base == 0 || hdr.s_base == base.fingerprint())
            && hdr.s_flags & SEGMENT_FLAG_PARTIAL == 0
            && !inode.is_inline()
            && inode.i_meta_config == base.i_meta_config
            && inode.i_last_cno == cno
            // More than validity: this is what makes the walk terminate. It goes
            // on from the taken-up inode's counter plus one, so a counter behind
            // `cno` sends it back to this same object, for ever -- and on memory
            // staging nothing in that loop yields, so an async timeout around it
            // never gets polled. `check_counters` implies this for other inodes, but not
            // for a device node, which keeps its rdev where the pointer goes.
            && inode.i_last_seq >= cno
            && inode.check_counters().is_ok()
            && BMap::<BlockIndex, V, BlockPtr, L, btree_ondisk::NullNodeCache>::read(
                &inode.i_bmap, meta.meta_block_size, loader, btree_ondisk::NullNodeCache).is_ok();
        Ok(fits.then_some(inode))
    }}

    /// Rebuild the map on the persisted inode after a flush lost a conflict, so the
    /// retry can put this session's dirty blocks on top of what was published.
    ///
    /// Reads the inode and nothing else: no segment is looked at, and nothing is
    /// written. Taking up a checkpoint that no inode points at is open's job; see
    /// [`Self::roll_forward`].
    fn refresh_bmap(&mut self) -> impl Future<Output = Result<SegmentId>> {async {
        // save current in memory state
        let curr_segid = self.inode().get_last_seq();
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let inode_state = self.staging().load_inode(&mut raw_inode.as_mut_u8_slice()).await?;
        raw_inode.check_counters()?;
        let od_last_seq = raw_inode.i_last_seq;

        if curr_segid.as_cno() == od_last_seq {
            debug!("REFRESH_BMAP - quit due: current segid {} == on disk segid {}", curr_segid, od_last_seq);
            return Ok(curr_segid);
        } else if curr_segid.as_cno() > od_last_seq {
            warn!("REFRESH_BMAP - current segid {} is ahead of on disk segid {}", curr_segid, od_last_seq);
            return Ok(curr_segid);
        }

        // get back block loader
        let meta_block_loader = self.bmap_get_block_loader();
        // get back node cache
        let node_cache = self.bmap_get_node_cache();

        // Borrowed out of the inode rather than copied to a local: `BMap::read`
        // wants an 8-byte-aligned buffer, and `BMapRawType` is `[u8; N]`, whose
        // own alignment is 1. `InodeRaw` is `align(8)` and `i_bmap` sits at an
        // 8-aligned offset within it, so this reference is aligned by
        // construction — a copy would be wherever the compiler put it.
        let mut bmap = BMap::<BlockIndex, V, BlockPtr, L, C>::read(&raw_inode.i_bmap, self.config().meta.meta_block_size, meta_block_loader, node_cache)?;

        let _permit = self.lock().await;

        // clear cached data
        let _ = self.clear_data_blocks_cache();

        let dirty_data_blocks = self.get_data_blocks_dirty();

        // no dirty blocks, just return
        if dirty_data_blocks.len() == 0 {
            return Ok(curr_segid);
        }

        // rebuild dirty bmap
        for (blk_idx, _) in dirty_data_blocks.data().iter() {
            // if blk index is already exists, overwrite it
            // A fresh map, not the session's: a failure here leaves the session's as
            // it was, so it is simply returned.
            let _ = Self::bmap_insert_dummy_value(&mut bmap, blk_idx).await?;
        }
        assert!(bmap.dirty() == true);

        // A number held for a checkpoint whose blocks are all still dirty can be
        // given up: another writer may have taken it, and nothing depends on it.
        // One with blocks already demoted cannot — the map just loaded does not
        // point at them — so a stream refuses rather than lose them.
        if let Some(p) = self.pending_parts_get() {
            if !p.demoted {
                self.pending_parts_clear();
            } else if self.config().meta.block_ptr_format.is_stream() {
                return Err(Error::new(ErrorKind::AlreadyExists, format!(
                    "another writer published while checkpoint {} was part-way out with blocks written \
                     as partials; they cannot be carried over", p.segid)));
            }
        }
        info!("REFRESH_BMAP - update in-memory bmap with inode from ondisk");
        // refresh inner bmap and inode but don't touch inode attr
        self.bmap_update(bmap);
        (*self.inode_mut()).i_last_seq = raw_inode.i_last_seq;
        (*self.inode_mut()).i_last_cno = raw_inode.i_last_cno;
        (*self.inode_mut()).i_last_ondisk_cno = raw_inode.i_last_cno;
        (*self.inode_mut()).i_ondisk_state = inode_state;
        self.inode_mut().set_base(&raw_inode);

        Ok(self.inode().get_last_seq())
    }}

    fn flush_process_pre_build_segment(&self) -> impl Future<Output = Result<(u64, DirtyDataBlocks<'_>)>> {async {
        let dirty_data_blocks = self.get_data_blocks_dirty();

        // A checkpoint part-way out still has to be finished, and after its
        // partials there is nothing dirty left to notice. Its part 0 — the summary
        // naming every object, the metadata, the inode — is what makes it openable
        // at all, so quitting here would leave the partials referenced by nothing
        // and the flush would have done nothing while reporting success.
        let complete_pending = self.pending_parts_get().is_some();

        if !complete_pending && dirty_data_blocks.len() == 0 && self.refs_snapshot().is_empty()
            && !self.bmap_dirty() {
            if self.inode().is_attr_dirty() {
                debug!("inode attr is dirty, flush inode ONLY");
                let b = self.bmap_get_raw();
                let raw_inode = self.attr_only_raw(b);
                let od_state = self.staging().flush_inode(raw_inode.as_u8_slice(), self.inode().get_ondisk_state(), FlushInodeFlag::Update).await?;
                self.inode_mut().clear_attr_dirty();
                self.inode_mut().set_ondisk_state(od_state);
                self.inode_mut().set_base(&raw_inode);
                // Nothing is dirty here: what this writes is the whole state.
                self.checkpoint_published(&raw_inode);
                let last_cno = self.inode().get_last_cno();
                self.inode_mut().set_last_ondisk_cno(last_cno);
            }
            debug!("flush quit, NO dirty data blocks amd bmap is NOT dirty");
            let segid = self.inode().get_last_seq();
            return Ok((segid.as_cno(), DirtyDataBlocks { inner: None, owned: None }));
        }
        Ok((0, dirty_data_blocks))
    }}

    /// Whether this container may answer memory pressure with a partial segment
    /// instead of a checkpoint nobody asked for.
    ///
    /// Both halves are needed and neither is enough. The format decides whether a
    /// partial is addressable at all and is fixed when the container is created;
    /// the switch decides whether to use it, and can be left off on a container
    /// whose format allows it.
    /// The most bytes one stored object may hold: the configured
    /// `segment_max_object_bytes`, within what the block-pointer format can
    /// address.
    fn max_object_bytes(&self) -> usize {
        let limit = self.config().meta.block_ptr_format.max_object_bytes();
        self.config().runtime.segment_max_object_bytes
            .clamp(crate::MIN_SEGMENT_MAX_OBJECT_BYTES, limit)
    }

    /// The bytes of dirty meta blocks and the number of data blocks, dirty or
    /// referenced, that the next segment built from this state would hold.
    fn checkpoint_shape(&self) -> (usize, usize) {
        let meta_bytes = self.bmap_lookup_dirty().iter().map(|n| n.size()).sum();
        let nblocks = self.get_data_blocks_dirty().len() + self.refs_snapshot().len();
        (meta_bytes, nblocks)
    }

    /// Lay a checkpoint's blocks out in objects of at most [`Self::max_object_bytes`].
    ///
    /// Returns how many data blocks each object takes: the first entry is the
    /// object holding the meta blocks and the inode, the rest are data-only parts.
    /// Blocks go in index order, the first ones with the meta blocks. A checkpoint
    /// that fits is one entry, the layout it has always had.
    ///
    /// Only the parted format can name a block in a second object, and only
    /// without `wal`, which keeps a segment in one buffer for replay; anything else
    /// that does not fit is refused here, before a number is taken or a pointer
    /// assigned. So is a checkpoint whose meta blocks alone do not fit one object.
    fn plan_objects(&self, meta_bytes: usize, nblocks: usize) -> Result<Vec<usize>> {
        let cap = self.max_object_bytes();
        let bs = self.config().meta.data_block_size;
        if segment::Writer::<T>::calc_ss_aligned_bytes(0) + meta_bytes > cap {
            return Err(Error::new(ErrorKind::InvalidInput, format!(
                "the {} bytes of index this checkpoint writes do not fit the {} bytes one \
                 object may hold; flush more often", meta_bytes, cap)));
        }
        let first = segment::Writer::<T>::blocks_that_fit(cap, meta_bytes, bs).min(nblocks);
        if first == nblocks {
            return Ok(vec![first]);
        }
        let fmt = self.config().meta.block_ptr_format;
        if !fmt.is_parted() || cfg!(feature = "wal") {
            let why = if fmt.is_parted() { "a build with the WAL".to_string() } else { format!("the {:?} block-pointer format", fmt) };
            return Err(Error::new(ErrorKind::InvalidInput, format!(
                "a checkpoint of {} blocks needs more than the {} bytes one object may hold, \
                 and {} writes a checkpoint as one object; flush more often", nblocks, cap, why)));
        }
        let per = segment::Writer::<T>::blocks_that_fit(cap, 0, bs);
        if per == 0 {
            return Err(Error::new(ErrorKind::InvalidInput, format!(
                "one object of {} bytes cannot hold a {}-byte block and its summary", cap, bs)));
        }
        let mut counts = vec![first];
        let mut rest = nblocks - first;
        while rest > 0 {
            let n = rest.min(per);
            counts.push(n);
            rest -= n;
        }
        Ok(counts)
    }

    /// Assign every pointer of a checkpoint laid out as `counts` (see
    /// [`Self::plan_objects`]) and build a writer per object: `main` for the one
    /// holding the meta blocks, `data_ids[k]` for data part `k`.
    ///
    /// All pointers are assigned before anything is serialized, so the meta blocks
    /// written hold the final pointer of every block, whichever object it is in.
    #[allow(clippy::type_complexity)]
    fn assign_layout(&self, main: SegmentId, main_partial: bool, data_ids: &[SegmentId], counts: &[usize],
            dirty_meta_vec: &[BtreeNodeDirty<'_, BlockIndex, V, BlockPtr>],
            ordered: &[(BlockIndex, Option<&DataBlock>)])
            -> impl Future<Output = Result<(segment::Writer<T>, Vec<segment::Writer<T>>)>>
    {async move {
        debug_assert_eq!(counts.len(), data_ids.len() + 1);
        debug_assert_eq!(counts.iter().sum::<usize>(), ordered.len());
        let bs = self.config().meta.data_block_size;
        let mut segwr = if main_partial {
            let mut w = self.staging().new_segwr_at(main, &self.config().meta);
            w.mark_partial();
            w
        } else {
            self.staging().new_segwr(main, &self.config().meta)
        };
        let mut parts: Vec<segment::Writer<T>> = data_ids.iter().map(|id| {
            let mut w = self.staging().new_segwr_at(*id, &self.config().meta);
            w.mark_partial();
            w
        }).collect();

        let mut file_off = segment::Writer::<T>::calc_ss_aligned_bytes(counts[0]);
        let mut block_seq = 0;
        for n in dirty_meta_vec {
            let blk_ptr = self.blk_ptr_encode(main, file_off, block_seq);
            debug!("assign block ptr for meta node: block ptr {}", self.blk_ptr_decode_display(&blk_ptr));
            self.bmap_assign_meta_node(blk_ptr, n.clone()).await?;
            segwr.inc_metablk();
            file_off += n.size();
            block_seq += 1;
        }
        let mut blocks = ordered.iter();
        for (blk_idx, n) in blocks.by_ref().take(counts[0]) {
            debug_assert!(n.is_none_or(|n| n.size() == bs));
            let blk_ptr = self.blk_ptr_encode(main, file_off, block_seq);
            debug!("assign block ptr for data node: block ptr {}, block index {}", self.blk_ptr_decode_display(&blk_ptr), blk_idx);
            self.bmap_assign_data_node(blk_idx, blk_ptr).await?;
            segwr.inc_datablk(blk_idx, &blk_ptr);
            file_off += bs;
            block_seq += 1;
        }
        for (k, w) in parts.iter_mut().enumerate() {
            let id = data_ids[k];
            let mut off = segment::Writer::<T>::calc_ss_aligned_bytes(counts[k + 1]);
            for (seq, (blk_idx, n)) in blocks.by_ref().take(counts[k + 1]).enumerate() {
                debug_assert!(n.is_none_or(|n| n.size() == bs));
                let blk_ptr = self.blk_ptr_encode(id, off, seq);
                self.bmap_assign_data_node(blk_idx, blk_ptr).await?;
                w.inc_datablk(blk_idx, &blk_ptr);
                off += bs;
            }
        }
        debug_assert!(blocks.next().is_none());
        Ok((segwr, parts))
    }}

    /// Write the summaries, then the meta blocks and each block's bytes, into the
    /// writers [`Self::assign_layout`] built.
    ///
    /// `append_data` false leaves the data blocks to the caller: the WAL build's
    /// concurrent segment build copies them into its one buffer itself.
    #[allow(clippy::too_many_arguments)]
    fn fill_layout(&self, segwr: &mut segment::Writer<T>, parts: &mut [segment::Writer<T>], counts: &[usize],
            raw_inode: &InodeRaw, dirty_meta_vec: &[BtreeNodeDirty<'_, BlockIndex, V, BlockPtr>],
            ordered: &[(BlockIndex, Option<&DataBlock>)],
            src: &Option<std::sync::Arc<dyn crate::block_source::BlockSource>>, append_data: bool) -> Result<()>
    {
        let bs = self.config().meta.data_block_size;
        // The state this checkpoint is built on, for whoever finds it unpublished.
        let base = self.inode().base();
        segwr.set_base(base);
        segwr.realize_ss(0, raw_inode);
        for w in parts.iter_mut() {
            w.set_base(base);
            w.realize_ss(0, raw_inode);
        }
        for n in dirty_meta_vec {
            let _ = segwr.append(n.as_slice())?;
        }
        if !append_data {
            // Only for one object: the WAL writes a checkpoint as one.
            debug_assert!(parts.is_empty());
            return Ok(());
        }
        {
            let mut blocks = ordered.iter();
            let mut objects: Vec<&mut segment::Writer<T>> = std::iter::once(segwr).chain(parts.iter_mut()).collect();
            for (k, w) in objects.iter_mut().enumerate() {
                for (blk_idx, n) in blocks.by_ref().take(counts[k]) {
                    match n {
                        Some(n) => w.append_data_block(n)?,
                        // A reference: read from the source when this part uploads.
                        #[cfg(not(feature = "wal"))]
                        None => w.append_lazy(src.clone().expect("a reference has a source"),
                            *blk_idx * bs as u64, bs),
                        #[cfg(feature = "wal")]
                        None => { let _ = (blk_idx, src, bs); unreachable!("write_ref is refused with the WAL") }
                    }
                }
            }
        }
        Ok(())
    }

    /// The part the first data-only part of the next checkpoint takes, reserving
    /// the numbers it will use when the checkpoint is already part-way out.
    ///
    /// A checkpoint with partials keeps its number across retries, so a part name
    /// an attempt wrote would be taken for the next one; moving the stream past them
    /// now, before any is written, means no attempt reuses one. A checkpoint
    /// started fresh needs nothing: a retry takes a new number.
    fn reserve_data_parts(&mut self) -> Result<u16> {
        let (meta_bytes, nblocks) = self.checkpoint_shape();
        let counts = self.plan_objects(meta_bytes, nblocks)?;
        let ndata = counts.len() - 1;
        let pending = self.pending_parts_get();
        let first = pending.as_ref().map_or(segment::Segment::FIRST_DATA_PART, |p| p.next_part);
        if first as usize + ndata > segment::Segment::MAX_PARTS {
            return Err(Error::new(ErrorKind::InvalidInput, format!(
                "a checkpoint of {} blocks needs {} data parts after part {}, past the {} a \
                 pointer can name; flush more often", nblocks, ndata, first, segment::Segment::MAX_PARTS)));
        }
        if let Some(p) = pending {
            if ndata > 0 {
                self.pending_parts_advance(p.segid.at_part(first + ndata as u16 - 1));
            }
        }
        Ok(first)
    }

    fn parted_segment_enabled(&self) -> bool {
        self.config().meta.block_ptr_format.is_parted()
            && self.config().runtime.parted_segment_enabled
    }

    /// Whether there is enough dirty data for a partial segment to be worth the
    /// object it costs and the read it breaks.
    fn worth_a_partial_segment(&self) -> bool {
        let dirty_bytes = self.get_data_blocks_dirty().len() * self.config().meta.data_block_size;
        dirty_bytes >= self.config().runtime.data_cache_dirty_min_bytes_to_part
    }

    /// Write the dirty set out as one piece of a checkpoint that is not finished.
    ///
    /// For the moments that today force a publish nobody asked for: a dirty-data
    /// threshold, the flush interval, or — under a transaction or `Barrier` — the
    /// point at which a write is refused with `OutOfMemory` because nothing may
    /// publish. None of those is a consistency point, so none of them should make
    /// one; this writes the data and the metadata describing it, leaves the
    /// checkpoint open, and keeps accumulating.
    ///
    /// The object has the same shape as any other segment — its own summary, the
    /// inode inline, its metadata blocks, its data blocks — so anything that can
    /// parse a segment can parse this. A dirty set larger than one object (see
    /// `segment_max_object_bytes`) is written as several: data-only parts, then the
    /// one with the metadata, as a flush lays out a checkpoint. Since every currently dirty node is written,
    /// the bmap root in that inode reaches every block the checkpoint holds so far,
    /// which makes the partial a complete state and not a fragment.
    ///
    /// What it is not is durable. The inode *object* is untouched, so a reader that
    /// opens the container normally still sees the previous checkpoint, and a crash
    /// here leaves these objects referenced by nothing. Recovering to a partial
    /// means opening it by name, which is a repair tool's business and not
    /// `open`'s: the caller's flush is the consistency point, and reading past it
    /// would take that away.
    ///
    /// Ordering, and one step of it is the whole safety argument:
    ///
    /// Metadata pointers are assigned before the nodes are serialized, because a
    /// parent has to contain its children's pointers. So the data in the same
    /// object cannot be uploaded before its pointers are assigned either.
    ///
    /// What protects a reader instead is that a block does not leave the dirty tier
    /// until its upload has landed. Until then a read finds it in the cache and
    /// never resolves the pointer; afterwards the pointer resolves to bytes that
    /// are there. That is the last step, and moving it earlier would open exactly
    /// the window the single-object path needs its pinned buffer for.
    #[allow(clippy::type_complexity)]
    fn write_partial_segment(&mut self) -> impl Future<Output = Result<SegmentId>> {async move {
        if self.config().meta.block_ptr_format.is_stream() {
            return self.stream_flush(false).await;
        }
        if let Some(p) = self.pending_parts_get() {
            if p.next_part as usize >= segment::Segment::MAX_PARTS {
                return Err(Error::new(ErrorKind::OutOfMemory, format!(
                    "checkpoint {} has already been written as {} partial segments, \
                     which is all a pointer can name — flush to complete it",
                    p.segid, p.next_part)));
            }
        }

        // Laid out first: a dirty set larger than one object is written as data-only
        // parts followed by the part holding the meta blocks, all partials. Planned
        // before a number is taken, and every part number reserved before any is
        // written, so neither a refusal nor a failure part-way leaves a name the next
        // partial would collide with.
        // Before the blocks leave the dirty tier, judged now. A failure below
        // leaves it recorded, which can only make the checkpoint unmarked.
        self.fold_partial_into_compaction();
        let (meta_bytes, nblocks) = self.checkpoint_shape();
        let counts = self.plan_objects(meta_bytes, nblocks)?;
        let ndata = counts.len() - 1;
        let (number, first) = match self.pending_parts_get() {
            Some(p) => (p.segid, p.next_part),
            // The number this whole stream belongs to, taken once.
            None => (self.inode_mut().get_next_seq(), segment::Segment::FIRST_DATA_PART),
        };
        if first as usize + ndata >= segment::Segment::MAX_PARTS {
            return Err(Error::new(ErrorKind::OutOfMemory, format!(
                "the dirty set needs {} partial segments after part {}, past the {} a \
                 pointer can name — flush to complete the checkpoint", ndata + 1, first,
                segment::Segment::MAX_PARTS)));
        }
        let data_ids: Vec<SegmentId> = (0..ndata).map(|k| number.at_part(first + k as u16)).collect();
        let segid = number.at_part(first + ndata as u16);
        self.pending_parts_advance(segid);

        let dirty_meta_vec = self.bmap_lookup_dirty();
        let dirty_data_blocks = self.get_data_blocks_dirty();
        let refs = self.refs_snapshot();
        let ref_src = if refs.is_empty() { None } else {
            Some(self.block_source_get().expect("references exist only while a source is set"))
        };
        let mut ordered: Vec<(BlockIndex, Option<&DataBlock>)> =
            dirty_data_blocks.data().iter().map(|(i, n)| (*i, Some(*n))).collect();
        if !refs.is_empty() {
            ordered.extend(refs.iter().map(|i| (*i, None)));
            ordered.sort_by_key(|(i, _)| *i);
            assert!(ordered.windows(2).all(|w| w[0].0 != w[1].0),
                "a block is both referenced and dirty");
        }
        debug_assert_eq!(ordered.len(), nblocks);
        debug!("writing partial segment {}: dirty meta nodes {}, data blocks {}, in {} objects",
            segid, dirty_meta_vec.len(), ordered.len(), ndata + 1);

        let (mut segwr, mut parts) = self.assign_layout(segid, true, &data_ids, &counts,
            &dirty_meta_vec, &ordered).await?;

        let b = self.bmap_get_raw();
        let mut raw_inode = self.inode().to_raw(b);
        raw_inode.i_last_cno = segid.as_cno();
        raw_inode.set_compacted(false);
        self.fill_layout(&mut segwr, &mut parts, &counts, &raw_inode, &dirty_meta_vec, &ordered, &ref_src, true)?;

        // Collected so the blocks can be demoted after the upload; the tier's
        // borrow cannot be held across that.
        let written: Vec<BlockIndex> = ordered.iter().filter(|(_, n)| n.is_some()).map(|(i, _)| *i).collect();
        let written_refs: Vec<BlockIndex> = ordered.iter().filter(|(_, n)| n.is_none()).map(|(i, _)| *i).collect();
        drop(ordered);
        drop(dirty_data_blocks);

        // The data parts first, then the one whose meta blocks point into them.
        for part in parts {
            part.done().await?;
        }
        segwr.done().await?;

        // The upload landed, so these are readable from storage and may leave the
        // dirty tier. Nothing before this line may.
        for n in dirty_meta_vec {
            n.clear_dirty();
        }
        self.bmap_clear_dirty();
        self.demote_data_blocks_dirty(&written);
        self.refs_remove(&written_refs);
        self.set_last_flush();

        debug!("partial segment {} written, {} blocks left the dirty tier and {} references were resolved",
            segid, written.len(), written_refs.len());
        Ok(segid)
    }}

    /// Build the objects of a checkpoint: every pointer assigned, every writer
    /// filled. Returns the data-only parts, uploaded first, then the object holding
    /// the meta blocks and the inode, which is what makes it a checkpoint.
    ///
    /// `first_data_part` is the part the first data-only part takes. The caller
    /// passes it rather than it being read here: when the checkpoint is already
    /// part-way out, the caller reserves the numbers the parts take before this
    /// runs, so a retry never reuses a name an attempt has already written.
    #[allow(clippy::type_complexity)]
    fn flush_process_build_segment(&self, dirty_data_blocks: DirtyDataBlocks<'_>, first_data_part: u16)
            -> impl Future<Output = Result<(Vec<segment::Writer<T>>, segment::Writer<T>, SegmentId, InodeRaw, Vec<BtreeNodeDirty<'_, BlockIndex, V, BlockPtr>>)>>
    {async move {
        // prepare for a segment
        let _start = Instant::now();

        // 1. collect all dirty meta data
        //
        // Note the ordering constraint this imposes on everything that
        // dirties a data block: this set is fixed *here*, before the
        // loop below assigns pointers to data blocks. A bmap node that
        // is not already dirty at this point is never written, so the
        // pointer `bmap_assign_data_node` stores into it exists only in
        // memory, and a cold reader keeps following the old one.
        //
        // So every path that puts a block into the dirty tier must also
        // insert into the bmap for that index in the same flush window,
        // which is what marks the containing node dirty. The write
        // paths do it for every block they touch; see `block_mut` and
        // `truncate_last_data_block` for two places where omitting it
        // caused silent loss, visible only once the map had spilled out
        // of the inode's inline root — below that the root travels with
        // the inode, which every flush writes.
        let dirty_meta_vec = self.bmap_lookup_dirty();
        debug!("start to create a new segemtnt: dirty meta nodes {}, dirty data blocks {}",
            dirty_meta_vec.len(), dirty_data_blocks.len());

        let dirty_data_blocks = self.get_data_blocks_dirty();
        // Blocks written by reference go into the same segment, and their bytes
        // are read from the source as each part uploads.
        let refs = self.refs_snapshot();
        let ref_src = if refs.is_empty() { None } else {
            Some(self.block_source_get().expect("references exist only while a source is set"))
        };
        // Dirty and referenced blocks in one index order, so a checkpoint holding
        // references is laid out exactly as it would be had they been written by
        // value.
        let mut ordered: Vec<(BlockIndex, Option<&DataBlock>)> =
            dirty_data_blocks.data().iter().map(|(i, n)| (*i, Some(*n))).collect();
        if !refs.is_empty() {
            ordered.extend(refs.iter().map(|i| (*i, None)));
            ordered.sort_by_key(|(i, _)| *i);
            // Every path that changes a block takes its reference out first, so the
            // two tiers are disjoint; were they not, the block would be stored twice.
            assert!(ordered.windows(2).all(|w| w[0].0 != w[1].0),
                "a block is both referenced and dirty");
        }

        // Laid out before a number is taken or a pointer assigned, so a checkpoint
        // that cannot be written is refused with nothing changed.
        let meta_bytes: usize = dirty_meta_vec.iter().map(|n| n.size()).sum();
        let counts = self.plan_objects(meta_bytes, ordered.len())?;

        // A checkpoint already part-way out keeps its number: it is one checkpoint
        // however many objects it took, and asking for a fresh one here would
        // orphan every partial written under it.
        let segid = match self.pending_parts_get() {
            Some(p) => p.segid,
            None => self.inode_mut().get_next_seq(),
        };
        let data_ids: Vec<SegmentId> = (1..counts.len())
            .map(|k| segid.at_part(first_data_part + (k - 1) as u16)).collect();
        let (mut segwr, mut parts) = self.assign_layout(segid, false, &data_ids, &counts,
            &dirty_meta_vec, &ordered).await?;

        let _ = _start.elapsed();

        // prepare inode
        let b = self.bmap_get_raw();
        let mut raw_inode = self.inode().to_raw(b);
        raw_inode.i_last_cno = segid.as_cno();
        raw_inode.set_compacted(self.compaction_flag(segid));

        let _start = Instant::now();
        self.fill_layout(&mut segwr, &mut parts, &counts, &raw_inode, &dirty_meta_vec, &ordered, &ref_src, !cfg!(all(feature = "concurrent-segment-build", feature = "wal")))?;
        #[cfg(all(feature = "concurrent-segment-build", feature = "wal"))]
        {

        const TARGET_CHUNKS: usize = 50;
        let data_blocks = dirty_data_blocks.data()
                        .into_iter()
                        .map(|(_, block)| block)
                        .collect::<Vec<&DataBlock>>();
        // Split `data_blocks` into up to TARGET_CHUNKS chunks.
        //
        // Using ceil-div here is deliberate: floor-div of
        // `len / (TARGET_CHUNKS - 1)` yields 0 whenever
        // `len < TARGET_CHUNKS - 1`, and a `split_off(0)` loop
        // with chunk_size=0 never terminates. Ceil-div also
        // guarantees chunk_size >= 1 for any non-empty input,
        // so the number of chunks is at most TARGET_CHUNKS.
        let mut joins = Vec::new();
        if !data_blocks.is_empty() {
            let chunk_size = data_blocks.len().div_ceil(TARGET_CHUNKS);
            for chunk in data_blocks.chunks(chunk_size) {
                let j = segwr.spawn_append(chunk.to_vec());
                joins.push(j);
            }
        }

        // NOTE: the wait loop below is a CPU-bound busy-wait, not
        // an async wait. It keeps the caller's runtime pinned on
        // one core while the spawn_blocking workers make progress
        // on the blocking pool. That's tolerable on a multi-threaded
        // runtime (other workers remain free), and happens to work
        // on the reactor's current-thread runtime only because we
        // spawn onto the blocking pool (decoupled from the runtime's
        // worker).
        //
        // Known limitation: a panic inside the spawn_blocking
        // closure is silently swallowed. `is_finished()` returns
        // true for both "completed" and "panicked" handles, and we
        // drop the JoinHandle without calling `.await` or inspecting
        // its JoinError. For the current workload (memcpy into a
        // pre-sized buffer) a panic shouldn't happen, but this is
        // load-bearing only by accident.
        //
        // Replace with `tokio::task::JoinSet` when any of the
        // following becomes true:
        //   - panics in the copy closure need to surface as flush
        //     errors (today they're swallowed);
        //   - flush latency matters and the spinning core becomes a
        //     measurable cost;
        //   - we move the flush path to a single-threaded runtime
        //     with no blocking pool.
        //
        // Sketch of the replacement:
        //   let mut set = JoinSet::new();
        //   for chunk in data_blocks.chunks(chunk_size) {
        //       set.spawn_blocking(move || { /* copy */ });
        //   }
        //   while let Some(res) = set.join_next().await {
        //       res.map_err(|e| /* JoinError -> our Error */)?;
        //   }

        // wait all spawn append completed
        while let Some(res) = joins.pop() {
            let join = res?;
            if !join.is_finished() {
                joins.push(Ok(join));
            }
        }

        }
        let _ = _start.elapsed();
        Ok((parts, segwr, segid, raw_inode, dirty_meta_vec))
    }}

    /// Seal the records written since the last barrier, naming the checkpoint
    /// they belong to.
    ///
    /// Ordered deliberately: after every record is confirmed written, because a
    /// barrier claims the group is whole; and before the segment is built,
    /// because building it advances `last_seq` and the records are under the
    /// old one. Together those two make "a barrier exists for this segid" mean
    /// "the group is complete and was never published".
    ///
    /// Nothing to seal is not an error. A flush with no records behind it —
    /// metadata only, or a replay, which takes the log out while it runs —
    /// writes a barrier over an empty manifest, which recovery reads as a
    /// complete group of nothing.
    #[cfg(feature = "wal")]
    fn wal_seal_barrier(&mut self) -> impl Future<Output = Result<()>> {async {
        let segid = self.inode().get_last_seq();
        let fut = match self.wal_mut() {
            Some(wal) => wal.write_barrier(segid),
            None => return Ok(()),
        };
        fut.await
    }}

    /// The most bytes one object of a streamed checkpoint holds: the configured
    /// limit, but never less than one full leaf's data with the leaf, its summary
    /// entries and the header — so a leaf and its data always share an object, and
    /// every object describes itself.
    fn stream_object_bytes(&self) -> usize {
        use crate::stream_segment as ss;
        let ms = self.config().meta.meta_block_size;
        let bs = self.config().meta.data_block_size;
        // Above what a leaf can hold: a leaf entry is a key and a pointer.
        let per_leaf = ms / 16;
        let floor = per_leaf * bs + ms + per_leaf * ss::ENTRY_SIZE + ss::HEADER_SIZE;
        self.max_object_bytes().max(floor).min(self.config().meta.block_ptr_format.max_object_bytes())
    }

    /// Write the dirty set as objects laid out tail first; see
    /// [`crate::stream_segment`].
    ///
    /// Each object holds whole leaves of the block map with their dirty data, and
    /// describes itself: its data, those leaves, a summary and a header. A leaf is
    /// never split from its data, so the objects are cut at leaves.
    ///
    /// `commit` writes the checkpoint: data-only parts first, then part 0 with the
    /// index nodes above the leaves, the leaves that hold no new data, whatever
    /// data still fits, and the inode — the object that makes it a checkpoint —
    /// then the inode object. Nothing leaves the dirty tier until both have landed,
    /// so a failure changes nothing a retry depends on.
    ///
    /// Without `commit`, data-only parts are written as partials and their blocks
    /// leave the dirty tier as each call finishes; the index nodes above the leaves
    /// stay in memory until the checkpoint is written.
    ///
    /// The checkpoint's number is taken once and kept by every attempt, and each
    /// attempt reserves fresh part numbers before writing any, so none is reused.
    #[allow(clippy::type_complexity)]
    fn stream_flush(&mut self, commit: bool) -> impl Future<Output = Result<SegmentId>> {async move {
        use crate::stream_segment::{self as ss, TailBuilder};
        use crate::segment_body::SegmentBody;
        use btree_ondisk::node::{BtreeNode, BTREE_NODE_LEVEL_MIN};
        use futures::stream::{FuturesUnordered, StreamExt};

        // What was sent ahead is in by the time the rest is laid out.
        // Boxed for the reason `try_flush` boxes sending ahead.
        Box::pin(self.stream_settle(true)).await?;
        if commit {
            let (segid, _) = self.flush_process_pre_build_segment().await?;
            if segid > 0 {
                self.flush_timing().flush_count.fetch_add(1, Ordering::Relaxed);
                return Ok(SegmentId::new_from_cno(segid));
            }
        } else {
            // Before the blocks leave the dirty tier, judged now.
            self.fold_partial_into_compaction();
        }
        let bs = self.config().meta.data_block_size;
        let ms = self.config().meta.meta_block_size;
        let cap = self.stream_object_bytes();

        let (number, first) = match self.pending_parts_get() {
            Some(p) => (p.segid, p.next_part),
            None => (self.inode_mut().get_next_seq(), segment::Segment::FIRST_DATA_PART),
        };
        let dirty_meta = self.bmap_lookup_dirty();
        let dirty_data = self.get_data_blocks_dirty();
        let refs = self.refs_snapshot();
        let src = if refs.is_empty() { None } else {
            Some(self.block_source_get().expect("references exist only while a source is set"))
        };
        let mut ordered: Vec<(BlockIndex, Option<&DataBlock>)> =
            dirty_data.data().iter().map(|(i, n)| (*i, Some(*n))).collect();
        if !refs.is_empty() {
            ordered.extend(refs.iter().map(|i| (*i, None)));
            ordered.sort_by_key(|(i, _)| *i);
            assert!(ordered.windows(2).all(|w| w[0].0 != w[1].0), "a block is both referenced and dirty");
        }

        // Leaves, by their first key, and the index nodes above them.
        let mut leaves: Vec<(BlockIndex, usize)> = Vec::new();
        let mut upper: Vec<usize> = Vec::new();
        for (i, n) in dirty_meta.iter().enumerate() {
            let node = BtreeNode::<BlockIndex, BlockPtr, BlockPtr>::from_slice_ref(n.as_slice())?;
            if node.get_level() == BTREE_NODE_LEVEL_MIN && node.get_nchild() > 0 {
                leaves.push((*node.get_key(0), i));
            } else {
                upper.push(i);
            }
        }
        leaves.sort_by_key(|l| l.0);

        // Each block goes with the leaf holding it: the last whose first key is not
        // past it. A map without leaves — the inode's root holds the pointers —
        // has one group, with no leaf.
        struct Group { leaf: Option<usize>, start: usize, end: usize }
        let mut groups: Vec<Group> = Vec::new();
        let mut with_data = vec![false; leaves.len()];
        for (k, (blk_idx, _)) in ordered.iter().enumerate() {
            let at = leaves.partition_point(|l| l.0 <= *blk_idx);
            let leaf = at.checked_sub(1);
            match groups.last_mut() {
                Some(g) if g.leaf == leaf.map(|l| leaves[l].1) => g.end = k + 1,
                _ => groups.push(Group { leaf: leaf.map(|l| leaves[l].1), start: k, end: k + 1 }),
            }
            if let Some(l) = leaf {
                with_data[l] = true;
            }
        }
        let bare: Vec<usize> = leaves.iter().zip(with_data.iter())
            .filter(|(_, d)| !**d).map(|(l, _)| l.1).collect();
        let gsize = |g: &Group| (g.end - g.start) * (bs + ss::ENTRY_SIZE) + if g.leaf.is_some() { ms } else { 0 };
        if let Some(g) = groups.iter().find(|g| ss::HEADER_SIZE + gsize(g) > cap) {
            return Err(Error::new(ErrorKind::InvalidInput, format!(
                "{} blocks under one leaf need {} bytes, more than an object of {} holds",
                g.end - g.start, gsize(g), cap)));
        }

        // Objects: data-only parts from the front; with `commit`, whatever still
        // fits beside part 0's index nodes goes into part 0.
        let part0_fixed = (upper.len() + bare.len()) * ms + ss::HEADER_SIZE;
        if commit && part0_fixed > cap {
            return Err(Error::new(ErrorKind::InvalidInput, format!(
                "{} index nodes need {} bytes, more than an object of {} holds",
                upper.len() + bare.len(), part0_fixed, cap)));
        }
        let mut suffix = vec![0usize; groups.len() + 1];
        for g in (0..groups.len()).rev() {
            suffix[g] = suffix[g + 1] + gsize(&groups[g]);
        }
        let mut parts: Vec<Vec<usize>> = Vec::new();
        let mut in_part0: Vec<usize> = Vec::new();
        let mut cur: Vec<usize> = Vec::new();
        let mut cur_bytes = ss::HEADER_SIZE;
        for g in 0..groups.len() {
            if commit && cur.is_empty() && part0_fixed + suffix[g] <= cap {
                in_part0 = (g..groups.len()).collect();
                break;
            }
            if !cur.is_empty() && cur_bytes + gsize(&groups[g]) > cap {
                parts.push(std::mem::take(&mut cur));
                cur_bytes = ss::HEADER_SIZE;
                if commit && part0_fixed + suffix[g] <= cap {
                    in_part0 = (g..groups.len()).collect();
                    break;
                }
            }
            cur.push(g);
            cur_bytes += gsize(&groups[g]);
        }
        if !cur.is_empty() {
            parts.push(cur);
        }
        if first as usize + parts.len() >= segment::Segment::MAX_PARTS {
            return Err(Error::new(ErrorKind::OutOfMemory, format!(
                "checkpoint {} needs {} more parts after part {}, past the {} a pointer can name — flush \\
                 more often", number, parts.len(), first, segment::Segment::MAX_PARTS)));
        }
        // The number, and every part this attempt writes, held before any is
        // written: a retry keeps the number and writes under parts past these. A
        // refusal above is not retried, so it holds nothing.
        self.pending_parts_reserve(number, first + parts.len() as u16);
        self.flush_timing().stream_flush_bytes.fetch_add((ordered.len() * bs) as u64, Ordering::Relaxed);
        debug!("stream flush of {}: {} blocks, {} leaves, {} index nodes above them, {} data parts{}",
            number, ordered.len(), leaves.len(), upper.len(), parts.len(), if commit { " and part 0" } else { "" });

        let base = self.inode().base();
        let mut informative = self.inode().to_raw(self.bmap_get_raw());
        informative.i_last_cno = number.as_cno();
        informative.set_compacted(false);

        // Place one object's blocks and leaves; return its body so far and its tail.
        // A pointer's sequence is ignored by the parted encoding.
        let mut written: Vec<BlockIndex> = Vec::new();
        let mut written_refs: Vec<BlockIndex> = Vec::new();
        let mut written_leaves: Vec<usize> = Vec::new();
        let mut objects: Vec<(SegmentId, SegmentBody, TailBuilder)> = Vec::new();
        let layout: Vec<(SegmentId, &Vec<usize>, bool)> = parts.iter().enumerate()
            .map(|(k, p)| (number.at_part(first + k as u16), p, true))
            .chain(commit.then(|| (number.at_part(segment::Segment::SUMMARY_PART), &in_part0, false)))
            .collect();
        for (id, part, partial) in layout {
            let mut body = SegmentBody::new();
            let mut tail = TailBuilder::new(id, ms, bs, partial, base);
            let mut j = 0usize;
            for g in part.iter().map(|g| &groups[*g]) {
                for (blk_idx, n) in &ordered[g.start..g.end] {
                    let ptr = self.blk_ptr_encode(id, ss::data_off(j, bs), j);
                    self.bmap_assign_data_node(blk_idx, ptr).await?;
                    tail.data(*blk_idx, ptr);
                    match n {
                        Some(n) => { body.push(n.bytes_view()); written.push(*blk_idx); },
                        None => {
                            body.push_lazy(src.clone().expect("a reference has a source"), *blk_idx * bs as u64, bs);
                            written_refs.push(*blk_idx);
                        },
                    }
                    j += 1;
                }
            }
            let own_leaves: Vec<usize> = part.iter().filter_map(|g| groups[*g].leaf).collect();
            let mut nodes: Vec<usize> = own_leaves.clone();
            if !partial {
                nodes.extend(bare.iter().copied());
                nodes.extend(upper.iter().copied());
            }
            for (m, i) in nodes.iter().enumerate() {
                let ptr = self.blk_ptr_encode(id, j * bs + m * ms, j + m);
                self.bmap_assign_meta_node(ptr, dirty_meta[*i].clone()).await?;
                tail.meta();
            }
            written_leaves.extend(own_leaves);
            objects.push((id, body, tail));
            // The nodes' bytes are taken once every pointer into them is set, which
            // for part 0's is only after the last object's blocks: see below.
            let _ = nodes;
        }

        // Every pointer is set: the nodes' bytes are final. Part 0 also carries the
        // inode, whose map root is final now too.
        let mut raw_inode = informative;
        if commit {
            raw_inode = self.inode().to_raw(self.bmap_get_raw());
            raw_inode.i_last_cno = number.as_cno();
            raw_inode.set_compacted(self.compaction_flag(number));
        }
        let mut bodies: Vec<(SegmentId, SegmentBody)> = Vec::with_capacity(objects.len());
        let nparts = parts.len();
        for (k, (id, mut body, tail)) in objects.into_iter().enumerate() {
            let is_part0 = commit && k == nparts;
            let part: &Vec<usize> = if is_part0 { &in_part0 } else { &parts[k] };
            let mut nodes: Vec<usize> = part.iter().filter_map(|g| groups[*g].leaf).collect();
            if is_part0 {
                nodes.extend(bare.iter().copied());
                nodes.extend(upper.iter().copied());
            }
            for i in nodes {
                body.push(bytes::Bytes::copy_from_slice(dirty_meta[i].as_slice()));
            }
            body.push(bytes::Bytes::from(tail.finish(if is_part0 { &raw_inode } else { &informative })));
            bodies.push((id, body));
        }

        // Data-only parts, several at once; part 0 after every one has landed.
        let _start = Instant::now();
        let part0 = if commit { bodies.pop() } else { None };
        let limit = self.config().runtime.segment_mpu_max_concurrency.max(1);
        let mut inflight = FuturesUnordered::new();
        let mut queue = bodies.into_iter();
        loop {
            while inflight.len() < limit {
                let Some((id, body)) = queue.next() else { break };
                let st = self.staging().clone();
                inflight.push(async move { st.done_pieces(id, body).await });
            }
            match inflight.next().await {
                Some(r) => r?,
                None => break,
            }
        }
        let Some((id0, body0)) = part0 else {
            // Partials only: their blocks may leave the dirty tier now.
            for i in written_leaves {
                dirty_meta[i].clear_dirty();
            }
            drop(ordered);
            drop(dirty_data);
            drop(dirty_meta);
            self.demote_data_blocks_dirty(&written);
            self.refs_remove(&written_refs);
            if let Some(last) = parts.len().checked_sub(1) {
                self.pending_parts_advance(number.at_part(first + last as u16));
            }
            self.set_last_flush();
            debug!("stream partials of {} written: {} blocks left the dirty tier", number, written.len());
            return Ok(number.at_part(first + parts.len().saturating_sub(1) as u16));
        };
        self.staging().clone().done_pieces(id0, body0).await
            .map_err(|e| PublishIncomplete::maybe(e, number.checkpoint()))?;
        self.inode_mut().set_last_cno(number.as_cno());
        self.flush_timing().segment_done_ns.fetch_add(_start.elapsed().as_nanos() as u64, Ordering::Relaxed);

        let _start = Instant::now();
        let od_state = self.staging().flush_inode(raw_inode.as_u8_slice(), self.inode().get_ondisk_state(), FlushInodeFlag::Update).await
            .map_err(|e| PublishIncomplete::landed(e, number.checkpoint()))?;
        self.flush_timing().flush_inode_ns.fetch_add(_start.elapsed().as_nanos() as u64, Ordering::Relaxed);
        self.inode_mut().clear_attr_dirty();
        self.inode_mut().set_ondisk_state(od_state);
        self.inode_mut().set_base(&raw_inode);
        self.checkpoint_published(&raw_inode);
        let last_cno = self.inode().get_last_cno();
        self.inode_mut().set_last_ondisk_cno(last_cno);

        for n in dirty_meta.iter() {
            n.clear_dirty();
        }
        drop(ordered);
        drop(dirty_data);
        drop(dirty_meta);
        self.clear_data_blocks_dirty();
        self.refs_clear();
        self.bmap_clear_dirty();
        self.pending_parts_clear();
        self.set_last_flush();
        self.flush_timing().flush_count.fetch_add(1, Ordering::Relaxed);
        // See `flush_process`: the log prefixes this checkpoint supersedes.
        #[cfg(feature = "wal")]
        self.wal_drop_superseded_prefixes(number);
        Ok(SegmentId::new_from_cno(self.inode().get_last_ondisk_cno()))
    }}

    /// Under the stream format with the WAL: the checkpoint as one object, part 0,
    /// laid out tail first — the same layout a flush without the WAL writes — and
    /// built whole, for a WAL publish to pin while it uploads. Every block's and
    /// node's pointer is assigned; nothing is uploaded and nothing left dirty is
    /// cleared.
    #[cfg(feature = "wal")]
    #[allow(clippy::type_complexity)]
    fn stream_build_pinned(&self) -> impl Future<Output = Result<(segment::WalSegment<T>, SegmentId, InodeRaw,
        Vec<BtreeNodeDirty<'_, BlockIndex, V, BlockPtr>>)>>
    {async move {
        use crate::stream_segment::{self as ss, TailBuilder};
        let bs = self.config().meta.data_block_size;
        let ms = self.config().meta.meta_block_size;
        assert!(self.refs_snapshot().is_empty(), "write_ref is refused with the WAL");
        let dirty_meta = self.bmap_lookup_dirty();
        let dirty_data = self.get_data_blocks_dirty();
        let blocks: Vec<BlockIndex> = dirty_data.data().keys().copied().collect();
        let len = ss::object_len(blocks.len(), bs, dirty_meta.len(), ms);
        if len > self.stream_object_bytes() {
            return Err(Error::new(ErrorKind::InvalidInput, format!(
                "with the WAL a checkpoint is one object, and this one needs {} bytes, more than the {} \
                 one may hold; flush more often", len, self.stream_object_bytes())));
        }
        let number = match self.pending_parts_get() {
            Some(p) => p.segid,
            None => self.inode_mut().get_next_seq(),
        };
        let id = number.at_part(segment::Segment::SUMMARY_PART);
        let mut tail = TailBuilder::new(id, ms, bs, false, self.inode().base());
        for (k, blk_idx) in blocks.iter().enumerate() {
            let ptr = self.blk_ptr_encode(id, ss::data_off(k, bs), k);
            self.bmap_assign_data_node(blk_idx, ptr).await?;
            tail.data(*blk_idx, ptr);
        }
        let meta_off = blocks.len() * bs;
        for (m, n) in dirty_meta.iter().enumerate() {
            let ptr = self.blk_ptr_encode(id, meta_off + m * ms, blocks.len() + m);
            self.bmap_assign_meta_node(ptr, n.clone()).await?;
            tail.meta();
        }
        // Every pointer is set: the nodes' bytes and the map's root are final.
        let mut raw_inode = self.inode().to_raw(self.bmap_get_raw());
        raw_inode.i_last_cno = number.as_cno();
        raw_inode.set_compacted(self.compaction_flag(number));
        let mut data = Vec::with_capacity(len);
        {
            let d = dirty_data.data();
            for b in blocks.iter() {
                data.extend_from_slice(d[b].as_slice());
            }
        }
        for n in dirty_meta.iter() {
            data.extend_from_slice(n.as_slice());
        }
        data.extend_from_slice(&tail.finish(&raw_inode));
        debug_assert_eq!(data.len(), len);
        drop(dirty_data);
        let seg = segment::WalSegment::Stream { ctx: self.staging().clone(), object: id,
            data: std::sync::Arc::new(Box::pin(data)) };
        Ok((seg, number, raw_inode, dirty_meta))
    }}

    /// The segment a WAL publish uploads and pins: in the container's layout.
    #[cfg(feature = "wal")]
    #[allow(clippy::type_complexity)]
    fn wal_build_segment<'s>(&'s self, dirty_data_blocks: DirtyDataBlocks<'s>)
        -> impl Future<Output = Result<(segment::WalSegment<T>, SegmentId, InodeRaw, Vec<BtreeNodeDirty<'s, BlockIndex, V, BlockPtr>>)>>
    {async move {
        if self.config().meta.block_ptr_format.is_stream() {
            drop(dirty_data_blocks);
            return self.stream_build_pinned().await;
        }
        let (parts, segwr, segid, raw_inode, dirty_meta_vec) =
            self.flush_process_build_segment(dirty_data_blocks, segment::Segment::FIRST_DATA_PART).await?;
        // The WAL writes a checkpoint as one object: `plan_objects` refuses one
        // that would need more, so there are no data-only parts here.
        assert!(parts.is_empty());
        Ok((segment::WalSegment::Layout(segwr), segid, raw_inode, dirty_meta_vec))
    }}

    fn flush_process(&mut self) -> impl Future<Output = Result<SegmentId>> {async move {
        let fn_start = Instant::now();
        debug!("flush started");

        // Seal the log group before anything else: after every record is
        // confirmed written, and before the segment build advances `last_seq`
        // past the checkpoint those records are under. Sealing nothing is a
        // no-op, so the case where this flush turns out to have nothing to
        // publish costs nothing and writes no barrier.
        #[cfg(feature = "wal")]
        self.wal_seal_barrier().await?;
        if self.config().meta.block_ptr_format.is_stream() {
            return self.stream_flush(true).await;
        }
        let first_data_part = self.reserve_data_parts()?;
        let _start = Instant::now();
        let (segid, dirty_data_blocks) = self.flush_process_pre_build_segment().await?;
        self.flush_timing().pre_build_ns.fetch_add(
            _start.elapsed().as_nanos() as u64, Ordering::Relaxed);
        if segid > 0 {
            self.flush_timing().flush_count.fetch_add(1, Ordering::Relaxed);
            return Ok(SegmentId::new_from_cno(segid));
        }
        // Seal the log group before the segment exists, so a crash between the
        // two leaves a complete group to replay rather than an unpublished
        // segment with no way to tell whether its records were all there.
        let _start = Instant::now();
        let (parts, segwr, segid, raw_inode, dirty_meta_vec) = self.flush_process_build_segment(dirty_data_blocks, first_data_part).await?;
        self.flush_timing().build_segment_ns.fetch_add(
            _start.elapsed().as_nanos() as u64, Ordering::Relaxed);

        let _start = Instant::now();
        // Data-only parts first, then the part that makes it a checkpoint: an object
        // holding the inode on storage is a commit to whoever finds it, so every
        // block it points at has to be there before it is. None of these is a
        // commit, so a failure here leaves nothing to take up.
        for part in parts {
            part.done().await?;
        }
        segwr.done().await.map_err(|e| PublishIncomplete::maybe(e, segid.checkpoint()))?;
        // update last cno in memory after segment write out
        self.inode_mut().set_last_cno(segid.as_cno());
        self.flush_timing().segment_done_ns.fetch_add(
            _start.elapsed().as_nanos() as u64, Ordering::Relaxed);

        // flush inode after writeout segment
        let _start = Instant::now();
        let od_state = self.staging().flush_inode(raw_inode.as_u8_slice(), self.inode().get_ondisk_state(), FlushInodeFlag::Update).await
            .map_err(|e| PublishIncomplete::landed(e, segid.checkpoint()))?;
        self.flush_timing().flush_inode_ns.fetch_add(
            _start.elapsed().as_nanos() as u64, Ordering::Relaxed);
        self.inode_mut().clear_attr_dirty();
        self.inode_mut().set_ondisk_state(od_state);
        self.inode_mut().set_base(&raw_inode);
        self.checkpoint_published(&raw_inode);
        let last_cno = self.inode().get_last_cno();
        self.inode_mut().set_last_ondisk_cno(last_cno);

        // start to cleanup
        let _start = Instant::now();

        // clear dirty for all dirty meta node
        for n in dirty_meta_vec {
            n.clear_dirty();
        }

        self.clear_data_blocks_dirty();
        // Published: the referenced blocks are now read from storage like any other.
        self.refs_clear();

        // clear dirty for bmap
        self.bmap_clear_dirty();
        // The checkpoint is complete, so nothing is accumulating any more.
        self.pending_parts_clear();
        // reset last flush
        self.set_last_flush();

        self.flush_timing().cleanup_ns.fetch_add(
            _start.elapsed().as_nanos() as u64, Ordering::Relaxed);
        self.flush_timing().flush_count.fetch_add(1, Ordering::Relaxed);
        let _ = fn_start.elapsed();

        // WAL cleanup: the chunks feeding into this segment live
        // under (segid - 1) (flush called get_next_seq() when
        // allocating segid). Fire-and-forget; see the twin
        // wal_flush_done path for the same pattern on the reactor
        // WAL flush.
        #[cfg(feature = "wal")]
        {
            self.wal_drop_superseded_prefixes(segid);
        }

        Ok(SegmentId::new_from_cno(self.inode().get_last_ondisk_cno()))
    }}

    #[cfg(all(feature = "wal", feature = "reactor"))]
    fn wal_flush_process_reactor<'a: 'static>(&mut self, fh: ChannelGroup<FileContext<'a>>, lock: OwnedMutexGuard<()>)
            -> impl Future<Output = std::result::Result<SegmentId, (OwnedMutexGuard<()>, Error)>>
    {async move {
        let fn_start = Instant::now();
        debug!("flush started");

        // See `wal_seal_barrier`: after the records, before the segment.
        if let Err(e) = self.wal_seal_barrier().await {
            return Err((lock, e));
        }

        // A flush may be satisfied by the log alone. The data is durable — each
        // write made it so before returning — so what is deferred is the
        // checkpoint, not the data.
        //
        // The sequence still has to move, or the next group's records would land
        // under this checkpoint and the next barrier would overwrite this one's,
        // claiming two groups as one.
        //
        // The dirty-data thresholds stay on, and they are what bounds how much
        // can pile up here. Recovery replays each deferred group in turn, so a
        // mount after a crash pays what these flushes did not.
        if self.wal_may_defer_publish() {
            let segid = self.inode().get_last_seq();
            self.inode_mut().set_last_seq(segid.next());
            self.wal_count_deferred_barrier();
            self.set_last_flush();
            let last_cno = self.inode().get_last_cno();
            self.flush_unlock(lock);
            debug!("flush satisfied by the log, checkpoint deferred; last published {}", last_cno);
            return Ok(SegmentId::new_from_cno(last_cno));
        }

        let (segid, dirty_data_blocks) = match self.flush_process_pre_build_segment().await {
            Ok((segid, dirty_data_blocks)) => (segid, dirty_data_blocks),
            Err(e) => return Err((lock, e)),
        };
        if segid > 0 {
            // manually unlock
            self.set_last_flush();
            self.flush_unlock(lock);
            return Ok(SegmentId::new_from_cno(segid));
        }
        let (segwr, segid, raw_inode, dirty_meta_vec) = match self.wal_build_segment(dirty_data_blocks).await {
            Ok(built) => built,
            Err(e) => return Err((lock, e)),
        };

        let staging = self.staging().clone();
        let od_state = self.inode().get_ondisk_state().clone();
        // set bmap cache to unlimit, restore back until flush done
        let bmap_cache_limit = self.bmap_set_cache_unlimited();
        let (mem_segid, mem_segdata) = segwr.get_weak_data();
        self.wal_set_mem_segment(mem_segid, mem_segdata).await;
        let base = raw_inode.fingerprint();
        tokio::task::spawn(async move {
            let _start = Instant::now();
            match segwr.done().await {
                Ok(_) => {},
                Err(e) => {
                    warn!("segment write failed in wal flush process: {:?}", e);
                    let ctx = FileContext::new_wal_flush_recovery(lock);
                    let _ = fh.send_cb(ctx);
                    return;
                },
            }
            match staging.flush_inode(raw_inode.as_u8_slice(), &od_state, FlushInodeFlag::Update).await {
                Ok(od_state) => {
                    let ctx = FileContext::new_wal_flush_done(lock, segid, od_state.unwrap().clone(), bmap_cache_limit, base);
                    let _ = fh.send_cb(ctx);
                },
                Err(e) => {
                    warn!("flush inode failed in wal flush process: {:?}", e);
                    let ctx = FileContext::new_wal_flush_recovery(lock);
                    let _ = fh.send_cb(ctx);
                },
            }
        });
        self.inode_mut().set_last_cno(segid.as_cno());
        self.inode_mut().clear_attr_dirty();

        // start to cleanup
        let _start = Instant::now();

        // clear dirty for all dirty meta node
        for n in dirty_meta_vec {
            n.clear_dirty();
        }

        // set data blocks cache unlimited
        self.set_data_blocks_cache_unlimited();

        self.clear_data_blocks_dirty();

        // clear dirty for bmap
        self.bmap_clear_dirty();
        // The checkpoint is complete, so nothing is accumulating any more.
        self.pending_parts_clear();
        // reset last flush
        // defer set last flash in wal_flush_done

        let _ = _start.elapsed();
        let _ = fn_start.elapsed();
        // return cno in inode memory instead cno on disk
        Ok(SegmentId::new_from_cno(self.inode().get_last_cno()))
    }}

    #[cfg(all(feature = "wal", feature = "blocking"))]
    fn wal_flush_process_blocking<'a>(&mut self) -> impl Future<Output = Result<SegmentId>> {async move {
        let fn_start = Instant::now();
        debug!("flush started");

        let (segid, dirty_data_blocks) = self.flush_process_pre_build_segment().await?;
        if segid > 0 {
            return Ok(SegmentId::new_from_cno(segid));
        }
        // The WAL writes a checkpoint as one object; see the reactor twin.
        let (segwr, segid, raw_inode, dirty_meta_vec) = self.wal_build_segment(dirty_data_blocks).await?;

        let staging = self.staging().clone();
        let od_state = self.inode().get_ondisk_state().clone();
        let published = raw_inode;
        let join = tokio::task::spawn(async move {
            match segwr.done().await {
                Ok(_) => {},
                Err(e) => {
                    warn!("segment write failed in wal flush process: {:?}", e);
                    return Err(e);
                },
            }
            match staging.flush_inode(raw_inode.as_u8_slice(), &od_state, FlushInodeFlag::Update).await {
                Ok(od_state) => { return Ok(od_state); },
                Err(e) => {
                    warn!("flush inode failed in wal flush process: {:?}", e);
                    return Err(e);
                },
            }
        });

        let od_state = join.await??;

        self.inode_mut().set_last_cno(segid.as_cno());
        self.inode_mut().clear_attr_dirty();
        self.inode_mut().set_ondisk_state(od_state);
        self.inode_mut().set_base(&published);
        self.checkpoint_published(&published);
        let last_cno = self.inode().get_last_cno();
        self.inode_mut().set_last_ondisk_cno(last_cno);

        // start to cleanup
        let _start = Instant::now();

        // clear dirty for all dirty meta node
        for n in dirty_meta_vec {
            n.clear_dirty();
        }

        self.clear_data_blocks_dirty();

        // clear dirty for bmap
        self.bmap_clear_dirty();
        // The checkpoint is complete, so nothing is accumulating any more.
        self.pending_parts_clear();
        // reset last flush
        self.set_last_flush();

        let _ = _start.elapsed();
        let _ = fn_start.elapsed();
        Ok(SegmentId::new_from_cno(self.inode().get_last_ondisk_cno()))
    }}

    // flush out dirty data
    // original flush process
    fn _flush_process(&mut self) -> impl Future<Output = Result<SegmentId>> {async move {

        let fn_start = Instant::now();
        debug!("flush started");

        let dirty_data_blocks = self.get_data_blocks_dirty();

        // A checkpoint part-way out still has to be finished, and after its
        // partials there is nothing dirty left to notice. Its part 0 — the summary
        // naming every object, the metadata, the inode — is what makes it openable
        // at all, so quitting here would leave the partials referenced by nothing
        // and the flush would have done nothing while reporting success.
        let complete_pending = self.pending_parts_get().is_some();

        if !complete_pending && dirty_data_blocks.len() == 0 && !self.bmap_dirty() {
            if self.inode().is_attr_dirty() {
                debug!("inode attr is dirty, flush inode ONLY");
                let b = self.bmap_get_raw();
                let raw_inode = self.inode().to_raw(b);
                let od_state = self.staging().flush_inode(raw_inode.as_u8_slice(), self.inode().get_ondisk_state(), FlushInodeFlag::Update).await?;
                self.inode_mut().clear_attr_dirty();
                self.inode_mut().set_ondisk_state(od_state);
                let last_cno = self.inode().get_last_cno();
                self.inode_mut().set_last_ondisk_cno(last_cno);
            }
            debug!("flush quit, NO dirty data blocks amd bmap is NOT dirty");
            return Ok(self.inode().get_last_seq());
        }

        // prepare for a segment
        let _start = Instant::now();

        // 1. collect all dirty meta data
        //
        // Note the ordering constraint this imposes on everything that
        // dirties a data block: this set is fixed *here*, before the
        // loop below assigns pointers to data blocks. A bmap node that
        // is not already dirty at this point is never written, so the
        // pointer `bmap_assign_data_node` stores into it exists only in
        // memory, and a cold reader keeps following the old one.
        //
        // So every path that puts a block into the dirty tier must also
        // insert into the bmap for that index in the same flush window,
        // which is what marks the containing node dirty. The write
        // paths do it for every block they touch; see `block_mut` and
        // `truncate_last_data_block` for two places where omitting it
        // caused silent loss, visible only once the map had spilled out
        // of the inode's inline root — below that the root travels with
        // the inode, which every flush writes.
        let dirty_meta_vec = self.bmap_lookup_dirty();
        debug!("start to create a new segemtnt: dirty meta nodes {}, dirty data blocks {}",
            dirty_meta_vec.len(), dirty_data_blocks.len());

        // A checkpoint already part-way out keeps its number: it is one checkpoint
        // however many objects it took, and asking for a fresh one here would
        // orphan every partial written under it.
        let segid = match self.pending_parts_get() {
            Some(p) => p.segid,
            None => self.inode_mut().get_next_seq(),
        };
        let mut file_off = 0;
        let mut segwr = self.staging().new_segwr(segid, &self.config().meta);

        let dirty_data_blocks = self.get_data_blocks_dirty();
        let ndatadirty = dirty_data_blocks.len();
        file_off += segment::Writer::<T>::calc_ss_aligned_bytes(ndatadirty);

        let mut block_seq = 0;
        // assign real blk ptr to meta data
        for n in &dirty_meta_vec {
            let blk_ptr = self.blk_ptr_encode(segid, file_off, block_seq);
            let node_size = n.size();
            // use 0 as key, but it's useless
            debug!("assign block ptr for meta node: block ptr {}", self.blk_ptr_decode_display(&blk_ptr));
            self.bmap_assign_meta_node(blk_ptr, n.clone()).await?;
            segwr.inc_metablk();
            file_off += node_size;
            block_seq += 1;
        }

        // 2. collect all dirty data block

        // assign real blk ptr to data block
        for (blk_idx, n) in dirty_data_blocks.data().iter() {
            let blk_ptr = self.blk_ptr_encode(segid, file_off, block_seq);
            let block_size = n.size();
            debug!("assign block ptr for data node: block ptr {}, block index {}", self.blk_ptr_decode_display(&blk_ptr), blk_idx);
            self.bmap_assign_data_node(blk_idx, blk_ptr).await?;
            segwr.inc_datablk(blk_idx, &blk_ptr);
            file_off += block_size;
            block_seq += 1;
        }

        let _ = _start.elapsed();

        // prepare inode
        let b = self.bmap_get_raw();
        let mut raw_inode = self.inode().to_raw(b);
        raw_inode.i_last_cno = segid.as_cno();
        // TODO: calc segment checksum
        segwr.realize_ss(0, &raw_inode);

        // write out root node to staging file
        let _start = Instant::now();
        for n in &dirty_meta_vec {
            let _ = segwr.append(n.as_slice())?;
        }
        for (_, n) in dirty_data_blocks.data().iter() {
            let _ = segwr.append_data_block(n)?;
        }
        let _ = _start.elapsed();

        let _start = Instant::now();
        segwr.done().await?;
        // update last cno in memory after segment write out
        self.inode_mut().set_last_cno(segid.as_cno());
        let _ = _start.elapsed();

        // flush inode after writeout segment
        let _start = Instant::now();
        let od_state = self.staging().flush_inode(raw_inode.as_u8_slice(), self.inode().get_ondisk_state(), FlushInodeFlag::Update).await?;
        let _ = _start.elapsed();
        self.inode_mut().clear_attr_dirty();
        self.inode_mut().set_ondisk_state(od_state);
        let last_cno = self.inode().get_last_cno();
        self.inode_mut().set_last_ondisk_cno(last_cno);

        // start to cleanup
        let _start = Instant::now();

        // clear dirty for all dirty meta node
        for n in dirty_meta_vec {
            n.clear_dirty();
        }

        self.clear_data_blocks_dirty();

        // clear dirty for bmap
        self.bmap_clear_dirty();
        // The checkpoint is complete, so nothing is accumulating any more.
        self.pending_parts_clear();
        // reset last flush
        self.set_last_flush();

        let _ = _start.elapsed();
        let _ = fn_start.elapsed();
        Ok(SegmentId::new_from_cno(self.inode().get_last_ondisk_cno()))
    }}

    fn flush(&mut self) -> impl Future<Output = Result<SegmentId>> {async {
        use crate::config::FlushConflictPolicy;
        // A file that cannot publish must stop trying. Without this every
        // operation that flushes first — truncate among them — would reattempt
        // the failed publish and report its error rather than the read-only
        // state, and a steady stream of them would keep the retry cycle
        // running for good.
        self.check_writable()?;
        // Asking to publish inside an interval that asked for no publishing is
        // a contradiction, and answering `Ok` to it would tell the caller its
        // data is at a checkpoint when it is not.
        #[cfg(feature = "wal")]
        if self.in_txn_trait() {
            return Err(Error::new(ErrorKind::InvalidInput,
                "a transaction is open, so publishing was asked not to happen; commit it to publish"));
        }
        let policy = self.config().runtime.flush_conflict_policy;

        let mut retries = 0;
        let mut backoff = DEFAULT_FLUSH_BACKOFF_SECS;
        // What an earlier attempt left on storage. Each attempt takes a fresh
        // checkpoint number, so a later attempt can fail on something else entirely
        // -- or the retries run out -- and the error returned has to say it still,
        // or the rollback cannot abandon the checkpoint and the next open takes up
        // a flush its caller was told failed.
        let mut left: Option<(SegmentId, bool)> = None;
        while retries < DEFAULT_FLUSH_RETRIES {
            let lock = self.flush_lock().await;
            match self.flush_process().await {
                Ok(segid) => {
                    self.flush_unlock(lock);
                    return Ok(segid);
                },
                Err(err) => {
                    if let Some(inner) = err.get_ref().and_then(|i| i.downcast_ref::<PublishIncomplete>()) {
                        left = Some((inner.segid, inner.landed));
                    }
                    let err = PublishIncomplete::carry(err, left);
                    let kind = err.kind();

                    // AlreadyExists signals an OCC conflict from the storage
                    // layer (S3 412/409). Behavior is policy-controlled.
                    if kind == ErrorKind::AlreadyExists {
                        match policy {
                            FlushConflictPolicy::FailFast => {
                                // Do not retry: surface the conflict to the
                                // caller so it can re-read state and decide.
                                self.flush_unlock(lock);
                                return Err(err);
                            },
                            FlushConflictPolicy::RetryLastWriterWins => {
                                // Fall through to the retry path below.
                                warn!("{err}");
                            },
                        }
                    } else if kind != ErrorKind::ResourceBusy {
                        self.flush_unlock(lock);
                        return Err(err);
                    } else {
                        warn!("{err}");
                    }
                },
            }
            self.flush_unlock(lock);
            // back off sleep
            Self::sleep(Duration::from_secs(backoff)).await;
            let _ = self.refresh_bmap().await.map_err(|e| PublishIncomplete::carry(e, left))?;
            retries += 1;
            backoff += DEFAULT_FLUSH_BACKOFF_SECS;
        }
        let err_str = format!("FLUSH - reached max retries {}", retries);
        return Err(PublishIncomplete::carry(Error::new(ErrorKind::ResourceBusy, err_str), left));
    }}
}