Migrating task model and backend to use typed_sql.

Author: isoos

app/lib/database/database.dart

@@ -14,7 +14,9 @@ import 'package:pub_dev/database/schema.dart';
 import 'package:pub_dev/service/secret/backend.dart';
 import 'package:pub_dev/shared/configuration.dart';
 import 'package:pub_dev/shared/env_config.dart';
+import 'package:pub_dev/shared/exceptions.dart';
 import 'package:pub_dev/task/clock_control.dart';
+import 'package:retry/retry.dart';
 import 'package:typed_sql/typed_sql.dart';
 
 final _random = Random.secure();
@@ -180,3 +182,23 @@ Future<void> _dropCustomDatabase(String url, String dbName) async {
   await conn.execute('DROP DATABASE "$dbName";');
   await conn.close(force: true);
 }
+
+extension DatabaseExt on Database {
+  Future<K> transactWithRetry<K>(Future<K> Function() fn) async {
+    return await retry(
+      () async {
+        try {
+          return await transact(fn);
+        } on TransactionAbortedException catch (e) {
+          if (e.reason is ResponseException) {
+            // TODO: we should keep and use the original stacktrace in typed_sql's exception
+            throw e.reason;
+          }
+          rethrow;
+        }
+      },
+      maxAttempts: 3,
+      retryIf: (e) => e is DatabaseException,
+    );
+  }
+}

app/lib/search/backend.dart

@@ -85,6 +85,7 @@ void registerSearchIndex(SearchIndex index) =>
 /// Datastore-related access methods for the search service
 class SearchBackend {
   final DatastoreDB _db;
+
   final VersionedJsonStorage _snapshotStorage;
 
   SearchBackend(this._db, Bucket snapshotBucket)
@@ -258,7 +259,7 @@ class SearchBackend {
       addResult(e.name, e.updated);
     }
 
-    await for (final e in _db.tasks.listFinishedSince(updatedThreshold)) {
+    await for (final e in taskBackend.listFinishedSince(updatedThreshold)) {
       addResult(e.package, e.finished);
     }
 

app/lib/service/services.dart

@@ -338,6 +338,7 @@ Future<R> _withPubServices<R>(FutureOr<R> Function() fn) async {
         registerTaskBackend(
           TaskBackend(
             dbService,
+            primaryDatabase!.db,
             storageService.bucket(activeConfiguration.taskResultBucketName!),
           ),
         );