feat: two-phase upload/download with real-time progress + smart speed timeout

Split monolithic /session/start into two endpoints:
- POST /session/upload  — phase 1: file acquisition only, returns JSON
- POST /session/analyze — phase 2: backend analysis, returns HTML
- GET  /session/progress/{filename} — poll download progress

Uploads:
- Browser-native progress bar via XMLHttpRequest (real % + GiB)

Downloads:
- wget runs in background (Popen), frontend polls /session/progress
- HEAD request gets Content-Length before download starts
- Real-time speed (MB/s) and ETA displayed in the UI
- Smart timeout: after 30s, if ETA > 1 hour, kills download and
  suggests manual download to laptop + File Upload instead
- Absolute safety net at 4 hours

UI: clear phase transitions — 'Downloading... 2.3 GiB (4.5 MB/s) ~12 min'
→ 'Step 2/2: Analysing source image...' → result
This commit is contained in:
Claus Lohmar 2026-07-21 18:10:52 +00:00
parent 509eb97b57
commit af26da7b64
2 changed files with 408 additions and 106 deletions

View file

@ -19,10 +19,13 @@ import logging
import os
import shutil
import subprocess
import time
import uuid
from pathlib import Path
from typing import Optional
import requests as http_requests
from fastapi import FastAPI, Form, Request, UploadFile, File
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
@ -52,9 +55,14 @@ VM_ID_MIN = 21000
VM_ID_MAX = 21100
DEFAULT_STORAGE = "local-lvm"
MIN_FREE_DISK_GB = 2 # keep 2 GB headroom
DOWNLOAD_TIMEOUT = 7200 # 2 hours for very large files
DOWNLOAD_TIMEOUT = 14400 # 4 hours absolute max (safety net)
SPEED_CHECK_AFTER = 30 # wait N seconds before judging speed
MAX_ETA_SECONDS = 3600 # kill download if ETA > 1 hour
UPLOAD_PROGRESS_INTERVAL = 1024**3 # log every 1 GiB during upload
# Track active background downloads for progress polling
_active_downloads: dict[str, dict] = {}
def _validate_vmid(vmid: int) -> Optional[str]:
if not (VM_ID_MIN <= vmid <= VM_ID_MAX):
return f"VM ID must be between {VM_ID_MIN} and {VM_ID_MAX}."
@ -91,8 +99,8 @@ async def index(request: Request, vmid: Optional[int] = None, source: Optional[s
prefill_vmname="")
@app.post("/session/start", response_class=HTMLResponse)
async def start_session(
@app.post("/session/upload")
async def session_upload(
request: Request,
vmid: int = Form(...),
vm_name: str = Form(""),
@ -100,104 +108,230 @@ async def start_session(
source_file: Optional[UploadFile] = File(None),
source_url: Optional[str] = Form(None),
):
"""Upload or download source file, then call backend /analyze."""
filename = None
error = None
"""Phase 1 — acquire the source file. Returns JSON so the frontend can
show progress, then call /session/analyze separately."""
err = _validate_vmid(vmid)
if err:
return JSONResponse({"phase": "error", "error": err}, status_code=400)
# Validate VM ID range
vm_name = vm_name.strip()
# ── Upload ──────────────────────────────────────────────────────
if source_type == "upload":
if not source_file or not source_file.filename:
return JSONResponse({"phase": "error", "error": "No file uploaded."}, status_code=400)
filename = source_file.filename
dest = STAGING / filename
content_length = request.headers.get("content-length")
if content_length:
estimated_gb = int(content_length) / (1024**3)
err = _check_disk_space(dest, estimated_gb)
if err:
return JSONResponse({"phase": "error", "error": err}, status_code=400)
try:
logger.info("Receiving upload: %s (%s bytes)", filename, content_length or "unknown")
written = 0
with dest.open("wb") as f:
while True:
chunk = source_file.file.read(8 * 1024 * 1024)
if not chunk:
break
f.write(chunk)
written += len(chunk)
if written % UPLOAD_PROGRESS_INTERVAL < len(chunk):
logger.info("Upload progress: %s%.1f GiB", filename, written / (1024**3))
file_size_gb = round(written / (1024**3), 1)
logger.info("Upload complete: %s (%.1f GiB)", filename, file_size_gb)
except OSError as exc:
if dest.exists():
dest.unlink(missing_ok=True)
return JSONResponse({"phase": "error", "error": f"Upload failed (disk full?): {exc}"}, status_code=500)
except Exception as exc:
if dest.exists():
dest.unlink(missing_ok=True)
return JSONResponse({"phase": "error", "error": f"Upload failed: {exc}"}, status_code=500)
return JSONResponse({
"phase": "staged",
"filename": filename,
"vmid": vmid,
"vm_name": vm_name,
"file_size_gb": file_size_gb,
})
# ── Download ────────────────────────────────────────────────────
url = (source_url or "").strip()
if not url:
return JSONResponse({"phase": "error", "error": "No URL provided."}, status_code=400)
filename = Path(url).name or f"download_{uuid.uuid4().hex[:8]}"
dest = STAGING / filename
# Clean up any stale download with same name
_active_downloads.pop(filename, None)
usage = shutil.disk_usage(STAGING)
free_gb = usage.free / (1024**3)
if free_gb < 50:
logger.warning("Low disk: %.1f GB free — download may fail", free_gb)
# Try to get file size via HEAD request (for speed estimation)
content_length = 0
try:
head_resp = http_requests.head(url, timeout=10, allow_redirects=True)
cl = head_resp.headers.get("Content-Length")
if cl:
content_length = int(cl)
logger.info("Download size from HEAD: %.1f GiB", content_length / (1024**3))
except Exception:
logger.info("Could not determine download size (HEAD failed — will skip ETA check)")
logger.info("Starting background download: %s%s", url, dest)
try:
proc = subprocess.Popen(
["wget", "--progress=dot:giga", "-O", str(dest), url],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
)
except Exception as exc:
return JSONResponse({"phase": "error", "error": f"Failed to start download: {exc}"}, status_code=500)
_active_downloads[filename] = {
"proc": proc,
"dest": dest,
"vmid": vmid,
"vm_name": vm_name,
"start_time": time.time(),
"content_length": content_length,
}
return JSONResponse({
"phase": "downloading",
"filename": filename,
"vmid": vmid,
"vm_name": vm_name,
"content_length_gb": round(content_length / (1024**3), 1) if content_length else None,
})
@app.get("/session/progress/{filename}")
async def session_progress(filename: str):
"""Poll download progress — returns current file size and phase."""
info = _active_downloads.get(filename)
if not info:
# Check if file exists on disk (download already completed in a
# previous session, or it was an upload)
dest = STAGING / filename
if dest.exists():
return JSONResponse({
"phase": "complete",
"file_size_bytes": dest.stat().st_size,
"file_size_gb": round(dest.stat().st_size / (1024**3), 1),
})
return JSONResponse({"phase": "unknown", "error": "No active download for this file."}, status_code=404)
proc = info["proc"]
dest = info["dest"]
# Current bytes on disk
current_bytes = dest.stat().st_size if dest.exists() else 0
# Check if process still running
poll = proc.poll()
if poll is not None:
# Process exited
_active_downloads.pop(filename, None)
if poll != 0:
if dest.exists():
dest.unlink(missing_ok=True)
return JSONResponse({
"phase": "error",
"error": f"Download failed (wget exited with code {poll}).",
"file_size_bytes": current_bytes,
})
# Success
final_bytes = dest.stat().st_size
return JSONResponse({
"phase": "complete",
"file_size_bytes": final_bytes,
"file_size_gb": round(final_bytes / (1024**3), 1),
})
# Still downloading — check speed and estimate ETA
elapsed = max(time.time() - info.get("start_time", 0), 1)
speed_bps = current_bytes / elapsed
speed_mbps = round(speed_bps / 1_000_000, 1)
content_length = info.get("content_length", 0)
eta_str = None
too_slow = False
if content_length > 0 and elapsed > SPEED_CHECK_AFTER:
remaining = content_length - current_bytes
eta = remaining / max(speed_bps, 1) # seconds
if eta > MAX_ETA_SECONDS:
too_slow = True
eta_hours = round(eta / 3600, 1)
# Kill the download
try:
proc.kill()
except Exception:
pass
_active_downloads.pop(filename, None)
if dest.exists():
dest.unlink(missing_ok=True)
logger.warning(
"Download %s killed: ETA %.1f h at %.1f MB/s (threshold %d h)",
filename, eta_hours, speed_mbps, MAX_ETA_SECONDS // 3600,
)
return JSONResponse({
"phase": "too_slow",
"speed_mbps": speed_mbps,
"eta_hours": eta_hours,
"content_length_gb": round(content_length / (1024**3), 1),
"downloaded_gb": round(current_bytes / (1024**3), 1),
"message": (
f"Download would take ~{eta_hours} hours at {speed_mbps} MB/s "
f"(file is {round(content_length / (1024**3), 1)} GiB). "
f"Consider downloading to your computer manually, then use File Upload."
),
})
eta_str = f"~{round(eta / 60)} min remaining"
if too_slow:
# Already handled above; this line is unreachable but kept for clarity
pass
return JSONResponse({
"phase": "downloading",
"file_size_bytes": current_bytes,
"file_size_gb": round(current_bytes / (1024**3), 1),
"speed_mbps": speed_mbps,
"eta": eta_str,
"content_length_gb": round(content_length / (1024**3), 1) if content_length else None,
})
@app.post("/session/analyze", response_class=HTMLResponse)
async def session_analyze(
request: Request,
vmid: int = Form(...),
filename: str = Form(...),
vm_name: str = Form(""),
):
"""Phase 2 — call the backend /analyze endpoint and render the result."""
err = _validate_vmid(vmid)
if err:
return render("_analysis.html", request=request, error=err)
# --- Get the file into /mnt/converter/in/ ---
if source_type == "upload":
if not source_file or not source_file.filename:
error = "No file uploaded."
else:
filename = source_file.filename
dest = STAGING / filename
# Estimate size from Content-Length header (may be None)
content_length = request.headers.get("content-length")
if content_length:
estimated_gb = int(content_length) / (1024**3)
err = _check_disk_space(dest, estimated_gb)
if err:
return render("_analysis.html", request=request, error=err)
try:
logger.info("Receiving upload: %s (%s bytes)", filename, content_length or "unknown")
with dest.open("wb") as f:
written = 0
while True:
chunk = source_file.file.read(8 * 1024 * 1024) # 8 MiB chunks
if not chunk:
break
f.write(chunk)
written += len(chunk)
if written % UPLOAD_PROGRESS_INTERVAL < len(chunk):
logger.info("Upload progress: %s%.1f GiB written", filename, written / (1024**3))
logger.info("Upload complete: %s (%.1f GiB)", filename, written / (1024**3))
except OSError as exc:
# Clean up partial file on disk-full or other IO error
if dest.exists():
dest.unlink(missing_ok=True)
error = f"Upload failed (disk full?): {exc}"
except Exception as exc:
if dest.exists():
dest.unlink(missing_ok=True)
error = f"Upload failed: {exc}"
else:
url = (source_url or "").strip()
if not url:
error = "No URL provided."
else:
filename = Path(url).name or f"download_{uuid.uuid4().hex[:8]}"
dest = STAGING / filename
# Warn if staging is tight, but don't block (don't know file size)
usage = shutil.disk_usage(STAGING)
free_gb = usage.free / (1024**3)
if free_gb < 50:
logger.warning("Low disk space on %s: %.1f GB free — download may fail for large files",
STAGING, free_gb)
logger.info("Starting download: %s%s", url, dest)
try:
# --progress=dot:giga prints one dot per 64 KiB downloaded, minimal output
# Redirection to stderr keeps stdout clean for error capture
result = subprocess.run(
["wget", "--progress=dot:giga", "-O", str(dest), url],
capture_output=True, text=True, timeout=DOWNLOAD_TIMEOUT,
)
if result.returncode != 0:
error = f"Download failed: {result.stderr.strip()[:300]}"
else:
file_size = dest.stat().st_size / (1024**3) if dest.exists() else 0
logger.info("Download complete: %s (%.1f GiB)", filename, file_size)
except subprocess.TimeoutExpired:
if dest.exists():
dest.unlink(missing_ok=True)
error = (
f"Download timed out after {DOWNLOAD_TIMEOUT // 3600} hours. "
f"The file may be too large for your network speed."
)
except Exception as exc:
if dest.exists():
dest.unlink(missing_ok=True)
error = f"Download failed: {exc}"
if error:
return render("_analysis.html", request=request, error=error)
# --- Call backend /analyze ---
try:
analysis = api.analyze(vmid=vmid, filename=filename)
except ApiError as exc:
return render("_analysis.html", request=request,
error=f"Backend analysis failed: {exc.detail}")
vm_name = vm_name.strip() if vm_name else ""
vm_name = vm_name.strip()
if not vm_name:
vm_name = (analysis.get("os_type") or "vm") + f"-{vmid}"

View file

@ -43,7 +43,13 @@
<button type="submit" id="start-btn">Analyse Source Image</button>
</form>
<div id="session-status" class="hidden" style="margin-top:1rem;"></div>
<!-- Progress area (hidden until form submit) -->
<div id="session-status" class="hidden" style="margin-top:1rem;">
<div id="phase-label" style="margin-bottom:0.5rem;"></div>
<div class="progress-bar">
<div class="progress-fill" id="progress-fill" style="width:0%">0%</div>
</div>
</div>
</div>
<!-- Step 2: Analysis Result (populated by JavaScript) -->
@ -56,51 +62,213 @@ function toggleSourceInput(e) {
document.getElementById('url-group').classList.toggle('hidden', type !== 'url');
}
function setPhase(label, pct, isSpinner) {
document.getElementById('phase-label').innerHTML =
(isSpinner ? '<span class="spinner"></span> ' : '') + label;
const fill = document.getElementById('progress-fill');
fill.style.width = pct + '%';
fill.textContent = pct > 0 ? pct + '%' : '';
}
async function startSession(e) {
e.preventDefault();
const btn = document.getElementById('start-btn');
const status = document.getElementById('session-status');
const analysis = document.getElementById('analysis-section');
btn.disabled = true;
analysis.innerHTML = '';
// Reset progress area (clear old errors)
status.innerHTML = `
<div id="phase-label" style="margin-bottom:0.5rem;"></div>
<div class="progress-bar">
<div class="progress-fill" id="progress-fill" style="width:0%">0%</div>
</div>
`;
status.classList.remove('hidden');
status.innerHTML = '<div class="spinner"></div> Analysing source image...';
const vmid = document.getElementById('vmid').value;
const vmName = document.getElementById('vm-name').value.trim();
const sourceType = document.getElementById('source-type').value;
const formData = new FormData();
formData.append('vmid', document.getElementById('vmid').value);
formData.append('vm_name', document.getElementById('vm-name').value);
const sourceType = document.getElementById('source-type').value;
formData.append('vmid', vmid);
formData.append('vm_name', vmName);
formData.append('source_type', sourceType);
if (sourceType === 'upload') {
const fileInput = document.getElementById('source-file');
if (fileInput.files.length > 0) {
formData.append('source_file', fileInput.files[0]);
} else {
status.innerHTML = '<div class="error">Please select a file.</div>';
if (fileInput.files.length === 0) {
setPhase('Please select a file.', 0, false);
status.innerHTML += '<div class="error" style="margin-top:0.5rem;">No file selected.</div>';
btn.disabled = false;
return;
}
formData.append('source_file', fileInput.files[0]);
await handleUpload(formData);
} else {
const url = document.getElementById('source-url').value.trim();
if (!url) {
status.innerHTML = '<div class="error">Please enter a URL.</div>';
setPhase('Please enter a URL.', 0, false);
status.innerHTML += '<div class="error" style="margin-top:0.5rem;">No URL provided.</div>';
btn.disabled = false;
return;
}
formData.append('source_url', url);
await handleDownload(formData);
}
btn.disabled = false;
}
// ── Upload (browser-native progress) ──────────────────────────────
async function handleUpload(formData) {
return new Promise((resolve) => {
const xhr = new XMLHttpRequest();
xhr.open('POST', '/session/upload');
xhr.upload.addEventListener('progress', (e) => {
if (e.lengthComputable) {
const pct = Math.round((e.loaded / e.total) * 100);
const gb = (e.loaded / (1024**3)).toFixed(1);
setPhase('Uploading source file... ' + gb + ' GiB', pct, false);
}
});
xhr.addEventListener('load', async () => {
if (xhr.status === 200) {
const data = JSON.parse(xhr.responseText);
if (data.phase === 'staged') {
await runAnalysis(data.vmid, data.filename, data.vm_name);
} else {
setPhase('Upload failed', 0, false);
showError(data.error || 'Unknown error');
}
} else {
try {
const err = JSON.parse(xhr.responseText);
setPhase('Upload failed', 0, false);
showError(err.error || 'Server error ' + xhr.status);
} catch (_) {
setPhase('Upload failed', 0, false);
showError('Server error ' + xhr.status);
}
}
resolve();
});
xhr.addEventListener('error', () => {
setPhase('Upload failed', 0, false);
showError('Network error during upload.');
resolve();
});
xhr.send(formData);
});
}
// ── Download (polled progress) ────────────────────────────────────
async function handleDownload(formData) {
setPhase('Starting download...', 0, false);
let resp;
try {
const resp = await fetch('/session/start', { method: 'POST', body: formData });
resp = await fetch('/session/upload', { method: 'POST', body: formData });
} catch (err) {
setPhase('Download failed', 0, false);
showError('Network error: ' + err.message);
return;
}
const data = await resp.json();
if (data.phase === 'error') {
setPhase('Download failed', 0, false);
showError(data.error);
return;
}
if (data.phase !== 'downloading') {
setPhase('Unexpected phase: ' + data.phase, 0, false);
return;
}
// Poll for progress
const filename = data.filename;
setPhase('Downloading source file...', 0, true);
for (;;) {
await sleep(2000);
try {
const pr = await fetch('/session/progress/' + encodeURIComponent(filename));
const pdata = await pr.json();
if (pdata.phase === 'complete') {
setPhase('Download complete (' + (pdata.file_size_gb || 0) + ' GiB)', 100, false);
await sleep(500);
await runAnalysis(data.vmid, filename, data.vm_name);
return;
}
if (pdata.phase === 'error') {
setPhase('Download failed', 0, false);
showError(pdata.error);
return;
}
if (pdata.phase === 'too_slow') {
setPhase('Download too slow', 0, false);
showError(
'<strong>Download cancelled — it would take too long.</strong><br><br>' +
(pdata.message || '') +
'<br><br>Suggested next step:' +
'<ol style="margin-top:0.5rem; padding-left:1.2rem;">' +
'<li>Stop this download and download the file manually to your local computer.</li>' +
'<li>Switch to <strong>File Upload</strong> above and upload the file.</li>' +
'</ol>'
);
return;
}
if (pdata.phase === 'downloading') {
const gb = (pdata.file_size_gb || 0).toFixed(1);
let label = 'Downloading source file... ' + gb + ' GiB';
if (pdata.speed_mbps) label += ' (' + pdata.speed_mbps + ' MB/s)';
if (pdata.eta) label += ' — ' + pdata.eta;
setPhase(label, 0, true);
}
} catch (_) {
// Polling error — keep trying
}
}
}
// ── Phase 2: Analysis ─────────────────────────────────────────────
async function runAnalysis(vmid, filename, vmName) {
setPhase('Step 2/2: Analysing source image...', 0, true);
const fd = new FormData();
fd.append('vmid', vmid);
fd.append('filename', filename);
fd.append('vm_name', vmName);
try {
const resp = await fetch('/session/analyze', { method: 'POST', body: fd });
const html = await resp.text();
document.getElementById('analysis-section').innerHTML = html;
status.classList.add('hidden');
document.getElementById('session-status').classList.add('hidden');
} catch (err) {
status.innerHTML = `<div class="error">Failed: ${err.message}</div>`;
} finally {
btn.disabled = false;
setPhase('Analysis failed', 0, false);
showError('Failed to reach analysis endpoint: ' + err.message);
}
}
function showError(msg) {
const status = document.getElementById('session-status');
status.innerHTML += '<div class="error" style="margin-top:0.5rem;">' + msg + '</div>';
}
function sleep(ms) {
return new Promise(resolve => setTimeout(resolve, ms));
}
</script>
{% endblock %}