diff --git a/desktop/app.go b/desktop/app.go index 58ba6f7..27c8523 100644 --- a/desktop/app.go +++ b/desktop/app.go @@ -332,6 +332,29 @@ func (a *App) SaveDownload(name string, data []byte) (string, error) { return platform.SaveDownload(platform.ResolveDownloadDir(), name, data) } +// BeginDownload / AppendDownload / FinalizeDownload / AbortDownload are the +// streaming counterpart of SaveDownload, bound to the WebView. The receiver's +// hybrid sink keeps a file in memory until it crosses a cap, then spills the +// overflow here so large files don't sit in the WebView heap (the wails:// no-OPFS +// deadlock). Begin opens a temp file in the download dir, Append streams batches +// (base64 over the bridge), Finalize renames it into place and returns the path, +// Abort discards it on cancel / failure. +func (a *App) BeginDownload(sessionId string) error { + return platform.BeginStreamingDownload(platform.ResolveDownloadDir(), sessionId) +} + +func (a *App) AppendDownload(sessionId string, data []byte) error { + return platform.AppendStreamingDownload(sessionId, data) +} + +func (a *App) FinalizeDownload(sessionId, name string) (string, error) { + return platform.FinalizeStreamingDownload(sessionId, name) +} + +func (a *App) AbortDownload(sessionId string) { + platform.AbortStreamingDownload(sessionId) +} + // ChooseDownloadDir opens a native directory picker (title localised by the // caller) and returns the chosen path, or "" when the user cancels. The page // persists the result via SaveSettings. diff --git a/desktop/platform/download.go b/desktop/platform/download.go index 7fcad34..9167602 100644 --- a/desktop/platform/download.go +++ b/desktop/platform/download.go @@ -7,6 +7,7 @@ import ( "path/filepath" "runtime" "strings" + "sync" ) // SaveDownload writes a received file's bytes into dir (empty → system Downloads), @@ -102,7 +103,12 @@ func uniquePath(dir, name string) string { // revealInFileManager opens the OS file manager with the saved file selected. // Best-effort: the save already succeeded, so any error here is ignored. +// CDROP_NO_REVEAL skips the side effect — for headless / CI / test runs that +// shouldn't pop a Finder/Explorer window. func revealInFileManager(path string) { + if os.Getenv("CDROP_NO_REVEAL") != "" { + return + } switch runtime.GOOS { case "darwin": _ = exec.Command("open", "-R", path).Start() @@ -110,3 +116,159 @@ func revealInFileManager(path string) { _ = exec.Command("explorer", "/select,"+path).Start() } } + +// Streaming downloads — the receiver's hybrid sink (web/.../incomingSink.ts) keeps +// a file in memory until it crosses a cap (default 256 MB), then spills to disk +// through this manager instead of carrying the whole thing in the WebView heap. We +// can't use SaveDownload for that: it takes the complete bytes at once. The reason +// the whole hybrid exists is the wails:// scheme has no persistent WebView storage, +// so OPFS degrades and starves the receive loop (the PWA→Mac ~15 MB deadlock); the +// fix is to stage bounded memory and stream the overflow into a Go-owned temp file. +// +// The temp file lives in the SAME directory as the final target so finalize is an +// atomic rename, not a cross-device copy. The writable dir is chosen up front (with +// the same unwritable→system-Downloads fallback as SaveDownload) because once GBs +// are streamed we can't cheaply switch volumes. + +type streamingDownload struct { + file *os.File + dir string // the writable dir chosen at Begin; the final file is renamed into it + tmp string // the .part temp path under dir +} + +var ( + streamMu sync.Mutex + streamDownloads = map[string]*streamingDownload{} +) + +// BeginStreamingDownload opens a .part temp file under a writable download dir for +// the session. dir == "" → system Downloads; an unwritable dir falls back to system +// Downloads (mirrors SaveDownload). A stale entry for the same session id is closed +// and removed first, so a retried transfer never leaks a handle or temp file. +func BeginStreamingDownload(dir, sessionId string) error { + if dir == "" { + dir = DefaultDownloadDir() + } + // Drop any stale handle for this session FIRST (a retried transfer). Its temp + // path equals the new one (both derive from the session id), so removing it + // after opening the fresh file would delete the file we just created. Clean + // up first; openStreamTemp then re-creates the temp via O_TRUNC. + streamMu.Lock() + if prev, ok := streamDownloads[sessionId]; ok { + _ = prev.file.Close() + _ = os.Remove(prev.tmp) + delete(streamDownloads, sessionId) + } + streamMu.Unlock() + + f, dirUsed, tmp, err := openStreamTemp(dir, sessionId) + if err != nil { + return err + } + streamMu.Lock() + streamDownloads[sessionId] = &streamingDownload{file: f, dir: dirUsed, tmp: tmp} + streamMu.Unlock() + return nil +} + +// openStreamTemp creates a writable .part temp file under dir, retrying into the +// system Downloads dir if dir is unwritable (mirrors SaveDownload's fallback). +func openStreamTemp(dir, sessionId string) (*os.File, string, string, error) { + f, tmp, err := tryStreamTemp(dir, sessionId) + if err == nil { + return f, dir, tmp, nil + } + if def := DefaultDownloadDir(); def != dir { + if f2, tmp2, ferr := tryStreamTemp(def, sessionId); ferr == nil { + return f2, def, tmp2, nil + } + } + return nil, "", "", err +} + +func tryStreamTemp(dir, sessionId string) (*os.File, string, error) { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, "", err + } + tmp := filepath.Join(dir, ".cdrop-"+sanitizeSessionID(sessionId)+".part") + f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) + if err != nil { + return nil, "", err + } + return f, tmp, nil +} + +// sanitizeSessionID reduces the peer-supplied session id to filename-safe chars so +// it can name the .part temp file. It only has to be unique and path-safe, not +// reversible — the real filename is applied at finalize. +func sanitizeSessionID(id string) string { + var b strings.Builder + for _, r := range id { + switch { + case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_': + b.WriteRune(r) + } + } + s := b.String() + if s == "" { + s = "session" + } + if len(s) > 64 { + s = s[:64] + } + return s +} + +// AppendStreamingDownload writes a batch of bytes to the session's temp file. +func AppendStreamingDownload(sessionId string, data []byte) error { + streamMu.Lock() + d := streamDownloads[sessionId] + streamMu.Unlock() + if d == nil { + return fmt.Errorf("no streaming download for session %q", sessionId) + } + _, err := d.file.Write(data) + return err +} + +// FinalizeStreamingDownload closes the temp file and renames it to a non-colliding +// final name under the same dir (name sanitized like SaveDownload), then reveals +// it. Returns the absolute path written. +func FinalizeStreamingDownload(sessionId, name string) (string, error) { + streamMu.Lock() + d := streamDownloads[sessionId] + delete(streamDownloads, sessionId) + streamMu.Unlock() + if d == nil { + return "", fmt.Errorf("no streaming download for session %q", sessionId) + } + if err := d.file.Close(); err != nil { + _ = os.Remove(d.tmp) + return "", err + } + safe := sanitizeFileName(name) + if safe == "" { + safe = "cdrop-download" + } + target := uniquePath(d.dir, safe) + if err := os.Rename(d.tmp, target); err != nil { + _ = os.Remove(d.tmp) + return "", err + } + revealInFileManager(target) + return target, nil +} + +// AbortStreamingDownload closes and deletes the session's temp file (cancel / +// failure path). No-op when the session isn't streaming. +func AbortStreamingDownload(sessionId string) { + streamMu.Lock() + d := streamDownloads[sessionId] + delete(streamDownloads, sessionId) + streamMu.Unlock() + if d == nil { + return + } + _ = d.file.Close() + _ = os.Remove(d.tmp) +} diff --git a/desktop/platform/download_stream_test.go b/desktop/platform/download_stream_test.go new file mode 100644 index 0000000..eff0f10 --- /dev/null +++ b/desktop/platform/download_stream_test.go @@ -0,0 +1,136 @@ +package platform + +import ( + "bytes" + "os" + "path/filepath" + "testing" +) + +// streamAll runs the full Begin → Append* → Finalize cycle for one session into +// dir and returns the final path. CDROP_NO_REVEAL keeps Finder/Explorer shut. +func streamAll(t *testing.T, dir, sessionID, name string, chunks ...[]byte) string { + t.Helper() + t.Setenv("CDROP_NO_REVEAL", "1") + if err := BeginStreamingDownload(dir, sessionID); err != nil { + t.Fatalf("Begin: %v", err) + } + for i, c := range chunks { + if err := AppendStreamingDownload(sessionID, c); err != nil { + t.Fatalf("Append %d: %v", i, err) + } + } + path, err := FinalizeStreamingDownload(sessionID, name) + if err != nil { + t.Fatalf("Finalize: %v", err) + } + return path +} + +func TestStreamingDownload_WritesAllBytes(t *testing.T) { + dir := t.TempDir() + path := streamAll(t, dir, "sess-1", "greeting.txt", + []byte("hello "), []byte("streamed "), []byte("world")) + + if got, want := path, filepath.Join(dir, "greeting.txt"); got != want { + t.Errorf("path = %q, want %q", got, want) + } + got, err := os.ReadFile(path) + if err != nil { + t.Fatalf("ReadFile: %v", err) + } + if want := "hello streamed world"; string(got) != want { + t.Errorf("content = %q, want %q", got, want) + } + // The .part temp must be gone after a successful finalize. + if _, err := os.Stat(filepath.Join(dir, ".cdrop-sess-1.part")); !os.IsNotExist(err) { + t.Errorf("temp file not cleaned up: %v", err) + } +} + +func TestStreamingDownload_NoCollisionOverwrite(t *testing.T) { + dir := t.TempDir() + // Pre-existing file with the same name must not be overwritten. + if err := os.WriteFile(filepath.Join(dir, "f.bin"), []byte("original"), 0o644); err != nil { + t.Fatal(err) + } + path := streamAll(t, dir, "sess-2", "f.bin", []byte("new")) + + if got, want := path, filepath.Join(dir, "f (1).bin"); got != want { + t.Errorf("path = %q, want %q", got, want) + } + orig, _ := os.ReadFile(filepath.Join(dir, "f.bin")) + if string(orig) != "original" { + t.Errorf("original file was overwritten: %q", orig) + } +} + +func TestStreamingDownload_SanitizesTraversalName(t *testing.T) { + dir := t.TempDir() + // A crafted name with path components must collapse to a bare filename under + // dir — it must never escape via "../". + path := streamAll(t, dir, "sess-3", "../../etc/evil", []byte("x")) + + if filepath.Dir(path) != dir { + t.Errorf("escaped dir: path = %q, dir = %q", path, dir) + } + if got := filepath.Base(path); got != "evil" { + t.Errorf("base = %q, want %q", got, "evil") + } +} + +func TestStreamingDownload_AbortRemovesTemp(t *testing.T) { + dir := t.TempDir() + t.Setenv("CDROP_NO_REVEAL", "1") + if err := BeginStreamingDownload(dir, "sess-4"); err != nil { + t.Fatalf("Begin: %v", err) + } + if err := AppendStreamingDownload("sess-4", []byte("partial")); err != nil { + t.Fatalf("Append: %v", err) + } + tmp := filepath.Join(dir, ".cdrop-sess-4.part") + if _, err := os.Stat(tmp); err != nil { + t.Fatalf("temp should exist mid-stream: %v", err) + } + AbortStreamingDownload("sess-4") + if _, err := os.Stat(tmp); !os.IsNotExist(err) { + t.Errorf("temp file not removed after abort: %v", err) + } + // A second append after abort must error, not panic or write a stray file. + if err := AppendStreamingDownload("sess-4", []byte("late")); err == nil { + t.Error("Append after abort: want error, got nil") + } +} + +func TestStreamingDownload_ReBeginReplacesStale(t *testing.T) { + dir := t.TempDir() + t.Setenv("CDROP_NO_REVEAL", "1") + // First Begin + partial write, then a second Begin for the same session id + // (a retried transfer) must truncate, not append to, the stale temp. + if err := BeginStreamingDownload(dir, "sess-5"); err != nil { + t.Fatalf("Begin 1: %v", err) + } + if err := AppendStreamingDownload("sess-5", []byte("STALE")); err != nil { + t.Fatalf("Append 1: %v", err) + } + if err := BeginStreamingDownload(dir, "sess-5"); err != nil { + t.Fatalf("Begin 2: %v", err) + } + if err := AppendStreamingDownload("sess-5", []byte("fresh")); err != nil { + t.Fatalf("Append 2: %v", err) + } + path, err := FinalizeStreamingDownload("sess-5", "out.bin") + if err != nil { + t.Fatalf("Finalize: %v", err) + } + got, _ := os.ReadFile(path) + if !bytes.Equal(got, []byte("fresh")) { + t.Errorf("content = %q, want %q (stale bytes leaked)", got, "fresh") + } +} + +func TestStreamingDownload_FinalizeUnknownSession(t *testing.T) { + if _, err := FinalizeStreamingDownload("never-began", "x"); err == nil { + t.Error("Finalize unknown session: want error, got nil") + } +} diff --git a/web/src/features/clipboard/ClipboardPanel.tsx b/web/src/features/clipboard/ClipboardPanel.tsx index 4b44ff2..9720085 100644 --- a/web/src/features/clipboard/ClipboardPanel.tsx +++ b/web/src/features/clipboard/ClipboardPanel.tsx @@ -7,7 +7,7 @@ import { RefreshCw, Trash2, } from "lucide-react"; -import { Button, IconButton, Panel, Tooltip } from "../../ui/primitives"; +import { Button, DynText, IconButton, Panel, Tooltip } from "../../ui/primitives"; import { toast } from "../../ui/feedback"; import type { ClipboardState } from "../../store"; import { t } from "../../i18n"; @@ -159,7 +159,7 @@ export function ClipboardPanel(props: ClipboardPanelProps)
{sizeText} - {fromText} + {fromText}
{revealed ? ( diff --git a/web/src/features/devices/DeviceStrip.tsx b/web/src/features/devices/DeviceStrip.tsx index 6e4ae7c..f2d519d 100644 --- a/web/src/features/devices/DeviceStrip.tsx +++ b/web/src/features/devices/DeviceStrip.tsx @@ -2,6 +2,7 @@ import { useMemo, useState, type DragEvent } from "react"; import { Check, ScanLine } from "lucide-react"; import { DeviceTypeIcon, StateDot } from "../../ui/glyphs"; import type { DeviceKind } from "../../ui/glyphs"; +import { DynText } from "../../ui/primitives"; import type { DeviceInfo } from "../../store"; import { formatRelative } from "../../utils/format"; import { t } from "../../i18n"; @@ -41,7 +42,7 @@ export function DeviceStrip(props: DeviceStripProps)
{t("home.deviceList.title")} - {t("home.deviceList.onlineSuffix", { count: onlineCount })} + {t("home.deviceList.onlineSuffix", { count: onlineCount })}
@@ -51,9 +52,11 @@ export function DeviceStrip(props: DeviceStripProps) className={s.headerSwitch} onClick={() => setShowOffline((v) => !v)} > - {showOffline - ? t("home.deviceList.hideOffline") - : t("home.deviceList.showOffline", { count: offlineCount })} + + {showOffline + ? t("home.deviceList.hideOffline") + : t("home.deviceList.showOffline", { count: offlineCount })} + )} {onLinkDevice && ( @@ -148,9 +151,11 @@ function DevicePort(props: DevicePortProps)
- {device.online - ? t("settings.online") - : t("settings.lastSeen", { time: formatRelative(device.lastSeen) })} + + {device.online + ? t("settings.online") + : t("settings.lastSeen", { time: formatRelative(device.lastSeen) })} +
diff --git a/web/src/features/messaging/MessageList.tsx b/web/src/features/messaging/MessageList.tsx index c096b91..bfeae03 100644 --- a/web/src/features/messaging/MessageList.tsx +++ b/web/src/features/messaging/MessageList.tsx @@ -1,5 +1,5 @@ import { X } from "lucide-react"; -import { Button } from "../../ui/primitives"; +import { Button, DynText } from "../../ui/primitives"; import type { MessageRecord } from "../../store"; import { t } from "../../i18n"; import { formatRelative } from "../../utils/format"; @@ -23,7 +23,7 @@ export function MessageList(props: MessageListProps)
{t("home.message.title")} - {t("home.message.countSuffix", { count: messages.length })} + {t("home.message.countSuffix", { count: messages.length })}
) : ( )}
diff --git a/web/src/features/transfer/TransferRow.tsx b/web/src/features/transfer/TransferRow.tsx index 6f86184..928ce76 100644 --- a/web/src/features/transfer/TransferRow.tsx +++ b/web/src/features/transfer/TransferRow.tsx @@ -4,6 +4,7 @@ import type { TransferPhase, TransferRecord } from "../../store"; import { useAppStore } from "../../store"; import { t, type TranslationKey } from "../../i18n"; import { formatBytes } from "../../utils/format"; +import { DynText } from "../../ui/primitives"; import { cancelTransfer, skipWaitRelay } from "./transfer"; import { TransferDetails } from "./TransferDetails"; import s from "./TransferRow.module.css"; @@ -45,6 +46,8 @@ export function TransferRow(props: TransferRowProps) && FLOW_PHASES.has(tr.phase) && sinceMs >= STALLED_THRESHOLD_MS; + const status = userFriendlyStatus(tr, stalled, sinceMs); + const speed = speedText(tr, stalled); const pct = computeProgress(tr); const failed = tr.state === "FAILED"; const done = tr.state === "DONE"; @@ -92,13 +95,22 @@ export function TransferRow(props: TransferRowProps)
- {userFriendlyStatus(tr, stalled, sinceMs)} + {/* 状态文案含 CJK 且每秒 / 随 phase 就地更新;DynText 用 key 让其整体重挂, + 避免聚珍 shim 拆出的碎片累积(见 DynText)。 */} + {status} {bytesText(tr) && ( <> · {bytesText(tr)} )} + {speed && ( + <> + · + {/* 速度含 CJK「/秒」且高频更新,同样过 DynText */} + {speed} + + )}
@@ -176,6 +188,17 @@ function bytesText(tr: TransferRecord): string }); } +// 瞬时速度文案。仅在字节流动阶段、未卡死、且有正有限速率时给出——卡死时既有「似乎 +// 卡住了」状态已表达停滞,再显示一个陈旧的非零速度会误导。速率由 store 做 EMA 平滑。 +function speedText(tr: TransferRecord, stalled: boolean): string +{ + if (stalled) { return ""; } + if (!FLOW_PHASES.has(tr.phase ?? "initializing")) { return ""; } + const rate = tr.bytesPerSec; + if (!rate || !Number.isFinite(rate) || rate <= 0) { return ""; } + return t("transfer.bytesRate", { rate: formatBytes(rate) }); +} + function computeProgress(tr: TransferRecord): number { if (tr.state === "DONE") { return 100; } diff --git a/web/src/features/transfer/incomingSink.ts b/web/src/features/transfer/incomingSink.ts index 7e9f5fe..60dd187 100644 --- a/web/src/features/transfer/incomingSink.ts +++ b/web/src/features/transfer/incomingSink.ts @@ -1,33 +1,65 @@ // 接收端的字节落地槽,用来避免把整个文件累积到 JS 堆。 // -// 背景:iOS Safari 单 tab 内存预算紧;超过后系统会节流 / 暂停 JS 执行, -// 导致 SCTP 不再 ack → sender bufferedAmount 撑满 → 死锁(用户实测 24.4 MB -// 卡住的根因)。改成流式写入 Origin Private File System(OPFS)后内存只占 -// 一个 chunk,iOS 16+ / Chrome / Firefox 都已支持 createWritable。 +// 三条路径,按运行环境选: +// - 浏览器(含 iOS PWA):流式写 Origin Private File System(OPFS),内存只占 +// 一个 chunk。iOS Safari 单 tab 内存预算紧,超过即被系统节流 / 暂停 JS → SCTP +// 不再 ack → sender bufferedAmount 撑满死锁(实测 24.4 MB 卡住的根因),故必须 +// 流式。Safari 16+ / Chrome 102+ / Firefox 111+ 都已支持 createWritable。 +// - 桌面(Wails WKWebView):混合有界内存槽。wails:// 自定义 scheme 无持久 WebView +// 存储,OPFS 在此降级、createWritable().write() 慢且回压不畅,会把接收 promise +// 链堵住、拖垮主线程到 dc.onmessage 饿死 → 接收方 SCTP rwnd=0 → 发送方 +// bufferedAmount 卡死在 16 MiB → 传输死锁(实测 PWA→Mac 约 15 MB 卡住)。故桌面 +// 不碰 OPFS:小文件(暂存 < CAP)纯内存累积、close 时整文件过 Go 写盘;越过内存 +// 上限(CAP=256 MB)的大文件改为流式 spill 到 Go 临时文件(每 64 MB 一批),内存 +// 恒定有界。桌面 RAM 充裕,整体不会撑爆堆,又彻底避开 OPFS 那条饿死链路。 +// - 老浏览器 / OPFS 不可用:退回纯内存累积。 // -// 老浏览器 / OPFS 不可用时退回到老的内存累积方案;移动端可能仍然受影响, -// 但桌面端无碍。 +// close() 返回 SinkResult:浏览器 / 桌面小文件返回 Blob(走 downloadBlob——浏览器 +// 原生下载 / 桌面整文件过 Go 写盘);桌面已 spill 的大文件返回已落盘绝对路径(文件 +// 已在 Go 侧写好,跳过 Blob)。 import { t } from "../../i18n"; -import { isDesktop, saveIncomingFileDesktop } from "../../net/desktop"; +import { + abortIncomingDownloadDesktop, + appendIncomingDownloadDesktop, + beginIncomingDownloadDesktop, + finalizeIncomingDownloadDesktop, + isDesktop, + saveIncomingFileDesktop, +} from "../../net/desktop"; import { toast } from "../../ui/feedback"; +// close() 两态结果:blob=整文件在内存,交给 downloadBlob;saved=桌面大文件已流式 +// 落盘,path 是最终绝对路径(无需再过 Blob)。 +export type SinkResult = + | { kind: "blob"; blob: Blob } + | { kind: "saved"; path: string }; + export interface IncomingSink { /** 顺序写一个 chunk;返回的 Promise 只有完全落盘后才 resolve,自然 backpressure */ write(chunk: ArrayBuffer): Promise; - /** 收到 done 帧后调用:finalize + 取出最终 Blob 给浏览器下载 */ - close(): Promise; - /** 失败 / 取消路径上调用,确保 OPFS 临时文件被删 */ + /** 收到 done 帧后调用:finalize 并返回最终结果(Blob 或已落盘路径) */ + close(): Promise; + /** 失败 / 取消路径上调用,确保 OPFS / 流式临时文件被删 */ cancel(): Promise; } +// 桌面混合 sink 的内存上限与 spill 批大小:暂存达 CAP 即转流式落盘,之后每满一批刷。 +const DESKTOP_MEMORY_CAP = 256 * 1024 * 1024; +const DESKTOP_FLUSH_BATCH = 64 * 1024 * 1024; + /** - * 优先 OPFS,失败回退内存。sessionId 用作 OPFS 临时文件名(确保两台设备 - * 同时收不同会话不会撞文件)。 + * 按环境选 sink:桌面走混合有界内存槽(见文件头注释),浏览器优先 OPFS、失败回退 + * 内存。sessionId 用作 OPFS 临时文件名 / 桌面流式落盘 key(确保并发会话不撞文件); + * fileName 供桌面 spill 落盘后命名(仅大文件路径用到)。 */ -export async function openIncomingSink(sessionId: string): Promise +export async function openIncomingSink(sessionId: string, fileName: string): Promise { + if (isDesktop()) + { + return openHybridDesktopSink(sessionId, fileName); + } // 注意:直接 navigator.storage?.getDirectory 在 TS 下永远 truthy(optional // chaining 取的是函数引用,不会调用)。要真探测必须 try-catch 调用。 if (typeof navigator !== "undefined" && navigator.storage) @@ -42,6 +74,99 @@ export async function openIncomingSink(sessionId: string): Promise return openMemorySink(); } +/** + * 桌面混合有界内存槽。小文件(暂存 < CAP)纯内存累积,close 时返回 Blob 交给 + * downloadBlob(= 既有桌面整文件过 Go 写盘路径);一旦暂存越过 CAP,转流式落盘—— + * 把已暂存的全部按 ≤批大小刷给 Go、释放内存,之后每满一批再刷,内存恒定有界。 + * + * write() 由调用方(p2p / relay 的 receiveQueue promise 链)严格串行调用,故内部 + * 无需再加锁。 + */ +function openHybridDesktopSink(sessionId: string, fileName: string): IncomingSink +{ + let pending: ArrayBuffer[] = []; + let pendingBytes = 0; + let spilled = false; + let closed = false; + + // 从 pending 头部取出至多 DESKTOP_FLUSH_BATCH 字节拼成一块 append 给 Go。 + // drainAll=true 把剩余全部刷完(spill 起始 / close);否则只在满一批时刷。 + const flush = async (drainAll: boolean): Promise => + { + while (pendingBytes >= DESKTOP_FLUSH_BATCH || (drainAll && pendingBytes > 0)) + { + let take = 0; + const batch: ArrayBuffer[] = []; + while (pending.length > 0 + && (take === 0 || take + pending[0].byteLength <= DESKTOP_FLUSH_BATCH)) + { + const c = pending.shift() as ArrayBuffer; + batch.push(c); + take += c.byteLength; + } + pendingBytes -= take; + await appendIncomingDownloadDesktop(sessionId, concatChunks(batch, take)); + } + }; + + return { + async write(chunk) + { + if (closed) { return; } + // 复制一份:底层 ArrayBuffer 来自 dc.onmessage 的 event.data,异步链上 + // 保留独立副本避免被回收 / 复用。 + pending.push(chunk.slice(0)); + pendingBytes += chunk.byteLength; + + if (!spilled) + { + if (pendingBytes >= DESKTOP_MEMORY_CAP) + { + // 越过内存上限:开始流式落盘,把已暂存的全部刷给 Go,转滚动批。 + spilled = true; + await beginIncomingDownloadDesktop(sessionId); + await flush(true); + } + return; + } + await flush(false); + }, + async close() + { + if (closed) { throw new Error("sink already closed"); } + closed = true; + if (!spilled) + { + // 小文件:从未落盘,整文件还在内存 → Blob 交给 downloadBlob。 + return { kind: "blob", blob: new Blob(pending) }; + } + await flush(true); + const path = await finalizeIncomingDownloadDesktop(sessionId, fileName); + return { kind: "saved", path }; + }, + async cancel() + { + closed = true; + pending = []; + pendingBytes = 0; + if (spilled) { await abortIncomingDownloadDesktop(sessionId); } + }, + }; +} + +// concatChunks 把若干 ArrayBuffer 拼成一个 Uint8Array(供 base64 过桥)。 +function concatChunks(parts: ArrayBuffer[], totalBytes: number): Uint8Array +{ + const out = new Uint8Array(totalBytes); + let off = 0; + for (const p of parts) + { + out.set(new Uint8Array(p), off); + off += p.byteLength; + } + return out; +} + async function openOPFSSink(sessionId: string): Promise { const dir = await navigator.storage.getDirectory(); @@ -70,7 +195,10 @@ async function openOPFSSink(sessionId: string): Promise await writable.close(); const file = await handle.getFile(); // 文件在 OPFS 里仍占空间;download 触发后再删(finalizeIncoming 里调) - return new Blob([file], { type: file.type || "application/octet-stream" }); + return { + kind: "blob", + blob: new Blob([file], { type: file.type || "application/octet-stream" }), + }; }, async cancel() { @@ -97,7 +225,7 @@ function openMemorySink(): IncomingSink async close() { closed = true; - return new Blob(chunks); + return { kind: "blob", blob: new Blob(chunks) }; }, async cancel() { @@ -107,6 +235,19 @@ function openMemorySink(): IncomingSink }; } +// deliverIncoming 把 sink.close() 的结果落到用户:已落盘(桌面大文件)只提示路径, +// 文件已在 Go 侧写好;Blob(浏览器 / 桌面小文件)交给 downloadBlob(浏览器原生下载 +// / 桌面整文件过 Go 写盘)。p2p / relay 的接收完成路径都走这里。 +export function deliverIncoming(result: SinkResult, fileName: string, sessionId: string): void +{ + if (result.kind === "saved") + { + toast.ok(t("transfer.savedTo", { path: result.path })); + return; + } + downloadBlob(result.blob, fileName, sessionId); +} + /** * 触发浏览器下载并把 OPFS 临时文件 + ObjectURL 都安排清理。 * 60s 给浏览器把 blob 落到用户磁盘的窗口,过短可能导致下载读到一半源被删。 diff --git a/web/src/features/transfer/p2p.ts b/web/src/features/transfer/p2p.ts index 4153032..aeb9797 100644 --- a/web/src/features/transfer/p2p.ts +++ b/web/src/features/transfer/p2p.ts @@ -1,7 +1,7 @@ import { apiFetch } from "../../net/api"; import { getICEServers } from "./iceServers"; import { useAppStore, type CandidateBreakdown, type IceStats, type TransferPhase } from "../../store"; -import { downloadBlob, openIncomingSink, type IncomingSink } from "./incomingSink"; +import { deliverIncoming, openIncomingSink, type IncomingSink } from "./incomingSink"; // WebRTC client wired to the brief §2 invariants: // - iceServers: pulled from lib/iceServers (Cloudflare Realtime TURN over @@ -91,6 +91,9 @@ class Session // → sender 的 bufferedAmount 永远卡在 HIGH。把 store push 限到 10 Hz。 private storePushScheduled: number | null = null; private lastMessageAt = 0; + // 发送进度轮询:streamFile 期间每 250ms 刷一次,让"已交付字节"在 backpressure + // 等待期间也能平滑爬升、不误报 stalled(见 streamFile / senderDeliveredBytes)。 + private sendProgressPollId: number | null = null; constructor(public sessionId: string, public peerName: string, public role: "sender" | "receiver") { @@ -207,7 +210,7 @@ class Session if (!cur) { return; } const isReceiver = this.role === "receiver"; - const bytes = isReceiver ? this.receivedBytes : this.bytesSent; + const bytes = isReceiver ? this.receivedBytes : this.senderDeliveredBytes(); const lastProgressAt = bytes > (cur.bytesTransferred ?? 0) ? Date.now() : cur.lastProgressAt; @@ -237,11 +240,44 @@ class Session private emitProgress(): void { const total = this.role === "sender" ? this.outgoingTotal : (this.receivedMeta?.size ?? 0); - const bytes = this.role === "sender" ? this.bytesSent : this.receivedBytes; + const bytes = this.role === "sender" ? this.senderDeliveredBytes() : this.receivedBytes; for (const cb of this.progressListeners) { cb({ bytes, total }); } this.throttledPushUpdate(); } + /** + * 发送端"已交付字节"=已离开本地 buffer 的量(bytesSent − bufferedAmount)。 + * bytesSent 只是"已塞进 dc 的本地 16MiB buffer",会被 64KB 一片秒满到 16MiB 后 + * 定格;真正送达网络的是离开 buffer 的部分。用它做进度,进度随接收端实际收程 + * 平滑爬升,而非跳到 buffer 满就冻住,也避免 backpressure 等待期被误判 stalled。 + * 单调非降:send 时 bytesSent 与 bufferedAmount 同增(delivered 持平),其余时刻 + * buffer 抽干使 bufferedAmount 降(delivered 升)。 + */ + private senderDeliveredBytes(): number + { + const buffered = this.dc?.bufferedAmount ?? 0; + return Math.max(0, this.bytesSent - buffered); + } + + private startSendProgressPoll(): void + { + if (this.sendProgressPollId !== null) { return; } + this.sendProgressPollId = window.setInterval(() => + { + if (this.canceled) { this.stopSendProgressPoll(); return; } + this.emitProgress(); + }, 250); + } + + private stopSendProgressPoll(): void + { + if (this.sendProgressPollId !== null) + { + window.clearInterval(this.sendProgressPollId); + this.sendProgressPollId = null; + } + } + private classifyAndCount(c: RTCIceCandidate | RTCIceCandidateInit, into: CandidateBreakdown): void { const sdp = (c as RTCIceCandidate).candidate ?? (c as RTCIceCandidateInit).candidate ?? ""; @@ -297,8 +333,9 @@ class Session if (msg.type === "meta") { this.receivedMeta = { name: msg.name, size: msg.size, sha256: msg.sha256 }; - // 提前打开 sink;后续 chunk 入队等 sink 就绪 - this.receiveQueue = openIncomingSink(this.sessionId); + // 提前打开 sink;后续 chunk 入队等 sink 就绪。带上文件名,桌面大文件 + // spill 落盘后据此命名。 + this.receiveQueue = openIncomingSink(this.sessionId, msg.name); this.emitProgress(); } else if (msg.type === "done") @@ -364,8 +401,8 @@ class Session { const sink = await this.receiveQueue; if (!sink) { throw new Error("receive sink missing"); } - const blob = await sink.close(); - downloadBlob(blob, this.receivedMeta.name, this.sessionId); + const result = await sink.close(); + deliverIncoming(result, this.receivedMeta.name, this.sessionId); this.setState("completed"); void this.markServerDone(this.receivedBytes); } @@ -433,6 +470,7 @@ class Session { // eslint-disable-next-line no-console console.error("p2p send failed", e); + this.stopSendProgressPoll(); this.setState("failed"); }); }; @@ -455,6 +493,11 @@ class Session const meta: FileMeta = { name: file.name, size: file.size }; dc.send(JSON.stringify({ type: "meta", ...meta })); + // buffer 会被 64KB 一片秒满到 16MiB,之后发送循环长时间阻塞在 waitForBufferLow + // 等抽干,期间没有 send、也没有 emitProgress。这个轮询在等待期持续刷新进度,让 + // "已交付字节"随接收端实际收程平滑爬升、lastProgressAt 保持新鲜,不误报"疑似卡住"。 + this.startSendProgressPoll(); + let offset = 0; while (offset < file.size) { @@ -501,8 +544,17 @@ class Session } dc.send(JSON.stringify({ type: "done" })); - this.setState("completed"); this.setPhase("completing"); + // 抽干整个 buffer(数据 + done 帧)再宣告完成:dc.send 只是入本地 buffer,真正 + // 送达要等 SCTP 发出。早先送完入 buffer 即 setState("completed"),会在还有多达 + // 16MiB 在途时就标记完成,与接收端"仍在下载"不一致。等 bufferedAmount 落到 0= + // 字节已全部交付网络,接收端随后 finalize 并 POST /done(那条权威 /done 仍由接收 + // 端发,这里只在抽干后驱动发送端自身 UI 收尾)。 + await waitForBufferLow(dc, 0); + this.stopSendProgressPoll(); + this.bytesSent = file.size; + this.emitProgress(); + this.setState("completed"); // /done is intentionally posted by the *receiver* only — they are the // authoritative "got it" signal. Doubling it up here just produces // 409 console noise once the receiver's call lands first. @@ -657,6 +709,7 @@ class Session window.clearInterval(this.selectedPairPollId); this.selectedPairPollId = null; } + this.stopSendProgressPoll(); this.dc?.close(); this.pc.close(); this.setState("closed"); diff --git a/web/src/features/transfer/relay.ts b/web/src/features/transfer/relay.ts index bad887b..1aa8f4e 100644 --- a/web/src/features/transfer/relay.ts +++ b/web/src/features/transfer/relay.ts @@ -1,5 +1,5 @@ import { apiFetch } from "../../net/api"; -import { downloadBlob, openIncomingSink } from "./incomingSink"; +import { deliverIncoming, openIncomingSink } from "./incomingSink"; // Relay path (PROJECT_BRIEF.md §M5/M9). Sender uploads sequential 1 MiB chunks // via POST and receiver streams the response via GET. Streaming POST is NOT @@ -169,7 +169,7 @@ export async function relayReceiveFile( expectedSize: number, onProgress?: (p: RelayProgress) => void, abortSignal?: AbortSignal, -): Promise +): Promise { const url = `/api/relay/${encodeURIComponent(sessionId)}/stream`; const r = await apiFetch(url, { method: "GET", signal: abortSignal }); @@ -181,7 +181,7 @@ export async function relayReceiveFile( if (!r.body) { throw new Error("relay: no response body"); } const reader = r.body.getReader(); - const sink = await openIncomingSink(sessionId); + const sink = await openIncomingSink(sessionId, fileName); let received = 0; try { @@ -207,7 +207,6 @@ export async function relayReceiveFile( try { reader.cancel(); } catch { /* noop */ } } - const blob = await sink.close(); - downloadBlob(blob, fileName, sessionId); - return blob; + const result = await sink.close(); + deliverIncoming(result, fileName, sessionId); } diff --git a/web/src/net/desktop.ts b/web/src/net/desktop.ts index 99303fe..7e4aa6f 100644 --- a/web/src/net/desktop.ts +++ b/web/src/net/desktop.ts @@ -41,6 +41,12 @@ interface DesktopBridge // data 为 base64:Wails 把 JS 字符串作为 JSON 字符串编码,Go 端 []byte 参数按 // base64 解码。返回实际写入的绝对路径。 SaveDownload: (name: string, data: string) => Promise; + // 流式落盘(接收端混合 sink 越过内存上限后走这条):Begin 开临时文件、Append + // 追加批次(base64)、Finalize 改名落定返回绝对路径、Abort 取消时丢弃临时文件。 + BeginDownload: (sessionId: string) => Promise; + AppendDownload: (sessionId: string, data: string) => Promise; + FinalizeDownload: (sessionId: string, name: string) => Promise; + AbortDownload: (sessionId: string) => Promise; ChooseDownloadDir: (title: string) => Promise; EffectiveDownloadDir: () => Promise; // 弹原生系统通知(macOS UNUserNotificationCenter / Windows toast)。标题、正文 @@ -275,6 +281,38 @@ export async function saveIncomingFileDesktop(name: string, blob: Blob): Promise return app.SaveDownload(name, b64); } +// 流式落盘封装:接收端混合 sink 越过内存上限后,把溢出批次交给 Go 写入临时文件, +// finalize 时改名落定。begin / finalize 失败抛错(由 sink 走失败路径);append 在 +// 批次边界调用,base64 过桥同 SaveDownload;abort 吞错(取消路径不应再抛)。 +export async function beginIncomingDownloadDesktop(sessionId: string): Promise +{ + const app = bridge(); + if (!app) { throw new Error("desktop bridge unavailable"); } + await app.BeginDownload(sessionId); +} + +export async function appendIncomingDownloadDesktop(sessionId: string, bytes: Uint8Array): Promise +{ + const app = bridge(); + if (!app) { throw new Error("desktop bridge unavailable"); } + await app.AppendDownload(sessionId, bytesToBase64(bytes)); +} + +export async function finalizeIncomingDownloadDesktop(sessionId: string, name: string): Promise +{ + const app = bridge(); + if (!app) { throw new Error("desktop bridge unavailable"); } + return app.FinalizeDownload(sessionId, name); +} + +export async function abortIncomingDownloadDesktop(sessionId: string): Promise +{ + const app = bridge(); + if (!app) { return; } + try { await app.AbortDownload(sessionId); } + catch { /* 取消 / 失败路径,吞掉 */ } +} + // chooseDownloadDir 打开原生目录选择框(标题本地化由调用方传入),返回所选目录; // 用户取消返回空串。浏览器里 bridge() 为 null,返回空串。 export async function chooseDownloadDir(title: string): Promise diff --git a/web/src/routes/index.tsx b/web/src/routes/index.tsx index 05e5375..99872a6 100644 --- a/web/src/routes/index.tsx +++ b/web/src/routes/index.tsx @@ -12,6 +12,7 @@ import { startOutgoingTransfer } from "../features/transfer/transfer"; import { sendMessage } from "../features/messaging/messaging"; import { fetchClipboard, CLIPBOARD_MAX_BYTES } from "../features/clipboard/clipboard"; import { t } from "../i18n"; +import { DynText } from "../ui/primitives"; import { toast } from "../ui/feedback"; import s from "./index.module.css"; @@ -89,7 +90,7 @@ function HomePage()

