Node.js với database

Phần lớn kiến thức về database không phụ thuộc ngôn ngữ: index, transaction, isolation level, query plan (MySQL, Database). Trang này nói về những điểm riêng khi gọi MySQL và PostgreSQL từ Node.js: connection pool trong một process chạy bất đồng bộ, giữ đúng một connection cho cả transaction, truyền transaction xuống các hàm bên dưới, kiểu dữ liệu mà driver trả về, và stream kết quả lớn. Ví dụ dùng pg (node-postgres) và mysql2, hai driver phổ biến nhất; ORM như Prisma, TypeORM, Sequelize, Drizzle, Knex đều dựng trên các driver này hoặc driver tương tự.

Connection pool

Mở một connection tới database tốn vài ms tới hàng chục ms (TCP, TLS, xác thực), nên ứng dụng giữ sẵn một pool connection đã mở. Mỗi query mượn một connection, chạy, rồi trả lại (MySQL chương 10).

Trong Node, một process phục vụ hàng nghìn request cùng lúc trên một thread, nhưng mỗi connection chỉ chạy được một query tại một thời điểm. Vì vậy số query chạy song song bị giới hạn bởi kích thước pool, và request vượt quá phải xếp hàng chờ connection. Trên lab, 200 query mỗi query chờ 20 ms trong database mất khoảng 4,7–4,9 giây với pool 1 connection, khoảng 0,9 giây với pool 5, khoảng 230 ms với pool 20, và khoảng 90 ms với pool 50.

Nhưng pool lớn hơn không phải lúc nào cũng tốt hơn:

Tổng connection của mọi instance phải nhỏ hơn max_connections. Postgres mặc định cho 100 connection. Trên lab, 6 instance mỗi instance pool 20 (tổng 120) thì 20 query nhận lỗi sorry, too many clients already. Autoscale từ 5 lên 20 pod là cách hay gặp để vượt giới hạn này (Scale Node.js).
Database có số core giới hạn. Với query tốn CPU hay I/O ở phía database, thêm connection quá số core chỉ làm các query tranh nhau và chậm hơn. Pool lớn chỉ có lợi khi query chủ yếu là chờ (lock, đĩa, pg_sleep như trên lab).
Có nhiều instance thì đặt pool nhỏ ở mỗi instance, và dùng connection pooler ở giữa (PgBouncer cho Postgres, ProxySQL cho MySQL) khi số instance lớn.

Mô phỏng tính kích thước pool theo tải nằm ở MySQL chương 10.

Các tuỳ chọn nên đặt:

Tuỳ chọn	pg	mysql2	Vì sao
Kích thước pool	`max` (mặc định 10)	`connectionLimit` (mặc định 10)	Theo tải và `max_connections`
Chờ connection tối đa	`connectionTimeoutMillis` (mặc định 0: chờ mãi)	`queueLimit`, `acquireTimeout` tuỳ phiên bản	Pool cạn thì báo lỗi nhanh thay vì treo request
Thời gian chạy query tối đa	`statement_timeout`, `query_timeout`	`timeout` của từng query	Một query chậm không giữ connection mãi
Đóng connection rảnh	`idleTimeoutMillis`	`idleTimeout`, `maxIdle`	Trả connection lại cho database khi tải giảm

Trả connection về pool

Code mượn connection bằng tay (pool.connect() của pg, pool.getConnection() của mysql2) thì phải trả bằng release(), và phải trả trong finally. Tài liệu node-postgres cảnh báo rằng quên release thì ứng dụng nhanh chóng dùng hết client rảnh trong pool, và mọi lời gọi sau đó sẽ chờ mãi.

const client = await pool.connect()
try {
  await client.query('SELECT …')
  if (!ok) throw new Error('lỗi nghiệp vụ')  // không có finally thì release() bên dưới không bao giờ chạy
} finally {
  client.release()
}

