Merge pull request #56905 from mihir-kandoi/pg-lock-races-and-advisory-valuation

fix(stock): close postgres locking races; gate batch valuation with a txn advisory lock

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

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)

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

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

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)

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)