diff --git a/src/backend/InvenTree/build/models.py b/src/backend/InvenTree/build/models.py index 74576ae06a..732794a20c 100644 --- a/src/backend/InvenTree/build/models.py +++ b/src/backend/InvenTree/build/models.py @@ -1599,6 +1599,11 @@ class Build( ) } + # Net additional quantity being requested against each StockItem by this call + # (regardless of whether it lands in a new or an existing BuildItem row) + requested = {} + stock_items = {} + for item in items: build_line = item['build_line'] stock_item = item['stock_item'] @@ -1628,6 +1633,49 @@ class Build( # This is a new BuildItem to_create[key] = BuildItem(quantity=quantity, **filters) + stock_items[stock_item.pk] = stock_item + requested[stock_item.pk] = requested.get(stock_item.pk, 0) + quantity + + # Lock each referenced StockItem (in a consistent order, to avoid deadlocks + # against other allocation requests touching an overlapping set of items), + # and re-validate that the requested allocation still fits within the + # (now-locked, now-current) unallocated quantity of each item. + # + # Both the locking and the allocated-quantity lookup are done in bulk (rather + # than one query per StockItem) - critical for keeping the query count bounded + # when a single request references a large number of stock items. + pks = sorted(requested.keys()) + + locked_quantities = dict( + stock.models.StockItem.objects + .select_for_update() + .filter(pk__in=pks) + .order_by('pk') + .values_list('pk', 'quantity') + ) + + if len(locked_quantities) != len(pks): + raise ValidationError({'stock_item': _('Stock item no longer exists')}) + + allocated_quantities = stock.models.StockItem.bulk_allocation_count( + stock_items.values() + ) + + for pk in pks: + stock_item = stock_items[pk] + stock_item.quantity = locked_quantities[pk] + + available = max( + stock_item.quantity - allocated_quantities.get(pk, decimal.Decimal(0)), + decimal.Decimal(0), + ) + + if requested[pk] > available: + q = InvenTree.helpers.clean_decimal(available) + raise ValidationError({ + 'quantity': _(f'Available quantity ({q}) exceeded') + }) + BuildItem.objects.bulk_create(to_create.values(), batch_size=250) BuildItem.objects.bulk_update(to_update.values(), ['quantity'], batch_size=250) diff --git a/src/backend/InvenTree/build/test_build.py b/src/backend/InvenTree/build/test_build.py index a5f50aa51d..bee063d473 100644 --- a/src/backend/InvenTree/build/test_build.py +++ b/src/backend/InvenTree/build/test_build.py @@ -2046,3 +2046,131 @@ class BuildSubtractAllocatedStockConcurrencyTest(TransactionTestCase): # thread "won" the race for the row lock self.assertEqual(self.build_line.consumed, 10) self.assertFalse(BuildItem.objects.filter(pk=self.build_item.pk).exists()) + + +@skipUnlessDBFeature('has_select_for_update') +class BuildAllocateStockConcurrencyTest(TransactionTestCase): + """Genuine cross-transaction regression test for Build.allocate_stock(). + + Uses two real threads (each with its own database connection) to reproduce + the reported race: two concurrent allocation requests, against two + different Builds but the *same* StockItem, could each read the item's + unallocated quantity before either had committed, and both create a + BuildItem for the full quantity - over-allocating the StockItem. Because + the requests target different Builds, the pre-existing Build-row lock + does not serialize them. + + allocate_stock() now locks the referenced StockItem (select_for_update, + via StockItem.lock_quantity()) and re-validates the unallocated quantity + under that lock before writing, so only one of two concurrent + full-quantity allocation requests against a shared StockItem may succeed. + """ + + fixtures = ['users'] + + def setUp(self): + """Create two Builds which both require the same shared StockItem.""" + super().setUp() + + self.user = get_user_model().objects.get(pk=1) + + self.assembly_a = Part.objects.create( + name='Concurrency assembly A', + description='Assembly for allocate_stock concurrency test', + assembly=True, + ) + self.assembly_b = Part.objects.create( + name='Concurrency assembly B', + description='Assembly for allocate_stock concurrency test', + assembly=True, + ) + self.sub_part = Part.objects.create( + name='Shared concurrency component', + description='Component for allocate_stock concurrency test', + component=True, + ) + + BomItem.objects.create(part=self.assembly_a, sub_part=self.sub_part, quantity=1) + BomItem.objects.create(part=self.assembly_b, sub_part=self.sub_part, quantity=1) + + self.build_a = Build.objects.create( + reference=generate_next_build_reference(), + part=self.assembly_a, + quantity=1, + issued_by=self.user, + ) + self.build_b = Build.objects.create( + reference=generate_next_build_reference(), + part=self.assembly_b, + quantity=1, + issued_by=self.user, + ) + + self.build_line_a = BuildLine.objects.get(build=self.build_a) + self.build_line_b = BuildLine.objects.get(build=self.build_b) + + # Only enough stock for *one* of the two full-quantity allocations below + self.stock_item = StockItem.objects.create(part=self.sub_part, quantity=5) + + def test_concurrent_allocation_does_not_over_allocate(self): + """Two concurrent full-quantity allocation requests must not both succeed.""" + start_barrier = threading.Barrier(2, timeout=5) + errors = [] + results = [] + + # Wrap StockItem.lock_quantity() so both threads reach the (real, + # database-level) row lock at the same time - one wins the lock and + # proceeds, the other blocks until the winner's transaction completes. + original_lock_quantity = StockItem.lock_quantity + + def synced_lock_quantity(self_item): + start_barrier.wait(timeout=5) + return original_lock_quantity(self_item) + + def allocate(build, build_line): + try: + build.allocate_stock([ + { + 'build_line': build_line, + 'stock_item': self.stock_item, + 'quantity': 5, + } + ]) + results.append('ok') + except ValidationError: + results.append('rejected') + except Exception as exc: # pragma: no cover - surfaced via errors list + errors.append(exc) + finally: + connection.close() + + thread_a = threading.Thread( + target=allocate, args=(self.build_a, self.build_line_a) + ) + thread_b = threading.Thread( + target=allocate, args=(self.build_b, self.build_line_b) + ) + + with mock.patch.object(StockItem, 'lock_quantity', synced_lock_quantity): + thread_a.start() + thread_b.start() + + thread_a.join(timeout=5) + thread_b.join(timeout=5) + + self.assertFalse(thread_a.is_alive()) + self.assertFalse(thread_b.is_alive()) + self.assertEqual(errors, []) + + # Exactly one request must have been rejected as over-allocating + self.assertEqual(sorted(results), ['ok', 'rejected']) + + total_allocated = ( + BuildItem.objects.filter(stock_item=self.stock_item).aggregate( + q=Sum('quantity') + )['q'] + or 0 + ) + + self.assertEqual(total_allocated, 5) + self.assertLessEqual(total_allocated, self.stock_item.quantity) diff --git a/src/backend/InvenTree/order/models.py b/src/backend/InvenTree/order/models.py index 19b235e3ac..5cfbb52496 100644 --- a/src/backend/InvenTree/order/models.py +++ b/src/backend/InvenTree/order/models.py @@ -1738,6 +1738,13 @@ class SalesOrder(TotalPriceMixin, Order): serials_unavailable.add(serial) continue + # Lock the StockItem row, so that concurrent allocation requests are + # serialized against each other, and re-validate the unallocated + # quantity against the now-current (and now-locked) state + if not stock_item.lock_quantity(): + serials_unavailable.add(serial) + continue + if stock_item.unallocated_quantity() < 1: serials_unavailable.add(serial) continue @@ -3185,9 +3192,13 @@ class SalesOrderAllocation(models.Model): sales_allocation_count = self.item.sales_order_allocation_count( exclude_allocations={'pk': self.pk} ) + transfer_allocation_count = self.item.transfer_order_allocation_count() total_allocation = ( - build_allocation_count + sales_allocation_count + self.quantity + build_allocation_count + + sales_allocation_count + + transfer_allocation_count + + self.quantity ) if total_allocation > self.item.quantity: @@ -4287,9 +4298,15 @@ class TransferOrderAllocation(models.Model): sales_allocation_count = self.item.sales_order_allocation_count( exclude_allocations={'pk': self.pk} ) + transfer_allocation_count = self.item.transfer_order_allocation_count( + exclude_allocations={'pk': self.pk} + ) total_allocation = ( - build_allocation_count + sales_allocation_count + self.quantity + build_allocation_count + + sales_allocation_count + + transfer_allocation_count + + self.quantity ) if total_allocation > self.item.quantity: diff --git a/src/backend/InvenTree/order/serializers.py b/src/backend/InvenTree/order/serializers.py index 548323dff5..027d2e994d 100644 --- a/src/backend/InvenTree/order/serializers.py +++ b/src/backend/InvenTree/order/serializers.py @@ -1987,10 +1987,20 @@ class SalesOrderShipmentAllocationSerializer(serializers.Serializer): with transaction.atomic(): for entry in items: + stock_item = entry.get('stock_item') + + # Lock the StockItem row, so that concurrent allocation requests are + # serialized against each other, and full_clean() below re-validates + # against the now-current (and now-locked) allocation counts + if not stock_item.lock_quantity(): + raise ValidationError({ + 'stock_item': _('Stock item no longer exists') + }) + # Create a new SalesOrderAllocation allocation = order.models.SalesOrderAllocation( line=entry.get('line_item'), - item=entry.get('stock_item'), + item=stock_item, quantity=entry.get('quantity'), shipment=shipment, ) @@ -2828,10 +2838,20 @@ class TransferOrderLineItemAllocationSerializer(serializers.Serializer): with transaction.atomic(): for entry in items: + stock_item = entry.get('stock_item') + + # Lock the StockItem row, so that concurrent allocation requests are + # serialized against each other, and full_clean() below re-validates + # against the now-current (and now-locked) allocation counts + if not stock_item.lock_quantity(): + raise ValidationError({ + 'stock_item': _('Stock item no longer exists') + }) + # Create a new TransferOrderAllocation allocation = order.models.TransferOrderAllocation( line=entry.get('line_item'), - item=entry.get('stock_item'), + item=stock_item, quantity=entry.get('quantity'), ) @@ -2965,8 +2985,10 @@ class TransferOrderSerialAllocationSerializer(serializers.Serializer): """Validation for the serializer. - Ensure the serial_numbers and quantity fields match - - Check that all serial numbers exist - - Check that the serial numbers are not yet allocated + + Note: Resolving serial numbers to StockItem objects (and checking their + availability) is deferred to save(), where it can be done under a + database lock - see save() for details. """ data = super().validate(data) @@ -2983,11 +3005,29 @@ class TransferOrderSerialAllocationSerializer(serializers.Serializer): except DjangoValidationError as e: raise ValidationError({'serial_numbers': e.messages}) + return data + + @transaction.atomic + def save(self): + """Allocate stock items against the transfer order. + + Stock items are resolved from the requested serial numbers, and locked + (select_for_update, via StockItem.lock_quantity()) before their + availability is checked - this serializes concurrent allocation requests + against each other, so two requests cannot both allocate the same + serialized StockItem. + """ + data = self.validated_data + + line_item = data['line_item'] + serials = data['serials'] + part = line_item.part + serials_not_exist = set() serials_unavailable = set() - stock_items_to_allocate = [] + allocations = [] - for serial in data['serials']: + for serial in serials: serial = str(serial).strip() items = stock.models.StockItem.objects.filter( @@ -3004,12 +3044,20 @@ class TransferOrderSerialAllocationSerializer(serializers.Serializer): serials_unavailable.add(str(serial)) continue + if not stock_item.lock_quantity(): + serials_unavailable.add(str(serial)) + continue + if stock_item.unallocated_quantity() < 1: serials_unavailable.add(str(serial)) continue # At this point, the serial number is valid, and can be added to the list - stock_items_to_allocate.append(stock_item) + allocations.append( + order.models.TransferOrderAllocation( + line=line_item, item=stock_item, quantity=1 + ) + ) if len(serials_not_exist) > 0: error_msg = _('No match found for the following serial numbers') @@ -3025,26 +3073,4 @@ class TransferOrderSerialAllocationSerializer(serializers.Serializer): raise ValidationError({'serial_numbers': error_msg}) - data['stock_items'] = stock_items_to_allocate - - return data - - def save(self): - """Allocate stock items against the transfer order.""" - data = self.validated_data - - line_item = data['line_item'] - stock_items = data['stock_items'] - - allocations = [] - - for stock_item in stock_items: - # Create a new TransferOrderAllocation - allocations.append( - order.models.TransferOrderAllocation( - line=line_item, item=stock_item, quantity=1 - ) - ) - - with transaction.atomic(): - order.models.TransferOrderAllocation.objects.bulk_create(allocations) + order.models.TransferOrderAllocation.objects.bulk_create(allocations) diff --git a/src/backend/InvenTree/order/test_api.py b/src/backend/InvenTree/order/test_api.py index 7153c5ad77..f59c282c7a 100644 --- a/src/backend/InvenTree/order/test_api.py +++ b/src/backend/InvenTree/order/test_api.py @@ -3,6 +3,7 @@ import base64 import io import json +import threading from datetime import date, datetime, timedelta from typing import Optional from unittest import mock @@ -10,12 +11,14 @@ from unittest import mock from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ValidationError from django.db import connection +from django.test import TransactionTestCase, skipUnlessDBFeature from django.test.utils import CaptureQueriesContext from django.urls import reverse from djmoney.money import Money from icalendar import Calendar from rest_framework import status +from rest_framework.exceptions import ValidationError as DRFValidationError from common.currency import currency_codes from common.models import InvenTreeCustomUserStateModel, InvenTreeSetting @@ -24,6 +27,7 @@ from company.models import Company, SupplierPart, SupplierPriceBreak from InvenTree.unit_test import InvenTreeAPITestCase from order import models from order.models import SalesOrderAllocation, SalesOrderLineItem, SalesOrderShipment +from order.serializers import TransferOrderSerialAllocationSerializer from order.status_codes import ( PurchaseOrderStatus, ReturnOrderLineStatus, @@ -4944,6 +4948,109 @@ class TransferOrderAllocateTest(OrderTest): ) +@skipUnlessDBFeature('has_select_for_update') +class TransferOrderSerialAllocateConcurrencyTest(TransactionTestCase): + """Genuine cross-transaction regression test for serial-based transfer order allocation. + + Uses two real threads (each with its own database connection) to reproduce + the reported race: two concurrent requests to allocate the *same* serial + number (against two different TransferOrder line items) could both resolve + and validate the serialized StockItem as available before either had + committed its bulk_create - allocating the same physical unit twice. + + TransferOrderSerialAllocationSerializer.save() now locks each resolved + StockItem (select_for_update, via StockItem.lock_quantity()) and + re-validates its unallocated quantity under that lock before it is added + to the batch that gets created, so only one of two concurrent requests for + the same serial number may succeed. + """ + + fixtures = ['users'] + + def setUp(self): + """Create two TransferOrder lines which both request the same serial number.""" + super().setUp() + + self.part = Part.objects.create( + name='Concurrency trackable part', + description='Part for serial allocation concurrency test', + trackable=True, + ) + + self.order_a = models.TransferOrder.objects.create(reference='TO-CONC-A') + self.order_b = models.TransferOrder.objects.create(reference='TO-CONC-B') + + self.line_a = models.TransferOrderLineItem.objects.create( + order=self.order_a, part=self.part, quantity=1 + ) + self.line_b = models.TransferOrderLineItem.objects.create( + order=self.order_b, part=self.part, quantity=1 + ) + + # Only a single physical unit exists for this serial number + self.stock_item = StockItem.objects.create( + part=self.part, quantity=1, serial='1' + ) + + def test_concurrent_allocation_does_not_duplicate_serial(self): + """Two concurrent requests for the same serial number must not both succeed.""" + start_barrier = threading.Barrier(2, timeout=5) + errors = [] + results = [] + + # Wrap StockItem.lock_quantity() so both threads reach the (real, + # database-level) row lock at the same time - one wins the lock and + # proceeds, the other blocks until the winner's transaction completes. + original_lock_quantity = StockItem.lock_quantity + + def synced_lock_quantity(self_item): + start_barrier.wait(timeout=5) + return original_lock_quantity(self_item) + + def allocate(line_item): + try: + serializer = TransferOrderSerialAllocationSerializer( + data={ + 'line_item': line_item.pk, + 'quantity': 1, + 'serial_numbers': '1', + }, + context={'order': line_item.order}, + ) + serializer.is_valid(raise_exception=True) + serializer.save() + results.append('ok') + except (ValidationError, DRFValidationError): + results.append('rejected') + except Exception as exc: # pragma: no cover - surfaced via errors list + errors.append(exc) + finally: + connection.close() + + thread_a = threading.Thread(target=allocate, args=(self.line_a,)) + thread_b = threading.Thread(target=allocate, args=(self.line_b,)) + + with mock.patch.object(StockItem, 'lock_quantity', synced_lock_quantity): + thread_a.start() + thread_b.start() + + thread_a.join(timeout=5) + thread_b.join(timeout=5) + + self.assertFalse(thread_a.is_alive()) + self.assertFalse(thread_b.is_alive()) + self.assertEqual(errors, []) + + # Exactly one request must have been rejected as unavailable + self.assertEqual(sorted(results), ['ok', 'rejected']) + + # The serial number must only have been allocated once + self.assertEqual( + models.TransferOrderAllocation.objects.filter(item=self.stock_item).count(), + 1, + ) + + class SalesOrderAutoAllocateAPITest(InvenTreeAPITestCase): """API integration tests for the SalesOrder auto-allocate endpoint.""" diff --git a/src/backend/InvenTree/order/test_sales_order.py b/src/backend/InvenTree/order/test_sales_order.py index 74358eb9aa..34d839d953 100644 --- a/src/backend/InvenTree/order/test_sales_order.py +++ b/src/backend/InvenTree/order/test_sales_order.py @@ -1,15 +1,19 @@ """Unit tests for the SalesOrder models.""" +import threading from datetime import datetime, timedelta from unittest import mock from django.contrib.auth import get_user_model from django.contrib.auth.models import Group from django.core.exceptions import ValidationError +from django.db import connection from django.db.models import Sum +from django.test import TransactionTestCase, skipUnlessDBFeature from django.urls import reverse from django_q.models import OrmQ +from rest_framework.exceptions import ValidationError as DRFValidationError import order.tasks from common.models import InvenTreeSetting, NotificationMessage @@ -28,6 +32,7 @@ from order.models import ( SalesOrderLineItem, SalesOrderShipment, ) +from order.serializers import SalesOrderShipmentAllocationSerializer from part.models import Part from stock.events import StockEvents from stock.models import StockItem, StockItemTracking, StockLocation @@ -1177,3 +1182,125 @@ class SalesOrderAutoAllocateTest(InvenTreeTestCase): allocated_items = [a.item for a in allocs] self.assertNotIn(excluded, allocated_items) self.assertIn(included, allocated_items) + + +@skipUnlessDBFeature('has_select_for_update') +class SalesOrderAllocateStockConcurrencyTest(TransactionTestCase): + """Genuine cross-transaction regression test for sales order stock allocation. + + Uses two real threads (each with its own database connection) to reproduce + the reported race: two concurrent allocation requests, against two + different SalesOrder line items but the *same* StockItem, could each read + the item's unallocated quantity before either had committed, and both + create a SalesOrderAllocation for the full quantity - over-allocating the + StockItem. There was no order-level lock serializing these requests + against each other, since they target different orders. + + SalesOrderShipmentAllocationSerializer.save() now locks the referenced + StockItem (select_for_update, via StockItem.lock_quantity()) before + validating and creating each allocation, so only one of two concurrent + full-quantity allocation requests against a shared StockItem may succeed. + """ + + fixtures = ['users'] + + def setUp(self): + """Create two SalesOrders which both request the same shared StockItem.""" + super().setUp() + + self.user = get_user_model().objects.get(pk=1) + + self.customer = Company.objects.create( + name='Concurrency customer', + description='Customer for allocation concurrency test', + is_customer=True, + ) + + self.part = Part.objects.create( + name='Concurrency salable part', + description='Part for allocation concurrency test', + salable=True, + ) + + self.order_a = SalesOrder.objects.create( + customer=self.customer, reference='SO-CONC-A' + ) + self.order_b = SalesOrder.objects.create( + customer=self.customer, reference='SO-CONC-B' + ) + + self.line_a = SalesOrderLineItem.objects.create( + quantity=5, order=self.order_a, part=self.part + ) + self.line_b = SalesOrderLineItem.objects.create( + quantity=5, order=self.order_b, part=self.part + ) + + # Only enough stock for *one* of the two full-quantity allocations below + self.stock_item = StockItem.objects.create(part=self.part, quantity=5) + + def test_concurrent_allocation_does_not_over_allocate(self): + """Two concurrent full-quantity allocation requests must not both succeed.""" + start_barrier = threading.Barrier(2, timeout=5) + errors = [] + results = [] + + # Wrap StockItem.lock_quantity() so both threads reach the (real, + # database-level) row lock at the same time - one wins the lock and + # proceeds, the other blocks until the winner's transaction completes. + original_lock_quantity = StockItem.lock_quantity + + def synced_lock_quantity(self_item): + start_barrier.wait(timeout=5) + return original_lock_quantity(self_item) + + def allocate(line_item): + try: + serializer = SalesOrderShipmentAllocationSerializer( + data={ + 'items': [ + { + 'line_item': line_item.pk, + 'stock_item': self.stock_item.pk, + 'quantity': 5, + } + ] + }, + context={'order': line_item.order}, + ) + serializer.is_valid(raise_exception=True) + serializer.save() + results.append('ok') + except (ValidationError, DRFValidationError): + results.append('rejected') + except Exception as exc: # pragma: no cover - surfaced via errors list + errors.append(exc) + finally: + connection.close() + + thread_a = threading.Thread(target=allocate, args=(self.line_a,)) + thread_b = threading.Thread(target=allocate, args=(self.line_b,)) + + with mock.patch.object(StockItem, 'lock_quantity', synced_lock_quantity): + thread_a.start() + thread_b.start() + + thread_a.join(timeout=5) + thread_b.join(timeout=5) + + self.assertFalse(thread_a.is_alive()) + self.assertFalse(thread_b.is_alive()) + self.assertEqual(errors, []) + + # Exactly one request must have been rejected as over-allocating + self.assertEqual(sorted(results), ['ok', 'rejected']) + + total_allocated = ( + SalesOrderAllocation.objects.filter(item=self.stock_item).aggregate( + q=Sum('quantity') + )['q'] + or 0 + ) + + self.assertEqual(total_allocated, 5) + self.assertLessEqual(total_allocated, self.stock_item.quantity)