Undo some parallelism

Author: Porges

src/ApiService/ApiService/Functions/TimerTasks.cs

@@ -27,28 +27,33 @@ public async Async.Task Run([TimerTrigger("00:00:15")] TimerInfo myTimer) {
         var parallelOptions = new ParallelOptions { MaxDegreeOfParallelism = 10 };
 
         var expiredTasks = _taskOperations.SearchExpired();
-        await Parallel.ForEachAsync(expiredTasks, parallelOptions, async (task, _cancel) => {
+        // marking one task stopping can also mark other tasks in the same job as failed
+        // this could make parallel updates stomp on each other, so don't do it in parallel
+        await foreach (var task in expiredTasks) {
             _logger.LogInformation("stopping expired task. job_id:{JobId} task_id:{TaskId}", task.JobId, task.TaskId);
             await _taskOperations.MarkStopping(task, "task is expired");
-        });
+        }
 
         var expiredJobs = _jobOperations.SearchExpired();
+        // job updates are all distinct and only update tasks owned by that job, can be performed in parallel
         await Parallel.ForEachAsync(expiredJobs, parallelOptions, async (job, _cancel) => {
             _logger.LogInformation("stopping expired job. job_id:{JobId}", job.JobId);
             _ = await _jobOperations.Stopping(job);
         });
 
         var jobs = _jobOperations.SearchState(states: JobStateHelper.NeedsWork);
+        // job updates are okay to do in parallel
         await Parallel.ForEachAsync(jobs, parallelOptions, async (job, _cancel) => {
             _logger.LogInformation("update job: {JobId}", job.JobId);
             _ = await _jobOperations.ProcessStateUpdates(job);
         });
 
         var tasks = _taskOperations.SearchStates(states: TaskStateHelper.NeedsWorkStates);
-        await Parallel.ForEachAsync(tasks, parallelOptions, async (task, _cancel) => {
+        // task state transitions might affect the job, so parallel updates could stomp on each other
+        await foreach (var task in tasks) {
             _logger.LogInformation("update task: {TaskId}", task.TaskId);
             _ = await _taskOperations.ProcessStateUpdate(task);
-        });
+        }
 
         await _scheduler.ScheduleTasks();
         await _jobOperations.StopNeverStartedJobs();

src/ApiService/ApiService/onefuzzlib/JobOperations.cs

@@ -105,7 +105,7 @@ public async Async.Task<Job> Init(Job job) {
 
     public async Async.Task<Job> Stopping(Job job) {
         job = job with { State = JobState.Stopping };
-        var tasks = await _context.TaskOperations.QueryAsync(Query.PartitionKey(job.JobId.ToString())).ToListAsync();
+        var tasks = await _context.TaskOperations.GetByJobId(job.JobId).ToListAsync();
         var taskNotStopped = tasks.ToLookup(task => task.State != TaskState.Stopped);
 
         var notStopped = taskNotStopped[true];