lcpfs 2026.1.102

//! OSD - Object Storage Daemon
//!
//! This module implements the Object Storage Daemon (OSD), the workhorse of
//! the LCPFS distributed cluster. Each OSD manages a local storage device
//! and handles read/write operations for placement groups assigned to it.
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────┐
//! │                          OSD                                 │
//! │  ┌─────────────────┐  ┌─────────────────┐                   │
//! │  │   OSD Frontend  │  │   OSD Backend   │                   │
//! │  │  (RPC Handler)  │  │  (Local Store)  │                   │
//! │  └────────┬────────┘  └────────┬────────┘                   │
//! │           │                    │                             │
//! │           ▼                    ▼                             │
//! │  ┌─────────────────────────────────────────┐                │
//! │  │           Placement Groups              │                │
//! │  │  ┌─────┐  ┌─────┐  ┌─────┐  ┌─────┐   │                │
//! │  │  │PG 1 │  │PG 5 │  │PG 12│  │PG 99│   │                │
//! │  │  └─────┘  └─────┘  └─────┘  └─────┘   │                │
//! │  └─────────────────────────────────────────┘                │
//! │                       │                                      │
//! │                       ▼                                      │
//! │  ┌─────────────────────────────────────────┐                │
//! │  │         Local LCPFS Store               │                │
//! │  │    (Block device with LCPFS format)     │                │
//! │  └─────────────────────────────────────────┘                │
//! └─────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Responsibilities
//!
//! - Store and retrieve objects for assigned PGs
//! - Replicate writes to secondary OSDs (when acting as primary)
//! - Participate in recovery when PGs are degraded
//! - Report health and capacity to monitors
//! - Scrub data for integrity verification

#![cfg_attr(not(feature = "std"), no_std)]

extern crate alloc;

use alloc::boxed::Box;
use alloc::collections::BTreeMap;
use alloc::string::{String, ToString};
use alloc::vec;
use alloc::vec::Vec;
use core::fmt;

#[cfg(feature = "std")]
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

#[cfg(not(feature = "std"))]
use core::sync::atomic::{AtomicBool, AtomicU64, Ordering};

// ============================================================================
// OSD Configuration
// ============================================================================

/// Configuration for an OSD instance.
#[derive(Debug, Clone)]
pub struct OsdConfig {
    /// OSD ID (unique across cluster).
    pub id: u64,
    /// Cluster name.
    pub cluster_name: String,
    /// Data directory/device path.
    pub data_path: String,
    /// Journal device path (optional, for write-ahead log).
    pub journal_path: Option<String>,
    /// Maximum concurrent operations.
    pub max_ops: usize,
    /// Heartbeat interval in milliseconds.
    pub heartbeat_interval_ms: u64,
    /// Replication timeout in milliseconds.
    pub replication_timeout_ms: u64,
    /// Enable scrubbing.
    pub scrub_enabled: bool,
    /// Scrub interval in seconds.
    pub scrub_interval_secs: u64,
}

impl Default for OsdConfig {
    fn default() -> Self {
        Self {
            id: 0,
            cluster_name: "lcpfs".to_string(),
            data_path: "/var/lib/lcpfs/osd.0".to_string(),
            journal_path: None,
            max_ops: 128,
            heartbeat_interval_ms: 1000,
            replication_timeout_ms: 30000,
            scrub_enabled: true,
            scrub_interval_secs: 86400, // Daily
        }
    }
}

// ============================================================================
// OSD Operations
// ============================================================================

/// Operations that can be performed on an OSD.
#[derive(Debug, Clone)]
pub enum OsdOp {
    /// Read object data.
    Read {
        /// Object ID.
        oid: u64,
        /// Offset within object.
        offset: u64,
        /// Number of bytes to read.
        length: u64,
    },
    /// Write object data.
    Write {
        /// Object ID.
        oid: u64,
        /// Offset within object.
        offset: u64,
        /// Data to write.
        data: Vec<u8>,
    },
    /// Write full object (replace).
    WriteFull {
        /// Object ID.
        oid: u64,
        /// Complete object data.
        data: Vec<u8>,
    },
    /// Delete an object.
    Delete {
        /// Object ID.
        oid: u64,
    },
    /// Truncate object to size.
    Truncate {
        /// Object ID.
        oid: u64,
        /// New size.
        size: u64,
    },
    /// Get object stats.
    Stat {
        /// Object ID.
        oid: u64,
    },
    /// Set extended attribute.
    SetXattr {
        /// Object ID.
        oid: u64,
        /// Attribute name.
        name: String,
        /// Attribute value.
        value: Vec<u8>,
    },
    /// Get extended attribute.
    GetXattr {
        /// Object ID.
        oid: u64,
        /// Attribute name.
        name: String,
    },
    /// Remove extended attribute.
    RemoveXattr {
        /// Object ID.
        oid: u64,
        /// Attribute name.
        name: String,
    },
    /// List objects in PG.
    List {
        /// Maximum objects to return.
        max: usize,
        /// Continuation token.
        marker: Option<u64>,
    },
    /// Replicate object to another OSD.
    Replicate {
        /// Object ID.
        oid: u64,
        /// Target OSD ID.
        target_osd: u64,
    },
    /// Copy object from another OSD (for recovery).
    Pull {
        /// Object ID.
        oid: u64,
        /// Source OSD ID.
        source_osd: u64,
    },
    /// Scrub object (verify checksum).
    Scrub {
        /// Object ID (None = scrub all in PG).
        oid: Option<u64>,
    },
}

