nixevol/NixMsg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 按 K-00 约定修复 JS SDK 断线重交与退避

Browse Source
This commit is contained in:
Nixevol
2026-09-30 16:24:13 +08:00
parent f6f8ccf269
commit 55aa0ccf53
10 changed files with 1049 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk/js/src/client.ts
+390 -119
View File
@@ -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<typeof setTimeout>;
};
type SendItem = {
frame: Record<string, unknown>;
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<T> {
private map = new Map<string, T>();
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<string, { resolve: (r: RespFrame) => void; reject: (e: unknown) => void }>();
private pending = new Map<string, Pending>();
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<string, DedupState>();
private dedupOrd: string[] = [];
private receiptSeen = new Set<string>();
private store = new LRUMap<DedupState>(10000);
private cbChain: Promise<void> = Promise.resolve();
private timers = new Set<ReturnType<typeof setTimeout>>();
private watch?: { ids?: string[]; all?: boolean };
private onSession?: (token: string) => void;
private onMessage?: (msg: Message) => void | Promise<void>;
@@ -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<void> {
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<DedupState>(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<void> {
const sentAt = Date.now();
const req: Record<string, unknown> = {
@@ -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<void> {
if (!this.watch) return;
const req: Record<string, unknown> = { 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<string, unknown>;
send_at_ms: number;
}): Promise<void> {
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<void>((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<void>((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<void> {
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<void> {
@@ -389,13 +522,18 @@ export class Client {
}
private async handleReceipt(r: Receipt & { receipt_id: string }): Promise<void> {
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<void> {
@@ -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<SendResult> {
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<SendResult>((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<void> {
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<void> {
private async dispatchSend(item: SendItem, epoch: number): Promise<void> {
const rid = String(item.frame.rid);
const tr = this.transport!;
const tr = this.transport;
if (!tr) return;
try {
const data = await new Promise<unknown>((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<void> {
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<RecallResult> {
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<void> {
if (ids === "all") this.watch = { all: true };
else this.watch = { ids: [...ids] };
const req: Record<string, unknown> = { 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<void> {
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<void> {
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<void> {
return new Promise((r) => setTimeout(r, ms));
return new Promise((r) => {
const t = setTimeout(r, ms);
t.unref?.();
});
}
export async function register(