plato_relay_tidepool/

lib.rs

//! plato-relay-tidepool — TidePoolLayer adapter for plato-relay
//!
//! Bridges plato-relay's trust-weighted BFS routing to the Ship Interconnection
//! Protocol's TidePoolLayer trait. Messages are enqueued, routed through trust
//! weighting, and dequeued for delivery.
//!
//! Sprint 3 Task S3-3: implement TidePoolLayer for plato-relay.

use std::collections::{HashMap, VecDeque};

// ── TidePool Trait ───────────────────────────────────────

/// Tide pool layer: async message routing with buffering.
/// Matches plato-ship-protocol::TidePoolLayer exactly.
pub trait TidePoolLayer {
    fn enqueue(&mut self, msg: &[u8]) -> bool;
    fn dequeue(&mut self) -> Option<Vec<u8>>;
    fn buffer_len(&self) -> usize;
}

// ── Queued Message ───────────────────────────────────────

#[derive(Debug, Clone)]
pub struct QueuedMessage {
    pub payload: Vec<u8>,
    pub priority: f32,       // 0.0-1.0, higher = delivered first
    pub sender_trust: f32,   // sender's trust level
    pub enqueued_at: u64,    // nanosecond timestamp
    pub hops: u32,
}

impl QueuedMessage {
    pub fn new(payload: &[u8]) -> Self {
        Self {
            payload: payload.to_vec(),
            priority: 0.5,
            sender_trust: 0.5,
            enqueued_at: nanos_now(),
            hops: 0,
        }
    }

    pub fn with_priority(mut self, p: f32) -> Self {
        self.priority = p.max(0.0).min(1.0);
        self
    }

    pub fn with_trust(mut self, t: f32) -> Self {
        self.sender_trust = t.max(0.0).min(1.0);
        self
    }

    /// Effective priority: base priority weighted by sender trust
    pub fn effective_priority(&self) -> f32 {
        self.priority * (0.3 + 0.7 * self.sender_trust)
    }
}

// ── TidePool Adapter ────────────────────────────────────

/// Adapts trust-weighted message routing into a TidePoolLayer.
/// Messages are buffered, scored by trust, and dequeued in priority order.
#[derive(Debug, Clone)]
pub struct TidePoolAdapter {
    buffer: VecDeque<QueuedMessage>,
    capacity: usize,
    total_enqueued: u64,
    total_dequeued: u64,
    total_dropped: u64,
}

impl TidePoolAdapter {
    pub fn new() -> Self {
        Self { buffer: VecDeque::new(), capacity: 1024, total_enqueued: 0, total_dequeued: 0, total_dropped: 0 }
    }

    pub fn with_capacity(cap: usize) -> Self {
        Self { buffer: VecDeque::new(), capacity: cap, total_enqueued: 0, total_dequeued: 0, total_dropped: 0 }
    }

    /// Enqueue with trust and priority metadata
    pub fn enqueue_trusted(&mut self, payload: &[u8], trust: f32, priority: f32) -> bool {
        if self.buffer.len() >= self.capacity {
            self.total_dropped += 1;
            return false;
        }
        let msg = QueuedMessage::new(payload)
            .with_trust(trust)
            .with_priority(priority);
        self.buffer.push_back(msg);
        self.total_enqueued += 1;
        true
    }

    /// Dequeue highest-priority message (trust-weighted)
    pub fn dequeue_priority(&mut self) -> Option<QueuedMessage> {
        if self.buffer.is_empty() { return None; }

        // Find highest effective priority
        let best_idx = (0..self.buffer.len())
            .max_by(|a, b| {
                self.buffer[*a].effective_priority()
                    .partial_cmp(&self.buffer[*b].effective_priority())
                    .unwrap_or(std::cmp::Ordering::Equal)
            })?;

        let msg = self.buffer.remove(best_idx)?;
        self.total_dequeued += 1;
        Some(msg)
    }

    /// Stats
    pub fn total_enqueued(&self) -> u64 { self.total_enqueued }
    pub fn total_dequeued(&self) -> u64 { self.total_dequeued }
    pub fn total_dropped(&self) -> u64 { self.total_dropped }

    /// Average trust of buffered messages
    pub fn avg_buffer_trust(&self) -> f32 {
        if self.buffer.is_empty() { return 0.0; }
        let sum: f32 = self.buffer.iter().map(|m| m.sender_trust).sum();
        sum / self.buffer.len() as f32
    }