/// Message wrapper for OSD operations.
#[derive(Debug, Clone)]
pub struct OsdMessage {
    /// Cluster epoch (for stale message detection).
    pub epoch: u64,
    /// Placement group ID.
    pub pgid: u64,
    /// The operation to perform.
    pub op: OsdOp,
    /// Unique request ID.
    pub request_id: u64,
    /// Source OSD ID (for replication).
    pub source_osd: Option<u64>,
    /// Flags.
    pub flags: OsdOpFlags,
}

/// Flags for OSD operations.
#[derive(Debug, Clone, Copy, Default)]
pub struct OsdOpFlags {
    /// Acknowledge before replication complete.
    pub ack_on_primary: bool,
    /// Skip cache, read from disk.
    pub skip_cache: bool,
    /// This is a recovery operation.
    pub recovery: bool,
    /// This is a scrub operation.
    pub scrub: bool,
    /// Ignore PG state (for admin operations).
    pub ignore_state: bool,
}

/// Response from an OSD operation.
#[derive(Debug, Clone)]
pub struct OsdResponse {
    /// Request ID this responds to.
    pub request_id: u64,
    /// Result of the operation.
    pub result: OsdResult,
    /// Epoch at time of operation.
    pub epoch: u64,
}

/// Result of an OSD operation.
#[derive(Debug, Clone)]
pub enum OsdResult {
    /// Operation succeeded with optional data.
    Success(Option<Vec<u8>>),
    /// Object statistics.
    Stat(ObjectStat),
    /// Object listing.
    List(Vec<ObjectInfo>),
    /// Extended attribute value.
    Xattr(Vec<u8>),
    /// Operation failed.
    Error(OsdError),
}

/// Object statistics.
#[derive(Debug, Clone)]
pub struct ObjectStat {
    /// Object ID.
    pub oid: u64,
    /// Object size in bytes.
    pub size: u64,
    /// Modification time (Unix timestamp).
    pub mtime: u64,
    /// Object version.
    pub version: u64,
    /// Checksum.
    pub checksum: [u8; 32],
}

/// Object information for listings.
#[derive(Debug, Clone)]
pub struct ObjectInfo {
    /// Object ID.
    pub oid: u64,
    /// Object size.
    pub size: u64,
    /// Object version.
    pub version: u64,
}

// ============================================================================
// OSD Errors
// ============================================================================

/// Errors that can occur during OSD operations.
#[derive(Debug, Clone)]
pub enum OsdError {
    /// Object not found.
    ObjectNotFound(u64),
    /// PG not found or not owned by this OSD.
    PgNotFound(u64),
    /// PG is not active (degraded, recovering, etc.).
    PgNotActive(u64, PgState),
    /// Not the primary OSD for this PG.
    NotPrimary(u64),
    /// Stale epoch (cluster map changed).
    StaleEpoch {
        /// Expected epoch number.
        expected: u64,
        /// Received epoch number.
        got: u64,
    },
    /// I/O error.
    IoError(String),
    /// Replication failed.
    ReplicationFailed(String),
    /// Timeout.
    Timeout,
    /// OSD is full.
    OsdFull,
    /// Invalid operation.
    InvalidOp(String),
    /// Checksum mismatch.
    ChecksumMismatch {
        /// Object ID.
        oid: u64,
        /// Expected checksum.
        expected: [u8; 32],
        /// Actual checksum.
        got: [u8; 32],
    },
}

impl fmt::Display for OsdError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            OsdError::ObjectNotFound(oid) => write!(f, "Object {} not found", oid),
            OsdError::PgNotFound(pgid) => write!(f, "PG {} not found", pgid),
            OsdError::PgNotActive(pgid, state) => write!(f, "PG {} is {:?}", pgid, state),
            OsdError::NotPrimary(pgid) => write!(f, "Not primary for PG {}", pgid),
            OsdError::StaleEpoch { expected, got } => {
                write!(f, "Stale epoch: expected {}, got {}", expected, got)
            }
            OsdError::IoError(msg) => write!(f, "I/O error: {}", msg),
            OsdError::ReplicationFailed(msg) => write!(f, "Replication failed: {}", msg),
            OsdError::Timeout => write!(f, "Operation timeout"),
            OsdError::OsdFull => write!(f, "OSD is full"),
            OsdError::InvalidOp(msg) => write!(f, "Invalid operation: {}", msg),
            OsdError::ChecksumMismatch { oid, .. } => {
                write!(f, "Checksum mismatch for object {}", oid)
            }
        }
    }
}

// ============================================================================
// Placement Group State
// ============================================================================

/// State of a placement group.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PgState {
    /// PG is healthy and serving requests.
    Active,
    /// PG has fewer replicas than required.
    Degraded,
    /// PG is recovering missing data.
    Recovering,
    /// PG is backfilling to new OSD.
    Backfilling,
    /// PG is establishing consensus after OSD changes.
    Peering,
    /// PG is not serving requests (waiting for peering).
    Inactive,
    /// PG is being created.
    Creating,
    /// PG is incomplete (not enough OSDs).
    Incomplete,
    /// PG is stale (no heartbeat from primary).
    Stale,
    /// PG is being remapped to different OSDs.
    Remapped,
    /// PG has inconsistent data across replicas.
    Inconsistent,
}

