tape-sdk 0.4.4

//! Stream write implementation.

use std::time::Duration;

use futures::stream::{self, FuturesOrdered, Stream, StreamExt};
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::sync::{mpsc, oneshot, Mutex};
use tokio::time::sleep;
use tracing::{debug, warn};

use tape_rpc::{CommitmentLevel, Rpc};
use tape_api::program::tapedrive::track_pda;
use tape_api::state::Tape;
use tape_core::object::object_etag;
use tape_core::prelude::CompressedTrack;
use tape_core::track::data::{track_key, BlobDataSlice};
use tape_core::track::mirror::ArchiveMirror;
use tape_core::track::types::CompressedTrackProof;
use tape_core::types::ContentType;
use tape_core::types::{StorageUnits, TrackNumber};
use tape_crypto::hash::hash;
use tape_crypto::Hash;
use tape_protocol::Api;
use tape_protocol::api::CertifyRes;

use crate::error::TapedriveError;
use crate::keys::operator::TapeOperator;
use crate::keys::tape_key::TapeKey;
use crate::metrics::{Operation, Phase};
use crate::tapedrive::Tapedrive;
use crate::track::write::{
    certified_track, certify_submit_with_retry, certify_with_retry, coded_identity,
    collect_certification, encode_blob, ensure_track_matches, finish_coded_track, inline_write_fits,
    needs_fresh_signatures, plan_blob, register_blob_processed, resolve_sent_blob,
    should_retry_certification, submit_blob_with_logical_size, submit_certification_with_proof,
    submit_raw_with_logical_size, upload_with_retry, wait_for_certified_track, UploadPlan,
    WrittenTrack, UNNAMED_TRACK, UNTYPED_TRACK,
};
use crate::transfer::certify::CollectedSignatures;

use super::error::StreamError;
use super::manifest::{
    ChunkEntry, ChunkManifest, MAX_TRACK_SIZE, MAX_TRACKS_PER_TAPE, MANIFEST_VERSION,
};
use super::receipt::StreamReceipt;

/// Maximum track slots in a tape (2^TRACK_TREE_HEIGHT).
const MAX_TRACKS: TrackNumber = TrackNumber(MAX_TRACKS_PER_TAPE);

/// Encoded chunks buffered ahead of the serial register submits, so encoding
/// (seconds of CPU per chunk) overlaps register confirmations.
const ENCODE_AHEAD: usize = 2;

/// Concurrent chunk encodes, bounded by available cores. Each in-flight
/// encode holds its input and coded slices (~3x the chunk), so four workers
/// add up to ~768 MiB peak on top of the store stage.
const MAX_ENCODE_WORKERS: usize = 4;

/// Track-written event fetches kept in flight behind the register submits.
/// Events need confirmed level, so this hides that wait while the registers
/// themselves only pay a processed-level wait each.
const RESOLVE_CONCURRENCY: usize = 3;

/// Stored chunks whose signatures are collected ahead of the serial certify
/// submits. Collections hold no slice data, so lookahead is cheap.
const COLLECT_LOOKAHEAD: usize = 2;

/// Attempts to resubmit a conflicted certify against refetched chain state
/// before falling back to the peer-proof path.
const CERTIFY_CONFLICT_ATTEMPTS: usize = 3;

/// Delay before each conflict refetch; the interfering transaction may be
/// processed but not yet confirmed, so an immediate refetch can miss it.
const CERTIFY_CONFLICT_DELAY: Duration = Duration::from_millis(400);

/// Root comparisons made at the end of a write before the mirror is dropped.
const ROOT_CHECK_ATTEMPTS: usize = 10;

/// Flat delay between root comparisons, the confirmed root only trails briefly
const ROOT_CHECK_DELAY: Duration = Duration::from_millis(400);

impl<Blockchain: Rpc, Cluster: Api> Tapedrive<Blockchain, Cluster> {
    /// Store a named byte stream and return before final manifest certification.
    ///
    /// The returned manifest and receipts can be passed to
    /// [`Self::certify_with_receipts`]. Data chunks have already been certified;
    /// only the stream's final manifest may remain to certify.
    pub async fn store_named_stream<Reader: AsyncRead + Unpin>(
        &self,
        tape_key: &TapeKey,
        name: impl AsRef<[u8]>,
        content_type: ContentType,
        size: StorageUnits,
        reader: Reader,
    ) -> Result<(WrittenTrack, Vec<CertifyRes>), TapedriveError> {
        self.store_named_stream_as(tape_key, name, content_type, size, reader)
            .await
    }

    /// Store a named byte stream as its owner or an authorized delegate.
    ///
    /// Data chunks are certified before this returns. The final manifest has
    /// landed on storage peers but may still require certification through
    /// [`Self::certify_with_receipts_as`].
    pub async fn store_named_stream_as<Reader: AsyncRead + Unpin>(
        &self,
        operator: &impl TapeOperator,
        name: impl AsRef<[u8]>,
        content_type: ContentType,
        size: StorageUnits,
        reader: Reader,
    ) -> Result<(WrittenTrack, Vec<CertifyRes>), TapedriveError> {
        let timer = self
            .timer(Operation::WriteStream, Phase::Total)
            .bytes(size.to_bytes());
        let result = store_stream(
            self,
            operator,
            name.as_ref(),
            content_type,
            size,
            reader,
        )
        .await;
        timer.finish_result(&result);
        result.map(|manifest| (manifest.written, manifest.receipts))
    }
}

/// Write in-memory bytes as a multi-track stream.
pub async fn write_bytes<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    data: &[u8],
) -> Result<StreamReceipt, TapedriveError> {
    let manifest =
        store_bytes(client, tape_key, name, content_type, data).await?;
    complete_stream(client, tape_key, &manifest).await
}

/// Write bytes from an async reader as a multi-track stream.
pub async fn write_stream<Blockchain: Rpc, Cluster: Api, Reader: AsyncRead + Unpin>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    size: StorageUnits,
    reader: Reader,
) -> Result<StreamReceipt, TapedriveError> {
    let manifest =
        store_stream(client, tape_key, name, content_type, size, reader).await?;
    complete_stream(client, tape_key, &manifest).await
}

struct StoredManifest {
    written: WrittenTrack,
    receipts: Vec<CertifyRes>,
    etag: Hash,
}

// Chunks are internal fragments addressed by track number, never by name.
struct PendingChunk {
    pub entry: ChunkEntry,
    pub written: WrittenTrack,
    pub receipts: Vec<CertifyRes>,
}

