fix(ops): decouple action execution to background tasks and stabilize live progress tracking (#24)
Decoupled action execution in FastAPI to background tasks and stabilized live modal progress tracking.
This commit is contained in:
parent
54886753b7
commit
1a272ce1ff
|
|
@ -10,7 +10,7 @@ from email.message import EmailMessage
|
||||||
from typing import Any, Literal, Optional
|
from typing import Any, Literal, Optional
|
||||||
|
|
||||||
from email_validator import EmailNotValidError, validate_email
|
from email_validator import EmailNotValidError, validate_email
|
||||||
from fastapi import Body, Depends, FastAPI, Header, HTTPException, Query, Request, status
|
from fastapi import BackgroundTasks, Body, Depends, FastAPI, Header, HTTPException, Query, Request, status
|
||||||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||||||
from sqlalchemy import func, or_, select, text
|
from sqlalchemy import func, or_, select, text
|
||||||
from sqlalchemy.exc import IntegrityError
|
from sqlalchemy.exc import IntegrityError
|
||||||
|
|
@ -3069,9 +3069,56 @@ def ops_get_support_ticket(
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
def _run_ops_action_in_background(
|
||||||
|
action_id: uuid.UUID,
|
||||||
|
operator_identity: str,
|
||||||
|
operator_role: str,
|
||||||
|
bind: Any = None,
|
||||||
|
helper_override: str | None = None,
|
||||||
|
) -> None:
|
||||||
|
if not os.environ.get("OPS_ACTION_MOCK_EXECUTION"):
|
||||||
|
# Grace period (200ms) ensuring the HTTP response is fully flushed to Next.js and client browser
|
||||||
|
time.sleep(0.2)
|
||||||
|
try:
|
||||||
|
session_bind = bind or engine
|
||||||
|
with DbSession(session_bind) as db:
|
||||||
|
action = db.get(OpsAction, action_id)
|
||||||
|
if not action:
|
||||||
|
logger.error("Background ops action %s not found", action_id)
|
||||||
|
return
|
||||||
|
|
||||||
|
action = ops_actions_service.execute_action_sync(action, db, helper_override=helper_override)
|
||||||
|
|
||||||
|
try:
|
||||||
|
ops_audit(
|
||||||
|
db=db,
|
||||||
|
action=f"ops.action_{action.state}",
|
||||||
|
actor_identity=operator_identity,
|
||||||
|
actor_role=operator_role,
|
||||||
|
target_resource=f"action:{action.id}",
|
||||||
|
reason_code=(action.failure_code or "")[:64] if action.state != "succeeded" else None,
|
||||||
|
result="success" if action.state == "succeeded" else "failed",
|
||||||
|
metadata={
|
||||||
|
"action_type": action.action_type,
|
||||||
|
"failure_code": action.failure_code,
|
||||||
|
"deployed_revision": action.deployed_revision,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
db.commit()
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"Error recording action completion audit: {e}")
|
||||||
|
try:
|
||||||
|
db.rollback()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
except Exception as exc:
|
||||||
|
logger.exception("Unexpected error in background ops action execution: %s", exc)
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/v1/ops/actions", response_model=OpsActionRead)
|
@app.post("/api/v1/ops/actions", response_model=OpsActionRead)
|
||||||
def ops_create_action(
|
def ops_create_action(
|
||||||
req: OpsActionCreate,
|
req: OpsActionCreate,
|
||||||
|
background_tasks: BackgroundTasks,
|
||||||
db: DbSession = Depends(get_db),
|
db: DbSession = Depends(get_db),
|
||||||
auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("owner")),
|
auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("owner")),
|
||||||
):
|
):
|
||||||
|
|
@ -3135,13 +3182,15 @@ def ops_create_action(
|
||||||
)
|
)
|
||||||
|
|
||||||
# Create new action record
|
# Create new action record
|
||||||
|
now_utc = datetime.now(timezone.utc)
|
||||||
try:
|
try:
|
||||||
action = OpsAction(
|
action = OpsAction(
|
||||||
id=uuid.uuid4(),
|
id=uuid.uuid4(),
|
||||||
action_type=req.action_type,
|
action_type=req.action_type,
|
||||||
requested_by_id=operator.id,
|
requested_by_id=operator.id,
|
||||||
request_timestamp=datetime.now(timezone.utc),
|
request_timestamp=now_utc,
|
||||||
state="queued",
|
start_timestamp=now_utc,
|
||||||
|
state="running",
|
||||||
idempotency_key=req.idempotency_key,
|
idempotency_key=req.idempotency_key,
|
||||||
)
|
)
|
||||||
db.add(action)
|
db.add(action)
|
||||||
|
|
@ -3174,31 +3223,15 @@ def ops_create_action(
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
# Execute action via helper service
|
# Launch execution asynchronously in background
|
||||||
action = ops_actions_service.execute_action_sync(action, db)
|
background_tasks.add_task(
|
||||||
|
_run_ops_action_in_background,
|
||||||
|
action.id,
|
||||||
|
operator.identity,
|
||||||
|
operator.role,
|
||||||
|
bind=db.get_bind(),
|
||||||
|
)
|
||||||
|
|
||||||
try:
|
|
||||||
ops_audit(
|
|
||||||
db=db,
|
|
||||||
action=f"ops.action_{action.state}",
|
|
||||||
actor_identity=operator.identity,
|
|
||||||
actor_role=operator.role,
|
|
||||||
target_resource=f"action:{action.id}",
|
|
||||||
reason_code=(action.failure_code or "")[:64] if action.state != "succeeded" else None,
|
|
||||||
result="success" if action.state == "succeeded" else "failed",
|
|
||||||
metadata={
|
|
||||||
"action_type": action.action_type,
|
|
||||||
"failure_code": action.failure_code,
|
|
||||||
"deployed_revision": action.deployed_revision,
|
|
||||||
},
|
|
||||||
)
|
|
||||||
db.commit()
|
|
||||||
except Exception as e:
|
|
||||||
logger.warning(f"Error recording action completion audit: {e}")
|
|
||||||
try:
|
|
||||||
db.rollback()
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
return action
|
return action
|
||||||
|
|
||||||
@app.get("/api/v1/ops/actions", response_model=list[OpsActionRead])
|
@app.get("/api/v1/ops/actions", response_model=list[OpsActionRead])
|
||||||
|
|
|
||||||
|
|
@ -343,8 +343,14 @@ class TestOpsActions(unittest.TestCase):
|
||||||
)
|
)
|
||||||
self.assertEqual(resp.status_code, 200)
|
self.assertEqual(resp.status_code, 200)
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
self.assertEqual(data["state"], "succeeded")
|
self.assertEqual(data["state"], "running")
|
||||||
self.assertEqual(data["deployed_revision"], "d69d6119d026b1df2b7649489d1d79275013f283")
|
|
||||||
|
# Background task updates terminal state
|
||||||
|
poll_resp = self.client.get(f"/api/v1/ops/actions/{data['id']}", headers=self.owner_headers)
|
||||||
|
self.assertEqual(poll_resp.status_code, 200)
|
||||||
|
final_data = poll_resp.json()
|
||||||
|
self.assertEqual(final_data["state"], "succeeded")
|
||||||
|
self.assertEqual(final_data["deployed_revision"], "d69d6119d026b1df2b7649489d1d79275013f283")
|
||||||
|
|
||||||
# Verify audit event for success
|
# Verify audit event for success
|
||||||
with DbSession(self.engine) as db:
|
with DbSession(self.engine) as db:
|
||||||
|
|
@ -406,8 +412,14 @@ class TestOpsActions(unittest.TestCase):
|
||||||
)
|
)
|
||||||
self.assertEqual(resp.status_code, 200)
|
self.assertEqual(resp.status_code, 200)
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
self.assertEqual(data["state"], "failed")
|
self.assertEqual(data["state"], "running")
|
||||||
self.assertEqual(data["failure_code"], "build_error")
|
|
||||||
|
# Verify background execution failure
|
||||||
|
poll_resp = self.client.get(f"/api/v1/ops/actions/{data['id']}", headers=self.owner_headers)
|
||||||
|
self.assertEqual(poll_resp.status_code, 200)
|
||||||
|
final_data = poll_resp.json()
|
||||||
|
self.assertEqual(final_data["state"], "failed")
|
||||||
|
self.assertEqual(final_data["failure_code"], "build_error")
|
||||||
|
|
||||||
# Verify audit event for failure
|
# Verify audit event for failure
|
||||||
with DbSession(self.engine) as db:
|
with DbSession(self.engine) as db:
|
||||||
|
|
@ -434,8 +446,14 @@ class TestOpsActions(unittest.TestCase):
|
||||||
)
|
)
|
||||||
self.assertEqual(resp.status_code, 200)
|
self.assertEqual(resp.status_code, 200)
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
self.assertEqual(data["state"], "rolled_back")
|
self.assertEqual(data["state"], "running")
|
||||||
self.assertEqual(data["rollback_status"], "rolled_back")
|
|
||||||
|
# Verify rollback completion
|
||||||
|
poll_resp = self.client.get(f"/api/v1/ops/actions/{data['id']}", headers=self.owner_headers)
|
||||||
|
self.assertEqual(poll_resp.status_code, 200)
|
||||||
|
final_data = poll_resp.json()
|
||||||
|
self.assertEqual(final_data["state"], "rolled_back")
|
||||||
|
self.assertEqual(final_data["rollback_status"], "rolled_back")
|
||||||
finally:
|
finally:
|
||||||
os.environ.pop("OPS_ACTION_MOCK_EXECUTION", None)
|
os.environ.pop("OPS_ACTION_MOCK_EXECUTION", None)
|
||||||
|
|
||||||
|
|
@ -452,8 +470,14 @@ class TestOpsActions(unittest.TestCase):
|
||||||
)
|
)
|
||||||
self.assertEqual(resp.status_code, 200)
|
self.assertEqual(resp.status_code, 200)
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
self.assertEqual(data["state"], "failed")
|
self.assertEqual(data["state"], "running")
|
||||||
self.assertEqual(data["failure_code"], "timeout")
|
|
||||||
|
# Verify timeout failure
|
||||||
|
poll_resp = self.client.get(f"/api/v1/ops/actions/{data['id']}", headers=self.owner_headers)
|
||||||
|
self.assertEqual(poll_resp.status_code, 200)
|
||||||
|
final_data = poll_resp.json()
|
||||||
|
self.assertEqual(final_data["state"], "failed")
|
||||||
|
self.assertEqual(final_data["failure_code"], "timeout")
|
||||||
finally:
|
finally:
|
||||||
os.environ.pop("OPS_ACTION_MOCK_EXECUTION", None)
|
os.environ.pop("OPS_ACTION_MOCK_EXECUTION", None)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -110,22 +110,43 @@ export function OperationsActions({ role, onActionCompleted }: OperationsActions
|
||||||
const pollTimeoutRef = useRef<NodeJS.Timeout | null>(null);
|
const pollTimeoutRef = useRef<NodeJS.Timeout | null>(null);
|
||||||
const isOwner = role === 'owner';
|
const isOwner = role === 'owner';
|
||||||
|
|
||||||
// Live Timer: Increments elapsed seconds when an action is in-flight or active
|
const actionStartTimeRef = useRef<number | null>(null);
|
||||||
|
|
||||||
|
// Live Timer: Calculates real elapsed seconds from wall clock via setTimeout recursion
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
let timer: NodeJS.Timeout | null = null;
|
let timer: NodeJS.Timeout | null = null;
|
||||||
if (inFlightAction || activeAction) {
|
|
||||||
const tick = () => {
|
if (inFlightAction) {
|
||||||
setElapsedSeconds((prev) => prev + 1);
|
if (!actionStartTimeRef.current) {
|
||||||
timer = setTimeout(() => tick(), 1000);
|
actionStartTimeRef.current = inFlightAction.startTime;
|
||||||
};
|
}
|
||||||
timer = setTimeout(() => tick(), 1000);
|
} else if (activeAction) {
|
||||||
|
if (!actionStartTimeRef.current) {
|
||||||
|
const parsed = activeAction.request_timestamp
|
||||||
|
? new Date(activeAction.request_timestamp).getTime()
|
||||||
|
: Date.now();
|
||||||
|
actionStartTimeRef.current = isNaN(parsed) ? Date.now() : parsed;
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
|
actionStartTimeRef.current = null;
|
||||||
setElapsedSeconds(0);
|
setElapsedSeconds(0);
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const tick = () => {
|
||||||
|
if (actionStartTimeRef.current) {
|
||||||
|
const sec = Math.max(0, Math.floor((Date.now() - actionStartTimeRef.current) / 1000));
|
||||||
|
setElapsedSeconds(sec);
|
||||||
|
}
|
||||||
|
timer = setTimeout(() => tick(), 1000);
|
||||||
|
};
|
||||||
|
|
||||||
|
tick();
|
||||||
|
|
||||||
return () => {
|
return () => {
|
||||||
if (timer) clearTimeout(timer);
|
if (timer) clearTimeout(timer);
|
||||||
};
|
};
|
||||||
}, [inFlightAction, activeAction]);
|
}, [inFlightAction?.startTime, activeAction?.id, activeAction?.request_timestamp]);
|
||||||
|
|
||||||
// Fetch capabilities & recent action history
|
// Fetch capabilities & recent action history
|
||||||
const fetchActions = useCallback(async () => {
|
const fetchActions = useCallback(async () => {
|
||||||
|
|
@ -198,6 +219,7 @@ export function OperationsActions({ role, onActionCompleted }: OperationsActions
|
||||||
// Action finished!
|
// Action finished!
|
||||||
setActiveAction(null);
|
setActiveAction(null);
|
||||||
setInFlightAction(null);
|
setInFlightAction(null);
|
||||||
|
actionStartTimeRef.current = null;
|
||||||
setCompletedAction(updated);
|
setCompletedAction(updated);
|
||||||
setRecentBannerAction(updated);
|
setRecentBannerAction(updated);
|
||||||
setModalMode('completed');
|
setModalMode('completed');
|
||||||
|
|
@ -209,7 +231,7 @@ export function OperationsActions({ role, onActionCompleted }: OperationsActions
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} catch {
|
} catch {
|
||||||
// Non-blocking error
|
// Non-blocking error across service cycling
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!isCancelled) {
|
if (!isCancelled) {
|
||||||
|
|
@ -231,7 +253,7 @@ export function OperationsActions({ role, onActionCompleted }: OperationsActions
|
||||||
pollTimeoutRef.current = null;
|
pollTimeoutRef.current = null;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
}, [activeAction, fetchActions, onActionCompleted]);
|
}, [activeAction?.id, activeAction?.state, fetchActions, onActionCompleted]);
|
||||||
|
|
||||||
const handleOpenModal = (actionType: OpsActionType) => {
|
const handleOpenModal = (actionType: OpsActionType) => {
|
||||||
if (!isOwner || activeAction || inFlightAction) return;
|
if (!isOwner || activeAction || inFlightAction) return;
|
||||||
|
|
@ -361,8 +383,11 @@ export function OperationsActions({ role, onActionCompleted }: OperationsActions
|
||||||
if (elapsedSeconds < 4) return 1;
|
if (elapsedSeconds < 4) return 1;
|
||||||
if (elapsedSeconds < 14) return 2;
|
if (elapsedSeconds < 14) return 2;
|
||||||
return 3;
|
return 3;
|
||||||
|
} else {
|
||||||
|
if (elapsedSeconds < 2) return 1;
|
||||||
|
if (elapsedSeconds < 5) return 2;
|
||||||
|
return 3;
|
||||||
}
|
}
|
||||||
return 1;
|
|
||||||
}
|
}
|
||||||
if (state === 'preflight' || elapsedSeconds > 1) {
|
if (state === 'preflight' || elapsedSeconds > 1) {
|
||||||
return 1;
|
return 1;
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue