diff --git a/modules/app_upload.py b/modules/app_upload.py index 51b1621..51acaba 100644 --- a/modules/app_upload.py +++ b/modules/app_upload.py @@ -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 diff --git a/server/app.py b/server/app.py index ea4072a..b657633 100644 --- a/server/app.py +++ b/server/app.py @@ -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)