// A registered chunk whose encoded slices still need to be stored.
struct RegisteredChunk {
    entry: ChunkEntry,
    written: WrittenTrack,
    plan: UploadPlan,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ManifestWriteMode {
    Inline,
    Coded,
}

/// Validate stream-level input before any track writes begin.
fn validate_stream_size(size: StorageUnits) -> Result<(), StreamError> {
    if size.is_zero() {
        return Err(StreamError::InvalidInput(
            "empty streams are not supported".into(),
        ));
    }

    Ok(())
}

/// Compute the number of chunks required for a stream.
fn chunk_count_for_size(size: StorageUnits) -> Result<TrackNumber, StreamError> {
    let chunk_count = size.to_bytes().div_ceil(MAX_TRACK_SIZE as u64);
    Ok(TrackNumber(chunk_count))
}

/// Return the byte offset for a chunk index.
fn chunk_offset(chunk_index: usize) -> Result<StorageUnits, StreamError> {
    let chunk_index = u64::try_from(chunk_index)
        .map_err(|_| StreamError::InvalidInput("stream has too many chunks".into()))?;
    let offset = chunk_index
        .checked_mul(MAX_TRACK_SIZE as u64)
        .ok_or_else(|| StreamError::InvalidInput("stream size overflow".into()))?;
    Ok(StorageUnits::from_bytes(offset))
}

/// Return the stored byte size for a chunk at the given index.
fn chunk_size(
    chunk_index: usize,
    chunk_count: TrackNumber,
    total_size: StorageUnits,
) -> Result<StorageUnits, StreamError> {
    if chunk_index + 1 == chunk_count.as_usize() {
        let offset = chunk_offset(chunk_index)?;
        total_size
            .checked_sub(offset)
            .ok_or_else(|| StreamError::InvalidInput("stream size underflow".into()))
    } else {
        Ok(StorageUnits::from_bytes(MAX_TRACK_SIZE as u64))
    }
}

/// Build deterministic manifest entries before or after upload.
fn build_entries(
    start_track_number: TrackNumber,
    chunk_count: TrackNumber,
    total_size: StorageUnits,
) -> Result<Vec<ChunkEntry>, StreamError> {
    let mut entries = Vec::with_capacity(chunk_count.as_usize());

    for chunk_index in 0..chunk_count.as_usize() {
        let track_number = start_track_number
            .checked_add(TrackNumber(chunk_index as u64))
            .ok_or_else(|| StreamError::InvalidInput("chunk track number overflow".into()))?;

        entries.push(ChunkEntry {
            track_number,
            offset: chunk_offset(chunk_index)?,
            size: chunk_size(chunk_index, chunk_count, total_size)?,
        });
    }

    Ok(entries)
}

/// Build a chunk manifest from ordered entries.
fn build_manifest(
    key: Hash,
    total_size: StorageUnits,
    entries: Vec<ChunkEntry>,
) -> Result<ChunkManifest, StreamError> {
    let chunk_count = TrackNumber(
        u64::try_from(entries.len())
            .map_err(|_| StreamError::InvalidInput("stream has too many chunks".into()))?,
    );

    Ok(ChunkManifest {
        version: MANIFEST_VERSION,
        total_size,
        chunk_count,
        chunk_size: StorageUnits::from_bytes(MAX_TRACK_SIZE as u64),
        key,
        chunks: entries,
    })
}

/// Verify the tape has enough capacity and track slots for the stream.
fn preflight(
    tape: &Tape,
    total_required_bytes: StorageUnits,
    tracks_needed: TrackNumber,
) -> Result<(), TapedriveError> {
    let available_capacity = tape.capacity.saturating_sub(tape.used);
    if total_required_bytes > available_capacity {
        return Err(TapedriveError::InsufficientCapacity {
            need: total_required_bytes,
            available: available_capacity,
        });
    }

    let used_tracks = tape.tracks.next_number();
    let available_tracks = MAX_TRACKS.saturating_sub(used_tracks);

    if tracks_needed > available_tracks {
        let chunk_count = tracks_needed
            .checked_prev()
            .ok_or_else(|| stream_error(StreamError::InvalidInput("stream needs no data tracks".into())))?;

        return Err(stream_error(StreamError::InsufficientTrackSlots {
            available: available_tracks,
            needed: tracks_needed,
            chunks: chunk_count,
        }));
    }

    Ok(())
}

async fn store_bytes<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    data: &[u8],
) -> Result<StoredManifest, TapedriveError> {
    let size = StorageUnits::from_bytes(data.len() as u64);

    // A prior interrupted write of this stream leaves a matching prefix of tracks
    // starting at 0; finish it in place rather than appending a duplicate. Only a
    // tape whose first track is already this stream's first chunk resumes, so a
    // fresh tape and a shared bucket both fall through to the write pipeline.
    let chunk_count = chunk_count_for_size(size).map_err(stream_error)?;
    let first = stream_chunk(data, 0, chunk_count, size)?;
    if is_stream_resume(client, tape_key, first).await? {
        return resume_stream(client, tape_key, name, content_type, data, size, chunk_count).await;
    }

    let (tape, chunk_count) = prepare_write(client, tape_key, name, size).await?;
    let chunk_sources = stream::iter(
        data.chunks(MAX_TRACK_SIZE)
            .map(|chunk| Ok::<_, TapedriveError>(chunk.to_vec())),
    );
    let pending_chunks =
        pipeline_chunks(client, tape_key, &tape, size, chunk_count, chunk_sources).await?;

    store_manifest_for_chunks(client, tape_key, name, content_type, size, pending_chunks).await
}

async fn store_stream<Blockchain: Rpc, Cluster: Api, Reader: AsyncRead + Unpin>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    size: StorageUnits,
    mut reader: Reader,
) -> Result<StoredManifest, TapedriveError> {
    validate_stream_size(size).map_err(stream_error)?;
    let chunk_count = chunk_count_for_size(size).map_err(stream_error)?;

    // Read the first chunk to detect an interrupted prior write of this stream.
    // On resume this chunk is finished; on a fresh write it is prepended back
    // onto the pipeline so nothing is lost to the peek.
    let first = read_chunk(&mut reader, 0, chunk_count, size).await?;
    if is_stream_resume(client, tape_key, &first).await? {
        return resume_stream_reader(
            client, tape_key, name, content_type, size, chunk_count, first, reader,
        )
        .await;
    }

    let (tape, chunk_count) = prepare_write(client, tape_key, name, size).await?;
    let chunk_sources = stream::once(async move { Ok::<_, TapedriveError>(first) }).chain(
        stream::unfold((&mut reader, 1usize), move |(reader, chunk_index)| async move {
            if chunk_index >= chunk_count.as_usize() {
                return None;
            }
            let result = read_chunk(reader, chunk_index, chunk_count, size).await;
            Some((result, (reader, chunk_index + 1)))
        }),
    );
    let pending_chunks =
        pipeline_chunks(client, tape_key, &tape, size, chunk_count, chunk_sources).await?;

    verify_stream_drained(&mut reader).await?;
    store_manifest_for_chunks(client, tape_key, name, content_type, size, pending_chunks).await
}

