Race condition trong Node.js

"Node.js chạy một thread nên không có race condition" là một hiểu lầm phổ biến. Đúng là hai đoạn JavaScript không bao giờ chạy cùng một lúc trong một process. Nhưng một thao tác nghiệp vụ hiếm khi là một đoạn code đồng bộ liền mạch: nó đọc database, chờ, rồi mới ghi. Trong lúc chờ, request khác chạy. Trang này giải thích chỗ nào trong code Node có thể bị chen ngang, cách chặn trong một process và giữa nhiều process, cách giới hạn số việc chạy song song, và cách gom các request trùng nhau.

Mỗi await là một chỗ có thể bị chen ngang

Event loop của Node là cooperative: một đoạn code đồng bộ chạy liền một mạch tới khi xong hoặc gặp await (Process, thread). Vì vậy:

• Code đồng bộ không bị chen ngang. counter++ trên một biến trong bộ nhớ là an toàn: không request nào chạy xen vào giữa lúc đọc và lúc ghi.
• Mỗi await là một điểm mà request khác có thể chạy xen vào. Đọc một giá trị, await một việc gì đó, rồi ghi lại dựa trên giá trị đã đọc, thì giá trị đó có thể đã cũ.

async function withdraw(userId, amount) {
  const balance = await getBalance(userId)  // request khác chạy ở đây
  if (balance < amount) throw new Error('Không đủ tiền')
  await setBalance(userId, balance - amount) // ghi đè lên lần trừ của request khác
}

Trên lab, tài khoản có 100 đồng và 20 request cùng rút 10 đồng. Lẽ ra chỉ 10 lần thành công. Thực tế cả 20 lần đều thành công trong mọi lượt chạy, và số dư cuối vẫn là 80 hay 90 đồng: hai mẫu lỗi check-then-act và lost update xảy ra cùng lúc (Chống trùng & race condition). Mô phỏng cùng mẫu lỗi đó, mỗi await là một chỗ có thể đổi sang request kia:

Thử: hai request chạy xen kẽ

Cách viết:

Request A

Bước tiếp theo:

SELECT stock FROM products WHERE id = 1

Request B

Bước tiếp theo:

SELECT stock FROM products WHERE id = 1

products.stock1

orderstrống

Gợi ý: bấm A, B, A, B… để hai request chen nhau. Với cách viết có lỗi, chạy lần lượt thì đúng nhưng chạy xen kẽ thì sai; đó là lý do race condition hiếm khi lộ ra khi test bằng tay. Ở cách FOR UPDATE, B bị chặn ở bước đầu cho tới khi A COMMIT.

Không chỉ database. Bất kỳ thao tác nào đọc trạng thái, await, rồi ghi lại đều có thể bị như vậy: cập nhật một object cấu hình trong bộ nhớ sau khi tải dữ liệu, kiểm tra một file có chưa rồi mới tạo, nạp cache khi miss.

Mutex trong process chỉ đủ khi chỉ có một process

Một mutex trong process (tự viết bằng chuỗi promise, hoặc thư viện async-mutex) bắt các thao tác chờ nhau, nên đoạn đọc rồi ghi chạy tuần tự:

function createMutex() {
  let tail = Promise.resolve()
  return async (fn) => {
    const prev = tail
    let release
    tail = new Promise((r) => (release = r))
    await prev                      // chờ người trước xong
    try { return await fn() } finally { release() }
  }
}
const lock = createMutex()
await lock(() => withdraw(userId, 10))

Trên lab, có mutex thì đúng 10 lần rút thành công, số dư cuối là 0. Nhưng production gần như luôn chạy nhiều process: nhiều instance sau load balancer, cluster, hay nhiều pod (Scale Node.js). Mỗi process có mutex riêng, không biết gì về mutex của process kia. Với hai process, mỗi process có mutex của mình, lab ghi nhận 16 tới 20 lần rút thành công, và số dư cuối vẫn là 0: một lỗi còn khó thấy hơn, vì số dư trông hợp lý trong khi tiền đã bị rút quá.

Vì vậy với dữ liệu dùng chung giữa các process, chốt chặn phải nằm ở nơi lưu dữ liệu:

• Ghi có điều kiện trong một câu lệnh: UPDATE accounts SET balance = balance - ? WHERE id = ? AND balance >= ?, rồi kiểm tra số dòng bị ảnh hưởng. Trên lab, cách này cho đúng 10 lần thành công với cả hai process.
• Ràng buộc (UNIQUE, CHECK), lock dòng (SELECT … FOR UPDATE), optimistic lock bằng cột version (Chống trùng & race condition).
• Distributed lock cho việc không nằm trong database (Distributed lock).

