moonpool_sim/sim/

storage_ops.rs

//! Storage I/O operations for the simulation.
//!
//! This module contains storage-related event handlers and methods extracted from
//! `world.rs` to improve code organization. It handles file operations like
//! open, read, write, sync, and provides fault injection for testing storage reliability.

use std::collections::HashMap;
use std::net::IpAddr;
use std::task::Waker;
use std::time::Duration;
use tracing::instrument;

use crate::assert_reachable;
use crate::storage::StorageError;

use crate::chaos::fault_events::SimFaultEvent;

use super::{
    events::{Event, ScheduledEvent, StorageOperation},
    rng::{sim_random, sim_random_range},
    state::{DiskDegradationState, DiskEpisodeKind, FileId, PendingOpType, PendingStorageOp},
    world::{SimInner, SimWorld},
};

// =============================================================================
// Storage Event Handlers
// =============================================================================

/// Find and remove the first pending operation of the given type for a file.
///
/// Returns the sequence number and operation if found, or None if no such operation exists.
fn take_pending_op(
    inner: &mut SimInner,
    file_id: FileId,
    op_type: PendingOpType,
) -> Option<(u64, PendingStorageOp)> {
    let file_state = inner.storage.files.get_mut(&file_id)?;

    let op_seq = file_state
        .pending_ops
        .iter()
        .find(|(_, op)| op.op_type == op_type)
        .map(|(&seq, _)| seq)?;

    let op = file_state.pending_ops.remove(&op_seq)?;
    Some((op_seq, op))
}

/// Advance an owning process's disk-degradation episode state machine and return
/// the episode (if any) active for the operation now being scheduled.
///
/// Episodes are scoped per owner (keyed by process IP): a single stall/throttle
/// window applies to every file that process owns, modelling device-level
/// degradation that hits all of a machine's disks together (issue #147).
///
/// On each call: expire an episode whose window has elapsed, keep an episode
/// that is still active, or — when none is active and a family is enabled —
/// probabilistically enter a stall or throttle episode (stall takes precedence).
///
/// **Off-by-default guarantee:** when both probabilities are zero this returns
/// without drawing any simulation RNG, keeping the RNG stream byte-identical to
/// steady-state runs (so existing latency/determinism tests are unaffected).
fn update_disk_episode(
    episodes: &mut HashMap<IpAddr, DiskDegradationState>,
    owner_ip: IpAddr,
    now: Duration,
    stall_probability: f64,
    stall_duration: Duration,
    throttle_probability: f64,
    throttle_duration: Duration,
) -> Option<DiskDegradationState> {
    // Expire an episode whose window has elapsed.
    if let Some(episode) = episodes.get(&owner_ip).copied()
        && now >= episode.expires_at
    {
        episodes.remove(&owner_ip);
    }

    // An episode is still active: keep it, drawing no new randomness.
    if let Some(episode) = episodes.get(&owner_ip).copied() {
        return Some(episode);
    }

    // Off by default: never perturb the RNG stream when both families are disabled.
    if stall_probability <= 0.0 && throttle_probability <= 0.0 {
        return None;
    }

    // Probabilistically enter a new episode using a single draw.
    let roll = sim_random::<f64>();
    let episode = if roll < stall_probability {
        assert_reachable!("disk: stall episode entered");
        Some(DiskDegradationState {
            kind: DiskEpisodeKind::Stall,
            expires_at: now + stall_duration,
        })
    } else if roll < stall_probability + throttle_probability {
        assert_reachable!("disk: throttle episode entered");
        Some(DiskDegradationState {
            kind: DiskEpisodeKind::Throttle,
            expires_at: now + throttle_duration,
        })
    } else {
        None
    };
    if let Some(episode) = episode {
        episodes.insert(owner_ip, episode);
    }
    episode
}

/// Read the disk-episode knobs for an owning process into a copyable tuple, so
/// the immutable borrow of `inner.storage` ends before `inner.storage.disk_episodes`
/// is re-borrowed mutably for the episode update.
fn disk_episode_knobs(inner: &SimInner, owner_ip: IpAddr) -> (f64, Duration, f64, Duration) {
    let config = inner.storage.config_for(owner_ip);
    (
        config.disk_stall_probability,
        config.disk_stall_duration,
        config.disk_throttle_probability,
        config.disk_throttle_duration,
    )
}

