Add chunked, resumable dashcam upload to work around Cloudflare's 100MB body limit
Confirmed against the real deployed server: Cloudflare Tunnel's free tier caps request bodies at ~100MB and returns a plain 413 - a direct LAN upload of the same 250MB test file succeeds, the identical request through yaris.ceaf.dev does not. Real dashcam clips average ~320MB, so no dashcam video could ever complete an upload through the tunnel as it stood, regardless of the earlier memory-streaming fix. Server: new /api/v1/dashcam/upload/chunk (accepts one chunk at a time, keyed by relative_path+sha256+chunk_index/total_chunks, reassembles and verifies the sha256 once every chunk has arrived) and /api/v1/dashcam/upload/status (reports which chunk indices a given upload already has, or "already_uploaded" if it's done) endpoints. Chunks are staged under STORAGE_DIR/.upload_tmp/<device>/<sha256>/ and cleaned up after reassembly; a stale partial set left by a differently -chunked earlier attempt is detected via a small marker file and wiped before restarting. The original single-shot /upload endpoint is left in place unchanged. Client (modules/app_upload.py): _upload_dashcam_files() now calls _upload_video_chunked(), which checks /status first (so a prior partial upload resumes instead of restarting from byte 0) and then POSTs any missing 40MB chunks (DASHCAM_CHUNK_SIZE, comfortably under Cloudflare's cap) via the same streaming MultipartEncoder approach as before. Co-authored-by: Copilot <[email protected]>
This commit is contained in:
Mathieu Moran
+68
-23
@@ -20,6 +20,16 @@ they simply wait for connectivity to return, same as before.
|
||||
*** default (YARIS_CELLULAR_ENABLED unset) so this has zero effect on the
|
||||
*** already-verified WiFi upload path unless explicitly configured.
|
||||
|
||||
Dashcam files are sent in DASHCAM_CHUNK_SIZE pieces via
|
||||
/api/v1/dashcam/upload/chunk rather than one request per file - confirmed
|
||||
against the real deployed server that Cloudflare Tunnel's free-tier ~100MB
|
||||
request-body limit otherwise rejects real dashcam clips (300+MB) outright
|
||||
with a 413, and that memory-safe streaming alone (MultipartEncoder) doesn't
|
||||
help there since the *proxy*, not this app, is what enforces the cap.
|
||||
Chunking also makes uploads resumable: a /status check before sending finds
|
||||
out which chunks the server already has, so an interrupted transfer picks
|
||||
back up instead of restarting from byte 0.
|
||||
|
||||
Credentials/server URL are loaded from db/upload_creds.env (gitignored,
|
||||
never commit real values - mirrors the db/creds.env pattern already used by
|
||||
app_db.py for MongoDB).
|
||||
@@ -48,6 +58,13 @@ TRACKING_BATCH_SIZE = 500
|
||||
BACKOFF_INITIAL = 5 # seconds, doubles on each consecutive failure
|
||||
BACKOFF_MAX = 300
|
||||
|
||||
# Real dashcam clips run 300+MB - Cloudflare Tunnel's free-tier ~100MB max
|
||||
# request body size rejects those outright with a 413 (confirmed against
|
||||
# the real tunnel: a 250MB file uploads fine direct-LAN but 413s through
|
||||
# yaris.ceaf.dev). Files are split into chunks this size and sent one at a
|
||||
# time via /api/v1/dashcam/upload/chunk instead, comfortably under that cap.
|
||||
DASHCAM_CHUNK_SIZE = 40 * 1024 * 1024
|
||||
|
||||
DASHCAM_UPLOAD_SETTING_KEY = "dashcam_upload_enabled"
|
||||
|
||||
_upload_thread = None
|
||||
@@ -161,6 +178,56 @@ def _upload_tracking_batch():
|
||||
return len(rows)
|
||||
|
||||
|
||||
def _upload_video_chunked(abs_path, rel_path, sha256):
|
||||
"""Upload one dashcam video file in DASHCAM_CHUNK_SIZE pieces, resuming
|
||||
from whatever the server already has for this exact (path, sha256)
|
||||
combo. Raises on any request-level failure - the caller's normal
|
||||
backlog/backoff handling takes it from there; a retry will simply
|
||||
re-query /status and resume from the first missing chunk instead of
|
||||
starting over.
|
||||
"""
|
||||
status_resp = requests.get(
|
||||
f"{SERVER_URL}/api/v1/dashcam/upload/status",
|
||||
params={"relative_path": rel_path, "sha256": sha256},
|
||||
headers=_headers(),
|
||||
timeout=15,
|
||||
)
|
||||
status_resp.raise_for_status()
|
||||
status = status_resp.json()
|
||||
if status.get("status") == "already_uploaded":
|
||||
return
|
||||
|
||||
already_received = set(status.get("received_chunk_indices", []))
|
||||
|
||||
size = os.path.getsize(abs_path)
|
||||
total_chunks = max(1, -(-size // DASHCAM_CHUNK_SIZE)) # ceil division
|
||||
|
||||
with open(abs_path, "rb") as f:
|
||||
for idx in range(total_chunks):
|
||||
if idx in already_received:
|
||||
continue
|
||||
|
||||
f.seek(idx * DASHCAM_CHUNK_SIZE)
|
||||
chunk_bytes = f.read(DASHCAM_CHUNK_SIZE)
|
||||
|
||||
encoder = MultipartEncoder(
|
||||
fields={
|
||||
"relative_path": rel_path,
|
||||
"sha256": sha256,
|
||||
"chunk_index": str(idx),
|
||||
"total_chunks": str(total_chunks),
|
||||
"file": (f"chunk_{idx:06d}", chunk_bytes, "application/octet-stream"),
|
||||
}
|
||||
)
|
||||
resp = requests.post(
|
||||
f"{SERVER_URL}/api/v1/dashcam/upload/chunk",
|
||||
data=encoder,
|
||||
headers={**_headers(), "Content-Type": encoder.content_type},
|
||||
timeout=120,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
|
||||
|
||||
def _upload_dashcam_files():
|
||||
"""Upload any not-yet-uploaded dashcam files, one at a time. Returns count sent.
|
||||
|
||||
@@ -176,29 +243,7 @@ def _upload_dashcam_files():
|
||||
if app_sql.dashcam_file_marked_uploaded(rel_path, sha256):
|
||||
continue
|
||||
|
||||
# Stream the multipart body via MultipartEncoder instead of requests'
|
||||
# plain files=/data= kwargs. requests builds a non-streamed files=
|
||||
# upload by reading the *entire* file into memory up front to
|
||||
# assemble the multipart body - fine for small files, but real
|
||||
# dashcam clips run 300+MB each, and on a ~1GB-RAM Pi 3 that alone is
|
||||
# enough to trigger the OOM killer (confirmed against real hardware:
|
||||
# RSS spiked in lockstep with dashcam file size). MultipartEncoder
|
||||
# reads the file in small chunks as the request is sent instead.
|
||||
with open(abs_path, "rb") as f:
|
||||
encoder = MultipartEncoder(
|
||||
fields={
|
||||
"relative_path": rel_path,
|
||||
"sha256": sha256,
|
||||
"file": (os.path.basename(abs_path), f, "application/octet-stream"),
|
||||
}
|
||||
)
|
||||
resp = requests.post(
|
||||
f"{SERVER_URL}/api/v1/dashcam/upload",
|
||||
data=encoder,
|
||||
headers={**_headers(), "Content-Type": encoder.content_type},
|
||||
timeout=300,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
_upload_video_chunked(abs_path, rel_path, sha256)
|
||||
app_sql.mark_dashcam_file_uploaded(rel_path, sha256)
|
||||
uploaded_count += 1
|
||||
|
||||
|
||||
+142
@@ -3,6 +3,7 @@ uploaded by the Pi, storing them in Postgres + on-disk file storage.
|
||||
"""
|
||||
import hashlib
|
||||
import os
|
||||
import shutil
|
||||
|
||||
from flask import Flask, request, jsonify, g
|
||||
|
||||
@@ -112,6 +113,147 @@ def upload_dashcam_file():
|
||||
return jsonify(status="ok", size_bytes=size_bytes), 201
|
||||
|
||||
|
||||
# --- Chunked dashcam upload ---------------------------------------------
|
||||
# Real dashcam clips run 300+MB, well past Cloudflare Tunnel's ~100MB max
|
||||
# request body size (confirmed with a real 250MB file: a direct LAN upload
|
||||
# succeeds, the same request through the tunnel comes back 413). The Pi
|
||||
# client splits files into small chunks (see CHUNK_SIZE in
|
||||
# modules/app_upload.py) and PUTs them one at a time through this endpoint
|
||||
# instead, staying comfortably under that limit. Chunks for a given upload
|
||||
# are staged on disk under _CHUNK_TMP_DIR and reassembled once complete.
|
||||
|
||||
_CHUNK_TMP_DIRNAME = ".upload_tmp"
|
||||
|
||||
|
||||
def _chunk_tmp_dir(device_name, sha256):
|
||||
return os.path.join(STORAGE_DIR, _CHUNK_TMP_DIRNAME, device_name, sha256)
|
||||
|
||||
|
||||
def _received_chunk_indices(tmp_dir):
|
||||
if not os.path.isdir(tmp_dir):
|
||||
return set()
|
||||
indices = set()
|
||||
for name in os.listdir(tmp_dir):
|
||||
if name.endswith(".chunk"):
|
||||
try:
|
||||
indices.add(int(name[: -len(".chunk")]))
|
||||
except ValueError:
|
||||
continue
|
||||
return indices
|
||||
|
||||
|
||||
@app.get("/api/v1/dashcam/upload/status")
|
||||
@require_device_auth
|
||||
def dashcam_upload_status():
|
||||
"""Report upload progress for (relative_path, sha256) so the Pi client
|
||||
can resume a chunked upload without re-sending chunks the server
|
||||
already has (e.g. after a network drop or app restart mid-transfer)."""
|
||||
relative_path = request.args.get("relative_path", "").strip()
|
||||
sha256 = request.args.get("sha256", "").strip().lower()
|
||||
if not relative_path or not sha256:
|
||||
return jsonify(error="missing 'relative_path' or 'sha256' query param"), 400
|
||||
|
||||
if db.dashcam_file_exists(g.device_id, relative_path, sha256):
|
||||
return jsonify(status="already_uploaded"), 200
|
||||
|
||||
tmp_dir = _chunk_tmp_dir(g.device_name, sha256)
|
||||
received = sorted(_received_chunk_indices(tmp_dir))
|
||||
return jsonify(status="in_progress" if received else "not_started", received_chunk_indices=received), 200
|
||||
|
||||
|
||||
@app.post("/api/v1/dashcam/upload/chunk")
|
||||
@require_device_auth
|
||||
def dashcam_upload_chunk():
|
||||
"""Receive one chunk of a dashcam video file.
|
||||
|
||||
Multipart form fields:
|
||||
file: the chunk's bytes
|
||||
relative_path: original sub-folder path on the dashcam SD card
|
||||
sha256: hex digest of the *whole* (reassembled) file
|
||||
chunk_index: 0-based index of this chunk
|
||||
total_chunks: total number of chunks for this file
|
||||
|
||||
Once every chunk 0..total_chunks-1 has arrived, the chunks are
|
||||
concatenated in order, the combined sha256 is verified, and the result
|
||||
is recorded exactly like the single-shot /upload endpoint.
|
||||
"""
|
||||
if "file" not in request.files:
|
||||
return jsonify(error="missing 'file' in multipart form"), 400
|
||||
|
||||
relative_path = request.form.get("relative_path", "").strip()
|
||||
expected_sha256 = request.form.get("sha256", "").strip().lower()
|
||||
try:
|
||||
chunk_index = int(request.form.get("chunk_index", ""))
|
||||
total_chunks = int(request.form.get("total_chunks", ""))
|
||||
except ValueError:
|
||||
return jsonify(error="'chunk_index'/'total_chunks' must be integers"), 400
|
||||
|
||||
if not relative_path or not expected_sha256 or total_chunks < 1 or not (0 <= chunk_index < total_chunks):
|
||||
return jsonify(error="missing/invalid form fields"), 400
|
||||
|
||||
device_id = g.device_id
|
||||
if db.dashcam_file_exists(device_id, relative_path, expected_sha256):
|
||||
return jsonify(status="already_uploaded"), 200
|
||||
|
||||
device_dir = os.path.join(STORAGE_DIR, g.device_name)
|
||||
dest_path = os.path.normpath(os.path.join(device_dir, relative_path))
|
||||
if os.path.commonpath([device_dir, dest_path]) != os.path.normpath(device_dir):
|
||||
return jsonify(error="invalid relative_path"), 400
|
||||
|
||||
tmp_dir = _chunk_tmp_dir(g.device_name, expected_sha256)
|
||||
# A stale tmp_dir left over from a previous attempt with a *different*
|
||||
# total_chunks (e.g. the client's chunk size changed between app
|
||||
# versions) would reassemble incorrectly - start clean if that happens.
|
||||
marker_path = os.path.join(tmp_dir, ".total_chunks")
|
||||
if os.path.isdir(tmp_dir) and os.path.exists(marker_path):
|
||||
with open(marker_path) as marker_file:
|
||||
if marker_file.read().strip() != str(total_chunks):
|
||||
shutil.rmtree(tmp_dir)
|
||||
|
||||
os.makedirs(tmp_dir, exist_ok=True)
|
||||
with open(marker_path, "w") as marker_file:
|
||||
marker_file.write(str(total_chunks))
|
||||
|
||||
chunk_path = os.path.join(tmp_dir, f"{chunk_index:06d}.chunk")
|
||||
uploaded = request.files["file"]
|
||||
with open(chunk_path, "wb") as f:
|
||||
while True:
|
||||
data = uploaded.stream.read(1024 * 1024)
|
||||
if not data:
|
||||
break
|
||||
f.write(data)
|
||||
|
||||
received = _received_chunk_indices(tmp_dir)
|
||||
if len(received) < total_chunks:
|
||||
return jsonify(status="chunk_received", received_chunks=len(received), total_chunks=total_chunks), 202
|
||||
|
||||
# All chunks in - reassemble, verify, and store.
|
||||
hasher = hashlib.sha256()
|
||||
tmp_path = dest_path + ".part"
|
||||
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
|
||||
size_bytes = 0
|
||||
with open(tmp_path, "wb") as out:
|
||||
for idx in range(total_chunks):
|
||||
with open(os.path.join(tmp_dir, f"{idx:06d}.chunk"), "rb") as chunk_file:
|
||||
while True:
|
||||
data = chunk_file.read(1024 * 1024)
|
||||
if not data:
|
||||
break
|
||||
hasher.update(data)
|
||||
size_bytes += len(data)
|
||||
out.write(data)
|
||||
|
||||
actual_sha256 = hasher.hexdigest()
|
||||
shutil.rmtree(tmp_dir, ignore_errors=True)
|
||||
if actual_sha256 != expected_sha256:
|
||||
os.remove(tmp_path)
|
||||
return jsonify(error="sha256 mismatch, upload corrupted in transit"), 422
|
||||
|
||||
os.replace(tmp_path, dest_path)
|
||||
db.insert_dashcam_file(device_id, relative_path, size_bytes, actual_sha256, dest_path)
|
||||
return jsonify(status="complete", size_bytes=size_bytes), 201
|
||||
|
||||
|
||||
def main():
|
||||
app.run(host="0.0.0.0", port=8098)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user