Recoverable tasks [version 1]

Signed-off-by: Bogdan Nechyporenko <bnechyporenko@bol.com>
Signed-off-by: bnechyporenko <bnechyporenko@bol.com>
This commit is contained in:
bnechyporenko
2024-01-14 16:18:42 +01:00
parent 56d3373c82
commit 11b9a08e92
31 changed files with 473 additions and 62 deletions
+1 -1
View File
@@ -105,7 +105,7 @@ export type ScaffolderTaskOutput = {
* @public
*/
export type LogEvent = {
type: 'log' | 'completion' | 'cancelled';
type: 'log' | 'completion' | 'cancelled' | 'recovered';
body: {
message: string;
stepId?: string;
@@ -62,12 +62,14 @@ type ReducerLogEntry = {
message: string;
output?: ScaffolderTaskOutput;
error?: Error;
recoverStrategy?: 'none' | 'startOver';
};
};
type ReducerAction =
| { type: 'INIT'; data: ScaffolderTask }
| { type: 'CANCELLED' }
| { type: 'RECOVERED'; data: ReducerLogEntry }
| { type: 'LOGS'; data: ReducerLogEntry[] }
| { type: 'COMPLETED'; data: ReducerLogEntry }
| { type: 'ERROR'; data: Error };
@@ -105,17 +107,19 @@ function reducer(draft: TaskStream, action: ReducerAction) {
const currentStepLog = draft.stepLogs?.[entry.body.stepId];
const currentStep = draft.steps?.[entry.body.stepId];
if (entry.body.status && entry.body.status !== currentStep.status) {
currentStep.status = entry.body.status;
if (currentStep) {
if (entry.body.status && entry.body.status !== currentStep.status) {
currentStep.status = entry.body.status;
if (currentStep.status === 'processing') {
currentStep.startedAt = entry.createdAt;
}
if (currentStep.status === 'processing') {
currentStep.startedAt = entry.createdAt;
}
if (
['cancelled', 'completed', 'failed'].includes(currentStep.status)
) {
currentStep.endedAt = entry.createdAt;
if (
['cancelled', 'completed', 'failed'].includes(currentStep.status)
) {
currentStep.endedAt = entry.createdAt;
}
}
}
@@ -138,6 +142,17 @@ function reducer(draft: TaskStream, action: ReducerAction) {
return;
}
case 'RECOVERED': {
for (const stepId in draft.steps) {
if (draft.steps.hasOwnProperty(stepId)) {
draft.steps[stepId].startedAt = undefined;
draft.steps[stepId].endedAt = undefined;
draft.steps[stepId].status = 'open';
}
}
return;
}
case 'ERROR': {
draft.error = action.data;
draft.loading = false;
@@ -202,6 +217,7 @@ export const useTaskEventStream = (taskId: string): TaskStream => {
subscription = observable.subscribe({
next: event => {
retryCount = 1;
switch (event.type) {
case 'log':
return collectedLogEvents.push(event);
@@ -212,6 +228,9 @@ export const useTaskEventStream = (taskId: string): TaskStream => {
emitLogs();
dispatch({ type: 'COMPLETED', data: event });
return undefined;
case 'recovered':
dispatch({ type: 'RECOVERED', data: event });
return undefined;
default:
throw new Error(
`Unhandled event type ${event.type} in observer`,
@@ -226,16 +245,18 @@ export const useTaskEventStream = (taskId: string): TaskStream => {
// just to restart the fetch process
// details here https://github.com/backstage/backstage/issues/15002
const maxRetries = 3;
if (!error.message) {
error.message = `We cannot connect at the moment, trying again in some seconds... Retrying (${retryCount}/3 retries)`;
error.message = `We cannot connect at the moment, trying again in some seconds... Retrying (${
retryCount > maxRetries ? maxRetries : retryCount
}/${maxRetries} retries)`;
}
if (retryCount <= 3) {
setTimeout(() => {
retryCount += 1;
startStreamLogProcess();
}, 15000);
}
setTimeout(() => {
retryCount += 1;
void startStreamLogProcess();
}, 15000);
dispatch({ type: 'ERROR', data: error });
},
@@ -247,7 +268,7 @@ export const useTaskEventStream = (taskId: string): TaskStream => {
}
},
);
startStreamLogProcess();
void startStreamLogProcess();
return () => {
didCancel = true;
if (subscription) {