Stream session video ranges directly from storage

.dockerignore

@@ -11,4 +11,4 @@
 !start.py
 !storage_listing.py
 !playback_timing.py
-!backfill_playback_times.py
+!video_stream.py

Dockerfile

@@ -17,7 +17,7 @@ COPY requirements.txt ./
 RUN pip install --no-cache-dir -r requirements.txt
 COPY --from=node-build --chown=appuser:appuser /app/dist ./dist
 COPY --from=node-build --chown=appuser:appuser /app/node_modules ./node_modules
-COPY --chown=appuser:appuser package.json app.py start.py storage_listing.py playback_timing.py ./
+COPY --chown=appuser:appuser package.json app.py start.py storage_listing.py playback_timing.py video_stream.py ./
 USER appuser
 EXPOSE 7860
 CMD ["python", "start.py"]

app.py

@@ -32,6 +32,8 @@ from huggingface_hub import HfApi, hf_hub_download, snapshot_download
 from huggingface_hub.errors import RemoteEntryNotFoundError
 from storage_listing import session_files
 from playback_timing import action_times, frame_times
+from video_stream import video_response
+from starlette.concurrency import run_in_threadpool
 
 load_dotenv(Path(__file__).with_name(".env"))
 jobs = {}
@@ -774,8 +776,8 @@ def cleanup_deleted_session_files(session_id, storage_path, job_paths):
         logger.info("Finished background storage cleanup for session %s", session_id)
 
 
-def session_row(session_id, user_id, is_admin=False):
-    row = backend(f"/{session_id}", user_id, is_admin=is_admin)
+def session_row(session_id, user_id, is_admin=False, playback_times=True):
+    row = backend(f"/{session_id}" + ("" if playback_times else "?playbackTimes=false"), user_id, is_admin=is_admin)
     if row["path"] != f"sessions/{session_id}":
         raise HTTPException(409, "Session uses a different storage layout")
     return row
@@ -1025,18 +1027,17 @@ def results(session_id: UUID, request: Request):
 
 
 @app.api_route("/storage/{session_id}/videos/{filename}", methods=["GET", "HEAD"])
-def video(session_id: UUID, filename: str, request: Request):
-    row = session_row(session_id, request.state.user_id, request.state.is_admin)
+async def video(session_id: UUID, filename: str, request: Request):
     if Path(filename).name != filename or "\\" in filename:
         raise HTTPException(404, "Video not found")
-    path = hf_hub_download(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
-        filename=f'{row["path"]}/videos/{filename}', token=os.environ["HF_TOKEN"])
-    return FileResponse(path)
+    row = await run_in_threadpool(session_row, session_id, request.state.user_id, request.state.is_admin, False)
+    return await video_response(request, os.environ["STORAGE_REPO_ID"],
+                                f'{row["path"]}/videos/{filename}', os.environ["HF_TOKEN"])
 
 
 @app.api_route("/storage/{session_id}/posters/{filename}", methods=["GET", "HEAD"])
 def video_poster(session_id: UUID, filename: str, request: Request):
-    row = session_row(session_id, request.state.user_id, request.state.is_admin)
+    row = session_row(session_id, request.state.user_id, request.state.is_admin, playback_times=False)
     if Path(filename).name != filename or "\\" in filename or not filename.endswith(".webp"):
         raise HTTPException(404, "Preview not found")
     try:

src/analysis.ts

