fix(protocol): recover highway uploads after missing responses

Try another available upload endpoint when a transfer stops responding,
while preserving confirmed progress and the current upload session.
Report unconfirmed transfers clearly and cover recovery and publication
boundaries with regression tests.

Fixes #552

packages/core/tests/actions/group-file.test.ts

@@ -305,6 +305,51 @@ describe('apis/group-file', () => {
         expect(publishSpy).toHaveBeenCalledOnce();
     });
 
+    it.each([true, false])(
+        'publishes only after the upload is confirmed (confirmed=%s)',
+        async (confirmed) => {
+            const bridge = mockBridge();
+            bridge.sendRawPacket.mockResolvedValueOnce(
+                packResponse(
+                    protobuf_encode<OidbBase<OidbGroupFileResp>>({
+                        body: {
+                            upload: {
+                                fileId: 'fid-confirm',
+                                uploadIp: '127.0.0.1',
+                                uploadPort: 443,
+                                fileKey: new Uint8Array([1]),
+                                checkKey: new Uint8Array([2]),
+                            },
+                        },
+                    }),
+                ),
+            );
+            const api = new GroupFileApi(bridge as any);
+            const publishSpy = vi.spyOn(api, 'publish').mockResolvedValue();
+            const failure = new Error('highway upload transport failed; current chunk unconfirmed');
+            vi.mocked(highwayClient.uploadHighwayHttp).mockImplementationOnce(async () => {
+                expect(publishSpy).not.toHaveBeenCalled();
+
+                if (!confirmed) {
+                    throw failure;
+                }
+            });
+            const upload = api.upload(12345, 'base64://AQID', 'confirmed.txt');
+
+            if (confirmed) {
+                await expect(upload).resolves.toEqual({ fileId: 'fid-confirm' });
+                expect(publishSpy).toHaveBeenCalledExactlyOnceWith(12345, 'fid-confirm');
+            } else {
+                await expect(upload).rejects.toBe(failure);
+                expect(publishSpy).not.toHaveBeenCalled();
+            }
+
+            expect(highwayClient.uploadHighwayHttp).toHaveBeenCalledOnce();
+            expect(bridge.sendRawPacket).toHaveBeenCalledOnce();
+            expect(highwayClient.fetchHighwaySession).toHaveBeenCalledOnce();
+        },
+    );
+
     it('upload caches the upload metadata for later resend by file_id', async () => {
         const bridge = mockBridge();
         bridge.sendRawPacket.mockResolvedValueOnce(

packages/core/tests/highway/highway-client.test.ts

@@ -15,8 +15,9 @@
 //     一连接，但仍全部成功。
 
 import { getLogLevel, setLogLevel, subscribeLogs, type LogEntry } from '@snowluma/common/logger';
-import type { HttpConn0x6FF501Response } from '@snowluma/proto-defs/highway';
-import { protobuf_encode } from '@snowluma/proton';
+import type { HttpConn0x6FF501Response, ReqDataHighwayHead } from '@snowluma/proto-defs/highway';
+import { protobuf_decode, protobuf_encode } from '@snowluma/proton';
+import { unpackHighwayFrame } from '@snowluma/protocol/highway/utils';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 import { EventEmitter } from 'events';
 
@@ -79,6 +80,8 @@ class FakeSocket extends EventEmitter {
     writable = true;
     posts = 0;
     headers: string[] = [];
+    frames: Buffer[] = [];
+    closeBeforeResponse = false;
     connectFailure?: Error | 'timeout';
     private connectTimer?: ReturnType<typeof setTimeout>;
     private buf = Buffer.alloc(0);
@@ -141,6 +144,7 @@ class FakeSocket extends EventEmitter {
             }
 
             this.headers.push(s.slice(0, he));
+            this.frames.push(Buffer.from(this.buf.subarray(he + 4, total)));
             this.buf = this.buf.subarray(total);
             this.posts += 1;
             const thisPost = this.posts;
@@ -157,6 +161,14 @@ class FakeSocket extends EventEmitter {
                     return;
                 }
 
+                if (this.closeBeforeResponse) {
+                    this.writable = false;
+                    this.readableEnded = true;
+                    this.emit('end');
+                    this.emit('close');
+                    return;
+                }
+
                 if (thisPost === this.errorAfterPartialOnPost) {
                     const full = httpWrap(this.responseFrame);
                     this.emit('data', full.subarray(0, full.length - 3));
@@ -443,7 +455,7 @@ describe('fetchHighwaySession trace lifecycle', () => {
     });
 });
 
-describe('Highway initial connection selection', () => {
+describe('Highway connection selection', () => {
     afterEach(() => {
         vi.useRealTimers();
         vi.restoreAllMocks();
@@ -582,24 +594,160 @@ describe('Highway initial connection selection', () => {
         expect(source.closeCount).toBe(1);
     });
 
-    it('stays on the selected node after data was sent without an acknowledgement', async () => {
+    it.each([0, HIGHWAY_BLOCK_SIZE])(
+        'retries only the unconfirmed chunk at offset %i on the next node',
+        async (offset) => {
+            vi.useFakeTimers();
+            installFactory(() => {
+                const host = createConnectionMock.mock.lastCall?.[0].host;
+                const socket = new FakeSocket(
+                    buildEmptyHighwayResponseFrame(),
+                    'keepalive',
+                    host === '192.0.2.1' && offset > 0 ? 2 : -1,
+                );
+                socket.closeBeforeResponse = host === '192.0.2.1' && offset === 0;
+                return socket;
+            });
+            const session = sessionWithCandidates();
+            const original = structuredClone(session);
+            const source = new RecordingChunkSource(offset + 80);
+            const extend = new Uint8Array([0x12, 0x34]);
+            const upload = uploadHighwayHttp(
+                makeBridge(),
+                session,
+                71,
+                source,
+                new Uint8Array(16),
+                extend,
+            );
+            const completed = expect(upload).resolves.toBeUndefined();
+            await Promise.all([completed, vi.runAllTimersAsync()]);
+            expect(createConnectionMock.mock.calls.map(([options]) => options)).toEqual(
+                session.servers,
+            );
+            const [failed, recovered] = createdSockets as FakeSocket[];
+            expect(recovered!.posts).toBe(1);
+            const first = unpackHighwayFrame(failed!.frames.at(-1)!);
+            const retry = unpackHighwayFrame(recovered!.frames[0]!);
+            const firstHead = protobuf_decode<ReqDataHighwayHead>(first.head);
+            const retryHead = protobuf_decode<ReqDataHighwayHead>(retry.head);
+            expect(retry.body).toEqual(first.body);
+            expect(retryHead).toEqual({
+                ...firstHead,
+                msgBaseHead: { ...firstHead.msgBaseHead, retryTimes: 1 },
+            });
+            expect(retryHead.msgSegHead?.dataOffset ?? 0n).toBe(BigInt(offset));
+            expect(retryHead.bytesReqExtendInfo).toEqual(Buffer.from(extend));
+            expect(source.reads).toEqual(
+                offset === 0
+                    ? [{ offset: 0, length: 80 }]
+                    : [
+                          { offset: 0, length: HIGHWAY_BLOCK_SIZE },
+                          { offset, length: 80 },
+                      ],
+            );
+            expect(source.closeCount).toBe(1);
+            expect(session).toEqual(original);
+            expect((createdSockets as FakeSocket[]).every((socket) => socket.destroyed)).toBe(true);
+        },
+    );
+
+    it('bounds retries across nodes and reports the unconfirmed chunk without advancing progress', async () => {
         vi.useFakeTimers();
-        installFactory(
-            (idx) =>
-                new FakeSocket(buildEmptyHighwayResponseFrame(), 'keepalive', idx === 0 ? 1 : -1),
-        );
+        installFactory(() => {
+            const socket = new FakeSocket(buildEmptyHighwayResponseFrame(), 'keepalive');
+            socket.closeBeforeResponse = true;
+            return socket;
+        });
         const session = sessionWithCandidates();
+        const source = new RecordingChunkSource(80);
+        const entries: LogEntry[] = [];
+        setLogLevel('trace');
+        const unsubscribe = subscribeLogs((entry) => entries.push(entry));
+
+        try {
+            const upload = uploadHighwayHttp(
+                makeBridge(),
+                session,
+                71,
+                source,
+                new Uint8Array(16),
+                new Uint8Array(),
+            );
+            const failed = expect(upload).rejects.toThrow(
+                /transport failed after 3 attempts.*acknowledgedBytes=0.*unconfirmed/,
+            );
+            await Promise.all([failed, vi.runAllTimersAsync()]);
+            expect(createConnectionMock.mock.calls.map(([options]) => options)).toEqual([
+                session.servers[0],
+                session.servers[1],
+                session.servers[0],
+            ]);
+            expect(source.reads).toEqual([{ offset: 0, length: 80 }]);
+            expect(source.closeCount).toBe(1);
+            const terminals = highwayTrace(entries).filter((entry) =>
+                entry.message.startsWith('highway_put_terminal '),
+            );
+            expect(terminals).toHaveLength(1);
+            expect(terminals[0]!.message).toContain('outcome=failed reason=transport_exhausted');
+            expect(terminals[0]!.message).toContain('uploadedBytes=0');
+            expect((createdSockets as FakeSocket[]).every((socket) => socket.destroyed)).toBe(true);
+        } finally {
+            unsubscribe();
+        }
+    });
+
+    it('continues after the actual failed node and skips candidates that cannot reconnect', async () => {
+        vi.useFakeTimers();
+        installFactory(() => {
+            const host = createConnectionMock.mock.lastCall?.[0].host;
+            const socket = new FakeSocket(buildEmptyHighwayResponseFrame(), 'keepalive');
+
+            if (host === '192.0.2.1') {
+                socket.connectFailure = new Error('ECONNREFUSED');
+            } else if (host === '192.0.2.2') {
+                socket.closeBeforeResponse = true;
+            } else if (host === '192.0.2.3') {
+                socket.connectFailure = 'timeout';
+            }
+
+            return socket;
+        });
+        const session = {
+            ...makeSession(),
+            servers: Array.from({ length: 4 }, (_, i) => ({ host: `192.0.2.${i + 1}`, port: 443 })),
+        };
+        const source = new RecordingChunkSource(80);
         const upload = uploadHighwayHttp(
             makeBridge(),
             session,
-            1004,
-            new RecordingChunkSource(1),
+            71,
+            source,
+            new Uint8Array(16),
+            new Uint8Array(),
+        );
+        await Promise.all([expect(upload).resolves.toBeUndefined(), vi.runAllTimersAsync()]);
+        expect(createConnectionMock.mock.calls.map(([options]) => options)).toEqual(
+            session.servers,
+        );
+        expect((createdSockets as FakeSocket[]).map((socket) => socket.posts)).toEqual([
+            0, 1, 0, 1,
+        ]);
+        expect(source.reads).toEqual([{ offset: 0, length: 80 }]);
+        expect(source.closeCount).toBe(1);
+    });
+
+    it('keeps the successful node when the peer closes after acknowledgement', async () => {
+        installFactory(() => new FakeSocket(buildEmptyHighwayResponseFrame(), 'closeafter'));
+        const session = sessionWithCandidates();
+        await uploadHighwayHttp(
+            makeBridge(),
+            session,
+            71,
+            new RecordingChunkSource(HIGHWAY_BLOCK_SIZE + 80),
             new Uint8Array(16),
             new Uint8Array(),
         );
-        const completed = expect(upload).resolves.toBeUndefined();
-        await vi.runAllTimersAsync();
-        await completed;
         expect(createConnectionMock.mock.calls.map(([options]) => options)).toEqual([
             session.servers[0],
             session.servers[0],

packages/protocol/src/highway/highway-client.ts

@@ -336,11 +336,14 @@ function tcpConnect(host: string, port: number, timeoutMs = 10000): Promise<net.
 
 async function connectHighway(
     servers: HighwayServer[],
-): Promise<{ socket: net.Socket; server: HighwayServer }> {
+    startIndex = 0,
+): Promise<{ socket: net.Socket; server: HighwayServer; serverIndex: number }> {
     const deadline = Date.now() + HIGHWAY_CONNECT_BUDGET_MS;
     const errors: Error[] = [];
 
-    for (const server of servers) {
+    for (let i = 0; i < servers.length; i++) {
+        const serverIndex = (startIndex + i) % servers.length;
+        const server = servers[serverIndex]!;
         const remaining = deadline - Date.now();
 
         if (remaining <= 0) {
@@ -353,7 +356,7 @@ async function connectHighway(
                 server.port,
                 Math.min(HIGHWAY_CONNECT_TIMEOUT_MS, remaining),
             );
-            return { socket, server };
+            return { socket, server, serverIndex };
         } catch (cause) {
             const error = new Error(`${server.host}:${server.port}: ${highwayErrorText(cause)}`, {
                 cause,
@@ -698,6 +701,7 @@ async function uploadHighwayHttpOperation(
     let socket: net.Socket | null = null;
     let connectCount = 0;
     let server = session.servers[0]!;
+    let serverIndex = 0;
 
     const dropSocket = (): void => {
         if (socket) {
@@ -771,13 +775,10 @@ async function uploadHighwayHttpOperation(
 
                 try {
                     if (!socket) {
-                        // Select only before the first connection succeeds. Zero acknowledged
-                        // bytes does not mean no data was sent to an already connected node.
-                        if (connectCount === 0) {
-                            ({ socket, server } = await connectHighway(session.servers));
-                        } else {
-                            socket = await tcpConnect(server.host, server.port);
-                        }
+                        ({ socket, server, serverIndex } = await connectHighway(
+                            session.servers,
+                            serverIndex,
+                        ));
 
                         connectCount += 1;
                         trace.connectCount = connectCount;
@@ -824,17 +825,23 @@ async function uploadHighwayHttpOperation(
                         trace.failureReason = 'transport_exhausted';
                         const exhausted = new Error(
                             `highway upload transport failed after ${transportAttempts} attempts ` +
-                                `(cmdId=${commandId} chunk=${chunkSize}/${totalSize} offset=${offset}): ${String(err)}`,
+                                `(cmdId=${commandId} chunk=${chunkSize}/${totalSize} offset=${offset}` +
+                                ` acknowledgedBytes=${offset}; current chunk unconfirmed): ${String(err)}`,
+                            { cause: err },
                         );
                         trace.error = exhausted;
                         throw exhausted;
                     }
 
+                    // Keep the current session and chunk intact when changing nodes.
+                    // Missing acknowledgement does not prove the previous node received nothing.
+                    serverIndex = (serverIndex + 1) % session.servers.length;
                     log.trace(
-                        'highway_put_branch branch=chunk_retry offset=%d bytes=%d attempt=%d error=%j',
+                        'highway_put_branch branch=chunk_retry offset=%d bytes=%d attempt=%d nextServer=%j error=%j',
                         offset,
                         chunkSize,
                         transportAttempts,
+                        session.servers[serverIndex],
                         highwayErrorText(err),
                     );
                     await sleepMs(HIGHWAY_RETRY_BASE_MS * transportAttempts);