传输修复:P2P 接收混合有界内存 sink + 发送进度/完成语义修正 + 前端 CJK 文字累积修复 + 传输速度显示
- P2P 接收端(桌面)改用混合有界内存 sink:wails:// 自定义 scheme 无持久 WebView 存储致 OPFS 降级、createWritable 回压不畅,把接收链堵死、饿死 dc.onmessage → 接收方 rwnd=0 → 发送方 bufferedAmount 卡死 16MiB(PWA→Mac 约 15MB 死锁)。小文件(暂存 < 256MB)纯内存、 close 整文件过 Go 写盘;超上限大文件流式 spill 到 Go 临时文件(每 64MB 一批),内存恒定 有界。浏览器 / iOS 接收仍走 OPFS - 新增桌面流式落盘:download.go 的 Begin/Append/Finalize/AbortStreamingDownload(复用 SaveDownload 的目录解析 / 文件名 sanitize / 不可写回退 / reveal,临时文件同目录便于原子 rename),app.go 加 4 个绑定 + net/desktop.ts 封装;含单测 download_stream_test.go; revealInFileManager 加 CDROP_NO_REVEAL 守卫(headless / 测试不弹 Finder) - IncomingSink.close() 返回 SinkResult(blob | 已落盘 path),新增 deliverIncoming 统一 落地;relay 接收一并改用,桌面大文件经中继也有界落盘 - P2P 发送端进度改按“已交付字节”(bytesSent − bufferedAmount)计,而非已塞进本地 16MiB buffer 的量——后者在 buffer 秒满后定格、误报“疑似卡住”;并加 250ms 轮询在 backpressure 等待期持续刷新进度 - P2P 发送端送完后等 bufferedAmount 抽干到 0 再宣告 completed(原先入 buffer 即标完成, 与接收端仍在下载的状态不一致) - 前端动态 CJK 文本累积修复:聚珍 shim 把就地更新的 CJK 文本节点 charify 拆段后,旧碎片 残留累积(典型“发送到 X”切设备多出“到”)。新增 DynText(key=文本内容触发整体重挂、 保留中西排版),应用于发送按钮 / 设备名 / 在线数 / 相对时间 / 会话与令牌活跃时间 / 消息 计数 / 剪贴板来源 / 问候语等就地更新处 - 传输卡片新增实时速度:store upsertTransfer 据相邻样本差分 + EMA 平滑算 bytesPerSec, TransferRow 在字节流动且未卡死时以“X/秒”展示(点分隔 meta 行追加)
This commit is contained in:
@@ -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");
|
||||
|
||||
Reference in New Issue
Block a user