/// Handle storage I/O events.
///
/// Storage events represent the completion of I/O operations.
/// Processing applies faults and wakes waiting tasks.
pub(crate) fn handle_storage_event(
    inner: &mut SimInner,
    file_id: u64,
    operation: StorageOperation,
) {
    let file_id = FileId(file_id);

    match operation {
        StorageOperation::ReadComplete { len: _ } => {
            handle_read_complete(inner, file_id);
        }
        StorageOperation::WriteComplete { len: _ } => {
            handle_write_complete(inner, file_id);
        }
        StorageOperation::SyncComplete => {
            handle_sync_complete(inner, file_id);
        }
        StorageOperation::OpenComplete => {
            handle_open_complete(inner, file_id);
        }
        StorageOperation::SetLenComplete { new_len } => {
            handle_set_len_complete(inner, file_id, new_len);
        }
    }
}

/// Handle read operation completion.
fn handle_read_complete(inner: &mut SimInner, file_id: FileId) {
    let read_fault_probability = inner.storage.files.get(&file_id).map_or(0.0, |f| {
        inner.storage.config_for(f.owner_ip).read_fault_probability
    });

    // Find and remove the oldest pending read operation
    let Some((op_seq, op)) = take_pending_op(inner, file_id, PendingOpType::Read) else {
        tracing::warn!("ReadComplete for unknown file {:?}", file_id);
        return;
    };

    let (offset, len) = (op.offset, op.len);

    // Apply read fault injection - mark sectors as faulted based on probability
    let mut read_faulted = false;
    if read_fault_probability > 0.0
        && let Some(file_state) = inner.storage.files.get_mut(&file_id)
    {
        let offset_usize = usize::try_from(offset).expect("offset fits in usize");
        let start_sector = offset_usize / crate::storage::SECTOR_SIZE;
        let end_sector = (offset_usize + len).div_ceil(crate::storage::SECTOR_SIZE);

        for sector in start_sector..end_sector {
            if sim_random::<f64>() < read_fault_probability {
                file_state.storage.set_fault(sector);
                read_faulted = true;
                tracing::info!(
                    "Read fault injected for file {:?}, sector {}",
                    file_id,
                    sector
                );
            }
        }
    }
    if read_faulted {
        let ip = inner.storage.files.get(&file_id).map(|f| f.owner_ip);
        if let Some(ip) = ip {
            inner.record_fault(SimFaultEvent::StorageReadFault {
                ip: ip.to_string(),
                file_id: file_id.0,
            });
        }
    }

    // Wake the waker for this operation
    if let Some(waker) = inner.wakers.storage_ops.remove(&(file_id, op_seq)) {
        tracing::trace!("Waking read waker for file {:?}, op {}", file_id, op_seq);
        waker.wake();
    }
}

/// Handle write operation completion.
///
/// Applies the write to storage with potential fault injection:
/// - `phantom_write_probability`: write appears to succeed but isn't persisted
/// - `misdirect_write_probability`: write lands at wrong location
fn handle_write_complete(inner: &mut SimInner, file_id: FileId) {
    let config = inner
        .storage
        .files
        .get(&file_id)
        .map(|f| inner.storage.config_for(f.owner_ip).clone())
        .unwrap_or_default();

    let owner_ip = inner.storage.files.get(&file_id).map(|f| f.owner_ip);

    // Find and remove the oldest pending write operation
    let Some((op_seq, op)) = take_pending_op(inner, file_id, PendingOpType::Write) else {
        tracing::warn!("WriteComplete for unknown file {:?}", file_id);
        return;
    };

    let (offset, data_opt) = (op.offset, op.data);

    // Apply the write with potential fault injection
    let mut write_fault_kind: Option<&str> = None;
    if let Some(data) = data_opt
        && let Some(file_state) = inner.storage.files.get_mut(&file_id)
    {
        // Check for phantom write (write appears to succeed but doesn't persist)
        if sim_random::<f64>() < config.phantom_write_probability {
            tracing::info!(
                "Phantom write injected for file {:?}, offset {}, len {}",
                file_id,
                offset,
                data.len()
            );
            file_state.storage.record_phantom_write(offset, &data);
            write_fault_kind = Some("phantom");
        }
        // Check for misdirected write
        else if sim_random::<f64>() < config.misdirect_write_probability {
            // Pick a random different offset
            let max_offset = file_state.storage.size().saturating_sub(data.len() as u64);
            let mistaken_offset = if max_offset > 0 {
                sim_random_range(0..max_offset)
            } else {
                0
            };
            tracing::info!(
                "Misdirected write injected for file {:?}: intended={}, actual={}",
                file_id,
                offset,
                mistaken_offset
            );
            if let Err(e) =
                file_state
                    .storage
                    .apply_misdirected_write(offset, mistaken_offset, &data)
            {
                tracing::warn!("Failed to apply misdirected write: {}", e);
            }
            write_fault_kind = Some("misdirected");
        }
        // Normal write (not synced - may be lost on crash)
        else if let Err(e) = file_state.storage.write(offset, &data, false) {
            tracing::warn!("Write failed for file {:?}: {}", file_id, e);
        } else {
            // Check for write corruption - mark sectors as faulted after successful write
            if config.write_fault_probability > 0.0 {
                let offset_usize = usize::try_from(offset).expect("offset fits in usize");
                let start_sector = offset_usize / crate::storage::SECTOR_SIZE;
                let end_sector = (offset_usize + data.len()).div_ceil(crate::storage::SECTOR_SIZE);

                for sector in start_sector..end_sector {
                    if sim_random::<f64>() < config.write_fault_probability {
                        file_state.storage.set_fault(sector);
                        write_fault_kind = Some("corruption");
                        tracing::info!(
                            "Write fault injected for file {:?}, sector {}",
                            file_id,
                            sector
                        );
                    }
                }
            }
        }
    }
    if let (Some(kind), Some(ip)) = (write_fault_kind, owner_ip) {
        inner.record_fault(SimFaultEvent::StorageWriteFault {
            ip: ip.to_string(),
            file_id: file_id.0,
            write_kind: kind.to_string(),
        });
    }

    // Wake the waker for this operation
    if let Some(waker) = inner.wakers.storage_ops.remove(&(file_id, op_seq)) {
        tracing::trace!("Waking write waker for file {:?}, op {}", file_id, op_seq);
        waker.wake();
    }
}

