Refactor + separately test the instance deletion logic in task scheduling (#9097)

Author: isoos

app/lib/task/backend.dart

@@ -24,6 +24,7 @@ import 'package:pub_dev/package/backend.dart';
 import 'package:pub_dev/package/models.dart';
 import 'package:pub_dev/package/upload_signer_service.dart';
 import 'package:pub_dev/scorecard/backend.dart';
+import 'package:pub_dev/shared/configuration.dart';
 import 'package:pub_dev/shared/datastore.dart';
 import 'package:pub_dev/shared/exceptions.dart';
 import 'package:pub_dev/shared/redis_cache.dart';
@@ -43,6 +44,7 @@ import 'package:pub_dev/task/clock_control.dart';
 import 'package:pub_dev/task/cloudcompute/cloudcompute.dart';
 import 'package:pub_dev/task/global_lock.dart';
 import 'package:pub_dev/task/handlers.dart';
+import 'package:pub_dev/task/loops/delete_instances.dart';
 import 'package:pub_dev/task/loops/scan_packages_updated.dart';
 import 'package:pub_dev/task/models.dart'
     show
@@ -101,6 +103,7 @@ class TaskBackend {
 
   ScanPackagesUpdatedState _scanPackagesUpdatedState =
       ScanPackagesUpdatedState.init();
+  DeleteInstancesState _deleteInstancesState = DeleteInstancesState.init();
 
   TaskBackend(this._db, this._bucket);
 
@@ -128,6 +131,13 @@ class TaskBackend {
         await _scanForPackageUpdates(claim, abort: aborted);
       },
     );
+    final deleteLoop = _createLoop(
+      name: 'delete-instances',
+      aborted: aborted,
+      fn: (claim, aborted) async {
+        await _scanForInstanceDeletion(claim, abort: aborted);
+      },
+    );
     final scheduleLoop = _createLoop(
       name: 'schedule',
       aborted: aborted,
@@ -138,7 +148,7 @@ class TaskBackend {
 
     scheduleMicrotask(() async {
       // Wait for background process to finish
-      await Future.wait([scanLoop, scheduleLoop]);
+      await Future.wait([scanLoop, deleteLoop, scheduleLoop]);
 
       // Report background processes as stopped
       stopped.complete();
@@ -321,6 +331,40 @@ class TaskBackend {
     }
   }
 
+  /// Scan for compute instances that can be deleted until [abort] is resolved,
+  /// or [claim] is lost.
+  Future<void> _scanForInstanceDeletion(
+    GlobalLockClaim claim, {
+    Completer<void>? abort,
+  }) async {
+    abort ??= Completer<void>();
+
+    bool isAbortedFn() => !claim.valid || abort!.isCompleted;
+    while (!isAbortedFn()) {
+      await _runOneInstanceDeletion(isAbortedFn: isAbortedFn);
+
+      if (isAbortedFn()) {
+        return;
+      }
+      // Wait until aborted or 10 minutes before scanning again!
+      await abort.future.timeoutWithClock(
+        Duration(minutes: 10),
+        onTimeout: () => null,
+      );
+    }
+  }
+
+  Future<void> _runOneInstanceDeletion({
+    required bool Function() isAbortedFn,
+  }) async {
+    _deleteInstancesState = await scanAndDeleteInstances(
+      _deleteInstancesState,
+      taskWorkerCloudCompute,
+      isAbortedFn,
+      maxTaskRunHours: activeConfiguration.maxTaskRunHours,
+    );
+  }
+
   Future<void> trackPackage(
     String packageName, {
     bool updateDependents = false,

app/lib/task/cloudcompute/fakecloudcompute.dart

@@ -26,6 +26,9 @@ final class FakeCloudCompute extends CloudCompute {
   FakeCloudCompute({InstanceRunner? instanceRunner})
     : _instanceRunner = instanceRunner ?? _defaultInstanceRunner;
 
+  factory FakeCloudCompute.noop() =>
+      FakeCloudCompute(instanceRunner: (_) async {});
+
   @override
   final List<String> zones = const ['zone-a', 'zone-b'];
 
@@ -105,6 +108,26 @@ final class FakeCloudCompute extends CloudCompute {
     }
   }
 
+  /// Creates a fake instance with a running (or terminated) state.
+  /// There is no scheduling or other state update, it only registers the instance.
+  void fakeCreateRunningInstance({
+    required String zone,
+    required String instanceName,
+    required Duration ago,
+    bool isTerminated = false,
+  }) {
+    final instance = FakeCloudInstance._(
+      instanceName: instanceName,
+      zone: zone,
+      created: clock.now().subtract(ago),
+      state: isTerminated ? InstanceState.terminated : InstanceState.pending,
+      dockerImage: 'fake-docker-image',
+      arguments: [],
+      description: 'fake-description',
+    );
+    _instances.add(instance);
+  }
+
   /// Change state of instance with [instanceName] to [InstanceState.running].
   ///
   /// This does not start any subprocess or run any code. Just changes the state

app/lib/task/loops/delete_instances.dart

@@ -0,0 +1,85 @@
+// Copyright (c) 2025, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+
+import 'package:basics/basics.dart';
+import 'package:clock/clock.dart';
+import 'package:logging/logging.dart';
+import 'package:pub_dev/task/cloudcompute/cloudcompute.dart';
+
+final _log = Logger('pub.task.scan_instances');
+
+/// The internal state for scanning and deleting instances.
+final class DeleteInstancesState {
+  // Maps the `CloudInstance.instanceName` to the deletion
+  // start timestamp.
+  final Map<String, DateTime> deletions;
+
+  DeleteInstancesState({required this.deletions});
+
+  factory DeleteInstancesState.init() => DeleteInstancesState(deletions: {});
+}
+
+/// Calculates the next state of delete instances loop by processing
+/// the input from [cloudCompute].
+Future<DeleteInstancesState> scanAndDeleteInstances(
+  DeleteInstancesState state,
+  CloudCompute cloudCompute,
+  bool Function() isAbortedFn, {
+  required int maxTaskRunHours,
+}) async {
+  // Purge deletionsInProgress after 5 minutes, if instance has not disappeared
+  // after 5 minutes we'll try to delete it again.
+  final keepTreshold = clock.ago(minutes: 5);
+  final deletionInProgress = {
+    ...state.deletions.whereValue((v) => v.isAfter(keepTreshold)),
+  };
+
+  final futures = <Future>[];
+  await for (final instance in cloudCompute.listInstances()) {
+    if (isAbortedFn()) {
+      break;
+    }
+
+    // Prevent multiple calls to delete the same instance.
+    if (deletionInProgress.containsKey(instance.instanceName)) {
+      continue;
+    }
+
+    // If terminated or older than maxInstanceAge, delete the instance...
+    final isTerminated = instance.state == InstanceState.terminated;
+    final isTooOld = instance.created
+        .add(Duration(hours: maxTaskRunHours))
+        .isBefore(clock.now());
+
+    if (isTooOld) {
+      // This indicates that something is wrong the with the instance,
+      // ideally it should have detected its own deadline being violated
+      // and terminated on its own. Of course, this can fail for arbitrary
+      // reasons in a distributed system.
+      _log.warning('terminating $instance for being too old!');
+    } else if (isTerminated) {
+      _log.info('deleting $instance as it has terminated.');
+    } else {
+      // Do not delete this instance
+      continue;
+    }
+
+    deletionInProgress[instance.instanceName] = clock.now();
+
+    futures.add(
+      Future.microtask(() async {
+        try {
+          await cloudCompute.delete(instance.zone, instance.instanceName);
+        } catch (e, st) {
+          _log.severe('Failed to delete $instance', e, st);
+        }
+      }),
+    );
+  }
+
+  await Future.wait(futures);
+  return DeleteInstancesState(deletions: deletionInProgress);
+}

app/lib/task/scheduler.dart

@@ -60,64 +60,25 @@ Future<void> schedule(
     }
   }
 
-  // Set of `CloudInstance.instanceName`s currently being deleted.
-  // This to avoid deleting instances where the deletion process is still
-  // running.
-  final deletionInProgress = <String>{};
-
   // Create a fast RNG with random seed for picking zones.
   final rng = Random(Random.secure().nextInt(2 << 31));
 
+  bool isAbortedFn() => !claim.valid || abort.isCompleted;
+
   // Run scheduling iterations, so long as we have a valid claim
-  while (claim.valid && !abort.isCompleted) {
+  while (!isAbortedFn()) {
     final iterationStart = clock.now();
     _log.info('Starting scheduling cycle');
 
-    // Count number of instances, and delete old instances
-    var instances = 0;
-    await for (final instance in compute.listInstances()) {
-      instances += 1; // count the instance
-
-      // If terminated or older than maxInstanceAge, delete the instance...
-      final isTerminated = instance.state == InstanceState.terminated;
-      final isTooOld = instance.created
-          .add(Duration(hours: activeConfiguration.maxTaskRunHours))
-          .isBefore(clock.now());
-      // Also check deletionInProgress to prevent multiple calls to delete the
-      // same instance
-      final isBeingDeleted = deletionInProgress.contains(instance.instanceName);
-      if ((isTerminated || isTooOld) && !isBeingDeleted) {
-        if (isTooOld) {
-          // This indicates that something is wrong the with the instance,
-          // ideally it should have detected its own deadline being violated
-          // and terminated on its own. Of course, this can fail for arbitrary
-          // reasons in a distributed system.
-          _log.warning('terminating $instance for being too old!');
-        } else if (isTerminated) {
-          _log.info('deleting $instance as it has terminated.');
-        }
-
-        deletionInProgress.add(instance.instanceName);
-        scheduleMicrotask(() async {
-          final deletionStart = clock.now();
-          try {
-            await compute.delete(instance.zone, instance.instanceName);
-          } catch (e, st) {
-            _log.severe('Failed to delete $instance', e, st);
-          } finally {
-            // Wait at-least 5 minutes from start of deletion until we remove
-            // it from [deletionInProgress] that way we give the API some time
-            // reconcile state.
-            await sleepOrAborted(Duration(minutes: 5), since: deletionStart);
-            deletionInProgress.remove(instance.instanceName);
-          }
-        });
-      }
+    final instances = await compute.listInstances().toList();
+    if (isAbortedFn()) {
+      break;
     }
+
     _log.info('Found $instances instances');
 
     // If we are not allowed to create new instances within the allowed quota,
-    if (activeConfiguration.maxTaskInstances <= instances) {
+    if (activeConfiguration.maxTaskInstances <= instances.length) {
       _log.info('Reached instance limit, trying again in 30s');
       // Wait 30 seconds then list instances again, so that we can count them
       await sleepOrAborted(Duration(seconds: 30), since: iterationStart);
@@ -145,7 +106,7 @@ Future<void> schedule(
     var pendingPackagesReviewed = 0;
     final selectLimit = min(
       _maxInstancesPerIteration,
-      max(0, activeConfiguration.maxTaskInstances - instances),
+      max(0, activeConfiguration.maxTaskInstances - instances.length),
     );
 
     Future<void> scheduleInstance(({String package}) selected) async {
@@ -249,14 +210,14 @@ Future<void> schedule(
 
     // If there was no pending packages reviewed, and no instances currently
     // running, then we can easily sleep 5 minutes before we poll again.
-    if (instances == 0 && pendingPackagesReviewed == 0) {
+    if (instances.isEmpty && pendingPackagesReviewed == 0) {
       await sleepOrAborted(Duration(minutes: 5));
       continue;
     }
 
     // If more tasks is available and quota wasn't used up, we only sleep 10s
     if (pendingPackagesReviewed >= _maxInstancesPerIteration &&
-        activeConfiguration.maxTaskInstances > instances) {
+        activeConfiguration.maxTaskInstances > instances.length) {
       await sleepOrAborted(Duration(seconds: 10), since: iterationStart);
       continue;
     }