Trên lab, pool 3 connection, mỗi request lỗi giữa chừng mà không release: sau 3 request, pool không còn connection rảnh, và một SELECT 1 bình thường lỗi timeout exceeded when trying to connect sau 1 giây (nhờ đặt connectionTimeoutMillis; không đặt thì nó chờ mãi). Tệ hơn, pool.end() chờ mọi client đang bị mượn được trả về, nên với client bị leak nó treo mãi: graceful shutdown của process cũng treo theo, tới khi bị SIGKILL (Xử lý lỗi & graceful shutdown).

Với query đơn lẻ, dùng pool.query(): pool tự mượn và tự trả.

Transaction phải chạy trên một connection

Transaction là trạng thái của một connection: BEGIN mở transaction trên connection đó, và chỉ những lệnh gửi qua connection đó mới nằm trong transaction. Đây là cái bẫy riêng của code async với pool:

// SAI: mỗi pool.query có thể chạy trên một connection khác
await pool.query('BEGIN')
await pool.query('UPDATE accounts SET balance = balance - 10 WHERE id = 1')
await pool.query('COMMIT')

Tài liệu node-postgres nói thẳng: phải dùng cùng một client cho mọi lệnh trong transaction, và không dùng transaction với pool.query. Khi chỉ có một request, lỗi này không hiện ra, vì pool trả lại đúng connection vừa dùng. Khi có nhiều request cùng lúc, các lệnh của chúng trộn vào nhau:

UPDATE rơi vào một connection không có transaction: lệnh tự commit ngay, ROLLBACK sau đó không hoàn tác được nó.
ROLLBACK của request này đóng transaction của request khác.
Một connection trở về pool khi vẫn đang mở transaction ("idle in transaction"), giữ row lock và thay đổi chưa commit; request kế tiếp mượn nó chạy bên trong transaction của người khác.

Trên lab, 20 request cùng chạy BEGIN, trừ 1 đồng, ROLLBACK bằng pool.query: không request nào báo lỗi, nhưng số dư cuối là 97 hoặc 99 thay vì 100. Làm đúng bằng một client cho cả transaction thì số dư giữ nguyên 100. Chạy từng bước để thấy lệnh nào rơi vào connection nào:

Pool 3 connection

3 request cùng lúc

Request sau đến trễ 0,5 lệnh

Mỗi request: BEGIN; UPDATE acct SET balance = balance - 1; ROLLBACK. Số dư ban đầu 100 và phải giữ nguyên 100.

Bước 1 / 11, thời điểm 0

trong transaction

mở bởi A

không có transaction

100số dư đã commit

A: BEGIN chạy trên C1.

Mô hình đơn giản hoá: mỗi lệnh mất đúng 1 đơn vị thời gian, không có độ trễ mạng; pool lấy connection vừa được trả gần nhất như pg-pool.

Cách làm đúng:

async function withTransaction(fn) {
  const client = await pool.connect()
  try {
    await client.query('BEGIN')
    const result = await fn(client)
    await client.query('COMMIT')
    return result
  } catch (e) {
    await client.query('ROLLBACK')
    throw e
  } finally {
    client.release()
  }
}

ORM có sẵn hàm tương tự: prisma.$transaction(async (tx) => …), sequelize.transaction(async (t) => …), dataSource.transaction(async (manager) => …) của TypeORM. Cái bẫy chuyển sang chỗ khác: trong callback phải dùng đúng tx được truyền vào, dùng prisma hay repository toàn cục thì lệnh chạy ngoài transaction (Bẫy của ORM › Quên truyền transaction).

Giữ transaction ngắn: không gọi API bên ngoài, không chờ người dùng bên trong transaction. Mỗi transaction giữ một connection của pool và có thể giữ lock trong suốt thời gian đó (MySQL chương 6).

Truyền transaction bằng AsyncLocalStorage

