120 lines
5.0 KiB
Python
120 lines
5.0 KiB
Python
"""Durable, read-only recovery progress for writes with uncertain outcomes."""
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
from pigway_frame_flow.errors import ErrorCode
|
|
from pigway_frame_flow.storage.database import encode, now_iso
|
|
|
|
|
|
class RecoveryMonitor:
|
|
def __init__(self, service):
|
|
self.service, self.db = service, service.db
|
|
|
|
def observe(self, op, reason):
|
|
timestamp = now_iso()
|
|
age = (
|
|
datetime.now(UTC) - datetime.fromisoformat(op["created_at"])
|
|
).total_seconds()
|
|
review = age >= self.service.settings.sync_timeout + 300
|
|
with self.db.transaction() as tx:
|
|
previous = tx.execute(
|
|
"SELECT * FROM operation_recovery WHERE operation_id=?", (op["id"],)
|
|
).fetchone()
|
|
# Once escalated, wall-clock corrections must not clear the alert.
|
|
review = review or bool(previous and previous["state"] == "manual_review")
|
|
state = "manual_review" if review else "checking"
|
|
tx.execute(
|
|
"INSERT INTO operation_recovery VALUES(?,?,?,?,?,1) ON CONFLICT(operation_id) DO UPDATE SET state=excluded.state,reason=excluded.reason,last_checked_at=excluded.last_checked_at,checks=operation_recovery.checks+1",
|
|
(op["id"], state, reason, timestamp, timestamp),
|
|
)
|
|
if review and (not previous or previous["state"] != "manual_review"):
|
|
tx.execute(
|
|
"INSERT INTO events VALUES(?,?,?,?)",
|
|
(
|
|
uuid4().hex,
|
|
"operation.blocked",
|
|
encode(
|
|
{
|
|
"operation_id": op["id"],
|
|
"device_id": op["device_id"],
|
|
"reason": reason,
|
|
"requires_manual_review": True,
|
|
}
|
|
),
|
|
timestamp,
|
|
),
|
|
)
|
|
tx.execute(
|
|
"INSERT INTO history(client_id,request_id,operation_id,device_id,action,result,timestamp) VALUES(?,?,?,?,?,?,?)",
|
|
(
|
|
op["client_id"],
|
|
op["request_id"],
|
|
op["id"],
|
|
op["device_id"],
|
|
op["action"],
|
|
"MANUAL_REVIEW_REQUIRED",
|
|
timestamp,
|
|
),
|
|
)
|
|
|
|
def resolve(self, operation_id):
|
|
self.db.execute(
|
|
"UPDATE operation_recovery SET state='resolved' WHERE operation_id=? AND state!='resolved'",
|
|
(operation_id,),
|
|
)
|
|
|
|
def info(self, operation_id):
|
|
row = self.db.one(
|
|
"SELECT state,reason,first_checked_at,last_checked_at,checks FROM operation_recovery WHERE operation_id=?",
|
|
(operation_id,),
|
|
)
|
|
if row:
|
|
row["requires_manual_review"] = row["state"] == "manual_review"
|
|
return row
|
|
|
|
def close_superseded_displays(self):
|
|
for op in self.db.all(
|
|
"SELECT * FROM operations WHERE action='display' AND write_started=1 "
|
|
"AND status IN ('SYNCING','UNKNOWN')"
|
|
):
|
|
result = json.loads(op["result"] or "{}")
|
|
if not result.get("image_id") or json.loads(op["payload"]).get(
|
|
"options", {}
|
|
).get("restore_previous"):
|
|
continue
|
|
newer = self.db.one(
|
|
"SELECT id,result FROM operations WHERE device_id=? AND action='display' "
|
|
"AND status='COMPLETED' AND write_started=1 AND created_at>? "
|
|
"AND COALESCE(json_extract(payload,'$.options.restore_previous'),0)=0 "
|
|
"AND json_extract(result,'$.image_id') IS NOT NULL "
|
|
"ORDER BY created_at DESC LIMIT 1",
|
|
(op["device_id"], op["created_at"]),
|
|
)
|
|
if (
|
|
not newer
|
|
or json.loads(newer["result"])["image_id"] == result["image_id"]
|
|
):
|
|
continue
|
|
# A later confirmed permanent display ends this attempt, but cannot
|
|
# prove whether the earlier image was ever shown. Never replay it.
|
|
self.service.update(
|
|
op["id"],
|
|
"FAILED",
|
|
error={
|
|
"code": ErrorCode.DISPLAY_NOT_CONFIRMED,
|
|
"message": "Superseded by a later confirmed display; earlier display unconfirmed",
|
|
"request_id": op["request_id"],
|
|
"details": {
|
|
"superseded_by_operation_id": newer["id"],
|
|
"previous_error": json.loads(op["error"] or "null"),
|
|
},
|
|
},
|
|
)
|
|
self.db.history(op, "DISPLAY_NOT_CONFIRMED")
|
|
self.db.event(
|
|
"operation.failed",
|
|
{"operation_id": op["id"], "code": ErrorCode.DISPLAY_NOT_CONFIRMED},
|
|
)
|