Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x | /**
* Nettoyage des exécutions Temporal orphelines.
*
* Le type d'une exécution Temporal est FIGÉ au démarrage : aucune migration
* MongoDB ne peut le renommer. Un renommage de type (ex. `WorkflowWorkflow`
* → `WorkflowTW`, cf. migration `20260916_004_workflow_type_tw`) laisse donc
* les exécutions déjà démarrées sous l'ancien nom. Le worker ne peut plus les
* initialiser (« no such function is exported by the workflow bundle ») et
* retente la tâche indéfiniment (workflow `RUNNING` fantôme).
*
* On termine donc, au démarrage, les exécutions `Running` dont le type figure
* dans une liste EXPLICITE d'obsolètes. On ne termine jamais « tout type
* inconnu » : cela toucherait les workflows légitimes des autres services
* (ex. `SchedulerTW`). Toute nouvelle vague de renommage doit être ajoutée ici.
*/
// Types historiques remplacés par le suffixe `TW` (v2.43).
const OBSOLETE_WORKFLOW_TYPES = [
'WorkflowWorkflow',
'CrawlerOrchestrationWorkflow',
];
/**
* Termine les exécutions Temporal `Running` dont le type est obsolète.
*
* @param {{ workflowService: object }} connection Client ou Connection Temporal.
* @param {object} [options]
* @param {string} [options.namespace='default']
* @param {string[]} [options.obsoleteTypes=OBSOLETE_WORKFLOW_TYPES]
* @param {object} [options.logger=console]
* @returns {Promise<Array<{workflowId: string, type: string}>>} exécutions terminées.
*/
async function terminateObsoleteWorkflows(
connection,
{
namespace = 'default',
obsoleteTypes = OBSOLETE_WORKFLOW_TYPES,
logger = console,
} = {},
) {
const obsolete = new Set(obsoleteTypes);
const terminated = [];
let nextPageToken;
do {
const resp = await connection.workflowService.listWorkflowExecutions({
namespace,
query: "ExecutionStatus = 'Running'",
nextPageToken,
});
for (const exec of resp.executions || []) {
const type = exec.type && exec.type.name;
if (!obsolete.has(type)) continue;
const workflowId = exec.execution && exec.execution.workflowId;
const runId = exec.execution && exec.execution.runId;
try {
await connection.workflowService.terminateWorkflowExecution({
namespace,
workflowId,
runId,
reason: `Type obsolète '${type}' non servi par le bundle courant`,
});
terminated.push({ workflowId, type });
logger.log(
`[Temporal] Workflow orphelin terminé : ${workflowId} (type ${type})`,
);
} catch (e) {
logger.error(
`[Temporal] Échec de terminaison ${workflowId} (${type}): ${e.message}`,
);
}
}
nextPageToken = resp.nextPageToken;
} while (nextPageToken && nextPageToken.length > 0);
if (terminated.length === 0) {
logger.log('[Temporal] Aucun workflow orphelin à terminer');
}
return terminated;
}
module.exports = { terminateObsoleteWorkflows, OBSOLETE_WORKFLOW_TYPES };
|