relay_knowledge/watcher/engine/

mod.rs

use std::{
    future::Future,
    pin::Pin,
    sync::{Arc, atomic::AtomicU64},
    time::Duration,
};

use serde::{Deserialize, Serialize};
use tokio::sync::{RwLock, mpsc, oneshot, watch};

use super::{ContentHashCache, WatchedRepository, config::WatcherConfig};

mod diagnostics;
mod event_loop;
mod index_queue;
mod repository_registry;

const COMMAND_CHANNEL_CAPACITY: usize = 128;
const COMMAND_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);

type TaskQueueFuture = Pin<Box<dyn Future<Output = Result<(), String>> + Send>>;
type TaskQueueSink =
    Arc<dyn Fn(crate::storage::CodeIndexTaskSeed) -> TaskQueueFuture + Send + Sync>;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WatcherState {
    Disabled,
    Active,
    Degraded,
    Failed,
}

impl WatcherState {
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Disabled => "disabled",
            Self::Active => "active",
            Self::Degraded => "degraded",
            Self::Failed => "failed",
        }
    }

    pub fn parse(value: &str) -> Option<Self> {
        match value {
            "disabled" => Some(Self::Disabled),
            "active" => Some(Self::Active),
            "degraded" => Some(Self::Degraded),
            "failed" => Some(Self::Failed),
            _ => None,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WatcherDiagnostics {
    pub state: WatcherState,
    pub watched_repository_count: usize,
    pub total_events_received: u64,
    pub total_events_filtered: u64,
    pub total_index_tasks_queued: u64,
    pub total_commit_reconciliations: u64,
    pub total_commit_tasks_queued: u64,
    pub total_commit_reconcile_failures: u64,
    pub total_events_dropped: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_error: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub degraded_reason: Option<String>,
}

impl Default for WatcherDiagnostics {
    fn default() -> Self {
        Self {
            state: WatcherState::Disabled,
            watched_repository_count: 0,
            total_events_received: 0,
            total_events_filtered: 0,
            total_index_tasks_queued: 0,
            total_commit_reconciliations: 0,
            total_commit_tasks_queued: 0,
            total_commit_reconcile_failures: 0,
            total_events_dropped: 0,
            last_error: None,
            degraded_reason: None,
        }
    }
}

#[derive(Debug, Clone)]
pub struct WatcherHandle {
    diagnostics: watch::Receiver<WatcherDiagnostics>,
    shutdown: watch::Sender<bool>,
    state: Arc<RwLock<WatcherInternalState>>,
    command_tx: Option<mpsc::Sender<WatcherCommand>>,
}

impl WatcherHandle {
    pub fn diagnostics(&self) -> WatcherDiagnostics {
        self.diagnostics.borrow().clone()
    }

    pub async fn updated_diagnostics(&self) -> WatcherDiagnostics {
        let mut diagnostics = self.diagnostics.clone();
        let _ = diagnostics.changed().await;
        diagnostics.borrow().clone()
    }

    pub fn request_shutdown(&self) {
        let _ = self.shutdown.send(true);
    }

    pub async fn add_repository(&self, repository: WatchedRepository) -> bool {
        let Some(command_tx) = &self.command_tx else {
            return false;
        };
        let (response_tx, response_rx) = oneshot::channel();
        let command = WatcherCommand::Add {
            repository,
            response: response_tx,
        };
        if command_tx.send(command).await.is_err() {
            return false;
        }
        matches!(
            tokio::time::timeout(COMMAND_RESPONSE_TIMEOUT, response_rx).await,
            Ok(Ok(true))
        )
    }

    pub async fn remove_repository(&self, alias: &str) -> bool {
        let Some(command_tx) = &self.command_tx else {
            return false;
        };
        let (response_tx, response_rx) = oneshot::channel();
        let command = WatcherCommand::Remove {
            alias_or_id: alias.to_owned(),
            response: response_tx,
        };
        if command_tx.send(command).await.is_err() {
            return false;
        }
        matches!(
            tokio::time::timeout(COMMAND_RESPONSE_TIMEOUT, response_rx).await,
            Ok(Ok(true))
        )
    }

    pub async fn repository_count(&self) -> usize {
        self.state.read().await.repositories.len()
    }
}

pub struct FileWatcher {
    config: WatcherConfig,
}

impl FileWatcher {
    pub fn new(config: WatcherConfig) -> Self {
        Self { config }
    }

    pub fn start(self, repositories: Vec<WatchedRepository>) -> Result<WatcherHandle, String> {
        self.start_with_sink(repositories, |_| async { Ok(()) })
    }

    pub fn start_with_sink<F, Fut>(
        self,
        repositories: Vec<WatchedRepository>,
        task_sink: F,
    ) -> Result<WatcherHandle, String>
    where
        F: Fn(crate::storage::CodeIndexTaskSeed) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<(), String>> + Send + 'static,
    {
        if !self.config.enabled {
            return Ok(disabled_handle(self.config.hash_cache_capacity));
        }

        let (diagnostics_tx, diagnostics_rx) = watch::channel(WatcherDiagnostics {
            state: WatcherState::Active,
            ..WatcherDiagnostics::default()
        });
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let (command_tx, command_rx) = mpsc::channel(COMMAND_CHANNEL_CAPACITY);
        let state = Arc::new(RwLock::new(WatcherInternalState::new(
            self.config.hash_cache_capacity,
        )));
        let handle = WatcherHandle {
            diagnostics: diagnostics_rx,
            shutdown: shutdown_tx,
            state: Arc::clone(&state),
            command_tx: Some(command_tx),
        };
        let context = WatcherLoopContext {
            state,
            diagnostics_tx,
            dropped_events: Arc::new(AtomicU64::new(0)),
            debounce: self.config.debounce,
            commit_reconcile_interval: self.config.commit_reconcile_interval,
            max_watch_dirs: self.config.max_watch_dirs,
            task_sink: boxed_task_sink(task_sink),
        };
        tokio::spawn(event_loop::run(
            context,
            shutdown_rx,
            repositories,
            command_rx,
        ));

        Ok(handle)
    }
}

#[derive(Debug)]
struct WatcherInternalState {
    repositories: Vec<WatchedRepository>,
    hash_cache: ContentHashCache,
    deferred_changes: ContentHashCache,
    events_received: u64,
    events_filtered: u64,
    index_tasks_queued: u64,
    commit_reconciliations: u64,
    commit_tasks_queued: u64,
    commit_reconcile_failures: u64,
}

impl WatcherInternalState {
    fn new(hash_cache_capacity: usize) -> Self {
        Self {
            repositories: Vec::new(),
            hash_cache: ContentHashCache::new(hash_cache_capacity),
            deferred_changes: ContentHashCache::new(hash_cache_capacity),
            events_received: 0,
            events_filtered: 0,
            index_tasks_queued: 0,
            commit_reconciliations: 0,
            commit_tasks_queued: 0,
            commit_reconcile_failures: 0,
        }
    }
}

enum WatcherCommand {
    Add {
        repository: WatchedRepository,
        response: oneshot::Sender<bool>,
    },
    Remove {
        alias_or_id: String,
        response: oneshot::Sender<bool>,
    },
}

struct WatcherLoopContext {
    state: Arc<RwLock<WatcherInternalState>>,
    diagnostics_tx: watch::Sender<WatcherDiagnostics>,
    dropped_events: Arc<AtomicU64>,
    debounce: Duration,
    commit_reconcile_interval: Duration,
    max_watch_dirs: usize,
    task_sink: TaskQueueSink,
}

fn disabled_handle(hash_cache_capacity: usize) -> WatcherHandle {
    let (_diagnostics_tx, diagnostics_rx) = watch::channel(WatcherDiagnostics {
        state: WatcherState::Disabled,
        ..WatcherDiagnostics::default()
    });
    let (shutdown_tx, _) = watch::channel(false);
    WatcherHandle {
        diagnostics: diagnostics_rx,
        shutdown: shutdown_tx,
        state: Arc::new(RwLock::new(WatcherInternalState::new(hash_cache_capacity))),
        command_tx: None,
    }
}

fn boxed_task_sink<F, Fut>(task_sink: F) -> TaskQueueSink
where
    F: Fn(crate::storage::CodeIndexTaskSeed) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = Result<(), String>> + Send + 'static,
{
    Arc::new(move |task| Box::pin(task_sink(task)))
}

#[cfg(test)]
use self::{
    diagnostics::emit as emit_diagnostics,
    index_queue::{process_debounced_paths, reconcile_all_commit_heads, should_process_path},
};

#[cfg(test)]
#[path = "integration_tests.rs"]
mod tests;