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..e97bbd923ab 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,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()