From 019b02adcdf95a1f0c31ebff0f0440200160d9c3 Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Fri, 3 Jul 2026 13:00:00 +0530 Subject: [PATCH] fix: prevent repeatable read related concurrency errors Process Period Closing Voucher and Process Period Closing Voucher Details are trackers how the jobs are processed. Keep transactions on them very short. (cherry picked from commit 7e4045e8282714928989453529d679ff3bf4b6eb) --- .../process_period_closing_voucher.py | 99 +++++++++++-------- 1 file changed, 58 insertions(+), 41 deletions(-) diff --git a/erpnext/accounts/doctype/process_period_closing_voucher/process_period_closing_voucher.py b/erpnext/accounts/doctype/process_period_closing_voucher/process_period_closing_voucher.py index 966ded8f6fa..691948cabad 100644 --- a/erpnext/accounts/doctype/process_period_closing_voucher/process_period_closing_voucher.py +++ b/erpnext/accounts/doctype/process_period_closing_voucher/process_period_closing_voucher.py @@ -89,47 +89,55 @@ class ProcessPeriodClosingVoucher(Document): cancel_pcv_processing(self.name) +def initialize_parallel_threads(docname: str): + threads = 4 + timeout = frappe.db.get_single_value("Accounts Settings", "pcv_job_timeout") or 3600 + ppcvd = qb.DocType("Process Period Closing Voucher Detail") + + frappe.db.set_value("Process Period Closing Voucher", docname, "status", "Running") + + if normal_balances := ( + qb.from_(ppcvd) + .select(ppcvd.name, ppcvd.processing_date, ppcvd.report_type, ppcvd.parentfield) + .where(ppcvd.parent.eq(docname) & ppcvd.status.eq("Queued")) + .orderby(ppcvd.parentfield, ppcvd.idx, ppcvd.processing_date) + .limit(threads) + .for_update(skip_locked=True) + .run(as_dict=True) + ): + if not is_scheduler_inactive(): + for x in normal_balances: + frappe.db.set_value( + "Process Period Closing Voucher Detail", + x.name, + "status", + "Running", + ) + frappe.enqueue( + method="erpnext.accounts.doctype.process_period_closing_voucher.process_period_closing_voucher.process_individual_date", + queue="long", + timeout=timeout, + is_async=True, + enqueue_after_commit=True, + docname=docname, + row_name=x.name, + date=x.processing_date, + report_type=x.report_type, + parentfield=x.parentfield, + ) + # keep transaction on PPCV and PPCVD short + # prevents concurrency errors - REPEATABLE READ + if not frappe.in_test: + frappe.db.commit() + else: + frappe.db.set_value("Process Period Closing Voucher", docname, "status", "Completed") + + @frappe.whitelist() def start_pcv_processing(docname: str): if frappe.db.get_value("Process Period Closing Voucher", docname, "status") in ["Queued", "Running"]: frappe.has_permission("Process Period Closing Voucher", "write", doc=docname, throw=True) - frappe.db.set_value("Process Period Closing Voucher", docname, "status", "Running") - - timeout = frappe.db.get_single_value("Accounts Settings", "pcv_job_timeout") or 3600 - - ppcvd = qb.DocType("Process Period Closing Voucher Detail") - if normal_balances := ( - qb.from_(ppcvd) - .select(ppcvd.name, ppcvd.processing_date, ppcvd.report_type, ppcvd.parentfield) - .where(ppcvd.parent.eq(docname) & ppcvd.status.eq("Queued")) - .orderby(ppcvd.parentfield, ppcvd.idx, ppcvd.processing_date) - .limit(4) - .for_update(skip_locked=True) - .run(as_dict=True) - ): - if not is_scheduler_inactive(): - for x in normal_balances: - frappe.db.set_value( - "Process Period Closing Voucher Detail", - x.name, - "status", - "Running", - ) - frappe.enqueue( - method="erpnext.accounts.doctype.process_period_closing_voucher.process_period_closing_voucher.process_individual_date", - queue="long", - timeout=timeout, - is_async=True, - enqueue_after_commit=True, - docname=docname, - row_name=x.name, - date=x.processing_date, - report_type=x.report_type, - parentfield=x.parentfield, - ) - frappe.db.commit() - else: - frappe.db.set_value("Process Period Closing Voucher", docname, "status", "Completed") + initialize_parallel_threads(docname) @frappe.whitelist() @@ -247,8 +255,8 @@ def get_gle_for_closing_account(pcv, dimension_balance, dimensions): @frappe.whitelist() def schedule_next_date(docname: str): timeout = frappe.db.get_single_value("Accounts Settings", "pcv_job_timeout") or 3600 - ppcvd = qb.DocType("Process Period Closing Voucher Detail") + if to_process := ( qb.from_(ppcvd) .select(ppcvd.name, ppcvd.processing_date, ppcvd.report_type, ppcvd.parentfield) @@ -265,7 +273,11 @@ def schedule_next_date(docname: str): "status", "Running", ) - frappe.db.commit() + # keep transaction on PPCV and PPCVD short + # prevents concurrency errors - REPEATABLE READ + if not frappe.in_test: + frappe.db.commit() + frappe.enqueue( method="erpnext.accounts.doctype.process_period_closing_voucher.process_period_closing_voucher.process_individual_date", queue="long", @@ -438,7 +450,10 @@ def summarize_and_post_ledger_entries(docname): make_closing_entries(closing_entries, pcv.name, pcv.company, pcv.period_end_date) - frappe.db.commit() + # keep transaction on PPCV and PPCVD short + # prevents concurrency errors - REPEATABLE READ + if not frappe.in_test: + frappe.db.commit() frappe.db.set_value("Period Closing Voucher", pcv.name, "gle_processing_status", "Completed") frappe.db.set_value("Process Period Closing Voucher", docname, "status", "Completed") @@ -586,7 +601,9 @@ def process_individual_date(docname: str, row_name, date, report_type, parentfie "status", "Completed", ) - frappe.db.commit() + # commit heavy computation before touching PPCV or PPCVD + if not frappe.in_test: + frappe.db.commit() # chain call schedule_next_date(docname)