tag2upload_service_manager/

retry.rs

use crate::prelude::*;

/// Counts towards the retry count
///
/// Ie, there was another job at the same forge that succeeded,
/// since the last failure of *this* job.
#[derive(Debug, Clone, Copy)]
pub struct IsSalient;

#[derive(Debug, Clone, Copy)]
pub struct IsFinal;

#[derive(Debug)]
pub enum Instructions {
    /// Don't do anything until `retry_earliest`
    Wait {
        retry_earliest: TimeT,
    },
    /// Attempt now
    Attempt(AttemptInfo),
}

#[derive(Debug, derive_more::Display)]
#[display(
    "[{}{}]",
    is_salient.map(|IsSalient| "S").unwrap_or_default(),
    is_final.map(|()| "r").unwrap_or_else(|IsFinal| "F"),
)]
pub struct AttemptInfo {
    pub is_salient: Option<IsSalient>,
    pub is_final: Result<(), IsFinal>,
}

#[derive(Debug)]
pub struct HostState {
    last_failure: TimeT,

    /// Last success for this host
    ///
    /// We don't store this in the db.  That means that we might do
    /// one additional retry, for each job, for each restart.  That's OK.
    last_success: Option<HistEntId>,
}

const DEFAULT_HOST_STATE: HostState = HostState {
    last_failure: TimeT(0),
    last_success: None,
};

#[derive(Default, Debug)]
pub struct HostStates(HashMap<Hostname, retry::HostState>);

impl Globals {
    pub fn host_retry_state(&self) -> MutexGuard<'_, HostStates> {
        self.host_retry_state.lock()
            .unwrap_or_else(|pe| pe.into_inner())
    }
}

impl HostStates {
    pub fn get(&self, host: &Hostname) -> &HostState {
        self.0.get(host).unwrap_or(&DEFAULT_HOST_STATE)
    }
    pub fn get_mut(&mut self, host: &Hostname) -> &mut HostState {
        // Ideally we'd clone `host` only if we need to,
        // but the HashMap API doesn't allow that.
        self.0.entry(host.clone()).or_insert(DEFAULT_HOST_STATE)
    }
}

impl HostState {
    /// Calculate the job-specific earliest retry time
    ///
    /// Implements part of "Retry schedule"
    /// from TAG2UPLOAD-MANAGER-PROTOCOL.md.
    /// TODO consider moving that material here.
    ///
    /// This is the job-specific part.  The forge-specific retry limit is
    /// is not reified in the database (and is reset on every t2usm restart).
    ///
    /// Run before we make an attempt.
    pub fn calculate_retry(
        &self,
        dbt: &mut DbTransaction,
        now: TimeT,
        policy: &config::RetryPolicy,
        jid: JobId,
    ) -> Result<retry::Instructions, DbError<IE>> {
        let earliest_due_to_last_failure_same_host =
            self.last_failure.0
            .saturating_add(policy.timeout_initial.as_secs());
        if now.0 < earliest_due_to_last_failure_same_host {
            // Recent failure at the forge.  Don't even query the DB.
            trace!("need to wait for host, {}s",
                   earliest_due_to_last_failure_same_host - now.0);
            return Ok(Instructions::Wait {
                retry_earliest: TimeT(earliest_due_to_last_failure_same_host),
            })
        }

        let job_row: JobRow = dbt.bsql_query_1(&bsql!("
                        SELECT *
                          FROM jobs
                         WHERE jid = " jid "
               "))?;

        // Calculate number of attempts, which tells us the retry interval
        let (retry_count,) : (u32,) = dbt.bsql_query_1(&bsql!("
                        SELECT count(*)
                          FROM job_history
                         WHERE jid = " jid "
                           AND processing = " JOB_PROCESSING_TEMPFAIL "
               "))?;

        // If we haven't had a temporary failure yet, try now (and expect a retry)
        if retry_count == 0 {
            trace!("no attempts yet, run right away");
            return Ok(Instructions::Attempt(AttemptInfo {
                is_salient: Some(IsSalient),
                is_final: Ok(()),
            }));
        }

        let interval =
            policy.timeout_initial.as_secs() as f32 *
            policy.timeout_increase
            .powi(retry_count.try_into().unwrap_or(i32::MAX));
        let interval: u64 = cmp::min(
            interval as u64,
            policy.timeout_mintotal.as_secs(),
        );

        // Calculate the retry time based on the previous attempt

        let mut get_attempt = |asc_desc: &dyn Bsql|
            -> Result<JobHistoryRow, _>
        {
            dbt.bsql_query_1(&bsql!("
                        SELECT *
                          FROM job_history
                         WHERE jid = " jid "
                           AND processing = 'tempfail'
                      ORDER BY histent " {asc_desc} "
                         LIMIT 1
            "))
        };

        let first_attempt = get_attempt(&bsql!("ASC"))?;
        let previous_attempt = get_attempt(&bsql!("DESC"))?;

        trace!(?first_attempt, ?previous_attempt,
               "job-specific calculation");

        let retry_earliest = previous_attempt.last_update + interval;
        if now >= retry_earliest {
            return Ok(Instructions::Wait { retry_earliest })
        }

        // It's time to retry.  But, is it salient and/or final?

        let is_salient = (|| {
            if previous_attempt.histent >= self.last_success? {
                return None
            }
            Some(IsSalient)
        })();

        let is_final = if (
            {
                retry_count + 1 >= policy.min_retries 
            } && {
                let salient_count =
                    job_row.s.retry_salient_count +
                    (is_salient.is_some() as u32);
                salient_count >= policy.min_salient_retries
            } && {
                now >=
                    first_attempt.last_update
                    + policy.timeout_mintotal.as_secs()
            }
        ) {
            Err(IsFinal)
        } else {
            Ok(())
        };

        Ok(Instructions::Attempt(AttemptInfo { is_salient, is_final }))
    }

    pub fn note_success(&mut self, histent: HistEntId) {
        self.last_success.track_maximum_seen(Some(histent))
    }

    pub fn note_retriable_failure(&mut self, now: TimeT) {
        self.last_failure.track_maximum_seen(now);
    }
}