impl PgState {
    /// Check if PG can serve read requests.
    pub fn can_read(&self) -> bool {
        matches!(
            self,
            PgState::Active | PgState::Degraded | PgState::Recovering | PgState::Backfilling
        )
    }

    /// Check if PG can serve write requests.
    pub fn can_write(&self) -> bool {
        matches!(
            self,
            PgState::Active | PgState::Degraded | PgState::Recovering
        )
    }

    /// Check if PG is healthy.
    pub fn is_healthy(&self) -> bool {
        matches!(self, PgState::Active)
    }
}

/// Placement group managed by this OSD.
#[derive(Debug, Clone)]
pub struct PlacementGroup {
    /// PG ID.
    pub pgid: u64,
    /// Pool ID.
    pub pool_id: u64,
    /// Current state.
    pub state: PgState,
    /// Primary OSD.
    pub primary: u64,
    /// Acting set (OSDs currently serving this PG).
    pub acting: Vec<u64>,
    /// Up set (OSDs that should serve this PG per CRUSH).
    pub up: Vec<u64>,
    /// Current epoch.
    pub epoch: u64,
    /// Last epoch with successful peering.
    pub last_peered_epoch: u64,
    /// Object count.
    pub object_count: u64,
    /// Total bytes.
    pub total_bytes: u64,
    /// Last scrub time.
    pub last_scrub: u64,
    /// Last deep scrub time.
    pub last_deep_scrub: u64,
}

impl PlacementGroup {
    /// Create a new PG.
    pub fn new(pgid: u64, pool_id: u64, primary: u64, replicas: Vec<u64>) -> Self {
        let mut acting = vec![primary];
        acting.extend(replicas.iter());
        let up = acting.clone();

        Self {
            pgid,
            pool_id,
            state: PgState::Creating,
            primary,
            acting,
            up,
            epoch: 0,
            last_peered_epoch: 0,
            object_count: 0,
            total_bytes: 0,
            last_scrub: 0,
            last_deep_scrub: 0,
        }
    }

    /// Check if this OSD is primary for this PG.
    pub fn is_primary(&self, osd_id: u64) -> bool {
        self.primary == osd_id
    }

    /// Get the acting set excluding primary.
    pub fn replicas(&self) -> Vec<u64> {
        self.acting.iter().skip(1).copied().collect()
    }
}

// ============================================================================
// Local Object Store
// ============================================================================

/// Object stored locally.
#[derive(Debug, Clone)]
pub struct LocalObject {
    /// Object ID.
    pub oid: u64,
    /// Object data.
    pub data: Vec<u8>,
    /// Object version.
    pub version: u64,
    /// Modification time.
    pub mtime: u64,
    /// Extended attributes.
    pub xattrs: BTreeMap<String, Vec<u8>>,
    /// Checksum.
    pub checksum: [u8; 32],
}

impl LocalObject {
    /// Create a new object.
    pub fn new(oid: u64, data: Vec<u8>) -> Self {
        let checksum = Self::compute_checksum(&data);
        Self {
            oid,
            data,
            version: 1,
            mtime: 0,
            xattrs: BTreeMap::new(),
            checksum,
        }
    }

    /// Compute BLAKE3 checksum of data.
    pub fn compute_checksum(data: &[u8]) -> [u8; 32] {
        // Use BLAKE3 for fast, secure hashing
        let mut hasher = blake3::Hasher::new();
        hasher.update(data);
        *hasher.finalize().as_bytes()
    }

    /// Verify checksum.
    pub fn verify(&self) -> bool {
        self.checksum == Self::compute_checksum(&self.data)
    }

    /// Get size.
    pub fn size(&self) -> u64 {
        self.data.len() as u64
    }
}

/// Local object store (per-PG).
#[derive(Debug)]
pub struct LocalStore {
    /// Objects indexed by OID.
    objects: BTreeMap<u64, LocalObject>,
    /// Total bytes stored.
    total_bytes: u64,
    /// Object count.
    object_count: u64,
}

impl LocalStore {
    /// Create a new local store.
    pub fn new() -> Self {
        Self {
            objects: BTreeMap::new(),
            total_bytes: 0,
            object_count: 0,
        }
    }

    /// Write an object.
    pub fn write(&mut self, oid: u64, data: Vec<u8>) -> Result<u64, OsdError> {
        let size = data.len() as u64;

        if let Some(existing) = self.objects.get_mut(&oid) {
            // Update existing object
            self.total_bytes -= existing.data.len() as u64;
            existing.data = data;
            existing.version += 1;
            existing.checksum = LocalObject::compute_checksum(&existing.data);
            self.total_bytes += size;
            Ok(existing.version)
        } else {
            // Create new object
            let obj = LocalObject::new(oid, data);
            let version = obj.version;
            self.objects.insert(oid, obj);
            self.total_bytes += size;
            self.object_count += 1;
            Ok(version)
        }
    }

