way/pigway-frame-flow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 29 Wiki Activity

Move favorites to local library and keep confirmed task heads in cloud

Browse Source
This commit is contained in:
Pigway
2026-10-06 09:31:09 +08:00
parent c005a9444a
commit 180b6ae100
16 changed files with 1187 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pigway_frame_flow/display/cloud_library.py
+132
View File
@@ -0,0 +1,132 @@
"""Keep confirmed task heads and locally starred uploads; never retry deletion."""
import json
import time
from datetime import UTC, datetime
class CloudLibrary:
def __init__(self, service):
self.service = service
self.db = service.db
self.checked = 0.0
def register(self, device_id, image_id, asset_id, operation_id):
meta = self.service.assets.info(asset_id)
self.db.execute(
"INSERT INTO managed_cloud_uploads VALUES(?,?,?,?,?,?,NULL,NULL)",
(
device_id,
image_id,
asset_id,
meta.get("job_id"),
operation_id,
datetime.now(UTC).isoformat(),
),
)
def confirm(self, operation_id):
self.db.execute(
"UPDATE managed_cloud_uploads SET confirmed_at=? WHERE operation_id=?",
(datetime.now(UTC).isoformat(), operation_id),
)
self.checked = 0.0
def retained(self, row):
meta = self.db.one(
"SELECT starred FROM asset_metadata WHERE asset_id=?", (row["asset_id"],)
)
if meta and meta["starred"]:
head = self.db.one(
"SELECT image_id FROM managed_cloud_uploads WHERE device_id=? AND asset_id=? AND confirmed_at IS NOT NULL ORDER BY uploaded_at DESC,image_id DESC LIMIT 1",
(row["device_id"], row["asset_id"]),
)
if not head or head["image_id"] == row["image_id"]:
return True
if row["job_id"]:
head = self.db.one(
"SELECT image_id FROM managed_cloud_uploads WHERE device_id=? AND job_id=? AND confirmed_at IS NOT NULL ORDER BY uploaded_at DESC,image_id DESC LIMIT 1",
(row["device_id"], row["job_id"]),
)
if head and head["image_id"] == row["image_id"]:
return True
return False
def eligible(self, device_id, image_id):
row = self.db.one(
"SELECT * FROM managed_cloud_uploads WHERE device_id=? AND image_id=?",
(device_id, image_id),
)
if not row or self.retained(row):
return False
# A confirmed replacement is mandatory; failed/unknown uploads cannot
# evict the last known usable cloud image.
newer = self.db.one(
"SELECT image_id FROM managed_cloud_uploads WHERE device_id=? AND confirmed_at IS NOT NULL AND uploaded_at>? LIMIT 1",
(device_id, row["uploaded_at"]),
)
if not newer:
return False
state = self.service.state(device_id)
if state["base_image_id"] == image_id:
return False
if self.db.one(
"SELECT id FROM overrides WHERE device_id=? AND image_id=?",
(device_id, image_id),
):
return False
for r in self.db.all(
"SELECT payload FROM playlists UNION ALL SELECT payload FROM jobs"
):
data = json.loads(r["payload"])
if data.get("device_id") == device_id and (
image_id in data.get("image_ids", [])
or data.get("source", {}).get("image_id") == image_id
):
return False
return True
def sweep(self):
if time.monotonic() - self.checked < 60:
return
self.checked = time.monotonic()
with self.service.lock:
if (
not self.service.settings.mock
and not self.service.settings.write_enabled
):
return
for row in self.db.all(
"SELECT * FROM managed_cloud_uploads WHERE cleanup_operation IS NULL ORDER BY uploaded_at"
):
device_id, image_id = row["device_id"], row["image_id"]
if self.service.blocked(device_id) or not self.eligible(
device_id, image_id
):
continue
device = self.service.registry.get(device_id).device(refresh=True)
if (
device.status != "online"
or not device.current_content.synced
or device.current_content.id == image_id
):
continue
id = self.service.submit(
"delete",
{
"image_id": image_id,
"confirm_current": False,
"cloud_cleanup": True,
},
device_id,
"cloud-library",
"cloud-cleanup-" + row["operation_id"],
"cloud-cleanup-"
+ row["operation_id"]
+ "-"
+ datetime.now(UTC).strftime("%Y%m%d%H%M"),
)
self.db.execute(
"UPDATE managed_cloud_uploads SET cleanup_operation=? WHERE device_id=? AND image_id=?",
(id, device_id, image_id),
)
pigway_frame_flow/display/service.py
+41 -1
View File
@@ -5,6 +5,7 @@ import time
from datetime import UTC, datetime, timedelta
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.drivers.base import DriverRegistry
from pigway_frame_flow.errors import ErrorCode as E
@@ -38,6 +39,7 @@ class DisplayService:
)
self.lock = threading.RLock()
self.retention = Retention(self)
self.cloud_library = CloudLibrary(self)
self.expiration_checked: dict[str, float] = {}
# A killed write cannot be replayed safely; pre-write work can resume.
db.execute(
@@ -383,6 +385,7 @@ class DisplayService:
asset = self.content_asset(
TypeAdapter(Content).validate_python(payload), device_id, id
)
self.assets.record_origin(asset["asset_id"], op)
if action == "local-upload":
asset = self.assets.promote(asset["asset_id"])
self.update(
@@ -402,6 +405,26 @@ class DisplayService:
state = self.state(device_id)
if action == "delete":
image_id = payload["image_id"]
if payload.get("cloud_cleanup"):
if (
not self.cloud_library.eligible(device_id, image_id)
or previous == image_id
):
self.db.execute(
"UPDATE managed_cloud_uploads SET cleanup_operation=NULL WHERE device_id=? AND image_id=?",
(device_id, image_id),
)
self.update(
id, "COMPLETED", {"image_id": image_id, "deferred": True}
)
return
# The confirmed replacement supersedes this task-owned restore
# reference; unrelated artwork and active overrides stay protected.
self.db.execute(
"UPDATE device_state SET previous_image_id=NULL WHERE device_id=? AND previous_image_id=?",
(device_id, image_id),
)
state = self.state(device_id)
if payload.get("retention") and (
not self.retention.eligible(device_id, image_id) or image_id == previous
):
@@ -418,12 +441,26 @@ class DisplayService:
"SELECT image_id FROM overrides WHERE device_id=?", (device_id,)
)
}
if image_id in protected:
if image_id in protected and not payload.get("purge"):
raise FlowError(E.CONFLICT, "Image is needed for restoration", 409)
if image_id == previous and not payload["confirm_current"]:
raise FlowError(E.CONFLICT, "Current image needs confirmation", 409)
self._start_write(id)
driver.delete(image_id)
self.db.execute(
"DELETE FROM managed_cloud_uploads WHERE device_id=? AND image_id=?",
(device_id, image_id),
)
if payload.get("purge"):
self.db.execute(
"DELETE FROM overrides WHERE device_id=? AND image_id=?",
(device_id, image_id),
)
for field in ("previous_image_id", "base_image_id", "managed_image_id"):
self.db.execute(
f"UPDATE device_state SET {field}=NULL WHERE device_id=? AND {field}=?",
(device_id, image_id),
)
for suffix in [".original", "-thumb.jpg"]:
(
self.settings.data_dir / "cache" / f"{device_id}-{image_id}{suffix}"
@@ -476,6 +513,7 @@ class DisplayService:
image_id = matches[0].image_id
else:
asset = self.content_asset(content, device_id, id)
self.assets.record_origin(asset["asset_id"], op)
image_id = None
self.db.execute(
"UPDATE device_state SET previous_image_id=? WHERE device_id=?",
@@ -494,6 +532,7 @@ class DisplayService:
if asset:
self.update(id, "UPLOADING")
image_id = driver.upload(self.assets.path(asset["asset_id"]), asset["name"])
self.cloud_library.register(device_id, image_id, asset["asset_id"], id)
if (
request.options.cleanup_unstarred
and isinstance(content, PhotoSourceContent)
@@ -561,6 +600,7 @@ class DisplayService:
and device.current_content.synced
):
self.update(op["id"], "COMPLETED")
self.cloud_library.confirm(op["id"])
self.db.history(op, "COMPLETED")
self.db.event("operation.completed", {"operation_id": op["id"]})
elif (