ic-query 0.91.0

//! Module: sns::report::cache_attempt
//!
//! Responsibility: shared SNS refresh context and attempt-sidecar lifecycle.
//! Does not own: family page fetching, cache publication, or text rendering.
//! Boundary: one resolved context and attempt contract serves neuron refreshes.

use crate::cache_file::{ManagedReadBudget, write_managed_json_pretty_atomically};
use crate::report::CollectionSourceRequest;
use crate::{
    HostCacheError,
    cache::CacheRefreshAttemptStatus,
    snapshot_cache::{
        PagedCollectionPage, PagedCollectionState, SNAPSHOT_REFRESH_ATTEMPT_SCHEMA_VERSION,
        SnapshotRefreshAttempt, SnapshotRefreshProgress, current_attempt_timestamp,
        read_snapshot_refresh_attempt_strict, validate_snapshot_refresh_attempt,
    },
    sns::report::{
        SNS_CACHE_COMPONENT, SnsHostError, SnsRefreshAttemptStatus, SnsSnapshotRefreshRequest,
        cache_paths::sns_governance_from_cache_path,
    },
};
use serde::{Deserialize as SerdeDeserialize, Serialize};
use std::path::Path;

///
/// SnsRefreshAttemptMetadata
///
/// Snapshot refresh-attempt metadata shared by SNS neuron cache
/// refresh sidecars.
///

#[derive(Clone, Debug, Eq, PartialEq, SerdeDeserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub(in crate::sns::report) struct SnsRefreshAttemptMetadata {
    pub(in crate::sns::report) governance_canister_id: String,
}

pub(in crate::sns::report) type SnsRefreshAttempt =
    SnapshotRefreshAttempt<SnsRefreshAttemptMetadata>;

///
/// SnsRefreshContext
///
/// Resolved inputs and paging policy shared by one SNS collection refresh.
///

#[derive(Clone, Copy)]
pub(in crate::sns::report) struct SnsRefreshContext<'a> {
    pub(in crate::sns::report) path: &'a Path,
    pub(in crate::sns::report) request: &'a SnsSnapshotRefreshRequest,
    pub(in crate::sns::report) fetch_request: &'a CollectionSourceRequest,
    pub(in crate::sns::report) governance_canister_id: &'a str,
}

impl SnsRefreshContext<'_> {
    pub(in crate::sns::report) fn max_pages_reached(self, page_count: u32) -> bool {
        self.request
            .max_pages
            .is_some_and(|max_pages| page_count >= max_pages)
    }

    pub(in crate::sns::report) fn incomplete_refresh_error(
        page_count: u32,
        row_count: usize,
        reason: &'static str,
    ) -> SnsHostError {
        SnsHostError::IncompleteRefresh {
            pages_fetched: page_count,
            rows_fetched: row_count,
            reason: reason.to_string(),
        }
    }

    pub(in crate::sns::report) fn ingest_page<Row, Cursor>(
        self,
        pages: &mut PagedCollectionState<Row, Cursor>,
        rows: Vec<Row>,
        raw_page_len: usize,
        next_cursor: Option<Cursor>,
    ) -> PagedCollectionPage {
        // The API page length precedes removal of a repeated neuron boundary row.
        let exhausted = raw_page_len
            < usize::try_from(self.request.page_size).unwrap_or(usize::MAX)
            || next_cursor.is_none();
        let page = PagedCollectionPage::new(rows.len(), exhausted);
        pages.ingest(rows, next_cursor);
        page
    }
}

pub(in crate::sns::report) fn write_starting_sns_refresh_attempt(
    context: SnsRefreshContext<'_>,
) -> Result<(), SnsHostError> {
    write_sns_refresh_attempt_status(
        context,
        CacheRefreshAttemptStatus::Running,
        SnapshotRefreshProgress::default(),
        None,
    )
}

/// Write the running attempt evidence produced by one retained SNS page.
pub(in crate::sns::report) fn write_running_sns_refresh_page(
    context: SnsRefreshContext<'_>,
    page_count: u32,
    row_count: usize,
    last_cursor: Option<String>,
) -> Result<(), SnsHostError> {
    write_sns_refresh_attempt_status(
        context,
        CacheRefreshAttemptStatus::Running,
        SnapshotRefreshProgress::new(page_count, row_count, last_cursor),
        None,
    )
}

