Log Today recap time concurrently within the foreground deadline #608
103
src/main.py
103
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}
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user