Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Introduced `ResumableTaskStore` interface and its `level` implementation. - Introduced `ResumableTaskManager`. - Refactoring to `RecordsDelete` handler.
- Loading branch information
1 parent
40567c9
commit 0311b9e
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,113 @@ | ||
| import type { StorageController } from '../store/storage-controller.js'; | ||
| import type { ManagedResumableTask, ResumableTaskStore } from '../types/resumable-task-store.js'; | ||
|
|
||
| export enum ResumableTaskName { | ||
| RecordsDelete = 'RecordsDelete', | ||
| } | ||
|
|
||
| export type ResumableTask = { | ||
| name: ResumableTaskName; | ||
| data: any; | ||
| }; | ||
|
|
||
|
|
||
| export class ResumableTaskManager { | ||
|
|
||
| /** | ||
| * The frequency at which the automatic timeout extension is requested for a resumable task. | ||
| */ | ||
| public static readonly timeoutExtensionFrequencyInSeconds = 30; | ||
|
|
||
| private resumableTaskBatchSize = 100; | ||
| private resumableTaskHandlers: { [key:string]: (taskData: any) => Promise<void> }; | ||
|
|
||
| public constructor(private resumableTaskStore: ResumableTaskStore, storageController: StorageController) { | ||
| // assign resumable task handlers | ||
| this.resumableTaskHandlers = { | ||
| // NOTE: The arrow function is IMPORTANT here, else the `this` context will be lost within the invoked method. | ||
| // e.g. code within performRecordsDelete() won't know `this` refers to the `storageController` instance. | ||
| [ResumableTaskName.RecordsDelete]: async (task): Promise<void> => await storageController.performRecordsDelete(task), | ||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * Runs a new resumable task. | ||
| */ | ||
| public async run(task: ResumableTask): Promise<void> { | ||
| const timeoutInSeconds = ResumableTaskManager.timeoutExtensionFrequencyInSeconds * 2; // give ample time for extension to take place | ||
|
|
||
| // register the new resumable task before running it so that it can be resumed if it times out for any reason | ||
| const managedResumableTask = await this.resumableTaskStore.register(task, timeoutInSeconds); | ||
| await this.runWithAutomaticTimeoutExtension(managedResumableTask); | ||
| } | ||
|
|
||
| /** | ||
| * Runs a resumable task with automatic timeout extension. | ||
| * Deletes the task from the resumable task store once it is completed. | ||
| */ | ||
| private async runWithAutomaticTimeoutExtension(managedTask: ManagedResumableTask): Promise<void> { | ||
| const timeoutInSeconds = ResumableTaskManager.timeoutExtensionFrequencyInSeconds * 2; // give ample time for extension to take place | ||
|
|
||
| let timer!: NodeJS.Timer; | ||
| try { | ||
| // start a timer loop to keep extending the timeout of the task until it is completed | ||
| timer = setInterval(() => { | ||
| this.resumableTaskStore.extend(managedTask.id, timeoutInSeconds); | ||
| }, ResumableTaskManager.timeoutExtensionFrequencyInSeconds * 1000); | ||
|
|
||
| const handler = this.resumableTaskHandlers[managedTask.task.name]; | ||
| await handler(managedTask.task.data); | ||
| await this.resumableTaskStore.delete(managedTask.id); | ||
| } finally { | ||
| ResumableTaskManager.clearTimeoutExtensionTimer(timer); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Removes the specified timeout extension loop timer. | ||
| * NOTE: created mainly for testing purposes so we can spy on this specific method without needing to filter out other `clearInterval` calls. | ||
| */ | ||
| public static clearTimeoutExtensionTimer(timer: NodeJS.Timer): void { | ||
| clearInterval(timer); | ||
| } | ||
|
|
||
| /** | ||
| * Resumes the execution of resumable tasks until all are completed successfully. | ||
| */ | ||
| public async resumeTasksAndWaitForCompletion(): Promise<void> { | ||
| while (true) { | ||
| const resumableTasks = await this.resumableTaskStore.grab(this.resumableTaskBatchSize); | ||
|
|
||
| if (resumableTasks === undefined || resumableTasks.length === 0) { | ||
| break; | ||
| } | ||
|
|
||
| // Handle this batch of tasks before grabbing the next batch. | ||
| await this.retryTasksUntilCompletion(resumableTasks); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Repeatedly retry the given tasks until all are completed successfully. | ||
| */ | ||
| private async retryTasksUntilCompletion(resumableTasks: ManagedResumableTask[]): Promise<void> { | ||
|
|
||
| let managedTasks = resumableTasks; | ||
| while (managedTasks.length > 0) { | ||
| const managedTasksCopy = managedTasks; | ||
| managedTasks = []; | ||
|
|
||
| const allTaskPromises = managedTasksCopy.map(async (managedTask) => { | ||
| try { | ||
| await this.runWithAutomaticTimeoutExtension(managedTask); | ||
| } catch (error) { | ||
| console.error('Error while running resumable task:', error); | ||
| console.error('Resumable task:', resumableTasks); | ||
| managedTasks.push(managedTask); | ||
| } | ||
| }); | ||
|
|
||
| await Promise.all(allTaskPromises); | ||
| } | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.