Implement lock-free versioned (revisionId) and Set-based Redis caching for /tasks#5103
Implement lock-free versioned (revisionId) and Set-based Redis caching for /tasks#5103eyebrowsoffire wants to merge 3 commits into
Conversation
…g 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).
|
This pull request is not mergeable in its current state, likely because of a merge conflict. Pre-submit CI jobs were not triggered. Pushing a new commit to this branch that resolves the issue will result in pre-submit jobs being scheduled. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a lock-free, versioned, and set-based caching strategy for Firestore tasks in Cocoon to resolve read bottlenecks. It adds optimistic concurrency tracking via a revisionId on tasks, introduces batch and set operations to the CacheService (with Redis and in-memory implementations), and integrates caching into FirestoreQueries. The review feedback highlights critical issues: a memory leak in Redis where revision IDs are stored in a non-expiring shared hash, incorrect caching behavior when tasks are deleted, and a performance bottleneck caused by fetching missing tasks sequentially in a loop instead of in parallel.
|
|
||
| local revKey = "revisions/" .. key | ||
| local existingRev = tonumber(redis.call("get", revKey) or 0) | ||
| if rev > existingRev or (not redis.call("exists", key)) then |
There was a problem hiding this comment.
Gemini suggests a truthy take for "if the incoming revision is newer, OR if the main key is missing entirely, write it.":
I don't know REDIS or Lua all that well, but it suggests that "key exists == 1" and that "not 0" is false.
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| /// ### 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 | ||
| /// [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]. |
There was a problem hiding this comment.
I think this could have used a mermaidjs sequence diagram or something. It was a lot to read and I'm fairly certain I understood about 25% of it via reading redis.io docs.
| return Uint8List.fromList( | ||
| utf8.encode( |
There was a problem hiding this comment.
utf8.encode already returns a Uint8List: https://api.dart.dev/dart-convert/Utf8Codec/encode.html
| } | ||
|
|
||
| static Task _deserializeTask(Uint8List data) { | ||
| final jsonMap = json.decode(utf8.decode(data)) as Map<String, dynamic>; |
There was a problem hiding this comment.
inner matan: Map<String, Object?>
| /// Caches the individual [Task] payloads into the `tasks` subcache. | ||
| Future<void> _cacheTaskDocuments( | ||
| List<Task> tasks, { | ||
| Duration ttl = const Duration(hours: 12), |
There was a problem hiding this comment.
is const Duration(hours: 12) a number we every think we'll want to tweak or change on the fly? If so, consider adding it to the dynamic config with a default of 12 hours.
| transaction: transaction, | ||
| ); | ||
| final tasks = documents.map(Task.fromDocument).toList(); | ||
| if (cacheResults && transaction == null && cache != null) { |
There was a problem hiding this comment.
I want a dynamic flag that overrides this and keeps the old path. That way we're talking about ~5 minutes to rollout or rollback.
| Transaction? transaction, | ||
| }) async { | ||
| if (transaction == null && cache != null) { | ||
| return await _queryTasksByCommitCached( |
There was a problem hiding this comment.
same here; caching shoudl be flagged in case we need to turn it off - we should set a date on when to remove the flag.
| if (_entries.length >= maxEntries && | ||
| !_entries.containsKey(cacheKey)) { | ||
| _entries.remove(_entries.keys.first); | ||
| } |
There was a problem hiding this comment.
this makes the third copy of this code; move to a new function and update the other two identical versions of this
| _entries[cacheKey] = _InMemoryCacheEntry( | ||
| entry.value, | ||
| DateTime.now().add(entry.ttl), | ||
| revisionId: entry.revisionId, | ||
| ); |
There was a problem hiding this comment.
Do you know if this cache service was trying to do an LRU cache? If so, shouldn't we remove and re-insert when read/writing to the cache so it updates in insertion order?
| 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<String, dynamic>; | ||
| return Task.fromDocument(Document.fromJson(jsonMap)); | ||
| } | ||
|
|
There was a problem hiding this comment.
These are duplicates in firestore.dart.
2 participants
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):
Lock-Free Versioned (revisionId) Payload Caching (tasks subcache):
Native Set-Based Commit Task Indexing (tasks_by_commit_ids):
Partial Cache Recovery (_queryTasksByCommitCached):
Modular & Explicit Cache Handlers (FirestoreQueries):