/// Resume an interrupted reader stream: finish the already-read first chunk,
/// then read and finish the rest in order, then the manifest.
async fn resume_stream_reader<Blockchain: Rpc, Cluster: Api, Reader: AsyncRead + Unpin>(
    client: &Tapedrive<Blockchain, Cluster>,
    operator: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    size: StorageUnits,
    chunk_count: TrackNumber,
    first: Vec<u8>,
    mut reader: Reader,
) -> Result<StoredManifest, TapedriveError> {
    let timer = client
        .timer(Operation::WriteStream, Phase::Total)
        .bytes(size.to_bytes());
    let result = async {
        resume_stream_chunk(client, operator, &first, TrackNumber(0)).await?;
        for chunk_index in 1..chunk_count.as_usize() {
            let chunk = read_chunk(&mut reader, chunk_index, chunk_count, size).await?;
            resume_stream_chunk(client, operator, &chunk, TrackNumber(chunk_index as u64)).await?;
        }
        verify_stream_drained(&mut reader).await?;
        store_resumed_manifest(client, operator, name, content_type, size, chunk_count).await
    }
    .await;
    timer.finish_result(&result);
    result
}

/// Validate the write upfront, returning the fetched tape and the chunk
/// count. The tape carries the pre-stream track tree the pipeline seeds its
/// mirror from, so the pipeline does not refetch it.
async fn prepare_write<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    name: &[u8],
    size: StorageUnits,
) -> Result<(Tape, TrackNumber), TapedriveError> {
    let timer = client
        .timer(Operation::WriteStream, Phase::Preflight)
        .bytes(size.to_bytes());

    let result = async {
        validate_stream_size(size).map_err(stream_error)?;
        let chunk_count = chunk_count_for_size(size).map_err(stream_error)?;
        let tracks_needed = chunk_count.checked_next().ok_or_else(|| {
            stream_error(StreamError::InvalidInput("stream has too many chunks".into()))
        })?;

        let entries = build_entries(TrackNumber(0), chunk_count, size).map_err(stream_error)?;
        let manifest = build_manifest(hash(name), size, entries).map_err(stream_error)?;
        let manifest_bytes = manifest.to_bytes().map_err(stream_error)?;
        let total_size = size
            .checked_add(StorageUnits::from_bytes(manifest_bytes.len() as u64))
            .ok_or_else(|| stream_error(StreamError::InvalidInput("stream size overflow".into())))?;

        let tape = client.get_tape(&tape_key.address()).await?;
        preflight(&tape, total_size, tracks_needed)?;
        Ok((tape, chunk_count))
    }
    .await;
    timer.finish_result(&result);
    result
}

/// Whether this stream's data already sits on the tape as an interrupted prior
/// write. True only when the tape has tracks and its first track is this
/// stream's first chunk, so a fresh tape and an unrelated bucket both fall
/// through to the write pipeline.
async fn is_stream_resume<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    operator: &impl TapeOperator,
    first_chunk: &[u8],
) -> Result<bool, TapedriveError> {
    let tape = client.get_tape(&operator.address()).await?;
    if tape.tracks.next_number() == TrackNumber(0) {
        return Ok(false);
    }
    stream_chunk_matches(client, operator, TrackNumber(0), first_chunk).await
}

/// The byte slice of chunk `chunk_index` in the source data.
fn stream_chunk(
    data: &[u8],
    chunk_index: usize,
    chunk_count: TrackNumber,
    size: StorageUnits,
) -> Result<&[u8], TapedriveError> {
    let start = chunk_offset(chunk_index).map_err(stream_error)?.as_usize();
    let len = chunk_size(chunk_index, chunk_count, size)
        .map_err(stream_error)?
        .as_usize();
    data.get(start..start + len)
        .ok_or_else(|| stream_error(StreamError::InvalidInput("chunk out of range".into())))
}

/// Whether track `track_number` is already the coded chunk `chunk_data` encodes
/// to. Chunks are unnamed and content addressed, so identity is the encoded key
/// and value hash.
async fn stream_chunk_matches<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    track_number: TrackNumber,
    chunk_data: &[u8],
) -> Result<bool, TapedriveError> {
    let existing = match client
        .get_track_by_number(&tape_key.address(), track_number)
        .await
    {
        Ok(track) => track,
        Err(TapedriveError::NotFound) => return Ok(false),
        Err(other) => return Err(other),
    };
    // Cheap reject before the expensive encode: a chunk is a coded track whose
    // stored size equals the chunk's byte length. A different object at this
    // position (e.g. a named object at track 0 of a shared bucket) fails here
    // without paying an up-to-64-MiB erasure encode.
    if !existing.is_coded() || existing.size.to_bytes() != chunk_data.len() as u64 {
        return Ok(false);
    }
    let (_, key, value_hash) = coded_identity(client, UNNAMED_TRACK, chunk_data, Operation::WriteStream).await?;
    Ok(existing.key == key && existing.value_hash == value_hash)
}

/// Resume an interrupted stream: finish or write every data chunk in order,
/// then the manifest. Each track completes independently, so a re-run converges
/// to a single certified stream without duplicating chunks or re-spending
/// capacity on the ones already written.
async fn resume_stream<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    data: &[u8],
    size: StorageUnits,
    chunk_count: TrackNumber,
) -> Result<StoredManifest, TapedriveError> {
    let timer = client
        .timer(Operation::WriteStream, Phase::Total)
        .bytes(size.to_bytes());
    let result = async {
        for chunk_index in 0..chunk_count.as_usize() {
            let chunk = stream_chunk(data, chunk_index, chunk_count, size)?;
            resume_stream_chunk(client, tape_key, chunk, TrackNumber(chunk_index as u64)).await?;
        }
        store_resumed_manifest(client, tape_key, name, content_type, size, chunk_count).await
    }
    .await;
    timer.finish_result(&result);
    result
}

