Merge pull request #57202 from mihir-kandoi/pg-read-committed-gates

fix(stock): serialize postgres stock writes per (item, warehouse); block GL inserts during account rename

erpnext/accounts/doctype/account/account.py

@@ -659,8 +659,15 @@ def _ensure_idle_system():
 
 	last_gl_update = None
 	try:
-		# We also lock inserts to GL entry table with for_update here.
-		last_gl_update = frappe.db.get_value("GL Entry", {}, "modified", for_update=True, wait=False)
+		if frappe.db.db_type == "postgres":
+			# The MariaDB branch blocks new GL inserts via the gap lock its for_update read takes;
+			# a postgres row lock never blocks inserts, so take an EXCLUSIVE table lock instead --
+			# writers block until the rename commits, readers don't. NOWAIT mirrors wait=False.
+			frappe.db.sql("LOCK TABLE `tabGL Entry` IN EXCLUSIVE MODE NOWAIT")
+			last_gl_update = frappe.db.get_value("GL Entry", {}, "modified")
+		else:
+			# We also lock inserts to GL entry table with for_update here.
+			last_gl_update = frappe.db.get_value("GL Entry", {}, "modified", for_update=True, wait=False)
 	except frappe.QueryTimeoutError:
 		# wait=False fails immediately if there's an active transaction.
 		last_gl_update = add_to_date(None, seconds=-1)

erpnext/stock/doctype/pick_list/pick_list.py

@@ -550,6 +550,15 @@ class PickList(TransactionBase):
 	def set_item_locations(self, save: bool = False):
 		self.validate_for_qty()
 		items = self.aggregate_item_qty()
+
+		# Serialize concurrent allocations per item on postgres. MariaDB's gap locks on the
+		# picked-items locking read below already make two simultaneous allocations take turns;
+		# postgres locking reads can't see the rows another in-flight allocation is inserting, so
+		# both could claim the same stock. Sorted so overlapping documents can't deadlock.
+		if frappe.db.db_type == "postgres" and hasattr(frappe.db, "transaction_advisory_lock"):
+			for item_code in sorted({d.item_code for d in items}):
+				frappe.db.transaction_advisory_lock(("pick-allocate", item_code))
+
 		picked_items_details = self.get_picked_items_details(items)
 		self.item_location_map = frappe._dict()
 

erpnext/stock/doctype/pick_list/test_pick_list.py

@@ -29,6 +29,23 @@ from erpnext.tests.utils import ERPNextTestSuite
 
 
 class TestPickList(ERPNextTestSuite):
+	def test_pick_list_allocation_takes_advisory_gate(self):
+		if frappe.db.db_type != "postgres":
+			return
+
+		item = make_item(properties={"is_stock_item": 1}).name
+		make_stock_entry(item=item, to_warehouse="_Test Warehouse - _TC", qty=5, basic_rate=100)
+		sales_order = make_sales_order(item_code=item, warehouse="_Test Warehouse - _TC", qty=2, 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_pick_list(sales_order.name)
+		self.assertGreater(held_advisory_locks(), before)
+
 	def test_pick_list_picks_warehouse_for_each_item(self):
 		item_code = make_item().name
 		try:

erpnext/stock/doctype/stock_ledger_entry/test_stock_ledger_entry.py

@@ -34,6 +34,21 @@ class TestStockLedgerEntry(ERPNextTestSuite, StockTestMixin):
 		create_items()
 		reset("Stock Entry")
 
+	def test_stock_write_takes_sle_advisory_gate(self):
+		if frappe.db.db_type != "postgres":
+			return
+
+		item = make_item(properties={"is_stock_item": 1}).name
+
+		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()
+		make_stock_entry(item_code=item, target="_Test Warehouse - _TC", qty=1, rate=10)
+		self.assertGreater(held_advisory_locks(), before)
+
 	def test_incoming_value_for_transferred_serial_no_is_deterministic(self):
 		"""get_incoming_value_for_serial_nos picks the latest SLE (posting_date desc, limit 1) for a
 		serial transferred to another company. posting_date alone is non-total, so two same-date SLEs

erpnext/stock/stock_ledger.py

@@ -115,6 +115,10 @@ def make_sl_entries(sl_entries, allow_negative_stock=False, via_landed_cost_vouc
 	from erpnext.controllers.stock_controller import future_sle_exists
 
 	if sl_entries:
+		# Sorted so two vouchers touching the same pairs can't take the gates in opposite order.
+		for pair in sorted({(d.get("item_code"), d.get("warehouse")) for d in sl_entries}):
+			sle_processing_gate(*pair)
+
 		cancelled = sl_entries[0].get("is_cancelled")
 		if cancelled:
 			validate_cancellation(sl_entries)
@@ -285,6 +289,18 @@ def repost_gate(item_code, warehouse):
 	return nullcontext()
 
 
+def sle_processing_gate(item_code, warehouse):
+	"""Serialize all stock writes for an (item, warehouse) on postgres. MariaDB gets this from the
+	gap locks its previous-SLE locking reads take (which also block, then reveal, concurrent
+	inserts); postgres locking reads never see rows another transaction is inserting, so without
+	this gate two concurrent writers compute from the same stale previous SLE and the loser's Bin
+	write is lost. Txn-scoped and re-entrant; released at commit/rollback. hasattr keeps a frappe
+	predating transaction_advisory_lock on the status quo (serialization-failure retries) instead
+	of breaking every stock submission."""
+	if frappe.db.db_type == "postgres" and hasattr(frappe.db, "transaction_advisory_lock"):
+		frappe.db.transaction_advisory_lock(("stock-sle", item_code, warehouse), timeout=REPOST_LOCK_TIMEOUT)
+
+
 def repost_future_sle(
 	items_to_be_repost=None,
 	voucher_type=None,
@@ -594,6 +610,8 @@ class update_entries_after:
 		if self.args.sle_id:
 			self.args["name"] = self.args.sle_id
 
+		sle_processing_gate(self.item_code, self.args.warehouse)
+
 		self.prev_sle_dict = frappe._dict({})
 		self.company = frappe.get_cached_value("Warehouse", self.args.warehouse, "company")
 		self.set_precision()