feldera_types/

checkpoint.rs

use std::{collections::BTreeSet, time::Duration};

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use uuid::Uuid;

use crate::suspend::TemporarySuspendError;

/// Checkpoint status returned by the `/checkpoint_status` endpoint.
#[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)]
pub struct CheckpointStatus {
    /// Most recently successful checkpoint.
    pub success: Option<u64>,

    /// Most recently failed checkpoint, and the associated error.
    ///
    /// This tracks transient checkpoint failures (e.g. I/O errors during
    /// writing).  A subsequent successful checkpoint will not clear this
    /// field — it always reflects the *last* failure that occurred.
    pub failure: Option<CheckpointFailure>,
}

/// Current checkpoint activity state.
#[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)]
#[serde(tag = "status", rename_all = "snake_case")]
pub enum CheckpointActivity {
    /// No checkpoint is pending or in progress.
    #[default]
    Idle,

    /// A checkpoint has been requested but is delayed for temporary reasons
    /// (e.g. replaying, bootstrapping, transaction in progress, or input
    /// endpoint barriers that require the coordinator to run steps).
    Delayed {
        /// Why the checkpoint cannot proceed yet.
        reasons: Vec<TemporarySuspendError>,
        /// When the delay started (serialized as ISO 8601).
        delayed_since: DateTime<Utc>,
    },

    /// A checkpoint is currently being written to storage.
    InProgress {
        /// When the checkpoint write started (serialized as ISO 8601).
        started_at: DateTime<Utc>,
    },
}

impl CheckpointActivity {
    /// Returns `delayed_since` if this is `CheckpointActivity::Delayed`.
    pub fn delayed_since(&self) -> Option<DateTime<Utc>> {
        match self {
            Self::Delayed { delayed_since, .. } => Some(*delayed_since),
            _ => None,
        }
    }
}

/// Information about a failed checkpoint.
#[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)]
pub struct CheckpointFailure {
    /// Sequence number of the failed checkpoint.
    pub sequence_number: u64,

    /// Error message associated with the failure.
    pub error: String,

    /// When the failure occurred (serialized as ISO 8601).
    pub failed_at: DateTime<Utc>,
}

/// Response to a checkpoint request.
#[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)]
pub struct CheckpointResponse {
    pub checkpoint_sequence_number: u64,

    /// Pass back to `/checkpoint_status` to detect pipeline restarts.
    ///
    /// Only `None` if sent by an older pipeline.
    pub incarnation_uuid: Option<Uuid>,
}

impl CheckpointResponse {
    pub fn new(checkpoint_sequence_number: u64, incarnation_uuid: Uuid) -> Self {
        Self {
            checkpoint_sequence_number,
            incarnation_uuid: Some(incarnation_uuid),
        }
    }
}

/// Response to a sync checkpoint request.
#[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)]
pub struct CheckpointSyncResponse {
    pub checkpoint_uuid: Uuid,

    /// Pass back to `/checkpoint/sync_status` to detect pipeline restarts.
    ///
    /// Only `None` if sent by an older pipeline.
    pub incarnation_uuid: Option<Uuid>,
}

impl CheckpointSyncResponse {
    pub fn new(checkpoint_uuid: Uuid, incarnation_uuid: Uuid) -> Self {
        Self {
            checkpoint_uuid,
            incarnation_uuid: Some(incarnation_uuid),
        }
    }
}

/// Query parameters for `/checkpoint_status` and `/checkpoint/sync_status`.
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct CheckpointStatusQuery {
    /// Pass the value from the initial response, to detect pipeline restarts.
    ///
    /// Optional to allow older clients to continue working.
    pub incarnation_uuid: Option<Uuid>,
}

/// Checkpoint status returned by the `/checkpoint/sync_status` endpoint.
#[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)]
#[serde(default)]
pub struct CheckpointSyncStatus {
    /// Most recently successful checkpoint sync.
    ///
    /// If `success` and `failure` would otherwise name the same UUID, then the
    /// most recent result is set and the other is cleared.
    pub success: Option<Uuid>,

    /// Most recently failed checkpoint sync, and the associated error.
    pub failure: Option<CheckpointSyncFailure>,

    /// Most recently successful automated periodic checkpoint sync.
    pub periodic: Option<Uuid>,

    /// Checkpoint syncs running right now.
    ///
    /// A UUID leaves `running` and lands in `success` or `failure` at the same
    /// moment.
    ///
    /// Periodic syncs do not appear here.
    pub running: BTreeSet<Uuid>,
}

