Fix aborted token handling in task scheduler. (#8511)

Author: isoos

app/lib/task/backend.dart

@@ -482,7 +482,6 @@ class TaskBackend {
             ),
         });
       state.derivePendingAt();
-      state.abortedTokens?.removeWhere((t) => t.expires.isAfter(clock.now()));
 
       _log.info('Update state tracking for $packageName');
       tx.insert(state);

app/test/task/task_test.dart

@@ -21,6 +21,7 @@ import 'package:pub_dev/tool/test_profile/importer.dart';
 import 'package:pub_dev/tool/test_profile/models.dart';
 import 'package:test/test.dart';
 
+import '../shared/handlers_test_utils.dart';
 import '../shared/test_services.dart';
 import 'fake_time.dart';
 
@@ -695,6 +696,88 @@ void main() {
           TestUser(email: '[email protected]', likes: []),
         ],
       ));
+
+  testWithFakeTime(
+    'new version during ongoing analysis',
+    testProfile: TestProfile(
+      defaultUser: '[email protected]',
+      packages: [
+        TestPackage(
+          name: 'neon',
+          versions: [TestVersion(version: '1.0.0')],
+        ),
+      ],
+      users: [
+        TestUser(email: '[email protected]', likes: []),
+      ],
+    ),
+    (fakeTime) async {
+      await taskBackend.backfillTrackingState();
+      await taskBackend.start();
+      await fakeTime.elapse(minutes: 15);
+      {
+        final instances = await cloud.listInstances().toList();
+        // There is only one package, so we should only get one instance
+        expect(instances, hasLength(1));
+
+        final instance = instances.first;
+        final payload = instance.payload;
+
+        // There should only be one version
+        expect(payload.versions, hasLength(1));
+        final v = payload.versions.first;
+
+        // Create new versions, removing the token from the first version
+        await importProfile(
+          source: ImportSource.autoGenerated(),
+          profile: TestProfile(
+            defaultUser: '[email protected]',
+            packages: [
+              TestPackage(
+                name: 'neon',
+                versions: [
+                  TestVersion(version: '1.1.0'),
+                  TestVersion(version: '1.2.0'),
+                  TestVersion(version: '2.0.0'),
+                  TestVersion(version: '3.0.0'),
+                  TestVersion(version: '4.0.0'),
+                  TestVersion(version: '5.0.0'),
+                ],
+              ),
+            ],
+          ),
+        );
+
+        await fakeTime.elapse(minutes: 15);
+
+        // Use token to get the upload information
+        final api = createPubApiClient(authToken: v.token);
+        await expectApiException(
+          api.taskUploadResult(
+            payload.package,
+            v.version,
+          ),
+          status: 400,
+          code: 'TaskAborted',
+          message: 'neon/1.0.0 has been aborted.',
+        );
+
+        // Report the task as finished
+        await expectApiException(
+          api.taskUploadFinished(payload.package, v.version),
+          status: 400,
+          code: 'TaskAborted',
+          message: 'neon/1.0.0 has been aborted.',
+        );
+      }
+      // Leave time for the instance to be deleted (takes 1 min in fake cloud)
+      await fakeTime.elapse(minutes: 5);
+
+      await taskBackend.stop();
+
+      await fakeTime.elapse(minutes: 10);
+    },
+  );
 }
 
 extension<T> on Stream<T> {