/// Store (or verify) the manifest for a resumed stream.
/// Both resume paths build the same manifest from the fixed chunk layout after
/// finishing every data chunk.
async fn store_resumed_manifest<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    operator: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    size: StorageUnits,
    chunk_count: TrackNumber,
) -> Result<StoredManifest, TapedriveError> {
    let entries = build_entries(TrackNumber(0), chunk_count, size).map_err(stream_error)?;
    let manifest = build_manifest(hash(name), size, entries).map_err(stream_error)?;
    let manifest_bytes = manifest.to_bytes().map_err(stream_error)?;
    ensure_stream_manifest(
        client,
        operator,
        name,
        content_type,
        size,
        &manifest_bytes,
        chunk_count,
    )
    .await
}

/// Finish or write one coded data chunk at its track number. Chunks are always
/// coded (the write pipeline never inlines them), so resume treats even a small
/// final chunk as coded to keep the kind consistent.
async fn resume_stream_chunk<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    chunk_data: &[u8],
    track_number: TrackNumber,
) -> Result<CompressedTrack, TapedriveError> {
    let tape = tape_key.address();
    match client.get_track_by_number(&tape, track_number).await {
        Ok(existing) => {
            let (plan, key, value_hash) =
                coded_identity(client, UNNAMED_TRACK, chunk_data, Operation::WriteStream).await?;
            ensure_track_matches(&existing, key, value_hash)?;
            finish_coded_track(client, tape_key, existing, &plan, Operation::WriteStream).await
        }
        Err(TapedriveError::NotFound) => {
            let logical_size = StorageUnits::from_bytes(chunk_data.len() as u64);
            let (written, plan) = submit_blob_with_logical_size(
                client,
                tape_key,
                UNNAMED_TRACK,
                UNTYPED_TRACK,
                logical_size,
                chunk_data,
                Operation::WriteStream,
                client.write_options.visibility,
            )
            .await?;
            verify_track_number(&written, track_number)?;
            let receipts =
                upload_with_retry(client, &written, &plan, Operation::WriteStream).await?;
            certify_with_retry(client, tape_key, &written, Operation::WriteStream, &receipts).await
        }
        Err(other) => Err(other),
    }
}

/// Finish or write the manifest track, mirroring the write path's inline/coded
/// choice so a resumed manifest matches an interrupted one exactly.
async fn ensure_stream_manifest<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    size: StorageUnits,
    manifest_bytes: &[u8],
    track_number: TrackNumber,
) -> Result<StoredManifest, TapedriveError> {
    let tape = tape_key.address();
    let existing = match client.get_track_by_number(&tape, track_number).await {
        Ok(track) => track,
        Err(TapedriveError::NotFound) => {
            let manifest =
                store_manifest(client, tape_key, name, content_type, size, manifest_bytes).await?;
            verify_track_number(&manifest.written, track_number)?;
            return Ok(manifest);
        }
        Err(other) => return Err(other),
    };

    match manifest_write_mode(name, manifest_bytes) {
        ManifestWriteMode::Inline => {
            let slice = BlobDataSlice::Inline(manifest_bytes);
            let meta = slice.meta().ok_or_else(|| {
                TapedriveError::Encoding("inline manifest has no commitment".into())
            })?;
            ensure_track_matches(&existing, track_key(name, &slice), meta.value_hash)?;
            // Inline tracks certify at register, so a matching one is complete.
            Ok(StoredManifest {
                written: WrittenTrack {
                    address: track_pda(tape, track_number).0,
                    track: existing,
                },
                receipts: Vec::new(),
                etag: object_etag(&existing, None),
            })
        }
        ManifestWriteMode::Coded => {
            let (plan, key, value_hash) =
                coded_identity(client, name, manifest_bytes, Operation::WriteStream).await?;
            ensure_track_matches(&existing, key, value_hash)?;
            let written = WrittenTrack {
                address: track_pda(tape, track_number).0,
                track: existing,
            };
            let receipts = if written.track.is_certified() {
                Vec::new()
            } else {
                upload_with_retry(client, &written, &plan, Operation::WriteStream).await?
            };
            Ok(StoredManifest {
                etag: object_etag(&written.track, Some(&plan_blob(&plan))),
                written,
                receipts,
            })
        }
    }
}

/// A register during resume must land at the track number resume expects; a
/// mismatch means the tape gained a track between processes, so refuse.
fn verify_track_number(
    written: &WrittenTrack,
    expected: TrackNumber,
) -> Result<(), TapedriveError> {
    if written.track.track_number == expected {
        Ok(())
    } else {
        Err(stream_error(StreamError::Integrity(format!(
            "resume expected track {expected}, registered at {}",
            written.track.track_number
        ))))
    }
}

