batchRoute: Better handle job cancellation and deletion

packages/transition-backend/src/services/transitRouting/TrRoutingBatch.ts

@@ -195,12 +195,9 @@ class TrRoutingBatch {
             checkpointTracker.completed();
 
             this.options.progressEmitter.emit('progress', { name: 'BatchRouting', progress: 1.0 });
-            this.options.progressEmitter.emit('progress', { name: 'StoppingRoutingParallelServers', progress: 0.0 });
-
-            const stopStatus = await TrRoutingProcessManager.stopBatch();
 
-            this.options.progressEmitter.emit('progress', { name: 'StoppingRoutingParallelServers', progress: 1.0 });
-            console.log('trRouting multiple stopStatus', stopStatus);
+            // FIXME Should we return here if the job is cancelled? Or we still
+            // generate the results that have been calculated since now?
 
             // Generate the output files
             this.options.progressEmitter.emit('progress', { name: 'GeneratingBatchRoutingResults', progress: 0.0 });
@@ -253,6 +250,14 @@ class TrRoutingBatch {
                 console.error(`Error in batch routing calculation job ${this.options.jobId}: ${error}`);
                 throw error;
             }
+        } finally {
+            // Make sure to stop the trRouting processes, even if an error occurred
+            this.options.progressEmitter.emit('progress', { name: 'StoppingRoutingParallelServers', progress: 0.0 });
+
+            const stopStatus = await TrRoutingProcessManager.stopBatch();
+
+            this.options.progressEmitter.emit('progress', { name: 'StoppingRoutingParallelServers', progress: 1.0 });
+            console.log('trRouting multiple stopStatus', stopStatus);
         }
     };
 

packages/transition-backend/src/tasks/TransitionWorkerPool.ts

@@ -30,8 +30,14 @@ function newProgressEmitter(task: ExecutableJob<JobDataType>) {
     });
     eventEmitter.on('checkpoint', (checkpoint: number) => {
         console.log('Task received checkpoint ', checkpoint);
-        task.attributes.internal_data.checkpoint = checkpoint;
-        task.save();
+        // Refresh the task before saving the checkpoint
+        task.refresh()
+            .then(() => {
+                // Add checkpoint, then save the task
+                task.attributes.internal_data.checkpoint = checkpoint;
+                task.save().catch(() => console.error('Error saving task after checkpoint'));
+            })
+            .catch(() => console.error('Error refreshing task before saving checkpoint')); // This will catch deleted jobs
     });
     return eventEmitter;
 }
@@ -65,7 +71,7 @@ const getTaskCancelledFct = (task: ExecutableJob<JobDataType>) => {
                     clearInterval(intervalObj);
                 }
             })
-            .catch(() => (refreshError = true));
+            .catch(() => (refreshError = true)); // This will catch deleted jobs
     }, 5000);
     return () => refreshError || task.status === 'cancelled';
 };