feat(core): Expand crash recovery to cover queue mode

packages/cli/src/databases/repositories/execution.repository.ts

@@ -261,7 +261,9 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
 		return String(executionId);
 	}
 
-	async markAsCrashed(executionIds: string[]) {
+	async markAsCrashed(executionIds: string | string[]) {
+		if (!Array.isArray(executionIds)) executionIds = [executionIds];
+
 		await this.update(
 			{ id: In(executionIds) },
 			{
@@ -749,4 +751,13 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
 
 		return executions.map(({ id }) => id);
 	}
+
+	async allCurrentExecutionIds() {
+		const executions = await this.find({
+			select: ['id'],
+			where: { status: In(['new', 'running']) },
+		});
+
+		return executions.map(({ id }) => id);
+	}
 }

packages/cli/src/eventbus/MessageEventBus/MessageEventBus.ts

@@ -178,18 +178,8 @@ export class MessageEventBus extends EventEmitter {
 					// start actual recovery process and write recovery process flag file
 					this.logWriter?.startRecoveryProcess();
 					for (const executionId of unfinishedExecutionIds) {
-						this.logger.warn(`Attempting to recover execution ${executionId}`);
-						if (!unsentAndUnfinished.unfinishedExecutions[executionId]?.length) {
-							this.logger.debug(
-								`No event messages found, marking execution ${executionId} as 'crashed'`,
-							);
-							await this.executionRepository.markAsCrashed([executionId]);
-						} else {
-							await this.recoveryService.recover(
-								executionId,
-								unsentAndUnfinished.unfinishedExecutions[executionId],
-							);
-						}
+						const logMesssages = unsentAndUnfinished.unfinishedExecutions[executionId];
+						await this.recoveryService.recoverFromLogs(executionId, logMesssages);
 					}
 				}
 				// remove the recovery process flag file