Fix datastore task update for post-submit builds (#2964)

This PR fixes a couple of things:

1) skips task updates for build `scheduled` case
    a) All tasks will be marked as `in progress` whenever scheduled. This skips the 400 errors: `Failed to process Instance of 'PushMessage'. (400) null is unknown`
    b) This skips datastore updates when a `scheduled` message comes after a `completed` message. Our pub/sub subscriber was created with `Message ordering: Disabled`, and we need to make sure [`idempotency`](https://en.wikipedia.org/wiki/Idempotence#Computer_science_meaning) from our consumer side.
2) skips task updates for task which has already finished
3) Removes the buggy logic to reset task status as `New`. 
4) Adds explicit task status in log before and after the update.

Helps: flutter/flutter#131192

app_dart/lib/src/model/appengine/task.dart

@@ -233,6 +233,14 @@ class Task extends Model<int> {
     statusSucceeded,
   ];
 
+  static const List<String> finishedStatusValues = <String>[
+    statusCancelled,
+    statusFailed,
+    statusInfraFailure,
+    statusSkipped,
+    statusSucceeded,
+  ];
+
   /// The key of the commit that owns this task.
   @ModelKeyProperty(propertyName: 'ChecklistKey')
   @JsonKey(name: 'ChecklistKey')
@@ -385,10 +393,6 @@ class Task extends Model<int> {
     final int currentBuildNumber = int.parse(buildAddress.split('/').last);
     if (buildNumber == null || buildNumber! < currentBuildNumber) {
       buildNumber = currentBuildNumber;
-      status = statusNew; // Reset status
-      createTimestamp = null;
-      endTimestamp = null;
-      startTimestamp = null;
     } else if (currentBuildNumber < buildNumber!) {
       log.fine('Skipping message as build number is before the current task');
       return;

app_dart/lib/src/request_handlers/postsubmit_luci_subscription.dart

@@ -80,9 +80,21 @@ class PostsubmitLuciSubscription extends SubscriptionHandler {
     }
     log.fine('Found $task');
 
+    // No need to process if
+    // 1) the build is `scheduled`. Task is marked as `In Progress`
+    //    whenever scheduled, either from scheduler/backfiller/rerun. We need to update
+    //    task in datastore only for
+    //    a) `started`: update info like builder number.
+    //    b) `completed`: update info like status.
+    // 2) the task is already complemeted.
+    if (build.status == Status.scheduled || Task.finishedStatusValues.contains(task.status)) {
+      log.fine('skip processing for build with status scheduled or task with status finished.');
+      return Body.empty;
+    }
+    final String oldTaskStatus = task.status;
     task.updateFromBuild(build);
     await datastore.insert(<Task>[task]);
-    log.fine('Updated datastore');
+    log.fine('Updated datastore from $oldTaskStatus to ${task.status}');
 
     final Commit commit = await datastore.lookupByValue<Commit>(commitKey);
     final CiYaml ciYaml = await scheduler.getCiYaml(commit);

app_dart/test/model/task_test.dart

@@ -82,22 +82,23 @@ void main() {
         final pm.Build build = generatePushMessageBuild(
           1,
           buildNumber: 2,
-          status: pm.Status.started,
+          status: pm.Status.completed,
+          result: pm.Result.success,
         );
         final Task task = generateTask(
           1,
           buildNumber: 1,
-          status: Task.statusSucceeded,
+          status: Task.statusInProgress,
         );
 
         expect(task.buildNumberList, '1');
-        expect(task.status, Task.statusSucceeded);
+        expect(task.status, Task.statusInProgress);
 
         task.updateFromBuild(build);
 
         expect(task.buildNumber, 2);
         expect(task.buildNumberList, '1,2');
-        expect(task.status, Task.statusInProgress);
+        expect(task.status, Task.statusSucceeded);
       });
 
       test('does not duplicate build numbers on multiple messages', () {

app_dart/test/request_handlers/postsubmit_luci_subscription_test.dart

@@ -110,6 +110,52 @@ void main() {
     expect(task.endTimestamp, 1565049193786);
   });
 
+  test('skips task processing when build is with scheduled status', () async {
+    final Commit commit = generateCommit(1, sha: '87f88734747805589f2131753620d61b22922822');
+    final Task task = generateTask(
+      4507531199512576,
+      name: 'Linux A',
+      parent: commit,
+      status: Task.statusInProgress,
+    );
+    config.db.values[task.key] = task;
+    config.db.values[commit.key] = commit;
+    tester.message = createBuildbucketPushMessage(
+      'SCHEDULED',
+      builderName: 'Linux A',
+      result: null,
+      userData: '{\\"task_key\\":\\"${task.key.id}\\", \\"commit_key\\":\\"${task.key.parent?.id}\\"}',
+    );
+
+    expect(task.status, Task.statusInProgress);
+    expect(task.attempts, 1);
+    expect(await tester.post(handler), Body.empty);
+    expect(task.status, Task.statusInProgress);
+  });
+
+  test('skips task processing when task has already finished', () async {
+    final Commit commit = generateCommit(1, sha: '87f88734747805589f2131753620d61b22922822');
+    final Task task = generateTask(
+      4507531199512576,
+      name: 'Linux A',
+      parent: commit,
+      status: Task.statusSucceeded,
+    );
+    config.db.values[task.key] = task;
+    config.db.values[commit.key] = commit;
+    tester.message = createBuildbucketPushMessage(
+      'STARTED',
+      builderName: 'Linux A',
+      result: null,
+      userData: '{\\"task_key\\":\\"${task.key.id}\\", \\"commit_key\\":\\"${task.key.parent?.id}\\"}',
+    );
+
+    expect(task.status, Task.statusSucceeded);
+    expect(task.attempts, 1);
+    expect(await tester.post(handler), Body.empty);
+    expect(task.status, Task.statusSucceeded);
+  });
+
   test('does not fail on empty user data', () async {
     tester.message = createBuildbucketPushMessage(
       'COMPLETED',