@@ -119,6 +119,7 @@ export function analysisRouter(db: SupabaseClient, secret: string, onCompleted?:
       const { data, error } = await query.maybeSingle();
       if (error) return res.status(503).json({ error: "Unable to load session" });
       if (!data) return res.sendStatus(404);
+      if (req.query.playbackTimes === "false") return res.json(data);
       const { data: actionTimes, error: timesError } = await db.from("actions")
         .select("action_id,start_time_seconds,end_time_seconds").eq("session_id", sessionId);
       if (timesError) {

src/server.ts

@@ -1243,7 +1243,7 @@ async function proxyAnalysisRequest(req: Request, res: Response) {
   }
   res.status(upstream.status);
   res.set("Cross-Origin-Resource-Policy", "cross-origin");
-  for (const name of ["content-type", "content-disposition", "cache-control", "accept-ranges", "content-range", "last-modified"]) {
+  for (const name of ["content-type", "content-length", "content-disposition", "cache-control", "accept-ranges", "content-range", "etag", "last-modified"]) {
     const value = upstream.headers.get(name);
     if (value) res.set(name, value);
   }

tests/test_video_stream.py

@@ -0,0 +1,120 @@
+import unittest
+from unittest.mock import AsyncMock, patch
+from uuid import UUID
+
+import httpx
+from fastapi import HTTPException
+from starlette.requests import Request
+
+import video_stream
+
+
+class ProbeStream(httpx.AsyncByteStream):
+    def __init__(self):
+        self.reads = 0
+        self.closed = False
+
+    async def __aiter__(self):
+        for chunk in (b"first", b"second"):
+            self.reads += 1
+            yield chunk
+
+    async def aclose(self):
+        self.closed = True
+
+
+def request(method="GET", headers=None):
+    return Request({"type": "http", "method": method, "path": "/storage/test/videos/camera_1.mp4",
+                    "headers": [(key.encode(), value.encode()) for key, value in (headers or {}).items()]})
+
+
+class VideoStreamingTest(unittest.IsolatedAsyncioTestCase):
+    async def test_streams_ranges_lazily_and_closes_after_client_cancels(self):
+        body = ProbeStream()
+        seen = []
+
+        def handler(req):
+            seen.append(req)
+            return httpx.Response(206, headers={"Content-Range": "bytes 0-10/1000000", "Content-Length": "11",
+                                                "Accept-Ranges": "bytes", "Content-Type": "video/mp4"}, stream=body)
+
+        client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+        with patch.object(video_stream.httpx, "AsyncClient", return_value=client):
+            response = await video_stream.video_response(request(headers={"range": "bytes=0-10", "cookie": "private-cookie"}),
+                                                         "test/storage", "sessions/id/videos/camera_1.mp4", "storage-token")
+        self.assertEqual(body.reads, 0)
+        self.assertEqual(response.status_code, 206)
+        self.assertEqual(response.headers["content-range"], "bytes 0-10/1000000")
+        self.assertEqual(seen[0].headers["range"], "bytes=0-10")
+        self.assertNotIn("cookie", seen[0].headers)
+        self.assertEqual(await anext(response.body_iterator), b"first")
+        await response.body_iterator.aclose()
+        await response.background()
+        self.assertEqual(body.reads, 1)
+        self.assertTrue(body.closed)
+        self.assertTrue(client.is_closed)
+
+    async def test_cdn_redirect_strips_authorization_and_keeps_range(self):
+        seen = []
+
+        def handler(req):
+            seen.append(req)
+            if req.url.host == "huggingface.co":
+                return httpx.Response(302, headers={"Location": "https://cdn.example/video?signature=test"})
+            return httpx.Response(206, headers={"Content-Range": "bytes 0-1/10"}, stream=ProbeStream())
+
+        client = httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=True)
+        with patch.object(video_stream.httpx, "AsyncClient", return_value=client):
+            response = await video_stream.video_response(request(headers={"range": "bytes=0-1"}),
+                                                         "test/storage", "video.mp4", "storage-token")
+        self.assertEqual(seen[0].headers["authorization"], "Bearer storage-token")
+        self.assertNotIn("authorization", seen[1].headers)
+        self.assertEqual(seen[1].headers["range"], "bytes=0-1")
+        await response.background()
+
+    async def test_head_preserves_size_without_reading_video(self):
+        body = ProbeStream()
+        client = httpx.AsyncClient(transport=httpx.MockTransport(lambda req: httpx.Response(
+            200, headers={"Content-Length": "1000000", "Accept-Ranges": "bytes"}, stream=body)))
+        with patch.object(video_stream.httpx, "AsyncClient", return_value=client):
+            response = await video_stream.video_response(request("HEAD"), "test/storage", "video.mp4", "token")
+        self.assertEqual(response.headers["content-length"], "1000000")
+        self.assertEqual(response.body, b"")
+        self.assertEqual(body.reads, 0)
+        self.assertTrue(client.is_closed)
+
+    async def test_invalid_range_keeps_total_size_and_omits_storage_error_body(self):
+        client = httpx.AsyncClient(transport=httpx.MockTransport(lambda req: httpx.Response(
+            416, headers={"Content-Range": "bytes */100", "Content-Length": "100"}, stream=ProbeStream())))
+        with patch.object(video_stream.httpx, "AsyncClient", return_value=client):
+            response = await video_stream.video_response(request(headers={"range": "bytes=999-"}), "test/storage", "video.mp4", "token")
+        self.assertEqual(response.status_code, 416)
+        self.assertEqual(response.headers["content-range"], "bytes */100")
+        self.assertEqual(response.headers["content-length"], "0")
+        self.assertTrue(client.is_closed)
+
+    async def test_upstream_errors_close_connections_and_hide_storage_details(self):
+        for status, expected in ((404, 404), (401, 502), (503, 502)):
+            with self.subTest(status=status):
+                client = httpx.AsyncClient(transport=httpx.MockTransport(lambda req: httpx.Response(status, stream=ProbeStream())))
+                with patch.object(video_stream.httpx, "AsyncClient", return_value=client):
+                    with self.assertRaises(HTTPException) as error:
+                        await video_stream.video_response(request(), "test/storage", "video.mp4", "private-token")
+                self.assertEqual(error.exception.status_code, expected)
+                self.assertNotIn("private-token", error.exception.detail)
+                self.assertTrue(client.is_closed)
+
+    async def test_video_route_checks_ownership_before_streaming(self):
+        import app as analysis_app
+        req = request()
+        req.state.user_id = "11111111-1111-4111-8111-111111111111"
+        req.state.is_admin = False
+        with patch.object(analysis_app, "session_row", side_effect=HTTPException(404, "Session not found")), \
+                patch.object(analysis_app, "video_response", new_callable=AsyncMock) as stream:
+            with self.assertRaises(HTTPException):
+                await analysis_app.video(UUID("22222222-2222-4222-8222-222222222222"), "camera_1.mp4", req)
+            stream.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/video-session.test.ts

@@ -0,0 +1,58 @@
+import assert from "node:assert/strict";
+import { createServer } from "node:http";
+import type { AddressInfo } from "node:net";
+import { test } from "node:test";
+import express from "express";
+import type { SupabaseClient } from "@supabase/supabase-js";
+
+test("media lookups skip timing data while retaining session ownership checks", async () => {
+  Object.assign(process.env, {
+    NODE_ENV: "test", FRONTEND_URL: "http://127.0.0.1:3000", BACKEND_URL: "http://127.0.0.1:4000",
+    ANALYSIS_PROXY_SECRET: "test-analysis-proxy-secret-32-chars", SUPABASE_URL: "http://127.0.0.1:54321",
+    SUPABASE_ANON_KEY: "test-anon-key", SUPABASE_SERVICE_ROLE_KEY: "test-service-role-key",
+  });
+  const { analysisRouter } = await import("../src/analysis.js");
+  const sessionId = "11111111-1111-4111-8111-111111111111";
+  const userId = "22222222-2222-4222-8222-222222222222";
+  const times = [{ action_id: sessionId, start_time_seconds: 1, end_time_seconds: 2 }];
+  const filters: [string, string][] = [];
+  const tables: string[] = [];
+  let owned = true;
+  const db = {
+    from(table: string) {
+      tables.push(table);
+      const query = {
+        select() { return query; },
+        eq(column: string, value: string) { filters.push([column, value]); return query; },
+        async maybeSingle() { return { data: owned ? { session_id: sessionId, user_id: userId } : null, error: null }; },
+        then(resolve: (value: unknown) => void) { return Promise.resolve({ data: times, error: null }).then(resolve); },
+      };
+      return query;
+    },
+  } as unknown as SupabaseClient;
+  const app = express();
+  const secret = process.env.ANALYSIS_PROXY_SECRET!;
+  app.use("/internal/analysis", analysisRouter(db, secret));
+  const server = createServer(app);
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+  const url = `http://127.0.0.1:${(server.address() as AddressInfo).port}/internal/analysis/${sessionId}`;
+  const headers = { "x-sporalize-proxy-secret": secret, "x-sporalize-user-id": userId };
+  try {
+    const media = await fetch(`${url}?playbackTimes=false`, { headers });
+    assert.equal(media.status, 200);
+    assert.deepEqual(tables, ["sessions"]);
+    assert.ok(filters.some(([column, value]) => column === "user_id" && value === userId));
+    tables.length = 0;
+    const results = await fetch(url, { headers });
+    assert.equal(results.status, 200);
+    assert.deepEqual((await results.json()).action_times, times);
+    assert.deepEqual(tables, ["sessions", "actions"]);
+    owned = false;
+    tables.length = 0;
+    const denied = await fetch(`${url}?playbackTimes=false`, { headers });
+    assert.equal(denied.status, 404);
+    assert.deepEqual(tables, ["sessions"]);
+  } finally {
+    await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
+  }
+});

video_stream.py

@@ -0,0 +1,53 @@
+"""Stream authenticated storage video ranges without downloading the full file first."""
+
+import httpx
+from fastapi import HTTPException, Request
+from huggingface_hub import hf_hub_url
+from starlette.background import BackgroundTask
+from starlette.responses import Response, StreamingResponse
+
+
+async def video_response(request: Request, repo_id: str, path: str, token: str):
+    # Browser cookies and user headers never go to the storage service.
+    headers = {"Authorization": f"Bearer {token}", "Accept-Encoding": "identity"}
+    for name in ("range", "if-range"):
+        if request.headers.get(name):
+            headers[name] = request.headers[name]
+    client = httpx.AsyncClient(follow_redirects=True, timeout=httpx.Timeout(60, connect=15))
+    try:
+        upstream = await client.send(client.build_request(
+            request.method, hf_hub_url(repo_id, path, repo_type="dataset"), headers=headers), stream=True)
+    except httpx.HTTPError:
+        await client.aclose()
+        raise HTTPException(502, "Video storage is temporarily unavailable") from None
+
+    async def close():
+        try:
+            await upstream.aclose()
+        finally:
+            await client.aclose()
+
+    if upstream.status_code not in (200, 206, 416):
+        await close()
+        raise HTTPException(404 if upstream.status_code == 404 else 502,
+                            "Video not found" if upstream.status_code == 404 else "Video storage is temporarily unavailable")
+    response_headers = {name: upstream.headers[name] for name in (
+        "content-type", "content-length", "content-range", "accept-ranges", "etag", "last-modified"
+    ) if name in upstream.headers}
+    response_headers["Cache-Control"] = "private, no-store"
+    if request.method == "HEAD" or upstream.status_code == 416:
+        await close()
+        # A 416 may have a storage error body: don't forward its body length with an empty response.
+        if upstream.status_code == 416:
+            response_headers.pop("content-length", None)
+        return Response(status_code=upstream.status_code, headers=response_headers)
+
+    async def chunks():
+        try:
+            async for chunk in upstream.aiter_raw():
+                yield chunk
+        finally:
+            await close()
+
+    return StreamingResponse(chunks(), status_code=upstream.status_code, headers=response_headers,
+                             background=BackgroundTask(close))