Merge pull request #2474 from scpwiki/WJ-1225-prune-expired

[WJ-1225] Add job to prune expired uploads

Author: emmiegit

deepwell/config.example.toml

@@ -170,6 +170,19 @@ max-delay-poll-secs = 360  # 6 minutes
 # It is merely for clearing the database of already-expired tokens.
 prune-session-secs = 3600  # 1 hour
 
+# The period, in seconds, to prune all expired file uploads.
+#
+# Uploading files is a two step process:
+# 1. Upload the data, which creates a blob_pending row in the database.
+# 2. Reference that ID when adding a file revision.
+#
+# After this point, or if the upload is aborted, the blob_pending row
+# is no longer needed and can be deleted. This is mediated through the
+# expires_at column in that table, which is the point when we delete
+# the row. This also serves as the limit for how long a pending blob can
+# be (re)used to upload a file.
+prune-upload-secs = 7200  # 2 hours
+
 # The period, in seconds, to prune all unused text rows.
 #
 # The text table deduplicates identical text objects, and

deepwell/src/config/file.rs

@@ -123,6 +123,7 @@ struct Job {
     min_delay_poll_secs: u64,
     max_delay_poll_secs: u64,
     prune_session_secs: u64,
+    prune_upload_secs: u64,
     prune_text_secs: u64,
     name_change_refill_secs: u64,
     lift_expired_punishments_secs: u64,
@@ -284,6 +285,7 @@ impl ConfigFile {
                     min_delay_poll_secs: job_min_poll_delay_secs,
                     max_delay_poll_secs: job_max_poll_delay_secs,
                     prune_session_secs: job_prune_session_secs,
+                    prune_upload_secs: job_prune_upload_secs,
                     prune_text_secs: job_prune_text_secs,
                     name_change_refill_secs: job_name_change_refill_secs,
                     lift_expired_punishments_secs: job_lift_expired_punishments_secs,
@@ -340,6 +342,10 @@ impl ConfigFile {
             job_prune_session_secs < RSMQ_DELAY_LIMIT,
             "Session prune job period time too long",
         );
+        assert!(
+            job_prune_upload_secs < RSMQ_DELAY_LIMIT,
+            "Pending upload prune job period time too long",
+        );
         assert!(
             job_prune_text_secs < RSMQ_DELAY_LIMIT,
             "Text prune job period time too long",
@@ -404,6 +410,7 @@ impl ConfigFile {
             job_min_poll_delay: StdDuration::from_secs(job_min_poll_delay_secs),
             job_max_poll_delay: StdDuration::from_secs(job_max_poll_delay_secs),
             job_prune_session: StdDuration::from_secs(job_prune_session_secs),
+            job_prune_uploads: StdDuration::from_secs(job_prune_upload_secs),
             job_prune_text: StdDuration::from_secs(job_prune_text_secs),
             job_name_change_refill: StdDuration::from_secs(job_name_change_refill_secs),
             job_lift_expired_punishments: StdDuration::from_secs(

deepwell/src/config/object.rs

@@ -134,6 +134,9 @@ pub struct Config {
     /// How often to run the "prune expired sessions" recurring job.
     pub job_prune_session: StdDuration,
 
+    /// How often to run the "prune pending uploads" recurring job.
+    pub job_prune_uploads: StdDuration,
+
     /// How often to run the "prune unused text" recurring job.
     pub job_prune_text: StdDuration,
 

deepwell/src/redis.rs

@@ -19,7 +19,8 @@
  */
 
 use crate::services::job::{
-    JOB_QUEUE_DELAY, JOB_QUEUE_MAXIMUM_SIZE, JOB_QUEUE_NAME, JOB_QUEUE_PROCESS_TIME,
+    Job, JobService, JOB_QUEUE_DELAY, JOB_QUEUE_MAXIMUM_SIZE, JOB_QUEUE_NAME,
+    JOB_QUEUE_PROCESS_TIME,
 };
 use anyhow::Result;
 use redis::aio::MultiplexedConnection;
@@ -51,6 +52,17 @@ pub async fn connect(redis_uri: &str) -> Result<(MultiplexedConnection, Rsmq)> {
             JOB_QUEUE_MAXIMUM_SIZE,
         )
         .await?;
+
+        // Then add initial repeating jobs
+        macro_rules! queue_job {
+            ($job_case:ident) => {
+                JobService::queue_job_inner(&mut rsmq, &Job::$job_case, None).await?
+            };
+        }
+
+        queue_job!(PruneSessions);
+        queue_job!(PrunePendingUploads);
+        queue_job!(PruneText);
     }
 
     Ok((connection, rsmq))

deepwell/src/services/blob/service.rs

@@ -451,6 +451,44 @@ impl BlobService {
         Ok(output)
     }
 
+    // Prune operations
+
+    /// Deletes all expired pending blobs from the database and S3.
+    pub async fn prune(ctx: &ServiceContext<'_>) -> Result<()> {
+        let txn = ctx.transaction();
+        let bucket = ctx.s3_files_bucket();
+        info!("Pruning expired pending blobs from database and S3");
+
+        // Fetch all expired pending blobs
+        let pending_blobs = BlobPending::find()
+            .select_only()
+            .column(blob_pending::Column::ExternalId)
+            .column(blob_pending::Column::S3Path)
+            .filter(blob_pending::Column::ExpiresAt.lte(now()))
+            .into_tuple::<(String, String)>()
+            .all(txn)
+            .await?;
+
+        // Delete from the S3 bucket
+        for (_, s3_path) in &pending_blobs {
+            // Only try to delete if the object exists,
+            // ignore missing objects.
+            if Self::exists(ctx, s3_path).await? {
+                bucket.delete_object(&s3_path).await?;
+            }
+        }
+
+        // Delete from the database
+        let blob_ids = pending_blobs.into_iter().map(|(id, _)| id);
+
+        BlobPending::delete_many()
+            .filter(blob_pending::Column::ExternalId.is_in(blob_ids))
+            .exec(txn)
+            .await?;
+
+        Ok(())
+    }
+
     // Hard-deletion operations
 
     /// Does a dry run on a blob hard deletion, showing what would have been changed.
@@ -872,20 +910,6 @@ impl BlobService {
         find_or_error!(Self::get_metadata_optional(ctx, hash), Blob)
     }
 
-    #[allow(dead_code)] // TEMP
-    pub async fn exists(ctx: &ServiceContext<'_>, hash: &[u8]) -> Result<bool> {
-        // Special handling for the empty blob
-        if hash == EMPTY_BLOB_HASH {
-            debug!("Checking existence of the empty blob");
-            return Ok(true);
-        }
-
-        // Fetch existence from S3
-        let hex_hash = blob_hash_to_hex(hash);
-        let result = Self::head(ctx, &hex_hash).await?;
-        Ok(result.is_some())
-    }
-
     /// Possibly retrieve blob contents, if a flag is set.
     ///
     /// This utility conditionally retrieves the
@@ -922,6 +946,11 @@ impl BlobService {
         }
     }
 
+    async fn exists(ctx: &ServiceContext<'_>, path: &str) -> Result<bool> {
+        let head = Self::head(ctx, path).await?;
+        Ok(head.is_some())
+    }
+
     pub async fn hard_delete(ctx: &ServiceContext<'_>, hash: &[u8]) -> Result<()> {
         // Special handling for empty blobs
         //

deepwell/src/services/caddy/test.rs

@@ -71,6 +71,7 @@ fn build_config(main_domain: &str, files_domain: &str) -> Config {
         job_min_poll_delay: StdDuration::from_secs(0),
         job_max_poll_delay: StdDuration::from_secs(0),
         job_prune_session: StdDuration::from_secs(0),
+        job_prune_uploads: StdDuration::from_secs(0),
         job_prune_text: StdDuration::from_secs(0),
         job_name_change_refill: StdDuration::from_secs(0),
         job_lift_expired_punishments: StdDuration::from_secs(0),

deepwell/src/services/job/service.rs

@@ -19,7 +19,7 @@
  */
 
 use super::prelude::*;
-use rsmq_async::RsmqConnection;
+use rsmq_async::{Rsmq, RsmqConnection};
 use std::time::Duration;
 
 pub const JOB_QUEUE_NAME: &str = "job";
@@ -57,13 +57,19 @@ impl JobService {
         ctx: &ServiceContext<'_>,
         job: &Job,
         delay: Option<Duration>,
+    ) -> Result<()> {
+        let mut rsmq = ctx.rsmq();
+        Self::queue_job_inner(&mut rsmq, job, delay).await
+    }
+
+    pub async fn queue_job_inner(
+        rsmq: &mut Rsmq,
+        job: &Job,
+        delay: Option<Duration>,
     ) -> Result<()> {
         info!("Queuing job {job:?} (delay {delay:?})");
         let payload = serde_json::to_vec(job)?;
-        ctx.rsmq()
-            .send_message(JOB_QUEUE_NAME, payload, delay)
-            .await?;
-
+        rsmq.send_message(JOB_QUEUE_NAME, payload, delay).await?;
         Ok(())
     }
 

deepwell/src/services/job/structs.rs

@@ -27,8 +27,8 @@ pub enum Job {
         depth: u32,
     },
     PruneSessions,
+    PrunePendingUploads,
     PruneText,
-    // TODO add job for pruning incomplete uploads (pending_blob table and corresponding columns for foreign keys)
     NameChangeRefill,
     LiftExpiredPunishments,
 }

deepwell/src/services/job/worker.rs

@@ -22,7 +22,9 @@
 
 use super::prelude::*;
 use crate::api::ServerState;
-use crate::services::{PageRevisionService, SessionService, TextService, UserService};
+use crate::services::{
+    BlobService, PageRevisionService, SessionService, TextService, UserService,
+};
 use crate::utils::debug_pointer;
 use rsmq_async::{Rsmq, RsmqConnection, RsmqMessage};
 use sea_orm::TransactionTrait;
@@ -194,6 +196,14 @@ impl JobWorker {
                     delay: Some(self.state.config.job_prune_session),
                 }
             }
+            Job::PrunePendingUploads => {
+                debug!("Pruning all expired pending uploads from database and S3");
+                BlobService::prune(ctx).await?;
+                NextJob::Next {
+                    job: Job::PrunePendingUploads,
+                    delay: Some(self.state.config.job_prune_uploads),
+                }
+            }
             Job::PruneText => {
                 debug!("Pruning all unused text items from database");
                 TextService::prune(ctx).await?;
@@ -241,7 +251,6 @@ impl JobWorker {
                 debug!("Job execution finished, follow-up job has been produced");
                 trace!("* Job:   {job:?}");
                 trace!("* Delay: {delay:?}");
-
                 JobService::queue_job(ctx, &job, delay).await?;
             }
         }

install/dev/deepwell/deepwell.toml

@@ -35,7 +35,8 @@ max-attempts = 3
 delay-ms = 5
 min-delay-poll-secs = 10  # 10 seconds
 max-delay-poll-secs = 360  # 6 minutes
-prune-session-secs = 600  # 5 minutes
+prune-session-secs = 300  # 5 minutes
+prune-upload-secs = 300  # 5 minutes
 prune-text-secs = 86400  # 1 day
 name-change-refill-secs = 86400  # 1 day
 lift-expired-punishments-secs = 86400  # 1 day