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.

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: forge::Namever,
    #[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,
}

/// `jobs` table row
#[derive(Debug, Deftly, Clone)]
#[derive_deftly(FromSqlRow, AsBsqlRow, UpdateSqlRow)]
#[derive_deftly(UiMap)]
pub struct JobRow {
    #[deftly(bsql(rowid))]
    pub jid: JobId,
    #[deftly(bsql(flatten), ui(flatten))]
    pub data: JobData,
    pub received: TimeT,

    /// Time of the earliest retry
    ///
    /// This is calculated by the retry scheduling algorithm
    /// based on information in `job_history`.
    /// When it's reached, we recalculate (in case of config changes, etc.).
    /// So this stored value is allowed to be too early.
    ///
    /// `0` if this job has never failed
    pub retry_earliest: TreatZeroAsNone<TimeT>,

    #[deftly(ui(skip))]
    pub tag_data: NoneIsEmpty<TagObjectData>,
    #[deftly(bsql(flatten), ui(flatten))]
    pub s: JobState,
}

/// Primary state of a job
///
/// Preserved at each state change, for the job history in the web UI.
/// Doesn't include the tag data.
#[derive(Debug, Deftly, Clone)]
#[derive_deftly(FromSqlRow, AsBsqlRow, UpdateSqlRow)]
#[derive_deftly(UiMap, UpdateWorkerReport)]
pub struct JobState {
    #[deftly(worker_report)]
    pub status: JobStatus,
    pub last_update: TimeT,
    /// Not `None` iff we are currently actually processing this job somehow
    ///
    /// `"tempfail"` for a temporary failure;
    /// in the job_history table, exactly once for each such failure.
    pub processing: NoneIsEmpty<ProcessingInfo>,
    #[deftly(worker_report)]
    pub info: String,
    pub duplicate_of: Option<JobId>,

    /// Number of attempts that "count"
    ///
    /// Incremented just before an attempt.
    ///
    /// If we are insisting on looking for successes in between failures,
    /// we may retry more often, but those aren't included.
    pub retry_salient_count: u32,
}

#[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,
}

pub const JOB_PROCESSING_TEMPFAIL: &str = "tempfail";

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

//==================== jobs_history table ====================

/// `jobs_history` table row
#[derive(Debug, Deftly, Clone, Deref)]
#[derive_deftly(FromSqlRow)]
// maintained by a TRIGGER, so not AsBsqlRow UpdateSqlRow
pub struct JobHistoryRow {
    pub histent: HistEntId,
    pub jid: JobId,
    #[deftly(bsql(flatten))]
    #[deref]
    pub s: JobState,
}

//==================== 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_state(row: &JobState) -> 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,
}