History

Fix two more CI e2e flakes: picker hydration race and dev-db ghost affinity (#1658)

* Retry reveal-style clicks that race console hydration * Repair dev-db queue affinity abandoned by mid-execution disconnects

Rhys Sullivan committed 2 months ago

Commit a5f34001 parent 739805f7 files · +140 −44Browse files

Unified · Split

7 files
  1. apps/cloud/scripts/dev-db.ts +14 −8
  2. apps/cloud/src/db/dev-db-socket-concurrency.node.test.ts +82 −23
  3. e2e/scenarios/google-health-checks.test.ts +2 −6
  4. e2e/scenarios/google-photos-preset-ui.test.ts +2 −2
  5. e2e/scenarios/provider-plugins-ui.test.ts +5 −2
  6. e2e/src/surfaces/browser.ts +30 −1
  7. patches/@electric-sql%2Fpglite-socket@0.1.4.patch +5 −2
apps/cloud/scripts/dev-db.ts+14−8
@@ -104,14 +104,20 @@ await migrate(drizzle(db), { migrationsFolder: MIGRATIONS_FOLDER }); · 103 unmodified lines
104104 // but prepared statement requires M" -> random 500s on whichever request lost
105105 // the race). The patch in patches/@electric-sql%2Fpglite-socket@0.1.4.patch
106106 // batches each socket data event into one queue entry and holds handler
107−// affinity while a pipeline is open. The patch also fixes the queue's failure
108−// path: stock 0.1.4 `return`ed out of the drain loop when a query REJECTED at
109−// the JS level, leaving its `processing` flag latched true — after one such
110−// throw nothing was ever dequeued again, so new connections' startup packets
111−// sat unanswered (postgres.js CONNECT_TIMEOUT) and the whole stack was bricked
112−// until restart: the CI e2e "cloud signIn: callback set no session (500)"
113−// cascade. src/db/dev-db-socket-concurrency.node.test.ts is the regression
114−// test for all of the above.
107+// affinity while a pipeline is open. The patch also fixes the queue's two
108+// self-bricking failure paths — both surfaced in CI as the e2e "cloud signIn:
109+// callback set no session (500)" cascade, where new connections' startup
110+// packets sat unanswered (postgres.js CONNECT_TIMEOUT) until restart:
111+// 1. Stock 0.1.4 `return`ed out of the drain loop when a query REJECTED at
112+// the JS level, leaving its `processing` flag latched true; nothing was
113+// ever dequeued again.
114+// 2. A client whose socket died WHILE its pipeline-opening entry executed:
115+// detach() cleared affinity before the entry finished, the queue then
116+// took affinity for the already-dead handler, and no timer was left to
117+// release it. The queue now tracks detached handlers and repairs any
118+// transaction or pipeline affinity they can no longer release.
119+// src/db/dev-db-socket-concurrency.node.test.ts is the regression test for
120+// all of the above.
115121 const server = new PGLiteSocketServer({
116122 db,
117123 port: PORT,
apps/cloud/src/db/dev-db-socket-concurrency.node.test.ts+82−23
@@ -42,6 +42,40 @@ const makeClient = (port: number, connectTimeout = 5) => · 41 unmodified lines
4242 onnotice: () => undefined,
4343 });
4444
45+// Hand-rolled wire client: connect and complete the trust-auth startup, so a
46+// test can then speak raw protocol frames (e.g. a lone Parse) that postgres.js
47+// would never emit on its own. Resolves after ReadyForQuery so the next write
48+// is its own data event — and its own queue entry — on the server.
49+const openWireClient = async (port: number): Promise<Socket> => {
50+ const socket: Socket = connect(port, "127.0.0.1");
51+ await new Promise<void>((res, rej) => {
52+ socket.once("connect", res);
53+ socket.once("error", rej);
54+ });
55+ const startupBody = Buffer.concat([
56+ Buffer.from([0, 3, 0, 0]),
57+ Buffer.from("user\0postgres\0database\0postgres\0\0"),
58+ ]);
59+ const startup = Buffer.concat([Buffer.alloc(4), startupBody]);
60+ startup.writeInt32BE(startup.length, 0);
61+ socket.write(startup);
62+ await new Promise<void>((res) => {
63+ socket.on("data", (chunk: Buffer) => {
64+ if (chunk.includes(0x5a)) res(); // 'Z' = ReadyForQuery
65+ });
66+ });
67+ return socket;
68+};
69+
70+// A Parse frame for an unnamed statement: opens an extended-protocol pipeline
71+// that only a later Sync (or the server's recovery) closes.
72+const parseFrame = (query: string): Buffer => {
73+ const body = Buffer.concat([Buffer.from(`\0${query}\0`), Buffer.from([0, 0])]);
74+ const frame = Buffer.concat([Buffer.from("P"), Buffer.alloc(4), body]);
75+ frame.writeInt32BE(4 + body.length, 1);
76+ return frame;
77+};
78+
4579 describe("dev-db PGlite socket under concurrent connections", () => {
4680 it(
4781 "serves interleaved multi-connection pipelines without protocol corruption",
@@ -207,29 +241,8 @@ describe("dev-db PGlite socket under concurrent connections", () => { · 159 unmodified lines
207241 // Hand-rolled wire client: complete the trust-auth startup, then send a
208242 // lone Parse. Its last frame type ('P') marks the pipeline open, so the
209243 // handler takes affinity and every other connection queues behind it.
210− const staller: Socket = connect(port, "127.0.0.1");
211− await new Promise<void>((res, rej) => {
212− staller.once("connect", res);
213− staller.once("error", rej);
214− });
215− const startupBody = Buffer.concat([
216− Buffer.from([0, 3, 0, 0]),
217− Buffer.from("user\0postgres\0database\0postgres\0\0"),
218− ]);
219− const startup = Buffer.concat([Buffer.alloc(4), startupBody]);
220− startup.writeInt32BE(startup.length, 0);
221− staller.write(startup);
222− // Wait for AuthenticationOk + ReadyForQuery before opening the pipeline,
223− // so the Parse is its own data event (and its own queue entry).
224− await new Promise<void>((res) => {
225− staller.on("data", (chunk: Buffer) => {
226− if (chunk.includes(0x5a)) res(); // 'Z' = ReadyForQuery
227− });
228− });
229− const parseBody = Buffer.from("\0select 1\0\0\0");
230− const parse = Buffer.concat([Buffer.from("P"), Buffer.alloc(4), parseBody]);
231− parse.writeInt32BE(4 + parseBody.length, 1);
232− staller.write(parse);
244+ const staller = await openWireClient(port);
245+ staller.write(parseFrame("select 1"));
233246
234247 const bystander = makeClient(port, 10);
235248 // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
@@ -245,4 +258,50 @@ describe("dev-db PGlite socket under concurrent connections", () => { · 9 unmodified lines
245258 }
246259 },
247260 );
261+
262+ // Regression for the second wedge mode behind the same CI cascade: a client
263+ // whose socket dies WHILE its pipeline-opening entry is executing. detach()
264+ // clears pipeline affinity before the entry finishes, so the queue then
265+ // assigned affinity to the already-dead handler — and nothing ever cleared
266+ // it: the dead handler has no timers left, and every other connection
267+ // (including fresh startups) queued behind the ghost forever. The queue now
268+ // tracks detached handlers and repairs affinity they can no longer release.
269+ it(
270+ "a client that dies mid-execution does not leave the queue pinned to its ghost",
271+ { timeout: 30_000 },
272+ async () => {
273+ const port = 45994;
274+ const db = await PGlite.create();
275+
276+ // Hold the marker query in flight long enough that the disconnect below
277+ // reliably lands while the entry is EXECUTING (after detach's cleanup,
278+ // before the queue takes affinity for it).
279+ const real = db.execProtocolRawStream.bind(db);
280+ (db as { execProtocolRawStream: typeof real }).execProtocolRawStream = async (...args) => {
281+ if (Buffer.from(args[0]).includes("ghost_marker")) await sleep(300);
282+ return real(...args);
283+ };
284+
285+ const server = new PGLiteSocketServer({ db, port, host: "127.0.0.1", maxConnections: 100 });
286+ await server.start();
287+
288+ const ghost = await openWireClient(port);
289+ ghost.write(parseFrame("select 'ghost_marker'"));
290+ // Give the data event time to reach the queue and start executing, then
291+ // die without a trace mid-flight.
292+ await sleep(100);
293+ ghost.destroy();
294+
295+ const bystander = makeClient(port);
296+ // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
297+ try {
298+ expect((await bystander.unsafe(`select 5 as five`))[0]).toEqual({ five: 5 });
299+ } finally {
300+ // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
301+ await bystander.end({ timeout: 5 }).catch(() => {});
302+ await server.stop();
303+ await db.close();
304+ }
305+ },
306+ );
248307 });
e2e/scenarios/google-health-checks.test.ts+2−6
@@ -16,7 +16,7 @@ import { · 15 unmodified lines
1616 import { scenario } from "../src/scenario";
1717 import { Api, Browser, Target } from "../src/services";
1818 import type { Identity, Target as TargetShape } from "../src/target";
19−import type { BrowserSurface } from "../src/surfaces/browser";
19+import { clickToReveal, type BrowserSurface } from "../src/surfaces/browser";
2020
2121 const api = composePluginApi([openApiHttpPlugin()] as const);
2222 type Client = HttpApiClient.ForApi<typeof api>;
@@ -88,12 +88,8 @@ const addGooglePresetFromCatalog = ( · 65 unmodified lines
8888 browser.session(identity, async ({ page, step }) => {
8989 await step(`Open ${presetName} from the connect catalog`, async () => {
9090 await page.goto("/integrations", { waitUntil: "networkidle" });
91− await page
92− .getByRole("button", { name: /Connect/ })
93− .first()
94− .click();
9591 const dialog = page.getByRole("dialog", { name: "Connect an integration" });
96− await dialog.waitFor();
92+ await clickToReveal(page.getByRole("button", { name: /Connect/ }).first(), dialog);
9793 await dialog.getByPlaceholder(/Search or paste a URL/).fill(presetName);
9894 await dialog.getByRole("link", { name: new RegExp(`^${presetName}\\b`) }).click();
9995 });
e2e/scenarios/google-photos-preset-ui.test.ts+2−2
@@ -3,6 +3,7 @@ import { Effect } from "effect"; · 2 unmodified lines
33
44 import { scenario } from "../src/scenario";
55 import { Browser, Target } from "../src/services";
6+import { clickToReveal } from "../src/surfaces/browser";
67
78 scenario(
89 "Google Photos: separated catalog presets open a Photos service add flow",
@@ -17,9 +18,8 @@ scenario( · 8 unmodified lines
1718 "Find the separated Google Photos presets from the integrations picker",
1819 async () => {
1920 await page.goto("/integrations", { waitUntil: "networkidle" });
20− await page.getByRole("button", { name: "Connect" }).click();
2121 const dialog = page.getByRole("dialog", { name: "Connect an integration" });
22− await dialog.waitFor();
22+ await clickToReveal(page.getByRole("button", { name: "Connect" }), dialog);
2323 await dialog.getByPlaceholder(/Search or paste a URL/).fill("google photos");
2424 await dialog.getByRole("link", { name: /^Google Photos Library\b/ }).waitFor();
2525 await dialog.getByRole("link", { name: /^Google Photos Picker\b/ }).waitFor();
e2e/scenarios/provider-plugins-ui.test.ts+5−2
@@ -3,6 +3,7 @@ import { Effect } from "effect"; · 2 unmodified lines
33
44 import { scenario } from "../src/scenario";
55 import { Browser, Target } from "../src/services";
6+import { clickToReveal } from "../src/surfaces/browser";
67
78 scenario(
89 "Provider catalog · Google and Microsoft services are OpenAPI presets",
@@ -15,8 +16,10 @@ scenario( · 6 unmodified lines
1516 yield* browser.session(identity, async ({ page, step }) => {
1617 await step("Open the integrations picker", async () => {
1718 await page.goto("/integrations", { waitUntil: "networkidle" });
18− await page.getByRole("button", { name: "Connect" }).click();
19− await page.getByRole("dialog", { name: "Connect an integration" }).waitFor();
19+ await clickToReveal(
20+ page.getByRole("button", { name: "Connect" }),
21+ page.getByRole("dialog", { name: "Connect an integration" }),
22+ );
2023 });
2124
2225 await step("The picker exposes OpenAPI plus provider service presets", async () => {
e2e/src/surfaces/browser.ts+30−1
@@ -9,7 +9,7 @@ import { join } from "node:path"; · 8 unmodified lines
99 import { promisify } from "node:util";
1010
1111 import { Effect } from "effect";
12−import { chromium, type Page } from "playwright";
12+import { chromium, type Locator, type Page } from "playwright";
1313
1414 import { beat, enterFocus, markNavigation, markRecordingStart } from "../timeline";
1515 import { appendTraces, type TraceEntry } from "../trace-harvest";
@@ -36,6 +36,35 @@ const slug = (text: string): string => · 20 unmodified lines
3636 .replace(/^-+|-+$/g, "")
3737 .slice(0, 60);
3838
39+/**
40+ * Click `trigger` until `revealed` is visible.
41+ *
42+ * `waitUntil: "networkidle"` does not mean the console has hydrated: a click
43+ * that lands between the SSR paint and React attaching the handler is
44+ * swallowed without a trace, and whatever the click was meant to open never
45+ * appears (the "Connect an integration" dialog no-show flake). Re-clicking a
46+ * reveal-style trigger is idempotent, so retry until the result is actually
47+ * on screen; the final attempt waits with the full timeout so the failure
48+ * surfaces as the ordinary locator error.
49+ */
50+export const clickToReveal = async (
51+ trigger: Locator,
52+ revealed: Locator,
53+ { attempts = 5, revealTimeoutMs = 4_000 }: { attempts?: number; revealTimeoutMs?: number } = {},
54+): Promise<void> => {
55+ for (let attempt = 1; attempt < attempts; attempt++) {
56+ await trigger.click();
57+ const shown = await revealed
58+ .waitFor({ timeout: revealTimeoutMs })
59+ .then(() => true)
60+ // oxlint-disable-next-line executor/no-promise-catch -- retry boundary: a missed reveal is the signal to click again, not a failure
61+ .catch(() => false);
62+ if (shown) return;
63+ }
64+ await trigger.click();
65+ await revealed.waitFor({ timeout: revealTimeoutMs });
66+};
67+
3968 // acquireUseRelease so a vitest timeout (fiber interruption) still closes the
4069 // browser and flushes video + trace — a bare promise would leak Chromium.
4170 export const makeBrowserSurface = (dir: string, target: Target): BrowserSurface => ({
patches/@electric-sql%2Fpglite-socket@0.1.4.patch+5−2
@@ -4,13 +4,16 @@ index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2 · 3 unmodified lines
44 diff --git a/node_modules/@electric-sql/pglite-socket/.bun-tag-eaa11f63ffd98a26 b/.bun-tag-eaa11f63ffd98a26
55 new file mode 100644
66 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
7+diff --git a/node_modules/@electric-sql/pglite-socket/.bun-tag-fbab1bb0bfbef953 b/.bun-tag-fbab1bb0bfbef953
8+new file mode 100644
9+index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
710 diff --git a/dist/chunk-NSUMFCRM.js b/dist/chunk-NSUMFCRM.js
8−index 37d45ebc5150c43c919fbdb1c7fffb51b18fda8c..cc2a8f6e35c6a676f4fdecb7d580fc1202b74ace 100644
11+index 37d45ebc5150c43c919fbdb1c7fffb51b18fda8c..1cdfaaa69b568a0bed7618fe527ab2c10b1b9403 100644
912 --- a/dist/chunk-NSUMFCRM.js
1013 +++ b/dist/chunk-NSUMFCRM.js
1114 @@ -1,3 +1,3 @@
1215 -import{createServer as m}from"net";var b=6e4,c=class{constructor(s,e=!1){this.queue=[];this.processing=!1;this.lastHandlerId=null;this.db=s,this.debug=e}log(s,...e){this.debug&&console.log(`[QueryQueueManager] ${s}`,...e)}async enqueue(s,e,i){return new Promise((t,r)=>{let o={handlerId:s,message:e,resolve:t,reject:r,timestamp:Date.now(),onData:i};this.queue.push(o),this.log(`enqueued query from handler #${s}, queue size: ${this.queue.length}`),this.processing||this.processQueue()})}async processQueue(){if(!(this.processing||this.queue.length===0)){for(this.processing=!0;this.queue.length>0;){let s;if(this.db.isInTransaction()&&this.lastHandlerId){let t=this.queue.findIndex(r=>r.handlerId===this.lastHandlerId);t===-1?(this.log("transaction started, but no query from the same handler id found in queue",this.lastHandlerId),s=null):s=this.queue.splice(t,1)[0]}else s=this.queue.shift();if(!s)break;let e=Date.now()-s.timestamp;this.log(`processing query from handler #${s.handlerId} (waited ${e}ms)`);let i=0;try{await this.db.runExclusive(async()=>await this.db.execProtocolRawStream(s.message,{onRawData:t=>{i+=t.length,s.onData(t)}}))}catch(t){this.log(`query from handler #${s.handlerId} failed:`,t),s.reject(t);return}this.log(`query from handler #${s.handlerId} completed, ${i} bytes`),this.lastHandlerId=s.handlerId,s.resolve(i)}this.processing=!1,this.log("queue processing complete, queue length is",this.queue.length)}}getQueueLength(){return this.queue.length}clearQueueForHandler(s){let e=this.queue.length;this.queue=this.queue.filter(t=>t.handlerId===s?(t.reject(new Error("Handler disconnected")),!1):!0);let i=e-this.queue.length;i>0&&this.log(`cleared ${i} queries for handler #${s}`)}async clearTransactionIfNeeded(s){this.db.isInTransaction()&&this.lastHandlerId===s&&(await this.db.exec("ROLLBACK"),this.lastHandlerId=null,await this.processQueue())}},l=class l extends EventTarget{constructor(e){super();this.socket=null;this.active=!1;this.messageBuffer=Buffer.alloc(0);this.lastActivityTime=Date.now();this.queryQueue=e.queryQueue,this.closeOnDetach=e.closeOnDetach??!1,this.inspect=e.inspect??!1,this.debug=e.debug??!1,this.idleTimeout=e.idleTimeout??0,this.id=l.nextHandlerId++,this.log("constructor: created new handler")}get handlerId(){return this.id}log(e,...i){this.debug&&console.log(`[PGLiteSocketHandler#${this.id}] ${e}`,...i)}async attach(e){if(this.log(`attach: attaching socket from ${e.remoteAddress}:${e.remotePort}`),this.socket)throw new Error("Socket already attached");return this.socket=e,this.active=!0,this.lastActivityTime=Date.now(),e.setNoDelay(!0),this.idleTimeout>0&&this.resetIdleTimer(),this.log("attach: setting up socket event handlers"),e.on("data",i=>{this.lastActivityTime=Date.now(),this.resetIdleTimer(),setImmediate(async()=>{try{await this.handleData(i)}catch(t){this.log("socket on data error: ",t),this.handleError(t)}})}),e.on("error",i=>{setImmediate(()=>this.handleError(i))}),e.on("close",()=>{setImmediate(()=>this.handleClose())}),this.log("attach: socket handler ready"),this}resetIdleTimer(){this.idleTimeout<=0||(this.idleTimer&&clearTimeout(this.idleTimer),this.idleTimer=setTimeout(()=>{let e=Date.now()-this.lastActivityTime;this.log(`idle timeout after ${e}ms`),this.handleError(new Error("Idle timeout"))},this.idleTimeout))}async detach(e){if(this.log(`detach: detaching socket, close=${e??this.closeOnDetach}`),this.idleTimer&&(clearTimeout(this.idleTimer),this.idleTimer=void 0),this.queryQueue.clearQueueForHandler(this.id),await this.queryQueue.clearTransactionIfNeeded(this.id),!this.socket)return this.log("detach: no socket attached, nothing to do"),this;if(this.socket.removeAllListeners("data"),this.socket.removeAllListeners("error"),this.socket.removeAllListeners("close"),(e??this.closeOnDetach)&&this.socket.writable){this.log("detach: closing socket");try{this.socket.end(),this.socket.destroy()}catch(i){this.log("detach: error closing socket:",i)}}return this.socket=null,this.active=!1,this.messageBuffer=Buffer.alloc(0),this.log("detach: handler cleaned up"),this}get isAttached(){return this.socket!==null}async handleData(e){if(!this.socket||!this.active)return this.log("handleData: no active socket, ignoring data"),0;this.log(`handleData: received ${e.length} bytes`),this.messageBuffer=Buffer.concat([this.messageBuffer,e]),this.inspectData("incoming",e);try{let i=0;for(;this.messageBuffer.length>0;){let t=0,r=!1;if(this.messageBuffer.length>=4){let n=this.messageBuffer.readInt32BE(0);if(this.messageBuffer.length>=8){let a=this.messageBuffer.readInt32BE(4);(a===196608||a===196608)&&(t=n,r=this.messageBuffer.length>=t)}!r&&this.messageBuffer.length>=5&&(t=1+this.messageBuffer.readInt32BE(1),r=this.messageBuffer.length>=t)}if(!r||t===0){this.log(`handleData: incomplete message, buffering ${this.messageBuffer.length} bytes`);break}let o=this.messageBuffer.slice(0,t);if(this.messageBuffer=this.messageBuffer.slice(t),this.log(`handleData: processing message of ${o.length} bytes`),!this.active||!this.socket){this.log("handleData: socket no longer active, stopping processing");break}let h;if(await this.queryQueue.enqueue(this.id,new Uint8Array(o),n=>{this.log(`handleData: received ${n.length} bytes from PGlite`),this.inspectData("outgoing",n),n.length>0&&this.socket&&this.socket.writable&&this.active&&(this.log("handleData: writing response to socket"),this.socket?.writable?this.socket.write(Buffer.from(n),a=>{a?(this.log("handleData: error writing to socket:",a),h=a):this.log(`handleData: socket sent: ${n.length} bytes`)}):this.log("handleData: socket no longer writable")),i+=n.length}),h)throw h}return this.dispatchEvent(new CustomEvent("data",{detail:{incoming:e.length,outgoing:i}})),i}catch(i){throw this.log("handleData: error processing data:",i),i}}handleError(e){if(!this.active){this.log("handleError: handler not active, ignoring error");return}e.message?.includes("ECONNRESET")?this.log("handleError: client disconnected (ECONNRESET) - normal behavior"):e.message?.includes("Idle timeout")?this.log("handleError: connection idle timeout"):this.log("handleError:",e),this.active=!1,this.dispatchEvent(new CustomEvent("error",{detail:e})),this.detach(!0)}handleClose(){this.log("handleClose: socket closed"),this.active=!1,this.dispatchEvent(new CustomEvent("close")),this.detach(!1)}inspectData(e,i){if(this.inspect){console.log("-".repeat(75)),console.log(e==="incoming"?"-> incoming":"<- outgoing",i.length,"bytes");for(let t=0;t<i.length;t+=16){let r=Math.min(16,i.length-t),o="";for(let n=0;n<16;n++)if(n<r){let a=i[t+n];o+=a.toString(16).padStart(2,"0")+" "}else o+=" ";let h="";for(let n=0;n<r;n++){let a=i[t+n];h+=a>=32&&a<=126?String.fromCharCode(a):"."}console.log(`${t.toString(16).padStart(8,"0")} ${o} ${h}`)}}}};l.nextHandlerId=1;var d=l,u=class extends EventTarget{constructor(e){super();this.server=null;this.active=!1;this.handlers=new Set;this.db=e.db,e.path?this.path=e.path:(typeof e.port=="number"?this.port=e.port??e.port:this.port=5432,this.host=e.host||"127.0.0.1"),this.inspect=e.inspect??!1,this.debug=e.debug??!1,this.idleTimeout=e.idleTimeout??0,this.maxConnections=e.maxConnections??1,this.queryQueue=new c(this.db,this.debug),this.log(`constructor: created server on ${this.getServerConn()}`),this.log(`constructor: max connections: ${this.maxConnections}`),this.idleTimeout>0&&this.log(`constructor: idle timeout: ${this.idleTimeout}ms`)}log(e,...i){this.debug&&console.log(`[PGLiteSocketServer] ${e}`,...i)}async start(){if(this.log(`start: starting server on ${this.getServerConn()}`),this.server)throw new Error("Socket server already started");return await this.db.waitReady,this.active=!0,this.server=m(e=>{setImmediate(()=>this.handleConnection(e))}),this.server.maxConnections=this.maxConnections,new Promise((e,i)=>{if(!this.server)return i(new Error("Server not initialized"));if(this.server.on("error",t=>{this.log("start: server error:",t),this.dispatchEvent(new CustomEvent("error",{detail:t})),this.active||i(t)}),this.path)this.server.listen(this.path,()=>{this.log(`start: server listening on ${this.getServerConn()}`),this.dispatchEvent(new CustomEvent("listening",{detail:{path:this.path}})),e()});else{let t=this.server;t.listen(this.port,this.host,()=>{let r=t.address();if(r===null||typeof r!="object")throw Error("Expected address info");this.port=r.port,this.log(`start: server listening on ${this.getServerConn()}`),this.dispatchEvent(new CustomEvent("listening",{detail:{port:this.port,host:this.host}})),e()})}})}getServerConn(){return this.path?this.path:`${this.host}:${this.port}`}async stop(){this.log("stop: stopping server"),this.active=!1,this.log(`stop: detaching ${this.handlers.size} handlers`);for(let e of this.handlers)e.detach(!0);return this.handlers.clear(),this.server?new Promise(e=>{if(!this.server)return e();this.server.close(()=>{this.log("stop: server closed"),this.server=null,this.dispatchEvent(new CustomEvent("close")),e()})}):(this.log("stop: server not running, nothing to do"),Promise.resolve())}async handleConnection(e){let i={clientAddress:e.remoteAddress||"unknown",clientPort:e.remotePort||0};if(this.log(`handleConnection: new connection from ${i.clientAddress}:${i.clientPort}`),this.log(`handleConnection: active connections: ${this.handlers.size}, queued queries: ${this.queryQueue.getQueueLength()}`),!this.active){this.log("handleConnection: server not active, closing connection");try{e.end()}catch(r){this.log("handleConnection: error closing socket:",r)}return}if(this.handlers.size>=this.maxConnections){this.log("handleConnection: max connections reached, rejecting"),e.write(Buffer.from(`Too many connections
13−+import{createServer as m}from"net";var b=6e4,c=class{constructor(s,e=!1){this.queue=[];this.processing=!1;this.lastHandlerId=null;this.pipelineHandlerId=null;this.db=s,this.debug=e}log(s,...e){this.debug&&console.log(`[QueryQueueManager] ${s}`,...e)}async enqueue(s,e,i,S=!0){return new Promise((t,r)=>{let o={handlerId:s,message:e,resolve:t,reject:r,timestamp:Date.now(),onData:i,closes:S};this.queue.push(o),this.log(`enqueued query from handler #${s}, queue size: ${this.queue.length}`),this.processing||this.processQueue()})}async processQueue(){if(!(this.processing||this.queue.length===0)){for(this.processing=!0;this.queue.length>0;){let s;let __affine=this.db.isInTransaction()&&this.lastHandlerId?this.lastHandlerId:this.pipelineHandlerId;if(__affine){let t=this.queue.findIndex(r=>r.handlerId===__affine);t===-1?(this.log("affinity held, waiting for handler",__affine),s=null):s=this.queue.splice(t,1)[0]}else s=this.queue.shift();if(!s)break;let e=Date.now()-s.timestamp;this.log(`processing query from handler #${s.handlerId} (waited ${e}ms)`);let i=0;try{await this.db.runExclusive(async()=>await this.db.execProtocolRawStream(s.message,{onRawData:t=>{i+=t.length,s.onData(t)}}))}catch(t){this.log(`query from handler #${s.handlerId} failed:`,t),s.reject(t),this.pipelineHandlerId=null;continue}this.log(`query from handler #${s.handlerId} completed, ${i} bytes`),this.lastHandlerId=s.handlerId,this.pipelineHandlerId=s.closes?null:s.handlerId,s.resolve(i)}this.processing=!1,this.log("queue processing complete, queue length is",this.queue.length)}}getQueueLength(){return this.queue.length}holdsAffinity(s){return this.pipelineHandlerId===s||this.db.isInTransaction()&&this.lastHandlerId===s}clearQueueForHandler(s){let e=this.queue.length;this.queue=this.queue.filter(t=>t.handlerId===s?(t.reject(new Error("Handler disconnected")),!1):!0);let i=e-this.queue.length;i>0&&this.log(`cleared ${i} queries for handler #${s}`)}async clearPipelineIfNeeded(s){this.pipelineHandlerId===s&&(this.pipelineHandlerId=null,await this.db.runExclusive(async()=>await this.db.execProtocolRawStream(new Uint8Array([83,0,0,0,4]),{onRawData:()=>{}})).catch(()=>{}),this.processQueue())}async clearTransactionIfNeeded(s){this.db.isInTransaction()&&this.lastHandlerId===s&&(await this.db.exec("ROLLBACK"),this.lastHandlerId=null,await this.processQueue())}},l=class l extends EventTarget{constructor(e){super();this.socket=null;this.active=!1;this.messageBuffer=Buffer.alloc(0);this.lastActivityTime=Date.now();this.queryQueue=e.queryQueue,this.closeOnDetach=e.closeOnDetach??!1,this.inspect=e.inspect??!1,this.debug=e.debug??!1,this.idleTimeout=e.idleTimeout??0,this.id=l.nextHandlerId++,this.log("constructor: created new handler")}get handlerId(){return this.id}log(e,...i){this.debug&&console.log(`[PGLiteSocketHandler#${this.id}] ${e}`,...i)}async attach(e){if(this.log(`attach: attaching socket from ${e.remoteAddress}:${e.remotePort}`),this.socket)throw new Error("Socket already attached");return this.socket=e,this.active=!0,this.lastActivityTime=Date.now(),e.setNoDelay(!0),this.idleTimeout>0&&this.resetIdleTimer(),this.log("attach: setting up socket event handlers"),e.on("data",i=>{this.lastActivityTime=Date.now(),this.resetIdleTimer(),setImmediate(async()=>{try{await this.handleData(i)}catch(t){this.log("socket on data error: ",t),this.handleError(t)}})}),e.on("error",i=>{setImmediate(()=>this.handleError(i))}),e.on("close",()=>{setImmediate(()=>this.handleClose())}),this.log("attach: socket handler ready"),this}resetIdleTimer(){this.idleTimeout<=0||(this.idleTimer&&clearTimeout(this.idleTimer),this.idleTimer=setTimeout(()=>{if(!this.queryQueue.holdsAffinity(this.id)){this.resetIdleTimer();return}let e=Date.now()-this.lastActivityTime;this.log(`idle timeout after ${e}ms`),this.handleError(new Error("Idle timeout"))},this.idleTimeout))}async detach(e){if(this.log(`detach: detaching socket, close=${e??this.closeOnDetach}`),this.idleTimer&&(clearTimeout(this.idleTimer),this.idleTimer=void 0),this.queryQueue.clearQueueForHandler(this.id),await this.queryQueue.clearTransactionIfNeeded(this.id),await this.queryQueue.clearPipelineIfNeeded(this.id),!this.socket)return this.log("detach: no socket attached, nothing to do"),this;if(this.socket.removeAllListeners("data"),this.socket.removeAllListeners("error"),this.socket.removeAllListeners("close"),(e??this.closeOnDetach)&&this.socket.writable){this.log("detach: closing socket");try{this.socket.end(),this.socket.destroy()}catch(i){this.log("detach: error closing socket:",i)}}return this.socket=null,this.active=!1,this.messageBuffer=Buffer.alloc(0),this.log("detach: handler cleaned up"),this}get isAttached(){return this.socket!==null}async handleData(e){if(!this.socket||!this.active)return this.log("handleData: no active socket, ignoring data"),0;this.log(`handleData: received ${e.length} bytes`),this.messageBuffer=Buffer.concat([this.messageBuffer,e]),this.inspectData("incoming",e);try{let i=0;const __frames=[];for(;this.messageBuffer.length>0;){let t=0,r=!1;if(this.messageBuffer.length>=4){let n=this.messageBuffer.readInt32BE(0);if(this.messageBuffer.length>=8){let a=this.messageBuffer.readInt32BE(4);(a===196608||a===196608)&&(t=n,r=this.messageBuffer.length>=t)}!r&&this.messageBuffer.length>=5&&(t=1+this.messageBuffer.readInt32BE(1),r=this.messageBuffer.length>=t)}if(!r||t===0){this.log(`handleData: incomplete message, buffering ${this.messageBuffer.length} bytes`);break}let o=this.messageBuffer.slice(0,t);if(this.messageBuffer=this.messageBuffer.slice(t),this.log(`handleData: processing message of ${o.length} bytes`),!this.active||!this.socket){this.log("handleData: socket no longer active, stopping processing");break}__frames.push(o)}if(__frames.length===0)return this.dispatchEvent(new CustomEvent("data",{detail:{incoming:e.length,outgoing:i}})),i;{let o=__frames.length===1?__frames[0]:Buffer.concat(__frames);const __lastF=__frames[__frames.length-1];const __lt=__lastF[0]>=65?__lastF[0]:null;const __closes=__lt===null||__lt===83||__lt===81||__lt===88;let h;if(await this.queryQueue.enqueue(this.id,new Uint8Array(o),n=>{this.log(`handleData: received ${n.length} bytes from PGlite`),this.inspectData("outgoing",n),n.length>0&&this.socket&&this.socket.writable&&this.active&&(this.log("handleData: writing response to socket"),this.socket?.writable?this.socket.write(Buffer.from(n),a=>{a?(this.log("handleData: error writing to socket:",a),h=a):this.log(`handleData: socket sent: ${n.length} bytes`)}):this.log("handleData: socket no longer writable")),i+=n.length},__closes),h)throw h}return this.dispatchEvent(new CustomEvent("data",{detail:{incoming:e.length,outgoing:i}})),i}catch(i){throw this.log("handleData: error processing data:",i),i}}handleError(e){if(!this.active){this.log("handleError: handler not active, ignoring error");return}e.message?.includes("ECONNRESET")?this.log("handleError: client disconnected (ECONNRESET) - normal behavior"):e.message?.includes("Idle timeout")?this.log("handleError: connection idle timeout"):this.log("handleError:",e),this.active=!1,this.dispatchEvent(new CustomEvent("error",{detail:e})),this.detach(!0).catch(()=>{})}handleClose(){this.log("handleClose: socket closed"),this.active=!1,this.dispatchEvent(new CustomEvent("close")),this.detach(!1).catch(()=>{})}inspectData(e,i){if(this.inspect){console.log("-".repeat(75)),console.log(e==="incoming"?"-> incoming":"<- outgoing",i.length,"bytes");for(let t=0;t<i.length;t+=16){let r=Math.min(16,i.length-t),o="";for(let n=0;n<16;n++)if(n<r){let a=i[t+n];o+=a.toString(16).padStart(2,"0")+" "}else o+=" ";let h="";for(let n=0;n<r;n++){let a=i[t+n];h+=a>=32&&a<=126?String.fromCharCode(a):"."}console.log(`${t.toString(16).padStart(8,"0")} ${o} ${h}`)}}}};l.nextHandlerId=1;var d=l,u=class extends EventTarget{constructor(e){super();this.server=null;this.active=!1;this.handlers=new Set;this.db=e.db,e.path?this.path=e.path:(typeof e.port=="number"?this.port=e.port??e.port:this.port=5432,this.host=e.host||"127.0.0.1"),this.inspect=e.inspect??!1,this.debug=e.debug??!1,this.idleTimeout=e.idleTimeout??0,this.maxConnections=e.maxConnections??1,this.queryQueue=new c(this.db,this.debug),this.log(`constructor: created server on ${this.getServerConn()}`),this.log(`constructor: max connections: ${this.maxConnections}`),this.idleTimeout>0&&this.log(`constructor: idle timeout: ${this.idleTimeout}ms`)}log(e,...i){this.debug&&console.log(`[PGLiteSocketServer] ${e}`,...i)}async start(){if(this.log(`start: starting server on ${this.getServerConn()}`),this.server)throw new Error("Socket server already started");return await this.db.waitReady,this.active=!0,this.server=m(e=>{setImmediate(()=>this.handleConnection(e))}),this.server.maxConnections=this.maxConnections,new Promise((e,i)=>{if(!this.server)return i(new Error("Server not initialized"));if(this.server.on("error",t=>{this.log("start: server error:",t),this.dispatchEvent(new CustomEvent("error",{detail:t})),this.active||i(t)}),this.path)this.server.listen(this.path,()=>{this.log(`start: server listening on ${this.getServerConn()}`),this.dispatchEvent(new CustomEvent("listening",{detail:{path:this.path}})),e()});else{let t=this.server;t.listen(this.port,this.host,()=>{let r=t.address();if(r===null||typeof r!="object")throw Error("Expected address info");this.port=r.port,this.log(`start: server listening on ${this.getServerConn()}`),this.dispatchEvent(new CustomEvent("listening",{detail:{port:this.port,host:this.host}})),e()})}})}getServerConn(){return this.path?this.path:`${this.host}:${this.port}`}async stop(){this.log("stop: stopping server"),this.active=!1,this.log(`stop: detaching ${this.handlers.size} handlers`);for(let e of this.handlers)e.detach(!0).catch(()=>{});return this.handlers.clear(),this.server?new Promise(e=>{if(!this.server)return e();this.server.close(()=>{this.log("stop: server closed"),this.server=null,this.dispatchEvent(new CustomEvent("close")),e()})}):(this.log("stop: server not running, nothing to do"),Promise.resolve())}async handleConnection(e){let i={clientAddress:e.remoteAddress||"unknown",clientPort:e.remotePort||0};if(this.log(`handleConnection: new connection from ${i.clientAddress}:${i.clientPort}`),this.log(`handleConnection: active connections: ${this.handlers.size}, queued queries: ${this.queryQueue.getQueueLength()}`),!this.active){this.log("handleConnection: server not active, closing connection");try{e.end()}catch(r){this.log("handleConnection: error closing socket:",r)}return}if(this.handlers.size>=this.maxConnections){this.log("handleConnection: max connections reached, rejecting"),e.write(Buffer.from(`Too many connections
16++import{createServer as m}from"net";var b=6e4,c=class{constructor(s,e=!1){this.queue=[];this.processing=!1;this.lastHandlerId=null;this.pipelineHandlerId=null;this.dead=new Set();this.db=s,this.debug=e}log(s,...e){this.debug&&console.log(`[QueryQueueManager] ${s}`,...e)}async enqueue(s,e,i,S=!0){return new Promise((t,r)=>{let o={handlerId:s,message:e,resolve:t,reject:r,timestamp:Date.now(),onData:i,closes:S};this.queue.push(o),this.log(`enqueued query from handler #${s}, queue size: ${this.queue.length}`),this.processing||this.processQueue()})}async processQueue(){if(!(this.processing||this.queue.length===0)){for(this.processing=!0;this.queue.length>0;){let s;let __affine=this.db.isInTransaction()&&this.lastHandlerId?this.lastHandlerId:this.pipelineHandlerId;if(__affine&&this.dead.has(__affine)){this.log("affinity held by detached handler, recovering",__affine);if(this.db.isInTransaction()&&this.lastHandlerId===__affine){await this.db.exec("ROLLBACK").catch(()=>{});this.lastHandlerId=null}if(this.pipelineHandlerId===__affine){this.pipelineHandlerId=null;await this.db.runExclusive(async()=>await this.db.execProtocolRawStream(new Uint8Array([83,0,0,0,4]),{onRawData:()=>{}})).catch(()=>{})}continue}if(__affine){let t=this.queue.findIndex(r=>r.handlerId===__affine);t===-1?(this.log("affinity held, waiting for handler",__affine),s=null):s=this.queue.splice(t,1)[0]}else s=this.queue.shift();if(!s)break;let e=Date.now()-s.timestamp;this.log(`processing query from handler #${s.handlerId} (waited ${e}ms)`);let i=0;try{await this.db.runExclusive(async()=>await this.db.execProtocolRawStream(s.message,{onRawData:t=>{i+=t.length,s.onData(t)}}))}catch(t){this.log(`query from handler #${s.handlerId} failed:`,t),s.reject(t),this.pipelineHandlerId=null;continue}this.log(`query from handler #${s.handlerId} completed, ${i} bytes`),this.lastHandlerId=s.handlerId,this.pipelineHandlerId=s.closes?null:s.handlerId,s.resolve(i)}this.processing=!1,this.log("queue processing complete, queue length is",this.queue.length)}}getQueueLength(){return this.queue.length}holdsAffinity(s){return this.pipelineHandlerId===s||this.db.isInTransaction()&&this.lastHandlerId===s}clearQueueForHandler(s){this.dead.add(s);let e=this.queue.length;this.queue=this.queue.filter(t=>t.handlerId===s?(t.reject(new Error("Handler disconnected")),!1):!0);let i=e-this.queue.length;i>0&&this.log(`cleared ${i} queries for handler #${s}`)}async clearPipelineIfNeeded(s){this.pipelineHandlerId===s&&(this.pipelineHandlerId=null,await this.db.runExclusive(async()=>await this.db.execProtocolRawStream(new Uint8Array([83,0,0,0,4]),{onRawData:()=>{}})).catch(()=>{}),this.processQueue())}async clearTransactionIfNeeded(s){this.db.isInTransaction()&&this.lastHandlerId===s&&(await this.db.exec("ROLLBACK"),this.lastHandlerId=null,await this.processQueue())}},l=class l extends EventTarget{constructor(e){super();this.socket=null;this.active=!1;this.messageBuffer=Buffer.alloc(0);this.lastActivityTime=Date.now();this.queryQueue=e.queryQueue,this.closeOnDetach=e.closeOnDetach??!1,this.inspect=e.inspect??!1,this.debug=e.debug??!1,this.idleTimeout=e.idleTimeout??0,this.id=l.nextHandlerId++,this.log("constructor: created new handler")}get handlerId(){return this.id}log(e,...i){this.debug&&console.log(`[PGLiteSocketHandler#${this.id}] ${e}`,...i)}async attach(e){if(this.log(`attach: attaching socket from ${e.remoteAddress}:${e.remotePort}`),this.socket)throw new Error("Socket already attached");return this.socket=e,this.active=!0,this.lastActivityTime=Date.now(),e.setNoDelay(!0),this.idleTimeout>0&&this.resetIdleTimer(),this.log("attach: setting up socket event handlers"),e.on("data",i=>{this.lastActivityTime=Date.now(),this.resetIdleTimer(),setImmediate(async()=>{try{await this.handleData(i)}catch(t){this.log("socket on data error: ",t),this.handleError(t)}})}),e.on("error",i=>{setImmediate(()=>this.handleError(i))}),e.on("close",()=>{setImmediate(()=>this.handleClose())}),this.log("attach: socket handler ready"),this}resetIdleTimer(){this.idleTimeout<=0||(this.idleTimer&&clearTimeout(this.idleTimer),this.idleTimer=setTimeout(()=>{if(!this.queryQueue.holdsAffinity(this.id)){this.resetIdleTimer();return}let e=Date.now()-this.lastActivityTime;this.log(`idle timeout after ${e}ms`),this.handleError(new Error("Idle timeout"))},this.idleTimeout))}async detach(e){if(this.log(`detach: detaching socket, close=${e??this.closeOnDetach}`),this.idleTimer&&(clearTimeout(this.idleTimer),this.idleTimer=void 0),this.queryQueue.clearQueueForHandler(this.id),await this.queryQueue.clearTransactionIfNeeded(this.id),await this.queryQueue.clearPipelineIfNeeded(this.id),!this.socket)return this.log("detach: no socket attached, nothing to do"),this;if(this.socket.removeAllListeners("data"),this.socket.removeAllListeners("error"),this.socket.removeAllListeners("close"),(e??this.closeOnDetach)&&this.socket.writable){this.log("detach: closing socket");try{this.socket.end(),this.socket.destroy()}catch(i){this.log("detach: error closing socket:",i)}}return this.socket=null,this.active=!1,this.messageBuffer=Buffer.alloc(0),this.log("detach: handler cleaned up"),this}get isAttached(){return this.socket!==null}async handleData(e){if(!this.socket||!this.active)return this.log("handleData: no active socket, ignoring data"),0;this.log(`handleData: received ${e.length} bytes`),this.messageBuffer=Buffer.concat([this.messageBuffer,e]),this.inspectData("incoming",e);try{let i=0;const __frames=[];for(;this.messageBuffer.length>0;){let t=0,r=!1;if(this.messageBuffer.length>=4){let n=this.messageBuffer.readInt32BE(0);if(this.messageBuffer.length>=8){let a=this.messageBuffer.readInt32BE(4);(a===196608||a===196608)&&(t=n,r=this.messageBuffer.length>=t)}!r&&this.messageBuffer.length>=5&&(t=1+this.messageBuffer.readInt32BE(1),r=this.messageBuffer.length>=t)}if(!r||t===0){this.log(`handleData: incomplete message, buffering ${this.messageBuffer.length} bytes`);break}let o=this.messageBuffer.slice(0,t);if(this.messageBuffer=this.messageBuffer.slice(t),this.log(`handleData: processing message of ${o.length} bytes`),!this.active||!this.socket){this.log("handleData: socket no longer active, stopping processing");break}__frames.push(o)}if(__frames.length===0)return this.dispatchEvent(new CustomEvent("data",{detail:{incoming:e.length,outgoing:i}})),i;{let o=__frames.length===1?__frames[0]:Buffer.concat(__frames);const __lastF=__frames[__frames.length-1];const __lt=__lastF[0]>=65?__lastF[0]:null;const __closes=__lt===null||__lt===83||__lt===81||__lt===88;let h;if(await this.queryQueue.enqueue(this.id,new Uint8Array(o),n=>{this.log(`handleData: received ${n.length} bytes from PGlite`),this.inspectData("outgoing",n),n.length>0&&this.socket&&this.socket.writable&&this.active&&(this.log("handleData: writing response to socket"),this.socket?.writable?this.socket.write(Buffer.from(n),a=>{a?(this.log("handleData: error writing to socket:",a),h=a):this.log(`handleData: socket sent: ${n.length} bytes`)}):this.log("handleData: socket no longer writable")),i+=n.length},__closes),h)throw h}return this.dispatchEvent(new CustomEvent("data",{detail:{incoming:e.length,outgoing:i}})),i}catch(i){throw this.log("handleData: error processing data:",i),i}}handleError(e){if(!this.active){this.log("handleError: handler not active, ignoring error");return}e.message?.includes("ECONNRESET")?this.log("handleError: client disconnected (ECONNRESET) - normal behavior"):e.message?.includes("Idle timeout")?this.log("handleError: connection idle timeout"):this.log("handleError:",e),this.active=!1,this.dispatchEvent(new CustomEvent("error",{detail:e})),this.detach(!0).catch(()=>{})}handleClose(){this.log("handleClose: socket closed"),this.active=!1,this.dispatchEvent(new CustomEvent("close")),this.detach(!1).catch(()=>{})}inspectData(e,i){if(this.inspect){console.log("-".repeat(75)),console.log(e==="incoming"?"-> incoming":"<- outgoing",i.length,"bytes");for(let t=0;t<i.length;t+=16){let r=Math.min(16,i.length-t),o="";for(let n=0;n<16;n++)if(n<r){let a=i[t+n];o+=a.toString(16).padStart(2,"0")+" "}else o+=" ";let h="";for(let n=0;n<r;n++){let a=i[t+n];h+=a>=32&&a<=126?String.fromCharCode(a):"."}console.log(`${t.toString(16).padStart(8,"0")} ${o} ${h}`)}}}};l.nextHandlerId=1;var d=l,u=class extends EventTarget{constructor(e){super();this.server=null;this.active=!1;this.handlers=new Set;this.db=e.db,e.path?this.path=e.path:(typeof e.port=="number"?this.port=e.port??e.port:this.port=5432,this.host=e.host||"127.0.0.1"),this.inspect=e.inspect??!1,this.debug=e.debug??!1,this.idleTimeout=e.idleTimeout??0,this.maxConnections=e.maxConnections??1,this.queryQueue=new c(this.db,this.debug),this.log(`constructor: created server on ${this.getServerConn()}`),this.log(`constructor: max connections: ${this.maxConnections}`),this.idleTimeout>0&&this.log(`constructor: idle timeout: ${this.idleTimeout}ms`)}log(e,...i){this.debug&&console.log(`[PGLiteSocketServer] ${e}`,...i)}async start(){if(this.log(`start: starting server on ${this.getServerConn()}`),this.server)throw new Error("Socket server already started");return await this.db.waitReady,this.active=!0,this.server=m(e=>{setImmediate(()=>this.handleConnection(e))}),this.server.maxConnections=this.maxConnections,new Promise((e,i)=>{if(!this.server)return i(new Error("Server not initialized"));if(this.server.on("error",t=>{this.log("start: server error:",t),this.dispatchEvent(new CustomEvent("error",{detail:t})),this.active||i(t)}),this.path)this.server.listen(this.path,()=>{this.log(`start: server listening on ${this.getServerConn()}`),this.dispatchEvent(new CustomEvent("listening",{detail:{path:this.path}})),e()});else{let t=this.server;t.listen(this.port,this.host,()=>{let r=t.address();if(r===null||typeof r!="object")throw Error("Expected address info");this.port=r.port,this.log(`start: server listening on ${this.getServerConn()}`),this.dispatchEvent(new CustomEvent("listening",{detail:{port:this.port,host:this.host}})),e()})}})}getServerConn(){return this.path?this.path:`${this.host}:${this.port}`}async stop(){this.log("stop: stopping server"),this.active=!1,this.log(`stop: detaching ${this.handlers.size} handlers`);for(let e of this.handlers)e.detach(!0).catch(()=>{});return this.handlers.clear(),this.server?new Promise(e=>{if(!this.server)return e();this.server.close(()=>{this.log("stop: server closed"),this.server=null,this.dispatchEvent(new CustomEvent("close")),e()})}):(this.log("stop: server not running, nothing to do"),Promise.resolve())}async handleConnection(e){let i={clientAddress:e.remoteAddress||"unknown",clientPort:e.remotePort||0};if(this.log(`handleConnection: new connection from ${i.clientAddress}:${i.clientPort}`),this.log(`handleConnection: active connections: ${this.handlers.size}, queued queries: ${this.queryQueue.getQueueLength()}`),!this.active){this.log("handleConnection: server not active, closing connection");try{e.end()}catch(r){this.log("handleConnection: error closing socket:",r)}return}if(this.handlers.size>=this.maxConnections){this.log("handleConnection: max connections reached, rejecting"),e.write(Buffer.from(`Too many connections
1417 `)),e.end();return}let t=new d({queryQueue:this.queryQueue,closeOnDetach:!0,inspect:this.inspect,debug:this.debug,idleTimeout:this.idleTimeout});this.handlers.add(t),t.addEventListener("error",r=>{let o=r.detail;o?.message?.includes("ECONNRESET")?this.log(`handler #${t.handlerId}: client disconnected (ECONNRESET)`):o?.message?.includes("Idle timeout")?this.log(`handler #${t.handlerId}: idle timeout`):this.log(`handler #${t.handlerId}: error:`,o)}),t.addEventListener("close",()=>{this.log(`handler #${t.handlerId}: closed`),this.handlers.delete(t),this.log(`handleConnection: active connections: ${this.handlers.size}`)});try{await t.attach(e),this.dispatchEvent(new CustomEvent("connection",{detail:i}))}catch(r){this.log("handleConnection: error attaching socket:",r),this.handlers.delete(t),this.dispatchEvent(new CustomEvent("error",{detail:r}));try{e.end()}catch(o){this.log("handleConnection: error closing socket:",o)}}}getStats(){return{activeConnections:this.handlers.size,queuedQueries:this.queryQueue.getQueueLength(),maxConnections:this.maxConnections}}};export{b as a,d as b,u as c};
1518 //# sourceMappingURL=chunk-NSUMFCRM.js.map
1619 \ No newline at end of file

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t