feat(browser): self-heal sessions after camofox restarts (BR-3)
- Any session operation that hits 404/410 (or finds no open tab) reconciles
tabs; if none survive it re-injects the account cookies (BR-2), opens a new
tab under the same userId/sessionKey, returns to the recorded URL (site home
otherwise), marks old tabs lost and retries the operation once. Failure
returns 503 browser_reconnect_failed with the reason, with a 10s backoff.
- Restart detection: umt plugin GET /umt/instance (fallback /health fields);
a changed instance id marks every open session needsReconnect.
- POST /browser/sessions/{id}/reconnect (idempotent); session view exposes
needsReconnect and reconnect {status, trigger, at, reason, humanNotice}.
- Controller and lease are kept; reconnecting during takeover asks the human
to double-check the reloaded page.
- Web: screenshot and desktop modes show "reconnecting", auto-reconnect once,
pause polling, offer a reconnect button with the reason on failure; desktop
mode rebuilds the Selkies connection after a reconnect.
- Mock camofox restart(), tests, smoke step and runbook section 10.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
da24380870
commit
838aea3ebd
19 files changed
+977
-45
No files matched your search
@@ -204,7 +204,7 @@ describe.skipIf(!up)("浏览器 HTTP 端点", () => {
|
||||
expect(d2.status).toBe("resolved");
|
||||
});
|
||||
|
||||
it("服务重启:按数据库恢复会话,旧引用失效;上游回收的标签页标 lost", async () => {
|
||||
it("服务重启:按数据库恢复会话,旧引用失效;上游回收的标签页标 lost 并自动重开", async () => {
|
||||
// 新模块实例 = 进程重启(适配器内存为空)
|
||||
const fresh = createBrowserModule({ stagingDir: h.stagingDir });
|
||||
const reg = (await import("../../agent/capability-registry.ts")).createCapabilityRegistry({
|
||||
@@ -225,9 +225,10 @@ describe.skipIf(!up)("浏览器 HTTP 端点", () => {
|
||||
await h.db.selectFrom("browser_tab").selectAll().where("session_id", "=", sid).executeTakeFirstOrThrow()
|
||||
).upstream_tab_id;
|
||||
h.mock.dropTab(tab);
|
||||
await expect(reg.invoke("browser.snapshot", { sessionId: sid }, h.agentCtx())).rejects.toMatchObject({
|
||||
code: "conflict",
|
||||
});
|
||||
// BR-3:上游回收后自愈 —— 同一会话新开标签页回到原地址,重试一次
|
||||
const healed = (await reg.invoke("browser.snapshot", { sessionId: sid }, h.agentCtx())) as Loose;
|
||||
expect(healed.snapshot.text).toContain("种子列表");
|
||||
expect(healed.snapshot.tabId).not.toBe(tab);
|
||||
const t = await h.db
|
||||
.selectFrom("browser_tab")
|
||||
.select("status")
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { type BrowserHarness, createBrowserHarness, type Loose, ORIGIN, pgAvailable } from "./harness.ts";
|
||||
|
||||
// BR-3 浏览器服务重启后自愈(对 testkit mock camofox):容器重启 → 旧标签页 410、GET /tabs 为空、实例标识换新、
|
||||
// profile 登录态丢失(SIGKILL 未落盘)。任一会话操作 → 注入账号 Cookie → 同一会话新开标签页回到原地址 → 重试原操作;
|
||||
// 失败 503 browser_reconnect_failed;POST .../reconnect 幂等;人工接管中重连控制权不变并提示核对。
|
||||
|
||||
const up = await pgAvailable();
|
||||
const GOOD = "good-session-value-br3";
|
||||
|
||||
describe.skipIf(!up)("BR-3 浏览器服务重启后自动重连", () => {
|
||||
let h: BrowserHarness;
|
||||
|
||||
const setCookieAccount = async (cookie: string) => {
|
||||
const ref = await h.secrets.create({ purpose: "site_cookie", plaintext: cookie });
|
||||
await h.db
|
||||
.updateTable("site_account")
|
||||
.set({ auth_kind: "cookie", secret_ref: ref, session_secret_ref: null, status: "active", user_info: "{}" })
|
||||
.where("id", "=", h.accountId)
|
||||
.execute();
|
||||
};
|
||||
const open = async (path: string) =>
|
||||
(await h.req("/browser/sessions", { body: { siteId: h.siteId, accountId: h.accountId, url: `${ORIGIN}${path}` } }))
|
||||
.body.data as Loose;
|
||||
const closeAll = async () => {
|
||||
for (const s of (await h.req("/browser/sessions")).body.data.items as Loose[])
|
||||
await h.req(`/browser/sessions/${s.id}`, { method: "DELETE" });
|
||||
};
|
||||
const tabRow = (upstream: string) =>
|
||||
h.db.selectFrom("browser_tab").selectAll().where("upstream_tab_id", "=", upstream).executeTakeFirstOrThrow();
|
||||
const cookiePosts = (profileKey: string) =>
|
||||
h.mock.callsTo(`/umt/sessions/${encodeURIComponent(profileKey)}/cookies`).filter((c) => c.method === "POST").length;
|
||||
|
||||
beforeAll(async () => {
|
||||
h = await createBrowserHarness("br3", {
|
||||
sessionCookie: { name: "c_secure_pass", valid: [GOOD], fresh: "fresh-br3" },
|
||||
});
|
||||
await setCookieAccount(`c_secure_uid=1001; c_secure_pass=${GOOD}`);
|
||||
}, 120_000);
|
||||
afterAll(async () => {
|
||||
await h?.close();
|
||||
});
|
||||
|
||||
it("重启后截图:标签页 410 → 注入 Cookie、新开标签页回到原地址、重试成功;旧标签页标 lost", async () => {
|
||||
const s = await open("/details.php?id=42");
|
||||
expect(s.currentUrl).toBe(`${ORIGIN}/details.php?id=42`);
|
||||
const old = s.currentTabId as string;
|
||||
const injected = cookiePosts(s.profileKey);
|
||||
|
||||
h.mock.restart({ loseProfiles: true });
|
||||
const r = await h.req(`/browser/sessions/${s.id}/screenshot`);
|
||||
expect(r.status).toBe(200);
|
||||
expect(r.body.data.url).toBe(`${ORIGIN}/details.php?id=42`);
|
||||
|
||||
// Cookie 注入先于新标签页;profile 丢了登录态也能回到受保护页(不是登录页)
|
||||
expect(cookiePosts(s.profileKey)).toBe(injected + 1);
|
||||
const calls = h.mock.calls.map((c) => `${c.method} ${c.path}`);
|
||||
expect(calls.lastIndexOf(`POST /umt/sessions/${encodeURIComponent(s.profileKey)}/cookies`)).toBeLessThan(
|
||||
calls.lastIndexOf("POST /tabs"),
|
||||
);
|
||||
const tab = [...h.mock.tabs.values()].find((t) => t.userId === s.profileKey)!;
|
||||
expect(tab.sessionKey).toBe(s.id);
|
||||
expect(tab.url).toBe(`${ORIGIN}/details.php?id=42`);
|
||||
expect(h.mock.loggedIn.has(s.profileKey)).toBe(true);
|
||||
|
||||
const v = (await h.req(`/browser/sessions/${s.id}`)).body.data as Loose;
|
||||
expect(v.currentTabId).toBe(tab.tabId);
|
||||
expect(v.currentTabId).not.toBe(old);
|
||||
expect(v).toMatchObject({ controller: "automation", needsReconnect: false, needsSnapshot: true });
|
||||
expect(v.reconnect).toMatchObject({ status: "ok", humanNotice: false, reason: null });
|
||||
expect((await tabRow(old)).status).toBe("lost");
|
||||
expect(JSON.stringify(v)).not.toContain(GOOD);
|
||||
await closeAll();
|
||||
});
|
||||
|
||||
it("重启检测:实例标识变了 → 会话标需要重连;POST .../reconnect 幂等", async () => {
|
||||
const s = await open("/torrents.php");
|
||||
h.mock.restart();
|
||||
// 会话详情轮询发现重启(实例标识缓存 5 秒:用显式 reconnect 强制刷新)
|
||||
const r1 = await h.req(`/browser/sessions/${s.id}/reconnect`, { body: {} });
|
||||
expect(r1.status).toBe(200);
|
||||
expect(r1.body.data).toMatchObject({
|
||||
needsReconnect: false,
|
||||
reconnect: { status: "ok", trigger: "provider_restarted" },
|
||||
});
|
||||
expect(r1.body.data.currentUrl).toBe(`${ORIGIN}/torrents.php`);
|
||||
|
||||
const opened = h.mock.callsTo(/^\/tabs$/).filter((c) => c.method === "POST").length;
|
||||
const r2 = await h.req(`/browser/sessions/${s.id}/reconnect`, { body: {} });
|
||||
expect(r2.status).toBe(200);
|
||||
expect(r2.body.data.currentTabId).toBe(r1.body.data.currentTabId);
|
||||
expect(h.mock.callsTo(/^\/tabs$/).filter((c) => c.method === "POST").length).toBe(opened);
|
||||
|
||||
// Agent 快照也走同一条自愈
|
||||
h.mock.restart();
|
||||
const snap = await h.req(`/browser/sessions/${s.id}/snapshot`);
|
||||
expect(snap.status).toBe(200);
|
||||
expect(snap.body.data.url).toBe(`${ORIGIN}/torrents.php`);
|
||||
await closeAll();
|
||||
});
|
||||
|
||||
it("人工接管中重连:控制者与租约不变,提示页面已重新加载", async () => {
|
||||
const s = await open("/torrents.php");
|
||||
const acq = await h.req(`/browser/sessions/${s.id}/control/acquire`, { body: { reason: "核对" } });
|
||||
const leaseId = acq.body.data.lease.id;
|
||||
h.mock.restart();
|
||||
const r = await h.req(`/browser/sessions/${s.id}/screenshot`);
|
||||
expect(r.status).toBe(200);
|
||||
expect(r.body.data.controller).toBe("human");
|
||||
const v = (await h.req(`/browser/sessions/${s.id}`)).body.data as Loose;
|
||||
expect(v.lease).toMatchObject({ id: leaseId, status: "active" });
|
||||
expect(v.reconnect).toMatchObject({ status: "ok", humanNotice: true });
|
||||
// 接管者的输入照常可用
|
||||
const nav = await h.req(`/browser/sessions/${s.id}/navigate`, { body: { url: `${ORIGIN}/index.php` } });
|
||||
expect(nav.status).toBe(200);
|
||||
await closeAll();
|
||||
});
|
||||
|
||||
it("重连失败:503 browser_reconnect_failed 带原因;短时间内不反复重连;恢复后手动重连成功", async () => {
|
||||
const s = await open("/torrents.php");
|
||||
h.mock.restart();
|
||||
h.mock.fault(/^\/tabs$/, { status: 500, times: 10 });
|
||||
const r = await h.req(`/browser/sessions/${s.id}/screenshot`);
|
||||
expect(r.status).toBe(503);
|
||||
expect(r.body.error.code).toBe("browser_reconnect_failed");
|
||||
expect(r.body.error.message).toContain("HTTP 500");
|
||||
const v = (await h.req(`/browser/sessions/${s.id}`)).body.data as Loose;
|
||||
expect(v).toMatchObject({ needsReconnect: true, reconnect: { status: "failed", trigger: "tab_gone" } });
|
||||
|
||||
// 退避期内的截图直接报上次原因,不再打上游
|
||||
const before = h.mock.calls.length;
|
||||
const again = await h.req(`/browser/sessions/${s.id}/screenshot`);
|
||||
expect(again.status).toBe(503);
|
||||
expect(h.mock.calls.slice(before).some((c) => c.path === "/tabs")).toBe(false);
|
||||
|
||||
h.mock.clearFaults();
|
||||
const ok = await h.req(`/browser/sessions/${s.id}/reconnect`, { body: {} });
|
||||
expect(ok.status).toBe(200);
|
||||
expect(ok.body.data).toMatchObject({ needsReconnect: false, reconnect: { status: "ok" } });
|
||||
expect((await h.req(`/browser/sessions/${s.id}/screenshot`)).status).toBe(200);
|
||||
await closeAll();
|
||||
});
|
||||
});
|
||||
@@ -15,6 +15,7 @@ const CODE = {
|
||||
bad_request: "precondition",
|
||||
upstream: "precondition",
|
||||
rate_limited: "budget",
|
||||
reconnect_failed: "precondition",
|
||||
} as const;
|
||||
|
||||
async function guard<T>(fn: () => Promise<T>): Promise<T> {
|
||||
|
||||
@@ -35,6 +35,7 @@ const CODE = {
|
||||
bad_request: "bad_request",
|
||||
upstream: "upstream_unavailable",
|
||||
rate_limited: "rate_limited",
|
||||
reconnect_failed: "browser_reconnect_failed",
|
||||
} as const;
|
||||
|
||||
function route<D extends EndpointDef>(app: ModuleApp, def: D, handler: (input: Input<D>) => Promise<unknown>) {
|
||||
@@ -284,6 +285,12 @@ export function registerBrowserRoutes(app: ModuleApp, svc: () => BrowserService)
|
||||
};
|
||||
});
|
||||
|
||||
// BR-3:浏览器服务重启后重新连接(幂等;原站视图收到 404/410/503 时调用一次)。不要求接管:控制权不变
|
||||
route(app, B.reconnect, async ({ params }) => {
|
||||
await svc().reconnectIfNeeded(params.id);
|
||||
return svc().view(params.id);
|
||||
});
|
||||
|
||||
route(app, B.acquire, async ({ params, body, c }) => {
|
||||
const lease = await svc().acquire(params.id, human(c).holder, body.reason, body.ttlMs);
|
||||
return { lease, session: await svc().view(params.id) };
|
||||
|
||||
@@ -32,6 +32,7 @@ import type {
|
||||
BrowserEvidenceRef,
|
||||
BrowserHealth,
|
||||
BrowserProviderView,
|
||||
BrowserReconnectState,
|
||||
BrowserSession,
|
||||
BrowserSessionView,
|
||||
BrowserTab,
|
||||
@@ -88,7 +89,14 @@ export interface ProviderDeps {
|
||||
/** 业务错误(路由层转 ApiException,能力层转 CapabilityError) */
|
||||
export class BrowserServiceError extends Error {
|
||||
constructor(
|
||||
readonly code: "not_found" | "conflict" | "precondition" | "bad_request" | "upstream" | "rate_limited",
|
||||
readonly code:
|
||||
| "not_found"
|
||||
| "conflict"
|
||||
| "precondition"
|
||||
| "bad_request"
|
||||
| "upstream"
|
||||
| "rate_limited"
|
||||
| "reconnect_failed",
|
||||
message: string,
|
||||
) {
|
||||
super(message);
|
||||
@@ -147,6 +155,26 @@ export interface RecoveryState {
|
||||
downloads?: { sha256: string; artifactId: string; sourceReleaseId: string | null; planId: string | null }[];
|
||||
/** BR-2:打开时的凭据注入结果与最近一次回写检查 */
|
||||
credential?: BrowserCredentialState;
|
||||
/** BR-3:上游上下文已不在(浏览器服务重启),下一次操作前先重连 */
|
||||
needsReconnect?: boolean;
|
||||
reconnectTrigger?: BrowserReconnectState["trigger"];
|
||||
/** 打开 / 重连时浏览器服务的实例标识;变了 = 浏览器服务重启过 */
|
||||
instanceId?: string | null;
|
||||
reconnect?: BrowserReconnectState;
|
||||
}
|
||||
|
||||
/** 同一浏览器服务的实例标识多久查一次 */
|
||||
const INSTANCE_CHECK_MS = 5_000;
|
||||
/** 自动重连失败后,操作触发的重连至少间隔多久 */
|
||||
const RECONNECT_BACKOFF_MS = 10_000;
|
||||
|
||||
/** 上游说标签页 / 会话不在了:404 / 410,或适配器内存里已没有打开的标签页 */
|
||||
function isGone(e: unknown): boolean {
|
||||
return (
|
||||
e instanceof BrowserError &&
|
||||
e.kind === "not_found" &&
|
||||
(e.httpStatus === undefined || e.httpStatus === 404 || e.httpStatus === 410)
|
||||
);
|
||||
}
|
||||
|
||||
/** 有活动的会话每隔多久从浏览器读回一次 Cookie */
|
||||
@@ -631,6 +659,9 @@ export class BrowserService {
|
||||
async view(id: string): Promise<BrowserSessionView> {
|
||||
let row = await this.sessionRow(id);
|
||||
row = await this.expireLeases(row);
|
||||
// 原站视图轮询会话详情:顺带发现浏览器服务重启(节流,失败不影响详情)
|
||||
if (row.controller !== "closed" && (await this.checkInstance(row.provider_id).catch(() => false)))
|
||||
row = await this.sessionRow(id);
|
||||
const [tabs, lease] = await Promise.all([this.tabRows(id), this.activeLease(id)]);
|
||||
const rs = obj(row.recovery_state) as RecoveryState;
|
||||
return {
|
||||
@@ -641,6 +672,8 @@ export class BrowserService {
|
||||
needsSnapshot: rs.needsSnapshot === true,
|
||||
waitingReason: row.controller === "waiting" || row.controller === "human" ? (rs.waitingReason ?? null) : null,
|
||||
credential: rs.credential ?? null,
|
||||
needsReconnect: rs.needsReconnect === true,
|
||||
reconnect: rs.reconnect ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -716,6 +749,7 @@ export class BrowserService {
|
||||
throw fromBrowserError(e);
|
||||
}
|
||||
const s = res.session;
|
||||
const instanceId = await this.currentInstance(prow, true).catch(() => null);
|
||||
const credential: BrowserCredentialState = { ...prep.state };
|
||||
if (res.cookies) {
|
||||
credential.status = res.cookies.status;
|
||||
@@ -745,7 +779,12 @@ export class BrowserService {
|
||||
current_url: s.currentUrl,
|
||||
last_activity_at: new Date(s.lastActivityAt),
|
||||
upstream_ref: JSON.stringify(s.upstreamRef),
|
||||
recovery_state: JSON.stringify({ currentTabId: res.tab.upstreamTabId, needsSnapshot: true, credential }),
|
||||
recovery_state: JSON.stringify({
|
||||
currentTabId: res.tab.upstreamTabId,
|
||||
needsSnapshot: true,
|
||||
credential,
|
||||
instanceId,
|
||||
}),
|
||||
})
|
||||
.execute();
|
||||
await trx
|
||||
@@ -1173,7 +1212,18 @@ export class BrowserService {
|
||||
);
|
||||
}
|
||||
}
|
||||
const loaded = await this.loaded(row);
|
||||
// BR-3:浏览器服务重启过(实例标识变了)→ 先重建上游上下文再操作
|
||||
if (await this.checkInstance(row.provider_id).catch(() => false)) row = await this.sessionRow(id);
|
||||
const pending = obj(row.recovery_state) as RecoveryState;
|
||||
if (pending.needsReconnect) {
|
||||
// 刚失败过:短时间内不反复重连(原站视图 1.5 秒一次截图),直接报上次的原因;显式 POST .../reconnect 不受限
|
||||
const last = pending.reconnect;
|
||||
if (last?.status === "failed" && this.now().getTime() - new Date(last.at).getTime() < RECONNECT_BACKOFF_MS)
|
||||
throw new BrowserServiceError("reconnect_failed", `浏览器已重启,重新连接失败:${last.reason ?? "未知原因"}`);
|
||||
await this.reconnect(id, pending.reconnectTrigger ?? "provider_restarted");
|
||||
row = await this.sessionRow(id);
|
||||
}
|
||||
let loaded = await this.loaded(row);
|
||||
row = loaded.row;
|
||||
const prow = await this.providerRow(row.provider_id).catch(() => null);
|
||||
const site = await this.liveSite(row.site_id);
|
||||
@@ -1185,19 +1235,211 @@ export class BrowserService {
|
||||
try {
|
||||
out = await fn(loaded.provider, ref, { row, site, budget });
|
||||
} catch (e) {
|
||||
if (e instanceof BrowserError && e.kind === "not_found" && (e.httpStatus === 404 || e.httpStatus === 410)) {
|
||||
const r = await loaded.provider.reconcileTabs(id).catch(() => null);
|
||||
await this.persist(id, loaded.provider, { needsSnapshot: true });
|
||||
if (r?.lost.length) throw new BrowserServiceError("conflict", "标签页已被浏览器服务回收,重新打开页面");
|
||||
if (!isGone(e)) {
|
||||
await this.persist(id, loaded.provider).catch(() => {});
|
||||
throw fromBrowserError(e);
|
||||
}
|
||||
// 上游说标签页 / 会话不在了:先对账。当前标签页还活着 → 指定的旧标签页没了照旧报错,否则在当前标签页重试一次;
|
||||
// 全没了(浏览器服务重启 / 回收)→ 注入 Cookie、新开标签页回到原地址,再重试一次(失败 503 browser_reconnect_failed)
|
||||
const r = await loaded.provider.reconcileTabs(id).catch(() => null);
|
||||
await this.persist(id, loaded.provider, { needsSnapshot: true }).catch(() => {});
|
||||
const alive = r ? (loaded.provider.sessionInfo(id)?.currentTabId ?? null) : null;
|
||||
if (alive && opts.tabId && opts.tabId !== alive) {
|
||||
if (r?.lost.includes(opts.tabId))
|
||||
throw new BrowserServiceError("conflict", "标签页已被浏览器服务回收,重新打开页面");
|
||||
throw fromBrowserError(e);
|
||||
}
|
||||
if (!alive) {
|
||||
await this.markReconnect(id, "tab_gone");
|
||||
await this.reconnect(id, "tab_gone");
|
||||
loaded = await this.loaded(await this.sessionRow(id));
|
||||
row = loaded.row;
|
||||
}
|
||||
const retry: TabRef = { sessionId: id, as: actor.kind };
|
||||
try {
|
||||
out = await fn(loaded.provider, retry, { row, site, budget });
|
||||
} catch (e2) {
|
||||
await this.persist(id, loaded.provider).catch(() => {});
|
||||
throw fromBrowserError(e2);
|
||||
}
|
||||
await this.persist(id, loaded.provider).catch(() => {});
|
||||
throw fromBrowserError(e);
|
||||
}
|
||||
if (opts.persist !== false) await this.persist(id, loaded.provider, opts.patch?.(out) ?? {});
|
||||
this.maybeSyncCookies(row);
|
||||
return out;
|
||||
}
|
||||
|
||||
// ---------- BR-3 浏览器服务重启后自愈 ----------
|
||||
|
||||
/** 各浏览器服务最近一次读到的实例标识(节流缓存;只在进程内) */
|
||||
private readonly instances = new Map<string, { at: number; id: string | null; pending?: Promise<string | null> }>();
|
||||
/** 各浏览器服务已按哪个实例标识对过会话(同一标识不重复写库) */
|
||||
private readonly applied = new Map<string, string>();
|
||||
/** 进行中的重连:同一会话并发的操作共用一次 */
|
||||
private readonly reconnecting = new Map<string, Promise<void>>();
|
||||
|
||||
/** 浏览器服务当前实例标识:INSTANCE_CHECK_MS 内复用上次的结果;取不到(旧镜像)为 null */
|
||||
private async currentInstance(prow: BrowserProviderRow, force = false): Promise<string | null> {
|
||||
const provider = this.providerFor(prow);
|
||||
if (!provider.instanceId) return null;
|
||||
const hit = this.instances.get(prow.id);
|
||||
if (hit?.pending) return hit.pending;
|
||||
if (hit && !force && Date.now() - hit.at < INSTANCE_CHECK_MS) return hit.id;
|
||||
const pending = provider.instanceId().then(
|
||||
(id) => {
|
||||
this.instances.set(prow.id, { at: Date.now(), id });
|
||||
return id;
|
||||
},
|
||||
(e) => {
|
||||
// 连不上:沿用旧值,不当成重启(真重启时标签页 404/410 会被动发现)
|
||||
this.instances.set(prow.id, { at: Date.now(), id: hit?.id ?? null });
|
||||
throw e;
|
||||
},
|
||||
);
|
||||
this.instances.set(prow.id, { at: hit?.at ?? 0, id: hit?.id ?? null, pending });
|
||||
return pending;
|
||||
}
|
||||
|
||||
/**
|
||||
* 重启检测:浏览器服务实例标识变了 → 该服务下全部未关闭会话标 needsReconnect(下一次操作 / 原站视图触发重连)。
|
||||
* 没记录实例标识的旧会话补记当前值。返回本次是否有会话被标记。
|
||||
*/
|
||||
async checkInstance(providerId: string, force = false): Promise<boolean> {
|
||||
const prow = await this.ctx.db
|
||||
.selectFrom("browser_provider")
|
||||
.selectAll()
|
||||
.where("id", "=", providerId)
|
||||
.executeTakeFirst();
|
||||
if (!prow?.enabled || prow.kind === "none") return false;
|
||||
const id = await this.currentInstance(prow, force);
|
||||
if (!id || (!force && this.applied.get(providerId) === id)) return false;
|
||||
const db = this.ctx.db;
|
||||
await db
|
||||
.updateTable("browser_session")
|
||||
.set({ recovery_state: sql`recovery_state || ${JSON.stringify({ instanceId: id })}::jsonb` })
|
||||
.where("provider_id", "=", providerId)
|
||||
.where("controller", "<>", "closed")
|
||||
.where(sql<boolean>`recovery_state->>'instanceId' is null`)
|
||||
.execute();
|
||||
const r = await db
|
||||
.updateTable("browser_session")
|
||||
.set({
|
||||
recovery_state: sql`recovery_state || ${JSON.stringify({ needsReconnect: true, reconnectTrigger: "provider_restarted" })}::jsonb`,
|
||||
updated_at: this.now(),
|
||||
})
|
||||
.where("provider_id", "=", providerId)
|
||||
.where("controller", "<>", "closed")
|
||||
.where(sql<boolean>`recovery_state->>'instanceId' <> ${id}`)
|
||||
.where(sql<boolean>`coalesce(recovery_state->>'needsReconnect', 'false') <> 'true'`)
|
||||
.executeTakeFirst();
|
||||
this.applied.set(providerId, id);
|
||||
const n = Number(r.numUpdatedRows ?? 0);
|
||||
if (n > 0)
|
||||
this.ctx.logger.info(
|
||||
{ component: "browser", providerId, sessions: n },
|
||||
"浏览器服务重启过,未关闭的会话标为需要重新连接",
|
||||
);
|
||||
return n > 0;
|
||||
}
|
||||
|
||||
private async markReconnect(id: string, trigger: BrowserReconnectState["trigger"]): Promise<void> {
|
||||
await this.patchRecovery(await this.sessionRow(id), { needsReconnect: true, reconnectTrigger: trigger });
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /browser/sessions/{id}/reconnect(幂等):先查实例标识;没标需要重连时按上游对账,当前标签页还活着就什么都不做;
|
||||
* 否则重建。失败抛 reconnect_failed(503)。
|
||||
*/
|
||||
async reconnectIfNeeded(id: string): Promise<void> {
|
||||
let row = await this.sessionRow(id);
|
||||
if (row.controller === "closed") throw new BrowserServiceError("conflict", "会话已关闭");
|
||||
if (await this.checkInstance(row.provider_id, true).catch(() => false)) row = await this.sessionRow(id);
|
||||
const rs = obj(row.recovery_state) as RecoveryState;
|
||||
if (!rs.needsReconnect) {
|
||||
const { provider } = await this.loaded(row);
|
||||
try {
|
||||
await provider.reconcileTabs(id);
|
||||
} catch (e) {
|
||||
throw new BrowserServiceError(
|
||||
"reconnect_failed",
|
||||
`浏览器服务不可用:${e instanceof Error ? e.message : String(e)}`,
|
||||
);
|
||||
}
|
||||
await this.persist(id, provider).catch(() => {});
|
||||
if (provider.sessionInfo(id)?.currentTabId) return;
|
||||
}
|
||||
await this.reconnect(id, rs.needsReconnect ? (rs.reconnectTrigger ?? "provider_restarted") : "manual");
|
||||
}
|
||||
|
||||
/** 重建会话的上游上下文(同一会话并发调用只做一次) */
|
||||
reconnect(id: string, trigger: BrowserReconnectState["trigger"]): Promise<void> {
|
||||
const running = this.reconnecting.get(id);
|
||||
if (running) return running;
|
||||
const p = this.doReconnect(id, trigger).finally(() => this.reconnecting.delete(id));
|
||||
this.reconnecting.set(id, p);
|
||||
return p;
|
||||
}
|
||||
|
||||
/**
|
||||
* 按会话的站点 / 账号:注入账号 Cookie(BR-2 同一套准备)→ 同一 userId 开新标签页 → 导航到会话记录的地址(没有或不在站点域名内
|
||||
* 则站点首页)→ 旧标签页标 lost、换当前标签页。控制者与租约不变;人工接管中重连记 humanNotice(页面已重新加载,请核对)。
|
||||
*/
|
||||
private async doReconnect(id: string, trigger: BrowserReconnectState["trigger"]): Promise<void> {
|
||||
let row = await this.sessionRow(id);
|
||||
if (row.controller === "closed") throw new BrowserServiceError("conflict", "会话已关闭");
|
||||
const loaded = await this.loaded(row);
|
||||
row = loaded.row;
|
||||
const provider = loaded.provider;
|
||||
const prow = await this.providerRow(row.provider_id);
|
||||
const site = await this.liveSite(row.site_id);
|
||||
const budget = this.budgetFor(site, prow);
|
||||
const at = this.now().toISOString();
|
||||
const fail = async (reason: string) => {
|
||||
const cur = await this.sessionRow(id);
|
||||
await this.patchRecovery(cur, {
|
||||
needsReconnect: true,
|
||||
reconnectTrigger: trigger,
|
||||
reconnect: { status: "failed", trigger, at, reason, humanNotice: false },
|
||||
});
|
||||
this.ctx.logger.warn({ component: "browser", sessionId: id, trigger, reason }, "浏览器会话重新连接失败");
|
||||
return new BrowserServiceError("reconnect_failed", `浏览器已重启,重新连接失败:${reason}`);
|
||||
};
|
||||
if (!provider.reopenSession) throw await fail("浏览器服务不支持重新连接");
|
||||
const url = restoreUrl(row.current_url, budget) ?? site.base_url;
|
||||
const prep = await this.prepareCookies(site, row.account_id).catch(() => null);
|
||||
let res: Awaited<ReturnType<NonNullable<BrowserProvider["reopenSession"]>>>;
|
||||
try {
|
||||
await this.rateLimit(site);
|
||||
res = await provider.reopenSession(id, {
|
||||
url,
|
||||
budget,
|
||||
...(prep?.cookies.length ? { cookies: prep.cookies } : {}),
|
||||
});
|
||||
} catch (e) {
|
||||
throw await fail(e instanceof Error ? e.message : String(e));
|
||||
}
|
||||
const instanceId = await this.currentInstance(prow, true).catch(() => null);
|
||||
const rs = obj((await this.sessionRow(id)).recovery_state) as RecoveryState;
|
||||
await this.persist(id, provider, {
|
||||
needsReconnect: false,
|
||||
needsSnapshot: true,
|
||||
instanceId: instanceId ?? rs.instanceId ?? null,
|
||||
reconnect: { status: "ok", trigger, at, reason: null, humanNotice: row.controller === "human" },
|
||||
});
|
||||
this.coordSpaces.delete(id);
|
||||
this.ctx.logger.info(
|
||||
{
|
||||
component: "browser",
|
||||
sessionId: id,
|
||||
trigger,
|
||||
lostTabs: res.lost.length,
|
||||
cookies: res.cookies?.status ?? "none",
|
||||
cookieCount: res.cookies?.count ?? 0,
|
||||
controller: row.controller,
|
||||
},
|
||||
"浏览器会话已重新连接",
|
||||
);
|
||||
}
|
||||
|
||||
/** 有活动时每 COOKIE_SYNC_INTERVAL_MS 读回一次登录态(后台,不拖慢动作;同一会话同时只跑一个) */
|
||||
private readonly syncing = new Set<string>();
|
||||
private maybeSyncCookies(row: BrowserSessionRow): void {
|
||||
@@ -1557,6 +1799,14 @@ const REASON_TEXT: Record<string, string> = {
|
||||
human_released_waiting: "人工交还时选择继续等待",
|
||||
};
|
||||
|
||||
/** 重连回到的地址:会话记录的地址是 http(s)、在站点域名内、没有被脱敏过才用;否则 null(调用方退回站点首页) */
|
||||
function restoreUrl(url: string | null, budget: NavigationBudget): string | null {
|
||||
const u = url ? safeUrl(url) : null;
|
||||
if (!u || !/^https?:$/.test(u.protocol) || !hostAllowed(u.hostname, budget.domainAllowlist)) return null;
|
||||
if (url!.includes("[REDACTED")) return null;
|
||||
return url;
|
||||
}
|
||||
|
||||
function safeUrl(s: string): URL | null {
|
||||
try {
|
||||
return new URL(s);
|
||||
|
||||
Reference in new issue
Block a user