Fix consume_build_stock (#13003)

Fixes concurrent or repeated calls to consume_build_stock, making this method idempotent

Also adds regression testing
This commit is contained in:
Oliver authored and GitHub committed 2026-10-13 16:58:25 +11:00
1 parent e0f3d293d3
commit fa04bcc455
4 files changed
+184 -2

No files matched your search

+15 -1
View File
@@ -1011,14 +1011,28 @@ class BuildConsume(BuildOrderContextMixin, CreateAPI):
lines = data.get('lines', []) lines = data.get('lines', [])
notes = data.get('notes', '') 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 # Offload the task to the background worker
task_id = offload_task( task_id = offload_task(
consume_build_stock, consume_build_stock,
build.pk, 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}, items={item['build_item'].pk: item['quantity'] for item in items},
user_id=self.request.user.pk, user_id=self.request.user.pk,
notes=notes, notes=notes,
expected=expected,
) )
response = common.serializers.TaskDetailSerializer.from_task(task_id).data response = common.serializers.TaskDetailSerializer.from_task(task_id).data
+30
View File
@@ -668,6 +668,7 @@ class Build(
quantities: Optional[dict] = None, quantities: Optional[dict] = None,
notes: Optional[str] = None, notes: Optional[str] = None,
user: Optional[User] = None, user: Optional[User] = None,
expected_quantities: Optional[dict] = None,
): ):
"""Complete the allocation of stock for the given build lines. """Complete the allocation of stock for the given build lines.
@@ -676,6 +677,12 @@ class Build(
quantities: Optional dict of quantities to allocate quantities: Optional dict of quantities to allocate
notes: Optional notes for the allocation notes: Optional notes for the allocation
user: The user completing 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: Notes:
This unrolls what would otherwise be a per-BuildItem call, so that the underlying StockItem, 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 # Lock the distinct StockItem rows (in pk order, to avoid deadlocks) and
# re-read their quantities, so the clamping arithmetic below operates on # re-read their quantities, so the clamping arithmetic below operates on
# the current committed values. Stock adjustments elsewhere lock these # the current committed values. Stock adjustments elsewhere lock these
+12 -1
View File
@@ -40,6 +40,7 @@ def consume_build_stock(
lines: Optional[list[int]] = None, lines: Optional[list[int]] = None,
items: Optional[dict] = None, items: Optional[dict] = None,
user_id: int | None = None, user_id: int | None = None,
expected: Optional[dict] = None,
**kwargs, **kwargs,
): ):
"""Consume stock for the specified BuildOrder. """Consume stock for the specified BuildOrder.
@@ -49,6 +50,10 @@ def consume_build_stock(
lines: Optional list of BuildLine IDs to consume lines: Optional list of BuildLine IDs to consume
items: Optional dict of BuildItem IDs (and quantities)to consume items: Optional dict of BuildItem IDs (and quantities)to consume
user_id: The ID of the user who initiated the stock consumption 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 from build.models import Build, BuildItem
@@ -71,7 +76,13 @@ def consume_build_stock(
.distinct() .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') @tracer.start_as_current_span('complete_build_allocations')
+127
View File
@@ -2358,6 +2358,133 @@ class BuildConsumeTest(BuildAPITest):
for line in self.build.build_lines.all(): for line in self.build.build_lines.all():
self.assertEqual(line.consumed, 100) 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): def test_consume_many_lines(self):
"""Test consuming stock against a build order with a large number of BOM lines. """Test consuming stock against a build order with a large number of BOM lines.