From 01c1b50541b0dc1d42cbdaa90052f6b94ceba20c Mon Sep 17 00:00:00 2001 From: Christian Cleberg Date: Sat, 11 Apr 2026 18:45:54 -0500 Subject: feat: prioritize recent-window backfill before full history --- src/srht_contrib/services/aggregator.py | 9 +++++++ src/srht_contrib/services/git.py | 39 +++++++++++++++++++++++------- src/srht_contrib/services/todo.py | 42 +++++++++++++++++++++++++++++---- 3 files changed, 77 insertions(+), 13 deletions(-) (limited to 'src/srht_contrib/services') diff --git a/src/srht_contrib/services/aggregator.py b/src/srht_contrib/services/aggregator.py index 1c8af86..4973f16 100644 --- a/src/srht_contrib/services/aggregator.py +++ b/src/srht_contrib/services/aggregator.py @@ -62,6 +62,12 @@ class ContributionAggregator: is_indexed=calendar.is_indexed, last_polled_at=calendar.last_polled_at, indexing_state=calendar.indexing_state, + is_recent_window_backfilled=calendar.is_recent_window_backfilled, + recent_backfill_state=calendar.recent_backfill_state, + recent_backfill_completed_at=calendar.recent_backfill_completed_at, + is_backfilled=calendar.is_backfilled, + backfill_state=calendar.backfill_state, + backfill_completed_at=calendar.backfill_completed_at, ) def _index_metadata(self, db: Session, actor: str) -> ContributionIndexMetadata: @@ -81,6 +87,9 @@ class ContributionAggregator: is_indexed=is_indexed, last_polled_at=tracked_actor.last_polled_at if tracked_actor is not None else None, indexing_state=indexing_state, + is_recent_window_backfilled=(tracked_actor.recent_backfill_status == "completed") if tracked_actor is not None else False, + recent_backfill_state=(tracked_actor.recent_backfill_status if tracked_actor is not None else "pending"), + recent_backfill_completed_at=tracked_actor.recent_backfill_completed_at if tracked_actor is not None else None, is_backfilled=(tracked_actor.backfill_status == "completed") if tracked_actor is not None else False, backfill_state=(tracked_actor.backfill_status if tracked_actor is not None else "pending"), backfill_completed_at=tracked_actor.backfill_completed_at if tracked_actor is not None else None, diff --git a/src/srht_contrib/services/git.py b/src/srht_contrib/services/git.py index ebb823e..bf7493b 100644 --- a/src/srht_contrib/services/git.py +++ b/src/srht_contrib/services/git.py @@ -120,6 +120,24 @@ class GitIngestionService: return repositories def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult: + return self._fetch_backfill_batch(actor, cursor_state, since=None) + + def fetch_recent_backfill_batch( + self, + actor: str, + cursor_state: dict | None = None, + *, + since: datetime, + ) -> BackfillBatchResult: + return self._fetch_backfill_batch(actor, cursor_state, since=since) + + def _fetch_backfill_batch( + self, + actor: str, + cursor_state: dict | None, + *, + since: datetime | None, + ) -> BackfillBatchResult: state = { "discovery_cursor": None, "discovery_complete": False, @@ -186,13 +204,18 @@ class GitIngestionService: log_page = repository.get("log") or {} commits = log_page.get("results") or [] next_cursor = log_page.get("cursor") - events = [ - normalized - for commit in commits - if isinstance(commit, dict) - for normalized in [self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit)] - if normalized is not None - ] + events: list[NormalizedEvent] = [] + stop_repository = False + for commit in commits: + if not isinstance(commit, dict): + continue + commit_time = parse_datetime((commit.get("author") or {}).get("time")) + if since is not None and commit_time < since: + stop_repository = True + break + normalized = self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit) + if normalized is not None: + events.append(normalized) logger.info( "git backfill actor=%s repository=%s commits=%s next_cursor=%s", actor, @@ -200,7 +223,7 @@ class GitIngestionService: len(commits), bool(next_cursor), ) - if next_cursor: + if next_cursor and not stop_repository: state["current_repository"]["cursor"] = next_cursor else: state["current_repository"] = None diff --git a/src/srht_contrib/services/todo.py b/src/srht_contrib/services/todo.py index b7209b1..9f2ece6 100644 --- a/src/srht_contrib/services/todo.py +++ b/src/srht_contrib/services/todo.py @@ -296,6 +296,24 @@ class TodoIngestionService: return TodoPollResult(events=tracker_events, cursor=cursor_time) def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult: + return self._fetch_backfill_batch(actor, cursor_state, since=None) + + def fetch_recent_backfill_batch( + self, + actor: str, + cursor_state: dict | None = None, + *, + since: datetime, + ) -> BackfillBatchResult: + return self._fetch_backfill_batch(actor, cursor_state, since=since) + + def _fetch_backfill_batch( + self, + actor: str, + cursor_state: dict | None, + *, + since: datetime | None, + ) -> BackfillBatchResult: state = { "trackers_cursor": None, "tracker_queue": [], @@ -349,10 +367,20 @@ class TodoIngestionService: tracker = data.get("tracker") or {} tickets_page = tracker.get("tickets") or {} tickets = tickets_page.get("results") or [] - current_tracker["tickets_cursor"] = tickets_page.get("cursor") - current_tracker["pending_tickets"].extend( - [{"id": int(ticket["id"]), "ref": str(ticket.get("ref") or ticket["id"])} for ticket in tickets if isinstance(ticket, dict)] - ) + next_tickets_cursor = tickets_page.get("cursor") + stop_tracker_paging = False + for ticket in tickets: + if not isinstance(ticket, dict): + continue + if since is not None: + updated = parse_datetime(ticket["updated"]) + if updated < since: + stop_tracker_paging = True + continue + current_tracker["pending_tickets"].append( + {"id": int(ticket["id"]), "ref": str(ticket.get("ref") or ticket["id"])} + ) + current_tracker["tickets_cursor"] = None if stop_tracker_paging else next_tickets_cursor logger.info( "todo backfill tracker=%s ticket page count=%s next_cursor=%s pending_tickets=%s", current_tracker.get("name") or current_tracker.get("id"), @@ -391,6 +419,7 @@ class TodoIngestionService: ) events: list[NormalizedEvent] = [] + stop_ticket_paging = False for event in page_events: if not isinstance(event, dict): continue @@ -402,6 +431,9 @@ class TodoIngestionService: "tracker": {"name": current_tracker.get("name")}, } occurred_at = parse_datetime(event["created"]) + if since is not None and occurred_at < since: + stop_ticket_paging = True + continue for change in event.get("changes") or []: if not isinstance(change, dict): continue @@ -415,7 +447,7 @@ class TodoIngestionService: if normalized is not None: events.append(normalized) - if next_cursor: + if next_cursor and not stop_ticket_paging: state["current_ticket"]["cursor"] = next_cursor else: state["current_ticket"] = None -- cgit v1.2.3