add temporary self-graph indexing priority

10 files changed, 199 insertions, 6 deletions

diff --git a/API.md b/API.md
index 0117df2..43fb75f 100644
--- a/API.md
+++ b/API.md
@@ -44,6 +44,7 @@ Contribution ranges:
 
 - Contribution read endpoints return zero-filled days, so clients do not need to patch missing dates.
 - Public contribution reads also register the actor for background indexing. A first lookup may therefore return an empty graph while the scheduler catches up.
+- Clients may add `prioritize_self=true` on contribution read endpoints to explicitly request temporary indexing priority for the signed-in user's own graph.
 - Incremental indexing and one-year backfill are separate. An actor can be recently indexed before the retained one-year window is fully filled in.
 - The service only retains and backfills the most recent 365 days of activity.
 
@@ -101,6 +102,7 @@ Query parameters:
 - `year` integer, optional
 - `from` string `YYYY-MM-DD`, optional
 - `to` string `YYYY-MM-DD`, optional
+- `prioritize_self` boolean, optional
 
 Rules:
 
@@ -112,6 +114,7 @@ Behavior notes:
 
 - This endpoint resolves aliases to a canonical actor before querying data.
 - This endpoint also registers the actor for background indexing and updates the actor's `last_requested_at` timestamp.
+- When `prioritize_self=true`, registration also applies a temporary scheduler boost so that due polls for that actor run ahead of the normal due queue.
 - The response is always immediate; it does not wait for SourceHut polling to finish.
 - One-year backfill runs in bounded background batches and may take multiple scheduler passes to complete.
 
@@ -211,10 +214,12 @@ Query parameters:
 - `year` integer, optional
 - `from` string `YYYY-MM-DD`, optional
 - `to` string `YYYY-MM-DD`, optional
+- `prioritize_self` boolean, optional
 
 Behavior notes:
 
 - This endpoint has the same actor-registration and alias-resolution behavior as the calendar endpoint.
+- When `prioritize_self=true`, registration also applies the same temporary scheduler boost as the calendar endpoint.
 - This endpoint returns immediately and does not block on SourceHut polling.
 - This endpoint also reflects whether the retained one-year history window has been fully backfilled yet.
 
