tag2upload_service_manager/

expire.rs

use crate::prelude::*;

use std::io::BufWriter;
use std::fs::File;

pub fn expire_now(gl: &Arc<Globals>) -> Result<(), IE> {
    let overflow_msg = "implausible intervals.expiry or system time";

    let now = gl.now_systemtime();

    let oldest_to_keep: TimeT = {
        TimeT::try_from(
            now
                .checked_sub(*gl.config.intervals.expire)
                .ok_or_else(|| internal!("{overflow_msg}"))?
        )
            .into_internal(overflow_msg)?
    };

    let archive_dir = &gl.config.files.archive_dir;
    let timestamp = format!("{}", HtTimeT(now.into()));
    let archive_leaf = format!("expired-{timestamp}.dump.txt");
    let archive_subdir = timestamp.split_once('-')
        .ok_or_else(|| internal!(
            "unexpected archive timestamp format {timestamp}"
        ))?
        .0;
    let archive_subdir = format!("{archive_dir}/{archive_subdir}");
    let archive_file = format!("{archive_subdir}/{archive_leaf}");
    let archive_file_tmp = format!("{archive_file}.tmp");

    let sync_directory = |d| (|| {
        File::open(d).context("open")?
            .sync_all().context("sync")
    })()
        .with_context(|| format!("{d:?}"))
        .into_internal("fsync archive (sub)directory");

    let (n_expired, archived_to) = db_transaction(TN::Expiry, |dbt| {

        // Clear out any existing *.tmp files

        let clearout_glob = format!("{archive_dir}/*/expired-*.dump.txt.tmp");
        (|| {
            for clearout in glob::glob(&clearout_glob).context("glob start")? {
                let clearout = clearout.context("glob")?;
                fs::remove_file(&clearout)
                    .with_context(|| clearout.display().to_string())
                    .context("delete")?;
            }
            Ok::<_, AE>(())
        })().with_context(|| clearout_glob.clone())
            .into_internal("clear out old tmp files")?;

        // Create the subdirectory

        fs::create_dir(&archive_subdir).or_else(|e| match e {
            e if e.kind() == io::ErrorKind::AlreadyExists => Ok(()),
            other => Err(other),
        }).into_internal("create archive subdirectory")?;

        // Prepare to write an archive file

        let fwriter = deferred_file_creation::Writer::new_uncreated(
            archive_file_tmp.clone(),
            fs::OpenOptions::new().create_new(true).clone(),
        );
        let mut archiver = mini_sqlite_dump::Archiver::start(
            dbt, fwriter, ["jobs"],
        ).into_internal("prepare to archive expired data")?;

        let mut n_expired = 0;

        dbt.bsql_exec(&bsql!("
            DELETE FROM last_expiry
        "))?;
        dbt.bsql_exec(&bsql!("
            INSERT INTO last_expiry " .insert_row(LastExpiryRow {
                run: now.into(),
                threshold: oldest_to_keep,
            })"
        "))?;

        let archived_to = dbt.bsql_query_rows_concurrent(

            &bsql!(
                " SELECT * FROM jobs
                          WHERE last_update < " oldest_to_keep "
            "),

            |mut rows| {

                let mut tarchiver = archiver.start_table("jobs")
                    .into_internal("prepare to archive expired rows")?;

                /// `fn handle_io_err<E>(msg: impl Display)
                ///     -> impl FnOnce(E) -> DbError<IE>;`
                macro_rules! handle_io_err { { $msg:expr } => {
                    |e| {
                        // Trickery to ensure that `e` is safely eraseable,
                        Err::<Void, _>(e)
                            .context(archive_file_tmp.clone())
                            .context($msg.to_string())
                            .into_internal("archive file")
                            .void_unwrap_err()
                    }
                } }

                let mut expired_counts = HashMap::<_, i64>::default();

                while let Some(row) = rows.next()
                    .db_context("fetch row to expire")?
                {
                    let jid: JobId = row.get("jid")
                        .db_context("extract jid")?;

                    n_expired += 1;
                    tarchiver.writer_mut().ensure_created()
                        .map_err(handle_io_err!("start file"))?;
                    tarchiver.write_row(row)
                        .map_err(handle_io_err!("write row"))?;

                    // We don't archive all the intermediate states,
                    // only the final one.
                    dbt.bsql_exec_concurrent(&bsql!(
                        "DELETE FROM job_history WHERE jid = " jid
                    ))?;
                    dbt.bsql_exec_concurrent(&bsql!(
                        "DELETE FROM jobs WHERE jid = " jid
                    ))?;

                    match JobRow::from_sql_row(&row) {
                        Ok(row) => {
                            let shown = ShownJobStatus::from_state(&row.s);
                            *expired_counts
                                .entry(shown)
                                .or_default()
                                += 1;
                        },
                        Err(e) => error!(%e,
                     "error interpreting expired row, for status"
                        ),
                    }
                }

                for (shown, n_jobs) in expired_counts {
                    dbt.bsql_exec_concurrent(&bsql!("
                             UPDATE stats_by_shown_status_expired
                                SET n_jobs = n_jobs + " n_jobs "
                              WHERE shown_status = " shown "
                            "))?;
                }

                // If we actually did any rows, flush and install

                let fwriter = archiver.finish_with_writer()
                    .map_err(handle_io_err!("complete"))?;

                if let Some::<(File, String)>((file, _)) = {
                    fwriter.finish_created()
                        .map_err(handle_io_err!("finish"))?
                } {
                    file.sync_all()
                        .map_err(handle_io_err!("fsync"))?;

                    fs::rename(
                        &archive_file_tmp,
                        &archive_file,
                    )
                        .with_context(|| archive_file.clone())
                        .with_context(|| archive_file_tmp.clone())
                        .into_internal("install archive file")?;

                    sync_directory(&archive_subdir)?;
                    sync_directory(&archive_dir)?;

                    Ok::<_, DbError<IE>>(Some(archive_leaf.clone()))
                } else {
                    assert_eq!(n_expired, 0);
                    Ok::<_, DbError<IE>>(None)
                }
            }
        )?;

        // By returning `Ok` we instruct db_transaction to COMMIT.
        // We have already installed the new archive file, above.
        //
        // COMMIT isn't supposed to be likely to fail since our TN::Expiry
        // causes us to use BEGIN IMMEDIATE, so there oughtn't to e conflicts.
        //
        // If thins *do* go badly, we have written the archive,
        // but not actually expired the rows, which is the better failure.
        Ok((n_expired, archived_to))
    })?;

    if let Some(archive) = archived_to {
        info!(n_expired, %oldest_to_keep, ?archive, "expired");
    } else {
        info!(n_expired, %oldest_to_keep, "nothing expired");
    }

    Ok(())
}

impl LastExpiryRow {
    pub fn get(dbt: &mut DbTransaction)
        -> Result<LastExpiryRow, DbError<IE>>
    {
        let dummy_time: TimeT = 0.into();

        let row = dbt.bsql_query_01(&bsql!("SELECT * FROM last_expiry"))?
            .unwrap_or_else(|| LastExpiryRow {
                run: dummy_time,
                threshold: dummy_time,
            });
        Ok(row)
    }

    pub fn for_ui(dbt: &mut DbTransaction)
                  -> Result<UiSerializeMap<LastExpiryRow>, DbError<IE>>
    {
        Self::get(dbt).map(UiSerializeMap)
    }
}

pub fn start_task(gl: &Arc<Globals>) {
    gl.spawn_task_running("expiry", {
        let gl = gl.clone();
        async move {
            loop {
                let expire_every = *gl.config.intervals.expire_every;
                select! {
                    () = tokio::time::sleep(expire_every) => {},
                    sd = gl.await_shutdown() => return Err(sd.into()),
                };
                expire_now(&gl)?;
            }
        }
    })
}

//---------- deferred file writer ----------

pub mod deferred_file_creation {

    use super::*;

    define_derive_deftly! {
        DelegateWrite for struct, expect items:

        ${define FIELD { $( ${when fmeta(delegate_write)} $fname ) }}

        impl io::Write for $ttype {
            fn write(&mut self, data: &[u8]) -> io::Result<usize> {
                io::Write::write(&mut self.$FIELD, data)
            }
            fn flush(&mut self) -> io::Result<()> {
                io::Write::flush(&mut self.$FIELD)
            }
        }
    }

    #[derive(Debug, Deftly)]
    #[derive_deftly(DelegateWrite)]
    pub struct Writer {
        filename: String,
        #[deftly(delegate_write)]
        file: Either<NotYetCreated, BufWriter<File>>,
    }

    #[derive(Debug, Deftly)]
    #[derive_deftly(DelegateWrite)]
    struct NotYetCreated {
        #[deftly(delegate_write)]
        buf: Vec<u8>,
        open_options: fs::OpenOptions,
    }

    impl Writer {
        pub fn new_uncreated(
            filename: String,
            mut open_options: fs::OpenOptions,
        ) -> Self {
            open_options.write(true);
            let file = Either::Left(NotYetCreated {
                buf: vec![],
                open_options,
            });
            Writer { filename, file }
        }

        /// Make the file exist (idempotent)
        pub fn ensure_created(&mut self) -> io::Result<()> {
            match &mut self.file {
                Either::Left(nyc) => {
                    let file = nyc.open_options.open(&self.filename)?;
                    let mut file = BufWriter::new(file);
                    file.write_all(&mem::take(&mut nyc.buf))?;
                    self.file = Either::Right(file);
                }
                Either::Right(bw) => {
                    let _: &mut BufWriter<File> = bw;
                }
            }
            Ok(())
        }

        /// Finish up
        ///
        /// If the file was created, flush it and return the filename.
        /// Otherwise, return None.
        pub fn finish_created(self) -> io::Result<Option<(File, String)>> {
            match self.file {
                Either::Left(nyc) => {
                    let NotYetCreated { .. } = nyc;
                    Ok(None)
                },
                Either::Right(file) => {
                    let file = file.into_inner()
                        .map_err(|e| e.into_error())?;
                    Ok(Some((file, self.filename)))
                }
            }
        }
    }
}