diff --git a/src/compaction/src/lib.rs b/src/compaction/src/lib.rs index c08936ba2a..7699d72c94 100644 --- a/src/compaction/src/lib.rs +++ b/src/compaction/src/lib.rs @@ -325,10 +325,16 @@ pub async fn run_merge(job_tx: mpsc::Sender) -> Result<(), any need_done_ids.push(job.id); // the data will be deleted by retention, just skip continue; } - // every job of a dynamic stream's closed day reads the files the day's job merges - let dynamic_day = is_past_day(job.offsets) - && get_stream_setting_dynamic_merge(stream_type, &Some(stream_settings.clone())); - if partition_time_level == PartitionTimeLevel::Daily || dynamic_day { + // the day's merge job reads every hour of the day: the day's other jobs go first + if is_dynamic_day_job( + job.offsets, + get_stream_setting_dynamic_merge(stream_type, &Some(stream_settings.as_ref())), + ) && day_jobs_unfinished(&org_id, stream_type, &stream_name, job.offsets).await + { + need_release_ids.push(job.id); + continue; + } + if partition_time_level == PartitionTimeLevel::Daily { // check if this stream need process by this node let Some(node_name) = get_node_from_consistent_hash(&stream_name, &Role::Compactor, None).await @@ -460,6 +466,30 @@ pub async fn run_delay_deletion() -> Result<(), anyhow::Error> { Ok(()) } +/// Whether a job of the day past its 00:00 is pending or running; a failed lookup counts as one. +async fn day_jobs_unfinished( + org_id: &str, + stream_type: StreamType, + stream_name: &str, + day_start: i64, +) -> bool { + let range = (day_start + 1, day_start + day_micros(1) - 1); + match infra_file_list::has_unfinished_jobs(org_id, stream_type, stream_name, range).await { + Ok(unfinished) => unfinished, + Err(e) => { + log::error!( + "[COMPACTOR] has_unfinished_jobs [{org_id}/{stream_type}/{stream_name}] failed: {e}" + ); + true + } + } +} + +/// The 00:00 job of an enabled stream's closed day: the one that merges the day into `DD/00`. +pub(crate) fn is_dynamic_day_job(offset: i64, dynamic_merge: bool) -> bool { + dynamic_merge && offset % day_micros(1) == 0 && is_past_day(offset) +} + /// A day is closed once its last hour is past: no hour of it can receive files any more. pub(crate) fn is_past_day(offset: i64) -> bool { let day = day_micros(1); @@ -472,3 +502,33 @@ pub(crate) fn is_past_hour(offset: i64) -> bool { let offset_hour_end = offset - offset % hour + hour; now - offset_hour_end > second_micros(get_config().limit.max_file_retention_time as i64) * 3 } + +#[cfg(test)] +mod tests { + use super::*; + + /// The day's merge job waits for the day's other jobs, never for its own row or a done one. + #[tokio::test] + async fn test_day_jobs_unfinished_sees_only_the_day_past_its_00_job() { + infra_file_list::create_table().await.unwrap(); + // the sqlite file outlives a test run + let stream = &format!("day_jobs_{}", config::ider::generate()); + let day = day_micros(20_000); + let unfinished = || day_jobs_unfinished("o", StreamType::Logs, stream, day); + + infra_file_list::add_job("o", StreamType::Logs, stream, day) + .await + .unwrap(); + infra_file_list::add_job("o", StreamType::Logs, stream, day + day_micros(1)) + .await + .unwrap(); + assert!(!unfinished().await); + + let hour23 = infra_file_list::add_job("o", StreamType::Logs, stream, day + hour_micros(23)) + .await + .unwrap(); + assert!(unfinished().await); + infra_file_list::set_job_done(&[hour23]).await.unwrap(); + assert!(!unfinished().await); + } +} diff --git a/src/compaction/src/merge/dynamic.rs b/src/compaction/src/merge/dynamic.rs index f27d8ab9d5..8001551ea2 100644 --- a/src/compaction/src/merge/dynamic.rs +++ b/src/compaction/src/merge/dynamic.rs @@ -15,7 +15,10 @@ //! Dynamic merge: what the 00:00 job of a closed day does with one partition of that day. -use config::meta::stream::{FileKey, MergeStrategy}; +use config::{ + meta::stream::{FileKey, MergeStrategy}, + utils::time::{day_micros, hour_micros}, +}; use search::datafusion::merge::MergeMode; use super::{ @@ -30,14 +33,6 @@ const DAY_MERGE_SLACK: usize = 4; /// Position of the hour directory in `files/{org}/{type}/{stream}/YYYY/MM/DD/HH/...`. const HOUR_SEGMENT: usize = 7; -/// What the dynamic merge job does with one partition of its day. -pub(super) enum DayPlan { - /// The day holds many more files than its size needs: these batches go to `DD/00`. - Merge(Vec), - /// The day stays; its hour-00 files still get their hourly merge. - Hour00(Vec), -} - /// `prefix` with its hour directory replaced by `00`, where a merged day is written. pub(super) fn day_prefix(prefix: &str) -> String { prefix @@ -48,9 +43,26 @@ pub(super) fn day_prefix(prefix: &str) -> String { .join("/") } -pub(super) fn plan_day(files: Vec, mode: &MergeMode, limits: &BatchLimits<'_>) -> DayPlan { - if !should_merge_day(&files, limits.max_file_size) { - return DayPlan::Hour00(hour00_files(files, mode, limits)); +/// Whether the day was merged by day: only that writes a file of `DD/00` reaching past the hour. +pub(super) fn merged_by_day(files: &[FileKey]) -> bool { + files.iter().any(day_output) +} + +/// Batches of one partition of the day, all written to `DD/00`; none when the day stays as it is. +pub(super) fn plan_day( + files: Vec, + mode: &MergeMode, + limits: &BatchLimits<'_>, +) -> Vec { + let outputs = files.iter().filter(|f| day_output(f)).count(); + let act = if outputs > 0 { + // a merged day stays in one place: what arrived since joins it, whatever its count + outputs < files.len() + } else { + should_merge_day(&files, limits.max_file_size) + }; + if !act { + return Vec::new(); } let (mode_files, legacy_files) = split_legacy_metrics(files, mode); let mut batches: Vec = time_ordered_groups(legacy_files, limits) @@ -69,31 +81,19 @@ pub(super) fn plan_day(files: Vec, mode: &MergeMode, limits: &BatchLimi .map(|group| (group, mode.clone())), ); } - DayPlan::Merge(batches) + batches +} + +/// A file of `DD/00` reaching past its hour; no file reaches past its day. +fn day_output(file: &FileKey) -> bool { + file.key.split('/').nth(HOUR_SEGMENT) == Some("00") + && file.meta.max_ts % day_micros(1) >= hour_micros(1) } fn should_merge_day(files: &[FileKey], max_file_size: usize) -> bool { files.len() > DAY_MERGE_FACTOR * ideal_file_count(files, max_file_size) + DAY_MERGE_SLACK } -/// The files an hourly merge of hour 00 would have queried. -fn hour00_files( - mut files: Vec, - mode: &MergeMode, - limits: &BatchLimits<'_>, -) -> Vec { - let max_original_size = if mode.merges_whole_batch() { - i64::MAX - } else { - limits.merge_max_original_size - }; - files.retain(|f| { - f.key.split('/').nth(HOUR_SEGMENT) == Some("00") - && f.meta.original_size <= max_original_size - }); - files -} - /// Size-bounded groups in time order, so each output covers a contiguous range of the day. fn time_ordered_groups(mut files: Vec, limits: &BatchLimits<'_>) -> Vec> { files.retain(|f| f.meta.original_size <= limits.merge_max_original_size); @@ -108,7 +108,7 @@ fn time_ordered_groups(mut files: Vec, limits: &BatchLimits<'_>) -> Vec #[cfg(test)] mod tests { - use config::{meta::stream::FileMeta, utils::time::hour_micros}; + use config::meta::stream::FileMeta; use super::*; @@ -146,18 +146,19 @@ mod tests { .collect() } - fn merged(plan: DayPlan) -> Vec { - match plan { - DayPlan::Merge(batches) => batches, - DayPlan::Hour00(_) => panic!("the day was not merged"), - } - } - - fn hour00(plan: DayPlan) -> Vec { - match plan { - DayPlan::Hour00(files) => files, - DayPlan::Merge(_) => panic!("the day was merged"), - } + /// What a merged day looks like: each batch became one file in `DD/00` spanning its hours. + fn outputs(batches: &[PlannedBatch]) -> Vec { + batches + .iter() + .enumerate() + .map(|(i, (files, _))| { + let mut out = file("logs", 0, &format!("merged{i}"), 0); + out.meta.min_ts = files.iter().map(|f| f.meta.min_ts).min().unwrap(); + out.meta.max_ts = files.iter().map(|f| f.meta.max_ts).max().unwrap(); + out.meta.original_size = files.iter().map(|f| f.meta.original_size).sum(); + out + }) + .collect() } #[test] @@ -172,6 +173,18 @@ mod tests { ); } + /// Only a file of hour 00 reaching past the hour marks a merged day. + #[test] + fn test_merged_by_day_needs_a_file_of_hour_00_past_the_hour() { + assert!(!merged_by_day(&day("logs", "f", MB))); + let mut spanning = file("logs", 5, "late", MB); + spanning.meta.max_ts = hour_micros(23); + assert!(!merged_by_day(&[spanning])); + let mut merged = file("logs", 0, "merged", MB); + merged.meta.max_ts = hour_micros(23); + assert!(merged_by_day(&[merged])); + } + /// The design's table: one file per hour against the default 2 GB target. #[test] fn test_should_merge_day_follows_the_data_volume() { @@ -205,7 +218,7 @@ mod tests { let mut files = day("logs", "f", 300 * MB); files.reverse(); files.swap(3, 17); - let batches = merged(plan_day(files, &MergeMode::Classic, &limits)); + let batches = plan_day(files, &MergeMode::Classic, &limits); assert_eq!(batches.len(), 4); let ranges: Vec<(i64, i64)> = batches .iter() @@ -220,47 +233,62 @@ mod tests { assert!(batches.iter().all(|(files, _)| files.len() == 6)); } + /// A day with the files its size needs stays as it is, late data or not. + #[test] + fn test_plan_day_leaves_a_large_day_alone() { + let strategy = MergeStrategy::FileTime; + let limits = limits(&strategy); + let mut files = day("logs", "f", GB); + files.push(file("logs", 0, "late", MB)); + files.push(file("logs", 5, "late", MB)); + assert!(plan_day(files, &MergeMode::Classic, &limits).is_empty()); + } + /// Running the job again on the merged day does nothing: the policy is idempotent. #[test] fn test_plan_day_second_run_leaves_a_merged_day_alone() { let strategy = MergeStrategy::FileTime; let limits = limits(&strategy); - let batches = merged(plan_day( + let batches = plan_day(day("logs", "f", 300 * MB), &MergeMode::Classic, &limits); + let merged = outputs(&batches); + assert_eq!(merged.len(), 4); + assert!(plan_day(merged, &MergeMode::Classic, &limits).is_empty()); + } + + /// A merged day takes its late files without the threshold; they join an unfilled output. + #[test] + fn test_plan_day_merged_day_takes_late_files_into_hour_00() { + let strategy = MergeStrategy::FileTime; + let limits = limits(&strategy); + let mut merged = outputs(&plan_day( day("logs", "f", 300 * MB), &MergeMode::Classic, &limits, )); - // each batch becomes one file in DD/00 that spans the batch's hours - let outputs: Vec = batches - .iter() - .enumerate() - .map(|(i, (files, _))| { - let mut out = file("logs", 0, &format!("merged{i}"), 0); - out.meta.min_ts = files.iter().map(|f| f.meta.min_ts).min().unwrap(); - out.meta.max_ts = files.iter().map(|f| f.meta.max_ts).max().unwrap(); - out.meta.original_size = files.iter().map(|f| f.meta.original_size).sum(); - out - }) - .collect(); - let left = hour00(plan_day(outputs, &MergeMode::Classic, &limits)); - assert_eq!(left.len(), 4); - } - - #[test] - fn test_plan_day_skip_keeps_only_the_hour_00_merge_candidates() { - let strategy = MergeStrategy::FileTime; - let limits = limits(&strategy); - // a large stream, plus late data and a full-size file in hour 00 - let mut files = day("logs", "f", GB); - files.push(file("logs", 0, "late", MB)); - files.push(file("logs", 0, "full", 2 * GB)); - files.push(file("logs", 5, "late", MB)); - let files = hour00(plan_day(files, &MergeMode::Classic, &limits)); - let names: Vec<&str> = files + // the first three outputs are full, the last has room + for out in merged.iter_mut().take(3) { + out.meta.original_size = 2 * GB; + } + let mut files = merged.clone(); + files.push(file("logs", 5, "late5", MB)); + files.push(file("logs", 9, "late9", MB)); + let batches = plan_day(files, &MergeMode::Classic, &limits); + assert_eq!(batches.len(), 1); + let names: Vec<&str> = batches[0] + .0 .iter() .map(|f| f.key.rsplit('/').next().unwrap()) .collect(); - assert_eq!(names, vec!["f0.parquet", "late.parquet"]); + assert_eq!( + names, + vec!["late5.parquet", "late9.parquet", "merged3.parquet"] + ); + // a single late file with no output to join stays where it is + for out in merged.iter_mut() { + out.meta.original_size = 2 * GB; + } + merged.push(file("logs", 5, "late5", MB)); + assert!(plan_day(merged, &MergeMode::Classic, &limits).is_empty()); } /// Indexed metrics keep every series in one batch; legacy files never join it. @@ -271,7 +299,7 @@ mod tests { let mut files = day("metrics", "indexed-v1-", 100 * MB); files.push(file("metrics", 3, "legacy-a", MB)); files.push(file("metrics", 9, "legacy-b", MB)); - let batches = merged(plan_day(files, &MergeMode::MetricsIndexed, &limits)); + let batches = plan_day(files, &MergeMode::MetricsIndexed, &limits); assert_eq!(batches.len(), 2); let (legacy, indexed): (Vec<_>, Vec<_>) = batches .iter() diff --git a/src/compaction/src/merge/job.rs b/src/compaction/src/merge/job.rs index 3bf5143cbd..8ed167eef2 100644 --- a/src/compaction/src/merge/job.rs +++ b/src/compaction/src/merge/job.rs @@ -13,19 +13,26 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use std::collections::{BTreeSet, HashMap}; + use chrono::{DateTime, Datelike, Duration, TimeZone, Utc}; use config::{ cluster::LOCAL_NODE, get_config, meta::stream::{PartitionTimeLevel, StreamType}, - utils::time::{day_micros, hour_micros}, + utils::time::{HourFormat, day_micros, get_ymdh_from_micros, hour_micros}, }; use infra::{ cluster::get_node_by_uuid, dist_lock, file_list as infra_file_list, - schema::{get_dynamic_merge, get_partition_time_level, unwrap_stream_created_at}, + schema::{ + get_dynamic_merge, get_partition_time_level, get_stream_setting_dynamic_merge, + unwrap_stream_created_at, + }, }; +use super::dynamic::merged_by_day; + /// What the merge job at `offset` covers. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(super) enum JobScope { @@ -245,6 +252,10 @@ pub async fn generate_old_data_job_by_stream( .await?; // generate merging job + let dynamic_merge = + get_stream_setting_dynamic_merge(stream_type, &Some(stream_settings.as_ref())); + let mut merged_days = HashMap::new(); + let mut offsets = BTreeSet::new(); for hour in hours { let column = hour.split('/').collect::>(); if column.len() != 4 { @@ -257,10 +268,18 @@ pub async fn generate_old_data_job_by_stream( column[0], column[1], column[2], column[3] ))? .with_timezone(&Utc); - let offset = offset.timestamp_micros(); + let mut offset = offset.timestamp_micros(); + if dynamic_merge { + offset = + merged_day_job_offset(org_id, stream_type, stream_name, offset, &mut merged_days) + .await?; + } log::debug!( "[COMPACTOR] generate_old_data_job_by_stream [{org_id}/{stream_type}/{stream_name}] hours: {hour}, offset: {offset}" ); + offsets.insert(offset); + } + for offset in offsets { if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await { return Err(anyhow::anyhow!( "[COMPACTOR] add file_list_jobs for old data failed: {e}" @@ -391,8 +410,40 @@ fn dynamic_merge_offset(offset: i64) -> Option { (offset % day_micros(1) == 0).then(|| offset - day_micros(1)) } +/// An old hour of a closed day merged by day re-arms the day's 00:00 job: its files go to `DD/00`. +async fn merged_day_job_offset( + org_id: &str, + stream_type: StreamType, + stream_name: &str, + offset: i64, + merged_days: &mut HashMap, +) -> Result { + let day = offset - offset % day_micros(1); + if offset == day || !crate::is_past_day(offset) { + return Ok(offset); + } + let merged = match merged_days.get(&day) { + Some(merged) => *merged, + None => { + let hour00 = get_ymdh_from_micros(day, HourFormat::Zero); + let files = infra_file_list::query_for_merge( + org_id, + stream_type, + stream_name, + (hour00.clone(), hour00), + i64::MAX, + ) + .await?; + *merged_days.entry(day).or_insert(merged_by_day(&files)) + } + }; + Ok(if merged { day } else { offset }) +} + #[cfg(test)] mod tests { + use config::meta::stream::{FileKey, FileMeta}; + use super::*; #[test] @@ -471,4 +522,45 @@ mod tests { assert_eq!(dynamic_merge_offset(day), Some(day - day_micros(1))); assert_eq!(dynamic_merge_offset(day + hour_micros(1)), None); } + + /// An old hour of a day merged by day re-arms the day's 00:00 job instead of its own. + #[tokio::test] + async fn test_merged_day_job_offset_follows_the_files_of_hour_00() { + infra_file_list::create_table().await.unwrap(); + // the sqlite file outlives a test run + let stream = &format!("merged_day_{}", config::ider::generate()); + let day = day_start(); + let hour15 = day + hour_micros(15); + let hour00_file = |name: &str, max_ts: i64| { + FileKey::new( + 0, + "default".to_string(), + format!("files/old/logs/{stream}/2026/08/18/00/{name}.parquet"), + FileMeta { + min_ts: day + 1, + max_ts, + records: 1, + original_size: 1, + compressed_size: 1, + ..Default::default() + }, + false, + ) + }; + let offset = |offset: i64| async move { + merged_day_job_offset("old", StreamType::Logs, stream, offset, &mut HashMap::new()) + .await + .unwrap() + }; + + // an hourly file of hour 00 does not make a merged day + let hourly = hour00_file("hourly", day + hour_micros(1) - 2); + infra_file_list::batch_add(&[hourly]).await.unwrap(); + assert_eq!(offset(hour15).await, hour15); + + let merged = hour00_file("merged", day + hour_micros(24) - 2); + infra_file_list::batch_add(&[merged]).await.unwrap(); + assert_eq!(offset(hour15).await, day); + assert_eq!(offset(day).await, day); + } } diff --git a/src/compaction/src/merge/stream.rs b/src/compaction/src/merge/stream.rs index b7655abed8..d0f75b4460 100644 --- a/src/compaction/src/merge/stream.rs +++ b/src/compaction/src/merge/stream.rs @@ -19,10 +19,7 @@ use config::{ cluster, get_config, meta::stream::{FileKey, FileListDeleted, MergeStrategy, PartitionTimeLevel, StreamType}, metrics, - utils::{ - parquet::parse_file_key_columns, - time::{day_micros, hour_micros}, - }, + utils::{parquet::parse_file_key_columns, time::hour_micros}, }; use hashbrown::{HashMap, HashSet}; use infra::{cache::file_data, file_list as infra_file_list}; @@ -36,7 +33,7 @@ use tokio::{ }; use super::{ - dynamic::{DayPlan, day_prefix, plan_day}, + dynamic::{day_prefix, plan_day}, job::{JobScope, job_range_end, job_scope}, plan::{BatchLimits, PlannedBatch, plan_batches}, }; @@ -89,9 +86,10 @@ pub async fn merge_by_stream( let is_incremental = !crate::is_past_hour(offset); // check offset - let dynamic_day = offset % day_micros(1) == 0 - && crate::is_past_day(offset) - && infra::schema::get_dynamic_merge(org_id, stream_name, stream_type).await; + let dynamic_day = crate::is_dynamic_day_job( + offset, + infra::schema::get_dynamic_merge(org_id, stream_name, stream_type).await, + ); let scope = job_scope( stream_type, offset, @@ -443,27 +441,22 @@ fn plan_partition( stream: &str, prefix: &str, ) -> Vec { - let hourly = |files| { - plan_batches( + if scope != JobScope::DynamicDay { + return plan_batches( sort_by_strategy(files, limits.strategy), mode, limits, stream, - ) - }; - if scope != JobScope::DynamicDay { - return hourly(files); + ); } let day_files = files.len(); - match plan_day(files, mode, limits) { - DayPlan::Merge(batches) => { - log::info!( - "[COMPACTOR] merge_by_stream [{stream}] dynamic merge of {day_files} files into {prefix}" - ); - batches - } - DayPlan::Hour00(files) => hourly(files), + let batches = plan_day(files, mode, limits); + if !batches.is_empty() { + log::info!( + "[COMPACTOR] merge_by_stream [{stream}] dynamic merge of {day_files} files into {prefix}" + ); } + batches } /// The order the hourly planner groups files in. diff --git a/src/infra/src/file_list/mod.rs b/src/infra/src/file_list/mod.rs index ff9aa2100c..177ce3c61f 100644 --- a/src/infra/src/file_list/mod.rs +++ b/src/infra/src/file_list/mod.rs @@ -277,6 +277,13 @@ pub trait FileList: Sync + Send + 'static { fast_mode: bool, ) -> Result>; async fn get_pending_jobs_count(&self) -> Result>>; + async fn has_unfinished_jobs( + &self, + org_id: &str, + stream_type: StreamType, + stream: &str, + offset_range: (i64, i64), + ) -> Result; async fn set_job_pending(&self, ids: &[i64], offsets: i64, stream: Option<&str>) -> Result; async fn set_job_done(&self, ids: &[i64]) -> Result<()>; @@ -777,6 +784,19 @@ pub async fn get_pending_jobs_count() -> Result Result { + CLIENT + .has_unfinished_jobs(org_id, stream_type, stream, offset_range) + .await +} + #[inline] pub async fn set_job_pending(ids: &[i64], offsets: i64, stream: Option<&str>) -> Result { CLIENT.set_job_pending(ids, offsets, stream).await diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs index 3d144f9465..8a7939a068 100644 --- a/src/infra/src/file_list/postgres.rs +++ b/src/infra/src/file_list/postgres.rs @@ -1564,6 +1564,30 @@ DO UPDATE SET Ok(ret) } + async fn has_unfinished_jobs( + &self, + org_id: &str, + stream_type: StreamType, + stream: &str, + offset_range: (i64, i64), + ) -> Result { + let stream_key = format!("{org_id}/{stream_type}/{stream}"); + let pool = CLIENT_RO.clone(); + DB_QUERY_NUMS + .with_label_values(&["query", "file_list_jobs"]) + .inc(); + let ret = sqlx::query( + "SELECT id FROM file_list_jobs WHERE stream = $1 AND offsets >= $2 AND offsets <= $3 AND status <> $4 LIMIT 1;", + ) + .bind(stream_key) + .bind(offset_range.0) + .bind(offset_range.1) + .bind(super::FileListJobStatus::Done) + .fetch_optional(&pool) + .await?; + Ok(ret.is_some()) + } + async fn set_job_pending( &self, ids: &[i64], diff --git a/src/infra/src/file_list/sqlite.rs b/src/infra/src/file_list/sqlite.rs index 773710b2ed..c6f5bd5993 100644 --- a/src/infra/src/file_list/sqlite.rs +++ b/src/infra/src/file_list/sqlite.rs @@ -1230,6 +1230,27 @@ SELECT stream, max(id) as id, COUNT(*) AS num Ok(ret) } + async fn has_unfinished_jobs( + &self, + org_id: &str, + stream_type: StreamType, + stream: &str, + offset_range: (i64, i64), + ) -> Result { + let stream_key = format!("{org_id}/{stream_type}/{stream}"); + let client = CLIENT_RO.clone(); + let ret = sqlx::query( + "SELECT id FROM file_list_jobs WHERE stream = $1 AND offsets >= $2 AND offsets <= $3 AND status <> $4 LIMIT 1;", + ) + .bind(stream_key) + .bind(offset_range.0) + .bind(offset_range.1) + .bind(super::FileListJobStatus::Done) + .fetch_optional(&client) + .await?; + Ok(ret.is_some()) + } + async fn set_job_pending( &self, ids: &[i64],