diff --git a/README.md b/README.md
index 695d222..03b2cec 100644
--- a/README.md
+++ b/README.md
@@ -175,6 +175,8 @@ Scheduled polling only runs when `ENABLE_SCHEDULER=true`. The scheduler seeds `D
 
 The scheduler now drains actors gradually instead of polling every tracked actor on every pass. It only claims due actors, up to `DISCOVERY_BATCH_SIZE` per run, then reschedules indexed actors with `INDEXED_ACTOR_REPOLL_SECONDS` and failed actors with backoff based on `DISCOVERY_ERROR_BACKOFF_SECONDS`.
 
+Clients can explicitly signal that a public contribution read is for the signed-in user's own graph by sending `prioritize_self=true` on the read request. That temporarily boosts the actor to the front of the due queue for the next indexing pass, then clears the boost after the poll completes.
+
 ## Bulk Enqueue Without Immediate Indexing
 
 To durably queue a large username list without polling it immediately:
diff --git a/alembic/versions/20260412_0007_actor_priority_boost.py b/alembic/versions/20260412_0007_actor_priority_boost.py
new file mode 100644
index 0000000..ecceca6
--- /dev/null
+++ b/alembic/versions/20260412_0007_actor_priority_boost.py
@@ -0,0 +1,29 @@
+"""temporary actor priority boost marker"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy import inspect
+
+
+revision = "20260412_0007"
+down_revision = "20260411_0006"
+branch_labels = None
+depends_on = None
+
+
+def _column_names(table_name: str) -> set[str]:
+    return {column["name"] for column in inspect(op.get_bind()).get_columns(table_name)}
+
+
+def upgrade() -> None:
+    columns = _column_names("tracked_actors")
+    if "priority_boosted_at" not in columns:
+        op.add_column("tracked_actors", sa.Column("priority_boosted_at", sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade() -> None:
+    columns = _column_names("tracked_actors")
+    if "priority_boosted_at" in columns:
+        op.drop_column("tracked_actors", "priority_boosted_at")
diff --git a/src/srht_contrib/api/routes_contributions.py b/src/srht_contrib/api/routes_contributions.py
index 25bc3c2..82b1948 100644
--- a/src/srht_contrib/api/routes_contributions.py
+++ b/src/srht_contrib/api/routes_contributions.py
@@ -41,13 +41,14 @@ def get_contributions(
     year: int | None = Query(default=None, ge=1970, le=3000),
     from_date: str | None = Query(default=None, alias="from"),
     to_date: str | None = Query(default=None, alias="to"),
+    prioritize_self: bool = Query(default=False),
     poller: PollerService = Depends(get_poller),
     db: Session = Depends(get_db),
     actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver),
 ) -> ContributionCalendarResponse:
     start, end = _resolve_range(year, from_date, to_date)
     canonical_actor = actor_identity_resolver.canonicalize(actor, db=db)
-    poller.track_actor_request(db, canonical_actor)
+    poller.track_actor_request(db, canonical_actor, prioritize=prioritize_self)
     response = ContributionAggregator().build_calendar(db, canonical_actor, start, end)
     db.commit()
     return response
@@ -59,13 +60,14 @@ def get_contribution_stats(
     year: int | None = Query(default=None, ge=1970, le=3000),
     from_date: str | None = Query(default=None, alias="from"),
     to_date: str | None = Query(default=None, alias="to"),
+    prioritize_self: bool = Query(default=False),
     poller: PollerService = Depends(get_poller),
     db: Session = Depends(get_db),
     actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver),
 ) -> ContributionStatsResponse:
     start, end = _resolve_range(year, from_date, to_date)
     canonical_actor = actor_identity_resolver.canonicalize(actor, db=db)
-    poller.track_actor_request(db, canonical_actor)
+    poller.track_actor_request(db, canonical_actor, prioritize=prioritize_self)
     response = ContributionAggregator().build_stats(db, canonical_actor, start, end)
     db.commit()
     return response
diff --git a/src/srht_contrib/jobs/poller.py b/src/srht_contrib/jobs/poller.py
index 5e62b55..76d2fb0 100644
--- a/src/srht_contrib/jobs/poller.py
+++ b/src/srht_contrib/jobs/poller.py
@@ -65,6 +65,8 @@ class PollerService:
                 )
             )
             .order_by(
+                TrackedActor.priority_boosted_at.is_not(None).desc(),
+                TrackedActor.priority_boosted_at.desc(),
                 TrackedActor.next_poll_after.is_(None).desc(),
                 TrackedActor.next_poll_after,
                 TrackedActor.queued_for_discovery_at,
@@ -90,7 +92,14 @@ class PollerService:
             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:
+    def track_actor_request(
+        self,
+        db: Session,
+        actor: str,
+        *,
+        update_last_requested: bool = True,
+        prioritize: bool = False,
+    ) -> TrackedActor:
         tracked_actor = db.scalar(select(TrackedActor).where(TrackedActor.actor == actor))
         now = datetime.now(tz=UTC)
         if tracked_actor is None:
@@ -111,6 +120,9 @@ class PollerService:
             tracked_actor.next_poll_after = now
         if update_last_requested:
             tracked_actor.last_requested_at = now
+        if prioritize:
+            tracked_actor.priority_boosted_at = now
+            tracked_actor.next_poll_after = now
         db.flush()
         return tracked_actor
 
@@ -207,11 +219,13 @@ class PollerService:
             tracked_actor.discovery_state = "indexed"
             tracked_actor.last_polled_at = now
             tracked_actor.next_poll_after = now + timedelta(seconds=self.settings.indexed_actor_repoll_seconds)
+            tracked_actor.priority_boosted_at = None
         elif status == "error":
             tracked_actor.discovery_state = "error"
             tracked_actor.next_poll_after = now + timedelta(
                 seconds=self.settings.discovery_error_backoff_seconds * max(tracked_actor.poll_attempts, 1)
             )
+            tracked_actor.priority_boosted_at = None
         db.add(tracked_actor)
         db.flush()
 
diff --git a/src/srht_contrib/models.py b/src/srht_contrib/models.py
index 441b6ec..401e945 100644
--- a/src/srht_contrib/models.py
+++ b/src/srht_contrib/models.py
@@ -69,6 +69,7 @@ class TrackedActor(Base):
     is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
     discovery_state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued")
     queued_for_discovery_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    priority_boosted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     next_poll_after: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     last_claimed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     poll_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
diff --git a/tests/test_contributions_api.py b/tests/test_contributions_api.py
index e568feb..29b19fb 100644
--- a/tests/test_contributions_api.py
+++ b/tests/test_contributions_api.py
@@ -7,6 +7,11 @@ from srht_contrib.main import create_app
 from srht_contrib.models import ContributionEvent, TrackedActor
 
 
+class _Closable:
+    def close(self) -> None:
+        return None
+
+
 def test_read_only_contribution_routes_are_public_and_write_routes_require_api_key(settings, db_engine, session_factory) -> None:
     app = create_app(settings, engine=db_engine, session_factory=session_factory)
     with TestClient(app) as open_client:
@@ -72,6 +77,24 @@ def test_public_read_registers_actor_for_lazy_indexing(client: TestClient, db_se
     assert tracked_actor is not None
     assert tracked_actor.is_active is True
     assert tracked_actor.last_requested_at is not None
+    assert tracked_actor.priority_boosted_at is None
+
+
+class RecordingPriorityPoller:
+    def __init__(self) -> None:
+        service = type("Service", (), {"client": _Closable()})()
+        self.todo_service = service
+        self.git_service = service
+        self.calls: list[tuple[str, bool]] = []
+
+    def track_actor_request(self, db, actor: str, *, update_last_requested: bool = True, prioritize: bool = False):
+        self.calls.append((actor, prioritize))
+
+    def poll_all(self, db, actor: str) -> int:
+        return 0
+
+    def poll_tracked_actors(self, db, default_actor: str | None = None) -> dict[str, int]:
+        return {}
 
 
 def test_contribution_stats_api(client: TestClient, db_session) -> None:
@@ -148,3 +171,25 @@ def test_contribution_routes_use_settings_backed_alias_resolution(settings, db_e
 
     assert response.status_code == 200
     assert response.json()["actor"] == "~ccleberg"
+
+
+def test_contribution_route_passes_explicit_priority_signal(settings, db_engine, session_factory) -> None:
+    poller = RecordingPriorityPoller()
+    app = create_app(settings, engine=db_engine, session_factory=session_factory, poller=poller)
+
+    with TestClient(app) as client:
+        response = client.get("/api/contributions/~ccleberg?from=2026-03-28&to=2026-03-30&prioritize_self=true")
+
+    assert response.status_code == 200
+    assert poller.calls == [("~ccleberg", True)]
+
+
+def test_contribution_stats_route_keeps_non_prioritized_registration_by_default(settings, db_engine, session_factory) -> None:
+    poller = RecordingPriorityPoller()
+    app = create_app(settings, engine=db_engine, session_factory=session_factory, poller=poller)
+
+    with TestClient(app) as client:
+        response = client.get("/api/contributions/~ccleberg/stats?from=2026-03-28&to=2026-03-30")
+
+    assert response.status_code == 200
+    assert poller.calls == [("~ccleberg", False)]
diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py
index 6a1a162..6c06359 100644
--- a/tests/test_ingestion.py
+++ b/tests/test_ingestion.py
@@ -711,6 +711,94 @@ def test_poll_tracked_actors_limits_to_due_batch_size(db_session) -> None:
     assert actors["~c"].poll_attempts == 0
 
 
+def test_track_actor_request_prioritize_marks_actor_boosted_and_due_now(db_session) -> None:
+    settings = make_settings(INDEXED_ACTOR_REPOLL_SECONDS=3600)
+    poller = PollerService(todo_service=RecordingTodoService(events_by_call=[[]]), git_service=EmptyGitService(), settings=settings)
+    future_due = datetime.now(tz=UTC) + timedelta(hours=2)
+    db_session.add(
+        TrackedActor(
+            actor="~self",
+            is_active=True,
+            discovery_state="indexed",
+            queued_for_discovery_at=datetime.now(tz=UTC) - timedelta(hours=1),
+            next_poll_after=future_due,
+            recent_backfill_status="completed",
+        )
+    )
+    db_session.commit()
+
+    tracked_actor = poller.track_actor_request(db_session, "~self", prioritize=True)
+
+    assert tracked_actor.priority_boosted_at is not None
+    assert tracked_actor.next_poll_after is not None
+    assert tracked_actor.next_poll_after <= tracked_actor.priority_boosted_at
+
+
+def test_poll_tracked_actors_prioritizes_boosted_due_actor_first(db_session) -> None:
+    settings = make_settings(DISCOVERY_BATCH_SIZE=1, INDEXED_ACTOR_REPOLL_SECONDS=3600)
+    todo_service = RecordingTodoService(events_by_call=[[]])
+    poller = PollerService(todo_service=todo_service, git_service=EmptyGitService(), settings=settings)
+
+    now = datetime.now(tz=UTC)
+    db_session.add_all(
+        [
+            TrackedActor(
+                actor="~normal",
+                is_active=True,
+                discovery_state="queued",
+                queued_for_discovery_at=now - timedelta(minutes=10),
+                next_poll_after=now - timedelta(minutes=10),
+                recent_backfill_status="completed",
+            ),
+            TrackedActor(
+                actor="~self",
+                is_active=True,
+                discovery_state="queued",
+                queued_for_discovery_at=now - timedelta(minutes=1),
+                next_poll_after=now - timedelta(minutes=1),
+                priority_boosted_at=now,
+                recent_backfill_status="completed",
+            ),
+        ]
+    )
+    db_session.commit()
+
+    results = poller.poll_tracked_actors(db_session)
+
+    assert list(results) == ["~self"]
+    remaining = {
+        actor.actor: actor.discovery_state
+        for actor in db_session.scalars(select(TrackedActor).order_by(TrackedActor.actor)).all()
+    }
+    assert remaining["~self"] == "indexed"
+    assert remaining["~normal"] == "queued"
+
+
+def test_successful_poll_clears_temporary_priority_boost(db_session) -> None:
+    settings = make_settings(INDEXED_ACTOR_REPOLL_SECONDS=3600)
+    poller = PollerService(todo_service=RecordingTodoService(events_by_call=[[]]), git_service=EmptyGitService(), settings=settings)
+    now = datetime.now(tz=UTC)
+    db_session.add(
+        TrackedActor(
+            actor="~self",
+            is_active=True,
+            discovery_state="queued",
+            queued_for_discovery_at=now - timedelta(minutes=1),
+            next_poll_after=now - timedelta(minutes=1),
+            priority_boosted_at=now - timedelta(seconds=30),
+            recent_backfill_status="completed",
+        )
+    )
+    db_session.commit()
+
+    poller.poll_all(db_session, "~self")
+
+    tracked_actor = db_session.scalar(select(TrackedActor).where(TrackedActor.actor == "~self"))
+    assert tracked_actor is not None
+    assert tracked_actor.discovery_state == "indexed"
+    assert tracked_actor.priority_boosted_at is None
+
+
 def test_enqueue_actors_staggers_without_polling(tmp_path, monkeypatch) -> None:
     database_path = tmp_path / "enqueue.db"
     username_path = tmp_path / "srht_usernames.txt"
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index 3ce842b..e646ccf 100644
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@@ -122,7 +122,14 @@ def test_alembic_upgrade_adopts_legacy_schema(tmp_path) -> None:
     assert "discovered_repositories" in inspector.get_table_names()
     assert "tracked_actors" in inspector.get_table_names()
     assert "service_backfill_states" in inspector.get_table_names()
-    assert {"discovery_state", "queued_for_discovery_at", "next_poll_after", "last_claimed_at", "poll_attempts"} <= tracked_actor_columns
+    assert {
+        "discovery_state",
+        "queued_for_discovery_at",
+        "priority_boosted_at",
+        "next_poll_after",
+        "last_claimed_at",
+        "poll_attempts",
+    } <= tracked_actor_columns
 
 
 def test_alembic_prefers_database_url_from_environment(tmp_path, monkeypatch) -> None:
diff --git a/tests/test_polling_api.py b/tests/test_polling_api.py
index b752524..94188da 100644
--- a/tests/test_polling_api.py
+++ b/tests/test_polling_api.py
@@ -21,7 +21,7 @@ class InsertingPoller:
         self.git_service = service
         self.tracked_poll_calls: list[str] = []
 
-    def track_actor_request(self, db, actor: str, *, update_last_requested: bool = True):
+    def track_actor_request(self, db, actor: str, *, update_last_requested: bool = True, prioritize: bool = False):
         tracked_actor = db.scalar(select(TrackedActor).where(TrackedActor.actor == actor))
         if tracked_actor is None:
             tracked_actor = TrackedActor(actor=actor, is_active=True)
@@ -61,7 +61,7 @@ class FailingPoller:
         self.todo_service = service
         self.git_service = service
 
-    def track_actor_request(self, db, actor: str, *, update_last_requested: bool = True):
+    def track_actor_request(self, db, actor: str, *, update_last_requested: bool = True, prioritize: bool = False):
         tracked_actor = db.scalar(select(TrackedActor).where(TrackedActor.actor == actor))
         if tracked_actor is None:
             tracked_actor = TrackedActor(actor=actor, is_active=True)