fix(transfer): an abort left N-1 unhandled rejections behind
Some checks failed
Test / test (push) Has been cancelled
Some checks failed
Test / test (push) Has been cancelled
Second flake in the same hunt, and a real bug rather than a test artifact.
`runRoundRobinUpload` and its resumed twin ended with `await Promise.all(workers)`.
`Promise.all` rejects the moment one worker throws and leaves the rest running
with nobody holding their rejections — so an abort, which fails every lane at
once, produces one propagated error and N-1 unhandled ones.
Bun counts an unhandled rejection as a test failure, which is how
(fail) Resume protocol — kill-restart-verify
→ sender crash mid-transfer → resumeUpload completes the same stream
TransferAbortError: Aborted during attempt
failed about one run in fourteen, with an error the test had already caught
properly through `handle1.done()`. The test was right; the engine was dropping
rejections on the floor beside it.
`settleWorkers` waits for every lane, then reports the first real failure —
preferring a genuine error over `TransferAbortError` when both happened, since
an abort is not the interesting one when something actually broke. Four call
sites.
This is not only a test problem. In production the same shape means a
cancelled or failed multi-lane transfer emits unhandled rejections into
whatever process is running the engine, and on a strict runtime that is a
crash rather than a log line.
Verified: 1198 pass / 0 fail.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014489bKUtUEY1Zgs9xN9mt7
This commit is contained in:
@@ -451,7 +451,7 @@ export class TransferEngine {
|
|||||||
for (const q of queues.values()) q.abort(err);
|
for (const q of queues.values()) q.abort(err);
|
||||||
throw err;
|
throw err;
|
||||||
}
|
}
|
||||||
await Promise.all(workers);
|
await settleWorkers(workers);
|
||||||
}
|
}
|
||||||
|
|
||||||
private async runRoundRobinUploadResumed(
|
private async runRoundRobinUploadResumed(
|
||||||
@@ -518,7 +518,7 @@ export class TransferEngine {
|
|||||||
for (const q of queues.values()) q.abort(err);
|
for (const q of queues.values()) q.abort(err);
|
||||||
throw err;
|
throw err;
|
||||||
}
|
}
|
||||||
await Promise.all(workers);
|
await settleWorkers(workers);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -774,7 +774,7 @@ export class TransferEngine {
|
|||||||
for (const q of queues.values()) q.abort(err);
|
for (const q of queues.values()) q.abort(err);
|
||||||
throw err;
|
throw err;
|
||||||
}
|
}
|
||||||
await Promise.all(workers);
|
await settleWorkers(workers);
|
||||||
}
|
}
|
||||||
|
|
||||||
private async runRoundRobinUpload(state: OutgoingState): Promise<void> {
|
private async runRoundRobinUpload(state: OutgoingState): Promise<void> {
|
||||||
@@ -851,7 +851,7 @@ export class TransferEngine {
|
|||||||
for (const q of queues.values()) q.abort(err);
|
for (const q of queues.values()) q.abort(err);
|
||||||
throw err;
|
throw err;
|
||||||
}
|
}
|
||||||
await Promise.all(workers);
|
await settleWorkers(workers);
|
||||||
}
|
}
|
||||||
|
|
||||||
private async runLaneWorker(
|
private async runLaneWorker(
|
||||||
@@ -1596,3 +1596,25 @@ function snapshotIncomingLanes(
|
|||||||
}
|
}
|
||||||
return out;
|
return out;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Wait for every lane worker, then report the first real failure.
|
||||||
|
*
|
||||||
|
* `Promise.all` rejects the moment one worker throws and leaves the others
|
||||||
|
* running with nobody holding their rejections — so an abort, which fails
|
||||||
|
* every lane at once, produces one propagated error and N-1 unhandled ones.
|
||||||
|
* Bun counts an unhandled rejection as a test failure, which is how the resume
|
||||||
|
* test failed roughly one run in fourteen on 08.09.2026 with a
|
||||||
|
* `TransferAbortError` the test had already caught through the handle.
|
||||||
|
*
|
||||||
|
* An abort is not the interesting failure when a real one is present, so a
|
||||||
|
* genuine error is preferred over `TransferAbortError` when both occurred.
|
||||||
|
*/
|
||||||
|
async function settleWorkers(workers: Promise<unknown>[]): Promise<void> {
|
||||||
|
const results = await Promise.allSettled(workers);
|
||||||
|
const errors = results
|
||||||
|
.filter((r): r is PromiseRejectedResult => r.status === 'rejected')
|
||||||
|
.map((r) => r.reason);
|
||||||
|
if (errors.length === 0) return;
|
||||||
|
throw errors.find((e) => !(e instanceof TransferAbortError)) ?? errors[0];
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user