Truyền client hay tx qua mọi tầng (controller, service, repository) làm mọi hàm phải nhận thêm một tham số, và quên truyền ở một chỗ là lệnh chạy ngoài transaction. AsyncLocalStorage của Node giữ một giá trị gắn với chuỗi lời gọi bất đồng bộ hiện tại: mọi await, setTimeout, promise bên trong als.run() đều thấy cùng một giá trị, còn request khác chạy song song thấy giá trị của riêng nó.

import { AsyncLocalStorage } from 'node:async_hooks'
const als = new AsyncLocalStorage()
const db = () => als.getStore() ?? pool   // trong transaction thì dùng client của nó

async function withTransaction(fn) {
  const client = await pool.connect()
  try {
    await client.query('BEGIN')
    const r = await als.run(client, fn)    // mọi lời gọi db() bên trong fn dùng client này
    await client.query('COMMIT')
    return r
  } catch (e) { await client.query('ROLLBACK'); throw e } finally { client.release() }
}

// repository không cần nhận tham số connection
const debit = (id, amount) => db().query('UPDATE accounts SET balance = balance - $1 WHERE id = $2', [amount, id])
await withTransaction(async () => { await debit(1, 10); await credit(2, 10) })

Trên lab, hai transaction chạy song song, mỗi transaction gọi ba hàm có await và setTimeout xen giữa: ba hàm trong cùng một transaction luôn chạy trên cùng một connection (cùng pg_backend_pid()), hai transaction trên hai connection khác nhau, và transaction lỗi giữa chừng được rollback đúng. Thư viện như nestjs-cls, typeorm-transactional, và các extension của Prisma dùng đúng cơ chế này. AsyncLocalStorage cũng là cách gắn request id vào mọi dòng log của một request.

Kiểu dữ liệu: BIGINT và DECIMAL

Số của JavaScript là số thực 64 bit, chỉ biểu diễn chính xác số nguyên tới 2^53 − 1 (9.007.199.254.740.991) và không biểu diễn chính xác được số thập phân như 0,1 (Kiểu dữ liệu). Trong khi đó BIGINT của database tới 2^63 − 1, còn DECIMAL/NUMERIC là số thập phân chính xác. Mỗi driver chọn một cách khác nhau:

Kiểu	pg (mặc định)	mysql2 (mặc định)
`INT`	`number`	`number`
`BIGINT`, kể cả `COUNT(*)` của Postgres	chuỗi	`number`, mất chính xác khi vượt 2^53
`DECIMAL` / `NUMERIC`	chuỗi	chuỗi

Trên lab, 9007199254740993 qua pg là chuỗi "9007199254740993" (đúng), còn qua mysql2 mặc định là số 9007199254740992: sai một đơn vị mà không có lỗi hay cảnh báo nào. Với mysql2, bật supportBigNumbers: true cùng bigNumberStrings: true để nhận chuỗi.

Bẫy ngược lại ở pg: COUNT(*) trả kiểu bigint, nên là chuỗi. Trên lab, n + 1 với n là kết quả COUNT(*) bằng 2 ra "21", không phải 3. Cách xử lý: ép kiểu trong SQL (COUNT(*)::int) khi biết chắc giá trị nhỏ, hoặc Number() trong code.

Với tiền, đừng đổi DECIMAL thành number để tính toán. mysql2 có decimalNumbers: true (trên lab, 12.30 thành 12.3), nhưng phép cộng số thực sẽ sai ở số lẻ. Giữ dạng chuỗi rồi dùng thư viện số thập phân (decimal.js, big.js), hoặc lưu số nguyên theo đơn vị nhỏ nhất (đồng, xu) và để database làm phép tính (MySQL chương 2, Sổ cái).

Placeholder, không ghép chuỗi

Mọi giá trị từ người dùng phải đi qua placeholder, không ghép vào chuỗi SQL (Bảo mật database › SQL injection):

await pool.query('SELECT * FROM users WHERE email = $1', [email])           // pg
await conn.execute('SELECT * FROM users WHERE email = ?', [email])          // mysql2

