Merge pull request #130 from depot/task-locks

Clear completed tasks at the start of the loop

Author: jacobwgillespie

src/handlers/state.ts

@@ -11,6 +11,7 @@ import {
   reconcile as reconcileFly,
 } from '../utils/fly/reconcile'
 import {client} from '../utils/grpc'
+import {clearCompletedTasks} from '../utils/scheduler'
 
 interface CloudProvider<T extends AwsCurrentState | FlyCurrentState> {
   getCurrentState(): Promise<T>
@@ -33,6 +34,9 @@ export async function startStateStream<T extends AwsCurrentState | FlyCurrentSta
 ) {
   while (!signal.aborted) {
     try {
+      // Allow any completed tasks to be scheduled again if needed
+      clearCompletedTasks()
+
       const currentState = await provider.getCurrentState()
 
       const response = await client.getDesiredState(

src/utils/scheduler.ts

@@ -1,13 +1,14 @@
 import {reportError} from './errors'
 
 const inProgressTasks = new Set<string>()
+const completedTasks = new Set<string>()
 
 /**
  * Schedule an update to run, ensuring that only one update is running at a time.
  */
 export async function scheduleTask(key: string, task: () => Promise<void>) {
-  if (inProgressTasks.has(key)) {
-    console.log(`Skipping ${key} because it is already in progress`)
+  if (inProgressTasks.has(key) || completedTasks.has(key)) {
+    console.log(`Skipping ${key} because it is already in progress or completed`)
     return
   }
 
@@ -21,7 +22,20 @@ export async function scheduleTask(key: string, task: () => Promise<void>) {
     await reportError(err)
   } finally {
     inProgressTasks.delete(key)
+    completedTasks.add(key)
     const duration = new Date().getTime() - start.getTime()
     console.log(`Task ${key} completed (${duration}ms)`)
   }
 }
+
+/**
+ * Clear completed tasks, allowing them to be scheduled again.
+ * This should be called at the beginning of each reconciliation loop.
+ */
+export function clearCompletedTasks() {
+  const count = completedTasks.size
+  if (count > 0) {
+    console.log(`Clearing ${count} completed task(s)`)
+    completedTasks.clear()
+  }
+}