diff --git a/src/main.py b/src/main.py index eb63ff4..ba8a15e 100644 --- a/src/main.py +++ b/src/main.py @@ -1999,60 +1999,63 @@ async def save_today_recap_and_log_time( except (OSError, sqlite3.Error): raise HTTPException(status_code=503, detail="Today recap could not be saved", headers={"Retry-After": "1"}) - results = [] - for item in selected: - try: - state = await asyncio.to_thread( - _today_store().begin_time_log, login, payload.session_id, item.identity, item.actual_minutes - ) - if state == "claimed": - repository, number = gitea_proxy.issue_time_target(item.identity) - journal = _security_event_store() - try: - operation_id = await asyncio.to_thread( - journal.reserve, - "gitea_time_logged", - target=f"{repository}#{number}", - ) - except SecurityEventStoreError: - await asyncio.to_thread( - _today_store().finish_time_log, - login, payload.session_id, item.identity, succeeded=False, - ) - results.append({"identity": item.identity, "status": "retry"}) - continue - try: - await gitea_proxy.log_issue_time(item.identity, item.actual_minutes * 60) - except Exception as error: - if gitea_proxy.time_log_failure_is_retryable(error): + semaphore = asyncio.Semaphore(3) + + async def log_selected_time(item): + async with semaphore: + result = {"identity": item.identity, "status": "retry"} + try: + state = await asyncio.to_thread( + _today_store().begin_time_log, login, payload.session_id, item.identity, item.actual_minutes + ) + if state == "claimed": + repository, number = gitea_proxy.issue_time_target(item.identity) + journal = _security_event_store() + try: + operation_id = await asyncio.to_thread( + journal.reserve, + "gitea_time_logged", + target=f"{repository}#{number}", + ) + except SecurityEventStoreError: await asyncio.to_thread( _today_store().finish_time_log, login, payload.session_id, item.identity, succeeded=False, ) - try: - await asyncio.to_thread(journal.discard, operation_id) - except SecurityEventStoreError: - pass - status = "retry" - else: - status = "verify" - results.append({"identity": item.identity, "status": status}) - continue - await asyncio.to_thread( - _today_store().finish_time_log, - login, payload.session_id, item.identity, succeeded=True, - ) - try: - await asyncio.to_thread(journal.finalize, operation_id) - except SecurityEventStoreError: - pass - state = "logged" - results.append({ - "identity": item.identity, - "status": "logged" if state == "logged" else ("verify" if state == "pending" else "retry"), - }) - except (ValueError, OSError, sqlite3.Error): - results.append({"identity": item.identity, "status": "retry"}) + return result + try: + await gitea_proxy.log_issue_time(item.identity, item.actual_minutes * 60) + except Exception as error: + if gitea_proxy.time_log_failure_is_retryable(error): + await asyncio.to_thread( + _today_store().finish_time_log, + login, payload.session_id, item.identity, succeeded=False, + ) + try: + await asyncio.to_thread(journal.discard, operation_id) + except SecurityEventStoreError: + pass + status = "retry" + else: + status = "verify" + return {"identity": item.identity, "status": status} + await asyncio.to_thread( + _today_store().finish_time_log, + login, payload.session_id, item.identity, succeeded=True, + ) + try: + await asyncio.to_thread(journal.finalize, operation_id) + except SecurityEventStoreError: + pass + state = "logged" + return { + "identity": item.identity, + "status": "logged" if state == "logged" else ("verify" if state == "pending" else "retry"), + } + except (ValueError, OSError, sqlite3.Error): + return result + + results = await asyncio.gather(*(log_selected_time(item) for item in selected)) response.headers["Cache-Control"] = "no-store" return {**recap, "time_logs": results} diff --git a/tests/test_today_recap.py b/tests/test_today_recap.py index b9bcee9..435a456 100644 --- a/tests/test_today_recap.py +++ b/tests/test_today_recap.py @@ -1,5 +1,7 @@ +import asyncio import json import subprocess +import time from pathlib import Path import httpx @@ -250,15 +252,76 @@ async def test_recap_time_logging_journals_confirmed_ambiguous_and_retryable_out {"identity": "issue:stackchain/dashboard:591:", "status": "retry"}, ] time_events = [event for event in events if event["kind"] == "gitea_time_logged"] - assert [(event["target"], event["status"]) for event in time_events] == [ + assert {(event["target"], event["status"]) for event in time_events} == { ("stackchain/dashboard#590", "pending"), ("stackchain/dashboard#589", "completed"), - ] + } persisted = security_db.read_bytes() assert b"private-recap-session" not in persisted assert challenge.json()["detail"]["target"].encode() not in persisted +@pytest.mark.anyio +async def test_recap_time_logging_attempts_multiple_entries_with_bounded_concurrency( + monkeypatch, tmp_path +): + monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator") + monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple") + monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy") + monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3")) + monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login.sqlite3")) + monkeypatch.setenv("STACKCHAIN_TODAY_DB", str(tmp_path / "today.sqlite3")) + monkeypatch.setenv("STACKCHAIN_SECURITY_EVENT_DB", str(tmp_path / "security.sqlite3")) + active = 0 + peak = 0 + attempted = [] + + async def user(): + return {"id": 1, "login": "Timmy"} + + async def log_time(identity, _seconds): + nonlocal active, peak + active += 1 + peak = max(peak, active) + attempted.append(identity) + try: + await asyncio.sleep(0.1) + finally: + active -= 1 + + monkeypatch.setattr(main, "current_user", user) + monkeypatch.setattr(main.gitea_proxy, "log_issue_time", log_time) + identities = [f"issue:stackchain/dashboard:{number}:" for number in range(607, 612)] + payload = { + "session_id": "concurrent-recap-session", + "items": [ + {"identity": identity, "estimate_minutes": 10, "actual_minutes": 11} + for identity in identities + ], + "log_identities": identities, + } + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="https://test") as client: + await client.post("/api/v1/session", json={"access_token": "correct horse battery staple"}) + headers = {"Origin": "https://test", "X-CSRF-Token": client.cookies["stackchain_csrf"]} + grant = await recap_time_grant(client, headers, payload) + started = time.perf_counter() + response = await client.post( + "/api/v1/today/recaps/log-time", + json=payload, + headers={**headers, "X-Step-Up-Grant": grant}, + ) + elapsed = time.perf_counter() - started + + assert response.status_code == 200 + assert elapsed < 0.35 + assert 1 < peak <= 3 + assert set(attempted) == set(identities) + assert response.json()["time_logs"] == [ + {"identity": identity, "status": "logged"} for identity in identities + ] + + @pytest.mark.anyio async def test_explicit_recap_time_logging_posts_corrected_seconds_once(monkeypatch, tmp_path): monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")