Với mysql2, execute() dùng prepared statement thật ở phía server, còn query() với placeholder thì driver tự escape giá trị rồi ghép vào chuỗi ở phía client. Cả hai đều an toàn với giá trị, nhưng không cái nào nhận được tên cột hay tên bảng làm placeholder: những chỗ đó phải kiểm tra theo một danh sách cho phép (Những chỗ không dùng được placeholder). Template literal của ORM như prisma.$queryRaw`… ${email}` vẫn dùng placeholder, còn $queryRawUnsafe thì không.

Stream kết quả lớn

await pool.query('SELECT * FROM big') nạp toàn bộ kết quả vào một mảng trong bộ nhớ trước khi trả về. Với vài triệu dòng, đó là vài trăm MB tới vài GB heap, và process có thể chết vì hết bộ nhớ (Bộ nhớ & GC). Đọc kết quả dạng stream thì bộ nhớ không đổi theo số dòng:

// mysql2: dùng connection kiểu callback, .stream() trả về một Readable
for await (const row of conn.query('SELECT * FROM orders').stream()) {
  await handle(row)
}
// pg: thư viện pg-query-stream dùng cursor của Postgres

Trên lab, đọc 500.000 dòng từ MySQL bằng stream làm RSS tăng khoảng 11–13 MB, còn nạp hết vào mảng thì tăng 128–162 MB; thời gian gần như bằng nhau. Kết hợp với pipeline() để stream thẳng ra file hay HTTP response (Stream & backpressure), hoặc đọc theo lô bằng khoá chính khi cần xử lý từng phần (Xử lý dữ liệu lớn theo lô).

ORM và N+1

ORM tiết kiệm code nhưng giấu câu SQL thật. Lỗi hay gặp nhất là N+1 query: lấy 50 đơn hàng, rồi trong vòng lặp lấy khách hàng của từng đơn, thành 51 query. Với Node, một biến thể khác là Promise.all(orders.map((o) => getCustomer(o.customerId))): 50 query chạy song song, chiếm hết pool. Cách sửa và các bẫy khác của ORM ở Bẫy của ORM. Bật log SQL của ORM ở môi trường dev để thấy số query mỗi request.

Ghi nhớ

Mỗi connection chỉ chạy một query một lúc: pool giới hạn số query song song; tổng pool của mọi instance phải dưới max_connections. Đặt connectionTimeoutMillis và statement_timeout.
Mượn connection bằng tay thì release() trong finally; client bị leak làm pool cạn và pool.end() treo mãi.
Transaction phải chạy trên một connection: không dùng pool.query cho BEGIN/COMMIT; dùng withTransaction hay hàm transaction của ORM, và dùng đúng tx bên trong.
AsyncLocalStorage truyền transaction (hay request id) xuống các hàm bên dưới mà không cần thêm tham số.
pg trả BIGINT (kể cả COUNT(*)) và NUMERIC dạng chuỗi; mysql2 mặc định trả BIGINT dạng số và mất chính xác quá 2^53. Stream kết quả lớn thay vì nạp hết.

Nguồn tham khảo

Đã đối chiếu (đọc nguồn ngày 2026-10-26):

Nguồn	Dùng cho phần
node-postgres, Pooling	Phải release client; quên release thì pool cạn và mọi lời gọi sau đó chờ
node-postgres, Transactions	Dùng cùng một client cho mọi lệnh trong transaction; không dùng transaction với `pool.query`
mysqljs/mysql, Readme: Connection options	`supportBigNumbers`, `bigNumberStrings` (mysql2 dùng cùng tên tuỳ chọn)

