Support connect/reconnect

[ITsPro-CN]

Environment

Node.js: 22.14.0
Nuxt: 4.1.3

Reproduction

• nuxt.config.ts

export default defineNuxtConfig({
    nitro: {
    	experimental: {
    		database: true,
    	},
    	database: {
    		default: {
    			connector: "postgresql",
    			options: {
    				host: "container-name",
    				port: 5432,
    				user: "...",
    				password: "...",
    				database: "..."
    			},
    		}
    	}
  }
})

• /server/api/index.ts

export default defineEventHandler(async (event) => {
	const db = useDatabase();
	const {rows} = await db.sql`...`;
	return rows;
});

Describe the bug

I use Nitro's experimental features in Nuxt4 and often get errors a few hours after the service starts.

H3Error: Client has encountered a connection error and is not queryable
at /app/.output/server/node_modules/.pnpm/pg@8.16.3/node_modules/pg/lib/client.js:545:17
  ... 5 lines matching cause stack trace ...
  at async Server.toNodeHandle (file:///app/.output/server/chunks/nitro/nitro.mjs:1934:7) {
cause: Error: Client has encountered a connection error and is not queryable
  at /app/.output/server/node_modules/.pnpm/pg@8.16.3/node_modules/pg/lib/client.js:545:17
  at process.processTicksAndRejections (node:internal/process/task_queues:105:5)
  at async StatementWrapper.all (file:///app/.output/server/chunks/nitro/nitro.mjs:5436:15)
  at async Object.sql (file:///app/.output/server/chunks/nitro/nitro.mjs:5338:18)
  at async Object.handler (file:///app/.output/server/chunks/routes/api/index.get.mjs:16:28)
  at async Object.handler (file:///app/.output/server/chunks/nitro/nitro.mjs:1663:19)
  at async Server.toNodeHandle (file:///app/.output/server/chunks/nitro/nitro.mjs:1934:7),

Additional context

No response

Logs

[ITsPro-CN]

In Nuxt4, each time I use useDatabase() returns the same instance, and I cannot find an elegant way to auto-reconnect.

Tried

const db = useDatabase();
const client = await db.getInstance();
client.on("error", async (err) => {
	console.log("An error occurred with client =>", err);
	db.dispose();
});

End in This database instance has been disposed and cannot be used.

Or

const db = useDatabase();
const client = await db.getInstance();
client.on("error", async (err) => {
	console.log("An error occurred with client =>", err);
	await client.end();
});

Also end in Error: Client has encountered a connection error and is not queryable

[pi0]

Thanks for issue. We should introduce a way to allow reconnecting instances indeed.

[ITsPro-CN]

Thanks for issue. We should introduce a way to allow reconnecting instances indeed.

Thanks a lot. I will use a proxy for now. Looking forward to the update.

[systems-center]

Thanks for issue. We should introduce a way to allow reconnecting instances indeed.

this will also help me a lot
see:#175

thanks

[ITsPro-CN]

Thanks for issue. We should introduce a way to allow reconnecting instances indeed.

this will also help me a lot see:#175

thanks

For Nuxt, simply overwrite the useDatabase function.

/server/utils/db/db.ts

import {Connector, createDatabase} from "db0";
import mysql2Connector from "./connectors/mysql2";

const connectors: Record<string, (opts: any) => Connector> = {
	mysql2: mysql2Connector
};

const connectionConfigs: Record<string, any> = useRuntimeConfig().db;
const instances = /* @__PURE__ */ Object.create(null);
export const useDatabase = (name = "default") => {
	if (instances[name]) return instances[name];
	if (!connectionConfigs[name]) throw new Error(`Database connection "${name}" not configured.`);
	if (!connectionConfigs[name].connector || !connectors[connectionConfigs[name].connector]) throw new Error(`Database connection "${name}" cannot be created, connector "${connectionConfigs[name].connector}" not found`);
	const connector = connectors[connectionConfigs[name].connector];
	return (instances[name] = createDatabase(connector(connectionConfigs[name].options || {})));
};

/server/utils/db/connectors/mysql2.ts

import mysql from "mysql2/promise";
import type {Connector, Primitive} from "db0";
import {BoundableStatement} from "./statement";

type ConnectorOptions = mysql.ConnectionOptions;

type InternalQuery = (sql: string, params?: unknown[]) => Promise<mysql.QueryResult>;

export default function mysql2Connector(opts: ConnectorOptions): Connector<mysql.Connection> {
	let _connection: mysql.Connection | undefined;
	const getConnection = async () => {
		if (_connection) {
			return _connection;
		}
		console.log("NEW");
		_connection = await mysql.createConnection({
			...opts,
		});
		_connection.on("error", (err) => {
			console.log("Reconnect");
			dispose();
			getConnection();
		});
		return _connection;
	};

	const dispose = async () => {
		await _connection?.end?.();
		_connection = undefined;
	};

	const query: InternalQuery = (sql, params) =>
		getConnection()
			.then((c) => c.query(sql, params))
			.then((res) => res[0]);

	return {
		name: "mysql2",
		dialect: "mysql",
		getInstance: () => getConnection(),
		exec: (sql) => query(sql),
		prepare: (sql) => new StatementWrapper(sql, query),
		dispose,
	};
}

class StatementWrapper extends BoundableStatement<void> {
	#query: InternalQuery;
	#sql: string;

	constructor(sql: string, query: InternalQuery) {
		super();
		this.#sql = sql;
		this.#query = query;
	}

	async all(...params: Primitive[]) {
		const res = (await this.#query(this.#sql, params)) as mysql.RowDataPacket[];
		return res;
	}

	async run(...params: Primitive[]) {
		const res = (await this.#query(this.#sql, params)) as mysql.RowDataPacket[];
		return {
			success: true,
			...res,
		};
	}

	async get(...params: Primitive[]) {
		const res = (await this.#query(this.#sql, params)) as mysql.RowDataPacket[];
		return res[0];
	}
}

/server/utils/db/connectors/statement.ts

import type {Primitive, Statement, PreparedStatement} from "db0";
export abstract class BoundableStatement<T> implements Statement {
	_statement: T;
	constructor(rawStmt: T) {
		this._statement = rawStmt;
	}
	bind(...params: Primitive[]): PreparedStatement {
		return new BoundStatement(this, params);
	}
	abstract all(...params: Primitive[]): Promise<unknown[]>;
	abstract run(...params: Primitive[]): Promise<{success: boolean}>;
	abstract get(...params: Primitive[]): Promise<unknown>;
}
class BoundStatement<S extends Statement> implements PreparedStatement {
	#statement: S;
	#params: Primitive[];
	constructor(statement: S, params: Primitive[]) {
		this.#statement = statement;
		this.#params = params;
	}
	bind(...params: Primitive[]): BoundStatement<S> {
		return new BoundStatement(this.#statement, params);
	}
	all(): Promise<unknown[]> {
		return this.#statement.all(...this.#params);
	}
	run(): Promise<{success: boolean}> {
		return this.#statement.run(...this.#params);
	}
	get(): Promise<unknown> {
		return this.#statement.get(...this.#params);
	}
}

And replace the Nitro experimental Configuration with:

runtimeConfig: {
	db: {
		default: {
			connector: "mysql2",
			options: {
				host: process.env.MYSQL_HOST,
				port: Number(process.env.MYSQL_PORT),
				database: process.env.MYSQL_NAME,
				user: process.env.MYSQL_USER,
				password: process.env.MYSQL_PASSWORD,
			},
		},
	},
},

Also, there is a pool version:

import mysql from "mysql2/promise";
import type {Connector, Primitive} from "db0";
import {BoundableStatement} from "./statement";

type ConnectorOptions = mysql.PoolOptions;

type InternalQuery = (sql: string, params?: unknown[]) => Promise<mysql.QueryResult>;

export default function mysql2PoolConnector(opts: ConnectorOptions): Connector<mysql.Connection> {
	let _pool: mysql.Pool | undefined;
	const getConnection = async () => {
		if (!_pool) {
			_pool = mysql.createPool({
				...opts,
			});
		}
		return await _pool.getConnection();
	};
	const query: InternalQuery = async (sql, params) => {
		const connection = await getConnection();
		try {
			const result = await connection.query(sql, params);
			return result[0];
		} finally {
			connection.release();
		}
	};
	return {
		name: "mysql2pool",
		dialect: "mysql",
		getInstance: () => getConnection(),
		exec: (sql) => query(sql),
		prepare: (sql) => new StatementWrapper(sql, query),
		dispose: async () => {
			await _pool?.end?.();
			_pool = undefined;
		},
	};
}

class StatementWrapper extends BoundableStatement<void> {
	#query: InternalQuery;
	#sql: string;

	constructor(sql: string, query: InternalQuery) {
		super();
		this.#sql = sql;
		this.#query = query;
	}

	async all(...params: Primitive[]) {
		const res = (await this.#query(this.#sql, params)) as mysql.RowDataPacket[];
		return res;
	}

	async run(...params: Primitive[]) {
		const res = (await this.#query(this.#sql, params)) as mysql.RowDataPacket[];
		return {
			success: true,
			...res,
		};
	}

	async get(...params: Primitive[]) {
		const res = (await this.#query(this.#sql, params)) as mysql.RowDataPacket[];
		return res[0];
	}
}

[systems-center]

Thanks for issue. We should introduce a way to allow reconnecting instances indeed.

this will also help me a lot see:#175
thanks

For Nuxt, simply overwrite the useDatabase function.

/server/utils/db/db.ts

Thank you very much
I will try soon

By the way,
the MYSQL2 package has released v3.16.0 in which they added a state getter to track connection lifecycle status

[ITsPro-CN]

@systems-center Thanks for the reminder. I’ll check and see what I can do.
I recommend adopting the pool version, as it resolves the 'not queryable' issue simply by setting maxIdle: 0. And be cautious of the limitation on transactions with the pool connection.
I’ve submitted PR #203, which adds pool connectors for both mysql2and pg-nodejs. The complete connector code changes are available.