diff options
| -rw-r--r-- | index.js | 107 | ||||
| -rw-r--r-- | package.json | 2 |
2 files changed, 83 insertions, 26 deletions
@@ -1,4 +1,4 @@ -// PiRC extension 0.1.9-dev+931a2f6e036e. Third-party code in this file: @bufbuild/protobuf@2.16.0, ws@8.22.0; see NOTICE. +// PiRC extension 0.1.9-dev+3bc256ad5274. Third-party code in this file: @bufbuild/protobuf@2.16.0, ws@8.22.0; see NOTICE. import { createRequire as __pircCreateRequire } from "node:module"; const require = __pircCreateRequire(import.meta.url); var __create = Object.create; @@ -8400,7 +8400,9 @@ function hex(b) { function sameKey(a, b) { return a.length === b.length && a.every((x, i) => x === b[i]); } -var Host = class { +var HANDOVER_MS = 3e4; +var HANDOVER_AFTER_ANNOUNCE_MS = 2e3; +var Host = class _Host { constructor(instanceId, versions) { this.instanceId = instanceId; this.versions = versions; @@ -8425,9 +8427,56 @@ var Host = class { // Where the process runs, announced with its session; checked again on // activity (bind, a turn, a reconnect, a client's request). places = new PlaceTracker( - () => this.live?.onAnnouncementChanged(), + // Not connected, an event would be lost, leaving a gap in the sequence + // numbers; the announcement on connecting carries the place anyway. + () => { + if (this.state.kind === "ready") this.live?.onAnnouncementChanged(); + }, (e) => this.fail("place", e) ); + // The host this one took over from, until it is let go (HANDOVER_MS). + predecessor; + // A host for this process, taking over from the one another version of + // the extension made, if any: the same instance, with sequence numbers + // carrying on, so clients see the session continue. The old host keeps + // its connection, and its session (answering requests BUSY, as during + // any reload), until this one has announced: the relay ends the sessions + // of a connection that closes, but lets a newer one take a session over. + static succeeding(previous, instanceId, versions) { + const p = previous; + const host = new _Host(p?.instanceId ?? instanceId, versions); + if (!p) return host; + for (const [id, seq] of p.seqs ?? []) host.seqs.set(id, seq); + for (const live of [p.suspended, p.live]) if (live) host.seqs.set(live.sessionId, live.lastSeq); + const timer = setTimeout(() => host.releasePredecessor(), HANDOVER_MS); + timer.unref(); + host.predecessor = { host: p, timer, announced: false }; + return host; + } + releasePredecessor() { + const p = this.predecessor; + if (!p) return; + this.predecessor = void 0; + clearTimeout(p.timer); + const old = p.host; + try { + const lives = [old.suspended, old.live]; + old.live = old.suspended = void 0; + for (const live of lives) live?.dispose(); + old.conn?.stop(); + } catch (e) { + this.fail("previous version", e); + } + } + // Lets the predecessor go shortly after this host first announces. + announced() { + const p = this.predecessor; + if (!p || p.announced) return; + p.announced = true; + clearTimeout(p.timer); + p.timer = setTimeout(() => this.releasePredecessor(), HANDOVER_AFTER_ANNOUNCE_MS); + p.timer.unref(); + } // The live session that `pi` belongs to, if it is still current. liveFor(pi) { return this.live?.pi === pi ? this.live : void 0; @@ -8475,7 +8524,10 @@ var Host = class { this.previousSessionId = live.sessionId; live.onShutdown(reason); this.endSession(live); - if (reason === "quit") this.conn?.stop(); + if (reason === "quit") { + this.conn?.stop(); + this.releasePredecessor(); + } } endSession(live) { this.seqs.set(live.sessionId, live.lastSeq); @@ -8584,13 +8636,16 @@ var Host = class { } announce(live) { if (this.state.kind !== "ready") return; + let sent; try { - this.conn?.send(create(FrameSchema, { + sent = this.conn?.send(create(FrameSchema, { body: { case: "sessionAnnounce", value: { session: live.ref, payload: live.announcementBytes() } } })); } catch (e) { this.fail("announce", e); + return; } + if (sent) this.announced(); } place() { return this.places.current; @@ -8662,7 +8717,7 @@ function requestIdOf(m) { } // extension/src/index.ts -var EXTENSION_VERSION = true ? "0.1.9-dev+931a2f6e036e" : "dev"; +var EXTENSION_VERSION = true ? "0.1.9-dev+3bc256ad5274" : "dev"; function piVersion() { try { let dir = dirname3(realpathSync(process.argv[1])); @@ -8677,38 +8732,40 @@ function piVersion() { } return "unknown"; } +var LOAD = true ? "0.1.9-dev+3bc256ad5274" : `dev ${randomBytes3(4).toString("hex")}`; var KEY = /* @__PURE__ */ Symbol.for("pirc.extension"); -var proc = globalThis[KEY] ??= { - host: new Host(randomBytes3(8).toString("hex"), { pi: piVersion(), extension: EXTENSION_VERSION }) -}; +var proc = globalThis[KEY] ??= {}; +if (!proc.host || proc.load !== LOAD) { + proc.host = Host.succeeding(proc.host, randomBytes3(8).toString("hex"), { pi: piVersion(), extension: EXTENSION_VERSION }); + proc.load = LOAD; +} var USAGE = "usage: /pirc [status | enrol <relay-url> [token] | reconnect]"; function index_default(pi) { - const host = proc.host; let active = false; const on = (where, fn) => (e) => { - const live = host.liveFor(pi); + const live = proc.host.liveFor(pi); if (!live) return; try { fn(live, e); } catch (err) { - host.fail(where, err); + proc.host.fail(where, err); } }; pi.on("session_start", (e, ctx) => { if (ctx.mode !== "tui") return; active = true; try { - host.bind(pi, ctx, e); + proc.host.bind(pi, ctx, e); } catch (err) { - host.fail("session_start", err); + proc.host.fail("session_start", err); } }); pi.on("session_shutdown", (e) => { if (!active) return; try { - host.unbind(pi, e.reason); + proc.host.unbind(pi, e.reason); } catch (err) { - host.fail("session_shutdown", err); + proc.host.fail("session_shutdown", err); } }); pi.on("message_start", on("message_start", (live, e) => live.onMessageStart(e.message))); @@ -8718,18 +8775,18 @@ function index_default(pi) { pi.on("tool_execution_end", on("tool_execution_end", (live, e) => live.onToolEnd(e.toolCallId))); pi.on("agent_start", on("agent_start", (live) => { live.onWorking(true); - void host.places.refresh(); + void proc.host.places.refresh(); })); pi.on("agent_settled", on("agent_settled", (live) => { live.onWorking(false); - void host.places.refresh(); + void proc.host.places.refresh(); })); proc.stopPlace?.(); proc.stopPlace = pi.events.on(PLACE_EVENT, (data) => { try { - host.places.give(data); + proc.host.places.give(data); } catch (err) { - host.fail(PLACE_EVENT, err); + proc.host.fail(PLACE_EVENT, err); } }); pi.on("session_info_changed", on("session_info_changed", (live) => live.onAnnouncementChanged())); @@ -8743,9 +8800,9 @@ function index_default(pi) { handler: async (_args, ctx) => { try { const r = await ctx.newSession(); - if (r.cancelled) host.liveFor(pi)?.onReplaceFailed("another extension cancelled the new session"); + if (r.cancelled) proc.host.liveFor(pi)?.onReplaceFailed("another extension cancelled the new session"); } catch (err) { - host.liveFor(pi)?.onReplaceFailed(err?.message ?? String(err)); + proc.host.liveFor(pi)?.onReplaceFailed(err?.message ?? String(err)); } } }); @@ -8755,7 +8812,7 @@ function index_default(pi) { const [sub = "status", ...rest] = args.trim().split(/\s+/).filter(Boolean); switch (sub) { case "status": - ctx.ui.notify(host.statusReport(), "info"); + ctx.ui.notify(proc.host.statusReport(), "info"); return; case "enrol": case "enroll": { @@ -8767,12 +8824,12 @@ function index_default(pi) { } catch (err) { return ctx.ui.notify(`pirc: ${err.message}`, "error"); } - host.connect(token); + proc.host.connect(token); ctx.ui.notify(`pirc: enrolling with ${url}; /pirc status shows the result`, "info"); return; } case "reconnect": - host.connect(); + proc.host.connect(); ctx.ui.notify("pirc: reconnecting", "info"); return; default: diff --git a/package.json b/package.json index 7c6f8db..6e3d708 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "pirc-extension", - "version": "0.1.9-dev+931a2f6e036e", + "version": "0.1.9-dev+3bc256ad5274", "description": "PiRC: puts every interactive pi session on a PiRC relay", "license": "MIT", "homepage": "https://pirc.dev", |