Kiểm bằng lab: transaction bằng pool.query so với một client, pool cạn khi quên release, pool.end() treo, thời gian theo kích thước pool, vượt max_connections, kiểu trả về của pg và mysql2 với ba cấu hình, AsyncLocalStorage, stream so với nạp hết. Từ kiến thức chung, chưa đối chiếu lần này: giá trị mặc định của các tuỳ chọn pool, execute() so với query() của mysql2, pg-query-stream, tên hàm transaction của các ORM, nestjs-cls và typeorm-transactional.

Số đo chi tiết

Kết quả chạy lab/nodejs-database/run.mjs (PostgreSQL 17, MySQL 8.0 trong Docker, pg 8.23, mysql2 3.24, Node 24.14, hai lượt)

Phép thử	Kết quả
20 request `BEGIN`, trừ 1, `ROLLBACK` bằng `pool.query` (pool 10)	Không lỗi; số dư 99 / 97 thay vì 100
Cùng việc đó, một client cho cả transaction	Số dư 100
Pool 3, 3 request lỗi không release	`totalCount 3, idleCount 0`; request thứ 4 lỗi `timeout exceeded when trying to connect` sau 1.001–1.002 ms
`pool.end()` khi còn client bị leak	Vẫn treo sau 2 giây
200 × `pg_sleep(0.02)`, pool 1 / 5 / 20 / 50	4.695–4.872 ms / 912–969 ms / 227–236 ms / 91 ms
6 pool × 20, `max_connections = 100`	100/120 query thành công, còn lại `sorry, too many clients already`
pg: `BIGINT`, `NUMERIC(10,2)`, `COUNT(*)`, `INT`	`"9007199254740993"`, `"12.30"`, `"2"` (`n + 1 = "21"`), `2`
mysql2 mặc định: `BIGINT`, `DECIMAL`, `COUNT(*)`	`9007199254740992` (sai), `"12.30"`, `2`
mysql2 `supportBigNumbers` + `bigNumberStrings`	`"9007199254740993"`, `"12.30"`, `"2"`
mysql2 `decimalNumbers: true`	`9007199254740992`, `12.3`, `2`
`AsyncLocalStorage`, hai transaction song song	Mỗi transaction một `pg_backend_pid()` cho cả ba hàm; transaction lỗi được rollback
500.000 dòng MySQL: stream / nạp hết	RSS tăng 11–13 MB / 128–162 MB; 345–375 ms / 327–336 ms

Code đã dùng để chạy thử

Cần MySQL và PostgreSQL của lab (npm run up trong lab/, hoặc docker compose up -d --wait mysql postgres). Chạy bằng npm run lab -- nodejs-database, khoảng 20 giây. Lab tạo database nodedb và xoá bảng 500.000 dòng khi xong. Xem Lab.

lab/nodejs-database/run.mjs

// Trang nodejs/database. Cần MySQL và PostgreSQL của lab: npm run up (hoặc docker compose up -d --wait mysql postgres).
//   1. Transaction bằng pool.query (BEGIN, UPDATE, ROLLBACK có thể chạy trên ba connection khác nhau) so với một client
//   2. Quên release() connection khi có lỗi: pool cạn
//   3. Kích thước pool: thời gian chạy 200 query chờ 20 ms; nhiều pool cộng lại vượt max_connections
//   4. Kiểu dữ liệu: BIGINT, DECIMAL, COUNT(*) trả về dạng gì với pg và mysql2
//   5. AsyncLocalStorage: các hàm bên dưới tự dùng connection của transaction đang chạy
//   6. Stream 500.000 dòng từ MySQL so với nạp hết vào bộ nhớ
import pg from 'pg'
import mysql from 'mysql2/promise'
import mysqlCb from 'mysql2'
import { AsyncLocalStorage } from 'node:async_hooks'
import { MYSQL, PG, sleep } from '../common/config.mjs'

