From 55aa0ccf53773e20af02d1956d2bcd42f35fcddf Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 15:17:28 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8C=89=20K-00=20=E7=BA=A6=E5=AE=9A?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20JS=20SDK=20=E6=96=AD=E7=BA=BF=E9=87=8D?= =?UTF-8?q?=E4=BA=A4=E4=B8=8E=E9=80=80=E9=81=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/DEVIATIONS.md | 9 + sdk/js/example/minimal.mjs | 2 +- sdk/js/src/client.ts | 509 ++++++++++++++++++++++++++++--------- sdk/js/src/fake.ts | 31 ++- sdk/js/src/index.ts | 7 +- sdk/js/src/mqtt.ts | 68 +++-- sdk/js/src/types.ts | 109 +++++--- sdk/js/test/client.test.ts | 3 +- sdk/js/test/k00.test.ts | 269 ++++++++++++++++++++ sdk/js/test/k02.test.ts | 226 ++++++++++++++++ 10 files changed, 1049 insertions(+), 184 deletions(-) create mode 100644 sdk/js/test/k00.test.ts create mode 100644 sdk/js/test/k02.test.ts diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 5abc191..c9d50bd 100644 --- a/docs/DEVIATIONS.md +++ b/docs/DEVIATIONS.md @@ -1171,6 +1171,15 @@ - 备选方案：照搬旧 S1.3 / JS 双重翻倍；否决。 - 影响：仅 sdk/go。 +### 复审修复 K-02 + +- 日期：2026-09-30 +- 原条款：issue #59 及第二轮补充；DEVELOPMENT 第 9 节附录。 +- 实际做法：ReconnectBackoff 只维护计数 n，mqtt 不再 attempt++ 与 markOffline 双重翻倍；断线后在途发送新 rid 重交；回调成功即置 acked 再发 ack；publishUp 不用 Buffer；keepalive 30；顶号 `taken_over`；0x8B 可重试；connect 超时不永久关闭 Client；logout 把请求失败返回应用；假传输不从 index 导出；示例只打印令牌前缀。未加 Playwright 浏览器冒烟（sdk/js 无该依赖，单测覆盖无 Buffer 的字符串上行）。 +- 原因：与 K-00 对齐并修 Node 未处理拒绝、浏览器 Buffer、断线永不重交。 +- 备选方案：沿用 attempt+base 双重翻倍；否决。 +- 影响：仅 sdk/js。 + ## SDK 二 S2 ### S2-PY/JAVA 1–3 2026-09-30 diff --git a/sdk/js/example/minimal.mjs b/sdk/js/example/minimal.mjs index cf5dc3b..52d039c 100644 --- a/sdk/js/example/minimal.mjs +++ b/sdk/js/example/minimal.mjs @@ -6,7 +6,7 @@ const pass = process.env.NIXMSG_PASSWORD || "secret"; const peer = process.env.NIXMSG_PEER || "device-2"; const c = new Client(); -c.onSessionHandler((tok) => console.log("session", tok)); +c.onSessionHandler((tok) => console.log("session", String(tok).slice(0, 8))); c.onMessageHandler((msg) => console.log("msg", msg.from, msg.body.data)); c.onConnectionHandler((ev) => console.log("conn", ev.state, ev.reason)); diff --git a/sdk/js/src/client.ts b/sdk/js/src/client.ts index 3752427..bd0cbb1 100644 --- a/sdk/js/src/client.ts +++ b/sdk/js/src/client.ts @@ -22,7 +22,9 @@ import { SendResult, Target, Transport, + applyJitter, marshalJSON, + nominalDelay, registerURLFromConnect, } from "./types.js"; import { MqttTransport } from "./mqtt.js"; @@ -33,15 +35,54 @@ type RespFrame = { error?: { code: string; message: string }; }; +type Pending = { + resolve: (r: RespFrame) => void; + reject: (e: unknown) => void; + isSend?: boolean; + timer?: ReturnType; +}; + type SendItem = { frame: Record; payload: string; id: string; result: { resolve: (v: SendResult) => void; reject: (e: unknown) => void }; inflight: boolean; + epoch: number; + rateN: number; + abandoned?: boolean; + abort?: () => void; }; -type DedupState = "delivered" | "acked"; +type DedupState = "delivered" | "acked" | "revoked"; + +class LRUMap { + private map = new Map(); + constructor(private cap: number) {} + get(k: string): T | undefined { + const v = this.map.get(k); + if (v !== undefined) { + this.map.delete(k); + this.map.set(k, v); + } + return v; + } + has(k: string): boolean { + return this.map.has(k); + } + put(k: string, v: T): void { + if (this.map.has(k)) this.map.delete(k); + this.map.set(k, v); + while (this.map.size > this.cap) { + const first = this.map.keys().next().value as string | undefined; + if (first === undefined) break; + this.map.delete(first); + } + } + delete(k: string): void { + this.map.delete(k); + } +} export class Client { private opts: Required< @@ -70,12 +111,14 @@ export class Client { private backoff = new ReconnectBackoff(); private endpointId = ""; private ridSeq = 0; - private pending = new Map void; reject: (e: unknown) => void }>(); + private pending = new Map(); private sendQ: SendItem[] = []; private inflight = 0; private handshook = false; private stopReconnect = false; private closed = false; + private lastStopCode = ""; + private lastStopErr?: APIError; private state: ConnectionState = "offline"; private limits: HandshakeLimits = { server_time_ms: 0, @@ -88,10 +131,10 @@ export class Client { ack_timeout_seconds: 300, }; private clockSkew = 0; - private dedup = new Map(); - private dedupOrd: string[] = []; - private receiptSeen = new Set(); + private store = new LRUMap(10000); private cbChain: Promise = Promise.resolve(); + private timers = new Set>(); + private watch?: { ids?: string[]; all?: boolean }; private onSession?: (token: string) => void; private onMessage?: (msg: Message) => void | Promise; @@ -123,7 +166,6 @@ export class Client { this.onConnection = h; } - /** 别名，贴近文档 onSession(handler) */ onSessionCb(h: (token: string) => void): void { this.onSessionHandler(h); } @@ -136,6 +178,14 @@ export class Client { return { ...this.limits }; } + lastStopCodeForTest(): string { + return this.lastStopCode; + } + + resendPayloadForTest(): string | undefined { + return this.sendQ[0]?.payload; + } + async connect( url: string, endpointId: string, @@ -144,6 +194,9 @@ export class Client { ): Promise { if (this.closed) throw new APIError("closed", "已关闭"); if (this.transport) throw new APIError("bad_request", "已在连接中"); + if (options.maxReceiveBytes != null && options.maxReceiveBytes > 0 && options.maxReceiveBytes < 1024) { + throw new APIError("bad_request", "max_receive_bytes 小于 1024"); + } this.opts = { ...this.opts, ...options, @@ -155,9 +208,12 @@ export class Client { manualAck: options.manualAck ?? false, allowTcp: options.allowTcp ?? false, }; + this.store = new LRUMap(this.opts.dedupCapacity!); this.endpointId = endpointId; this.stopReconnect = false; this.handshook = false; + this.lastStopCode = ""; + this.lastStopErr = undefined; this.backoff = new ReconnectBackoff(); const pass = credential.sessionToken ?? credential.password ?? ""; const tr = options.transport ?? new MqttTransport(); @@ -165,48 +221,106 @@ export class Client { tr.setCredential(pass); this.setState("connecting"); - await tr.start({ - url, - endpointId, - connectTimeoutMs: this.opts.connectTimeoutMs!, - allowTcp: !!this.opts.allowTcp, - backoff: this.backoff, - onDown: (p) => this.handleDown(p), - onOffline: () => { - this.handshook = false; - if (!this.stopReconnect && !this.closed) this.setState("reconnecting"); - }, - onAuthFailed: (r) => this.failAuth(r), - onKicked: () => this.failKicked(), - mqttReady: () => this.doHello(), - }); + void tr + .start({ + url, + endpointId, + connectTimeoutMs: this.opts.connectTimeoutMs!, + allowTcp: !!this.opts.allowTcp, + backoff: this.backoff, + onDown: (p) => this.handleDown(p), + onOffline: () => this.handleOffline(), + onAuthFailed: (r) => this.failAuth(r), + onKicked: () => this.failKicked(), + mqttReady: () => this.doHello(), + }) + .catch(() => {}); const deadline = Date.now() + this.opts.connectTimeoutMs!; while (Date.now() < deadline) { if (this.handshook) return; if (this.stopReconnect || this.state === "auth_failed" || this.state === "kicked") { - throw new APIError("auth_failed", this.state); + throw this.lastStopErr ?? new APIError(this.lastStopCode || "auth_failed", this.state); } await sleep(20); } - await this.close(); + await this.teardown(false, new APIError("not_connected", "连接超时")); throw new APIError("not_connected", "连接超时"); } private failAuth(reason: AuthReason): void { + if (this.stopReconnect && this.lastStopCode) return; + const err = new APIError(reason, "认证失败，停止重连"); this.stopReconnect = true; this.handshook = false; + this.lastStopCode = reason; + this.lastStopErr = err; this.setState("auth_failed", reason); - this.failQueued(new APIError(reason, "认证失败，停止重连")); + this.failQueued(err); + this.failPending(err, true); void this.transport?.stop(); + this.transport = undefined; } private failKicked(): void { + if (this.stopReconnect && this.lastStopCode === "taken_over") return; + const err = new APIError("taken_over", "被顶号，停止重连"); this.stopReconnect = true; this.handshook = false; - this.setState("kicked", "0x8E"); - this.failQueued(new APIError("kicked", "被顶号，停止重连")); + this.lastStopCode = "taken_over"; + this.lastStopErr = err; + this.setState("kicked", "taken_over"); + this.failQueued(err); + this.failPending(err, true); void this.transport?.stop(); + this.transport = undefined; + } + + private handleFatal(reason: string): void { + if (this.stopReconnect && this.lastStopCode) return; + const code = reason || "fatal"; + const err = new APIError(code, "致命错误，停止重连"); + this.stopReconnect = true; + this.handshook = false; + this.lastStopCode = code; + this.lastStopErr = err; + this.setState("auth_failed", code); + this.failQueued(err); + this.failPending(err, true); + void this.transport?.stop(); + this.transport = undefined; + } + + private handleOffline(): void { + this.handshook = false; + if (this.stopReconnect || this.closed) { + this.failPending(new APIError("not_connected", "未连接"), true); + return; + } + this.requeueInflight(); + this.failPending(new APIError("not_connected", "未连接"), false); + this.setState("reconnecting"); + } + + private requeueInflight(): void { + for (const it of this.sendQ) { + if (!it.inflight) continue; + it.epoch++; + it.inflight = false; + const rid = String(it.frame.rid ?? ""); + const p = this.pending.get(rid); + if (p) { + this.clearPendingTimer(p); + this.pending.delete(rid); + } + this.regenerateSend(it); + } + this.inflight = 0; + } + + private regenerateSend(it: SendItem): void { + it.frame.rid = this.nextRid(); + it.payload = marshalJSON(it.frame); } private setState(state: ConnectionState, reason?: string): void { @@ -229,6 +343,13 @@ export class Client { return String(this.ridSeq); } + private stopErr(): APIError { + if (this.lastStopErr) return this.lastStopErr; + if (this.lastStopCode) return new APIError(this.lastStopCode); + if (this.closed) return new APIError("closed", "已关闭"); + return new APIError("not_connected"); + } + private async doHello(): Promise { const sentAt = Date.now(); const req: Record = { @@ -261,9 +382,24 @@ export class Client { this.transport?.setCredential(token); this.enqueueCb(() => this.onSession?.(token)); } + if (this.watch) { + void this.restoreWatch(); + } void this.drainSendQueue(); } + private async restoreWatch(): Promise { + if (!this.watch) return; + const req: Record = { v: 1, type: "presence.watch", rid: this.nextRid() }; + if (this.watch.all) req.all = true; + else req.ids = this.watch.ids ?? []; + try { + await this.request(req, false); + } catch { + /* 重连后尽力恢复 */ + } + } + private handleDown(payload: Uint8Array): void { let head: { type?: string; rid?: string }; try { @@ -277,16 +413,17 @@ export class Client { const rf = JSON.parse(text) as RespFrame & { rid: string }; const p = this.pending.get(rf.rid ?? head.rid!); if (p) { + this.clearPendingTimer(p); this.pending.delete(rf.rid ?? head.rid!); p.resolve(rf); } break; } case "msg": - void this.handleMsg(JSON.parse(text)); + void this.handleMsg(JSON.parse(text)).catch(() => {}); break; case "receipt": - void this.handleReceipt(JSON.parse(text)); + void this.handleReceipt(JSON.parse(text)).catch(() => {}); break; case "revoked": this.handleRevoked(JSON.parse(text)); @@ -303,10 +440,7 @@ export class Client { } case "fatal": { const f = JSON.parse(text) as { reason?: string }; - this.stopReconnect = true; - this.setState("auth_failed", f.reason); - this.failQueued(new APIError("fatal", f.reason ?? "")); - void this.transport?.stop(); + this.handleFatal(f.reason ?? "fatal"); break; } } @@ -320,59 +454,58 @@ export class Client { meta?: Record; send_at_ms: number; }): Promise { - const key = `${m.from}\0${m.id}`; - const ent = this.dedup.get(key); - if (ent === "acked") { - await this.sendAckFrame(m.from, m.id); - return; - } - if (ent === "delivered") return; - this.rememberDedup(key, "delivered"); - - const msg: Message = { - id: m.id, - from: m.from, - to: m.to, - body: m.body, - meta: m.meta, - send_at_ms: m.send_at_ms, - }; - - let cbErr: unknown; - await new Promise((resolve) => { - this.enqueueCb(async () => { - try { - await this.onMessage?.(msg); - } catch (e) { - cbErr = e; - } - resolve(); - }); - }); - - if (this.opts.manualAck) return; - if (cbErr) { - this.dedup.delete(key); - return; - } - await this.sendAckFrame(m.from, m.id); - this.dedup.set(key, "acked"); - } - - private rememberDedup(key: string, st: DedupState): void { - if (!this.dedup.has(key)) { - this.dedupOrd.push(key); - while (this.dedupOrd.length > this.opts.dedupCapacity!) { - const old = this.dedupOrd.shift()!; - this.dedup.delete(old); + try { + const key = `m\0${m.from}\0${m.id}`; + const ent = this.store.get(key); + if (ent === "acked") { + void this.sendAckFrame(m.from, m.id).catch(() => {}); + return; } + if (ent === "delivered" || ent === "revoked") return; + this.store.put(key, "delivered"); + + const msg: Message = { + id: m.id, + from: m.from, + to: m.to, + body: m.body, + meta: m.meta, + send_at_ms: m.send_at_ms, + }; + + let cbErr: unknown; + await new Promise((resolve) => { + this.enqueueCb(async () => { + if (this.store.get(key) === "revoked") { + resolve(); + return; + } + try { + await this.onMessage?.(msg); + } catch (e) { + cbErr = e; + } + resolve(); + }); + }); + + if (this.store.get(key) === "revoked") return; + if (this.opts.manualAck) return; + if (cbErr) { + this.store.delete(key); + return; + } + // S-06：回调成功即置 acked，再发 ack + this.store.put(key, "acked"); + void this.sendAckFrame(m.from, m.id).catch(() => {}); + } catch { + /* S-04：确认路径不得产生未处理拒绝 */ } - this.dedup.set(key, st); } async ack(msg: Message): Promise { await this.sendAckFrame(msg.from, msg.id); - this.rememberDedup(`${msg.from}\0${msg.id}`, "acked"); + this.store.put(`m\0${msg.from}\0${msg.id}`, "acked"); } private async sendAckFrame(from: string, id: string): Promise { @@ -389,13 +522,18 @@ export class Client { } private async handleReceipt(r: Receipt & { receipt_id: string }): Promise { - if (this.receiptSeen.has(r.receipt_id)) { - await this.sendReceiptAck(r.receipt_id); - return; + try { + const key = `r\0${r.receipt_id}`; + if (this.store.has(key)) { + void this.sendReceiptAck(r.receipt_id).catch(() => {}); + return; + } + this.store.put(key, "acked"); + this.enqueueCb(() => this.onReceipt?.(r)); + void this.sendReceiptAck(r.receipt_id).catch(() => {}); + } catch { + /* S-04 */ } - this.receiptSeen.add(r.receipt_id); - this.enqueueCb(() => this.onReceipt?.(r)); - await this.sendReceiptAck(r.receipt_id); } private async sendReceiptAck(receiptId: string): Promise { @@ -403,10 +541,10 @@ export class Client { } private handleRevoked(r: RevokedEvent): void { - const key = `${r.from}\0${r.id}`; - const ent = this.dedup.get(key); - if (!ent || ent === "acked") return; - this.dedup.delete(key); + const key = `m\0${r.from}\0${r.id}`; + const ent = this.store.get(key); + if (ent === "acked" || ent === "revoked") return; + this.store.put(key, "revoked"); this.enqueueCb(() => this.onRevoked?.(r)); } @@ -419,24 +557,39 @@ export class Client { frame.rid = rid; const payload = marshalJSON(frame); return new Promise((resolve, reject) => { - this.pending.set(rid, { + const p: Pending = { resolve: (rf) => { + this.clearPendingTimer(p); if (!rf.ok) { reject(new APIError(rf.error?.code ?? "bad_request", rf.error?.message ?? "")); return; } resolve(rf.data); }, - reject, - }); + reject: (e) => { + this.clearPendingTimer(p); + reject(e); + }, + isSend: false, + }; + p.timer = setTimeout(() => { + this.pending.delete(rid); + p.reject(new APIError("not_connected", "请求超时")); + }, 60_000); + p.timer.unref?.(); + this.timers.add(p.timer); + this.pending.set(rid, p); void tr.publishUp(payload).catch((e) => { this.pending.delete(rid); - reject(e); + this.clearPendingTimer(p); + reject(e instanceof APIError ? e : new APIError("not_connected", String(e))); }); }); } async send(to: Target, body: Body, opt: SendOptions = {}): Promise { + if (this.closed || this.stopReconnect) throw this.stopErr(); + const enc = body.enc || "utf8"; const b: Body = { enc, @@ -477,65 +630,119 @@ export class Client { const payload = marshalJSON(frame); const maxFrame = this.limits.max_frame_bytes || 786432; - if (this.handshook && payload.length > maxFrame) { + if (utf8Len(payload) > maxFrame) { throw new APIError("frame_too_large", "整帧超限"); } - if (this.closed || this.stopReconnect) throw new APIError("closed"); if (this.sendQ.length >= this.opts.sendQueueSize!) { throw new APIError("queue_full", "发送队列已满"); } return new Promise((resolve, reject) => { - this.sendQ.push({ frame, payload, id, result: { resolve, reject }, inflight: false }); + const item: SendItem = { + frame, + payload, + id, + result: { resolve, reject }, + inflight: false, + epoch: 0, + rateN: 0, + }; + if (opt.signal) { + const onAbort = () => { + if (item.inflight) { + item.abandoned = true; + reject(new APIError("result_unknown", "结果未知，请用同一消息号重试")); + return; + } + this.sendQ = this.sendQ.filter((x) => x !== item); + reject(new APIError("closed", "已取消")); + }; + if (opt.signal.aborted) { + onAbort(); + return; + } + opt.signal.addEventListener("abort", onAbort, { once: true }); + item.abort = () => opt.signal?.removeEventListener("abort", onAbort); + } + this.sendQ.push(item); void this.drainSendQueue(); }); } private async drainSendQueue(): Promise { while (true) { - if (!this.handshook || !this.transport) return; - const next = this.sendQ.find((x) => !x.inflight); + if (!this.handshook || !this.transport || this.stopReconnect) return; + const maxFrame = this.limits.max_frame_bytes || 786432; + const next = this.sendQ.find((x) => !x.inflight && !x.abandoned); if (!next || this.inflight >= this.opts.maxInflight!) return; + if (utf8Len(next.payload) > maxFrame) { + this.finishSendErr(next, new APIError("frame_too_large", "整帧超限")); + continue; + } + next.epoch++; next.inflight = true; this.inflight++; - void this.dispatchSend(next); + void this.dispatchSend(next, next.epoch); } } - private async dispatchSend(item: SendItem): Promise { + private async dispatchSend(item: SendItem, epoch: number): Promise { const rid = String(item.frame.rid); - const tr = this.transport!; + const tr = this.transport; + if (!tr) return; try { const data = await new Promise((resolve, reject) => { - this.pending.set(rid, { + if (item.epoch !== epoch || !item.inflight) { + reject(new Error("stale")); + return; + } + const p: Pending = { resolve: (rf) => { + this.clearPendingTimer(p); if (!rf.ok) { reject(new APIError(rf.error?.code ?? "bad_request", rf.error?.message ?? "")); return; } resolve(rf.data); }, - reject, - }); + reject: (e) => { + this.clearPendingTimer(p); + reject(e); + }, + isSend: true, + }; + this.pending.set(rid, p); void tr.publishUp(item.payload).catch((e) => { this.pending.delete(rid); + this.clearPendingTimer(p); reject(e); }); }); + if (item.epoch !== epoch) return; const sd = (data ?? {}) as SendResult; this.finishSend(item, { id: sd.id || item.id, send_at_ms: sd.send_at_ms, state: sd.state }); } catch (e) { + if (item.epoch !== epoch) return; if (e instanceof APIError && e.code === "rate_limited") { item.inflight = false; this.inflight = Math.max(0, this.inflight - 1); this.pending.delete(rid); - setTimeout(() => void this.drainSendQueue(), 1000); + item.rateN++; + this.regenerateSend(item); + const wait = applyJitter(nominalDelay(item.rateN)); + this.schedule(() => void this.drainSendQueue(), wait); return; } - // 网络错误：保留队列 + if (e instanceof APIError && e.code === "_retry") return; if (!(e instanceof APIError)) { + if (this.stopReconnect) { + this.finishSendErr(item, this.stopErr()); + return; + } item.inflight = false; this.inflight = Math.max(0, this.inflight - 1); + this.pending.delete(rid); + this.regenerateSend(item); return; } this.finishSendErr(item, e); @@ -548,7 +755,10 @@ export class Client { this.inflight = Math.max(0, this.inflight - 1); item.inflight = false; } - item.result.resolve(res); + const rid = String(item.frame.rid ?? ""); + this.pending.delete(rid); + item.abort?.(); + if (!item.abandoned) item.result.resolve(res); void this.drainSendQueue(); } @@ -558,16 +768,71 @@ export class Client { this.inflight = Math.max(0, this.inflight - 1); item.inflight = false; } - item.result.reject(err); + const rid = String(item.frame.rid ?? ""); + this.pending.delete(rid); + item.abort?.(); + if (!item.abandoned) item.result.reject(err); void this.drainSendQueue(); } private failQueued(err: unknown): void { - for (const it of this.sendQ) it.result.reject(err); + for (const it of this.sendQ) { + it.abort?.(); + if (!it.abandoned) it.result.reject(err); + } this.sendQ = []; this.inflight = 0; } + private failPending(err: unknown, all: boolean): void { + for (const [rid, p] of [...this.pending]) { + this.clearPendingTimer(p); + this.pending.delete(rid); + if (!all && p.isSend) { + p.reject(new APIError("_retry", "disconnected")); + continue; + } + p.reject(err); + } + } + + private clearPendingTimer(p: Pending): void { + if (p.timer) { + clearTimeout(p.timer); + this.timers.delete(p.timer); + p.timer = undefined; + } + } + + private schedule(fn: () => void, ms: number): void { + const t = setTimeout(() => { + this.timers.delete(t); + fn(); + }, ms); + t.unref?.(); + this.timers.add(t); + } + + private clearTimers(): void { + for (const t of this.timers) clearTimeout(t); + this.timers.clear(); + } + + private async teardown(setClosed: boolean, stopErr: APIError): Promise { + if (setClosed) this.closed = true; + this.stopReconnect = true; + this.lastStopErr = stopErr; + this.lastStopCode = stopErr.code; + this.failQueued(stopErr); + this.failPending(stopErr, true); + this.handshook = false; + this.clearTimers(); + this.setState("offline", this.lastStopCode); + const tr = this.transport; + this.transport = undefined; + await tr?.stop(); + } + async recall(id: string): Promise { return (await this.request({ v: 1, type: "recall", rid: this.nextRid(), id }, false)) as RecallResult; } @@ -599,6 +864,8 @@ export class Client { } async watchPresence(ids: string[] | "all"): Promise { + if (ids === "all") this.watch = { all: true }; + else this.watch = { ids: [...ids] }; const req: Record = { v: 1, type: "presence.watch", rid: this.nextRid() }; if (ids === "all") req.all = true; else req.ids = ids; @@ -716,21 +983,18 @@ export class Client { } async logout(): Promise { + let err: unknown; try { await this.request({ v: 1, type: "self.logout", rid: this.nextRid() }, false); - } finally { - this.stopReconnect = true; - await this.transport?.stop(); + } catch (e) { + err = e; } + await this.teardown(false, new APIError("logged_out", "已退出")); + if (err) throw err; } async close(): Promise { - this.closed = true; - this.stopReconnect = true; - this.failQueued(new APIError("closed", "已关闭")); - this.setState("offline"); - await this.transport?.stop(); - this.transport = undefined; + await this.teardown(true, new APIError("closed", "已关闭")); } } @@ -742,8 +1006,15 @@ function bodyDecodedLen(b: Body): number { return new TextEncoder().encode(b.data).length; } +function utf8Len(s: string): number { + return new TextEncoder().encode(s).length; +} + function sleep(ms: number): Promise { - return new Promise((r) => setTimeout(r, ms)); + return new Promise((r) => { + const t = setTimeout(r, ms); + t.unref?.(); + }); } export async function register( diff --git a/sdk/js/src/fake.ts b/sdk/js/src/fake.ts index afb4be0..59b3610 100644 --- a/sdk/js/src/fake.ts +++ b/sdk/js/src/fake.ts @@ -11,9 +11,11 @@ export interface FakeConnect { cleanStart: boolean; sessionExpiry: number; password: string; + receiveMaximumSet: boolean; + keepalive: number; } -/** 单测用假 MQTT。 */ +/** 单测用假 MQTT，不从包入口导出。 */ export class FakeTransport implements Transport { private cfg?: TransportConfig; private cred = ""; @@ -25,6 +27,7 @@ export class FakeTransport implements Transport { maxBodyBytes = 262144; maxMetaBytes = 4096; maxFrameBytes = 786432; + publishUpImpl?: (payload: string) => Promise; setCredential(passwordOrToken: string): void { this.cred = passwordOrToken; @@ -32,11 +35,16 @@ export class FakeTransport implements Transport { async start(cfg: TransportConfig): Promise { this.cfg = cfg; - await this.simulateConnectOK(); + queueMicrotask(() => { + void this.simulateConnectOK().catch(() => {}); + }); } async publishUp(payload: Uint8Array | string): Promise { const s = typeof payload === "string" ? payload : new TextDecoder().decode(payload); + if (this.publishUpImpl) { + await this.publishUpImpl(s); + } this.up.push(s); if (this.autoHello) { try { @@ -49,7 +57,7 @@ export class FakeTransport implements Transport { } async stop(): Promise { - this.cfg?.onOffline(); + /* 不触发 onOffline，避免 close 后误重连 */ } private replyHello(rid: string): void { @@ -75,7 +83,13 @@ export class FakeTransport implements Transport { async simulateConnectOK(): Promise { const { cleanStart, sessionExpiry } = buildCleanConnectFlags(); - this.connects.push({ cleanStart, sessionExpiry, password: this.cred }); + this.connects.push({ + cleanStart, + sessionExpiry, + password: this.cred, + receiveMaximumSet: false, + keepalive: 30, + }); this.cfg?.onConnectPacket?.(cleanStart, sessionExpiry); this.cfg?.backoff.markOnline(); await this.cfg?.mqttReady(); @@ -95,6 +109,15 @@ export class FakeTransport implements Transport { this.cfg?.onKicked(); } + simulateServerDisconnect(code: number): void { + if (code === 0x8e) { + this.cfg?.onKicked(); + return; + } + this.cfg?.backoff.markOffline(); + this.cfg?.onOffline(); + } + injectDown(payload: string | Uint8Array): void { const bytes = typeof payload === "string" ? new TextEncoder().encode(payload) : payload; this.cfg?.onDown(bytes); diff --git a/sdk/js/src/index.ts b/sdk/js/src/index.ts index 8641274..8dc3104 100644 --- a/sdk/js/src/index.ts +++ b/sdk/js/src/index.ts @@ -4,8 +4,11 @@ export { registerURLFromConnect, marshalJSON, ReconnectBackoff, + nominalDelay, + disableJitterForTest, + restoreJitterForTest, + applyJitter, } from "./types.js"; export type * from "./types.js"; export { Client, register } from "./client.js"; -export { FakeTransport } from "./fake.js"; -export { MqttTransport } from "./mqtt.js"; +export { MqttTransport, DEFAULT_KEEPALIVE, normalizeMQTTURL } from "./mqtt.js"; diff --git a/sdk/js/src/mqtt.ts b/sdk/js/src/mqtt.ts index 3db060e..6db3053 100644 --- a/sdk/js/src/mqtt.ts +++ b/sdk/js/src/mqtt.ts @@ -6,6 +6,8 @@ import { buildCleanConnectFlags, } from "./types.js"; +export const DEFAULT_KEEPALIVE = 30; + /** 基于 MQTT.js 5 的真实传输。 */ export class MqttTransport implements Transport { private client?: MqttClient; @@ -14,7 +16,8 @@ export class MqttTransport implements Transport { private stopped = false; private upTopic = ""; private downTopic = ""; - private attempt = 0; + private gen = 0; + private connectTimer?: ReturnType; setCredential(passwordOrToken: string): void { this.cred = passwordOrToken; @@ -25,25 +28,23 @@ export class MqttTransport implements Transport { this.stopped = false; this.upTopic = `nix/c/${cfg.endpointId}/up`; this.downTopic = `nix/c/${cfg.endpointId}/down`; - await this.connectLoop(); + // 不阻塞 connect() 超时轮询；连接循环自行退避重试。 + void this.connectLoop(); } private async connectLoop(): Promise { const cfg = this.cfg!; while (!this.stopped) { - const wait = cfg.backoff.delay(this.attempt); + const wait = cfg.backoff.nextWait(); if (wait > 0) await sleep(wait); if (this.stopped) return; try { await this.oneConnect(); - this.attempt = 0; return; - } catch (e) { - this.attempt++; + } catch { cfg.backoff.markOffline(); cfg.onOffline(); if (this.stopped) return; - // 认证失败等会设 stopped } } } @@ -53,48 +54,57 @@ export class MqttTransport implements Transport { const { cleanStart, sessionExpiry } = buildCleanConnectFlags(); cfg.onConnectPacket?.(cleanStart, sessionExpiry); - const url = normalizeWSURL(cfg.url, cfg.allowTcp); + const url = normalizeMQTTURL(cfg.url, cfg.allowTcp); const opts: IClientOptions = { clientId: cfg.endpointId, username: cfg.endpointId, password: this.cred, protocolVersion: 5, clean: cleanStart, + keepalive: DEFAULT_KEEPALIVE, connectTimeout: cfg.connectTimeoutMs, - reconnectPeriod: 0, // 自行退避 + reconnectPeriod: 0, properties: { sessionExpiryInterval: sessionExpiry, }, wsOptions: {}, }; + const myGen = ++this.gen; return new Promise((resolve, reject) => { const client = mqtt.connect(url, opts); this.client = client; let settled = false; const fail = (err: Error) => { - if (settled) return; + if (settled || this.gen !== myGen) return; settled = true; + this.clearConnectTimer(); client.end(true); reject(err); }; + this.connectTimer = setTimeout(() => { + if (!settled) fail(new Error("connect timeout")); + }, cfg.connectTimeoutMs); + this.connectTimer.unref?.(); + client.on("connect", async () => { try { cfg.backoff.markOnline(); await client.subscribeAsync(this.downTopic, { qos: 1 }); await cfg.mqttReady(); - if (!settled) { + if (!settled && this.gen === myGen) { settled = true; + this.clearConnectTimer(); resolve(); } - // 断线后自行重连 client.on("close", () => { - if (this.stopped) return; + if (this.stopped || this.gen !== myGen) return; + this.gen++; + void client.end(true); cfg.backoff.markOffline(); cfg.onOffline(); - this.attempt++; void this.connectLoop(); }); } catch (e) { @@ -107,9 +117,8 @@ export class MqttTransport implements Transport { }); client.on("error", (err) => { - // MQTT.js 用 error 报告部分失败 const msg = String(err?.message ?? err); - if (/Not authorized|Bad user|banned|0x86|0x87|0x8A/i.test(msg)) { + if (/0x86|0x87|0x8A|Not authorized|Bad user|banned/i.test(msg)) { const reason: AuthReason = this.cred.startsWith("nst_") ? "session_invalid" : "bad_credentials"; @@ -123,32 +132,40 @@ export class MqttTransport implements Transport { const code = packet?.reasonCode; if (code === 0x8e) { this.stopped = true; + this.gen++; cfg.onKicked(); + return; } + // 0x8B 及其余按可重试，交给 close 重连 }); - - setTimeout(() => { - if (!settled) fail(new Error("connect timeout")); - }, cfg.connectTimeoutMs); }); } + private clearConnectTimer(): void { + if (this.connectTimer) { + clearTimeout(this.connectTimer); + this.connectTimer = undefined; + } + } + async publishUp(payload: Uint8Array | string): Promise { const c = this.client; if (!c || !c.connected) throw new Error("not connected"); - const buf = typeof payload === "string" ? Buffer.from(payload) : Buffer.from(payload); - await c.publishAsync(this.upTopic, buf, { qos: 1 }); + const data = typeof payload === "string" ? payload : new TextDecoder().decode(payload); + await c.publishAsync(this.upTopic, data, { qos: 1 }); } async stop(): Promise { this.stopped = true; + this.gen++; + this.clearConnectTimer(); const c = this.client; this.client = undefined; if (c) await new Promise((r) => c.end(true, {}, () => r())); } } -function normalizeWSURL(raw: string, allowTcp: boolean): string { +export function normalizeMQTTURL(raw: string, allowTcp: boolean): string { const u = new URL(raw); if (u.protocol === "http:") u.protocol = "ws:"; if (u.protocol === "https:") u.protocol = "wss:"; @@ -164,5 +181,8 @@ function normalizeWSURL(raw: string, allowTcp: boolean): string { } function sleep(ms: number): Promise { - return new Promise((r) => setTimeout(r, ms)); + return new Promise((r) => { + const t = setTimeout(r, ms); + t.unref?.(); + }); } diff --git a/sdk/js/src/types.ts b/sdk/js/src/types.ts index 03d2266..f19cbc5 100644 --- a/sdk/js/src/types.ts +++ b/sdk/js/src/types.ts @@ -36,6 +36,8 @@ export interface SendOptions { contentType?: string; meta?: Record; id?: string; + /** 尚未发出时取消即出队；已发出则 result_unknown */ + signal?: AbortSignal; } export interface SendResult { @@ -164,58 +166,99 @@ export interface TransportConfig { onConnectPacket?: (cleanStart: boolean, sessionExpiry: number) => void; } -export class ReconnectBackoff { - private base = 1000; - private onlineAt = 0; - private online = false; - private stable = false; - private timer: ReturnType | null = null; +/** 第 n 次标称等待：min(1s×2^(n-1), 30s)，n≥1。 */ +export function nominalDelay(n: number): number { + if (n < 1) return 0; + if (n > 6) return 30000; + return Math.min(1000 * 2 ** (n - 1), 30000); +} - delay(attempt: number): number { - if (attempt <= 0) return 0; - let d = this.base; - for (let i = 1; i < attempt; i++) { - d = Math.min(d * 2, 30000); +let jitterFn = (d: number): number => { + if (d <= 0) return 0; + return Math.floor(d * (0.7 + Math.random() * 0.6)); +}; + +export function disableJitterForTest(): void { + jitterFn = (d) => d; +} + +export function restoreJitterForTest(): void { + jitterFn = (d) => { + if (d <= 0) return 0; + return Math.floor(d * (0.7 + Math.random() * 0.6)); + }; +} + +export function applyJitter(d: number): number { + return jitterFn(d); +} + +/** + * 只维护一个连续失败计数 n。connect 后第一次等待为 0； + * 之后第 n 次为 nominalDelay(n)×抖动。 + */ +export class ReconnectBackoff { + private n = 0; + private skipFirst = true; + private online = false; + private onlineAt = 0; + private counted = false; + + nextWait(): number { + this.counted = false; + if (this.skipFirst) { + this.skipFirst = false; + return 0; } - const f = 0.7 + Math.random() * 0.6; - return Math.floor(d * f); + const n = this.n < 1 ? 1 : this.n; + return jitterFn(nominalDelay(n)); + } + + nextWaitNoJitter(): number { + this.counted = false; + if (this.skipFirst) { + this.skipFirst = false; + return 0; + } + const n = this.n < 1 ? 1 : this.n; + return nominalDelay(n); + } + + /** 兼容旧调用；忽略 attempt，只用内部 n。 */ + delay(_attempt: number): number { + return this.nextWait(); } markOnline(): void { this.online = true; this.onlineAt = Date.now(); - this.stable = false; - if (this.timer) clearTimeout(this.timer); - this.timer = setTimeout(() => { - if (this.online) { - this.stable = true; - this.base = 1000; - } - }, 60_000); + this.counted = false; } markOffline(): void { - if (this.timer) { - clearTimeout(this.timer); - this.timer = null; - } + if (this.counted) return; + this.counted = true; const was = this.online; + const at = this.onlineAt; this.online = false; if (!was) { - this.base = Math.min(Math.max(this.base * 2, 1000), 30000); + this.n++; return; } - if (this.stable || Date.now() - this.onlineAt >= 60_000) { - this.base = 1000; - this.stable = false; + if (Date.now() - at >= 60_000) { + this.n = 1; return; } - this.base = Math.min(this.base * 2, 30000); - this.stable = false; + this.n++; } - getBase(): number { - return this.base; + getN(): number { + return this.n; + } + + setOnlineAtForTest(ms: number): void { + this.online = true; + this.onlineAt = ms; } } diff --git a/sdk/js/test/client.test.ts b/sdk/js/test/client.test.ts index d73c575..ad083f1 100644 --- a/sdk/js/test/client.test.ts +++ b/sdk/js/test/client.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; -import { Client, FakeTransport, APIError, buildCleanConnectFlags, register } from "../src/index.js"; +import { Client, APIError, buildCleanConnectFlags, register } from "../src/index.js"; +import { FakeTransport } from "../src/fake.js"; import { createServer } from "node:http"; async function connectFake(fake: FakeTransport): Promise { diff --git a/sdk/js/test/k00.test.ts b/sdk/js/test/k00.test.ts new file mode 100644 index 0000000..9118d38 --- /dev/null +++ b/sdk/js/test/k00.test.ts @@ -0,0 +1,269 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { + APIError, + Client, + DEFAULT_KEEPALIVE, + ReconnectBackoff, + disableJitterForTest, + normalizeMQTTURL, + restoreJitterForTest, +} from "../src/index.js"; +import { FakeTransport } from "../src/fake.js"; + +afterEach(() => { + restoreJitterForTest(); +}); + +async function connectFake(fake: FakeTransport, opts: Record = {}): Promise { + const c = new Client(); + await c.connect("ws://example.test/mqtt", "ep1", { password: "p" }, { transport: fake, ...opts }); + return c; +} + +describe("K00", () => { + it("TestK00FirstConnectTimeout", async () => { + const fake = new FakeTransport(); + fake.autoHello = false; + const c = new Client(); + await expect( + c.connect("ws://example.test/mqtt", "ep1", { password: "p" }, { + transport: fake, + connectTimeoutMs: 150, + }), + ).rejects.toMatchObject({ code: "not_connected" }); + fake.autoHello = true; + await c.connect("ws://example.test/mqtt", "ep1", { password: "p" }, { + transport: fake, + connectTimeoutMs: 2000, + }); + await c.close(); + }); + + it("TestK00AuthErrorCodes", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + fake.simulateAuthFail("bad_credentials"); + const deadline = Date.now() + 1000; + while (Date.now() < deadline && c.lastStopCodeForTest() !== "bad_credentials") { + await new Promise((r) => setTimeout(r, 5)); + } + expect(c.lastStopCodeForTest()).toBe("bad_credentials"); + await expect( + c.send({ kind: "endpoint", id: "b" }, { enc: "utf8", data: "x" }), + ).rejects.toMatchObject({ code: "bad_credentials" }); + }); + + it("TestK00TakenOverReason", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + let got = ""; + c.onConnectionHandler((ev) => { + if (ev.state === "kicked") got = ev.reason ?? ""; + }); + fake.simulateKick(); + const deadline = Date.now() + 1000; + while (Date.now() < deadline && got !== "taken_over") { + await new Promise((r) => setTimeout(r, 5)); + } + expect(got).toBe("taken_over"); + expect(c.lastStopCodeForTest()).toBe("taken_over"); + }); + + it("TestK00Disconnect8BRetryable", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + fake.simulateServerDisconnect(0x8b); + await new Promise((r) => setTimeout(r, 30)); + expect(c.lastStopCodeForTest()).not.toBe("taken_over"); + await fake.simulateConnectOK(); + await c.close(); + }); + + it("TestK00QueueFull", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake, { sendQueueSize: 1 }); + const p = c.send({ kind: "endpoint", id: "b" }, { enc: "utf8", data: "1" }); + await new Promise((r) => setTimeout(r, 20)); + await expect( + c.send({ kind: "endpoint", id: "b" }, { enc: "utf8", data: "2" }), + ).rejects.toMatchObject({ code: "queue_full" }); + fake.replyOK(String(fake.findUp("send")[0]?.rid), { id: "x", state: "accepted" }); + await p; + await c.close(); + }); + + it("TestK00RequestReturnsData", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + const timer = setInterval(() => { + for (const fr of fake.findUp("self.get")) { + fake.replyOK(String(fr.rid), { id: "ep1", name: "n", default_delay_ms: 0 }); + } + }, 5); + const info = (await c.getSelf()) as { id: string; name: string }; + clearInterval(timer); + expect(info.id).toBe("ep1"); + expect(info.name).toBe("n"); + await c.close(); + }); + + it("TestK00SendAtAndDelayConflict", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + await expect( + c.send( + { kind: "endpoint", id: "b" }, + { enc: "utf8", data: "x" }, + { sendAt: new Date(1), delayMs: 1000 }, + ), + ).rejects.toMatchObject({ code: "bad_request" }); + await c.close(); + }); + + it("TestK00SendAfterStopped", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + fake.simulateKick(); + await new Promise((r) => setTimeout(r, 30)); + await expect( + c.send({ kind: "endpoint", id: "b" }, { enc: "utf8", data: "x" }), + ).rejects.toMatchObject({ code: "taken_over" }); + }); + + it("TestK00LogoutReturnsError", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + fake.simulateServerDisconnect(0x8b); + await new Promise((r) => setTimeout(r, 20)); + await expect(c.logout()).rejects.toMatchObject({ code: "not_connected" }); + await expect( + c.send({ kind: "endpoint", id: "b" }, { enc: "utf8", data: "x" }), + ).rejects.toBeInstanceOf(APIError); + }); + + it("TestK00DurationInt64", () => { + const sd = JSON.parse(`{"id":"m1","send_at_ms":123,"state":"scheduled"}`) as { + id: string; + send_at_ms: number; + state: string; + }; + expect(sd.id).toBe("m1"); + expect(sd.send_at_ms).toBe(123); + expect(sd.state).toBe("scheduled"); + const ms = 30 * 24 * 3600 * 1000; + expect(ms).toBe(2592000000); + }); + + it("TestK00MaxReceiveBytesMin", async () => { + const fake = new FakeTransport(); + const c = new Client(); + await expect( + c.connect("ws://example.test/mqtt", "ep1", { password: "p" }, { + transport: fake, + maxReceiveBytes: 512, + }), + ).rejects.toMatchObject({ code: "bad_request" }); + }); + + it("TestK00URLMapping", () => { + const a = new URL(normalizeMQTTURL("https://host:7443/", false)); + expect(a.protocol).toBe("wss:"); + expect(a.pathname).toBe("/mqtt"); + const b = new URL(normalizeMQTTURL("http://host/app", false)); + expect(b.protocol).toBe("ws:"); + expect(b.pathname).toBe("/app"); + expect(() => normalizeMQTTURL("mqtt://host:1883", false)).toThrow(); + expect(normalizeMQTTURL("mqtt://host:1883", true)).toContain("mqtt:"); + }); + + it("TestK00CancelUnsent", async () => { + const fake = new FakeTransport(); + fake.autoHello = false; + const c = new Client(); + const connecting = c + .connect("ws://example.test/mqtt", "ep1", { password: "p" }, { + transport: fake, + connectTimeoutMs: 2000, + }) + .catch(() => {}); + await new Promise((r) => setTimeout(r, 40)); + const ac = new AbortController(); + const p = c.send({ kind: "endpoint", id: "b" }, { enc: "utf8", data: "x" }, { signal: ac.signal }); + ac.abort(); + await expect(p).rejects.toBeTruthy(); + expect(c.resendPayloadForTest()).toBeUndefined(); + await c.close(); + await connecting; + }); + + it("TestK00RateLimitedBackoff", async () => { + disableJitterForTest(); + const fake = new FakeTransport(); + const c = await connectFake(fake); + const rids: string[] = []; + let id0 = ""; + let sendAt: unknown; + const timer = setInterval(() => { + const sends = fake.findUp("send"); + if (!sends.length) return; + const last = sends[sends.length - 1]!; + const rid = String(last.rid); + if (!rids.length) { + id0 = String(last.id); + sendAt = last.send_at_ms; + rids.push(rid); + fake.replyErr(rid, "rate_limited", "slow"); + return; + } + if (rid === rids[rids.length - 1]) return; + rids.push(rid); + if (rids.length < 3) { + fake.replyErr(rid, "rate_limited", "slow"); + return; + } + expect(last.id).toBe(id0); + expect(last.send_at_ms).toBe(sendAt); + fake.replyOK(rid, { id: id0, send_at_ms: sendAt, state: "scheduled" }); + }, 5); + const at = new Date(1_700_000_000_000); + await c.send({ kind: "endpoint", id: "b" }, { enc: "utf8", data: "hi" }, { sendAt: at }); + clearInterval(timer); + expect(rids).toHaveLength(3); + expect(new Set(rids).size).toBe(3); + await c.close(); + }, 10000); + + it("TestK00ReconnectBackoff", () => { + const b = new ReconnectBackoff(); + expect(b.nextWaitNoJitter()).toBe(0); + const got: number[] = []; + for (let i = 0; i < 6; i++) { + b.markOffline(); + got.push(b.nextWaitNoJitter()); + } + expect(got).toEqual([1000, 2000, 4000, 8000, 16000, 30000]); + b.markOnline(); + b.setOnlineAtForTest(Date.now()); + b.markOffline(); + expect(b.nextWaitNoJitter()).toBe(30000); + const b2 = new ReconnectBackoff(); + b2.nextWaitNoJitter(); + b2.markOnline(); + b2.setOnlineAtForTest(Date.now() - 61_000); + b2.markOffline(); + expect(b2.nextWaitNoJitter()).toBe(1000); + }); + + it("TestK00KeepaliveDefault", () => { + expect(DEFAULT_KEEPALIVE).toBe(30); + }); + + it("TestK00NoReceiveMaximum", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + const cs = fake.getConnects(); + expect(cs.length).toBeGreaterThan(0); + expect(cs[0]?.receiveMaximumSet).toBe(false); + await c.close(); + }); +}); diff --git a/sdk/js/test/k02.test.ts b/sdk/js/test/k02.test.ts new file mode 100644 index 0000000..345b40e --- /dev/null +++ b/sdk/js/test/k02.test.ts @@ -0,0 +1,226 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { Client, disableJitterForTest, restoreJitterForTest } from "../src/index.js"; +import { FakeTransport } from "../src/fake.js"; + +afterEach(() => { + restoreJitterForTest(); +}); + +async function connectFake(fake: FakeTransport): Promise { + const c = new Client(); + await c.connect("ws://example.test/mqtt", "ep1", { password: "p" }, { transport: fake }); + return c; +} + +describe("K02", () => { + it("TestK01InflightResendAfterDisconnect", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + let firstRID = ""; + let firstID = ""; + let firstSendAt: unknown; + const timer = setInterval(() => { + const sends = fake.findUp("send"); + if (!sends.length) return; + const last = sends[sends.length - 1]!; + const rid = String(last.rid); + if (!firstRID) { + firstRID = rid; + firstID = String(last.id); + firstSendAt = last.send_at_ms; + void fake.simulateReconnect(); + return; + } + if (rid !== firstRID) { + expect(last.id).toBe(firstID); + expect(last.send_at_ms).toBe(firstSendAt); + fake.replyOK(rid, { id: firstID, send_at_ms: firstSendAt, state: "accepted" }); + } + }, 5); + const at = new Date(1_700_000_000_111); + await c.send({ kind: "endpoint", id: "b" }, { enc: "utf8", data: "hi" }, { sendAt: at }); + clearInterval(timer); + await c.close(); + }, 10000); + + it("unhandledRejection stays zero on ack publish throw", async () => { + const rejections: unknown[] = []; + const onRej = (e: unknown) => { + rejections.push(e); + }; + process.on("unhandledRejection", onRej); + const fake = new FakeTransport(); + const c = await connectFake(fake); + fake.publishUpImpl = async (s) => { + if (s.includes('"type":"ack"')) throw new Error("ack publish fail"); + }; + c.onMessageHandler(() => {}); + fake.injectDown( + JSON.stringify({ + v: 1, + type: "msg", + id: "m1", + from: "a", + to: { kind: "endpoint", id: "ep1" }, + body: { enc: "utf8", data: "hi" }, + send_at_ms: 1, + }), + ); + await new Promise((r) => setTimeout(r, 80)); + process.off("unhandledRejection", onRej); + expect(rejections).toHaveLength(0); + await c.close(); + }); + + it("ack fail then same msg re-acks without second callback", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + let calls = 0; + c.onMessageHandler(() => { + calls++; + }); + const msg = JSON.stringify({ + v: 1, + type: "msg", + id: "m1", + from: "a", + to: { kind: "endpoint", id: "ep1" }, + body: { enc: "utf8", data: "hi" }, + send_at_ms: 1, + }); + fake.injectDown(msg); + await new Promise((r) => setTimeout(r, 40)); + for (const fr of fake.findUp("ack")) { + fake.replyErr(String(fr.rid), "not_found", "gone"); + } + fake.injectDown(msg); + await new Promise((r) => setTimeout(r, 80)); + expect(calls).toBe(1); + expect(fake.findUp("ack").length).toBeGreaterThanOrEqual(2); + await c.close(); + }); + + it("revoked after msg skips callback", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + let msgs = 0; + let revs = 0; + c.onMessageHandler(() => { + msgs++; + }); + c.onRevokedHandler(() => { + revs++; + }); + fake.injectDown( + JSON.stringify({ + v: 1, + type: "msg", + id: "m1", + from: "a", + to: { kind: "endpoint", id: "ep1" }, + body: { enc: "utf8", data: "hi" }, + send_at_ms: 1, + }), + ); + fake.injectDown(JSON.stringify({ v: 1, type: "revoked", id: "m1", from: "a", reason: "recalled" })); + await new Promise((r) => setTimeout(r, 80)); + expect(msgs).toBe(0); + expect(revs).toBe(1); + await c.close(); + }); + + it("TestK01WatchRestored", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + const timer = setInterval(() => { + for (const fr of fake.findUp("presence.watch")) { + fake.replyOK(String(fr.rid), {}); + } + }, 5); + await c.watchPresence(["a", "b"]); + const n1 = fake.findUp("presence.watch").length; + await fake.simulateReconnect(); + const deadline = Date.now() + 2000; + while (Date.now() < deadline && fake.findUp("presence.watch").length <= n1) { + await new Promise((r) => setTimeout(r, 10)); + } + clearInterval(timer); + expect(fake.findUp("presence.watch").length).toBeGreaterThan(n1); + await c.close(); + }); + + it("TestK01FatalOnce", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + let n = 0; + c.onConnectionHandler((ev) => { + if (ev.state === "auth_failed" && ev.reason === "disabled") n++; + }); + const fatal = JSON.stringify({ v: 1, type: "fatal", reason: "disabled" }); + fake.injectDown(fatal); + fake.injectDown(fatal); + await new Promise((r) => setTimeout(r, 50)); + expect(n).toBe(1); + }); + + it("TestK01CallbackNoDeadlock", async () => { + const fake = new FakeTransport(); + const c = await connectFake(fake); + const timer = setInterval(() => { + for (const typ of ["ack", "self.login_password"]) { + for (const fr of fake.findUp(typ)) { + fake.replyOK(String(fr.rid), typ === "ack" ? { result: "accepted" } : {}); + } + } + }, 5); + let done = false; + c.onMessageHandler(async (msg) => { + await c.changeLoginPassword("old", "newpass12"); + await c.ack(msg); + done = true; + }); + fake.injectDown( + JSON.stringify({ + v: 1, + type: "msg", + id: "m1", + from: "a", + to: { kind: "endpoint", id: "ep1" }, + body: { enc: "utf8", data: "hi" }, + send_at_ms: 1, + }), + ); + const deadline = Date.now() + 2000; + while (Date.now() < deadline && !done) { + await new Promise((r) => setTimeout(r, 10)); + } + clearInterval(timer); + expect(done).toBe(true); + await c.close(); + }); + + it("rate_limited rids unique with disabled jitter", async () => { + disableJitterForTest(); + const fake = new FakeTransport(); + const c = await connectFake(fake); + const rids: string[] = []; + const timer = setInterval(() => { + const sends = fake.findUp("send"); + if (!sends.length) return; + const last = sends[sends.length - 1]!; + const rid = String(last.rid); + if (rids.includes(rid)) return; + rids.push(rid); + if (rids.length < 3) fake.replyErr(rid, "rate_limited", "slow"); + else fake.replyOK(rid, { id: last.id, send_at_ms: last.send_at_ms, state: "scheduled" }); + }, 5); + await c.send( + { kind: "endpoint", id: "b" }, + { enc: "utf8", data: "hi" }, + { sendAt: new Date(1_700_000_000_000) }, + ); + clearInterval(timer); + expect(new Set(rids).size).toBe(3); + await c.close(); + }, 10000); +});