fdu-core 0.3.0

//! Blocking opened-root polling over the index's one exact commit history.

use std::collections::BTreeSet;
use std::sync::{Condvar, Mutex};
use std::time::Instant;

use super::OpenedIndex;
use crate::{
    ChangeOutcome, ChangePoll, ChangeRequest, EngineVersion, Error, Impact, ImpactDomain,
    IndexState, Result, Since, Work,
};

pub(super) struct JournalWait {
    state: Mutex<WaitState>,
    changed: Condvar,
}

#[derive(Default)]
struct WaitState {
    closed: bool,
    /// Polls blocked in the wait right now, so a session golden can observe that none
    /// outlives close instead of asserting it.
    #[cfg(all(test, feature = "watch"))]
    waiters: usize,
}

impl JournalWait {
    pub(super) fn new() -> Self {
        Self { state: Mutex::new(WaitState::default()), changed: Condvar::new() }
    }

    /// Wake every poller after an exact commit has entered the index journal.
    pub(super) fn notify_commit(&self) {
        self.wake();
    }

    /// Wake every poller so it re-examines the root: a worker has failed, and a poller
    /// waiting for its commits must answer for that now rather than at its timeout.
    pub(super) fn wake(&self) {
        let guard = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        self.changed.notify_all();
        drop(guard);
    }

    /// Wake every poller permanently during joined owner shutdown.
    pub(super) fn close(&self) {
        let mut state = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        state.closed = true;
        self.changed.notify_all();
    }

    #[cfg(all(test, feature = "watch"))]
    pub(super) fn waiters(&self) -> usize {
        self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner).waiters
    }
}

struct JournalSnapshot {
    version: EngineVersion,
    state: IndexState,
    since: Since,
}

pub(super) fn poll(opened: &OpenedIndex, request: ChangeRequest) -> Result<ChangePoll> {
    let started = Instant::now();
    let mut wait = opened.state.journal.state.lock().map_err(|_| Error::OpenedJournalPoisoned)?;

    loop {
        if wait.closed || opened.state.cancellation.is_cancelled() {
            return Err(Error::OpenedIndexClosed);
        }

        let read = opened.state.index.read_with(|index| {
            let scope = index.scope();
            let version = EngineVersion {
                session: opened.state.session,
                sequence: index.clock(),
                scope: scope.entry_scope(),
                semantics: scope.semantic_identity(),
            };
            validate_cursor(request.after, version)?;
            let since = index.since(request.after.sequence);
            debug_assert_eq!(since.clock, version.sequence);
            debug_assert_eq!(since.state, index.state());
            Ok(JournalSnapshot { version, state: since.state, since })
        });
        let snapshot = match read {
            Ok(snapshot) => snapshot?,
            Err(Error::IndexLockPoisoned) => {
                // A worker that panics inside a commit, the likeliest place for a panic,
                // poisons the index lock as it unwinds. Once the panic is recorded it is
                // the cause close reports, and the poisoning is only its trace.
                return Err(opened
                    .state
                    .failures
                    .panicked()
                    .map_or(Error::IndexLockPoisoned, |worker| Error::OpenedWorkerPanicked {
                        worker,
                    }));
            }
            Err(error) => return Err(error),
        };

        if opened.state.cancellation.is_cancelled() {
            return Err(Error::OpenedIndexClosed);
        }

        if snapshot.since.truncated {
            return Ok(reset_at(&snapshot));
        }
        if !snapshot.since.commits.is_empty() {
            return Ok(changes_at(snapshot));
        }
        if let Some(worker) = opened.state.failures.panicked() {
            // Nothing will commit again; waiting would end only at the timeout. This is
            // the cause close reports, checked here rather than first so that commits
            // retained before the panic are still delivered.
            return Err(Error::OpenedWorkerPanicked { worker });
        }

        let remaining = request.timeout.saturating_sub(started.elapsed());
        if remaining.is_zero() {
            return Ok(idle_at(request.after, &snapshot));
        }
        #[cfg(test)]
        opened.state.test_controls.reach(super::TestPoint::BeforeJournalWait);
        #[cfg(all(test, feature = "watch"))]
        {
            wait.waiters += 1;
        }
        let (next, _) = opened
            .state
            .journal
            .changed
            .wait_timeout(wait, remaining)
            .map_err(|_| Error::OpenedJournalPoisoned)?;
        wait = next;
        #[cfg(all(test, feature = "watch"))]
        {
            wait.waiters -= 1;
        }
    }
}

fn validate_cursor(requested: EngineVersion, current: EngineVersion) -> Result<()> {
    if requested.session != current.session
        || requested.scope != current.scope
        || requested.semantics != current.semantics
        || requested.sequence > current.sequence
    {
        return Err(Error::ChangeCursorUnavailable {
            requested: Box::new(requested),
            current: Box::new(current),
        });
    }
    Ok(())
}

fn changes_at(snapshot: JournalSnapshot) -> ChangePoll {
    let count = u64::try_from(snapshot.since.commits.len()).unwrap_or(u64::MAX);
    let impact = combined_impact(&snapshot.since.commits);
    ChangePoll {
        cursor: snapshot.version,
        version: snapshot.version,
        state: snapshot.state,
        outcome: ChangeOutcome::Changes { commits: snapshot.since.commits, impact },
        work: Work { commits_visited: count, commits_returned: count, ..Work::default() },
    }
}

fn idle_at(after: EngineVersion, snapshot: &JournalSnapshot) -> ChangePoll {
    debug_assert_eq!(after.sequence, snapshot.version.sequence);
    ChangePoll {
        cursor: after,
        version: snapshot.version,
        state: snapshot.state,
        outcome: ChangeOutcome::Idle,
        work: Work::default(),
    }
}

fn reset_at(snapshot: &JournalSnapshot) -> ChangePoll {
    let visited = u64::try_from(snapshot.since.commits.len()).unwrap_or(u64::MAX);
    ChangePoll {
        cursor: snapshot.version,
        version: snapshot.version,
        state: snapshot.state,
        outcome: ChangeOutcome::Reset { impact: reset_impact() },
        work: Work { commits_visited: visited, ..Work::default() },
    }
}

pub(super) fn interval_impact(since: &Since) -> Impact {
    if since.truncated { reset_impact() } else { combined_impact(&since.commits) }
}

fn reset_impact() -> Impact {
    Impact {
        domains: vec![
            ImpactDomain::Topology,
            ImpactDomain::Metadata,
            ImpactDomain::Classification,
            ImpactDomain::Aggregates,
            ImpactDomain::Content,
            ImpactDomain::State,
        ],
        dirty_paths: Vec::new(),
        all_dirty: true,
    }
}

fn combined_impact(commits: &[crate::Commit]) -> Impact {
    let mut domains = BTreeSet::new();
    let mut paths = BTreeSet::new();
    let mut all_dirty = false;
    for commit in commits {
        domains.extend(commit.impact.domains.iter().copied());
        if all_dirty || commit.impact.all_dirty {
            all_dirty = true;
            paths.clear();
            continue;
        }
        for path in &commit.impact.dirty_paths {
            paths.insert(path.clone());
            if paths.len() > crate::MAX_DIRTY_PATHS {
                all_dirty = true;
                paths.clear();
                break;
            }
        }
    }
    Impact {
        domains: domains.into_iter().collect(),
        dirty_paths: paths.into_iter().collect(),
        all_dirty,
    }
}