From db58858c6879050c647d91224624756d65478799 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Sun, 5 Jul 2026 22:31:29 +0530 Subject: [PATCH 1/2] fix(stock): close postgres lock-then-read races in pick list and stock reservation Postgres has no gap locks, so the lock-then-read pattern (plain SELECT ... FOR UPDATE before a grouped read) only serializes on rows that already exist. Two sites had reachable races where the lock set is empty or disjoint: - Pick list: two pick lists against the same SO item submitted concurrently lock only docstatus=1 rows, so with no previously-submitted picks their lock sets are disjoint and both pass validate_picked_qty (over-pick; picked_qty last-writer-wins). Gate on the referenced Sales Order Item / Packed Item rows, which always exist. - Stock reservation: the first concurrent reservations for an (item, warehouse) find no SRE rows to lock, so both pass and reserved qty can exceed actual. Gate on the Bin row, which exists once there is stock. MariaDB is unchanged (its gap locks already serialize both; the gates are postgres-only). Also: ORDER BY on the small-set postgres lock selects for deterministic lock order, and the repost pre-lock in get_future_stock_vouchers selects a constant instead of shipping every matching SLE name to the client. --- erpnext/accounts/utils.py | 6 ++++-- erpnext/stock/doctype/pick_list/pick_list.py | 18 +++++++++++++++--- .../stock_reservation_entry.py | 15 ++++++++++++--- 3 files changed, 31 insertions(+), 8 deletions(-) diff --git a/erpnext/accounts/utils.py b/erpnext/accounts/utils.py index d6cda9cf548..9b50751e95a 100644 --- a/erpnext/accounts/utils.py +++ b/erpnext/accounts/utils.py @@ -13,6 +13,7 @@ from frappe.desk.reportview import build_match_conditions from frappe.model.meta import get_field_precision from frappe.model.naming import determine_consecutive_week_number from frappe.query_builder import AliasedQuery, Case, Criterion, Field, Table +from frappe.query_builder.custom import ConstantColumn from frappe.query_builder.functions import Count, IfNull, Max, Min, Round, Sum from frappe.query_builder.utils import DocType from frappe.utils import ( @@ -1791,9 +1792,10 @@ def get_future_stock_vouchers(posting_date, posting_time, for_warehouses=None, f # transaction can't modify them mid-flight (the original DISTINCT ... FOR UPDATE did this). # MariaDB carries the lock on the grouped query below; postgres rejects FOR UPDATE alongside # GROUP BY, so lock the matching rows in a separate pass first -- the row locks are held until - # the surrounding transaction ends, giving the same protection. + # the surrounding transaction ends, giving the same protection. Select a constant, not the + # name: a deep backdated repost can match millions of rows and only the locks are needed. if frappe.db.db_type == "postgres": - frappe.qb.from_(SLE).select(SLE.name).where(conditions).for_update().run() + frappe.qb.from_(SLE).select(ConstantColumn(1)).where(conditions).for_update().run() # distinct vouchers in chronological order; expressed as GROUP BY + Min() so it's valid on # postgres (SELECT DISTINCT can't ORDER BY non-selected cols, and FOR UPDATE is invalid with both). diff --git a/erpnext/stock/doctype/pick_list/pick_list.py b/erpnext/stock/doctype/pick_list/pick_list.py index 846c020de72..53e515b0f8e 100644 --- a/erpnext/stock/doctype/pick_list/pick_list.py +++ b/erpnext/stock/doctype/pick_list/pick_list.py @@ -952,10 +952,22 @@ def get_picked_items_qty(items, contains_packed_items=False) -> list[dict]: ) # Lock the picked-qty rows so a concurrent pick can't change them mid-transaction. 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). + # the lock on the grouped query (its gap locks also block rows other in-flight picks are about to + # submit); postgres has no gap locks, so first serialize on the referenced SO/packed item rows + # (they always exist), then lock the matching picked rows in a separate plain SELECT. if frappe.db.db_type == "postgres": - frappe.qb.from_(pi_item).select(pi_item.name).where(conditions).for_update().run() + parent = frappe.qb.DocType("Packed Item" if contains_packed_items else "Sales Order Item") + ( + frappe.qb.from_(parent) + .select(parent.name) + .where(parent.name.isin(items)) + .orderby(parent.name) + .for_update() + .run() + ) + frappe.qb.from_(pi_item).select(pi_item.name).where(conditions).orderby( + pi_item.name + ).for_update().run() else: query = query.for_update() diff --git a/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py b/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py index ad6e965b186..3dbdc2419c9 100644 --- a/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py +++ b/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py @@ -716,10 +716,19 @@ def get_available_qty_to_reserve( conditions &= sre.name != ignore_sre # Lock the rows being aggregated so a concurrent reservation can't change them mid-transaction. - # MariaDB carries the lock on the aggregate query itself; postgres rejects FOR UPDATE with an - # aggregate, so on postgres lock the same rows in a separate plain SELECT first (held for the txn). + # MariaDB carries the lock on the aggregate query itself (its gap locks also serialize two + # FIRST reservations, when no SRE rows exist yet); postgres has no gap locks, so gate on the + # Bin row (exists once there is stock), then lock the matching SREs in a plain SELECT. if frappe.db.db_type == "postgres": - frappe.qb.from_(sre).select(sre.name).where(conditions).for_update().run() + bin_table = frappe.qb.DocType("Bin") + ( + frappe.qb.from_(bin_table) + .select(bin_table.name) + .where((bin_table.item_code == item_code) & (bin_table.warehouse == warehouse)) + .for_update() + .run() + ) + frappe.qb.from_(sre).select(sre.name).where(conditions).orderby(sre.name).for_update().run() query = ( frappe.qb.from_(sre) From 2ec469257e4474d3d766bbaad86d748be9856673 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Sun, 5 Jul 2026 22:31:45 +0530 Subject: [PATCH 2/2] 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)