"""Durable, job-local upload retries; scheduled execution remains authoritative.""" import json from datetime import datetime, timedelta from pigway_frame_flow.errors import FlowError from pigway_frame_flow.storage.database import encode class UploadRetries: def __init__(self, service): self.service, self.db = service, service.db def load(self, job_id): row = self.db.one( "SELECT payload FROM app_config WHERE key=?", ("upload-retry:" + job_id,) ) return json.loads(row["payload"]) if row else None def save(self, job_id, plan): if plan is None: self.db.execute( "DELETE FROM app_config WHERE key=?", ("upload-retry:" + job_id,) ) else: self.db.execute( "INSERT OR REPLACE INTO app_config VALUES(?,?)", ("upload-retry:" + job_id, encode(plan)), ) @staticmethod def reuse(payload, asset_id): return {**payload, "content": {"type": "image", "asset_id": asset_id}} def scheduled_payload(self, job_id, payload): plan = self.load(job_id) if plan and plan["phase"] in ("extra", "scheduled", "scheduled_running"): return self.reuse(payload, plan["asset_id"]) return payload def scheduled_submitted(self, job_id, operation_id): plan = self.load(job_id) if plan and plan["phase"] in ("extra", "scheduled", "scheduled_running"): plan.update(phase="scheduled_running", active=operation_id) self.save(job_id, plan) def close_unknown(self, operation): if operation["status"] == "UNKNOWN": raw = self.db.one( "SELECT * FROM operations WHERE id=?", (operation["operation_id"],) ) driver = self.service.registry.get(operation["device_id"]) self.service.upload_reconciliation.check( raw, driver, driver.device(refresh=True) ) self.service.reconcile(operation["operation_id"], acknowledge=True) return self.service.operation(operation["operation_id"]) return operation def discard(self, asset_id): row = self.db.one("SELECT kind FROM assets WHERE id=?", (asset_id,)) meta = self.db.one( "SELECT starred FROM asset_metadata WHERE asset_id=?", (asset_id,) ) # Only disposable generated staging files are owned by this recovery. if row and row["kind"] == "cache" and not (meta and meta["starred"]): self.service.libraries.remove_asset(asset_id) def tick(self, current): for job in self.db.all("SELECT * FROM jobs WHERE enabled=1"): try: self.check(job, current) except FlowError: # A read failure leaves the durable plan in place for a later check. continue def check(self, job, current): config = json.loads(job["payload"]) if config["source"]["type"] not in ( "photo_source", "auto_collage", "template", "provider", ): return plan = self.load(job["id"]) if plan and plan["source"] != config["source"]: self.save(job["id"], None) plan = None if not plan: if not job["last_result"] or not job["next_run"]: return operation = self.service.operation(job["last_result"]) asset_id = (operation["result"] or {}).get("rendered_asset_id") if operation["status"] not in ("FAILED", "UNKNOWN") or not asset_id: return deadline = datetime.fromisoformat(job["next_run"]) remaining = (deadline - current).total_seconds() if remaining <= 0: times = [] else: # Five increasing delays occupy the first half of the available # window, leaving time for upload/sync before the next schedule. unit = remaining / 62 times = [ (current + timedelta(seconds=unit * n)).isoformat() for n in (1, 3, 7, 15, 31) ] plan = dict( root=operation["operation_id"], active=operation["operation_id"], asset_id=asset_id, source=config["source"], times=times, count=0, deadline=deadline.isoformat(), phase="extra", ) self.save(job["id"], plan) operation = self.service.operation(plan["active"]) deadline = min( datetime.fromisoformat(plan["deadline"]), datetime.fromisoformat(job["next_run"]), ) if ( plan["phase"] == "extra" and current >= deadline and operation["status"] != "COMPLETED" ): if operation["status"] == "SYNCING": self.service.update( operation["operation_id"], "UNKNOWN", operation["result"], operation["error"], ) operation = self.service.operation(operation["operation_id"]) operation = self.close_unknown(operation) if operation["status"] == "QUEUED": self.service.update( operation["operation_id"], "FAILED", error={ "code": "CONFLICT", "message": "Upload retry window closed", "request_id": operation["request_id"], "details": {"reason": "RETRY_WINDOW_CLOSED"}, }, ) if operation["status"] != "COMPLETED": plan["phase"] = "scheduled" self.save(job["id"], plan) return if operation["status"] == "COMPLETED": self.save(job["id"], None) # Extra retries are not job_runs and do not move the next plan time. self.db.execute( "UPDATE jobs SET last_result=? WHERE id=?", (operation["operation_id"], job["id"]), ) return if operation["status"] not in ("FAILED", "UNKNOWN"): return if plan["phase"] == "replacement": # A failed replacement gets its own bounded recovery window. self.db.execute( "UPDATE jobs SET last_result=? WHERE id=?", (operation["operation_id"], job["id"]), ) self.save(job["id"], None) return if plan["phase"] == "scheduled_running": operation = self.close_unknown(operation) if operation["status"] == "COMPLETED": return self.discard(plan["asset_id"]) payload = self.payload(config) key = "replacement-" + plan["root"] op_id = self.service.submit( "display", payload, config["device_id"], "scheduler-retry", key, key ) plan.update(phase="replacement", active=op_id) self.save(job["id"], plan) self.db.execute( "UPDATE jobs SET last_result=? WHERE id=?", (op_id, job["id"]) ) self.db.event( "upload.regenerated", { "job_id": job["id"], "operation_id": op_id, "failed_asset_id": plan["asset_id"], }, ) return if current >= deadline or plan["count"] >= len(plan["times"]): plan["phase"] = "scheduled" self.save(job["id"], plan) return if current < datetime.fromisoformat(plan["times"][plan["count"]]): return operation = self.close_unknown(operation) if operation["status"] == "COMPLETED": return number = plan["count"] + 1 key = f"upload-retry-{plan['root']}-{number}" op_id = self.service.submit( "display", self.reuse(self.payload(config), plan["asset_id"]), config["device_id"], "scheduler-retry", key, key, ) plan.update(active=op_id, count=number) self.save(job["id"], plan) self.db.execute("UPDATE jobs SET last_result=? WHERE id=?", (op_id, job["id"])) self.db.event( "upload.retry", { "job_id": job["id"], "operation_id": op_id, "attempt": number, "deadline": deadline.isoformat(), }, ) @staticmethod def payload(config): return { "device_id": config["device_id"], "content": config["source"], "options": { "duration": config["display_duration"], "restore_previous": config["restore_previous"], "priority": config["priority"], "cleanup_unstarred": config.get("cleanup_unstarred", False), }, }