    /// Read an object.
    pub fn read(&self, oid: u64, offset: u64, length: u64) -> Result<Vec<u8>, OsdError> {
        let obj = self
            .objects
            .get(&oid)
            .ok_or(OsdError::ObjectNotFound(oid))?;

        let start = offset as usize;
        let end = core::cmp::min(start + length as usize, obj.data.len());

        if start >= obj.data.len() {
            return Ok(Vec::new());
        }

        Ok(obj.data[start..end].to_vec())
    }

    /// Read full object.
    pub fn read_full(&self, oid: u64) -> Result<Vec<u8>, OsdError> {
        let obj = self
            .objects
            .get(&oid)
            .ok_or(OsdError::ObjectNotFound(oid))?;
        Ok(obj.data.clone())
    }

    /// Delete an object.
    pub fn delete(&mut self, oid: u64) -> Result<(), OsdError> {
        if let Some(obj) = self.objects.remove(&oid) {
            self.total_bytes -= obj.data.len() as u64;
            self.object_count -= 1;
            Ok(())
        } else {
            Err(OsdError::ObjectNotFound(oid))
        }
    }

    /// Get object stats.
    pub fn stat(&self, oid: u64) -> Result<ObjectStat, OsdError> {
        let obj = self
            .objects
            .get(&oid)
            .ok_or(OsdError::ObjectNotFound(oid))?;
        Ok(ObjectStat {
            oid: obj.oid,
            size: obj.data.len() as u64,
            mtime: obj.mtime,
            version: obj.version,
            checksum: obj.checksum,
        })
    }

    /// Set extended attribute.
    pub fn set_xattr(&mut self, oid: u64, name: &str, value: Vec<u8>) -> Result<(), OsdError> {
        let obj = self
            .objects
            .get_mut(&oid)
            .ok_or(OsdError::ObjectNotFound(oid))?;
        obj.xattrs.insert(name.to_string(), value);
        Ok(())
    }

    /// Get extended attribute.
    pub fn get_xattr(&self, oid: u64, name: &str) -> Result<Vec<u8>, OsdError> {
        let obj = self
            .objects
            .get(&oid)
            .ok_or(OsdError::ObjectNotFound(oid))?;
        obj.xattrs
            .get(name)
            .cloned()
            .ok_or(OsdError::ObjectNotFound(oid))
    }

    /// Remove extended attribute.
    pub fn remove_xattr(&mut self, oid: u64, name: &str) -> Result<(), OsdError> {
        let obj = self
            .objects
            .get_mut(&oid)
            .ok_or(OsdError::ObjectNotFound(oid))?;
        obj.xattrs.remove(name);
        Ok(())
    }

    /// List objects.
    pub fn list(&self, max: usize, marker: Option<u64>) -> Vec<ObjectInfo> {
        let iter: Box<dyn Iterator<Item = _>> = if let Some(m) = marker {
            Box::new(
                self.objects
                    .range((core::ops::Bound::Excluded(m), core::ops::Bound::Unbounded)),
            )
        } else {
            Box::new(self.objects.iter())
        };

        iter.take(max)
            .map(|(&oid, obj)| ObjectInfo {
                oid,
                size: obj.data.len() as u64,
                version: obj.version,
            })
            .collect()
    }

    /// Get total bytes.
    pub fn total_bytes(&self) -> u64 {
        self.total_bytes
    }

    /// Get object count.
    pub fn object_count(&self) -> u64 {
        self.object_count
    }

    /// Verify all objects.
    pub fn verify_all(&self) -> Vec<u64> {
        let mut corrupt = Vec::new();
        for (&oid, obj) in &self.objects {
            if !obj.verify() {
                corrupt.push(oid);
            }
        }
        corrupt
    }
}

impl Default for LocalStore {
    fn default() -> Self {
        Self::new()
    }
}

// ============================================================================
// OSD Instance
// ============================================================================

/// OSD state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OsdState {
    /// OSD is booting.
    Booting,
    /// OSD is active and serving requests.
    Active,
    /// OSD is stopping.
    Stopping,
    /// OSD is stopped.
    Stopped,
    /// OSD encountered an error.
    Error,
}

/// Statistics for an OSD.
#[derive(Debug, Clone, Default)]
pub struct OsdStats {
    /// Total operations processed.
    pub ops_total: u64,
    /// Read operations.
    pub ops_read: u64,
    /// Write operations.
    pub ops_write: u64,
    /// Bytes read.
    pub bytes_read: u64,
    /// Bytes written.
    pub bytes_written: u64,
    /// Operations in progress.
    pub ops_in_progress: u64,
    /// Replication operations sent.
    pub replication_sent: u64,
    /// Replication operations received.
    pub replication_received: u64,
    /// Recovery operations.
    pub recovery_ops: u64,
    /// Scrub operations.
    pub scrub_ops: u64,
}

/// OSD instance.
pub struct Osd {
    /// OSD configuration.
    config: OsdConfig,
    /// OSD ID.
    id: u64,
    /// Current state.
    state: OsdState,
    /// Current cluster epoch.
    epoch: AtomicU64,
    /// Placement groups managed by this OSD.
    pgs: BTreeMap<u64, PlacementGroup>,
    /// Local stores (one per PG).
    stores: BTreeMap<u64, LocalStore>,
    /// Statistics.
    stats: OsdStats,
    /// Is this OSD the primary for any PG?
    is_primary: AtomicBool,
    /// Pending replication acknowledgments.
    pending_acks: BTreeMap<u64, PendingReplication>,
}

