tag2upload_service_manager/

db_data.rs

//! Database
//!
//! See `JobStatus` for the states of a job,
//! and progression through the db.
//!
//! # Coalescing
//!
//!  1. We don't do, simultaneously in two jobs
//!
//!      - work on the same objectid
//!      - work on the same repo URL
//!
//!     Implemented in [`JobInWorkflow::start`].
//!
//!  2. We don't try fetching for an objectid, if we have other jobs
//!    for the same objectid which do have the tag data,
//!    depending on the other job's state:
//!
//!     1. **Noticed / Queued / Building**:
//!       They'll eventually progress to a different state.
//!       This job should wait.
//!       Implemented in [`JobInWorkflow::start_for_forge`].
//!
//!     2. **Irrecoverable / Uploaded**:
//!       This job is (presumably) Noticed or Queued.
//!       It will be marked a duplicate of that one,
//!       and takes on its status.
//!       Implemented in [`db_workflow::coalesce_completed`].
//!
//!     3. **Failed**:
//!       This job may proceed.
//!
//! # Replay / manual retry:
//!
//! We look at the timestamp (in the `tagger` line).
//! We don't accept tags more than SLOP In the future,
//! or more than MAX_AGE in the past.
//! We expire things from our database no earlier than MAX_AGE.
//!
//! If we get the same tag again (by git object id), it's ignored.
//! (see above).
//! There is no facility for manually forcing a retry.
//! The uploader should use a new version number.
//!
//! # Forge up status
//!
//! TODO forge up status is not implemented
//!
//! We remember for each forge whether we think it's up.
//!
//! We start out thinking the forge is down.
//! When we think the forge is down, we do a preflight
//! check (https request) to see if it seems to have come up.
//! We retry that check at increasing intervals.
//!
//! Whenever a job fails, we go back to thinking the forge is down.

use crate::prelude::*;

//==================== jobs table ====================

/// Guarantees:
///
/// * The configured repository and forge and calling IP address were fine
/// * The tag name is plausible
#[derive(Debug, Clone)]
#[derive(Deftly)]
#[derive_deftly(FromSqlRow, AsBSqlRow, UpdateSqlRow, UiMap)]
pub struct JobData {
    pub repo_git_url: String,
    pub tag_objectid: GitObjectId,
    pub tag_name: String,
    pub forge_host: Hostname,
    pub forge_namever: ForgeNamever,
    #[deftly(ui(skip))]
    pub forge_data: ForgeData,
    // We don't use this operationally, but it's very useful for reporting
    #[deftly(bsql(flatten), ui(flatten))]
    pub tag_meta: t2umeta::Parsed,
}

#[derive(Debug, Deftly, Clone)]
#[derive_deftly(FromSqlRow, AsBSqlRow, UpdateSqlRow)]
#[derive_deftly(UiMap, UpdateWorkerReport)]
pub struct JobRow {
    #[deftly(bsql(rowid))]
    pub jid: JobId,
    #[deftly(bsql(flatten), ui(flatten))]
    pub data: JobData,
    pub received: TimeT,
    pub last_update: TimeT,
    #[deftly(ui(skip))]
    pub tag_data: NoneIsEmpty<TagObjectData>,
    #[deftly(worker_report)]
    pub status: JobStatus,
    /// Not `None` iff we are currently actually processing this job somehow
    pub processing: NoneIsEmpty<ProcessingInfo>,
    #[deftly(worker_report)]
    pub info: String,
    pub duplicate_of: Option<JobId>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
#[derive(Deftly, strum::EnumIter)]
#[derive_deftly(FromSqlStrEnum, ToSqlEnum, UiDisplayEnum)]
#[cfg_attr(test, derive_deftly_adhoc)]
pub enum JobStatus {
    /// Webhook request has been received and queued.
    ///
    /// The webhook details including the tag name look plausible.
    /// We're happy it's at the right forge etc.
    ///
    /// Next state is Queued` or NotForUs
    Noticed,

    /// Tag object has been fetched, job is ready for an Oracle worker
    ///
    /// `JobRow.tag_data` is nonempty, and
    /// `JobRow.tag_objectid` has been confirmed locally.
    Queued,

    /// Job has been given to an Oracle worker for processing.
    ///
    /// This state ought to be accompanied by a connection
    /// from the worker, through which we will get the outcome.
    /// If it isn't (eg after restart), the job is irrecoverable.
    Building,

    /// Tag is *not* a (current) instruction to us
    NotForUs,

    /// Job has failed; other attempts for the same tag may work
    ///
    /// This is a possible next state from any of the other states.
    Failed,

    /// Job has failed; other attempts for the same tag are doomed
    ///
    /// This is a possible next state from any of the other states.
    Irrecoverable,

    /// Job has been completed successfully
    Uploaded,
}

#[cfg(test)] pub(crate) use derive_deftly_driver_JobStatus;

//==================== shown_status field ====================

define_derive_deftly! {
    RecursiveEnumIter:

    impl $ttype {
        pub fn iter() -> impl Iterator<Item = Self> {
            chain!( $(
                ${if v_is_unit {
                    [$vtype]
                } else {
                    $( $ftype::iter() ).map($vtype)
                }},
            ))
        }
    }
}

#[derive(Clone, Copy, Eq, PartialEq, Hash, Debug)]
#[derive(derive_more::Display, Deftly)]
#[derive_deftly(RecursiveEnumIter, UiDisplayEnum, FromSqlStrEnum, ToSqlEnum)]
pub enum ShownJobStatus {
    #[display("{_0:?}")]
    JobStatus(JobStatus),
    Duplicate,
}
impl ShownJobStatus {
    pub fn new(this_status: JobStatus, duplicate_of: Option<JobId>) -> Self {
        if let Some(_) = duplicate_of {
            ShownJobStatus::Duplicate
        } else {
            ShownJobStatus::JobStatus(this_status)
        }
    }
}
impl ShownJobStatus {
    pub fn from_row(row: &JobRow) -> Self {
        ShownJobStatus::new(row.status, row.duplicate_of)
    }
}
impl Ord for ShownJobStatus {
    fn cmp(&self, other: &Self) -> cmp::Ordering {
        use ShownJobStatus as S;
        use JobStatus as J;
        let o = |v| match v {
            S::JobStatus(s @ J::Uploaded) => (30, Some(s)),
            S::Duplicate                  => (20, None),
            S::JobStatus(other)           => (10, Some(other)),
        };
        o(*self).cmp(&o(*other))
    }
}
impl PartialOrd for ShownJobStatus {
    fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
        Some(self.cmp(other))
    }
}

//==================== last_expiry ====================

#[derive(Deftly, Ord, PartialOrd, Eq, PartialEq)]
#[derive_deftly(UiMap, FromSqlRow, AsBSqlRow, UpdateSqlRow)]
pub struct LastExpiryRow {
    pub run: TimeT,
    pub threshold: TimeT,
}

//==================== stats_by_shown_status* tables ====================

#[derive(Deftly, Ord, PartialOrd, Eq, PartialEq)]
#[derive_deftly(UiMap, FromSqlRow, AsBSqlRow, UpdateSqlRow)]
pub struct StatsByShownStatusRow {
    pub shown_status: ShownJobStatus,
    pub n_jobs: i64,
}

//==================== pause_insns table ====================

#[derive(Debug, Clone)]
#[derive(Deftly)]
#[derive_deftly(FromSqlRow, AsBSqlRow, UpdateSqlRow)]
pub struct PauseInsn {
    pub pause_id: PauseId,
    pub pause_key: String,
    pub pause_info: String,
    pub throttle: bool,
}