Serve stored progressive playback objects

Author: richiemcilroy

scripts/backfill-public-playback-aliases.mjs

@@ -31,6 +31,7 @@ Options:
   --prefix PREFIX            Public alias prefix. Defaults to REND_PUBLIC_PLAYBACK_ALIAS_PREFIX or v.
   --acl public-read|inherit  Alias object ACL. Defaults to REND_PUBLIC_PLAYBACK_ALIAS_ACL or public-read.
   --public-base-url URL      Optional unauthenticated GET verification base URL.
+  --backfill-progressive     Create hls/<rendition>/progressive.mp4 source objects before alias copy.
   --env-file PATH            Load storage credentials from a specific env file.
   --dry-run                  List counts without writing aliases.
 `);
@@ -44,6 +45,7 @@ function parseArgs(argv) {
     prefix: process.env.REND_PUBLIC_PLAYBACK_ALIAS_PREFIX || "v",
     acl: process.env.REND_PUBLIC_PLAYBACK_ALIAS_ACL || "public-read",
     publicBaseUrl: "",
+    backfillProgressive: false,
     envFile: "",
     dryRun: false,
   };
@@ -84,6 +86,10 @@ function parseArgs(argv) {
       index += 1;
       continue;
     }
+    if (arg === "--backfill-progressive") {
+      options.backfillProgressive = true;
+      continue;
+    }
     if (arg === "--env-file") {
       options.envFile = String(next || "");
       index += 1;
@@ -438,6 +444,7 @@ function playbackArtifactKeysForAsset(allKeys, assetId) {
         relative.endsWith(".m3u8") ||
         relative.endsWith(".m4s") ||
         relative.endsWith(".ts") ||
+        relative.endsWith("/progressive.mp4") ||
         /\/init_[A-Za-z0-9_-]+\.mp4$/.test(relative)
       );
     })
@@ -466,6 +473,97 @@ function cacheControlForKey(key) {
   return "public, max-age=31536000, immutable";
 }
 
+const startupRenditions = ["360p", "480p", "720p", "1080p", "2k", "4k"];
+
+function mediaSegmentNamesFromPlaylist(playlistText) {
+  return playlistText
+    .split(/\r?\n/)
+    .map((line) => line.trim())
+    .filter((line) => {
+      if (!line || line.startsWith("#")) return false;
+      if (line.includes("/") || line.includes("\\") || line.includes("..")) {
+        return false;
+      }
+      return /^segment_[0-9]+\.m4s$/.test(line);
+    });
+}
+
+async function getObjectBuffer(env, key, bucket = env.S3_BUCKET) {
+  const response = await s3Fetch(env, {
+    method: "GET",
+    key,
+    bucket,
+  });
+  return Buffer.from(await response.arrayBuffer());
+}
+
+async function putObjectBuffer(
+  env,
+  key,
+  body,
+  contentType,
+  bucket = env.S3_BUCKET,
+) {
+  await s3Fetch(env, {
+    method: "PUT",
+    key,
+    bucket,
+    headers: {
+      "content-type": contentType,
+    },
+    body,
+  });
+}
+
+async function backfillProgressiveObjects(env, options, assetId, allKeys) {
+  if (!options.backfillProgressive) return allKeys;
+  const knownKeys = new Set(allKeys);
+  const writtenKeys = [];
+
+  for (const rendition of startupRenditions) {
+    const prefix = `videos/${assetId}/hls/${rendition}`;
+    const targetKey = `${prefix}/progressive.mp4`;
+    if (knownKeys.has(targetKey)) continue;
+
+    const playlistKey = `${prefix}/index.m3u8`;
+    const initKey = `${prefix}/init_${rendition}.mp4`;
+    if (!knownKeys.has(playlistKey) || !knownKeys.has(initKey)) continue;
+
+    const playlist = String(await getObjectBuffer(env, playlistKey));
+    const segmentNames = mediaSegmentNamesFromPlaylist(playlist);
+    const segmentKeys = segmentNames.map((name) => `${prefix}/${name}`);
+    if (!segmentKeys.length || segmentKeys.some((key) => !knownKeys.has(key))) {
+      continue;
+    }
+
+    log(
+      `asset=${assetId} progressive_plan rendition=${rendition} segments=${segmentKeys.length} dry_run=${options.dryRun}`,
+    );
+    if (options.dryRun) {
+      knownKeys.add(targetKey);
+      writtenKeys.push(targetKey);
+      continue;
+    }
+
+    const parts = [await getObjectBuffer(env, initKey)];
+    for (const segmentKey of segmentKeys) {
+      parts.push(await getObjectBuffer(env, segmentKey));
+    }
+    const body = Buffer.concat(parts);
+    await putObjectBuffer(env, targetKey, body, "video/mp4");
+    knownKeys.add(targetKey);
+    writtenKeys.push(targetKey);
+    log(
+      `asset=${assetId} progressive_written rendition=${rendition} bytes=${body.byteLength}`,
+    );
+  }
+
+  if (writtenKeys.length) {
+    return [...knownKeys].sort();
+  }
+  return allKeys;
+}
+
 async function mapLimit(items, limit, mapper) {
   let nextIndex = 0;
   const results = [];
@@ -540,7 +638,12 @@ async function verifyPublicGet(options, assetId) {
 
 async function backfillAsset(env, options, assetId) {
   const sourcePrefix = `videos/${assetId}/`;
-  const allKeys = await listObjectKeys(env, sourcePrefix, env.S3_BUCKET);
+  const allKeys = await backfillProgressiveObjects(
+    env,
+    options,
+    assetId,
+    await listObjectKeys(env, sourcePrefix, env.S3_BUCKET),
+  );
   const artifactKeys = playbackArtifactKeysForAsset(allKeys, assetId);
   if (!artifactKeys.some((key) => key.endsWith("/hls/master.m3u8"))) {
     throw new Error(`asset=${assetId} has no hls/master.m3u8 object`);
@@ -559,10 +662,13 @@ async function backfillAsset(env, options, assetId) {
   );
   if (options.dryRun) return;
 
-  let copiedBytes = 0;
-  await mapLimit(planned, 6, async ({ objectKey, aliasKey }) => {
-    copiedBytes += await putAlias(env, options, objectKey, aliasKey);
-  });
+  const copiedSizes = await mapLimit(
+    planned,
+    6,
+    async ({ objectKey, aliasKey }) =>
+      putAlias(env, options, objectKey, aliasKey),
+  );
+  const copiedBytes = copiedSizes.reduce((total, size) => total + size, 0);
   log(
     `asset=${assetId} aliases_written=${planned.length} bytes=${copiedBytes}`,
   );

services/rend-api/src/main.rs

@@ -2562,10 +2562,6 @@ async fn playback_origin_artifact_inner(
     )
     .map_err(|_| AppError::unauthorized("unauthorized playback token"))?;
 
-    if hls_progressive_rendition(&artifact.artifact_path).is_some() {
-        return origin_playback_progressive_fmp4_response(state, artifact).await;
-    }
-
     let range_header = if is_hls_manifest_artifact_path(&artifact.artifact_path) {
         None
     } else {
@@ -2575,6 +2571,24 @@ async fn playback_origin_artifact_inner(
             .and_then(normalize_single_byte_range_header)
     };
 
+    if hls_progressive_rendition(&artifact.artifact_path).is_some() {
+        match origin_playback_artifact_full_bytes(state.as_ref(), &artifact).await {
+            Ok((bytes, content_type, cache_status)) => {
+                return Ok(origin_playback_artifact_bytes_response(
+                    artifact,
+                    bytes,
+                    content_type,
+                    cache_status,
+                    range_header.as_deref(),
+                ));
+            }
+            Err(error) if error.status == StatusCode::NOT_FOUND => {
+                return origin_playback_progressive_fmp4_response(state, artifact).await;
+            }
+            Err(error) => return Err(error),
+        }
+    }
+
     let cache_ttl = origin_playback_cache_ttl(&artifact);
     if cache_ttl.is_some() {
         let (bytes, content_type, cache_status) =

services/rend-api/src/media.rs

@@ -10,7 +10,12 @@ use anyhow::{Context, Result};
 use aws_sdk_s3::{Client as S3Client, primitives::ByteStream, types::ObjectCannedAcl};
 use serde::Deserialize;
 use sqlx::{PgPool, Postgres, Transaction};
-use tokio::{fs, io, process::Command, time};
+use tokio::{
+    fs,
+    io::{self, AsyncWriteExt},
+    process::Command,
+    time,
+};
 
 use crate::{
     billing,
@@ -588,6 +593,56 @@ async fn generate_and_upload_hls(
             rendition.name
         );
 
+        let init_segment_path = segment_paths
+            .iter()
+            .find(|path| {
+                path.file_name()
+                    .and_then(|name| name.to_str())
+                    .is_some_and(is_hls_init_segment_name)
+            })
+            .cloned();
+        let media_segment_paths = segment_paths
+            .iter()
+            .filter(|path| {
+                path.file_name()
+                    .and_then(|name| name.to_str())
+                    .is_some_and(is_hls_media_segment_name)
+            })
+            .cloned()
+            .collect::<Vec<_>>();
+
+        if let Some(init_segment_path) = init_segment_path.as_ref()
+            && !media_segment_paths.is_empty()
+        {
+            let progressive_path = variant_dir.join("progressive.mp4");
+            write_progressive_fmp4(&progressive_path, init_segment_path, &media_segment_paths)
+                .await
+                .with_context(|| format!("failed to create {} progressive fMP4", rendition.name))?;
+            let progressive_duration_ms = media_segment_paths
+                .iter()
+                .map(|segment_path| {
+                    let file_name = segment_path.file_name()?.to_string_lossy();
+                    let duration_key = format!("{}/{}", rendition.name, file_name);
+                    segment_durations
+                        .get(&duration_key)
+                        .copied()
+                        .or(Some(i64::from(HLS_TARGET_SEGMENT_SECONDS) * 1_000))
+                })
+                .flatten()
+                .sum::<i64>();
+            let artifact = upload_generated_file(
+                request,
+                &progressive_path,
+                hls_progressive_object_key(&request.asset_id, rendition.name),
+                "video/mp4",
+                "segment",
+                Some(progressive_duration_ms),
+                Some(rendition.resolution_tier),
+            )
+            .await?;
+            artifacts.push(artifact);
+        }
+
         for segment_path in segment_paths {
             let file_name = segment_path
                 .file_name()
@@ -621,6 +676,40 @@ async fn generate_and_upload_hls(
     Ok(artifacts)
 }
 
+async fn write_progressive_fmp4(
+    output_path: &Path,
+    init_segment_path: &Path,
+    media_segment_paths: &[PathBuf],
+) -> Result<()> {
+    let mut output = fs::File::create(output_path)
+        .await
+        .with_context(|| format!("failed to create {}", output_path.display()))?;
+    append_file(&mut output, init_segment_path).await?;
+    for segment_path in media_segment_paths {
+        append_file(&mut output, segment_path).await?;
+    }
+    output
+        .flush()
+        .await
+        .with_context(|| format!("failed to flush {}", output_path.display()))?;
+    Ok(())
+}
+
+async fn append_file(output: &mut fs::File, input_path: &Path) -> Result<()> {
+    let mut input = fs::File::open(input_path)
+        .await
+        .with_context(|| format!("failed to open {}", input_path.display()))?;
+    let copied = io::copy(&mut input, output)
+        .await
+        .with_context(|| format!("failed to append {}", input_path.display()))?;
+    anyhow::ensure!(
+        copied > 0,
+        "media fragment {} is empty",
+        input_path.display()
+    );
+    Ok(())
+}
+
 async fn upload_generated_file(
     request: &ProcessMediaRequest,
     path: &Path,
@@ -1243,6 +1332,10 @@ pub fn hls_segment_object_key(asset_id: &str, rendition_name: &str, segment_name
     format!("videos/{asset_id}/hls/{rendition_name}/{segment_name}")
 }
 
+pub fn hls_progressive_object_key(asset_id: &str, rendition_name: &str) -> String {
+    format!("videos/{asset_id}/hls/{rendition_name}/progressive.mp4")
+}
+
 pub fn normalize_public_playback_alias_prefix(value: &str) -> Result<String> {
     let prefix = value.trim().trim_matches('/');
     anyhow::ensure!(
@@ -1360,6 +1453,10 @@ mod tests {
             hls_segment_object_key("asset-123", "720p", "segment_00000.m4s"),
             "videos/asset-123/hls/720p/segment_00000.m4s"
         );
+        assert_eq!(
+            hls_progressive_object_key("asset-123", "720p"),
+            "videos/asset-123/hls/720p/progressive.mp4"
+        );
     }
 
     #[test]