diff --git a/erpnext/accounts/utils.py b/erpnext/accounts/utils.py index d6cda9cf548..9b50751e95a 100644 --- a/erpnext/accounts/utils.py +++ b/erpnext/accounts/utils.py @@ -13,6 +13,7 @@ from frappe.desk.reportview import build_match_conditions from frappe.model.meta import get_field_precision from frappe.model.naming import determine_consecutive_week_number from frappe.query_builder import AliasedQuery, Case, Criterion, Field, Table +from frappe.query_builder.custom import ConstantColumn from frappe.query_builder.functions import Count, IfNull, Max, Min, Round, Sum from frappe.query_builder.utils import DocType from frappe.utils import ( @@ -1791,9 +1792,10 @@ def get_future_stock_vouchers(posting_date, posting_time, for_warehouses=None, f # transaction can't modify them mid-flight (the original DISTINCT ... FOR UPDATE did this). # MariaDB carries the lock on the grouped query below; postgres rejects FOR UPDATE alongside # GROUP BY, so lock the matching rows in a separate pass first -- the row locks are held until - # the surrounding transaction ends, giving the same protection. + # the surrounding transaction ends, giving the same protection. Select a constant, not the + # name: a deep backdated repost can match millions of rows and only the locks are needed. if frappe.db.db_type == "postgres": - frappe.qb.from_(SLE).select(SLE.name).where(conditions).for_update().run() + frappe.qb.from_(SLE).select(ConstantColumn(1)).where(conditions).for_update().run() # distinct vouchers in chronological order; expressed as GROUP BY + Min() so it's valid on # postgres (SELECT DISTINCT can't ORDER BY non-selected cols, and FOR UPDATE is invalid with both). diff --git a/erpnext/stock/deprecated_serial_batch.py b/erpnext/stock/deprecated_serial_batch.py index b2010411644..9e097099f01 100644 --- a/erpnext/stock/deprecated_serial_batch.py +++ b/erpnext/stock/deprecated_serial_batch.py @@ -144,12 +144,8 @@ class DeprecatedBatchNoValuation: if self.sle.name: conditions &= sle.name != self.sle.name - # Lock the scanned SLE rows so a concurrent stock posting can't change them mid-valuation. - # MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so - # lock the same rows in a separate plain SELECT first (held for the transaction). - if frappe.db.db_type == "postgres": - frappe.qb.from_(sle).select(sle.name).where(conditions).for_update().run() - + # MariaDB carries a row lock on the grouped query below; on postgres the caller + # (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse). query = ( frappe.qb.from_(sle) .select( @@ -269,13 +265,8 @@ class DeprecatedBatchNoValuation: if self.sle.name: conditions &= sle.name != self.sle.name - # Lock the scanned SLE rows so a concurrent stock posting can't change them mid-valuation. - # MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so - # lock the same SLE rows in a separate plain SELECT first. The batch.use_batchwise_valuation - # refinement below only narrows the set, so locking without the join is a safe superset. - if frappe.db.db_type == "postgres": - frappe.qb.from_(sle).select(sle.name).where(conditions).for_update().run() - + # MariaDB carries a row lock on the grouped query below; on postgres the caller + # (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse). query = ( frappe.qb.from_(sle) .inner_join(batch) @@ -402,21 +393,8 @@ class DeprecatedBatchNoValuation: conditions &= bundle.name != self.sle.serial_and_batch_bundle conditions &= bundle.voucher_type != "Pick List" - # Lock the scanned bundle rows so a concurrent stock posting can't change them mid-valuation. - # MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so - # lock the same rows in a separate plain SELECT first (the batch.use_batchwise_valuation - # refinement below only narrows the set, so omitting that join is a safe superset). - if frappe.db.db_type == "postgres": - ( - frappe.qb.from_(bundle) - .inner_join(bundle_child) - .on(bundle.name == bundle_child.parent) - .select(bundle_child.name) - .where(conditions) - .for_update() - .run() - ) - + # MariaDB carries a row lock on the grouped query below; on postgres the caller + # (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse). query = ( frappe.qb.from_(bundle) .inner_join(bundle_child) diff --git a/erpnext/stock/doctype/pick_list/pick_list.py b/erpnext/stock/doctype/pick_list/pick_list.py index 846c020de72..53e515b0f8e 100644 --- a/erpnext/stock/doctype/pick_list/pick_list.py +++ b/erpnext/stock/doctype/pick_list/pick_list.py @@ -952,10 +952,22 @@ def get_picked_items_qty(items, contains_packed_items=False) -> list[dict]: ) # Lock the picked-qty rows so a concurrent pick can't change them mid-transaction. MariaDB carries - # the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so lock the same rows - # in a separate plain SELECT first (held for the transaction). + # the lock on the grouped query (its gap locks also block rows other in-flight picks are about to + # submit); postgres has no gap locks, so first serialize on the referenced SO/packed item rows + # (they always exist), then lock the matching picked rows in a separate plain SELECT. if frappe.db.db_type == "postgres": - frappe.qb.from_(pi_item).select(pi_item.name).where(conditions).for_update().run() + parent = frappe.qb.DocType("Packed Item" if contains_packed_items else "Sales Order Item") + ( + frappe.qb.from_(parent) + .select(parent.name) + .where(parent.name.isin(items)) + .orderby(parent.name) + .for_update() + .run() + ) + frappe.qb.from_(pi_item).select(pi_item.name).where(conditions).orderby( + pi_item.name + ).for_update().run() else: query = query.for_update() diff --git a/erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py b/erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py index 4ad8a4f4136..a491c805aa3 100644 --- a/erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py +++ b/erpnext/stock/doctype/serial_and_batch_bundle/test_serial_and_batch_bundle.py @@ -201,6 +201,33 @@ class TestSerialandBatchBundle(ERPNextTestSuite): self.assertEqual(flt(stock_value_difference, 2), -5000) + def test_outward_batch_valuation_takes_transaction_advisory_lock(self): + if frappe.db.db_type != "postgres": + return + + from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note + from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt + + item_code = make_item( + properties={ + "has_batch_no": 1, + "create_new_batch": 1, + "batch_number_series": "TEST-ADV-LCK-.#####", + "is_stock_item": 1, + }, + ).name + + make_purchase_receipt(item_code=item_code, warehouse="_Test Warehouse - _TC", qty=5, rate=100) + + def held_advisory_locks(): + return frappe.db.sql( + "SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND pid = pg_backend_pid()" + )[0][0] + + before = held_advisory_locks() + create_delivery_note(item_code=item_code, warehouse="_Test Warehouse - _TC", qty=2, rate=200) + self.assertGreater(held_advisory_locks(), before) + def test_old_batch_valuation(self): frappe.flags.ignore_serial_batch_bundle_validation = True frappe.flags.use_serial_and_batch_fields = True diff --git a/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py b/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py index ad6e965b186..3dbdc2419c9 100644 --- a/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py +++ b/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py @@ -716,10 +716,19 @@ def get_available_qty_to_reserve( conditions &= sre.name != ignore_sre # Lock the rows being aggregated so a concurrent reservation can't change them mid-transaction. - # MariaDB carries the lock on the aggregate query itself; postgres rejects FOR UPDATE with an - # aggregate, so on postgres lock the same rows in a separate plain SELECT first (held for the txn). + # MariaDB carries the lock on the aggregate query itself (its gap locks also serialize two + # FIRST reservations, when no SRE rows exist yet); postgres has no gap locks, so gate on the + # Bin row (exists once there is stock), then lock the matching SREs in a plain SELECT. if frappe.db.db_type == "postgres": - frappe.qb.from_(sre).select(sre.name).where(conditions).for_update().run() + bin_table = frappe.qb.DocType("Bin") + ( + frappe.qb.from_(bin_table) + .select(bin_table.name) + .where((bin_table.item_code == item_code) & (bin_table.warehouse == warehouse)) + .for_update() + .run() + ) + frappe.qb.from_(sre).select(sre.name).where(conditions).orderby(sre.name).for_update().run() query = ( frappe.qb.from_(sre) diff --git a/erpnext/stock/serial_batch_bundle.py b/erpnext/stock/serial_batch_bundle.py index 5eef60dcc1a..56812b1bfed 100644 --- a/erpnext/stock/serial_batch_bundle.py +++ b/erpnext/stock/serial_batch_bundle.py @@ -821,6 +821,15 @@ class BatchNoValuation(DeprecatedBatchNoValuation): "Serial and Batch Bundle", self.sle.serial_and_batch_bundle, "total_amount" ) else: + # Serialize concurrent valuations of this (item, warehouse) on postgres. MariaDB's + # grouped FOR UPDATE + gap locks do this via the history reads below; postgres has no + # gap locks, and row-locking the whole history writes a lock marker on every tuple -- + # a txn-scoped advisory lock (released at commit/rollback) serializes without either. + if frappe.db.db_type == "postgres": + frappe.db.transaction_advisory_lock( + ("batch-valuation", self.sle.item_code, self.sle.warehouse) + ) + entries = self.get_batch_stock_before_date() self.stock_value_change = 0.0 self.batch_avg_rate = defaultdict(float) @@ -869,12 +878,9 @@ class BatchNoValuation(DeprecatedBatchNoValuation): if timestamp_condition: conditions &= timestamp_condition - # Lock the scanned rows so a concurrent stock transaction can't change them mid-valuation. - # MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so - # lock the same rows in a separate plain SELECT first (held for the transaction). - if frappe.db.db_type == "postgres": - frappe.qb.from_(child).select(child.name).where(conditions).for_update().run() - + # MariaDB carries a row lock on the grouped query below; on postgres the caller + # (calculate_avg_rate) serializes via a txn-scoped advisory lock on (item, warehouse) + # instead of row-locking the whole history (FOR UPDATE is invalid with GROUP BY there). query = ( frappe.qb.from_(child) .select( @@ -1561,7 +1567,7 @@ def update_batch_qty(voucher_type, voucher_no, docstatus, via_landed_cost_vouche return precision = frappe.get_precision("Batch", "batch_qty") - for batch, qty in batches.items(): + for batch, qty in sorted(batches.items()): current_qty = get_batch_current_qty(batch) current_qty += flt(qty, precision) * (-1 if docstatus == 2 else 1)