diff --git a/packages/transport-http/src/transport.test.ts b/packages/transport-http/src/transport.test.ts index 4b2168c5..f1712964 100644 --- a/packages/transport-http/src/transport.test.ts +++ b/packages/transport-http/src/transport.test.ts @@ -1,7 +1,26 @@ -import { describe, vi, expect, it, beforeEach, afterEach } from "vitest"; +import { describe, vi, expect, it, beforeEach, afterEach, type MockInstance } from "vitest"; import { runTransportContract } from "../../../tests/utils/transportContract"; import { TransportHTTP } from "./transport"; +let abortTimeoutSpy: MockInstance | undefined; +beforeEach(() => { + abortTimeoutSpy = vi.spyOn( + globalThis.AbortSignal as unknown as { timeout(ms: number): AbortSignal }, + "timeout", + ).mockImplementation((ms: number) => { + const ctrl = new AbortController(); + const abort = () => + ctrl.abort(new DOMException("Timeout reached", "TimeoutError")); + // Uses setTimeout so vi.useFakeTimers() can fast-forward it + setTimeout(abort, ms); + return ctrl.signal; + }); +}); + +afterEach(() => { + abortTimeoutSpy?.mockRestore(); +}); + function stubFetch() { const inbox: Uint8Array[] = []; let lastWritten: ArrayBuffer | undefined; diff --git a/packages/transport-http/src/transport.ts b/packages/transport-http/src/transport.ts index 7aa29c61..d13b7b26 100644 --- a/packages/transport-http/src/transport.ts +++ b/packages/transport-http/src/transport.ts @@ -2,7 +2,6 @@ import { Types } from "@meshtastic/core"; const FETCH_INTERVAL_MS = 3000; const READ_TIMEOUT_MS = 7000; -const READ_TIMEOUT_BUFFER_MS = 250; const WRITE_TIMEOUT_MS = 4000; function toArrayBuffer(uint8array: Uint8Array): ArrayBuffer { @@ -16,38 +15,6 @@ function toArrayBuffer(uint8array: Uint8Array): ArrayBuffer { return uint8array.slice().buffer; } -function createTimeoutController(ms: number): { - controller: AbortController; - clear: () => void; -} { - const controller = new AbortController(); - const id = setTimeout(() => controller.abort(), ms); - const clear = () => clearTimeout(id); - return { controller, clear }; -} - -function composeAbortSignals( - a: AbortSignal, - b: AbortSignal, -): { signal: AbortSignal; cleanup: () => void } { - const controller = new AbortController(); - const abortIfNeeded = () => controller.abort(); - - if (a.aborted || b.aborted) { - controller.abort(); - } - - a.addEventListener("abort", abortIfNeeded); - b.addEventListener("abort", abortIfNeeded); - - const cleanup = () => { - a.removeEventListener("abort", abortIfNeeded); - b.removeEventListener("abort", abortIfNeeded); - }; - - return { signal: controller.signal, cleanup }; -} - /** * Provides HTTP(S) transport for Meshtastic devices. * @@ -108,7 +75,8 @@ export class TransportHTTP implements Types.Transport { if (!this._closingByUser) { this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, - error instanceof DOMException && error.name === "AbortError" + error instanceof DOMException && + (error.name === "AbortError" || error.name === "TimeoutError") ? "write-timeout" : "write-error", ); @@ -133,15 +101,6 @@ export class TransportHTTP implements Types.Transport { this.interval = setInterval(async () => { if (this.fetching) { - if ( - this._inflightReadController && - Date.now() - this._inflightReadStartedAt > - READ_TIMEOUT_MS + READ_TIMEOUT_BUFFER_MS - ) { - try { - this._inflightReadController.abort(); - } catch {} - } return; } @@ -152,7 +111,8 @@ export class TransportHTTP implements Types.Transport { if (!this._closingByUser) { this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, - error instanceof DOMException && error.name === "AbortError" + error instanceof DOMException && + (error.name === "AbortError" || error.name === "TimeoutError") ? "read-timeout" : "read-error", ); @@ -172,13 +132,10 @@ export class TransportHTTP implements Types.Transport { this._inflightReadController = inflight; this._inflightReadStartedAt = Date.now(); - const { controller: timeoutCtrl, clear } = - createTimeoutController(READ_TIMEOUT_MS); - - const { signal, cleanup } = composeAbortSignals( + const signal = AbortSignal.any([ inflight.signal, - timeoutCtrl.signal, - ); + AbortSignal.timeout(READ_TIMEOUT_MS), + ]); try { const response = await fetch( @@ -208,8 +165,6 @@ export class TransportHTTP implements Types.Transport { }); } } finally { - cleanup(); - clear(); this._inflightReadController = undefined; } } @@ -217,15 +172,12 @@ export class TransportHTTP implements Types.Transport { /** Write a protobuf-encoded request to `/api/v1/toradio`. */ private async writeToRadio(data: Uint8Array): Promise { - const { controller: timeoutCtrl, clear } = - createTimeoutController(WRITE_TIMEOUT_MS); - try { const response = await fetch(`${this.url}/api/v1/toradio`, { method: "PUT", headers: { "Content-Type": "application/x-protobuf" }, body: toArrayBuffer(data), - signal: timeoutCtrl.signal, + signal: AbortSignal.timeout(WRITE_TIMEOUT_MS), }); if (!response.ok) { throw new Error(`toradio ${response.status} ${response.statusText}`); @@ -234,14 +186,13 @@ export class TransportHTTP implements Types.Transport { if (!this._closingByUser) { this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, - error instanceof DOMException && error.name === "AbortError" + error instanceof DOMException && + (error.name === "AbortError" || error.name === "TimeoutError") ? "write-timeout" : "write-error", ); } throw error; - } finally { - clear(); } } diff --git a/packages/transport-node/src/transport.ts b/packages/transport-node/src/transport.ts index 8a0f259a..4b011bdb 100644 --- a/packages/transport-node/src/transport.ts +++ b/packages/transport-node/src/transport.ts @@ -94,13 +94,18 @@ export class TransportNode implements Types.Transport { ctrl.enqueue(value); } } - } catch { + ctrl.close(); + } catch (error) { if (!this._closingByUser) { this.emitStatus( Types.DeviceStatusEnum.DeviceDisconnected, "read-error", ); } + ctrl.error(error instanceof Error ? error : new Error(String(error))); + try { + await transformed.cancel(); + } catch {} } finally { reader.releaseLock(); }