Fix webhook starvation and isolate scheduled job failures for v0.1.2 [skip ci]

2026-10-16 09:48:51 +08:00
parent e1b001a431
commit 8a3c616b5f
8 changed files with 158 additions and 64 deletions
@@ -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)
@@ -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()),
@@ -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"
@@ -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(),)
@@ -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.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>
+<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.2</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,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]
@@ -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
@@ -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"
+    )