/// Run chunk writes as a pipeline: register chunks one at a time at processed
/// level, resolve their track numbers concurrently in track order, keep up to
/// the configured store depth of slice uploads in flight, and certify stored
/// chunks strictly in track order behind the uploads. A local mirror of the tape's
/// track tree supplies certify proofs without per-chunk refetches. Stage
/// errors cancel the whole pipeline; incomplete tracks are left for the
/// recovery worker.
async fn pipeline_chunks<Blockchain, Cluster, Chunks>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    tape: &Tape,
    size: StorageUnits,
    chunk_count: TrackNumber,
    chunk_sources: Chunks,
) -> Result<Vec<PendingChunk>, TapedriveError>
where
    Blockchain: Rpc,
    Cluster: Api,
    Chunks: Stream<Item = Result<Vec<u8>, TapedriveError>>,
{
    // The mirror seeds from the pre-stream track tree; the resolve stage
    // appends every registered track and the certify stage proves against
    // and updates the same tree, so both share it behind a mutex.
    let mirror = Mutex::new(ArchiveMirror::new(&tape.tracks));
    let mirror = &mirror;

    let (encoded_sender, mut encoded_receiver) = mpsc::channel(ENCODE_AHEAD);
    let (sent_sender, mut sent_receiver) = mpsc::channel(1);
    let (registered_sender, mut registered_receiver) = mpsc::channel(1);
    let (stored_sender, mut stored_receiver) = mpsc::channel(chunk_count.as_usize().max(1));

    // Registers and certifies rewrite the same track tree, so a certify
    // proof built while a register is still in flight is guaranteed stale.
    // The resolve stage fires this once its loop completes, which is the
    // moment every register is mirrored; the certify stage holds its first
    // submit until then.
    let (registers_mirrored_sender, registers_mirrored_receiver) = oneshot::channel::<()>();

    // Encoding is CPU-bound and single-threaded per chunk; encode several
    // chunks on blocking threads at once so idle cores shorten the encode
    // chain. Plans still reach the register stage in stream order.
    let encode_stage = async move {
        let workers = std::thread::available_parallelism()
            .map(|cores| cores.get())
            .unwrap_or(1)
            .min(MAX_ENCODE_WORKERS);
        let mut in_flight = FuturesOrdered::new();
        let mut chunk_sources = std::pin::pin!(chunk_sources);
        let mut chunk_index = 0usize;
        while let Some(chunk_data) = chunk_sources.next().await {
            if in_flight.len() >= workers {
                let Some(encoded) = in_flight.next().await else { break };
                if encoded_sender.send(encoded?).await.is_err() {
                    return Ok(());
                }
            }
            let data = chunk_data?;
            let index = chunk_index;
            in_flight.push_back(async move {
                encode_blob(client, data, Operation::WriteStream)
                    .await
                    .map(|plan| (index, plan))
            });
            chunk_index += 1;
        }
        while let Some(encoded) = in_flight.next().await {
            if encoded_sender.send(encoded?).await.is_err() {
                break;
            }
        }
        Ok::<_, TapedriveError>(())
    };

    // Registers submit at processed level: the processed wait is enough to
    // keep track numbers assigned in stream order, and the confirmed-level
    // event wait moves into the resolve stage.
    let register_stage = async move {
        while let Some((chunk_index, plan)) = encoded_receiver.recv().await {
            let logical_size = plan.storage_units;
            let sent = register_blob_processed(
                client,
                tape_key,
                UNNAMED_TRACK,
                UNTYPED_TRACK,
                logical_size,
                plan,
                Operation::WriteStream,
                client.write_options.visibility,
            )
            .await?;
            if sent_sender.send((chunk_index, sent)).await.is_err() {
                break;
            }
        }
        Ok::<_, TapedriveError>(())
    };

    // Resolving fetches each register's track-written event at confirmed
    // level; FuturesOrdered overlaps those waits but emits chunks in track
    // order, which the mirror appends require.
    let resolve_stage = async move {
        let mut in_flight = FuturesOrdered::new();
        let mut is_registering = true;
        while is_registering || !in_flight.is_empty() {
            tokio::select! {
                // Safe: recv is cancellation-safe and a chunk only leaves the
                // channel when this branch completes.
                sent = sent_receiver.recv(),
                    if is_registering && in_flight.len() < RESOLVE_CONCURRENCY =>
                {
                    match sent {
                        Some((chunk_index, sent)) => in_flight.push_back(async move {
                            resolve_sent_blob(client, sent)
                                .await
                                .map(|resolved| (chunk_index, resolved))
                        }),
                        None => is_registering = false,
                    }
                }
                // Safe: FuturesOrdered::next only removes a future once it
                // completes; a cancelled poll leaves every resolve in place.
                resolved = in_flight.next(), if !in_flight.is_empty() => {
                    let Some(resolved) = resolved else { continue };
                    let (chunk_index, (written, plan)) = resolved?;
                    append_to_mirror(client, tape_key, mirror, &written).await?;
                    let registered = RegisteredChunk {
                        entry: ChunkEntry {
                            track_number: written.track.track_number,
                            offset: chunk_offset(chunk_index).map_err(stream_error)?,
                            size: chunk_size(chunk_index, chunk_count, size)
                                .map_err(stream_error)?,
                        },
                        written,
                        plan,
                    };
                    if registered_sender.send(registered).await.is_err() {
                        break;
                    }
                }
            }
        }
        // The loop only completes once every register sent has been
        // mirrored; error exits drop the sender without firing.
        let _ = registers_mirrored_sender.send(());
        Ok::<_, TapedriveError>(())
    };

    let store_stage = async move {
        // Each in-flight chunk holds ~2x its bytes encoded, so the depth also
        // bounds peak memory.
        let store_depth = client.write_options.store_depth.max(1);
        let mut in_flight = FuturesOrdered::new();
        let mut is_registering = true;
        while is_registering || !in_flight.is_empty() {
            tokio::select! {
                // Safe: recv is cancellation-safe and a chunk only leaves the
                // channel when this branch completes.
                registered = registered_receiver.recv(),
                    if is_registering && in_flight.len() < store_depth =>
                {
                    match registered {
                        Some(registered) => in_flight.push_back(store_chunk(client, registered)),
                        None => is_registering = false,
                    }
                }
                // Safe: FuturesOrdered::next only removes a future once it
                // completes; a cancelled poll leaves every upload in place.
                stored = in_flight.next(), if !in_flight.is_empty() => {
                    let Some(stored) = stored else { continue };
                    if stored_sender.send(stored?).await.is_err() {
                        break;
                    }
                }
            }
        }
        Ok::<_, TapedriveError>(())
    };

    // Signatures sign the track's leaf hash, so collection for the next chunk
    // can run while the previous chunk's certify transaction confirms.
    let (collected_sender, mut collected_receiver) = mpsc::channel(COLLECT_LOOKAHEAD);
    let collect_stage = async move {
        while let Some(pending) = stored_receiver.recv().await {
            let collected =
                collect_certification(
                    client,
                    &pending.written,
                    Operation::WriteStream,
                    &pending.receipts,
                )
                .await?;
            if collected_sender.send((pending, collected)).await.is_err() {
                break;
            }
        }
        Ok::<_, TapedriveError>(())
    };

    // Each certify mutates the tape's track tree, so a proof is only valid
    // against the root left by the previous certify; proofs come from the
    // shared mirror. Do not parallelize the submits.
    let certify_stage = async move {
        // Hold the first proof until the chain's register sequence is final;
        // after that the tree gains no new leaves until the manifest write,
        // so mirror proofs cannot be staled by this stream's own registers.
        // Collections keep running ahead; only proof generation and the
        // submits wait.
        registers_mirrored_receiver.await.map_err(|_| {
            stream_error(StreamError::Chunk(
                "resolve stage ended before mirroring every register".into(),
            ))
        })?;

        let mut pending_chunks = Vec::with_capacity(chunk_count.as_usize());
        while let Some((pending, collected)) = collected_receiver.recv().await {
            certify_chunk(
                client,
                tape_key,
                mirror,
                &pending.written,
                collected,
                &pending.receipts,
                Operation::WriteStream,
            )
            .await?;
            pending_chunks.push(pending);
        }

        // Every certify is on-chain; confirm peer visibility for all chunks at
        // once instead of once per chunk inside the serial loop.
        let visible = client
            .timer(Operation::WriteStream, Phase::CertifyVisible)
            .chunks(pending_chunks.len() as u64);
        let tape_address = tape_key.address();
        let result = futures::future::try_join_all(pending_chunks.iter().map(|pending| {
            wait_for_certified_track(client, &tape_address, pending.written.track.track_number)
        }))
        .await;
        visible.finish_result(&result);
        result?;

        // Peers already report every track certified, so the confirmed root
        // has had time to catch up with the processed certifies; a lasting
        // mismatch means an external writer touched the tape mid-stream.
        verify_mirror_root(client, tape_key, mirror).await?;

        Ok::<_, TapedriveError>(pending_chunks)
    };

    let ((), (), (), (), (), pending_chunks) = tokio::try_join!(
        encode_stage,
        register_stage,
        resolve_stage,
        store_stage,
        collect_stage,
        certify_stage
    )?;
    Ok(pending_chunks)
}