/// Handle sync operation completion.
///
/// Applies `sync_failure_probability` fault injection.
fn handle_sync_complete(inner: &mut SimInner, file_id: FileId) {
    let sync_failure_prob = inner.storage.files.get(&file_id).map_or(0.0, |f| {
        inner
            .storage
            .config_for(f.owner_ip)
            .sync_failure_probability
    });

    // Find and remove the oldest pending sync operation
    let Some((op_seq, _)) = take_pending_op(inner, file_id, PendingOpType::Sync) else {
        tracing::warn!("SyncComplete for unknown file {:?}", file_id);
        return;
    };

    // Check for sync failure
    if sim_random::<f64>() < sync_failure_prob {
        tracing::info!("Sync failure injected for file {:?}", file_id);
        // Record the failure so SyncFuture can return an error
        inner.storage.sync_failures.insert((file_id, op_seq));
        // On sync failure, we don't call storage.sync()
        // Data remains in pending state and may be lost on crash
        let ip = inner.storage.files.get(&file_id).map(|f| f.owner_ip);
        if let Some(ip) = ip {
            inner.record_fault(SimFaultEvent::StorageSyncFault {
                ip: ip.to_string(),
                file_id: file_id.0,
            });
        }
    } else if let Some(file_state) = inner.storage.files.get_mut(&file_id) {
        // Successful sync - make all pending writes durable
        file_state.storage.sync();
    }

    // Wake the waker for this operation
    if let Some(waker) = inner.wakers.storage_ops.remove(&(file_id, op_seq)) {
        tracing::trace!("Waking sync waker for file {:?}, op {}", file_id, op_seq);
        waker.wake();
    }
}

/// Handle open operation completion.
fn handle_open_complete(inner: &mut SimInner, file_id: FileId) {
    // Find and remove the oldest pending open operation
    let Some((op_seq, _)) = take_pending_op(inner, file_id, PendingOpType::Open) else {
        // File might not have pending open op (it was already "open" on creation)
        tracing::trace!("OpenComplete for file {:?} (no pending op)", file_id);
        return;
    };

    // Wake the waker for this operation
    if let Some(waker) = inner.wakers.storage_ops.remove(&(file_id, op_seq)) {
        tracing::trace!("Waking open waker for file {:?}, op {}", file_id, op_seq);
        waker.wake();
    }
}

/// Handle `set_len` operation completion.
fn handle_set_len_complete(inner: &mut SimInner, file_id: FileId, new_len: u64) {
    // Find and remove the oldest pending set_len operation
    let Some((op_seq, _)) = take_pending_op(inner, file_id, PendingOpType::SetLen) else {
        tracing::warn!("SetLenComplete for unknown file {:?}", file_id);
        return;
    };

    // Resize the storage (preserves seed, written/fault bitmaps, and overlays)
    if let Some(file_state) = inner.storage.files.get_mut(&file_id) {
        file_state.storage.resize(new_len);
    }

    // Wake the waker for this operation
    if let Some(waker) = inner.wakers.storage_ops.remove(&(file_id, op_seq)) {
        tracing::trace!(
            "Waking set_len waker for file {:?}, op {}, new_len={}",
            file_id,
            op_seq,
            new_len
        );
        waker.wake();
    }
}

