Reconcile lost upload responses without replaying frame writes

pigway_frame_flow/display/service.py

@@ -7,6 +7,7 @@ from uuid import uuid4
 
 from pigway_frame_flow.display.cloud_library import CloudLibrary
 from pigway_frame_flow.display.retention import Retention
+from pigway_frame_flow.display.upload_reconciliation import UploadReconciliation
 from pigway_frame_flow.drivers.base import DriverRegistry
 from pigway_frame_flow.errors import ErrorCode as E
 from pigway_frame_flow.errors import FlowError
@@ -49,6 +50,7 @@ class DisplayService:
         self.lock = threading.RLock()
         self.retention = Retention(self)
         self.cloud_library = CloudLibrary(self)
+        self.upload_reconciliation = UploadReconciliation(self)
         self.expiration_checked: dict[str, float] = {}
         self.unknown_checked: dict[str, float] = {}
         # A killed write cannot be replayed safely; pre-write work can resume.
@@ -169,8 +171,9 @@ class DisplayService:
         if result:
             row = self.db.one("SELECT result FROM operations WHERE id=?", (id,))
             previous = json.loads(row["result"]) if row and row["result"] else {}
-            if "photo_attempts" in previous:
-                result = {**result, "photo_attempts": previous["photo_attempts"]}
+            for key in ("photo_attempts", "upload_intent"):
+                if key in previous:
+                    result = {**result, key: previous[key]}
         self.db.execute(
             "UPDATE operations SET status=?,updated_at=?,result=COALESCE(?,result),error=? WHERE id=?",
             (
@@ -627,6 +630,8 @@ class DisplayService:
             raise FlowError(
                 E.CONFLICT, "Temporary override requires an existing image", 409
             )
+        if asset:
+            self.upload_reconciliation.prepare(id, driver, asset, previous)
         self._start_write(id)
         if asset:
             self.update(id, "UPLOADING")
@@ -690,22 +695,31 @@ class DisplayService:
         )
 
     def sync(self):
+        with self.lock:
+            self._sync()
+
+    def _sync(self):
         # Late acknowledgements only resolve a known display target. Never replay
         # writes or infer success from gallery presence alone.
         for op in self.db.all(
             "SELECT * FROM operations WHERE status='UNKNOWN' AND action='display' AND write_started=1"
         ):
             expected = json.loads(op["result"] or "{}").get("image_id")
-            if not expected:
+            intent = json.loads(op["result"] or "{}").get("upload_intent")
+            if not expected and not intent:
                 continue
             checked = time.monotonic()
             if checked - self.unknown_checked.get(op["id"], float("-inf")) < 60:
                 continue
             self.unknown_checked[op["id"]] = checked
             try:
-                device = self.registry.get(op["device_id"]).device(refresh=True)
+                driver = self.registry.get(op["device_id"])
+                device = driver.device(refresh=True)
+                if intent:
+                    expected = self.upload_reconciliation.check(op, driver, device)
                 if (
-                    device.current_content.id == expected
+                    expected
+                    and device.current_content.id == expected
                     and device.current_content.synced
                 ):
                     self.update(op["id"], "COMPLETED")

pigway_frame_flow/display/upload_reconciliation.py

@@ -0,0 +1,157 @@
+"""Resolve missing upload responses using read-only, exact-content evidence."""
+
+import hashlib
+import json
+from datetime import UTC, datetime, timedelta
+
+from pigway_frame_flow.errors import ErrorCode
+from pigway_frame_flow.storage.database import encode, now_iso
+
+
+class UploadReconciliation:
+    def __init__(self, service):
+        self.service = service
+
+    def prepare(self, operation_id, driver, asset, previous):
+        path = self.service.assets.path(asset["asset_id"])
+        self.service.update(
+            operation_id,
+            "RENDERING",
+            {
+                "rendered_asset_id": asset["asset_id"],
+                "upload_intent": {
+                    "name": asset["name"],
+                    "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
+                    "size": path.stat().st_size,
+                    "before_ids": [image.image_id for image in driver.images()],
+                    "previous_image_id": previous,
+                    "started_at": now_iso(),
+                },
+            },
+        )
+
+    def check(self, op, driver, device):
+        service = self.service
+        result = json.loads(op["result"] or "{}")
+        intent = result.get("upload_intent")
+        if not intent:
+            return None
+        expected = result.get("image_id")
+        if not expected:
+            start = datetime.fromisoformat(intent["started_at"])
+            candidates = []
+            for image in driver.images():
+                if (
+                    image.image_id in intent["before_ids"]
+                    or image.name != intent["name"]
+                ):
+                    continue
+                if image.size_bytes is not None and image.size_bytes != intent["size"]:
+                    continue
+                if not image.created_at:
+                    continue
+                try:
+                    created = datetime.fromisoformat(image.created_at)
+                    age = (created - start).total_seconds()
+                except (ValueError, TypeError):
+                    continue
+                if 0 <= age <= service.settings.sync_timeout + 300:
+                    candidates.append(image)
+            # Ambiguous matches never release a write lock.
+            if len(candidates) != 1:
+                return None
+            expected = candidates[0].image_id
+            data = driver.download(expected)
+            if (
+                len(data) != intent["size"]
+                or hashlib.sha256(data).hexdigest() != intent["sha256"]
+            ):
+                return None
+            result["image_id"] = expected
+            result["upload_confirmed"] = True
+            service.update(op["id"], "UNKNOWN", result, json.loads(op["error"] or "{}"))
+            service.assets.promote(result["rendered_asset_id"])
+            source = service.db.one(
+                "SELECT payload FROM asset_sources WHERE asset_id=?",
+                (result["rendered_asset_id"],),
+            )
+            service.db.execute(
+                "INSERT OR REPLACE INTO image_sources VALUES(?,?,?,?,?)",
+                (
+                    op["device_id"],
+                    expected,
+                    result["rendered_asset_id"],
+                    op["id"],
+                    source["payload"] if source else encode({"type": "image"}),
+                ),
+            )
+            if not service.db.one(
+                "SELECT image_id FROM managed_cloud_uploads WHERE operation_id=?",
+                (op["id"],),
+            ):
+                service.cloud_library.register(
+                    op["device_id"], expected, result["rendered_asset_id"], op["id"]
+                )
+        if device.current_content.id == expected and device.current_content.synced:
+            options = json.loads(op["payload"]).get("options", {})
+            service.db.execute(
+                "UPDATE device_state SET managed_image_id=? WHERE device_id=?",
+                (expected, op["device_id"]),
+            )
+            if options.get("restore_previous"):
+                expires = datetime.fromisoformat(intent["started_at"]) + timedelta(
+                    seconds=options["duration"]
+                )
+                service.db.execute(
+                    "INSERT OR IGNORE INTO overrides VALUES(?,?,?,?,?,?)",
+                    (
+                        op["id"],
+                        op["device_id"],
+                        expected,
+                        options["priority"],
+                        expires.isoformat(),
+                        intent["started_at"],
+                    ),
+                )
+            return expected
+        elapsed = (
+            datetime.now(UTC) - datetime.fromisoformat(intent["started_at"])
+        ).total_seconds()
+        if (
+            result.get("upload_confirmed")
+            and not json.loads(op["payload"]).get("options", {}).get("restore_previous")
+            and elapsed > service.settings.sync_timeout + 300
+            and device.current_content.id == intent["previous_image_id"]
+            and device.current_content.synced
+        ):
+            # Upload is proven; display is not. Two separated stable observations
+            # close the failed display, without replaying either vendor write.
+            observed = result.get("unchanged_observed_at")
+            if not observed:
+                result["unchanged_observed_at"] = now_iso()
+                service.update(
+                    op["id"], "UNKNOWN", result, json.loads(op["error"] or "{}")
+                )
+            elif (
+                datetime.now(UTC) - datetime.fromisoformat(observed)
+            ).total_seconds() >= 60:
+                service.update(
+                    op["id"],
+                    "FAILED",
+                    result,
+                    {
+                        "code": ErrorCode.DISPLAY_NOT_CONFIRMED,
+                        "message": "Upload confirmed; frame remained on previous image",
+                        "request_id": op["request_id"],
+                        "details": {},
+                    },
+                )
+                service.db.history(op, "DISPLAY_NOT_CONFIRMED")
+                service.db.event(
+                    "operation.failed",
+                    {"operation_id": op["id"], "code": ErrorCode.DISPLAY_NOT_CONFIRMED},
+                )
+        elif result.get("unchanged_observed_at"):
+            result.pop("unchanged_observed_at")
+            service.update(op["id"], "UNKNOWN", result, json.loads(op["error"] or "{}"))
+        return expected

pigway_frame_flow/errors.py

@@ -11,6 +11,7 @@ class ErrorCode(StrEnum):
     UNSUPPORTED_CAPABILITY = "UNSUPPORTED_CAPABILITY"
     RENDER_FAILED = "RENDER_FAILED"
     UPLOAD_FAILED = "UPLOAD_FAILED"
+    DISPLAY_NOT_CONFIRMED = "DISPLAY_NOT_CONFIRMED"
     WRITE_RESULT_UNKNOWN = "WRITE_RESULT_UNKNOWN"
     RATE_LIMITED = "RATE_LIMITED"
     INTERNAL_ERROR = "INTERNAL_ERROR"