Compare commits

...
3 Commits
11 changed files with 222 additions and 6 deletions
+13
View File
@@ -363,3 +363,16 @@ the audit client's normal HTTP freshness rules. It is not a permanent offline
snapshot or a cross-job cache. Raw network exceptions are omitted because proxy
URLs may contain credentials. Logs identify the advisory host and whether the
complete audit succeeded.
### Production image retention
After a deployment passes verification and its success journal is saved, Frame Flow
removes obsolete `release-<commit>-<owner>` and `rollback-<owner>` images from the
configured production image repository. It retains the running version and the
previous deployment baseline for one-step rollback. Images referenced by any
container (including stopped containers), images with unrelated tags, and CI
images are protected. Retention does not prune build cache, volumes, or data.
Failed deployments do not run retention; cleanup failure produces a warning and
never rolls back a successfully verified deployment. Historical deployment
journals remain available for audit, but only the latest rollback image is
retained locally; older releases require rebuilding before use.
+2 -2
View File
@@ -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.0",
version="0.1.1",
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.0",
version="0.1.1",
mock=settings.mock,
write_enabled=settings.write_enabled,
api_key_configured=bool(settings.api_key)
+29
View File
@@ -50,6 +50,7 @@ class DisplayService:
self.retention = Retention(self)
self.cloud_library = CloudLibrary(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.
db.execute(
"UPDATE operations SET status='UNKNOWN', updated_at=? WHERE write_started=1 AND status IN ('UPLOADING','RENDERING')",
@@ -689,6 +690,31 @@ class DisplayService:
)
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:
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)
if (
device.current_content.id == expected
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"]})
self.unknown_checked.pop(op["id"], None)
except FlowError:
pass
for op in self.db.all("SELECT * FROM operations WHERE status='SYNCING'"):
try:
device = self.registry.get(op["device_id"]).device()
@@ -824,6 +850,9 @@ class DisplayService:
{"image_id": device.current_content.id},
)
resolved = True
if confirmed and op["action"] == "display":
self.cloud_library.confirm(id)
self.db.event("operation.completed", {"operation_id": id})
self.db.history(
dict(
id=id,
+2 -2
View File
@@ -336,7 +336,7 @@ class SettingsView(BaseModel):
class Health(BaseModel):
status: str
version: str = "0.1.0"
version: str = "0.1.1"
class Readiness(BaseModel):
@@ -366,7 +366,7 @@ class RestoreRequest(StrictModel):
class Version(BaseModel):
product: str = "Pigway Frame Flow"
version: str = "0.1.0"
version: str = "0.1.1"
api_version: str = "v1"
commit: str = "development"
build_time: str = "unknown"
+3
View File
@@ -269,6 +269,9 @@ class Scheduler:
def run(self):
while not self.stop_event.is_set():
try:
# Resolve late sync before submitting due jobs: a write lock
# otherwise makes tick fail before reconciliation can run.
self.service.sync()
self.tick()
self.service.expire()
self.service.cloud_library.sweep()
+1 -1
View File
@@ -1,3 +1,3 @@
<!doctype html>
<html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><meta name="color-scheme" content="light dark"><title>Pigway Frame Flow</title><link rel="stylesheet" href="/ui/style.css"></head>
<body><aside class="sidebar"><a class="brand" href="/"><span class="brand-icon">F<span>↗</span></span><span>Frame Flow<small>Pigway · Smart Frame Automation</small></span></a><nav id="nav"></nav><div class="sidebar-bottom"><small>Version 0.1.0</small></div></aside><div class="workspace"><header><div><span class="eyebrow">PIGWAY FRAME FLOW</span><h1 id="page-title"></h1></div><div class="toolbar"><select id="device" aria-label="Device"></select><button id="lang" class="icon-button" aria-label="Change language">EN</button><button id="theme" class="icon-button" aria-label="Change theme">◐</button></div></header><main id="main"></main><footer id="footer"></footer></div><div id="toast" role="status" aria-live="polite"></div><dialog id="modal"><div id="modal-content"></div></dialog><script src="/ui/i18n.js"></script><script src="/ui/features.js"></script><script src="/ui/extensions.js"></script><script src="/ui/app.js"></script></body></html>
<body><aside class="sidebar"><a class="brand" href="/"><span class="brand-icon">F<span>↗</span></span><span>Frame Flow<small>Pigway · Smart Frame Automation</small></span></a><nav id="nav"></nav><div class="sidebar-bottom"><small>Version 0.1.1</small></div></aside><div class="workspace"><header><div><span class="eyebrow">PIGWAY FRAME FLOW</span><h1 id="page-title"></h1></div><div class="toolbar"><select id="device" aria-label="Device"></select><button id="lang" class="icon-button" aria-label="Change language">EN</button><button id="theme" class="icon-button" aria-label="Change theme">◐</button></div></header><main id="main"></main><footer id="footer"></footer></div><div id="toast" role="status" aria-live="polite"></div><dialog id="modal"><div id="modal-content"></div></dialog><script src="/ui/i18n.js"></script><script src="/ui/features.js"></script><script src="/ui/extensions.js"></script><script src="/ui/app.js"></script></body></html>
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "pigway-frame-flow"
version = "0.1.0"
version = "0.1.1"
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]
+45
View File
@@ -723,6 +723,50 @@ class Deploy:
text = re.sub(r"https?://[^\s\"'<>]+", "[REDACTED-URL]", text)
print("Scoped container logs (redacted):", text, flush=True)
def cleanup_images(self):
"""Retain container references and the verified deployment's rollback baseline."""
try:
protected = {self.state["previous_image"]}
container_ids = self.run("ps", "-aq").stdout.decode().split()
if container_ids:
containers = json.loads(self.run("inspect", *container_ids).stdout)
protected.update(container["Image"] for container in containers)
repo = self.env["PROD_IMAGE_REPO"]
image_ids = (
self.run("image", "ls", "--no-trunc", "-q", repo)
.stdout.decode()
.split()
)
if not image_ids:
return
images = json.loads(
self.run("image", "inspect", *sorted(set(image_ids))).stdout
)
removed = 0
for image in images:
tags = image.get("RepoTags") or []
# Shared images and unrelated tags are never cleanup targets.
if image["Id"] in protected or not tags:
continue
if not all(
re.fullmatch(
re.escape(repo)
+ r":(?:release-[0-9a-f]{40}-[0-9a-f]{8}|rollback-[0-9a-f]{32})",
tag,
)
for tag in tags
):
continue
result = self.run("image", "rm", *tags, check=False)
if result.returncode:
print("IMAGE RETENTION: referenced image retained", flush=True)
else:
removed += 1
print(f"IMAGE RETENTION: removed {removed} obsolete images", flush=True)
except Exception:
# Cleanup failure must never roll back an already verified deployment.
print("IMAGE RETENTION WARNING: cleanup deferred", flush=True)
def deploy(self):
previous = self.preflight()
self.lock()
@@ -769,6 +813,7 @@ class Deploy:
self.state["status"] = "success"
self.state_io(self.state)
print("DEPLOY VERIFIED", flush=True)
self.cleanup_images()
except BaseException as exc:
print("DEPLOY FAILED:", type(exc).__name__, flush=True)
self.diagnostics()
+52
View File
@@ -152,3 +152,55 @@ def test_offline(service):
id = submit(service)
service.process_one()
assert service.operation(id)["error"]["code"] == "DEVICE_OFFLINE"
def test_late_sync_resolves_unknown_without_upload_retry(service, monkeypatch):
id = submit(service)
service.process_one()
service.update(id, "UNKNOWN")
driver = service.registry.get("living-room")
calls = []
device = driver.device
monkeypatch.setattr(driver, "device", lambda **kw: calls.append(kw) or device(**kw))
service.sync()
assert service.operation(id)["status"] == "COMPLETED"
assert driver.client.uploads == 1
assert calls == [{"refresh": True}]
assert not service.blocked("living-room")
assert service.db.one(
"SELECT confirmed_at FROM managed_cloud_uploads WHERE operation_id=?", (id,)
)["confirmed_at"]
def test_unknown_wrong_target_stays_locked_and_queries_at_most_once_per_minute(
service, monkeypatch
):
id = submit(service)
service.process_one()
service.update(id, "UNKNOWN")
driver = service.registry.get("living-room")
driver.client.display(1)
driver.invalidate()
calls = []
device = driver.device
monkeypatch.setattr(driver, "device", lambda **kw: calls.append(kw) or device(**kw))
service.sync()
service.sync()
assert service.operation(id)["status"] == "UNKNOWN"
assert service.blocked("living-room")
assert driver.client.uploads == 1
assert len(calls) == 1
def test_unknown_matching_image_without_sync_stays_locked(service, monkeypatch):
id = submit(service)
service.process_one()
service.update(id, "UNKNOWN")
driver = service.registry.get("living-room")
device = driver.device()
device.current_content.synced = False
monkeypatch.setattr(driver, "device", lambda **kw: device)
service.sync()
assert service.operation(id)["status"] == "UNKNOWN"
assert service.blocked("living-room")
assert driver.client.uploads == 1
+56
View File
@@ -111,6 +111,10 @@ def test_deploy_rolls_back_exact_baseline_without_deleting_data(
monkeypatch.setattr(
tool, "state_io", lambda state: saved.append(json.loads(json.dumps(state)))
)
cleanup = []
monkeypatch.setattr(
tool, "cleanup_images", lambda: cleanup.append(saved[-1]["status"])
)
verifies = []
def verify(config):
@@ -125,11 +129,13 @@ def test_deploy_rolls_back_exact_baseline_without_deleting_data(
if failure:
with pytest.raises(RuntimeError):
tool.deploy()
assert cleanup == []
assert activated == [new, old]
assert saved[0]["previous"] == old
assert ("unlock" in calls) is (failure != "rollback")
else:
tool.deploy()
assert cleanup == ["success"]
assert activated == [new]
assert saved[-1]["status"] == "success"
assert not any(
@@ -243,3 +249,53 @@ def test_lock_conflict_never_activates_production(tool, monkeypatch):
monkeypatch.setattr(tool, "run", fail)
with pytest.raises(RuntimeError):
tool.deploy()
def test_image_retention_protects_containers_rollback_and_other_tags(tool, monkeypatch):
tool.state = {"previous_image": "previous"}
release = "pigway-frame-flow:release-" + "a" * 40 + "-" + "b" * 8
rollback = "pigway-frame-flow:rollback-" + "c" * 32
images = [
{"Id": "current", "RepoTags": [release]},
{"Id": "previous", "RepoTags": [rollback]},
{"Id": "stopped", "RepoTags": [release]},
{"Id": "obsolete", "RepoTags": [release, rollback]},
{"Id": "shared", "RepoTags": [release, "other:latest"]},
{"Id": "ci", "RepoTags": ["ffci-fixture:latest"]},
]
calls = []
def run(*args, **kwargs):
calls.append(args)
if args == ("ps", "-aq"):
data = "container1 container2"
elif args[0] == "inspect":
data = json.dumps([{"Image": "current"}, {"Image": "stopped"}])
elif args[:2] == ("image", "ls"):
data = "current previous stopped obsolete shared ci"
elif args[:2] == ("image", "inspect"):
data = json.dumps(images)
else:
data = ""
return SimpleNamespace(stdout=data.encode(), returncode=0)
monkeypatch.setattr(tool, "run", run)
tool.cleanup_images()
assert [c for c in calls if c[:2] == ("image", "rm")] == [
("image", "rm", release, rollback)
]
def test_image_retention_failure_does_not_fail_verified_deploy(
tool, monkeypatch, capsys
):
tool.state = {"previous_image": "previous"}
def unavailable(*args, **kwargs):
raise RuntimeError("private diagnostic")
monkeypatch.setattr(tool, "run", unavailable)
tool.cleanup_images()
output = capsys.readouterr().out
assert "cleanup deferred" in output
assert "private diagnostic" not in output
+18
View File
@@ -75,3 +75,21 @@ def test_playlist(app, service):
def test_invalid_cron():
with pytest.raises(FlowError):
next_run(Schedule(type="cron", cron="* * * * * *"))
def test_worker_checks_late_sync_before_due_tasks(app, service, monkeypatch):
scheduler = app.state.scheduler
calls = []
monkeypatch.setattr(service, "sync", lambda: calls.append("sync"))
def tick():
calls.append("tick")
scheduler.stop_event.set()
monkeypatch.setattr(scheduler, "tick", tick)
monkeypatch.setattr(service, "expire", lambda: None)
monkeypatch.setattr(service.cloud_library, "sweep", lambda: None)
monkeypatch.setattr(service.local_retention, "sweep", lambda: None)
monkeypatch.setattr(service, "process_one", lambda: False)
scheduler.run()
assert calls[:2] == ["sync", "tick"]