nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-08-16 16:08:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(stock): gate batch valuation with a txn advisory lock on postgres

Browse Source 
On postgres, outward batch valuation row-locked the item's ENTIRE SLE / Serial and Batch Entry history (a separate plain SELECT ... FOR UPDATE per site, since FOR UPDATE is invalid with GROUP BY there). That writes a lock marker (xmax + WAL) on every historical tuple per outward movement - write amplification that grows with history forever - and locks nothing at all when the history is empty (negative-stock edge: two concurrent outwards don't serialize).

Replace all four history-wide postgres lock statements with one frappe.db.transaction_advisory_lock(("batch-valuation", item_code, warehouse)) at the top of BatchNoValuation.calculate_avg_rate's outward branch - every valuation read (bundle + the three deprecated paths) is downstream of it. Released at commit/rollback, participates in deadlock detection, and serializes regardless of history size, so the empty-history edge is closed by construction. Batch qty updates iterate in sorted order so concurrent vouchers lock Batch rows in the same sequence.

MariaDB is unchanged: it keeps the original grouped FOR UPDATE row locks (and its gap locks). Requires frappe#40621.

Test: outward delivery of a batched item must leave the xact advisory lock visible in pg_locks for the submitting transaction.
This commit is contained in:
Mihir Kandoi
2026-07-05 22:31:45 +05:30
parent db58858c68
commit 2ec469257e
3 changed files with 46 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
erpnext/stock/deprecated_serial_batch.py
View File

@@ -144,12 +144,8 @@ class DeprecatedBatchNoValuation:
if self.sle.name:
conditions &= sle.name != self.sle.name
# Lock the scanned SLE rows so a concurrent stock posting can't change them mid-valuation.
# MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so
# lock the same rows in a separate plain SELECT first (held for the transaction).
if frappe.db.db_type == "postgres":
frappe.qb.from_(sle).select(sle.name).where(conditions).for_update().run()
# MariaDB carries a row lock on the grouped query below; on postgres the caller
# (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse).
query = (
frappe.qb.from_(sle)
.select(
@@ -269,13 +265,8 @@ class DeprecatedBatchNoValuation:
if self.sle.name:
conditions &= sle.name != self.sle.name
# Lock the scanned SLE rows so a concurrent stock posting can't change them mid-valuation.
# MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so
# lock the same SLE rows in a separate plain SELECT first. The batch.use_batchwise_valuation
# refinement below only narrows the set, so locking without the join is a safe superset.
if frappe.db.db_type == "postgres":
frappe.qb.from_(sle).select(sle.name).where(conditions).for_update().run()
# MariaDB carries a row lock on the grouped query below; on postgres the caller
# (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse).
query = (
frappe.qb.from_(sle)
.inner_join(batch)
@@ -402,21 +393,8 @@ class DeprecatedBatchNoValuation:
conditions &= bundle.name != self.sle.serial_and_batch_bundle
conditions &= bundle.voucher_type != "Pick List"
# Lock the scanned bundle rows so a concurrent stock posting can't change them mid-valuation.
# MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so
# lock the same rows in a separate plain SELECT first (the batch.use_batchwise_valuation
# refinement below only narrows the set, so omitting that join is a safe superset).
if frappe.db.db_type == "postgres":
(
frappe.qb.from_(bundle)
.inner_join(bundle_child)
.on(bundle.name == bundle_child.parent)
.select(bundle_child.name)
.where(conditions)
.for_update()
.run()
)
# MariaDB carries a row lock on the grouped query below; on postgres the caller
# (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse).
query = (
frappe.qb.from_(bundle)
.inner_join(bundle_child)

27
erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py
View File

@@ -201,6 +201,33 @@ class TestSerialandBatchBundle(ERPNextTestSuite):
self.assertEqual(flt(stock_value_difference, 2), -5000)
def test_outward_batch_valuation_takes_transaction_advisory_lock(self):
if frappe.db.db_type != "postgres":
return
from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt
item_code = make_item(
properties={
"has_batch_no": 1,
"create_new_batch": 1,
"batch_number_series": "TEST-ADV-LCK-.#####",
"is_stock_item": 1,
},
).name
make_purchase_receipt(item_code=item_code, warehouse="_Test Warehouse - _TC", qty=5, rate=100)
def held_advisory_locks():
return frappe.db.sql(
"SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND pid = pg_backend_pid()"
)[0][0]
before = held_advisory_locks()
create_delivery_note(item_code=item_code, warehouse="_Test Warehouse - _TC", qty=2, rate=200)
self.assertGreater(held_advisory_locks(), before)
def test_old_batch_valuation(self):
frappe.flags.ignore_serial_batch_bundle_validation = True
frappe.flags.use_serial_and_batch_fields = True

20
erpnext/stock/serial_batch_bundle.py
View File

@@ -821,6 +821,15 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
"Serial and Batch Bundle", self.sle.serial_and_batch_bundle, "total_amount"
)
else:
# Serialize concurrent valuations of this (item, warehouse) on postgres. MariaDB's
# grouped FOR UPDATE + gap locks do this via the history reads below; postgres has no
# gap locks, and row-locking the whole history writes a lock marker on every tuple --
# a txn-scoped advisory lock (released at commit/rollback) serializes without either.
if frappe.db.db_type == "postgres":
frappe.db.transaction_advisory_lock(
("batch-valuation", self.sle.item_code, self.sle.warehouse)
)
entries = self.get_batch_stock_before_date()
self.stock_value_change = 0.0
self.batch_avg_rate = defaultdict(float)
@@ -869,12 +878,9 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
if timestamp_condition:
conditions &= timestamp_condition
# Lock the scanned rows so a concurrent stock transaction can't change them mid-valuation.
# MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so
# lock the same rows in a separate plain SELECT first (held for the transaction).
if frappe.db.db_type == "postgres":
frappe.qb.from_(child).select(child.name).where(conditions).for_update().run()
# MariaDB carries a row lock on the grouped query below; on postgres the caller
# (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse)
# instead of row-locking the whole history (FOR UPDATE is invalid with GROUP BY there).
query = (
frappe.qb.from_(child)
.select(
@@ -1561,7 +1567,7 @@ def update_batch_qty(voucher_type, voucher_no, docstatus, via_landed_cost_vouche
return
precision = frappe.get_precision("Batch", "batch_qty")
for batch, qty in batches.items():
for batch, qty in sorted(batches.items()):
current_qty = get_batch_current_qty(batch)
current_qty += flt(qty, precision) * (-1 if docstatus == 2 else 1)