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 <pk> ... 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) <noreply@anthropic.com>

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()
 

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)

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()
 

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):