const log = (label, v) => console.log(`${label.padEnd(58)} ${v}`)
const DB = 'nodedb'
{
  const c = new pg.Client(PG); await c.connect()
  const { rowCount } = await c.query('SELECT 1 FROM pg_database WHERE datname = $1', [DB])
  if (!rowCount) await c.query(`CREATE DATABASE ${DB}`)
  await c.end()
}
const PGDB = { ...PG, database: DB }
const reset = async (p) => p.query('DROP TABLE IF EXISTS acct; CREATE TABLE acct (id int PRIMARY KEY, balance int NOT NULL); INSERT INTO acct VALUES (1, 100), (2, 100)')
const balance = async (p) => (await p.query('SELECT balance FROM acct WHERE id = 1')).rows[0].balance

console.log('\n=== 1. Transaction: 20 request cùng lúc, mỗi request BEGIN, trừ 1 đồng, ROLLBACK (số dư phải giữ nguyên 100) ===')
{
  const pool = new pg.Pool({ ...PGDB, max: 10, statement_timeout: 2000 })
  await reset(pool)
  const t = performance.now()
  const rs = await Promise.allSettled(Array.from({ length: 20 }, async () => {
    await pool.query('BEGIN')
    await pool.query('UPDATE acct SET balance = balance - 1 WHERE id = 1')
    await pool.query('ROLLBACK')
  }))
  const errs = rs.filter((r) => r.status === 'rejected').map((r) => r.reason.message)
  const watch = new pg.Client(PGDB); await watch.connect()
  const { rows: [st] } = await watch.query(`SELECT count(*) FILTER (WHERE state = 'idle in transaction') AS idle_tx FROM pg_stat_activity WHERE datname = $1`, [DB])
  const { rows: [b] } = await watch.query('SELECT balance FROM acct WHERE id = 1')
  log('pool.query cho từng lệnh', `${Math.round(performance.now() - t)} ms; ${errs.length} request lỗi${errs.length ? ` (${[...new Set(errs)].join('; ')})` : ''}; số dư thấy từ connection khác: ${b.balance}; ${st.idle_tx} connection kẹt "idle in transaction"`)
  await watch.end()
  await pool.end()
  const pool2 = new pg.Pool({ ...PGDB, max: 10 })
  const after = await balance(pool2)
  log('Sau khi đóng pool (Postgres rollback các transaction dở)', `số dư ${after}`)
  await reset(pool2)
  await Promise.all(Array.from({ length: 20 }, async () => {
    const client = await pool2.connect()
    try {
      await client.query('BEGIN')
      await client.query('UPDATE acct SET balance = balance - 1 WHERE id = 1')
      await client.query('ROLLBACK')
    } catch (e) { await client.query('ROLLBACK'); throw e } finally { client.release() }
  }))
  log('pool.connect(), cả transaction trên một client, release trong finally', `số dư ${await balance(pool2)}`)
  await pool2.end()
}

console.log('\n=== 2. Quên release() khi có lỗi: pool 3 connection, connectionTimeoutMillis 1000 ===')
{
  const pool = new pg.Pool({ ...PGDB, max: 3, connectionTimeoutMillis: 1000 })
  const leaked = [] // chỉ để lab tự dọn ở cuối; code thật bị leak thì không còn tham chiếu nào
  const leaky = async () => {
    const client = await pool.connect()
    leaked.push(client)
    await client.query('SELECT 1')
    throw new Error('lỗi nghiệp vụ')   // client.release() nằm sau dòng này nên không bao giờ chạy
  }
  for (let i = 0; i < 3; i++) await leaky().catch(() => {})
  log('Sau 3 request lỗi', `totalCount ${pool.totalCount}, idleCount ${pool.idleCount}`)
  const t = performance.now()
  const r = await pool.query('SELECT 1').then(() => 'thành công', (e) => `lỗi "${e.message}"`)
  log('Request thứ 4, một SELECT 1 bình thường', `${r} sau ${Math.round(performance.now() - t)} ms`)
  // pool.end() chờ mọi client đang bị mượn được trả về: client bị leak làm nó treo mãi (graceful shutdown treo theo)
  const t2 = performance.now()
  const ended = pool.end()
  const r2 = await Promise.race([ended.then(() => 'xong'), sleep(2000).then(() => 'vẫn treo')])
  log('pool.end() khi còn client chưa release', `${r2} sau ${Math.round(performance.now() - t2)} ms`)
  for (const c of leaked) c.release(true) // release(true): huỷ hẳn connection thay vì trả về pool
  await ended
}

