fix(protocol): improve upload recovery and time limits

Apply configurable idle and total transfer deadlines, cover supplied upload candidates, and retain recent endpoint outcomes per account and upload type. Preserve failed action responses and resource cleanup.

Refs #637

AGENTS.md

@@ -84,6 +84,7 @@ Use focused tests while iterating. Before handing off a code change, run the ful
 - **MCP execution services:** keep background tasks and event connections opt-in. Persist task state before effects, never replay interrupted writes, and finish durable cleanup before draining the protocol on shutdown. Input continuations must retain their original arguments and execute at most once. Event buffers require explicit account binding, bounded storage and visible gaps. Build the bundled message preview through `packages/mcp/scripts/build-message-app.mjs`; loading a preview must not send a message. Test restart, cancellation and legacy compatibility using isolated transports, including the compiled CLI.
 - **Protocol codecs:** keep the existing Proton integration in tests so they exercise the same transformations as production. Import real implementations; do not duplicate the algorithm in the test.
 - **Video uploads:** preserve the single-connection default and the additive global setting in [HIGHWAY_UPLOADS.md](packages/protocol/HIGHWAY_UPLOADS.md). Only video bodies opt into parallel chunks. Confirm every chunk before completion, bound both concurrency and lookahead, and drain outstanding reads and requests before closing the source. Keep retries bounded and backpressure changes visible. Test the actual uploader with isolated transports; passing tests do not prove live throughput.
