diff --git a/erpnext/hooks.py b/erpnext/hooks.py index d614d8b6356..fbc8d6c8687 100644 --- a/erpnext/hooks.py +++ b/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 diff --git a/erpnext/stock/doctype/repost_item_valuation/repost_item_valuation.py b/erpnext/stock/doctype/repost_item_valuation/repost_item_valuation.py index d741d923547..160a887d851 100644 --- a/erpnext/stock/doctype/repost_item_valuation/repost_item_valuation.py +++ b/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() - - rq_jobs = frappe.get_all( - "RQ Job", - fields=["arguments"], - filters={ - "status": ("like", "%started%"), - "job_name": "erpnext.stock.doctype.repost_item_valuation.repost_item_valuation.execute_reposting_entry", - }, - ) + riv_entries = get_repost_item_valuation_entries(limit=no_of_parallel_reposting * 100) + entries_in_progress = get_entries_with_active_jobs() + items = get_items_with_active_reposting(entries_in_progress) for row in riv_entries: - if rq_jobs: - if job_running_for_entry(row.name, rq_jobs): - continue + if row.name in entries_in_progress: + 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( - execute_reposting_entry, - name=row.name, - queue="long", - timeout=1800, - ) + enqueue_reposting_entry(row.name) + + +def enqueue_reposting_entry(name): + 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 diff --git a/erpnext/stock/doctype/repost_item_valuation/test_repost_item_valuation.py b/erpnext/stock/doctype/repost_item_valuation/test_repost_item_valuation.py index 8241ca6b6ed..a27c8d49ea2 100644 --- a/erpnext/stock/doctype/repost_item_valuation/test_repost_item_valuation.py +++ b/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"])