[metrics_center] await future in metrics_center (#1202)

.cirrus.yml

@@ -55,7 +55,7 @@ task:
       always:
         format_script: ./script/tool_runner.sh format --fail-on-change
         license_script: dart pub global run flutter_plugin_tools license-check
-        analyze_script: ./script/tool_runner.sh analyze --custom-analysis=web_benchmarks/testing/test_app,flutter_lints/example,rfw/example
+        analyze_script: ./script/tool_runner.sh analyze --custom-analysis=web_benchmarks/testing/test_app,flutter_lints/example,rfw/example,metrics_center
         pubspec_script: ./script/tool_runner.sh pubspec-check
     - name: publishable
       env:

packages/metrics_center/CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.0.4
+
+- Fix un-await-ed Future in `SkiaPerfDestination.update`.
+
 ## 1.0.3
 
 - Filter out host_name, load_avg and caches keys from context

packages/metrics_center/analysis_options.yaml

@@ -0,0 +1,5 @@
+include: ../../analysis_options.yaml
+
+linter:
+  rules:
+    unawaited_futures: true

packages/metrics_center/lib/src/skiaperf.dart

@@ -407,6 +407,9 @@ class SkiaPerfDestination extends MetricDestination {
       }
     }
 
+    // All created locks must be released before returning
+    final List<Future<void>> lockFutures = <Future<void>>[];
+
     // 2nd, read existing points from the gcs object and update with new ones.
     for (final String repo in pointMap.keys) {
       for (final String? revision in pointMap[repo]!.keys) {
@@ -418,18 +421,21 @@ class SkiaPerfDestination extends MetricDestination {
         // json files according to task names. Skia perf read all json files in
         // the directory so one can use arbitrary names for those sharded json
         // file names.
-        _lock!.protectedRun('$objectName.lock', () async {
-          final List<SkiaPerfPoint> oldPoints =
-              await _gcs.readPoints(objectName);
-          for (final SkiaPerfPoint p in oldPoints) {
-            if (newPoints![p.id] == null) {
-              newPoints[p.id] = p;
+        lockFutures.add(
+          _lock!.protectedRun('$objectName.lock', () async {
+            final List<SkiaPerfPoint> oldPoints =
+                await _gcs.readPoints(objectName);
+            for (final SkiaPerfPoint p in oldPoints) {
+              if (newPoints![p.id] == null) {
+                newPoints[p.id] = p;
+              }
             }
-          }
-          await _gcs.writePoints(objectName, newPoints!.values.toList());
-        });
+            await _gcs.writePoints(objectName, newPoints!.values.toList());
+          }),
+        );
       }
     }
+    await Future.wait(lockFutures);
   }
 
   final SkiaPerfGcsAdaptor _gcs;

packages/metrics_center/pubspec.yaml

@@ -1,5 +1,5 @@
 name: metrics_center
-version: 1.0.3
+version: 1.0.4
 description:
   Support multiple performance metrics sources/formats and destinations.
 repository: https://github.com/flutter/packages/tree/main/packages/metrics_center

packages/metrics_center/test/flutter_test.dart

@@ -44,7 +44,7 @@ void main() {
     final FlutterDestination dst =
         await FlutterDestination.makeFromCredentialsJson(credentialsJson!,
             isTesting: true);
-    dst.update(<FlutterEngineMetricPoint>[simplePoint],
+    await dst.update(<FlutterEngineMetricPoint>[simplePoint],
         DateTime.fromMillisecondsSinceEpoch(123), 'test');
   }, skip: credentialsJson == null);
 }

packages/metrics_center/test/skiaperf_test.dart

@@ -4,6 +4,7 @@
 
 @Timeout(Duration(seconds: 3600))
 
+import 'dart:async';
 import 'dart:convert';
 
 import 'package:gcloud/storage.dart';
@@ -27,6 +28,12 @@ class MockGcsLock implements GcsLock {
 }
 
 class MockSkiaPerfGcsAdaptor implements SkiaPerfGcsAdaptor {
+  MockSkiaPerfGcsAdaptor({
+    this.writePointsOverride,
+  });
+
+  final Future<void> Function()? writePointsOverride;
+
   @override
   Future<List<SkiaPerfPoint>> readPoints(String objectName) async {
     return _storage[objectName] ?? <SkiaPerfPoint>[];
@@ -35,6 +42,9 @@ class MockSkiaPerfGcsAdaptor implements SkiaPerfGcsAdaptor {
   @override
   Future<void> writePoints(
       String objectName, List<SkiaPerfPoint> points) async {
+    if (writePointsOverride != null) {
+      return writePointsOverride!();
+    }
     _storage[objectName] = points.toList();
   }
 
@@ -545,6 +555,33 @@ Future<void> main() async {
     skip: testBucket == null,
   );
 
+  test('SkiaPerfDestination.update awaits locks', () async {
+    bool updateCompleted = false;
+    final Completer<void> callbackCompleter = Completer<void>();
+    final SkiaPerfGcsAdaptor mockGcs = MockSkiaPerfGcsAdaptor(
+      writePointsOverride: () => callbackCompleter.future,
+    );
+    final GcsLock mockLock = MockGcsLock();
+    final SkiaPerfDestination dst = SkiaPerfDestination(mockGcs, mockLock);
+    final Future<void> updateFuture = dst.update(
+      <MetricPoint>[cocoonPointRev1Metric1],
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    );
+    final Future<void> markedUpdateCompleted = updateFuture.then<void>((_) {
+      updateCompleted = true;
+    });
+
+    // spin event loop to make sure function hasn't done anything yet
+    await (Completer<void>()..complete()).future;
+
+    // Ensure that the .update() method is waiting for callbackCompleter
+    expect(updateCompleted, false);
+    callbackCompleter.complete();
+    await markedUpdateCompleted;
+    expect(updateCompleted, true);
+  });
+
   test('SkiaPerfDestination correctly updates points', () async {
     final SkiaPerfGcsAdaptor mockGcs = MockSkiaPerfGcsAdaptor();
     final GcsLock mockLock = MockGcsLock();