/// Pending replication waiting for acknowledgment.
#[derive(Debug)]
struct PendingReplication {
    /// Request ID.
    request_id: u64,
    /// PG ID.
    pgid: u64,
    /// Object ID.
    oid: u64,
    /// Target OSDs.
    targets: Vec<u64>,
    /// Received acknowledgments.
    acks: Vec<u64>,
    /// Required acknowledgments (for quorum).
    required: usize,
    /// Timestamp.
    timestamp: u64,
}

impl Osd {
    /// Create a new OSD with configuration.
    pub fn new(config: OsdConfig) -> Self {
        let id = config.id;
        Self {
            config,
            id,
            state: OsdState::Booting,
            epoch: AtomicU64::new(0),
            pgs: BTreeMap::new(),
            stores: BTreeMap::new(),
            stats: OsdStats::default(),
            is_primary: AtomicBool::new(false),
            pending_acks: BTreeMap::new(),
        }
    }

    /// Get OSD ID.
    pub fn id(&self) -> u64 {
        self.id
    }

    /// Get current state.
    pub fn state(&self) -> OsdState {
        self.state
    }

    /// Get current epoch.
    pub fn epoch(&self) -> u64 {
        self.epoch.load(Ordering::SeqCst)
    }

    /// Set epoch.
    pub fn set_epoch(&self, epoch: u64) {
        self.epoch.store(epoch, Ordering::SeqCst);
    }

    /// Get statistics.
    pub fn stats(&self) -> &OsdStats {
        &self.stats
    }

    /// Boot the OSD.
    pub fn boot(&mut self) -> Result<(), OsdError> {
        self.state = OsdState::Active;
        Ok(())
    }

    /// Shutdown the OSD.
    pub fn shutdown(&mut self) {
        self.state = OsdState::Stopping;
        // Flush pending operations
        self.state = OsdState::Stopped;
    }

    /// Add a PG to this OSD.
    pub fn add_pg(&mut self, pg: PlacementGroup) {
        let pgid = pg.pgid;
        if pg.is_primary(self.id) {
            self.is_primary.store(true, Ordering::SeqCst);
        }
        self.pgs.insert(pgid, pg);
        self.stores.entry(pgid).or_default();
    }

    /// Remove a PG from this OSD.
    pub fn remove_pg(&mut self, pgid: u64) {
        self.pgs.remove(&pgid);
        self.stores.remove(&pgid);

        // Update primary status
        let still_primary = self.pgs.values().any(|pg| pg.is_primary(self.id));
        self.is_primary.store(still_primary, Ordering::SeqCst);
    }

    /// Get a PG.
    pub fn get_pg(&self, pgid: u64) -> Option<&PlacementGroup> {
        self.pgs.get(&pgid)
    }

    /// Get a mutable PG.
    pub fn get_pg_mut(&mut self, pgid: u64) -> Option<&mut PlacementGroup> {
        self.pgs.get_mut(&pgid)
    }

    /// Handle an incoming operation.
    pub fn handle_op(&mut self, msg: OsdMessage) -> OsdResponse {
        let request_id = msg.request_id;
        let current_epoch = self.epoch();

        // Check epoch
        if msg.epoch < current_epoch && !msg.flags.ignore_state {
            return OsdResponse {
                request_id,
                result: OsdResult::Error(OsdError::StaleEpoch {
                    expected: current_epoch,
                    got: msg.epoch,
                }),
                epoch: current_epoch,
            };
        }

        // Check PG
        let pg = match self.pgs.get(&msg.pgid) {
            Some(pg) => pg,
            None => {
                return OsdResponse {
                    request_id,
                    result: OsdResult::Error(OsdError::PgNotFound(msg.pgid)),
                    epoch: current_epoch,
                };
            }
        };

        // Check PG state for writes
        let is_write = matches!(
            msg.op,
            OsdOp::Write { .. }
                | OsdOp::WriteFull { .. }
                | OsdOp::Delete { .. }
                | OsdOp::Truncate { .. }
        );

        if is_write && !pg.state.can_write() && !msg.flags.ignore_state {
            return OsdResponse {
                request_id,
                result: OsdResult::Error(OsdError::PgNotActive(msg.pgid, pg.state)),
                epoch: current_epoch,
            };
        }

        if !is_write && !pg.state.can_read() && !msg.flags.ignore_state {
            return OsdResponse {
                request_id,
                result: OsdResult::Error(OsdError::PgNotActive(msg.pgid, pg.state)),
                epoch: current_epoch,
            };
        }

        // Execute operation
        let result = self.execute_op(msg.pgid, msg.op, msg.flags);

        self.stats.ops_total += 1;

        OsdResponse {
            request_id,
            result,
            epoch: current_epoch,
        }
    }

