mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-03 23:23:21 +00:00
fix(omp): release closed sessions retained by mount notices
Keep per-session notice identity without keeping the session and its runtime reachable from the client-wide ledger. Co-authored-by: luke <luopeihuan@gmail.com>
This commit is contained in:
@@ -734,6 +734,8 @@ export class OmpAgentSession implements AgentSession {
|
||||
private readonly mountNotices: OmpMountNoticeLedgers;
|
||||
private readonly agentId: string | undefined;
|
||||
private readonly sessionMountNotices = new OmpMountNoticeLedger();
|
||||
// The client-wide ledger outlives this session; retain identity without its runtime resources.
|
||||
private readonly mountNoticeDeliverer = {};
|
||||
private state: OmpSessionState;
|
||||
private currentModeId: string | null;
|
||||
private runtimeDead = false;
|
||||
@@ -1019,8 +1021,8 @@ export class OmpAgentSession implements AgentSession {
|
||||
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
|
||||
// This history seeds a fresh projection: it keeps the mount notices this session already
|
||||
// delivered into it, and none from earlier sessions.
|
||||
if (this.agentId) this.mountNotices.seedFromHistory(this.agentId, this);
|
||||
else this.sessionMountNotices.retainDeliveredBy(this);
|
||||
if (this.agentId) this.mountNotices.seedFromHistory(this.agentId, this.mountNoticeDeliverer);
|
||||
else this.sessionMountNotices.retainDeliveredBy(this.mountNoticeDeliverer);
|
||||
yield* streamOmpHistory({
|
||||
sessionFile: this.state.sessionFile,
|
||||
runtimeSession: this.runtimeSession,
|
||||
@@ -1049,7 +1051,7 @@ export class OmpAgentSession implements AgentSession {
|
||||
if (!notice || this.subscribers.size === 0) return false;
|
||||
const ledger = this.mountNoticeLedger();
|
||||
if (!ledger.announces(notice)) return true;
|
||||
ledger.recordDelivered(notice, this);
|
||||
ledger.recordDelivered(notice, this.mountNoticeDeliverer);
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
import { mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { setImmediate as waitForImmediate } from "node:timers/promises";
|
||||
@@ -162,6 +164,19 @@ describe("parseOmpMountNotice", () => {
|
||||
});
|
||||
|
||||
describe("OMP xd:// mount notices", () => {
|
||||
test("retaining announced mounts does not retain closed OMP sessions", () => {
|
||||
const fixture = fileURLToPath(new URL("./test-utils/mount-notice-memory.ts", import.meta.url));
|
||||
const output = execFileSync(process.execPath, ["--expose-gc", "--import", "tsx", fixture], {
|
||||
encoding: "utf8",
|
||||
timeout: 20_000,
|
||||
});
|
||||
expect(JSON.parse(output)).toEqual({
|
||||
closedWithoutNoticeRetained: false,
|
||||
closedWithNoticeRetained: false,
|
||||
noticesAfterRelaunch: 1,
|
||||
});
|
||||
});
|
||||
|
||||
test("a relaunched OMP process does not repeat a mount set the timeline already shows", async () => {
|
||||
const omp = await harness();
|
||||
await omp.announce(BROWSER_AND_LINEAR);
|
||||
@@ -324,7 +339,14 @@ describe("OMP xd:// mount notices", () => {
|
||||
class DaemonTimelineHarness {
|
||||
readonly omp = new FakeOmp();
|
||||
readonly manager = new AgentManager({
|
||||
clients: { omp: new OmpAgentClient({ logger: createTestLogger(), runtime: this.omp }) },
|
||||
clients: {
|
||||
omp: new OmpAgentClient({
|
||||
logger: createTestLogger(),
|
||||
runtime: this.omp,
|
||||
// FakeOmp supplies the runtime; the manager still checks command availability.
|
||||
runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } },
|
||||
}),
|
||||
},
|
||||
logger: createTestLogger(),
|
||||
});
|
||||
readonly live: Array<{
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
import { setImmediate as nextTask } from "node:timers/promises";
|
||||
import pino from "pino";
|
||||
import { OmpAgentClient } from "../agent.js";
|
||||
import { FakeOmp } from "./fake-omp.js";
|
||||
|
||||
if (!global.gc) throw new Error("Run with --expose-gc");
|
||||
const runtime = new FakeOmp();
|
||||
const client = new OmpAgentClient({ logger: pino({ level: "silent" }), runtime });
|
||||
const notices: string[] = [];
|
||||
const notice = {
|
||||
type: "notice",
|
||||
level: "info",
|
||||
source: "xdev",
|
||||
message: "xd://: mounted browser",
|
||||
} as const;
|
||||
const agentId = "a7969ccf-b200-429a-a36a-e61ac01e59ab";
|
||||
|
||||
async function closedSession(announce: boolean, id: string) {
|
||||
const session = await client.createSession(
|
||||
{ provider: "omp", cwd: process.cwd() },
|
||||
{ agentId: id },
|
||||
);
|
||||
session.subscribe((event) => {
|
||||
if (event.type === "timeline" && event.item.type === "notification") {
|
||||
notices.push(event.item.message);
|
||||
}
|
||||
});
|
||||
if (announce) runtime.latestSession().emit(notice);
|
||||
await nextTask();
|
||||
const persistence = session.describePersistence();
|
||||
if (!persistence) throw new Error("Session has no persistence handle");
|
||||
await session.close();
|
||||
return { reference: new WeakRef(session), persistence };
|
||||
}
|
||||
|
||||
const withoutNotice = await closedSession(false, "cd39924e-3984-4d10-b2ec-8a4e4ba029f5");
|
||||
const withNotice = await closedSession(true, agentId);
|
||||
// Move to another job before collecting: WeakRef targets survive their creation job.
|
||||
for (let i = 0; i < 5; i++) {
|
||||
await nextTask();
|
||||
global.gc();
|
||||
}
|
||||
const retained = {
|
||||
closedWithoutNoticeRetained: withoutNotice.reference.deref() !== undefined,
|
||||
closedWithNoticeRetained: withNotice.reference.deref() !== undefined,
|
||||
};
|
||||
|
||||
const resumed = await client.resumeSession(withNotice.persistence, undefined, { agentId });
|
||||
try {
|
||||
resumed.subscribe((event) => {
|
||||
if (event.type === "timeline" && event.item.type === "notification") {
|
||||
notices.push(event.item.message);
|
||||
}
|
||||
});
|
||||
runtime.latestSession().emit(notice);
|
||||
await nextTask();
|
||||
console.log(JSON.stringify({ ...retained, noticesAfterRelaunch: notices.length }));
|
||||
} finally {
|
||||
await resumed.close();
|
||||
}
|
||||
Reference in New Issue
Block a user