+- **Highway upload recovery:** all chunks and workers share one file transfer deadline. Keep node observations bounded, short-lived and isolated by account session, upload type and full host/port; select only current server-supplied candidates. Only validated acknowledgements establish success. Preserve terminal failure reporting and drain pending file reads after cancellation. Test deadline cleanup and candidate rotation with isolated transports, including real loopback sockets; see [upload policy](packages/protocol/HIGHWAY_UPLOADS.md#upload-timeouts-and-node-selection).
 - **Native coverage:** `pnpm test:native` is separate from default protocol tests and requires compatible native artifacts. A missing artifact or unavailable target is a reported limitation, not permission to bypass the check.
 - **OneBot SSE:** HTTP event streams share the HTTP adapter's binding and authentication. Preserve account isolation, bounded replay, visible expired-cursor errors, report filtering and slow-reader limits described in [HTTP_SSE.md](packages/onebot/HTTP_SSE.md). Token/report-policy changes revoke existing streams and invalidate replay; test shutdown and reload with active connections.
 - **Live QQ behavior:** fixture tests and a successful API response do not prove delivery or client rendering. State what was tested locally and what still needs a real client.

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

@@ -82,8 +82,12 @@ class FakeSocket extends EventEmitter {
     headers: string[] = [];
     frames: Buffer[] = [];
     closeBeforeResponse = false;
+    idleBeforeResponse = false;
+    stallWrites = false;
+    responseDelayMs = 0;
     connectFailure?: Error | 'timeout';
     private connectTimer?: ReturnType<typeof setTimeout>;
+    private responseTimer?: ReturnType<typeof setTimeout>;
     private buf = Buffer.alloc(0);
     constructor(
         private readonly responseFrame: Buffer,
@@ -105,6 +109,10 @@ class FakeSocket extends EventEmitter {
     }
 
     write(data: unknown, cb?: (err?: Error) => void): boolean {
+        if (this.stallWrites) {
+            return false;
+        }
+
         if (this.destroyed || !this.writable) {
             const err = new Error('write after socket end');
 
@@ -148,11 +156,16 @@ class FakeSocket extends EventEmitter {
             this.buf = this.buf.subarray(total);
             this.posts += 1;
             const thisPost = this.posts;
-            setImmediate(() => {
+
+            const respond = () => {
                 if (this.destroyed) {
                     return;
                 }
 
+                if (this.idleBeforeResponse) {
+                    return;
+                }
+
                 if (thisPost === this.failOnPost) {
                     this.writable = false;
                     this.readableEnded = true;
@@ -199,12 +212,19 @@ class FakeSocket extends EventEmitter {
                     this.emit('end');
                     this.emit('close');
                 }
-            });
+            };
+
+            if (this.responseDelayMs > 0) {
+                this.responseTimer = setTimeout(respond, this.responseDelayMs);
+            } else {
+                setImmediate(respond);
+            }
         }
     }
 
     destroy() {
         clearTimeout(this.connectTimer);
+        clearTimeout(this.responseTimer);
         this.destroyed = true;
         this.writable = false;
     }
@@ -239,8 +259,8 @@ function highwayTrace(entries: LogEntry[]): LogEntry[] {
     return entries.filter((entry) => entry.level === 'trace' && entry.scope === 'Highway');
 }
 
-function makeBridge(): BridgeContext {
+function makeBridge(uin = '10001'): BridgeContext {
-    return { identity: { uin: '10001' } as unknown } as unknown as BridgeContext;
+    return { identity: { uin } as unknown } as unknown as BridgeContext;
 }
 
 function makeSession(): HighwaySession {
@@ -459,6 +479,7 @@ describe('Highway connection selection', () => {
     afterEach(() => {
         vi.useRealTimers();
         vi.restoreAllMocks();
+        vi.unstubAllEnvs();
         setLogLevel(previousLogLevel);
     });
 
@@ -472,6 +493,201 @@ describe('Highway connection selection', () => {
         };
     }
 
+    it.each([1003, 1004])(
+        'reaches the fourth candidate after three idle peers for command %i (#637)',
+        async (commandId) => {
+            vi.useFakeTimers();
+            vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS', '100');
+            vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '5000');
+            installFactory(() => {
+                const socket = new FakeSocket(buildEmptyHighwayResponseFrame(), 'keepalive');
+                socket.idleBeforeResponse =
+                    createConnectionMock.mock.lastCall?.[0].host !== '192.0.2.4';
+                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 completed = expect(
+                uploadHighwayHttp(
+                    makeBridge(),
+                    session,
+                    commandId,
+                    source,
+                    new Uint8Array(16),
+                    new Uint8Array(),
+                ),
+            ).resolves.toBeUndefined();
+            await Promise.all([completed, vi.runAllTimersAsync()]);
+            expect(createConnectionMock.mock.calls.map(([options]) => options)).toEqual(
+                session.servers,
+            );
+            expect(source.reads).toEqual([{ offset: 0, length: 80 }]);
+            expect(source.closeCount).toBe(1);
+            expect((createdSockets as FakeSocket[]).every((socket) => socket.destroyed)).toBe(true);
+        },
+    );
+
+    it('uses recent successful host and port only for the same account and upload type', async () => {
+        vi.useFakeTimers();
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS', '100');
+        const session = {
+            ...makeSession(),
+            servers: [
+                { host: '192.0.2.1', port: 14000 },
+                { host: '192.0.2.1', port: 80 },
+            ],
+        };
+        const bridge = makeBridge();
+
+        const send = async (owner: BridgeContext, commandId = 1004, candidates = session) => {
+            installFactory(() => {
+                const socket = new FakeSocket(buildEmptyHighwayResponseFrame(), 'keepalive');
+                socket.idleBeforeResponse = createConnectionMock.mock.lastCall?.[0].port === 14000;
+                return socket;
+            });
+            await Promise.all([
+                expect(
+                    uploadHighwayHttp(
+                        owner,
+                        candidates,
+                        commandId,
+                        new RecordingChunkSource(80),
+                        new Uint8Array(16),
+                        new Uint8Array(),
+                    ),
+                ).resolves.toBeUndefined(),
+                vi.runAllTimersAsync(),
+            ]);
+            return createConnectionMock.mock.calls.map(([options]) => options);
+        };
+
+        expect(await send(bridge)).toEqual(session.servers);
+        expect(await send(bridge)).toEqual([session.servers[1]]);
+        expect(await send(makeBridge('10002'))).toEqual(session.servers);
+        expect(await send(bridge, 1003)).toEqual(session.servers);
+        const replacement = { ...session, servers: [{ host: '192.0.2.2', port: 80 }] };
+        expect(await send(bridge, 1004, replacement)).toEqual(replacement.servers);
+        await vi.advanceTimersByTimeAsync(5 * 60_000);
+        expect(await send(bridge)).toEqual(session.servers);
+    });
+
+    it.each(['response', 'write', 'retry'])(
+        'stops a pending %s at the shared upload deadline and releases resources',
+        async (phase) => {
+            vi.useFakeTimers();
+            vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '150');
+            vi.stubEnv(
+                'SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS',
+                phase === 'retry' ? '50' : '1000',
+            );
+            installFactory(() => {
+                const socket = new FakeSocket(buildEmptyHighwayResponseFrame(), 'keepalive');
+                socket.idleBeforeResponse = true;
+                socket.stallWrites = phase === 'write';
+                return socket;
+            });
+            const entries: LogEntry[] = [];
+            setLogLevel('trace');
+            const unsubscribe = subscribeLogs((entry) => entries.push(entry));
+            const source = new RecordingChunkSource(80);
+            const start = Date.now();
+
+            try {
+                const failed = expect(
+                    uploadHighwayHttp(
+                        makeBridge(),
+                        sessionWithCandidates(),
+                        1004,
+                        source,
+                        new Uint8Array(16),
+                        new Uint8Array(),
+                    ),
+                ).rejects.toThrow('timed out after 150ms');
+                await Promise.all([failed, vi.runAllTimersAsync()]);
+                expect(Date.now() - start).toBe(150);
+                expect(createConnectionMock).toHaveBeenCalledTimes(1);
+                expect(source.closeCount).toBe(1);
+                expect(
+                    (createdSockets as FakeSocket[]).every(
+                        (socket) =>
+                            socket.destroyed &&
+                            socket.listenerCount('data') === 0 &&
+                            socket.listenerCount('close') === 0,
+                    ),
+                ).toBe(true);
+                const terminal = highwayTrace(entries).filter((entry) =>
+                    entry.message.startsWith('highway_put_terminal '),
+                );
+                expect(terminal).toHaveLength(1);
+                expect(terminal[0]!.message).toContain('reason=upload_timeout');
+                expect(vi.getTimerCount()).toBe(0);
+
+                if (phase === 'retry') {
+                    expect(
+                        entries.some(
+                            (entry) =>
+                                entry.message.includes('phase=read_response') &&
+                                entry.message.includes('received=0B'),
+                        ),
+                    ).toBe(true);
+                }
+            } finally {
+                unsubscribe();
+            }
+        },
+    );
+
+    it('does not renew the total deadline when a chunk succeeds', async () => {
+        vi.useFakeTimers();
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '250');
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS', '200');
+        installFactory(() => {
+            const socket = new FakeSocket(buildEmptyHighwayResponseFrame(), 'keepalive');
+            socket.responseDelayMs = 100;
+            return socket;
+        });
+        const source = new RecordingChunkSource(2 * HIGHWAY_BLOCK_SIZE + 80);
+        const start = Date.now();
+        const failed = expect(
+            uploadHighwayHttp(
+                makeBridge(),
+                makeSession(),
+                1004,
+                source,
+                new Uint8Array(16),
+                new Uint8Array(),
+            ),
+        ).rejects.toThrow('acknowledgedBytes=2097152/2097232');
+        await Promise.all([failed, vi.runAllTimersAsync()]);
+        expect(Date.now() - start).toBe(250);
+        expect(source.closeCount).toBe(1);
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('rejects invalid timeout configuration before connecting and closes the source', async () => {
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '0');
+        installFactory(() => new FakeSocket(buildEmptyHighwayResponseFrame(), 'keepalive'));
+        const source = new RecordingChunkSource(80);
+        await expect(
+            uploadHighwayHttp(
+                makeBridge(),
+                makeSession(),
+                1004,
+                source,
+                new Uint8Array(16),
+                new Uint8Array(),
+            ),
+        ).rejects.toThrow('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS');
+        expect(createConnectionMock).not.toHaveBeenCalled();
+        expect(source.closeCount).toBe(1);
+    });
+
     it.each(['timeout', 'ECONNREFUSED', 'ENETUNREACH'])(
         'switches on %s and uploads over the successful connection',
         async (failure) => {
@@ -515,8 +731,9 @@ describe('Highway connection selection', () => {
         },
     );
 
-    it('bounds initial selection to ten seconds without repeating exhausted candidates', async () => {
+    it('bounds initial selection by the configured total deadline without repeating candidates', async () => {
         vi.useFakeTimers();
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '10000');
         installFactory(() => {
             const socket = new FakeSocket(buildEmptyHighwayResponseFrame(), 'keepalive');
             socket.connectFailure = 'timeout';
@@ -544,7 +761,7 @@ describe('Highway connection selection', () => {
                 new Uint8Array(16),
                 new Uint8Array(),
             );
-            const failed = expect(upload).rejects.toThrow(/connection failed.*TCP connect timeout/);
+            const failed = expect(upload).rejects.toThrow('timed out after 10000ms');
             await vi.runAllTimersAsync();
             await failed;
             expect(Date.now() - start).toBe(10000);
@@ -561,7 +778,7 @@ describe('Highway connection selection', () => {
                 entry.message.startsWith('highway_put_terminal '),
             );
             expect(terminals).toHaveLength(1);
-            expect(terminals[0]!.message).toContain('reason=connect_exhausted');
+            expect(terminals[0]!.message).toContain('reason=upload_timeout');
             expect(terminals[0]!.message).toContain('connections=0');
         } finally {
             unsubscribe();

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

@@ -177,6 +177,8 @@ beforeEach(() => {
 });
 afterEach(() => {
     setLogLevel(level);
+    vi.unstubAllEnvs();
+    vi.useRealTimers();
 
     for (const socket of sockets) {
         socket.destroy();
@@ -184,6 +186,64 @@ afterEach(() => {
 });
 
 describe('bounded parallel highway uploads', () => {
+    it('does not let trickled response bytes extend the whole-file deadline', async () => {
+        vi.useFakeTimers();
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '100');
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS', '30');
+
+        onRequest = ({ socket }) => {
+            const timer = setInterval(() => socket.emit('data', Buffer.from('x')), 10);
+            socket.once('close', () => clearInterval(timer));
+        };
+
+        const source = new Source(80);
+        const start = Date.now();
+        const failed = expect(upload(source, 1)).rejects.toThrow('timed out after 100ms');
+        await Promise.all([failed, vi.runAllTimersAsync()]);
+        expect(Date.now() - start).toBe(100);
+        expect(source.closes).toBe(1);
+        expect(active.size).toBe(0);
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('drains a pending source read after the deadline without sending its bytes', async () => {
+        vi.useFakeTimers();
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '50');
+        const source = new Source(80);
+        let reading = false;
+        const read = source.read.bind(source);
+
+        source.read = async (offset, length) => {
+            reading = true;
+            await new Promise<void>((resolve) => setTimeout(resolve, 100));
+            reading = false;
+            return read(offset, length);
+        };
+
+        const close = vi.spyOn(source, 'close').mockImplementation(async () => {
+            expect(reading).toBe(false);
+        });
+        const failed = expect(upload(source, 1)).rejects.toThrow('timed out after 50ms');
+        await Promise.all([failed, vi.runAllTimersAsync()]);
+        expect(connect).not.toHaveBeenCalled();
+        expect(close).toHaveBeenCalledTimes(1);
+        expect(vi.getTimerCount()).toBe(0);
+    });
+
+    it('aborts all parallel requests at one whole-file deadline', async () => {
+        vi.useFakeTimers();
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '1000');
+        const source = new Source(6 * BLOCK);
+        const start = Date.now();
+        const failed = expect(upload(source, 3)).rejects.toThrow('timed out after 1000ms');
+        await Promise.all([failed, vi.runAllTimersAsync()]);
+        expect(Date.now() - start).toBe(1000);
+        expect(requests).toHaveLength(3);
+        expect(sockets.every((socket) => socket.destroyed)).toBe(true);
+        expect(active.size).toBe(0);
+        expect(source.closes).toBe(1);
+        expect(vi.getTimerCount()).toBe(0);
+    });
     it.each([2, 3, 4])(
         'uses %i connections, accepts out-of-order acknowledgements and sends the tail last',
         async (concurrency) => {

packages/protocol/HIGHWAY_UPLOADS.md

@@ -1,4 +1,6 @@
-# Video upload connections
+# Media transfer settings
+
+## Video upload connections
 
 Video uploads use one connection by default. In **Settings → Global configuration →
 Video upload connections**, choose 1, 2, 3 or 4 connections. The equivalent setting
@@ -36,6 +38,54 @@ Regression tests use isolated transports and cover concurrency bounds,
 out-of-order acknowledgements, a stalled early chunk, retry exhaustion, backpressure
 and cleanup. They do not establish real QQ delivery or upload throughput.
 
+## Upload timeouts and node selection
+
+All Highway HTTP uploads (including images, voice, video and files) snapshot
+these deployment-wide environment variables when each file starts transferring:
+
+| Variable | Default | Meaning |
+| --- | --- | --- |
+| `SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS` | `30000` | Maximum wait without response bytes during one request, in milliseconds. Resets when response bytes arrive. |
+| `SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS` | `1800000` | Total transfer budget for one file, in milliseconds (30 minutes). Shared by all chunks, connections, writes, response waits and retry delays. |
+
+For example, a deployment sending small images can use:
+
+```sh
+SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS=5000
+SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS=120000
+```
+
+Restart the process/container with these variables. Both values must be integers
+from 1 to 2147483647; invalid values fail the upload visibly rather than silently
+using a default. The total budget does not reset after a successful chunk, after
+incoming bytes, or on a different node. The idle timer starts when the HTTP request
+begins, including writes; slow connections may need a larger value. Source downloads,
+hashing and QQ upload preparation precede this transfer budget. Video bodies and
+their covers are separate files with separate budgets.
+
+Transport failures retry only the unconfirmed chunk. Each chunk allows at least
+three attempts, or enough attempts to cover all distinct supplied candidates when
+there are more than three, subject to the file's total budget. Each connection has
+a three-second handshake limit. An expired total budget aborts all workers and
+closes their connections; outstanding filesystem reads are drained before closing
+the source, so cleanup can finish after that deadline. Explicit server rejection
+and invalid acknowledgements still fail immediately. Backpressure retries remain
+bounded separately and share the same file deadline.
+
+Node observations belong to the account session and upload type, keyed by the
+complete host and port. A validated successful upload acknowledgement is preferred
+for five minutes; a transport failure lowers that endpoint's priority for thirty
+seconds. Unknown endpoints remain eligible ahead of recent failures, and failures
+never permanently exclude a candidate. Only destinations supplied for the current
+upload can be selected. The history holds at most 128 observations per account
+session, contains no tickets or payloads, and is released when that account's
+identity object is discarded. A process restart clears it.
+
+TRACE logs show the effective timeouts and candidate order. Read timeouts include
+the current write/read phase, completed write byte count, received byte count and
+whether response headers were seen. Total-budget exhaustion is reported separately
+as `upload_timeout`; failed transfers remain failed OneBot actions.
+
 ## Media download limits
 
 Two deployment-wide knobs bound HTTP media downloads in `highway/utils.ts`. They are

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

@@ -18,14 +18,18 @@ import { protobuf_decode, protobuf_encode } from '@snowluma/proton';
 import net from 'net';
 import { promises as fsp } from 'fs';
 import type { BridgeContext } from '../bridge-context';
-import { validateUploadConcurrency } from './upload-settings';
+import {
+    highwayUploadTimeouts,
+    validateUploadConcurrency,
+    type HighwayUploadTimeouts,
+} from './upload-settings';
+import { uploadNodeHistory } from './upload-nodes';
 import { computeMd5, packHighwayFrame, unpackHighwayFrame } from './utils';
 
 const log = createLogger('Highway');
 const HIGHWAY_APP_ID = 1600001604;
 const HIGHWAY_BLOCK_SIZE = 1024 * 1024;
 const HIGHWAY_CONNECT_TIMEOUT_MS = 3000;
-const HIGHWAY_CONNECT_BUDGET_MS = 10000;
 
 // Per-chunk transport retry: a transient connection failure retries only
 // the unconfirmed offset, preserving acknowledged data (issue #118).
@@ -44,14 +48,6 @@ const HIGHWAY_DELAY_RETRY_CODES = new Set([102902, 302902]);
 const HIGHWAY_DELAY_RETRY_MS = 250;
 const HIGHWAY_MAX_DELAY_RETRIES = 8;
 
-// Idle read timeout for a highway response. tcpConnect() clears the socket's
-// own timeout once connected, so a peer that accepts the connection but then
-// neither responds nor FINs would hang the whole upload forever. The single
-// persistent connection (issue #211) keeps a socket alive across the whole
-// file, widening that exposure, so the response reader arms its own idle timer
-// and treats a stall as a retryable transport failure.
-const HIGHWAY_READ_IDLE_MS = 30000;
-
 function sleepMs(ms: number, signal: AbortSignal): Promise<void> {
     signal.throwIfAborted();
     return new Promise((resolve, reject) => {
@@ -95,6 +91,7 @@ type HighwayPutFailureReason =
     | 'server_rejected'
     | 'source_close_failed'
     | 'source_read_failed'
+    | 'upload_timeout'
     | 'transport_exhausted';
 
 interface HighwayPutTraceState {
@@ -359,29 +356,25 @@ async function connectHighway(
     servers: HighwayServer[],
     startIndex: number,
     signal: AbortSignal,
+    onFailure: (server: HighwayServer) => void,
 ): Promise<{ socket: net.Socket; server: HighwayServer; serverIndex: number }> {
-    const deadline = Date.now() + HIGHWAY_CONNECT_BUDGET_MS;
     const errors: Error[] = [];
 
     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) {
-            break;
-        }
 
         try {
             const socket = await tcpConnect(
                 server.host,
                 server.port,
-                Math.min(HIGHWAY_CONNECT_TIMEOUT_MS, remaining),
+                HIGHWAY_CONNECT_TIMEOUT_MS,
                 signal,
             );
             return { socket, server, serverIndex };
         } catch (cause) {
             signal.throwIfAborted();
+            onFailure(server);
             const error = new Error(`${server.host}:${server.port}: ${highwayErrorText(cause)}`, {
                 cause,
             });
@@ -396,13 +389,41 @@ async function connectHighway(
     );
 }
 
-function socketWrite(socket: net.Socket, data: Uint8Array): Promise<void> {
+function socketWrite(socket: net.Socket, data: Uint8Array, signal: AbortSignal): Promise<void> {
+    signal.throwIfAborted();
     return new Promise((resolve, reject) => {
-        socket.write(data, (err) => (err ? reject(err) : resolve()));
+        const finish = (error?: unknown) => {
+            signal.removeEventListener('abort', onAbort);
+
+            if (error !== undefined) {
+                reject(error);
+            } else {
+                resolve();
+            }
+        };
+
+        const onAbort = () => finish(signal.reason);
+        signal.addEventListener('abort', onAbort, { once: true });
+
+        try {
+            socket.write(data, (error) => finish(error ?? undefined));
+        } catch (error) {
+            finish(error);
+        }
     });
 }
 
-function readHttpResponseBody(socket: net.Socket, signal: AbortSignal): Promise<Uint8Array> {
+interface RequestProgress {
+    phase: 'write_headers' | 'write_body' | 'read_response';
+    writtenBytes: number;
+}
+
+function readHttpResponseBody(
+    socket: net.Socket,
+    signal: AbortSignal,
+    idleMs: number,
+    progress: RequestProgress,
+): Promise<Uint8Array> {
     signal.throwIfAborted();
     return new Promise((resolve, reject) => {
         const chunks: Buffer[] = [];
@@ -410,6 +431,7 @@ function readHttpResponseBody(socket: net.Socket, signal: AbortSignal): Promise<
         let contentLength = 0;
         let totalNeeded = 0;
         let settled = false;
+        let receivedBytes = 0;
 
         const partialResponseBody = (): Uint8Array | undefined => {
             if (headerEnd < 0 || getLogLevel() !== 'trace') {
@@ -433,12 +455,12 @@ function readHttpResponseBody(socket: net.Socket, signal: AbortSignal): Promise<
                     finish(() =>
                         reject(
                             new HighwayResponseReadError(
-                                'highway response read timeout (peer idle)',
+                                `highway response read timeout (peer idle; phase=${progress.phase}, written=${progress.writtenBytes}B, received=${receivedBytes}B, headerSeen=${headerEnd >= 0}, expected=${totalNeeded}B)`,
                                 partialResponseBody(),
                             ),
                         ),
                     ),
-                HIGHWAY_READ_IDLE_MS,
+                idleMs,
             );
         };
 
@@ -467,6 +489,7 @@ function readHttpResponseBody(socket: net.Socket, signal: AbortSignal): Promise<
         const onData = (chunk: Buffer) => {
             armIdle();
             chunks.push(chunk);
+            receivedBytes += chunk.length;
             const buf = Buffer.concat(chunks);
 
             if (headerEnd < 0) {
@@ -542,6 +565,7 @@ async function httpPostFrame(
     path: string,
     body: Uint8Array,
     signal: AbortSignal,
+    idleMs: number,
 ): Promise<Uint8Array> {
     const header = `POST ${path} HTTP/1.1\r\nHost: ${host}\r\nConnection: keep-alive\r\nAccept-Encoding: identity\r\nUser-Agent: Mozilla/5.0 (compatible; MSIE 10.0; Windows NT 6.2)\r\nContent-Length: ${body.length}\r\n\r\n`;
     const responseAbort = new AbortController();
@@ -554,14 +578,20 @@ async function httpPostFrame(
     signal.throwIfAborted();
     signal.addEventListener('abort', onAbort, { once: true });
     // Arm the reader before writing: a fast peer may reply before write callbacks.
-    const response = readHttpResponseBody(socket, responseAbort.signal);
+    const progress: RequestProgress = { phase: 'write_headers', writtenBytes: 0 };
+    const response = readHttpResponseBody(socket, responseAbort.signal, idleMs, progress);
     const write = (async () => {
-        await socketWrite(socket, Buffer.from(header, 'ascii'));
+        await socketWrite(socket, Buffer.from(header, 'ascii'), responseAbort.signal);
+        progress.writtenBytes += Buffer.byteLength(header, 'ascii');
         signal.throwIfAborted();
 
         if (body.length > 0) {
-            await socketWrite(socket, body);
+            progress.phase = 'write_body';
+            await socketWrite(socket, body, responseAbort.signal);
+            progress.writtenBytes += body.byteLength;
         }
+
+        progress.phase = 'read_response';
     })();
 
     try {
@@ -739,13 +769,16 @@ async function uploadHighwayHttpOperation(
 
     signal.addEventListener('abort', wake, { once: true });
 
-    const fail = (reason: HighwayPutFailureReason, error: unknown): never => {
+    const abortUpload = (reason: HighwayPutFailureReason, error: unknown): void => {
         if (!signal.aborted) {
             trace.failureReason = reason;
             trace.error = error;
             abort.abort(error);
         }
+    };
 
+    const fail = (reason: HighwayPutFailureReason, error: unknown): never => {
+        abortUpload(reason, error);
         throw signal.reason;
     };
 
@@ -766,6 +799,11 @@ async function uploadHighwayHttpOperation(
     let retries = 0;
     let contiguous = 0;
     const confirmed = new Set<number>();
+    const history = uploadNodeHistory(bridge.identity);
+    const servers = history.order(commandId, session.servers);
+    const maxAttempts = Math.max(HIGHWAY_MAX_CHUNK_ATTEMPTS, servers.length);
+    let timeouts: HighwayUploadTimeouts;
+    let deadlineTimer: ReturnType<typeof setTimeout> | undefined;
 
     const dropSocket = (worker: Worker) => {
         worker.socket?.destroy();
@@ -854,9 +892,10 @@ async function uploadHighwayHttpOperation(
 
                     if (!worker.socket) {
                         const connected = await connectHighway(
-                            session.servers,
+                            servers,
                             worker.serverIndex,
                             signal,
+                            (server) => history.record(commandId, server, false),
                         );
                         worker.socket = connected.socket;
                         worker.serverIndex = connected.serverIndex;
@@ -896,10 +935,11 @@ async function uploadHighwayHttpOperation(
                     );
                     body = await httpPostFrame(
                         worker.socket,
-                        session.servers[worker.serverIndex]!.host,
+                        servers[worker.serverIndex]!.host,
                         pathStr,
                         packHighwayFrame(head, chunk),
                         signal,
+                        timeouts.idleMs,
                     );
                 } catch (error) {
                     signal.throwIfAborted();
@@ -919,9 +959,10 @@ async function uploadHighwayHttpOperation(
                     }
 
                     dropSocket(worker);
+                    history.record(commandId, servers[worker.serverIndex]!, false);
                     transportAttempts += 1;
 
-                    if (transportAttempts >= HIGHWAY_MAX_CHUNK_ATTEMPTS) {
+                    if (transportAttempts >= maxAttempts) {
                         return fail(
                             'transport_exhausted',
                             new Error(
@@ -932,13 +973,13 @@ async function uploadHighwayHttpOperation(
                     }
 
                     retries += 1;
-                    worker.serverIndex = (worker.serverIndex + 1) % session.servers.length;
+                    worker.serverIndex = (worker.serverIndex + 1) % servers.length;
                     log.trace(
                         'highway_put_branch branch=chunk_retry offset=%d bytes=%d attempt=%d nextServer=%j error=%j',
                         offset,
                         chunkSize,
                         transportAttempts,
-                        session.servers[worker.serverIndex],
+                        servers[worker.serverIndex],
                         highwayErrorText(error),
                     );
                     await sleepMs(HIGHWAY_RETRY_BASE_MS * transportAttempts, signal);
@@ -991,7 +1032,7 @@ async function uploadHighwayHttpOperation(
                         effectiveConcurrency = 1;
                     }
 
-                    worker.serverIndex = (worker.serverIndex + 1) % session.servers.length;
+                    worker.serverIndex = (worker.serverIndex + 1) % servers.length;
                     const delayMs = HIGHWAY_DELAY_RETRY_MS;
                     log.trace(
                         'highway_put_branch branch=chunk_delay_retry offset=%d bytes=%d attempt=%d errorCode=%d delayMs=%d',
@@ -1029,6 +1070,7 @@ async function uploadHighwayHttpOperation(
                     );
                 }
 
+                history.record(commandId, servers[worker.serverIndex]!, true);
                 break;
             } finally {
                 activeRequests -= 1;
@@ -1063,13 +1105,30 @@ async function uploadHighwayHttpOperation(
 
     try {
         try {
+            timeouts = highwayUploadTimeouts();
             concurrency = validateUploadConcurrency(
                 options.concurrency === undefined ? 1 : options.concurrency,
             );
         } catch (error) {
-            fail('invalid_options', error);
+            return fail('invalid_options', error);
         }
 
+        deadlineTimer = setTimeout(() => {
+            abortUpload(
+                'upload_timeout',
+                new Error(
+                    `highway upload timed out after ${timeouts.totalMs}ms (cmdId=${commandId} acknowledgedBytes=${trace.uploadedBytes}/${totalSize})`,
+                ),
+            );
+        }, timeouts.totalMs);
+        log.trace(
+            'highway_put_policy idleTimeoutMs=%d totalTimeoutMs=%d maxAttempts=%d servers=%j',
+            timeouts.idleMs,
+            timeouts.totalMs,
+            maxAttempts,
+            servers,
+        );
+
         const pieces = Math.ceil(totalSize / HIGHWAY_BLOCK_SIZE);
         // The final piece commits the file only after every preceding piece is
         // acknowledged. One/two-piece files therefore need just one connection.
@@ -1142,6 +1201,7 @@ async function uploadHighwayHttpOperation(
             Date.now() - trace.startedAt,
         );
     } finally {
+        clearTimeout(deadlineTimer);
         abort.abort(new Error('highway upload stopped'));
 
         for (const worker of workers) {

packages/protocol/src/highway/upload-nodes.ts

@@ -0,0 +1,79 @@
+import type { BridgeContext } from '../bridge-context';
+import type { HighwayServer } from './highway-client';
+
+const SUCCESS_TTL_MS = 5 * 60_000;
+const FAILURE_TTL_MS = 30_000;
+const MAX_OBSERVATIONS = 128;
+
+interface Observation {
+    succeeded: boolean;
+    expiresAt: number;
+}
+
+/** Short-lived scheduling evidence, never credentials or additional upload destinations. */
+export class UploadNodeHistory {
+    private readonly observations = new Map<string, Observation>();
+
+    private key(commandId: number, server: HighwayServer): string {
+        return JSON.stringify([commandId, server.host, server.port]);
+    }
+
+    private prune(now: number): void {
+        for (const [key, observation] of this.observations) {
+            if (observation.expiresAt <= now) {
+                this.observations.delete(key);
+            }
+        }
+    }
+
+    order(commandId: number, candidates: readonly HighwayServer[]): HighwayServer[] {
+        this.prune(Date.now());
+        const unique = new Map(candidates.map((server) => [this.key(commandId, server), server]));
+
+        const rank = (server: HighwayServer): number => {
+            const observation = this.observations.get(this.key(commandId, server));
+            return observation ? (observation.succeeded ? 0 : 2) : 1;
+        };
+
+        // Stable ties retain server-provided order. Failed endpoints remain eligible:
+        // a temporary observation must not permanently exclude the only working node.
+        return [...unique.values()].sort((a, b) => rank(a) - rank(b));
+    }
+
+    record(commandId: number, server: HighwayServer, succeeded: boolean): void {
+        const now = Date.now();
+        this.prune(now);
+        const key = this.key(commandId, server);
+        this.observations.delete(key);
+        this.observations.set(key, {
+            succeeded,
+            expiresAt: now + (succeeded ? SUCCESS_TTL_MS : FAILURE_TTL_MS),
+        });
+
+        if (this.observations.size > MAX_OBSERVATIONS) {
+            const oldest = this.observations.keys().next();
+
+            if (!oldest.done) {
+                this.observations.delete(oldest.value);
+            }
+        }
+    }
+}
+
+// IdentityService belongs to one account's Bridge. Weak ownership also releases
+// observations when that account session is discarded, without a global UIN registry.
+const histories = new WeakMap<
+    BridgeContext['identity'],
+    { uin: string; history: UploadNodeHistory }
+>();
+
+export function uploadNodeHistory(identity: BridgeContext['identity']): UploadNodeHistory {
+    let entry = histories.get(identity);
+
+    if (!entry || entry.uin !== identity.uin) {
+        entry = { uin: identity.uin, history: new UploadNodeHistory() };
+        histories.set(identity, entry);
+    }
+
+    return entry.history;
+}

packages/protocol/src/highway/upload-settings.ts

@@ -1,6 +1,36 @@
 /** Deployment-wide policy. Each video snapshots it before preparing its bytes. */
 let videoUploadConcurrency = 1;
 
+export interface HighwayUploadTimeouts {
+    idleMs: number;
+    totalMs: number;
+}
+
+/** Snapshot once per file; a deployment change must not alter an in-flight upload. */
+export function highwayUploadTimeouts(env: NodeJS.ProcessEnv = process.env): HighwayUploadTimeouts {
+    const read = (name: string, defaultMs: number): number => {
+        const raw = env[name];
+
+        if (raw === undefined) {
+            return defaultMs;
+        }
+
+        const value = raw.trim();
+        const parsed = /^[1-9]\d*$/.test(value) ? Number(value) : Number.NaN;
+
+        if (!Number.isSafeInteger(parsed) || parsed > 2_147_483_647) {
+            throw new RangeError(`${name} must be an integer from 1 to 2147483647 milliseconds`);
+        }
+
+        return parsed;
+    };
+
+    return {
+        idleMs: read('SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS', 30_000),
+        totalMs: read('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', 30 * 60_000),
+    };
+}
+
 export function validateUploadConcurrency(value: unknown): number {
     if (typeof value !== 'number' || !Number.isInteger(value) || value < 1 || value > 4) {
         throw new RangeError('videoUploadConcurrency must be an integer from 1 to 4');

packages/protocol/tests/highway/upload-nodes.test.ts

@@ -0,0 +1,56 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { UploadNodeHistory } from '../../src/highway/upload-nodes';
+
+const a = { host: '192.0.2.1', port: 14000 };
+const b = { host: '192.0.2.1', port: 80 };
+const c = { host: '192.0.2.2', port: 80 };
+
+afterEach(() => vi.useRealTimers());
+
+describe('upload node observations', () => {
+    it('orders current candidates by confirmed success, unknown, then recent failure', () => {
+        const history = new UploadNodeHistory();
+        history.record(1004, a, false);
+        history.record(1004, b, true);
+        const original = [a, c, b, a];
+        expect(history.order(1004, original)).toEqual([b, c, a]);
+        expect(original).toEqual([a, c, b, a]);
+        expect(history.order(1003, [a, c, b])).toEqual([a, c, b]);
+        expect(history.order(1004, [c])).toEqual([c]);
+    });
+
+    it('expires failures and successes and still allows every failed candidate', () => {
+        vi.useFakeTimers();
+        const history = new UploadNodeHistory();
+        history.record(1004, a, false);
+        history.record(1004, b, false);
+        expect(history.order(1004, [a, b])).toEqual([a, b]);
+        expect(history.order(1004, [a, c])).toEqual([c, a]);
+        vi.advanceTimersByTime(30_000);
+        expect(history.order(1004, [a, c])).toEqual([a, c]);
+        history.record(1004, b, true);
+        expect(history.order(1004, [a, b])).toEqual([b, a]);
+        vi.advanceTimersByTime(5 * 60_000);
+        expect(history.order(1004, [a, b])).toEqual([a, b]);
+    });
+
+    it('lets a new failure replace success and a subsequent success recover', () => {
+        const history = new UploadNodeHistory();
+        history.record(1004, b, true);
+        history.record(1004, b, false);
+        expect(history.order(1004, [b, a])).toEqual([a, b]);
+        history.record(1004, b, true);
+        expect(history.order(1004, [a, b])).toEqual([b, a]);
+    });
+
+    it('evicts old observations when many distinct endpoints arrive', () => {
+        const history = new UploadNodeHistory();
+        history.record(1004, a, false);
+
+        for (let port = 10000; port < 10200; port++) {
+            history.record(1004, { host: '192.0.2.3', port }, false);
+        }
+
+        expect(history.order(1004, [a, c])).toEqual([a, c]);
+    });
+});

packages/protocol/tests/highway/upload-settings.test.ts

@@ -0,0 +1,33 @@
+import { describe, expect, it } from 'vitest';
+import { highwayUploadTimeouts } from '../../src/highway/upload-settings';
+
+describe('highway upload timeouts', () => {
+    it('retains the idle default and supplies a finite whole-file deadline', () => {
+        expect(highwayUploadTimeouts({})).toEqual({ idleMs: 30_000, totalMs: 1_800_000 });
+    });
+
+    it('reads independent millisecond budgets without clamping', () => {
+        expect(
+            highwayUploadTimeouts({
+                SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS: ' 5000 ',
+                SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS: '120000',
+            }),
+        ).toEqual({ idleMs: 5000, totalMs: 120000 });
+        expect(highwayUploadTimeouts({ SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS: '1' }).totalMs).toBe(1);
+        expect(
+            highwayUploadTimeouts({ SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS: '2147483647' }).totalMs,
+        ).toBe(2147483647);
+    });
+
+    it.each(['', ' ', '0', '-1', '1.5', 'NaN', 'Infinity', '1e3', '200ms', '2147483648'])(
+        'rejects invalid configuration %j',
+        (value) => {
+            for (const name of [
+                'SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS',
+                'SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS',
+            ]) {
+                expect(() => highwayUploadTimeouts({ [name]: value })).toThrow(name);
+            }
+        },
+    );
+});

packages/protocol/tests/highway/upload-transport.test.ts

@@ -0,0 +1,133 @@
+import net from 'node:net';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import type { BridgeContext } from '../../src/bridge-context';
+import {
+    BufferChunkSource,
+    uploadHighwayHttp,
+    type HighwayServer,
+} from '../../src/highway/highway-client';
+import { packHighwayFrame } from '../../src/highway/utils';
+
+const servers: net.Server[] = [];
+const sockets = new Set<net.Socket>();
+
+afterEach(async () => {
+    vi.unstubAllEnvs();
+
+    for (const socket of sockets) {
+        socket.destroy();
+    }
+
+    await Promise.all(
+        servers.splice(0).map(
+            (server) =>
+                new Promise<void>((resolve, reject) => {
+                    server.close((error) => (error ? reject(error) : resolve()));
+                }),
+        ),
+    );
+});
+
+async function listen(
+    reply: boolean,
+): Promise<{ address: HighwayServer; connections: () => number }> {
+    let connections = 0;
+    const server = net.createServer((socket) => {
+        connections++;
+        sockets.add(socket);
+        socket.once('close', () => sockets.delete(socket));
+        let received = Buffer.alloc(0);
+        socket.on('data', (bytes: Buffer) => {
+            received = Buffer.concat([received, bytes]);
+            const end = received.indexOf('\r\n\r\n');
+
+            if (!reply || end < 0) {
+                return;
+            }
+
+            const header = received.subarray(0, end).toString('ascii');
+            const length = Number(header.match(/Content-Length: (\d+)/i)?.[1]);
+
+            if (received.length < end + 4 + length) {
+                return;
+            }
+
+            const frame = packHighwayFrame(new Uint8Array(), new Uint8Array());
+            socket.write(
+                Buffer.concat([
+                    Buffer.from(`HTTP/1.1 200 OK\r\nContent-Length: ${frame.byteLength}\r\n\r\n`),
+                    frame,
+                ]),
+            );
+            received = received.subarray(end + 4 + length);
+        });
+    });
+    await new Promise<void>((resolve, reject) => {
+        server.once('error', reject);
+        server.listen(0, '127.0.0.1', () => {
+            server.off('error', reject);
+            resolve();
+        });
+    });
+    servers.push(server);
+    const address = server.address();
+
+    if (!address || typeof address === 'string') {
+        throw new Error('missing loopback address');
+    }
+
+    return {
+        address: { host: address.address, port: address.port },
+        connections: () => connections,
+    };
+}
+
+function bridge(uin: string): BridgeContext {
+    return { identity: { uin } } as unknown as BridgeContext;
+}
+
+function send(
+    owner: BridgeContext,
+    candidates: HighwayServer[],
+    source = new BufferChunkSource(new Uint8Array([1, 2, 3])),
+) {
+    return uploadHighwayHttp(
+        owner,
+        { sigSession: new Uint8Array([1]), sessionKey: new Uint8Array(), servers: candidates },
+        1004,
+        source,
+        new Uint8Array(16),
+        new Uint8Array(),
+    );
+}
+
+describe('highway upload over real loopback sockets', () => {
+    it('switches idle ports and remembers only confirmed success for that account', async () => {
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS', '100');
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '5000');
+        const idle = await listen(false);
+        const healthy = await listen(true);
+        const owner = bridge('10001');
+        const candidates = [idle.address, healthy.address];
+        await send(owner, candidates);
+        await send(owner, candidates);
+        expect(idle.connections()).toBe(1);
+        expect(healthy.connections()).toBe(2);
+        await send(bridge('10002'), candidates);
+        expect(idle.connections()).toBe(2);
+        expect(healthy.connections()).toBe(3);
+    });
+
+    it('rejects an unresponsive peer at the total deadline and closes its source once', async () => {
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_IDLE_TIMEOUT_MS', '10000');
+        vi.stubEnv('SNOWLUMA_HIGHWAY_UPLOAD_TIMEOUT_MS', '100');
+        const idle = await listen(false);
+        const source = new BufferChunkSource(new Uint8Array([1, 2, 3]));
+        const close = vi.spyOn(source, 'close');
+        await expect(send(bridge('10003'), [idle.address], source)).rejects.toThrow(
+            'timed out after 100ms',
+        );
+        expect(close).toHaveBeenCalledTimes(1);
+        expect(idle.connections()).toBe(1);
+    });
+});