    /// Execute an operation.
    fn execute_op(&mut self, pgid: u64, op: OsdOp, _flags: OsdOpFlags) -> OsdResult {
        let store = match self.stores.get_mut(&pgid) {
            Some(s) => s,
            None => return OsdResult::Error(OsdError::PgNotFound(pgid)),
        };

        match op {
            OsdOp::Read {
                oid,
                offset,
                length,
            } => {
                self.stats.ops_read += 1;
                match store.read(oid, offset, length) {
                    Ok(data) => {
                        self.stats.bytes_read += data.len() as u64;
                        OsdResult::Success(Some(data))
                    }
                    Err(e) => OsdResult::Error(e),
                }
            }

            OsdOp::Write { oid, offset, data } => {
                self.stats.ops_write += 1;
                self.stats.bytes_written += data.len() as u64;

                // Read existing data, modify, write back
                let existing = store.read_full(oid).unwrap_or_default();
                let mut new_data = existing;

                let end = offset as usize + data.len();
                if new_data.len() < end {
                    new_data.resize(end, 0);
                }
                new_data[offset as usize..end].copy_from_slice(&data);

                match store.write(oid, new_data) {
                    Ok(_) => OsdResult::Success(None),
                    Err(e) => OsdResult::Error(e),
                }
            }

            OsdOp::WriteFull { oid, data } => {
                self.stats.ops_write += 1;
                self.stats.bytes_written += data.len() as u64;

                match store.write(oid, data) {
                    Ok(_) => OsdResult::Success(None),
                    Err(e) => OsdResult::Error(e),
                }
            }

            OsdOp::Delete { oid } => match store.delete(oid) {
                Ok(_) => OsdResult::Success(None),
                Err(e) => OsdResult::Error(e),
            },

            OsdOp::Truncate { oid, size } => match store.read_full(oid) {
                Ok(mut data) => {
                    data.truncate(size as usize);
                    match store.write(oid, data) {
                        Ok(_) => OsdResult::Success(None),
                        Err(e) => OsdResult::Error(e),
                    }
                }
                Err(e) => OsdResult::Error(e),
            },

            OsdOp::Stat { oid } => match store.stat(oid) {
                Ok(stat) => OsdResult::Stat(stat),
                Err(e) => OsdResult::Error(e),
            },

            OsdOp::SetXattr { oid, name, value } => match store.set_xattr(oid, &name, value) {
                Ok(_) => OsdResult::Success(None),
                Err(e) => OsdResult::Error(e),
            },

            OsdOp::GetXattr { oid, name } => match store.get_xattr(oid, &name) {
                Ok(value) => OsdResult::Xattr(value),
                Err(e) => OsdResult::Error(e),
            },

            OsdOp::RemoveXattr { oid, name } => match store.remove_xattr(oid, &name) {
                Ok(_) => OsdResult::Success(None),
                Err(e) => OsdResult::Error(e),
            },

            OsdOp::List { max, marker } => {
                let objects = store.list(max, marker);
                OsdResult::List(objects)
            }

            OsdOp::Replicate { oid, target_osd: _ } => {
                // This would normally send data to another OSD
                // For now, just acknowledge
                self.stats.replication_sent += 1;
                match store.read_full(oid) {
                    Ok(_data) => {
                        // In real implementation: send data to target_osd
                        OsdResult::Success(None)
                    }
                    Err(e) => OsdResult::Error(e),
                }
            }

            OsdOp::Pull {
                oid: _,
                source_osd: _,
            } => {
                // This would normally fetch data from another OSD
                self.stats.recovery_ops += 1;
                // In real implementation: request data from source_osd
                OsdResult::Success(None)
            }

            OsdOp::Scrub { oid } => {
                self.stats.scrub_ops += 1;
                let corrupt = if let Some(oid) = oid {
                    match store.objects.get(&oid) {
                        Some(obj) if !obj.verify() => vec![oid],
                        _ => vec![],
                    }
                } else {
                    store.verify_all()
                };

                if corrupt.is_empty() {
                    OsdResult::Success(None)
                } else {
                    // Return first corrupt object as error
                    let oid = corrupt[0];
                    let obj = store.objects.get(&oid).unwrap();
                    OsdResult::Error(OsdError::ChecksumMismatch {
                        oid,
                        expected: obj.checksum,
                        got: LocalObject::compute_checksum(&obj.data),
                    })
                }
            }
        }
    }

    /// Primary write with replication.
    pub fn primary_write(
        &mut self,
        pgid: u64,
        oid: u64,
        data: Vec<u8>,
        request_id: u64,
    ) -> Result<(), OsdError> {
        let pg = self.pgs.get(&pgid).ok_or(OsdError::PgNotFound(pgid))?;

        if !pg.is_primary(self.id) {
            return Err(OsdError::NotPrimary(pgid));
        }

        // Write locally first
        let store = self
            .stores
            .get_mut(&pgid)
            .ok_or(OsdError::PgNotFound(pgid))?;
        store.write(oid, data.clone())?;

        // Get replica OSDs
        let replicas = pg.replicas();

        if replicas.is_empty() {
            // No replicas, we're done
            return Ok(());
        }

        // Track pending replication
        let pending = PendingReplication {
            request_id,
            pgid,
            oid,
            targets: replicas.clone(),
            acks: vec![self.id], // We count ourselves as acked
            required: replicas.len().div_ceil(2) + 1, // Majority quorum
            timestamp: 0,        // Would be set to current time
        };

        self.pending_acks.insert(request_id, pending);
        self.stats.replication_sent += replicas.len() as u64;

        // In real implementation: send replication messages to replicas
        // For now, we simulate immediate success

        Ok(())
    }