console.log('\n=== 3. Kích thước pool: 200 query, mỗi query chờ 20 ms trong database (pg_sleep) ===')
{
  for (const max of [1, 5, 20, 50]) {
    const pool = new pg.Pool({ ...PGDB, max })
    await Promise.all(Array.from({ length: max }, () => pool.query('SELECT 1'))) // mở sẵn connection
    const t = performance.now()
    await Promise.all(Array.from({ length: 200 }, () => pool.query('SELECT pg_sleep(0.02)')))
    log(`Pool ${max} connection`, `${Math.round(performance.now() - t)} ms`)
    await pool.end()
  }
  const c = new pg.Client(PGDB); await c.connect()
  const max = (await c.query('SHOW max_connections')).rows[0].max_connections
  await c.end()
  const pools = Array.from({ length: 6 }, () => new pg.Pool({ ...PGDB, max: 20 }))
  const rs = await Promise.allSettled(pools.flatMap((p) => Array.from({ length: 20 }, () => p.query('SELECT pg_sleep(0.5)'))))
  const errs = rs.filter((r) => r.status === 'rejected')
  log(`6 instance, mỗi instance pool 20 (max_connections = ${max})`, `${rs.length - errs.length}/${rs.length} query thành công; lỗi: ${[...new Set(errs.map((e) => e.reason.message))].join('; ') || 'không'}`)
  await Promise.all(pools.map((p) => p.end()))
}

console.log('\n=== 4. Kiểu dữ liệu trả về ===')
{
  const show = (v) => `${JSON.stringify(v)} (${typeof v})`
  const c = new pg.Client(PGDB); await c.connect()
  await reset(c)
  const { rows: [r] } = await c.query('SELECT 9007199254740993::bigint AS big, 12.30::numeric(10,2) AS dec, (SELECT count(*) FROM acct) AS n, 2::int AS i')
  log('pg: BIGINT 9007199254740993', show(r.big))
  log('pg: NUMERIC(10,2) 12.30', show(r.dec))
  log('pg: COUNT(*) (kiểu bigint), rồi n + 1', `${show(r.n)}, n + 1 = ${JSON.stringify(r.n + 1)}`)
  log('pg: INT', show(r.i))
  await c.end()
  const sql = 'SELECT CAST(9007199254740993 AS SIGNED) AS big, CAST(12.30 AS DECIMAL(10,2)) AS `dec`, COUNT(*) AS n FROM (SELECT 1 UNION ALL SELECT 2) t'
  for (const [label, opts] of [['mysql2 mặc định', {}], ['mysql2 supportBigNumbers + bigNumberStrings', { supportBigNumbers: true, bigNumberStrings: true }], ['mysql2 decimalNumbers: true', { decimalNumbers: true }]]) {
    const m = await mysql.createConnection({ ...MYSQL, ...opts })
    const [[x]] = await m.query(sql)
    log(`${label}: BIGINT, DECIMAL, COUNT(*)`, `${show(x.big)}, ${show(x.dec)}, ${show(x.n)}`)
    await m.end()
  }
}