/// Mirror a resolved register, reseeding from chain state when the register sequence moved on
pub(crate) async fn append_to_mirror<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    mirror: &Mutex<ArchiveMirror>,
    written: &WrittenTrack,
) -> Result<(), TapedriveError> {
    {
        let mut guard = mirror.lock().await;
        if written.track.track_number < guard.next_number() {
            return Ok(());
        }
        if guard.append(&written.track).is_ok() {
            return Ok(());
        }
    }

    let reseeded = reseed_mirror(client, tape_key).await?;
    *mirror.lock().await = reseeded;
    debug!(
        tape = %tape_key.address(),
        track = %written.address,
        "mirror could not follow a foreign register; later proofs come from storage nodes"
    );
    Ok(())
}

/// Certify one stored chunk: mirror proof first, stale proof retried, dead signatures re-collected on the peer path
pub(crate) async fn certify_chunk<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    mirror: &Mutex<ArchiveMirror>,
    written: &WrittenTrack,
    collected: CollectedSignatures,
    banked: &[CertifyRes],
    operation: Operation,
) -> Result<(), TapedriveError> {
    let track_number = written.track.track_number;
    let certified = certified_track(&written.track);
    let mut has_valid_signatures = true;

    // Tracks a reseeded mirror can no longer prove go straight to the
    // fallback.
    let proof = mirror.lock().await.proof_for(track_number);
    if let Ok(proof) = proof {
        let submitted = submit_certification_with_proof(
            client,
            tape_key,
            proof,
            &collected,
            CommitmentLevel::Processed,
            operation,
        )
        .await;
        match submitted {
            Ok(()) => {
                return apply_certified_to_mirror(client, tape_key, mirror, &certified, Some(&proof))
                    .await
            }
            // A stale proof is regenerated; rejected signatures sign a past epoch and need collecting again
            Err(err) if should_retry_certification(&err) => {
                if needs_fresh_signatures(&err) {
                    has_valid_signatures = false;
                    client.state().invalidate();
                } else {
                    let done = recertify_after_conflict(
                        client,
                        tape_key,
                        mirror,
                        &certified,
                        &collected,
                        operation,
                    )
                    .await?;
                    if done {
                        return Ok(());
                    }
                }
            }
            Err(err) => return Err(err),
        }
    }

    let signatures = has_valid_signatures.then_some(collected);
    certify_submit_with_retry(client, tape_key, written, operation, signatures, banked).await?;
    apply_certified_to_mirror(client, tape_key, mirror, &certified, None).await
}

/// Retry a conflicted mirror certify against refetched chain state, reusing
/// the already-collected signatures. An interfering writer stales the proof
/// between generation and execution; a mirror still in lockstep with the
/// chain regenerates a valid proof, while any tape-tree mutation the mirror
/// never saw forces a reseed that drops the track out of its provable range.
/// Returns false when the caller must fall back to the peer-proof path.
async fn recertify_after_conflict<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    mirror: &Mutex<ArchiveMirror>,
    certified: &CompressedTrack,
    collected: &CollectedSignatures,
    operation: Operation,
) -> Result<bool, TapedriveError> {
    for _ in 0..CERTIFY_CONFLICT_ATTEMPTS {
        // The interfering transaction may only be processed, so give the
        // confirmed view time to include it before refetching.
        sleep(CERTIFY_CONFLICT_DELAY).await;

        let reseeded = reseed_mirror(client, tape_key).await?;
        let proof = {
            let mut mirror = mirror.lock().await;
            if reseeded.next_number() != mirror.next_number()
                || reseeded.root() != mirror.root()
            {
                *mirror = reseeded;
                return Ok(false);
            }
            mirror.proof_for(certified.track_number)
        };
        let Ok(proof) = proof else {
            return Ok(false);
        };

        let submitted = submit_certification_with_proof(
            client,
            tape_key,
            proof,
            collected,
            CommitmentLevel::Processed,
            operation,
        )
        .await;
        match submitted {
            Ok(()) => {
                apply_certified_to_mirror(client, tape_key, mirror, certified, Some(&proof))
                    .await?;
                return Ok(true);
            }
            Err(err) if should_retry_certification(&err) => {
                if needs_fresh_signatures(&err) {
                    return Ok(false);
                }
            }
            Err(err) => return Err(err),
        }
    }

    Ok(false)
}

/// Build a replacement mirror from refetched chain state. The certify stage
/// is the only mirror user once certifies begin (every register is mirrored
/// before the first proof), so the fetch runs before the caller takes the
/// mirror lock and no append can land in between; no lock is held across
/// the network await.
async fn reseed_mirror<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
) -> Result<ArchiveMirror, TapedriveError> {
    let tape = client.get_tape(&tape_key.address()).await?;
    Ok(ArchiveMirror::new(&tape.tracks))
}

/// Replay a landed certify on the mirror, reseeding when a covered track fails to apply
async fn apply_certified_to_mirror<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    mirror: &Mutex<ArchiveMirror>,
    certified: &CompressedTrack,
    proof: Option<&CompressedTrackProof>,
) -> Result<(), TapedriveError> {
    let applied = {
        let mut mirror = mirror.lock().await;
        if !mirror.covers(certified.track_number) {
            return Ok(());
        }
        match proof {
            Some(proof) => {
                mirror.apply_certified_with_proof(certified.track_number, certified, proof)
            }
            None => mirror.apply_certified(certified.track_number, certified),
        }
    };
    if applied.is_ok() {
        return Ok(());
    }

    match reseed_mirror(client, tape_key).await {
        Ok(reseeded) => *mirror.lock().await = reseeded,
        Err(err) => warn!(tape = %tape_key.address(), %err, "mirror reseed failed; later proofs fall back"),
    }
    Ok(())
}