    /// Handle replication acknowledgment.
    pub fn handle_ack(&mut self, request_id: u64, from_osd: u64) -> bool {
        if let Some(pending) = self.pending_acks.get_mut(&request_id) {
            if !pending.acks.contains(&from_osd) {
                pending.acks.push(from_osd);
            }

            if pending.acks.len() >= pending.required {
                self.pending_acks.remove(&request_id);
                return true; // Quorum reached
            }
        }
        false
    }

    /// Get PG stats.
    pub fn pg_stats(&self, pgid: u64) -> Option<(u64, u64)> {
        self.stores
            .get(&pgid)
            .map(|s| (s.object_count(), s.total_bytes()))
    }

    /// Get all PG IDs.
    pub fn pg_ids(&self) -> Vec<u64> {
        self.pgs.keys().copied().collect()
    }

    /// Update PG state.
    pub fn set_pg_state(&mut self, pgid: u64, state: PgState) {
        if let Some(pg) = self.pgs.get_mut(&pgid) {
            pg.state = state;
        }
    }
}

// ============================================================================
// OSD Network Provider Trait
// ============================================================================

/// Trait for OSD network communication.
/// Implemented by the platform (std or LunaOS kernel).
pub trait OsdNetwork: Send + Sync {
    /// Send a message to another OSD.
    fn send(&self, target_osd: u64, message: &[u8]) -> Result<Vec<u8>, OsdError>;

    /// Broadcast to all OSDs.
    fn broadcast(&self, message: &[u8]) -> Vec<(u64, Result<Vec<u8>, OsdError>)>;

    /// Register to receive messages.
    fn listen(&self) -> Result<(), OsdError>;
}

/// No-op network provider for standalone/testing.
#[derive(Debug, Default)]
pub struct NoOpOsdNetwork;

impl OsdNetwork for NoOpOsdNetwork {
    fn send(&self, _target: u64, _message: &[u8]) -> Result<Vec<u8>, OsdError> {
        // No-op: pretend success
        Ok(Vec::new())
    }

    fn broadcast(&self, _message: &[u8]) -> Vec<(u64, Result<Vec<u8>, OsdError>)> {
        Vec::new()
    }

    fn listen(&self) -> Result<(), OsdError> {
        Ok(())
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;

    fn create_test_osd() -> Osd {
        let config = OsdConfig {
            id: 0,
            ..Default::default()
        };
        let mut osd = Osd::new(config);
        osd.boot().unwrap();
        osd
    }

    #[test]
    fn test_osd_creation() {
        let osd = create_test_osd();
        assert_eq!(osd.id(), 0);
        assert_eq!(osd.state(), OsdState::Active);
    }

    #[test]
    fn test_add_pg() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![1, 2]);
        osd.add_pg(pg);

