From ae43f47c39c240d422692f681ea8ac278f4d66cc Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Fri, 3 Jul 2026 11:29:18 +0530 Subject: [PATCH 1/4] fix: race condition and repeatable read in process pcv - Update using child table name to avoid scanning whole table, which eventually leads to mariadb 1020 (REPEATABLE READ). - Avoid race condition in final summarization (cherry picked from commit ff6881764b843a3046934fddbbbfd8a907637ef8) --- .../process_period_closing_voucher.py | 33 +++++++++---------- 1 file changed, 15 insertions(+), 18 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 b5ca3331b71..43813fb98af 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 @@ -97,7 +97,7 @@ def start_pcv_processing(docname: str): ppcvd = qb.DocType("Process Period Closing Voucher Detail") if normal_balances := ( qb.from_(ppcvd) - .select(ppcvd.processing_date, ppcvd.report_type, ppcvd.parentfield) + .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) @@ -108,12 +108,7 @@ def start_pcv_processing(docname: str): for x in normal_balances: frappe.db.set_value( "Process Period Closing Voucher Detail", - { - "processing_date": x.processing_date, - "parent": docname, - "report_type": x.report_type, - "parentfield": x.parentfield, - }, + x.name, "status", "Running", ) @@ -124,10 +119,12 @@ def start_pcv_processing(docname: str): 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") @@ -251,7 +248,7 @@ def schedule_next_date(docname: str): ppcvd = qb.DocType("Process Period Closing Voucher Detail") if to_process := ( qb.from_(ppcvd) - .select(ppcvd.processing_date, ppcvd.report_type, ppcvd.parentfield) + .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(1) @@ -261,15 +258,11 @@ def schedule_next_date(docname: str): if not is_scheduler_inactive(): frappe.db.set_value( "Process Period Closing Voucher Detail", - { - "processing_date": to_process[0].processing_date, - "parent": docname, - "report_type": to_process[0].report_type, - "parentfield": to_process[0].parentfield, - }, + to_process[0].name, "status", "Running", ) + frappe.db.commit() frappe.enqueue( method="erpnext.accounts.doctype.process_period_closing_voucher.process_period_closing_voucher.process_individual_date", queue="long", @@ -277,6 +270,7 @@ def schedule_next_date(docname: str): is_async=True, enqueue_after_commit=True, docname=docname, + row_name=to_process[0].name, date=to_process[0].processing_date, report_type=to_process[0].report_type, parentfield=to_process[0].parentfield, @@ -441,6 +435,8 @@ def summarize_and_post_ledger_entries(docname): make_closing_entries(closing_entries, pcv.name, pcv.company, pcv.period_end_date) + 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") @@ -526,10 +522,10 @@ def build_dimension_wise_balance_dict(gl_entries): return dimension_balances -def process_individual_date(docname: str, date, report_type, parentfield): +def process_individual_date(docname: str, row_name, date, report_type, parentfield): current_date_status = frappe.db.get_value( "Process Period Closing Voucher Detail", - {"processing_date": date, "report_type": report_type, "parentfield": parentfield}, + row_name, "status", ) if current_date_status != "Running": @@ -576,17 +572,18 @@ def process_individual_date(docname: str, date, report_type, parentfield): # save results frappe.db.set_value( "Process Period Closing Voucher Detail", - {"processing_date": date, "parent": docname, "report_type": report_type, "parentfield": parentfield}, + row_name, "closing_balance", frappe.json.dumps(res), ) frappe.db.set_value( "Process Period Closing Voucher Detail", - {"processing_date": date, "parent": docname, "report_type": report_type, "parentfield": parentfield}, + row_name, "status", "Completed", ) + frappe.db.commit() # chain call schedule_next_date(docname) From 1ee8904a94b21f60d46ee925de02cf0321ce5193 Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Fri, 3 Jul 2026 13:00:00 +0530 Subject: [PATCH 2/4] 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 43813fb98af..a0e59d96be4 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 @@ -86,47 +86,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() @@ -244,8 +252,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) @@ -262,7 +270,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", @@ -435,7 +447,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") @@ -583,7 +598,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) From 6945e1292b58348210280b4519f1fc3f37d5cea0 Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Fri, 3 Jul 2026 15:51:18 +0530 Subject: [PATCH 3/4] refactor: prevent whole table scan while scheduling next date - helps in concurrency isolation (cherry picked from commit 21f4603144d3ffbe5e1b871dbaa02631e61919c2) --- .../process_period_closing_voucher_detail.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/erpnext/accounts/doctype/process_period_closing_voucher_detail/process_period_closing_voucher_detail.py b/erpnext/accounts/doctype/process_period_closing_voucher_detail/process_period_closing_voucher_detail.py index f3a8302ac5b..0e0b905c96a 100644 --- a/erpnext/accounts/doctype/process_period_closing_voucher_detail/process_period_closing_voucher_detail.py +++ b/erpnext/accounts/doctype/process_period_closing_voucher_detail/process_period_closing_voucher_detail.py @@ -1,7 +1,7 @@ # Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors # For license information, please see license.txt -# import frappe +import frappe from frappe.model.document import Document @@ -24,3 +24,10 @@ class ProcessPeriodClosingVoucherDetail(Document): # end: auto-generated types pass + + +def on_doctype_update(): + frappe.db.add_index( + "Process Period Closing Voucher Detail", + ["parent", "status", "parentfield", "idx", "processing_date"], + ) From dcea09ea8f2a30f621cb364482899432bd96ab2d Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Fri, 3 Jul 2026 17:04:51 +0530 Subject: [PATCH 4/4] chore: linter fix (cherry picked from commit a9ffdac8062de9b2e2e68a9c457e9e96e8ced37b) --- .../process_period_closing_voucher.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 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 a0e59d96be4..6315560b89f 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 @@ -125,7 +125,7 @@ def initialize_parallel_threads(docname: str): # keep transaction on PPCV and PPCVD short # prevents concurrency errors - REPEATABLE READ if not frappe.in_test: - frappe.db.commit() + frappe.db.commit() # nosemgrep else: frappe.db.set_value("Process Period Closing Voucher", docname, "status", "Completed") @@ -273,7 +273,7 @@ def schedule_next_date(docname: str): # keep transaction on PPCV and PPCVD short # prevents concurrency errors - REPEATABLE READ if not frappe.in_test: - frappe.db.commit() + frappe.db.commit() # nosemgrep frappe.enqueue( method="erpnext.accounts.doctype.process_period_closing_voucher.process_period_closing_voucher.process_individual_date", @@ -450,7 +450,7 @@ def summarize_and_post_ledger_entries(docname): # keep transaction on PPCV and PPCVD short # prevents concurrency errors - REPEATABLE READ if not frappe.in_test: - frappe.db.commit() + frappe.db.commit() # nosemgrep frappe.db.set_value("Period Closing Voucher", pcv.name, "gle_processing_status", "Completed") frappe.db.set_value("Process Period Closing Voucher", docname, "status", "Completed") @@ -600,7 +600,7 @@ def process_individual_date(docname: str, row_name, date, report_type, parentfie ) # commit heavy computation before touching PPCV or PPCVD if not frappe.in_test: - frappe.db.commit() + frappe.db.commit() # nosemgrep # chain call schedule_next_date(docname)