fix: parallel reposting stalls between scheduler ticks (backport #57220) (#57248)

fix: parallel reposting stalls between scheduler ticks (#57220)

(cherry picked from commit 40f861c0a0)

erpnext/hooks.py

@@ -433,8 +433,6 @@ scheduler_events = {
 	"cron": {
 		"0/15 * * * *": [
 			"erpnext.manufacturing.doctype.bom_update_log.bom_update_log.resume_bom_cost_update_jobs",
-		],
-		"0/30 * * * *": [
 			"erpnext.stock.doctype.repost_item_valuation.repost_item_valuation.run_parallel_reposting",
 		],
 		# Hourly but offset by 30 minutes

erpnext/stock/doctype/repost_item_valuation/repost_item_valuation.py

@@ -600,8 +600,13 @@ def get_recipients():
 	return recipients
 
 
+REPOSTING_JOB_ID_PREFIX = "repost_item_valuation_entry_"
+
+
 def run_parallel_reposting():
-	# This function is called every 15 minutes via hooks.py
+	# This function is called every 15 minutes via hooks.py as a recovery net;
+	# each reposting job re-triggers it on completion to pick the next queued
+	# entry, so the queue drains continuously without waiting for the cron
 
 	if not frappe.db.get_single_value("Stock Reposting Settings", "enable_parallel_reposting"):
 		return
@@ -609,26 +614,17 @@ def run_parallel_reposting():
 	if not in_configured_timeslot():
 		return
 
-	items = set()
 	no_of_parallel_reposting = (
 		frappe.db.get_single_value("Stock Reposting Settings", "no_of_parallel_reposting") or 4
 	)
 
-	riv_entries = get_repost_item_valuation_entries()
+	riv_entries = get_repost_item_valuation_entries(limit=no_of_parallel_reposting * 100)
-
+	entries_in_progress = get_entries_with_active_jobs()
-	rq_jobs = frappe.get_all(
+	items = get_items_with_active_reposting(entries_in_progress)
-		"RQ Job",
-		fields=["arguments"],
-		filters={
-			"status": ("like", "%started%"),
-			"job_name": "erpnext.stock.doctype.repost_item_valuation.repost_item_valuation.execute_reposting_entry",
-		},
-	)
 
 	for row in riv_entries:
-		if rq_jobs:
+		if row.name in entries_in_progress:
-			if job_running_for_entry(row.name, rq_jobs):
+			continue
-				continue
 
 		if row.based_on != "Item and Warehouse" or row.repost_only_accounting_ledgers:
 			execute_reposting_entry(row.name)
@@ -641,12 +637,52 @@ def run_parallel_reposting():
 		if len(items) > no_of_parallel_reposting:
 			break
 
-		frappe.enqueue(
+		enqueue_reposting_entry(row.name)
-			execute_reposting_entry,
+
-			name=row.name,
+
-			queue="long",
+def enqueue_reposting_entry(name):
-			timeout=1800,
+	frappe.enqueue(
-		)
+		execute_reposting_entry,
+		name=name,
+		continue_reposting=True,
+		queue="long",
+		timeout=1800,
+		job_id=f"{REPOSTING_JOB_ID_PREFIX}{name}",
+		deduplicate=True,
+	)
+
+
+def enqueue_parallel_reposting():
+	frappe.enqueue(
+		run_parallel_reposting,
+		queue="long",
+		timeout=1800,
+		job_id="run_parallel_reposting",
+		deduplicate=True,
+	)
+
+
+def get_entries_with_active_jobs() -> set:
+	from frappe.utils.background_jobs import get_queue
+
+	queue = get_queue("long")
+	job_ids = list(queue.get_job_ids()) + list(queue.started_job_registry.get_job_ids())
+
+	prefix = f"{frappe.local.site}||{REPOSTING_JOB_ID_PREFIX}"
+	return {job_id[len(prefix) :] for job_id in job_ids if job_id.startswith(prefix)}
+
+
+def get_items_with_active_reposting(entries_in_progress) -> set:
+	if not entries_in_progress:
+		return set()
+
+	items = frappe.get_all(
+		"Repost Item Valuation",
+		filters={"name": ("in", list(entries_in_progress))},
+		pluck="item_code",
+	)
+
+	return {item_code for item_code in items if item_code}
 
 
 def repost_entries():
@@ -664,7 +700,15 @@ def repost_entries():
 		execute_reposting_entry(row.name)
 
 
-def execute_reposting_entry(name):
+def execute_reposting_entry(name, continue_reposting=False):
+	try:
+		_execute_reposting_entry(name)
+	finally:
+		if continue_reposting:
+			enqueue_parallel_reposting()
+
+
+def _execute_reposting_entry(name):
 	doc = frappe.get_doc("Repost Item Valuation", name)
 	if (
 		doc.repost_only_accounting_ledgers
@@ -679,7 +723,7 @@ def execute_reposting_entry(name):
 		doc.deduplicate_similar_repost()
 
 
-def get_repost_item_valuation_entries():
+def get_repost_item_valuation_entries(limit=None):
 	doctype = frappe.qb.DocType("Repost Item Valuation")
 
 	query = (
@@ -695,6 +739,9 @@ def get_repost_item_valuation_entries():
 		.orderby(doctype.status, order=frappe.qb.asc)
 	)
 
+	if limit:
+		query = query.limit(cint(limit))
+
 	return query.run(as_dict=True)
 
 
@@ -778,19 +825,3 @@ def get_existing_reposting_only_gl_entries(reposting_reference):
 		reposting_map[key] = d.reposting_reference
 
 	return reposting_map
-
-
-def job_running_for_entry(reposting_entry, rq_jobs):
-	for job in rq_jobs:
-		if not job.arguments:
-			continue
-
-		try:
-			job_args = json.loads(job.arguments)
-		except (TypeError, json.JSONDecodeError):
-			continue
-
-		if isinstance(job_args, dict) and job_args.get("kwargs", {}).get("name") == reposting_entry:
-			return True
-
-	return False

erpnext/stock/doctype/repost_item_valuation/test_repost_item_valuation.py

@@ -2,7 +2,7 @@
 # See license.txt
 
 
-from unittest.mock import MagicMock, call
+from unittest.mock import MagicMock, call, patch
 
 import frappe
 from frappe.utils import add_days, add_to_date, now, nowdate, today
@@ -13,7 +13,11 @@ from erpnext.controllers.stock_controller import create_item_wise_repost_entries
 from erpnext.stock.doctype.item.test_item import make_item
 from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt
 from erpnext.stock.doctype.repost_item_valuation.repost_item_valuation import (
+	REPOSTING_JOB_ID_PREFIX,
+	enqueue_reposting_entry,
+	execute_reposting_entry,
 	in_configured_timeslot,
+	run_parallel_reposting,
 )
 from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
 from erpnext.stock.tests.test_utils import StockTestMixin
@@ -510,3 +514,80 @@ class TestRepostItemValuation(ERPNextTestSuite, StockTestMixin):
 						"name",
 					)
 				)
+
+	@ERPNextTestSuite.change_settings(
+		"Stock Reposting Settings",
+		{"item_based_reposting": 1, "enable_parallel_reposting": 1, "no_of_parallel_reposting": 2},
+	)
+	def test_parallel_reposting_excludes_items_with_active_jobs(self):
+		module = "erpnext.stock.doctype.repost_item_valuation.repost_item_valuation"
+		entries = [
+			frappe._dict(
+				name="RIV-1",
+				based_on="Item and Warehouse",
+				item_code="ITEM-A",
+				repost_only_accounting_ledgers=0,
+			),
+			frappe._dict(
+				name="RIV-2",
+				based_on="Item and Warehouse",
+				item_code="ITEM-A",
+				repost_only_accounting_ledgers=0,
+			),
+			frappe._dict(
+				name="RIV-3", based_on="Transaction", item_code=None, repost_only_accounting_ledgers=0
+			),
+			frappe._dict(
+				name="RIV-4",
+				based_on="Item and Warehouse",
+				item_code="ITEM-B",
+				repost_only_accounting_ledgers=0,
+			),
+			frappe._dict(
+				name="RIV-5",
+				based_on="Item and Warehouse",
+				item_code="ITEM-C",
+				repost_only_accounting_ledgers=0,
+			),
+		]
+
+		with (
+			patch(f"{module}.get_repost_item_valuation_entries", return_value=entries) as entries_mock,
+			patch(f"{module}.get_entries_with_active_jobs", return_value={"RIV-1"}),
+			patch(f"{module}.get_items_with_active_reposting", return_value={"ITEM-A"}),
+			patch(f"{module}.execute_reposting_entry") as execute_mock,
+			patch(f"{module}.enqueue_reposting_entry") as enqueue_mock,
+		):
+			run_parallel_reposting()
+
+		entries_mock.assert_called_once_with(limit=200)
+		execute_mock.assert_called_once_with("RIV-3")
+		enqueue_mock.assert_called_once_with("RIV-4")
+
+	def test_reposting_entry_continues_with_next_batch(self):
+		module = "erpnext.stock.doctype.repost_item_valuation.repost_item_valuation"
+
+		with (
+			patch(f"{module}._execute_reposting_entry", side_effect=Exception("boom")),
+			patch(f"{module}.enqueue_parallel_reposting") as chain_mock,
+		):
+			self.assertRaises(Exception, execute_reposting_entry, "RIV-X", continue_reposting=True)
+
+		chain_mock.assert_called_once()
+
+		with (
+			patch(f"{module}._execute_reposting_entry"),
+			patch(f"{module}.enqueue_parallel_reposting") as chain_mock,
+		):
+			execute_reposting_entry("RIV-X")
+
+		chain_mock.assert_not_called()
+
+	def test_enqueue_reposting_entry_is_deduplicated(self):
+		with patch("frappe.enqueue") as enqueue_mock:
+			enqueue_reposting_entry("RIV-X")
+
+		kwargs = enqueue_mock.call_args.kwargs
+		self.assertEqual(kwargs["job_id"], f"{REPOSTING_JOB_ID_PREFIX}RIV-X")
+		self.assertTrue(kwargs["deduplicate"])
+		self.assertTrue(kwargs["continue_reposting"])