Mutex trong process vẫn có chỗ dùng: bảo vệ tài nguyên chỉ thuộc về process đó, như một file cục bộ, một object cấu hình trong bộ nhớ, hay để chỉ một lần làm mới token chạy tại một thời điểm.

Giới hạn số việc chạy song song

Ngược với race condition là chạy quá nhiều việc cùng lúc. Promise.all(items.map(callApi)) với 10.000 phần tử gửi 10.000 request cùng lúc: dịch vụ phía sau quá tải, pool database cạn, hoặc API bên thứ ba trả 429 Too Many Requests.

Trên lab, một dịch vụ chỉ chịu được 10 request cùng lúc (quá thì trả 503). Gửi 500 việc bằng Promise.all không giới hạn thì chỉ 10 tới 20 việc thành công, còn lại nhận 503. Giới hạn ở phía gọi 10 việc cùng lúc thì cả 500 việc thành công trong khoảng 1,1 giây, đúng bằng sức chịu của dịch vụ. Giới hạn 50 thì vẫn hỏng gần hết, vì vẫn vượt sức chịu.

function pLimit(k) {
  let active = 0
  const queue = []
  const next = () => {
    if (active >= k || !queue.length) return
    active++
    const { fn, resolve, reject } = queue.shift()
    fn().then(resolve, reject).finally(() => { active--; next() })
  }
  return (fn) => new Promise((resolve, reject) => { queue.push({ fn, resolve, reject }); next() })
}

const limit = pLimit(10)
await Promise.all(items.map((item) => limit(() => callApi(item))))

Thư viện p-limit, p-map (concurrency), p-queue làm đúng việc này. Con số giới hạn nên dựa vào sức chịu của phía sau: số connection của pool database, rate limit của API bên thứ ba. Kéo thanh trượt để thấy vì sao retry không thay được giới hạn:

Phía gọi chạy tối đa 10 việc cùng lúc

500 việc

Retry khi nhận 503 (chờ khoảng 50 ms, có jitter)

500 / 500việc thành công

0việc thất bại

500request đã gửi

1.000 msxong tất cả

Phía gọi không bao giờ gửi quá 10 request cùng lúc, nên dịch vụ không phải từ chối ai. 500 việc × 20 ms / 10 = 1.000 ms là nhanh nhất có thể: không cách nào nhanh hơn sức chịu của dịch vụ.

Dịch vụ chịu tối đa 10 request cùng lúc, mỗi request 20 ms, quá thì trả 503 sau 0,5 ms; giống lab/concurrency.

Hàng đợi trong process (một mảng cộng một vòng lặp) chỉ hợp với việc phụ được phép mất khi process chết. Việc quan trọng thì đưa vào hàng đợi bền vững như BullMQ trên Redis, hay hàng đợi bằng database (Hàng đợi bằng database).

Gom các request trùng nhau

Một dạng race khác: 100 request cùng hỏi một key cache vừa hết hạn, cả 100 cùng thấy miss, cùng query database và cùng ghi lại cache. Đây là thundering herd (Sự cố cache). Trong một process, gom chúng lại bằng cách lưu chính promise đang chạy theo key, để các request sau chờ cùng một promise thay vì tự query lại (singleflight):

const inflight = new Map()
function loadOnce(key, loader) {
  if (inflight.has(key)) return inflight.get(key)
  const p = loader(key).finally(() => inflight.delete(key))
  inflight.set(key, p)
  return p
}

Cách này chỉ gom được trong một process: 20 instance vẫn gửi 20 query. Muốn gom giữa các instance thì cần lock trong Redis hay cơ chế ở tầng cache.

Ghi nhớ

1. Code đồng bộ không bị chen ngang, nhưng mỗi await là một chỗ request khác chạy xen vào: đọc, await, rồi ghi dựa trên giá trị đã đọc là race condition.
2. Mutex trong process chỉ đúng khi chỉ có một process; với nhiều instance, chặn ở database bằng ghi có điều kiện, ràng buộc hay lock.
3. Hai process mỗi cái một mutex vẫn rút quá tiền, và số dư cuối có thể trông hợp lý.
4. Giới hạn số việc song song theo sức chịu của phía sau (p-limit, p-map); Promise.all không giới hạn làm phía sau quá tải.
5. Gom request trùng bằng cách lưu promise đang chạy theo key; chỉ có tác dụng trong một process.

Nguồn tham khảo