// =============================================================================
// Storage Methods for SimWorld
// =============================================================================

impl SimWorld {
    /// Access storage configuration for the simulation.
    ///
    /// # Panics
    ///
    /// Panics if the simulation lock is poisoned by a prior task panic.
    pub fn with_storage_config<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&crate::storage::StorageConfiguration) -> R,
    {
        let inner = self
            .inner
            .read()
            .expect("RwLock poisoned: prior task panicked");
        f(&inner.storage.config)
    }

    /// Open a file in the simulation.
    ///
    /// Creates a new file or opens an existing one based on the options.
    /// Files are tagged with the `owner_ip` for per-process storage fault injection.
    /// Schedules an `OpenComplete` event and returns the file ID.
    pub(crate) fn open_file(
        &self,
        path: &str,
        options: moonpool_core::OpenOptions,
        initial_size: u64,
        owner_ip: IpAddr,
    ) -> Result<FileId, StorageError> {
        use crate::storage::InMemoryStorage;

        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");
        let path_str = path.to_string();

        // Check create_new semantics - fail if file exists
        if options.is_create_new() && inner.storage.path_to_file.contains_key(&path_str) {
            return Err(StorageError::AlreadyExists { path: path_str });
        }

        // Check if file was deleted and create is not set
        if inner.storage.deleted_paths.contains(&path_str) && !options.is_create() {
            return Err(StorageError::NotFound { path: path_str });
        }

        // If file already exists and we're opening it, return existing file ID
        if let Some(&existing_id) = inner.storage.path_to_file.get(&path_str) {
            if let Some(file_state) = inner.storage.files.get_mut(&existing_id) {
                // If truncate is set, reset the storage
                if options.is_truncate() {
                    let seed = sim_random::<u64>();
                    file_state.storage = InMemoryStorage::new(0, seed);
                    file_state.position = 0;
                } else if options.is_append() {
                    // For append mode, seek to end
                    file_state.position = file_state.storage.size();
                } else {
                    // For normal reopen, reset position to start
                    file_state.position = 0;
                }
                // Update options for the new open
                file_state.options = options;
                file_state.is_closed = false;
            }
            return Ok(existing_id);
        }

        // File doesn't exist - check if we're allowed to create it
        if !options.is_create() && !options.is_create_new() {
            return Err(StorageError::NotFound { path: path_str });
        }

        // Create new file
        let file_id = FileId(inner.storage.next_file_id);
        inner.storage.next_file_id += 1;

        // Remove from deleted paths if re-creating
        inner.storage.deleted_paths.remove(&path_str);

        // Create in-memory storage with deterministic seed
        let seed = sim_random::<u64>();
        let storage = InMemoryStorage::new(initial_size, seed);

        let file_state = super::state::StorageFileState::new(
            file_id,
            path_str.clone(),
            options,
            storage,
            owner_ip,
        );

        inner.storage.files.insert(file_id, file_state);
        inner.storage.path_to_file.insert(path_str, file_id);

        // Schedule OpenComplete event with minimal latency
        let open_latency = Duration::from_micros(1);
        let scheduled_time = inner.current_time + open_latency;
        let sequence = inner.next_sequence;
        inner.next_sequence += 1;
        let event = Event::Storage {
            file_id: file_id.0,
            operation: StorageOperation::OpenComplete,
        };
        inner
            .event_queue
            .schedule(ScheduledEvent::new(scheduled_time, event, sequence));

        tracing::debug!("Opened file {:?} with id {:?}", path, file_id);
        Ok(file_id)
    }

    /// Check if a file exists at the given path.
    pub(crate) fn file_exists(&self, path: &str) -> bool {
        let inner = self
            .inner
            .read()
            .expect("RwLock poisoned: prior task panicked");
        let path_str = path.to_string();
        inner.storage.path_to_file.contains_key(&path_str)
            && !inner.storage.deleted_paths.contains(&path_str)
    }

    /// Delete a file at the given path.
    pub(crate) fn delete_file(&self, path: &str) -> Result<(), StorageError> {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");
        let path_str = path.to_string();

        if let Some(file_id) = inner.storage.path_to_file.remove(&path_str) {
            // Mark file as closed and remove it
            if let Some(file_state) = inner.storage.files.get_mut(&file_id) {
                file_state.is_closed = true;
            }
            inner.storage.files.remove(&file_id);
            inner.storage.deleted_paths.insert(path_str);
            tracing::debug!("Deleted file {:?}", path);
            Ok(())
        } else {
            Err(StorageError::NotFound { path: path_str })
        }
    }

    /// Rename a file from one path to another.
    pub(crate) fn rename_file(&self, from: &str, to: &str) -> Result<(), StorageError> {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");
        let from_str = from.to_string();
        let to_str = to.to_string();

        if let Some(file_id) = inner.storage.path_to_file.remove(&from_str) {
            // Update the path in the file state
            if let Some(file_state) = inner.storage.files.get_mut(&file_id) {
                file_state.path.clone_from(&to_str);
            }
            inner.storage.path_to_file.insert(to_str, file_id);
            inner.storage.deleted_paths.remove(&from_str);
            tracing::debug!("Renamed file {:?} to {:?}", from, to);
            Ok(())
        } else {
            Err(StorageError::NotFound { path: from_str })
        }
    }

    /// Schedule a read operation on a file.
    ///
    /// Returns an operation sequence number that can be used to check completion.
    pub(crate) fn schedule_read(
        &self,
        file_id: FileId,
        offset: u64,
        len: usize,
    ) -> Result<u64, StorageError> {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");

        let file_state = inner
            .storage
            .files
            .get_mut(&file_id)
            .ok_or(StorageError::InvalidFileHandle { file_id })?;

        if file_state.is_closed {
            return Err(StorageError::FileClosed { file_id });
        }

        let op_seq = file_state.next_op_seq;
        file_state.next_op_seq += 1;

        // Store the pending operation
        file_state.pending_ops.insert(
            op_seq,
            PendingStorageOp {
                op_type: PendingOpType::Read,
                offset,
                len,
                data: None,
            },
        );

        // Advance the disk-degradation episode, then compute latency.
        let owner_ip = file_state.owner_ip;
        let now = inner.current_time;
        let (stall_p, stall_dur, throttle_p, throttle_dur) = disk_episode_knobs(&inner, owner_ip);
        let episode = update_disk_episode(
            &mut inner.storage.disk_episodes,
            owner_ip,
            now,
            stall_p,
            stall_dur,
            throttle_p,
            throttle_dur,
        );
        let config = inner.storage.config_for(owner_ip);
        let latency = Self::calculate_storage_latency(config, len, false, episode, now);
        let scheduled_time = now + latency;
        let sequence = inner.next_sequence;
        inner.next_sequence += 1;

        let event = Event::Storage {
            file_id: file_id.0,
            operation: StorageOperation::ReadComplete {
                len: u32::try_from(len).expect("read length fits in u32"),
            },
        };
        inner
            .event_queue
            .schedule(ScheduledEvent::new(scheduled_time, event, sequence));

        tracing::trace!(
            "Scheduled read: file={:?}, offset={}, len={}, op_seq={}",
            file_id,
            offset,
            len,
            op_seq
        );

        Ok(op_seq)
    }

    /// Schedule a write operation on a file.
    ///
    /// Returns an operation sequence number that can be used to check completion.
    pub(crate) fn schedule_write(
        &self,
        file_id: FileId,
        offset: u64,
        data: Vec<u8>,
    ) -> Result<u64, StorageError> {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");

        let file_state = inner
            .storage
            .files
            .get_mut(&file_id)
            .ok_or(StorageError::InvalidFileHandle { file_id })?;

        if file_state.is_closed {
            return Err(StorageError::FileClosed { file_id });
        }

        let op_seq = file_state.next_op_seq;
        file_state.next_op_seq += 1;
        let len = data.len();

        // Store the pending operation with the data
        file_state.pending_ops.insert(
            op_seq,
            PendingStorageOp {
                op_type: PendingOpType::Write,
                offset,
                len,
                data: Some(data),
            },
        );

        // Advance the disk-degradation episode, then compute latency.
        let owner_ip = file_state.owner_ip;
        let now = inner.current_time;
        let (stall_p, stall_dur, throttle_p, throttle_dur) = disk_episode_knobs(&inner, owner_ip);
        let episode = update_disk_episode(
            &mut inner.storage.disk_episodes,
            owner_ip,
            now,
            stall_p,
            stall_dur,
            throttle_p,
            throttle_dur,
        );
        let config = inner.storage.config_for(owner_ip);
        let latency = Self::calculate_storage_latency(config, len, true, episode, now);
        let scheduled_time = now + latency;
        let sequence = inner.next_sequence;
        inner.next_sequence += 1;

        let event = Event::Storage {
            file_id: file_id.0,
            operation: StorageOperation::WriteComplete {
                len: u32::try_from(len).expect("write length fits in u32"),
            },
        };
        inner
            .event_queue
            .schedule(ScheduledEvent::new(scheduled_time, event, sequence));

        tracing::trace!(
            "Scheduled write: file={:?}, offset={}, len={}, op_seq={}",
            file_id,
            offset,
            len,
            op_seq
        );

        Ok(op_seq)
    }

    /// Schedule a sync operation on a file.
    ///
    /// Returns an operation sequence number that can be used to check completion.
    pub(crate) fn schedule_sync(&self, file_id: FileId) -> Result<u64, StorageError> {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");

        let file_state = inner
            .storage
            .files
            .get_mut(&file_id)
            .ok_or(StorageError::InvalidFileHandle { file_id })?;

        if file_state.is_closed {
            return Err(StorageError::FileClosed { file_id });
        }

        let op_seq = file_state.next_op_seq;
        file_state.next_op_seq += 1;

        // Store the pending operation
        file_state.pending_ops.insert(
            op_seq,
            PendingStorageOp {
                op_type: PendingOpType::Sync,
                offset: 0,
                len: 0,
                data: None,
            },
        );

        // Advance the disk-degradation episode, then compute sync latency. Sync
        // is affected by stalls (the disk is frozen until expiry); a throttle
        // scales IOPS/bandwidth, which sync latency does not use.
        let owner_ip = file_state.owner_ip;
        let now = inner.current_time;
        let (stall_p, stall_dur, throttle_p, throttle_dur) = disk_episode_knobs(&inner, owner_ip);
        let episode = update_disk_episode(
            &mut inner.storage.disk_episodes,
            owner_ip,
            now,
            stall_p,
            stall_dur,
            throttle_p,
            throttle_dur,
        );
        let config = inner.storage.config_for(owner_ip);
        let mut latency = crate::network::sample_latency(&config.sync_latency);
        if let Some(DiskDegradationState {
            kind: DiskEpisodeKind::Stall,
            expires_at,
        }) = episode
        {
            latency += expires_at.saturating_sub(now);
        }
        let scheduled_time = now + latency;
        let sequence = inner.next_sequence;
        inner.next_sequence += 1;

        let event = Event::Storage {
            file_id: file_id.0,
            operation: StorageOperation::SyncComplete,
        };
        inner
            .event_queue
            .schedule(ScheduledEvent::new(scheduled_time, event, sequence));

        tracing::trace!("Scheduled sync: file={:?}, op_seq={}", file_id, op_seq);

        Ok(op_seq)
    }

    /// Schedule a `set_len` operation on a file.
    ///
    /// Returns an operation sequence number that can be used to check completion.
    pub(crate) fn schedule_set_len(
        &self,
        file_id: FileId,
        new_len: u64,
    ) -> Result<u64, StorageError> {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");

        let file_state = inner
            .storage
            .files
            .get_mut(&file_id)
            .ok_or(StorageError::InvalidFileHandle { file_id })?;

        if file_state.is_closed {
            return Err(StorageError::FileClosed { file_id });
        }

        let op_seq = file_state.next_op_seq;
        file_state.next_op_seq += 1;

        // Store the pending operation
        file_state.pending_ops.insert(
            op_seq,
            PendingStorageOp {
                op_type: PendingOpType::SetLen,
                offset: new_len,
                len: 0,
                data: None,
            },
        );

        // Use write latency from per-process config
        let owner_ip = file_state.owner_ip;
        let config = inner.storage.config_for(owner_ip);
        let latency = crate::network::sample_latency(&config.write_latency);
        let scheduled_time = inner.current_time + latency;
        let sequence = inner.next_sequence;
        inner.next_sequence += 1;

        let event = Event::Storage {
            file_id: file_id.0,
            operation: StorageOperation::SetLenComplete { new_len },
        };
        inner
            .event_queue
            .schedule(ScheduledEvent::new(scheduled_time, event, sequence));

        tracing::trace!(
            "Scheduled set_len: file={:?}, new_len={}, op_seq={}",
            file_id,
            new_len,
            op_seq
        );

        Ok(op_seq)
    }

    /// Check if a storage operation is complete.
    pub(crate) fn is_storage_op_complete(&self, file_id: FileId, op_seq: u64) -> bool {
        let inner = self
            .inner
            .read()
            .expect("RwLock poisoned: prior task panicked");
        if let Some(file_state) = inner.storage.files.get(&file_id) {
            // Operation is complete when it's no longer in pending_ops
            !file_state.pending_ops.contains_key(&op_seq)
        } else {
            // File not found means operation is effectively "complete" (failed)
            true
        }
    }

    /// Check if a sync operation failed and clear the failure flag.
    ///
    /// Returns true if the sync failed due to fault injection.
    pub(crate) fn take_sync_failure(&self, file_id: FileId, op_seq: u64) -> bool {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");
        inner.storage.sync_failures.remove(&(file_id, op_seq))
    }

    /// Register a waker for a storage operation.
    pub(crate) fn register_storage_waker(&self, file_id: FileId, op_seq: u64, waker: Waker) {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");
        inner.wakers.storage_ops.insert((file_id, op_seq), waker);
    }

    /// Read data from a file at the given offset.
    ///
    /// This is called after `ReadComplete` to actually fetch the data.
    pub(crate) fn read_from_file(
        &self,
        file_id: FileId,
        offset: u64,
        buf: &mut [u8],
    ) -> Result<usize, StorageError> {
        let inner = self
            .inner
            .read()
            .expect("RwLock poisoned: prior task panicked");

        let file_state = inner
            .storage
            .files
            .get(&file_id)
            .ok_or(StorageError::InvalidFileHandle { file_id })?;

        if file_state.is_closed {
            return Err(StorageError::FileClosed { file_id });
        }

        // Read from the in-memory storage
        file_state
            .storage
            .read(offset, buf)
            .map_err(|e| StorageError::Io {
                file_id,
                kind: e.kind(),
                message: e.to_string(),
            })?;

        Ok(buf.len())
    }

    /// Get the current file position.
    pub(crate) fn file_position(&self, file_id: FileId) -> Result<u64, StorageError> {
        let inner = self
            .inner
            .read()
            .expect("RwLock poisoned: prior task panicked");
        inner
            .storage
            .files
            .get(&file_id)
            .map(|f| f.position)
            .ok_or(StorageError::InvalidFileHandle { file_id })
    }

    /// Set the current file position.
    pub(crate) fn set_file_position(
        &self,
        file_id: FileId,
        position: u64,
    ) -> Result<(), StorageError> {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");
        if let Some(file_state) = inner.storage.files.get_mut(&file_id) {
            file_state.position = position;
            Ok(())
        } else {
            Err(StorageError::InvalidFileHandle { file_id })
        }
    }

    /// Get the size of a file.
    pub(crate) fn file_size(&self, file_id: FileId) -> Result<u64, StorageError> {
        let inner = self
            .inner
            .read()
            .expect("RwLock poisoned: prior task panicked");
        inner
            .storage
            .files
            .get(&file_id)
            .map(|f| f.storage.size())
            .ok_or(StorageError::InvalidFileHandle { file_id })
    }

    /// Calculate storage latency using FDB formula, adjusted for any active
    /// disk-degradation episode.
    ///
    /// Latency = `base_latency` + `iops_overhead` + `transfer_time`, where a
    /// **throttle** episode divides effective IOPS/bandwidth by the configured
    /// multipliers and a **stall** episode adds the time remaining until the
    /// stall expires (the disk is frozen until then).
    fn calculate_storage_latency(
        config: &crate::storage::StorageConfiguration,
        size: usize,
        is_write: bool,
        episode: Option<DiskDegradationState>,
        now: Duration,
    ) -> Duration {
        // Sample base latency from config range
        let base_range = if is_write {
            &config.write_latency
        } else {
            &config.read_latency
        };
        let base = crate::network::sample_latency(base_range);

        // During a throttle episode, effective IOPS/bandwidth are divided by the
        // configured multipliers (>= 1.0); otherwise the divisors are 1.0.
        let (iops_divisor, bandwidth_divisor) = match episode {
            Some(DiskDegradationState {
                kind: DiskEpisodeKind::Throttle,
                ..
            }) => (
                config.disk_throttle_iops_multiplier.max(1.0),
                config.disk_throttle_bandwidth_multiplier.max(1.0),
            ),
            _ => (1.0, 1.0),
        };

        // IOPS overhead: divisor/iops seconds per operation.
        // Precision loss is acceptable: iops is typically << 2^52.
        let iops_f64 = u32::try_from(config.iops).map_or(f64::from(u32::MAX), f64::from);
        let iops_overhead = Duration::from_secs_f64(iops_divisor / iops_f64);

        // Transfer time: size * divisor / bandwidth seconds.
        // Precision loss is acceptable: simulated sizes/bandwidths fit in 2^52.
        let size_f64 = u32::try_from(size).map_or(f64::from(u32::MAX), f64::from);
        let bandwidth_f64 = u32::try_from(config.bandwidth).map_or(f64::from(u32::MAX), f64::from);
        let transfer = Duration::from_secs_f64(size_f64 * bandwidth_divisor / bandwidth_f64);

        let steady = base + iops_overhead + transfer;

        // During a stall the disk is frozen until expiry: the op waits out the
        // remaining window, then takes its normal steady-state latency.
        match episode {
            Some(DiskDegradationState {
                kind: DiskEpisodeKind::Stall,
                expires_at,
            }) => steady + expires_at.saturating_sub(now),
            _ => steady,
        }
    }

    /// Simulate a crash affecting storage for a specific process.
    ///
    /// Only affects files owned by the given IP address:
    /// 1. Calls `apply_crash()` on matching `InMemoryStorage` instances
    /// 2. Clears pending operations (lost in crash)
    /// 3. Optionally marks files as closed
    /// 4. Wakes all storage wakers (operations will fail)
    ///
    /// Files owned by other IPs are unaffected.
    ///
    /// # Panics
    ///
    /// Panics if the simulation lock is poisoned by a prior task panic.
    #[instrument(skip(self))]
    pub fn simulate_crash_for_process(&self, ip: IpAddr, close_files: bool) {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");
        let crash_probability = inner.storage.config_for(ip).crash_fault_probability;

        // Collect all wakers to wake in one pass (to avoid borrow conflict)
        let mut wakers_to_wake = Vec::new();
        let file_ids: Vec<FileId> = inner
            .storage
            .files
            .iter()
            .filter(|(_, f)| f.owner_ip == ip)
            .map(|(id, _)| *id)
            .collect();

        for file_id in &file_ids {
            if let Some(file_state) = inner.storage.files.get_mut(file_id) {
                // Apply crash to in-memory storage (may corrupt pending writes)
                file_state.storage.apply_crash(crash_probability);

                // Collect lost op sequence numbers
                let lost_ops: Vec<u64> = file_state.pending_ops.keys().copied().collect();

                // Clear pending ops - they're lost in crash
                file_state.pending_ops.clear();

                // Collect waker keys for later removal
                for op_seq in lost_ops {
                    wakers_to_wake.push((*file_id, op_seq));
                }

                // Optionally close files
                if close_files {
                    file_state.is_closed = true;
                }
            }
        }

        // Wake all collected wakers (after file iteration is complete)
        for key in wakers_to_wake {
            if let Some(waker) = inner.wakers.storage_ops.remove(&key) {
                waker.wake();
            }
        }

        inner.record_fault(SimFaultEvent::StorageCrash { ip: ip.to_string() });

        tracing::info!(
            "Storage crash simulated for {}: {} files affected, close_files={}",
            ip,
            file_ids.len(),
            close_files
        );
    }

    /// Wipe all storage for a specific process.
    ///
    /// Deletes all files owned by the given IP address. Used by `CrashAndWipe`
    /// reboot to simulate total data loss. After wipe, the process can create
    /// new files at the same paths.
    ///
    /// Files owned by other IPs are unaffected.
    ///
    /// # Panics
    ///
    /// Panics if the simulation lock is poisoned by a prior task panic.
    #[instrument(skip(self))]
    pub fn wipe_storage_for_process(&self, ip: IpAddr) {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");

        // Collect files owned by this IP
        let file_ids: Vec<(FileId, String)> = inner
            .storage
            .files
            .iter()
            .filter(|(_, f)| f.owner_ip == ip)
            .map(|(id, f)| (*id, f.path.clone()))
            .collect();

        // Collect wakers to wake
        let mut wakers_to_wake = Vec::new();

        for (file_id, path) in &file_ids {
            if let Some(file_state) = inner.storage.files.remove(file_id) {
                for op_seq in file_state.pending_ops.keys() {
                    wakers_to_wake.push((*file_id, *op_seq));
                }
            }
            inner.storage.path_to_file.remove(path);
            inner.storage.deleted_paths.insert(path.clone());
        }

        // Wake all collected wakers
        for key in wakers_to_wake {
            if let Some(waker) = inner.wakers.storage_ops.remove(&key) {
                waker.wake();
            }
        }

        inner.record_fault(SimFaultEvent::StorageWipe { ip: ip.to_string() });

        tracing::info!("Storage wiped for {}: {} files deleted", ip, file_ids.len(),);
    }

    /// Set storage configuration for a specific process.
    ///
    /// Files owned by this IP will use this configuration for fault injection
    /// and latency calculations. Takes effect immediately, even for files
    /// already open.
    ///
    /// # Panics
    ///
    /// Panics if the simulation lock is poisoned by a prior task panic.
    #[instrument(skip(self, config))]
    pub fn set_process_storage_config(
        &self,
        ip: IpAddr,
        config: crate::storage::StorageConfiguration,
    ) {
        let mut inner = self
            .inner
            .write()
            .expect("RwLock poisoned: prior task panicked");
        inner.storage.per_process_configs.insert(ip, config);
    }
}