/// Poll the chain root against the mirror and log a lasting mismatch, every track is already certified
pub(crate) async fn verify_mirror_root<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    mirror: &Mutex<ArchiveMirror>,
) -> Result<(), TapedriveError> {
    // A mirror reseeded from chain state holds none of this client's tracks,
    // so its root is a chain snapshot and comparing it proves nothing.
    if mirror.lock().await.is_empty() {
        debug!(tape = %tape_key.address(), "mirror holds no appended tracks; skipping root check");
        return Ok(());
    }

    let mut expected = Hash::default();
    let mut observed = Hash::default();
    for attempt in 0..ROOT_CHECK_ATTEMPTS {
        if attempt > 0 {
            sleep(ROOT_CHECK_DELAY).await;
        }

        // A certify landing during the poll moves the mirror, so reread it
        expected = mirror.lock().await.root();
        let tape = match client.get_tape(&tape_key.address()).await {
            Ok(tape) => tape,
            Err(err) => {
                warn!(tape = %tape_key.address(), %err, "mirror root check skipped; tape fetch failed");
                return Ok(());
            }
        };
        observed = tape.tracks.tree.root();
        if observed == expected {
            return Ok(());
        }
    }

    warn!(
        tape = %tape_key.address(),
        mirror_root = %expected,
        chain_root = %observed,
        "tape track tree diverged from the mirror; another writer touched the tape"
    );
    Ok(())
}

/// Read one chunk's bytes from the source reader.
async fn read_chunk<Reader: AsyncRead + Unpin>(
    reader: &mut Reader,
    chunk_index: usize,
    chunk_count: TrackNumber,
    total_size: StorageUnits,
) -> Result<Vec<u8>, TapedriveError> {
    let expected_chunk_size =
        chunk_size(chunk_index, chunk_count, total_size).map_err(stream_error)?;
    let mut chunk_data = vec![0u8; expected_chunk_size.as_usize()];
    read_chunk_exact(reader, &mut chunk_data).await?;
    Ok(chunk_data)
}

/// Read exactly one chunk from the source reader.
async fn read_chunk_exact<Reader: AsyncRead + Unpin>(
    reader: &mut Reader,
    chunk_data: &mut [u8],
) -> Result<(), TapedriveError> {
    match reader.read_exact(chunk_data).await {
        Ok(_) => Ok(()),
        Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => Err(stream_error(
            StreamError::InvalidInput("stream ended before declared size".into()),
        )),
        Err(error) => Err(TapedriveError::Io(error)),
    }
}

/// Ensure the source reader does not contain extra bytes beyond the declared size.
async fn verify_stream_drained<Reader: AsyncRead + Unpin>(
    reader: &mut Reader,
) -> Result<(), TapedriveError> {
    let mut extra = [0u8; 1];
    if reader.read(&mut extra).await? != 0 {
        return Err(stream_error(StreamError::InvalidInput(
            "stream exceeded declared size".into(),
        )));
    }

    Ok(())
}

/// Upload a registered chunk's slices, dropping them once stored.
async fn store_chunk<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    registered: RegisteredChunk,
) -> Result<PendingChunk, TapedriveError> {
    let receipts = upload_with_retry(
        client,
        &registered.written,
        &registered.plan,
        Operation::WriteStream,
    )
    .await?;

    Ok(PendingChunk {
        entry: registered.entry,
        written: registered.written,
        receipts,
    })
}

fn manifest_write_mode(name: &[u8], manifest_bytes: &[u8]) -> ManifestWriteMode {
    if inline_write_fits(name, manifest_bytes.len()) {
        ManifestWriteMode::Inline
    } else {
        ManifestWriteMode::Coded
    }
}

/// Store the manifest inline when the transaction stays small; otherwise upload it as a blob.
async fn store_manifest<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    logical_size: StorageUnits,
    manifest_bytes: &[u8],
) -> Result<StoredManifest, TapedriveError> {
    if manifest_write_mode(name, manifest_bytes) == ManifestWriteMode::Inline {
        let written = submit_raw_with_logical_size(
            client,
            tape_key,
            name,
            content_type,
            logical_size,
            manifest_bytes,
            Operation::WriteStream,
            client.write_options.visibility,
        )
        .await?;
        return Ok(StoredManifest {
            etag: object_etag(&written.track, None),
            written,
            receipts: Vec::new(),
        });
    }

    let (written, plan) = submit_blob_with_logical_size(
        client,
        tape_key,
        name,
        content_type,
        logical_size,
        manifest_bytes,
        Operation::WriteStream,
        client.write_options.visibility,
    )
    .await?;
    let receipts = upload_with_retry(client, &written, &plan, Operation::WriteStream).await?;
    Ok(StoredManifest {
        etag: object_etag(&written.track, Some(&plan_blob(&plan))),
        written,
        receipts,
    })
}

/// Store the final manifest after every chunk is stored and certified.
async fn store_manifest_for_chunks<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    name: &[u8],
    content_type: ContentType,
    size: StorageUnits,
    pending_chunks: Vec<PendingChunk>,
) -> Result<StoredManifest, TapedriveError> {
    let entries = pending_chunks
        .into_iter()
        .map(|pending_chunk| pending_chunk.entry)
        .collect();
    let manifest = build_manifest(hash(name), size, entries).map_err(stream_error)?;
    let manifest_bytes = manifest.to_bytes().map_err(stream_error)?;

    store_manifest(client, tape_key, name, content_type, size, &manifest_bytes).await
}

async fn complete_stream<Blockchain: Rpc, Cluster: Api>(
    client: &Tapedrive<Blockchain, Cluster>,
    tape_key: &impl TapeOperator,
    manifest: &StoredManifest,
) -> Result<StreamReceipt, TapedriveError> {
    let track = if manifest.written.track.is_certified() {
        manifest.written.track
    } else {
        certify_with_retry(
            client,
            tape_key,
            &manifest.written,
            Operation::WriteStream,
            &manifest.receipts,
        )
        .await?
    };
    Ok(StreamReceipt::from_manifest_track(&track, manifest.etag))
}

