From 2ec469257e4474d3d766bbaad86d748be9856673 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Sun, 5 Jul 2026 22:31:45 +0530 Subject: [PATCH] perf(stock): gate batch valuation with a txn advisory lock on postgres 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. --- erpnext/stock/deprecated_serial_batch.py | 34 ++++--------------- .../test_serial_and_batch_bundle.py | 27 +++++++++++++++ erpnext/stock/serial_batch_bundle.py | 20 +++++++---- 3 files changed, 46 insertions(+), 35 deletions(-) diff --git a/erpnext/stock/deprecated_serial_batch.py b/erpnext/stock/deprecated_serial_batch.py index b2010411644..9e097099f01 100644 --- a/erpnext/stock/deprecated_serial_batch.py +++ b/erpnext/stock/deprecated_serial_batch.py @@ -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) diff --git a/erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py b/erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py index 4ad8a4f4136..a491c805aa3 100644 --- a/erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py +++ b/erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py @@ -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 diff --git a/erpnext/stock/serial_batch_bundle.py b/erpnext/stock/serial_batch_bundle.py index 5eef60dcc1a..56812b1bfed 100644 --- a/erpnext/stock/serial_batch_bundle.py +++ b/erpnext/stock/serial_batch_bundle.py @@ -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)