Job task counters are inflated after restoring an open job with multiple submissions

When the server restores an open job that received more than one submission, every task that was already in a terminal state is added to the job counters once per submission. The task list is still correct after the restore, but job info and job list report wrong counts, and the waiting count can wrap around.

Version: v0.27.0. restore.rs and job.rs have the same blobs on current main.

Steps to reproduce

hq --server-dir ./sd server start --journal ./journal.bin &
hq --server-dir ./sd worker start --cpus 1 &

hq --server-dir ./sd job open
hq --server-dir ./sd submit --job 1 --array 0-1 -- /bin/true
# wait until tasks 0 and 1 are finished
hq --server-dir ./sd worker stop all
hq --server-dir ./sd submit --job 1 -- /bin/true
hq --server-dir ./sd submit --job 1 -- /bin/true

hq --server-dir ./sd --output-mode json job info 1   # task_stats before restart

hq --server-dir ./sd server stop
hq --server-dir ./sd server start --journal ./journal.bin &
hq --server-dir ./sd --output-mode json job info 1   # task_stats after restart
hq --server-dir ./sd task list 1

Before the restart:

{"aborted":0,"canceled":0,"failed":0,"finished":2,"running":0,"waiting":2}

After the restart:

{"aborted":0,"canceled":0,"failed":0,"finished":6,"running":0,"waiting":4294967294}

task list 1 after the restart still shows tasks 0 and 1 as FINISHED and tasks 2 and 3 as WAITING, which is correct. With one finished task and three submissions, the result is finished: 3, waiting: 0, so there is no wrap-around, but the counts are still wrong. After the restart in the wrap-around case, job list reports the open job as WAITING. In the case without wrap-around, it reports OPENED.

We noticed this on a long-running server with a journal. After restarts, the job info task counts of open jobs with many submissions no longer matched task list.

Cause

In RestorerJob::restore_job, the loop over job.tasks that updates job.counters sits inside the loop over self.submit_descs:

hyperqueue/crates/hyperqueue/src/server/restore.rs

Lines 64 to 105 in c41ad8d

    
           for submit in self.submit_descs { 
        
               if let Some(e) = validate_submit(state.get_job(job_id), &submit.description().task_desc) 
        
               { 
        
                   return Err(HqError::GenericError(format!( 
        
                       "Job validation failed {e:?}" 
        
                   ))); 
        
               } 
        
               let mut new_tasks = submit_job_desc( 
        
                   state, 
        
                   server_ref, 
        
                   job_id, 
        
                   submit.description().clone(), 
        
                   submit.submitted_at(), 
        
               ); 
        
               let job = state.get_job_mut(job_id).unwrap(); 
        
               new_tasks.tasks.retain_mut(|t| { 
        
                   t.task_deps 
        
                       .retain(|d| !is_task_completed(&self.tasks, d.job_task_id())); 
        
                   !is_task_completed(&self.tasks, t.id.job_task_id()) 
        
               }); 
        
               for (task_id, job_task) in job.tasks.iter_mut() { 
        
                   if let Some(task) = self.tasks.get_mut(task_id) { 
        
                       if task.crash_counter > 0 || task.instance_id.is_some() { 
        
                           new_tasks.adjust_instance_id_and_crash_counters.insert( 
        
                               TaskId::new(job_id, *task_id), 
        
                               ( 
        
                                   task.instance_id.map(|x| x.as_num() + 1).unwrap_or(0).into(), 
        
                                   task.crash_counter, 
        
                               ), 
        
                           ); 
        
                       } 
        
                       match &task.state { 
        
                           JobTaskState::Waiting | JobTaskState::Running { .. } => continue, 
        
                           JobTaskState::Finished { .. } => job.counters.n_finished_tasks += 1, 
        
                           JobTaskState::Failed { .. } => job.counters.n_failed_tasks += 1, 
        
                           JobTaskState::Canceled { .. } => job.counters.n_canceled_tasks += 1, 
        
                           JobTaskState::Aborted { .. } => job.counters.n_aborted_tasks += 1, 
        
                       } 
        
                       job_task.state = task.state.clone(); 
        
                   }

Each iteration visits every task the job has accumulated so far, including tasks created by earlier submissions. So a task from the first submission is counted again for every later submission. n_waiting_tasks then subtracts the inflated counts from the number of tasks with plain u32 arithmetic, which wraps in release builds:

hyperqueue/crates/hyperqueue/src/server/job.rs

Lines 97 to 104 in c41ad8d

    
           pub fn n_waiting_tasks(&self, n_tasks: JobTaskCount) -> JobTaskCount { 
        
               n_tasks 
        
                   - self.n_running_tasks 
        
                   - self.n_finished_tasks 
        
                   - self.n_failed_tasks 
        
                   - self.n_canceled_tasks 
        
                   - self.n_aborted_tasks 
        
           }

Possible fix

Count each restored task once. Either visit only the tasks created by the current submission inside the loop, or move the counter update after the loop over submissions. The instance id and crash counter adjustment in the same loop probably has the same repetition, although it looks harmless because it only overwrites map entries. I can open a PR if that helps.

Navigation Menu

Repository navigation

Job task counters are inflated after restoring an open job with multiple submissions #1156

Steps to reproduce

Cause

Possible fix

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	for submit in self.submit_descs {
	if let Some(e) = validate_submit(state.get_job(job_id), &submit.description().task_desc)
	{
	return Err(HqError::GenericError(format!(
	"Job validation failed {e:?}"
	)));
	}
	let mut new_tasks = submit_job_desc(
	state,
	server_ref,
	job_id,
	submit.description().clone(),
	submit.submitted_at(),
	);
	let job = state.get_job_mut(job_id).unwrap();

	new_tasks.tasks.retain_mut(\|t\| {
	t.task_deps
	.retain(\|d\| !is_task_completed(&self.tasks, d.job_task_id()));
	!is_task_completed(&self.tasks, t.id.job_task_id())
	});

	for (task_id, job_task) in job.tasks.iter_mut() {
	if let Some(task) = self.tasks.get_mut(task_id) {
	if task.crash_counter > 0 \|\| task.instance_id.is_some() {
	new_tasks.adjust_instance_id_and_crash_counters.insert(
	TaskId::new(job_id, *task_id),
	(
	task.instance_id.map(\|x\| x.as_num() + 1).unwrap_or(0).into(),
	task.crash_counter,
	),
	);
	}
	match &task.state {
	JobTaskState::Waiting \| JobTaskState::Running { .. } => continue,
	JobTaskState::Finished { .. } => job.counters.n_finished_tasks += 1,
	JobTaskState::Failed { .. } => job.counters.n_failed_tasks += 1,
	JobTaskState::Canceled { .. } => job.counters.n_canceled_tasks += 1,
	JobTaskState::Aborted { .. } => job.counters.n_aborted_tasks += 1,
	}
	job_task.state = task.state.clone();
	}

	pub fn n_waiting_tasks(&self, n_tasks: JobTaskCount) -> JobTaskCount {
	n_tasks
	- self.n_running_tasks
	- self.n_finished_tasks
	- self.n_failed_tasks
	- self.n_canceled_tasks
	- self.n_aborted_tasks
	}

Job task counters are inflated after restoring an open job with multiple submissions #1156

Description

Steps to reproduce

Cause

Possible fix

Activity

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions