Skip to content

delivery-kid: port Coconut V2 finalize (httpstream) + disable broken fast path #94

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jMyles merged 1 commit into from
Jun 1, 2026
Merged

delivery-kid: port Coconut V2 finalize (httpstream) + disable broken fast path #94

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 20 additions & 21 deletions delivery-kid/pinning-service/app/routes/content.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -623,27 +623,26 @@ async def send_event(event: str, data: dict):
video_files = [f for f in state.files if f.media_type == "video"]
wants_transcode = len(state.files) == 1 and video_files and _should_transcode_video(request)

# === Fast path: preview already done, no trim ===
if wants_transcode and state.preview_cid and not has_trim:
logger.info("[content:%s] Using existing preview HLS: %s", draft_id[:8], state.preview_cid)
yield await send_event("progress", {
"stage": "transcode",
"message": "AV1 HLS already transcoded — using preview.",
"progress": 80
})

gateway_url = f"{settings.ipfs_gateway_url}/ipfs/{state.preview_cid}"

state.status = "finalized"
yield await send_event("complete", {
"cid": state.preview_cid,
"gateway_url": gateway_url,
"title": request.title,
"file_type": request.file_type,
"subsequent_to": request.subsequent_to,
})
pin_success = True
return
# === Fast path DISABLED for V2 migration ===
# Pre-V2-migration this branch reused state.preview_cid as the final
# Release CID. That was safe when preview produced full HLS variants.
#
# Under V2 the preview path now submits a single 'mp4' output
# (rendered inline on the wiki ReleaseDraft page), and the finalize
# path submits an 'httpstream' HLS tree — they produce different
# artifacts. Reusing preview_cid here would put an MP4 at the
# Release CID instead of HLS variants. Worse, when the V2 webhook
# download path hadn't been ported, preview_cid pointed at an IPFS
# dir containing only metadata.json — finalize blindly reused that
# and minted Release pages with no playable content (the Nine
# Pound Hammer regression).
#
# Always go through the finalize transcode path until we have a
# reliable signal that preview_cid points at finalize-grade HLS.
# When/if Coconut's V2 schema lets us produce HLS at preview time
# too, this fast path can be reinstated guarded by that.
# if wants_transcode and state.preview_cid and not has_trim:
# ... (see git blame for the V1-era body)

# === Coconut cloud transcoding (with trim, or no preview available) ===
if wants_transcode and _should_use_coconut(request, settings):
Expand Down
184 changes: 130 additions & 54 deletions delivery-kid/pinning-service/app/services/coconut.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,71 +63,70 @@ async def submit_to_coconut(
trim_end: float | None = None,
include_preview: bool = False,
) -> dict:
"""Submit a video to Coconut for transcoding.
"""Submit a video to Coconut V2 for transcoding.

Coconut V2 schema (see https://docs.coconut.co/jobs/api):
V2 schema (verified against api.coconut.co/v2/jobs — HTTP 201 on each):

{
"input": {"url": "..."},
"storage": {"service": "coconut"}, # host outputs themselves
"storage": {"service": "coconut"}, # Coconut hosts outputs
"outputs": {"<format-spec>": {"path": "..."}},
"notification": {"type": "http", "url": "..."}
}

The ``<format-spec>`` key encodes the container/quality (e.g. ``mp4``,
The ``<format-spec>`` key encodes the container/quality (``mp4``,
``mp4:480p``, ``httpstream``). Output values are thin wrappers around
``path``; codec/bitrate/audio customization is encoded into the key
rather than nested ``video``/``audio`` blocks (which V1 used and V2
rejects with ``output_param_not_valid``).
``path``; nested codec/bitrate ``video``/``audio`` blocks from V1 are
rejected as ``output_param_not_valid``.

With ``storage.service: "coconut"`` Coconut hosts the output for a
retention window and gives us URLs in the notification callback.
``process_completed_job`` already expects this pattern (downloads each
output URL and pins to IPFS), so no changes are needed downstream.

**Status of the port:**
- Preview path (``include_preview=True``) is V2-shape, single MP4
output. Verified end-to-end with a real ``curl`` to V2 — 201 from
Coconut.
- HLS / finalize path (``qualities``, ``trim_start``, ``trim_end``)
is not yet ported. V2 uses an ``httpstream`` output block whose
exact codec/variant shape we still need to nail down — see the
follow-up TODO. Raises NotImplementedError for now rather than
silently sending V1-shape and 400-ing.
retention window and posts the URLs in the notification callback.

Two job shapes today:
- **Preview** (``include_preview=True``): single ``mp4`` output. Light,
lets the wiki render an inline player on the ReleaseDraft page.
- **Finalize** (default): single ``httpstream`` output → HLS playlist
tree. Quality variant / trim / codec customization isn't yet exposed
because the V2 schema for those inside ``httpstream`` is still
undocumented and we get HTTP 201 on the minimal shape. Falls back to
Coconut's defaults for now.

Args:
source_url: Public URL of the source video (IPFS gateway URL).
source_url: Public URL of the source video.
api_key: Coconut API key.
webhook_url: URL Coconut will POST to on completion.
qualities: V1-era. Currently unsupported on V2 — pending HLS port.
trim_start: V1-era. Currently unsupported on V2 — pending HLS port.
trim_end: V1-era. Currently unsupported on V2 — pending HLS port.
include_preview: If True, submit a single 480p MP4 preview job.
This is the only supported shape today.
qualities: V1-era list of output heights (e.g. ``[720, 480]``).
Currently ignored on V2 — logged as a warning when non-default.
trim_start: V1-era trim offset seconds. Currently ignored on V2.
trim_end: V1-era trim end seconds. Currently ignored on V2.
include_preview: Switches between preview-mp4 and finalize-httpstream
output shapes (see above).

Returns:
Coconut API response dict (job id, status, etc.)
Coconut API response dict (job id, status, outputs array, etc.)

Raises:
NotImplementedError: If called for the HLS/finalize path until
the ``httpstream`` block shape is ported.
httpx.HTTPStatusError: If Coconut rejects the job. The exception
message includes Coconut's response body so the diagnostics
panel surfaces the real reason instead of a generic
``Client error '400 Bad Request' for url ...``.
"""
if not include_preview:
raise NotImplementedError(
"Coconut V2 HLS via 'httpstream' output block is not yet ported. "
"Only the preview MP4 path (include_preview=True) works today. "
"See follow-up issue for the finalize-side HLS port."
)
if include_preview:
outputs = {"mp4": {"path": "/preview.mp4"}}
else:
outputs = {"httpstream": {"hls": {"path": "/hls"}}}
if qualities or trim_start is not None or trim_end is not None:
logger.warning(
"submit_to_coconut: V2 finalize path doesn't yet pass "
"qualities/trim through to Coconut — using V2 defaults. "
"Ignored: qualities=%s trim_start=%s trim_end=%s",
qualities, trim_start, trim_end,
)

# Preview path — minimal V2 shape verified against api.coconut.co/v2/jobs.
job_config = {
"input": {"url": source_url},
"storage": {"service": "coconut"},
"outputs": {"mp4": {"path": "/preview.mp4"}},
"outputs": outputs,
"notification": {"type": "http", "url": webhook_url},
}

Expand All @@ -154,45 +153,122 @@ async def submit_to_coconut(
return resp.json()


def _normalize_outputs(outputs) -> list[dict]:
"""Normalise Coconut V1 dict-shape and V2 array-shape into a single
list of ``{"key": ..., "url": ..., ...}`` entries.

- V1 webhook callback: ``{"hls_av1_720p": {"url": ...}, ...}``
- V2 webhook callback: ``[{"key": "httpstream", "url": ..., ...}]``

Coconut's outputs schema changed when V2 rolled out; both shapes may
appear in flight until we've fully purged V1 references. Tolerate
either rather than failing closed.
"""
if isinstance(outputs, dict):
return [{"key": k, **(v if isinstance(v, dict) else {})}
for k, v in outputs.items()]
if isinstance(outputs, list):
return [o for o in outputs if isinstance(o, dict)]
return []


async def download_hls_outputs(
outputs: dict,
outputs,
hls_dir: Path,
) -> None:
"""Download HLS playlists and segments from Coconut output URLs.

Accepts both the V1 dict shape (``{"hls_av1_720p": {"url": ...}}``) and
the V2 list shape (``[{"key": "httpstream", "url": ...}]``). For V2's
single ``httpstream`` output, expects ``url`` to point at the master
playlist; we then chase variants + segments from there.

Args:
outputs: Dict of output key -> {url: ...} from Coconut webhook
hls_dir: Local directory to write files into
outputs: Coconut webhook ``outputs`` field — dict (V1) or list (V2).
hls_dir: Local directory to write files into.
"""
hls_dir.mkdir(parents=True, exist_ok=True)
normalized = _normalize_outputs(outputs)

async with httpx.AsyncClient(timeout=120.0) as client:
for key, output in outputs.items():
for output in normalized:
key = output.get("key", "")
url = output.get("url")
if not url:
continue

if key == "hls_master":
local_path = hls_dir / "master.m3u8"
if key == "httpstream":
# V2: single httpstream output points at the master playlist.
# Pull it, then chase variant playlists + segments.
logger.info("Downloading httpstream master from %s", url)
resp = await client.get(url)
if not resp.is_success:
logger.warning("Failed to download httpstream master: %s",
resp.status_code)
continue
(hls_dir / "master.m3u8").write_text(resp.text)
await _download_hls_variants_from_master(client, resp.text, url, hls_dir)
elif key == "hls_master":
# V1 legacy.
logger.info("Downloading hls_master from %s", url)
resp = await client.get(url)
if resp.is_success:
(hls_dir / "master.m3u8").write_text(resp.text)
elif key.startswith("hls_av1_"):
# V1 legacy: per-variant playlist.
quality = key.replace("hls_av1_", "").replace("p", "")
quality_dir = hls_dir / f"{quality}p"
quality_dir.mkdir(parents=True, exist_ok=True)
local_path = quality_dir / "playlist.m3u8"
else:
continue
logger.info("Downloading %s from %s", key, url)
resp = await client.get(url)
if resp.is_success:
local_path.write_text(resp.text)
await _download_segments(client, resp.text, url, quality_dir)


# Download playlist
logger.info("Downloading %s from %s", key, url)
resp = await client.get(url)
async def _download_hls_variants_from_master(
client: httpx.AsyncClient,
master_text: str,
master_url: str,
hls_dir: Path,
) -> None:
"""For V2 httpstream: parse master.m3u8, fetch each variant playlist
referenced by ``#EXT-X-STREAM-INF``, then fetch each variant's segments.

Master entries look like::
#EXT-X-STREAM-INF:BANDWIDTH=...,RESOLUTION=1280x720,...
720p/playlist.m3u8

The line after STREAM-INF is the (possibly-relative) variant URL.
"""
from urllib.parse import urljoin

lines = master_text.splitlines()
for i, line in enumerate(lines):
if not line.startswith("#EXT-X-STREAM-INF"):
continue
if i + 1 >= len(lines):
continue
variant_rel = lines[i + 1].strip()
if not variant_rel or variant_rel.startswith("#"):
continue
variant_url = urljoin(master_url, variant_rel)
# Use the directory part as the local subdir (e.g. "720p")
variant_subdir = variant_rel.rsplit("/", 1)[0] if "/" in variant_rel else ""
variant_dir = (hls_dir / variant_subdir) if variant_subdir else hls_dir
variant_dir.mkdir(parents=True, exist_ok=True)
variant_playlist_name = variant_rel.rsplit("/", 1)[-1]
try:
resp = await client.get(variant_url)
if not resp.is_success:
logger.warning("Failed to download %s: %s", key, resp.status_code)
logger.warning("Failed to fetch variant %s: %s",
variant_rel, resp.status_code)
continue
local_path.write_text(resp.text)

# Download segments referenced in playlist
if local_path.name.endswith(".m3u8") and key != "hls_master":
await _download_segments(client, resp.text, url, local_path.parent)
(variant_dir / variant_playlist_name).write_text(resp.text)
await _download_segments(client, resp.text, variant_url, variant_dir)
except Exception as e:
logger.warning("Error fetching variant %s: %s", variant_rel, e)


async def _download_segments(
Expand Down