diff --git a/erpnext/stock/stock_ledger.py b/erpnext/stock/stock_ledger.py index f03e75c2b51..db5bc5b5f66 100644 --- a/erpnext/stock/stock_ledger.py +++ b/erpnext/stock/stock_ledger.py @@ -5,6 +5,7 @@ import copy import gzip import json from collections import deque +from contextlib import nullcontext import frappe from frappe import _, bold, scrub @@ -261,6 +262,28 @@ def make_entry(args, allow_negative_stock=False, via_landed_cost_voucher=False): return sle +# A repost waits this long for another repost's per-(item, warehouse) gate before giving up. Kept +# well under the 1800s repost job timeout so a wait can't burn the whole budget, and short enough +# that a contended worker re-queues (recoverable QueryTimeoutError) and frees the slot for other +# items instead of pinning it. +REPOST_LOCK_TIMEOUT = 300 + + +def repost_gate(item_code, warehouse): + """Serialize concurrent background reposts of the same (item, warehouse) with a session-level + advisory lock taken before the inner `... for update` row locks, so they take turns instead of + racing into a lock-order deadlock. Row locks still enforce correctness; this only cuts the + deadlock/retry churn. Scope is repost-vs-repost only -- the synchronous repost_current_voucher + submit path is deliberately not gated (blocking a submit behind a background repost would be a + worse regression) and keeps relying on the existing deadlock retry. No advisory locks, no gate.""" + # hasattr keeps this a graceful opt-in: on an ERPNext predating frappe.db.advisory_lock, fall + # back to no gate rather than raising and marking the Repost Item Valuation permanently Failed. + if frappe.db.db_type in ("postgres", "mariadb") and hasattr(frappe.db, "advisory_lock"): + # Tuple key: a colon in item_code/warehouse can't collide two distinct pairs onto one lock. + return frappe.db.advisory_lock(("stock_repost", item_code, warehouse), timeout=REPOST_LOCK_TIMEOUT) + return nullcontext() + + def repost_future_sle( items_to_be_repost=None, voucher_type=None, @@ -289,22 +312,25 @@ def repost_future_sle( while index < len(items_to_be_repost): validate_item_warehouse(items_to_be_repost[index]) - obj = update_entries_after( - { - "item_code": items_to_be_repost[index].get("item_code"), - "warehouse": items_to_be_repost[index].get("warehouse"), - "posting_date": items_to_be_repost[index].get("posting_date"), - "posting_time": items_to_be_repost[index].get("posting_time"), - "creation": items_to_be_repost[index].get("creation"), - "current_idx": index, - "items_to_be_repost": items_to_be_repost, - "repost_doc": doc, - "repost_affected_transaction": repost_affected_transaction, - "item_wh_wise_last_posted_sle": resume_item_wh_wise_last_posted_sle, - }, - allow_negative_stock=allow_negative_stock, - via_landed_cost_voucher=via_landed_cost_voucher, - ) + item_code = items_to_be_repost[index].get("item_code") + warehouse = items_to_be_repost[index].get("warehouse") + with repost_gate(item_code, warehouse): + obj = update_entries_after( + { + "item_code": item_code, + "warehouse": warehouse, + "posting_date": items_to_be_repost[index].get("posting_date"), + "posting_time": items_to_be_repost[index].get("posting_time"), + "creation": items_to_be_repost[index].get("creation"), + "current_idx": index, + "items_to_be_repost": items_to_be_repost, + "repost_doc": doc, + "repost_affected_transaction": repost_affected_transaction, + "item_wh_wise_last_posted_sle": resume_item_wh_wise_last_posted_sle, + }, + allow_negative_stock=allow_negative_stock, + via_landed_cost_voucher=via_landed_cost_voucher, + ) index += 1