diff options
| -rw-r--r-- | src/srht_contrib/jobs/poller.py | 3 | ||||
| -rw-r--r-- | src/srht_contrib/services/git.py | 3 | ||||
| -rw-r--r-- | src/srht_contrib/services/todo.py | 3 | ||||
| -rw-r--r-- | tests/test_ingestion.py | 71 |
4 files changed, 77 insertions, 3 deletions
diff --git a/src/srht_contrib/jobs/poller.py b/src/srht_contrib/jobs/poller.py index 2b00090..45992e4 100644 --- a/src/srht_contrib/jobs/poller.py +++ b/src/srht_contrib/jobs/poller.py @@ -1,5 +1,6 @@ from __future__ import annotations +import copy import logging from datetime import UTC, datetime, timedelta @@ -224,7 +225,7 @@ class PollerService: result = fetcher(actor=actor, cursor_state=state.cursor_json) inserted = self._insert_events(db, result.events) total_inserted += inserted - state.cursor_json = result.cursor_state + state.cursor_json = copy.deepcopy(result.cursor_state) state.last_error = None state.updated_at = datetime.now(tz=UTC) if result.complete: diff --git a/src/srht_contrib/services/git.py b/src/srht_contrib/services/git.py index 69a6c1c..ebb823e 100644 --- a/src/srht_contrib/services/git.py +++ b/src/srht_contrib/services/git.py @@ -1,5 +1,6 @@ from __future__ import annotations +import copy from dataclasses import dataclass from datetime import UTC, datetime, timedelta import logging @@ -131,7 +132,7 @@ class GitIngestionService: "current_repository": None, } if cursor_state: - state.update(cursor_state) + state.update(copy.deepcopy(cursor_state)) if not state["discovery_complete"]: data = self.client.execute( diff --git a/src/srht_contrib/services/todo.py b/src/srht_contrib/services/todo.py index 0142005..b7209b1 100644 --- a/src/srht_contrib/services/todo.py +++ b/src/srht_contrib/services/todo.py @@ -1,5 +1,6 @@ from __future__ import annotations +import copy from dataclasses import dataclass from datetime import UTC, datetime, timedelta import logging @@ -303,7 +304,7 @@ class TodoIngestionService: "trackers_loaded": False, } if cursor_state: - state.update(cursor_state) + state.update(copy.deepcopy(cursor_state)) if not state["trackers_loaded"]: data = self.client.execute(TODO_TRACKERS_QUERY, {"cursor": state["trackers_cursor"]}) diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py index 857f1aa..0cb2abc 100644 --- a/tests/test_ingestion.py +++ b/tests/test_ingestion.py @@ -83,6 +83,52 @@ class BackfillingTodoService: return BackfillBatchResult(events=[event], cursor_state=None, complete=True) +class QueueShrinkingTodoService: + service_name = "todo" + + def fetch_recent_events(self, actor: str, since: datetime | None = None) -> TodoPollResult: + return TodoPollResult(events=[], cursor=datetime(2026, 3, 31, tzinfo=UTC).isoformat()) + + 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} + if cursor_state: + state.update(copy.deepcopy(cursor_state)) + if not state["tracker_queue"]: + return BackfillBatchResult(events=[], cursor_state=None, complete=True) + state["tracker_queue"].pop(0) + return BackfillBatchResult(events=[], cursor_state=state, complete=False) + + +class QueueShrinkingGitService: + service_name = "git" + + def __init__(self) -> None: + self.settings = Settings( + SRHT_TOKEN="x", + DATABASE_URL="sqlite://", + DEFAULT_ACTOR="~ccleberg", + TODO_SRHT_ENDPOINT="https://todo.sr.ht/query", + GIT_SRHT_ENDPOINT="https://git.sr.ht/query", + POLL_INTERVAL_SECONDS=60, + ) + + def fetch_recent_events(self, actor: str, since: datetime | None = None, repositories=None) -> GitPollResult: + return GitPollResult(events=[], cursor=datetime(2026, 3, 31, tzinfo=UTC).isoformat()) + + 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} + if cursor_state: + state.update(copy.deepcopy(cursor_state)) + if not state["repository_queue"]: + return BackfillBatchResult(events=[], cursor_state=None, complete=True) + state["repository_queue"].pop(0) + return BackfillBatchResult(events=[], cursor_state=state, complete=False) + + def make_settings(**overrides) -> Settings: values = { "API_KEY": "test-api-key", @@ -470,3 +516,28 @@ def test_poll_marks_backfill_complete_and_persists_service_state(db_session) -> assert tracked_actor.backfill_completed_at is not None assert [state.service for state in service_states] == ["git", "todo"] assert all(state.status == "completed" for state in service_states) + + +def test_backfill_cursor_state_shrinks_across_repeated_polls(db_session) -> None: + poller = PollerService(todo_service=QueueShrinkingTodoService(), git_service=QueueShrinkingGitService()) + + poller.poll_all(db_session, "~ccleberg") + first_states = { + state.service: state.cursor_json + for state in db_session.scalars( + select(ServiceBackfillState).where(ServiceBackfillState.actor == "~ccleberg") + ).all() + } + + poller.poll_all(db_session, "~ccleberg") + second_states = { + state.service: state.cursor_json + for state in db_session.scalars( + select(ServiceBackfillState).where(ServiceBackfillState.actor == "~ccleberg") + ).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"] == [] |