From a8991830877c35d9668aa3d7cd68e3b181023584 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Wed, 17 Jun 2026 08:50:33 +0530 Subject: [PATCH] fix(postgres): keep row locks via lock-then-read (address review) Acquire the same row locks on postgres that MariaDB takes, via a separate plain SELECT ... FOR UPDATE before each grouped/aggregate read (FOR UPDATE is invalid with GROUP BY on postgres). Applied to all 6 aggregate lock sites. Co-Authored-By: Claude Opus 4.8 (1M context) --- erpnext/stock/deprecated_serial_batch.py | 112 +++++++++++------- erpnext/stock/doctype/pick_list/pick_list.py | 23 ++-- .../stock_reservation_entry.py | 27 +++-- erpnext/stock/serial_batch_bundle.py | 44 ++++--- 4 files changed, 119 insertions(+), 87 deletions(-) diff --git a/erpnext/stock/deprecated_serial_batch.py b/erpnext/stock/deprecated_serial_batch.py index 4652f53ff7b..b2010411644 100644 --- a/erpnext/stock/deprecated_serial_batch.py +++ b/erpnext/stock/deprecated_serial_batch.py @@ -132,6 +132,24 @@ class DeprecatedBatchNoValuation: sle.creation < self.sle.creation ) + conditions = ( + (sle.item_code == self.sle.item_code) + & (sle.warehouse == self.sle.warehouse) + & (sle.batch_no.isin(self.batchwise_valuation_batches)) + & (sle.batch_no.isnotnull()) + & (sle.is_cancelled == 0) + ) + if timestamp_condition: + conditions &= timestamp_condition + 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() + query = ( frappe.qb.from_(sle) .select( @@ -139,23 +157,9 @@ class DeprecatedBatchNoValuation: Sum(sle.stock_value_difference).as_("batch_value"), Sum(sle.actual_qty).as_("batch_qty"), ) - .where( - (sle.item_code == self.sle.item_code) - & (sle.warehouse == self.sle.warehouse) - & (sle.batch_no.isin(self.batchwise_valuation_batches)) - & (sle.batch_no.isnotnull()) - & (sle.is_cancelled == 0) - ) + .where(conditions) .groupby(sle.batch_no) ) - - if timestamp_condition: - query = query.where(timestamp_condition) - - if self.sle.name: - query = query.where(sle.name != self.sle.name) - - # lock scanned rows on MariaDB; FOR UPDATE is invalid with GROUP BY on postgres if frappe.db.db_type != "postgres": query = query.for_update() @@ -254,6 +258,24 @@ class DeprecatedBatchNoValuation: sle.creation < self.sle.creation ) + conditions = ( + (sle.item_code == self.sle.item_code) + & (sle.warehouse == self.sle.warehouse) + & (sle.batch_no.isnotnull()) + & (sle.is_cancelled == 0) + & (sle.batch_no.isin(self.non_batchwise_valuation_batches)) + & timestamp_condition + ) + 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() + query = ( frappe.qb.from_(sle) .inner_join(batch) @@ -263,20 +285,10 @@ class DeprecatedBatchNoValuation: Sum(sle.actual_qty).as_("batch_qty"), Sum(sle.stock_value_difference).as_("batch_value"), ) - .where( - (sle.item_code == self.sle.item_code) - & (sle.warehouse == self.sle.warehouse) - & (sle.batch_no.isnotnull()) - & (sle.is_cancelled == 0) - & (sle.batch_no.isin(self.non_batchwise_valuation_batches)) - ) - .where(timestamp_condition) + .where(conditions) .groupby(sle.batch_no) ) - if self.sle.name: - query = query.where(sle.name != self.sle.name) - # Moving Average items with no Use Batch wise Valuation but want to use batch wise valuation moving_avg_item_non_batch_value = False if valuation_method := self.get_valuation_method(self.sle.item_code): @@ -286,7 +298,6 @@ class DeprecatedBatchNoValuation: query = query.where(batch.use_batchwise_valuation == 0) moving_avg_item_non_batch_value = True - # lock scanned rows on MariaDB; FOR UPDATE is invalid with GROUP BY on postgres if frappe.db.db_type != "postgres": query = query.for_update() @@ -377,6 +388,35 @@ class DeprecatedBatchNoValuation: bundle.creation < self.sle.creation ) + conditions = ( + (bundle.item_code == self.sle.item_code) + & (bundle.warehouse == self.sle.warehouse) + & (bundle_child.batch_no.isnotnull()) + & (bundle.is_cancelled == 0) + & (bundle.docstatus == 1) + & (bundle.type_of_transaction.isin(["Inward", "Outward"])) + & (bundle_child.batch_no.isin(self.non_batchwise_valuation_batches)) + & timestamp_condition + ) + if self.sle.serial_and_batch_bundle: + 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() + ) + query = ( frappe.qb.from_(bundle) .inner_join(bundle_child) @@ -388,25 +428,10 @@ class DeprecatedBatchNoValuation: Sum(bundle_child.qty).as_("batch_qty"), Sum(bundle_child.stock_value_difference).as_("batch_value"), ) - .where( - (bundle.item_code == self.sle.item_code) - & (bundle.warehouse == self.sle.warehouse) - & (bundle_child.batch_no.isnotnull()) - & (bundle.is_cancelled == 0) - & (bundle.docstatus == 1) - & (bundle.type_of_transaction.isin(["Inward", "Outward"])) - & (bundle_child.batch_no.isin(self.non_batchwise_valuation_batches)) - ) - # FOR UPDATE is invalid with GROUP BY on postgres (deprecated valuation path) - .where(timestamp_condition) + .where(conditions) .groupby(bundle_child.batch_no) ) - if self.sle.serial_and_batch_bundle: - query = query.where(bundle.name != self.sle.serial_and_batch_bundle) - - query = query.where(bundle.voucher_type != "Pick List") - # Moving Average items with no Use Batch wise Valuation but want to use batch wise valuation moving_avg_item_non_batch_value = False if valuation_method := self.get_valuation_method(self.sle.item_code): @@ -416,7 +441,6 @@ class DeprecatedBatchNoValuation: query = query.where(batch.use_batchwise_valuation == 0) moving_avg_item_non_batch_value = True - # lock scanned rows on MariaDB; FOR UPDATE is invalid with GROUP BY on postgres if frappe.db.db_type != "postgres": query = query.for_update() diff --git a/erpnext/stock/doctype/pick_list/pick_list.py b/erpnext/stock/doctype/pick_list/pick_list.py index 6ee1b9dede6..2429970dd50 100644 --- a/erpnext/stock/doctype/pick_list/pick_list.py +++ b/erpnext/stock/doctype/pick_list/pick_list.py @@ -902,6 +902,9 @@ def update_pick_list_status(pick_list): def get_picked_items_qty(items, contains_packed_items=False) -> list[dict]: pi_item = frappe.qb.DocType("Pick List Item") + group_field = pi_item.product_bundle_item if contains_packed_items else pi_item.sales_order_item + conditions = (pi_item.docstatus == 1) & group_field.isin(items) + query = ( frappe.qb.from_(pi_item) .select( @@ -914,22 +917,16 @@ def get_picked_items_qty(items, contains_packed_items=False) -> list[dict]: Sum(pi_item.stock_qty).as_("stock_qty"), Sum(pi_item.picked_qty).as_("picked_qty"), ) - .where(pi_item.docstatus == 1) + .where(conditions) + .groupby(group_field, pi_item.sales_order) ) - if contains_packed_items: - query = query.groupby( - pi_item.product_bundle_item, - pi_item.sales_order, - ).where(pi_item.product_bundle_item.isin(items)) + # 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). + if frappe.db.db_type == "postgres": + frappe.qb.from_(pi_item).select(pi_item.name).where(conditions).for_update().run() else: - query = query.groupby( - pi_item.sales_order_item, - pi_item.sales_order, - ).where(pi_item.sales_order_item.isin(items)) - - # FOR UPDATE is invalid with GROUP BY on postgres; lock scanned rows on MariaDB only - if frappe.db.db_type != "postgres": query = query.for_update() return query.run(as_dict=True) 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 2c5c89596ad..5c586a1fd53 100644 --- a/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py +++ b/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py @@ -706,21 +706,26 @@ def get_available_qty_to_reserve( if available_qty: sre = frappe.qb.DocType("Stock Reservation Entry") + conditions = ( + (sre.docstatus == 1) + & (sre.item_code == item_code) + & (sre.warehouse == warehouse) + & (sre.delivered_qty < sre.reserved_qty) + ) + if ignore_sre: + 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). + if frappe.db.db_type == "postgres": + frappe.qb.from_(sre).select(sre.name).where(conditions).for_update().run() + query = ( frappe.qb.from_(sre) .select(Sum(sre.reserved_qty - sre.delivered_qty - sre.transferred_qty - sre.consumed_qty)) - .where( - (sre.docstatus == 1) - & (sre.item_code == item_code) - & (sre.warehouse == warehouse) - & (sre.delivered_qty < sre.reserved_qty) - ) + .where(conditions) ) - - if ignore_sre: - query = query.where(sre.name != ignore_sre) - - # FOR UPDATE is invalid with aggregates on postgres; lock scanned rows on MariaDB only if frappe.db.db_type != "postgres": query = query.for_update() diff --git a/erpnext/stock/serial_batch_bundle.py b/erpnext/stock/serial_batch_bundle.py index 2e9a9a0d97b..1144f32f848 100644 --- a/erpnext/stock/serial_batch_bundle.py +++ b/erpnext/stock/serial_batch_bundle.py @@ -851,6 +851,30 @@ class BatchNoValuation(DeprecatedBatchNoValuation): child.creation < self.sle.creation ) + conditions = ( + (child.item_code == self.sle.item_code) + & (child.warehouse == self.sle.warehouse) + & (child.batch_no.isin(self.batchwise_valuation_batches)) + & (child.docstatus == 1) + & (child.type_of_transaction.isin(["Inward", "Outward"])) + ) + + # Important to exclude the current voucher detail no / voucher no to calculate the correct stock value difference + if self.sle.voucher_detail_no: + conditions &= child.voucher_detail_no != self.sle.voucher_detail_no + elif self.sle.voucher_no: + conditions &= child.voucher_no != self.sle.voucher_no + + conditions &= child.voucher_type != "Pick List" + 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() + query = ( frappe.qb.from_(child) .select( @@ -858,30 +882,12 @@ class BatchNoValuation(DeprecatedBatchNoValuation): Sum(child.stock_value_difference).as_("incoming_rate"), Sum(child.qty).as_("qty"), ) - .where( - (child.item_code == self.sle.item_code) - & (child.warehouse == self.sle.warehouse) - & (child.batch_no.isin(self.batchwise_valuation_batches)) - & (child.docstatus == 1) - & (child.type_of_transaction.isin(["Inward", "Outward"])) - ) + .where(conditions) .groupby(child.batch_no) ) - - # FOR UPDATE is invalid with GROUP BY on postgres; lock scanned rows on MariaDB only if frappe.db.db_type != "postgres": query = query.for_update() - # Important to exclude the current voucher detail no / voucher no to calculate the correct stock value difference - if self.sle.voucher_detail_no: - query = query.where(child.voucher_detail_no != self.sle.voucher_detail_no) - elif self.sle.voucher_no: - query = query.where(child.voucher_no != self.sle.voucher_no) - - query = query.where(child.voucher_type != "Pick List") - if timestamp_condition: - query = query.where(timestamp_condition) - return query.run(as_dict=True) def prepare_batches(self):