fn stream_error(error: StreamError) -> TapedriveError {
    TapedriveError::Stream(error.to_string())
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use arc_swap::ArcSwap;
    use bytemuck::Zeroable;
    use peer_memory::MemoryApi;
    use rpc_litesvm::LiteSvmRpc;
    use tape_api::state::Tape;
    use tape_core::spooler::GroupIndex;
    use tape_core::system::Member;
    use tape_core::track::types::{TrackState, TrackVisibility};
    use tape_core::types::coin::TAPE;
    use tape_core::types::{EpochNumber, StorageUnits, TapeNumber, TrackNumber};
    use tape_crypto::address::Address;
    use tape_peer_manager::PeerManager;
    use tape_protocol::api::{ApiError, GetTrackByNumberRes, PeerReq, PeerRes};
    use tape_protocol::ProtocolState;
    use tape_rpc_client::RpcClient;

    use crate::track::write::content_etag_named;

    use super::*;

    fn manifest_client(track: CompressedTrack) -> Tapedrive<LiteSvmRpc, MemoryApi> {
        let api = MemoryApi::new(move |_, request| {
            if let PeerReq::GetTrackByNumber(request) = request {
                assert_eq!(request.tape, track.tape);
                assert_eq!(request.track_number, track.track_number);
                return PeerRes::GetTrackByNumber(Ok(GetTrackByNumberRes { track }));
            }
            PeerRes::GetTrackByNumber(Err(ApiError::Other("unexpected request".into())))
        });
        let mut state = ProtocolState::default();
        state.current.committee.push(Member::new(Address::new_unique(), TAPE(1)));
        state.touch();
        Tapedrive::from_parts(
            ArcSwap::from_pointee(state),
            Arc::new(PeerManager::new()),
            Arc::new(api),
            Arc::new(RpcClient::from_rpc(LiteSvmRpc::new())),
            None,
        )
    }

    fn make_tape(capacity_bytes: u64, used_bytes: u64, next_track_number: u64) -> Tape {
        let mut tape = Tape::zeroed();
        tape.id = TapeNumber(1);
        tape.authority = Address::new_unique();
        tape.capacity = StorageUnits::from_bytes(capacity_bytes);
        tape.used = StorageUnits::from_bytes(used_bytes);
        tape.active_epoch = EpochNumber(1);
        tape.expiry_epoch = EpochNumber(2);
        tape.tracks.next_number = TrackNumber(next_track_number);
        tape
    }

    fn sample_manifest_bytes(chunk_count: u64) -> Vec<u8> {
        let key = Hash::from([0x11; 32]);
        let total_size = StorageUnits::from_bytes(MAX_TRACK_SIZE as u64 * chunk_count);
        let entries = build_entries(TrackNumber(0), TrackNumber(chunk_count), total_size)
            .expect("build entries");
        build_manifest(key, total_size, entries)
            .expect("build manifest")
            .to_bytes()
            .expect("serialize manifest")
    }

    #[test]
    fn small_manifest() {
        let manifest_bytes = sample_manifest_bytes(1);

        assert_eq!(
            manifest_write_mode(b"roms/small.bin", &manifest_bytes),
            ManifestWriteMode::Inline
        );
    }

    #[test]
    fn large_manifest() {
        let manifest_bytes = sample_manifest_bytes(256);

        assert_eq!(
            manifest_write_mode(b"roms/large.bin", &manifest_bytes),
            ManifestWriteMode::Coded
        );
    }

    // Resuming a certified manifest returns the same ETag as the object index for both storage modes.
    #[tokio::test]
    async fn resumed_etags() {
        let name = b"backup.tar";
        for chunk_count in [1, 256] {
            let tape_key = TapeKey::generate();
            let bytes = sample_manifest_bytes(chunk_count);
            let computed = content_etag_named(name, &bytes).await.expect("encode manifest");
            let blob = computed.plan.as_ref().map(plan_blob);
            let data = match blob {
                Some(blob) => BlobDataSlice::Coded(blob),
                None => BlobDataSlice::Inline(&bytes),
            };
            let metadata = data.meta().expect("valid manifest metadata");
            let track = CompressedTrack {
                tape: tape_key.address(),
                track_number: TrackNumber(chunk_count),
                key: track_key(name, &data),
                kind: metadata.kind as u64,
                state: TrackState::Certified as u64,
                visibility: TrackVisibility::Public as u64,
                size: metadata.size,
                group: GroupIndex(0),
                value_hash: metadata.value_hash,
            };
            let client = manifest_client(track);

            let manifest = ensure_stream_manifest(
                &client,
                &tape_key,
                name,
                ContentType::Unknown,
                StorageUnits::from_bytes(MAX_TRACK_SIZE as u64 * chunk_count),
                &bytes,
                track.track_number,
            ).await.expect("resume manifest");
            assert!(manifest.receipts.is_empty());
            let receipt = complete_stream(&client, &tape_key, &manifest)
                .await
                .expect("complete stream");

            assert_eq!(receipt.manifest, track_pda(track.tape, track.track_number).0);
            assert_eq!(receipt.manifest_track_number, track.track_number);
            assert_eq!(receipt.etag, object_etag(&track, blob.as_ref()));
            assert_eq!(track.is_coded(), chunk_count == 256);
            if track.is_coded() {
                assert_ne!(receipt.etag, track.value_hash);
            } else {
                assert_eq!(receipt.etag, track.value_hash);
            }
        }
    }

    // capacity checks include the serialized manifest bytes.
    #[test]
    fn manifest_size() {
        let key = Hash::from([0x11; 32]);
        let total_size = StorageUnits::from_bytes(MAX_TRACK_SIZE as u64);
        let entries = build_entries(TrackNumber(0), TrackNumber(1), total_size).expect("build entries");
        let manifest = build_manifest(key, total_size, entries).expect("build manifest");
        let manifest_bytes = manifest.to_bytes().expect("serialize manifest");
        let total_required = total_size + StorageUnits::from_bytes(manifest_bytes.len() as u64);
        let tape = make_tape(total_required.to_bytes() - 1, 0, 0);

        let error = preflight(&tape, total_required, TrackNumber(2)).expect_err("preflight should fail");

        match error {
            TapedriveError::InsufficientCapacity { need, available } => {
                assert_eq!(need, total_required);
                assert_eq!(available.to_bytes(), total_required.to_bytes() - 1);
            }
            other => panic!("expected insufficient capacity, got {other}"),
        }
    }

    // empty streams are rejected before chunk planning.
    #[test]
    fn empty_stream() {
        let error = validate_stream_size(StorageUnits::zero()).expect_err("empty stream should fail");

        match error {
            StreamError::InvalidInput(message) => {
                assert_eq!(message, "empty streams are not supported");
            }
            other => panic!("expected empty stream error, got {other}"),
        }
    }

    // manifest entries reject track number overflow.
    #[test]
    fn overflow() {
        let error = build_entries(
            TrackNumber(u64::MAX),
            TrackNumber(2),
            StorageUnits::from_bytes(MAX_TRACK_SIZE as u64 * 2),
        )
            .expect_err("entries should fail");

        match error {
            StreamError::InvalidInput(message) => {
                assert_eq!(message, "chunk track number overflow");
            }
            other => panic!("expected chunk track number overflow, got {other}"),
        }
    }
}