Kiểm bằng lab: mọi kết quả trong trang. Từ kiến thức chung, chưa đối chiếu lần này: API của async-mutex, p-limit, p-map, p-queue, BullMQ.

Số đo chi tiết

Kết quả chạy lab/concurrency/race.mjs (Node 24.14, hai lượt, mỗi cách 20 lần lặp)

Tài khoản 100 đồng, 20 lần rút 10 đồng cùng lúc; "database" là một process riêng, mỗi lệnh có độ trễ ngẫu nhiên 0–2 ms:

Cách	Số lần rút thành công (đúng là 10)	Số dư cuối	Lượt sai
Một process, đọc rồi ghi qua await	20	80–90	20/20
Một process, mutex trong process	10	0	0/20
Hai process, mỗi process một mutex	16–20	0	20/20
Hai process, đọc rồi ghi	20	80–90	20/20
Hai process, ghi có điều kiện ở database	10	0	0/20

Dịch vụ chịu 10 request cùng lúc (quá thì 503), 500 việc, mỗi việc 20 ms:

Phía gọi	Thành công	Lỗi 503	Thời gian
Promise.all không giới hạn	10–20	480–490	29–75 ms
Giới hạn 10 việc cùng lúc	500	0	1.085–1.098 ms
Giới hạn 50 việc cùng lúc	20	480	43 ms

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

Chạy bằng npm run lab -- concurrency, chỉ cần Node.js, khoảng 10 giây. Xem Lab.

lab/concurrency/race.mjs

// Trang nodejs/concurrency. Chỉ cần Node.js, khoảng 10 giây.
//   1. Check-then-act qua await: 20 lần rút 10 đồng cùng lúc từ tài khoản 100 đồng, trong một process
//   2. Mutex trong process chặn được trong một process, không chặn được khi có hai process
//   3. Ghi có điều kiện ở phía "database" chặn được cả hai trường hợp
//   4. Giới hạn số việc chạy song song khi gọi một dịch vụ chỉ chịu được 10 request cùng lúc
// Khi chạy với đối số "worker", script là một process app thứ hai dùng chung "database".
import http from 'node:http'
import { fork } from 'node:child_process'

const agent = new http.Agent({ keepAlive: true, maxSockets: 256 })
const call = (method, path) => new Promise((resolve, reject) => {
  const req = http.request({ host: '127.0.0.1', port: Number(process.env.STORE_PORT), method, path, agent }, (r) => { let b = ''; r.on('data', (c) => (b += c)); r.on('end', () => resolve({ status: r.statusCode, body: b })) })
  req.on('error', reject); req.end()
})
const getBalance = async () => Number((await call('GET', '/balance')).body)
const setBalance = (v) => call('POST', `/balance?v=${v}`)

// Mutex trong process: chuỗi promise, mỗi người chờ người trước xong
function createMutex() {
  let tail = Promise.resolve()
  return async (fn) => {
    const prev = tail
    let release
    tail = new Promise((r) => (release = r))
    await prev
    try { return await fn() } finally { release() }
  }
}
const lock = createMutex()

async function withdrawNaive(amount) {
  const b = await getBalance()          // request khác chen vào ở đây
  if (b < amount) return false
  await setBalance(b - amount)
  return true
}
const withdrawLocked = (amount) => lock(() => withdrawNaive(amount))
const withdrawAtomic = async (amount) => (await call('POST', `/withdraw?amount=${amount}`)).body === '1'
const STRATEGIES = { naive: withdrawNaive, locked: withdrawLocked, atomic: withdrawAtomic }