    /// Drain all messages (for batch processing)
    pub fn drain(&mut self) -> Vec<QueuedMessage> {
        let msgs: Vec<_> = self.buffer.drain(..).collect();
        self.total_dequeued += msgs.len() as u64;
        msgs
    }
}

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

impl TidePoolLayer for TidePoolAdapter {
    fn enqueue(&mut self, msg: &[u8]) -> bool {
        if self.buffer.len() >= self.capacity {
            self.total_dropped += 1;
            return false;
        }
        self.buffer.push_back(QueuedMessage::new(msg));
        self.total_enqueued += 1;
        true
    }

    fn dequeue(&mut self) -> Option<Vec<u8>> {
        let msg = self.buffer.pop_front()?;
        self.total_dequeued += 1;
        Some(msg.payload)
    }

    fn buffer_len(&self) -> usize {
        self.buffer.len()
    }
}

// ── Trust-Weighted Router ───────────────────────────────

/// Routes messages between named agents with trust scoring.
/// Integrates TidePoolLayer with trust-based priority.
#[derive(Debug, Clone)]
pub struct TrustRouter {
    pools: HashMap<String, TidePoolAdapter>,
    trust: HashMap<String, f32>, // agent_id → trust score
    default_trust: f32,
}

impl TrustRouter {
    pub fn new() -> Self {
        Self { pools: HashMap::new(), trust: HashMap::new(), default_trust: 0.5 }
    }

    pub fn with_default_trust(t: f32) -> Self {
        Self { pools: HashMap::new(), trust: HashMap::new(), default_trust: t }
    }

    /// Register an agent with a pool
    pub fn register_agent(&mut self, agent_id: &str, capacity: usize) {
        self.pools.insert(agent_id.to_string(), TidePoolAdapter::with_capacity(capacity));
    }

    /// Set trust level for an agent
    pub fn set_trust(&mut self, agent_id: &str, trust: f32) {
        self.trust.insert(agent_id.to_string(), trust.max(0.0).min(1.0));
    }

    /// Get trust for an agent
    pub fn get_trust(&self, agent_id: &str) -> f32 {
        *self.trust.get(agent_id).unwrap_or(&self.default_trust)
    }

    /// Route a message to an agent's pool with trust-weighted priority
    pub fn route(&mut self, to_agent: &str, payload: &[u8], from_trust: f32) -> bool {
        if let Some(pool) = self.pools.get_mut(to_agent) {
            pool.enqueue_trusted(payload, from_trust, 0.5)
        } else {
            false
        }
    }

    /// Get next message for an agent (highest trust-weighted priority)
    pub fn next_for(&mut self, agent_id: &str) -> Option<QueuedMessage> {
        self.pools.get_mut(agent_id)?.dequeue_priority()
    }

    /// Buffer size for an agent
    pub fn buffer_size(&self, agent_id: &str) -> usize {
        self.pools.get(agent_id).map(|p| p.buffer_len()).unwrap_or(0)
    }

    /// Total messages across all pools
    pub fn total_buffered(&self) -> usize {
        self.pools.values().map(|p| p.buffer_len()).sum()
    }

    /// Agent count
    pub fn agent_count(&self) -> usize { self.pools.len() }
}

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

fn nanos_now() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_nanos() as u64).unwrap_or(0)
}

