@ -1,3 +1,4 @@
use crate ::config ::ConfigManager ;
use crate ::inventory ::{ InventoryDb , NewTaskRecord , TaskRecord , TaskStatus , TaskUpdate } ;
use crate ::tasks ::download ::DownloadTask ;
use crate ::tasks ::types ::{ TaskKind , TaskPayload , TaskProgress } ;
@ -6,14 +7,17 @@ use anyhow::{Context, Result, anyhow};
use cloudreve_api ::Client ;
use dashmap ::DashMap ;
use serde_json ::Value ;
use std ::path ::PathBuf ;
use std ::io ::ErrorKind ;
use std ::path ::{ Path , PathBuf } ;
use std ::sync ::Arc ;
use std ::sync ::atomic ::{ AtomicBool , AtomicUsize , Ordering } ;
use std ::time ::{ Duration , SystemTime } ;
use tokio ::sync ::{
Mutex , Notify , Semaphore ,
mpsc ::{ self , UnboundedReceiver , UnboundedSender } ,
} ;
use tokio ::task ::JoinHandle ;
use tokio ::time ::{ Instant , sleep } ;
use tracing ::{ debug , error , info , warn } ;
use uuid ::Uuid ;
@ -48,6 +52,12 @@ pub struct TaskQueue {
task_paths : DashMap < String , String > ,
}
#[ derive(Debug, Clone, PartialEq, Eq) ]
struct FileChangeSignature {
size : u64 ,
modified : SystemTime ,
}
impl TaskQueue {
pub async fn new (
drive_id : impl Into < String > ,
@ -501,6 +511,11 @@ impl TaskQueue {
match & task . payload . kind {
TaskKind ::Upload = > {
match self . wait_for_upload_stability ( task ) . await ? {
TaskRunState ::Cancelled = > return Ok ( TaskRunState ::Cancelled ) ,
TaskRunState ::Completed = > { }
}
let mut task_executor = UploadTask ::new (
self . inventory . clone ( ) ,
self . cr_client . clone ( ) ,
@ -567,6 +582,174 @@ impl TaskQueue {
Ok ( TaskRunState ::Completed )
}
fn is_task_marked_inactive ( & self , task_id : & str ) -> bool {
match self . inventory . get_task_status ( task_id ) {
Ok ( Some ( TaskStatus ::Cancelled ) ) = > true ,
Ok ( Some ( status ) ) = > ! status . is_active ( ) ,
Err ( err ) = > {
warn ! (
target : "tasks::queue" ,
drive = % self . drive_id ,
task_id = % task_id ,
error = % err ,
"Failed to check task status while waiting for delayed upload"
) ;
false
}
_ = > false ,
}
}
fn capture_file_signature ( path : & Path ) -> Result < Option < FileChangeSignature > > {
let metadata = match std ::fs ::metadata ( path ) {
Ok ( metadata ) = > metadata ,
Err ( err ) if err . kind ( ) = = ErrorKind ::NotFound = > return Ok ( None ) ,
Err ( err ) = > {
return Err ( err )
. with_context ( | | format! ( "failed to read metadata for {}" , path . display ( ) ) ) ;
}
} ;
if ! metadata . is_file ( ) {
return Ok ( None ) ;
}
let modified = metadata
. modified ( )
. with_context ( | | format! ( "failed to read modified time for {}" , path . display ( ) ) ) ? ;
Ok ( Some ( FileChangeSignature {
size : metadata . len ( ) ,
modified ,
} ) )
}
async fn wait_for_upload_stability ( & self , task : & QueuedTask ) -> Result < TaskRunState > {
let delay_seconds = ConfigManager ::try_get ( )
. map ( | config | config . sync_delay_seconds ( ) )
. unwrap_or ( 0 ) ;
if delay_seconds = = 0 {
return Ok ( TaskRunState ::Completed ) ;
}
let local_path = & task . payload . local_path ;
let mut signature = match Self ::capture_file_signature ( local_path . as_path ( ) ) {
Ok ( Some ( signature ) ) = > signature ,
Ok ( None ) = > return Ok ( TaskRunState ::Completed ) ,
Err ( err ) = > {
warn ! (
target : "tasks::queue" ,
drive = % self . drive_id ,
task_id = % task . task_id ,
path = % local_path . display ( ) ,
error = % err ,
"Failed to capture initial file signature, skipping delayed upload wait"
) ;
return Ok ( TaskRunState ::Completed ) ;
}
} ;
info ! (
target : "tasks::queue" ,
drive = % self . drive_id ,
task_id = % task . task_id ,
path = % local_path . display ( ) ,
delay_seconds = delay_seconds ,
"Waiting for file to stop changing before upload"
) ;
let required_stable = Duration ::from_secs ( delay_seconds ) ;
// A continuously-growing file must not block a worker slot forever:
// give up waiting after 4x the configured delay (min 10 minutes) and
// upload whatever is on disk, rather than starving the queue.
let max_total_wait = required_stable
. saturating_mul ( 4 )
. max ( Duration ::from_secs ( 600 ) ) ;
let wait_started = Instant ::now ( ) ;
let poll_interval = if delay_seconds < = 30 {
Duration ::from_secs ( 1 )
} else if delay_seconds < = 120 {
Duration ::from_secs ( 2 )
} else {
Duration ::from_secs ( 3 )
} ;
let status_check_interval = Duration ::from_secs ( 5 ) ;
let mut unchanged_since = Instant ::now ( ) ;
let mut next_status_check = Instant ::now ( ) ;
while unchanged_since . elapsed ( ) < required_stable {
if wait_started . elapsed ( ) > = max_total_wait {
warn ! (
target : "tasks::queue" ,
drive = % self . drive_id ,
task_id = % task . task_id ,
path = % local_path . display ( ) ,
"File still changing after maximum wait, proceeding with upload"
) ;
return Ok ( TaskRunState ::Completed ) ;
}
if self . cancel_requested . load ( Ordering ::SeqCst ) {
debug ! (
target : "tasks::queue" ,
drive = % self . drive_id ,
task_id = % task . task_id ,
path = % local_path . display ( ) ,
"Task cancelled while waiting for upload delay"
) ;
return Ok ( TaskRunState ::Cancelled ) ;
}
if Instant ::now ( ) > = next_status_check {
if self . is_task_marked_inactive ( & task . task_id ) {
debug ! (
target : "tasks::queue" ,
drive = % self . drive_id ,
task_id = % task . task_id ,
path = % local_path . display ( ) ,
"Task became inactive while waiting for upload delay"
) ;
return Ok ( TaskRunState ::Cancelled ) ;
}
next_status_check = Instant ::now ( ) + status_check_interval ;
}
sleep ( poll_interval ) . await ;
let current_signature = match Self ::capture_file_signature ( local_path . as_path ( ) ) {
Ok ( Some ( signature ) ) = > signature ,
Ok ( None ) = > return Ok ( TaskRunState ::Completed ) ,
Err ( err ) = > {
warn ! (
target : "tasks::queue" ,
drive = % self . drive_id ,
task_id = % task . task_id ,
path = % local_path . display ( ) ,
error = % err ,
"Failed to capture file signature while waiting, continuing upload"
) ;
return Ok ( TaskRunState ::Completed ) ;
}
} ;
if current_signature ! = signature {
signature = current_signature ;
unchanged_since = Instant ::now ( ) ;
}
}
debug ! (
target : "tasks::queue" ,
drive = % self . drive_id ,
task_id = % task . task_id ,
path = % local_path . display ( ) ,
delay_seconds = delay_seconds ,
"File remained unchanged for configured delay, proceeding with upload"
) ;
Ok ( TaskRunState ::Completed )
}
#[ allow(dead_code) ]
async fn wait_for_idle ( & self ) {
while self . inflight . load ( Ordering ::SeqCst ) > 0 {
@ -716,3 +899,39 @@ pub struct QueuedTask {
pub task_id : String ,
pub payload : TaskPayload ,
}
#[ cfg(test) ]
mod tests {
use super ::* ;
#[ test ]
fn file_signature_captures_size_and_mtime ( ) {
let dir = tempfile ::tempdir ( ) . unwrap ( ) ;
let file = dir . path ( ) . join ( "data.bin" ) ;
std ::fs ::write ( & file , b" hello " ) . unwrap ( ) ;
let signature = TaskQueue ::capture_file_signature ( & file )
. unwrap ( )
. expect ( "file should produce a signature" ) ;
assert_eq! ( signature . size , 5 ) ;
std ::fs ::write ( & file , b" hello world " ) . unwrap ( ) ;
let updated = TaskQueue ::capture_file_signature ( & file )
. unwrap ( )
. unwrap ( ) ;
assert_ne! ( signature , updated ) ;
}
#[ test ]
fn file_signature_none_for_dir_and_missing ( ) {
let dir = tempfile ::tempdir ( ) . unwrap ( ) ;
assert! ( TaskQueue ::capture_file_signature ( dir . path ( ) )
. unwrap ( )
. is_none ( ) ) ;
assert! ( TaskQueue ::capture_file_signature (
& dir . path ( ) . join ( "missing.bin" )
)
. unwrap ( )
. is_none ( ) ) ;
}
}
Tomas Dvorak
2 weeks ago