YBF/trade-message-center
1
0
Fork 0
mirror of https://github.com/sinanyuntu/trade-message-center.git synced 2026-09-17 13:22:11 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(onetalk): harden mind send liveness

Browse Source
This commit is contained in:
YBF
2026-09-12 15:35:15 +08:00
parent bb4c7afd17
commit 6e77de7506
7 changed files with 106 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/mind-test-harness/src/harness/errors.ts
+1
View File
@@ -19,6 +19,7 @@ export const harnessErrorsScript = String.raw` const errorText = (err
oss_upload_invalid_request: '上传文件或元数据无效',
oss_upload_origin_not_allowed: '当前页面 Origin 不被 Bright 允许上传',
oss_upload_unavailable: 'Bright 未配置 OSS 上传能力',
request_timeout: '请求超过 30 秒未完成',
invalid_response: 'Bright 响应不符合读取契约',
http_error: 'Bright 返回了未知 HTTP 错误'
};
apps/mind-test-harness/src/harness/reading.ts
+14 -13
View File
@@ -48,19 +48,20 @@ export const harnessReadingScript = String.raw` const requestedChanne
};
const readResponse = async (url, requestHeaders = {}) => {
const response = await fetch(url, { headers: requestHeaders, credentials: 'include' });
let body = null;
try { body = await response.json(); } catch (_) { body = null; }
if (!response.ok) {
const code = isReadErrorResponse(body) ? body.error.code : 'http_error';
const error = new Error(code);
error.code = code;
const retryAfter = response.headers.get('retry-after');
if (/^\d+$/.test(retryAfter || '')) error.retryAfter = Number(retryAfter);
throw error;
}
if (!isRecord(body)) throw new Error('invalid_response');
return body;
return fetchWithDeadline(url, { headers: requestHeaders, credentials: 'include' }, async (response) => {
let body = null;
try { body = await response.json(); } catch (_) { body = null; }
if (!response.ok) {
const code = isReadErrorResponse(body) ? body.error.code : 'http_error';
const error = new Error(code);
error.code = code;
const retryAfter = response.headers.get('retry-after');
if (/^\d+$/.test(retryAfter || '')) error.retryAfter = Number(retryAfter);
throw error;
}
if (!isRecord(body)) throw new Error('invalid_response');
return body;
});
};
const fillConversations = () => {
apps/mind-test-harness/src/harness/runtime.ts
+27
View File
@@ -1,6 +1,8 @@
// 页面脚本分片：页面状态与控件引用（由 harness/index.ts 组装进主 IIFE）
export const harnessRuntimeScript = String.raw` const heartbeatIntervalMs = 25000;
const heartbeatAckTimeoutMs = 75000;
const requestTimeoutMs = 30000;
const state = {
scope: null,
conversationId: '',
@@ -17,6 +19,7 @@ export const harnessRuntimeScript = String.raw` const heartbeatInterv
socketAccepted: false,
helloRequestId: null,
heartbeatTimer: null,
heartbeatAckTimer: null,
heartbeatRequestId: null,
heartbeatSentAtMs: null,
requestSequence: 0,
@@ -67,4 +70,28 @@ export const harnessRuntimeScript = String.raw` const heartbeatInterv
const setStatus = (target, text, stateName) => {
target.dataset.state = stateName;
target.querySelector('span').textContent = text;
};
const fetchWithDeadline = async (url, options, consumeResponse) => {
const controller = new AbortController();
const timeout = window.setTimeout(() => controller.abort(), requestTimeoutMs);
try {
const response = await fetch(url, { ...options, signal: controller.signal });
const result = await consumeResponse(response);
if (controller.signal.aborted) {
const timeoutError = new Error('request_timeout');
timeoutError.code = 'request_timeout';
throw timeoutError;
}
return result;
} catch (error) {
if (controller.signal.aborted) {
const timeoutError = new Error('request_timeout');
timeoutError.code = 'request_timeout';
throw timeoutError;
}
throw error;
} finally {
window.clearTimeout(timeout);
}
};`;
apps/mind-test-harness/src/harness/upload.ts
+20 -19
View File
@@ -31,30 +31,31 @@ export const harnessUploadScript = String.raw` const setUploadStatus
setUploadStatus('上传中…', 'warn');
updateUploadAvailability();
try {
const response = await fetch(uploadUrlFor(file.name, mimeType), {
await fetchWithDeadline(uploadUrlFor(file.name, mimeType), {
method: 'POST',
headers: { 'content-type': 'application/octet-stream' },
credentials: 'include',
body: file
}, async (response) => {
let body = null;
try { body = await response.json(); } catch (_) { body = null; }
if (!response.ok) {
const code = isUploadErrorResponse(body) ? body.error.code : 'http_error';
const error = new Error(code);
error.code = code;
throw error;
}
if (!isUploadResponse(body)) throw new Error('invalid_response');
state.uploadedUrl = body.downloadUrl;
fields.uploadDownloadUrl.value = body.downloadUrl;
fields.fileSourceUrl.value = body.downloadUrl;
fields.fileName.value = body.fileName;
fields.fileMimeType.value = body.mimeType;
if (body.mimeType.startsWith('image/') && imageContentFromUrl(body.downloadUrl)) {
fields.imageSourceUrl.value = body.downloadUrl;
}
setUploadStatus('上传成功：已填入图片或附件发送字段', 'ok');
});
let body = null;
try { body = await response.json(); } catch (_) { body = null; }
if (!response.ok) {
const code = isUploadErrorResponse(body) ? body.error.code : 'http_error';
const error = new Error(code);
error.code = code;
throw error;
}
if (!isUploadResponse(body)) throw new Error('invalid_response');
state.uploadedUrl = body.downloadUrl;
fields.uploadDownloadUrl.value = body.downloadUrl;
fields.fileSourceUrl.value = body.downloadUrl;
fields.fileName.value = body.fileName;
fields.fileMimeType.value = body.mimeType;
if (body.mimeType.startsWith('image/') && imageContentFromUrl(body.downloadUrl)) {
fields.imageSourceUrl.value = body.downloadUrl;
}
setUploadStatus('上传成功：已填入图片或附件发送字段', 'ok');
} catch (error) {
setUploadStatus(errorText(error), 'error');
} finally {
apps/mind-test-harness/src/harness/websocket.ts
+32 -3
View File
@@ -99,16 +99,25 @@ export const harnessWebsocketScript = String.raw` const frameMatchesF
const stopHeartbeat = () => {
if (state.heartbeatTimer !== null) window.clearInterval(state.heartbeatTimer);
if (state.heartbeatAckTimer !== null) window.clearTimeout(state.heartbeatAckTimer);
state.heartbeatTimer = null;
state.heartbeatAckTimer = null;
state.heartbeatRequestId = null;
state.heartbeatSentAtMs = null;
};
const settlePendingSendAsUnknown = () => {
if (state.pendingSendRequestId === null) return;
state.pendingSendRequestId = null;
setStatus(fields.syncStatus, '本地连接不可用；发送未确认，结果未知（不会自动重试）', 'error');
updateSendAvailability();
};
const clearSocketState = () => {
stopHeartbeat();
state.socketAccepted = false;
state.helloRequestId = null;
state.pendingSendRequestId = null;
settlePendingSendAsUnknown();
setPluginStatus('offline');
};
@@ -123,6 +132,16 @@ export const harnessWebsocketScript = String.raw` const frameMatchesF
updateSendAvailability();
};
const invalidateSocket = (socket, status) => {
if (state.socket !== socket) return;
state.socket = null;
clearSocketState();
socket.onclose = null;
socket.close();
setStatus(fields.connectionStatus, status, 'error');
updateSendAvailability();
};
const nextRequestId = (prefix) => {
state.requestSequence += 1;
return prefix + '-' + state.requestSequence;
@@ -139,7 +158,8 @@ export const harnessWebsocketScript = String.raw` const frameMatchesF
state.socket !== socket ||
!state.socketAccepted ||
!matchesCurrentScope(scope) ||
socket.readyState !== WebSocket.OPEN
socket.readyState !== WebSocket.OPEN ||
state.heartbeatRequestId !== null
) {
return;
}
@@ -147,6 +167,10 @@ export const harnessWebsocketScript = String.raw` const frameMatchesF
const requestId = nextRequestId('heartbeat');
state.heartbeatRequestId = requestId;
state.heartbeatSentAtMs = sentAtMs;
state.heartbeatAckTimer = window.setTimeout(() => {
if (state.socket !== socket || state.heartbeatRequestId !== requestId) return;
invalidateSocket(socket, '心跳确认超时；点击“消息恢复后重连”');
}, heartbeatAckTimeoutMs);
sendWsFrame(socket, {
protocolVersion,
connectionType: 'mind_page',
@@ -229,6 +253,8 @@ export const harnessWebsocketScript = String.raw` const frameMatchesF
frame.requestId === state.heartbeatRequestId &&
frame.payload.sentAtMs === state.heartbeatSentAtMs
) {
if (state.heartbeatAckTimer !== null) window.clearTimeout(state.heartbeatAckTimer);
state.heartbeatAckTimer = null;
state.heartbeatRequestId = null;
state.heartbeatSentAtMs = null;
}
@@ -256,7 +282,10 @@ export const harnessWebsocketScript = String.raw` const frameMatchesF
setStatus(fields.syncStatus, status + ' · ' + text + (frame.payload.reason ? ' · ' + frame.payload.reason : ''), status === 'confirmed_sent' ? 'ok' : status === 'rejected_before_send' ? 'warn' : 'error');
}
};
socket.onerror = () => setStatus(fields.connectionStatus, '连接错误；消息仍可读取', 'error');
socket.onerror = () => {
if (state.socket !== socket) return;
setStatus(fields.connectionStatus, '连接错误；消息仍可读取', 'error');
};
socket.onclose = () => {
if (state.socket !== socket) return;
state.socket = null;
apps/server/src/websocket/connection-store.ts
+9
View File
@@ -66,6 +66,7 @@ export type OneTalkConnectionStore = {
register: (connection: OneTalkRegisteredConnection) => () => void;
unregister: (socket: WebSocket) => void;
isPluginOnline: (scope: OneTalkMindScope) => boolean;
isPluginLeaseFresh: (connection: OneTalkRegisteredConnection) => boolean;
getConnections: () => OneTalkRegisteredConnection[];
getGeneration: (socket: WebSocket) => number | undefined;
currentEpoch: () => number;
@@ -243,6 +244,13 @@ export const createOneTalkConnectionStore = (options: {
};
const isPluginOnline = (scope: OneTalkMindScope): boolean => hasPluginForScope(scope);
const isPluginLeaseFresh = (connection: OneTalkRegisteredConnection): boolean => {
return (
connection.connectionType === "plugin" &&
connections.get(connection.socket) === connection &&
(connection.lastHeartbeatAtMs ?? 0) > now() - heartbeatTimeoutMs
);
};
const getConnections = (): OneTalkRegisteredConnection[] => [...connections.values()];
const getGeneration = (socket: WebSocket): number | undefined => generations.get(socket);
const recordHeartbeat = (socket: WebSocket): boolean => {
@@ -279,6 +287,7 @@ export const createOneTalkConnectionStore = (options: {
register,
unregister,
isPluginOnline,
isPluginLeaseFresh,
getConnections,
getGeneration,
recordHeartbeat,
apps/server/src/websocket/pending-send-coordinator.ts
+3 -1
View File
@@ -145,7 +145,8 @@ export const createOneTalkPendingSendCoordinator = (options: {
isSameOneTalkScope(connection.mindScope, mind.mindScope) &&
connection.binding === mind.binding &&
connection.permissions.includes("send") &&
connection.socket.readyState === WEBSOCKET_OPEN,
connection.socket.readyState === WEBSOCKET_OPEN &&
options.store.isPluginLeaseFresh(connection),
);
if (candidates.length === 0)
return { status: "rejected_before_send", reason: "waiting_for_page" };
@@ -245,6 +246,7 @@ export const createOneTalkPendingSendCoordinator = (options: {
options.store.getGeneration(plugin.socket) !== pluginGeneration ||
mind.socket.readyState !== WEBSOCKET_OPEN ||
plugin.socket.readyState !== WEBSOCKET_OPEN ||
!options.store.isPluginLeaseFresh(plugin) ||
!options.store.epochIsCurrent(policyEpoch) ||
!options.store.policyAdmits("mind_page") ||
!options.store.policyAdmits("plugin")