// ── Tests ────────────────────────────────────────────────

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

    #[test]
    fn test_enqueue_dequeue() {
        let mut pool = TidePoolAdapter::new();
        assert!(pool.enqueue(b"hello"));
        assert!(pool.enqueue(b"world"));
        assert_eq!(pool.buffer_len(), 2);

        let msg = pool.dequeue().unwrap();
        assert_eq!(msg, b"hello");
        assert_eq!(pool.buffer_len(), 1);
    }

    #[test]
    fn test_dequeue_empty() {
        let mut pool = TidePoolAdapter::new();
        assert!(pool.dequeue().is_none());
    }

    #[test]
    fn test_fifo_order() {
        let mut pool = TidePoolAdapter::new();
        pool.enqueue(b"first");
        pool.enqueue(b"second");
        pool.enqueue(b"third");

        assert_eq!(pool.dequeue().unwrap(), b"first");
        assert_eq!(pool.dequeue().unwrap(), b"second");
        assert_eq!(pool.dequeue().unwrap(), b"third");
    }

    #[test]
    fn test_capacity_limit() {
        let mut pool = TidePoolAdapter::with_capacity(2);
        assert!(pool.enqueue(b"a"));
        assert!(pool.enqueue(b"b"));
        assert!(!pool.enqueue(b"c")); // over capacity
        assert_eq!(pool.total_dropped(), 1);
    }

    #[test]
    fn test_trusted_enqueue() {
        let mut pool = TidePoolAdapter::new();
        pool.enqueue_trusted(b"trusted", 0.9, 0.5);
        pool.enqueue_trusted(b"untrusted", 0.1, 0.5);

        // dequeue_priority should return trusted first
        let first = pool.dequeue_priority().unwrap();
        assert_eq!(first.payload, b"trusted");
    }

    #[test]
    fn test_effective_priority() {
        let high_trust = QueuedMessage::new(b"hi")
            .with_trust(0.9).with_priority(0.5);
        let low_trust = QueuedMessage::new(b"hi")
            .with_trust(0.1).with_priority(0.5);

        // 0.5 * (0.3 + 0.7 * 0.9) = 0.5 * 0.93 = 0.465
        assert!((high_trust.effective_priority() - 0.465).abs() < 0.01);
        // 0.5 * (0.3 + 0.7 * 0.1) = 0.5 * 0.37 = 0.185
        assert!((low_trust.effective_priority() - 0.185).abs() < 0.01);
    }

    #[test]
    fn test_priority_ordering() {
        let mut pool = TidePoolAdapter::new();
        pool.enqueue_trusted(b"low", 0.1, 0.5);
        pool.enqueue_trusted(b"high", 0.9, 0.8);
        pool.enqueue_trusted(b"mid", 0.5, 0.5);

        let first = pool.dequeue_priority().unwrap();
        assert_eq!(first.payload, b"high");
    }

    #[test]
    fn test_drain() {
        let mut pool = TidePoolAdapter::new();
        pool.enqueue(b"a");
        pool.enqueue(b"b");
        pool.enqueue(b"c");

        let msgs = pool.drain();
        assert_eq!(msgs.len(), 3);
        assert!(pool.buffer.is_empty());
        assert_eq!(pool.total_dequeued(), 3);
    }

    #[test]
    fn test_stats() {
        let mut pool = TidePoolAdapter::with_capacity(1);
        pool.enqueue(b"a");
        pool.enqueue(b"b"); // dropped (over capacity)
        pool.dequeue();

        assert_eq!(pool.total_enqueued(), 1);  // only 'a' was actually enqueued
        assert_eq!(pool.total_dequeued(), 1);
        assert_eq!(pool.total_dropped(), 1);   // 'b' was dropped
    }

    #[test]
    fn test_avg_buffer_trust() {
        let mut pool = TidePoolAdapter::new();
        pool.enqueue_trusted(b"a", 0.8, 0.5);
        pool.enqueue_trusted(b"b", 0.4, 0.5);

        let avg = pool.avg_buffer_trust();
        assert!((avg - 0.6).abs() < 0.01);
    }

    #[test]
    fn test_trust_router() {
        let mut router = TrustRouter::new();
        router.register_agent("oracle1", 100);
        router.register_agent("jc1", 100);
        router.set_trust("oracle1", 0.9);

        assert!(router.route("oracle1", b"hello", 0.8));
        assert_eq!(router.buffer_size("oracle1"), 1);
        assert_eq!(router.buffer_size("jc1"), 0);

        let msg = router.next_for("oracle1").unwrap();
        assert_eq!(msg.payload, b"hello");
    }

    #[test]
    fn test_trust_router_unknown_agent() {
        let mut router = TrustRouter::new();
        assert!(!router.route("nonexistent", b"msg", 0.5));
    }

    #[test]
    fn test_trust_router_default_trust() {
        let router = TrustRouter::with_default_trust(0.3);
        assert_eq!(router.get_trust("unknown"), 0.3);
        assert_eq!(router.get_trust("known"), 0.3); // not set
    }

    #[test]
    fn test_trust_router_trust_weighted_delivery() {
        let mut router = TrustRouter::new();
        router.register_agent("dest", 100);

        // Low trust sender
        router.route("dest", b"low_priority", 0.1);
        // High trust sender
        router.route("dest", b"high_priority", 0.9);

        let first = router.next_for("dest").unwrap();
        assert_eq!(first.payload, b"high_priority");
    }

    #[test]
    fn test_queued_message_clamping() {
        let msg = QueuedMessage::new(b"test").with_priority(2.0).with_trust(-0.5);
        assert_eq!(msg.priority, 1.0);
        assert_eq!(msg.sender_trust, 0.0);
    }
}