add Fabric read model rollback activation
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c
This commit is contained in:
parent
cf40c7bb4e
commit
cdff3b7e08
5 changed files with 133 additions and 0 deletions
|
|
@ -1,5 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
|
@ -10,6 +11,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
|||
from api.database import get_session
|
||||
from api.models.fabric_graph import FabricGraphEdge, FabricGraphImport, FabricGraphNode
|
||||
from api.schemas.fabric_graph import (
|
||||
FabricGraphActivationRequest,
|
||||
FabricGraphActivationResult,
|
||||
FabricGraphEdgeRead,
|
||||
FabricGraphImportRead,
|
||||
FabricGraphIngestResult,
|
||||
|
|
@ -19,6 +22,7 @@ from api.schemas.fabric_graph import (
|
|||
)
|
||||
from api.services.fabric_graph import (
|
||||
FabricGraphValidationError,
|
||||
activate_fabric_graph_import,
|
||||
ingest_fabric_graph_export,
|
||||
record_fabric_graph_error,
|
||||
split_graph_ingest_body,
|
||||
|
|
@ -132,6 +136,34 @@ async def latest_graph_import(
|
|||
return FabricGraphImportRead.model_validate(import_run)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/graph-exports/{import_id}/activate",
|
||||
response_model=FabricGraphActivationResult,
|
||||
)
|
||||
async def activate_graph_import(
|
||||
import_id: uuid.UUID,
|
||||
body: FabricGraphActivationRequest | None = None,
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> FabricGraphActivationResult:
|
||||
request = body or FabricGraphActivationRequest()
|
||||
try:
|
||||
import_run, previous_import_id, activated = await activate_fabric_graph_import(
|
||||
session,
|
||||
import_id,
|
||||
requested_by=request.requested_by,
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||||
if import_run is None:
|
||||
raise HTTPException(status_code=404, detail=f"Fabric graph import '{import_id}' not found")
|
||||
return FabricGraphActivationResult(
|
||||
import_run=FabricGraphImportRead.model_validate(import_run),
|
||||
previous_import_id=previous_import_id,
|
||||
activated=activated,
|
||||
idempotent=not activated,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/graph/nodes", response_model=list[FabricGraphNodeRead])
|
||||
async def list_graph_nodes(
|
||||
source_repo_slug: str = "railiance-fabric",
|
||||
|
|
|
|||
|
|
@ -261,6 +261,17 @@ class FabricGraphIngestResult(BaseModel):
|
|||
edge_count: int
|
||||
|
||||
|
||||
class FabricGraphActivationRequest(BaseModel):
|
||||
requested_by: str = "operator"
|
||||
|
||||
|
||||
class FabricGraphActivationResult(BaseModel):
|
||||
import_run: FabricGraphImportRead
|
||||
previous_import_id: uuid.UUID | None = None
|
||||
activated: bool
|
||||
idempotent: bool
|
||||
|
||||
|
||||
class FabricGraphNodeRead(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
|
|
|
|||
|
|
@ -221,6 +221,55 @@ async def ingest_fabric_graph_export(
|
|||
return import_run, True, False
|
||||
|
||||
|
||||
async def activate_fabric_graph_import(
|
||||
session: AsyncSession,
|
||||
import_id: Any,
|
||||
*,
|
||||
requested_by: str,
|
||||
) -> tuple[FabricGraphImport | None, Any | None, bool]:
|
||||
"""Select a retained valid import as the active read model.
|
||||
|
||||
Imports are immutable. Activation only moves the per-source ``is_latest``
|
||||
marker, which makes cutover rollback fast and avoids re-fetching an older
|
||||
authority payload during an incident.
|
||||
"""
|
||||
result = await session.execute(
|
||||
select(FabricGraphImport).where(FabricGraphImport.id == import_id)
|
||||
)
|
||||
import_run = result.scalar_one_or_none()
|
||||
if import_run is None:
|
||||
return None, None, False
|
||||
if import_run.validation_status != "valid":
|
||||
raise ValueError("Only a valid Fabric graph import can be activated.")
|
||||
|
||||
previous_result = await session.execute(
|
||||
select(FabricGraphImport).where(
|
||||
FabricGraphImport.source_repo_slug == import_run.source_repo_slug,
|
||||
FabricGraphImport.is_latest.is_(True),
|
||||
)
|
||||
)
|
||||
previous = previous_result.scalars().first()
|
||||
if previous is not None and previous.id == import_run.id:
|
||||
return import_run, previous.id, False
|
||||
|
||||
await _mark_latest(session, import_run)
|
||||
import_run.last_seen_at = datetime.now(timezone.utc)
|
||||
await _record_progress(
|
||||
session,
|
||||
"Fabric graph read model activation changed.",
|
||||
{
|
||||
"source_repo_slug": import_run.source_repo_slug,
|
||||
"import_id": str(import_run.id),
|
||||
"previous_import_id": str(previous.id) if previous else None,
|
||||
"content_hash": import_run.content_hash,
|
||||
"requested_by": requested_by,
|
||||
},
|
||||
)
|
||||
await session.commit()
|
||||
await session.refresh(import_run)
|
||||
return import_run, previous.id if previous else None, True
|
||||
|
||||
|
||||
def validate_fabric_graph_export(payload: dict[str, Any]) -> FabricGraphExportPayload:
|
||||
try:
|
||||
export = FabricGraphExportPayload.model_validate(payload)
|
||||
|
|
|
|||
|
|
@ -61,6 +61,20 @@ curl -s http://127.0.0.1:8000/fabric/graph-exports/latest
|
|||
curl -s http://127.0.0.1:8000/fabric/graph/summary
|
||||
```
|
||||
|
||||
Imports are immutable and retained. To roll the read model back to a previously
|
||||
validated import without depending on the source service, activate its import
|
||||
record; repeat the operation with the newer import id to roll forward again:
|
||||
|
||||
```bash
|
||||
curl -s -X POST \
|
||||
http://127.0.0.1:8000/fabric/graph-exports/<import-id>/activate \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"requested_by":"operator-rollback"}'
|
||||
```
|
||||
|
||||
Activation only moves the per-source latest marker. It does not alter or delete
|
||||
either graph snapshot, and activating the current import is idempotent.
|
||||
|
||||
Representative relationship queries:
|
||||
|
||||
```bash
|
||||
|
|
|
|||
|
|
@ -1545,6 +1545,33 @@ class TestFabricGraphReadModel:
|
|||
assert latest_flags[first_id] is False
|
||||
assert latest_flags[second_id] is True
|
||||
|
||||
async def test_retained_valid_import_can_be_activated_for_rollback(self, client):
|
||||
r = await client.post("/fabric/graph-exports", json=_fabric_graph_export())
|
||||
first_id = r.json()["import_run"]["id"]
|
||||
r = await client.post("/fabric/graph-exports", json=_fabric_graph_export(extra_node=True))
|
||||
second_id = r.json()["import_run"]["id"]
|
||||
|
||||
r = await client.post(
|
||||
f"/fabric/graph-exports/{first_id}/activate",
|
||||
json={"requested_by": "cutover-test"},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["activated"] is True
|
||||
assert r.json()["idempotent"] is False
|
||||
assert r.json()["previous_import_id"] == second_id
|
||||
assert r.json()["import_run"]["id"] == first_id
|
||||
assert (await client.get("/fabric/graph/summary")).json()["node_count"] == 3
|
||||
|
||||
r = await client.post(f"/fabric/graph-exports/{first_id}/activate")
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["activated"] is False
|
||||
assert r.json()["idempotent"] is True
|
||||
|
||||
r = await client.post(f"/fabric/graph-exports/{second_id}/activate")
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["previous_import_id"] == first_id
|
||||
assert (await client.get("/fabric/graph/summary")).json()["node_count"] == 4
|
||||
|
||||
async def test_read_only_queries_filter_graph_without_mutating_state_hub_entities(self, client):
|
||||
await _create_domain(client)
|
||||
topic = await _create_topic(client)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue