Durable fetch queue - #418
Open
helq wants to merge 9 commits into
Open
Conversation
This was referenced Aug 11, 2026
Closed
Add a submitted_at field to Event::RawMboxSubmitted, stamped with the server's current time when a raw mbox is submitted via POST /api/submit (Inject variant). This timestamp is propagated through metadata.received_date and used as the patchset date instead of the email's Date: header. This prevents stale mbox timestamps (which can be arbitrarily far in the past) from skewing queue ordering. The original email date is preserved in the messages table for display purposes. The Thread and Remote submission paths are unaffected as they already use server-side timestamps via create_fetching_patchset(). Signed-off-by: Elkin Cruz <elkin@google.com>
Define PriorityRule and CompiledPriorityRule structs for regex-based patchset priority classification. Add a custom serde deserializer (deserialize_indexed_vec) to handle both TOML array and env-var indexed-map representations. Add the priority_rules field to ReviewSettings with serde(default) so existing configs are unaffected. Signed-off-by: Elkin Cruz <elkin@google.com>
Add a priority INTEGER DEFAULT 500 column to the patchsets table and a composite index idx_patchsets_status_priority_date on (status, priority DESC, date ASC) for efficient priority-ordered queries. The column defaults to 500 so existing patchsets are unaffected. Signed-off-by: Elkin Cruz <elkin@google.com>
Add migration to create the priority column and composite index. Introduce create_patchset_with_priority() which threads an explicit priority through all INSERT/UPDATE paths, with MIN(priority, ?) semantics to preserve manual deprioritization. Refactor create_patchset() to delegate with default None. Change get_pending_patchsets() ordering to priority DESC, date ASC. Add calculate_priority() for evaluating compiled regex rules against subjects (last match wins). Signed-off-by: Elkin Cruz <elkin@google.com>
Compile priority_rules from settings at startup and thread them through the DB worker into process_parsed_article(). Use calculate_priority() to compute priority from the patchset subject before calling create_patchset_with_priority(). API callers pass None for priority to create_fetching_patchset(). Signed-off-by: Elkin Cruz <elkin@google.com>
Expose the per-remote async mutex so that the fetcher module can acquire it before running git fetch, preventing concurrent fetches against the same remote (which causes duplicate multi-million-object downloads on large mirrors like the kernel repo). Signed-off-by: Elkin Cruz <elkin@google.com>
Introduce a fetch_queue table plus the DB operations that back the new durable fetch worker, modeled on the existing patchwork_outbox pattern. The table schema and FetchStatus/FetchQueueRow types are accompanied by the full set of queue operations: enqueue_fetch (idempotent), lock_pending_fetch (atomic claim under lease), mark_fetch_done, set_fetch_retry_at (backoff scheduling), mark_fetch_failed (terminal failure), sweep_ghost_fetches (reclaim expired leases), and get_stuck_fetch_placeholders (startup recovery for orphans). Also adds repo_url and priority columns to patchsets for crash recovery of in-flight fetches. This is the storage layer only; wiring lands in a follow-up. Signed-off-by: Elkin Cruz <elkin@google.com>
Add standalone utility functions that the upcoming FetchWorker will use, alongside the existing FetchAgent code: - PermanentFetchError: marker error for failures that should not be retried (e.g. missing commit with no remote to fetch from) - commit_hash_from_placeholder(): parse placeholder message ids (sha@sashiko.local or mr-N-base..head) into git commit/range - now_secs(): current unix timestamp helper - backoff_secs(): exponential backoff calculator (30s base, 10m cap) - looks_like_object_id(): detect full SHA vs ref name for efficient git fetch strategies Each function has unit tests. Signed-off-by: Elkin Cruz <elkin@google.com>
Rework git fetching around the fetch_queue table so requests survive restarts, cancellation, and worker crashes. FetchWorker polls the queue, claims due rows under a lease, ensures the commits are present (with an optimistic early-return when objects are already local, deliberately avoiding a full-remote fetch fallback that was pathologically slow on large mirrors), extracts patches, and emits PatchSubmitted events. Transient failures are retried with exponential backoff (30s doubling to a 10m cap); a fetch is only marked Failed after a continuous 24h retry window is exhausted. A periodic sweep reclaims ghost leases from dead workers. The per-remote lock (get_remote_lock) is acquired before every fetch to prevent concurrent fetches against the same remote, which previously caused duplicate multi-million-object downloads. The API submit paths now persist fetches via enqueue_fetch instead of pushing onto an mpsc channel, and the fetch_sender is removed from AppState/build_router/run_server. On startup, main backfills any patchset stuck in Fetching with no active queue row so in-flight work from before this change is recovered. Signed-off-by: Elkin Cruz <elkin@google.com>
helq
force-pushed
the
upstream/durable-fetcher
branch
from
August 21, 2026 18:37
29cc94e to
8a9c06c
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
On top of #415
Commits 9403451 to 29cc94e.
Replace the in-memory FetchAgent (mpsc channel) with a durable, database-backed FetchWorker so that fetch requests survive restarts, cancellation, and worker crashes.
Inspired on the architecture used for sending emails.
We replace FetchAgent with FetchWorker. The worker polls the queue, claims due rows under a lease, ensures commits are present (with an optimistic early-return when objects are already local), extracts patches, and emits PatchSubmitted events. Transient failures retry with exponential backoff; permanent failure after 24h. API paths now call enqueue_fetch() instead of pushing onto an mpsc channel. On startup, orphaned fetches are backfilled.
Call out: fetches have to be made sequential, thus why we make get_remote_lock() public. Allowing fetches to happen concurrently of each other broke our internal service for days.