diff --git a/deploy/Dockerfile.runtime-update b/deploy/Dockerfile.runtime-update
index 7b46d9a..8dc7a7e 100644
--- a/deploy/Dockerfile.runtime-update
+++ b/deploy/Dockerfile.runtime-update
@@ -29,4 +29,4 @@ CMD ["python", "-m", "uvicorn", "pigway_frame_flow.main:app", "--host", "0.0.0.0
ARG BUILD_COMMIT
ARG BUILD_TIME
ENV BUILD_COMMIT=$BUILD_COMMIT BUILD_TIME=$BUILD_TIME
-LABEL org.opencontainers.image.title="Pigway Frame Flow" org.opencontainers.image.version="0.1.14" org.opencontainers.image.revision=$BUILD_COMMIT
+LABEL org.opencontainers.image.title="Pigway Frame Flow" org.opencontainers.image.version="0.1.15" org.opencontainers.image.revision=$BUILD_COMMIT
diff --git a/docs/releases/v0.1.15.md b/docs/releases/v0.1.15.md
new file mode 100644
index 0000000..5189cb2
--- /dev/null
+++ b/docs/releases/v0.1.15.md
@@ -0,0 +1,9 @@
+# v0.1.15
+
+自动照片、拼版、模板及数据模板任务在上传失败后，持久保存同一张暂存图片，并在下一次计划之前安排最多 5 次额外重试。间隔按 1、2、4、8、16 权重递增，随剩余窗口缩放，计划时间集中在窗口前半段，为上传与设备同步留出时间。重试不移动原定计划，也不复用旧操作 ID；每次独立记录操作及事件。成功后取消剩余重试，重启后沿用已有计数与截止时间；停用任务不再追加重试，删除任务取消其重试计划。
+
+若额外重试均失败，下一次计划仍先上传同一张图；若这一次仍失败，只删除该任务未入库、未星标的生成缓存，再按照片源生成并上传新图。NAS 原照片、用户上传文件和成功图库图片不由此流程删除。新图仍失败时，进入它自己的下一轮恢复窗口。
+
+上传结果未知不再阻塞后续独立上传。每次重试前先只读核对设备及云图库，精确确认已成功的上传不再重传；需继续尝试时关闭旧尝试并保留错误和核对审计。删除、恢复及自动清理仍保留各自的保护条件。重试窗口到期后不再启动额外上传，待同步尝试经核对结束后交给原计划执行。
+
+任务生成图片先保存在缓存，确认任务成功后才进入本地图库；失败不会产生看似成功的本地图库记录。额外重试仍归属原任务的云端保留规则，任务列表显示最近尝试结果。
diff --git a/pigway_frame_flow/api/app.py b/pigway_frame_flow/api/app.py
index 1f181fd..a2bc1a2 100644
--- a/pigway_frame_flow/api/app.py
+++ b/pigway_frame_flow/api/app.py
@@ -126,7 +126,7 @@ def create_app(settings=None, start_worker=True):
app = FastAPI(
title="Pigway Frame Flow API",
docs_url=None,
- version="0.1.14",
+ version="0.1.15",
lifespan=lifespan,
responses=error_responses,
)
@@ -944,6 +944,7 @@ def create_app(settings=None, start_worker=True):
)
def delete_job(job_id: str):
db.execute("DELETE FROM jobs WHERE id=?", (job_id,))
+ db.execute("DELETE FROM app_config WHERE key=?", ("upload-retry:" + job_id,))
return Response(status_code=204)
@app.get(
@@ -1032,7 +1033,7 @@ def create_app(settings=None, start_worker=True):
supports_upload=True,
),
],
- version="0.1.14",
+ version="0.1.15",
mock=settings.mock,
write_enabled=settings.write_enabled,
api_key_configured=bool(settings.api_key)
diff --git a/pigway_frame_flow/display/cloud_library.py b/pigway_frame_flow/display/cloud_library.py
index ce502a0..768230e 100644
--- a/pigway_frame_flow/display/cloud_library.py
+++ b/pigway_frame_flow/display/cloud_library.py
@@ -13,18 +13,29 @@ class CloudLibrary:
def register(self, device_id, image_id, asset_id, operation_id):
operation = self.db.one(
- "SELECT payload FROM operations WHERE id=?", (operation_id,)
+ "SELECT payload,client_id FROM operations WHERE id=?", (operation_id,)
)
payload = json.loads(operation["payload"])
content = payload.get("content", {})
job = self.db.one(
"SELECT job_id FROM job_runs WHERE operation_id=?", (operation_id,)
)
+ if not job and operation["client_id"] == "scheduler-retry":
+ job = self.db.one(
+ "SELECT job_id FROM asset_metadata WHERE asset_id=? AND job_id IS NOT NULL",
+ (asset_id,),
+ )
+ job_config = (
+ self.db.one("SELECT payload FROM jobs WHERE id=?", (job["job_id"],))
+ if job
+ else None
+ )
+ source = json.loads(job_config["payload"])["source"] if job_config else content
job_id = (
job["job_id"]
if job
- and content.get("type") in ("photo_source", "auto_collage")
- and content.get("item_id") is None
+ and source.get("type") in ("photo_source", "auto_collage")
+ and source.get("item_id") is None
else None
)
self.db.execute(
diff --git a/pigway_frame_flow/display/service.py b/pigway_frame_flow/display/service.py
index 727f1a4..da6e5d6 100644
--- a/pigway_frame_flow/display/service.py
+++ b/pigway_frame_flow/display/service.py
@@ -115,7 +115,7 @@ class DisplayService:
)
return existing["operation_id"]
if (
- action not in ("render", "render-ai", "local-upload")
+ action not in ("display", "render", "render-ai", "local-upload")
and tx.execute(
"SELECT id FROM operations WHERE device_id=? AND status='UNKNOWN' AND write_started=1 AND action!='render-ai' LIMIT 1",
(device_id,),
@@ -187,6 +187,13 @@ class DisplayService:
id,
),
)
+ if status == "COMPLETED":
+ row = self.db.one("SELECT result FROM operations WHERE id=?", (id,))
+ asset_id = json.loads(row["result"] or "{}").get("rendered_asset_id")
+ if asset_id and self.db.one(
+ "SELECT id FROM assets WHERE id=?", (asset_id,)
+ ):
+ self.assets.promote(asset_id)
if status in ("COMPLETED", "FAILED"):
self.recovery.resolve(id)
@@ -226,12 +233,17 @@ class DisplayService:
(device_id, image_id, image_id, device_id, device_id),
)
- def content_asset(self, content, device_id=None, operation_id=None):
+ def content_asset(
+ self, content, device_id=None, operation_id=None, *, staging=False
+ ):
if isinstance(content, AutoCollageContent):
from pigway_frame_flow.extensions.auto_collage import prepare
return self.content_asset(
- prepare(self, content, device_id), device_id, operation_id
+ prepare(self, content, device_id),
+ device_id,
+ operation_id,
+ staging=staging,
)
if isinstance(content, ProviderContent):
values = {**content.data, **self.providers.get(content.provider_id)}
@@ -244,11 +256,14 @@ class DisplayService:
),
device_id,
operation_id,
+ staging=staging,
)
if isinstance(content, TemplateContent):
values = self.renderer.resolve(content.template, content.data)
data = self.renderer.render(content.template, values, content.profile)
- asset = self.assets.put(data, content.template + ".jpg", "generated")
+ asset = self.assets.put(
+ data, content.template + ".jpg", "cache" if staging else "generated"
+ )
source = {
"type": "template",
"template": content.template,
@@ -377,7 +392,7 @@ class DisplayService:
"UPDATE operations SET result=? WHERE id=?",
(encode({"photo_attempts": attempts}), operation_id),
)
- return self.assets.put(rendered, name)
+ return self.assets.put(rendered, name, "cache" if staging else "images")
raise FlowError(
E.INVALID_REQUEST, "No decodable photo found within 5 candidates"
)
@@ -390,7 +405,9 @@ class DisplayService:
else:
raise FlowError(E.INVALID_REQUEST, "Content requires renderable image data")
return self.assets.put(
- convert(data, content.fit, content.profile), content.name
+ convert(data, content.fit, content.profile),
+ content.name,
+ "cache" if staging else "images",
)
def current_content(self, device_id):
@@ -446,10 +463,38 @@ class DisplayService:
yield from rows
cursor_time, cursor_id = rows[-1]["created_at"], rows[-1]["id"]
+ def upload_retry_expired(self, op):
+ if op["client_id"] != "scheduler-retry" or not op["request_id"].startswith(
+ "upload-retry-"
+ ):
+ return False
+ root = op["request_id"].split("-")[-2]
+ for row in self.db.all(
+ "SELECT payload FROM app_config WHERE key LIKE 'upload-retry:%'"
+ ):
+ plan = json.loads(row["payload"])
+ if plan["root"] == root:
+ return now_iso() >= plan["deadline"]
+ return True
+
def process_one(self):
with self.lock:
for op in self.queued_operations():
+ if self.upload_retry_expired(op):
+ self.update(
+ op["id"],
+ "FAILED",
+ error={
+ "code": E.CONFLICT,
+ "message": "Upload retry window closed",
+ "request_id": op["request_id"],
+ "details": {"reason": "RETRY_WINDOW_CLOSED"},
+ },
+ )
+ self.db.history(op, "RETRY_WINDOW_CLOSED")
+ continue
if op["action"] not in (
+ "display",
"render",
"render-ai",
"local-upload",
@@ -632,6 +677,13 @@ class DisplayService:
self.update(id, "SYNCING", {"image_id": target})
self.db.history(op, "ACCEPTED")
return
+ # A new upload is independent of a previous failed write. Read back its
+ # outcome before closing the old attempt; never replay that operation.
+ for pending in self.db.all(
+ "SELECT id FROM operations WHERE device_id=? AND status='UNKNOWN' AND write_started=1 AND action='display'",
+ (device_id,),
+ ):
+ self.reconcile(pending["id"], acknowledge=True)
request = DisplayRequest.model_validate(payload)
content = request.content
self.update(id, "RENDERING")
@@ -648,7 +700,7 @@ class DisplayService:
)
image_id = matches[0].image_id
else:
- asset = self.content_asset(content, device_id, id)
+ asset = self.content_asset(content, device_id, id, staging=True)
self.assets.record_origin(asset["asset_id"], op)
image_id = None
self.db.execute(
@@ -683,7 +735,6 @@ class DisplayService:
driver.display(image_id)
result = {"image_id": image_id}
if asset:
- self.assets.promote(asset["asset_id"])
result["rendered_asset_id"] = asset["asset_id"]
source = self.db.one(
"SELECT payload FROM asset_sources WHERE asset_id=?",
diff --git a/pigway_frame_flow/display/upload_reconciliation.py b/pigway_frame_flow/display/upload_reconciliation.py
index eb86423..3c526e4 100644
--- a/pigway_frame_flow/display/upload_reconciliation.py
+++ b/pigway_frame_flow/display/upload_reconciliation.py
@@ -77,7 +77,6 @@ class UploadReconciliation:
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"],),
diff --git a/pigway_frame_flow/extensions/storage.py b/pigway_frame_flow/extensions/storage.py
index 687426f..f2cb38b 100644
--- a/pigway_frame_flow/extensions/storage.py
+++ b/pigway_frame_flow/extensions/storage.py
@@ -43,6 +43,7 @@ class LocalRetention:
"SELECT payload FROM jobs",
"SELECT payload FROM template_config",
"SELECT payload FROM operations WHERE status IN ('QUEUED','RENDERING','UPLOADING','SYNCING','UNKNOWN')",
+ "SELECT payload FROM app_config WHERE key LIKE 'upload-retry:%'",
]:
payloads += [row["payload"] for row in db.all(query)]
for payload in payloads:
diff --git a/pigway_frame_flow/models.py b/pigway_frame_flow/models.py
index 855fee7..4b53358 100644
--- a/pigway_frame_flow/models.py
+++ b/pigway_frame_flow/models.py
@@ -366,7 +366,7 @@ class SettingsView(BaseModel):
class Health(BaseModel):
status: str
- version: str = "0.1.14"
+ version: str = "0.1.15"
class Readiness(BaseModel):
@@ -396,7 +396,7 @@ class RestoreRequest(StrictModel):
class Version(BaseModel):
product: str = "Pigway Frame Flow"
- version: str = "0.1.14"
+ version: str = "0.1.15"
api_version: str = "v1"
commit: str = "development"
build_time: str = "unknown"
diff --git a/pigway_frame_flow/renderers/images.py b/pigway_frame_flow/renderers/images.py
index 80b038b..0771856 100644
--- a/pigway_frame_flow/renderers/images.py
+++ b/pigway_frame_flow/renderers/images.py
@@ -1,5 +1,6 @@
import base64
import ipaddress
+import json
import socket
from io import BytesIO
from urllib.parse import urlsplit
@@ -115,9 +116,17 @@ class Assets:
def promote(self, asset_id):
row = self.db.one("SELECT * FROM assets WHERE id=?", (asset_id,))
if row and row["kind"] == "cache":
- target = self.settings.data_dir / "images" / (asset_id + ".jpg")
+ source = self.db.one(
+ "SELECT payload FROM asset_sources WHERE asset_id=?", (asset_id,)
+ )
+ kind = (
+ "generated"
+ if source and json.loads(source["payload"]).get("type") == "template"
+ else "images"
+ )
+ target = self.settings.data_dir / kind / (asset_id + ".jpg")
self.path(asset_id).replace(target)
- self.db.execute("UPDATE assets SET kind='images' WHERE id=?", (asset_id,))
+ self.db.execute("UPDATE assets SET kind=? WHERE id=?", (kind, asset_id))
return self.info(asset_id)
def info(self, asset_id):
diff --git a/pigway_frame_flow/scheduler/engine.py b/pigway_frame_flow/scheduler/engine.py
index ed5f17c..b8c0705 100644
--- a/pigway_frame_flow/scheduler/engine.py
+++ b/pigway_frame_flow/scheduler/engine.py
@@ -18,6 +18,7 @@ from pigway_frame_flow.models import (
PlaylistInput,
RestoreContent,
)
+from pigway_frame_flow.scheduler.upload_retries import UploadRetries
from pigway_frame_flow.storage.database import encode, now_iso
@@ -60,6 +61,7 @@ class Scheduler:
self.thread = None
self.webhook_thread = None
self.tick_lock = threading.Lock()
+ self.upload_retries = UploadRetries(service)
def jobs(self):
rows = [
@@ -147,6 +149,7 @@ class Scheduler:
def tick(self, current=None):
current = current or datetime.now(UTC)
with self.tick_lock:
+ self.upload_retries.tick(current)
for row in self.db.all(
"SELECT * FROM jobs WHERE enabled=1 AND next_run<=?",
(current.isoformat(),),
@@ -183,6 +186,10 @@ class Scheduler:
),
).model_dump(mode="json")
)
+ if action == "display":
+ payload = self.upload_retries.scheduled_payload(
+ row["id"], payload
+ )
operation_id = self.service.submit(
action,
payload,
@@ -191,6 +198,10 @@ class Scheduler:
"schedule-" + uuid4().hex,
"job-" + row["id"] + "-" + slot,
)
+ if action == "display":
+ self.upload_retries.scheduled_submitted(
+ row["id"], operation_id
+ )
following = next_run(job.schedule, current)
enabled = bool(following)
updated = job.model_copy(update={"enabled": enabled}).model_dump(
diff --git a/pigway_frame_flow/scheduler/upload_retries.py b/pigway_frame_flow/scheduler/upload_retries.py
new file mode 100644
index 0000000..2fbfc3b
--- /dev/null
+++ b/pigway_frame_flow/scheduler/upload_retries.py
@@ -0,0 +1,246 @@
+"""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.assets.path(asset_id).unlink(missing_ok=True)
+ with self.db.transaction() as tx:
+ for table, field in (
+ ("assets", "id"),
+ ("asset_sources", "asset_id"),
+ ("asset_metadata", "asset_id"),
+ ):
+ tx.execute(f"DELETE FROM {table} WHERE {field}=?", (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),
+ },
+ }
diff --git a/pigway_frame_flow/web/index.html b/pigway_frame_flow/web/index.html
index f9192b5..54a71eb 100644
--- a/pigway_frame_flow/web/index.html
+++ b/pigway_frame_flow/web/index.html
@@ -1,3 +1,3 @@
Pigway Frame Flow
-
+