Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aabbff8127 | ||
|
|
7cdf712829 | ||
|
|
327fa53d7b |
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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,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
@@ -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]
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"]
|
||||
|
||||
Reference in New Issue
Block a user