mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
[chore] logical-gaps: release approve async dispatch (202) — kill the 40min synchronous HTTP 504
The approve route ran the whole fail-closed execute inline: clone(600s) + gate(1800s) + CI poll(2400s) + publish(300s) ≈ up to 85min worst case. nginx (the single :3000 entry point, ~60s read timeout) 504'd long before it finished, so the CEO's approve always appeared to fail even when the release succeeded server-side — the structured ReleaseResult was unreachable over the wire. dispatch_approve spawns the execute in a background task with a fresh session (built from the request session's engine) and the route returns 202 'accepted' immediately; _INFLIGHT_APPROVES tracks the dispatched task for observability (self-cleans via done-callback; the Redis mutex still refuses a double-execute on a second click). The panel already polls GET /proposal every 30s, so it observes the final status (COMPLETED on published/already_published, else the proposal stays open for retry); the card's approve toast now treats 'accepted' as an info 'dispatched, running in the background' instead of the old 'Release halted' warning. TDD: 2 route tests red→green (approve returns 202 'accepted' + the proposal transitions to COMPLETED / stays PENDING once the background faked execute completes; the dispatched task is awaited while the executor patch is live). 83 release tests green; ruff/mypy clean; panel typecheck+lint+format+test green.
This commit is contained in:
@@ -65,6 +65,14 @@ export function ReleaseProposalCard({ className }: { className?: string }) {
|
|||||||
`Published v${result.version}` +
|
`Published v${result.version}` +
|
||||||
(result.release_url ? "" : " (no release URL returned)"),
|
(result.release_url ? "" : " (no release URL returned)"),
|
||||||
);
|
);
|
||||||
|
} else if (result.status === "accepted") {
|
||||||
|
// The execute runs in the background (a synchronous request would 504
|
||||||
|
// at nginx before the ~40min fail-closed gate/CI/publish finished).
|
||||||
|
// This card polls GET /proposal every 30s and reflects the final
|
||||||
|
// outcome (COMPLETED on a publish, else the proposal stays open).
|
||||||
|
toast.info(
|
||||||
|
"Release execute dispatched — running in the background. This card updates as it progresses.",
|
||||||
|
);
|
||||||
} else {
|
} else {
|
||||||
toast.warning(`Release halted (${result.status}): ${result.detail}`);
|
toast.warning(`Release halted (${result.status}): ${result.detail}`);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ keeps the proposal held. Nothing here publishes without the CEO's explicit POST.
|
|||||||
from typing import TYPE_CHECKING, cast
|
from typing import TYPE_CHECKING, cast
|
||||||
|
|
||||||
from fastapi import APIRouter, HTTPException, status
|
from fastapi import APIRouter, HTTPException, status
|
||||||
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||||
|
|
||||||
from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
|
from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
|
||||||
from roboco.api.schemas.release import (
|
from roboco.api.schemas.release import (
|
||||||
@@ -18,7 +19,10 @@ from roboco.api.schemas.release import (
|
|||||||
ReleaseReportModel,
|
ReleaseReportModel,
|
||||||
)
|
)
|
||||||
from roboco.foundation.policy.content import markers
|
from roboco.foundation.policy.content import markers
|
||||||
from roboco.services.release_proposal import get_release_proposal_service
|
from roboco.services.release_proposal import (
|
||||||
|
dispatch_approve,
|
||||||
|
get_release_proposal_service,
|
||||||
|
)
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
@@ -71,11 +75,25 @@ async def get_release_proposal(
|
|||||||
return _to_response(task)
|
return _to_response(task)
|
||||||
|
|
||||||
|
|
||||||
@router.post("/proposal/approve", response_model=ReleaseExecuteResponse)
|
@router.post(
|
||||||
|
"/proposal/approve",
|
||||||
|
response_model=ReleaseExecuteResponse,
|
||||||
|
status_code=status.HTTP_202_ACCEPTED,
|
||||||
|
)
|
||||||
async def approve_release_proposal(
|
async def approve_release_proposal(
|
||||||
db: DbSession, agent: CurrentAgentContext
|
db: DbSession, agent: CurrentAgentContext
|
||||||
) -> ReleaseExecuteResponse:
|
) -> ReleaseExecuteResponse:
|
||||||
"""Approve the held proposal → run the fail-closed executor."""
|
"""Approve the held proposal → dispatch the fail-closed executor async.
|
||||||
|
|
||||||
|
The execute is a ~40min clone→gate→CI→publish pipeline; running it inline
|
||||||
|
would 504 at nginx (the single :3000 entry point, ~60s read timeout) before
|
||||||
|
it finished, so the CEO's approve always appeared to fail even when the
|
||||||
|
release succeeded server-side. The route dispatches the execute in a
|
||||||
|
background task with a fresh session and returns 202 immediately; the panel
|
||||||
|
polls ``GET /proposal`` to observe the final status (COMPLETED on
|
||||||
|
published/already_published, else the proposal stays open for retry). A
|
||||||
|
second click is refused by the Redis mutex (``already_in_progress``).
|
||||||
|
"""
|
||||||
_require_ceo(agent)
|
_require_ceo(agent)
|
||||||
svc = get_release_proposal_service(db)
|
svc = get_release_proposal_service(db)
|
||||||
task = await svc.open_proposal()
|
task = await svc.open_proposal()
|
||||||
@@ -83,20 +101,24 @@ async def approve_release_proposal(
|
|||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_404_NOT_FOUND, detail="No open release proposal"
|
status_code=status.HTTP_404_NOT_FOUND, detail="No open release proposal"
|
||||||
)
|
)
|
||||||
result = await svc.approve(cast("UUID", task.id))
|
# Materialize the proposal for the background session (a no-op in prod,
|
||||||
if result is None:
|
# where the release-manager engine already committed it; tests seed it only
|
||||||
raise HTTPException(
|
# flushed into the request session).
|
||||||
status_code=status.HTTP_400_BAD_REQUEST,
|
|
||||||
detail="Proposal has no stored readiness report",
|
|
||||||
)
|
|
||||||
await db.commit()
|
await db.commit()
|
||||||
|
factory = async_sessionmaker(
|
||||||
|
bind=db.bind, class_=AsyncSession, expire_on_commit=False, autoflush=False
|
||||||
|
)
|
||||||
|
dispatch_approve(cast("UUID", task.id), factory)
|
||||||
return ReleaseExecuteResponse(
|
return ReleaseExecuteResponse(
|
||||||
status=result.status,
|
status="accepted",
|
||||||
version=result.version,
|
version="",
|
||||||
files_changed=result.files_changed,
|
files_changed=[],
|
||||||
commit_sha=result.commit_sha,
|
commit_sha=None,
|
||||||
release_url=result.release_url,
|
release_url=None,
|
||||||
detail=result.detail,
|
detail=(
|
||||||
|
"Release execute dispatched in the background; poll"
|
||||||
|
" /api/release/proposal for the final status."
|
||||||
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -29,7 +29,7 @@ from roboco.services.task import RELEASE_MANAGER_SOURCE, get_task_service
|
|||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
|
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||||
|
|
||||||
from roboco.db.tables import TaskTable
|
from roboco.db.tables import TaskTable
|
||||||
|
|
||||||
@@ -290,3 +290,47 @@ class ReleaseProposalService(BaseService):
|
|||||||
def get_release_proposal_service(session: AsyncSession) -> ReleaseProposalService:
|
def get_release_proposal_service(session: AsyncSession) -> ReleaseProposalService:
|
||||||
"""Construct a ReleaseProposalService bound to ``session``."""
|
"""Construct a ReleaseProposalService bound to ``session``."""
|
||||||
return ReleaseProposalService(session)
|
return ReleaseProposalService(session)
|
||||||
|
|
||||||
|
|
||||||
|
# In-flight background approves keyed by proposal task id. The HTTP approve
|
||||||
|
# route dispatches the ~40min execute asynchronously (a synchronous request
|
||||||
|
# would 504 at any proxy before the fail-closed gate/CI/publish finished) and
|
||||||
|
# returns 202 immediately; the panel polls GET /proposal for the final status.
|
||||||
|
# This registry lets a status endpoint / tests await the dispatched execute;
|
||||||
|
# it self-cleans via a done-callback and the Redis mutex still prevents a
|
||||||
|
# double-execute on a second click.
|
||||||
|
_INFLIGHT_APPROVES: dict[UUID, asyncio.Task[None]] = {}
|
||||||
|
|
||||||
|
|
||||||
|
async def _run_approve_background(
|
||||||
|
task_id: UUID, session_factory: async_sessionmaker[AsyncSession]
|
||||||
|
) -> None:
|
||||||
|
"""Run ``approve`` in a background task with a fresh session (the request
|
||||||
|
session closes when the 202 returns). Commits the outcome; a failure logs
|
||||||
|
and rolls back — the proposal stays open for the CEO to retry."""
|
||||||
|
async with session_factory() as bg_db:
|
||||||
|
try:
|
||||||
|
result = await get_release_proposal_service(bg_db).approve(task_id)
|
||||||
|
logger.info(
|
||||||
|
"release approve completed task_id=%s status=%s",
|
||||||
|
task_id,
|
||||||
|
result.status if result is not None else "no_report",
|
||||||
|
)
|
||||||
|
await bg_db.commit()
|
||||||
|
except Exception:
|
||||||
|
logger.exception(
|
||||||
|
"release approve background task failed task_id=%s", task_id
|
||||||
|
)
|
||||||
|
await bg_db.rollback()
|
||||||
|
|
||||||
|
|
||||||
|
def dispatch_approve(
|
||||||
|
task_id: UUID, session_factory: async_sessionmaker[AsyncSession]
|
||||||
|
) -> asyncio.Task[None]:
|
||||||
|
"""Spawn the long release execute in a background task so the HTTP approve
|
||||||
|
route can return 202 immediately. Registered in ``_INFLIGHT_APPROVES`` for
|
||||||
|
observability (done-callback removes the entry)."""
|
||||||
|
bg_task = asyncio.create_task(_run_approve_background(task_id, session_factory))
|
||||||
|
_INFLIGHT_APPROVES[task_id] = bg_task
|
||||||
|
bg_task.add_done_callback(lambda _t: _INFLIGHT_APPROVES.pop(task_id, None))
|
||||||
|
return bg_task
|
||||||
|
|||||||
@@ -12,18 +12,22 @@ import pytest_asyncio
|
|||||||
from fastapi import FastAPI
|
from fastapi import FastAPI
|
||||||
from httpx import ASGITransport, AsyncClient
|
from httpx import ASGITransport, AsyncClient
|
||||||
from roboco.api.deps import get_agent_context, get_db
|
from roboco.api.deps import get_agent_context, get_db
|
||||||
|
from roboco.api.routes import release as release_route
|
||||||
from roboco.api.routes.release import router as release_router
|
from roboco.api.routes.release import router as release_router
|
||||||
from roboco.db.tables import AgentTable, ProjectTable, TaskTable
|
from roboco.db.tables import AgentTable, ProjectTable, TaskTable
|
||||||
from roboco.models import AgentRole, AgentStatus, Team
|
from roboco.models import AgentRole, AgentStatus, Team
|
||||||
from roboco.models.base import TaskNature, TaskStatus, TaskType
|
from roboco.models.base import TaskNature, TaskStatus, TaskType
|
||||||
from roboco.models.permissions import AgentContext
|
from roboco.models.permissions import AgentContext
|
||||||
from roboco.services.release_executor import ReleaseResult
|
from roboco.services.release_executor import ReleaseResult
|
||||||
|
from roboco.services.release_proposal import ReleaseProposalService
|
||||||
from roboco.services.release_readiness import ReleaseReadinessReport, report_to_dict
|
from roboco.services.release_readiness import ReleaseReadinessReport, report_to_dict
|
||||||
from roboco.services.task import RELEASE_MANAGER_SOURCE
|
from roboco.services.task import RELEASE_MANAGER_SOURCE
|
||||||
from sqlalchemy import delete
|
from sqlalchemy import delete
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
|
import asyncio
|
||||||
from collections.abc import AsyncIterator
|
from collections.abc import AsyncIterator
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
@@ -148,9 +152,15 @@ async def test_get_proposal_404_when_none(ceo_client: AsyncClient) -> None:
|
|||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_approve_runs_executor_and_completes(
|
async def test_approve_dispatches_async_and_completes(
|
||||||
db_session: AsyncSession, ceo_client: AsyncClient
|
db_session: AsyncSession, ceo_client: AsyncClient
|
||||||
) -> None:
|
) -> None:
|
||||||
|
"""#324: the approve route dispatches the ~40min execute in a background
|
||||||
|
task and returns 202 immediately (a synchronous request would 504 at nginx
|
||||||
|
before the fail-closed gate/CI/publish finished). The panel polls
|
||||||
|
GET /proposal for the final status; here we await the dispatched task and
|
||||||
|
assert the proposal transitions to COMPLETED once the (faked) publish
|
||||||
|
succeeds."""
|
||||||
task = await _seed_proposal(db_session)
|
task = await _seed_proposal(db_session)
|
||||||
published = ReleaseResult(
|
published = ReleaseResult(
|
||||||
status="published",
|
status="published",
|
||||||
@@ -162,23 +172,55 @@ async def test_approve_runs_executor_and_completes(
|
|||||||
)
|
)
|
||||||
fake_executor = AsyncMock()
|
fake_executor = AsyncMock()
|
||||||
fake_executor.execute = AsyncMock(return_value=published)
|
fake_executor.execute = AsyncMock(return_value=published)
|
||||||
with patch(
|
captured: dict[str, asyncio.Task[None]] = {}
|
||||||
"roboco.services.release_proposal.get_release_executor",
|
real_dispatch = release_route.dispatch_approve
|
||||||
AsyncMock(return_value=fake_executor),
|
|
||||||
|
def _capturing_dispatch(task_id: UUID, factory: Any) -> asyncio.Task[None]:
|
||||||
|
bg = real_dispatch(task_id, factory)
|
||||||
|
captured["task"] = bg
|
||||||
|
return bg
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch(
|
||||||
|
"roboco.services.release_proposal.get_release_executor",
|
||||||
|
AsyncMock(return_value=fake_executor),
|
||||||
|
),
|
||||||
|
patch(
|
||||||
|
"roboco.api.routes.release.dispatch_approve",
|
||||||
|
side_effect=_capturing_dispatch,
|
||||||
|
),
|
||||||
|
patch.object(
|
||||||
|
ReleaseProposalService, "_acquire_release_lock", AsyncMock(return_value="t")
|
||||||
|
),
|
||||||
|
patch.object(
|
||||||
|
ReleaseProposalService,
|
||||||
|
"_release_release_lock",
|
||||||
|
AsyncMock(return_value=None),
|
||||||
|
),
|
||||||
|
patch.object(
|
||||||
|
ReleaseProposalService,
|
||||||
|
"_heartbeat_release_lock",
|
||||||
|
AsyncMock(return_value=True),
|
||||||
|
),
|
||||||
):
|
):
|
||||||
resp = await ceo_client.post("/api/release/proposal/approve")
|
resp = await ceo_client.post("/api/release/proposal/approve")
|
||||||
assert resp.status_code == HTTPStatus.OK
|
assert resp.status_code == HTTPStatus.ACCEPTED
|
||||||
assert resp.json()["status"] == "published"
|
assert resp.json()["status"] == "accepted"
|
||||||
|
# Await the background execute WHILE the executor patch is still active
|
||||||
|
# (the dispatched task runs the faked publish).
|
||||||
|
bg = captured["task"]
|
||||||
|
await bg
|
||||||
fake_executor.execute.assert_awaited_once()
|
fake_executor.execute.assert_awaited_once()
|
||||||
refreshed = await db_session.get(TaskTable, task.id)
|
await db_session.refresh(task)
|
||||||
assert refreshed is not None
|
assert task.status == TaskStatus.COMPLETED
|
||||||
assert refreshed.status == TaskStatus.COMPLETED
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_approve_gate_failure_keeps_proposal_open(
|
async def test_approve_gate_failure_keeps_proposal_open_async(
|
||||||
db_session: AsyncSession, ceo_client: AsyncClient
|
db_session: AsyncSession, ceo_client: AsyncClient
|
||||||
) -> None:
|
) -> None:
|
||||||
|
"""A gate failure in the background execute leaves the proposal open (the
|
||||||
|
CEO retries after the cause is fixed); the route still returned 202."""
|
||||||
task = await _seed_proposal(db_session)
|
task = await _seed_proposal(db_session)
|
||||||
failed = ReleaseResult(
|
failed = ReleaseResult(
|
||||||
status="gate_failed",
|
status="gate_failed",
|
||||||
@@ -190,16 +232,43 @@ async def test_approve_gate_failure_keeps_proposal_open(
|
|||||||
)
|
)
|
||||||
fake_executor = AsyncMock()
|
fake_executor = AsyncMock()
|
||||||
fake_executor.execute = AsyncMock(return_value=failed)
|
fake_executor.execute = AsyncMock(return_value=failed)
|
||||||
with patch(
|
captured: dict[str, asyncio.Task[None]] = {}
|
||||||
"roboco.services.release_proposal.get_release_executor",
|
real_dispatch = release_route.dispatch_approve
|
||||||
AsyncMock(return_value=fake_executor),
|
|
||||||
|
def _capturing_dispatch(task_id: UUID, factory: Any) -> asyncio.Task[None]:
|
||||||
|
bg = real_dispatch(task_id, factory)
|
||||||
|
captured["task"] = bg
|
||||||
|
return bg
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch(
|
||||||
|
"roboco.services.release_proposal.get_release_executor",
|
||||||
|
AsyncMock(return_value=fake_executor),
|
||||||
|
),
|
||||||
|
patch(
|
||||||
|
"roboco.api.routes.release.dispatch_approve",
|
||||||
|
side_effect=_capturing_dispatch,
|
||||||
|
),
|
||||||
|
patch.object(
|
||||||
|
ReleaseProposalService, "_acquire_release_lock", AsyncMock(return_value="t")
|
||||||
|
),
|
||||||
|
patch.object(
|
||||||
|
ReleaseProposalService,
|
||||||
|
"_release_release_lock",
|
||||||
|
AsyncMock(return_value=None),
|
||||||
|
),
|
||||||
|
patch.object(
|
||||||
|
ReleaseProposalService,
|
||||||
|
"_heartbeat_release_lock",
|
||||||
|
AsyncMock(return_value=True),
|
||||||
|
),
|
||||||
):
|
):
|
||||||
resp = await ceo_client.post("/api/release/proposal/approve")
|
resp = await ceo_client.post("/api/release/proposal/approve")
|
||||||
assert resp.status_code == HTTPStatus.OK
|
assert resp.status_code == HTTPStatus.ACCEPTED
|
||||||
assert resp.json()["status"] == "gate_failed"
|
assert resp.json()["status"] == "accepted"
|
||||||
refreshed = await db_session.get(TaskTable, task.id)
|
await captured["task"]
|
||||||
assert refreshed is not None
|
await db_session.refresh(task)
|
||||||
assert refreshed.status == TaskStatus.PENDING # still held
|
assert task.status == TaskStatus.PENDING # still held for retry
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|||||||
Reference in New Issue
Block a user