fix(stock): serialize stock writes per (item, warehouse) with a txn advisory lock on postgres

Postgres locking reads never see rows a concurrent transaction is inserting
(MariaDB's gap locks block the insert, then its locking reads return the fresh
row), so two concurrent writers for the same (item, warehouse) compute from the
same stale previous SLE and the loser overwrites Bin with a wrong absolute qty.
Today only the REPEATABLE READ serialization-failure retry catches this; the
gate makes correctness lock-based, covers the empty-history first-transaction
case (nothing exists to row-lock), and keeps negative-stock validation accurate
against concurrently inserted SLEs. Taken at the top of make_sl_entries (sorted
pairs, before the future_sle_exists cache warms) and in
update_entries_after.__init__ for the repost paths; re-entrant, released at
commit. MariaDB paths unchanged.

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,16 @@ 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."""
+	if frappe.db.db_type == "postgres":
+		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 +608,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()