Decrease time to cleanup resumable uploads

server/controllers/api/videos/upload.ts

@@ -8,6 +8,7 @@ import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent'
 import { getLocalVideoActivityPubUrl } from '@server/lib/activitypub/url'
 import { generateWebTorrentVideoFilename } from '@server/lib/paths'
 import { Redis } from '@server/lib/redis'
+import { uploadx } from '@server/lib/uploadx'
 import {
   addMoveToObjectStorageJob,
   addOptimizeOrMergeAudioJob,
@@ -19,7 +20,6 @@ import { VideoPathManager } from '@server/lib/video-path-manager'
 import { buildNextVideoState } from '@server/lib/video-state'
 import { openapiOperationDoc } from '@server/middlewares/doc'
 import { MVideo, MVideoFile, MVideoFullLight } from '@server/types/models'
-import { Uploadx } from '@uploadx/core'
 import { VideoCreate, VideoState } from '../../../../shared'
 import { HttpStatusCode } from '../../../../shared/models'
 import { auditLoggerFactory, getAuditIdFromRes, VideoAuditView } from '../../../helpers/audit-logger'
@@ -41,8 +41,8 @@ import {
   authenticate,
   videosAddLegacyValidator,
   videosAddResumableInitValidator,
-  videosResumableUploadIdValidator,
+  videosAddResumableValidator,
-  videosAddResumableValidator
+  videosResumableUploadIdValidator
 } from '../../../middlewares'
 import { ScheduleVideoUpdateModel } from '../../../models/video/schedule-video-update'
 import { VideoModel } from '../../../models/video/video'
@@ -52,9 +52,6 @@ const lTags = loggerTagsFactory('api', 'video')
 const auditLogger = auditLoggerFactory('videos')
 const uploadRouter = express.Router()
 
-const uploadx = new Uploadx({ directory: getResumableUploadPath() })
-uploadx.getUserId = (_, res: express.Response) => res.locals.oauth?.token.user.id
-
 const reqVideoFileAdd = createReqFiles(
   [ 'videofile', 'thumbnailfile', 'previewfile' ],
   Object.assign({}, MIMETYPES.VIDEO.MIMETYPE_EXT, MIMETYPES.IMAGE.MIMETYPE_EXT),

server/initializers/constants.ts

@@ -223,7 +223,7 @@ const SCHEDULER_INTERVALS_MS = {
   REMOVE_OLD_VIEWS: 60000 * 60 * 24, // 1 day
   REMOVE_OLD_HISTORY: 60000 * 60 * 24, // 1 day
   UPDATE_INBOX_STATS: 1000 * 60, // 1 minute
-  REMOVE_DANGLING_RESUMABLE_UPLOADS: 60000 * 60 * 16 // 16 hours
+  REMOVE_DANGLING_RESUMABLE_UPLOADS: 60000 * 60 // 1 hour
 }
 
 // ---------------------------------------------------------------------------

server/lib/schedulers/remove-dangling-resumable-uploads-scheduler.ts

@@ -1,9 +1,7 @@
-import { map } from 'bluebird'
+
-import { readdir, remove, stat } from 'fs-extra'
 import { logger, loggerTagsFactory } from '@server/helpers/logger'
-import { getResumableUploadPath } from '@server/helpers/upload'
 import { SCHEDULER_INTERVALS_MS } from '@server/initializers/constants'
-import { METAFILE_EXTNAME } from '@uploadx/core'
+import { uploadx } from '../uploadx'
 import { AbstractScheduler } from './abstract-scheduler'
 
 const lTags = loggerTagsFactory('scheduler', 'resumable-upload', 'cleaner')
@@ -22,36 +20,17 @@ export class RemoveDanglingResumableUploadsScheduler extends AbstractScheduler {
   }
 
   protected async internalExecute () {
-    const path = getResumableUploadPath()
+    logger.debug('Removing dangling resumable uploads', lTags())
-    const files = await readdir(path)
 
-    const metafiles = files.filter(f => f.endsWith(METAFILE_EXTNAME))
+    const now = new Date().getTime()
-
-    if (metafiles.length === 0) return
-
-    logger.debug('Reading resumable video upload folder %s with %d files', path, metafiles.length, lTags())
 
     try {
-      await map(metafiles, metafile => {
+      // Remove files that were not updated since the last execution
-        return this.deleteIfOlderThan(metafile, this.lastExecutionTimeMs)
+      await uploadx.storage.purge(now - this.lastExecutionTimeMs)
-      }, { concurrency: 5 })
     } catch (error) {
       logger.error('Failed to handle file during resumable video upload folder cleanup', { error, ...lTags() })
     } finally {
-      this.lastExecutionTimeMs = new Date().getTime()
+      this.lastExecutionTimeMs = now
-    }
-  }
-
-  private async deleteIfOlderThan (metafile: string, olderThan: number) {
-    const metafilePath = getResumableUploadPath(metafile)
-    const statResult = await stat(metafilePath)
-
-    // Delete uploads that started since a long time
-    if (statResult.ctimeMs < olderThan) {
-      await remove(metafilePath)
-
-      const datafile = metafilePath.replace(new RegExp(`${METAFILE_EXTNAME}$`), '')
-      await remove(datafile)
     }
   }
 

server/lib/uploadx.ts

@@ -0,0 +1,10 @@
+import express from 'express'
+import { getResumableUploadPath } from '@server/helpers/upload'
+import { Uploadx } from '@uploadx/core'
+
+const uploadx = new Uploadx({ directory: getResumableUploadPath() })
+uploadx.getUserId = (_, res: express.Response) => res.locals.oauth?.token.user.id
+
+export {
+  uploadx
+}