[bug] Fix concurrency issues for allocation (#12478)

* Fix concurrency issues for allocation

* Remove duplicate atomic call

* Reduce query count

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)
 

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)

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:

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)

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."""
 

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)