From 7c1e52f64aa32a9131355cbffcd89a0c1833ec73 Mon Sep 17 00:00:00 2001 From: Jackson Gardner Date: Fri, 10 Jul 2026 14:40:50 -0700 Subject: [PATCH 1/7] Implement lock-free versioned (revisionId) and Set-based Redis caching for /tasks To eliminate read bottlenecks against the Firestore /tasks (kTaskCollectionId = 'tasks') collection without distributed locking (tryLock), this change introduces an optimistic, multi-tier caching layer across Cocoon (Task, FirestoreQueries, and CacheService): 1. Lock-Free Versioned (revisionId) Payload Caching (tasks subcache): - Every Task document contains a monotonically increasing revisionId integer. - CacheService.insertVersioned(subcacheName, entries) executes an atomic check-and-set via Redis Lua script (EVAL) verifying entry.revisionId > cachedRevisionId before updating payloads. - Chunked batching (batchSize = 20) guarantees <0.1ms Lua execution without starving concurrent readers (MGET). 2. Native Set-Based Commit Task Indexing (tasks_by_commit_ids): - Replaces monolithic JSON array strings with native Redis Sets (SMEMBERS, SADD) via getSet, updateSet, and addToSetIfExists. - Leverages two domain invariants: 1. Immutable commitSha: Task mutations (updateCacheForTaskMutations) never change a commit's task list membership and do not touch or invalidate tasks_by_commit_ids. 2. Monotonically Increasing Set: Task attempts only grow over time, so unioning IDs via SADD (updateCacheForCreatedTasks) safely converges without lock coordination. 3. Partial Cache Recovery (_queryTasksByCommitCached): - When reading commit tasks, _queryTasksByCommitCached retrieves cached task IDs (SMEMBERS) and performs a batch lookup (MGET). - If individual payloads are expired or missing (missingDocIds), _queryTasksByCommitCached selectively queries getDocument only for missing entries and merges them with foundTasks, eliminating redundant full-commit queries. 4. Modular & Explicit Cache Handlers (FirestoreQueries): - _cacheTaskDocuments(tasks): Reusable helper for versioned Task payload insertions. - _fetchAndCacheCommitTasks(commitSha): Reusable slow-path query helper shared across read-miss and task-creation-miss paths. - updateCacheForCreatedTasks(tasks) and updateCacheForTaskMutations(writes): Explicit domain handlers replacing generic ad-hoc cache invalidations. - _subcacheRecentTasksIds and _queryRecentTasksByNameCached removed in favor of _cacheTaskDocuments warming on query execution (~65 lines simplified). --- app_dart/bin/gae_server.dart | 5 +- app_dart/lib/src/model/firestore/task.dart | 62 ++- app_dart/lib/src/service/cache_service.dart | 330 ++++++++++++++- app_dart/lib/src/service/firestore.dart | 400 ++++++++++++++++-- app_dart/lib/src/service/scheduler.dart | 8 +- .../test/service/firestore_cache_test.dart | 222 ++++++++++ .../lib/src/fakes/fake_cache_service.dart | 32 ++ .../lib/src/fakes/fake_firestore_service.dart | 61 ++- 8 files changed, 1060 insertions(+), 60 deletions(-) create mode 100644 app_dart/test/service/firestore_cache_test.dart diff --git a/app_dart/bin/gae_server.dart b/app_dart/bin/gae_server.dart index 7c65385fe1..3f03cb9fb6 100644 --- a/app_dart/bin/gae_server.dart +++ b/app_dart/bin/gae_server.dart @@ -43,7 +43,10 @@ Future main() async { } final cache = CacheService.redis(); - final firestore = await FirestoreService.from(const GoogleAuthProvider()); + final firestore = await FirestoreService.from( + const GoogleAuthProvider(), + cache: cache, + ); final bigQuery = await BigQueryService.from(const GoogleAuthProvider()); // Start with a fresh copy of the DynamicConfig. If this throws, the server diff --git a/app_dart/lib/src/model/firestore/task.dart b/app_dart/lib/src/model/firestore/task.dart index e049e8e700..0035d70676 100644 --- a/app_dart/lib/src/model/firestore/task.dart +++ b/app_dart/lib/src/model/firestore/task.dart @@ -5,6 +5,9 @@ /// @docImport 'commit.dart'; library; +import 'dart:convert'; +import 'dart:typed_data'; + import 'package:buildbucket/buildbucket_pb.dart' as bbv2; import 'package:cocoon_common/task_status.dart'; import 'package:googleapis/firestore/v1.dart' hide Status; @@ -113,6 +116,7 @@ final class Task extends AppDocument { static const fieldStatus = 'status'; static const fieldTestFlaky = 'testFlaky'; static const fieldAttempt = 'attempt'; + static const fieldRevisionId = 'revisionId'; /// Returns a document ID for a task from the given parameters. static AppDocumentId documentIdFor({ @@ -136,6 +140,19 @@ final class Task extends AppDocument { fromDocument: Task.fromDocument, ); + static Uint8List _serializeTask(Task task) { + return Uint8List.fromList( + utf8.encode( + json.encode(Document(name: task.name, fields: task.fields).toJson()), + ), + ); + } + + static Task _deserializeTask(Uint8List data) { + final jsonMap = json.decode(utf8.decode(data)) as Map; + return Task.fromDocument(Document.fromJson(jsonMap)); + } + /// Lookup [Task] from Firestore. /// /// `documentName` follows `/projects/{project}/databases/{database}/documents/{document_path}` @@ -143,10 +160,32 @@ final class Task extends AppDocument { FirestoreService firestoreService, AppDocumentId id, ) async { + if (firestoreService.cache != null) { + final cachedTask = await firestoreService.cache!.get( + 'tasks', + id.documentId, + ); + if (cachedTask != null) { + return _deserializeTask(cachedTask); + } + } + final document = await firestoreService.getDocument( p.posix.join(kDatabase, 'documents', kTaskCollectionId, id.documentId), ); - return Task.fromDocument(document); + final task = Task.fromDocument(document); + + if (firestoreService.cache != null) { + await firestoreService.cache!.insertVersioned('tasks', [ + VersionedCacheEntry( + key: id.documentId, + value: _serializeTask(task), + revisionId: task.revisionId, + ttl: const Duration(hours: 12), + ), + ]); + } + return task; } factory Task({ @@ -178,6 +217,7 @@ final class Task extends AppDocument { fieldStatus: status.value.toValue(), fieldTestFlaky: testFlaky.toValue(), fieldAttempt: currentAttempt.toValue(), + fieldRevisionId: 1.toValue(), }, name: p.posix.join( kDatabase, @@ -229,12 +269,26 @@ final class Task extends AppDocument { fields: {fieldStatus: Value(stringValue: status.value)}, ), updateMask: DocumentMask(fieldPaths: [fieldStatus]), + updateTransforms: [ + FieldTransform( + fieldPath: fieldRevisionId, + increment: Value(integerValue: '1'), + ), + ], ); } /// The task was run successfully. static const statusSucceeded = TaskStatus.succeeded; + int get revisionId => fields.containsKey(fieldRevisionId) + ? int.parse(fields[fieldRevisionId]!.integerValue!) + : 1; + + void incrementRevisionId() { + fields[fieldRevisionId] = (revisionId + 1).toValue(); + } + /// The timestamp (in milliseconds since the Epoch) that this task was /// created. /// @@ -308,14 +362,17 @@ final class Task extends AppDocument { void setStatus(TaskStatus status) { fields[fieldStatus] = status.value.toValue(); + incrementRevisionId(); } void setEndTimestamp(int endTimestamp) { fields[fieldEndTimestamp] = endTimestamp.toValue(); + incrementRevisionId(); } void setTestFlaky(bool testFlaky) { fields[fieldTestFlaky] = testFlaky.toValue(); + incrementRevisionId(); } void updateFromBuild(bbv2.Build build) { @@ -335,6 +392,7 @@ final class Task extends AppDocument { .toValue(); _setStatusFromLuciStatus(build); + incrementRevisionId(); } void resetAsRetry({int? attempt, DateTime? now}) { @@ -360,11 +418,13 @@ final class Task extends AppDocument { fieldTestFlaky: false.toValue(), fieldCommitSha: commitSha.toValue(), fieldAttempt: attempt.toValue(), + fieldRevisionId: 1.toValue(), }; } void setBuildNumber(int buildNumber) { fields[fieldBuildNumber] = buildNumber.toValue(); + incrementRevisionId(); } void _setStatusFromLuciStatus(bbv2.Build build) { diff --git a/app_dart/lib/src/service/cache_service.dart b/app_dart/lib/src/service/cache_service.dart index d3263b0cc9..c3510ab823 100644 --- a/app_dart/lib/src/service/cache_service.dart +++ b/app_dart/lib/src/service/cache_service.dart @@ -12,6 +12,20 @@ import 'package:meta/meta.dart'; import 'package:mutex/mutex.dart'; import 'package:redis/redis.dart'; +class VersionedCacheEntry { + const VersionedCacheEntry({ + required this.key, + required this.value, + required this.revisionId, + this.ttl = const Duration(hours: 12), + }); + + final String key; + final Uint8List value; + final int revisionId; + final Duration ttl; +} + /// Service for reading and writing values to a cache for quick access of data. abstract class CacheService { CacheService(); @@ -28,6 +42,9 @@ abstract class CacheService { /// Get value of [key] from the subcache [subcacheName]. Future get(String subcacheName, String key); + /// Get values for multiple [keys] from the subcache [subcacheName] in a single batch API call. + Future> getMulti(String subcacheName, List keys); + /// Set [value] for [key] in the subcache [subcacheName] with [ttl]. Future set( String subcacheName, @@ -48,6 +65,31 @@ abstract class CacheService { Duration ttl = const Duration(minutes: 1), }); + /// Atomically inserts multiple [entries] into [subcacheName] in a single batch API call + /// if and only if their [VersionedCacheEntry.revisionId] is strictly greater than any + /// existing cached revision for that key. + Future insertVersioned( + String subcacheName, + List entries, + ); + + /// Get the set of string values for [key] from the subcache [subcacheName]. + Future> getSet(String subcacheName, String key); + + /// Atomically adds [values] to the set at [key] in [subcacheName]. + /// If the set does not yet exist, it is created with [ttl]. + /// If the set already exists, all [values] are added to it without altering its TTL. + Future updateSet( + String subcacheName, + String key, + Set values, { + Duration ttl = const Duration(hours: 12), + }); + + /// Atomically adds [value] to the set at [key] in [subcacheName] if and only if the set already exists. + /// Returns `true` if the set existed and [value] was added, or `false` if the set did not exist. + Future addToSetIfExists(String subcacheName, String key, String value); + /// Get value of [key] from the subcache [subcacheName]. If the key has no /// value, call [createFn] to create a value for it, set it, and return it. Future getOrCreate( @@ -183,6 +225,163 @@ class RedisCacheService extends CacheService { } } + @override + Future> getMulti( + String subcacheName, + List keys, + ) async { + if (keys.isEmpty) return const []; + try { + final redisKeys = keys.map((k) => '$subcacheName/$k').toList(); + final values = await _runCommand( + (client) => client.send_object(['MGET', ...redisKeys]), + ); + if (values is! List) { + return List.filled(keys.length, null); + } + return values.map((value) { + if (value == null) return null; + return base64.decode(value as String); + }).toList(); + } catch (e) { + log.warn('Unable to retrieve multi-values from cache.', e); + return List.filled(keys.length, null); + } + } + + @override + Future insertVersioned( + String subcacheName, + List entries, + ) async { + if (entries.isEmpty) return; + const insertVersionedScript = ''' + local numKeys = tonumber(ARGV[1]) + local subcache = ARGV[2] + for i = 1, numKeys do + local key = KEYS[i] + local val = ARGV[2 + (i - 1) * 3 + 1] + local rev = tonumber(ARGV[2 + (i - 1) * 3 + 2]) + local ttl = tonumber(ARGV[2 + (i - 1) * 3 + 3]) + + local existingRev = tonumber(redis.call("hget", "revisions:" .. subcache, key) or 0) + if rev > existingRev or (not redis.call("exists", key)) then + redis.call("set", key, val, "PX", ttl) + redis.call("hset", "revisions:" .. subcache, key, rev) + end + end + return 1 + '''; + const batchSize = 20; + for (var i = 0; i < entries.length; i += batchSize) { + final chunk = entries.sublist(i, min(i + batchSize, entries.length)); + try { + final keys = chunk.map((e) => '$subcacheName/${e.key}').toList(); + final args = [ + chunk.length.toString(), + subcacheName, + for (final e in chunk) ...[ + base64.encode(e.value), + e.revisionId.toString(), + e.ttl.inMilliseconds.toString(), + ], + ]; + await _runCommand( + (client) => client.send_object([ + 'EVAL', + insertVersionedScript, + keys.length.toString(), + ...keys, + ...args, + ]), + ); + } catch (e) { + log.warn('Unable to insert versioned entries into cache.', e); + } + } + } + + @override + Future> getSet(String subcacheName, String key) async { + final redisKey = '$subcacheName/$key'; + try { + final values = await _runCommand( + (client) => client.send_object(['SMEMBERS', redisKey]), + ); + if (values is! List || values.isEmpty) { + return const {}; + } + return values.map((e) => e.toString()).toSet(); + } catch (e) { + log.warn('Unable to retrieve set for $key from cache.', e); + return const {}; + } + } + + @override + Future updateSet( + String subcacheName, + String key, + Set values, { + Duration ttl = const Duration(hours: 12), + }) async { + if (values.isEmpty) return; + const updateSetScript = ''' + if redis.call("exists", KEYS[1]) == 1 then + for i = 1, #ARGV - 1 do + redis.call("sadd", KEYS[1], ARGV[i]) + end + else + for i = 1, #ARGV - 1 do + redis.call("sadd", KEYS[1], ARGV[i]) + end + redis.call("pexpire", KEYS[1], tonumber(ARGV[#ARGV])) + end + return 1 + '''; + final redisKey = '$subcacheName/$key'; + try { + final args = [...values, ttl.inMilliseconds.toString()]; + await _runCommand( + (client) => client.send_object([ + 'EVAL', + updateSetScript, + '1', + redisKey, + ...args, + ]), + ); + } catch (e) { + log.warn('Unable to update set for $key in cache.', e); + } + } + + @override + Future addToSetIfExists( + String subcacheName, + String key, + String value, + ) async { + const addToSetScript = ''' + if redis.call("exists", KEYS[1]) == 1 then + redis.call("sadd", KEYS[1], ARGV[1]) + return 1 + end + return 0 + '''; + final redisKey = '$subcacheName/$key'; + try { + final response = await _runCommand( + (client) => + client.send_object(['EVAL', addToSetScript, '1', redisKey, value]), + ); + return response == 1 || response == '1'; + } catch (e) { + log.warn('Unable to add to set for $key in cache.', e); + return false; + } + } + @override Future set( String subcacheName, @@ -246,6 +445,10 @@ class RedisCacheService extends CacheService { final redisKey = '$subcacheName/$key'; try { await _runCommand((client) => client.send_object(['DEL', redisKey])); + await _runCommand( + (client) => + client.send_object(['HDEL', 'revisions:$subcacheName', redisKey]), + ); } catch (e) { log.warn('Unable to purge value for $key from cache.', e); } @@ -308,6 +511,7 @@ class InMemoryCacheService extends CacheService { final int maxEntries; final Map _entries = {}; + final Map _sets = {}; final Map _locks = {}; final _mutex = Mutex(); @@ -328,6 +532,119 @@ class InMemoryCacheService extends CacheService { } } + @override + Future> getMulti( + String subcacheName, + List keys, + ) async { + await _mutex.acquire(); + try { + return keys.map((key) { + final cacheKey = '$subcacheName/$key'; + final entry = _entries[cacheKey]; + if (entry == null || entry.isExpired) { + _entries.remove(cacheKey); + return null; + } + return entry.value; + }).toList(); + } finally { + _mutex.release(); + } + } + + @override + Future insertVersioned( + String subcacheName, + List entries, + ) async { + if (entries.isEmpty) return; + await _mutex.acquire(); + try { + _entries.removeWhere((k, v) => v.isExpired); + for (final entry in entries) { + final cacheKey = '$subcacheName/${entry.key}'; + final existing = _entries[cacheKey]; + if (existing == null || + existing.isExpired || + entry.revisionId > existing.revisionId) { + if (_entries.length >= maxEntries && + !_entries.containsKey(cacheKey)) { + _entries.remove(_entries.keys.first); + } + _entries[cacheKey] = _InMemoryCacheEntry( + entry.value, + DateTime.now().add(entry.ttl), + revisionId: entry.revisionId, + ); + } + } + } finally { + _mutex.release(); + } + } + + @override + Future> getSet(String subcacheName, String key) async { + await _mutex.acquire(); + try { + final cacheKey = '$subcacheName/$key'; + final entry = _sets[cacheKey]; + if (entry == null || entry.isExpired) { + _sets.remove(cacheKey); + return const {}; + } + return Set.of(entry.values); + } finally { + _mutex.release(); + } + } + + @override + Future updateSet( + String subcacheName, + String key, + Set values, { + Duration ttl = const Duration(hours: 12), + }) async { + if (values.isEmpty) return; + await _mutex.acquire(); + try { + final cacheKey = '$subcacheName/$key'; + final existing = _sets[cacheKey]; + if (existing != null && !existing.isExpired) { + existing.values.addAll(values); + } else { + _sets[cacheKey] = _InMemorySetEntry( + Set.of(values), + DateTime.now().add(ttl), + ); + } + } finally { + _mutex.release(); + } + } + + @override + Future addToSetIfExists( + String subcacheName, + String key, + String value, + ) async { + await _mutex.acquire(); + try { + final cacheKey = '$subcacheName/$key'; + final existing = _sets[cacheKey]; + if (existing != null && !existing.isExpired) { + existing.values.add(value); + return true; + } + return false; + } finally { + _mutex.release(); + } + } + @override Future set( String subcacheName, @@ -391,6 +708,7 @@ class InMemoryCacheService extends CacheService { try { final cacheKey = '$subcacheName/$key'; _entries.remove(cacheKey); + _sets.remove(cacheKey); } finally { _mutex.release(); } @@ -430,10 +748,20 @@ class InMemoryCacheService extends CacheService { } class _InMemoryCacheEntry { - _InMemoryCacheEntry(this.value, this.expiresAt); + _InMemoryCacheEntry(this.value, this.expiresAt, {this.revisionId = 0}); final Uint8List value; final DateTime expiresAt; + final int revisionId; + + bool get isExpired => DateTime.now().isAfter(expiresAt); +} + +class _InMemorySetEntry { + _InMemorySetEntry(this.values, this.expiresAt); + + final Set values; + final DateTime expiresAt; bool get isExpired => DateTime.now().isAfter(expiresAt); } diff --git a/app_dart/lib/src/service/firestore.dart b/app_dart/lib/src/service/firestore.dart index a9fa9cf7c4..e1532c6f8f 100644 --- a/app_dart/lib/src/service/firestore.dart +++ b/app_dart/lib/src/service/firestore.dart @@ -3,6 +3,8 @@ // found in the LICENSE file. import 'dart:async'; +import 'dart:convert'; +import 'dart:typed_data'; import 'package:cocoon_common/core_extensions.dart'; import 'package:cocoon_common/task_status.dart'; @@ -11,6 +13,7 @@ import 'package:cocoon_server/google_auth_provider.dart'; import 'package:github/github.dart'; import 'package:googleapis/firestore/v1.dart'; import 'package:meta/meta.dart'; +import 'package:path/path.dart' as p; import '../../cocoon_service.dart'; import '../model/firestore/github_build_status.dart'; @@ -49,7 +52,203 @@ final kFieldMapRegExp = RegExp( ); @visibleForTesting +/// Mixin that encapsulates Firestore query and caching operations across Cocoon. +/// +/// ## Lock-Free Versioned & Set-Based Caching Strategy +/// +/// To resolve read bottlenecks against the Firestore `/tasks` collection without incurring +/// distributed locking overhead, this mixin implements a two-tier optimistic caching architecture: +/// +/// ### 1. Individual Task Payload Caching (`tasks` subcache) +/// - **Single Source of Truth**: Full [Task] document payloads are serialized to JSON (`_serializeTask`) +/// and stored in the `tasks` subcache keyed by exact document ID (`$commitSha_$taskName_$attempt`). +/// - **Optimistic Concurrency (`revisionId`)**: Every task document contains a monotonically increasing integer +/// `revisionId`. Updates and insertions use [CacheService.insertVersioned], which runs an atomic check-and-set +/// ensuring payload updates are only applied if `newRevisionId > cachedRevisionId`. +/// - **Chunked Batching**: Multi-task insertions (`_cacheTaskDocuments`) are chunked in batches of 20 (`batchSize = 20`) +/// via atomic Redis Lua (`EVAL`) scripts, guaranteeing execution in `<0.1ms` without starving concurrent readers. +/// +/// ### 2. Commit Task List Set Caching (`tasks_by_commit_ids` subcache) +/// - **Native Set Operations**: Instead of storing monolithic serialized JSON arrays of task IDs, commit task indices +/// are stored in native Redis Sets (`SMEMBERS`, `SADD`), backed by two fundamental domain invariants: +/// 1. **Immutable `commitSha`**: A task's commit association never changes after creation. Therefore, status or field +/// mutations (`patchStatus`) never alter the membership of a commit's task list (`updateCacheForTaskMutations` +/// updates individual `tasks/$docId` entries in-place lock-free and never touches or invalidates `tasks_by_commit_ids`). +/// 2. **Monotonically Increasing Set**: Task attempts for a commit only grow over time. Adding task IDs via `SADD` +/// ([CacheService.updateSet], [CacheService.addToSetIfExists]) safely converges to database state without locks. +/// +/// ### 3. Partial Cache Recovery (`_queryTasksByCommitCached`) +/// - When querying all tasks for a commit, we retrieve the cached Set of document IDs (`SMEMBERS`) and perform a batch +/// payload lookup across `tasks` (`MGET`). +/// - If any individual task entries have expired or are missing from `tasks`, successfully retrieved cached tasks +/// ([foundTasks]) are preserved immediately. Missing document IDs (`missingDocIds`) are queried individually via +/// [getDocument], inserted into `tasks` via `insertVersioned`, and combined with [foundTasks]—avoiding a full table query. +/// - If `tasks_by_commit_ids` is missing entirely (`docIds.isEmpty`), we execute a full query (`_fetchAndCacheCommitTasks`), +/// populate `tasks` for all items, and initialize the commit Set via [CacheService.updateSet]. mixin FirestoreQueries { + CacheService? get cache => null; + Future getDocument(String name, {Transaction? transaction}); + + static const String _subcacheTasksByCommitIds = 'tasks_by_commit_ids'; + + static Uint8List _serializeTask(Task task) { + return Uint8List.fromList( + utf8.encode( + json.encode(Document(name: task.name, fields: task.fields).toJson()), + ), + ); + } + + static Task _deserializeTask(Uint8List data) { + final jsonMap = json.decode(utf8.decode(data)) as Map; + return Task.fromDocument(Document.fromJson(jsonMap)); + } + + /// Caches the individual [Task] payloads into the `tasks` subcache. + Future _cacheTaskDocuments( + List tasks, { + Duration ttl = const Duration(hours: 12), + }) async { + if (cache == null || tasks.isEmpty) return; + final entries = tasks + .map( + (t) => VersionedCacheEntry( + key: p.basename(t.name!), + value: _serializeTask(t), + revisionId: t.revisionId, + ttl: ttl, + ), + ) + .toList(); + await cache!.insertVersioned('tasks', entries); + } + + /// Queries all tasks for [commitSha] from Firestore, updates their individual cache entries, + /// and populates the `tasks_by_commit_ids` set. + Future> _fetchAndCacheCommitTasks(String commitSha) async { + final documents = await query( + kTaskCollectionId, + {'${Task.fieldCommitSha} =': commitSha}, + orderMap: {Task.fieldCreateTimestamp: kQueryOrderDescending}, + ); + final tasks = documents.map(Task.fromDocument).toList(); + await _cacheTaskDocuments(tasks); + final docIds = tasks.map((t) => p.basename(t.name!)).toSet(); + if (docIds.isNotEmpty && cache != null) { + await cache!.updateSet( + _subcacheTasksByCommitIds, + commitSha, + docIds, + ttl: const Duration(hours: 12), + ); + } + return tasks; + } + + /// Explicitly updates the cache when new [Task]s are created. + Future updateCacheForCreatedTasks(List tasks) async { + if (cache == null || tasks.isEmpty) return; + await _cacheTaskDocuments(tasks); + for (final task in tasks) { + if (task.commitSha.isNotEmpty) { + final docId = p.basename(task.name!); + final added = await cache!.addToSetIfExists( + _subcacheTasksByCommitIds, + task.commitSha, + docId, + ); + if (!added) { + await _fetchAndCacheCommitTasks(task.commitSha); + } + } + } + } + + /// Explicitly updates the cache when existing [Task]s are mutated (e.g. status updates). + /// + /// Because commit membership is immutable, mutations never alter `tasks_by_commit_ids`. + Future updateCacheForTaskMutations(List writes) async { + if (cache == null || writes.isEmpty) return; + final mutatedTasks = []; + + for (final write in writes) { + final docName = + write.update?.name ?? write.delete ?? write.transform?.document; + if (docName == null || + !docName.contains('/documents/$kTaskCollectionId/')) { + continue; + } + final docId = p.basename(docName); + + if (write.update != null && + write.updateMask == null && + write.update!.fields != null && + write.update!.fields!.containsKey(Task.fieldCreateTimestamp)) { + try { + mutatedTasks.add(Task.fromDocument(write.update!)); + } catch (_) {} + } else { + final cachedBytes = await cache!.get('tasks', docId); + if (cachedBytes != null) { + try { + final cachedTask = _deserializeTask(cachedBytes); + if (write.update?.fields != null) { + cachedTask.fields.addAll(write.update!.fields!); + } + if (write.updateTransforms != null) { + for (final transform in write.updateTransforms!) { + if (transform.fieldPath == Task.fieldRevisionId && + transform.increment != null) { + cachedTask.incrementRevisionId(); + } + } + } else { + cachedTask.incrementRevisionId(); + } + mutatedTasks.add(cachedTask); + } catch (_) {} + } + } + } + + if (mutatedTasks.isNotEmpty) { + await _cacheTaskDocuments(mutatedTasks); + } + } + + @visibleForTesting + Future invalidateCacheForWrites(List writes) async { + if (cache == null || writes.isEmpty) return; + final createdTasks = []; + final mutationWrites = []; + + for (final write in writes) { + final docName = + write.update?.name ?? write.delete ?? write.transform?.document; + if (docName == null || + !docName.contains('/documents/$kTaskCollectionId/')) { + continue; + } + if (write.update != null && + write.updateMask == null && + write.update!.fields != null && + write.update!.fields!.containsKey(Task.fieldCreateTimestamp)) { + try { + createdTasks.add(Task.fromDocument(write.update!)); + } catch (_) {} + } else { + mutationWrites.add(write); + } + } + + if (createdTasks.isNotEmpty) { + await updateCacheForCreatedTasks(createdTasks); + } + if (mutationWrites.isNotEmpty) { + await updateCacheForTaskMutations(mutationWrites); + } + } + /// Wrapper to simplify Firestore query. /// /// The [filterMap] follows format: @@ -115,21 +314,20 @@ mixin FirestoreQueries { return [...documents.map(Commit.fromDocument)]; } - Future> _queryTasks({ - required int? limit, + Future> _queryTasksFromFirestore({ + int? limit, String? name, TaskStatus? status, String? commitSha, Transaction? transaction, + bool cacheResults = true, }) async { - final filterMap = { + final filterMap = { '${Task.fieldName} =': ?name, if (status != null) '${Task.fieldStatus} =': status.value, '${Task.fieldCommitSha} =': ?commitSha, }; - // Avoid a full table-scan. - // TODO(matanlurey): Debatably this should be in the root query method. if (limit == null && filterMap.isEmpty) { throw ArgumentError.value( limit, @@ -138,31 +336,145 @@ mixin FirestoreQueries { ); } - // For tasks, therer is no reason to _not_ order this way. final orderMap = {Task.fieldCreateTimestamp: kQueryOrderDescending}; final documents = await query( kTaskCollectionId, filterMap, orderMap: orderMap, + limit: limit, + transaction: transaction, + ); + final tasks = documents.map(Task.fromDocument).toList(); + if (cacheResults && transaction == null && cache != null) { + await _cacheTaskDocuments(tasks, ttl: const Duration(hours: 4)); + } + return tasks; + } + + Future> _queryTasksByCommit({ + required String commitSha, + String? name, + TaskStatus? status, + int? limit, + Transaction? transaction, + }) async { + if (transaction == null && cache != null) { + return await _queryTasksByCommitCached( + commitSha: commitSha, + name: name, + status: status, + limit: limit, + ); + } + return await _queryTasksFromFirestore( + commitSha: commitSha, + name: name, + status: status, + limit: limit, transaction: transaction, + cacheResults: false, ); - return [...documents.map(Task.fromDocument)]; } - /// Queries for recent [Task]s. + Future> _queryTasksByCommitCached({ + required String commitSha, + String? name, + TaskStatus? status, + int? limit, + }) async { + List? tasks; + final docIds = await cache!.getSet(_subcacheTasksByCommitIds, commitSha); + if (docIds.isNotEmpty) { + final docIdList = docIds.toList(); + final cachedTasksData = await cache!.getMulti('tasks', docIdList); + final foundTasks = []; + final missingDocIds = []; + + for (var i = 0; i < docIdList.length; i++) { + final data = cachedTasksData[i]; + if (data != null) { + try { + foundTasks.add(_deserializeTask(data)); + } catch (_) { + missingDocIds.add(docIdList[i]); + } + } else { + missingDocIds.add(docIdList[i]); + } + } + + if (missingDocIds.isEmpty) { + tasks = foundTasks; + } else { + final fetchedMissingTasks = []; + for (final missingId in missingDocIds) { + try { + final document = await getDocument( + p.posix.join( + kDatabase, + 'documents', + kTaskCollectionId, + missingId, + ), + ); + fetchedMissingTasks.add(Task.fromDocument(document)); + } catch (_) {} + } + if (fetchedMissingTasks.isNotEmpty) { + await _cacheTaskDocuments(fetchedMissingTasks); + foundTasks.addAll(fetchedMissingTasks); + } + tasks = foundTasks; + } + } + + tasks ??= await _fetchAndCacheCommitTasks(commitSha); + + var result = tasks.toList(); + if (name != null) { + result = result.where((t) => t.taskName == name).toList(); + } + if (status != null) { + result = result.where((t) => t.status == status).toList(); + } + result.sort((a, b) => b.createTimestamp.compareTo(a.createTimestamp)); + if (limit != null) { + result = result.take(limit).toList(); + } + return result; + } + + /// Queries for recent [Task]s across commits matching [name]. /// - /// If other named arguments are provided, they are used as a query filter. - Future> queryRecentTasks({ + /// If [status] is provided, only tasks matching the status are returned. + Future> queryRecentTasksByName({ + required String name, int limit = 100, - String? name, TaskStatus? status, - String? commitSha, }) async { - return await _queryTasks( + return await _queryTasksFromFirestore( limit: limit, name: name, status: status, + ); + } + + /// Queries [Task]s running against the specified [commitSha]. + /// + /// If [name] is provided, only tasks matching the task name are returned. + /// If [status] is provided, only tasks matching the status are returned. + /// If [limit] is provided, at most [limit] tasks are returned. + Future> queryRecentTasksByCommit({ + required String commitSha, + String? name, + TaskStatus? status, + int? limit, + }) async { + return await _queryTasksByCommit( commitSha: commitSha, + name: name, + status: status, + limit: limit, ); } @@ -173,8 +485,7 @@ mixin FirestoreQueries { String? name, Transaction? transaction, }) async { - return await _queryTasks( - limit: null, + return await _queryTasksByCommit( commitSha: commitSha, status: status, name: name, @@ -249,17 +560,12 @@ mixin FirestoreQueries { required String commitSha, required String builderName, }) async { - final tasks = await query( - kTaskCollectionId, - { - '${Task.fieldCommitSha} =': commitSha, - '${Task.fieldName} =': builderName, - }, - // Assumes the invariant where the newest task has the highest attempt #. - orderMap: {Task.fieldCreateTimestamp: kQueryOrderDescending}, + final tasks = await _queryTasksByCommit( + commitSha: commitSha, + name: builderName, limit: 1, ); - return tasks.isEmpty ? null : Task.fromDocument(tasks.first); + return tasks.isEmpty ? null : tasks.first; } /// Queries the last updated build status for the [slug], [prNumber], and [head]. @@ -310,7 +616,10 @@ mixin FirestoreQueries { class FirestoreService with FirestoreQueries { /// Creates a [FirestoreService] using Google API authentication. - static Future from(GoogleAuthProvider authProvider) async { + static Future from( + GoogleAuthProvider authProvider, { + CacheService? cache, + }) async { final client = await authProvider.createClient( scopes: const [FirestoreApi.datastoreScope], baseClient: FirestoreBaseClient( @@ -318,13 +627,17 @@ class FirestoreService with FirestoreQueries { databaseId: Config.flutterGcpFirestoreDatabase, ), ); - return FirestoreService._(FirestoreApi(client)); + return FirestoreService._(FirestoreApi(client), cache: cache); } - const FirestoreService._(this._api); + const FirestoreService._(this._api, {this.cache}); final FirestoreApi _api; + @override + final CacheService? cache; + /// Gets a document based on name. + @override Future getDocument(String name, {Transaction? transaction}) async { return _api.projects.databases.documents.get( name, @@ -342,12 +655,16 @@ class FirestoreService with FirestoreQueries { required String collectionId, String? documentId, }) async { - return _api.projects.databases.documents.createDocument( + final result = await _api.projects.databases.documents.createDocument( document, '$kDatabase/documents', collectionId, documentId: documentId, ); + if (cache != null && collectionId == kTaskCollectionId) { + await invalidateCacheForWrites([Write(update: result)]); + } + return result; } /// Batch writes documents to Firestore. @@ -360,7 +677,14 @@ class FirestoreService with FirestoreQueries { BatchWriteRequest request, String database, ) async { - return _api.projects.databases.documents.batchWrite(request, database); + final result = await _api.projects.databases.documents.batchWrite( + request, + database, + ); + if (cache != null && request.writes != null) { + await invalidateCacheForWrites(request.writes!); + } + return result; } /// Begins a read-write transaction. @@ -384,7 +708,14 @@ class FirestoreService with FirestoreQueries { transaction: transaction.identifier, writes: writes, ); - return await _api.projects.databases.documents.commit(request, kDatabase); + final result = await _api.projects.databases.documents.commit( + request, + kDatabase, + ); + if (cache != null) { + await invalidateCacheForWrites(writes); + } + return result; } /// Rolls back a transaction. @@ -406,7 +737,14 @@ class FirestoreService with FirestoreQueries { transaction: beginTransactionResponse.transaction, writes: writes, ); - return _api.projects.databases.documents.commit(commitRequest, kDatabase); + final result = await _api.projects.databases.documents.commit( + commitRequest, + kDatabase, + ); + if (cache != null) { + await invalidateCacheForWrites(writes); + } + return result; } /// Returns Firestore [Value] based on corresponding object type. diff --git a/app_dart/lib/src/service/scheduler.dart b/app_dart/lib/src/service/scheduler.dart index ced9831a60..85786a3266 100644 --- a/app_dart/lib/src/service/scheduler.dart +++ b/app_dart/lib/src/service/scheduler.dart @@ -214,7 +214,9 @@ class Scheduler { final priority = await target.schedulerPolicy.triggerPriority( taskName: task.taskName, commitSha: commit.sha, - recentTasks: await _firestore.queryRecentTasks(name: task.taskName), + recentTasks: await _firestore.queryRecentTasksByName( + name: task.taskName, + ), ); if (priority != null) { // Mark task as in progress to ensure it isn't scheduled over @@ -1727,10 +1729,10 @@ $stacktrace final fs.Task fsTask; // Query the lastest run of the `checkName` againt commit `sha`. - final fsTasks = await _firestore.queryRecentTasks( - limit: 1, + final fsTasks = await _firestore.queryRecentTasksByCommit( commitSha: fsCommit.sha, name: checkName, + limit: 1, ); if (fsTasks.isEmpty) { throw StateError('Expected 1+ tasks for $checkName'); diff --git a/app_dart/test/service/firestore_cache_test.dart b/app_dart/test/service/firestore_cache_test.dart new file mode 100644 index 0000000000..903eebade3 --- /dev/null +++ b/app_dart/test/service/firestore_cache_test.dart @@ -0,0 +1,222 @@ +// Copyright 2026 The Flutter Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +import 'package:cocoon_common/task_status.dart'; +import 'package:cocoon_integration_test/testing.dart'; +import 'package:cocoon_service/src/model/firestore/task.dart'; +import 'package:cocoon_service/src/service/cache_service.dart'; +import 'package:cocoon_service/src/service/firestore.dart'; +import 'package:googleapis/firestore/v1.dart'; +import 'package:test/test.dart'; + +void main() { + group('Task Firestore Cache', () { + late FakeFirestoreService firestore; + late CacheService cache; + const commitSha = 'testSha'; + + setUp(() { + cache = CacheService.inMemory(); + firestore = FakeFirestoreService(cache: cache); + }); + + test( + 'Task.fromFirestore caches task and returns from cache on second call', + () async { + final task = generateFirestoreTask( + 1, + name: 'Linux A', + attempts: 1, + status: TaskStatus.inProgress, + ); + firestore.putDocument(task); + + final fetched1 = await Task.fromFirestore( + firestore, + TaskId( + commitSha: commitSha, + taskName: task.taskName, + currentAttempt: task.currentAttempt, + ), + ); + expect(fetched1.status, TaskStatus.inProgress); + expect(fetched1.revisionId, 1); + + // Directly update firestore without using a service write so we know if cache is used on second read + final rawDoc = firestore.documents.firstWhere( + (d) => d.name == task.name, + ); + rawDoc.fields![Task.fieldStatus] = Value( + stringValue: TaskStatus.succeeded.value, + ); + + final fetched2 = await Task.fromFirestore( + firestore, + TaskId( + commitSha: commitSha, + taskName: task.taskName, + currentAttempt: task.currentAttempt, + ), + ); + // Because fetched2 comes from cache, it should still have status inProgress and rev 1 + expect(fetched2.status, TaskStatus.inProgress); + expect(fetched2.revisionId, 1); + }, + ); + + test( + 'queryAllTasksForCommit caches commit task IDs and serves subsequent queries from versioned task cache', + () async { + final task1 = generateFirestoreTask( + 1, + name: 'Linux A', + attempts: 1, + status: TaskStatus.inProgress, + ); + final task2 = generateFirestoreTask( + 2, + name: 'Mac B', + attempts: 1, + status: TaskStatus.succeeded, + ); + firestore.putDocument(task1); + firestore.putDocument(task2); + + final initialTasks = await firestore.queryAllTasksForCommit( + commitSha: commitSha, + ); + expect(initialTasks.length, 2); + expect( + await cache.getSet('tasks_by_commit_ids', commitSha), + isNotEmpty, + ); + + // Verify cached task lookup without hitting firestore query + firestore.reset(); + final cachedTasks = await firestore.queryAllTasksForCommit( + commitSha: commitSha, + ); + expect(cachedTasks.length, 2); + }, + ); + + test( + 'queryAllTasksForCommit recovers missing individual task entries via partial query when commit set is populated', + () async { + final task1 = generateFirestoreTask( + 1, + name: 'Linux A', + attempts: 1, + status: TaskStatus.inProgress, + ); + final task2 = generateFirestoreTask( + 2, + name: 'Mac B', + attempts: 1, + status: TaskStatus.succeeded, + ); + firestore.putDocument(task1); + firestore.putDocument(task2); + + await firestore.queryAllTasksForCommit(commitSha: commitSha); + await Future.delayed(const Duration(milliseconds: 10)); + await cache.purge('tasks', '${commitSha}_Mac B_1'); + expect(await cache.get('tasks', '${commitSha}_Mac B_1'), isNull); + + final recoveredTasks = await firestore.queryAllTasksForCommit( + commitSha: commitSha, + ); + expect(recoveredTasks.length, 2); + expect(await cache.get('tasks', '${commitSha}_Mac B_1'), isNotNull); + }, + ); + + test( + 'batchWriteDocuments updates versioned task cache in-place lock-free', + () async { + final task1 = generateFirestoreTask( + 1, + name: 'Linux A', + attempts: 1, + status: TaskStatus.inProgress, + ); + firestore.putDocument(task1); + + // Populate commit cache and task cache via queryAllTasksForCommit + final initialTasks = await firestore.queryAllTasksForCommit( + commitSha: commitSha, + ); + expect(initialTasks.first.status, TaskStatus.inProgress); + expect( + await cache.getSet('tasks_by_commit_ids', commitSha), + isNotEmpty, + ); + + // Patch task status via batchWriteDocuments + final patchWrite = Task.patchStatus( + TaskId( + commitSha: commitSha, + taskName: task1.taskName, + currentAttempt: task1.currentAttempt, + ), + TaskStatus.succeeded, + ); + await firestore.batchWriteDocuments( + BatchWriteRequest(writes: [patchWrite]), + kDatabase, + ); + + // Verify that tasks_by_commit_ids remains intact (no purge needed!) + expect( + await cache.getSet('tasks_by_commit_ids', commitSha), + isNotEmpty, + ); + + // Verify that single-task fetch and commit query return the updated version right from cache + final updatedTask = await Task.fromFirestore( + firestore, + TaskId( + commitSha: commitSha, + taskName: task1.taskName, + currentAttempt: task1.currentAttempt, + ), + ); + expect(updatedTask.status, TaskStatus.succeeded); + expect(updatedTask.revisionId, 2); + }, + ); + + test( + 'writeViaTransaction updates versioned task cache in-place lock-free', + () async { + final task = generateFirestoreTask( + 1, + name: 'Linux B', + attempts: 1, + status: TaskStatus.inProgress, + ); + firestore.putDocument(task); + await firestore.queryAllTasksForCommit(commitSha: commitSha); + + task.setStatus(TaskStatus.succeeded); + await firestore.writeViaTransaction([ + Write( + update: Document(name: task.name, fields: task.fields), + ), + ]); + + final updatedTask = await Task.fromFirestore( + firestore, + TaskId( + commitSha: commitSha, + taskName: task.taskName, + currentAttempt: task.currentAttempt, + ), + ); + expect(updatedTask.status, TaskStatus.succeeded); + expect(updatedTask.revisionId, 2); + }, + ); + }); +} diff --git a/packages/cocoon_integration_test/lib/src/fakes/fake_cache_service.dart b/packages/cocoon_integration_test/lib/src/fakes/fake_cache_service.dart index 4796cdab3a..fd77960a64 100644 --- a/packages/cocoon_integration_test/lib/src/fakes/fake_cache_service.dart +++ b/packages/cocoon_integration_test/lib/src/fakes/fake_cache_service.dart @@ -14,6 +14,38 @@ class FakeCacheService extends CacheService { @override Future get(String subcacheName, String key) async => null; + @override + Future> getMulti( + String subcacheName, + List keys, + ) async { + return List.filled(keys.length, null); + } + + @override + Future insertVersioned( + String subcacheName, + List entries, + ) async {} + + @override + Future> getSet(String subcacheName, String key) async => const {}; + + @override + Future updateSet( + String subcacheName, + String key, + Set values, { + Duration ttl = const Duration(hours: 12), + }) async {} + + @override + Future addToSetIfExists( + String subcacheName, + String key, + String value, + ) async => false; + @override Future dispose() async {} diff --git a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart index bcd6c4a1a2..6f889b1b53 100644 --- a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart +++ b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart @@ -2,10 +2,13 @@ // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. +import 'dart:async'; import 'dart:convert'; import 'dart:io'; import 'package:cocoon_service/src/model/firestore/base.dart'; +import 'package:cocoon_service/src/model/firestore/task.dart'; +import 'package:cocoon_service/src/service/cache_service.dart'; import 'package:cocoon_service/src/service/config.dart'; import 'package:cocoon_service/src/service/firestore.dart'; import 'package:googleapis/firestore/v1.dart'; @@ -22,6 +25,11 @@ final queryKeyValidator = RegExp(r'(^[a-zA-Z_][a-zA-Z_0-9]*)$'); abstract base class _FakeInMemoryFirestoreService with FirestoreQueries implements FirestoreService { + _FakeInMemoryFirestoreService({this.cache}); + + @override + final CacheService? cache; + /// Every document currently stored in the fake. Iterable get documents => _documents.values; final _documents = {}; @@ -195,36 +203,31 @@ abstract base class _FakeInMemoryFirestoreService for (final transform in updateTransforms ?? const []) { final field = fields[transform.fieldPath]; - if (field == null) { - if (transform.appendMissingElements != null) { - // firestore: Append the given elements in order if they are not - // already present in the current field value. If the field is not an - // array, or if the field does not yet exist, it is first set to the - // empty array. - // https://firebase.google.com/docs/firestore/reference/rest/v1beta1/Write#FieldTransform + if (transform.appendMissingElements?.values case final elements?) { + if (field?.arrayValue?.values case final fieldArray?) { + for (final element in elements) { + if (fieldArray.contains(element)) continue; + fieldArray.add(element); + } + } else { fields[transform.fieldPath!] = Value( - arrayValue: transform.appendMissingElements!, + arrayValue: ArrayValue(values: elements), ); - continue; } - // this is most certainly wrong as the real firestore probably(?) updates - // the field. We're not using it that way in the tests, so if you find - // yourself getting this error - congrats and welcome to the team. + } else if (transform.increment case final inc?) { + if (field?.integerValue case final oldVal?) { + final newVal = int.parse(oldVal) + int.parse(inc.integerValue!); + fields[transform.fieldPath!] = Value(integerValue: newVal.toString()); + } else { + fields[transform.fieldPath!] = Value(integerValue: inc.integerValue!); + } + } else if (field == null) { throw ArgumentError.value( transform.fieldPath, 'field', 'not found, cannot patch', ); } - // The following are "union" members; only one operation per transform. - if (transform.appendMissingElements?.values case final elements?) { - if (field.arrayValue?.values case final fieldArray?) { - for (final element in elements) { - if (fieldArray.contains(element)) continue; - fieldArray.add(element); - } - } - } } _checkDatabaseName(name); @@ -237,6 +240,9 @@ abstract base class _FakeInMemoryFirestoreService _now().toUtc().toIso8601String(), updateTime: (updated ?? _now()).toUtc().toIso8601String(), ); + if (cache != null && collection == kTaskCollectionId) { + unawaited(invalidateCacheForWrites([Write(update: _documents[name]!)])); + } return _clone(_documents[name]!); } @@ -312,9 +318,13 @@ abstract base class _FakeInMemoryFirestoreService if (p.split(database).length != 4) { throw StateError('Unexpected database: $database'); } - return BatchWriteResponse( + final response = BatchWriteResponse( status: _batchWriteSync(request.writes ?? const []), ); + if (cache != null && request.writes != null) { + await invalidateCacheForWrites(request.writes!); + } + return response; } /// Same as [batchWriteDocuments], but does not yield the microtask loop. @@ -441,6 +451,9 @@ abstract base class _FakeInMemoryFirestoreService } final updated = _now().toUtc().toIso8601String(); + if (cache != null) { + await invalidateCacheForWrites(writes); + } return CommitResponse( commitTime: updated, writeResults: List.generate( @@ -667,7 +680,9 @@ abstract base class _FakeInMemoryFirestoreService } /// A fake implementation of [FirestoreService]. -final class FakeFirestoreService extends _FakeInMemoryFirestoreService {} +final class FakeFirestoreService extends _FakeInMemoryFirestoreService { + FakeFirestoreService({super.cache}); +} /// Checks that the models described by [metadata] match storage of [matcher]. /// From a5a92d1febfa5630f2cac53ab296fadbc83b8fde Mon Sep 17 00:00:00 2001 From: Jackson Gardner Date: Fri, 10 Jul 2026 15:52:09 -0700 Subject: [PATCH 2/7] Addressed Gemini comments. --- app_dart/lib/src/service/cache_service.dart | 18 ++++---- app_dart/lib/src/service/firestore.dart | 46 +++++++++++++++---- .../test/service/firestore_cache_test.dart | 3 ++ .../lib/src/fakes/fake_firestore_service.dart | 12 +++++ 4 files changed, 59 insertions(+), 20 deletions(-) diff --git a/app_dart/lib/src/service/cache_service.dart b/app_dart/lib/src/service/cache_service.dart index c3510ab823..efff34965d 100644 --- a/app_dart/lib/src/service/cache_service.dart +++ b/app_dart/lib/src/service/cache_service.dart @@ -257,17 +257,17 @@ class RedisCacheService extends CacheService { if (entries.isEmpty) return; const insertVersionedScript = ''' local numKeys = tonumber(ARGV[1]) - local subcache = ARGV[2] for i = 1, numKeys do local key = KEYS[i] - local val = ARGV[2 + (i - 1) * 3 + 1] - local rev = tonumber(ARGV[2 + (i - 1) * 3 + 2]) - local ttl = tonumber(ARGV[2 + (i - 1) * 3 + 3]) - - local existingRev = tonumber(redis.call("hget", "revisions:" .. subcache, key) or 0) + local val = ARGV[1 + (i - 1) * 3 + 1] + local rev = tonumber(ARGV[1 + (i - 1) * 3 + 2]) + local ttl = tonumber(ARGV[1 + (i - 1) * 3 + 3]) + + local revKey = "revisions/" .. key + local existingRev = tonumber(redis.call("get", revKey) or 0) if rev > existingRev or (not redis.call("exists", key)) then redis.call("set", key, val, "PX", ttl) - redis.call("hset", "revisions:" .. subcache, key, rev) + redis.call("set", revKey, rev, "PX", ttl) end end return 1 @@ -279,7 +279,6 @@ class RedisCacheService extends CacheService { final keys = chunk.map((e) => '$subcacheName/${e.key}').toList(); final args = [ chunk.length.toString(), - subcacheName, for (final e in chunk) ...[ base64.encode(e.value), e.revisionId.toString(), @@ -444,10 +443,9 @@ class RedisCacheService extends CacheService { Future purge(String subcacheName, String key) async { final redisKey = '$subcacheName/$key'; try { - await _runCommand((client) => client.send_object(['DEL', redisKey])); await _runCommand( (client) => - client.send_object(['HDEL', 'revisions:$subcacheName', redisKey]), + client.send_object(['DEL', redisKey, 'revisions/$redisKey']), ); } catch (e) { log.warn('Unable to purge value for $key from cache.', e); diff --git a/app_dart/lib/src/service/firestore.dart b/app_dart/lib/src/service/firestore.dart index 7c6177881b..1dbb9deeee 100644 --- a/app_dart/lib/src/service/firestore.dart +++ b/app_dart/lib/src/service/firestore.dart @@ -82,12 +82,13 @@ final kFieldMapRegExp = RegExp( /// payload lookup across `tasks` (`MGET`). /// - If any individual task entries have expired or are missing from `tasks`, successfully retrieved cached tasks /// ([foundTasks]) are preserved immediately. Missing document IDs (`missingDocIds`) are queried individually via -/// [getDocument], inserted into `tasks` via `insertVersioned`, and combined with [foundTasks]—avoiding a full table query. +/// [batchGetDocuments], inserted into `tasks` via `insertVersioned`, and combined with [foundTasks]—avoiding a full table query. /// - If `tasks_by_commit_ids` is missing entirely (`docIds.isEmpty`), we execute a full query (`_fetchAndCacheCommitTasks`), /// populate `tasks` for all items, and initialize the commit Set via [CacheService.updateSet]. mixin FirestoreQueries { CacheService? get cache => null; Future getDocument(String name, {Transaction? transaction}); + Future> batchGetDocuments(List names); static const String _subcacheTasksByCommitIds = 'tasks_by_commit_ids'; @@ -180,6 +181,11 @@ mixin FirestoreQueries { } final docId = p.basename(docName); + if (write.delete != null) { + await cache!.purge('tasks', docId); + continue; + } + if (write.update != null && write.updateMask == null && write.update!.fields != null && @@ -229,6 +235,13 @@ mixin FirestoreQueries { !docName.contains('/documents/$kTaskCollectionId/')) { continue; } + final docId = p.basename(docName); + + if (write.delete != null) { + await cache!.purge('tasks', docId); + continue; + } + if (write.update != null && write.updateMask == null && write.update!.fields != null && @@ -421,20 +434,18 @@ mixin FirestoreQueries { if (missingDocIds.isEmpty) { tasks = foundTasks; } else { - final fetchedMissingTasks = []; - for (final missingId in missingDocIds) { - try { - final document = await getDocument( - p.posix.join( + final missingNames = missingDocIds + .map( + (missingId) => p.posix.join( kDatabase, 'documents', kTaskCollectionId, missingId, ), - ); - fetchedMissingTasks.add(Task.fromDocument(document)); - } catch (_) {} - } + ) + .toList(); + final documents = await batchGetDocuments(missingNames); + final fetchedMissingTasks = documents.map(Task.fromDocument).toList(); if (fetchedMissingTasks.isNotEmpty) { await _cacheTaskDocuments(fetchedMissingTasks); foundTasks.addAll(fetchedMissingTasks); @@ -666,6 +677,21 @@ class FirestoreService with FirestoreQueries { ); } + /// Gets multiple documents based on a list of names in a single batch request. + @override + Future> batchGetDocuments(List names) async { + if (names.isEmpty) return const []; + final request = BatchGetDocumentsRequest(documents: names); + final response = await _api.projects.databases.documents.batchGet( + request, + kDatabase, + ); + return response + .map((element) => element.found) + .whereType() + .toList(); + } + /// Creates a document. /// /// A document name is automatically generated if [documentId] is omitted. diff --git a/app_dart/test/service/firestore_cache_test.dart b/app_dart/test/service/firestore_cache_test.dart index 903eebade3..35cdf693f5 100644 --- a/app_dart/test/service/firestore_cache_test.dart +++ b/app_dart/test/service/firestore_cache_test.dart @@ -4,6 +4,7 @@ import 'package:cocoon_common/task_status.dart'; import 'package:cocoon_integration_test/testing.dart'; +import 'package:cocoon_server_test/test_logging.dart'; import 'package:cocoon_service/src/model/firestore/task.dart'; import 'package:cocoon_service/src/service/cache_service.dart'; import 'package:cocoon_service/src/service/firestore.dart'; @@ -11,6 +12,8 @@ import 'package:googleapis/firestore/v1.dart'; import 'package:test/test.dart'; void main() { + useTestLoggerPerTest(); + group('Task Firestore Cache', () { late FakeFirestoreService firestore; late CacheService cache; diff --git a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart index 6f889b1b53..bcf5631ea7 100644 --- a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart +++ b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart @@ -387,6 +387,18 @@ abstract base class _FakeInMemoryFirestoreService return result; } + @override + Future> batchGetDocuments(List names) async { + final results = []; + for (final name in names) { + final doc = tryPeekDocumentByName(name); + if (doc != null) { + results.add(doc); + } + } + return results; + } + @override Future createDocument( Document document, { From 6cb63c6e13344da6677bda2973d8f6776f4c7bc5 Mon Sep 17 00:00:00 2001 From: Jackson Gardner Date: Wed, 22 Jul 2026 13:53:23 -0700 Subject: [PATCH 3/7] Refactor task writes to use transactional full documents and versioned Redis updates --- app_dart/lib/src/model/firestore/task.dart | 58 ++-- .../src/request_handlers/rerun_prod_task.dart | 19 +- .../scheduler/batch_backfiller.dart | 63 ++-- app_dart/lib/src/service/cache_service.dart | 34 ++- app_dart/lib/src/service/config.dart | 3 +- app_dart/lib/src/service/firestore.dart | 285 ++++++++---------- .../lib/src/service/flags/dynamic_config.dart | 18 ++ .../src/service/flags/dynamic_config.g.dart | 4 + .../test/service/firestore_cache_test.dart | 55 ++-- .../lib/src/fakes/fake_firestore_service.dart | 11 +- 10 files changed, 283 insertions(+), 267 deletions(-) diff --git a/app_dart/lib/src/model/firestore/task.dart b/app_dart/lib/src/model/firestore/task.dart index 0035d70676..8cc657bd0e 100644 --- a/app_dart/lib/src/model/firestore/task.dart +++ b/app_dart/lib/src/model/firestore/task.dart @@ -140,16 +140,16 @@ final class Task extends AppDocument { fromDocument: Task.fromDocument, ); - static Uint8List _serializeTask(Task task) { - return Uint8List.fromList( - utf8.encode( - json.encode(Document(name: task.name, fields: task.fields).toJson()), - ), + /// Serializes a [Task] document into bytes for caching. + static Uint8List serialize(Task task) { + return utf8.encode( + json.encode(Document(name: task.name, fields: task.fields).toJson()), ); } - static Task _deserializeTask(Uint8List data) { - final jsonMap = json.decode(utf8.decode(data)) as Map; + /// Deserializes bytes into a [Task] document. + static Task deserialize(Uint8List data) { + final jsonMap = json.decode(utf8.decode(data)) as Map; return Task.fromDocument(Document.fromJson(jsonMap)); } @@ -166,7 +166,7 @@ final class Task extends AppDocument { id.documentId, ); if (cachedTask != null) { - return _deserializeTask(cachedTask); + return deserialize(cachedTask); } } @@ -179,7 +179,7 @@ final class Task extends AppDocument { await firestoreService.cache!.insertVersioned('tasks', [ VersionedCacheEntry( key: id.documentId, - value: _serializeTask(task), + value: serialize(task), revisionId: task.revisionId, ttl: const Duration(hours: 12), ), @@ -254,30 +254,6 @@ final class Task extends AppDocument { ..name = name; } - /// Returns a Firestore [Write] that patches the [status] field for [id]. - @useResult - static Write patchStatus(AppDocumentId id, TaskStatus status) { - return Write( - currentDocument: Precondition(exists: true), - update: Document( - name: p.posix.join( - kDatabase, - 'documents', - kTaskCollectionId, - id.documentId, - ), - fields: {fieldStatus: Value(stringValue: status.value)}, - ), - updateMask: DocumentMask(fieldPaths: [fieldStatus]), - updateTransforms: [ - FieldTransform( - fieldPath: fieldRevisionId, - increment: Value(integerValue: '1'), - ), - ], - ); - } - /// The task was run successfully. static const statusSucceeded = TaskStatus.succeeded; @@ -395,6 +371,22 @@ final class Task extends AppDocument { incrementRevisionId(); } + Task createRetry({DateTime? now}) { + now ??= DateTime.now(); + return Task( + builderName: taskName, + currentAttempt: currentAttempt + 1, + commitSha: commitSha, + bringup: bringup, + createTimestamp: now.millisecondsSinceEpoch, + startTimestamp: 0, + endTimestamp: 0, + status: TaskStatus.waitingForBackfill, + testFlaky: false, + buildNumber: null, + ); + } + void resetAsRetry({int? attempt, DateTime? now}) { attempt ??= currentAttempt + 1; name = p.posix.join( diff --git a/app_dart/lib/src/request_handlers/rerun_prod_task.dart b/app_dart/lib/src/request_handlers/rerun_prod_task.dart index f458337e4e..9b4eed723f 100644 --- a/app_dart/lib/src/request_handlers/rerun_prod_task.dart +++ b/app_dart/lib/src/request_handlers/rerun_prod_task.dart @@ -210,23 +210,22 @@ final class RerunProdTask extends ApiRequestHandler { // If it appears the task was in progress, cancel any running builders // and crease a _new_ task (to represent a new run). if (task.status == TaskStatus.inProgress) { - // Mark cancelled. + // Mark current attempt cancelled with incremented revisionId + task.setStatus(TaskStatus.cancelled); documentWrites.add( - fs.Task.patchStatus( - fs.TaskId( - commitSha: task.commitSha, - currentAttempt: task.currentAttempt, - taskName: task.taskName, + g.Write( + update: g.Document( + name: task.name, + fields: Map.from(task.fields), ), - TaskStatus.cancelled, ), ); } - // Start a new task. - task.resetAsRetry(now: _now()); + // Start a new task attempt + final retryTask = task.createRetry(now: _now()); documentWrites.add( - g.Write(currentDocument: g.Precondition(exists: false), update: task), + g.Write(currentDocument: g.Precondition(exists: false), update: retryTask), ); } diff --git a/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart b/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart index da9394ab4b..32ae66bff7 100644 --- a/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart +++ b/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart @@ -9,6 +9,7 @@ import 'package:cocoon_common/task_status.dart'; import 'package:cocoon_server/logging.dart'; import 'package:github/github.dart'; import 'package:meta/meta.dart'; +import 'package:path/path.dart' as p; import '../../../cocoon_service.dart'; import '../../model/ci_yaml/ci_yaml.dart'; @@ -175,30 +176,44 @@ final class BatchBackfiller extends ApiRequestHandler { Iterable skip, ) async { log.debug('Querying ${schedule.length} tasks in Firestore...'); - await _firestore.writeViaTransaction([ - ...schedule.map((toUpdate) { - final BackfillTask(:task) = toUpdate; - return fs.Task.patchStatus( - fs.TaskId( - commitSha: task.commitSha, - taskName: task.name, - currentAttempt: task.currentAttempt, - ), - TaskStatus.inProgress, - ); - }), - ...skip.map((toSkip) { - final SkippableTask(:task) = toSkip; - return fs.Task.patchStatus( - fs.TaskId( - commitSha: task.commitSha, - taskName: task.name, - currentAttempt: task.currentAttempt, - ), - TaskStatus.skipped, - ); - }), - ]); + final taskRefs = [ + ...schedule.map((toUpdate) => (toUpdate.task, TaskStatus.inProgress)), + ...skip.map((toSkip) => (toSkip.task, TaskStatus.skipped)), + ]; + if (taskRefs.isEmpty) return; + + final docIds = taskRefs.map((item) { + final (ref, _) = item; + return p.posix.join( + kDatabase, + 'documents', + kTaskCollectionId, + fs.TaskId( + commitSha: ref.commitSha, + taskName: ref.name, + currentAttempt: ref.currentAttempt, + ).documentId, + ); + }).toList(); + + final transaction = await _firestore.beginTransaction(); + final docs = await _firestore.batchGetDocuments( + docIds, + transaction: transaction, + ); + final tasksToUpdate = []; + + for (var i = 0; i < docs.length; i++) { + final task = fs.Task.fromDocument(docs[i]); + final (_, targetStatus) = taskRefs[i]; + task.setStatus(targetStatus); + tasksToUpdate.add(task); + } + + await _firestore.commit( + transaction, + documentsToWrites(tasksToUpdate), + ); log.debug('Wrote to Firestore for backfill'); } diff --git a/app_dart/lib/src/service/cache_service.dart b/app_dart/lib/src/service/cache_service.dart index efff34965d..8d3445203e 100644 --- a/app_dart/lib/src/service/cache_service.dart +++ b/app_dart/lib/src/service/cache_service.dart @@ -265,7 +265,7 @@ class RedisCacheService extends CacheService { local revKey = "revisions/" .. key local existingRev = tonumber(redis.call("get", revKey) or 0) - if rev > existingRev or (not redis.call("exists", key)) then + if rev > existingRev or redis.call("exists", key) == 0 then redis.call("set", key, val, "PX", ttl) redis.call("set", revKey, rev, "PX", ttl) end @@ -513,6 +513,19 @@ class InMemoryCacheService extends CacheService { final Map _locks = {}; final _mutex = Mutex(); + void _evictIfFull(String cacheKey) { + if (_entries.length >= maxEntries && !_entries.containsKey(cacheKey)) { + _entries.remove(_entries.keys.first); + } + } + + void _touch(String cacheKey) { + final entry = _entries.remove(cacheKey); + if (entry != null) { + _entries[cacheKey] = entry; + } + } + @override Future get(String subcacheName, String key) async { await _mutex.acquire(); @@ -524,6 +537,7 @@ class InMemoryCacheService extends CacheService { _entries.remove(cacheKey); return null; } + _touch(cacheKey); return entry.value; } finally { _mutex.release(); @@ -544,6 +558,7 @@ class InMemoryCacheService extends CacheService { _entries.remove(cacheKey); return null; } + _touch(cacheKey); return entry.value; }).toList(); } finally { @@ -566,10 +581,8 @@ class InMemoryCacheService extends CacheService { if (existing == null || existing.isExpired || entry.revisionId > existing.revisionId) { - if (_entries.length >= maxEntries && - !_entries.containsKey(cacheKey)) { - _entries.remove(_entries.keys.first); - } + _evictIfFull(cacheKey); + _entries.remove(cacheKey); _entries[cacheKey] = _InMemoryCacheEntry( entry.value, DateTime.now().add(entry.ttl), @@ -660,11 +673,9 @@ class InMemoryCacheService extends CacheService { final cacheKey = '$subcacheName/$key'; _entries.removeWhere((k, v) => v.isExpired); + _evictIfFull(cacheKey); - if (_entries.length >= maxEntries && !_entries.containsKey(cacheKey)) { - _entries.remove(_entries.keys.first); - } - + _entries.remove(cacheKey); _entries[cacheKey] = _InMemoryCacheEntry(value, DateTime.now().add(ttl)); return value; } finally { @@ -688,10 +699,7 @@ class InMemoryCacheService extends CacheService { } _entries.removeWhere((k, v) => v.isExpired); - - if (_entries.length >= maxEntries && !_entries.containsKey(cacheKey)) { - _entries.remove(_entries.keys.first); - } + _evictIfFull(cacheKey); _entries[cacheKey] = _InMemoryCacheEntry(value, DateTime.now().add(ttl)); return true; diff --git a/app_dart/lib/src/service/config.dart b/app_dart/lib/src/service/config.dart index 26a1d137ca..cbb51d92bc 100644 --- a/app_dart/lib/src/service/config.dart +++ b/app_dart/lib/src/service/config.dart @@ -9,7 +9,8 @@ import 'package:cocoon_server/generate_github_jws.dart'; import 'package:cocoon_server/logging.dart'; import 'package:cocoon_server/secret_manager.dart'; import 'package:github/github.dart' as gh; -import 'package:graphql/client.dart' hide JsonSerializable; +import 'package:graphql/client.dart' + show AuthLink, GraphQLCache, GraphQLClient, HttpLink; import 'package:http/http.dart' as http; import 'package:meta/meta.dart'; import 'package:retry/retry.dart'; diff --git a/app_dart/lib/src/service/firestore.dart b/app_dart/lib/src/service/firestore.dart index 1dbb9deeee..ff3cf8583b 100644 --- a/app_dart/lib/src/service/firestore.dart +++ b/app_dart/lib/src/service/firestore.dart @@ -3,13 +3,12 @@ // found in the LICENSE file. import 'dart:async'; -import 'dart:convert'; -import 'dart:typed_data'; import 'package:cocoon_common/core_extensions.dart'; import 'package:cocoon_common/task_status.dart'; import 'package:cocoon_server/access_client_provider.dart'; import 'package:cocoon_server/google_auth_provider.dart'; +import 'package:cocoon_server/logging.dart'; import 'package:github/github.dart'; import 'package:googleapis/firestore/v1.dart'; import 'package:meta/meta.dart'; @@ -59,68 +58,70 @@ final kFieldMapRegExp = RegExp( /// To resolve read bottlenecks against the Firestore `/tasks` collection without incurring /// distributed locking overhead, this mixin implements a two-tier optimistic caching architecture: /// -/// ### 1. Individual Task Payload Caching (`tasks` subcache) -/// - **Single Source of Truth**: Full [Task] document payloads are serialized to JSON (`_serializeTask`) -/// and stored in the `tasks` subcache keyed by exact document ID (`$commitSha_$taskName_$attempt`). -/// - **Optimistic Concurrency (`revisionId`)**: Every task document contains a monotonically increasing integer -/// `revisionId`. Updates and insertions use [CacheService.insertVersioned], which runs an atomic check-and-set -/// ensuring payload updates are only applied if `newRevisionId > cachedRevisionId`. -/// - **Chunked Batching**: Multi-task insertions (`_cacheTaskDocuments`) are chunked in batches of 20 (`batchSize = 20`) -/// via atomic Redis Lua (`EVAL`) scripts, guaranteeing execution in `<0.1ms` without starving concurrent readers. +/// ```mermaid +/// sequenceDiagram +/// autonumber +/// participant Client +/// participant FirestoreQueries +/// participant Redis as Redis Cache +/// participant DB as Firestore DB /// -/// ### 2. Commit Task List Set Caching (`tasks_by_commit_ids` subcache) -/// - **Native Set Operations**: Instead of storing monolithic serialized JSON arrays of task IDs, commit task indices -/// are stored in native Redis Sets (`SMEMBERS`, `SADD`), backed by two fundamental domain invariants: -/// 1. **Immutable `commitSha`**: A task's commit association never changes after creation. Therefore, status or field -/// mutations (`patchStatus`) never alter the membership of a commit's task list (`updateCacheForTaskMutations` -/// updates individual `tasks/$docId` entries in-place lock-free and never touches or invalidates `tasks_by_commit_ids`). -/// 2. **Monotonically Increasing Set**: Task attempts for a commit only grow over time. Adding task IDs via `SADD` -/// ([CacheService.updateSet], [CacheService.addToSetIfExists]) safely converges to database state without locks. +/// Note over Client,DB: Scenario 1: Cache Hit Flow +/// Client->>FirestoreQueries: queryTasks(commitSha) +/// FirestoreQueries->>Redis: SMEMBERS tasks_by_commit_ids/$sha +/// Redis-->>FirestoreQueries: [docId1, docId2] +/// FirestoreQueries->>Redis: MGET tasks/docId1, tasks/docId2 +/// Redis-->>FirestoreQueries: [payload1, payload2] +/// FirestoreQueries-->>Client: Return tasks /// -/// ### 3. Partial Cache Recovery (`_queryTasksByCommitCached`) -/// - When querying all tasks for a commit, we retrieve the cached Set of document IDs (`SMEMBERS`) and perform a batch -/// payload lookup across `tasks` (`MGET`). -/// - If any individual task entries have expired or are missing from `tasks`, successfully retrieved cached tasks -/// ([foundTasks]) are preserved immediately. Missing document IDs (`missingDocIds`) are queried individually via -/// [batchGetDocuments], inserted into `tasks` via `insertVersioned`, and combined with [foundTasks]—avoiding a full table query. -/// - If `tasks_by_commit_ids` is missing entirely (`docIds.isEmpty`), we execute a full query (`_fetchAndCacheCommitTasks`), -/// populate `tasks` for all items, and initialize the commit Set via [CacheService.updateSet]. +/// Note over Client,DB: Scenario 2: Partial Cache Recovery +/// Client->>FirestoreQueries: queryTasks(commitSha) +/// FirestoreQueries->>Redis: SMEMBERS tasks_by_commit_ids/$sha +/// Redis-->>FirestoreQueries: [docId1, docId2] +/// FirestoreQueries->>Redis: MGET tasks/docId1, tasks/docId2 +/// Redis-->>FirestoreQueries: [payload1, null (missing)] +/// FirestoreQueries->>DB: batchGetDocuments([docId2]) +/// DB-->>FirestoreQueries: [docId2 document (rev 5)] +/// FirestoreQueries->>Redis: insertVersioned(tasks/docId2, rev=5) +/// FirestoreQueries-->>Client: Return merged tasks +/// +/// Note over Client,DB: Scenario 3: Transactional Write-Back +/// Client->>FirestoreQueries: transaction write (mutate task) +/// FirestoreQueries->>DB: commit transaction (increment rev to 6) +/// DB-->>FirestoreQueries: Transaction commit result (rev 6) +/// FirestoreQueries->>Redis: insertVersioned(tasks/docId, rev=6) +/// Note over Redis: Stale in-flight reads with rev < 6 rejected +/// ``` mixin FirestoreQueries { CacheService? get cache => null; + Config? get config => null; Future getDocument(String name, {Transaction? transaction}); - Future> batchGetDocuments(List names); + Future> batchGetDocuments( + List names, { + Transaction? transaction, + }); static const String _subcacheTasksByCommitIds = 'tasks_by_commit_ids'; - static Uint8List _serializeTask(Task task) { - return Uint8List.fromList( - utf8.encode( - json.encode(Document(name: task.name, fields: task.fields).toJson()), - ), - ); - } - - static Task _deserializeTask(Uint8List data) { - final jsonMap = json.decode(utf8.decode(data)) as Map; - return Task.fromDocument(Document.fromJson(jsonMap)); - } + Duration get _defaultCacheTtl => + Duration(hours: config?.flags.taskCacheTtlInHours ?? 12); /// Caches the individual [Task] payloads into the `tasks` subcache. Future _cacheTaskDocuments( List tasks, { - Duration ttl = const Duration(hours: 12), + Duration? ttl, }) async { - if (cache == null || tasks.isEmpty) return; - final entries = tasks - .map( - (t) => VersionedCacheEntry( - key: p.basename(t.name!), - value: _serializeTask(t), - revisionId: t.revisionId, - ttl: ttl, - ), - ) - .toList(); + if (cache == null || !(config?.flags.taskCachingEnabled ?? true) || tasks.isEmpty) return; + final effectiveTtl = ttl ?? _defaultCacheTtl; + final entries = [ + for (final t in tasks) + VersionedCacheEntry( + key: p.basename(t.name!), + value: Task.serialize(t), + revisionId: t.revisionId, + ttl: effectiveTtl, + ), + ]; await cache!.insertVersioned('tasks', entries); } @@ -133,100 +134,57 @@ mixin FirestoreQueries { orderMap: {Task.fieldCreateTimestamp: kQueryOrderDescending}, ); final tasks = documents.map(Task.fromDocument).toList(); - await _cacheTaskDocuments(tasks); - final docIds = tasks.map((t) => p.basename(t.name!)).toSet(); - if (docIds.isNotEmpty && cache != null) { - await cache!.updateSet( - _subcacheTasksByCommitIds, - commitSha, - docIds, - ttl: const Duration(hours: 12), - ); + if (cache != null && (config?.flags.taskCachingEnabled ?? true)) { + final docIds = {for (final t in tasks) p.basename(t.name!)}; + await [ + _cacheTaskDocuments(tasks), + if (docIds.isNotEmpty) + cache!.updateSet( + _subcacheTasksByCommitIds, + commitSha, + docIds, + ttl: _defaultCacheTtl, + ), + ].wait; } return tasks; } /// Explicitly updates the cache when new [Task]s are created. Future updateCacheForCreatedTasks(List tasks) async { - if (cache == null || tasks.isEmpty) return; + if (cache == null || !(config?.flags.taskCachingEnabled ?? true) || tasks.isEmpty) return; await _cacheTaskDocuments(tasks); + + final tasksByCommit = >{}; for (final task in tasks) { if (task.commitSha.isNotEmpty) { final docId = p.basename(task.name!); - final added = await cache!.addToSetIfExists( - _subcacheTasksByCommitIds, - task.commitSha, - docId, - ); - if (!added) { - await _fetchAndCacheCommitTasks(task.commitSha); - } + tasksByCommit.putIfAbsent(task.commitSha, () => {}).add(docId); } } - } - - /// Explicitly updates the cache when existing [Task]s are mutated (e.g. status updates). - /// - /// Because commit membership is immutable, mutations never alter `tasks_by_commit_ids`. - Future updateCacheForTaskMutations(List writes) async { - if (cache == null || writes.isEmpty) return; - final mutatedTasks = []; - for (final write in writes) { - final docName = - write.update?.name ?? write.delete ?? write.transform?.document; - if (docName == null || - !docName.contains('/documents/$kTaskCollectionId/')) { - continue; - } - final docId = p.basename(docName); - - if (write.delete != null) { - await cache!.purge('tasks', docId); - continue; - } - - if (write.update != null && - write.updateMask == null && - write.update!.fields != null && - write.update!.fields!.containsKey(Task.fieldCreateTimestamp)) { - try { - mutatedTasks.add(Task.fromDocument(write.update!)); - } catch (_) {} - } else { - final cachedBytes = await cache!.get('tasks', docId); - if (cachedBytes != null) { - try { - final cachedTask = _deserializeTask(cachedBytes); - if (write.update?.fields != null) { - cachedTask.fields.addAll(write.update!.fields!); - } - if (write.updateTransforms != null) { - for (final transform in write.updateTransforms!) { - if (transform.fieldPath == Task.fieldRevisionId && - transform.increment != null) { - cachedTask.incrementRevisionId(); - } - } - } else { - cachedTask.incrementRevisionId(); - } - mutatedTasks.add(cachedTask); - } catch (_) {} + await Future.wait( + tasksByCommit.entries.map((entry) async { + final commitSha = entry.key; + final docIds = entry.value; + for (final docId in docIds) { + final added = await cache!.addToSetIfExists( + _subcacheTasksByCommitIds, + commitSha, + docId, + ); + if (!added) { + await _fetchAndCacheCommitTasks(commitSha); + break; + } } - } - } - - if (mutatedTasks.isNotEmpty) { - await _cacheTaskDocuments(mutatedTasks); - } + }), + ); } - @visibleForTesting - Future invalidateCacheForWrites(List writes) async { - if (cache == null || writes.isEmpty) return; - final createdTasks = []; - final mutationWrites = []; + Future handleWriteCacheInvalidation(List writes) async { + if (cache == null || !(config?.flags.taskCachingEnabled ?? true) || writes.isEmpty) return; + final updatedTasks = []; for (final write in writes) { final docName = @@ -242,23 +200,17 @@ mixin FirestoreQueries { continue; } - if (write.update != null && - write.updateMask == null && - write.update!.fields != null && - write.update!.fields!.containsKey(Task.fieldCreateTimestamp)) { + if (write.update != null && write.update!.fields != null) { try { - createdTasks.add(Task.fromDocument(write.update!)); - } catch (_) {} - } else { - mutationWrites.add(write); + updatedTasks.add(Task.fromDocument(write.update!)); + } catch (e) { + log.warn('Unable to parse task document for $docId', e); + } } } - if (createdTasks.isNotEmpty) { - await updateCacheForCreatedTasks(createdTasks); - } - if (mutationWrites.isNotEmpty) { - await updateCacheForTaskMutations(mutationWrites); + if (updatedTasks.isNotEmpty) { + await updateCacheForCreatedTasks(updatedTasks); } } @@ -373,8 +325,8 @@ mixin FirestoreQueries { transaction: transaction, ); final tasks = documents.map(Task.fromDocument).toList(); - if (cacheResults && transaction == null && cache != null) { - await _cacheTaskDocuments(tasks, ttl: const Duration(hours: 4)); + if (cacheResults && transaction == null && cache != null && (config?.flags.taskCachingEnabled ?? true)) { + await _cacheTaskDocuments(tasks); } return tasks; } @@ -386,7 +338,7 @@ mixin FirestoreQueries { int? limit, Transaction? transaction, }) async { - if (transaction == null && cache != null) { + if (transaction == null && cache != null && (config?.flags.taskCachingEnabled ?? true)) { return await _queryTasksByCommitCached( commitSha: commitSha, name: name, @@ -422,8 +374,9 @@ mixin FirestoreQueries { final data = cachedTasksData[i]; if (data != null) { try { - foundTasks.add(_deserializeTask(data)); - } catch (_) { + foundTasks.add(Task.deserialize(data)); + } catch (e) { + log.warn('Unable to deserialize cached task for ${docIdList[i]}', e); missingDocIds.add(docIdList[i]); } } else { @@ -651,6 +604,7 @@ class FirestoreService with FirestoreQueries { static Future from( GoogleAuthProvider authProvider, { CacheService? cache, + Config? config, }) async { final client = await authProvider.createClient( scopes: const [FirestoreApi.datastoreScope], @@ -659,15 +613,22 @@ class FirestoreService with FirestoreQueries { databaseId: Config.flutterGcpFirestoreDatabase, ), ); - return FirestoreService._(FirestoreApi(client), cache: cache); + return FirestoreService._( + FirestoreApi(client), + cache: cache, + config: config, + ); } - const FirestoreService._(this._api, {this.cache}); + const FirestoreService._(this._api, {this.cache, this.config}); final FirestoreApi _api; @override final CacheService? cache; + @override + final Config? config; + /// Gets a document based on name. @override Future getDocument(String name, {Transaction? transaction}) async { @@ -679,9 +640,15 @@ class FirestoreService with FirestoreQueries { /// Gets multiple documents based on a list of names in a single batch request. @override - Future> batchGetDocuments(List names) async { + Future> batchGetDocuments( + List names, { + Transaction? transaction, + }) async { if (names.isEmpty) return const []; - final request = BatchGetDocumentsRequest(documents: names); + final request = BatchGetDocumentsRequest( + documents: names, + transaction: transaction?.identifier, + ); final response = await _api.projects.databases.documents.batchGet( request, kDatabase, @@ -708,8 +675,8 @@ class FirestoreService with FirestoreQueries { collectionId, documentId: documentId, ); - if (cache != null && collectionId == kTaskCollectionId) { - await invalidateCacheForWrites([Write(update: result)]); + if (cache != null && (config?.flags.taskCachingEnabled ?? true) && collectionId == kTaskCollectionId) { + await updateCacheForCreatedTasks([Task.fromDocument(result)]); } return result; } @@ -728,8 +695,8 @@ class FirestoreService with FirestoreQueries { request, database, ); - if (cache != null && request.writes != null) { - await invalidateCacheForWrites(request.writes!); + if (cache != null && (config?.flags.taskCachingEnabled ?? true) && request.writes != null) { + await handleWriteCacheInvalidation(request.writes!); } return result; } @@ -759,8 +726,8 @@ class FirestoreService with FirestoreQueries { request, kDatabase, ); - if (cache != null) { - await invalidateCacheForWrites(writes); + if (cache != null && (config?.flags.taskCachingEnabled ?? true)) { + await handleWriteCacheInvalidation(writes); } return result; } @@ -788,8 +755,8 @@ class FirestoreService with FirestoreQueries { commitRequest, kDatabase, ); - if (cache != null) { - await invalidateCacheForWrites(writes); + if (cache != null && (config?.flags.taskCachingEnabled ?? true)) { + await handleWriteCacheInvalidation(writes); } return result; } diff --git a/app_dart/lib/src/service/flags/dynamic_config.dart b/app_dart/lib/src/service/flags/dynamic_config.dart index 0e735074a0..54e90f12af 100644 --- a/app_dart/lib/src/service/flags/dynamic_config.dart +++ b/app_dart/lib/src/service/flags/dynamic_config.dart @@ -43,6 +43,8 @@ final class DynamicConfig { orderedPresubmit: OrderedPresubmit.defaultInstance, dynamicTestSuppression: false, geminiModel: 'gemini-3-flash-preview', + taskCachingEnabled: true, + taskCacheTtlInHours: 12, ); /// Upper limit of commit rows to be backfilled in API call. @@ -85,6 +87,14 @@ final class DynamicConfig { @JsonKey() final String geminiModel; + /// Whether Redis caching for /tasks is enabled. + @JsonKey() + final bool taskCachingEnabled; + + /// Cache TTL duration in hours for task cache entries. + @JsonKey() + final int taskCacheTtlInHours; + const DynamicConfig._({ required this.backfillerCommitLimit, required this.ciYaml, @@ -95,6 +105,8 @@ final class DynamicConfig { required this.orderedPresubmit, required this.dynamicTestSuppression, required this.geminiModel, + required this.taskCachingEnabled, + required this.taskCacheTtlInHours, }); /// Creates [DynamicConfig] flags from a [json] object. @@ -110,6 +122,8 @@ final class DynamicConfig { OrderedPresubmit? orderedPresubmit, bool? dynamicTestSuppression, String? geminiModel, + bool? taskCachingEnabled, + int? taskCacheTtlInHours, }) { return DynamicConfig._( backfillerCommitLimit: @@ -128,6 +142,10 @@ final class DynamicConfig { dynamicTestSuppression: dynamicTestSuppression ?? defaultInstance.dynamicTestSuppression, geminiModel: geminiModel ?? defaultInstance.geminiModel, + taskCachingEnabled: + taskCachingEnabled ?? defaultInstance.taskCachingEnabled, + taskCacheTtlInHours: + taskCacheTtlInHours ?? defaultInstance.taskCacheTtlInHours, ); } diff --git a/app_dart/lib/src/service/flags/dynamic_config.g.dart b/app_dart/lib/src/service/flags/dynamic_config.g.dart index d958c97c10..51ca321757 100644 --- a/app_dart/lib/src/service/flags/dynamic_config.g.dart +++ b/app_dart/lib/src/service/flags/dynamic_config.g.dart @@ -33,6 +33,8 @@ DynamicConfig _$DynamicConfigFromJson(Map json) => ), dynamicTestSuppression: json['dynamicTestSuppression'] as bool?, geminiModel: json['geminiModel'] as String?, + taskCachingEnabled: json['taskCachingEnabled'] as bool?, + taskCacheTtlInHours: (json['taskCacheTtlInHours'] as num?)?.toInt(), ); Map _$DynamicConfigToJson(DynamicConfig instance) => @@ -46,4 +48,6 @@ Map _$DynamicConfigToJson(DynamicConfig instance) => 'orderedPresubmit': instance.orderedPresubmit.toJson(), 'dynamicTestSuppression': instance.dynamicTestSuppression, 'geminiModel': instance.geminiModel, + 'taskCachingEnabled': instance.taskCachingEnabled, + 'taskCacheTtlInHours': instance.taskCacheTtlInHours, }; diff --git a/app_dart/test/service/firestore_cache_test.dart b/app_dart/test/service/firestore_cache_test.dart index 35cdf693f5..25b62c19d2 100644 --- a/app_dart/test/service/firestore_cache_test.dart +++ b/app_dart/test/service/firestore_cache_test.dart @@ -9,6 +9,7 @@ import 'package:cocoon_service/src/model/firestore/task.dart'; import 'package:cocoon_service/src/service/cache_service.dart'; import 'package:cocoon_service/src/service/firestore.dart'; import 'package:googleapis/firestore/v1.dart'; +import 'package:path/path.dart' as p; import 'package:test/test.dart'; void main() { @@ -136,7 +137,7 @@ void main() { ); test( - 'batchWriteDocuments updates versioned task cache in-place lock-free', + 'batchWriteDocuments updates versioned task cache with full document and incremented revisionId', () async { final task1 = generateFirestoreTask( 1, @@ -151,42 +152,31 @@ void main() { commitSha: commitSha, ); expect(initialTasks.first.status, TaskStatus.inProgress); + expect(initialTasks.first.revisionId, 1); expect( await cache.getSet('tasks_by_commit_ids', commitSha), isNotEmpty, ); - // Patch task status via batchWriteDocuments - final patchWrite = Task.patchStatus( - TaskId( - commitSha: commitSha, - taskName: task1.taskName, - currentAttempt: task1.currentAttempt, - ), - TaskStatus.succeeded, - ); + // Mutate task status and increment revisionId + task1.setStatus(TaskStatus.succeeded); await firestore.batchWriteDocuments( - BatchWriteRequest(writes: [patchWrite]), + BatchWriteRequest(writes: documentsToWrites([task1])), kDatabase, ); - // Verify that tasks_by_commit_ids remains intact (no purge needed!) + // Verify that tasks_by_commit_ids remains intact expect( await cache.getSet('tasks_by_commit_ids', commitSha), isNotEmpty, ); - // Verify that single-task fetch and commit query return the updated version right from cache - final updatedTask = await Task.fromFirestore( - firestore, - TaskId( - commitSha: commitSha, - taskName: task1.taskName, - currentAttempt: task1.currentAttempt, - ), - ); - expect(updatedTask.status, TaskStatus.succeeded); - expect(updatedTask.revisionId, 2); + // Verify that single-task payload cache entry was updated in Redis with full document and rev 2 + final cachedBytes = await cache.get('tasks', p.basename(task1.name!)); + expect(cachedBytes, isNotNull); + final cachedTask = Task.deserialize(cachedBytes!); + expect(cachedTask.status, TaskStatus.succeeded); + expect(cachedTask.revisionId, 2); }, ); @@ -221,5 +211,24 @@ void main() { expect(updatedTask.revisionId, 2); }, ); + + test( + 'updateCacheForCreatedTasks groups tasks by commit SHA', + () async { + final task1 = generateFirestoreTask(1, commitSha: 'shaA', name: 'Linux A'); + final task2 = generateFirestoreTask(2, commitSha: 'shaA', name: 'Linux B'); + final task3 = generateFirestoreTask(3, commitSha: 'shaB', name: 'Linux C'); + + await firestore.updateCacheForCreatedTasks([task1, task2, task3]); + + final cachedTask1 = await cache.get('tasks', p.basename(task1.name!)); + final cachedTask2 = await cache.get('tasks', p.basename(task2.name!)); + final cachedTask3 = await cache.get('tasks', p.basename(task3.name!)); + + expect(cachedTask1, isNotNull); + expect(cachedTask2, isNotNull); + expect(cachedTask3, isNotNull); + }, + ); }); } diff --git a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart index bcf5631ea7..4903cb86da 100644 --- a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart +++ b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart @@ -241,7 +241,7 @@ abstract base class _FakeInMemoryFirestoreService updateTime: (updated ?? _now()).toUtc().toIso8601String(), ); if (cache != null && collection == kTaskCollectionId) { - unawaited(invalidateCacheForWrites([Write(update: _documents[name]!)])); + unawaited(updateCacheForCreatedTasks([Task.fromDocument(_documents[name]!)])); } return _clone(_documents[name]!); } @@ -322,7 +322,7 @@ abstract base class _FakeInMemoryFirestoreService status: _batchWriteSync(request.writes ?? const []), ); if (cache != null && request.writes != null) { - await invalidateCacheForWrites(request.writes!); + await handleWriteCacheInvalidation(request.writes!); } return response; } @@ -388,7 +388,10 @@ abstract base class _FakeInMemoryFirestoreService } @override - Future> batchGetDocuments(List names) async { + Future> batchGetDocuments( + List names, { + Transaction? transaction, + }) async { final results = []; for (final name in names) { final doc = tryPeekDocumentByName(name); @@ -464,7 +467,7 @@ abstract base class _FakeInMemoryFirestoreService final updated = _now().toUtc().toIso8601String(); if (cache != null) { - await invalidateCacheForWrites(writes); + await handleWriteCacheInvalidation(writes); } return CommitResponse( commitTime: updated, From 579a7d19660189bd46c0649168204ca0e041e9e4 Mon Sep 17 00:00:00 2001 From: Jackson Gardner Date: Thu, 23 Jul 2026 13:44:48 -0700 Subject: [PATCH 4/7] Addressed codefu's comments. --- app_dart/lib/src/model/firestore/task.dart | 27 +- .../dart_internal_subscription.dart | 7 +- .../postsubmit_luci_subscription.dart | 6 +- .../src/request_handlers/rerun_prod_task.dart | 28 +- .../scheduler/batch_backfiller.dart | 53 +--- .../scheduler/vacuum_stale_tasks.dart | 7 +- app_dart/lib/src/service/firestore.dart | 275 ++++++++---------- .../service/firestore/task_cache_service.dart | 212 ++++++++++++++ .../lib/src/service/luci_build_service.dart | 7 +- app_dart/lib/src/service/scheduler.dart | 1 + .../test/service/firestore_cache_test.dart | 15 +- .../test/service/task_cache_service_test.dart | 125 ++++++++ .../lib/src/fakes/fake_firestore_service.dart | 9 +- 13 files changed, 500 insertions(+), 272 deletions(-) create mode 100644 app_dart/lib/src/service/firestore/task_cache_service.dart create mode 100644 app_dart/test/service/task_cache_service_test.dart diff --git a/app_dart/lib/src/model/firestore/task.dart b/app_dart/lib/src/model/firestore/task.dart index 8cc657bd0e..07873e2c4f 100644 --- a/app_dart/lib/src/model/firestore/task.dart +++ b/app_dart/lib/src/model/firestore/task.dart @@ -160,32 +160,7 @@ final class Task extends AppDocument { FirestoreService firestoreService, AppDocumentId id, ) async { - if (firestoreService.cache != null) { - final cachedTask = await firestoreService.cache!.get( - 'tasks', - id.documentId, - ); - if (cachedTask != null) { - return deserialize(cachedTask); - } - } - - final document = await firestoreService.getDocument( - p.posix.join(kDatabase, 'documents', kTaskCollectionId, id.documentId), - ); - final task = Task.fromDocument(document); - - if (firestoreService.cache != null) { - await firestoreService.cache!.insertVersioned('tasks', [ - VersionedCacheEntry( - key: id.documentId, - value: serialize(task), - revisionId: task.revisionId, - ttl: const Duration(hours: 12), - ), - ]); - } - return task; + return firestoreService.getTask(id); } factory Task({ diff --git a/app_dart/lib/src/request_handlers/dart_internal_subscription.dart b/app_dart/lib/src/request_handlers/dart_internal_subscription.dart index d861a080a8..46fc5b0d77 100644 --- a/app_dart/lib/src/request_handlers/dart_internal_subscription.dart +++ b/app_dart/lib/src/request_handlers/dart_internal_subscription.dart @@ -7,7 +7,7 @@ import 'dart:convert'; import 'package:buildbucket/buildbucket_pb.dart' as bbv2; import 'package:cocoon_common/is_dart_internal.dart'; import 'package:cocoon_server/logging.dart'; -import 'package:googleapis/firestore/v1.dart'; + import '../../cocoon_service.dart'; import '../model/bbv2_extension.dart'; @@ -89,10 +89,7 @@ final class DartInternalSubscription extends SubscriptionHandler { } log.info('Inserting Task into Firestore: ${fsTask.toString()}'); - await _firestore.batchWriteDocuments( - BatchWriteRequest(writes: documentsToWrites([fsTask])), - kDatabase, - ); + await _firestore.updateTasks([fsTask]); return Response.json(fsTask.toString()); } diff --git a/app_dart/lib/src/request_handlers/postsubmit_luci_subscription.dart b/app_dart/lib/src/request_handlers/postsubmit_luci_subscription.dart index 1d305c0fff..fb6b1ad1e5 100644 --- a/app_dart/lib/src/request_handlers/postsubmit_luci_subscription.dart +++ b/app_dart/lib/src/request_handlers/postsubmit_luci_subscription.dart @@ -8,7 +8,6 @@ import 'package:archive/archive.dart'; import 'package:buildbucket/buildbucket_pb.dart' as bbv2; import 'package:cocoon_common/is_release_branch.dart'; import 'package:cocoon_server/logging.dart'; -import 'package:googleapis/firestore/v1.dart' hide Status; import '../../ci_yaml.dart'; import '../model/firestore/commit.dart' as fs; @@ -150,10 +149,7 @@ final class PostsubmitLuciSubscription extends SubscriptionHandler { Future _updateFirestore(fs.Task fsTask, bbv2.Build build) async { fsTask.updateFromBuild(build); - await _firestore.batchWriteDocuments( - BatchWriteRequest(writes: documentsToWrites([fsTask], exists: true)), - kDatabase, - ); + await _firestore.updateTasks([fsTask]); } // No need to update task in Firestore if diff --git a/app_dart/lib/src/request_handlers/rerun_prod_task.dart b/app_dart/lib/src/request_handlers/rerun_prod_task.dart index 9b4eed723f..a303728141 100644 --- a/app_dart/lib/src/request_handlers/rerun_prod_task.dart +++ b/app_dart/lib/src/request_handlers/rerun_prod_task.dart @@ -5,7 +5,6 @@ import 'package:cocoon_common/is_dart_internal.dart'; import 'package:cocoon_common/task_status.dart'; import 'package:cocoon_server/logging.dart'; -import 'package:googleapis/firestore/v1.dart' as g; import 'package:meta/meta.dart'; import '../model/ci_yaml/ci_yaml.dart'; @@ -184,9 +183,9 @@ final class RerunProdTask extends ApiRequestHandler { ).withMostRecentTaskOnly().tasks; final wasMarkedNew = []; - final documentWrites = []; + final tasksToUpdate = []; + final createdTasks = []; - // Wait for cancellations? final Future cancelRunningTasks; if (statusesToRerun.contains(TaskStatus.inProgress)) { cancelRunningTasks = _luciBuildService.cancelBuildsBySha( @@ -208,31 +207,28 @@ final class RerunProdTask extends ApiRequestHandler { } // If it appears the task was in progress, cancel any running builders - // and crease a _new_ task (to represent a new run). + // and create a _new_ task (to represent a new run). if (task.status == TaskStatus.inProgress) { // Mark current attempt cancelled with incremented revisionId task.setStatus(TaskStatus.cancelled); - documentWrites.add( - g.Write( - update: g.Document( - name: task.name, - fields: Map.from(task.fields), - ), - ), - ); + tasksToUpdate.add(task); } // Start a new task attempt final retryTask = task.createRetry(now: _now()); - documentWrites.add( - g.Write(currentDocument: g.Precondition(exists: false), update: retryTask), - ); + tasksToUpdate.add(retryTask); + createdTasks.add(retryTask); } + final writes = documentsToWrites(tasksToUpdate); await Future.wait([ cancelRunningTasks, - _firestore.commit(transaction, documentWrites), + _firestore.commit(transaction, writes), ]); + await _firestore.cacheTaskPayloads(tasksToUpdate); + if (createdTasks.isNotEmpty) { + await _firestore.updateCacheForCreatedTasks(createdTasks); + } return wasMarkedNew; } diff --git a/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart b/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart index 32ae66bff7..2fa4e5b44a 100644 --- a/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart +++ b/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart @@ -9,7 +9,6 @@ import 'package:cocoon_common/task_status.dart'; import 'package:cocoon_server/logging.dart'; import 'package:github/github.dart'; import 'package:meta/meta.dart'; -import 'package:path/path.dart' as p; import '../../../cocoon_service.dart'; import '../../model/ci_yaml/ci_yaml.dart'; @@ -175,45 +174,23 @@ final class BatchBackfiller extends ApiRequestHandler { Iterable schedule, Iterable skip, ) async { - log.debug('Querying ${schedule.length} tasks in Firestore...'); - final taskRefs = [ - ...schedule.map((toUpdate) => (toUpdate.task, TaskStatus.inProgress)), - ...skip.map((toSkip) => (toSkip.task, TaskStatus.skipped)), - ]; - if (taskRefs.isEmpty) return; - - final docIds = taskRefs.map((item) { - final (ref, _) = item; - return p.posix.join( - kDatabase, - 'documents', - kTaskCollectionId, + final taskStatusMap = { + for (final item in schedule) fs.TaskId( - commitSha: ref.commitSha, - taskName: ref.name, - currentAttempt: ref.currentAttempt, - ).documentId, - ); - }).toList(); - - final transaction = await _firestore.beginTransaction(); - final docs = await _firestore.batchGetDocuments( - docIds, - transaction: transaction, - ); - final tasksToUpdate = []; - - for (var i = 0; i < docs.length; i++) { - final task = fs.Task.fromDocument(docs[i]); - final (_, targetStatus) = taskRefs[i]; - task.setStatus(targetStatus); - tasksToUpdate.add(task); - } + commitSha: item.task.commitSha, + taskName: item.task.name, + currentAttempt: item.task.currentAttempt, + ): TaskStatus.inProgress, + for (final item in skip) + fs.TaskId( + commitSha: item.task.commitSha, + taskName: item.task.name, + currentAttempt: item.task.currentAttempt, + ): TaskStatus.skipped, + }; + if (taskStatusMap.isEmpty) return; - await _firestore.commit( - transaction, - documentsToWrites(tasksToUpdate), - ); + await _firestore.updateTaskStatuses(taskStatusMap); log.debug('Wrote to Firestore for backfill'); } diff --git a/app_dart/lib/src/request_handlers/scheduler/vacuum_stale_tasks.dart b/app_dart/lib/src/request_handlers/scheduler/vacuum_stale_tasks.dart index 10d767ca23..5d804585cd 100644 --- a/app_dart/lib/src/request_handlers/scheduler/vacuum_stale_tasks.dart +++ b/app_dart/lib/src/request_handlers/scheduler/vacuum_stale_tasks.dart @@ -9,7 +9,7 @@ import 'package:cocoon_common/is_release_branch.dart'; import 'package:cocoon_common/task_status.dart'; import 'package:cocoon_server/logging.dart'; import 'package:github/github.dart' as gh; -import 'package:googleapis/firestore/v1.dart'; + import 'package:meta/meta.dart'; import '../../../cocoon_service.dart'; @@ -139,10 +139,7 @@ final class VacuumStaleTasks extends ApiRequestHandler { } tasks.add(task); } - await _firestore.batchWriteDocuments( - BatchWriteRequest(writes: documentsToWrites(tasks)), - kDatabase, - ); + await _firestore.updateTasks(tasks); } } diff --git a/app_dart/lib/src/service/firestore.dart b/app_dart/lib/src/service/firestore.dart index ff3cf8583b..bb80ff6995 100644 --- a/app_dart/lib/src/service/firestore.dart +++ b/app_dart/lib/src/service/firestore.dart @@ -8,7 +8,6 @@ import 'package:cocoon_common/core_extensions.dart'; import 'package:cocoon_common/task_status.dart'; import 'package:cocoon_server/access_client_provider.dart'; import 'package:cocoon_server/google_auth_provider.dart'; -import 'package:cocoon_server/logging.dart'; import 'package:github/github.dart'; import 'package:googleapis/firestore/v1.dart'; import 'package:meta/meta.dart'; @@ -19,6 +18,7 @@ import '../model/firestore/github_build_status.dart'; import '../model/firestore/github_gold_status.dart'; import 'firestore/commit_and_tasks.dart'; +import 'firestore/task_cache_service.dart'; export '../model/common/firestore_extensions.dart'; @@ -53,76 +53,57 @@ final kFieldMapRegExp = RegExp( @visibleForTesting /// Mixin that encapsulates Firestore query and caching operations across Cocoon. /// -/// ## Lock-Free Versioned & Set-Based Caching Strategy -/// -/// To resolve read bottlenecks against the Firestore `/tasks` collection without incurring -/// distributed locking overhead, this mixin implements a two-tier optimistic caching architecture: -/// -/// ```mermaid -/// sequenceDiagram -/// autonumber -/// participant Client -/// participant FirestoreQueries -/// participant Redis as Redis Cache -/// participant DB as Firestore DB -/// -/// Note over Client,DB: Scenario 1: Cache Hit Flow -/// Client->>FirestoreQueries: queryTasks(commitSha) -/// FirestoreQueries->>Redis: SMEMBERS tasks_by_commit_ids/$sha -/// Redis-->>FirestoreQueries: [docId1, docId2] -/// FirestoreQueries->>Redis: MGET tasks/docId1, tasks/docId2 -/// Redis-->>FirestoreQueries: [payload1, payload2] -/// FirestoreQueries-->>Client: Return tasks -/// -/// Note over Client,DB: Scenario 2: Partial Cache Recovery -/// Client->>FirestoreQueries: queryTasks(commitSha) -/// FirestoreQueries->>Redis: SMEMBERS tasks_by_commit_ids/$sha -/// Redis-->>FirestoreQueries: [docId1, docId2] -/// FirestoreQueries->>Redis: MGET tasks/docId1, tasks/docId2 -/// Redis-->>FirestoreQueries: [payload1, null (missing)] -/// FirestoreQueries->>DB: batchGetDocuments([docId2]) -/// DB-->>FirestoreQueries: [docId2 document (rev 5)] -/// FirestoreQueries->>Redis: insertVersioned(tasks/docId2, rev=5) -/// FirestoreQueries-->>Client: Return merged tasks -/// -/// Note over Client,DB: Scenario 3: Transactional Write-Back -/// Client->>FirestoreQueries: transaction write (mutate task) -/// FirestoreQueries->>DB: commit transaction (increment rev to 6) -/// DB-->>FirestoreQueries: Transaction commit result (rev 6) -/// FirestoreQueries->>Redis: insertVersioned(tasks/docId, rev=6) -/// Note over Redis: Stale in-flight reads with rev < 6 rejected -/// ``` +/// Task caching orchestration is delegated to [TaskCacheService]. +/// See [TaskCacheService] for full architectural documentation and caching invariant flowcharts. mixin FirestoreQueries { CacheService? get cache => null; Config? get config => null; + + /// Lazy accessor for [TaskCacheService]. + TaskCacheService? get _taskCache { + if (cache == null || !(config?.flags.taskCachingEnabled ?? true)) { + return null; + } + return TaskCacheService(cache: cache!, config: config); + } + Future getDocument(String name, {Transaction? transaction}); + @protected + @visibleForTesting Future> batchGetDocuments( List names, { Transaction? transaction, }); + Future beginTransaction(); + Future commit(Transaction transaction, List writes); + Future batchWriteDocuments( + BatchWriteRequest request, + String database, + ); - static const String _subcacheTasksByCommitIds = 'tasks_by_commit_ids'; + /// Queries all tasks for [commitSha] from Firestore, updates their individual cache entries, + /// Retrieves a [Task] by its document ID, serving from versioned cache if available, + /// or fetching from Firestore DB on cache miss and populating the cache. + Future getTask(AppDocumentId id) async { + final docId = id.documentId; + if (_taskCache != null) { + final result = await _taskCache!.getTaskPayloads([docId]); + if (result.foundTasks.isNotEmpty) { + return result.foundTasks.first; + } + } - Duration get _defaultCacheTtl => - Duration(hours: config?.flags.taskCacheTtlInHours ?? 12); + final docName = p.posix.join(kDatabase, 'documents', kTaskCollectionId, docId); + final document = await getDocument(docName); + final task = Task.fromDocument(document); - /// Caches the individual [Task] payloads into the `tasks` subcache. - Future _cacheTaskDocuments( - List tasks, { - Duration? ttl, - }) async { - if (cache == null || !(config?.flags.taskCachingEnabled ?? true) || tasks.isEmpty) return; - final effectiveTtl = ttl ?? _defaultCacheTtl; - final entries = [ - for (final t in tasks) - VersionedCacheEntry( - key: p.basename(t.name!), - value: Task.serialize(t), - revisionId: t.revisionId, - ttl: effectiveTtl, - ), - ]; - await cache!.insertVersioned('tasks', entries); + await _taskCache?.cacheTaskPayloads([task]); + return task; + } + + /// Caches payloads for modified [Task] entities. + Future cacheTaskPayloads(Iterable tasks) async { + await _taskCache?.cacheTaskPayloads(tasks); } /// Queries all tasks for [commitSha] from Firestore, updates their individual cache entries, @@ -134,17 +115,20 @@ mixin FirestoreQueries { orderMap: {Task.fieldCreateTimestamp: kQueryOrderDescending}, ); final tasks = documents.map(Task.fromDocument).toList(); - if (cache != null && (config?.flags.taskCachingEnabled ?? true)) { + if (_taskCache != null) { final docIds = {for (final t in tasks) p.basename(t.name!)}; await [ - _cacheTaskDocuments(tasks), + _taskCache!.cacheTaskPayloads(tasks), if (docIds.isNotEmpty) - cache!.updateSet( - _subcacheTasksByCommitIds, - commitSha, - docIds, - ttl: _defaultCacheTtl, - ), + () async { + final initialized = await _taskCache!.initializeCommitTaskSet( + commitSha, + docIds, + ); + if (!initialized) { + await _taskCache!.addTasksToCommitSet(commitSha, docIds); + } + }(), ].wait; } return tasks; @@ -152,8 +136,8 @@ mixin FirestoreQueries { /// Explicitly updates the cache when new [Task]s are created. Future updateCacheForCreatedTasks(List tasks) async { - if (cache == null || !(config?.flags.taskCachingEnabled ?? true) || tasks.isEmpty) return; - await _cacheTaskDocuments(tasks); + if (_taskCache == null || tasks.isEmpty) return; + await _taskCache!.cacheTaskPayloads(tasks); final tasksByCommit = >{}; for (final task in tasks) { @@ -163,55 +147,65 @@ mixin FirestoreQueries { } } - await Future.wait( - tasksByCommit.entries.map((entry) async { - final commitSha = entry.key; - final docIds = entry.value; - for (final docId in docIds) { - final added = await cache!.addToSetIfExists( - _subcacheTasksByCommitIds, - commitSha, - docId, - ); - if (!added) { - await _fetchAndCacheCommitTasks(commitSha); - break; - } - } - }), - ); + for (final entry in tasksByCommit.entries) { + final commitSha = entry.key; + final docIds = entry.value; + final setExisted = await _taskCache!.addTasksToCommitSet( + commitSha, + docIds, + ); + if (!setExisted) { + // Set was missing in Redis: fall back to full database query to populate set + await _fetchAndCacheCommitTasks(commitSha); + } + } } - Future handleWriteCacheInvalidation(List writes) async { - if (cache == null || !(config?.flags.taskCachingEnabled ?? true) || writes.isEmpty) return; - final updatedTasks = []; + /// Saves or updates [Task] entities in Firestore and updates the versioned Redis cache. + Future updateTasks(List tasks) async { + if (tasks.isEmpty) return; + final writes = documentsToWrites(tasks); + await batchWriteDocuments(BatchWriteRequest(writes: writes), kDatabase); + await _taskCache?.cacheTaskPayloads(tasks); + } - for (final write in writes) { - final docName = - write.update?.name ?? write.delete ?? write.transform?.document; - if (docName == null || - !docName.contains('/documents/$kTaskCollectionId/')) { - continue; - } - final docId = p.basename(docName); + /// Updates the status of tasks identified by [taskStatusMap] within a single Firestore transaction + /// and updates the versioned task cache. + Future updateTaskStatuses(Map taskStatusMap) async { + if (taskStatusMap.isEmpty) return; - if (write.delete != null) { - await cache!.purge('tasks', docId); - continue; - } + final docIds = taskStatusMap.keys + .map( + (id) => p.posix.join( + kDatabase, + 'documents', + kTaskCollectionId, + id.documentId, + ), + ) + .toList(); - if (write.update != null && write.update!.fields != null) { - try { - updatedTasks.add(Task.fromDocument(write.update!)); - } catch (e) { - log.warn('Unable to parse task document for $docId', e); - } + final transaction = await beginTransaction(); + final docs = await batchGetDocuments(docIds, transaction: transaction); + final tasksToUpdate = []; + + for (final doc in docs) { + final task = Task.fromDocument(doc); + final id = TaskId( + commitSha: task.commitSha, + taskName: task.taskName, + currentAttempt: task.currentAttempt, + ); + final newStatus = taskStatusMap[id]; + if (newStatus != null) { + task.setStatus(newStatus); + tasksToUpdate.add(task); } } - if (updatedTasks.isNotEmpty) { - await updateCacheForCreatedTasks(updatedTasks); - } + final writes = documentsToWrites(tasksToUpdate); + await commit(transaction, writes); + await _taskCache?.cacheTaskPayloads(tasksToUpdate); } /// Wrapper to simplify Firestore query. @@ -325,8 +319,8 @@ mixin FirestoreQueries { transaction: transaction, ); final tasks = documents.map(Task.fromDocument).toList(); - if (cacheResults && transaction == null && cache != null && (config?.flags.taskCachingEnabled ?? true)) { - await _cacheTaskDocuments(tasks); + if (cacheResults && transaction == null && _taskCache != null) { + await _taskCache!.cacheTaskPayloads(tasks); } return tasks; } @@ -363,33 +357,17 @@ mixin FirestoreQueries { int? limit, }) async { List? tasks; - final docIds = await cache!.getSet(_subcacheTasksByCommitIds, commitSha); - if (docIds.isNotEmpty) { - final docIdList = docIds.toList(); - final cachedTasksData = await cache!.getMulti('tasks', docIdList); - final foundTasks = []; - final missingDocIds = []; - - for (var i = 0; i < docIdList.length; i++) { - final data = cachedTasksData[i]; - if (data != null) { - try { - foundTasks.add(Task.deserialize(data)); - } catch (e) { - log.warn('Unable to deserialize cached task for ${docIdList[i]}', e); - missingDocIds.add(docIdList[i]); - } - } else { - missingDocIds.add(docIdList[i]); - } - } + final docIds = await _taskCache!.getTaskIdsForCommit(commitSha); + if (docIds != null && docIds.isNotEmpty) { + final lookupResult = await _taskCache!.getTaskPayloads(docIds); + final foundTasks = lookupResult.foundTasks; - if (missingDocIds.isEmpty) { + if (lookupResult.missingDocIds.isEmpty) { tasks = foundTasks; } else { - final missingNames = missingDocIds + final missingNames = lookupResult.missingDocIds .map( - (missingId) => p.posix.join( + (String missingId) => p.posix.join( kDatabase, 'documents', kTaskCollectionId, @@ -400,7 +378,7 @@ mixin FirestoreQueries { final documents = await batchGetDocuments(missingNames); final fetchedMissingTasks = documents.map(Task.fromDocument).toList(); if (fetchedMissingTasks.isNotEmpty) { - await _cacheTaskDocuments(fetchedMissingTasks); + await _taskCache!.cacheTaskPayloads(fetchedMissingTasks); foundTasks.addAll(fetchedMissingTasks); } tasks = foundTasks; @@ -675,7 +653,7 @@ class FirestoreService with FirestoreQueries { collectionId, documentId: documentId, ); - if (cache != null && (config?.flags.taskCachingEnabled ?? true) && collectionId == kTaskCollectionId) { + if (_taskCache != null && collectionId == kTaskCollectionId) { await updateCacheForCreatedTasks([Task.fromDocument(result)]); } return result; @@ -687,21 +665,19 @@ class FirestoreService with FirestoreQueries { /// Each write succeeds or fails independently. /// /// https://firebase.google.com/docs/firestore/reference/rest/v1/projects.databases.documents/batchWrite + @override Future batchWriteDocuments( BatchWriteRequest request, String database, ) async { - final result = await _api.projects.databases.documents.batchWrite( + return await _api.projects.databases.documents.batchWrite( request, database, ); - if (cache != null && (config?.flags.taskCachingEnabled ?? true) && request.writes != null) { - await handleWriteCacheInvalidation(request.writes!); - } - return result; } /// Begins a read-write transaction. + @override Future beginTransaction() async { final request = BeginTransactionRequest( options: TransactionOptions(readWrite: ReadWrite()), @@ -714,6 +690,7 @@ class FirestoreService with FirestoreQueries { } /// Commits a transaction. + @override Future commit( Transaction transaction, List writes, @@ -722,14 +699,10 @@ class FirestoreService with FirestoreQueries { transaction: transaction.identifier, writes: writes, ); - final result = await _api.projects.databases.documents.commit( + return await _api.projects.databases.documents.commit( request, kDatabase, ); - if (cache != null && (config?.flags.taskCachingEnabled ?? true)) { - await handleWriteCacheInvalidation(writes); - } - return result; } /// Rolls back a transaction. @@ -751,14 +724,10 @@ class FirestoreService with FirestoreQueries { transaction: beginTransactionResponse.transaction, writes: writes, ); - final result = await _api.projects.databases.documents.commit( + return await _api.projects.databases.documents.commit( commitRequest, kDatabase, ); - if (cache != null && (config?.flags.taskCachingEnabled ?? true)) { - await handleWriteCacheInvalidation(writes); - } - return result; } /// Returns Firestore [Value] based on corresponding object type. diff --git a/app_dart/lib/src/service/firestore/task_cache_service.dart b/app_dart/lib/src/service/firestore/task_cache_service.dart new file mode 100644 index 0000000000..c3adbecc36 --- /dev/null +++ b/app_dart/lib/src/service/firestore/task_cache_service.dart @@ -0,0 +1,212 @@ +// Copyright 2019 The Flutter Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +import 'package:path/path.dart' as p; + +import '../../model/firestore/task.dart'; +import '../cache_service.dart'; +import '../config.dart'; + +/// Result container for task payload batch lookups. +final class TaskPayloadLookupResult { + TaskPayloadLookupResult({ + required this.foundTasks, + required this.missingDocIds, + }); + + /// Tasks successfully retrieved and deserialized from cache. + final List foundTasks; + + /// Document IDs whose payload was expired or missing in cache. + final List missingDocIds; +} + +/// Dedicated service encapsulating Redis caching logic for /tasks. +/// +/// ## Two-Tier Lock-Free Task Caching Architecture +/// +/// ### Invariant 1: Task Entity Payload Cache (`tasks/$docId`) +/// - **Monotonically Increasing Revision ID**: Database write transactions establish a strict total order +/// of document states, monotonically numbered by `revisionId`. +/// - **Optimistic Version Guarding**: Write-backs call `insertVersioned(tasks/$docId, payload, revisionId)`. +/// Redis atomically evaluates `newRev > existingRev` (or `EXISTS == 0`). +/// +/// #### Scenario 1.1: Read Task Entry +/// ```mermaid +/// flowchart TD +/// A["Query tasks/$docId Payload"] --> B{"Key Exists in Redis?"} +/// B -- Cache Hit --> C["Return Serialized JSON Payload"] +/// B -- Cache Miss --> D["Fetch Task from DB"] +/// D --> E["cacheTaskPayloads"] +/// E -- "!existingEntry || task.revisionId > existingEntry.revisionId" --> F[Cache write] +/// E -- "task.revisionId <= existingEntry.revisionId" --> G[No cache write] +/// F --> C +/// G --> C +/// ``` +/// +/// #### Scenario 1.2: Write Task Entry +/// ```mermaid +/// flowchart TD +/// A["Write/create task"] --> B[Write task to database, increment revisionId] +/// B --> C[cacheTaskPayloads] +/// C -- "!existingEntry || task.revisionId > existingEntry.revisionId" --> D[Cache write] +/// C -- "task.revisionId <= existingEntry.revisionId" --> E[No cache write] +/// D --> F[Return task] +/// E --> F[Return task] +/// ``` +/// +/// ### Invariant 2: Commit Task Index Set Cache (`tasks_by_commit_ids/$commitSha`) +/// - **Monotonically Growing Set & Immutable Commit Association**: A task's `commitSha` is permanent +/// upon creation and never changes. Task status updates **never alter set membership** and **never touch +/// `tasks_by_commit_ids`**. We only ever modify `tasks_by_commit_ids` when new tasks are created. +/// +/// #### Scenario 2.1: Read `taskIdsByCommitId` +/// ```mermaid +/// flowchart TD +/// A["getTaskIdsForCommit(commitSha)"] --> B["getTaskPayloads"] +/// B -- Cache hit --> C["Return Set of Task IDs (SMEMBERS)"] +/// B -- Cache miss --> D["initializeCommitTaskSet (setIfNotExists)"] +/// D -- "Success (Set Initialized)" --> E["Set Created with Task IDs"] +/// D -- "Failed (Intervening Initialization)" --> F[addTasksToCommitSet] +/// F --> C +/// E --> C +/// ``` +/// +/// #### Scenario 2.2: Create Task +/// ```mermaid +/// flowchart TD +/// A[Create task] --> B[addTasksToCommitSet] +/// B -- Cache entry exists, taskId added --> C[Return] +/// B -- Cache entry does not exist --> D[Fetch tasks for task.commitId] +/// D --> E[initializeCommitTaskSet] +/// E -- "Success (Set Initialized)" --> C +/// E -- "Failed (Intervening Initialization)" --> F[addTasksToCommitSet] +/// F --> C +/// ``` +final class TaskCacheService { + TaskCacheService({required this.cache, this.config}); + + final CacheService cache; + final Config? config; + + /// Returns whether caching is enabled in dynamic config. + bool get isEnabled => config?.flags.taskCachingEnabled ?? true; + + /// Returns the default cache TTL based on dynamic configuration. + Duration get defaultTtl => + Duration(hours: config?.flags.taskCacheTtlInHours ?? 12); + + // -------------------------------------------------------------------------- + // Invariant 1: Task Entity Payload Cache (`tasks/$docId`) + // -------------------------------------------------------------------------- + + /// Caches full [Task] document payloads using optimistic versioning (`revisionId`). + /// + /// The payload update only succeeds if `task.revisionId > existingRev` or the cache key is missing. + Future cacheTaskPayloads(Iterable tasks, {Duration? ttl}) async { + if (tasks.isEmpty || !isEnabled) return; + final effectiveTtl = ttl ?? defaultTtl; + final entries = [ + for (final t in tasks) + VersionedCacheEntry( + key: p.basename(t.name!), + value: Task.serialize(t), + revisionId: t.revisionId, + ttl: effectiveTtl, + ), + ]; + await cache.insertVersioned('tasks', entries); + } + + /// Evicts a deleted task payload from Redis. + Future evictTaskPayload(String docId) async { + if (!isEnabled) return; + await cache.purge('tasks', docId); + } + + /// Performs a batch lookup of task payloads by document IDs. + Future getTaskPayloads(Iterable docIds) async { + final docIdList = docIds.toList(); + if (docIdList.isEmpty || !isEnabled) { + return TaskPayloadLookupResult(foundTasks: [], missingDocIds: docIdList); + } + + final cachedData = await cache.getMulti('tasks', docIdList); + final foundTasks = []; + final missingDocIds = []; + + for (var i = 0; i < docIdList.length; i++) { + final data = cachedData[i]; + if (data != null) { + try { + foundTasks.add(Task.deserialize(data)); + } catch (_) { + missingDocIds.add(docIdList[i]); + } + } else { + missingDocIds.add(docIdList[i]); + } + } + + return TaskPayloadLookupResult( + foundTasks: foundTasks, + missingDocIds: missingDocIds, + ); + } + + // -------------------------------------------------------------------------- + // Invariant 2: Commit Task Index Set Cache (`tasks_by_commit_ids/$commitSha`) + // -------------------------------------------------------------------------- + + /// Retrieves the set of task document IDs associated with [commitSha]. + /// + /// Returns `null` if the set key does not exist in Redis (cache miss). + Future?> getTaskIdsForCommit(String commitSha) async { + if (!isEnabled) return null; + final set = await cache.getSet('tasks_by_commit_ids', commitSha); + if (set.isEmpty) return null; + return set; + } + + /// Adds [taskIds] to the commit set if the set already exists in Redis (`addToSetIfExists`). + /// + /// Returns `true` if the set existed and task IDs were added; `false` if the set key is missing. + Future addTasksToCommitSet( + String commitSha, + Iterable taskIds, { + Duration? ttl, + }) async { + if (taskIds.isEmpty || !isEnabled) return false; + var allAdded = true; + for (final taskId in taskIds) { + final added = await cache.addToSetIfExists( + 'tasks_by_commit_ids', + commitSha, + taskId, + ); + if (!added) { + allAdded = false; + } + } + return allAdded; + } + + /// Initializes the commit task set if it does NOT already exist (`setIfNotExists`). + /// + /// Returns `true` if the set was successfully updated/initialized. + Future initializeCommitTaskSet( + String commitSha, + Iterable taskIds, { + Duration? ttl, + }) async { + if (!isEnabled) return false; + await cache.updateSet( + 'tasks_by_commit_ids', + commitSha, + taskIds.toSet(), + ttl: ttl ?? defaultTtl, + ); + return true; + } +} diff --git a/app_dart/lib/src/service/luci_build_service.dart b/app_dart/lib/src/service/luci_build_service.dart index b834078b93..b831cca153 100644 --- a/app_dart/lib/src/service/luci_build_service.dart +++ b/app_dart/lib/src/service/luci_build_service.dart @@ -12,7 +12,7 @@ import 'package:cocoon_common/task_status.dart'; import 'package:cocoon_server/logging.dart'; import 'package:fixnum/fixnum.dart'; import 'package:github/github.dart'; -import 'package:googleapis/firestore/v1.dart' hide Status; + import 'package:meta/meta.dart'; import '../../cocoon_service.dart'; @@ -1169,10 +1169,7 @@ class LuciBuildService { task.resetAsRetry(); task.setStatus(TaskStatus.inProgress); - await _firestore.batchWriteDocuments( - BatchWriteRequest(writes: documentsToWrites([task], exists: false)), - kDatabase, - ); + await _firestore.updateTasks([task]); return task.currentAttempt; } diff --git a/app_dart/lib/src/service/scheduler.dart b/app_dart/lib/src/service/scheduler.dart index 1541616f5e..6f4f2199ab 100644 --- a/app_dart/lib/src/service/scheduler.dart +++ b/app_dart/lib/src/service/scheduler.dart @@ -257,6 +257,7 @@ class Scheduler { await _firestore.writeViaTransaction( documentsToWrites([...tasks, commit], exists: false), ); + await _firestore.updateCacheForCreatedTasks(tasks); } /// Schedule all builds in batch requests instead of a single request. diff --git a/app_dart/test/service/firestore_cache_test.dart b/app_dart/test/service/firestore_cache_test.dart index 25b62c19d2..d400a4372e 100644 --- a/app_dart/test/service/firestore_cache_test.dart +++ b/app_dart/test/service/firestore_cache_test.dart @@ -7,7 +7,7 @@ import 'package:cocoon_integration_test/testing.dart'; import 'package:cocoon_server_test/test_logging.dart'; import 'package:cocoon_service/src/model/firestore/task.dart'; import 'package:cocoon_service/src/service/cache_service.dart'; -import 'package:cocoon_service/src/service/firestore.dart'; + import 'package:googleapis/firestore/v1.dart'; import 'package:path/path.dart' as p; import 'package:test/test.dart'; @@ -160,10 +160,7 @@ void main() { // Mutate task status and increment revisionId task1.setStatus(TaskStatus.succeeded); - await firestore.batchWriteDocuments( - BatchWriteRequest(writes: documentsToWrites([task1])), - kDatabase, - ); + await firestore.updateTasks([task1]); // Verify that tasks_by_commit_ids remains intact expect( @@ -181,7 +178,7 @@ void main() { ); test( - 'writeViaTransaction updates versioned task cache in-place lock-free', + 'updateTasks updates versioned task cache in-place lock-free', () async { final task = generateFirestoreTask( 1, @@ -193,11 +190,7 @@ void main() { await firestore.queryAllTasksForCommit(commitSha: commitSha); task.setStatus(TaskStatus.succeeded); - await firestore.writeViaTransaction([ - Write( - update: Document(name: task.name, fields: task.fields), - ), - ]); + await firestore.updateTasks([task]); final updatedTask = await Task.fromFirestore( firestore, diff --git a/app_dart/test/service/task_cache_service_test.dart b/app_dart/test/service/task_cache_service_test.dart new file mode 100644 index 0000000000..c089c77dac --- /dev/null +++ b/app_dart/test/service/task_cache_service_test.dart @@ -0,0 +1,125 @@ +// Copyright 2019 The Flutter Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +import 'package:cocoon_common/task_status.dart'; +import 'package:cocoon_integration_test/testing.dart'; +import 'package:cocoon_service/src/model/firestore/task.dart'; +import 'package:cocoon_service/src/service/cache_service.dart'; +import 'package:cocoon_service/src/service/firestore/task_cache_service.dart'; +import 'package:cocoon_service/src/service/flags/dynamic_config.dart'; +import 'package:path/path.dart' as p; +import 'package:test/test.dart'; + +void main() { + late InMemoryCacheService cache; + late FakeConfig config; + late TaskCacheService taskCache; + + Task generateTestTask( + int attempt, { + String name = 'Linux A', + String commitSha = 'sha123', + TaskStatus status = TaskStatus.inProgress, + }) { + return Task( + builderName: name, + currentAttempt: attempt, + commitSha: commitSha, + bringup: false, + createTimestamp: 1000, + startTimestamp: 1000, + endTimestamp: 0, + status: status, + testFlaky: false, + buildNumber: null, + ); + } + + String docIdFor(Task task) => p.basename(task.name!); + + setUp(() { + cache = InMemoryCacheService(); + config = FakeConfig(dynamicConfig: DynamicConfig(taskCachingEnabled: true)); + taskCache = TaskCacheService(cache: cache, config: config); + }); + + group('TaskCacheService Payload Caching', () { + test('cacheTaskPayloads stores versioned payload and getTaskPayloads retrieves it', () async { + final task = generateTestTask(1); + await taskCache.cacheTaskPayloads([task]); + + final result = await taskCache.getTaskPayloads([docIdFor(task)]); + expect(result.foundTasks.length, equals(1)); + expect(result.foundTasks.first.taskName, equals('Linux A')); + expect(result.missingDocIds, isEmpty); + }); + + test('evictTaskPayload purges item from cache', () async { + final task = generateTestTask(1); + await taskCache.cacheTaskPayloads([task]); + + await taskCache.evictTaskPayload(docIdFor(task)); + final result = await taskCache.getTaskPayloads([docIdFor(task)]); + expect(result.foundTasks, isEmpty); + expect(result.missingDocIds, contains(docIdFor(task))); + }); + + test('getTaskPayloads correctly reports missing and found items', () async { + final task1 = generateTestTask(1, name: 'Linux A'); + await taskCache.cacheTaskPayloads([task1]); + + final result = await taskCache.getTaskPayloads([docIdFor(task1), 'non_existent_doc']); + expect(result.foundTasks.length, equals(1)); + expect(result.foundTasks.first.taskName, equals('Linux A')); + expect(result.missingDocIds, equals(['non_existent_doc'])); + }); + + test('cacheTaskPayloads respects taskCachingEnabled flag', () async { + config.dynamicConfig = DynamicConfig(taskCachingEnabled: false); + final task = generateTestTask(1); + await taskCache.cacheTaskPayloads([task]); + + final result = await taskCache.getTaskPayloads([docIdFor(task)]); + expect(result.foundTasks, isEmpty); + expect(result.missingDocIds, contains(docIdFor(task))); + }); + }); + + group('TaskCacheService Commit Set Caching', () { + test('getTaskIdsForCommit returns null on cache miss', () async { + final taskIds = await taskCache.getTaskIdsForCommit('sha123'); + expect(taskIds, isNull); + }); + + test('initializeCommitTaskSet populates set and getTaskIdsForCommit retrieves it', () async { + final created = await taskCache.initializeCommitTaskSet('sha123', ['doc1', 'doc2']); + expect(created, isTrue); + + final taskIds = await taskCache.getTaskIdsForCommit('sha123'); + expect(taskIds, equals({'doc1', 'doc2'})); + }); + + test('addTasksToCommitSet adds items only if set exists', () async { + // First attempt on missing set returns false + final addedMissing = await taskCache.addTasksToCommitSet('sha123', ['doc1']); + expect(addedMissing, isFalse); + + // Initialize set + await taskCache.initializeCommitTaskSet('sha123', ['doc1']); + + // Second attempt on existing set returns true and updates set + final addedExisting = await taskCache.addTasksToCommitSet('sha123', ['doc2']); + expect(addedExisting, isTrue); + + final taskIds = await taskCache.getTaskIdsForCommit('sha123'); + expect(taskIds, equals({'doc1', 'doc2'})); + }); + + test('initializeCommitTaskSet returns false if set flag is disabled', () async { + config.dynamicConfig = DynamicConfig(taskCachingEnabled: false); + final first = await taskCache.initializeCommitTaskSet('sha123', ['doc1']); + expect(first, isFalse); + }); + }); +} diff --git a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart index 4903cb86da..c7faa28a35 100644 --- a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart +++ b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart @@ -318,13 +318,9 @@ abstract base class _FakeInMemoryFirestoreService if (p.split(database).length != 4) { throw StateError('Unexpected database: $database'); } - final response = BatchWriteResponse( + return BatchWriteResponse( status: _batchWriteSync(request.writes ?? const []), ); - if (cache != null && request.writes != null) { - await handleWriteCacheInvalidation(request.writes!); - } - return response; } /// Same as [batchWriteDocuments], but does not yield the microtask loop. @@ -466,9 +462,6 @@ abstract base class _FakeInMemoryFirestoreService } final updated = _now().toUtc().toIso8601String(); - if (cache != null) { - await handleWriteCacheInvalidation(writes); - } return CommitResponse( commitTime: updated, writeResults: List.generate( From be02c2a4064d7ce5ca212cb9dbf1d6cb43516d16 Mon Sep 17 00:00:00 2001 From: Jackson Gardner Date: Thu, 23 Jul 2026 13:53:58 -0700 Subject: [PATCH 5/7] Fix analysis errors. --- app_dart/test/service/task_cache_service_test.dart | 3 +++ .../lib/src/fakes/fake_firestore_service.dart | 4 +++- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/app_dart/test/service/task_cache_service_test.dart b/app_dart/test/service/task_cache_service_test.dart index c089c77dac..0582a27cab 100644 --- a/app_dart/test/service/task_cache_service_test.dart +++ b/app_dart/test/service/task_cache_service_test.dart @@ -4,6 +4,7 @@ import 'package:cocoon_common/task_status.dart'; import 'package:cocoon_integration_test/testing.dart'; +import 'package:cocoon_server_test/test_logging.dart'; import 'package:cocoon_service/src/model/firestore/task.dart'; import 'package:cocoon_service/src/service/cache_service.dart'; import 'package:cocoon_service/src/service/firestore/task_cache_service.dart'; @@ -12,6 +13,8 @@ import 'package:path/path.dart' as p; import 'package:test/test.dart'; void main() { + useTestLoggerPerTest(); + late InMemoryCacheService cache; late FakeConfig config; late TaskCacheService taskCache; diff --git a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart index c7faa28a35..4d1e198a01 100644 --- a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart +++ b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart @@ -241,7 +241,9 @@ abstract base class _FakeInMemoryFirestoreService updateTime: (updated ?? _now()).toUtc().toIso8601String(), ); if (cache != null && collection == kTaskCollectionId) { - unawaited(updateCacheForCreatedTasks([Task.fromDocument(_documents[name]!)])); + unawaited( + updateCacheForCreatedTasks([Task.fromDocument(_documents[name]!)]), + ); } return _clone(_documents[name]!); } From 2f706abaf7ad16b34bbf4b5190b2d3eae513223d Mon Sep 17 00:00:00 2001 From: Jackson Gardner Date: Thu, 23 Jul 2026 13:57:12 -0700 Subject: [PATCH 6/7] Fix formatting. --- .../dart_internal_subscription.dart | 1 - app_dart/lib/src/service/firestore.dart | 16 +++-- .../service/firestore/task_cache_service.dart | 4 +- .../luci_build_service/user_data.g.dart | 25 +++---- .../scheduler/backfill_grid_test.dart | 8 +-- .../test/service/firestore_cache_test.dart | 45 +++++++----- app_dart/test/service/scheduler_test.dart | 41 +++++------ .../test/service/task_cache_service_test.dart | 69 ++++++++++++------- 8 files changed, 114 insertions(+), 95 deletions(-) diff --git a/app_dart/lib/src/request_handlers/dart_internal_subscription.dart b/app_dart/lib/src/request_handlers/dart_internal_subscription.dart index 46fc5b0d77..4f25a22697 100644 --- a/app_dart/lib/src/request_handlers/dart_internal_subscription.dart +++ b/app_dart/lib/src/request_handlers/dart_internal_subscription.dart @@ -8,7 +8,6 @@ import 'package:buildbucket/buildbucket_pb.dart' as bbv2; import 'package:cocoon_common/is_dart_internal.dart'; import 'package:cocoon_server/logging.dart'; - import '../../cocoon_service.dart'; import '../model/bbv2_extension.dart'; import '../model/firestore/task.dart' as fs; diff --git a/app_dart/lib/src/service/firestore.dart b/app_dart/lib/src/service/firestore.dart index bb80ff6995..33c2528f21 100644 --- a/app_dart/lib/src/service/firestore.dart +++ b/app_dart/lib/src/service/firestore.dart @@ -93,7 +93,12 @@ mixin FirestoreQueries { } } - final docName = p.posix.join(kDatabase, 'documents', kTaskCollectionId, docId); + final docName = p.posix.join( + kDatabase, + 'documents', + kTaskCollectionId, + docId, + ); final document = await getDocument(docName); final task = Task.fromDocument(document); @@ -332,7 +337,9 @@ mixin FirestoreQueries { int? limit, Transaction? transaction, }) async { - if (transaction == null && cache != null && (config?.flags.taskCachingEnabled ?? true)) { + if (transaction == null && + cache != null && + (config?.flags.taskCachingEnabled ?? true)) { return await _queryTasksByCommitCached( commitSha: commitSha, name: name, @@ -699,10 +706,7 @@ class FirestoreService with FirestoreQueries { transaction: transaction.identifier, writes: writes, ); - return await _api.projects.databases.documents.commit( - request, - kDatabase, - ); + return await _api.projects.databases.documents.commit(request, kDatabase); } /// Rolls back a transaction. diff --git a/app_dart/lib/src/service/firestore/task_cache_service.dart b/app_dart/lib/src/service/firestore/task_cache_service.dart index c3adbecc36..8b8b755818 100644 --- a/app_dart/lib/src/service/firestore/task_cache_service.dart +++ b/app_dart/lib/src/service/firestore/task_cache_service.dart @@ -126,7 +126,9 @@ final class TaskCacheService { } /// Performs a batch lookup of task payloads by document IDs. - Future getTaskPayloads(Iterable docIds) async { + Future getTaskPayloads( + Iterable docIds, + ) async { final docIdList = docIds.toList(); if (docIdList.isEmpty || !isEnabled) { return TaskPayloadLookupResult(foundTasks: [], missingDocIds: docIdList); diff --git a/app_dart/lib/src/service/luci_build_service/user_data.g.dart b/app_dart/lib/src/service/luci_build_service/user_data.g.dart index f67563c6b7..267b85c819 100644 --- a/app_dart/lib/src/service/luci_build_service/user_data.g.dart +++ b/app_dart/lib/src/service/luci_build_service/user_data.g.dart @@ -73,21 +73,16 @@ const _$CiStageEnumMap = { }; PostsubmitUserData _$PostsubmitUserDataFromJson(Map json) => - $checkedCreate( - 'PostsubmitUserData', - json, - ($checkedConvert) { - final val = PostsubmitUserData( - checkRunId: $checkedConvert( - 'check_run_id', - (v) => (v as num?)?.toInt(), - ), - taskId: $checkedConvert('task_id', (v) => TaskId.parse(v as String)), - ); - return val; - }, - fieldKeyMap: const {'checkRunId': 'check_run_id', 'taskId': 'task_id'}, - ); + $checkedCreate('PostsubmitUserData', json, ($checkedConvert) { + final val = PostsubmitUserData( + checkRunId: $checkedConvert( + 'check_run_id', + (v) => (v as num?)?.toInt(), + ), + taskId: $checkedConvert('task_id', (v) => TaskId.parse(v as String)), + ); + return val; + }, fieldKeyMap: const {'checkRunId': 'check_run_id', 'taskId': 'task_id'}); Map _$PostsubmitUserDataToJson(PostsubmitUserData instance) => { diff --git a/app_dart/test/request_handlers/scheduler/backfill_grid_test.dart b/app_dart/test/request_handlers/scheduler/backfill_grid_test.dart index 1178d2b48e..eca59fb123 100644 --- a/app_dart/test/request_handlers/scheduler/backfill_grid_test.dart +++ b/app_dart/test/request_handlers/scheduler/backfill_grid_test.dart @@ -100,11 +100,9 @@ void main() { reason: 'Target 1 is marked backfill: false, so it is not eligible', ); - expect( - grid.skippableTasks, - [isSkippableTask.hasTask(t1c2)], - reason: 'Target 1 is marked backfill: false, so it is skipped', - ); + expect(grid.skippableTasks, [ + isSkippableTask.hasTask(t1c2), + ], reason: 'Target 1 is marked backfill: false, so it is skipped'); }); test('filters out tasks that are missing from ToT', () { diff --git a/app_dart/test/service/firestore_cache_test.dart b/app_dart/test/service/firestore_cache_test.dart index d400a4372e..cf168861d8 100644 --- a/app_dart/test/service/firestore_cache_test.dart +++ b/app_dart/test/service/firestore_cache_test.dart @@ -205,23 +205,32 @@ void main() { }, ); - test( - 'updateCacheForCreatedTasks groups tasks by commit SHA', - () async { - final task1 = generateFirestoreTask(1, commitSha: 'shaA', name: 'Linux A'); - final task2 = generateFirestoreTask(2, commitSha: 'shaA', name: 'Linux B'); - final task3 = generateFirestoreTask(3, commitSha: 'shaB', name: 'Linux C'); - - await firestore.updateCacheForCreatedTasks([task1, task2, task3]); - - final cachedTask1 = await cache.get('tasks', p.basename(task1.name!)); - final cachedTask2 = await cache.get('tasks', p.basename(task2.name!)); - final cachedTask3 = await cache.get('tasks', p.basename(task3.name!)); - - expect(cachedTask1, isNotNull); - expect(cachedTask2, isNotNull); - expect(cachedTask3, isNotNull); - }, - ); + test('updateCacheForCreatedTasks groups tasks by commit SHA', () async { + final task1 = generateFirestoreTask( + 1, + commitSha: 'shaA', + name: 'Linux A', + ); + final task2 = generateFirestoreTask( + 2, + commitSha: 'shaA', + name: 'Linux B', + ); + final task3 = generateFirestoreTask( + 3, + commitSha: 'shaB', + name: 'Linux C', + ); + + await firestore.updateCacheForCreatedTasks([task1, task2, task3]); + + final cachedTask1 = await cache.get('tasks', p.basename(task1.name!)); + final cachedTask2 = await cache.get('tasks', p.basename(task2.name!)); + final cachedTask3 = await cache.get('tasks', p.basename(task3.name!)); + + expect(cachedTask1, isNotNull); + expect(cachedTask2, isNotNull); + expect(cachedTask3, isNotNull); + }); }); } diff --git a/app_dart/test/service/scheduler_test.dart b/app_dart/test/service/scheduler_test.dart index 5e56d325bd..af40888d30 100644 --- a/app_dart/test/service/scheduler_test.dart +++ b/app_dart/test/service/scheduler_test.dart @@ -4131,11 +4131,9 @@ targets: final pullRequest = generatePullRequest(authorLogin: 'joe-flutter'); await scheduler.triggerPresubmitTargets(pullRequest: pullRequest); - expect( - fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), - ['Linux engine_build'], - reason: 'Should still run engine phase', - ); + expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ + 'Linux engine_build', + ], reason: 'Should still run engine phase'); }); test('still runs engine builds (engine/**)', () async { @@ -4146,11 +4144,9 @@ targets: final pullRequest = generatePullRequest(authorLogin: 'joe-flutter'); await scheduler.triggerPresubmitTargets(pullRequest: pullRequest); - expect( - fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), - ['Linux engine_build'], - reason: 'Should still run engine phase', - ); + expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ + 'Linux engine_build', + ], reason: 'Should still run engine phase'); }); test( @@ -4167,11 +4163,9 @@ targets: ); await scheduler.triggerPresubmitTargets(pullRequest: pullRequest); - expect( - fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), - ['Linux engine_build'], - reason: 'Should still run engine phase', - ); + expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ + 'Linux engine_build', + ], reason: 'Should still run engine phase'); }, ); @@ -4187,11 +4181,10 @@ targets: ), reason: 'Should use the base ref for the engine artifacts', ); - expect( - fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), - ['Linux A', 'Linux analyze'], - reason: 'Should skip Linux engine_build', - ); + expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ + 'Linux A', + 'Linux analyze', + ], reason: 'Should skip Linux engine_build'); expect( firestore, @@ -4220,11 +4213,9 @@ targets: ), reason: 'Should use the base ref for the engine artifacts', ); - expect( - fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), - ['Linux analyze'], - reason: 'Only scheduled a special-cased build', - ); + expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ + 'Linux analyze', + ], reason: 'Only scheduled a special-cased build'); }); }); group('process unified check run', () { diff --git a/app_dart/test/service/task_cache_service_test.dart b/app_dart/test/service/task_cache_service_test.dart index 0582a27cab..9e10769814 100644 --- a/app_dart/test/service/task_cache_service_test.dart +++ b/app_dart/test/service/task_cache_service_test.dart @@ -48,15 +48,18 @@ void main() { }); group('TaskCacheService Payload Caching', () { - test('cacheTaskPayloads stores versioned payload and getTaskPayloads retrieves it', () async { - final task = generateTestTask(1); - await taskCache.cacheTaskPayloads([task]); - - final result = await taskCache.getTaskPayloads([docIdFor(task)]); - expect(result.foundTasks.length, equals(1)); - expect(result.foundTasks.first.taskName, equals('Linux A')); - expect(result.missingDocIds, isEmpty); - }); + test( + 'cacheTaskPayloads stores versioned payload and getTaskPayloads retrieves it', + () async { + final task = generateTestTask(1); + await taskCache.cacheTaskPayloads([task]); + + final result = await taskCache.getTaskPayloads([docIdFor(task)]); + expect(result.foundTasks.length, equals(1)); + expect(result.foundTasks.first.taskName, equals('Linux A')); + expect(result.missingDocIds, isEmpty); + }, + ); test('evictTaskPayload purges item from cache', () async { final task = generateTestTask(1); @@ -72,7 +75,10 @@ void main() { final task1 = generateTestTask(1, name: 'Linux A'); await taskCache.cacheTaskPayloads([task1]); - final result = await taskCache.getTaskPayloads([docIdFor(task1), 'non_existent_doc']); + final result = await taskCache.getTaskPayloads([ + docIdFor(task1), + 'non_existent_doc', + ]); expect(result.foundTasks.length, equals(1)); expect(result.foundTasks.first.taskName, equals('Linux A')); expect(result.missingDocIds, equals(['non_existent_doc'])); @@ -95,34 +101,49 @@ void main() { expect(taskIds, isNull); }); - test('initializeCommitTaskSet populates set and getTaskIdsForCommit retrieves it', () async { - final created = await taskCache.initializeCommitTaskSet('sha123', ['doc1', 'doc2']); - expect(created, isTrue); - - final taskIds = await taskCache.getTaskIdsForCommit('sha123'); - expect(taskIds, equals({'doc1', 'doc2'})); - }); + test( + 'initializeCommitTaskSet populates set and getTaskIdsForCommit retrieves it', + () async { + final created = await taskCache.initializeCommitTaskSet('sha123', [ + 'doc1', + 'doc2', + ]); + expect(created, isTrue); + + final taskIds = await taskCache.getTaskIdsForCommit('sha123'); + expect(taskIds, equals({'doc1', 'doc2'})); + }, + ); test('addTasksToCommitSet adds items only if set exists', () async { // First attempt on missing set returns false - final addedMissing = await taskCache.addTasksToCommitSet('sha123', ['doc1']); + final addedMissing = await taskCache.addTasksToCommitSet('sha123', [ + 'doc1', + ]); expect(addedMissing, isFalse); // Initialize set await taskCache.initializeCommitTaskSet('sha123', ['doc1']); // Second attempt on existing set returns true and updates set - final addedExisting = await taskCache.addTasksToCommitSet('sha123', ['doc2']); + final addedExisting = await taskCache.addTasksToCommitSet('sha123', [ + 'doc2', + ]); expect(addedExisting, isTrue); final taskIds = await taskCache.getTaskIdsForCommit('sha123'); expect(taskIds, equals({'doc1', 'doc2'})); }); - test('initializeCommitTaskSet returns false if set flag is disabled', () async { - config.dynamicConfig = DynamicConfig(taskCachingEnabled: false); - final first = await taskCache.initializeCommitTaskSet('sha123', ['doc1']); - expect(first, isFalse); - }); + test( + 'initializeCommitTaskSet returns false if set flag is disabled', + () async { + config.dynamicConfig = DynamicConfig(taskCachingEnabled: false); + final first = await taskCache.initializeCommitTaskSet('sha123', [ + 'doc1', + ]); + expect(first, isFalse); + }, + ); }); } From c8376530eb421212fb7d9fe18ed5b0c7bc3482ac Mon Sep 17 00:00:00 2001 From: Jackson Gardner Date: Thu, 23 Jul 2026 16:31:20 -0700 Subject: [PATCH 7/7] Fix formatting. --- .../luci_build_service/user_data.g.dart | 25 ++++++----- .../scheduler/backfill_grid_test.dart | 8 ++-- app_dart/test/service/scheduler_test.dart | 41 +++++++++++-------- 3 files changed, 45 insertions(+), 29 deletions(-) diff --git a/app_dart/lib/src/service/luci_build_service/user_data.g.dart b/app_dart/lib/src/service/luci_build_service/user_data.g.dart index 267b85c819..f67563c6b7 100644 --- a/app_dart/lib/src/service/luci_build_service/user_data.g.dart +++ b/app_dart/lib/src/service/luci_build_service/user_data.g.dart @@ -73,16 +73,21 @@ const _$CiStageEnumMap = { }; PostsubmitUserData _$PostsubmitUserDataFromJson(Map json) => - $checkedCreate('PostsubmitUserData', json, ($checkedConvert) { - final val = PostsubmitUserData( - checkRunId: $checkedConvert( - 'check_run_id', - (v) => (v as num?)?.toInt(), - ), - taskId: $checkedConvert('task_id', (v) => TaskId.parse(v as String)), - ); - return val; - }, fieldKeyMap: const {'checkRunId': 'check_run_id', 'taskId': 'task_id'}); + $checkedCreate( + 'PostsubmitUserData', + json, + ($checkedConvert) { + final val = PostsubmitUserData( + checkRunId: $checkedConvert( + 'check_run_id', + (v) => (v as num?)?.toInt(), + ), + taskId: $checkedConvert('task_id', (v) => TaskId.parse(v as String)), + ); + return val; + }, + fieldKeyMap: const {'checkRunId': 'check_run_id', 'taskId': 'task_id'}, + ); Map _$PostsubmitUserDataToJson(PostsubmitUserData instance) => { diff --git a/app_dart/test/request_handlers/scheduler/backfill_grid_test.dart b/app_dart/test/request_handlers/scheduler/backfill_grid_test.dart index eca59fb123..1178d2b48e 100644 --- a/app_dart/test/request_handlers/scheduler/backfill_grid_test.dart +++ b/app_dart/test/request_handlers/scheduler/backfill_grid_test.dart @@ -100,9 +100,11 @@ void main() { reason: 'Target 1 is marked backfill: false, so it is not eligible', ); - expect(grid.skippableTasks, [ - isSkippableTask.hasTask(t1c2), - ], reason: 'Target 1 is marked backfill: false, so it is skipped'); + expect( + grid.skippableTasks, + [isSkippableTask.hasTask(t1c2)], + reason: 'Target 1 is marked backfill: false, so it is skipped', + ); }); test('filters out tasks that are missing from ToT', () { diff --git a/app_dart/test/service/scheduler_test.dart b/app_dart/test/service/scheduler_test.dart index af40888d30..5e56d325bd 100644 --- a/app_dart/test/service/scheduler_test.dart +++ b/app_dart/test/service/scheduler_test.dart @@ -4131,9 +4131,11 @@ targets: final pullRequest = generatePullRequest(authorLogin: 'joe-flutter'); await scheduler.triggerPresubmitTargets(pullRequest: pullRequest); - expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ - 'Linux engine_build', - ], reason: 'Should still run engine phase'); + expect( + fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), + ['Linux engine_build'], + reason: 'Should still run engine phase', + ); }); test('still runs engine builds (engine/**)', () async { @@ -4144,9 +4146,11 @@ targets: final pullRequest = generatePullRequest(authorLogin: 'joe-flutter'); await scheduler.triggerPresubmitTargets(pullRequest: pullRequest); - expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ - 'Linux engine_build', - ], reason: 'Should still run engine phase'); + expect( + fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), + ['Linux engine_build'], + reason: 'Should still run engine phase', + ); }); test( @@ -4163,9 +4167,11 @@ targets: ); await scheduler.triggerPresubmitTargets(pullRequest: pullRequest); - expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ - 'Linux engine_build', - ], reason: 'Should still run engine phase'); + expect( + fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), + ['Linux engine_build'], + reason: 'Should still run engine phase', + ); }, ); @@ -4181,10 +4187,11 @@ targets: ), reason: 'Should use the base ref for the engine artifacts', ); - expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ - 'Linux A', - 'Linux analyze', - ], reason: 'Should skip Linux engine_build'); + expect( + fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), + ['Linux A', 'Linux analyze'], + reason: 'Should skip Linux engine_build', + ); expect( firestore, @@ -4213,9 +4220,11 @@ targets: ), reason: 'Should use the base ref for the engine artifacts', ); - expect(fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), [ - 'Linux analyze', - ], reason: 'Only scheduled a special-cased build'); + expect( + fakeLuciBuildService.scheduledTryBuilds.map((t) => t.name), + ['Linux analyze'], + reason: 'Only scheduled a special-cased build', + ); }); }); group('process unified check run', () {