/// Information about a failed checkpoint sync.
#[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)]
pub struct CheckpointSyncFailure {
    /// UUID of the failed checkpoint.
    pub uuid: Uuid,

    /// Error message associated with the failure.
    pub error: String,
}

/// Holds meta-data about a checkpoint that was taken for persistent storage
/// and recovery of a circuit's state.
#[derive(Debug, Clone, Default, Serialize, Deserialize, ToSchema, PartialEq, Eq)]
pub struct CheckpointMetadata {
    /// A unique identifier for the given checkpoint.
    ///
    /// This is used to identify the checkpoint in the file-system hierarchy.
    pub uuid: Uuid,
    /// An optional name for the checkpoint.
    pub identifier: Option<String>,
    /// Fingerprint of the circuit at the time of the checkpoint.
    // Uses the full 64-bit range.
    #[schema(format = "uint64")]
    pub fingerprint: u64,
    /// Total size of the checkpoint files in bytes.
    pub size: Option<u64>,
    /// Total number of steps made.
    pub steps: Option<u64>,
    /// Total number of records processed.
    pub processed_records: Option<u64>,
}

/// Identifies a host within a multihost pipeline.
///
/// Used to scope checkpoint sync operations (push/pull) to the correct
/// remote subdirectory.
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub struct HostInfo {
    /// Zero-based index of this host in the pipeline layout.
    pub host_idx: usize,
    /// Total number of hosts in the pipeline layout.
    pub n_hosts: usize,
}

impl HostInfo {
    /// Returns the remote storage subdirectory prefix for this host,
    /// e.g. `"host0"` for index 0.
    pub fn prefix(&self) -> String {
        if self.host_idx >= self.n_hosts {
            log::warn!(
                "HostInfo::prefix: host_idx {} >= n_hosts {}",
                self.host_idx,
                self.n_hosts
            );
        }
        format!("host{}", self.host_idx)
    }
}

/// Format of `pspine-batches-*.dat` in storage.
///
/// These files exist to be a simple format for higher-level code and outside
/// tools to parse.  The spine itself writes them for that purpose, but it does
/// not read them.
#[derive(Debug, Serialize, Deserialize)]
pub struct PSpineBatches {
    pub files: Vec<String>,
}

/// Serialized form of `dependencies.json` on disk.
///
/// Two formats. New checkpoints write the struct form (`V2`) carrying both
/// the batch list referenced at the storage root *and* the list of per-operator
/// state files inside the checkpoint dir. Old checkpoints stored only the
/// batch-filename array (`V1`); they remain readable so a rolling upgrade
/// across in-flight checkpoints is safe.
#[derive(Debug, Deserialize)]
#[serde(untagged)]
pub enum CheckpointDependencies {
    V2 {
        /// Batch filenames at the storage root (`w*.feldera`) that the
        /// checkpoint references for GC retention.
        batches: Vec<String>,
        /// Per-operator state filenames inside the checkpoint dir
        /// (e.g. `pspine-*.dat`, `z1-*.dat`, `CHECKPOINT`). Consumed by
        /// restore-time verification. Defaulted to empty for forward compat.
        #[serde(default)]
        state_files: Vec<String>,
    },
    /// Legacy form: JSON array of batch filenames at the storage root
    /// (`w*.feldera`). No state-file manifest.
    V1(Vec<String>),
}

impl CheckpointDependencies {
    /// Batch files the checkpoint references at the storage root
    /// (`w*.feldera`). Present in both V1 and V2 checkpoints.
    pub fn batches(&self) -> &[String] {
        match self {
            CheckpointDependencies::V2 { batches, .. } => batches,
            CheckpointDependencies::V1(batches) => batches,
        }
    }

    /// Per-operator state files the checkpoint owned at commit time. These
    /// live inside the checkpoint dir (e.g. `pspine-*.dat`, `z1-*.dat`).
    /// Empty for V1 checkpoints, which predate the state-file manifest.
    pub fn state_files(&self) -> &[String] {
        match self {
            CheckpointDependencies::V2 { state_files, .. } => state_files,
            CheckpointDependencies::V1(_) => &[],
        }
    }
}

/// Serialized form written to `dependencies.json`.  Always emits V2.
#[derive(Debug, Serialize)]
pub struct CheckpointDependenciesWrite<'a> {
    pub batches: &'a [String],
    pub state_files: &'a [String],
}

