feat: recover failed scheduled uploads with bounded persistent retries [skip ci]

tests/test_api.py

@@ -338,7 +338,8 @@ def test_unknown_same_idempotency_key_returns_original(client, service):
     fresh = client.post(
         "/api/v1/display", json=payload, headers={"Idempotency-Key": "new-key"}
     )
-    assert fresh.status_code == 409
+    assert fresh.status_code == 202
+    assert fresh.json()["operation_id"] != original.json()["operation_id"]
 
 
 def test_schema_contract(client):

tests/test_display.py

@@ -71,22 +71,24 @@ def test_nested_overrides(service):
     assert service.registry.get("living-room").get_current_image() == 1
 
 
-def test_no_retry_unknown(service):
+def test_unknown_does_not_block_new_independent_upload(service):
     driver = service.registry.get("living-room")
+    attempts = []
 
-    def timeout(*a):
+    def timeout(*args):
+        attempts.append(args)
         raise FlowError(ErrorCode.WRITE_RESULT_UNKNOWN, "Write outcome unknown", 502)
 
     driver.upload = timeout
-    id = submit(service)
+    first = submit(service)
     service.process_one()
-    assert service.operation(id)["status"] == "UNKNOWN"
-    with pytest.raises(FlowError):
-        submit(service)
-    assert service.process_one() is False
-    assert service.reconcile(id)["resolved"] is False
-    assert service.reconcile(id, True)["resolved"] is True
-    assert service.operation(id)["status"] == "FAILED"
+    assert service.operation(first)["status"] == "UNKNOWN"
+    second = submit(service, priority=10)
+    assert second != first
+    assert service.process_one()
+    assert service.operation(first)["status"] == "FAILED"
+    assert service.operation(second)["status"] == "UNKNOWN"
+    assert len(attempts) == 2
 
 
 def test_restart_preserves_override(service):

tests/test_upload_retries.py

