perf(core): parallelize highway chunk uploads (#211)
Group/private file uploads (and all highway media) topped out around 500 KB/s regardless of link speed. Two costs stacked up in uploadHighwayHttp: chunks were sent strictly one at a time (throughput capped at one 1 MiB block per round-trip), and each block opened a fresh connection that paid TCP slow-start from a cold congestion window — so a 1 MiB block finished before the window ramped and never reached line rate. Drive the (already offset-addressed, self-contained) blocks through a small worker pool instead of a sequential loop. Highway frames carry their own dataOffset/chunkMd5/fileMd5, so blocks are independent and the server reassembles by offset; a handful of connections overlap the round-trip stalls and ramp their windows together. Concurrency is kept modest (4) to stay friendly to the edge nodes' rate limiting, and the one-connection-per-block invariant from #118 is preserved — parallelism just puts several of those short-lived connections in flight at once. Correctness preserved and covered by new tests: - offset claim is synchronous (no await between read and advance of nextOffset), so every byte range is uploaded exactly once — no skip, no duplicate, ragged final block correct; - concurrent ChunkSource reads are race-free (positional FileHandle reads into per-call buffers; read-only Buffer subarrays); - first permanent block failure aborts the whole upload and rejects; peers stop claiming new blocks and drain the one in flight; - workers never reject, so Promise.all settles all in-flight reads/sockets before the source is closed exactly once — no close-during-read, no masked upload error; - 0-byte source still opens zero connections; peak memory bounded to concurrency x ~1 MiB, never the whole file.
This commit is contained in:
DrSmoothl
@@ -88,7 +88,7 @@ vi.mock('net', () => ({
|
||||
createConnection: createConnectionMock,
|
||||
}));
|
||||
|
||||
import { uploadHighwayHttp, BufferChunkSource, type HighwaySession } from '@snowluma/protocol/highway';
|
||||
import { uploadHighwayHttp, BufferChunkSource, type ChunkSource, type HighwaySession } from '@snowluma/protocol/highway';
|
||||
import type { BridgeContext } from '@snowluma/protocol/bridge-context';
|
||||
|
||||
const HIGHWAY_BLOCK_SIZE = 1024 * 1024;
|
||||
@@ -166,3 +166,140 @@ describe('uploadHighwayHttp connection lifecycle', () => {
|
||||
expect(createConnectionMock).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
// 并发上传（#211）的正确性回归。重点覆盖用户要求的四个维度：
|
||||
// 数据安全 —— 每个 offset 恰好读一次、大小正确、覆盖整段 [0, size)；
|
||||
// 原子性 —— 首个永久失败即中止，不再领新 chunk，整体 reject；
|
||||
// 回退 —— 失败/成功都只 close() 一次，且绝不在有 read 在飞时 close；
|
||||
// 一致性 —— 并发确实发生（同时在飞的连接 > 1）但不超过 chunk 数。
|
||||
describe('uploadHighwayHttp concurrent upload safety (#211)', () => {
|
||||
// 记录每次 read 的 offset/length，跟踪在飞 read 数与 close 时机。read 故意
|
||||
// 异步（await 一个 microtask），这样 close-while-reading 一旦发生就能被捕获。
|
||||
class RecordingChunkSource implements ChunkSource {
|
||||
reads: Array<{ offset: number; length: number }> = [];
|
||||
inFlight = 0;
|
||||
maxInFlight = 0;
|
||||
closeCount = 0;
|
||||
closedWhileReading = false;
|
||||
constructor(readonly size: number) {}
|
||||
async read(offset: number, length: number): Promise<Uint8Array> {
|
||||
this.inFlight += 1;
|
||||
this.maxInFlight = Math.max(this.maxInFlight, this.inFlight);
|
||||
this.reads.push({ offset, length });
|
||||
await Promise.resolve();
|
||||
this.inFlight -= 1;
|
||||
return new Uint8Array(length);
|
||||
}
|
||||
async close(): Promise<void> {
|
||||
if (this.inFlight > 0) this.closedWhileReading = true;
|
||||
this.closeCount += 1;
|
||||
}
|
||||
}
|
||||
|
||||
// errorCode 帧：RespDataHighwayHead.field3 (errorCode) = 921。
|
||||
// 定义即拒绝（不重试），用来制造“瞬时永久失败”而不吃重试 sleep。
|
||||
// varint(921) = 0x99 0x07，tag(field3,varint) = 0x18。
|
||||
function buildErrorCodeFrame(): Buffer {
|
||||
const head = Buffer.from([0x18, 0x99, 0x07]);
|
||||
const frame = Buffer.alloc(9 + head.length + 1);
|
||||
frame[0] = 0x28;
|
||||
frame.writeUInt32BE(head.length, 1);
|
||||
frame.writeUInt32BE(0, 5);
|
||||
head.copy(frame, 9);
|
||||
frame[9 + head.length] = 0x29;
|
||||
return frame;
|
||||
}
|
||||
|
||||
// 跟踪同时存活（已连接、未 destroy）的 socket 峰值，用来断言并发确实发生。
|
||||
let liveSockets = 0;
|
||||
let maxLiveSockets = 0;
|
||||
|
||||
function installMock(responseFrame: Buffer): void {
|
||||
createConnectionMock.mockReset();
|
||||
liveSockets = 0;
|
||||
maxLiveSockets = 0;
|
||||
createConnectionMock.mockImplementation((_opts: unknown, listener?: () => void) => {
|
||||
const sock = new FakeSocket(buildHttpResponse(responseFrame));
|
||||
liveSockets += 1;
|
||||
maxLiveSockets = Math.max(maxLiveSockets, liveSockets);
|
||||
const origDestroy = sock.destroy.bind(sock);
|
||||
sock.destroy = () => { if (!sock.destroyed) liveSockets -= 1; origDestroy(); };
|
||||
createdSockets.push(sock);
|
||||
if (listener) setImmediate(listener);
|
||||
return sock as unknown as ReturnType<typeof createConnectionMock>;
|
||||
});
|
||||
}
|
||||
|
||||
beforeEach(() => { createdSockets.length = 0; });
|
||||
afterEach(() => { vi.restoreAllMocks(); });
|
||||
|
||||
it('covers every offset exactly once and closes the source exactly once (data safety + consistency)', async () => {
|
||||
installMock(buildEmptyHighwayResponseFrame());
|
||||
const chunks = 8;
|
||||
const size = chunks * HIGHWAY_BLOCK_SIZE;
|
||||
const source = new RecordingChunkSource(size);
|
||||
|
||||
await uploadHighwayHttp(
|
||||
makeBridge(), makeSession(), 1004, source, new Uint8Array(16), new Uint8Array(0),
|
||||
);
|
||||
|
||||
// Exactly-once, correctly-sized, gap-free coverage of [0, size).
|
||||
const offsets = source.reads.map((r) => r.offset).sort((a, b) => a - b);
|
||||
expect(offsets).toEqual(Array.from({ length: chunks }, (_, i) => i * HIGHWAY_BLOCK_SIZE));
|
||||
expect(new Set(offsets).size).toBe(chunks); // no duplicate offset
|
||||
expect(source.reads.every((r) => r.length === HIGHWAY_BLOCK_SIZE)).toBe(true);
|
||||
// One connection per chunk; source closed once, never mid-read.
|
||||
expect(createConnectionMock).toHaveBeenCalledTimes(chunks);
|
||||
expect(source.closeCount).toBe(1);
|
||||
expect(source.closedWhileReading).toBe(false);
|
||||
});
|
||||
|
||||
it('actually uploads chunks in parallel but never exceeds the chunk count (concurrency)', async () => {
|
||||
installMock(buildEmptyHighwayResponseFrame());
|
||||
const chunks = 8;
|
||||
const source = new RecordingChunkSource(chunks * HIGHWAY_BLOCK_SIZE);
|
||||
|
||||
await uploadHighwayHttp(
|
||||
makeBridge(), makeSession(), 1004, source, new Uint8Array(16), new Uint8Array(0),
|
||||
);
|
||||
|
||||
// Parallelism happened (more than one connection in flight at once) …
|
||||
expect(maxLiveSockets).toBeGreaterThan(1);
|
||||
// … but the pool is bounded and never opens more than there are chunks.
|
||||
expect(maxLiveSockets).toBeLessThanOrEqual(chunks);
|
||||
});
|
||||
|
||||
it('the last (ragged) chunk carries the remainder length', async () => {
|
||||
installMock(buildEmptyHighwayResponseFrame());
|
||||
const size = 2 * HIGHWAY_BLOCK_SIZE + 12345; // 2 full + 1 partial
|
||||
const source = new RecordingChunkSource(size);
|
||||
|
||||
await uploadHighwayHttp(
|
||||
makeBridge(), makeSession(), 1004, source, new Uint8Array(16), new Uint8Array(0),
|
||||
);
|
||||
|
||||
const byOffset = new Map(source.reads.map((r) => [r.offset, r.length]));
|
||||
expect(byOffset.get(0)).toBe(HIGHWAY_BLOCK_SIZE);
|
||||
expect(byOffset.get(HIGHWAY_BLOCK_SIZE)).toBe(HIGHWAY_BLOCK_SIZE);
|
||||
expect(byOffset.get(2 * HIGHWAY_BLOCK_SIZE)).toBe(12345);
|
||||
const total = source.reads.reduce((n, r) => n + r.length, 0);
|
||||
expect(total).toBe(size); // every byte accounted for exactly once
|
||||
});
|
||||
|
||||
it('aborts the whole upload on a definitive server reject, closing the source once and never mid-read (atomicity + rollback)', async () => {
|
||||
// Every chunk gets error_code=921 → definitive reject, no retries. The first
|
||||
// failure must abort; the source must still be closed exactly once and never
|
||||
// while a read is in flight.
|
||||
installMock(buildErrorCodeFrame());
|
||||
const source = new RecordingChunkSource(8 * HIGHWAY_BLOCK_SIZE);
|
||||
|
||||
await expect(uploadHighwayHttp(
|
||||
makeBridge(), makeSession(), 1004, source, new Uint8Array(16), new Uint8Array(0),
|
||||
)).rejects.toThrow(/error_code=921/);
|
||||
|
||||
expect(source.closeCount).toBe(1);
|
||||
expect(source.closedWhileReading).toBe(false);
|
||||
// Atomicity: once aborted we stop claiming new chunks, so not all 8 were read.
|
||||
expect(source.reads.length).toBeLessThan(8);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -28,6 +28,18 @@ const HIGHWAY_BLOCK_SIZE = 1024 * 1024;
|
||||
const HIGHWAY_MAX_CHUNK_ATTEMPTS = 3;
|
||||
const HIGHWAY_RETRY_BASE_MS = 300;
|
||||
|
||||
// Concurrent chunk upload fan-out. A large file's chunks are independent —
|
||||
// each highway frame is self-contained (its own dataOffset, chunkMd5, fileMd5
|
||||
// and serviceTicket) and the server reassembles by offset, so chunks can be
|
||||
// PUT in any order over parallel connections. Uploading them one-at-a-time
|
||||
// (the old behavior) capped throughput at ~1 chunk / RTT AND paid TCP
|
||||
// slow-start on every fresh per-chunk connection, so a 1 MiB chunk finished
|
||||
// before its cold congestion window opened — the upload never reached line
|
||||
// rate (issue #211). A small pool overlaps the RTT stalls and lets several
|
||||
// connections ramp their windows at once. Kept modest so we stay a good
|
||||
// citizen on QQ's highway edge nodes (which rate-limit — see error_code=921).
|
||||
const HIGHWAY_UPLOAD_CONCURRENCY = 4;
|
||||
|
||||
const sleepMs = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
export const PRIVATE_IMAGE_CMD_ID = 1003;
|
||||
@@ -306,98 +318,138 @@ export async function uploadHighwayHttp(
|
||||
const pathStr = `/cgi-bin/httpconn?htcmd=0x6FF0087&uin=${bridge.identity.uin}`;
|
||||
const totalSize = source.size;
|
||||
|
||||
// 每个 chunk 一个独立的 TCP 连接。
|
||||
// Upload exactly one chunk. Extracted verbatim from the old sequential loop
|
||||
// body — the per-chunk wire behavior is unchanged, only the *scheduling*
|
||||
// across chunks becomes concurrent (see HIGHWAY_UPLOAD_CONCURRENCY).
|
||||
//
|
||||
// 每个 chunk 仍然一个独立的 TCP 连接。
|
||||
//
|
||||
// 旧实现整个文件复用同一个 socket，依赖 `Connection: keep-alive` 跑多次
|
||||
// POST。但 QQ highway 边缘节点（以及链路上的代理/NAT）经常在第一次响应
|
||||
// 之后立刻 FIN 关闭连接，导致第二个 chunk 写入一个已被对端关闭的 socket，
|
||||
// `readHttpResponseBody` 立刻收到 `close` 事件并抛 `connection closed
|
||||
// before response`。
|
||||
// before response`（#118）。所以每个 chunk 独立建连的不变量必须保留 ——
|
||||
// 并发只是让多个这样的一次性连接同时在飞，连接之间没有任何共享状态。
|
||||
//
|
||||
// 现象上的体现：≤1 MB 的图片/PTT 因为只发一个 chunk，不会触发；超过 1 MB
|
||||
// 的图片或视频（HIGHWAY_BLOCK_SIZE = 1 MiB）必然失败。
|
||||
// Data safety: `source.read` returns a fresh buffer per call (FileChunkSource
|
||||
// allocates; BufferChunkSource returns a read-only subarray), and reads are
|
||||
// positional (offset-addressed), so concurrent reads at distinct offsets are
|
||||
// race-free. Each chunk computes its own md5 over its own bytes.
|
||||
const uploadChunk = async (offset: number, chunkSize: number): Promise<void> => {
|
||||
const chunk = await source.read(offset, chunkSize);
|
||||
const chunkMd5 = computeMd5(chunk);
|
||||
const head = makeHighwayHead(
|
||||
bridge.identity.uin, commandId, totalSize, offset, chunkSize,
|
||||
chunkMd5, fileMd5, session.sigSession, extend,
|
||||
);
|
||||
const frame = packHighwayFrame(head, chunk);
|
||||
|
||||
// Retry the connect + POST for this chunk on transient transport errors
|
||||
// (peer FIN before response, ECONNRESET, connect refused/timeout). Each
|
||||
// attempt uses a fresh connection; re-sending the same offset range is
|
||||
// idempotent on the server.
|
||||
let responseBody: Uint8Array | undefined;
|
||||
for (let attempt = 1; ; attempt++) {
|
||||
let socket: net.Socket;
|
||||
try {
|
||||
socket = await tcpConnect(session.host, session.port);
|
||||
} catch (err) {
|
||||
if (attempt >= HIGHWAY_MAX_CHUNK_ATTEMPTS) {
|
||||
throw new Error(
|
||||
`highway connect failed after ${attempt} attempts ` +
|
||||
`(cmdId=${commandId} offset=${offset}/${totalSize}): ${String(err)}`,
|
||||
);
|
||||
}
|
||||
await sleepMs(HIGHWAY_RETRY_BASE_MS * attempt);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
responseBody = await httpPostFrame(socket, session.host, pathStr, frame);
|
||||
break;
|
||||
} catch (err) {
|
||||
if (attempt >= HIGHWAY_MAX_CHUNK_ATTEMPTS) {
|
||||
throw new Error(
|
||||
`highway upload transport failed after ${attempt} attempts ` +
|
||||
`(cmdId=${commandId} chunk=${chunkSize}/${totalSize} offset=${offset}): ${String(err)}`,
|
||||
);
|
||||
}
|
||||
log.trace('chunk offset=%d attempt %d failed (%s), retrying', offset, attempt, String(err));
|
||||
await sleepMs(HIGHWAY_RETRY_BASE_MS * attempt);
|
||||
} finally {
|
||||
socket.destroy();
|
||||
}
|
||||
}
|
||||
|
||||
// Unreachable: the retry loop only exits via break (responseBody set) or
|
||||
// throw — the guard just narrows the type for the compiler.
|
||||
if (!responseBody) throw new Error('highway upload: missing response');
|
||||
const { head: respHead } = unpackHighwayFrame(responseBody);
|
||||
const resp = protobuf_decode<RespDataHighwayHead>(respHead);
|
||||
if (resp?.errorCode && resp.errorCode !== 0) {
|
||||
// Surface every diagnostic the highway response carries so
|
||||
// user reports of `error_code=921` and friends include the
|
||||
// server-side context (segHead.retCode, chunk size, file md5)
|
||||
// — without these we can't tell apart a malformed-payload
|
||||
// reject, a session-ticket mismatch, or a per-account rate-
|
||||
// limit.
|
||||
const segRetCode = resp.msgSegHead?.retCode ?? 0;
|
||||
const fileMd5Hex = Buffer.from(fileMd5).toString('hex');
|
||||
throw new Error(
|
||||
`highway upload error_code=${resp.errorCode}` +
|
||||
` (cmdId=${commandId} chunk=${chunkSize}/${totalSize}` +
|
||||
` offset=${offset} segRetCode=${segRetCode}` +
|
||||
` fileMd5=${fileMd5Hex.slice(0, 16)}…)`,
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
// Concurrent worker pool over the chunk range.
|
||||
//
|
||||
// 改成每 chunk 一连接代价可忽略：
|
||||
// - 单个 1 MB chunk 的传输时间远大于 TCP 建连开销；
|
||||
// - QQ 服务端反正也不希望客户端长时间占用连接；
|
||||
// - 与 NapCat / Lagrange 在大文件上传时的连接生命周期一致。
|
||||
// Atomicity / consistency: an offset is claimed synchronously — there is NO
|
||||
// await between reading `nextOffset` and advancing it — so two workers can
|
||||
// never claim the same chunk, and every byte range is PUT exactly once. On
|
||||
// the first permanent chunk failure a worker sets `aborted`; peers stop
|
||||
// pulling new chunks but drain the one already in flight. `worker()` NEVER
|
||||
// rejects (it captures its error), so `Promise.all` always resolves only
|
||||
// after every in-flight read/socket has settled — that is what makes it safe
|
||||
// to `source.close()` in the finally without racing a live read.
|
||||
//
|
||||
// Rollback: a thrown upload leaves the server holding an incomplete set of
|
||||
// chunks that it never finalizes — identical to the pre-concurrency failure
|
||||
// mode. The caller's publish/finalize step (which makes the file visible)
|
||||
// only runs when this resolves, so a partial upload is never published.
|
||||
let nextOffset = 0;
|
||||
let uploaded = 0;
|
||||
let aborted = false;
|
||||
let firstError: Error | null = null;
|
||||
|
||||
const worker = async (): Promise<void> => {
|
||||
while (!aborted) {
|
||||
const offset = nextOffset;
|
||||
if (offset >= totalSize) return;
|
||||
const chunkSize = Math.min(HIGHWAY_BLOCK_SIZE, totalSize - offset);
|
||||
nextOffset = offset + chunkSize; // claim — synchronous, race-free
|
||||
try {
|
||||
await uploadChunk(offset, chunkSize);
|
||||
} catch (err) {
|
||||
aborted = true;
|
||||
firstError ??= err instanceof Error ? err : new Error(String(err));
|
||||
return;
|
||||
}
|
||||
uploaded += chunkSize;
|
||||
log.trace('uploaded %d/%d bytes', uploaded, totalSize);
|
||||
}
|
||||
};
|
||||
|
||||
// `source` may buffer the whole file (BufferChunkSource) or stream it from
|
||||
// disk (FileChunkSource); either way one chunk is read/held at a time. We
|
||||
// own the source and close it exactly once in the `finally` below.
|
||||
// disk (FileChunkSource). We own the source and close it exactly once below,
|
||||
// only after all workers have settled.
|
||||
let succeeded = false;
|
||||
try {
|
||||
let offset = 0;
|
||||
while (offset < totalSize) {
|
||||
const chunkSize = Math.min(HIGHWAY_BLOCK_SIZE, totalSize - offset);
|
||||
const chunk = await source.read(offset, chunkSize);
|
||||
const chunkMd5 = computeMd5(chunk);
|
||||
const head = makeHighwayHead(
|
||||
bridge.identity.uin, commandId, totalSize, offset, chunkSize,
|
||||
chunkMd5, fileMd5, session.sigSession, extend,
|
||||
);
|
||||
const frame = packHighwayFrame(head, chunk);
|
||||
|
||||
// Retry the connect + POST for this chunk on transient transport errors
|
||||
// (peer FIN before response, ECONNRESET, connect refused/timeout). Each
|
||||
// attempt uses a fresh connection; re-sending the same offset range is
|
||||
// idempotent on the server.
|
||||
let responseBody: Uint8Array | undefined;
|
||||
for (let attempt = 1; ; attempt++) {
|
||||
let socket: net.Socket;
|
||||
try {
|
||||
socket = await tcpConnect(session.host, session.port);
|
||||
} catch (err) {
|
||||
if (attempt >= HIGHWAY_MAX_CHUNK_ATTEMPTS) {
|
||||
throw new Error(
|
||||
`highway connect failed after ${attempt} attempts ` +
|
||||
`(cmdId=${commandId} offset=${offset}/${totalSize}): ${String(err)}`,
|
||||
);
|
||||
}
|
||||
await sleepMs(HIGHWAY_RETRY_BASE_MS * attempt);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
responseBody = await httpPostFrame(socket, session.host, pathStr, frame);
|
||||
break;
|
||||
} catch (err) {
|
||||
if (attempt >= HIGHWAY_MAX_CHUNK_ATTEMPTS) {
|
||||
throw new Error(
|
||||
`highway upload transport failed after ${attempt} attempts ` +
|
||||
`(cmdId=${commandId} chunk=${chunkSize}/${totalSize} offset=${offset}): ${String(err)}`,
|
||||
);
|
||||
}
|
||||
log.trace('chunk offset=%d attempt %d failed (%s), retrying', offset, attempt, String(err));
|
||||
await sleepMs(HIGHWAY_RETRY_BASE_MS * attempt);
|
||||
} finally {
|
||||
socket.destroy();
|
||||
}
|
||||
}
|
||||
|
||||
// Unreachable: the retry loop only exits via break (responseBody set) or
|
||||
// throw — the guard just narrows the type for the compiler.
|
||||
if (!responseBody) throw new Error('highway upload: missing response');
|
||||
const { head: respHead } = unpackHighwayFrame(responseBody);
|
||||
const resp = protobuf_decode<RespDataHighwayHead>(respHead);
|
||||
if (resp?.errorCode && resp.errorCode !== 0) {
|
||||
// Surface every diagnostic the highway response carries so
|
||||
// user reports of `error_code=921` and friends include the
|
||||
// server-side context (segHead.retCode, chunk size, file md5)
|
||||
// — without these we can't tell apart a malformed-payload
|
||||
// reject, a session-ticket mismatch, or a per-account rate-
|
||||
// limit.
|
||||
const segRetCode = resp.msgSegHead?.retCode ?? 0;
|
||||
const fileMd5Hex = Buffer.from(fileMd5).toString('hex');
|
||||
throw new Error(
|
||||
`highway upload error_code=${resp.errorCode}` +
|
||||
` (cmdId=${commandId} chunk=${chunkSize}/${totalSize}` +
|
||||
` offset=${offset} segRetCode=${segRetCode}` +
|
||||
` fileMd5=${fileMd5Hex.slice(0, 16)}…)`,
|
||||
);
|
||||
}
|
||||
offset += chunkSize;
|
||||
log.trace('uploaded %d/%d bytes', offset, totalSize);
|
||||
}
|
||||
const chunkCount = Math.max(1, Math.ceil(totalSize / HIGHWAY_BLOCK_SIZE));
|
||||
const workerCount = Math.min(HIGHWAY_UPLOAD_CONCURRENCY, chunkCount);
|
||||
await Promise.all(Array.from({ length: workerCount }, () => worker()));
|
||||
if (firstError) throw firstError;
|
||||
succeeded = true;
|
||||
} finally {
|
||||
// Best-effort close — never mask a primary upload error. A close failure
|
||||
|
||||
Reference in New Issue
Block a user