pub(in crate::sns::report) fn write_complete_sns_refresh_attempt(
    context: SnsRefreshContext<'_>,
    progress: SnapshotRefreshProgress,
) -> Result<(), SnsHostError> {
    write_sns_refresh_attempt_status(context, CacheRefreshAttemptStatus::Complete, progress, None)
}

pub(in crate::sns::report) fn write_failed_sns_refresh_attempt(
    context: SnsRefreshContext<'_>,
    error: &SnsHostError,
) {
    let latest = read_sns_refresh_attempt(
        &context.request.cache_root,
        context.path,
        &context.request.target.network,
        None,
    )
    .ok()
    .flatten();
    let progress = latest.map_or_else(
        SnapshotRefreshProgress::default,
        SnapshotRefreshProgress::from,
    );
    let _ = write_sns_refresh_attempt_status(
        context,
        CacheRefreshAttemptStatus::Failed,
        progress,
        Some(error.to_string()),
    );
}

fn write_sns_refresh_attempt_status(
    context: SnsRefreshContext<'_>,
    status: CacheRefreshAttemptStatus,
    progress: SnapshotRefreshProgress,
    last_error: Option<String>,
) -> Result<(), SnsHostError> {
    let attempt = SnsRefreshAttempt {
        schema_version: SNAPSHOT_REFRESH_ATTEMPT_SCHEMA_VERSION,
        network: context.request.target.network.clone(),
        source_endpoint: context.request.target.source_endpoint.clone(),
        started_at: context.fetch_request.fetched_at.clone(),
        updated_at: current_attempt_timestamp(&context.fetch_request.fetched_at),
        metadata: SnsRefreshAttemptMetadata {
            governance_canister_id: context.governance_canister_id.to_string(),
        },
        status,
        page_size: context.request.page_size,
        pages_fetched: progress.pages_fetched,
        rows_fetched: progress.rows_fetched,
        last_cursor: progress.last_cursor,
        last_error,
    };
    write_managed_json_pretty_atomically(
        &context.request.cache_root,
        context.path,
        &attempt,
        crate::cache_file::MAX_REFRESH_ATTEMPT_BYTES,
        |path, source| {
            SnsHostError::from(HostCacheError::serialize_cache(
                SNS_CACHE_COMPONENT,
                path,
                source,
            ))
        },
        |error| SnsHostError::from(HostCacheError::operation(SNS_CACHE_COMPONENT, error)),
    )
}

fn validate_sns_refresh_attempt(
    path: &Path,
    expected_network: &str,
    attempt: &SnapshotRefreshAttempt<SnsRefreshAttemptMetadata>,
) -> Result<(), SnsHostError> {
    let invalid = |reason| SnsHostError::InvalidRefreshAttempt {
        path: path.to_path_buf(),
        reason,
    };
    validate_snapshot_refresh_attempt(attempt, expected_network).map_err(invalid)?;
    let expected_governance = sns_governance_from_cache_path(path)
        .and_then(|name| name.to_str())
        .ok_or_else(|| {
            invalid("attempt path does not contain a Governance identity".to_string())
        })?;
    if attempt.metadata.governance_canister_id != expected_governance {
        return Err(invalid(format!(
            "governance_canister_id is {}, expected {expected_governance}",
            attempt.metadata.governance_canister_id
        )));
    }
    super::target::parse_governance_canister_input(expected_governance)
        .map_err(|error| invalid(error.to_string()))?;
    Ok(())
}

fn read_sns_refresh_attempt(
    cache_root: &Path,
    path: &Path,
    expected_network: &str,
    budget: Option<&mut ManagedReadBudget>,
) -> Result<Option<SnsRefreshAttempt>, SnsHostError> {
    read_snapshot_refresh_attempt_strict::<SnsRefreshAttempt>(cache_root, path, budget)
        .map_err(|error| {
            error.into_error(SNS_CACHE_COMPONENT, SnsHostError::from, |path, reason| {
                SnsHostError::InvalidRefreshAttempt { path, reason }
            })
        })?
        .map(|attempt| {
            validate_sns_refresh_attempt(path, expected_network, &attempt)?;
            Ok(attempt)
        })
        .transpose()
}

pub(in crate::sns::report) fn read_sns_refresh_attempt_status_strict(
    cache_root: &Path,
    path: &Path,
    expected_network: &str,
    budget: Option<&mut ManagedReadBudget>,
) -> Result<Option<SnsRefreshAttemptStatus>, SnsHostError> {
    Ok(
        read_sns_refresh_attempt(cache_root, path, expected_network, budget)?
            .map(SnsRefreshAttemptStatus::from_validated),
    )
}