if (process.argv[2] === 'worker') {
  // Process app thứ hai: chạy 10 lần rút theo chiến lược được giao, báo số lần thành công
  const fn = STRATEGIES[process.argv[3]]
  const ok = (await Promise.all(Array.from({ length: 10 }, () => fn(10)))).filter(Boolean).length
  process.send(ok)
  agent.destroy()
} else {
  const store = fork(new URL('./store.mjs', import.meta.url))
  const port = await new Promise((r) => store.once('message', r))
  process.env.STORE_PORT = String(port)
  const run = async (fn, n) => (await Promise.all(Array.from({ length: n }, () => fn(10)))).filter(Boolean).length
  const ROUNDS = 20
  const line = (label, results) => {
    const over = results.filter((r) => r.ok > 10).length
    console.log(`${label.padEnd(52)} rút thành công ${Math.min(...results.map((r) => r.ok))}–${Math.max(...results.map((r) => r.ok))} lần (đúng là 10), số dư cuối ${Math.min(...results.map((r) => r.end))}–${Math.max(...results.map((r) => r.end))}; ${over}/${results.length} lượt sai`)
  }
  console.log(`\n=== Tài khoản 100 đồng, 20 lần rút 10 đồng cùng lúc, ${ROUNDS} lượt mỗi cách ===`)
  for (const [label, name, procs] of [['Một process, đọc rồi ghi qua await', 'naive', 1], ['Một process, có mutex trong process', 'locked', 1],
    ['Hai process, mỗi process có mutex riêng', 'locked', 2], ['Hai process, đọc rồi ghi', 'naive', 2], ['Hai process, ghi có điều kiện ở database', 'atomic', 2]]) {
    const results = []
    for (let i = 0; i < ROUNDS; i++) {
      await call('POST', '/reset?v=100')
      let ok
      if (procs === 1) ok = await run(STRATEGIES[name], 20)
      else {
        const ws = [0, 1].map(() => fork(new URL(import.meta.url), ['worker', name], { env: { ...process.env, STORE_PORT: String(port) } }))
        ok = (await Promise.all(ws.map((w) => new Promise((r) => w.once('message', r))))).reduce((a, b) => a + b, 0)
      }
      results.push({ ok, end: await getBalance() })
    }
    line(label, results)
  }

  console.log('\n=== Gọi dịch vụ chịu tối đa 10 request cùng lúc (quá thì 503), 500 việc, mỗi việc 20 ms ===')
  const limit = (k) => {
    let active = 0
    const queue = []
    const next = () => { if (active < k && queue.length) { active++; const { fn, res, rej } = queue.shift(); fn().then(res, rej).finally(() => { active--; next() }) } }
    return (fn) => new Promise((res, rej) => { queue.push({ fn, res, rej }); next() })
  }
  for (const [label, k] of [['Promise.all không giới hạn', Infinity], ['Giới hạn 10 việc cùng lúc', 10], ['Giới hạn 50 việc cùng lúc', 50]]) {
    const l = limit(k)
    const t = performance.now()
    const rs = await Promise.all(Array.from({ length: 500 }, () => l(() => call('GET', '/slot'))))
    const peak = (await call('GET', '/peak')).body
    console.log(`${label.padEnd(52)} ${rs.filter((r) => r.status === 200).length}/500 thành công, ${rs.filter((r) => r.status === 503).length} lỗi 503, ${Math.round(performance.now() - t)} ms, dịch vụ chạy cùng lúc nhiều nhất ${peak}`)
  }
  agent.destroy()
  store.kill()
}

lab/concurrency/store.mjs: "database" giả lập chạy trong process riêng

// "Database" giả lập chạy trong process riêng, cho lab/concurrency: một tài khoản có số dư, ba endpoint
//   GET  /balance             đọc số dư
//   POST /balance?v=N         ghi số dư (ghi đè)
//   POST /withdraw?amount=N   trừ có điều kiện trong một bước, giống UPDATE … SET b = b - N WHERE b >= N
//   POST /reset?v=N           đặt lại số dư
//   GET  /slot                dịch vụ chỉ chịu được 10 request cùng lúc: quá thì trả 503, mỗi request mất 20 ms
// Mỗi request có độ trễ ngẫu nhiên 0–2 ms, như một lần đi qua mạng tới database.
import http from 'node:http'

let balance = 100
let inFlight = 0, peak = 0
const delay = () => new Promise((r) => setTimeout(r, Math.random() * 2))
http.createServer(async (req, res) => {
  const url = new URL(req.url, 'http://x')
  if (url.pathname === '/slot') {
    if (inFlight >= 10) { res.statusCode = 503; return res.end('busy') }
    inFlight++; peak = Math.max(peak, inFlight)
    await new Promise((r) => setTimeout(r, 20))
    inFlight--
    return res.end('ok')
  }
  if (url.pathname === '/peak') { res.end(String(peak)); peak = 0; return }
  await delay()
  if (url.pathname === '/balance' && req.method === 'GET') return res.end(String(balance))
  if (url.pathname === '/balance') { balance = Number(url.searchParams.get('v')); return res.end('ok') }
  if (url.pathname === '/reset') { balance = Number(url.searchParams.get('v')); return res.end('ok') }
  if (url.pathname === '/withdraw') {
    const n = Number(url.searchParams.get('amount'))
    if (balance >= n) { balance -= n; return res.end('1') } // một bước, không có await ở giữa: không ai chen vào được
    return res.end('0')
  }
  res.statusCode = 404; res.end()
}).listen(0, '127.0.0.1', function () { process.send(this.address().port) })