mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
feat(tlon): add durable ingress (#110910)
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
53ad69c6ee
commit
138d2d071c
@@ -1,5 +1,6 @@
|
||||
import { resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime";
|
||||
import { createChannelInboundEnvelopeBuilder } from "openclaw/plugin-sdk/channel-inbound";
|
||||
import { bindIngressLifecycleToReplyOptions } from "openclaw/plugin-sdk/channel-outbound";
|
||||
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime";
|
||||
import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime";
|
||||
import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env";
|
||||
@@ -27,11 +28,8 @@ import { resolveChannelAuthorization } from "./authorization.js";
|
||||
import { createTlonCitationResolver } from "./cites.js";
|
||||
import { fetchAllChannels, fetchInitData } from "./discovery.js";
|
||||
import { cacheMessage, fetchThreadHistory, getChannelHistory } from "./history.js";
|
||||
import { createTlonIngressMonitor, type TlonIngressLifecycle } from "./ingress.js";
|
||||
import { downloadMessageImages } from "./media.js";
|
||||
import {
|
||||
createProcessedMessageTracker,
|
||||
runWithProcessedMessageClaim,
|
||||
} from "./processed-messages.js";
|
||||
import {
|
||||
applyTlonSettingsOverrides,
|
||||
buildTlonSettingsMigrations,
|
||||
@@ -140,7 +138,6 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
},
|
||||
});
|
||||
|
||||
const processedTracker = createProcessedMessageTracker(2000);
|
||||
let groupChannels: string[] = [];
|
||||
let botNickname: string | null = null;
|
||||
|
||||
@@ -308,6 +305,7 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
timestamp: number;
|
||||
parentId?: string | null;
|
||||
isThreadReply?: boolean;
|
||||
turnAdoptionLifecycle?: TlonIngressLifecycle;
|
||||
}) => {
|
||||
const {
|
||||
messageId,
|
||||
@@ -320,6 +318,7 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
parentId,
|
||||
isThreadReply,
|
||||
messageContent,
|
||||
turnAdoptionLifecycle,
|
||||
} = params;
|
||||
const groupChannel = channelNest; // For compatibility
|
||||
let messageText = params.messageText;
|
||||
@@ -654,6 +653,9 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
responsePrefix,
|
||||
humanDelay,
|
||||
},
|
||||
...(turnAdoptionLifecycle
|
||||
? { replyOptions: bindIngressLifecycleToReplyOptions(turnAdoptionLifecycle) }
|
||||
: {}),
|
||||
record: {
|
||||
onRecordError: (err) => {
|
||||
runtime.error?.(`[tlon] failed updating session meta: ${String(err)}`);
|
||||
@@ -739,7 +741,10 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
});
|
||||
|
||||
// Firehose handler for all channel messages (/v2)
|
||||
const handleChannelsFirehose = async (event: unknown) => {
|
||||
const handleChannelsFirehose = async (
|
||||
event: unknown,
|
||||
turnAdoptionLifecycle?: TlonIngressLifecycle,
|
||||
) => {
|
||||
try {
|
||||
const eventRecord = asRecord(event);
|
||||
const nest = readString(eventRecord, "nest");
|
||||
@@ -780,119 +785,111 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
return;
|
||||
}
|
||||
|
||||
const processed = await runWithProcessedMessageClaim({
|
||||
tracker: processedTracker,
|
||||
const senderShip = normalizeShip(readString(content, "author") ?? "");
|
||||
if (!senderShip || senderShip === botShipName) {
|
||||
return;
|
||||
}
|
||||
|
||||
const rawText = extractMessageText(content.content);
|
||||
if (!rawText.trim()) {
|
||||
return;
|
||||
}
|
||||
|
||||
const contentBody = content.content;
|
||||
const sentAt = readNumber(content, "sent") ?? Date.now();
|
||||
|
||||
cacheMessage(nest, {
|
||||
author: senderShip,
|
||||
content: rawText,
|
||||
timestamp: sentAt,
|
||||
id: messageId,
|
||||
task: async () => {
|
||||
const senderShip = normalizeShip(readString(content, "author") ?? "");
|
||||
if (!senderShip || senderShip === botShipName) {
|
||||
return;
|
||||
}
|
||||
|
||||
const rawText = extractMessageText(content.content);
|
||||
if (!rawText.trim()) {
|
||||
return;
|
||||
}
|
||||
|
||||
const contentBody = content.content;
|
||||
const sentAt = readNumber(content, "sent") ?? Date.now();
|
||||
|
||||
cacheMessage(nest, {
|
||||
author: senderShip,
|
||||
content: rawText,
|
||||
timestamp: sentAt,
|
||||
id: messageId,
|
||||
});
|
||||
|
||||
// Get thread info early for participation check
|
||||
const seal = isThreadReply ? asRecord(replySet?.seal) : asRecord(set?.seal);
|
||||
const parentId = readString(seal, "parent-id") ?? readString(seal, "parent") ?? null;
|
||||
|
||||
// Check if we should respond:
|
||||
// 1. Direct mention always triggers response
|
||||
// 2. Thread replies where we've participated - respond if relevant (let agent decide)
|
||||
const mentioned = isBotMentioned(rawText, botShipName, botNickname ?? undefined);
|
||||
const inParticipatedThread =
|
||||
isThreadReply && parentId && participatedThreads.has(parentId);
|
||||
const mentionDecision = resolveTlonGroupMentionDecision({
|
||||
cfg,
|
||||
accountId: account.accountId,
|
||||
wasMentioned: mentioned,
|
||||
botParticipatedInThread: Boolean(inParticipatedThread),
|
||||
});
|
||||
|
||||
if (mentionDecision.shouldSkip) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Log why we're responding
|
||||
if (mentionDecision.implicitMention && !mentioned) {
|
||||
runtime.log?.(
|
||||
`[tlon] Responding to thread we participated in (no mention): ${parentId}`,
|
||||
);
|
||||
}
|
||||
|
||||
// Owner is always allowed
|
||||
if (isOwner(senderShip)) {
|
||||
runtime.log?.(`[tlon] Owner ${senderShip} is always allowed in channels`);
|
||||
} else {
|
||||
const { mode, allowedShips } = resolveChannelAuthorization(cfg, nest, currentSettings);
|
||||
if (mode === "restricted") {
|
||||
const normalizedAllowed = allowedShips.map(normalizeShip);
|
||||
if (!normalizedAllowed.includes(senderShip)) {
|
||||
// If owner is configured, queue approval request
|
||||
if (effectiveOwnerShip) {
|
||||
const approval = createPendingApproval({
|
||||
type: "channel",
|
||||
requestingShip: senderShip,
|
||||
channelNest: nest,
|
||||
messagePreview: sliceUtf16Safe(rawText, 0, 100),
|
||||
originalMessage: {
|
||||
messageId: messageId ?? "",
|
||||
messageText: rawText,
|
||||
messageContent: contentBody,
|
||||
timestamp: sentAt,
|
||||
parentId: parentId ?? undefined,
|
||||
isThreadReply,
|
||||
},
|
||||
});
|
||||
await queueApprovalRequest(approval);
|
||||
} else {
|
||||
runtime.log?.(
|
||||
`[tlon] Access denied: ${senderShip} in ${nest} (allowed: ${allowedShips.join(", ")})`,
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const messageText = await resolveAuthorizedMessageText({
|
||||
rawText,
|
||||
content: contentBody,
|
||||
authorizedForCites: true,
|
||||
resolveAllCites,
|
||||
});
|
||||
|
||||
const parsed = parseChannelNest(nest);
|
||||
await processMessage({
|
||||
messageId: messageId ?? "",
|
||||
senderShip,
|
||||
messageText,
|
||||
messageContent: contentBody, // Pass raw content for media extraction
|
||||
isGroup: true,
|
||||
channelNest: nest,
|
||||
hostShip: parsed?.hostShip,
|
||||
channelName: parsed?.channelName,
|
||||
timestamp: sentAt,
|
||||
parentId,
|
||||
isThreadReply,
|
||||
});
|
||||
},
|
||||
});
|
||||
void processed;
|
||||
|
||||
// Get thread info early for participation check
|
||||
const seal = isThreadReply ? asRecord(replySet?.seal) : asRecord(set?.seal);
|
||||
const parentId = readString(seal, "parent-id") ?? readString(seal, "parent") ?? null;
|
||||
|
||||
// Check if we should respond:
|
||||
// 1. Direct mention always triggers response
|
||||
// 2. Thread replies where we've participated - respond if relevant (let agent decide)
|
||||
const mentioned = isBotMentioned(rawText, botShipName, botNickname ?? undefined);
|
||||
const inParticipatedThread = isThreadReply && parentId && participatedThreads.has(parentId);
|
||||
const mentionDecision = resolveTlonGroupMentionDecision({
|
||||
cfg,
|
||||
accountId: account.accountId,
|
||||
wasMentioned: mentioned,
|
||||
botParticipatedInThread: Boolean(inParticipatedThread),
|
||||
});
|
||||
|
||||
if (mentionDecision.shouldSkip) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Log why we're responding
|
||||
if (mentionDecision.implicitMention && !mentioned) {
|
||||
runtime.log?.(`[tlon] Responding to thread we participated in (no mention): ${parentId}`);
|
||||
}
|
||||
|
||||
// Owner is always allowed
|
||||
if (isOwner(senderShip)) {
|
||||
runtime.log?.(`[tlon] Owner ${senderShip} is always allowed in channels`);
|
||||
} else {
|
||||
const { mode, allowedShips } = resolveChannelAuthorization(cfg, nest, currentSettings);
|
||||
if (mode === "restricted") {
|
||||
const normalizedAllowed = allowedShips.map(normalizeShip);
|
||||
if (!normalizedAllowed.includes(senderShip)) {
|
||||
// If owner is configured, queue approval request
|
||||
if (effectiveOwnerShip) {
|
||||
const approval = createPendingApproval({
|
||||
type: "channel",
|
||||
requestingShip: senderShip,
|
||||
channelNest: nest,
|
||||
messagePreview: sliceUtf16Safe(rawText, 0, 100),
|
||||
originalMessage: {
|
||||
messageId: messageId ?? "",
|
||||
messageText: rawText,
|
||||
messageContent: contentBody,
|
||||
timestamp: sentAt,
|
||||
parentId: parentId ?? undefined,
|
||||
isThreadReply,
|
||||
},
|
||||
});
|
||||
await queueApprovalRequest(approval);
|
||||
} else {
|
||||
runtime.log?.(
|
||||
`[tlon] Access denied: ${senderShip} in ${nest} (allowed: ${allowedShips.join(", ")})`,
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const messageText = await resolveAuthorizedMessageText({
|
||||
rawText,
|
||||
content: contentBody,
|
||||
authorizedForCites: true,
|
||||
resolveAllCites,
|
||||
});
|
||||
|
||||
const parsed = parseChannelNest(nest);
|
||||
await processMessage({
|
||||
messageId: messageId ?? "",
|
||||
senderShip,
|
||||
messageText,
|
||||
messageContent: contentBody, // Pass raw content for media extraction
|
||||
isGroup: true,
|
||||
channelNest: nest,
|
||||
hostShip: parsed?.hostShip,
|
||||
channelName: parsed?.channelName,
|
||||
timestamp: sentAt,
|
||||
parentId,
|
||||
isThreadReply,
|
||||
turnAdoptionLifecycle,
|
||||
});
|
||||
} catch (error: unknown) {
|
||||
runtime.error?.(`[tlon] Error handling channel firehose event: ${formatErrorMessage(error)}`);
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
|
||||
@@ -900,7 +897,10 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
// Track which DM invites we've already processed to avoid duplicate accepts
|
||||
const processedDmInvites = new Set<string>();
|
||||
|
||||
const handleChatFirehose = async (event: unknown) => {
|
||||
const handleChatFirehose = async (
|
||||
event: unknown,
|
||||
turnAdoptionLifecycle?: TlonIngressLifecycle,
|
||||
) => {
|
||||
try {
|
||||
// Handle DM invite lists (arrays)
|
||||
if (Array.isArray(event)) {
|
||||
@@ -976,116 +976,125 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
return;
|
||||
}
|
||||
|
||||
const processed = await runWithProcessedMessageClaim({
|
||||
tracker: processedTracker,
|
||||
id: messageId,
|
||||
task: async () => {
|
||||
const authorShip = normalizeShip(readString(essay, "author") ?? "");
|
||||
const partnerShip = extractDmPartnerShip(whom);
|
||||
const senderShip = partnerShip || authorShip;
|
||||
const authorShip = normalizeShip(readString(essay, "author") ?? "");
|
||||
const partnerShip = extractDmPartnerShip(whom);
|
||||
const senderShip = partnerShip || authorShip;
|
||||
|
||||
// Ignore the bot's own outbound DM events.
|
||||
if (authorShip === botShipName) {
|
||||
return;
|
||||
}
|
||||
if (!senderShip || senderShip === botShipName) {
|
||||
return;
|
||||
}
|
||||
// Ignore the bot's own outbound DM events.
|
||||
if (authorShip === botShipName) {
|
||||
return;
|
||||
}
|
||||
if (!senderShip || senderShip === botShipName) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Log mismatch between author and partner for debugging
|
||||
if (authorShip && partnerShip && authorShip !== partnerShip) {
|
||||
runtime.log?.(
|
||||
`[tlon] DM ship mismatch (author=${authorShip}, partner=${partnerShip}) - routing to partner`,
|
||||
);
|
||||
}
|
||||
// Log mismatch between author and partner for debugging
|
||||
if (authorShip && partnerShip && authorShip !== partnerShip) {
|
||||
runtime.log?.(
|
||||
`[tlon] DM ship mismatch (author=${authorShip}, partner=${partnerShip}) - routing to partner`,
|
||||
);
|
||||
}
|
||||
|
||||
const rawText = extractMessageText(essay.content);
|
||||
if (!rawText.trim()) {
|
||||
return;
|
||||
}
|
||||
const rawText = extractMessageText(essay.content);
|
||||
if (!rawText.trim()) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Check if this is the owner sending an approval response
|
||||
const messageText = rawText;
|
||||
if (isOwner(senderShip) && isApprovalResponse(messageText)) {
|
||||
const handled = await handleApprovalResponse(messageText);
|
||||
if (handled) {
|
||||
runtime.log?.(`[tlon] Processed approval response from owner: ${messageText}`);
|
||||
return;
|
||||
}
|
||||
}
|
||||
// Check if this is the owner sending an approval response
|
||||
const messageText = rawText;
|
||||
if (isOwner(senderShip) && isApprovalResponse(messageText)) {
|
||||
const handled = await handleApprovalResponse(messageText);
|
||||
if (handled) {
|
||||
runtime.log?.(`[tlon] Processed approval response from owner: ${messageText}`);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Check if this is the owner sending an admin command
|
||||
if (isOwner(senderShip) && isAdminCommand(messageText)) {
|
||||
const handled = await handleAdminCommand(messageText);
|
||||
if (handled) {
|
||||
runtime.log?.(`[tlon] Processed admin command from owner: ${messageText}`);
|
||||
return;
|
||||
}
|
||||
}
|
||||
// Check if this is the owner sending an admin command
|
||||
if (isOwner(senderShip) && isAdminCommand(messageText)) {
|
||||
const handled = await handleAdminCommand(messageText);
|
||||
if (handled) {
|
||||
runtime.log?.(`[tlon] Processed admin command from owner: ${messageText}`);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Owner is always allowed to DM (bypass allowlist)
|
||||
if (isOwner(senderShip)) {
|
||||
const resolvedMessageText = await resolveAuthorizedMessageText({
|
||||
rawText,
|
||||
content: essay.content,
|
||||
authorizedForCites: true,
|
||||
resolveAllCites,
|
||||
});
|
||||
runtime.log?.(`[tlon] Processing DM from owner ${senderShip}`);
|
||||
await processMessage({
|
||||
// Owner is always allowed to DM (bypass allowlist)
|
||||
if (isOwner(senderShip)) {
|
||||
const resolvedMessageText = await resolveAuthorizedMessageText({
|
||||
rawText,
|
||||
content: essay.content,
|
||||
authorizedForCites: true,
|
||||
resolveAllCites,
|
||||
});
|
||||
runtime.log?.(`[tlon] Processing DM from owner ${senderShip}`);
|
||||
await processMessage({
|
||||
messageId: messageId ?? "",
|
||||
senderShip,
|
||||
messageText: resolvedMessageText,
|
||||
messageContent: essay.content,
|
||||
isGroup: false,
|
||||
timestamp: readNumber(essay, "sent") ?? Date.now(),
|
||||
turnAdoptionLifecycle,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// For DMs from others, check allowlist
|
||||
if (!(await isDmAllowedWithIngress(senderShip, effectiveDmAllowlist))) {
|
||||
// If owner is configured, queue approval request
|
||||
if (effectiveOwnerShip) {
|
||||
const approval = createPendingApproval({
|
||||
type: "dm",
|
||||
requestingShip: senderShip,
|
||||
messagePreview: sliceUtf16Safe(messageText, 0, 100),
|
||||
originalMessage: {
|
||||
messageId: messageId ?? "",
|
||||
senderShip,
|
||||
messageText: resolvedMessageText,
|
||||
messageText,
|
||||
messageContent: essay.content,
|
||||
isGroup: false,
|
||||
timestamp: readNumber(essay, "sent") ?? Date.now(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// For DMs from others, check allowlist
|
||||
if (!(await isDmAllowedWithIngress(senderShip, effectiveDmAllowlist))) {
|
||||
// If owner is configured, queue approval request
|
||||
if (effectiveOwnerShip) {
|
||||
const approval = createPendingApproval({
|
||||
type: "dm",
|
||||
requestingShip: senderShip,
|
||||
messagePreview: sliceUtf16Safe(messageText, 0, 100),
|
||||
originalMessage: {
|
||||
messageId: messageId ?? "",
|
||||
messageText,
|
||||
messageContent: essay.content,
|
||||
timestamp: readNumber(essay, "sent") ?? Date.now(),
|
||||
},
|
||||
});
|
||||
await queueApprovalRequest(approval);
|
||||
} else {
|
||||
runtime.log?.(`[tlon] Blocked DM from ${senderShip}: not in allowlist`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
await processMessage({
|
||||
messageText: await resolveAuthorizedMessageText({
|
||||
rawText,
|
||||
content: essay.content,
|
||||
authorizedForCites: true,
|
||||
resolveAllCites,
|
||||
}),
|
||||
messageId: messageId ?? "",
|
||||
senderShip,
|
||||
messageContent: essay.content, // Pass raw content for media extraction
|
||||
isGroup: false,
|
||||
timestamp: readNumber(essay, "sent") ?? Date.now(),
|
||||
},
|
||||
});
|
||||
},
|
||||
await queueApprovalRequest(approval);
|
||||
} else {
|
||||
runtime.log?.(`[tlon] Blocked DM from ${senderShip}: not in allowlist`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
await processMessage({
|
||||
messageText: await resolveAuthorizedMessageText({
|
||||
rawText,
|
||||
content: essay.content,
|
||||
authorizedForCites: true,
|
||||
resolveAllCites,
|
||||
}),
|
||||
messageId: messageId ?? "",
|
||||
senderShip,
|
||||
messageContent: essay.content, // Pass raw content for media extraction
|
||||
isGroup: false,
|
||||
timestamp: readNumber(essay, "sent") ?? Date.now(),
|
||||
turnAdoptionLifecycle,
|
||||
});
|
||||
void processed;
|
||||
} catch (error: unknown) {
|
||||
runtime.error?.(`[tlon] Error handling chat firehose event: ${formatErrorMessage(error)}`);
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
|
||||
const ingress = createTlonIngressMonitor({
|
||||
accountId: account.accountId,
|
||||
runtime,
|
||||
abortSignal: opts.abortSignal,
|
||||
dispatch: async (source, event, turnAdoptionLifecycle) => {
|
||||
if (source === "channels") {
|
||||
await handleChannelsFirehose(event, turnAdoptionLifecycle);
|
||||
return;
|
||||
}
|
||||
await handleChatFirehose(event, turnAdoptionLifecycle);
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
runtime.log?.("[tlon] Subscribing to firehose updates...");
|
||||
|
||||
@@ -1093,8 +1102,11 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
await api.subscribe({
|
||||
app: "channels",
|
||||
path: "/v2",
|
||||
event: (event) => {
|
||||
void handleChannelsFirehose(event);
|
||||
event: async (event) => {
|
||||
const result = await ingress.receive({ source: "channels", event });
|
||||
if (result.kind === "ignored") {
|
||||
await handleChannelsFirehose(event);
|
||||
}
|
||||
},
|
||||
err: (error) => {
|
||||
runtime.error?.(`[tlon] Channels firehose error: ${String(error)}`);
|
||||
@@ -1109,8 +1121,11 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
await api.subscribe({
|
||||
app: "chat",
|
||||
path: "/v3",
|
||||
event: (event) => {
|
||||
void handleChatFirehose(event);
|
||||
event: async (event) => {
|
||||
const result = await ingress.receive({ source: "chat", event });
|
||||
if (result.kind === "ignored") {
|
||||
await handleChatFirehose(event);
|
||||
}
|
||||
},
|
||||
err: (error) => {
|
||||
runtime.error?.(`[tlon] Chat firehose error: ${String(error)}`);
|
||||
@@ -1474,6 +1489,7 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
|
||||
runtime.log?.("[tlon] All subscriptions registered, connecting to SSE stream...");
|
||||
await api.connect();
|
||||
ingress.start();
|
||||
runtime.log?.("[tlon] Connected! Firehose subscriptions active");
|
||||
|
||||
// Periodically refresh channel discovery
|
||||
@@ -1516,6 +1532,8 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
|
||||
await new Promise(() => {});
|
||||
}
|
||||
} finally {
|
||||
api?.stopReceiving();
|
||||
await ingress.stop();
|
||||
try {
|
||||
await api?.close();
|
||||
} catch (error: unknown) {
|
||||
|
||||
@@ -0,0 +1,375 @@
|
||||
// Tlon durable ingress tests cover append, recovery, tombstones, and guard parity.
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
createChannelIngressQueueForTests,
|
||||
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { UrbitHttpError } from "../urbit/errors.js";
|
||||
import { createTlonIngressMonitor } from "./ingress.js";
|
||||
|
||||
type TlonIngressQueue = NonNullable<Parameters<typeof createTlonIngressMonitor>[0]["queue"]>;
|
||||
type TlonIngressPayload = Parameters<TlonIngressQueue["enqueue"]>[1];
|
||||
type TlonIngressDispatch = Parameters<typeof createTlonIngressMonitor>[0]["dispatch"];
|
||||
|
||||
function channelEvent(params?: { id?: string; nest?: string; text?: string }) {
|
||||
return {
|
||||
nest: params?.nest ?? "chat/~zod/general",
|
||||
response: {
|
||||
post: {
|
||||
id: params?.id ?? "message-1",
|
||||
"r-post": {
|
||||
set: {
|
||||
essay: {
|
||||
author: "~nec",
|
||||
content: [{ text: params?.text ?? "hello" }],
|
||||
sent: 1_700_000_000_000,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function channelReplyEvent(params?: { id?: string; nest?: string; text?: string }) {
|
||||
return {
|
||||
nest: params?.nest ?? "chat/~zod/general",
|
||||
response: {
|
||||
post: {
|
||||
id: "parent-post",
|
||||
"r-post": {
|
||||
reply: {
|
||||
id: params?.id ?? "reply-1",
|
||||
"r-reply": {
|
||||
set: {
|
||||
memo: {
|
||||
author: "~nec",
|
||||
content: [{ text: params?.text ?? "hello" }],
|
||||
sent: 1_700_000_000_000,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function chatEvent(params?: { id?: string; peer?: string; text?: string }) {
|
||||
return {
|
||||
whom: params?.peer ?? "~nec",
|
||||
id: params?.id ?? "dm-1",
|
||||
response: {
|
||||
add: {
|
||||
essay: {
|
||||
author: params?.peer ?? "~nec",
|
||||
content: [{ text: params?.text ?? "hello" }],
|
||||
sent: 1_700_000_000_000,
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function createQueue(stateDir: string, accountId = "default"): TlonIngressQueue {
|
||||
return createChannelIngressQueueForTests<TlonIngressPayload>({
|
||||
channelId: "tlon",
|
||||
accountId,
|
||||
stateDir,
|
||||
});
|
||||
}
|
||||
|
||||
async function withQueue<T>(
|
||||
fn: (queue: TlonIngressQueue, stateDir: string) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-tlon-ingress-"));
|
||||
const stateDir = await fs.realpath(created);
|
||||
try {
|
||||
return await fn(createQueue(stateDir), stateDir);
|
||||
} finally {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
await fs.rm(stateDir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
function startMonitor(queue: TlonIngressQueue, dispatch: TlonIngressDispatch) {
|
||||
const monitor = createTlonIngressMonitor({
|
||||
accountId: "default",
|
||||
queue,
|
||||
dispatch,
|
||||
runtime: { error: vi.fn(), log: vi.fn() },
|
||||
pollIntervalMs: 60_000,
|
||||
adoptionStallTimeoutMs: 5_000,
|
||||
});
|
||||
monitor.start();
|
||||
return monitor;
|
||||
}
|
||||
|
||||
function deferred() {
|
||||
let resolve = () => {};
|
||||
const promise = new Promise<void>((done) => {
|
||||
resolve = done;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe("Tlon durable ingress", () => {
|
||||
it("propagates durable append failure before dispatch", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const appendError = new Error("sqlite unavailable");
|
||||
const failingQueue = {
|
||||
...queue,
|
||||
enqueue: vi.fn().mockRejectedValue(appendError),
|
||||
} satisfies TlonIngressQueue;
|
||||
const dispatch = vi.fn();
|
||||
const monitor = startMonitor(failingQueue, dispatch);
|
||||
try {
|
||||
await expect(monitor.receive({ source: "channels", event: channelEvent() })).rejects.toBe(
|
||||
appendError,
|
||||
);
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("recovers an uncompleted message with a fresh drain exactly once", async () => {
|
||||
await withQueue(async (queue, stateDir) => {
|
||||
const interruptedDispatch = vi.fn(async () => ({ kind: "deferred" }) as const);
|
||||
const interrupted = startMonitor(queue, interruptedDispatch);
|
||||
await interrupted.receive({
|
||||
source: "channels",
|
||||
event: channelEvent({ id: "message-restart" }),
|
||||
});
|
||||
await interrupted.waitForIdle();
|
||||
expect(await queue.listClaims()).toHaveLength(1);
|
||||
await interrupted.stop();
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
|
||||
const recoveredDispatch = vi.fn<TlonIngressDispatch>(async (_source, _event, lifecycle) => {
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
const recovered = startMonitor(createQueue(stateDir), recoveredDispatch);
|
||||
try {
|
||||
await recovered.waitForIdle();
|
||||
expect(recoveredDispatch).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
await recovered.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("retains completion so a duplicate message id cannot dispatch twice", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn<TlonIngressDispatch>(async (_source, _event, lifecycle) => {
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
const event = chatEvent({ id: "dm-completed" });
|
||||
await monitor.receive({ source: "chat", event });
|
||||
await monitor.waitForIdle();
|
||||
await monitor.receive({ source: "chat", event });
|
||||
await monitor.waitForIdle();
|
||||
expect(dispatch).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves the retired guard for reissued or edited message envelopes", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn<TlonIngressDispatch>(async (_source, _event, lifecycle) => {
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
await monitor.receive({
|
||||
source: "channels",
|
||||
event: channelEvent({ id: "message-guard", text: "first delivery" }),
|
||||
});
|
||||
await monitor.waitForIdle();
|
||||
await monitor.receive({
|
||||
source: "channels",
|
||||
event: channelEvent({ id: "message-guard", text: "edited redelivery" }),
|
||||
});
|
||||
await monitor.waitForIdle();
|
||||
expect(dispatch).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("retains completed logical ids by count rather than age", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
let now = 1_700_000_000_000;
|
||||
vi.spyOn(Date, "now").mockImplementation(() => now);
|
||||
const dispatch = vi.fn<TlonIngressDispatch>(async (_source, _event, lifecycle) => {
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
const original = channelEvent({ id: "message-aged" });
|
||||
await monitor.receive({ source: "channels", event: original });
|
||||
await monitor.waitForIdle();
|
||||
|
||||
now += 31 * 24 * 60 * 60 * 1_000;
|
||||
await monitor.receive({
|
||||
source: "channels",
|
||||
event: channelEvent({ id: "message-prune-trigger" }),
|
||||
});
|
||||
await monitor.waitForIdle();
|
||||
await monitor.receive({ source: "channels", event: original });
|
||||
await monitor.waitForIdle();
|
||||
|
||||
expect(dispatch).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("stores raw envelopes in per-conversation lanes", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn(async () => ({ kind: "deferred" }) as const);
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
const group = channelEvent({ id: "message-raw", nest: "chat/~zod/ops" });
|
||||
try {
|
||||
await monitor.receive({ source: "channels", event: group });
|
||||
await monitor.waitForIdle();
|
||||
expect(await queue.listClaims()).toEqual([
|
||||
expect.objectContaining({
|
||||
id: "message-raw",
|
||||
laneKey: "group:chat/~zod/ops",
|
||||
payload: expect.objectContaining({ rawEvent: JSON.stringify(group) }),
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("uses the nested reply id rather than the parent post id", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn(async () => ({ kind: "deferred" }) as const);
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
const reply = channelReplyEvent({ id: "reply-stable", nest: "chat/~zod/ops" });
|
||||
try {
|
||||
await monitor.receive({ source: "channels", event: reply });
|
||||
await monitor.waitForIdle();
|
||||
expect(await queue.listClaims()).toEqual([
|
||||
expect.objectContaining({ id: "reply-stable", laneKey: "group:chat/~zod/ops" }),
|
||||
]);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("dead-letters malformed persisted payloads and authentication failures", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
await queue.enqueue(
|
||||
"message-malformed",
|
||||
{ version: 1, receivedAt: 1, source: "channels", rawEvent: "{" },
|
||||
{ receivedAt: 1, laneKey: "group:chat/~zod/general" },
|
||||
);
|
||||
const dispatch = vi.fn<TlonIngressDispatch>(async () => {
|
||||
throw new UrbitHttpError({ operation: "Poke", status: 401 });
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
await monitor.waitForIdle();
|
||||
expect((await queue.enqueue("message-malformed", {} as TlonIngressPayload)).kind).toBe(
|
||||
"failed",
|
||||
);
|
||||
|
||||
await monitor.receive({
|
||||
source: "chat",
|
||||
event: chatEvent({ id: "message-auth" }),
|
||||
});
|
||||
await monitor.waitForIdle();
|
||||
expect((await queue.enqueue("message-auth", {} as TlonIngressPayload)).kind).toBe("failed");
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps unrelated authentication failures retryable", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn<TlonIngressDispatch>(async () => {
|
||||
throw Object.assign(new Error("model credentials expired"), { status: 401 });
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
await monitor.receive({
|
||||
source: "chat",
|
||||
event: chatEvent({ id: "message-provider-auth" }),
|
||||
});
|
||||
await monitor.waitForIdle();
|
||||
expect((await queue.listPending({ limit: "all" })).map((record) => record.id)).toEqual([
|
||||
"message-provider-auth",
|
||||
]);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("waits for admitted work and leaves it pending on repeated stop", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const stored = deferred();
|
||||
const release = deferred();
|
||||
const enqueue = queue.enqueue.bind(queue);
|
||||
queue.enqueue = async (...args) => {
|
||||
const result = await enqueue(...args);
|
||||
stored.resolve();
|
||||
await release.promise;
|
||||
return result;
|
||||
};
|
||||
const dispatch = vi.fn();
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
const admission = monitor.receive({
|
||||
source: "channels",
|
||||
event: channelEvent({ id: "message-stop-1" }),
|
||||
});
|
||||
await stored.promise;
|
||||
const queuedAdmission = monitor.receive({
|
||||
source: "channels",
|
||||
event: channelEvent({ id: "message-stop-2" }),
|
||||
});
|
||||
|
||||
let stopSettled = false;
|
||||
const stopping = monitor.stop().then(() => {
|
||||
stopSettled = true;
|
||||
});
|
||||
const stoppingAgain = monitor.stop();
|
||||
await Promise.resolve();
|
||||
expect(stopSettled).toBe(false);
|
||||
await expect(
|
||||
monitor.receive({ source: "channels", event: channelEvent({ id: "message-too-late" }) }),
|
||||
).rejects.toThrow("stopped before dispatch adoption");
|
||||
|
||||
release.resolve();
|
||||
await Promise.all([admission, queuedAdmission, stopping, stoppingAgain]);
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
expect((await queue.listPending({ limit: "all" })).map((record) => record.id)).toEqual([
|
||||
"message-stop-1",
|
||||
"message-stop-2",
|
||||
]);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,366 @@
|
||||
// Tlon plugin module owns raw Urbit firehose durable ingress mapping and draining.
|
||||
import {
|
||||
createChannelIngressDrain,
|
||||
DEFAULT_INGRESS_ADOPTION_STALL_MS,
|
||||
DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
|
||||
DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
|
||||
type ChannelIngressQueue,
|
||||
} from "openclaw/plugin-sdk/channel-outbound";
|
||||
import { collectErrorGraphCandidates, formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
|
||||
import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env";
|
||||
import { getTlonRuntime } from "../runtime.js";
|
||||
import { UrbitAuthError, UrbitHttpError } from "../urbit/errors.js";
|
||||
|
||||
const TLON_INGRESS_PAYLOAD_VERSION = 1;
|
||||
const TLON_INGRESS_POLL_INTERVAL_MS = 1_000;
|
||||
const TLON_INGRESS_PRUNE_INTERVAL_MS = 60 * 60 * 1_000;
|
||||
const TLON_INGRESS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1_000;
|
||||
// Preserve the retired process-local guard's full 2,000-message key window.
|
||||
const TLON_INGRESS_TOMBSTONE_MAX_ENTRIES = 2_000;
|
||||
|
||||
export type TlonIngressLifecycle = {
|
||||
abortSignal: AbortSignal;
|
||||
onAdopted: () => void | Promise<void>;
|
||||
onDeferred: () => void;
|
||||
onAdoptionFinalizing: () => void;
|
||||
onAbandoned: () => void | Promise<void>;
|
||||
};
|
||||
|
||||
type TlonIngressSource = "channels" | "chat";
|
||||
|
||||
type TlonIngressPayload = {
|
||||
version: 1;
|
||||
receivedAt: number;
|
||||
source: TlonIngressSource;
|
||||
rawEvent: string;
|
||||
};
|
||||
|
||||
type TlonIngressDrain = {
|
||||
drainOnce: (options?: { shouldStop?: () => boolean }) => Promise<{ started: number }>;
|
||||
waitForIdle: () => Promise<void>;
|
||||
dispose: () => void;
|
||||
};
|
||||
|
||||
type TlonIngressDispatchResult =
|
||||
| { kind: "completed" }
|
||||
| { kind: "deferred" }
|
||||
| { kind: "failed-retryable"; error: unknown };
|
||||
|
||||
type TlonIngressDispatch = (
|
||||
source: TlonIngressSource,
|
||||
event: unknown,
|
||||
lifecycle: TlonIngressLifecycle,
|
||||
) => Promise<TlonIngressDispatchResult | void> | TlonIngressDispatchResult | void;
|
||||
|
||||
class TlonIngressPermanentError extends Error {
|
||||
constructor(
|
||||
readonly reason: "invalid-event" | "tlon-auth",
|
||||
message: string,
|
||||
options?: ErrorOptions,
|
||||
) {
|
||||
super(message, options);
|
||||
this.name = "TlonIngressPermanentError";
|
||||
}
|
||||
}
|
||||
|
||||
class TlonIngressShutdownError extends Error {
|
||||
constructor() {
|
||||
super("Tlon ingress stopped before dispatch adoption.");
|
||||
this.name = "TlonIngressShutdownError";
|
||||
}
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function nonEmptyString(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim() ? value.trim() : null;
|
||||
}
|
||||
|
||||
function inspectChannelsEvent(event: unknown): { eventId: string; laneKey: string } | null {
|
||||
const envelope = isRecord(event) ? event : null;
|
||||
const nest = nonEmptyString(envelope?.nest);
|
||||
const response = isRecord(envelope?.response) ? envelope.response : null;
|
||||
const post = isRecord(response?.post) ? response.post : null;
|
||||
const rPost = isRecord(post?.["r-post"]) ? post["r-post"] : null;
|
||||
const set = isRecord(rPost?.set) ? rPost.set : null;
|
||||
const reply = isRecord(rPost?.reply) ? rPost.reply : null;
|
||||
const rReply = isRecord(reply?.["r-reply"]) ? reply["r-reply"] : null;
|
||||
const replySet = isRecord(rReply?.set) ? rReply.set : null;
|
||||
if (!nest || (!isRecord(set?.essay) && !isRecord(replySet?.memo))) {
|
||||
return null;
|
||||
}
|
||||
const eventId = nonEmptyString(isRecord(replySet?.memo) ? reply?.id : post?.id);
|
||||
return eventId ? { eventId, laneKey: `group:${nest}` } : null;
|
||||
}
|
||||
|
||||
function inspectChatEvent(event: unknown): { eventId: string; laneKey: string } | null {
|
||||
const envelope = isRecord(event) ? event : null;
|
||||
const response = isRecord(envelope?.response) ? envelope.response : null;
|
||||
const add = isRecord(response?.add) ? response.add : null;
|
||||
const essay = isRecord(add?.essay) ? add.essay : null;
|
||||
const eventId = nonEmptyString(envelope?.id);
|
||||
if (!essay || !eventId) {
|
||||
return null;
|
||||
}
|
||||
const whom = isRecord(envelope?.whom) ? nonEmptyString(envelope.whom.ship) : null;
|
||||
const peer = nonEmptyString(envelope?.whom) ?? whom ?? nonEmptyString(essay.author);
|
||||
return { eventId, laneKey: peer ? `direct:${peer}` : `event:${eventId}` };
|
||||
}
|
||||
|
||||
function inspectTlonIngressEvent(
|
||||
source: TlonIngressSource,
|
||||
event: unknown,
|
||||
): { eventId: string; laneKey: string } | null {
|
||||
// Urbit SSE ids belong to a disposable HTTP channel. The message id inside
|
||||
// each firehose envelope survives resubscription and preserves the retired guard key.
|
||||
return source === "channels" ? inspectChannelsEvent(event) : inspectChatEvent(event);
|
||||
}
|
||||
|
||||
function parseClaimedEvent(payload: TlonIngressPayload, claimedId: string): unknown {
|
||||
if (
|
||||
payload.version !== TLON_INGRESS_PAYLOAD_VERSION ||
|
||||
(payload.source !== "channels" && payload.source !== "chat") ||
|
||||
typeof payload.rawEvent !== "string"
|
||||
) {
|
||||
throw new TlonIngressPermanentError(
|
||||
"invalid-event",
|
||||
`Tlon ingress row ${claimedId} has an invalid payload.`,
|
||||
);
|
||||
}
|
||||
let event: unknown;
|
||||
try {
|
||||
event = JSON.parse(payload.rawEvent);
|
||||
} catch (error) {
|
||||
throw new TlonIngressPermanentError(
|
||||
"invalid-event",
|
||||
`Tlon ingress row ${claimedId} contains invalid JSON.`,
|
||||
{ cause: error },
|
||||
);
|
||||
}
|
||||
const facts = inspectTlonIngressEvent(payload.source, event);
|
||||
if (!facts || facts.eventId !== claimedId) {
|
||||
throw new TlonIngressPermanentError(
|
||||
"invalid-event",
|
||||
`Tlon ingress row ${claimedId} has invalid message identity.`,
|
||||
);
|
||||
}
|
||||
return event;
|
||||
}
|
||||
|
||||
function resolveTlonIngressNonRetryableFailure(error: unknown) {
|
||||
if (error instanceof TlonIngressPermanentError) {
|
||||
return { reason: error.reason, message: error.message };
|
||||
}
|
||||
for (const candidate of collectErrorGraphCandidates(error, (current) => [current.cause])) {
|
||||
if (
|
||||
candidate instanceof UrbitAuthError ||
|
||||
(candidate instanceof UrbitHttpError &&
|
||||
(candidate.status === 401 || candidate.status === 403))
|
||||
) {
|
||||
return { reason: "tlon-auth", message: formatErrorMessage(candidate) };
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
type TlonIngressMonitor = {
|
||||
receive: (params: {
|
||||
source: TlonIngressSource;
|
||||
event: unknown;
|
||||
}) => Promise<{ kind: "accepted" } | { kind: "ignored" }>;
|
||||
start: () => void;
|
||||
stop: () => Promise<void>;
|
||||
waitForIdle: () => Promise<void>;
|
||||
};
|
||||
|
||||
export function createTlonIngressMonitor(options: {
|
||||
accountId: string;
|
||||
queue?: ChannelIngressQueue<TlonIngressPayload>;
|
||||
dispatch: TlonIngressDispatch;
|
||||
runtime: Pick<RuntimeEnv, "error" | "log">;
|
||||
pollIntervalMs?: number;
|
||||
adoptionStallTimeoutMs?: number;
|
||||
abortSignal?: AbortSignal;
|
||||
}): TlonIngressMonitor {
|
||||
let queue = options.queue;
|
||||
let drain: TlonIngressDrain | undefined;
|
||||
let accepting = true;
|
||||
let running = false;
|
||||
let stopped = false;
|
||||
let requested = false;
|
||||
let pumping: Promise<void> | undefined;
|
||||
let pollTimer: ReturnType<typeof setInterval> | undefined;
|
||||
let lastPrunedAt = 0;
|
||||
let stopPromise: Promise<void> | undefined;
|
||||
|
||||
const getQueue = (): ChannelIngressQueue<TlonIngressPayload> => {
|
||||
queue ??= getTlonRuntime().state.openChannelIngressQueue<TlonIngressPayload>({
|
||||
accountId: options.accountId,
|
||||
});
|
||||
return queue;
|
||||
};
|
||||
|
||||
const getDrain = (): TlonIngressDrain => {
|
||||
drain ??= createChannelIngressDrain<TlonIngressPayload>({
|
||||
queue: getQueue(),
|
||||
adoptionStallTimeoutMs: options.adoptionStallTimeoutMs ?? DEFAULT_INGRESS_ADOPTION_STALL_MS,
|
||||
retryPolicy: {
|
||||
maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
|
||||
deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
|
||||
},
|
||||
resolveNonRetryableFailure: resolveTlonIngressNonRetryableFailure,
|
||||
...(options.abortSignal ? { abortSignal: options.abortSignal } : {}),
|
||||
onLog: (message) => options.runtime.log?.(`tlon ${message}`),
|
||||
dispatchClaimedEvent: async (record, lifecycle) => {
|
||||
if (!running || lifecycle.abortSignal.aborted || options.abortSignal?.aborted) {
|
||||
return { kind: "failed-retryable", error: new TlonIngressShutdownError() };
|
||||
}
|
||||
const event = parseClaimedEvent(record.payload, record.id);
|
||||
try {
|
||||
const result = await options.dispatch(record.payload.source, event, lifecycle);
|
||||
return !running || options.abortSignal?.aborted
|
||||
? { kind: "failed-retryable", error: new TlonIngressShutdownError() }
|
||||
: result;
|
||||
} catch (error) {
|
||||
if (!running || options.abortSignal?.aborted) {
|
||||
return { kind: "failed-retryable", error };
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
});
|
||||
return drain;
|
||||
};
|
||||
|
||||
const pruneIfDue = async (): Promise<void> => {
|
||||
const now = Date.now();
|
||||
if (now - lastPrunedAt < TLON_INGRESS_PRUNE_INTERVAL_MS) {
|
||||
return;
|
||||
}
|
||||
await getQueue().prune({
|
||||
completedMaxEntries: TLON_INGRESS_TOMBSTONE_MAX_ENTRIES,
|
||||
failedTtlMs: TLON_INGRESS_FAILED_TTL_MS,
|
||||
failedMaxEntries: TLON_INGRESS_TOMBSTONE_MAX_ENTRIES,
|
||||
now,
|
||||
});
|
||||
lastPrunedAt = now;
|
||||
};
|
||||
|
||||
const runPump = async (): Promise<void> => {
|
||||
try {
|
||||
for (;;) {
|
||||
requested = false;
|
||||
await pruneIfDue();
|
||||
// stop() may race the async prune; never create a live drain afterward.
|
||||
if (!running) {
|
||||
break;
|
||||
}
|
||||
const activeDrain = getDrain();
|
||||
const { started } = await activeDrain.drainOnce({ shouldStop: () => !running });
|
||||
await activeDrain.waitForIdle();
|
||||
if (!running || (!requested && started === 0)) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
options.runtime.error?.(`tlon ingress drain failed: ${formatErrorMessage(error)}`);
|
||||
} finally {
|
||||
pumping = undefined;
|
||||
if (running && requested) {
|
||||
requestDrain();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const requestDrain = (): void => {
|
||||
requested = true;
|
||||
if (!running || pumping) {
|
||||
return;
|
||||
}
|
||||
pumping = runPump();
|
||||
};
|
||||
|
||||
// Stream callbacks are awaited, but serialize direct test/caller admissions too.
|
||||
let admissionTail: Promise<void> = Promise.resolve();
|
||||
|
||||
const admitOnce = async (source: TlonIngressSource, event: unknown): Promise<boolean> => {
|
||||
const facts = inspectTlonIngressEvent(source, event);
|
||||
if (!facts) {
|
||||
return false;
|
||||
}
|
||||
const receivedAt = Date.now();
|
||||
await getQueue().enqueue(
|
||||
facts.eventId,
|
||||
{
|
||||
version: TLON_INGRESS_PAYLOAD_VERSION,
|
||||
receivedAt,
|
||||
source,
|
||||
rawEvent: JSON.stringify(event),
|
||||
},
|
||||
{ receivedAt, laneKey: facts.laneKey },
|
||||
);
|
||||
requestDrain();
|
||||
return true;
|
||||
};
|
||||
|
||||
return {
|
||||
receive: async ({ source, event }) => {
|
||||
if (!accepting) {
|
||||
throw new TlonIngressShutdownError();
|
||||
}
|
||||
let accepted = false;
|
||||
const admission = admissionTail.then(async () => {
|
||||
accepted = await admitOnce(source, event);
|
||||
});
|
||||
admissionTail = admission.catch(() => undefined);
|
||||
await admission;
|
||||
return { kind: accepted ? "accepted" : "ignored" };
|
||||
},
|
||||
start: () => {
|
||||
if (running || stopped) {
|
||||
return;
|
||||
}
|
||||
running = true;
|
||||
pollTimer = setInterval(
|
||||
requestDrain,
|
||||
options.pollIntervalMs ?? TLON_INGRESS_POLL_INTERVAL_MS,
|
||||
);
|
||||
pollTimer.unref?.();
|
||||
requestDrain();
|
||||
},
|
||||
stop: async () => {
|
||||
if (stopPromise) {
|
||||
await stopPromise;
|
||||
return;
|
||||
}
|
||||
accepting = false;
|
||||
running = false;
|
||||
stopped = true;
|
||||
if (pollTimer) {
|
||||
clearInterval(pollTimer);
|
||||
pollTimer = undefined;
|
||||
}
|
||||
stopPromise = (async () => {
|
||||
await admissionTail;
|
||||
drain?.dispose();
|
||||
await pumping;
|
||||
// A pump may have lazily created the drain before observing running=false.
|
||||
drain?.dispose();
|
||||
await drain?.waitForIdle();
|
||||
})();
|
||||
await stopPromise;
|
||||
},
|
||||
waitForIdle: async () => {
|
||||
for (;;) {
|
||||
const activePump = pumping;
|
||||
if (!activePump) {
|
||||
break;
|
||||
}
|
||||
await activePump;
|
||||
}
|
||||
await drain?.waitForIdle();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1,59 +0,0 @@
|
||||
// Tlon tests cover processed messages plugin behavior.
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
createProcessedMessageTracker,
|
||||
runWithProcessedMessageClaim,
|
||||
} from "./processed-messages.js";
|
||||
|
||||
describe("createProcessedMessageTracker", () => {
|
||||
it("dedupes and evicts oldest entries", () => {
|
||||
const tracker = createProcessedMessageTracker(3);
|
||||
|
||||
expect(tracker.mark("a")).toBe(true);
|
||||
expect(tracker.mark("a")).toBe(false);
|
||||
expect(tracker.has("a")).toBe(true);
|
||||
|
||||
tracker.mark("b");
|
||||
tracker.mark("c");
|
||||
expect(tracker.size()).toBe(3);
|
||||
|
||||
tracker.mark("d");
|
||||
expect(tracker.size()).toBe(3);
|
||||
expect(tracker.has("a")).toBe(false);
|
||||
expect(tracker.has("b")).toBe(true);
|
||||
expect(tracker.has("c")).toBe(true);
|
||||
expect(tracker.has("d")).toBe(true);
|
||||
});
|
||||
|
||||
it("releases failed claims so retries can run again", async () => {
|
||||
const tracker = createProcessedMessageTracker();
|
||||
|
||||
await expect(
|
||||
runWithProcessedMessageClaim({
|
||||
tracker,
|
||||
id: "evt-1",
|
||||
task: async () => {
|
||||
throw new Error("boom");
|
||||
},
|
||||
}),
|
||||
).rejects.toThrow("boom");
|
||||
|
||||
expect(tracker.has("evt-1")).toBe(false);
|
||||
expect(tracker.claim("evt-1")).toEqual({ kind: "claimed" });
|
||||
});
|
||||
|
||||
it("keeps successful claims deduped", async () => {
|
||||
const tracker = createProcessedMessageTracker();
|
||||
|
||||
await expect(
|
||||
runWithProcessedMessageClaim({
|
||||
tracker,
|
||||
id: "evt-2",
|
||||
task: async () => undefined,
|
||||
}),
|
||||
).resolves.toEqual({ kind: "processed", value: undefined });
|
||||
|
||||
expect(tracker.has("evt-2")).toBe(true);
|
||||
expect(tracker.claim("evt-2")).toEqual({ kind: "duplicate" });
|
||||
});
|
||||
});
|
||||
@@ -1,90 +0,0 @@
|
||||
// Tlon plugin module implements processed messages behavior.
|
||||
import { createDedupeCache } from "../../runtime-api.js";
|
||||
|
||||
type ProcessedMessageTracker = {
|
||||
claim: (id?: string | null) => { kind: "claimed" } | { kind: "duplicate" };
|
||||
commit: (id?: string | null) => void;
|
||||
release: (id?: string | null) => void;
|
||||
mark: (id?: string | null) => boolean;
|
||||
has: (id?: string | null) => boolean;
|
||||
size: () => number;
|
||||
};
|
||||
|
||||
export function createProcessedMessageTracker(limit = 2000): ProcessedMessageTracker {
|
||||
const dedupe = createDedupeCache({ ttlMs: 0, maxSize: limit });
|
||||
const inFlight = new Set<string>();
|
||||
|
||||
const claim = (id?: string | null) => {
|
||||
const trimmed = id?.trim();
|
||||
if (!trimmed) {
|
||||
return { kind: "claimed" } as const;
|
||||
}
|
||||
if (inFlight.has(trimmed) || dedupe.peek(trimmed)) {
|
||||
return { kind: "duplicate" } as const;
|
||||
}
|
||||
inFlight.add(trimmed);
|
||||
return { kind: "claimed" } as const;
|
||||
};
|
||||
|
||||
const commit = (id?: string | null) => {
|
||||
const trimmed = id?.trim();
|
||||
if (!trimmed) {
|
||||
return;
|
||||
}
|
||||
inFlight.delete(trimmed);
|
||||
dedupe.check(trimmed);
|
||||
};
|
||||
|
||||
const release = (id?: string | null) => {
|
||||
const trimmed = id?.trim();
|
||||
if (!trimmed) {
|
||||
return;
|
||||
}
|
||||
inFlight.delete(trimmed);
|
||||
};
|
||||
|
||||
const mark = (id?: string | null) => {
|
||||
const claimed = claim(id);
|
||||
if (claimed.kind === "duplicate") {
|
||||
return false;
|
||||
}
|
||||
commit(id);
|
||||
return true;
|
||||
};
|
||||
|
||||
const has = (id?: string | null) => {
|
||||
const trimmed = id?.trim();
|
||||
if (!trimmed) {
|
||||
return false;
|
||||
}
|
||||
return dedupe.peek(trimmed);
|
||||
};
|
||||
|
||||
return {
|
||||
claim,
|
||||
commit,
|
||||
release,
|
||||
mark,
|
||||
has,
|
||||
size: () => dedupe.size(),
|
||||
};
|
||||
}
|
||||
|
||||
export async function runWithProcessedMessageClaim<T>(params: {
|
||||
tracker: ProcessedMessageTracker;
|
||||
id?: string | null;
|
||||
task: () => Promise<T>;
|
||||
}): Promise<{ kind: "processed"; value: T } | { kind: "duplicate" }> {
|
||||
const claim = params.tracker.claim(params.id);
|
||||
if (claim.kind === "duplicate") {
|
||||
return claim;
|
||||
}
|
||||
try {
|
||||
const value = await params.task();
|
||||
params.tracker.commit(params.id);
|
||||
return { kind: "processed", value };
|
||||
} catch (error) {
|
||||
params.tracker.release(params.id);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
@@ -66,7 +66,11 @@ export async function pokeUrbitChannel(
|
||||
const errorText = await readResponseTextLimited(response, TLON_ERROR_BODY_LIMIT_BYTES).catch(
|
||||
() => "",
|
||||
);
|
||||
throw new Error(`Poke failed: ${response.status}${errorText ? ` - ${errorText}` : ""}`);
|
||||
throw new UrbitHttpError({
|
||||
operation: "Poke",
|
||||
status: response.status,
|
||||
bodyText: errorText || undefined,
|
||||
});
|
||||
}
|
||||
return pokeId;
|
||||
} finally {
|
||||
@@ -95,7 +99,10 @@ export async function scryUrbitPath(
|
||||
|
||||
try {
|
||||
if (!response.ok) {
|
||||
throw new Error(`Scry failed: ${response.status} for path ${params.path}`);
|
||||
throw new UrbitHttpError({
|
||||
operation: `Scry for path ${params.path}`,
|
||||
status: response.status,
|
||||
});
|
||||
}
|
||||
// Successful scry bodies come from a remote Urbit and have no protocol size bound.
|
||||
// Keep the shared JSON ceiling while retaining the path needed to identify the endpoint.
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
import { Readable } from "node:stream";
|
||||
import { MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { ensureUrbitChannelOpen } from "./channel-ops.js";
|
||||
import { urbitFetch } from "./fetch.js";
|
||||
import { UrbitSSEClient } from "./sse-client.js";
|
||||
|
||||
@@ -218,23 +219,88 @@ describe("UrbitSSEClient", () => {
|
||||
|
||||
expect(client.reconnectAttempts).toBe(0);
|
||||
});
|
||||
|
||||
it("reopens the same HTTP channel so unacked events can replay", async () => {
|
||||
vi.useFakeTimers();
|
||||
const mockUrbitFetch = vi.mocked(urbitFetch);
|
||||
mockUrbitFetch.mockResolvedValue({
|
||||
response: { ok: true, status: 200, body: null } as unknown as Response,
|
||||
finalUrl: "https://example.com",
|
||||
release: vi.fn().mockResolvedValue(undefined),
|
||||
});
|
||||
const onReconnect = vi.fn();
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123", {
|
||||
reconnectDelay: 1,
|
||||
maxReconnectDelay: 1,
|
||||
onReconnect,
|
||||
});
|
||||
const channelId = client.channelId;
|
||||
|
||||
const reconnecting = client.attemptReconnect();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await reconnecting;
|
||||
|
||||
expect(client.channelId).toBe(channelId);
|
||||
expect(onReconnect).toHaveBeenCalledOnce();
|
||||
const callArgs = requireFirstMockCall(mockUrbitFetch.mock.calls, "stream reconnect")[0] as
|
||||
| Parameters<typeof urbitFetch>[0]
|
||||
| undefined;
|
||||
expect(callArgs?.path).toBe(`/~/channel/${channelId}`);
|
||||
expect(callArgs?.init?.method).toBe("GET");
|
||||
expect(client.reconnectAttempts).toBe(0);
|
||||
});
|
||||
|
||||
it("replaces a server-deleted HTTP channel and restores subscriptions", async () => {
|
||||
vi.useFakeTimers();
|
||||
const mockUrbitFetch = vi.mocked(urbitFetch);
|
||||
mockUrbitFetch
|
||||
.mockResolvedValueOnce({
|
||||
response: { ok: false, status: 404, body: null } as unknown as Response,
|
||||
finalUrl: "https://example.com",
|
||||
release: vi.fn().mockResolvedValue(undefined),
|
||||
})
|
||||
.mockResolvedValueOnce({
|
||||
response: { ok: true, status: 200, body: null } as unknown as Response,
|
||||
finalUrl: "https://example.com",
|
||||
release: vi.fn().mockResolvedValue(undefined),
|
||||
});
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123", {
|
||||
reconnectDelay: 1,
|
||||
maxReconnectDelay: 1,
|
||||
});
|
||||
await client.subscribe({ app: "chat", path: "/dm/~zod", event: () => {} });
|
||||
const deletedChannelId = client.channelId;
|
||||
|
||||
const reconnecting = client.attemptReconnect();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await reconnecting;
|
||||
|
||||
expect(client.channelId).not.toBe(deletedChannelId);
|
||||
expect(ensureUrbitChannelOpen).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ channelId: client.channelId }),
|
||||
expect.objectContaining({ createBody: client.subscriptions }),
|
||||
);
|
||||
expect(mockUrbitFetch).toHaveBeenCalledTimes(2);
|
||||
expect(mockUrbitFetch.mock.calls[1]?.[0].path).toBe(`/~/channel/${client.channelId}`);
|
||||
expect(client.reconnectAttempts).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("event acking", () => {
|
||||
it("logs malformed SSE JSON with an owned parser error", () => {
|
||||
it("logs malformed SSE JSON with an owned parser error", async () => {
|
||||
const logger = { error: vi.fn() };
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123", {
|
||||
logger,
|
||||
});
|
||||
|
||||
client.processEvent("id: 1\ndata: {not json");
|
||||
await client.processEvent("id: 1\ndata: {not json");
|
||||
|
||||
expect(logger.error).toHaveBeenCalledWith(
|
||||
"Error parsing SSE event: Error: Tlon Urbit SSE event was malformed JSON",
|
||||
);
|
||||
});
|
||||
|
||||
it("guards JSON.parse against oversized SSE payload to prevent OOM", () => {
|
||||
it("guards JSON.parse against oversized SSE payload to prevent OOM", async () => {
|
||||
const errors: string[] = [];
|
||||
const logger = { error: (msg: string) => errors.push(msg) };
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123", { logger });
|
||||
@@ -247,7 +313,7 @@ describe("UrbitSSEClient", () => {
|
||||
const padLen = cap + 1024 - jsonOverhead;
|
||||
const hugeJson = prefix + "A".repeat(padLen) + suffix;
|
||||
|
||||
client.processEvent(`id: 1\ndata: ${hugeJson}`);
|
||||
await client.processEvent(`id: 1\ndata: ${hugeJson}`);
|
||||
|
||||
expect(errors).toHaveLength(1);
|
||||
expect(errors[0]).toBe(
|
||||
@@ -255,7 +321,7 @@ describe("UrbitSSEClient", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("accepts SSE payload at the 16 MiB boundary", () => {
|
||||
it("accepts SSE payload at the 16 MiB boundary", async () => {
|
||||
const cap = 16 * 1024 * 1024;
|
||||
// Allocate valid JSON whose byteLength exactly equals cap.
|
||||
const prefix = '{"json":{"ok":true,"x":"';
|
||||
@@ -268,7 +334,7 @@ describe("UrbitSSEClient", () => {
|
||||
const handler = vi.fn();
|
||||
client.eventHandlers.set(1, { event: handler });
|
||||
|
||||
client.processEvent(`id: 1\ndata: ${hugeJson}`);
|
||||
await client.processEvent(`id: 1\ndata: ${hugeJson}`);
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
const payload = handler.mock.calls[0]?.[0] as { ok?: boolean; x?: string } | undefined;
|
||||
expect(payload?.ok).toBe(true);
|
||||
@@ -372,13 +438,102 @@ describe("UrbitSSEClient", () => {
|
||||
});
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123");
|
||||
|
||||
client.processEvent('id: 25abc\ndata: {"json":{"ok":true}}');
|
||||
await Promise.resolve();
|
||||
await client.processEvent('id: 25abc\ndata: {"json":{"ok":true}}');
|
||||
|
||||
expect(mockUrbitFetch).not.toHaveBeenCalled();
|
||||
expect((client as unknown as { lastHeardEventId: number }).lastHeardEventId).toBe(-1);
|
||||
});
|
||||
|
||||
it("waits for durable admission before acknowledging the transport event", async () => {
|
||||
const mockUrbitFetch = vi.mocked(urbitFetch);
|
||||
mockUrbitFetch.mockResolvedValue({
|
||||
response: { ok: true, status: 200 } as unknown as Response,
|
||||
finalUrl: "https://example.com",
|
||||
release: vi.fn().mockResolvedValue(undefined),
|
||||
});
|
||||
let releaseAdmission = () => {};
|
||||
const admission = new Promise<void>((resolve) => {
|
||||
releaseAdmission = resolve;
|
||||
});
|
||||
const handler = vi.fn(async () => {
|
||||
await admission;
|
||||
});
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123");
|
||||
client.eventHandlers.set(1, { event: handler });
|
||||
|
||||
const processing = client.processEvent('id: 20\ndata: {"id":1,"json":{"ok":true}}');
|
||||
await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(1));
|
||||
expect(mockUrbitFetch).not.toHaveBeenCalled();
|
||||
|
||||
releaseAdmission();
|
||||
await processing;
|
||||
expect(mockUrbitFetch).toHaveBeenCalledTimes(1);
|
||||
const body = mockUrbitFetch.mock.calls[0]?.[0].init?.body;
|
||||
if (typeof body !== "string") {
|
||||
throw new Error("Expected string ACK request body");
|
||||
}
|
||||
expect(JSON.parse(body)).toEqual([{ id: expect.any(Number), action: "ack", "event-id": 20 }]);
|
||||
});
|
||||
|
||||
it("does not acknowledge a failed durable admission", async () => {
|
||||
const mockUrbitFetch = vi.mocked(urbitFetch);
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123");
|
||||
client.eventHandlers.set(1, {
|
||||
event: async () => {
|
||||
throw new Error("sqlite unavailable");
|
||||
},
|
||||
});
|
||||
|
||||
await expect(
|
||||
client.processEvent('id: 20\ndata: {"id":1,"json":{"ok":true}}'),
|
||||
).rejects.toThrow("sqlite unavailable");
|
||||
expect(mockUrbitFetch).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not advance the ack watermark when the ack request fails", async () => {
|
||||
const mockUrbitFetch = vi.mocked(urbitFetch);
|
||||
mockUrbitFetch.mockResolvedValue({
|
||||
response: { ok: false, status: 503 } as unknown as Response,
|
||||
finalUrl: "https://example.com",
|
||||
release: vi.fn().mockResolvedValue(undefined),
|
||||
});
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123");
|
||||
client.eventHandlers.set(1, { event: vi.fn() });
|
||||
|
||||
await expect(
|
||||
client.processEvent('id: 20\ndata: {"id":1,"json":{"ok":true}}'),
|
||||
).rejects.toThrow("Ack failed with status 503");
|
||||
expect(
|
||||
(client as unknown as { lastAcknowledgedEventId: number }).lastAcknowledgedEventId,
|
||||
).toBe(-1);
|
||||
});
|
||||
|
||||
it("retries a failed ack when the unacknowledged event replays", async () => {
|
||||
const mockUrbitFetch = vi.mocked(urbitFetch);
|
||||
mockUrbitFetch
|
||||
.mockResolvedValueOnce({
|
||||
response: { ok: false, status: 503 } as unknown as Response,
|
||||
finalUrl: "https://example.com",
|
||||
release: vi.fn().mockResolvedValue(undefined),
|
||||
})
|
||||
.mockResolvedValueOnce({
|
||||
response: { ok: true, status: 204 } as unknown as Response,
|
||||
finalUrl: "https://example.com",
|
||||
release: vi.fn().mockResolvedValue(undefined),
|
||||
});
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123");
|
||||
client.eventHandlers.set(1, { event: vi.fn() });
|
||||
const event = 'id: 20\ndata: {"id":1,"json":{"ok":true}}';
|
||||
|
||||
await expect(client.processEvent(event)).rejects.toThrow("Ack failed with status 503");
|
||||
await expect(client.processEvent(event)).resolves.toBeUndefined();
|
||||
|
||||
expect(mockUrbitFetch).toHaveBeenCalledTimes(2);
|
||||
expect(
|
||||
(client as unknown as { lastAcknowledgedEventId: number }).lastAcknowledgedEventId,
|
||||
).toBe(20);
|
||||
});
|
||||
|
||||
it("tracks lastHeardEventId and ackThreshold", () => {
|
||||
const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123");
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ import { readResponseTextLimited } from "openclaw/plugin-sdk/provider-http";
|
||||
import type { LookupFn, SsrFPolicy } from "openclaw/plugin-sdk/ssrf-runtime";
|
||||
import { ensureUrbitChannelOpen, pokeUrbitChannel, scryUrbitPath } from "./channel-ops.js";
|
||||
import { getUrbitContext, normalizeUrbitCookie } from "./context.js";
|
||||
import { UrbitHttpError } from "./errors.js";
|
||||
import { urbitFetch } from "./fetch.js";
|
||||
|
||||
type UrbitSseLogger = {
|
||||
@@ -63,7 +64,11 @@ export class UrbitSSEClient {
|
||||
}> = [];
|
||||
eventHandlers = new Map<
|
||||
number,
|
||||
{ event?: (data: unknown) => void; err?: (error: unknown) => void; quit?: () => void }
|
||||
{
|
||||
event?: (data: unknown) => Promise<void> | void;
|
||||
err?: (error: unknown) => void;
|
||||
quit?: () => void;
|
||||
}
|
||||
>();
|
||||
aborted = false;
|
||||
streamController: AbortController | null = null;
|
||||
@@ -80,7 +85,7 @@ export class UrbitSSEClient {
|
||||
fetchImpl?: (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>;
|
||||
streamRelease: (() => Promise<void>) | null = null;
|
||||
|
||||
// Event ack tracking - must ack every ~50 events to keep channel healthy
|
||||
// Event ack tracking keeps each HTTP channel's delivered window bounded.
|
||||
private lastHeardEventId = -1;
|
||||
private lastAcknowledgedEventId = -1;
|
||||
private readonly ackThreshold = 20;
|
||||
@@ -115,10 +120,24 @@ export class UrbitSSEClient {
|
||||
};
|
||||
}
|
||||
|
||||
private resetChannelIdentity(): void {
|
||||
this.channelId = `${Math.floor(Date.now() / 1000)}-${randomUUID()}`;
|
||||
this.channelUrl = new URL(`/~/channel/${this.channelId}`, this.url).toString();
|
||||
this.lastHeardEventId = -1;
|
||||
this.lastAcknowledgedEventId = -1;
|
||||
}
|
||||
|
||||
private async createCurrentChannel(): Promise<void> {
|
||||
await ensureUrbitChannelOpen(this.channelRequestContext(), {
|
||||
createBody: this.subscriptions,
|
||||
createAuditContext: "tlon-urbit-channel-create",
|
||||
});
|
||||
}
|
||||
|
||||
async subscribe(params: {
|
||||
app: string;
|
||||
path: string;
|
||||
event?: (data: unknown) => void;
|
||||
event?: (data: unknown) => Promise<void> | void;
|
||||
err?: (error: unknown) => void;
|
||||
quit?: () => void;
|
||||
}) {
|
||||
@@ -170,10 +189,10 @@ export class UrbitSSEClient {
|
||||
}
|
||||
|
||||
async connect() {
|
||||
await ensureUrbitChannelOpen(this.channelRequestContext(), {
|
||||
createBody: this.subscriptions,
|
||||
createAuditContext: "tlon-urbit-channel-create",
|
||||
});
|
||||
// A fresh HTTP channel owns an independent event-id and ack sequence.
|
||||
this.lastHeardEventId = -1;
|
||||
this.lastAcknowledgedEventId = -1;
|
||||
await this.createCurrentChannel();
|
||||
|
||||
await this.openStream();
|
||||
this.isConnected = true;
|
||||
@@ -218,7 +237,7 @@ export class UrbitSSEClient {
|
||||
if (!response.ok) {
|
||||
this.streamRelease = null;
|
||||
await release();
|
||||
throw new Error(`Stream connection failed: ${response.status}`);
|
||||
throw new UrbitHttpError({ operation: "Stream connection", status: response.status });
|
||||
}
|
||||
|
||||
this.processStream(response.body).catch((error: unknown) => {
|
||||
@@ -263,12 +282,12 @@ export class UrbitSSEClient {
|
||||
buffer += text;
|
||||
bufferBytes = nextBytes;
|
||||
};
|
||||
const consumeText = (text: string) => {
|
||||
const consumeText = async (text: string) => {
|
||||
let offset = 0;
|
||||
if (pendingDelimiterNewline && text.length > 0) {
|
||||
pendingDelimiterNewline = false;
|
||||
if (text.startsWith("\n")) {
|
||||
this.processEvent(buffer);
|
||||
await this.processEvent(buffer);
|
||||
buffer = "";
|
||||
bufferBytes = 0;
|
||||
offset = 1;
|
||||
@@ -287,7 +306,7 @@ export class UrbitSSEClient {
|
||||
return;
|
||||
}
|
||||
appendPending(text.slice(offset, eventEnd));
|
||||
this.processEvent(buffer);
|
||||
await this.processEvent(buffer);
|
||||
buffer = "";
|
||||
bufferBytes = 0;
|
||||
offset = eventEnd + 2;
|
||||
@@ -300,13 +319,13 @@ export class UrbitSSEClient {
|
||||
break;
|
||||
}
|
||||
if (typeof chunk === "string") {
|
||||
consumeText(decoder.decode());
|
||||
consumeText(chunk);
|
||||
await consumeText(decoder.decode());
|
||||
await consumeText(chunk);
|
||||
} else {
|
||||
consumeText(decoder.decode(chunk as Uint8Array, { stream: true }));
|
||||
await consumeText(decoder.decode(chunk as Uint8Array, { stream: true }));
|
||||
}
|
||||
}
|
||||
consumeText(decoder.decode());
|
||||
await consumeText(decoder.decode());
|
||||
} finally {
|
||||
if (this.streamRelease) {
|
||||
const release = this.streamRelease;
|
||||
@@ -322,7 +341,7 @@ export class UrbitSSEClient {
|
||||
}
|
||||
}
|
||||
|
||||
processEvent(eventData: string) {
|
||||
async processEvent(eventData: string): Promise<void> {
|
||||
const lines = eventData.split("\n");
|
||||
let data: string | null = null;
|
||||
let eventId: number | null = null;
|
||||
@@ -340,49 +359,54 @@ export class UrbitSSEClient {
|
||||
return;
|
||||
}
|
||||
|
||||
// Track event ID and send ack if needed
|
||||
if (eventId !== null && !Number.isNaN(eventId)) {
|
||||
if (eventId > this.lastHeardEventId) {
|
||||
this.lastHeardEventId = eventId;
|
||||
if (eventId - this.lastAcknowledgedEventId > this.ackThreshold) {
|
||||
this.logger.log?.(
|
||||
`[SSE] Acking event ${eventId} (last acked: ${this.lastAcknowledgedEventId})`,
|
||||
);
|
||||
this.ack(eventId).catch((err: unknown) => {
|
||||
this.logger.error?.(`Failed to ack event ${eventId}: ${String(err)}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let parsed: ReturnType<typeof parseUrbitSsePayload>;
|
||||
try {
|
||||
const parsed = parseUrbitSsePayload(data);
|
||||
|
||||
if (parsed.response === "quit") {
|
||||
if (parsed.id) {
|
||||
const handlers = this.eventHandlers.get(parsed.id);
|
||||
if (handlers?.quit) {
|
||||
handlers.quit();
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (parsed.id && this.eventHandlers.has(parsed.id)) {
|
||||
const { event } = this.eventHandlers.get(parsed.id) ?? {};
|
||||
if (event && parsed.json) {
|
||||
event(parsed.json);
|
||||
}
|
||||
} else if (parsed.json) {
|
||||
for (const { event } of this.eventHandlers.values()) {
|
||||
if (event) {
|
||||
event(parsed.json);
|
||||
}
|
||||
}
|
||||
}
|
||||
parsed = parseUrbitSsePayload(data);
|
||||
} catch (error) {
|
||||
// Malformed transport payloads are permanent. Count them handled so one
|
||||
// poison event cannot pin the Urbit channel forever.
|
||||
this.logger.error?.(`Error parsing SSE event: ${String(error)}`);
|
||||
await this.acknowledgeHandledEventIfNeeded(eventId);
|
||||
return;
|
||||
}
|
||||
|
||||
if (parsed.response === "quit") {
|
||||
if (parsed.id) {
|
||||
const handlers = this.eventHandlers.get(parsed.id);
|
||||
if (handlers?.quit) {
|
||||
handlers.quit();
|
||||
}
|
||||
}
|
||||
} else if (parsed.id && this.eventHandlers.has(parsed.id)) {
|
||||
const { event } = this.eventHandlers.get(parsed.id) ?? {};
|
||||
if (event && parsed.json) {
|
||||
await event(parsed.json);
|
||||
}
|
||||
} else if (parsed.json) {
|
||||
for (const { event } of this.eventHandlers.values()) {
|
||||
if (event) {
|
||||
await event(parsed.json);
|
||||
}
|
||||
}
|
||||
}
|
||||
// Handler failures propagate without ack. Durable callbacks resolve only after append.
|
||||
await this.acknowledgeHandledEventIfNeeded(eventId);
|
||||
}
|
||||
|
||||
private async acknowledgeHandledEventIfNeeded(eventId: number | null): Promise<void> {
|
||||
if (eventId === null || eventId <= this.lastAcknowledgedEventId) {
|
||||
return;
|
||||
}
|
||||
this.lastHeardEventId = Math.max(this.lastHeardEventId, eventId);
|
||||
if (this.lastHeardEventId - this.lastAcknowledgedEventId <= this.ackThreshold) {
|
||||
return;
|
||||
}
|
||||
this.logger.log?.(
|
||||
`[SSE] Acking event ${this.lastHeardEventId} (last acked: ${this.lastAcknowledgedEventId})`,
|
||||
);
|
||||
// The acknowledged watermark advances only after PUT succeeds. A replay
|
||||
// therefore retries a failed ack instead of leaving the subscription clogged.
|
||||
await this.ack(this.lastHeardEventId);
|
||||
}
|
||||
|
||||
async poke(params: { app: string; mark: string; json: unknown }) {
|
||||
@@ -414,8 +438,6 @@ export class UrbitSSEClient {
|
||||
}
|
||||
|
||||
private async ack(eventId: number): Promise<void> {
|
||||
this.lastAcknowledgedEventId = eventId;
|
||||
|
||||
const ackData = {
|
||||
id: Date.now(),
|
||||
action: "ack",
|
||||
@@ -431,6 +453,7 @@ export class UrbitSSEClient {
|
||||
if (!response.ok) {
|
||||
throw new Error(`Ack failed with status ${response.status}`);
|
||||
}
|
||||
this.lastAcknowledgedEventId = eventId;
|
||||
} finally {
|
||||
await release();
|
||||
}
|
||||
@@ -470,15 +493,31 @@ export class UrbitSSEClient {
|
||||
setTimeout(resolve, delay);
|
||||
});
|
||||
|
||||
try {
|
||||
this.channelId = `${Math.floor(Date.now() / 1000)}-${randomUUID()}`;
|
||||
this.channelUrl = new URL(`/~/channel/${this.channelId}`, this.url).toString();
|
||||
if (this.aborted || !this.autoReconnect) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
if (this.onReconnect) {
|
||||
await this.onReconnect(this);
|
||||
}
|
||||
|
||||
await this.connect();
|
||||
try {
|
||||
// Reopen the same Eyre channel. Its queue retains every unacked event;
|
||||
// switching ids here would discard the cursor and strand failed admission.
|
||||
await this.openStream();
|
||||
} catch (error) {
|
||||
if (!(error instanceof UrbitHttpError) || error.status !== 404) {
|
||||
throw error;
|
||||
}
|
||||
// Eyre deletes idle channels. Only a definitive missing-channel response
|
||||
// permits losing the old cursor and rebuilding every subscription.
|
||||
this.resetChannelIdentity();
|
||||
await this.createCurrentChannel();
|
||||
await this.openStream();
|
||||
}
|
||||
this.isConnected = true;
|
||||
this.reconnectAttempts = 0;
|
||||
this.logger.log?.("[SSE] Reconnection successful!");
|
||||
} catch (error) {
|
||||
this.logger.error?.(`[SSE] Reconnection failed: ${String(error)}`);
|
||||
@@ -486,10 +525,14 @@ export class UrbitSSEClient {
|
||||
}
|
||||
}
|
||||
|
||||
async close() {
|
||||
stopReceiving(): void {
|
||||
this.aborted = true;
|
||||
this.isConnected = false;
|
||||
this.streamController?.abort();
|
||||
}
|
||||
|
||||
async close() {
|
||||
this.stopReceiving();
|
||||
|
||||
try {
|
||||
const unsubscribes = this.subscriptions.map((sub) => ({
|
||||
|
||||
Reference in New Issue
Block a user