mirror of
https://github.com/by-sonic/tglock.git
synced 2026-11-05 15:21:55 +03:00
fix(worker): type-clean frame handling; name a Telegram close before any reply (#67)
Cloudflare's dashboard editor type-checks the script and underlined four lines of the Blob handling, because a separate `blob` flag did not narrow `event.data` (#66). binaryFrame() narrows each branch with instanceof; the file now type-checks cleanly against current @cloudflare/workers-types. When Telegram closes TCP without sending a byte, the Worker used to end the WebSocket with a plain 1000. It now closes with 1011 "Telegram closed before replying" and logs the destination and the bytes the client sent, so the client log and the Cloudflare log both say where the tunnel broke (#53). Co-authored-by: babin <Thartewerner536e@engineer.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
+46
-13
@@ -34,6 +34,32 @@ const CONNECT_TIMEOUT_MS = 3000;
|
|||||||
const MAX_PENDING_BYTES = 1024 * 1024;
|
const MAX_PENDING_BYTES = 1024 * 1024;
|
||||||
const MAX_PENDING_MESSAGES = 256;
|
const MAX_PENDING_MESSAGES = 256;
|
||||||
|
|
||||||
|
// Размер бинарного кадра и способ достать его байты; null — текстовый кадр.
|
||||||
|
//
|
||||||
|
// С compatibility date 2026-03-17 бинарные кадры приходят как Blob, если
|
||||||
|
// binaryType не "arraybuffer". Свойство выставлено при accept(); Blob всё
|
||||||
|
// равно принимается — на случай рантайма, который его не учитывает. Раньше
|
||||||
|
// такой кадр закрывал туннель кодом 1003 сразу после рукопожатия (#61).
|
||||||
|
//
|
||||||
|
// Каждая ветка проверяет тип напрямую, чтобы проверка типов в редакторе
|
||||||
|
// Cloudflare не подчёркивала доступ к полям (by-sonic/tglock#66).
|
||||||
|
/** @param {unknown} data */
|
||||||
|
function binaryFrame(data) {
|
||||||
|
if (typeof Blob !== "undefined" && data instanceof Blob) {
|
||||||
|
const blob = data;
|
||||||
|
return { size: blob.size, read: async () => new Uint8Array(await blob.arrayBuffer()) };
|
||||||
|
}
|
||||||
|
if (data instanceof ArrayBuffer) {
|
||||||
|
const bytes = new Uint8Array(data);
|
||||||
|
return { size: bytes.byteLength, read: async () => bytes };
|
||||||
|
}
|
||||||
|
if (ArrayBuffer.isView(data)) {
|
||||||
|
const bytes = new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
|
||||||
|
return { size: bytes.byteLength, read: async () => bytes };
|
||||||
|
}
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
export default {
|
export default {
|
||||||
async fetch(request, env = {}) {
|
async fetch(request, env = {}) {
|
||||||
const url = new URL(request.url);
|
const url = new URL(request.url);
|
||||||
@@ -114,23 +140,16 @@ export default {
|
|||||||
let pending = Promise.resolve();
|
let pending = Promise.resolve();
|
||||||
let pendingBytes = 0;
|
let pendingBytes = 0;
|
||||||
let pendingMessages = 0;
|
let pendingMessages = 0;
|
||||||
|
let uploaded = 0;
|
||||||
|
|
||||||
server.addEventListener("message", (event) => {
|
server.addEventListener("message", (event) => {
|
||||||
if (closed) return;
|
if (closed) return;
|
||||||
const data = event.data;
|
const frame = binaryFrame(event.data);
|
||||||
// С compatibility date 2026-03-17 бинарные кадры приходят как Blob, если
|
if (!frame) {
|
||||||
// binaryType не "arraybuffer". Свойство выставлено выше; Blob всё равно
|
|
||||||
// принимается — на случай рантайма, который его не учитывает. Раньше такой
|
|
||||||
// кадр закрывал туннель кодом 1003 сразу после рукопожатия.
|
|
||||||
const blob = typeof Blob !== "undefined" && data instanceof Blob;
|
|
||||||
if (!blob && !(data instanceof ArrayBuffer) && !ArrayBuffer.isView(data)) {
|
|
||||||
shutdown(1003, "binary messages required");
|
shutdown(1003, "binary messages required");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const chunk = blob ? null : data instanceof ArrayBuffer
|
const size = frame.size;
|
||||||
? new Uint8Array(data)
|
|
||||||
: new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
|
|
||||||
const size = blob ? data.size : chunk.byteLength;
|
|
||||||
if (pendingBytes + size > MAX_PENDING_BYTES || pendingMessages >= MAX_PENDING_MESSAGES) {
|
if (pendingBytes + size > MAX_PENDING_BYTES || pendingMessages >= MAX_PENDING_MESSAGES) {
|
||||||
shutdown(1009, "Telegram write queue full");
|
shutdown(1009, "Telegram write queue full");
|
||||||
return;
|
return;
|
||||||
@@ -140,9 +159,12 @@ export default {
|
|||||||
pending = pending
|
pending = pending
|
||||||
.then(async () => {
|
.then(async () => {
|
||||||
if (closed) return;
|
if (closed) return;
|
||||||
const bytes = chunk || new Uint8Array(await data.arrayBuffer());
|
const bytes = await frame.read();
|
||||||
await writer.ready;
|
await writer.ready;
|
||||||
if (!closed) await writer.write(bytes);
|
if (!closed) {
|
||||||
|
await writer.write(bytes);
|
||||||
|
uploaded += bytes.byteLength;
|
||||||
|
}
|
||||||
})
|
})
|
||||||
.catch(() => shutdown(1011, "Telegram write failed"))
|
.catch(() => shutdown(1011, "Telegram write failed"))
|
||||||
.finally(() => {
|
.finally(() => {
|
||||||
@@ -157,10 +179,12 @@ export default {
|
|||||||
// Обратное направление: всё, что приходит от Telegram, уходит клиенту.
|
// Обратное направление: всё, что приходит от Telegram, уходит клиенту.
|
||||||
(async () => {
|
(async () => {
|
||||||
const reader = upstream.readable.getReader();
|
const reader = upstream.readable.getReader();
|
||||||
|
let downloaded = 0;
|
||||||
try {
|
try {
|
||||||
for (;;) {
|
for (;;) {
|
||||||
const { value, done } = await reader.read();
|
const { value, done } = await reader.read();
|
||||||
if (done) break;
|
if (done) break;
|
||||||
|
downloaded += value.byteLength;
|
||||||
server.send(value);
|
server.send(value);
|
||||||
}
|
}
|
||||||
} catch {
|
} catch {
|
||||||
@@ -168,6 +192,15 @@ export default {
|
|||||||
} finally {
|
} finally {
|
||||||
reader.releaseLock();
|
reader.releaseLock();
|
||||||
}
|
}
|
||||||
|
// Telegram закрыл TCP, не ответив ни байтом. Раньше это было обычное
|
||||||
|
// закрытие 1000 без причины, и в журнале клиента не было видно, кто и на
|
||||||
|
// каком шаге оборвал туннель (by-sonic/tglock#53). Строка уходит и в
|
||||||
|
// журнал Worker'а в панели Cloudflare.
|
||||||
|
if (!downloaded && !closed) {
|
||||||
|
console.warn(`Telegram ${destination} closed TCP before replying (client sent ${uploaded} bytes)`);
|
||||||
|
shutdown(1011, "Telegram closed before replying");
|
||||||
|
return;
|
||||||
|
}
|
||||||
shutdown();
|
shutdown();
|
||||||
})();
|
})();
|
||||||
|
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ async function fixture({ opened = Promise.resolve(), write, timeout = false } =
|
|||||||
const closed = deferred();
|
const closed = deferred();
|
||||||
const writes = [];
|
const writes = [];
|
||||||
const events = new Map();
|
const events = new Map();
|
||||||
const state = { connects: 0, accepted: false, socketCloses: 0, sent: [], closes: [] };
|
const state = { connects: 0, accepted: false, socketCloses: 0, sent: [], closes: [], warnings: [] };
|
||||||
const server = {
|
const server = {
|
||||||
accept() { state.accepted = true; state.binaryTypeOnAccept = this.binaryType; },
|
accept() { state.accepted = true; state.binaryTypeOnAccept = this.binaryType; },
|
||||||
addEventListener(name, callback) { events.set(name, callback); },
|
addEventListener(name, callback) { events.set(name, callback); },
|
||||||
@@ -41,7 +41,7 @@ async function fixture({ opened = Promise.resolve(), write, timeout = false } =
|
|||||||
},
|
},
|
||||||
};
|
};
|
||||||
const context = vm.createContext({
|
const context = vm.createContext({
|
||||||
URL, Uint8Array, ArrayBuffer, Blob, setTimeout: timeout ? (fn) => setTimeout(fn, 0) : setTimeout, clearTimeout,
|
URL, Uint8Array, ArrayBuffer, Blob, console: { warn: (line) => state.warnings.push(line) }, setTimeout: timeout ? (fn) => setTimeout(fn, 0) : setTimeout, clearTimeout,
|
||||||
Response: class {
|
Response: class {
|
||||||
constructor(body, options) { this.body = body; Object.assign(this, options); }
|
constructor(body, options) { this.body = body; Object.assign(this, options); }
|
||||||
},
|
},
|
||||||
@@ -208,6 +208,20 @@ test("Telegram bytes reach WebSocket unchanged and EOF closes both directions",
|
|||||||
assert.equal(f.state.socketCloses, 1);
|
assert.equal(f.state.socketCloses, 1);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("Telegram closing TCP before any reply is named, not a silent 1000", async () => {
|
||||||
|
// A tunnel that never carried a byte from Telegram used to end as a plain
|
||||||
|
// 1000 close, which hid where it broke (#53). The reason also reaches the
|
||||||
|
// Worker log in the Cloudflare dashboard.
|
||||||
|
const f = await fixture();
|
||||||
|
await f.fetch();
|
||||||
|
f.message(new Uint8Array([1, 2, 3]).buffer);
|
||||||
|
await tick();
|
||||||
|
f.eof();
|
||||||
|
for (let i = 0; i < 5; i++) await tick();
|
||||||
|
assert.deepEqual(f.state.closes[0], { code: 1011, reason: "Telegram closed before replying" });
|
||||||
|
assert.match(f.state.warnings[0], /149\.154\.167\.51 closed TCP before replying \(client sent 3 bytes\)/);
|
||||||
|
});
|
||||||
|
|
||||||
test("CDN worker keeps the requested DC203 destination", async () => {
|
test("CDN worker keeps the requested DC203 destination", async () => {
|
||||||
const f = await fixture();
|
const f = await fixture();
|
||||||
await f.fetch("/apiws?dst=91.105.192.100&dc=203");
|
await f.fetch("/apiws?dst=91.105.192.100&dc=203");
|
||||||
|
|||||||
Reference in New Issue
Block a user