History

Fix dev-db socket server slot leak and add a wedge watchdog (#1744)

Rhys Sullivan committed 2 months ago

Commit 1bbd47e1 parent 91062c23 files · +188 −18Browse files

Unified · Split

3 files
  1. apps/cloud/scripts/dev-db.ts +131 −16
  2. apps/cloud/src/db/dev-db-socket-concurrency.node.test.ts +51 −0
  3. patches/@electric-sql%2Fpglite-socket@0.1.4.patch +6 −2
apps/cloud/scripts/dev-db.ts+131−16
@@ -15,6 +15,7 @@ import { PGlite } from "@electric-sql/pglite"; · 14 unmodified lines
1515 import { PGLiteSocketServer } from "@electric-sql/pglite-socket";
1616 import { drizzle } from "drizzle-orm/pglite";
1717 import { migrate } from "drizzle-orm/pglite/migrator";
18+import postgres from "postgres";
1819
1920 const __dirname = dirname(fileURLToPath(import.meta.url));
2021 // Port + data dir default to the dev values but are env-overridable so a second
@@ -118,27 +119,32 @@ await migrate(drizzle(db), { migrationsFolder: MIGRATIONS_FOLDER }); · 97 unmodified lines
118119 // transaction or pipeline affinity they can no longer release.
119120 // src/db/dev-db-socket-concurrency.node.test.ts is the regression test for
120121 // all of the above.
121−const server = new PGLiteSocketServer({
122− db,
123− port: PORT,
124− host: "127.0.0.1",
125− maxConnections: Number(process.env.DEV_DB_MAX_CONNECTIONS ?? 1000),
126− // Backstop for pipeline affinity: a client that stalls mid-pipeline (Parse
127− // sent, no Sync) with its socket still OPEN would hold the queue's handler
128− // affinity forever and starve every other connection, since affinity only
129− // releases on detach and detach needs close/error/idle-timeout. In ms; the
130− // timer resets on every data event. The patch scopes the reap to connections
131− // actually HOLDING affinity (open pipeline or transaction): an idle-at-rest
132− // connection is the normal state of a healthy postgres.js pool held by a
133− // long-lived scope (SSE), and reaping those raced live queries into
134− // sporadic `write CONNECTION_ENDED` 500s.
135− idleTimeout: Number(process.env.DEV_DB_IDLE_TIMEOUT_MS ?? 30_000),
136−});
122+const makeServer = () =>
123+ new PGLiteSocketServer({
124+ db,
125+ port: PORT,
126+ host: "127.0.0.1",
127+ maxConnections: Number(process.env.DEV_DB_MAX_CONNECTIONS ?? 1000),
128+ // Backstop for pipeline affinity: a client that stalls mid-pipeline (Parse
129+ // sent, no Sync) with its socket still OPEN would hold the queue's handler
130+ // affinity forever and starve every other connection, since affinity only
131+ // releases on detach and detach needs close/error/idle-timeout. In ms; the
132+ // timer resets on every data event. The patch scopes the reap to connections
133+ // actually HOLDING affinity (open pipeline or transaction): an idle-at-rest
134+ // connection is the normal state of a healthy postgres.js pool held by a
135+ // long-lived scope (SSE), and reaping those raced live queries into
136+ // sporadic `write CONNECTION_ENDED` 500s.
137+ idleTimeout: Number(process.env.DEV_DB_IDLE_TIMEOUT_MS ?? 30_000),
138+ });
137139
140+let server = makeServer();
138141 await server.start();
139142 console.log(`[dev-db] Listening on postgresql://postgres:postgres@127.0.0.1:${PORT}/postgres`);
140143
144+let stopping = false;
145+
141146 const shutdown = async () => {
147+ stopping = true;
142148 console.log("\n[dev-db] Shutting down");
143149 await server.stop();
144150 await db.close();
@@ -147,3 +153,112 @@ const shutdown = async () => { · 2 unmodified lines
147153
148154 process.on("SIGINT", shutdown);
149155 process.on("SIGTERM", shutdown);
156+
157+// ---------------------------------------------------------------------------
158+// Wedge watchdog
159+// ---------------------------------------------------------------------------
160+//
161+// Twice the socket server has shipped a state machine that could stop
162+// answering NEW connections while the process, the port, and PGlite all stayed
163+// up (the CI e2e "cloud signIn: callback set no session (500)" cascades: every
164+// in-flight query dies once, then every fresh connection's startup packet
165+// times out — CONNECT_TIMEOUT — for the rest of the shard). The known paths
166+// are patched with regression tests, but each recurrence so far has found a
167+// new path, and a wedged front-end turns ONE infra hiccup into a failure of
168+// every remaining test in the shard.
169+//
170+// So: probe the server the way the app does — a fresh TCP connection, real
171+// startup handshake, `select 1` — and when several consecutive probes fail,
172+// dump the server's internals to the boot log and swap in a fresh socket
173+// server on the same PGlite instance (all state is in PGlite; the front-end is
174+// stateless, so this drops only already-doomed connections). If a restart
175+// doesn't restore service, exit non-zero: the boot supervisor logs the exit
176+// loudly and the run fails fast with an attributable cause instead of minutes
177+// of anonymous CONNECT_TIMEOUTs. The wedge itself stays visible in the
178+// server-logs artifact via the [dev-db][watchdog] lines.
179+const WATCHDOG_INTERVAL_MS = Number(process.env.DEV_DB_WATCHDOG_INTERVAL_MS ?? 5_000);
180+// 3 consecutive failures ≈ 15s+ of hard unavailability. PGlite serves queries
181+// in milliseconds; even a deep queue clears in well under one probe interval,
182+// so consecutive startup failures this sustained only happen wedged.
183+const WATCHDOG_FAILURES_TO_RESTART = 3;
184+const WATCHDOG_MAX_RESTARTS = 3;
185+
186+const probe = async (): Promise<void> => {
187+ const sql = postgres(`postgres://postgres:postgres@127.0.0.1:${PORT}/postgres`, {
188+ max: 1,
189+ idle_timeout: 0,
190+ connect_timeout: 5,
191+ fetch_types: false,
192+ prepare: false,
193+ onnotice: () => undefined,
194+ });
195+ try {
196+ // connect_timeout only bounds the handshake; race the query too so a
197+ // post-startup wedge cannot hang the watchdog itself.
198+ await Promise.race([
199+ sql.unsafe("select 1"),
200+ sleep(10_000).then(() => {
201+ throw new Error("probe query timed out after 10s");
202+ }),
203+ ]);
204+ } finally {
205+ await sql.end({ timeout: 5 }).catch(() => {});
206+ }
207+};
208+
209+const watchdog = async () => {
210+ let consecutiveFailures = 0;
211+ let restarts = 0;
212+ for (;;) {
213+ await sleep(WATCHDOG_INTERVAL_MS);
214+ if (stopping) return;
215+ try {
216+ await probe();
217+ consecutiveFailures = 0;
218+ } catch (cause) {
219+ consecutiveFailures += 1;
220+ console.error(
221+ `[dev-db][watchdog] probe failed (${consecutiveFailures}/${WATCHDOG_FAILURES_TO_RESTART}): ${String(cause)}`,
222+ );
223+ if (consecutiveFailures < WATCHDOG_FAILURES_TO_RESTART) continue;
224+ console.error(
225+ `[dev-db][watchdog] socket server wedged; stats: ${JSON.stringify(server.getStats())}`,
226+ );
227+ if (restarts >= WATCHDOG_MAX_RESTARTS) {
228+ console.error(
229+ `[dev-db][watchdog] still wedged after ${restarts} restarts — giving up so the boot supervisor reports it`,
230+ );
231+ process.exit(1);
232+ }
233+ restarts += 1;
234+ consecutiveFailures = 0;
235+ console.error(
236+ `[dev-db][watchdog] restarting socket server (${restarts}/${WATCHDOG_MAX_RESTARTS})`,
237+ );
238+ // stop() itself goes through the query queue (detach rolls back open
239+ // transactions), so a wedge deep enough can hang the restart too —
240+ // bound it and treat that as fatal rather than hanging the watchdog.
241+ const restart = async () => {
242+ await server.stop();
243+ server = makeServer();
244+ await server.start();
245+ };
246+ try {
247+ await Promise.race([
248+ restart(),
249+ sleep(15_000).then(() => {
250+ throw new Error("restart timed out after 15s");
251+ }),
252+ ]);
253+ console.error(`[dev-db][watchdog] socket server restarted`);
254+ } catch (restartCause) {
255+ console.error(
256+ `[dev-db][watchdog] restart failed (${String(restartCause)}) — exiting so the boot supervisor reports it`,
257+ );
258+ process.exit(1);
259+ }
260+ }
261+ }
262+};
263+
264+void watchdog();
apps/cloud/src/db/dev-db-socket-concurrency.node.test.ts+51−0
@@ -259,6 +259,57 @@ describe("dev-db PGlite socket under concurrent connections", () => { · 258 unmodified lines
259259 },
260260 );
261261
262+ // Regression for the reap SLOT LEAK: detach(true) removes the socket's
263+ // listeners before destroying it, so a server-initiated teardown (the idle
264+ // backstop) never fired the server's 'close' bookkeeping — the reaped
265+ // handler stayed in the server's handlers set forever, burning one
266+ // maxConnections slot per reap. Enough reaps over a long run and the server
267+ // answers every NEW connection with "Too many connections" while the
268+ // process, the port, and PGlite are all healthy — postgres.js surfaces that
269+ // as the same CONNECT_TIMEOUT cascade as the queue wedges. The server now
270+ // drops the handler when it dispatches its terminal error.
271+ it("reaped handlers release their connection slots", { timeout: 30_000 }, async () => {
272+ const port = 45993;
273+ const db = await PGlite.create();
274+ const server = new PGLiteSocketServer({
275+ db,
276+ port,
277+ host: "127.0.0.1",
278+ maxConnections: 2,
279+ idleTimeout: 250,
280+ });
281+ await server.start();
282+
283+ // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
284+ try {
285+ // Burn through more reaps than there are slots: each staller opens a
286+ // pipeline and goes silent, so the idle backstop reaps it (the server
287+ // destroys the socket — its 'close' marks that reap complete).
288+ for (let i = 0; i < 3; i++) {
289+ const staller = await openWireClient(port);
290+ staller.write(parseFrame(`select ${i + 1}`));
291+ await new Promise<void>((res) => staller.once("close", res));
292+ }
293+
294+ expect(
295+ server.getStats().activeConnections,
296+ "reaped handlers stay counted against maxConnections",
297+ ).toBe(0);
298+
299+ const sql = makeClient(port);
300+ // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
301+ try {
302+ expect((await sql.unsafe(`select 6 as six`))[0]).toEqual({ six: 6 });
303+ } finally {
304+ // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
305+ await sql.end({ timeout: 5 }).catch(() => {});
306+ }
307+ } finally {
308+ await server.stop();
309+ await db.close();
310+ }
311+ });
312+
262313 // Regression for the second wedge mode behind the same CI cascade: a client
263314 // whose socket dies WHILE its pipeline-opening entry is executing. detach()
264315 // clears pipeline affinity before the entry finishes, so the queue then
patches/@electric-sql%2Fpglite-socket@0.1.4.patch+6−2
@@ -1,3 +1,6 @@
1+diff --git a/node_modules/@electric-sql/pglite-socket/.bun-tag-9dda1ac39b3d8eab b/.bun-tag-9dda1ac39b3d8eab
2+new file mode 100644
3+index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
14 diff --git a/node_modules/@electric-sql/pglite-socket/.bun-tag-a8fabe72c1056a8f b/.bun-tag-a8fabe72c1056a8f
25 new file mode 100644
36 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
@@ -8,12 +11,13 @@ diff --git a/node_modules/@electric-sql/pglite-socket/.bun-tag-fbab1bb0bfbef953 · 4 unmodified lines
811 new file mode 100644
912 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
1013 diff --git a/dist/chunk-NSUMFCRM.js b/dist/chunk-NSUMFCRM.js
11−index 37d45ebc5150c43c919fbdb1c7fffb51b18fda8c..1cdfaaa69b568a0bed7618fe527ab2c10b1b9403 100644
14+index 37d45ebc5150c43c919fbdb1c7fffb51b18fda8c..ab5ff4de0ca78cc8a4b7905fe30f93b655473c14 100644
1215 --- a/dist/chunk-NSUMFCRM.js
1316 +++ b/dist/chunk-NSUMFCRM.js
1417 @@ -1,3 +1,3 @@
1518 -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
19+-`)),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};
1620 +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
17− `)),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};
21++`)),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);this.handlers.delete(t)}),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};
1822 //# sourceMappingURL=chunk-NSUMFCRM.js.map
1923 \ 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