From fa04bcc455597735e3e703148486ce9fdb2ee95b Mon Sep 17 00:00:00 2001 From: Oliver Date: Tue, 13 Oct 2026 16:58:25 +1100 Subject: [PATCH] Fix consume_build_stock (#13003) Fixes concurrent or repeated calls to consume_build_stock, making this method idempotent Also adds regression testing --- src/backend/InvenTree/build/api.py | 16 ++- src/backend/InvenTree/build/models.py | 30 ++++++ src/backend/InvenTree/build/tasks.py | 13 ++- src/backend/InvenTree/build/test_api.py | 127 ++++++++++++++++++++++++ 4 files changed, 184 insertions(+), 2 deletions(-) diff --git a/src/backend/InvenTree/build/api.py b/src/backend/InvenTree/build/api.py index b440b2d876..2b0e82c358 100644 --- a/src/backend/InvenTree/build/api.py +++ b/src/backend/InvenTree/build/api.py @@ -1011,14 +1011,28 @@ class BuildConsume(BuildOrderContextMixin, CreateAPI): lines = data.get('lines', []) notes = data.get('notes', '') + line_ids = [line['build_line'].pk for line in lines] + item_ids = [item['build_item'].pk for item in items] + + # Snapshot the current allocated quantity of every affected BuildItem. + # The task only consumes against build items which still match this snapshot, + # making it idempotent against task redelivery (and stale double-submission) + expected = dict( + BuildItem.objects + .filter(Q(build_line__pk__in=line_ids) | Q(pk__in=item_ids)) + .filter(build_line__build=build) + .values_list('pk', 'quantity') + ) + # Offload the task to the background worker task_id = offload_task( consume_build_stock, build.pk, - lines=[line['build_line'].pk for line in lines], + lines=line_ids, items={item['build_item'].pk: item['quantity'] for item in items}, user_id=self.request.user.pk, notes=notes, + expected=expected, ) response = common.serializers.TaskDetailSerializer.from_task(task_id).data diff --git a/src/backend/InvenTree/build/models.py b/src/backend/InvenTree/build/models.py index 279c9fd88d..1ade9f3045 100644 --- a/src/backend/InvenTree/build/models.py +++ b/src/backend/InvenTree/build/models.py @@ -668,6 +668,7 @@ class Build( quantities: Optional[dict] = None, notes: Optional[str] = None, user: Optional[User] = None, + expected_quantities: Optional[dict] = None, ): """Complete the allocation of stock for the given build lines. @@ -676,6 +677,12 @@ class Build( quantities: Optional dict of quantities to allocate notes: Optional notes for the allocation user: The user completing the allocation + expected_quantities: Optional dict of {BuildItem pk: quantity}, a snapshot of + the allocated quantity of each build item at the time the request was made. + If provided, only build items in this dict whose current quantity still matches + the snapshot are processed - any other build item is skipped. This makes the + operation idempotent: a repeated (e.g. redelivered) request observes the + already-reduced quantities and does nothing. Notes: This unrolls what would otherwise be a per-BuildItem call, so that the underlying StockItem, @@ -700,6 +707,29 @@ class Build( ) ) + if expected_quantities is not None: + # Discard any build item which has changed since the request was made. + # This check is performed under the build row lock (above), so the + # build item quantities read here are current + expected = { + int(pk): decimal.Decimal(str(qty)) + for pk, qty in expected_quantities.items() + } + + stale = [ + item.pk + for item in build_items + if expected.get(item.pk) != item.quantity + ] + + if stale: + logger.warning( + 'Skipping consumption of BuildItems for BuildOrder <%s> - allocated quantity changed since request: %s', + self.pk, + stale, + ) + build_items = [item for item in build_items if item.pk not in stale] + # Lock the distinct StockItem rows (in pk order, to avoid deadlocks) and # re-read their quantities, so the clamping arithmetic below operates on # the current committed values. Stock adjustments elsewhere lock these diff --git a/src/backend/InvenTree/build/tasks.py b/src/backend/InvenTree/build/tasks.py index f1b5e425b4..612c056c77 100644 --- a/src/backend/InvenTree/build/tasks.py +++ b/src/backend/InvenTree/build/tasks.py @@ -40,6 +40,7 @@ def consume_build_stock( lines: Optional[list[int]] = None, items: Optional[dict] = None, user_id: int | None = None, + expected: Optional[dict] = None, **kwargs, ): """Consume stock for the specified BuildOrder. @@ -49,6 +50,10 @@ def consume_build_stock( lines: Optional list of BuildLine IDs to consume items: Optional dict of BuildItem IDs (and quantities)to consume user_id: The ID of the user who initiated the stock consumption + expected: Optional dict of BuildItem IDs (and their allocated quantities) at the + time of the request. Build items which no longer match are skipped, so that a + redelivered task does not consume stock a second time. + If not provided (e.g. a task queued by an older version), no check is performed. """ from build.models import Build, BuildItem @@ -71,7 +76,13 @@ def consume_build_stock( .distinct() ) - build.complete_allocations(build_items, quantities=items, notes=notes, user=user) + build.complete_allocations( + build_items, + quantities=items, + notes=notes, + user=user, + expected_quantities=expected, + ) @tracer.start_as_current_span('complete_build_allocations') diff --git a/src/backend/InvenTree/build/test_api.py b/src/backend/InvenTree/build/test_api.py index 4e4a33d14a..70fab6608d 100644 --- a/src/backend/InvenTree/build/test_api.py +++ b/src/backend/InvenTree/build/test_api.py @@ -2358,6 +2358,133 @@ class BuildConsumeTest(BuildAPITest): for line in self.build.build_lines.all(): self.assertEqual(line.consumed, 100) + def consume_snapshot(self) -> dict: + """Return the {BuildItem pk: quantity} snapshot which the consume API passes to the task.""" + return dict(self.build.allocated_stock.values_list('pk', 'quantity')) + + def test_consume_items_redelivery(self): + """A redelivered partial-quantity consume task must not consume stock twice.""" + from build.tasks import consume_build_stock + + self.allocate_stock() + + items = {item.pk: 30 for item in self.build.allocated_stock.all()} + expected = self.consume_snapshot() + + consume_build_stock(self.build.pk, items=items, expected=expected) + + n_tracking = StockItemTracking.objects.count() + n_stock = StockItem.objects.count() + + # Simulate redelivery of the exact same task payload + consume_build_stock(self.build.pk, items=items, expected=expected) + + self.assertEqual(StockItemTracking.objects.count(), n_tracking) + self.assertEqual(StockItem.objects.count(), n_stock) + self.assertEqual(self.build.consumed_stock.count(), 3) + + for item in self.build.allocated_stock.all(): + self.assertEqual(item.quantity, 70) + + for line in self.build.build_lines.all(): + self.assertEqual(line.consumed, 30) + + for si in self.stock_items: + si.refresh_from_db() + self.assertEqual(si.quantity, 970) + + def test_consume_lines_redelivery(self): + """A redelivered line consume task must not consume stock twice. + + The consumed quantity is clamped by the stock item quantity here, so a + (partially consumed) build item remains after the first run. + """ + from build.tasks import consume_build_stock + + self.allocate_stock() + + # Reduce the first stock item below the allocated quantity + StockItem.objects.filter(pk=self.stock_items[0].pk).update(quantity=40) + + lines = list(self.build.build_lines.values_list('pk', flat=True)) + expected = self.consume_snapshot() + + consume_build_stock(self.build.pk, lines=lines, expected=expected) + + remaining = self.build.allocated_stock.get() + self.assertEqual(remaining.quantity, 60) + + n_tracking = StockItemTracking.objects.count() + + # Simulate redelivery of the exact same task payload + consume_build_stock(self.build.pk, lines=lines, expected=expected) + + self.assertEqual(StockItemTracking.objects.count(), n_tracking) + + remaining.refresh_from_db() + self.assertEqual(remaining.quantity, 60) + self.assertEqual(remaining.build_line.consumed, 40) + + def test_consume_without_snapshot(self): + """Tasks queued without an 'expected' snapshot retain the previous behavior.""" + from build.tasks import consume_build_stock + + self.allocate_stock() + + items = {item.pk: 30 for item in self.build.allocated_stock.all()} + + consume_build_stock(self.build.pk, items=items) + consume_build_stock(self.build.pk, items=items) + + for line in self.build.build_lines.all(): + self.assertEqual(line.consumed, 60) + + def test_consume_stale_snapshot(self): + """A build item modified after the request was made is not consumed.""" + from build.tasks import consume_build_stock + + self.allocate_stock() + + items = {item.pk: 30 for item in self.build.allocated_stock.all()} + expected = self.consume_snapshot() + + # Another user modifies one allocation before the task runs + stale = self.build.allocated_stock.first() + BuildItem.objects.filter(pk=stale.pk).update(quantity=50) + + consume_build_stock(self.build.pk, items=items, expected=expected) + + stale.refresh_from_db() + self.assertEqual(stale.quantity, 50) + self.assertEqual(stale.build_line.consumed, 0) + + # The other (unmodified) build items are still consumed + for item in self.build.allocated_stock.exclude(pk=stale.pk): + self.assertEqual(item.quantity, 70) + self.assertEqual(item.build_line.consumed, 30) + + def test_consume_items_repeated_requests(self): + """Separate API requests for partial quantities are each applied.""" + self.allocate_stock() + + url = reverse('api-build-consume', kwargs={'pk': self.build.pk}) + + for _ in range(2): + data = { + 'items': [ + {'build_item': item.pk, 'quantity': 30} + for item in self.build.allocated_stock.all() + ] + } + + self.post(url, data, expected_code=200) + + for item in self.build.allocated_stock.all(): + self.assertEqual(item.quantity, 40) + + for line in self.build.build_lines.all(): + self.assertEqual(line.consumed, 60) + def test_consume_many_lines(self): """Test consuming stock against a build order with a large number of BOM lines.