From a843d4de9bc34a709a1261d61839e2bebd3b3054 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B8=A1=E5=8F=A3=E6=B5=AA=E4=BA=BA?= Date: Fri, 16 Oct 2026 12:25:16 +0800 Subject: [PATCH] Add documented KoKonna failure policy, bounded retries and durable recovery escalation [skip ci] --- docs/KOKONNA_RECOVERY.md | 61 +++++ docs/REAL_DEVICE_TEST.md | 4 + docs/RESEARCH.md | 5 +- docs/openapi.json | 163 ++++++++++++- pigway_frame_flow/api/app.py | 4 +- pigway_frame_flow/display/recovery.py | 73 ++++++ pigway_frame_flow/display/service.py | 36 ++- .../display/upload_reconciliation.py | 7 + pigway_frame_flow/drivers/kokonna/client.py | 222 ++++++++++++++---- pigway_frame_flow/extensions/webhooks.py | 1 + pigway_frame_flow/models.py | 14 +- pigway_frame_flow/storage/database.py | 1 + pigway_frame_flow/web/index.html | 2 +- pyproject.toml | 2 +- tests/schema-contract.json | 53 +++++ tests/test_client.py | 186 ++++++++++++++- tests/test_recovery.py | 109 +++++++++ 17 files changed, 873 insertions(+), 70 deletions(-) create mode 100644 docs/KOKONNA_RECOVERY.md create mode 100644 pigway_frame_flow/display/recovery.py create mode 100644 tests/test_recovery.py diff --git a/docs/KOKONNA_RECOVERY.md b/docs/KOKONNA_RECOVERY.md new file mode 100644 index 0000000..85a9ecf --- /dev/null +++ b/docs/KOKONNA_RECOVERY.md @@ -0,0 +1,61 @@ +# KoKonna 错误、重试与阻塞处理 + +核对日期：2026-10-16。依据：[KoKonna OpenAPI 官方文档](https://kokonna.art/pages/openapi)、[Requests 异常定义](https://requests.readthedocs.io/en/latest/_modules/requests/exceptions/)。本页区分厂商契约与 Frame Flow 的本地策略；没有文档保证的行为不作为自动重发依据。 + +## 官方接口契约 + +- 设备 API Key 使用 Bearer 鉴权；单设备每分钟上限 20 次，请求超限返回 429。 +- 上传成功返回新图片 ID 和计数，并立即激活新图；接口不提供幂等键或上传状态查询契约。 +- 删除可能改变当前图片。切换与删除都不能视为天然可重放。 +- 404 可表示方法不存在；图片下载时也可表示目标图片不存在。 +- 500 既可能是通用服务故障，也可能明确表示设备密钥无效、图片不存在或图片 ID 参数错误。实现只匹配官方明确列出的签名，其他 500 不猜测是否产生副作用。 +- 上传请求体上限 50 MB；本服务原图限制仍为 20 MiB，并保留像素上限和 base64 检查。 +- 官方文档未承诺服务处理时长、超时后的取消、事务回滚或 Retry-After 返回；不能靠等待固定时间证明写入没有执行。 +- Requests 将连接建立超时定义为可安全重试；读取超时不提供相同保证。一般连接异常也不能推定请求未发送。 + +## 已落实的本地策略 + +| 结果 | 处理 | 自动重试 | 设备阻塞 | +| --- | --- | --- | --- | +| 未配置密钥 | CONFIG_REQUIRED | 无 | 无 | +| 明确的设备密钥无效 | CONFIG_REQUIRED，保存固定错误分类 | 无 | 无 | +| 明确的图片不存在 | NOT_FOUND | 无 | 无 | +| 明确的图片 ID 参数错误 | INVALID_REQUEST | 无 | 无 | +| 明确的方法不存在 | UNSUPPORTED_CAPABILITY | 无 | 无 | +| 其他 HTTP 4xx 拒绝 | 保存状态码，结束失败 | 除 429 外无 | 无 | +| 本地请求预算耗尽 | 等待预算窗口 | 同一请求最多 3 次尝试 | 不创建 UNKNOWN | +| HTTP 429 | RATE_LIMITED，尊重服务端冷却时间 | 最多 3 次尝试 | 耗尽后失败，不创建 UNKNOWN | +| ConnectTimeout | 确定请求未发送 | 读写均最多 3 次尝试 | 耗尽后失败，不创建 UNKNOWN | +| 只读查询的临时网络错误、通用 5xx、无效 JSON | CLOUD_UNAVAILABLE | 最多 3 次尝试 | 不创建 UNKNOWN | +| TLS 异常 | 保存固定分类 | 无 | 写请求仍保守标为 UNKNOWN | +| 写请求读取超时、一般连接异常、通用 5xx | 无法确定写入结果 | 不重发，只读核对 | UNKNOWN，保留保护 | +| 写请求无效响应或无有效上传 ID | 无法确认成功 | 不重发，只读核对 | UNKNOWN，保留保护 | + +“最多 3 次”包含首次请求，重试发生在同一传输请求内，不重新渲染、不重新执行整个操作。普通退避为 1、2 秒；每次实际请求都计入本地 18 次/分钟预算。429 优先使用 Retry-After 秒数或 HTTP 日期；缺失/无效时按 60 秒。单次等待不超过 60 秒，超过此值保留冷却并结束当前尝试，不提前请求。没有无限循环，也不在重启后重放不确定的写入。 + +操作错误的 details 保存 attempts、retryable、write_outcome，以及 transport_error / http_status / failure_kind 等固定分类。不会保存异常原文、完整下载 URL、带密钥的厂商消息或响应正文。是否重试由结果类别决定，不能仅依据 HTTP 500。 + +## 未知写入的核对与升级 + +上传前记录资产、哈希、字节数、名称、上传前图库和时间。响应丢失后只读查找并校验唯一候选；图片相同且设备已同步才能记为成功。已证明上传、但设备持续停留原图时，沿用两次稳定观察的失败关闭规则。 + +没有候选、候选不唯一、内容不符、设备未同步、查询失败或缺少历史证据，都有独立恢复原因。图库没有目标图片不等于厂商保证从未接受请求，因此不自动把它改为失败或重传。 + +`GET /api/v1/operations/{operation_id}` 的 `recovery` 返回： + +- `state`：checking / manual_review / resolved。 +- `reason`：gallery_match_missing、gallery_match_ambiguous、content_mismatch、waiting_device_sync、target_mismatch、read_unavailable、reconciliation_error、missing_upload_evidence 或 manual_reconciliation_required。 +- `first_checked_at`、`last_checked_at`、`checks`、`requires_manual_review`。 + +只读核对每条操作至多每分钟一次。操作持续未知超过 sync_timeout + 300 秒（默认约 15 分钟）时，持久化一次 `operation.blocked` 事件和 `MANUAL_REVIEW_REQUIRED` 历史记录；服务重启、重复检查或原因改变均不会重复升级同一条未解决操作。既有检查继续运行，不自动取消设备保护。 + +管理员可在 Webhooks 中订阅 `operation.blocked`；未配置订阅不会自动向外发送通知。事件和 API 记录始终保留，但它们不能替代已配置的通知渠道。单个任务失败、单条未知操作的元数据异常均不应中断其他任务处理。 + +## 人工处理 + +1. 根据 operation_id 查看原始错误、恢复原因、当前设备和图库；不点击重新执行来猜测结果。 +2. 调用现有 reconcile 接口并传 `acknowledge=false` 进行核对。 +3. 无法证明成功、但管理员确认可以结束旧操作时，传 `acknowledge=true`：结束为失败，保留原始目标、上传意图和此前错误，记录观察到的图片与人工处理历史。该接口不会重发设备写入。 +4. 有目标一致且同步完成的证据才记为成功。处理后恢复原计划，错过的任务不补发。 + +仍可能因厂商无法给出明确写入结果而需要人工处理；本策略解决错误误判、可安全重试的暂时失败和长期阻塞缺乏解释的问题，不承诺网络故障时所有设备任务都能自动成功。 diff --git a/docs/REAL_DEVICE_TEST.md b/docs/REAL_DEVICE_TEST.md index 54a7d7a..6855a30 100644 --- a/docs/REAL_DEVICE_TEST.md +++ b/docs/REAL_DEVICE_TEST.md @@ -41,3 +41,7 @@ minute apart record `DISPLAY_NOT_CONFIRMED` and unblock future scheduled work. Temporary overrides retain their restore behavior and are not automatically closed as failed. Legacy operations without durable upload evidence require manual acknowledgement; absence from the gallery alone never proves failure. + +### 错误与有限重试 + +完整错误矩阵、重试边界、持久化恢复状态和人工处理见 [KoKonna 恢复策略](KOKONNA_RECOVERY.md)。自动重试只允许只读请求、连接建立超时或明确限流拒绝；读取超时等未知写入不会重发。 diff --git a/docs/RESEARCH.md b/docs/RESEARCH.md index 22c54eb..434b7fc 100644 --- a/docs/RESEARCH.md +++ b/docs/RESEARCH.md @@ -1,6 +1,7 @@ # KoKonna API research / 接口调查 -Checked 2026-10-01 against https://kokonna.art/pages/openapi. +Rechecked 2026-10-16 against https://kokonna.art/pages/openapi. +错误分类、重试和未知写入处理见 [KoKonna 恢复策略](KOKONNA_RECOVERY.md)。 Base: https://api.kokonna.art/openapi. No fallback to legacy galaxyguide.cn. | Official endpoint | Frame Flow capability | @@ -29,7 +30,7 @@ headroom at 18. Device state has a 60-second cache, gallery is loaded on demand. Uploads automatically activate a new image; there is no upload-without-activation flag. Gallery-only uploads are therefore local staging, not a fictitious cloud operation. Maximum upstream request size is 50 MB; this service accepts 20 MiB -image files with bounded decoded pixels. No non-idempotent cloud write is retried. +image files with bounded decoded pixels. Only proven-unsent connections or explicit rate-limit rejections may retry a write, within a three-attempt budget. Ambiguous writes are never replayed. Cloud acknowledgement is not physical refresh confirmation; /device synced and imageId drive the operation state. Unconfirmed or interrupted writes become UNKNOWN and block later device writes until read-only reconciliation/explicit acknowledgement. diff --git a/docs/openapi.json b/docs/openapi.json index f5cf9da..9f02bd9 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -2,7 +2,7 @@ "openapi": "3.1.0", "info": { "title": "Pigway Frame Flow API", - "version": "0.1.0" + "version": "0.1.4" }, "paths": { "/api/v1/devices/{device_id}/widgets/selection": { @@ -6397,7 +6397,15 @@ "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/TemplateContent" + "anyOf": [ + { + "$ref": "#/components/schemas/TemplateContent" + }, + { + "$ref": "#/components/schemas/AutoCollageContent" + } + ], + "title": "Payload" } } } @@ -14298,6 +14306,87 @@ ], "title": "AssetBatch" }, + "AutoCollageContent": { + "properties": { + "type": { + "type": "string", + "const": "auto_collage", + "title": "Type" + }, + "source_id": { + "anyOf": [ + { + "type": "string", + "pattern": "^[a-z0-9-]{1,60}$" + }, + { + "type": "null" + } + ], + "title": "Source Id" + }, + "count": { + "type": "integer", + "maximum": 12.0, + "minimum": 1.0, + "title": "Count", + "default": 4 + }, + "title": { + "type": "string", + "maxLength": 100, + "title": "Title", + "default": "" + }, + "message": { + "type": "string", + "maxLength": 500, + "title": "Message", + "default": "" + }, + "orientation": { + "type": "string", + "enum": [ + "auto", + "portrait", + "landscape" + ], + "title": "Orientation", + "default": "auto" + }, + "style": { + "type": "string", + "enum": [ + "grid", + "hero", + "poster" + ], + "title": "Style", + "default": "grid" + }, + "use_ai": { + "type": "boolean", + "title": "Use Ai", + "default": true + }, + "prompt": { + "type": "string", + "maxLength": 1500, + "title": "Prompt", + "default": "" + }, + "profile": { + "$ref": "#/components/schemas/Profile", + "default": "color" + } + }, + "additionalProperties": false, + "type": "object", + "required": [ + "type" + ], + "title": "AutoCollageContent" + }, "Body_import_images_api_v1_layout_images_post": { "properties": { "files": { @@ -14778,6 +14867,9 @@ { "$ref": "#/components/schemas/PhotoSourceContent" }, + { + "$ref": "#/components/schemas/AutoCollageContent" + }, { "$ref": "#/components/schemas/ProviderContent" }, @@ -14789,6 +14881,7 @@ "discriminator": { "propertyName": "type", "mapping": { + "auto_collage": "#/components/schemas/AutoCollageContent", "html": "#/components/schemas/FutureContent", "image": "#/components/schemas/ImageContent", "image_url": "#/components/schemas/URLContent", @@ -14852,6 +14945,7 @@ "UNSUPPORTED_CAPABILITY", "RENDER_FAILED", "UPLOAD_FAILED", + "DISPLAY_NOT_CONFIRMED", "WRITE_RESULT_UNKNOWN", "RATE_LIMITED", "INTERNAL_ERROR", @@ -15062,7 +15156,7 @@ "version": { "type": "string", "title": "Version", - "default": "0.1.0" + "default": "0.1.4" } }, "type": "object", @@ -15324,6 +15418,9 @@ { "$ref": "#/components/schemas/PhotoSourceContent" }, + { + "$ref": "#/components/schemas/AutoCollageContent" + }, { "$ref": "#/components/schemas/ProviderContent" }, @@ -15334,6 +15431,7 @@ "discriminator": { "propertyName": "type", "mapping": { + "auto_collage": "#/components/schemas/AutoCollageContent", "html": "#/components/schemas/FutureContent", "image": "#/components/schemas/ImageContent", "image_url": "#/components/schemas/URLContent", @@ -15469,6 +15567,9 @@ { "$ref": "#/components/schemas/PhotoSourceContent" }, + { + "$ref": "#/components/schemas/AutoCollageContent" + }, { "$ref": "#/components/schemas/ProviderContent" }, @@ -15479,6 +15580,7 @@ "discriminator": { "propertyName": "type", "mapping": { + "auto_collage": "#/components/schemas/AutoCollageContent", "html": "#/components/schemas/FutureContent", "image": "#/components/schemas/ImageContent", "image_url": "#/components/schemas/URLContent", @@ -15787,6 +15889,16 @@ }, "Operation": { "properties": { + "recovery": { + "anyOf": [ + { + "$ref": "#/components/schemas/RecoveryState" + }, + { + "type": "null" + } + ] + }, "operation_id": { "type": "string", "title": "Operation Id" @@ -16808,6 +16920,49 @@ ], "title": "Reconciliation" }, + "RecoveryState": { + "properties": { + "state": { + "type": "string", + "enum": [ + "checking", + "manual_review", + "resolved" + ], + "title": "State" + }, + "reason": { + "type": "string", + "title": "Reason" + }, + "first_checked_at": { + "type": "string", + "title": "First Checked At" + }, + "last_checked_at": { + "type": "string", + "title": "Last Checked At" + }, + "checks": { + "type": "integer", + "title": "Checks" + }, + "requires_manual_review": { + "type": "boolean", + "title": "Requires Manual Review" + } + }, + "type": "object", + "required": [ + "state", + "reason", + "first_checked_at", + "last_checked_at", + "checks", + "requires_manual_review" + ], + "title": "RecoveryState" + }, "Region": { "properties": { "asset_id": { @@ -17178,7 +17333,7 @@ "version": { "type": "string", "title": "Version", - "default": "0.1.0" + "default": "0.1.4" }, "api_version": { "type": "string", diff --git a/pigway_frame_flow/api/app.py b/pigway_frame_flow/api/app.py index f1e3924..5abfa76 100644 --- a/pigway_frame_flow/api/app.py +++ b/pigway_frame_flow/api/app.py @@ -124,7 +124,7 @@ def create_app(settings=None, start_worker=True): app = FastAPI( title="Pigway Frame Flow API", docs_url=None, - version="0.1.3", + version="0.1.4", lifespan=lifespan, responses=error_responses, ) @@ -1000,7 +1000,7 @@ def create_app(settings=None, start_worker=True): ) def settings_view(): return SettingsView( - version="0.1.3", + version="0.1.4", mock=settings.mock, write_enabled=settings.write_enabled, api_key_configured=bool(settings.api_key) diff --git a/pigway_frame_flow/display/recovery.py b/pigway_frame_flow/display/recovery.py new file mode 100644 index 0000000..b71dcbf --- /dev/null +++ b/pigway_frame_flow/display/recovery.py @@ -0,0 +1,73 @@ +"""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 diff --git a/pigway_frame_flow/display/service.py b/pigway_frame_flow/display/service.py index b55fe46..85d9344 100644 --- a/pigway_frame_flow/display/service.py +++ b/pigway_frame_flow/display/service.py @@ -6,6 +6,7 @@ from datetime import UTC, datetime, timedelta from uuid import uuid4 from pigway_frame_flow.display.cloud_library import CloudLibrary +from pigway_frame_flow.display.recovery import RecoveryMonitor 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 @@ -50,6 +51,7 @@ class DisplayService: self.lock = threading.RLock() self.retention = Retention(self) self.cloud_library = CloudLibrary(self) + self.recovery = RecoveryMonitor(self) self.upload_reconciliation = UploadReconciliation(self) self.expiration_checked: dict[str, float] = {} self.unknown_checked: dict[str, float] = {} @@ -165,6 +167,7 @@ class DisplayService: request_id=op["request_id"], client_id=op["client_id"], action=op["action"], + recovery=self.recovery.info(id), ) def update(self, id, status, result=None, error=None): @@ -184,6 +187,8 @@ class DisplayService: id, ), ) + if status in ("COMPLETED", "FAILED"): + self.recovery.resolve(id) def blocked(self, device_id): return ( @@ -702,21 +707,34 @@ class DisplayService: # 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" + "SELECT * FROM operations WHERE status='UNKNOWN' AND action!='render-ai' AND write_started=1" ): expected = json.loads(op["result"] or "{}").get("image_id") 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 + if op["action"] != "display" or (not expected and not intent): + self.recovery.observe( + op, + "missing_upload_evidence" + if op["action"] == "display" + else "manual_reconciliation_required", + ) + continue try: driver = self.registry.get(op["device_id"]) device = driver.device(refresh=True) if intent: expected = self.upload_reconciliation.check(op, driver, device) + if ( + self.db.one( + "SELECT status FROM operations WHERE id=?", (op["id"],) + )["status"] + == "FAILED" + ): + continue if ( expected and device.current_content.id == expected @@ -727,8 +745,18 @@ class DisplayService: self.db.history(op, "COMPLETED") self.db.event("operation.completed", {"operation_id": op["id"]}) self.unknown_checked.pop(op["id"], None) + elif expected: + self.recovery.observe( + op, + "waiting_device_sync" + if device.current_content.id == expected + else "target_mismatch", + ) except FlowError: - pass + self.recovery.observe(op, "read_unavailable") + except Exception: + # Bad vendor metadata for one operation must not stop the worker. + self.recovery.observe(op, "reconciliation_error") for op in self.db.all("SELECT * FROM operations WHERE status='SYNCING'"): try: device = self.registry.get(op["device_id"]).device() diff --git a/pigway_frame_flow/display/upload_reconciliation.py b/pigway_frame_flow/display/upload_reconciliation.py index 2a770f5..eb86423 100644 --- a/pigway_frame_flow/display/upload_reconciliation.py +++ b/pigway_frame_flow/display/upload_reconciliation.py @@ -59,6 +59,12 @@ class UploadReconciliation: candidates.append(image) # Ambiguous matches never release a write lock. if len(candidates) != 1: + service.recovery.observe( + op, + "gallery_match_missing" + if not candidates + else "gallery_match_ambiguous", + ) return None expected = candidates[0].image_id data = driver.download(expected) @@ -66,6 +72,7 @@ class UploadReconciliation: len(data) != intent["size"] or hashlib.sha256(data).hexdigest() != intent["sha256"] ): + service.recovery.observe(op, "content_mismatch") return None result["image_id"] = expected result["upload_confirmed"] = True diff --git a/pigway_frame_flow/drivers/kokonna/client.py b/pigway_frame_flow/drivers/kokonna/client.py index 1e7c3d6..36f4d06 100644 --- a/pigway_frame_flow/drivers/kokonna/client.py +++ b/pigway_frame_flow/drivers/kokonna/client.py @@ -1,7 +1,7 @@ """Hardened adapter over the MIT-0 guooscar 1.2.0 device/upload methods. No UI, scheduler, renderer, environment lookup, or constructor cloud calls. -Never retry a write; a timeout may have occurred after cloud acceptance. +Retry only read-only requests or writes proven unsent/rejected; never replay ambiguous writes. """ import threading @@ -19,28 +19,85 @@ BASE_URL = "https://api.kokonna.art/openapi" class SafeTransport: + MAX_ATTEMPTS = 3 + MAX_RETRY_DELAY = 60 + def __init__(self, key, session=None): self.key = key self.session = session or requests.Session() self.calls: deque[float] = deque() self.lock = threading.Lock() + self.cooldown_until = 0.0 def request(self, endpoint, payload=None, write=False, binary=False): - if not self.key: - raise FlowError(E.CONFIG_REQUIRED, "KoKonna configuration required", 503) + for attempt in range(1, self.MAX_ATTEMPTS + 1): + try: + return self._request_once(endpoint, payload, write, binary) + except FlowError as exc: + exc.details["attempts"] = attempt + delay = max( + 2 ** (attempt - 1), exc.details.get("retry_after_seconds", 0) + ) + if ( + not exc.details.get("retryable") + or attempt == self.MAX_ATTEMPTS + or delay > self.MAX_RETRY_DELAY + ): + if exc.details.get("retryable"): + exc.details["retry_exhausted"] = attempt == self.MAX_ATTEMPTS + exc.details["retry_deferred"] = delay > self.MAX_RETRY_DELAY + raise + time.sleep(delay) + raise AssertionError("Retry loop exhausted") + + def _reserve(self): with self.lock: current = time.monotonic() while self.calls and current - self.calls[0] >= 60: self.calls.popleft() - # Reserve capacity beneath the official 20/minute ceiling. + delay = max(0.0, self.cooldown_until - current) if len(self.calls) >= 18: - raise FlowError(E.RATE_LIMITED, "Device request budget exhausted", 429) + delay = max(delay, 60 - (current - self.calls[0])) + if delay: + raise FlowError( + E.RATE_LIMITED, + "Device request budget exhausted", + 429, + details={ + "retryable": True, + "retry_after_seconds": delay, + "write_outcome": "not_sent", + "failure_kind": "local_rate_limit", + }, + ) self.calls.append(current) - url = BASE_URL + endpoint + + @staticmethod + def _retry_after(value): + from datetime import UTC, datetime + from email.utils import parsedate_to_datetime + + try: + seconds = float(value) + except (ValueError, TypeError): + try: + seconds = ( + parsedate_to_datetime(value) - datetime.now(UTC) + ).total_seconds() + except (ValueError, TypeError, OverflowError): + return 60.0 + import math + + return max(0.0, seconds) if math.isfinite(seconds) else 60.0 + + def _request_once(self, endpoint, payload, write, binary): + if not self.key: + raise FlowError(E.CONFIG_REQUIRED, "KoKonna configuration required", 503) + self._reserve() try: response = self.session.request( "GET" if binary else "POST", - url, + BASE_URL + endpoint, headers={} if binary else { @@ -52,8 +109,7 @@ class SafeTransport: allow_redirects=False, ) except requests.RequestException as exc: - # Persist only fixed categories: exception text and download URLs may - # contain credentials. Unknown writes must still never be replayed. + # Never retain exception text or a URL: binary URLs contain the key. category = next( ( label @@ -68,50 +124,124 @@ class SafeTransport: ), "request_error", ) + not_sent = isinstance(exc, requests.ConnectTimeout) + uncertain = write and not not_sent raise FlowError( - E.WRITE_RESULT_UNKNOWN if write else E.CLOUD_UNAVAILABLE, + E.WRITE_RESULT_UNKNOWN if uncertain else E.CLOUD_UNAVAILABLE, "Cloud write outcome is unknown" - if write + if uncertain else "Device cloud is unavailable", 502, - details={"transport_error": category}, + details={ + "transport_error": category, + "retryable": (not write and category != "tls_error") or not_sent, + "write_outcome": "unknown" + if uncertain + else ("not_sent" if write else "read_only"), + }, ) from None - if response.status_code != 200: - code = ( - E.RATE_LIMITED - if response.status_code == 429 - else ( - E.WRITE_RESULT_UNKNOWN - if write and response.status_code >= 500 - else E.CLOUD_UNAVAILABLE - ) - ) - raise FlowError( - code, - "Device cloud request failed", - 429 if code == E.RATE_LIMITED else 502, - details={"http_status": response.status_code}, - ) - if binary: - if len(response.content) > 20 * 1024 * 1024: - raise FlowError(E.INVALID_REQUEST, "Cloud image exceeds size limit") - return response.content try: - value = response.json() - if not isinstance(value, dict): - raise ValueError() - # Cloud errors may echo credentials: never return raw messages. - if value.get("success") is False or ( - "message" in value and "id" not in value + value = None + if not binary or response.status_code != 200: + try: + value = response.json() + except ValueError: + pass + message = value.get("message") if isinstance(value, dict) else None + status = response.status_code + # Only exact documented business failures establish rejection. Generic + # 5xx responses may arrive after a write; those remain UNKNOWN. + business = None + if status in (404, 500) and isinstance(message, str): + if message == "image not found": + business = ( + E.NOT_FOUND, + "Cloud image not found", + 404, + "image_not_found", + ) + elif status == 500 and message == "imageId is empty": + business = ( + E.INVALID_REQUEST, + "Cloud image ID is invalid", + 400, + "invalid_image_id", + ) + elif status == 500 and message.startswith("can not find robot "): + business = ( + E.CONFIG_REQUIRED, + "Device key does not identify a frame", + 503, + "device_key_invalid", + ) + elif status == 404 and message.startswith("can not find method "): + business = ( + E.UNSUPPORTED_CAPABILITY, + "Cloud method not found", + 422, + "method_not_found", + ) + if business: + code, safe_message, local_status, kind = business + raise FlowError( + code, + safe_message, + local_status, + details={ + "http_status": status, + "failure_kind": kind, + "retryable": False, + "write_outcome": "rejected" if write else "read_only", + }, + ) + if status != 200: + limited = status == 429 + rejected = 400 <= status < 500 + uncertain = write and not rejected + details = { + "http_status": status, + "retryable": limited or (not write and status >= 500), + "write_outcome": "unknown" + if uncertain + else ("rejected" if write else "read_only"), + } + if limited: + delay = self._retry_after(response.headers.get("Retry-After")) + details["retry_after_seconds"] = delay + with self.lock: + self.cooldown_until = max( + self.cooldown_until, time.monotonic() + delay + ) + raise FlowError( + E.RATE_LIMITED + if limited + else (E.WRITE_RESULT_UNKNOWN if uncertain else E.CLOUD_UNAVAILABLE), + "Device cloud request failed", + 429 if limited else 502, + details=details, + ) + if binary: + if len(response.content) > 20 * 1024 * 1024: + raise FlowError(E.INVALID_REQUEST, "Cloud image exceeds size limit") + return response.content + if ( + not isinstance(value, dict) + or value.get("success") is False + or ("message" in value and "id" not in value) ): - raise ValueError() + raise FlowError( + E.WRITE_RESULT_UNKNOWN if write else E.CLOUD_UNAVAILABLE, + "Invalid device cloud response", + 502, + details={ + "failure_kind": "invalid_response", + "retryable": not write, + "write_outcome": "unknown" if write else "read_only", + }, + ) return value - except ValueError: - raise FlowError( - E.WRITE_RESULT_UNKNOWN if write else E.CLOUD_UNAVAILABLE, - "Invalid device cloud response", - 502, - ) from None + finally: + response.close() class KokonnaClient: @@ -136,7 +266,7 @@ class KokonnaClient: def upload(self, path, name): self.guard() value = guooscar_protocol.upload(self.transport, path.read_bytes(), name) - if not isinstance(value.get("id"), int) or value["id"] <= 0: + if type(value.get("id")) is not int or value["id"] <= 0: raise FlowError(E.WRITE_RESULT_UNKNOWN, "Upload ID not confirmed", 502) return value["id"] diff --git a/pigway_frame_flow/extensions/webhooks.py b/pigway_frame_flow/extensions/webhooks.py index ff9c44f..a6cfe4b 100644 --- a/pigway_frame_flow/extensions/webhooks.py +++ b/pigway_frame_flow/extensions/webhooks.py @@ -138,6 +138,7 @@ def install_webhooks(app, service): allowed = { "operation.completed", "operation.failed", + "operation.blocked", "display.changed", "device.online", "device.offline", diff --git a/pigway_frame_flow/models.py b/pigway_frame_flow/models.py index cd0647a..e1e8883 100644 --- a/pigway_frame_flow/models.py +++ b/pigway_frame_flow/models.py @@ -177,7 +177,17 @@ class OperationResult(BaseModel): deferred: bool = False +class RecoveryState(BaseModel): + state: Literal["checking", "manual_review", "resolved"] + reason: str + first_checked_at: str + last_checked_at: str + checks: int + requires_manual_review: bool + + class Operation(BaseModel): + recovery: RecoveryState | None = None operation_id: str status: OperationStatus device_id: str @@ -336,7 +346,7 @@ class SettingsView(BaseModel): class Health(BaseModel): status: str - version: str = "0.1.3" + version: str = "0.1.4" class Readiness(BaseModel): @@ -366,7 +376,7 @@ class RestoreRequest(StrictModel): class Version(BaseModel): product: str = "Pigway Frame Flow" - version: str = "0.1.3" + version: str = "0.1.4" api_version: str = "v1" commit: str = "development" build_time: str = "unknown" diff --git a/pigway_frame_flow/storage/database.py b/pigway_frame_flow/storage/database.py index 2cc61de..240bb5f 100644 --- a/pigway_frame_flow/storage/database.py +++ b/pigway_frame_flow/storage/database.py @@ -31,6 +31,7 @@ class Database: CREATE TABLE IF NOT EXISTS image_sources (device_id TEXT, image_id INTEGER, asset_id TEXT, operation_id TEXT, payload TEXT NOT NULL, PRIMARY KEY(device_id,image_id)); CREATE TABLE IF NOT EXISTS app_config (key TEXT PRIMARY KEY, payload TEXT NOT NULL); CREATE TABLE IF NOT EXISTS playlists (id TEXT PRIMARY KEY, payload TEXT, position INTEGER, next_run TEXT); + CREATE TABLE IF NOT EXISTS operation_recovery (operation_id TEXT PRIMARY KEY, state TEXT NOT NULL, reason TEXT NOT NULL, first_checked_at TEXT NOT NULL, last_checked_at TEXT NOT NULL, checks INTEGER NOT NULL); CREATE TABLE IF NOT EXISTS operations ( id TEXT PRIMARY KEY, status TEXT NOT NULL, action TEXT NOT NULL, payload TEXT NOT NULL, device_id TEXT NOT NULL, client_id TEXT NOT NULL, diff --git a/pigway_frame_flow/web/index.html b/pigway_frame_flow/web/index.html index 8f9698e..5ec3c81 100644 --- a/pigway_frame_flow/web/index.html +++ b/pigway_frame_flow/web/index.html @@ -1,3 +1,3 @@ Pigway Frame Flow -
PIGWAY FRAME FLOW

 +
