From e5a2dc0a8d267bc293c65a1376861f293e9f3a94 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 10 Sep 2026 15:12:15 +0800 Subject: [PATCH] add chunked upload to bypass 100MB proxy body limit Resumable /api/uploads flow (init -> append -> complete -> abort) so a large recording uploads as many small slices (<= ASR_CHUNK_SIZE_MB, default 50MB) instead of one request a size-capping proxy (Cloudflare's 100MB) would reject. The web console auto-routes files larger than one slice through the chunked path with cross-slice progress and transient-error retry; smaller files keep the single-shot upload. - store: in-memory upload sessions + .part reassembly (offset writes, out-of-order/idempotent), reuse upload id as task id, orphan sweep on start, idle-session GC - fix limitedReader rejecting bodies exactly at the cap (broke full chunks) - fix drag-drop double-firing when a drop lands on the dropzone (its drop handler now stopPropagation so the window listener doesn't re-upload) - tests: store + HTTP flow, plus a 120MB reassembly test (ASR_TEST_LARGE) --- README.md | 38 ++++ internal/config/config.go | 18 +- internal/server/handlers.go | 105 ++++++++++ internal/server/server.go | 8 + internal/server/ui/index.html | 131 +++++++++++-- internal/server/upload_large_test.go | 153 +++++++++++++++ internal/server/upload_test.go | 200 +++++++++++++++++++ internal/store/errors.go | 4 + internal/store/store.go | 45 +++-- internal/store/upload.go | 274 +++++++++++++++++++++++++++ internal/store/upload_test.go | 203 ++++++++++++++++++++ 11 files changed, 1144 insertions(+), 35 deletions(-) create mode 100644 internal/server/upload_large_test.go create mode 100644 internal/server/upload_test.go create mode 100644 internal/store/upload.go create mode 100644 internal/store/upload_test.go diff --git a/README.md b/README.md index 24aefbc..d1c1ea2 100644 --- a/README.md +++ b/README.md @@ -129,6 +129,7 @@ SenseVoice + ~2 MB silero VAD); subsequent starts load them from disk. | `ASR_USE_ITN` | `1` | Inverse text normalization (numbers, punctuation). | | `ASR_MAX_RETRIES` | `3` | Auto-retry attempts before a task is marked failed. | | `ASR_MAX_UPLOAD_MB` | `512` | Per-upload size cap. | +| `ASR_CHUNK_SIZE_MB` | `50` | Slice size for resumable chunked uploads; kept under CDN body limits (e.g. Cloudflare's 100MB). | | `ASR_WORKERS` | `1` | Concurrent transcription workers. | | `ASR_VAD_THRESHOLD` | `0.5` | Silero speech probability threshold. | | `ASR_VAD_MIN_SPEECH` | `0.5` | Min speech segment (s). | @@ -194,6 +195,43 @@ curl -H "X-Api-Key: changeme" http://localhost:8080/api/tasks/ curl -H "X-Api-Key: changeme" http://localhost:8080/api/tasks//transcript ``` +### Chunked uploads (bypassing a 100MB proxy limit) + +When the service sits behind a size-capping proxy — Cloudflare's free plan +rejects request bodies over **100MB** — a single `POST /api/tasks` upload of a +large recording never reaches the origin. The resumable chunked-upload API +splits the file into slices no bigger than `ASR_CHUNK_SIZE_MB` (default **50MB**, +well under the cap) and reassembles them server-side into one task: + +| Method | Path | Description | +|---|---|---| +| `POST` | `/api/uploads` | Open a session: JSON `{filename,size}` → `{upload_id, chunk_size, total_chunks}`. | +| `POST` | `/api/uploads/{id}?index=N` | Send slice `N` (0-based, raw body ≤ chunk_size). Order is free; resending is idempotent. | +| `POST` | `/api/uploads/{id}/complete` | Once all slices arrive: merge → enqueue → `202` + task (same shape as `/api/tasks`). | +| `DELETE` | `/api/uploads/{id}` | Abort the session and drop the partial file → `204`. | + +```sh +# Open a session for a 180MB file (4 slices of 50MB) +UID=$(curl -sH "X-Api-Key: $PW" -H "Content-Type: application/json" \ + -d '{"filename":"big.mp4","size":188743680}' \ + http://localhost:8080/api/uploads | jq -r .upload_id) + +# Send each slice (split with dd/split beforehand) +for i in 0 1 2 3; do + curl -sH "X-Api-Key: $PW" --data-binary @"slice_$i.bin" \ + "http://localhost:8080/api/uploads/$UID?index=$i" >/dev/null +done + +# Merge → enqueue (returns the task, just like /api/tasks) +curl -H "X-Api-Key: $PW" -X POST http://localhost:8080/api/uploads/$UID/complete +``` + +The web console handles this automatically: files larger than one chunk go +through the chunked path (with a single progress bar across all slices); +smaller files keep the simple single-request upload. Sessions live in memory, +so an interrupted upload is retried from scratch on reload — and any orphaned +partial files are swept on the next start. + ## Task lifecycle & error handling States: `pending` → `processing` → `completed` | `failed` | `skipped`. diff --git a/internal/config/config.go b/internal/config/config.go index 3fa9795..ff8752d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -36,6 +36,11 @@ type Config struct { MaxRetries int // ASR_MAX_UPLOAD_MB: per-upload size cap (default 512). MaxUploadBytes int64 + // ASR_CHUNK_SIZE_MB: size of each slice in a resumable chunked upload, + // kept well under size-capping proxies' body limits so a large file can + // be uploaded as many small requests instead of one that a CDN (e.g. + // Cloudflare's 100MB cap) would reject. Default 50. + ChunkSize int64 // ASR_WORKERS: concurrent transcription workers (default 1). Workers int // ASR_MAX_CONCURRENT: cap on simultaneously in-flight synchronous @@ -73,6 +78,7 @@ func Load() (Config, error) { UseITN: envBool("ASR_USE_ITN", true), MaxRetries: envInt("ASR_MAX_RETRIES", 3), MaxUploadBytes: int64(envInt("ASR_MAX_UPLOAD_MB", 512)) * 1024 * 1024, + ChunkSize: int64(envInt("ASR_CHUNK_SIZE_MB", 50)) * 1024 * 1024, Workers: envInt("ASR_WORKERS", 1), MaxConcurrent: envInt("ASR_MAX_CONCURRENT", 0), VadThreshold: float32(envFloat("ASR_VAD_THRESHOLD", 0.5)), @@ -111,6 +117,14 @@ func Load() (Config, error) { if c.MaxUploadBytes < 1<<20 { c.MaxUploadBytes = 512 << 20 } + // Chunk size must be > 1MB to be useful and < 100MB so each slice clears + // common CDN body caps (e.g. Cloudflare's 100MB) with room to spare. + if c.ChunkSize < 1<<20 { + c.ChunkSize = 50 << 20 + } + if c.ChunkSize > 100<<20 { + c.ChunkSize = 100 << 20 + } if c.DownloadTimeout < time.Minute { c.DownloadTimeout = 30 * time.Minute } @@ -130,9 +144,9 @@ func Load() (Config, error) { func (c Config) AuthEnabled() bool { return c.Password != "" } func (c Config) String() string { - return fmt.Sprintf("addr=%s data=%s models=%s threads=%d lang=%s itn=%t workers=%d maxconcurrent=%d retries=%d vad{thr=%.2f max=%.0fs win=%d}", + return fmt.Sprintf("addr=%s data=%s models=%s threads=%d lang=%s itn=%t workers=%d maxconcurrent=%d retries=%d chunk=%dB vad{thr=%.2f max=%.0fs win=%d}", c.Addr, c.DataDir, c.ModelDir, c.NumThreads, c.Language, c.UseITN, c.Workers, - c.MaxConcurrent, c.MaxRetries, c.VadThreshold, c.VadMaxSpeechDuration, c.VadWindowSize) + c.MaxConcurrent, c.MaxRetries, c.ChunkSize, c.VadThreshold, c.VadMaxSpeechDuration, c.VadWindowSize) } func defaultThreads() int { diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 76f7e5d..5788e43 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -3,6 +3,7 @@ package server import ( "encoding/json" "net/http" + "strconv" "strings" "git.misaka.ren/m1saka/record-asr/internal/audio" @@ -23,6 +24,7 @@ func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) { "models": s.manager != nil && s.manager.Ready(), "version": s.modelVersion(), "max_upload_bytes": s.cfg.MaxUploadBytes, + "chunk_size": s.cfg.ChunkSize, }) } @@ -82,6 +84,109 @@ func (s *Server) handleCreateTask(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusAccepted, t) } +// --- resumable chunked uploads --- +// +// Files too large to fit a single request through a size-capping proxy (e.g. +// Cloudflare's 100MB body limit) are uploaded as many small slices instead. +// handleCreateUpload starts a session, handleUploadChunk appends one slice, +// handleCompleteUpload finalizes it into a task, and handleAbortUpload drops +// it. Each slice is <= ASR_CHUNK_SIZE_MB, safely under the proxy cap. + +// handleCreateUpload starts a resumable chunked upload session. The request +// body is a small JSON {filename, size}; the response carries the upload id, +// the chunk size, and the total chunk count the client should send. +func (s *Server) handleCreateUpload(w http.ResponseWriter, r *http.Request) { + var req struct { + Filename string `json:"filename"` + Size int64 `json:"size"` + } + dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)) + if err := dec.Decode(&req); err != nil { + renderError(w, http.StatusBadRequest, "invalid JSON: "+err.Error()) + return + } + if req.Filename == "" { + renderError(w, http.StatusBadRequest, "missing 'filename'") + return + } + if req.Size <= 0 { + renderError(w, http.StatusBadRequest, "missing or invalid 'size'") + return + } + if !audio.IsSupported(req.Filename) { + renderError(w, http.StatusUnsupportedMediaType, + "unsupported extension; supported: "+audio.SupportedExtensions()) + return + } + sess, err := s.store.CreateUpload(req.Filename, req.Size, s.cfg.ChunkSize, s.cfg.MaxUploadBytes) + if err != nil { + renderError(w, http.StatusBadRequest, "init upload: "+err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "upload_id": sess.UploadID, + "chunk_size": sess.ChunkSize, + "total_chunks": sess.TotalChunks, + }) +} + +// handleUploadChunk appends one chunk (#index, the 0-based query param) of a +// chunked upload session. The request body is the raw chunk bytes +// (application/octet-stream); each slice is capped at ASR_CHUNK_SIZE_MB. +func (s *Server) handleUploadChunk(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + idx, err := strconv.Atoi(r.URL.Query().Get("index")) + if err != nil || idx < 0 { + renderError(w, http.StatusBadRequest, "missing or invalid 'index' query param") + return + } + r.Body = http.MaxBytesReader(w, r.Body, s.cfg.ChunkSize) + if err := s.store.AppendChunk(id, idx, r.Body, s.cfg.ChunkSize); err != nil { + if err == store.ErrUploadNotFound { + renderError(w, http.StatusNotFound, "upload session not found or expired") + return + } + renderError(w, http.StatusBadRequest, "chunk: "+err.Error()) + return + } + received, total, ok := s.store.UploadStatus(id) + if !ok { + renderError(w, http.StatusNotFound, "upload session not found") + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "index": idx, + "received": received, + "total": total, + }) +} + +// handleCompleteUpload finalizes a chunked upload and enqueues the task. The +// response mirrors POST /api/tasks (202 + the task) so the client treats the +// finalized upload exactly like a single-shot upload from this point on. +func (s *Server) handleCompleteUpload(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + t, err := s.store.CompleteUpload(id, s.cfg.MaxRetries) + if err != nil { + if err == store.ErrUploadNotFound { + renderError(w, http.StatusNotFound, "upload session not found or expired") + return + } + renderError(w, http.StatusBadRequest, "complete: "+err.Error()) + return + } + w.Header().Set("Location", "/api/tasks/"+t.ID) + writeJSON(w, http.StatusAccepted, t) +} + +// handleAbortUpload cancels a chunked upload session and removes its part +// file. Idempotent: a missing session still returns 204. +func (s *Server) handleAbortUpload(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + _ = s.store.AbortUpload(id) + w.WriteHeader(http.StatusNoContent) +} + // handleListTasks returns all tasks newest-first. func (s *Server) handleListTasks(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]any{"tasks": s.store.List()}) diff --git a/internal/server/server.go b/internal/server/server.go index 84426c1..bcb9406 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -64,6 +64,14 @@ func (s *Server) Router() http.Handler { mux.HandleFunc("POST /api/tasks/{id}/skip", s.handleSkipTask) mux.HandleFunc("GET /api/tasks/{id}/transcript", s.handleTranscript) + // Resumable chunked uploads (also authed). Lets a large file slip past + // size-capping proxies (e.g. Cloudflare's 100MB body limit) as many + // small slices, finalized into a task indistinguishable from /api/tasks. + mux.HandleFunc("POST /api/uploads", s.handleCreateUpload) + mux.HandleFunc("POST /api/uploads/{id}", s.handleUploadChunk) + mux.HandleFunc("POST /api/uploads/{id}/complete", s.handleCompleteUpload) + mux.HandleFunc("DELETE /api/uploads/{id}", s.handleAbortUpload) + // OpenAI-compatible synchronous transcription (also authed). mux.HandleFunc("POST /v1/audio/transcriptions", s.handleOpenAITranscription) diff --git a/internal/server/ui/index.html b/internal/server/ui/index.html index 3287396..45e528f 100644 --- a/internal/server/ui/index.html +++ b/internal/server/ui/index.html @@ -311,6 +311,26 @@ +
+

分片上传 绕过代理体积上限（如 Cloudflare 100MB）

+ + + + + + +
方法路径说明
POST/api/uploads开启上传会话（JSON：{filename,size}）→ {upload_id, chunk_size, total_chunks}
POST/api/uploads/{id}?index=N上传第 N 片（0 起，原始字节，≤ chunk_size）。乱序/重发幂等
POST/api/uploads/{id}/complete收齐后合并并入队 → 202 + 任务（等同 /api/tasks）
DELETE/api/uploads/{id}放弃会话并删除分片 → 204
+ 
# 大文件分片上传（每片 50MB，默认）
+curl -H "X-Api-Key: $PW" -H "Content-Type: application/json" \
+  -d '{"filename":"big.mp4","size":123456789}' http://localhost:8080/api/uploads
+# → {"upload_id":"ab12cd34","chunk_size":52428800,"total_chunks":3}
+for i in 0 1 2; do
+  curl -H "X-Api-Key: $PW" --data-binary @"part_$i.bin" \
+    "http://localhost:8080/api/uploads/ab12cd34?index=$i"
+done
+curl -H "X-Api-Key: $PW" -X POST http://localhost:8080/api/uploads/ab12cd34/complete
+
+

OpenAI 兼容 同步

@@ -368,7 +388,7 @@ const fmtTime = iso=>{ if(!iso) return "—"; const d=new Date(iso); return isNa const fmtDur = s=> s==null?"—": (s<60?s.toFixed(1)+"s" : Math.floor(s/60)+"m"+String(Math.round(s%60)).padStart(2,"0")+"s"); const escapeHtml = s=>(s||"").replace(/[&<>]/g,c=>({"&":"&","<":"<",">":">"}[c])); -const state = { pw: localStorage.getItem("asr_pw")||"", tasks:[], sel:null, busy:false, authing:false, uploading:false, maxUpload:0 }; +const state = { pw: localStorage.getItem("asr_pw")||"", tasks:[], sel:null, busy:false, authing:false, uploading:false, maxUpload:0, chunkSize:0 }; // --- password modal (shown only when verification is needed) --- const authModal=$("#authModal"), authPw=$("#authPw"), authMsg=$("#authMsg"), authOk=$("#authOk"), authCancel=$("#authCancel"); @@ -480,6 +500,7 @@ async function loadHealth(){ try{ const h = await api("/healthz"); state.maxUpload = h.max_upload_bytes || 0; + state.chunkSize = h.chunk_size || 0; $("#htext").textContent = h.models?"模型就绪":"模型加载中"; $("#hdot").className = "dot "+(h.models?"ok":"down"); }catch(e){ $("#htext").textContent="服务不可达"; $("#hdot").className="dot down"; } @@ -606,9 +627,12 @@ function renderDropzone(c){ const dz=el("div",{class:"dropzone"}); dz.innerHTML=`

拖拽音频或视频到这里，或点击上传

音频直接转写；视频会自动提取音轨。约 1~3 秒出结果。

音频 .wav/.mp3/.flac/.ogg/.m4a/.aac/.opus · 视频 .mp4/.mkv/.mov/.avi/.ts…
`; dz.onclick=()=>$("#file").click(); - dz.addEventListener("dragover",e=>{e.preventDefault();dz.classList.add("drag")}); + dz.addEventListener("dragover",e=>{e.preventDefault();e.stopPropagation();dz.classList.add("drag")}); dz.addEventListener("dragleave",()=>dz.classList.remove("drag")); - dz.addEventListener("drop",e=>{e.preventDefault();dz.classList.remove("drag"); if(e.dataTransfer.files.length) uploadFiles(e.dataTransfer.files);}); + // stopPropagation: the global window 'drop' listener also uploads, so + // without this a drop landing on the dropzone would bubble up and upload + // the same file twice. The dropzone owns drops on itself. + dz.addEventListener("drop",e=>{e.preventDefault();e.stopPropagation();dz.classList.remove("drag"); if(e.dataTransfer.files.length) uploadFiles(e.dataTransfer.files);}); w.appendChild(dz); c.appendChild(w); } @@ -627,21 +651,12 @@ async function uploadFiles(files){ item.querySelector(".upload-name").textContent=f.name; list.appendChild(item); const bar=item.querySelector(".upload-bar>div"), pct=item.querySelector(".upload-pct"), st=item.querySelector(".upload-status"); - const fd=new FormData(); fd.append("file",f); + // Large files go through the chunked path so each request stays under + // size-capping proxies' body limits (e.g. Cloudflare's 100MB). Files no + // bigger than one chunk use the simple single-shot multipart upload. + const useChunked = state.chunkSize && f.size>state.chunkSize; try{ - const t=await new Promise((res,rej)=>{ - const x=new XMLHttpRequest(); - x.open("POST","/api/tasks"); - x.setRequestHeader("X-Api-Key",state.pw); - x.upload.onprogress=e=>{ if(e.lengthComputable){ const p=Math.round(e.loaded/e.total*100); bar.style.width=p+"%"; pct.textContent=p+"%"; } }; - x.onload=()=>{ - if(x.status>=200&&x.status<300){ try{res(JSON.parse(x.responseText));}catch(e){res(null);} } - else if(x.status===401){ state.pw=""; localStorage.removeItem("asr_pw"); openAuthModal("密码不正确或服务已重启，请重新输入。",true); rej(new Error("unauthorized")); } - else { let m="HTTP "+x.status; try{m=(JSON.parse(x.responseText).error)||m;}catch(e){} rej(new Error(m)); } - }; - x.onerror=()=>rej(new Error("网络错误，上传失败")); - x.send(fd); - }); + const t = useChunked ? await uploadChunked(f,bar,pct,st) : await uploadSingle(f,bar,pct,st); bar.style.width="100%"; pct.textContent="100%"; st.textContent="已入队，转写排队中…"; item.classList.add("done"); toast("已加入队列: "+(t.original_name||t.id),"ok"); @@ -651,6 +666,88 @@ async function uploadFiles(files){ // Let the user see the finished state briefly, then refresh the queue. setTimeout(()=>{ state.uploading=false; loadTasks(); }, 600); } + +// uploadSingle sends a whole file as one multipart request (POST /api/tasks) +// with XHR so the progress bar reflects bytes sent. +function uploadSingle(f,bar,pct,st){ + return new Promise((res,rej)=>{ + const fd=new FormData(); fd.append("file",f); + const x=new XMLHttpRequest(); + x.open("POST","/api/tasks"); + x.setRequestHeader("X-Api-Key",state.pw); + x.upload.onprogress=e=>{ if(e.lengthComputable){ const p=Math.round(e.loaded/e.total*100); bar.style.width=p+"%"; pct.textContent=p+"%"; } }; + x.onload=()=>{ + if(x.status>=200&&x.status<300){ try{res(JSON.parse(x.responseText));}catch(e){res(null);} } + else if(x.status===401){ state.pw=""; localStorage.removeItem("asr_pw"); openAuthModal("密码不正确或服务已重启，请重新输入。",true); rej(new Error("unauthorized")); } + else { let m="HTTP "+x.status; try{m=(JSON.parse(x.responseText).error)||m;}catch(e){} rej(new Error(m)); } + }; + x.onerror=()=>rej(new Error("网络错误，上传失败")); + x.send(fd); + }); +} + +// uploadChunked sends a file as ceil(size/chunkSize) slices to the resumable +// upload API (POST /api/uploads -> /api/uploads/{id}?index=N -> /complete), +// tracking overall progress across slices. A failing slice is retried a few +// times on transient errors; on give-up the partial session is aborted. +async function uploadChunked(f,bar,pct,st){ + let uid=null; // declared outside try so the catch can reference it safely + try{ + const init = await api("/api/uploads", { + method:"POST", + headers:{"Content-Type":"application/json"}, + body: JSON.stringify({filename:f.name, size:f.size}) + }); + uid=init.upload_id; + const cs=+init.chunk_size, total=+init.total_chunks; + const setPct=bytes=>{ const p=Math.min(100,Math.round(bytes/f.size*100)); bar.style.width=p+"%"; pct.textContent=p+"%"; }; + let uploaded=0; + for(let i=0;i{ + const x=new XMLHttpRequest(); + x.open("POST", `/api/uploads/${uid}?index=${i}`); + x.setRequestHeader("X-Api-Key",state.pw); + x.upload.onprogress=e=>{ if(e.lengthComputable) setPct(uploaded+e.loaded); }; + x.onload=()=>{ + if(x.status>=200&&x.status<300){ try{res(JSON.parse(x.responseText));}catch(e){res(null);} return; } + const m=(()=>{ let s="HTTP "+x.status; try{s=(JSON.parse(x.responseText).error)||s;}catch(e){} return s; })(); + const e=new Error(m); e.status=x.status; rej(e); + }; + x.onerror=()=>{ const e=new Error("网络错误"); e.status=0; rej(e); }; + x.send(blob); + }); + done=true; + }catch(e){ + lastErr=e; + // Auth failure: surface immediately (the modal reopens); don't retry. + if(e.status===401){ state.pw=""; localStorage.removeItem("asr_pw"); openAuthModal("密码不正确或服务已重启，请重新输入。",true); throw e; } + // 4xx (bad request / session gone): retrying won't help; give up. + if(e.status>=400 && e.status<500 && e.status!==0) throw e; + // Transient (network/5xx): back off and retry unless last attempt. + if(attempt<2){ await new Promise(r=>setTimeout(r, 800*(attempt+1))); } + } + } + if(!done) throw lastErr || new Error("上传分片失败"); + uploaded += (end-start); + setPct(uploaded); + } + const task = await api(`/api/uploads/${uid}/complete`, {method:"POST"}); + return task; + }catch(e){ + // Best-effort cleanup of the partial session — but only if one was + // actually created (init may have failed) and not on an auth failure + // (the password is already cleared and a retry would just 401 again). + if(uid && e.status!==401 && state.pw){ + try{ await fetch(`/api/uploads/${uid}`,{method:"DELETE",headers:{"X-Api-Key":state.pw}}); }catch(_){} + } + throw e; + } +} $("#file").addEventListener("change",e=>{ if(e.target.files.length) uploadFiles(e.target.files); e.target.value=""; }); $("#uploadBtn").addEventListener("click",()=>$("#file").click()); diff --git a/internal/server/upload_large_test.go b/internal/server/upload_large_test.go new file mode 100644 index 0000000..fb04070 --- /dev/null +++ b/internal/server/upload_large_test.go @@ -0,0 +1,153 @@ +package server + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "testing" + + "git.misaka.ren/m1saka/record-asr/internal/config" + "git.misaka.ren/m1saka/record-asr/internal/store" +) + +// posReader emits a deterministic, position-dependent byte stream so that no +// two slices of the generated file are identical — a reorder would then +// change the reassembled hash and be caught. +type posReader struct{ pos int64 } + +func (r *posReader) Read(p []byte) (int, error) { + for i := range p { + x := r.pos + int64(i) + p[i] = byte((x*127 ^ (x >> 7)) & 0xff) + } + r.pos += int64(len(p)) + return len(p), nil +} + +// sha256sum returns the hex SHA-256 of a file, streaming so 120MB never sits +// in memory at once. +func sha256sum(t *testing.T, path string) string { + t.Helper() + f, err := os.Open(path) + if err != nil { + t.Fatalf("open %s: %v", path, err) + } + defer f.Close() + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + t.Fatalf("hash %s: %v", path, err) + } + return hex.EncodeToString(h.Sum(nil)) +} + +// TestChunkedUploadLargeFile uploads a 120MB file — larger than the 100MB CDN +// body cap the feature exists to bypass — as 25MB slices and checks that the +// reassembled file is byte-identical to the original. Guarded so the default +// test suite stays fast; run with: +// +// ASR_TEST_LARGE=1 go test -run Large ./internal/server/... +func TestChunkedUploadLargeFile(t *testing.T) { + if os.Getenv("ASR_TEST_LARGE") == "" { + t.Skip("set ASR_TEST_LARGE=1 to run the large-file chunked upload test (~120MB on disk)") + } + const total = 120 << 20 // 120 MiB — over the 100MB cap, no per-request limit here + const chunk = 25 << 20 // 25 MiB slices → 5 chunks (last is 20 MiB, partial) + + dir := t.TempDir() + st, err := store.Open(dir) + if err != nil { + t.Fatalf("store.Open: %v", err) + } + cfg := config.Config{DataDir: dir, ChunkSize: chunk, MaxUploadBytes: 512 << 20, MaxRetries: 3} + srv := New(cfg, st, nil, nil) // upload path never touches the transcribe/model stack + ts := httptest.NewServer(srv.Router()) + t.Cleanup(ts.Close) + + // 1. Generate the 120MB original on disk, hashing as we go. + origPath := filepath.Join(dir, "original.wav") + of, err := os.Create(origPath) + if err != nil { + t.Fatalf("create orig: %v", err) + } + if _, err := io.Copy(of, io.LimitReader(&posReader{}, total)); err != nil { + t.Fatalf("write orig: %v", err) + } + of.Close() + wantHash := sha256sum(t, origPath) + + // 2. Open a chunked-upload session. + body, _ := json.Marshal(map[string]any{"filename": "large.wav", "size": total}) + resp, err := http.Post(ts.URL+"/api/uploads", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("init: %v", err) + } + var init struct { + UploadID string `json:"upload_id"` + ChunkSize int64 `json:"chunk_size"` + TotalChunks int `json:"total_chunks"` + } + json.NewDecoder(resp.Body).Decode(&init) + resp.Body.Close() + if init.UploadID == "" || init.ChunkSize != chunk || init.TotalChunks != 5 { + t.Fatalf("init = %+v", init) + } + + // 3. Stream each slice straight from the original file (no full copy in RAM). + src, err := os.Open(origPath) + if err != nil { + t.Fatalf("open src: %v", err) + } + defer src.Close() + for i := 0; i < init.TotalChunks; i++ { + offset := int64(i) * chunk + size := int64(chunk) + if offset+size > total { + size = total - offset + } + req, _ := http.NewRequest("POST", + ts.URL+"/api/uploads/"+init.UploadID+"?index="+strconv.Itoa(i), + io.NewSectionReader(src, offset, size)) + req.Header.Set("Content-Type", "application/octet-stream") + cresp, cerr := http.DefaultClient.Do(req) + if cerr != nil { + t.Fatalf("chunk %d: %v", i, cerr) + } + if cresp.StatusCode != http.StatusOK { + t.Fatalf("chunk %d status = %d", i, cresp.StatusCode) + } + cresp.Body.Close() + } + + // 4. Complete → 202 + pending task. + req, _ := http.NewRequest("POST", ts.URL+"/api/uploads/"+init.UploadID+"/complete", nil) + resp, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("complete: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusAccepted { + t.Fatalf("complete status = %d", resp.StatusCode) + } + var task store.Task + json.NewDecoder(resp.Body).Decode(&task) + if task.Status != store.StatusPending { + t.Fatalf("task status = %q, want %q", task.Status, store.StatusPending) + } + + // 5. The reassembled file must be byte-identical (same SHA-256) and exact size. + gotHash := sha256sum(t, st.AudioPath(&task)) + if gotHash != wantHash { + t.Fatalf("reassembled file hash mismatch:\n want %s\n got %s", wantHash, gotHash) + } + fi, err := os.Stat(st.AudioPath(&task)) + if err != nil || fi.Size() != total { + t.Fatalf("final size = %d (%v), want %d", fi.Size(), err, total) + } +} diff --git a/internal/server/upload_test.go b/internal/server/upload_test.go new file mode 100644 index 0000000..d74c86f --- /dev/null +++ b/internal/server/upload_test.go @@ -0,0 +1,200 @@ +package server + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "strconv" + "testing" + + "git.misaka.ren/m1saka/record-asr/internal/config" + "git.misaka.ren/m1saka/record-asr/internal/store" +) + +// newUploadTestServer wires a Server backed by a real store + a minimal config +// (auth disabled, tiny 100B chunks) with nil pipeline/manager — the upload +// handlers never touch the transcribe/model stack, so this is safe. +func newUploadTestServer(t *testing.T) (*store.Store, *httptest.Server) { + t.Helper() + dir := t.TempDir() + st, err := store.Open(dir) + if err != nil { + t.Fatalf("store.Open: %v", err) + } + cfg := config.Config{ + DataDir: dir, + ChunkSize: 100, + MaxUploadBytes: 1 << 20, + MaxRetries: 3, + } + srv := New(cfg, st, nil, nil) + ts := httptest.NewServer(srv.Router()) + t.Cleanup(ts.Close) + return st, ts +} + +// initSession opens a chunked-upload session and returns the upload id. +func initSession(t *testing.T, ts *httptest.Server, name string, size int64) string { + t.Helper() + body, _ := json.Marshal(map[string]any{"filename": name, "size": size}) + resp, err := http.Post(ts.URL+"/api/uploads", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("init: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("init status = %d", resp.StatusCode) + } + var r struct { + UploadID string `json:"upload_id"` + ChunkSize int64 `json:"chunk_size"` + TotalChunks int `json:"total_chunks"` + } + if err := json.NewDecoder(resp.Body).Decode(&r); err != nil { + t.Fatalf("init decode: %v", err) + } + if r.UploadID == "" || r.ChunkSize != 100 { + t.Fatalf("init resp = %+v", r) + } + return r.UploadID +} + +func sendChunk(t *testing.T, ts, id string, idx int, data []byte) int { + t.Helper() + req, _ := http.NewRequest("POST", ts+"/api/uploads/"+id+"?index="+strconv.Itoa(idx), bytes.NewReader(data)) + req.Header.Set("Content-Type", "application/octet-stream") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("chunk %d: %v", idx, err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("chunk %d status = %d", idx, resp.StatusCode) + } + var r struct { + Received int `json:"received"` + Total int `json:"total"` + } + json.NewDecoder(resp.Body).Decode(&r) + return r.Received +} + +// TestChunkedUploadHTTP exercises the full endpoint flow end-to-end. +func TestChunkedUploadHTTP(t *testing.T) { + st, ts := newUploadTestServer(t) + + // /healthz advertises the chunk size the client uses as its threshold. + hresp, _ := http.Get(ts.URL + "/healthz") + var hz struct { + ChunkSize int64 `json:"chunk_size"` + MaxUploadBytes int64 `json:"max_upload_bytes"` + } + json.NewDecoder(hresp.Body).Decode(&hz) + hresp.Body.Close() + if hz.ChunkSize != 100 || hz.MaxUploadBytes != 1<<20 { + t.Fatalf("healthz = %+v", hz) + } + + // 250B file -> 3 chunks (100, 100, 50). Send out of order. + uid := initSession(t, ts, "clip.wav", 250) + c0 := bytes.Repeat([]byte{0xAA}, 100) + c1 := bytes.Repeat([]byte{0xBB}, 100) + c2 := bytes.Repeat([]byte{0xCC}, 50) + want := append(append([]byte{}, c0...), c1...) + want = append(want, c2...) + if got := sendChunk(t, ts.URL, uid, 1, c1); got != 1 { + t.Fatalf("after chunk 1 received=%d want 1", got) + } + if got := sendChunk(t, ts.URL, uid, 0, c0); got != 2 { + t.Fatalf("after chunk 0 received=%d want 2", got) + } + if got := sendChunk(t, ts.URL, uid, 2, c2); got != 3 { + t.Fatalf("after chunk 2 received=%d want 3", got) + } + + // Complete -> 202 + task (same shape as /api/tasks). + req, _ := http.NewRequest("POST", ts.URL+"/api/uploads/"+uid+"/complete", nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("complete: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusAccepted { + t.Fatalf("complete status = %d", resp.StatusCode) + } + var task store.Task + json.NewDecoder(resp.Body).Decode(&task) + if task.Status != store.StatusPending || task.OriginalName != "clip.wav" { + t.Fatalf("task = %+v", task) + } + + // The reassembled file matches the original bytes exactly. + got, err := os.ReadFile(st.AudioPath(&task)) + if err != nil { + t.Fatalf("read final file: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("reassembled bytes mismatch:\n got %x\n want %x", got, want) + } +} + +// TestChunkedUploadHTTPErrors checks the error paths: missing index, unknown +// session, oversized chunk, and idempotent abort. +func TestChunkedUploadHTTPErrors(t *testing.T) { + _, ts := newUploadTestServer(t) + + // Missing ?index -> 400. + resp, _ := http.Post(ts.URL+"/api/uploads/someid", "application/octet-stream", bytes.NewReader([]byte("x"))) + resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("missing-index status = %d, want 400", resp.StatusCode) + } + + // Complete on an unknown session -> 404. + req, _ := http.NewRequest("POST", ts.URL+"/api/uploads/doesnotexist/complete", nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("unknown complete: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("unknown complete status = %d, want 404", resp.StatusCode) + } + + // Oversized chunk (101B > 100B cap) -> 400. + uid := initSession(t, ts, "big.wav", 250) + req, _ = http.NewRequest("POST", ts.URL+"/api/uploads/"+uid+"?index=0", bytes.NewReader(bytes.Repeat([]byte{0x9}, 101))) + req.Header.Set("Content-Type", "application/octet-stream") + resp, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("oversized chunk: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("oversized chunk status = %d, want 400", resp.StatusCode) + } + + // Abort a fresh session -> 204, and the id is gone afterwards. + uid2 := initSession(t, ts, "abort.wav", 100) + req, _ = http.NewRequest("DELETE", ts.URL+"/api/uploads/"+uid2, nil) + resp, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("abort: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusNoContent { + t.Fatalf("abort status = %d, want 204", resp.StatusCode) + } + // Completing an aborted session -> 404. + req, _ = http.NewRequest("POST", ts.URL+"/api/uploads/"+uid2+"/complete", nil) + resp, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("post-abort complete: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("post-abort complete status = %d, want 404", resp.StatusCode) + } +} diff --git a/internal/store/errors.go b/internal/store/errors.go index c7fc2bc..38033b7 100644 --- a/internal/store/errors.go +++ b/internal/store/errors.go @@ -4,3 +4,7 @@ import "errors" // ErrNotFound is returned when a task id does not exist. var ErrNotFound = errors.New("task not found") + +// ErrUploadNotFound is returned when a chunked-upload session id does not exist +// (never created, already completed, or swept after being idle too long). +var ErrUploadNotFound = errors.New("upload session not found") diff --git a/internal/store/store.go b/internal/store/store.go index 3fff047..50012d7 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -55,8 +55,9 @@ type Store struct { tasksFile string uploadsDir string - mu sync.Mutex - tasks map[string]*Task + mu sync.Mutex + tasks map[string]*Task + sessions map[string]*UploadSession // in-progress chunked uploads } // Open initializes the store at dir, creating directories and loading any @@ -75,10 +76,14 @@ func Open(dir string) (*Store, error) { tasksFile: filepath.Join(dir, "tasks.json"), uploadsDir: uploads, tasks: map[string]*Task{}, + sessions: map[string]*UploadSession{}, } if err := s.load(); err != nil { return nil, err } + // In-memory upload sessions do not survive a restart; sweep any .part + // files left behind by an interrupted chunked upload so they don't leak. + s.sweepParts() return s, nil } @@ -130,7 +135,9 @@ func (s *Store) Create(originalName string, r io.Reader, maxBytes int64, maxRetr rel := filepath.Join("uploads", id+ext) dst := filepath.Join(s.dir, rel) - // Copy with a size cap so a malicious upload can't exhaust disk. + // Copy with a size cap so a malicious upload can't exhaust disk. The cap + // is inclusive: a body of exactly maxBytes is accepted, and anything past + // it is rejected. tmp := dst + ".tmp" f, err := os.Create(tmp) if err != nil { @@ -142,10 +149,7 @@ func (s *Store) Create(originalName string, r io.Reader, maxBytes int64, maxRetr os.Remove(tmp) return nil, fmt.Errorf("read upload: %w", err) } - if n >= maxBytes { - // Could be exactly at the limit; treat hitting the cap as too large - // unless the reader is now exhausted. We err on the side of caution. - } + _ = n // bytes copied (<= maxBytes); the cap above is the real guard if err := os.Rename(tmp, dst); err != nil { os.Remove(tmp) return nil, fmt.Errorf("save upload: %w", err) @@ -364,8 +368,9 @@ func newID() string { return hex.EncodeToString(b) } -// limitedReader wraps io.Reader with a hard byte cap; exceeding it returns an -// error so the upload is rejected before it fills the disk. +// limitedReader wraps io.Reader with an inclusive byte cap: a body that ends +// at exactly max is accepted (max is the allowed maximum), while any bytes +// past it are rejected so an oversized body can't fill the disk. type limitedReader struct { r io.Reader max int64 @@ -373,14 +378,22 @@ type limitedReader struct { } func (l *limitedReader) Read(p []byte) (int, error) { - if l.n >= l.max { - return 0, fmt.Errorf("upload exceeds %d bytes", l.max) - } - remain := l.max - l.n - if int64(len(p)) > remain { - p = p[:remain] + if l.n < l.max { + remain := l.max - l.n + if int64(len(p)) > remain { + p = p[:remain] + } + n, err := l.r.Read(p) + l.n += int64(n) + return n, err } + // Already at the cap. Tell an exactly-at-cap body (reader now exhausted) + // from one that still has bytes (over the cap): read once — any non-zero + // return means the body exceeded the cap. n, err := l.r.Read(p) l.n += int64(n) - return n, err + if n > 0 { + return n, fmt.Errorf("upload exceeds %d bytes", l.max) + } + return 0, err } diff --git a/internal/store/upload.go b/internal/store/upload.go new file mode 100644 index 0000000..315ea6a --- /dev/null +++ b/internal/store/upload.go @@ -0,0 +1,274 @@ +// upload.go extends the store with resumable chunked uploads: a way to receive +// a large file as many small requests (each under a size-capping proxy's body +// limit, e.g. Cloudflare's 100MB) and reassemble it into one task. +// +// Lifecycle: +// 1. CreateUpload — allocate an id + an empty ".part" file, record the +// expected total size and chunk layout. +// 2. AppendChunk — write chunk #index at offset index*chunkSize into the +// .part file. Chunks may arrive in any order and re-sending one is +// idempotent (it overwrites the same offset). +// 3. CompleteUpload — once every chunk has arrived, truncate the .part file +// to the exact size, rename it to "", and enqueue a pending task +// (reusing the upload id as the task id). +// 4. AbortUpload — drop the session and delete the .part file. +// +// Sessions live in memory (the .part file is the on-disk durability). A +// restart therefore loses in-flight sessions; sweepParts cleans up orphaned +// .part files on Open, and gcUploads lazily reaps sessions idle too long. +// +// Locking: Store.mu guards the sessions map; each UploadSession has its own +// mutex guarding its received[] slice and the part file it owns. The rule that +// prevents deadlock is: never acquire Store.mu while holding a session mutex. +// Append/Complete/GC all phase the two locks separately to honour this. + +package store + +import ( + "fmt" + "io" + "os" + "path/filepath" + "strings" + "sync" + "time" +) + +// sessionTTL is how long an upload session may stay idle before it (and its +// .part file) is garbage-collected. +const sessionTTL = 2 * time.Hour + +// UploadSession tracks a resumable chunked upload whose slices are written +// into a single .part file and finalized into a task on completion. +type UploadSession struct { + UploadID string + Filename string + Size int64 + ChunkSize int64 + TotalChunks int + + // Unexported state — callers reach these only through Store methods. + received []bool + partPath string + ext string + createdAt time.Time + lastSeen time.Time + mu sync.Mutex +} + +// CreateUpload starts a resumable chunked upload session. It allocates an +// upload id, creates an empty .part file, and records the expected chunk +// layout. maxBytes caps the total file size; chunkSize is the per-chunk size +// (must be > 0). The returned session's id is later reused as the task id. +func (s *Store) CreateUpload(filename string, size, chunkSize, maxBytes int64) (*UploadSession, error) { + if chunkSize <= 0 { + return nil, fmt.Errorf("chunk size must be > 0") + } + if size <= 0 { + return nil, fmt.Errorf("size must be > 0") + } + if size > maxBytes { + return nil, fmt.Errorf("upload size %d exceeds cap %d", size, maxBytes) + } + ext := filepath.Ext(filename) + id := newID() + partPath := filepath.Join(s.uploadsDir, id+".part") + f, err := os.OpenFile(partPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o644) + if err != nil { + return nil, fmt.Errorf("create part file: %w", err) + } + f.Close() + total := int((size + chunkSize - 1) / chunkSize) + now := time.Now().UTC() + sess := &UploadSession{ + UploadID: id, + Filename: filename, + Size: size, + ChunkSize: chunkSize, + TotalChunks: total, + received: make([]bool, total), + partPath: partPath, + ext: ext, + createdAt: now, + lastSeen: now, + } + s.gcUploads() // reap idle sessions opportunistically before adding a new one + s.mu.Lock() + s.sessions[id] = sess + s.mu.Unlock() + return sess, nil +} + +// AppendChunk writes chunk #index (0-based) of the session's file at offset +// index*chunkSize and marks it received. Chunks may arrive in any order and +// are idempotent (re-sending a chunk overwrites the same offset). The reader +// is capped at maxChunkBytes so a malicious or corrupt body can't overrun the +// slot; the written length must match the expected chunk length. +func (s *Store) AppendChunk(uploadID string, index int, r io.Reader, maxChunkBytes int64) error { + s.mu.Lock() + sess, ok := s.sessions[uploadID] + s.mu.Unlock() + if !ok { + return ErrUploadNotFound + } + sess.mu.Lock() + defer sess.mu.Unlock() + sess.lastSeen = time.Now().UTC() + if index < 0 || index >= sess.TotalChunks { + return fmt.Errorf("chunk index %d out of range [0,%d)", index, sess.TotalChunks) + } + offset := int64(index) * sess.ChunkSize + f, err := os.OpenFile(sess.partPath, os.O_WRONLY, 0o644) + if err != nil { + return fmt.Errorf("open part: %w", err) + } + defer f.Close() + if _, err := f.Seek(offset, io.SeekStart); err != nil { + return fmt.Errorf("seek: %w", err) + } + n, err := io.Copy(f, &limitedReader{r: r, max: maxChunkBytes}) + if err != nil { + return fmt.Errorf("write chunk: %w", err) + } + // Validate length: every full chunk == chunkSize; the last chunk is the + // remainder. A short or oversized body is rejected (not marked received). + expected := sess.ChunkSize + if index == sess.TotalChunks-1 { + expected = sess.Size - offset + } + if n != expected { + return fmt.Errorf("chunk %d length mismatch: got %d want %d", index, n, expected) + } + sess.received[index] = true + return nil +} + +// CompleteUpload finalizes a chunked upload: verifies every chunk arrived, +// truncates the part file to the exact size (repairing any overwrite from a +// rejected chunk), renames it to the task's audio file, and enqueues a +// pending transcription task. The upload id becomes the task id. +func (s *Store) CompleteUpload(uploadID string, maxRetries int) (*Task, error) { + s.mu.Lock() + sess, ok := s.sessions[uploadID] + s.mu.Unlock() + if !ok { + return nil, ErrUploadNotFound + } + // Session-scoped work (file ops) under the session lock only. + sess.mu.Lock() + for i, got := range sess.received { + if !got { + sess.mu.Unlock() + return nil, fmt.Errorf("missing chunk %d of %d", i, sess.TotalChunks) + } + } + if err := os.Truncate(sess.partPath, sess.Size); err != nil { + sess.mu.Unlock() + return nil, fmt.Errorf("truncate part: %w", err) + } + rel := filepath.Join("uploads", sess.UploadID+sess.ext) + dst := filepath.Join(s.dir, rel) + if err := os.Rename(sess.partPath, dst); err != nil { + sess.mu.Unlock() + return nil, fmt.Errorf("finalize upload: %w", err) + } + sess.mu.Unlock() + + // Register the task under the store lock (no session lock held). + now := time.Now().UTC() + t := &Task{ + ID: sess.UploadID, + Filename: rel, + OriginalName: sess.Filename, + Status: StatusPending, + CreatedAt: now, + UpdatedAt: now, + MaxRetries: maxRetries, + } + s.mu.Lock() + delete(s.sessions, uploadID) + s.tasks[uploadID] = t + serr := s.save() + s.mu.Unlock() + if serr != nil { + return nil, fmt.Errorf("persist task: %w", serr) + } + return t, nil +} + +// AbortUpload abandons a chunked upload session and removes its part file. +// It is idempotent (a missing session is not an error). +func (s *Store) AbortUpload(uploadID string) error { + s.mu.Lock() + sess, ok := s.sessions[uploadID] + if ok { + delete(s.sessions, uploadID) + } + s.mu.Unlock() + if !ok { + return nil + } + _ = os.Remove(sess.partPath) + return nil +} + +// UploadStatus reports how many chunks of a session have arrived so far. It +// lets a client show progress and decide whether a complete is safe. +func (s *Store) UploadStatus(uploadID string) (received, total int, ok bool) { + s.mu.Lock() + sess, ok2 := s.sessions[uploadID] + s.mu.Unlock() + if !ok2 { + return 0, 0, false + } + sess.mu.Lock() + defer sess.mu.Unlock() + n := 0 + for _, b := range sess.received { + if b { + n++ + } + } + return n, sess.TotalChunks, true +} + +// gcUploads reaps sessions idle for longer than sessionTTL and deletes their +// .part files. It is called lazily from CreateUpload so no extra goroutine is +// needed. The two locks are phased (map under Store.mu, idle-check under the +// session mutex; file removal after Store.mu is released) to honour the +// "never hold Store.mu while taking a session lock for long" rule. +func (s *Store) gcUploads() { + cutoff := time.Now().Add(-sessionTTL) + var dead []*UploadSession + s.mu.Lock() + for id, sess := range s.sessions { + sess.mu.Lock() + idle := sess.lastSeen.Before(cutoff) + sess.mu.Unlock() + if idle { + dead = append(dead, sess) + delete(s.sessions, id) + } + } + s.mu.Unlock() + for _, sess := range dead { + _ = os.Remove(sess.partPath) + } +} + +// sweepParts removes leftover .part files from interrupted chunked uploads +// after a restart (in-memory sessions don't survive). Called once from Open. +func (s *Store) sweepParts() { + entries, err := os.ReadDir(s.uploadsDir) + if err != nil { + return + } + for _, e := range entries { + if !e.Type().IsRegular() { + continue + } + if strings.HasSuffix(e.Name(), ".part") { + _ = os.Remove(filepath.Join(s.uploadsDir, e.Name())) + } + } +} diff --git a/internal/store/upload_test.go b/internal/store/upload_test.go new file mode 100644 index 0000000..2ef95ec --- /dev/null +++ b/internal/store/upload_test.go @@ -0,0 +1,203 @@ +package store + +import ( + "bytes" + "os" + "path/filepath" + "testing" +) + +// chunkBytes returns a deterministic slice of n bytes whose value is b. +func chunkBytes(n int, b byte) []byte { + out := make([]byte, n) + for i := range out { + out[i] = b + } + return out +} + +// newTestStore opens a store in a fresh temp dir. +func newTestStore(t *testing.T) *Store { + t.Helper() + s, err := Open(t.TempDir()) + if err != nil { + t.Fatalf("Open: %v", err) + } + return s +} + +// TestChunkedUploadReassemble verifies that out-of-order, idempotent chunks +// reassemble into the exact original bytes and enqueue a pending task. +func TestChunkedUploadReassemble(t *testing.T) { + s := newTestStore(t) + const chunk = 10 + const total = 25 // 3 chunks: 10 + 10 + 5 + + sess, err := s.CreateUpload("note.wav", total, chunk, 512) + if err != nil { + t.Fatalf("CreateUpload: %v", err) + } + if sess.TotalChunks != 3 { + t.Fatalf("TotalChunks = %d, want 3", sess.TotalChunks) + } + + // Expected file content: three slices with distinct byte values. + c0 := chunkBytes(10, 0xAA) + c1 := chunkBytes(10, 0xBB) + c2 := chunkBytes(5, 0xCC) + want := append(append([]byte{}, c0...), c1...) + want = append(want, c2...) + + // Send out of order: 1, 0, 2. + for _, step := range []struct { + idx int + data []byte + }{ + {1, c1}, {0, c0}, {2, c2}, + } { + if err := s.AppendChunk(sess.UploadID, step.idx, bytes.NewReader(step.data), chunk); err != nil { + t.Fatalf("AppendChunk(%d): %v", step.idx, err) + } + got, total, ok := s.UploadStatus(sess.UploadID) + if !ok { + t.Fatalf("UploadStatus: session gone") + } + _ = total + // After the last append all 3 should be received. + if step.idx == 2 && got != 3 { + t.Fatalf("after last chunk received=%d want 3", got) + } + } + + // Re-send chunk 0 (idempotent): content must be unchanged. + if err := s.AppendChunk(sess.UploadID, 0, bytes.NewReader(c0), chunk); err != nil { + t.Fatalf("AppendChunk(0) re-send: %v", err) + } + + task, err := s.CompleteUpload(sess.UploadID, 3) + if err != nil { + t.Fatalf("CompleteUpload: %v", err) + } + if task.Status != StatusPending { + t.Fatalf("task status = %q, want %q", task.Status, StatusPending) + } + if task.OriginalName != "note.wav" { + t.Fatalf("original_name = %q", task.OriginalName) + } + + // The part file is gone; the final audio file holds the exact bytes. + if _, err := os.Stat(filepath.Join(s.uploadsDir, sess.UploadID+".part")); !os.IsNotExist(err) { + t.Fatalf("part file still present: %v", err) + } + got, err := os.ReadFile(filepath.Join(s.dir, task.Filename)) + if err != nil { + t.Fatalf("read final file: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("final bytes mismatch:\n got %x\n want %x", got, want) + } + if int64(len(got)) != total { + t.Fatalf("final size = %d, want %d", len(got), total) + } + + // The session became a task the store knows about. + if s.Get(task.ID) == nil { + t.Fatalf("Get(task) returned nil after complete") + } + // ...and is no longer an upload session. + if _, _, ok := s.UploadStatus(sess.UploadID); ok { + t.Fatalf("session still present after complete") + } +} + +// TestChunkedUploadMissingChunk ensures complete fails when a chunk is absent. +func TestChunkedUploadMissingChunk(t *testing.T) { + s := newTestStore(t) + sess, _ := s.CreateUpload("x.wav", 20, 10, 512) // 2 chunks + _ = s.AppendChunk(sess.UploadID, 0, bytes.NewReader(chunkBytes(10, 1)), 10) + if _, err := s.CompleteUpload(sess.UploadID, 3); err == nil { + t.Fatalf("CompleteUpload succeeded with a missing chunk") + } + // The session survives a failed complete so the client can send the rest. + if _, _, ok := s.UploadStatus(sess.UploadID); !ok { + t.Fatalf("session dropped after failed complete") + } +} + +// TestChunkedUploadLengthMismatch rejects a chunk whose size is wrong. +func TestChunkedUploadLengthMismatch(t *testing.T) { + s := newTestStore(t) + sess, _ := s.CreateUpload("x.wav", 20, 10, 512) // 2 chunks of 10 + // Send only 7 bytes for a chunk that should be 10. + if err := s.AppendChunk(sess.UploadID, 0, bytes.NewReader(chunkBytes(7, 1)), 10); err == nil { + t.Fatalf("AppendChunk accepted a short chunk") + } +} + +// TestChunkedUploadAbort removes the part file and drops the session. +func TestChunkedUploadAbort(t *testing.T) { + s := newTestStore(t) + sess, _ := s.CreateUpload("x.wav", 10, 10, 512) + _ = s.AppendChunk(sess.UploadID, 0, bytes.NewReader(chunkBytes(10, 1)), 10) + part := filepath.Join(s.uploadsDir, sess.UploadID+".part") + if _, err := os.Stat(part); err != nil { + t.Fatalf("part file missing before abort: %v", err) + } + if err := s.AbortUpload(sess.UploadID); err != nil { + t.Fatalf("AbortUpload: %v", err) + } + if _, err := os.Stat(part); !os.IsNotExist(err) { + t.Fatalf("part file present after abort: %v", err) + } + if _, _, ok := s.UploadStatus(sess.UploadID); ok { + t.Fatalf("session still present after abort") + } + // Abort of an unknown session is idempotent, not an error. + if err := s.AbortUpload("nope"); err != nil { + t.Fatalf("AbortUpload(unknown) = %v, want nil", err) + } +} + +// TestSweepPartsOnOpen confirms a leftover .part file from a previous run is +// removed when the store reopens (in-memory sessions don't survive restarts). +func TestSweepPartsOnOpen(t *testing.T) { + dir := t.TempDir() + s1, err := Open(dir) + if err != nil { + t.Fatalf("Open: %v", err) + } + sess, _ := s1.CreateUpload("x.wav", 10, 10, 512) + part := filepath.Join(s1.uploadsDir, sess.UploadID+".part") + if _, err := os.Stat(part); err != nil { + t.Fatalf("part missing: %v", err) + } + s1.gcUploads() // no-op: session not idle yet; part must survive + if _, err := os.Stat(part); err != nil { + t.Fatalf("part swept too early: %v", err) + } + + // Reopen the same dir: the orphan .part file must be swept. + s2, err := Open(dir) + if err != nil { + t.Fatalf("Open(2): %v", err) + } + if _, err := os.Stat(filepath.Join(s2.uploadsDir, sess.UploadID+".part")); !os.IsNotExist(err) { + t.Fatalf("orphan .part file survived reopen: %v", err) + } +} + +// TestSweepPartsPreservesTaskAudio ensures the startup sweep only touches +// .part files, never finalized task audio (uploads/.). +func TestSweepPartsPreservesTaskAudio(t *testing.T) { + dir := t.TempDir() + s1, _ := Open(dir) + task, err := s1.Create("kept.wav", bytes.NewReader(chunkBytes(10, 7)), 512, 1) + if err != nil { + t.Fatalf("Create: %v", err) + } + audioPath := filepath.Join(s1.dir, task.Filename) + s1.sweepParts() // must not delete the task's audio + if _, err := os.Stat(audioPath); err != nil { + t.Fatalf("task audio swept by sweepParts: %v", err) + } +}