74 lines
3.0 KiB
Python
74 lines
3.0 KiB
Python
"""Durable, read-only recovery progress for writes with uncertain outcomes."""
|
|
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
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
|