- {t("home.greeting", { name: user?.name ?? "" })} + {t("home.greeting", { name: user?.name ?? "" })}

- {onlineLabel} · {kindLabel} · {lastUsedText} + {onlineLabel} · {kindLabel} · {lastUsedText} {onRevoke && ( diff --git a/web/src/store/slices/transfer.ts b/web/src/store/slices/transfer.ts index 3e250ff..4053e99 100644 --- a/web/src/store/slices/transfer.ts +++ b/web/src/store/slices/transfer.ts @@ -1,5 +1,5 @@ import type { StateCreator } from "zustand"; -import type { AppState } from "../types"; +import type { AppState, TransferRecord } from "../types"; import { normalizeTerminal } from "../helpers"; export type TransferSlice = Pick< @@ -7,6 +7,26 @@ export type TransferSlice = Pick< "activeTransfers" | "history" | "upsertTransfer" | "completeTransfer" | "removeTransfer" >; +// EMA 平滑系数:瞬时速率抖动大(10Hz 采样 + chunk 边界),取 0.3 让显示稳又不过钝。 +const RATE_EMA_ALPHA = 0.3; + +// withRate 在记录落库前算出瞬时传输速度。bytesTransferred 与 lastProgressAt 配对—— +// 后者只在前者增长时刷新为当时墙钟(见 p2p pushUpdate),故相邻两次「有增长」样本的 +// 差分即瞬时速率,再做 EMA 平滑。无增长(仅 iceStats / phase 等更新)则原样保留上次 +// 速率(next 由 {...cur} 展开本就带着它),不污染。 +function withRate(prev: TransferRecord | undefined, next: TransferRecord): TransferRecord +{ + if (!prev) { return next; } + const prevBytes = prev.bytesTransferred ?? 0; + const nextBytes = next.bytesTransferred ?? 0; + const prevAt = prev.lastProgressAt ?? 0; + const nextAt = next.lastProgressAt ?? 0; + if (nextBytes <= prevBytes || nextAt <= prevAt) { return next; } + const instant = (nextBytes - prevBytes) / ((nextAt - prevAt) / 1000); + const prevRate = prev.bytesPerSec ?? instant; + return { ...next, bytesPerSec: RATE_EMA_ALPHA * instant + (1 - RATE_EMA_ALPHA) * prevRate }; +} + export const createTransferSlice: StateCreator = (set) => ({ activeTransfers: {}, @@ -16,7 +36,7 @@ export const createTransferSlice: StateCreator { set((s) => ({ - activeTransfers: { ...s.activeTransfers, [t.sessionId]: t }, + activeTransfers: { ...s.activeTransfers, [t.sessionId]: withRate(s.activeTransfers[t.sessionId], t) }, })); }, diff --git a/web/src/store/types.ts b/web/src/store/types.ts index 0b817fb..bb88642 100644 --- a/web/src/store/types.ts +++ b/web/src/store/types.ts @@ -92,6 +92,9 @@ export interface TransferRecord bytesTransferred?: number; /** 上一次 bytesTransferred 增长的时刻(ms epoch),UI 据此判 stalled */ lastProgressAt?: number; + /** 瞬时传输速度(字节/秒,EMA 平滑)。由 upsertTransfer 据相邻样本差分算得, + * UI 仅在字节流动且未卡死时展示,避免显示误导性的 0 / 陈旧速率 */ + bytesPerSec?: number; iceStats?: IceStats; } diff --git a/web/src/ui/primitives/DynText.tsx b/web/src/ui/primitives/DynText.tsx new file mode 100644 index 0000000..8d48031 --- /dev/null +++ b/web/src/ui/primitives/DynText.tsx @@ -0,0 +1,19 @@ +// DynText —— 动态 CJK 文本的排版安全包装。 +// +// 聚珍(Juzhen)shim 会把含 CJK 的文本节点 charify 拆成多个 DOM 节点来做中西间距 / +// 标点对齐。当这段文本**就地更新**(React 改写同一个文本节点、元素本身仍挂载)时, +// shim 先前拆出的旧碎片成了 React 不再追踪的兄弟节点残留下来,每更新一次就累积一个 +// 多余字——典型如「发送到 X」切换设备后前面多出来的「到」,或每秒 tick 的相对时间。 +// +// 解法:用 key=文本内容。内容一变,React 就把旧 span(连同里面的 shim 碎片)整体卸载、 +// 挂载一个全新的 span,shim 再对新节点处理一次。既消除累积,又**保留**中西自动间距 +// (不像 data-jz-skip 那样把整段排版关掉)。代价是内容变化时该 span 会重挂,但只是 +// 一截短文本、成本可忽略。 +// +// 仅用于会就地变化的动态 CJK chrome(状态文案 / 计数 / 相对时间 / 速度 / 设备名等); +// 静态一次性文本不会累积,无需包。children 限定为 string —— key 必须是稳定可比较的 +// 原始值,且「内容变=key 变」正是触发重挂的依据。 +export function DynText(props: { children: string }) +{ + return {props.children}; +} diff --git a/web/src/ui/primitives/index.ts b/web/src/ui/primitives/index.ts index 54cddc7..2d72bc1 100644 --- a/web/src/ui/primitives/index.ts +++ b/web/src/ui/primitives/index.ts @@ -21,3 +21,5 @@ export type { TooltipProps } from "./Tooltip"; export { Callout } from "./Callout"; export type { CalloutProps, CalloutTone } from "./Callout"; + +export { DynText } from "./DynText";