diff --git a/pigway_frame_flow/api/app.py b/pigway_frame_flow/api/app.py index 6270c77..b964b28 100644 --- a/pigway_frame_flow/api/app.py +++ b/pigway_frame_flow/api/app.py @@ -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.1", + version="0.1.2", 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.1", + version="0.1.2", mock=settings.mock, write_enabled=settings.write_enabled, api_key_configured=bool(settings.api_key) diff --git a/pigway_frame_flow/extensions/webhooks.py b/pigway_frame_flow/extensions/webhooks.py index 4c6936f..ff9c44f 100644 --- a/pigway_frame_flow/extensions/webhooks.py +++ b/pigway_frame_flow/extensions/webhooks.py @@ -39,12 +39,13 @@ class Webhooks: db = self.service.db for sub in db.all("SELECT * FROM webhook_subscriptions WHERE enabled=1"): types = json.loads(sub["event_types"]) + if not types: + continue + placeholders = ",".join("?" for _ in types) for event in db.all( - "SELECT * FROM events e WHERE created_at>=? AND NOT EXISTS (SELECT 1 FROM webhook_deliveries d WHERE d.event_id=e.id AND d.subscription_id=?) ORDER BY created_at LIMIT 500", - (sub["created_at"], sub["id"]), + f"SELECT * FROM events e WHERE created_at>=? AND event_type IN ({placeholders}) AND NOT EXISTS (SELECT 1 FROM webhook_deliveries d WHERE d.event_id=e.id AND d.subscription_id=?) ORDER BY created_at LIMIT 500", + (sub["created_at"], *types, sub["id"]), ): - if event["event_type"] not in types: - continue db.execute( "INSERT OR IGNORE INTO webhook_deliveries VALUES(?,?,'QUEUED',0,?,NULL)", (sub["id"], event["id"], now_iso()), diff --git a/pigway_frame_flow/models.py b/pigway_frame_flow/models.py index fd814d3..04a86bf 100644 --- a/pigway_frame_flow/models.py +++ b/pigway_frame_flow/models.py @@ -336,7 +336,7 @@ class SettingsView(BaseModel): class Health(BaseModel): status: str - version: str = "0.1.1" + version: str = "0.1.2" class Readiness(BaseModel): @@ -366,7 +366,7 @@ class RestoreRequest(StrictModel): class Version(BaseModel): product: str = "Pigway Frame Flow" - version: str = "0.1.1" + version: str = "0.1.2" api_version: str = "v1" commit: str = "development" build_time: str = "unknown" diff --git a/pigway_frame_flow/scheduler/engine.py b/pigway_frame_flow/scheduler/engine.py index ab07c5b..39cec8a 100644 --- a/pigway_frame_flow/scheduler/engine.py +++ b/pigway_frame_flow/scheduler/engine.py @@ -147,61 +147,72 @@ class Scheduler: "SELECT * FROM jobs WHERE enabled=1 AND next_run<=?", (current.isoformat(),), ): - job = JobInput.model_validate_json(row["payload"]) - slot = row["next_run"] - if self.db.one( - "SELECT * FROM job_runs WHERE job_id=? AND scheduled_at=?", - (row["id"], slot), - ): - continue - # Skip overdue tasks (>5 minutes): avoid an unexpected artwork - # overwrite after downtime. One-time tasks are retired too. - late = (current - datetime.fromisoformat(slot)).total_seconds() > 300 - operation_id = None - if not late: - action = "restore" if job.source.type == "restore" else "display" - payload = ( - {} - if isinstance(job.source, RestoreContent) - else DisplayRequest( - device_id=job.device_id, - content=job.source, - options=Options( - duration=job.display_duration, - restore_previous=job.restore_previous, - priority=job.priority, - cleanup_unstarred=job.cleanup_unstarred, + try: + job = JobInput.model_validate_json(row["payload"]) + slot = row["next_run"] + if self.db.one( + "SELECT * FROM job_runs WHERE job_id=? AND scheduled_at=?", + (row["id"], slot), + ): + continue + # Skip overdue tasks (>5 minutes): avoid an unexpected artwork + # overwrite after downtime. One-time tasks are retired too. + late = ( + current - datetime.fromisoformat(slot) + ).total_seconds() > 300 + operation_id = None + if not late: + action = ( + "restore" if job.source.type == "restore" else "display" + ) + payload = ( + {} + if isinstance(job.source, RestoreContent) + else DisplayRequest( + device_id=job.device_id, + content=job.source, + options=Options( + duration=job.display_duration, + restore_previous=job.restore_previous, + priority=job.priority, + cleanup_unstarred=job.cleanup_unstarred, + ), + ).model_dump(mode="json") + ) + operation_id = self.service.submit( + action, + payload, + job.device_id, + "scheduler", + "schedule-" + uuid4().hex, + "job-" + row["id"] + "-" + slot, + ) + following = next_run(job.schedule, current) + enabled = bool(following) + updated = job.model_copy(update={"enabled": enabled}).model_dump( + mode="json" + ) + with self.db.transaction() as tx: + tx.execute( + "INSERT OR IGNORE INTO job_runs VALUES(?,?,?)", + (row["id"], slot, operation_id), + ) + tx.execute( + "UPDATE jobs SET payload=?,enabled=?,next_run=?,last_run=?,last_result=? WHERE id=?", + ( + encode(updated), + int(enabled), + following, + now_iso(), + "SKIPPED_OVERDUE" if late else operation_id, + row["id"], ), - ).model_dump(mode="json") - ) - operation_id = self.service.submit( - action, - payload, - job.device_id, - "scheduler", - "schedule-" + uuid4().hex, - "job-" + row["id"] + "-" + slot, - ) - following = next_run(job.schedule, current) - enabled = bool(following) - updated = job.model_copy(update={"enabled": enabled}).model_dump( - mode="json" - ) - with self.db.transaction() as tx: - tx.execute( - "INSERT OR IGNORE INTO job_runs VALUES(?,?,?)", - (row["id"], slot, operation_id), - ) - tx.execute( - "UPDATE jobs SET payload=?,enabled=?,next_run=?,last_run=?,last_result=? WHERE id=?", - ( - encode(updated), - int(enabled), - following, - now_iso(), - "SKIPPED_OVERDUE" if late else operation_id, - row["id"], - ), + ) + except Exception: + # Keep this slot pending for retry or the existing overdue policy. + # A single device must not prevent other jobs or queue consumption. + logging.getLogger("frame-flow").error( + "Scheduled job failed; details suppressed" ) for row in self.db.all( "SELECT * FROM playlists WHERE next_run<=?", (current.isoformat(),) diff --git a/pigway_frame_flow/web/index.html b/pigway_frame_flow/web/index.html index 6b9a513..5320e66 100644 --- a/pigway_frame_flow/web/index.html +++ b/pigway_frame_flow/web/index.html @@ -1,3 +1,3 @@ Pigway Frame Flow -
PIGWAY FRAME FLOW

 +
