diff --git a/pigway_frame_flow/display/service.py b/pigway_frame_flow/display/service.py index b0be75a..872e902 100644 --- a/pigway_frame_flow/display/service.py +++ b/pigway_frame_flow/display/service.py @@ -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, diff --git a/pigway_frame_flow/scheduler/engine.py b/pigway_frame_flow/scheduler/engine.py index ae525b1..ab07c5b 100644 --- a/pigway_frame_flow/scheduler/engine.py +++ b/pigway_frame_flow/scheduler/engine.py @@ -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() diff --git a/tests/test_display.py b/tests/test_display.py index bd5fd5a..965c689 100644 --- a/tests/test_display.py +++ b/tests/test_display.py @@ -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 diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 1ff9f68..3dae11e 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -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"]