PIGWAY FRAME FLOW

 diff --git a/pyproject.toml b/pyproject.toml index 0a4a51d..3d04f8d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pigway-frame-flow" -version = "0.1.3" +version = "0.1.4" requires-python = ">=3.12" dependencies = ["paho-mqtt==2.1.0", "python-dateutil==2.9.0.post0", "fastapi==0.142.2", "uvicorn==0.54.0", "pillow==12.3.0", "requests==2.34.2", "croniter==6.2.4", "python-multipart==0.0.32", "pyyaml==6.0.3"] [project.optional-dependencies] diff --git a/tests/schema-contract.json b/tests/schema-contract.json index fbcf436..42857a8 100644 --- a/tests/schema-contract.json +++ b/tests/schema-contract.json @@ -613,6 +613,16 @@ }, "Operation": { "properties": { + "recovery": { + "anyOf": [ + { + "$ref": "#/components/schemas/RecoveryState" + }, + { + "type": "null" + } + ] + }, "operation_id": { "type": "string", "title": "Operation Id" @@ -1422,5 +1432,48 @@ "type" ], "title": "AutoCollageContent" + }, + "RecoveryState": { + "properties": { + "state": { + "type": "string", + "enum": [ + "checking", + "manual_review", + "resolved" + ], + "title": "State" + }, + "reason": { + "type": "string", + "title": "Reason" + }, + "first_checked_at": { + "type": "string", + "title": "First Checked At" + }, + "last_checked_at": { + "type": "string", + "title": "Last Checked At" + }, + "checks": { + "type": "integer", + "title": "Checks" + }, + "requires_manual_review": { + "type": "boolean", + "title": "Requires Manual Review" + } + }, + "type": "object", + "required": [ + "state", + "reason", + "first_checked_at", + "last_checked_at", + "checks", + "requires_manual_review" + ], + "title": "RecoveryState" } } diff --git a/tests/test_client.py b/tests/test_client.py index 4f16722..50144bf 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,3 +1,4 @@ +from types import SimpleNamespace from unittest.mock import Mock import pytest @@ -7,11 +8,28 @@ from pigway_frame_flow.drivers.kokonna.client import KokonnaClient from pigway_frame_flow.errors import FlowError +@pytest.fixture(autouse=True) +def transport_clock(monkeypatch): + import pigway_frame_flow.drivers.kokonna.client as module + + clock = SimpleNamespace(now=0.0, waits=[]) + + def sleep(seconds): + clock.waits.append(seconds) + clock.now += seconds + + monkeypatch.setattr( + module, "time", SimpleNamespace(monotonic=lambda: clock.now, sleep=sleep) + ) + return clock + + def session(body=None, status=200): s = Mock() s.request.return_value.status_code = status s.request.return_value.json.return_value = body or {} s.request.return_value.content = b"image" + s.request.return_value.headers = {} return s @@ -88,7 +106,7 @@ def test_timeout_redaction_no_retry(write): c.display(1) if write else c.get_device() assert "secret" not in str(e.value) assert e.value.code == ("WRITE_RESULT_UNKNOWN" if write else "CLOUD_UNAVAILABLE") - assert s.request.call_count == 1 + assert s.request.call_count == (1 if write else 3) @pytest.mark.parametrize("status", [401, 429, 500, 530]) @@ -109,15 +127,14 @@ def test_invalid_upload_id(tmp_path): assert e.value.code == "WRITE_RESULT_UNKNOWN" -def test_rate_budget(): +def test_rate_budget(transport_clock): s = session({"nickname": "x"}) c = KokonnaClient("secret", session=s) for _ in range(18): c.get_device() - with pytest.raises(FlowError) as e: - c.get_device() - assert e.value.code == "RATE_LIMITED" - assert s.request.call_count == 18 + c.get_device() + assert s.request.call_count == 19 + assert transport_clock.waits == [60] @pytest.mark.parametrize( @@ -137,7 +154,160 @@ def test_transport_failure_records_safe_category_without_replay(exception, categ c = KokonnaClient("secret", True, s) with pytest.raises(FlowError) as error: c.display(1) - assert error.value.code == "WRITE_RESULT_UNKNOWN" - assert error.value.details == {"transport_error": category} + assert error.value.code == ( + "CLOUD_UNAVAILABLE" if category == "connect_timeout" else "WRITE_RESULT_UNKNOWN" + ) + assert error.value.details["transport_error"] == category + assert error.value.details["attempts"] == ( + 3 if category == "connect_timeout" else 1 + ) assert "secret" not in str(error.value.details) + assert s.request.call_count == (3 if category == "connect_timeout" else 1) + + +@pytest.mark.parametrize("write", [False, True]) +def test_connect_timeout_retries_then_succeeds(write, transport_clock): + s = session({"success": True}) + response = s.request.return_value + s.request.side_effect = [requests.ConnectTimeout("private"), response] + c = KokonnaClient("private", True, s) + c.display(1) if write else c.get_device() + assert s.request.call_count == 2 + assert transport_clock.waits == [1] + + +@pytest.mark.parametrize( + "message,code,kind", + [ + ("image not found", "NOT_FOUND", "image_not_found"), + ("imageId is empty", "INVALID_REQUEST", "invalid_image_id"), + ("can not find robot private", "CONFIG_REQUIRED", "device_key_invalid"), + ], +) +def test_documented_500_rejection_does_not_lock_or_retry(message, code, kind): + s = session({"message": message}, 500) + c = KokonnaClient("private", True, s) + with pytest.raises(FlowError) as error: + c.display(1) + assert error.value.code == code + assert error.value.details["write_outcome"] == "rejected" + assert error.value.details["failure_kind"] == kind + assert "private" not in str(error.value.details) assert s.request.call_count == 1 + + +@pytest.mark.parametrize("status", [500, 502, 503, 530]) +def test_generic_5xx_read_retries_but_write_never_replays(status): + for write in [False, True]: + s = session({"message": "unknown failure private"}, status) + c = KokonnaClient("private", True, s) + with pytest.raises(FlowError) as error: + c.display(1) if write else c.get_device() + assert s.request.call_count == (1 if write else 3) + assert error.value.code == ( + "WRITE_RESULT_UNKNOWN" if write else "CLOUD_UNAVAILABLE" + ) + assert "private" not in str(error.value.details) + + +def test_rate_limit_honors_retry_after_and_closes_response(transport_clock): + s = session({"message": "rate limited"}, 429) + first = s.request.return_value + first.headers = {"Retry-After": "7"} + second = session({"success": True}).request.return_value + s.request.side_effect = [first, second] + KokonnaClient("private", True, s).display(1) + assert s.request.call_count == 2 + assert transport_clock.waits == [7] + first.close.assert_called_once() + second.close.assert_called_once() + + +def test_long_retry_after_preserves_cooldown_without_early_request(transport_clock): + s = session({"message": "rate limited"}, 429) + s.request.return_value.headers = {"Retry-After": "120"} + c = KokonnaClient("private", True, s) + for _ in range(2): + with pytest.raises(FlowError) as error: + c.display(1) + assert error.value.code == "RATE_LIMITED" + assert s.request.call_count == 1 + assert transport_clock.waits == [] + + +@pytest.mark.parametrize("header", [None, "bad", "nan", "inf"]) +def test_invalid_retry_after_defaults_to_one_minute(header): + from pigway_frame_flow.drivers.kokonna.client import SafeTransport + + assert SafeTransport._retry_after(header) == 60 + + +def test_retry_after_accepts_http_date(): + from datetime import UTC, datetime, timedelta + from email.utils import format_datetime + + from pigway_frame_flow.drivers.kokonna.client import SafeTransport + + future = format_datetime(datetime.now(UTC) + timedelta(seconds=30), usegmt=True) + assert 28 <= SafeTransport._retry_after(future) <= 30 + + +def test_invalid_write_response_is_never_retried(): + s = session({"success": False, "message": "private"}) + with pytest.raises(FlowError) as error: + KokonnaClient("private", True, s).display(1) + assert error.value.code == "WRITE_RESULT_UNKNOWN" + assert error.value.details["failure_kind"] == "invalid_response" + assert s.request.call_count == 1 + + +@pytest.mark.parametrize( + "failure,expected", + [ + ("connect_timeout", "CLOUD_UNAVAILABLE"), + ("business_rejection", "CONFIG_REQUIRED"), + ("rate_limit", "RATE_LIMITED"), + ], +) +def test_proven_upload_failures_end_failed_without_device_lock( + service, monkeypatch, failure, expected +): + from test_display import submit + + s = session({"message": "can not find robot private"}, 500) + if failure == "rate_limit": + s.request.return_value.status_code = 429 + if failure == "connect_timeout": + s.request.side_effect = requests.ConnectTimeout("private") + c = KokonnaClient("private", True, s) + monkeypatch.setattr(service.registry.get("living-room"), "upload", c.upload) + operation = submit(service) + service.process_one() + op = service.operation(operation) + assert op["status"] == "FAILED" + assert op["error"]["code"] == expected + assert not service.blocked("living-room") + + +@pytest.mark.parametrize("value", [True, False, 0, -1, "12", None]) +def test_invalid_upload_id_never_becomes_confirmed(tmp_path, value): + path = tmp_path / "image.jpg" + path.write_bytes(b"jpeg") + s = session({"id": value}) + with pytest.raises(FlowError) as error: + KokonnaClient("private", True, s).upload(path, "image.jpg") + assert error.value.code == "WRITE_RESULT_UNKNOWN" + assert s.request.call_count == 1 + + +def test_ambiguous_failure_after_safe_rejection_stops_retrying(transport_clock): + s = session({"message": "limited"}, 429) + first = s.request.return_value + first.headers = {"Retry-After": "1"} + s.request.side_effect = [first, requests.ReadTimeout("private")] + with pytest.raises(FlowError) as error: + KokonnaClient("private", True, s).display(1) + assert error.value.code == "WRITE_RESULT_UNKNOWN" + assert error.value.details["attempts"] == 2 + assert s.request.call_count == 2 + assert transport_clock.waits == [1] diff --git a/tests/test_recovery.py b/tests/test_recovery.py new file mode 100644 index 0000000..f4d8417 --- /dev/null +++ b/tests/test_recovery.py @@ -0,0 +1,109 @@ +from datetime import UTC, datetime, timedelta + +from test_display import submit + +from pigway_frame_flow.display.service import DisplayService +from pigway_frame_flow.errors import ErrorCode, FlowError + + +def unknown_upload(service, monkeypatch): + def fail(*args): + raise FlowError(ErrorCode.WRITE_RESULT_UNKNOWN, "Response lost", 502) + + monkeypatch.setattr(service.registry.get("living-room"), "upload", fail) + operation = submit(service) + service.process_one() + return operation + + +def test_missing_upload_is_visible_without_unsafe_unlock(client, service, monkeypatch): + operation = unknown_upload(service, monkeypatch) + service.sync() + recovery = client.get("/api/v1/operations/" + operation).json()["recovery"] + assert recovery["reason"] == "gallery_match_missing" + assert recovery["state"] == "checking" + assert not recovery["requires_manual_review"] + assert service.blocked("living-room") + assert service.registry.get("living-room").client.uploads == 0 + + +def test_persistent_block_escalates_once_survives_restart_and_preserves_evidence( + service, monkeypatch +): + operation = unknown_upload(service, monkeypatch) + original = service.operation(operation) + service.db.execute( + "UPDATE operations SET created_at=? WHERE id=?", + ( + (datetime.now(UTC) - timedelta(hours=1)).isoformat(), + operation, + ), + ) + service.sync() + first = service.operation(operation)["recovery"] + assert first["state"] == "manual_review" + assert first["reason"] == "gallery_match_missing" + service.unknown_checked.clear() + service.sync() + restarted = DisplayService( + service.db, service.settings, service.registry, service.assets, service.renderer + ) + restarted.sync() + assert ( + len(service.db.all("SELECT * FROM events WHERE event_type='operation.blocked'")) + == 1 + ) + assert ( + service.db.one( + "SELECT COUNT(*) AS n FROM history WHERE result='MANUAL_REVIEW_REQUIRED'" + )["n"] + == 1 + ) + assert service.operation(operation)["recovery"]["checks"] == 3 + assert service.operation(operation)["result"] == original["result"] + service.reconcile(operation, acknowledge=True) + assert service.operation(operation)["recovery"]["state"] == "resolved" + assert not service.operation(operation)["recovery"]["requires_manual_review"] + assert not service.blocked("living-room") + + +def test_missing_legacy_evidence_escalates_without_cloud_write(service): + operation = submit(service) + service._start_write(operation) + service.update(operation, "UNKNOWN") + service.sync() + assert ( + service.operation(operation)["recovery"]["reason"] == "missing_upload_evidence" + ) + assert service.blocked("living-room") + + +def test_delete_unknown_requires_manual_review_without_replay(service): + operation = service.submit( + "delete", {"image_id": 1}, "living-room", "admin", "test" + ) + service._start_write(operation) + service.update(operation, "UNKNOWN") + service.sync() + assert ( + service.operation(operation)["recovery"]["reason"] + == "manual_reconciliation_required" + ) + assert service.blocked("living-room") + assert service.registry.get("living-room").images()[0].image_id == 1 + + +def test_malformed_reconciliation_does_not_abort_other_background_work( + service, monkeypatch +): + operation = unknown_upload(service, monkeypatch) + + def invalid_metadata(*args): + raise ValueError("vendor metadata with private data") + + monkeypatch.setattr(service.upload_reconciliation, "check", invalid_metadata) + service.sync() + op = service.operation(operation) + assert op["recovery"]["reason"] == "reconciliation_error" + assert "private" not in str(op["recovery"]) + assert service.blocked("living-room")