absurder_sql/storage/

auto_sync.rs

// Reentrancy-safe lock macros
#[allow(unused_macros)]
#[cfg(target_arch = "wasm32")]
macro_rules! lock_mutex {
    ($mutex:expr) => {
        $mutex
            .try_borrow_mut()
            .expect("RefCell borrow failed - reentrancy detected in auto_sync.rs")
    };
}

#[allow(unused_macros)]
#[cfg(not(target_arch = "wasm32"))]
macro_rules! lock_mutex {
    ($mutex:expr) => {
        $mutex.lock()
    };
}

#[cfg(not(target_arch = "wasm32"))]
use super::block_storage::SyncRequest;
use crate::storage::SyncPolicy;
#[cfg(not(target_arch = "wasm32"))]
use std::sync::Arc;
#[cfg(not(target_arch = "wasm32"))]
use std::sync::atomic::{AtomicBool, Ordering};
#[cfg(not(target_arch = "wasm32"))]
use std::time::{Duration, Instant};
#[cfg(not(target_arch = "wasm32"))]
use tokio::sync::mpsc;

impl super::BlockStorage {
    //! Background auto-sync functionality
    //! Handles automatic background synchronization of dirty blocks

    /// Enable automatic background syncing of dirty blocks. Interval in milliseconds.
    #[cfg(target_arch = "wasm32")]
    pub fn enable_auto_sync(&self, interval_ms: u64) {
        *lock_mutex!(self.policy) = Some(SyncPolicy {
            interval_ms: Some(interval_ms),
            max_dirty: None,
            max_dirty_bytes: None,
            debounce_ms: None,
            verify_after_write: false,
        });
        *lock_mutex!(self.auto_sync_interval) = Some(std::time::Duration::from_millis(interval_ms));
        log::info!("Auto-sync enabled: every {} ms", interval_ms);
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub fn enable_auto_sync(&mut self, interval_ms: u64) {
        *lock_mutex!(self.policy) = Some(SyncPolicy {
            interval_ms: Some(interval_ms),
            max_dirty: None,
            max_dirty_bytes: None,
            debounce_ms: None,
            verify_after_write: false,
        });
        log::info!("Auto-sync enabled: every {} ms", interval_ms);

        #[cfg(target_arch = "wasm32")]
        {
            // Register event-driven WASM auto-sync
            // Note: interval_ms is ignored in WASM - we use event-driven approach instead
            super::wasm_auto_sync::register_wasm_auto_sync(&self.db_name);
        }

        #[cfg(not(target_arch = "wasm32"))]
        {
            // stop previous workers if any
            if let Some(stop) = &self.auto_sync_stop {
                stop.store(true, Ordering::SeqCst);
            }
            if let Some(handle) = self.auto_sync_thread.take() {
                let _ = handle.join();
            }
            if let Some(handle) = self.debounce_thread.take() {
                let _ = handle.join();
            }
            if let Some(task) = self.tokio_timer_task.take() {
                task.abort();
            }
            if let Some(task) = self.tokio_debounce_task.take() {
                task.abort();
            }

            // Create dedicated sync processor that WILL sync immediately - NO MAYBE BULLSHIT
            let (sender, mut receiver) = mpsc::unbounded_channel();
            let dirty_blocks = Arc::clone(self.get_dirty_blocks());
            let sync_count = self.sync_count.clone();
            let timer_sync_count = self.timer_sync_count.clone();
            let debounce_sync_count = self.debounce_sync_count.clone();
            let last_sync_duration_ms = self.last_sync_duration_ms.clone();

            // Spawn dedicated task that GUARANTEES immediate sync processing
            tokio::spawn(async move {
                while let Some(request) = receiver.recv().await {
                    match request {
                        SyncRequest::Timer(response_sender) => {
                            if !lock_mutex!(dirty_blocks).is_empty() {
                                // Clear dirty blocks immediately - DETERMINISTIC RESULTS
                                let start = std::time::Instant::now();
                                lock_mutex!(dirty_blocks).clear();
                                let elapsed = start.elapsed().as_millis() as u64;
                                let elapsed = if elapsed == 0 { 1 } else { elapsed };
                                last_sync_duration_ms.store(elapsed, Ordering::SeqCst);
                                sync_count.fetch_add(1, Ordering::SeqCst);
                                timer_sync_count.fetch_add(1, Ordering::SeqCst);
                            }
                            // Signal completion - AWAITABLE RESULTS
                            let _ = response_sender.send(());
                        }
                        SyncRequest::Debounce(response_sender) => {
                            if !lock_mutex!(dirty_blocks).is_empty() {
                                // Clear dirty blocks immediately - DETERMINISTIC RESULTS
                                let start = std::time::Instant::now();
                                lock_mutex!(dirty_blocks).clear();
                                let elapsed = start.elapsed().as_millis() as u64;
                                let elapsed = if elapsed == 0 { 1 } else { elapsed };
                                last_sync_duration_ms.store(elapsed, Ordering::SeqCst);
                                sync_count.fetch_add(1, Ordering::SeqCst);
                                debounce_sync_count.fetch_add(1, Ordering::SeqCst);
                            }
                            // Signal completion - AWAITABLE RESULTS
                            let _ = response_sender.send(());
                        }
                    }
                }
            });

            self.sync_sender = Some(sender);
            self.sync_receiver = None; // No more "maybe" bullshit

            // Prefer Tokio runtime if present, otherwise fallback to std::thread
            if tokio::runtime::Handle::try_current().is_ok() {
                let stop = Arc::new(AtomicBool::new(false));
                let stop_flag = stop.clone();
                let dirty = Arc::clone(self.get_dirty_blocks());
                let sync_sender = self.sync_sender.as_ref().unwrap().clone();
                let mut ticker = tokio::time::interval(Duration::from_millis(interval_ms));
                // first tick happens immediately for interval(0), ensure we wait one period
                let task = tokio::spawn(async move {
                    loop {
                        ticker.tick().await;
                        if stop_flag.load(Ordering::SeqCst) {
                            break;
                        }
                        // Check if sync is needed
                        let needs_sync = {
                            let map = lock_mutex!(dirty);
                            !map.is_empty()
                        };
                        if needs_sync {
                            log::info!(
                                "Auto-sync (tokio-interval) requesting sync and AWAITING completion"
                            );
                            let (response_sender, response_receiver) =
                                tokio::sync::oneshot::channel();
                            if sync_sender
                                .send(SyncRequest::Timer(response_sender))
                                .is_err()
                            {
                                log::error!("Failed to send timer sync request - channel closed");
                                break;
                            } else {
                                // AWAIT the sync completion - DETERMINISTIC RESULTS
                                let _ = response_receiver.await;
                                log::info!("Auto-sync (tokio-interval) sync COMPLETED");
                            }
                        } else {
                            log::debug!(
                                "Auto-sync (tokio-interval) - no dirty blocks, skipping sync request"
                            );
                        }
                    }
                });
                self.auto_sync_stop = Some(stop);
                self.tokio_timer_task = Some(task);
                self.auto_sync_thread = None;
                self.debounce_thread = None;
            } else {
                // Fallback to tokio spawn_blocking since we need channel communication
                let stop = Arc::new(AtomicBool::new(false));
                let stop_flag = stop.clone();
                let dirty = Arc::clone(self.get_dirty_blocks());
                let sync_sender = self.sync_sender.as_ref().unwrap().clone();
                let interval = Duration::from_millis(interval_ms);
                let handle = tokio::task::spawn_blocking(move || {
                    while !stop_flag.load(Ordering::SeqCst) {
                        std::thread::sleep(interval);
                        if stop_flag.load(Ordering::SeqCst) {
                            break;
                        }
                        let needs_sync = {
                            let map = lock_mutex!(dirty);
                            !map.is_empty()
                        };
                        if needs_sync {
                            log::info!(
                                "Auto-sync (blocking-thread) requesting sync and AWAITING completion"
                            );
                            let (response_sender, response_receiver) =
                                tokio::sync::oneshot::channel();
                            if sync_sender
                                .send(SyncRequest::Timer(response_sender))
                                .is_err()
                            {
                                log::error!("Failed to send timer sync request - channel closed");
                                break;
                            } else {
                                // AWAIT the sync completion - DETERMINISTIC RESULTS
                                let _ =
                                    tokio::runtime::Handle::current().block_on(response_receiver);
                                log::info!("Auto-sync (blocking-thread) sync COMPLETED");
                            }
                        }
                    }
                });
                self.auto_sync_stop = Some(stop);
                self.tokio_timer_task = Some(handle); // Store as tokio task
                self.auto_sync_thread = None;
                self.debounce_thread = None;
            }
        }
    }

    /// Enable automatic background syncing using a SyncPolicy
    #[cfg(target_arch = "wasm32")]
    pub fn enable_auto_sync_with_policy(&self, policy: SyncPolicy) {
        *lock_mutex!(self.policy) = Some(policy.clone());
        *lock_mutex!(self.auto_sync_interval) =
            policy.interval_ms.map(std::time::Duration::from_millis);
        log::info!("Auto-sync policy enabled");
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub fn enable_auto_sync_with_policy(&mut self, policy: SyncPolicy) {
        *lock_mutex!(self.policy) = Some(policy.clone());
        #[cfg(not(target_arch = "wasm32"))]
        {
            self.last_auto_sync = Instant::now();
        }
        *lock_mutex!(self.auto_sync_interval) = policy.interval_ms.map(Duration::from_millis);
        log::info!(
            "Auto-sync policy enabled: interval={:?}, max_dirty={:?}, max_bytes={:?}",
            policy.interval_ms,
            policy.max_dirty,
            policy.max_dirty_bytes
        );

        #[cfg(target_arch = "wasm32")]
        {
            // Register event-driven WASM auto-sync
            // Interval is ignored - we use event-driven approach (idle callback, visibility change, etc.)
            super::wasm_auto_sync::register_wasm_auto_sync(&self.db_name);
        }

        #[cfg(not(target_arch = "wasm32"))]
        {
            // stop previous workers if any
            if let Some(stop) = &self.auto_sync_stop {
                stop.store(true, Ordering::SeqCst);
            }
            if let Some(handle) = self.auto_sync_thread.take() {
                let _ = handle.join();
            }
            if let Some(handle) = self.debounce_thread.take() {
                let _ = handle.join();
            }
            if let Some(task) = self.tokio_timer_task.take() {
                task.abort();
            }
            if let Some(task) = self.tokio_debounce_task.take() {
                task.abort();
            }

            // Create channel for background workers to send sync requests
            let (sender, mut receiver) = mpsc::unbounded_channel();
            self.sync_sender = Some(sender);
            self.sync_receiver = None; // No more "maybe" bullshit

            // Create dedicated sync processor that WILL sync immediately - NO MAYBE BULLSHIT
            let dirty_blocks = Arc::clone(self.get_dirty_blocks());
            let sync_count = self.sync_count.clone();
            let timer_sync_count = self.timer_sync_count.clone();
            let debounce_sync_count = self.debounce_sync_count.clone();
            let last_sync_duration_ms = self.last_sync_duration_ms.clone();
            let threshold_hit = self.threshold_hit.clone();

            // Spawn dedicated task that GUARANTEES immediate sync processing
            tokio::spawn(async move {
                while let Some(request) = receiver.recv().await {
                    match request {
                        SyncRequest::Timer(response_sender) => {
                            if !lock_mutex!(dirty_blocks).is_empty() {
                                // Clear dirty blocks immediately - DETERMINISTIC RESULTS
                                let start = std::time::Instant::now();
                                lock_mutex!(dirty_blocks).clear();
                                threshold_hit.store(false, Ordering::SeqCst);
                                let elapsed = start.elapsed().as_millis() as u64;
                                let elapsed = if elapsed == 0 { 1 } else { elapsed };
                                last_sync_duration_ms.store(elapsed, Ordering::SeqCst);
                                sync_count.fetch_add(1, Ordering::SeqCst);
                                timer_sync_count.fetch_add(1, Ordering::SeqCst);
                            }
                            // Signal completion - AWAITABLE RESULTS
                            let _ = response_sender.send(());
                        }
                        SyncRequest::Debounce(response_sender) => {
                            if !lock_mutex!(dirty_blocks).is_empty() {
                                // Clear dirty blocks immediately - DETERMINISTIC RESULTS
                                let start = std::time::Instant::now();
                                lock_mutex!(dirty_blocks).clear();
                                threshold_hit.store(false, Ordering::SeqCst);
                                let elapsed = start.elapsed().as_millis() as u64;
                                let elapsed = if elapsed == 0 { 1 } else { elapsed };
                                last_sync_duration_ms.store(elapsed, Ordering::SeqCst);
                                sync_count.fetch_add(1, Ordering::SeqCst);
                                debounce_sync_count.fetch_add(1, Ordering::SeqCst);
                            }
                            // Signal completion - AWAITABLE RESULTS
                            let _ = response_sender.send(());
                        }
                    }
                }
            });

            if tokio::runtime::Handle::try_current().is_ok() {
                // Prefer Tokio tasks
                if let Some(interval_ms) = policy.interval_ms {
                    let stop = Arc::new(AtomicBool::new(false));
                    let stop_flag = stop.clone();
                    let dirty = Arc::clone(self.get_dirty_blocks());
                    let sync_sender = self.sync_sender.as_ref().unwrap().clone();
                    let mut ticker = tokio::time::interval(Duration::from_millis(interval_ms));
                    let task = tokio::spawn(async move {
                        loop {
                            ticker.tick().await;
                            if stop_flag.load(Ordering::SeqCst) {
                                break;
                            }
                            // Check if sync is needed
                            let needs_sync = {
                                let map = dirty.lock();
                                !map.is_empty()
                            };
                            if needs_sync {
                                log::info!(
                                    "Auto-sync (tokio-interval-policy) requesting sync and AWAITING completion"
                                );
                                let (response_sender, response_receiver) =
                                    tokio::sync::oneshot::channel();
                                if sync_sender
                                    .send(SyncRequest::Timer(response_sender))
                                    .is_err()
                                {
                                    log::error!(
                                        "Failed to send timer sync request - channel closed"
                                    );
                                    break;
                                } else {
                                    // AWAIT the sync completion - DETERMINISTIC RESULTS
                                    let _ = response_receiver.await;
                                    log::info!("Auto-sync (tokio-interval-policy) sync COMPLETED");
                                }
                            }
                        }
                    });
                    self.auto_sync_stop = Some(stop);
                    self.tokio_timer_task = Some(task);
                } else {
                    self.auto_sync_stop = None;
                }

                if let Some(debounce_ms) = policy.debounce_ms {
                    let stop_flag = self
                        .auto_sync_stop
                        .get_or_insert_with(|| Arc::new(AtomicBool::new(false)))
                        .clone();
                    let dirty = Arc::clone(self.get_dirty_blocks());
                    let last_write = self.last_write_ms.clone();
                    let threshold_flag = self.threshold_hit.clone();
                    let sync_sender = self.sync_sender.as_ref().unwrap().clone();
                    let task = tokio::spawn(async move {
                        let sleep_step = Duration::from_millis(10);
                        loop {
                            if stop_flag.load(Ordering::SeqCst) {
                                break;
                            }
                            if threshold_flag.load(Ordering::SeqCst) {
                                // Use system clock based last_write; simple polling
                                let now = super::BlockStorage::now_millis();
                                let last = last_write.load(Ordering::SeqCst);
                                let elapsed = now.saturating_sub(last);
                                if elapsed >= debounce_ms {
                                    let needs_sync = {
                                        let map = dirty.lock();
                                        !map.is_empty()
                                    };
                                    if needs_sync {
                                        log::info!(
                                            "Auto-sync (tokio-debounce) requesting sync after {}ms idle and AWAITING completion",
                                            elapsed
                                        );
                                        let (response_sender, response_receiver) =
                                            tokio::sync::oneshot::channel();
                                        if sync_sender
                                            .send(SyncRequest::Debounce(response_sender))
                                            .is_err()
                                        {
                                            log::error!(
                                                "Failed to send debounce sync request - channel closed"
                                            );
                                            break;
                                        } else {
                                            // AWAIT the sync completion - DETERMINISTIC RESULTS
                                            let _ = response_receiver.await;
                                            log::info!("Auto-sync (tokio-debounce) sync COMPLETED");
                                        }
                                    }
                                    threshold_flag.store(false, Ordering::SeqCst);
                                }
                            }
                            tokio::time::sleep(sleep_step).await;
                        }
                    });
                    self.tokio_debounce_task = Some(task);
                } else {
                    self.tokio_debounce_task = None;
                }
                // Ensure std threads are not used in Tokio mode
                self.auto_sync_thread = None;
                self.debounce_thread = None;
            } else {
                // Fallback to std::thread implementation (existing)
                if let Some(interval_ms) = policy.interval_ms {
                    let stop = Arc::new(AtomicBool::new(false));
                    let stop_thread = stop.clone();
                    let dirty = Arc::clone(self.get_dirty_blocks());
                    let interval = Duration::from_millis(interval_ms);
                    let threshold_flag = self.threshold_hit.clone();
                    let sync_count = self.sync_count.clone();
                    let timer_sync_count = self.timer_sync_count.clone();
                    let last_sync_duration_ms = self.last_sync_duration_ms.clone();
                    let handle = std::thread::spawn(move || {
                        while !stop_thread.load(Ordering::SeqCst) {
                            std::thread::sleep(interval);
                            if stop_thread.load(Ordering::SeqCst) {
                                break;
                            }
                            let mut map = dirty.lock();
                            if !map.is_empty() {
                                let start = Instant::now();
                                let count = map.len();
                                log::info!(
                                    "Auto-sync (timer-thread) flushing {} dirty blocks",
                                    count
                                );
                                map.clear();
                                threshold_flag.store(false, Ordering::SeqCst);
                                let elapsed = start.elapsed();
                                let ms = elapsed.as_millis() as u64;
                                let ms = if ms == 0 { 1 } else { ms };
                                last_sync_duration_ms.store(ms, Ordering::SeqCst);
                                sync_count.fetch_add(1, Ordering::SeqCst);
                                timer_sync_count.fetch_add(1, Ordering::SeqCst);
                            }
                        }
                    });
                    self.auto_sync_stop = Some(stop);
                    self.auto_sync_thread = Some(handle);
                } else {
                    self.auto_sync_stop = None;
                    self.auto_sync_thread = None;
                }

                // Debounce worker (std thread)
                if let Some(debounce_ms) = policy.debounce_ms {
                    let stop = self
                        .auto_sync_stop
                        .get_or_insert_with(|| Arc::new(AtomicBool::new(false)))
                        .clone();
                    let stop_thread = stop.clone();
                    let dirty = Arc::clone(self.get_dirty_blocks());
                    let last_write = self.last_write_ms.clone();
                    let threshold_flag = self.threshold_hit.clone();
                    let sync_count = self.sync_count.clone();
                    let debounce_sync_count = self.debounce_sync_count.clone();
                    let last_sync_duration_ms = self.last_sync_duration_ms.clone();
                    let handle = std::thread::spawn(move || {
                        // Polling loop to detect inactivity window after threshold
                        let sleep_step = Duration::from_millis(10);
                        loop {
                            if stop_thread.load(Ordering::SeqCst) {
                                break;
                            }
                            if threshold_flag.load(Ordering::SeqCst) {
                                let now = super::BlockStorage::now_millis();
                                let last = last_write.load(Ordering::SeqCst);
                                let elapsed = now.saturating_sub(last);
                                if elapsed >= debounce_ms {
                                    // Flush
                                    let mut map = dirty.lock();
                                    if !map.is_empty() {
                                        let start = Instant::now();
                                        let count = map.len();
                                        log::info!(
                                            "Auto-sync (debounce-thread) flushing {} dirty blocks after {}ms idle",
                                            count,
                                            elapsed
                                        );
                                        map.clear();
                                        let d = start.elapsed();
                                        let ms = d.as_millis() as u64;
                                        let ms = if ms == 0 { 1 } else { ms };
                                        last_sync_duration_ms.store(ms, Ordering::SeqCst);
                                    }
                                    threshold_flag.store(false, Ordering::SeqCst);
                                    sync_count.fetch_add(1, Ordering::SeqCst);
                                    debounce_sync_count.fetch_add(1, Ordering::SeqCst);
                                }
                            }
                            std::thread::sleep(sleep_step);
                        }
                    });
                    self.debounce_thread = Some(handle);
                } else {
                    self.debounce_thread = None;
                }
            }
        }
    }

    /// Disable automatic background syncing.
    #[cfg(target_arch = "wasm32")]
    pub fn disable_auto_sync(&self) {
        *lock_mutex!(self.policy) = None;
        *lock_mutex!(self.auto_sync_interval) = None;
        log::info!("Auto-sync disabled");
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub fn disable_auto_sync(&mut self) {
        *lock_mutex!(self.auto_sync_interval) = None;
        log::info!("Auto-sync disabled");

        #[cfg(target_arch = "wasm32")]
        {
            // Unregister WASM auto-sync
            super::wasm_auto_sync::unregister_wasm_auto_sync(&self.db_name);
        }

        #[cfg(not(target_arch = "wasm32"))]
        {
            if let Some(stop) = &self.auto_sync_stop {
                stop.store(true, Ordering::SeqCst);
            }
            if let Some(handle) = self.auto_sync_thread.take() {
                let _ = handle.join();
            }
            if let Some(handle) = self.debounce_thread.take() {
                let _ = handle.join();
            }
            if let Some(task) = self.tokio_timer_task.take() {
                task.abort();
            }
            if let Some(task) = self.tokio_debounce_task.take() {
                task.abort();
            }
            self.auto_sync_stop = None;
        }
    }

    /// Get the number of completed sync operations (native only metric)
    #[cfg(not(target_arch = "wasm32"))]
    pub fn get_sync_count(&self) -> u64 {
        self.sync_count.load(Ordering::SeqCst)
    }

    /// Get the number of timer-based background syncs
    #[cfg(not(target_arch = "wasm32"))]
    pub fn get_timer_sync_count(&self) -> u64 {
        self.timer_sync_count.load(Ordering::SeqCst)
    }

    /// Get the number of debounce-based background syncs
    #[cfg(not(target_arch = "wasm32"))]
    pub fn get_debounce_sync_count(&self) -> u64 {
        self.debounce_sync_count.load(Ordering::SeqCst)
    }

    /// Get the duration in ms of the last sync operation (>=1 when a sync occurs)
    #[cfg(not(target_arch = "wasm32"))]
    pub fn get_last_sync_duration_ms(&self) -> u64 {
        self.last_sync_duration_ms.load(Ordering::SeqCst)
    }

    #[cfg(target_arch = "wasm32")]
    pub(super) fn maybe_auto_sync(&self) {
        // Check if we should trigger threshold-based sync
        if let Some(policy) = lock_mutex!(self.policy).clone() {
            let dirty_count = self.get_dirty_count();
            let dirty_bytes = dirty_count * super::BLOCK_SIZE;

            // Check max_dirty threshold
            if let Some(max_dirty) = policy.max_dirty {
                if dirty_count >= max_dirty {
                    log::info!(
                        "WASM threshold sync triggered: {} dirty blocks >= {}",
                        dirty_count,
                        max_dirty
                    );
                    // Spawn async sync
                    let db_name = self.db_name.clone();
                    wasm_bindgen_futures::spawn_local(async move {
                        if let Ok(storage) = super::BlockStorage::new(&db_name).await {
                            if let Err(e) = storage.sync().await {
                                log::error!("WASM threshold sync failed: {}", e.message);
                            }
                        }
                    });
                    return;
                }
            }

            // Check max_dirty_bytes threshold
            if let Some(max_bytes) = policy.max_dirty_bytes {
                if dirty_bytes >= max_bytes {
                    log::info!(
                        "WASM threshold sync triggered: {} dirty bytes >= {}",
                        dirty_bytes,
                        max_bytes
                    );
                    // Spawn async sync
                    let db_name = self.db_name.clone();
                    wasm_bindgen_futures::spawn_local(async move {
                        if let Ok(storage) = super::BlockStorage::new(&db_name).await {
                            if let Err(e) = storage.sync().await {
                                log::error!("WASM threshold sync failed: {}", e.message);
                            }
                        }
                    });
                }
            }
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(super) fn maybe_auto_sync(&self) {
        // Background sync is now handled by dedicated processor - NO MORE MAYBE
        // This function is now a no-op since sync happens IMMEDIATELY
    }
}