diff options
| -rw-r--r-- | API.md | 34 | ||||
| -rw-r--r-- | README.md | 11 | ||||
| -rw-r--r-- | src/srht_contrib/jobs/poller.py | 52 | ||||
| -rw-r--r-- | src/srht_contrib/schemas.py | 3 | ||||
| -rw-r--r-- | src/srht_contrib/services/aggregator.py | 6 | ||||
| -rw-r--r-- | src/srht_contrib/utils/retention.py | 17 | ||||
| -rw-r--r-- | tests/test_contributions_api.py | 4 | ||||
| -rw-r--r-- | tests/test_ingestion.py | 74 |
8 files changed, 107 insertions, 94 deletions
@@ -140,9 +140,6 @@ Response `200 OK`: "is_recent_window_backfilled": false, "recent_backfill_state": "in_progress", "recent_backfill_completed_at": null, - "is_backfilled": false, - "backfill_state": "in_progress", - "backfill_completed_at": null, "days": [ { "date": "2026-03-01", "count": 0, "score": 0.0 }, { "date": "2026-03-02", "count": 3, "score": 2.5 } @@ -158,12 +155,9 @@ Response fields: - `is_indexed` boolean: whether the service has already completed at least one successful recent/incremental poll for this actor - `last_polled_at` string or `null`: most recent successful poll time, if any - `indexing_state` string: one of `pending`, `indexed`, or `error` -- `is_recent_window_backfilled` boolean: whether the prioritized recent history window has completed backfill +- `is_recent_window_backfilled` boolean: whether the service has finished filling the retained one-year history window - `recent_backfill_state` string: one of `pending`, `in_progress`, `completed`, or `error` -- `recent_backfill_completed_at` string or `null`: when recent-window backfill completed, if it has -- `is_backfilled` boolean: whether historical backfill has completed for this actor -- `backfill_state` string: one of `pending`, `in_progress`, `completed`, or `error` -- `backfill_completed_at` string or `null`: when full historical backfill completed, if it has +- `recent_backfill_completed_at` string or `null`: when one-year backfill completed, if it has - `days` array: - `date` string `YYYY-MM-DD` - `count` integer contribution count for the day @@ -175,15 +169,19 @@ Indexing state semantics: - `indexed`: at least one successful poll has completed for the actor - `error`: the most recent poll attempt for the actor failed -Backfill state semantics: +Recent backfill semantics: -- recent-window fields: - - represent the prioritized visible-history window for client UX -- `pending`: the actor has not started historical backfill yet -- `in_progress`: historical backfill is actively progressing in bounded background batches -- `completed`: historical backfill has completed for all supported services +- the service only retains and backfills the most recent 365 days of activity +- `pending`: the actor has not started one-year backfill yet +- `in_progress`: one-year backfill is actively progressing in bounded background batches +- `completed`: the retained one-year window is fully backfilled - `error`: the most recent backfill attempt failed +Retention notes: + +- activity older than 365 days is not retained +- scheduled polling periodically prunes contribution rows older than the retained window + Possible errors: - `400 Bad Request` for invalid or conflicting date input @@ -218,7 +216,7 @@ Behavior notes: - This endpoint has the same actor-registration and alias-resolution behavior as the calendar endpoint. - This endpoint returns immediately and does not block on SourceHut polling. -- This endpoint also reflects historical backfill state so clients can distinguish recent indexing from complete history. +- This endpoint also reflects whether the retained one-year history window has been fully backfilled yet. Example: @@ -239,9 +237,6 @@ Response `200 OK`: "is_recent_window_backfilled": true, "recent_backfill_state": "completed", "recent_backfill_completed_at": "2026-04-11T18:02:00Z", - "is_backfilled": false, - "backfill_state": "in_progress", - "backfill_completed_at": null, "total_events": 126, "total_score": 116.75, "active_days": 14, @@ -261,9 +256,6 @@ Response fields: - `is_recent_window_backfilled` boolean - `recent_backfill_state` string - `recent_backfill_completed_at` string or `null` -- `is_backfilled` boolean -- `backfill_state` string -- `backfill_completed_at` string or `null` - `total_events` integer - `total_score` float - `active_days` integer @@ -211,9 +211,6 @@ Example response: "is_recent_window_backfilled": true, "recent_backfill_state": "completed", "recent_backfill_completed_at": "2026-04-11T18:02:00Z", - "is_backfilled": false, - "backfill_state": "in_progress", - "backfill_completed_at": null, "days": [ {"date": "2026-03-28", "count": 3, "score": 3.5}, {"date": "2026-03-29", "count": 0, "score": 0.0}, @@ -241,9 +238,6 @@ Example response: "is_recent_window_backfilled": true, "recent_backfill_state": "completed", "recent_backfill_completed_at": "2026-04-11T18:02:00Z", - "is_backfilled": false, - "backfill_state": "in_progress", - "backfill_completed_at": null, "total_events": 42, "total_score": 37.5, "active_days": 18, @@ -326,14 +320,15 @@ The SourceHut-specific assumptions are isolated to the service modules: - `src/srht_contrib/services/todo.py` uses the authenticated `events(cursor)` feed first, then falls back to tracker/ticket event traversal for reliable contribution discovery. - `src/srht_contrib/services/git.py` discovers owned repositories for an actor, polls each repository `log(cursor)`, and attributes commits through the configured alias map. +- the service only retains the most recent 365 days of contribution history and periodically prunes older rows ## Known Limitations - `git.sr.ht` polling assumes the actor's repositories are discoverable through the SourceHut GraphQL API - scheduled polling runs in-process, so it is not a distributed scheduler - newly requested actors are indexed asynchronously, so the first public read may be empty until a scheduler or manual poll runs -- the recent visible-history window is prioritized first, but deep-history backfill can still take many scheduler passes for active users -- full historical backfill can take many scheduler passes for active users because it runs in bounded batches +- the service is intentionally limited to a rolling one-year history window; older activity is not retained +- one-year backfill still runs in bounded batches, so a newly requested actor may take multiple scheduler passes before their visible graph is fully filled in - alias management is config-driven; there is no alias CRUD API yet - current deployment model is trusted-operator V1, not a public multi-tenant service diff --git a/src/srht_contrib/jobs/poller.py b/src/srht_contrib/jobs/poller.py index c41f3d4..ff0aaae 100644 --- a/src/srht_contrib/jobs/poller.py +++ b/src/srht_contrib/jobs/poller.py @@ -13,14 +13,13 @@ from srht_contrib.schemas import NormalizedEvent from srht_contrib.services.git import GitIngestionService from srht_contrib.services.srht_client import SourceHutClientError from srht_contrib.services.todo import TodoIngestionService +from srht_contrib.utils.retention import RETENTION_DAYS, prune_contribution_events from srht_contrib.utils.repositories import canonicalize_repository_name logger = logging.getLogger(__name__) SYNC_OVERLAP = timedelta(hours=24) -RECENT_BACKFILL_DAYS = 365 RECENT_BACKFILL_BATCHES_PER_SERVICE = 5 -FULL_BACKFILL_BATCHES_PER_SERVICE = 1 class PollerService: @@ -61,6 +60,9 @@ class PollerService: logger.exception("Scheduled poll failed for actor=%s", actor) except Exception: logger.exception("Unexpected scheduled poll failure for actor=%s", actor) + deleted = self.prune_old_events(db) + if deleted: + logger.info("Pruned %s contribution events older than %s days", deleted, RETENTION_DAYS) return results def track_actor_request(self, db: Session, actor: str, *, update_last_requested: bool = True) -> TrackedActor: @@ -70,7 +72,6 @@ class PollerService: actor=actor, is_active=True, recent_backfill_status="pending", - backfill_status="pending", ) db.add(tracked_actor) @@ -185,11 +186,11 @@ class PollerService: def _run_backfill_batches(self, db: Session, actor: str) -> int: tracked_actor = self.track_actor_request(db, actor, update_last_requested=False) - if tracked_actor.recent_backfill_status == "completed" and tracked_actor.backfill_status == "completed": + if tracked_actor.recent_backfill_status == "completed": return 0 now = datetime.now(tz=UTC) - recent_since = now - timedelta(days=RECENT_BACKFILL_DAYS) + recent_since = now - timedelta(days=RETENTION_DAYS) db.add(tracked_actor) db.flush() total_inserted = 0 @@ -223,34 +224,6 @@ class PollerService: db.add(tracked_actor) db.flush() - if tracked_actor.recent_backfill_status == "completed" and tracked_actor.backfill_status != "completed": - if tracked_actor.backfill_started_at is None: - tracked_actor.backfill_started_at = now - tracked_actor.backfill_status = "in_progress" - tracked_actor.last_backfill_error = None - total_inserted += self._run_backfill_scope( - db, - actor=actor, - scope="full", - services=[ - (self.todo_service.service_name, self.todo_service.fetch_backfill_batch), - (self.git_service.service_name, self.git_service.fetch_backfill_batch), - ], - batches_per_service=FULL_BACKFILL_BATCHES_PER_SERVICE, - since=None, - ) - full_statuses = db.scalars( - select(ServiceBackfillState.status) - .where(ServiceBackfillState.actor == actor) - .where(ServiceBackfillState.scope == "full") - ).all() - if full_statuses and all(status == "completed" for status in full_statuses): - tracked_actor.backfill_status = "completed" - tracked_actor.backfill_completed_at = datetime.now(tz=UTC) - tracked_actor.last_backfill_error = None - db.add(tracked_actor) - db.flush() - return total_inserted def _run_backfill_scope( @@ -328,12 +301,8 @@ class PollerService: state.status = "error" state.last_error = str(exc) state.updated_at = datetime.now(tz=UTC) - if scope == "recent": - tracked_actor.recent_backfill_status = "error" - tracked_actor.last_recent_backfill_error = str(exc) - else: - tracked_actor.backfill_status = "error" - tracked_actor.last_backfill_error = str(exc) + tracked_actor.recent_backfill_status = "error" + tracked_actor.last_recent_backfill_error = str(exc) db.add(state) db.add(tracked_actor) db.flush() @@ -343,3 +312,8 @@ class PollerService: db.flush() return total_inserted + + def prune_old_events(self, db: Session) -> int: + deleted = prune_contribution_events(db) + db.commit() + return deleted diff --git a/src/srht_contrib/schemas.py b/src/srht_contrib/schemas.py index 98d2a74..273cb26 100644 --- a/src/srht_contrib/schemas.py +++ b/src/srht_contrib/schemas.py @@ -31,9 +31,6 @@ class ContributionIndexMetadata(BaseModel): is_recent_window_backfilled: bool = False recent_backfill_state: Literal["pending", "in_progress", "completed", "error"] = "pending" recent_backfill_completed_at: datetime | None = None - is_backfilled: bool = False - backfill_state: Literal["pending", "in_progress", "completed", "error"] = "pending" - backfill_completed_at: datetime | None = None class ContributionCalendarResponse(ContributionIndexMetadata): diff --git a/src/srht_contrib/services/aggregator.py b/src/srht_contrib/services/aggregator.py index 4973f16..32baebd 100644 --- a/src/srht_contrib/services/aggregator.py +++ b/src/srht_contrib/services/aggregator.py @@ -65,9 +65,6 @@ class ContributionAggregator: 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: @@ -90,9 +87,6 @@ class ContributionAggregator: 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, ) def _query_daily_aggregates(self, db: Session, actor: str, start: date, end: date) -> list[DailyAggregate]: diff --git a/src/srht_contrib/utils/retention.py b/src/srht_contrib/utils/retention.py new file mode 100644 index 0000000..5018b32 --- /dev/null +++ b/src/srht_contrib/utils/retention.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +from sqlalchemy import delete +from sqlalchemy.orm import Session + +from srht_contrib.models import ContributionEvent + + +RETENTION_DAYS = 365 + + +def prune_contribution_events(db: Session, *, now: datetime | None = None) -> int: + cutoff = (now or datetime.now(tz=UTC)) - timedelta(days=RETENTION_DAYS) + result = db.execute(delete(ContributionEvent).where(ContributionEvent.occurred_at < cutoff)) + return int(result.rowcount or 0) diff --git a/tests/test_contributions_api.py b/tests/test_contributions_api.py index 15af7d8..e568feb 100644 --- a/tests/test_contributions_api.py +++ b/tests/test_contributions_api.py @@ -68,9 +68,6 @@ def test_public_read_registers_actor_for_lazy_indexing(client: TestClient, db_se assert response.json()["is_recent_window_backfilled"] is False assert response.json()["recent_backfill_state"] == "pending" assert response.json()["recent_backfill_completed_at"] is None - assert response.json()["is_backfilled"] is False - assert response.json()["backfill_state"] == "pending" - assert response.json()["backfill_completed_at"] is None assert response.json()["last_polled_at"] is None assert tracked_actor is not None assert tracked_actor.is_active is True @@ -117,7 +114,6 @@ def test_contribution_stats_api(client: TestClient, db_session) -> None: assert response.json()["indexing_state"] == "indexed" assert response.json()["is_recent_window_backfilled"] is False assert response.json()["recent_backfill_state"] == "pending" - assert response.json()["is_backfilled"] is False def test_invalid_date_input_returns_400(client: TestClient) -> None: diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py index 75e3710..aa79e3c 100644 --- a/tests/test_ingestion.py +++ b/tests/test_ingestion.py @@ -4,7 +4,7 @@ from sqlalchemy import select from srht_contrib.config import Settings from srht_contrib.jobs.poller import PollerService -from srht_contrib.models import ServiceBackfillState, SyncState, TrackedActor, TrackedRepository +from srht_contrib.models import ContributionEvent, ServiceBackfillState, SyncState, TrackedActor, TrackedRepository from srht_contrib.schemas import NormalizedEvent from srht_contrib.services.git import GitIngestionService, GitPollResult from srht_contrib.services.todo import TodoIngestionService, TodoPollResult @@ -94,7 +94,7 @@ class BackfillingTodoService: repo_name="todo", resource_id="backfill-ticket", external_uid=f"todo:backfill:{actor}", - occurred_at=datetime(2024, 1, 1, 12, 0, tzinfo=UTC), + occurred_at=datetime(2026, 1, 1, 12, 0, tzinfo=UTC), weight=1.0, raw_payload_json=None, ) @@ -119,7 +119,13 @@ class QueueShrinkingTodoService: def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult: import copy - state = {"tracker_queue": ["t1", "t2"], "current_tracker": None, "current_ticket": None, "trackers_loaded": True, "trackers_cursor": None} + state = { + "tracker_queue": ["t1", "t2", "t3", "t4", "t5", "t6"], + "current_tracker": None, + "current_ticket": None, + "trackers_loaded": True, + "trackers_cursor": None, + } if cursor_state: state.update(copy.deepcopy(cursor_state)) if not state["tracker_queue"]: @@ -156,7 +162,12 @@ class QueueShrinkingGitService: def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult: import copy - state = {"repository_queue": ["r1", "r2"], "current_repository": None, "discovery_complete": True, "discovery_cursor": None} + state = { + "repository_queue": ["r1", "r2", "r3", "r4", "r5", "r6"], + "current_repository": None, + "discovery_complete": True, + "discovery_cursor": None, + } if cursor_state: state.update(copy.deepcopy(cursor_state)) if not state["repository_queue"]: @@ -561,9 +572,7 @@ def test_poll_marks_backfill_complete_and_persists_service_state(db_session) -> assert tracked_actor is not None assert tracked_actor.recent_backfill_status == "completed" assert tracked_actor.recent_backfill_completed_at is not None - assert tracked_actor.backfill_status == "completed" - assert tracked_actor.backfill_completed_at is not None - assert [f"{state.scope}:{state.service}" for state in service_states] == ["full:git", "full:todo", "recent:git", "recent:todo"] + assert [f"{state.scope}:{state.service}" for state in service_states] == ["recent:git", "recent:todo"] assert all(state.status == "completed" for state in service_states) @@ -576,7 +585,7 @@ def test_backfill_cursor_state_shrinks_across_repeated_polls(db_session) -> None for state in db_session.scalars( select(ServiceBackfillState) .where(ServiceBackfillState.actor == "~ccleberg") - .where(ServiceBackfillState.scope == "full") + .where(ServiceBackfillState.scope == "recent") ).all() } @@ -586,11 +595,50 @@ def test_backfill_cursor_state_shrinks_across_repeated_polls(db_session) -> None for state in db_session.scalars( select(ServiceBackfillState) .where(ServiceBackfillState.actor == "~ccleberg") - .where(ServiceBackfillState.scope == "full") + .where(ServiceBackfillState.scope == "recent") ).all() } - assert first_states["git"]["repository_queue"] == ["r2"] - assert first_states["todo"]["tracker_queue"] == ["t2"] - assert second_states["git"]["repository_queue"] == [] - assert second_states["todo"]["tracker_queue"] == [] + assert first_states["git"]["repository_queue"] == ["r6"] + assert first_states["todo"]["tracker_queue"] == ["t6"] + assert second_states["git"] is None + assert second_states["todo"] is None + + +def test_prune_old_events_removes_data_older_than_one_year(db_session) -> None: + poller = PollerService(todo_service=BackfillingTodoService(), git_service=EmptyGitService()) + db_session.add_all( + [ + ContributionEvent( + service="todo", + event_type="ticket_created", + actor="~ccleberg", + repo_name="todo", + resource_id="old", + external_uid="todo:old", + occurred_at=datetime(2025, 1, 1, 12, 0, tzinfo=UTC), + weight=1.0, + raw_payload_json=None, + ), + ContributionEvent( + service="todo", + event_type="ticket_created", + actor="~ccleberg", + repo_name="todo", + resource_id="recent", + external_uid="todo:recent", + occurred_at=datetime(2026, 4, 1, 12, 0, tzinfo=UTC), + weight=1.0, + raw_payload_json=None, + ), + ] + ) + db_session.commit() + + deleted = poller.prune_old_events(db_session) + remaining = db_session.scalars( + select(ContributionEvent.external_uid).order_by(ContributionEvent.external_uid) + ).all() + + assert deleted == 1 + assert remaining == ["todo:recent"] |