From 233af41545704e335c9195205f3760150e9d4f06 Mon Sep 17 00:00:00 2001 From: Ada Bohm Date: Fri, 23 Oct 2026 15:46:06 +0200 Subject: [PATCH 1/2] Auto forget --- CHANGELOG.md | 2 + .../hyperqueue/src/client/commands/server.rs | 8 ++++ crates/hyperqueue/src/server/autoforget.rs | 39 ++++++++++++++++ crates/hyperqueue/src/server/bootstrap.rs | 13 ++++++ crates/hyperqueue/src/server/client/mod.rs | 45 +++++++++++-------- crates/hyperqueue/src/server/mod.rs | 1 + crates/hyperqueue/src/tests/server.rs | 1 + crates/pyhq/src/cluster/server.rs | 1 + docs/jobs/jobs.md | 13 ++++++ tests/job/test_job_forget.py | 19 ++++++++ 10 files changed, 124 insertions(+), 18 deletions(-) create mode 100644 crates/hyperqueue/src/server/autoforget.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 23f51254a..8633a933d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ ### New features * Python API `crash_limit` now accepts `"unlimited"` and `"never-restart"` (and `0` is rejected) +* `hq server start --auto-forget=` makes the server automatically forget completed jobs + that finished more than the given duration ago (e.g. `--auto-forget=7d`) ## Fixes diff --git a/crates/hyperqueue/src/client/commands/server.rs b/crates/hyperqueue/src/client/commands/server.rs index 8232bd227..9670a032f 100644 --- a/crates/hyperqueue/src/client/commands/server.rs +++ b/crates/hyperqueue/src/client/commands/server.rs @@ -138,6 +138,13 @@ pub struct ServerStartOpts { help = duration_doc!("Maximum time the scheduler's placement solve may run per round.") )] scheduler_time_limit: Duration, + + #[arg( + long, + value_parser = parse_hms_or_human_time, + help = duration_doc!("Automatically forget completed jobs that finished more than this duration ago.") + )] + auto_forget: Option, } #[derive(Parser)] @@ -231,6 +238,7 @@ async fn start_server(gsettings: &GlobalSettings, opts: ServerStartOpts) -> anyh }), server_uid: access_file.as_ref().map(|a| a.server_uid().to_string()), scheduler_mip_time_limit: opts.scheduler_time_limit, + auto_forget: opts.auto_forget, }; init_hq_server(gsettings, server_cfg).await diff --git a/crates/hyperqueue/src/server/autoforget.rs b/crates/hyperqueue/src/server/autoforget.rs new file mode 100644 index 000000000..b1d51220e --- /dev/null +++ b/crates/hyperqueue/src/server/autoforget.rs @@ -0,0 +1,39 @@ +use crate::server::Senders; +use crate::server::client::forget_jobs; +use crate::server::state::StateRef; +use chrono::{TimeDelta, Utc}; +use std::time::Duration; +use tako::JobId; + +pub async fn auto_forget_process(state_ref: StateRef, senders: Senders, max_age: Duration) { + let max_age_delta = TimeDelta::from_std(max_age).unwrap_or(TimeDelta::MAX); + let check_period = (max_age / 10).clamp(Duration::from_secs(1), Duration::from_secs(3600)); + let mut interval = tokio::time::interval(check_period); + loop { + interval.tick().await; + let forgotten_jobs = { + let mut state = state_ref.get_mut(); + let now = Utc::now(); + let job_ids: Vec = state + .jobs() + .filter(|job| { + job.is_terminated() + && job + .completion_date + .is_some_and(|date| now - date >= max_age_delta) + }) + .map(|job| job.job_id) + .collect(); + if job_ids.is_empty() { + continue; + } + log::info!("Auto-forgetting {} job(s)", job_ids.len()); + forget_jobs(&mut state, &senders, &job_ids) + }; + // Dropping jobs can take a long time, so we do it outside the event loop. + let _ = tokio::task::spawn_blocking(move || { + drop(forgotten_jobs); + }) + .await; + } +} diff --git a/crates/hyperqueue/src/server/bootstrap.rs b/crates/hyperqueue/src/server/bootstrap.rs index ad9c589f6..0c698a19e 100644 --- a/crates/hyperqueue/src/server/bootstrap.rs +++ b/crates/hyperqueue/src/server/bootstrap.rs @@ -17,6 +17,7 @@ use crate::common::serverdir::{ }; use crate::server::Senders; use crate::server::autoalloc::{QueueId, create_autoalloc_service}; +use crate::server::autoforget::auto_forget_process; use crate::server::event::journal::JournalWriter; use crate::server::event::journal::start_event_streaming; use crate::server::event::streamer::EventStreamer; @@ -50,6 +51,8 @@ pub struct ServerConfig { pub server_uid: Option, /// See `tako::server::SchedulerConfig::mip_time_limit`. pub scheduler_mip_time_limit: Duration, + /// Completed jobs older than this duration are automatically forgotten. + pub auto_forget: Option, } /// This function initializes the HQ server. @@ -267,6 +270,15 @@ pub async fn initialize_server( senders.clone(), ))); + let auto_forget_fut: Pin>> = match server_cfg.auto_forget { + Some(max_age) => Box::pin(auto_forget_process( + state_ref.clone(), + senders.clone(), + max_age, + )), + None => Box::pin(futures::future::pending()), + }; + let senders2 = senders.clone(); let fut = async move { tokio::pin! { @@ -287,6 +299,7 @@ pub async fn initialize_server( key ) => { Ok(()) } _ = &mut autoalloc_process => { Ok(()) } + _ = auto_forget_fut => { Ok(()) } r = comm_fut => { r.map_err(|e| e.into()) } }; diff --git a/crates/hyperqueue/src/server/client/mod.rs b/crates/hyperqueue/src/server/client/mod.rs index 464248400..8f303ac87 100644 --- a/crates/hyperqueue/src/server/client/mod.rs +++ b/crates/hyperqueue/src/server/client/mod.rs @@ -797,25 +797,22 @@ fn handle_job_forget( ) -> (ToClientMessage, Vec) { let mut state = state_ref.get_mut(); let job_ids: Vec = get_job_ids(&state, selector); - let mut forgotten: usize = 0; - let mut forgotten_jobs = Vec::with_capacity(job_ids.len()); - for &job_id in &job_ids { - let can_be_forgotten = state - .get_job(job_id) - .map(|j| { - j.is_terminated() && allowed_statuses.contains(&job_status(&j.make_job_info(false))) - }) - .unwrap_or(false); - if can_be_forgotten { - if let Some(job) = state.forget_job(job_id) { - forgotten_jobs.push(job); - } - forgotten += 1; - } - } - state.try_release_memory(); - senders.server_control.try_release_memory(); + let to_forget: Vec = job_ids + .iter() + .copied() + .filter(|&job_id| { + state + .get_job(job_id) + .map(|j| { + j.is_terminated() + && allowed_statuses.contains(&job_status(&j.make_job_info(false))) + }) + .unwrap_or(false) + }) + .collect(); + let forgotten = to_forget.len(); + let forgotten_jobs = forget_jobs(&mut state, senders, &to_forget); let ignored = job_ids.len() - forgotten; @@ -825,6 +822,18 @@ fn handle_job_forget( ) } +/// Removes the given (terminated) jobs from the state and releases memory. +/// The forgotten jobs are returned, so that the caller can drop them outside the event loop. +pub(crate) fn forget_jobs(state: &mut State, senders: &Senders, job_ids: &[JobId]) -> Vec { + let forgotten_jobs: Vec = job_ids + .iter() + .filter_map(|&job_id| state.forget_job(job_id)) + .collect(); + state.try_release_memory(); + senders.server_control.try_release_memory(); + forgotten_jobs +} + fn handle_get_list(state_ref: &StateRef, workers: bool) -> ToClientMessage { let state = state_ref.get(); diff --git a/crates/hyperqueue/src/server/mod.rs b/crates/hyperqueue/src/server/mod.rs index e7b95eb6b..b33b6ae3f 100644 --- a/crates/hyperqueue/src/server/mod.rs +++ b/crates/hyperqueue/src/server/mod.rs @@ -3,6 +3,7 @@ use crate::server::event::streamer::EventStreamer; use tako::control::ServerRef; pub mod autoalloc; +mod autoforget; pub mod bootstrap; pub mod client; pub mod event; diff --git a/crates/hyperqueue/src/tests/server.rs b/crates/hyperqueue/src/tests/server.rs index 9157c7b11..9ac4b9d09 100644 --- a/crates/hyperqueue/src/tests/server.rs +++ b/crates/hyperqueue/src/tests/server.rs @@ -58,6 +58,7 @@ where client_secret_key: None, server_uid: None, scheduler_mip_time_limit: Duration::from_secs(5), + auto_forget: None, }; let (fut, notify, _state, _senders) = initialize_server(&gsettings, server_cfg, 1.into(), 1, None) diff --git a/crates/pyhq/src/cluster/server.rs b/crates/pyhq/src/cluster/server.rs index 1b0a80fe6..955d33293 100644 --- a/crates/pyhq/src/cluster/server.rs +++ b/crates/pyhq/src/cluster/server.rs @@ -41,6 +41,7 @@ impl RunningServer { server_uid: None, journal_flush_period: Duration::from_secs(30), scheduler_mip_time_limit: Duration::from_secs(5), + auto_forget: None, }; let main_future = async move { diff --git a/docs/jobs/jobs.md b/docs/jobs/jobs.md index 84afeb063..04861d093 100644 --- a/docs/jobs/jobs.md +++ b/docs/jobs/jobs.md @@ -314,6 +314,19 @@ $ hq job forget all --status finished,aborted,canceled However, only jobs that are completed, i.e. that have been finished successfully, failed, aborted or have been canceled, can be forgotten. If you want to forget a waiting or a running job, [cancel](#cancelling-jobs) it first. +### Automatic forgetting + +Instead of calling `hq job forget` periodically, you can let the server forget old jobs automatically +by starting it with the `--auto-forget` option: + +```console +$ hq server start --auto-forget=7d +``` + +The server then forgets every completed job (finished/failed/aborted/canceled) that was completed +more than the given duration ago. Open jobs are never forgotten automatically; they have to be closed +and completed first. + Note that if you are using a journal, forgetting only free the memory of the server but the tasks remains in journal, run [`hq journal prune`](cli:hq.journal.prune) to remove completed jobs and workers from journal file. diff --git a/tests/job/test_job_forget.py b/tests/job/test_job_forget.py index 11f4a92c5..623074021 100644 --- a/tests/job/test_job_forget.py +++ b/tests/job/test_job_forget.py @@ -1,3 +1,4 @@ +import time from typing import List, Optional from ..conftest import HqEnv @@ -65,6 +66,24 @@ def test_forget_multiple_jobs(hq_env: HqEnv): wait_for_job_list_count(hq_env, 1) +def test_auto_forget(hq_env: HqEnv): + hq_env.start_server(args=["--auto-forget=3s"]) + hq_env.start_worker() + hq_env.command(["submit", "uname"]) + hq_env.command(["submit", "/non-existent"]) + hq_env.command(["submit", "sleep", "100"]) + hq_env.command(["job", "open"]) + wait_for_job_state(hq_env, 1, "FINISHED") + wait_for_job_state(hq_env, 2, "FAILED") + + # Jobs are not forgotten before they are old enough + wait_for_job_list_count(hq_env, 4) + # Running and open jobs are never forgotten + wait_for_job_list_count(hq_env, 2) + time.sleep(4) + wait_for_job_list_count(hq_env, 2) + + def forget_jobs( hq_env: HqEnv, selector: str, From f1ccd673249e37239fea96a0c3b44098b607c83b Mon Sep 17 00:00:00 2001 From: Ada Bohm Date: Tue, 27 Oct 2026 09:51:21 +0100 Subject: [PATCH 2/2] Periodic journal pruning --- CHANGELOG.md | 5 +- .../src/client/commands/journal/output.rs | 10 +++ .../src/client/commands/journal/report.rs | 3 +- .../hyperqueue/src/client/commands/server.rs | 9 +++ crates/hyperqueue/src/client/commands/wait.rs | 3 +- .../dashboard/data/timelines/job_timeline.rs | 3 +- .../data/timelines/worker_timeline.rs | 3 +- .../server/{autoforget.rs => autoclean.rs} | 17 ++++- crates/hyperqueue/src/server/bootstrap.rs | 14 +++- crates/hyperqueue/src/server/client/mod.rs | 30 ++++++-- .../src/server/event/journal/mod.rs | 2 +- .../src/server/event/journal/prune.rs | 13 ++++ .../src/server/event/journal/stream.rs | 6 +- crates/hyperqueue/src/server/event/payload.rs | 7 ++ .../hyperqueue/src/server/event/streamer.rs | 8 ++- crates/hyperqueue/src/server/mod.rs | 2 +- crates/hyperqueue/src/server/restore.rs | 7 ++ crates/hyperqueue/src/server/state.rs | 11 ++- crates/hyperqueue/src/tests/server.rs | 1 + crates/pyhq/src/cluster/server.rs | 1 + docs/deployment/server.md | 11 +++ docs/jobs/jobs.md | 4 ++ tests/test_journal.py | 72 +++++++++++++++++++ 23 files changed, 224 insertions(+), 18 deletions(-) rename crates/hyperqueue/src/server/{autoforget.rs => autoclean.rs} (75%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8633a933d..e40e27c84 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,10 +7,13 @@ * Python API `crash_limit` now accepts `"unlimited"` and `"never-restart"` (and `0` is rejected) * `hq server start --auto-forget=` makes the server automatically forget completed jobs that finished more than the given duration ago (e.g. `--auto-forget=7d`) +* `hq server start --journal-prune-period=` makes the server periodically prune the journal -## Fixes +### Fixes * Fixed crash in auto alloc query in the case of multi-node tasks +* Job and worker IDs are no longer reused after the server is restarted from a pruned journal + ## v0.27.0 diff --git a/crates/hyperqueue/src/client/commands/journal/output.rs b/crates/hyperqueue/src/client/commands/journal/output.rs index 0fbba9734..ad121380a 100644 --- a/crates/hyperqueue/src/client/commands/journal/output.rs +++ b/crates/hyperqueue/src/client/commands/journal/output.rs @@ -146,6 +146,16 @@ fn format_payload(event: EventPayload) -> serde_json::Value { "type": "server-stop", }) } + EventPayload::IdCounters { + max_job_id, + max_worker_id, + } => { + json!({ + "type": "id-counters", + "max_job_id": max_job_id, + "max_worker_id": max_worker_id, + }) + } EventPayload::JobOpen(job_id, job_desc) => { json!({ "type": "job-open", diff --git a/crates/hyperqueue/src/client/commands/journal/report.rs b/crates/hyperqueue/src/client/commands/journal/report.rs index e0195106a..91be67685 100644 --- a/crates/hyperqueue/src/client/commands/journal/report.rs +++ b/crates/hyperqueue/src/client/commands/journal/report.rs @@ -330,7 +330,8 @@ impl JournalStats { EventPayload::AllocationStarted(_, _) | EventPayload::JobIdle(_) | EventPayload::TaskNotify(_) - | EventPayload::ServerStart { .. } => {} + | EventPayload::ServerStart { .. } + | EventPayload::IdCounters { .. } => {} EventPayload::ServerStop => { for trace in jstats.running_workers.values_mut() { trace.add_value(time, 0); diff --git a/crates/hyperqueue/src/client/commands/server.rs b/crates/hyperqueue/src/client/commands/server.rs index 9670a032f..9c07ee012 100644 --- a/crates/hyperqueue/src/client/commands/server.rs +++ b/crates/hyperqueue/src/client/commands/server.rs @@ -113,6 +113,14 @@ pub struct ServerStartOpts { )] journal_flush_period: Duration, + #[arg( + long, + requires = "journal", + value_parser = parse_hms_or_human_time, + help = duration_doc!("Periodically prune the journal (remove completed jobs and lost workers).\n\nIf `--auto-forget` is set, only completed jobs that were already forgotten are removed.\n\nIf not set, the journal is never pruned automatically.") + )] + journal_prune_period: Option, + /// The path to an access file #[arg(long)] access_file: Option, @@ -216,6 +224,7 @@ async fn start_server(gsettings: &GlobalSettings, opts: ServerStartOpts) -> anyh worker_port, journal_path: opts.journal, journal_flush_period: opts.journal_flush_period, + journal_prune_period: opts.journal_prune_period, worker_secret_key: access_file .as_ref() .map(|a| a.worker_key().cloned()) diff --git a/crates/hyperqueue/src/client/commands/wait.rs b/crates/hyperqueue/src/client/commands/wait.rs index eac3a0e0a..7f2450da2 100644 --- a/crates/hyperqueue/src/client/commands/wait.rs +++ b/crates/hyperqueue/src/client/commands/wait.rs @@ -212,7 +212,8 @@ pub async fn wait_for_jobs_with_progress( | EventPayload::AllocationFinished(_, _) | EventPayload::ServerStart { .. } | EventPayload::ServerStop - | EventPayload::TaskNotify(..) => {} + | EventPayload::TaskNotify(..) + | EventPayload::IdCounters { .. } => {} }, _ => { log::warn!("Unexpected message from server: {:?}", msg); diff --git a/crates/hyperqueue/src/dashboard/data/timelines/job_timeline.rs b/crates/hyperqueue/src/dashboard/data/timelines/job_timeline.rs index 05f522db2..5fad7d998 100644 --- a/crates/hyperqueue/src/dashboard/data/timelines/job_timeline.rs +++ b/crates/hyperqueue/src/dashboard/data/timelines/job_timeline.rs @@ -150,7 +150,8 @@ impl JobTimeline { | EventPayload::AllocationFinished(_, _) | EventPayload::ServerStart { .. } | EventPayload::ServerStop - | EventPayload::TaskNotify(_) => {} + | EventPayload::TaskNotify(_) + | EventPayload::IdCounters { .. } => {} } } } diff --git a/crates/hyperqueue/src/dashboard/data/timelines/worker_timeline.rs b/crates/hyperqueue/src/dashboard/data/timelines/worker_timeline.rs index 463c1e8a4..0e5572a9b 100644 --- a/crates/hyperqueue/src/dashboard/data/timelines/worker_timeline.rs +++ b/crates/hyperqueue/src/dashboard/data/timelines/worker_timeline.rs @@ -88,7 +88,8 @@ impl WorkerTimeline { | EventPayload::AllocationFinished(_, _) | EventPayload::ServerStart { .. } | EventPayload::ServerStop - | EventPayload::TaskNotify(_) => {} + | EventPayload::TaskNotify(_) + | EventPayload::IdCounters { .. } => {} } } } diff --git a/crates/hyperqueue/src/server/autoforget.rs b/crates/hyperqueue/src/server/autoclean.rs similarity index 75% rename from crates/hyperqueue/src/server/autoforget.rs rename to crates/hyperqueue/src/server/autoclean.rs index b1d51220e..fe1d01555 100644 --- a/crates/hyperqueue/src/server/autoforget.rs +++ b/crates/hyperqueue/src/server/autoclean.rs @@ -1,9 +1,10 @@ use crate::server::Senders; -use crate::server::client::forget_jobs; +use crate::server::client::{forget_jobs, prune_journal}; use crate::server::state::StateRef; use chrono::{TimeDelta, Utc}; use std::time::Duration; use tako::JobId; +use tokio::time::Instant; pub async fn auto_forget_process(state_ref: StateRef, senders: Senders, max_age: Duration) { let max_age_delta = TimeDelta::from_std(max_age).unwrap_or(TimeDelta::MAX); @@ -37,3 +38,17 @@ pub async fn auto_forget_process(state_ref: StateRef, senders: Senders, max_age: .await; } } + +pub async fn journal_prune_process( + state_ref: StateRef, + senders: Senders, + period: Duration, + keep_completed_jobs: bool, +) { + let mut interval = tokio::time::interval_at(Instant::now() + period, period); + loop { + interval.tick().await; + log::debug!("Pruning journal"); + prune_journal(&state_ref, &senders, keep_completed_jobs).await; + } +} diff --git a/crates/hyperqueue/src/server/bootstrap.rs b/crates/hyperqueue/src/server/bootstrap.rs index 0c698a19e..1951e2df1 100644 --- a/crates/hyperqueue/src/server/bootstrap.rs +++ b/crates/hyperqueue/src/server/bootstrap.rs @@ -17,7 +17,7 @@ use crate::common::serverdir::{ }; use crate::server::Senders; use crate::server::autoalloc::{QueueId, create_autoalloc_service}; -use crate::server::autoforget::auto_forget_process; +use crate::server::autoclean::{auto_forget_process, journal_prune_process}; use crate::server::event::journal::JournalWriter; use crate::server::event::journal::start_event_streaming; use crate::server::event::streamer::EventStreamer; @@ -46,6 +46,7 @@ pub struct ServerConfig { pub worker_port: Option, pub journal_path: Option, pub journal_flush_period: Duration, + pub journal_prune_period: Option, pub worker_secret_key: Option>, pub client_secret_key: Option>, pub server_uid: Option, @@ -278,6 +279,16 @@ pub async fn initialize_server( )), None => Box::pin(futures::future::pending()), }; + let journal_prune_fut: Pin>> = match server_cfg.journal_prune_period + { + Some(period) => Box::pin(journal_prune_process( + state_ref.clone(), + senders.clone(), + period, + server_cfg.auto_forget.is_some(), + )), + None => Box::pin(futures::future::pending()), + }; let senders2 = senders.clone(); let fut = async move { @@ -300,6 +311,7 @@ pub async fn initialize_server( ) => { Ok(()) } _ = &mut autoalloc_process => { Ok(()) } _ = auto_forget_fut => { Ok(()) } + _ = journal_prune_fut => { Ok(()) } r = comm_fut => { r.map_err(|e| e.into()) } }; diff --git a/crates/hyperqueue/src/server/client/mod.rs b/crates/hyperqueue/src/server/client/mod.rs index 8f303ac87..889be3d0b 100644 --- a/crates/hyperqueue/src/server/client/mod.rs +++ b/crates/hyperqueue/src/server/client/mod.rs @@ -545,12 +545,24 @@ fn reconstruct_historical_events( async fn handle_prune_journal(state_ref: &StateRef, senders: &Senders) -> ToClientMessage { log::debug!("Client asked for journal prunning"); - let (live_jobs, live_workers) = { + prune_journal(state_ref, senders, false).await; + ToClientMessage::Finished +} + +/// Removes completed jobs and lost workers from the journal (if journal is enabled). +/// If `keep_completed_jobs` is true, completed jobs that are still in the server memory +/// (i.e. not forgotten yet) are kept in the journal. +pub(crate) async fn prune_journal( + state_ref: &StateRef, + senders: &Senders, + keep_completed_jobs: bool, +) { + let (live_jobs, live_workers, max_job_id, max_worker_id) = { let state = state_ref.get(); let live_jobs: Set<_> = state .jobs() .filter_map(|job| { - if job.is_terminated() { + if job.is_terminated() && !keep_completed_jobs { None } else { Some(job.job_id) @@ -568,12 +580,20 @@ async fn handle_prune_journal(state_ref: &StateRef, senders: &Senders) -> ToClie } }) .collect(); - (live_jobs, live_workers) + ( + live_jobs, + live_workers, + state.last_job_id(), + state.max_worker_id(), + ) }; - if let Some(receiver) = senders.events.prune_journal(live_jobs, live_workers) { + if let Some(receiver) = + senders + .events + .prune_journal(live_jobs, live_workers, max_job_id, max_worker_id) + { let _ = receiver.await; } - ToClientMessage::Finished } fn handle_worker_stop( diff --git a/crates/hyperqueue/src/server/event/journal/mod.rs b/crates/hyperqueue/src/server/event/journal/mod.rs index a24f08063..7b0c5fa03 100644 --- a/crates/hyperqueue/src/server/event/journal/mod.rs +++ b/crates/hyperqueue/src/server/event/journal/mod.rs @@ -11,7 +11,7 @@ pub use write::JournalWriter; const HQ_JOURNAL_HEADER: &[u8] = b"hqjl0002"; const HQ_JOURNAL_VERSION_MAJOR: u32 = 26; -const HQ_JOURNAL_VERSION_MINOR: u32 = 0; +const HQ_JOURNAL_VERSION_MINOR: u32 = 1; #[derive(Serialize, Deserialize)] struct JournalVersion { diff --git a/crates/hyperqueue/src/server/event/journal/prune.rs b/crates/hyperqueue/src/server/event/journal/prune.rs index 67f83e92b..0f9cfc306 100644 --- a/crates/hyperqueue/src/server/event/journal/prune.rs +++ b/crates/hyperqueue/src/server/event/journal/prune.rs @@ -1,5 +1,7 @@ +use crate::server::event::Event; use crate::server::event::journal::{JournalReader, JournalWriter}; use crate::server::event::payload::EventPayload; +use chrono::Utc; use tako::JobId; use tako::{Set, WorkerId}; @@ -8,7 +10,16 @@ pub(crate) fn prune_journal( writer: &mut JournalWriter, live_job_ids: &Set, live_worker_ids: &Set, + max_job_id: JobId, + max_worker_id: WorkerId, ) -> crate::Result<()> { + writer.store(Event { + time: Utc::now(), + payload: EventPayload::IdCounters { + max_job_id, + max_worker_id, + }, + })?; for event in reader { let mut event = event?; let event = match &mut event.payload { @@ -44,6 +55,8 @@ pub(crate) fn prune_journal( | EventPayload::TaskNotify(_) | EventPayload::JobIdle(_) | EventPayload::ServerStop => Some(event), + // A fresh one was written at the beginning + EventPayload::IdCounters { .. } => None, }; if let Some(event) = event { writer.store(event)?; diff --git a/crates/hyperqueue/src/server/event/journal/stream.rs b/crates/hyperqueue/src/server/event/journal/stream.rs index 540b573ee..ae4ce764a 100644 --- a/crates/hyperqueue/src/server/event/journal/stream.rs +++ b/crates/hyperqueue/src/server/event/journal/stream.rs @@ -20,6 +20,8 @@ pub enum EventStreamMessage { callback: tokio::sync::oneshot::Sender<()>, live_jobs: Set, live_workers: Set, + max_job_id: JobId, + max_worker_id: WorkerId, }, FlushJournal(tokio::sync::oneshot::Sender<()>), } @@ -99,7 +101,7 @@ async fn streaming_process( tx.send(event?)?; } }, - Some(EventStreamMessage::PruneJournal { live_jobs, live_workers, callback }) => { + Some(EventStreamMessage::PruneJournal { live_jobs, live_workers, max_job_id, max_worker_id, callback }) => { writer.flush()?; drop(writer); let mut tmp_path: OsString = journal_path.into(); @@ -108,7 +110,7 @@ async fn streaming_process( { let mut reader = JournalReader::open(journal_path)?; let mut writer = JournalWriter::create(&tmp_path)?; - if let Err(e) = prune_journal(&mut reader, &mut writer, &live_jobs, &live_workers) { + if let Err(e) = prune_journal(&mut reader, &mut writer, &live_jobs, &live_workers, max_job_id, max_worker_id) { remove_file(&tmp_path)?; return Err(e.into()) } diff --git a/crates/hyperqueue/src/server/event/payload.rs b/crates/hyperqueue/src/server/event/payload.rs index fd7fc2b74..4c58cfeeb 100644 --- a/crates/hyperqueue/src/server/event/payload.rs +++ b/crates/hyperqueue/src/server/event/payload.rs @@ -91,6 +91,13 @@ pub enum EventPayload { /// Task notification from a task (EPHEMERAL - not stored in the journal) TaskNotify(TaskNotification), + + /// Highest job/worker IDs used so far; written by journal pruning, so that IDs + /// are not reused after a restart from a pruned journal + IdCounters { + max_job_id: JobId, + max_worker_id: WorkerId, + }, } #[derive(Serialize, Deserialize, Debug, Clone)] diff --git a/crates/hyperqueue/src/server/event/streamer.rs b/crates/hyperqueue/src/server/event/streamer.rs index ac2c56fd8..474f2f625 100644 --- a/crates/hyperqueue/src/server/event/streamer.rs +++ b/crates/hyperqueue/src/server/event/streamer.rs @@ -134,7 +134,9 @@ impl EventFilter { | EventPayload::AllocationFinished(_, _) => { self.flags.contains(EventFilterFlags::ALLOCATION_EVENTS) } - EventPayload::ServerStart { .. } | EventPayload::ServerStop => true, + EventPayload::ServerStart { .. } + | EventPayload::ServerStop + | EventPayload::IdCounters { .. } => true, EventPayload::TaskNotify(notify) => { if !self.flags.contains(EventFilterFlags::NOTIFY_EVENTS) { false @@ -498,6 +500,8 @@ impl EventStreamer { &self, live_jobs: Set, live_workers: Set, + max_job_id: JobId, + max_worker_id: WorkerId, ) -> Option> { let inner = self.inner.get(); if let Some(ref streamer) = inner.storage_sender { @@ -506,6 +510,8 @@ impl EventStreamer { .send(EventStreamMessage::PruneJournal { live_jobs, live_workers, + max_job_id, + max_worker_id, callback: sender, }) .is_err() diff --git a/crates/hyperqueue/src/server/mod.rs b/crates/hyperqueue/src/server/mod.rs index b33b6ae3f..8f420465f 100644 --- a/crates/hyperqueue/src/server/mod.rs +++ b/crates/hyperqueue/src/server/mod.rs @@ -3,7 +3,7 @@ use crate::server::event::streamer::EventStreamer; use tako::control::ServerRef; pub mod autoalloc; -mod autoforget; +mod autoclean; pub mod bootstrap; pub mod client; pub mod event; diff --git a/crates/hyperqueue/src/server/restore.rs b/crates/hyperqueue/src/server/restore.rs index 2e80aa55c..fb35f5dc8 100644 --- a/crates/hyperqueue/src/server/restore.rs +++ b/crates/hyperqueue/src/server/restore.rs @@ -413,6 +413,13 @@ impl StateRestorer { EventPayload::AllocationFinished(_, _) => {} EventPayload::ServerStart { server_uid } => self.server_uid = server_uid, EventPayload::ServerStop => { /* Do nothing */ } + EventPayload::IdCounters { + max_job_id, + max_worker_id, + } => { + self.max_job_id = self.max_job_id.max(max_job_id.as_num()); + self.max_worker_id = self.max_worker_id.max(max_worker_id.as_num()); + } EventPayload::JobOpen(job_id, job_description) => { let job = RestorerJob::new(job_description, true); self.add_job(job_id, job); diff --git a/crates/hyperqueue/src/server/state.rs b/crates/hyperqueue/src/server/state.rs index ade274b72..befac9276 100644 --- a/crates/hyperqueue/src/server/state.rs +++ b/crates/hyperqueue/src/server/state.rs @@ -21,6 +21,8 @@ pub struct State { jobs: Map, workers: Map, job_id_counter: ::IdType, + /// Highest worker ID seen so far (including workers from before a restart) + max_worker_id: WorkerId, server_info: ServerInfo, } @@ -41,6 +43,7 @@ impl State { pub fn add_worker(&mut self, worker: Worker) { let worker_id = worker.worker_id(); + self.max_worker_id = self.max_worker_id.max(worker_id); assert!(self.workers.insert(worker_id, worker).is_none()) } @@ -98,6 +101,10 @@ impl State { JobId::new(self.job_id_counter - 1) } + pub fn max_worker_id(&self) -> WorkerId { + self.max_worker_id + } + pub fn get_workers(&self) -> &Map { &self.workers } @@ -223,7 +230,8 @@ impl State { } pub(crate) fn restore_state(&mut self, restorer: &StateRestorer) { - self.job_id_counter = restorer.job_id_counter() + self.job_id_counter = restorer.job_id_counter(); + self.max_worker_id = WorkerId::new(restorer.worker_id_counter().as_num() - 1); } } @@ -233,6 +241,7 @@ impl StateRef { jobs: Default::default(), workers: Default::default(), job_id_counter: 1, + max_worker_id: WorkerId::new(0), server_info, })) } diff --git a/crates/hyperqueue/src/tests/server.rs b/crates/hyperqueue/src/tests/server.rs index 9ac4b9d09..c6a6ed51b 100644 --- a/crates/hyperqueue/src/tests/server.rs +++ b/crates/hyperqueue/src/tests/server.rs @@ -54,6 +54,7 @@ where worker_port: None, journal_path: None, journal_flush_period: Duration::from_secs(30), + journal_prune_period: None, worker_secret_key: None, client_secret_key: None, server_uid: None, diff --git a/crates/pyhq/src/cluster/server.rs b/crates/pyhq/src/cluster/server.rs index 955d33293..dc8085eda 100644 --- a/crates/pyhq/src/cluster/server.rs +++ b/crates/pyhq/src/cluster/server.rs @@ -40,6 +40,7 @@ impl RunningServer { worker_secret_key: None, server_uid: None, journal_flush_period: Duration::from_secs(30), + journal_prune_period: None, scheduler_mip_time_limit: Duration::from_secs(5), auto_forget: None, }; diff --git a/docs/deployment/server.md b/docs/deployment/server.md index 25b251f8f..d6145dff7 100644 --- a/docs/deployment/server.md +++ b/docs/deployment/server.md @@ -134,6 +134,17 @@ whithin a given time frame. It takes duration and it is counted from the beginni The [`hq journal prune`](cli:hq.journal.prune) command removes all completed jobs and disconnected workers from the journal file, in order to reduce its size on disk. +The server can also prune the journal periodically by itself, using the `--journal-prune-period` option: + +```console +$ hq server start --journal /path/to/journal --journal-prune-period=1d +``` + +If this option is not set, the journal is never pruned automatically. If +[`--auto-forget`](../jobs/jobs.md#automatic-forgetting) is also set, periodic pruning removes only jobs that +the server has already forgotten, so the journal keeps the same jobs as the server memory. Pruning does not change job and +worker IDs: after a restart from a pruned journal, new jobs and workers still get IDs that were not used before. + ### Flushing the journal The [`hq journal flush`](cli:hq.journal.flush) command will force the server to flush the journal, so that the latest state of affairs is persisted to disk. It is mainly useful for testing or if you are going to run `hq journal export` while a server is running (however, it is usually better to use `hq journal stream`). diff --git a/docs/jobs/jobs.md b/docs/jobs/jobs.md index 04861d093..ef6200d4b 100644 --- a/docs/jobs/jobs.md +++ b/docs/jobs/jobs.md @@ -329,6 +329,10 @@ and completed first. Note that if you are using a journal, forgetting only free the memory of the server but the tasks remains in journal, run [`hq journal prune`](cli:hq.journal.prune) to remove completed jobs and workers from journal file. +You can also let the server prune the journal periodically, e.g. +`hq server start --journal /path/to/journal --auto-forget=7d --journal-prune-period=1d`. +When both options are set, periodic pruning removes only jobs that were already forgotten, so completed +jobs are kept in the journal (and survive a server restart) until they are auto-forgotten. ## Waiting for jobs diff --git a/tests/test_journal.py b/tests/test_journal.py index e45913e53..3043e93ff 100644 --- a/tests/test_journal.py +++ b/tests/test_journal.py @@ -10,6 +10,7 @@ from .autoalloc.utils import add_queue, remove_queue from .conftest import HqEnv from .utils import wait_for_job_state, wait_for_task_state +from .utils.wait import wait_until import os @@ -491,3 +492,74 @@ def test_journal_cancel_auto_flush(hq_env: HqEnv, tmp_path): hq_env.command(["job", "cancel", "1"]) hq_env.kill_server() assert any(event["event"]["type"] == "task-canceled" for event in read_events(hq_env, journal_path)) + + +def test_journal_prune_period(hq_env: HqEnv, tmp_path): + journal_path = os.path.join(tmp_path, "my.journal") + hq_env.start_server(args=["--journal", journal_path, "--journal-prune-period=2s"]) + hq_env.start_worker() + hq_env.command(["submit", "--", "sleep", "0"]) + hq_env.command(["submit", "--", "sleep", "100"]) + wait_for_job_state(hq_env, 1, "FINISHED") + wait_for_job_state(hq_env, 2, "RUNNING") + + def job_ids(): + hq_env.command(["journal", "flush"]) + events = read_events(hq_env, journal_path) + return {e["event"]["job"] for e in events if "job" in e["event"]} + + assert job_ids() == {1, 2} + wait_until(lambda: job_ids() == {2}) + + +def test_journal_prune_period_keeps_not_forgotten_jobs(hq_env: HqEnv, tmp_path): + journal_path = os.path.join(tmp_path, "my.journal") + hq_env.start_server(args=["--journal", journal_path, "--journal-prune-period=1s", "--auto-forget=4s"]) + hq_env.start_worker() + hq_env.command(["submit", "--", "sleep", "0"]) + wait_for_job_state(hq_env, 1, "FINISHED") + + def job_ids(): + hq_env.command(["journal", "flush"]) + events = read_events(hq_env, journal_path) + return {e["event"]["job"] for e in events if "job" in e["event"]} + + # Several prunes happen before the job is forgotten + time.sleep(2.5) + assert job_ids() == {1} + wait_until(lambda: job_ids() == set()) + + +def test_journal_prune_period_requires_journal(hq_env: HqEnv): + hq_env.command( + ["server", "start", "--journal-prune-period=2s"], + expect_fail="--journal ", + ) + + +def test_restore_ids_after_prune(hq_env: HqEnv, tmp_path): + journal_path = os.path.join(tmp_path, "my.journal") + hq_env.start_server(args=["--journal", journal_path]) + w = hq_env.start_worker() + hq_env.command(["submit", "--", "sleep", "0"]) + hq_env.command(["submit", "--", "sleep", "0"]) + wait_for_job_state(hq_env, [1, 2], "FINISHED") + hq_env.command(["worker", "stop", "1"]) + hq_env.check_process_exited(w) + hq_env.command(["journal", "prune"]) + hq_env.stop_server() + + # Pruning twice keeps the counters + hq_env.start_server(args=["--journal", journal_path]) + hq_env.command(["journal", "prune"]) + hq_env.stop_server() + + hq_env.start_server(args=["--journal", journal_path]) + hq_env.start_worker() + hq_env.command(["submit", "--", "sleep", "0"]) + wait_for_job_state(hq_env, 3, "FINISHED") + table = hq_env.command(["job", "list", "--all"], as_table=True) + table.check_columns_value(["ID"], 0, ["3"]) + # Same worker ID as after a restart without pruning (see test_restore_fully_finished) + table = hq_env.command(["worker", "list"], as_table=True) + table.check_columns_value(["ID", "State"], 0, ["3", "RUNNING"])