/// A checkpoint that exists in remote object storage.
#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
pub struct RemoteCheckpoint {
    /// UUID of the checkpoint.
    pub uuid: Uuid,
}

#[derive(Debug)]
pub struct CheckpointSyncMetrics {
    pub duration: Duration,
    pub speed: u64,
    pub bytes: u64,
}

/// Status of a `POST /coordination/checkpoint/pull` operation.
///
/// Returned by `GET /coordination/checkpoint/pull_status`.
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
#[serde(tag = "status", rename_all = "snake_case")]
pub enum CheckpointPullStatus {
    /// No pull has been requested yet.
    #[default]
    NotRequested,
    /// A pull is currently in progress.
    InProgress,
    /// The pull completed successfully.
    Ok,
    /// The pull failed.
    Error { error: String },
}

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

    /// A newer client's `CheckpointResponse`/`CheckpointSyncResponse` must
    /// still parse a response from an older pipeline that predates the
    /// `incarnation_uuid` field, yielding `None` rather than a
    /// deserialization error.
    #[test]
    fn deserialize_response_missing_incarnation_uuid() {
        let resp: CheckpointResponse =
            serde_json::from_str(r#"{"checkpoint_sequence_number": 3}"#).unwrap();
        assert_eq!(resp.checkpoint_sequence_number, 3);
        assert_eq!(resp.incarnation_uuid, None);

        let sync_resp: CheckpointSyncResponse =
            serde_json::from_str(r#"{"checkpoint_uuid": "00000000-0000-0000-0000-000000000001"}"#)
                .unwrap();
        assert_eq!(sync_resp.incarnation_uuid, None);
    }

    /// Legacy bare-array dependencies.json from older checkpoints must still
    /// parse, yielding an empty state-file list (no manifest verification).
    #[test]
    fn deserialize_v1_legacy_array() {
        let raw = r#"["w0-aaa.feldera", "w1-bbb.feldera"]"#;
        let deps: CheckpointDependencies = serde_json::from_str(raw).unwrap();
        assert!(deps.state_files().is_empty());
        assert_eq!(deps.batches(), &["w0-aaa.feldera", "w1-bbb.feldera"]);
    }

    /// Current struct form carries both lists.
    #[test]
    fn deserialize_v2_struct() {
        let raw = r#"{
            "batches": ["w0-aaa.feldera"],
            "state_files": ["pspine-0-zzz.dat", "CHECKPOINT"]
        }"#;
        let deps: CheckpointDependencies = serde_json::from_str(raw).unwrap();
        assert_eq!(deps.state_files(), &["pspine-0-zzz.dat", "CHECKPOINT"]);
        assert_eq!(deps.batches(), &["w0-aaa.feldera"]);
    }

    /// V2 without `state_files` (partial writer, partial migration)
    /// deserializes with an empty state-file list rather than failing.
    #[test]
    fn deserialize_v2_missing_state_files_defaults_to_empty() {
        let raw = r#"{"batches": ["w0-aaa.feldera"]}"#;
        let deps: CheckpointDependencies = serde_json::from_str(raw).unwrap();
        assert!(deps.state_files().is_empty());
        assert_eq!(deps.batches(), &["w0-aaa.feldera"]);
    }

    /// Writes emit V2 and round-trip back to the same content.
    #[test]
    fn write_v2_round_trips() {
        let batches = vec!["w0-x.feldera".to_string()];
        let state_files = vec!["pspine-0-y.dat".to_string()];
        let json = serde_json::to_string(&CheckpointDependenciesWrite {
            batches: &batches,
            state_files: &state_files,
        })
        .unwrap();
        let deps: CheckpointDependencies = serde_json::from_str(&json).unwrap();
        assert_eq!(deps.state_files(), state_files.as_slice());
        assert_eq!(deps.batches(), batches.as_slice());
    }

    #[test]
    fn host_info_prefix_formats_index() {
        assert_eq!(
            HostInfo {
                host_idx: 0,
                n_hosts: 2
            }
            .prefix(),
            "host0"
        );
        assert_eq!(
            HostInfo {
                host_idx: 1,
                n_hosts: 2
            }
            .prefix(),
            "host1"
        );
        assert_eq!(
            HostInfo {
                host_idx: 42,
                n_hosts: 100
            }
            .prefix(),
            "host42"
        );
    }
}