@@ -0,0 +1,154 @@
+from datetime import UTC, datetime, timedelta
+
+from pigway_frame_flow.errors import ErrorCode, FlowError
+from pigway_frame_flow.models import JobInput
+from pigway_frame_flow.scheduler.engine import Scheduler
+
+
+def failed_job(app, service, monkeypatch):
+    scheduler = app.state.scheduler
+    job = scheduler.save_job(
+        JobInput(
+            name="Retry",
+            schedule={"type": "interval", "interval_minutes": 120},
+            source={"type": "template", "template": "quote"},
+        )
+    )
+    now = datetime.now(UTC)
+    service.db.execute(
+        "UPDATE jobs SET next_run=? WHERE id=?",
+        ((now - timedelta(seconds=1)).isoformat(), job["id"]),
+    )
+    driver = service.registry.get("living-room")
+    upload = driver.upload
+
+    def fail(*a):
+        raise FlowError(ErrorCode.UPLOAD_FAILED, "Rejected", 502)
+
+    monkeypatch.setattr(driver, "upload", fail)
+    scheduler.tick(now)
+    service.process_one()
+    scheduler.tick(now + timedelta(seconds=1))
+    return scheduler, job, now, driver, upload
+
+
+def test_five_backoffs_then_one_scheduled_failure_regenerates(
+    app, service, monkeypatch, client
+):
+    scheduler, job, now, driver, upload = failed_job(app, service, monkeypatch)
+    policy = scheduler.upload_retries
+    plan = policy.load(job["id"])
+    asset = plan["asset_id"]
+    path = service.assets.path(asset)
+    original = path.read_bytes()
+    assert client.get("/api/v1/assets").json() == []
+    assert (
+        service.db.one("SELECT kind FROM assets WHERE id=?", (asset,))["kind"]
+        == "cache"
+    )
+    times = [datetime.fromisoformat(x) for x in plan["times"]]
+    gaps = [(times[0] - now).total_seconds()] + [
+        (b - a).total_seconds() for a, b in zip(times, times[1:], strict=False)
+    ]
+    assert gaps == sorted(gaps) and len(times) == 5
+    assert times[-1] < datetime.fromisoformat(plan["deadline"])
+    for number, when in enumerate(times, 1):
+        scheduler.tick(when + timedelta(milliseconds=1))
+        attempt = policy.load(job["id"])
+        assert attempt["count"] == number
+        op = service.operation(attempt["active"])
+        assert (
+            service.db.one(
+                "SELECT payload FROM operations WHERE id=?", (op["operation_id"],)
+            )["payload"].find(asset)
+            >= 0
+        )
+        service.process_one()
+        assert path.read_bytes() == original
+    assert len(service.db.all("SELECT * FROM job_runs")) == 1
+    # The plan survives a process restart, without resetting the five-attempt cap.
+    scheduler = Scheduler(service)
+    deadline = datetime.fromisoformat(plan["deadline"])
+    scheduler.tick(deadline + timedelta(milliseconds=1))
+    planned = scheduler.upload_retries.load(job["id"])
+    assert planned["phase"] == "scheduled_running"
+    service.process_one()
+    scheduler.tick(deadline + timedelta(seconds=1))
+    replacement = scheduler.upload_retries.load(job["id"])
+    assert replacement["phase"] == "replacement"
+    assert not path.exists()
+    assert service.db.one("SELECT id FROM assets WHERE id=?", (asset,)) is None
+    monkeypatch.setattr(driver, "upload", upload)
+    service.process_one()
+    service.sync()
+    op = service.operation(replacement["active"])
+    assert op["status"] == "COMPLETED"
+    assert op["result"]["rendered_asset_id"] != asset
+    assert len(service.db.all("SELECT * FROM job_runs")) == 2
+
+
+def test_retry_success_cancels_rest_and_keeps_schedule(app, service, monkeypatch):
+    scheduler, job, now, driver, upload = failed_job(app, service, monkeypatch)
+    policy = scheduler.upload_retries
+    plan = policy.load(job["id"])
+    deadline = service.db.one("SELECT next_run FROM jobs WHERE id=?", (job["id"],))[
+        "next_run"
+    ]
+    monkeypatch.setattr(driver, "upload", upload)
+    when = datetime.fromisoformat(plan["times"][0]) + timedelta(milliseconds=1)
+    scheduler.tick(when)
+    operation = policy.load(job["id"])["active"]
+    service.process_one()
+    service.sync()
+    scheduler.tick(when + timedelta(seconds=1))
+    assert policy.load(job["id"]) is None
+    assert service.operation(operation)["status"] == "COMPLETED"
+    assert (
+        service.db.one("SELECT next_run FROM jobs WHERE id=?", (job["id"],))["next_run"]
+        == deadline
+    )
+    assert (
+        service.db.one("SELECT kind FROM assets WHERE id=?", (plan["asset_id"],))[
+            "kind"
+        ]
+        == "generated"
+    )
+    assert driver.client.uploads == 1
+
+
+def test_expired_queued_retry_never_uploads(app, service, monkeypatch):
+    scheduler, job, now, driver, upload = failed_job(app, service, monkeypatch)
+    policy = scheduler.upload_retries
+    plan = policy.load(job["id"])
+    scheduler.tick(datetime.fromisoformat(plan["times"][0]) + timedelta(milliseconds=1))
+    plan = policy.load(job["id"])
+    plan["deadline"] = (datetime.now(UTC) - timedelta(seconds=1)).isoformat()
+    policy.save(job["id"], plan)
+
+    def unexpected(*a):
+        raise AssertionError("Expired retry must never upload")
+
+    monkeypatch.setattr(driver, "upload", unexpected)
+    service.process_one()
+    operation = service.operation(plan["active"])
+    assert operation["status"] == "FAILED"
+    assert operation["error"]["details"]["reason"] == "RETRY_WINDOW_CLOSED"
+    assert service.assets.path(plan["asset_id"]).exists()
+
+
+def test_lost_response_is_reconciled_before_retry_without_duplicate(
+    app, service, monkeypatch
+):
+    scheduler, job, now, driver, upload = failed_job(app, service, monkeypatch)
+    policy = scheduler.upload_retries
+    plan = policy.load(job["id"])
+    asset = service.assets.info(plan["asset_id"])
+    monkeypatch.setattr(driver, "upload", upload)
+    # The original request reached the cloud, but its response was lost.
+    driver.upload(service.assets.path(plan["asset_id"]), asset["name"])
+    service.update(plan["root"], "UNKNOWN")
+    scheduler.tick(datetime.fromisoformat(plan["times"][0]) + timedelta(milliseconds=1))
+    scheduler.tick(datetime.fromisoformat(plan["times"][0]) + timedelta(seconds=1))
+    assert service.operation(plan["root"])["status"] == "COMPLETED"
+    assert policy.load(job["id"]) is None
+    assert driver.client.uploads == 1