diff --git a/erpnext/accounts/doctype/account/account.py b/erpnext/accounts/doctype/account/account.py index ebfb2d0bcee..e67b29bc1be 100644 --- a/erpnext/accounts/doctype/account/account.py +++ b/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) diff --git a/erpnext/stock/doctype/pick_list/pick_list.py b/erpnext/stock/doctype/pick_list/pick_list.py index 53e515b0f8e..50a3a0ebfb4 100644 --- a/erpnext/stock/doctype/pick_list/pick_list.py +++ b/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() diff --git a/erpnext/stock/doctype/pick_list/test_pick_list.py b/erpnext/stock/doctype/pick_list/test_pick_list.py index e37cb0a3532..3ca41f29bb3 100644 --- a/erpnext/stock/doctype/pick_list/test_pick_list.py +++ b/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: diff --git a/erpnext/stock/doctype/stock_ledger_entry/test_stock_ledger_entry.py b/erpnext/stock/doctype/stock_ledger_entry/test_stock_ledger_entry.py index d0fcec592ad..0c4c368ad3e 100644 --- a/erpnext/stock/doctype/stock_ledger_entry/test_stock_ledger_entry.py +++ b/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 diff --git a/erpnext/stock/stock_ledger.py b/erpnext/stock/stock_ledger.py index c1ce66317bc..aaecd0a4e0c 100644 --- a/erpnext/stock/stock_ledger.py +++ b/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()