console.log('\n=== 5. AsyncLocalStorage: truyền transaction xuống các hàm bên dưới ===')
{
  const pool = new pg.Pool({ ...PGDB, max: 5 })
  const als = new AsyncLocalStorage()
  const db = () => als.getStore() ?? pool              // trong transaction thì dùng client của nó, không thì pool
  async function withTransaction(fn) {
    const client = await pool.connect()
    try {
      await client.query('BEGIN')
      const r = await als.run(client, fn)
      await client.query('COMMIT')
      return r
    } catch (e) { await client.query('ROLLBACK'); throw e } finally { client.release() }
  }
  // Các hàm "repository" không nhận tham số connection nào
  const pid = async () => (await db().query('SELECT pg_backend_pid() AS p')).rows[0].p
  const debit = async () => { await sleep(5); await db().query('UPDATE acct SET balance = balance - 10 WHERE id = 1'); return pid() }
  const credit = async () => { await new Promise((r) => setTimeout(r, 3)); await db().query('UPDATE acct SET balance = balance + 10 WHERE id = 2'); return pid() }
  await reset(pool)
  const [a, b] = await Promise.all([1, 2].map(() => withTransaction(async () => [await pid(), await debit(), await credit()])))
  log('Hai transaction chạy song song, pid ở 3 hàm trong mỗi transaction', `[${a.join(', ')}] và [${b.join(', ')}]`)
  const fail = await withTransaction(async () => { await debit(); throw new Error('lỗi giữa chừng') }).catch((e) => e.message)
  log('Transaction lỗi sau debit(): số dư id 1 (trước đó 80)', `${await balance(pool)} (${fail}, đã rollback)`)
  log('Ngoài transaction, db() là pool', `${db() === pool}`)
  await pool.end()
}

console.log('\n=== 6. 500.000 dòng từ MySQL: nạp hết so với stream ===')
{
  const setup = await mysql.createConnection({ ...MYSQL, multipleStatements: true })
  await setup.query(`CREATE DATABASE IF NOT EXISTS ${DB}; USE ${DB}; DROP TABLE IF EXISTS rows500k;
    CREATE TABLE rows500k (id INT PRIMARY KEY, email VARCHAR(50), note VARCHAR(100));
    SET SESSION cte_max_recursion_depth = 1000000;
    INSERT INTO rows500k WITH RECURSIVE s AS (SELECT 1 AS n UNION ALL SELECT n + 1 FROM s WHERE n < 500000)
      SELECT n, CONCAT('user', n, '@shop.vn'), REPEAT('x', 80) FROM s;`)
  await setup.end()
  const peakDuring = async (fn) => {
    global.gc?.()
    const base = process.memoryUsage().rss
    let peak = base
    const iv = setInterval(() => (peak = Math.max(peak, process.memoryUsage().rss)), 5)
    const t = performance.now()
    const n = await fn()
    clearInterval(iv)
    peak = Math.max(peak, process.memoryUsage().rss)
    return `${n.toLocaleString('vi-VN')} dòng, RSS tăng tối đa ${Math.round((peak - base) / 1e6)} MB, ${Math.round(performance.now() - t)} ms`
  }
  const sql = `SELECT * FROM ${DB}.rows500k`
  log('query(...).stream() rồi for await', await peakDuring(async () => {
    const c = mysqlCb.createConnection(MYSQL)
    let n = 0
    for await (const row of c.query(sql).stream({ highWaterMark: 100 })) if (row.id) n++
    await new Promise((r) => c.end(r))
    return n
  }))
  log('await query(...) nạp hết vào mảng', await peakDuring(async () => {
    const c = await mysql.createConnection(MYSQL)
    const [rows] = await c.query(sql)
    const n = rows.length
    await c.end()
    return n
  }))
  const drop = await mysql.createConnection(MYSQL)
  await drop.query(`DROP TABLE ${DB}.rows500k`)
  await drop.end()
}

Node.js với database ​

Connection pool ​

Trả connection về pool ​

Transaction phải chạy trên một connection ​

Truyền transaction bằng AsyncLocalStorage ​

Kiểu dữ liệu: BIGINT và DECIMAL ​

Placeholder, không ghép chuỗi ​

Stream kết quả lớn ​

ORM và N+1 ​

Nguồn tham khảo ​

Số đo chi tiết ​

Code đã dùng để chạy thử ​