SLICE-7: images pipeline end-to-end — fetch/host/serve + INV-12 over images (SD-0002 §7.2) #30
@@ -89,23 +89,28 @@ def _process_one(pool, store, image: dict, allow_private: bool) -> None:
|
|||||||
with pool.connection() as conn:
|
with pool.connection() as conn:
|
||||||
repo.mark_image_failed(conn, image_id, str(exc)[:200]); conn.commit()
|
repo.mark_image_failed(conn, image_id, str(exc)[:200]); conn.commit()
|
||||||
return
|
return
|
||||||
result = images_mod.process(data)
|
try:
|
||||||
if isinstance(result, images_mod.Rejected):
|
result = images_mod.process(data)
|
||||||
reason = ("below the resolution bar" if result.reason == "rejected_low_res"
|
if isinstance(result, images_mod.Rejected):
|
||||||
else "not a supported image")
|
reason = ("below the resolution bar" if result.reason == "rejected_low_res"
|
||||||
|
else "not a supported image")
|
||||||
|
with pool.connection() as conn:
|
||||||
|
repo.mark_image_rejected(conn, image_id, result.reason, reason); conn.commit()
|
||||||
|
return
|
||||||
|
keys = {"original": _key(sf, image_id, "original"),
|
||||||
|
"thumb": _key(sf, image_id, "thumb.webp"),
|
||||||
|
"card": _key(sf, image_id, "card.webp"),
|
||||||
|
"detail": _key(sf, image_id, "detail.webp")}
|
||||||
|
store.put(keys["original"], data, ctype)
|
||||||
|
store.put(keys["thumb"], result.renditions["thumb"], "image/webp")
|
||||||
|
store.put(keys["card"], result.renditions["card"], "image/webp")
|
||||||
|
store.put(keys["detail"], result.renditions["detail"], "image/webp")
|
||||||
with pool.connection() as conn:
|
with pool.connection() as conn:
|
||||||
repo.mark_image_rejected(conn, image_id, result.reason, reason); conn.commit()
|
repo.mark_image_fetched(conn, image_id, keys); conn.commit()
|
||||||
return
|
except Exception as exc: # noqa: BLE001 — one bad image must never wedge the run (review fix)
|
||||||
keys = {"original": _key(sf, image_id, "original"),
|
with pool.connection() as conn:
|
||||||
"thumb": _key(sf, image_id, "thumb.webp"),
|
repo.mark_image_failed(conn, image_id, f"processing error: {type(exc).__name__}")
|
||||||
"card": _key(sf, image_id, "card.webp"),
|
conn.commit()
|
||||||
"detail": _key(sf, image_id, "detail.webp")}
|
|
||||||
store.put(keys["original"], data, ctype)
|
|
||||||
store.put(keys["thumb"], result.renditions["thumb"], "image/webp")
|
|
||||||
store.put(keys["card"], result.renditions["card"], "image/webp")
|
|
||||||
store.put(keys["detail"], result.renditions["detail"], "image/webp")
|
|
||||||
with pool.connection() as conn:
|
|
||||||
repo.mark_image_fetched(conn, image_id, keys); conn.commit()
|
|
||||||
|
|
||||||
|
|
||||||
def run_image_phase(pool, store, run_id: int, allow_private: bool) -> None:
|
def run_image_phase(pool, store, run_id: int, allow_private: bool) -> None:
|
||||||
|
|||||||
@@ -60,7 +60,7 @@ def process(data: bytes) -> Processed | Rejected:
|
|||||||
height=height,
|
height=height,
|
||||||
renditions=renditions,
|
renditions=renditions,
|
||||||
)
|
)
|
||||||
except (UnidentifiedImageError, OSError, ValueError):
|
except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError):
|
||||||
return Rejected(reason="rejected_not_image")
|
return Rejected(reason="rejected_not_image")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -53,3 +53,13 @@ def test_not_an_image_rejected_not_image():
|
|||||||
def test_jpeg_source_format_preserved_in_result():
|
def test_jpeg_source_format_preserved_in_result():
|
||||||
result = images.process(_jpeg(800, 800))
|
result = images.process(_jpeg(800, 800))
|
||||||
assert result.source_format == "JPEG"
|
assert result.source_format == "JPEG"
|
||||||
|
|
||||||
|
|
||||||
|
def test_decompression_bomb_rejected_not_image(monkeypatch):
|
||||||
|
# A small file whose pixel count exceeds Pillow's bomb threshold must be a
|
||||||
|
# typed rejection, never an escaping exception.
|
||||||
|
from PIL import Image as PILImage
|
||||||
|
monkeypatch.setattr(PILImage, "MAX_IMAGE_PIXELS", 100) # 10x10 exceeds it
|
||||||
|
result = images.process(_png(800, 800))
|
||||||
|
assert isinstance(result, images.Rejected)
|
||||||
|
assert result.reason == "rejected_not_image"
|
||||||
|
|||||||
@@ -121,6 +121,28 @@ def test_run_phase_classifies_each_image(pool, host, tmp_path):
|
|||||||
assert _run_status(pool, rid) == "complete_with_problems"
|
assert _run_status(pool, rid) == "complete_with_problems"
|
||||||
|
|
||||||
|
|
||||||
|
def test_unexpected_processing_error_marks_image_failed_not_wedged(pool, host, tmp_path, monkeypatch):
|
||||||
|
# If anything after the fetch raises unexpectedly (e.g. objectstore.put),
|
||||||
|
# the image is marked failed and the run still reaches a terminal status —
|
||||||
|
# never stranded in fetching_images.
|
||||||
|
store = objectstore.LocalObjectStore(str(tmp_path))
|
||||||
|
sf, _acct, rid = _seed(pool)
|
||||||
|
p = _product(pool, sf, "lamp")
|
||||||
|
iid = _image(pool, p, f"{host}/good.png", rid, position=1)
|
||||||
|
|
||||||
|
def _boom(*a, **k):
|
||||||
|
raise RuntimeError("storage down")
|
||||||
|
monkeypatch.setattr(store, "put", _boom)
|
||||||
|
|
||||||
|
imagefetch.run_image_phase(pool, store, rid, allow_private=True)
|
||||||
|
with pool.connection() as conn:
|
||||||
|
counts = repo.run_image_counts(conn, rid)
|
||||||
|
assert counts["failed"] == 1 and counts["pending"] == 0
|
||||||
|
assert _run_status(pool, rid) == "complete_with_problems"
|
||||||
|
row = _img_row(pool, iid)
|
||||||
|
assert row["status"] == "failed"
|
||||||
|
|
||||||
|
|
||||||
def test_resume_after_kill_completes_remaining(pool, host, tmp_path):
|
def test_resume_after_kill_completes_remaining(pool, host, tmp_path):
|
||||||
store = objectstore.LocalObjectStore(str(tmp_path))
|
store = objectstore.LocalObjectStore(str(tmp_path))
|
||||||
sf, _acct, rid = _seed(pool)
|
sf, _acct, rid = _seed(pool)
|
||||||
|
|||||||
+17
-4
@@ -155,12 +155,25 @@ them concurrently:
|
|||||||
and multicast ranges are blocked); redirects are followed only within the
|
and multicast ranges are blocked); redirects are followed only within the
|
||||||
same guard — a redirect to a private IP is rejected mid-chain.
|
same guard — a redirect to a private IP is rejected mid-chain.
|
||||||
- **Bounds:** ≤ 20 MB response body, ≤ 30 s per fetch.
|
- **Bounds:** ≤ 20 MB response body, ≤ 30 s per fetch.
|
||||||
- **Per-image idempotent claim:** each `CatalogImage` row is claimed before
|
- **Per-image presence/idempotency check (`claim_image_for_fetch`):** before
|
||||||
fetch; a restart that finds a row already `claimed` skips it — making the
|
processing, each `CatalogImage` row is checked so an already-handled row is
|
||||||
phase resumable.
|
skipped. This is a presence guard adequate for the **single-worker,
|
||||||
|
in-process** model — the deployed app runs one uvicorn worker, recovery runs
|
||||||
|
at startup over prior-process runs, and a fresh confirm always targets a new
|
||||||
|
run, so two phase invocations never process the same run's images
|
||||||
|
concurrently. It is **not** a cross-process lock. A true multi-worker claim
|
||||||
|
would need a distinct in-flight status transition (`pending → fetching` with
|
||||||
|
`RETURNING`, or `SELECT … FOR UPDATE SKIP LOCKED`) — that is the future seam
|
||||||
|
if the fetch phase ever moves multi-process or onto a queue (§6.9).
|
||||||
- **Startup recovery scan:** on process start the app scans for runs stuck in
|
- **Startup recovery scan:** on process start the app scans for runs stuck in
|
||||||
`fetching_images` and re-enqueues their pending images (emits TEL-5). This
|
`fetching_images` and re-enqueues their pending images (emits TEL-5). This
|
||||||
covers crash/restart scenarios without a separate scheduler.
|
covers crash/restart scenarios without a separate scheduler. On a mid-fetch
|
||||||
|
process restart (e.g. a deploy) the phase is interrupted and the run is left
|
||||||
|
resumable: pending images stay `pending`, fetched ones stay `fetched`, and
|
||||||
|
the startup scan completes the run. Any pool-closed error in an in-flight
|
||||||
|
worker at shutdown is benign — the uncommitted image stays `pending` and is
|
||||||
|
re-fetched on resume. Mid-flight interruption is tolerated by construction
|
||||||
|
(§6.9).
|
||||||
- **Progress tracking:** the run row carries `image_progress`,
|
- **Progress tracking:** the run row carries `image_progress`,
|
||||||
`image_counts`, and `image_outcomes` (updated per image); these fields feed
|
`image_counts`, and `image_outcomes` (updated per image); these fields feed
|
||||||
the run-detail UI panel.
|
the run-detail UI panel.
|
||||||
|
|||||||
Reference in New Issue
Block a user