        assert!(osd.get_pg(1).is_some());
        assert!(osd.is_primary.load(Ordering::SeqCst));
    }

    #[test]
    fn test_write_read() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![]);
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Active);

        // Write
        let write_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::WriteFull {
                oid: 100,
                data: b"Hello, LCPFS!".to_vec(),
            },
            request_id: 1,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };

        let resp = osd.handle_op(write_msg);
        assert!(matches!(resp.result, OsdResult::Success(_)));

        // Read
        let read_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::Read {
                oid: 100,
                offset: 0,
                length: 100,
            },
            request_id: 2,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };

        let resp = osd.handle_op(read_msg);
        match resp.result {
            OsdResult::Success(Some(data)) => {
                assert_eq!(data, b"Hello, LCPFS!");
            }
            _ => panic!("Expected success with data"),
        }
    }

    #[test]
    fn test_partial_read() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![]);
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Active);

        // Write
        let write_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::WriteFull {
                oid: 100,
                data: b"Hello, World!".to_vec(),
            },
            request_id: 1,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        osd.handle_op(write_msg);

        // Partial read
        let read_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::Read {
                oid: 100,
                offset: 7,
                length: 5,
            },
            request_id: 2,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };

        let resp = osd.handle_op(read_msg);
        match resp.result {
            OsdResult::Success(Some(data)) => {
                assert_eq!(data, b"World");
            }
            _ => panic!("Expected success with data"),
        }
    }

    #[test]
    fn test_delete() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![]);
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Active);

        // Write
        let write_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::WriteFull {
                oid: 100,
                data: b"test".to_vec(),
            },
            request_id: 1,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        osd.handle_op(write_msg);

        // Delete
        let delete_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::Delete { oid: 100 },
            request_id: 2,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        let resp = osd.handle_op(delete_msg);
        assert!(matches!(resp.result, OsdResult::Success(_)));

        // Read should fail
        let read_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::Read {
                oid: 100,
                offset: 0,
                length: 100,
            },
            request_id: 3,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        let resp = osd.handle_op(read_msg);
        assert!(matches!(
            resp.result,
            OsdResult::Error(OsdError::ObjectNotFound(100))
        ));
    }

    #[test]
    fn test_stat() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![]);
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Active);

        // Write
        let data = b"test data for stat";
        let write_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::WriteFull {
                oid: 100,
                data: data.to_vec(),
            },
            request_id: 1,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        osd.handle_op(write_msg);

        // Stat
        let stat_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::Stat { oid: 100 },
            request_id: 2,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        let resp = osd.handle_op(stat_msg);

        match resp.result {
            OsdResult::Stat(stat) => {
                assert_eq!(stat.oid, 100);
                assert_eq!(stat.size, data.len() as u64);
                assert_eq!(stat.version, 1);
            }
            _ => panic!("Expected stat result"),
        }
    }

    #[test]
    fn test_xattrs() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![]);
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Active);

        // Create object first
        let write_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::WriteFull {
                oid: 100,
                data: b"data".to_vec(),
            },
            request_id: 1,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        osd.handle_op(write_msg);

        // Set xattr
        let set_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::SetXattr {
                oid: 100,
                name: "user.test".to_string(),
                value: b"xattr value".to_vec(),
            },
            request_id: 2,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        osd.handle_op(set_msg);

        // Get xattr
        let get_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::GetXattr {
                oid: 100,
                name: "user.test".to_string(),
            },
            request_id: 3,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        let resp = osd.handle_op(get_msg);

        match resp.result {
            OsdResult::Xattr(value) => {
                assert_eq!(value, b"xattr value");
            }
            _ => panic!("Expected xattr result"),
        }
    }

    #[test]
    fn test_list() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![]);
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Active);

        // Write multiple objects
        for i in 0..5 {
            let write_msg = OsdMessage {
                epoch: 0,
                pgid: 1,
                op: OsdOp::WriteFull {
                    oid: 100 + i,
                    data: vec![i as u8; 100],
                },
                request_id: i,
                source_osd: None,
                flags: OsdOpFlags::default(),
            };
            osd.handle_op(write_msg);
        }

        // List
        let list_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::List {
                max: 10,
                marker: None,
            },
            request_id: 100,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        let resp = osd.handle_op(list_msg);

        match resp.result {
            OsdResult::List(objects) => {
                assert_eq!(objects.len(), 5);
            }
            _ => panic!("Expected list result"),
        }
    }

    #[test]
    fn test_pg_state_check() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![]);
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Peering); // Not active

        // Write should fail
        let write_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::WriteFull {
                oid: 100,
                data: b"test".to_vec(),
            },
            request_id: 1,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        let resp = osd.handle_op(write_msg);

        assert!(matches!(
            resp.result,
            OsdResult::Error(OsdError::PgNotActive(1, PgState::Peering))
        ));
    }

    #[test]
    fn test_stale_epoch() {
        let mut osd = create_test_osd();
        osd.set_epoch(10);

        let pg = PlacementGroup::new(1, 0, 0, vec![]);
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Active);

        // Old epoch message
        let msg = OsdMessage {
            epoch: 5, // Stale
            pgid: 1,
            op: OsdOp::Stat { oid: 100 },
            request_id: 1,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        let resp = osd.handle_op(msg);

        match resp.result {
            OsdResult::Error(OsdError::StaleEpoch {
                expected: 10,
                got: 5,
            }) => {}
            _ => panic!("Expected stale epoch error"),
        }
    }

    #[test]
    fn test_primary_write() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![1, 2]); // Replicas on OSDs 1, 2
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Active);

        let result = osd.primary_write(1, 100, b"replicated data".to_vec(), 1);
        assert!(result.is_ok());

        // Verify data was written locally
        let (count, _) = osd.pg_stats(1).unwrap();
        assert_eq!(count, 1);
    }

    #[test]
    fn test_local_store() {
        let mut store = LocalStore::new();

        // Write
        store.write(1, b"hello".to_vec()).unwrap();
        store.write(2, b"world".to_vec()).unwrap();

        assert_eq!(store.object_count(), 2);
        assert_eq!(store.total_bytes(), 10);

        // Read
        let data = store.read_full(1).unwrap();
        assert_eq!(data, b"hello");

        // Delete
        store.delete(1).unwrap();
        assert_eq!(store.object_count(), 1);

        // Verify checksum
        assert!(store.verify_all().is_empty());
    }

    #[test]
    fn test_pg_state_transitions() {
        assert!(PgState::Active.can_read());
        assert!(PgState::Active.can_write());
        assert!(PgState::Active.is_healthy());

        assert!(PgState::Degraded.can_read());
        assert!(PgState::Degraded.can_write());
        assert!(!PgState::Degraded.is_healthy());

        assert!(!PgState::Peering.can_read());
        assert!(!PgState::Peering.can_write());

        assert!(PgState::Recovering.can_read());
        assert!(PgState::Recovering.can_write());
    }

    #[test]
    fn test_object_checksum() {
        let obj = LocalObject::new(1, b"test data".to_vec());
        assert!(obj.verify());

        // Corrupt the data
        let mut obj2 = obj.clone();
        obj2.data[0] = 0xFF;
        assert!(!obj2.verify());
    }

    #[test]
    fn test_scrub() {
        let mut osd = create_test_osd();

        let pg = PlacementGroup::new(1, 0, 0, vec![]);
        osd.add_pg(pg);
        osd.set_pg_state(1, PgState::Active);

        // Write
        let write_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::WriteFull {
                oid: 100,
                data: b"scrub test".to_vec(),
            },
            request_id: 1,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        osd.handle_op(write_msg);

        // Scrub should pass
        let scrub_msg = OsdMessage {
            epoch: 0,
            pgid: 1,
            op: OsdOp::Scrub { oid: Some(100) },
            request_id: 2,
            source_osd: None,
            flags: OsdOpFlags::default(),
        };
        let resp = osd.handle_op(scrub_msg);
        assert!(matches!(resp.result, OsdResult::Success(_)));
    }
}