PIGWAY FRAME FLOW

 diff --git a/pyproject.toml b/pyproject.toml index 7dddfcf..ff34200 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pigway-frame-flow" -version = "0.1.1" +version = "0.1.2" 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] diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 0113e88..d7e2d18 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -373,3 +373,30 @@ def test_ai_layout_sends_metadata_only_and_validates_response( assert b"test-ai-secret" not in calls[0]["body"] assert image.getvalue() not in calls[0]["body"] assert calls[0]["headers"]["Authorization"] == "Bearer test-ai-secret" + + +def test_webhook_matching_event_after_unrelated_backlog(client, service, monkeypatch): + import json + + service.settings.image_url_hosts.append("example.com") + response = client.post( + "/api/v1/webhooks", + json={ + "callback_url": "https://example.com/events", + "event_types": ["operation.failed"], + }, + ) + assert response.status_code == 201 + for _ in range(500): + service.db.event("display.changed", {}) + service.db.event("operation.failed", {"operation_id": "expected"}) + sent = [] + monkeypatch.setattr( + service.assets, "fetch_url", lambda url, **kwargs: sent.append(kwargs) or b"" + ) + service.webhooks.tick() + service.webhooks.last = 0 + service.webhooks.tick() + assert len(sent) == 1 + assert json.loads(sent[0]["body"])["data"] == {"operation_id": "expected"} + assert len(service.db.all("SELECT * FROM webhook_deliveries")) == 1 diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 3dae11e..af0fbf5 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -93,3 +93,58 @@ def test_worker_checks_late_sync_before_due_tasks(app, service, monkeypatch): monkeypatch.setattr(service, "process_one", lambda: False) scheduler.run() assert calls[:2] == ["sync", "tick"] + + +def test_blocked_job_keeps_retry_slot_without_blocking_worker( + app, service, monkeypatch +): + scheduler = app.state.scheduler + service.registry.drivers["other-room"] = service.registry.get("living-room") + jobs = [] + for device in ("living-room", "other-room"): + jobs.append( + scheduler.save_job( + JobInput( + name=device, + device_id=device, + schedule={"type": "daily", "time": "07:00"}, + source={"type": "restore"}, + ) + ) + ) + current = datetime.now(UTC) + due = (current - timedelta(seconds=2)).isoformat() + service.db.execute("UPDATE jobs SET next_run=?", (due,)) + unknown = service.submit("display", {}, "living-room", "test", "unknown") + service.db.execute( + "UPDATE operations SET status='UNKNOWN',write_started=1 WHERE id=?", (unknown,) + ) + queued = service.submit("render", {}, "living-room", "test", "render") + calls = [] + monkeypatch.setattr(service, "sync", lambda: None) + monkeypatch.setattr(service, "expire", lambda: calls.append("expire")) + monkeypatch.setattr(service.cloud_library, "sweep", lambda: None) + monkeypatch.setattr(service.local_retention, "sweep", lambda: None) + + def process(): + calls.append("process") + assert service.operation(queued)["status"] == "QUEUED" + scheduler.stop_event.set() + return False + + monkeypatch.setattr(service, "process_one", process) + monkeypatch.setattr( + scheduler.stop_event, "wait", lambda timeout: scheduler.stop_event.set() + ) + scheduler.run() + assert calls == ["expire", "process"] + blocked = service.db.one("SELECT * FROM jobs WHERE id=?", (jobs[0]["id"],)) + assert blocked["next_run"] == due + assert blocked["last_run"] is None + assert len(service.db.all("SELECT * FROM job_runs")) == 1 + assert service.db.one("SELECT * FROM job_runs")["job_id"] == jobs[1]["id"] + scheduler.tick(current + timedelta(minutes=6)) + assert ( + service.db.one("SELECT * FROM jobs WHERE id=?", (jobs[0]["id"],))["last_result"] + == "SKIPPED_OVERDUE" + )