Locking fixes (#12435)
* Secure row locks when trimming allocated stock * Secure row locks when completing a SalesOrder * Regression testing * Further improvements and regression testing * Ignore locking test for sqlite
This commit is contained in:
Oliver
GitHub
parent
4739d59c0b
commit
8b009e8d9f
|
|
@ -681,6 +681,16 @@ class Build(
|
|||
"""Action to be taken when a build is completed."""
|
||||
import build.tasks
|
||||
|
||||
# Lock this build against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous completion requests
|
||||
# could both offload the completion task (duplicate events and
|
||||
# notifications; the stock-consuming side effects are themselves
|
||||
# guarded against double-processing - see Build.subtract_allocated_stock()).
|
||||
self.status = Build.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.status == BuildStatus.COMPLETE.value:
|
||||
return
|
||||
|
||||
trim_allocated_stock = kwargs.pop('trim_allocated_stock', False)
|
||||
user = kwargs.pop('user', None)
|
||||
|
||||
|
|
@ -730,6 +740,12 @@ class Build(
|
|||
|
||||
def _action_issue(self, *args, **kwargs):
|
||||
"""Perform the action to mark this order as PRODUCTION."""
|
||||
# Lock this build against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the issue side effects
|
||||
# (duplicate events and offloaded stock-check tasks).
|
||||
self.status = Build.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_issue:
|
||||
self.status = BuildStatus.PRODUCTION.value
|
||||
self.save()
|
||||
|
|
@ -757,6 +773,12 @@ class Build(
|
|||
|
||||
def _action_hold(self, *args, **kwargs):
|
||||
"""Action to be taken when a build is placed on hold."""
|
||||
# Lock this build against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the hold side effects
|
||||
# (duplicate events).
|
||||
self.status = Build.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_hold:
|
||||
self.status = BuildStatus.ON_HOLD.value
|
||||
self.save()
|
||||
|
|
@ -784,6 +806,16 @@ class Build(
|
|||
"""Action to be taken when a build is cancelled."""
|
||||
import build.tasks
|
||||
|
||||
# Lock this build against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous cancellation requests
|
||||
# could both offload the cancellation task (duplicate events and
|
||||
# notifications; the stock-consuming side effects are themselves
|
||||
# guarded against double-processing - see Build.subtract_allocated_stock()).
|
||||
self.status = Build.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.status == BuildStatus.CANCELLED.value:
|
||||
return
|
||||
|
||||
user = kwargs.pop('user', None)
|
||||
|
||||
remove_allocated_stock = kwargs.get('remove_allocated_stock', False)
|
||||
|
|
@ -986,7 +1018,17 @@ class Build(
|
|||
continue
|
||||
|
||||
# Find BuildItem objects to trim
|
||||
for item in BuildItem.objects.filter(build_line=build_line):
|
||||
# Lock these rows (in pk order, to avoid deadlocks against other
|
||||
# allocation operations) so a concurrent allocation update cannot
|
||||
# be silently overwritten by the quantity decrement below
|
||||
items = (
|
||||
BuildItem.objects
|
||||
.select_for_update()
|
||||
.filter(build_line=build_line)
|
||||
.order_by('pk')
|
||||
)
|
||||
|
||||
for item in items:
|
||||
# Previous item completed the job
|
||||
if reduce_by <= 0:
|
||||
break
|
||||
|
|
@ -1013,11 +1055,20 @@ class Build(
|
|||
"""Returns a QuerySet object of all BuildItem objects which point back to this Build."""
|
||||
return BuildItem.objects.filter(build_line__build=self)
|
||||
|
||||
@transaction.atomic
|
||||
def subtract_allocated_stock(self, user) -> None:
|
||||
"""Removes the allocated untracked items from stock."""
|
||||
# Find all BuildItem objects which point to this build
|
||||
items = self.allocated_stock.filter(
|
||||
build_line__bom_item__sub_part__trackable=False
|
||||
# Lock these rows (in pk order, to avoid deadlocks against other
|
||||
# allocation operations), so a duplicated call to this method (e.g. a
|
||||
# redelivered background task, or a completion and cancellation racing
|
||||
# against each other) cannot process - and double-consume - the same
|
||||
# allocations
|
||||
items = list(
|
||||
self.allocated_stock
|
||||
.filter(build_line__bom_item__sub_part__trackable=False)
|
||||
.select_for_update()
|
||||
.order_by('pk')
|
||||
)
|
||||
|
||||
# Remove stock
|
||||
|
|
@ -1025,7 +1076,7 @@ class Build(
|
|||
item.complete_allocation(user=user)
|
||||
|
||||
# Delete allocation
|
||||
items.all().delete()
|
||||
BuildItem.objects.filter(pk__in=[item.pk for item in items]).delete()
|
||||
|
||||
@transaction.atomic
|
||||
def scrap_build_output(
|
||||
|
|
@ -2043,6 +2094,15 @@ class BuildItem(InvenTree.models.InvenTreeMetadataModel):
|
|||
|
||||
item = self.stock_item
|
||||
|
||||
# Lock the stock item's row and refresh its quantity, so the clamp and
|
||||
# split decision below operate on the current committed quantity rather
|
||||
# than a stale in-memory copy (concurrent allocations may have consumed
|
||||
# stock from this same item in the meantime)
|
||||
if not item.lock_quantity():
|
||||
# The stock item no longer exists - remove this (now invalid) allocation
|
||||
self.delete()
|
||||
return
|
||||
|
||||
# Ensure we are not allocating more than available
|
||||
if quantity > item.quantity:
|
||||
quantity = item.quantity
|
||||
|
|
|
|||
|
|
@ -1,12 +1,17 @@
|
|||
"""Unit tests for the 'build' models."""
|
||||
|
||||
import threading
|
||||
import time
|
||||
import uuid
|
||||
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, transaction
|
||||
from django.db.models import Sum
|
||||
from django.test import TransactionTestCase, skipUnlessDBFeature
|
||||
from django.test.utils import override_settings
|
||||
from django.urls import reverse
|
||||
|
||||
|
|
@ -586,6 +591,39 @@ class BuildTest(BuildTestBase):
|
|||
initial_output_count,
|
||||
)
|
||||
|
||||
def test_cancel_stale_instance_is_noop(self):
|
||||
"""A second cancellation attempt with a stale build instance must be a no-op.
|
||||
|
||||
Regression test: _action_cancel() offloaded the cancellation task
|
||||
unconditionally, using no idempotency check at all, so two concurrent
|
||||
cancellation requests (each holding its own instance of the build) could
|
||||
both offload the cancellation task (duplicate events and notifications).
|
||||
The status is now re-read (under lock) from the database, and a build
|
||||
which is already cancelled is skipped.
|
||||
"""
|
||||
self.build.issue_build()
|
||||
self.allocate_stock(None, {self.stock_1_2: 50})
|
||||
|
||||
# Two "concurrent" requests each hold their own instance of the build
|
||||
build_a = Build.objects.get(pk=self.build.pk)
|
||||
build_b = Build.objects.get(pk=self.build.pk)
|
||||
|
||||
build_a.cancel_build(None)
|
||||
|
||||
self.build.refresh_from_db()
|
||||
self.assertEqual(self.build.status, status.BuildStatus.CANCELLED)
|
||||
|
||||
# The second (stale) instance still believes the build is in production -
|
||||
# cancellation must be skipped based on the database state
|
||||
self.assertEqual(build_b.status, status.BuildStatus.PRODUCTION)
|
||||
|
||||
with mock.patch('build.models.trigger_event') as trigger:
|
||||
build_b.cancel_build(None)
|
||||
trigger.assert_not_called()
|
||||
|
||||
self.build.refresh_from_db()
|
||||
self.assertEqual(self.build.status, status.BuildStatus.CANCELLED)
|
||||
|
||||
def test_complete(self):
|
||||
"""Test completion of a build output."""
|
||||
self.stock_1_1.quantity = 1000
|
||||
|
|
@ -763,6 +801,46 @@ class BuildTest(BuildTestBase):
|
|||
self.line_1.refresh_from_db()
|
||||
self.assertEqual(self.line_1.consumed, 0)
|
||||
|
||||
def test_complete_allocation_stale_item_instance(self):
|
||||
"""complete_allocation() must not clamp/split using a stale stock_item.quantity.
|
||||
|
||||
Regression test: the allocated quantity was clamped, and the choice between
|
||||
consuming the item directly versus splitting off a portion of it, was made
|
||||
using self.stock_item.quantity as cached in memory on the BuildItem. If a
|
||||
concurrent operation reduced the stock item's quantity after the BuildItem
|
||||
was loaded, the stale value could select the "split" branch when only the
|
||||
entire (now-reduced) item is actually available - causing splitStock() to
|
||||
reject the now-oversized split. The stock item's row is now locked and its
|
||||
quantity refreshed before the clamp/split decision is made.
|
||||
"""
|
||||
self.build.issue_build()
|
||||
|
||||
# A partial allocation - less than the full stock item quantity
|
||||
alloc = BuildItem.objects.create(
|
||||
build_line=self.line_1, stock_item=self.stock_1_2, quantity=60
|
||||
)
|
||||
|
||||
# Load a second copy of the allocation, caching the stock item's original quantity
|
||||
stale_alloc = BuildItem.objects.get(pk=alloc.pk)
|
||||
self.assertEqual(stale_alloc.stock_item.quantity, 100)
|
||||
|
||||
# Reduce the available stock *after* the stale copy was loaded
|
||||
# (simulating a concurrent stock adjustment elsewhere)
|
||||
self.stock_1_2.quantity = 40
|
||||
self.stock_1_2.save()
|
||||
|
||||
# Completing the (stale) allocation must not raise, and must consume only
|
||||
# the quantity which is actually available
|
||||
stale_alloc.complete_allocation(user=self.user)
|
||||
|
||||
self.stock_1_2.refresh_from_db()
|
||||
self.line_1.refresh_from_db()
|
||||
|
||||
# The entire (now-reduced) item is consumed directly - no split occurs
|
||||
self.assertEqual(self.stock_1_2.quantity, 40)
|
||||
self.assertEqual(self.stock_1_2.consumed_by, self.build)
|
||||
self.assertEqual(self.line_1.consumed, 40)
|
||||
|
||||
def test_complete_with_required_tests(self):
|
||||
"""Test the prevention completion when a required test is missing feature."""
|
||||
# with required tests incompleted the save should fail
|
||||
|
|
@ -1671,3 +1749,218 @@ class BuildTaskTests(BuildTestBase):
|
|||
self.assertFalse(output.is_building)
|
||||
# status=None should resolve to StockStatus.OK (the model default)
|
||||
self.assertEqual(output.status, StockStatus.OK.value)
|
||||
|
||||
|
||||
@skipUnlessDBFeature('has_select_for_update')
|
||||
class BuildTrimAllocatedStockConcurrencyTest(TransactionTestCase):
|
||||
"""Genuine cross-transaction regression test for Build.trim_allocated_stock().
|
||||
|
||||
Uses two real threads (each with its own database connection) to reproduce
|
||||
an actual overlapping-transaction race, rather than the "stale in-memory
|
||||
instance" trick used elsewhere - trim_allocated_stock() always re-queries the
|
||||
BuildItem rows fresh, so a stale Python object cannot reproduce this bug.
|
||||
"""
|
||||
|
||||
fixtures = ['users']
|
||||
|
||||
def setUp(self):
|
||||
"""Create a minimal build/allocation setup for the concurrency test."""
|
||||
super().setUp()
|
||||
|
||||
self.user = get_user_model().objects.get(pk=1)
|
||||
|
||||
self.assembly = Part.objects.create(
|
||||
name='Concurrency assembly',
|
||||
description='Assembly for trim_allocated_stock concurrency test',
|
||||
assembly=True,
|
||||
)
|
||||
self.sub_part = Part.objects.create(
|
||||
name='Concurrency component',
|
||||
description='Component for trim_allocated_stock concurrency test',
|
||||
component=True,
|
||||
)
|
||||
|
||||
BomItem.objects.create(part=self.assembly, sub_part=self.sub_part, quantity=1)
|
||||
|
||||
self.build = Build.objects.create(
|
||||
reference=generate_next_build_reference(),
|
||||
part=self.assembly,
|
||||
quantity=1,
|
||||
issued_by=self.user,
|
||||
)
|
||||
|
||||
self.build_line = BuildLine.objects.get(build=self.build)
|
||||
|
||||
self.stock_item = StockItem.objects.create(part=self.sub_part, quantity=100)
|
||||
|
||||
# Over-allocate: only 1 is required, but 10 have been allocated
|
||||
self.build_item = BuildItem.objects.create(
|
||||
build_line=self.build_line, stock_item=self.stock_item, quantity=10
|
||||
)
|
||||
|
||||
def test_trim_allocated_stock_does_not_lose_concurrent_allocation(self):
|
||||
"""trim_allocated_stock() must not lose a concurrent allocation update.
|
||||
|
||||
Regression test: BuildItem rows were read via a plain (unlocked) queryset,
|
||||
reduced in Python, and written back with bulk_update(). A concurrent request
|
||||
that increased the same BuildItem's quantity (e.g. a new allocation, which
|
||||
locks the row via select_for_update while saving - see
|
||||
BuildAllocationSerializer) could commit in between the read and the write,
|
||||
and would be silently discarded by the stale trim calculation. The
|
||||
BuildItem rows are now locked (select_for_update) before being read, so
|
||||
the trim always operates on the latest committed quantity.
|
||||
"""
|
||||
lock_acquired = threading.Event()
|
||||
release_lock = threading.Event()
|
||||
errors = []
|
||||
|
||||
def concurrent_allocation_increase():
|
||||
"""Simulate a second request increasing the allocation.
|
||||
|
||||
Holds a row-level lock on the BuildItem (as the real allocation
|
||||
endpoint does) until told to proceed, so the trim thread below is
|
||||
forced to block on the same row.
|
||||
"""
|
||||
try:
|
||||
with transaction.atomic():
|
||||
item = BuildItem.objects.select_for_update().get(
|
||||
pk=self.build_item.pk
|
||||
)
|
||||
lock_acquired.set()
|
||||
release_lock.wait(timeout=5)
|
||||
item.quantity += 5
|
||||
item.save()
|
||||
except Exception as exc: # pragma: no cover - surfaced via errors list
|
||||
errors.append(exc)
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
def run_trim():
|
||||
try:
|
||||
self.build.trim_allocated_stock()
|
||||
except Exception as exc: # pragma: no cover - surfaced via errors list
|
||||
errors.append(exc)
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
alloc_thread = threading.Thread(target=concurrent_allocation_increase)
|
||||
alloc_thread.start()
|
||||
|
||||
# Wait until the other thread holds the row lock
|
||||
self.assertTrue(lock_acquired.wait(timeout=5))
|
||||
|
||||
trim_thread = threading.Thread(target=run_trim)
|
||||
trim_thread.start()
|
||||
|
||||
# Give the trim thread a moment to reach (and block on) the row lock
|
||||
time.sleep(0.5)
|
||||
|
||||
# Release the lock - the concurrent allocation increase commits first
|
||||
release_lock.set()
|
||||
|
||||
alloc_thread.join(timeout=5)
|
||||
trim_thread.join(timeout=5)
|
||||
|
||||
self.assertFalse(alloc_thread.is_alive())
|
||||
self.assertFalse(trim_thread.is_alive())
|
||||
self.assertEqual(errors, [])
|
||||
|
||||
self.build_item.refresh_from_db()
|
||||
|
||||
# The concurrent increase (+5, giving 15) must not have been lost: the
|
||||
# trim reduces by 9 (10 originally allocated - 1 actually needed) from
|
||||
# whatever is actually committed, leaving 15 - 9 = 6. A lost update
|
||||
# would instead leave 15 (the trim's reduction discarded) or 1 (the
|
||||
# concurrent increase discarded).
|
||||
self.assertEqual(self.build_item.quantity, 6)
|
||||
|
||||
|
||||
@skipUnlessDBFeature('has_select_for_update')
|
||||
class BuildSubtractAllocatedStockConcurrencyTest(TransactionTestCase):
|
||||
"""Genuine cross-transaction regression test for Build.subtract_allocated_stock().
|
||||
|
||||
Uses two real threads (each with its own database connection) to reproduce
|
||||
duplicated/overlapping execution - e.g. a redelivered 'complete_build' or
|
||||
'cancel_build' background task - which cannot be reproduced with a single
|
||||
stale Python instance, since this method always re-queries the BuildItem
|
||||
rows fresh.
|
||||
"""
|
||||
|
||||
fixtures = ['users']
|
||||
|
||||
def setUp(self):
|
||||
"""Create a minimal build/allocation setup for the concurrency test."""
|
||||
super().setUp()
|
||||
|
||||
self.user = get_user_model().objects.get(pk=1)
|
||||
|
||||
self.assembly = Part.objects.create(
|
||||
name='Concurrency assembly 2',
|
||||
description='Assembly for subtract_allocated_stock concurrency test',
|
||||
assembly=True,
|
||||
)
|
||||
self.sub_part = Part.objects.create(
|
||||
name='Concurrency component 2',
|
||||
description='Component for subtract_allocated_stock concurrency test',
|
||||
component=True,
|
||||
)
|
||||
|
||||
BomItem.objects.create(part=self.assembly, sub_part=self.sub_part, quantity=1)
|
||||
|
||||
self.build = Build.objects.create(
|
||||
reference=generate_next_build_reference(),
|
||||
part=self.assembly,
|
||||
quantity=1,
|
||||
issued_by=self.user,
|
||||
)
|
||||
|
||||
self.build_line = BuildLine.objects.get(build=self.build)
|
||||
|
||||
self.stock_item = StockItem.objects.create(part=self.sub_part, quantity=10)
|
||||
|
||||
self.build_item = BuildItem.objects.create(
|
||||
build_line=self.build_line, stock_item=self.stock_item, quantity=10
|
||||
)
|
||||
|
||||
def test_subtract_allocated_stock_is_not_processed_twice(self):
|
||||
"""A duplicated/concurrent call to subtract_allocated_stock() must not double-consume.
|
||||
|
||||
Regression test: BuildItem allocation rows were read via a plain (unlocked)
|
||||
queryset before being consumed and deleted. Two overlapping calls to this
|
||||
method (e.g. a redelivered completion/cancellation background task) could
|
||||
each read the same still-existing allocation and both call
|
||||
complete_allocation() on it - double-counting the BuildLine's 'consumed'
|
||||
quantity. The BuildItem rows are now locked (select_for_update) before
|
||||
being processed, so a duplicate call finds nothing left to do.
|
||||
"""
|
||||
start_barrier = threading.Barrier(2, timeout=5)
|
||||
errors = []
|
||||
|
||||
def run_subtract():
|
||||
try:
|
||||
start_barrier.wait()
|
||||
self.build.subtract_allocated_stock(self.user)
|
||||
except Exception as exc: # pragma: no cover - surfaced via errors list
|
||||
errors.append(exc)
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
thread_a = threading.Thread(target=run_subtract)
|
||||
thread_b = threading.Thread(target=run_subtract)
|
||||
|
||||
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, [])
|
||||
|
||||
self.build_line.refresh_from_db()
|
||||
|
||||
# The allocation must only have been consumed once, no matter which
|
||||
# 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())
|
||||
|
|
|
|||
|
|
@ -851,6 +851,12 @@ class PurchaseOrder(TotalPriceMixin, Order):
|
|||
|
||||
Order must be currently PENDING.
|
||||
"""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the transition side
|
||||
# effects (duplicate events and notifications).
|
||||
self.status = PurchaseOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_issue:
|
||||
self.status = PurchaseOrderStatus.PLACED.value
|
||||
self.issue_date = InvenTree.helpers.current_date()
|
||||
|
|
@ -872,6 +878,12 @@ class PurchaseOrder(TotalPriceMixin, Order):
|
|||
|
||||
Order must be currently PLACED.
|
||||
"""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe status=PLACED, and each would run the completion side effects
|
||||
# (duplicate events and pricing-update scheduling).
|
||||
self.status = PurchaseOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.status == PurchaseOrderStatus.PLACED:
|
||||
self.status = PurchaseOrderStatus.COMPLETE.value
|
||||
self.complete_date = InvenTree.helpers.current_date()
|
||||
|
|
@ -953,6 +965,12 @@ class PurchaseOrder(TotalPriceMixin, Order):
|
|||
|
||||
def _action_cancel(self, *args, **kwargs):
|
||||
"""Marks the PurchaseOrder as CANCELLED."""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the cancellation side
|
||||
# effects (duplicate events and notifications).
|
||||
self.status = PurchaseOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_cancel:
|
||||
self.status = PurchaseOrderStatus.CANCELLED.value
|
||||
self.save()
|
||||
|
|
@ -978,6 +996,12 @@ class PurchaseOrder(TotalPriceMixin, Order):
|
|||
|
||||
def _action_hold(self, *args, **kwargs):
|
||||
"""Mark this purchase order as 'on hold'."""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the hold side effects
|
||||
# (duplicate events).
|
||||
self.status = PurchaseOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_hold:
|
||||
self.status = PurchaseOrderStatus.ON_HOLD.value
|
||||
self.save()
|
||||
|
|
@ -1807,6 +1831,12 @@ class SalesOrder(TotalPriceMixin, Order):
|
|||
|
||||
def _action_place(self, *args, **kwargs):
|
||||
"""Change this order from 'PENDING' to 'IN_PROGRESS'."""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the transition side
|
||||
# effects (duplicate events and notifications).
|
||||
self.status = SalesOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_issue:
|
||||
self.status = SalesOrderStatus.IN_PROGRESS.value
|
||||
self.issue_date = InvenTree.helpers.current_date()
|
||||
|
|
@ -1833,6 +1863,12 @@ class SalesOrder(TotalPriceMixin, Order):
|
|||
|
||||
def _action_hold(self, *args, **kwargs):
|
||||
"""Mark this sales order as 'on hold'."""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the hold side effects
|
||||
# (duplicate events).
|
||||
self.status = SalesOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_hold:
|
||||
self.status = SalesOrderStatus.ON_HOLD.value
|
||||
self.save()
|
||||
|
|
@ -1844,6 +1880,13 @@ class SalesOrder(TotalPriceMixin, Order):
|
|||
"""Mark this order as "complete."""
|
||||
user = kwargs.pop('user', None)
|
||||
|
||||
# Lock this order against concurrent completion, and re-read the status
|
||||
# from the database. Without this, two simultaneous completion requests
|
||||
# can both observe an "open" status, and each would run the completion
|
||||
# side effects (duplicate events, pricing updates, and shipped quantity
|
||||
# updates against the virtual line items).
|
||||
self.status = SalesOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if not self.can_complete(**kwargs):
|
||||
return False
|
||||
|
||||
|
|
@ -1888,6 +1931,13 @@ class SalesOrder(TotalPriceMixin, Order):
|
|||
- Mark the order as 'cancelled'
|
||||
- Delete any StockItems which have been allocated
|
||||
"""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an "open" status, and each would run the cancellation side
|
||||
# effects (duplicate events; the allocation deletion itself is
|
||||
# idempotent).
|
||||
self.status = SalesOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if not self.can_cancel:
|
||||
return False
|
||||
|
||||
|
|
@ -3176,6 +3226,12 @@ class ReturnOrder(TotalPriceMixin, Order):
|
|||
|
||||
def _action_hold(self, *args, **kwargs):
|
||||
"""Mark this order as 'on hold' (if allowed)."""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the hold side effects
|
||||
# (duplicate events).
|
||||
self.status = ReturnOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_hold:
|
||||
self.status = ReturnOrderStatus.ON_HOLD.value
|
||||
self.save()
|
||||
|
|
@ -3189,6 +3245,12 @@ class ReturnOrder(TotalPriceMixin, Order):
|
|||
|
||||
def _action_cancel(self, *args, **kwargs):
|
||||
"""Cancel this ReturnOrder (if not already cancelled)."""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the cancellation side
|
||||
# effects (duplicate events and notifications).
|
||||
self.status = ReturnOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_cancel:
|
||||
self.status = ReturnOrderStatus.CANCELLED.value
|
||||
self.save()
|
||||
|
|
@ -3233,6 +3295,12 @@ class ReturnOrder(TotalPriceMixin, Order):
|
|||
|
||||
def _action_place(self, *args, **kwargs):
|
||||
"""Issue this ReturnOrder (if currently pending)."""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the issue side effects
|
||||
# (duplicate events and notifications).
|
||||
self.status = ReturnOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_issue:
|
||||
self.status = ReturnOrderStatus.IN_PROGRESS.value
|
||||
self.issue_date = InvenTree.helpers.current_date()
|
||||
|
|
@ -3704,6 +3772,12 @@ class TransferOrder(Order):
|
|||
|
||||
Order must be currently PENDING.
|
||||
"""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the issue side effects
|
||||
# (duplicate events and notifications).
|
||||
self.status = TransferOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_issue:
|
||||
self.status = TransferOrderStatus.ISSUED.value
|
||||
self.issue_date = InvenTree.helpers.current_date()
|
||||
|
|
@ -3730,6 +3804,12 @@ class TransferOrder(Order):
|
|||
|
||||
def _action_hold(self, *args, **kwargs):
|
||||
"""Mark this transfer order as 'on hold'."""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an eligible status, and each would run the hold side effects
|
||||
# (duplicate events).
|
||||
self.status = TransferOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if self.can_hold:
|
||||
self.status = TransferOrderStatus.ON_HOLD.value
|
||||
self.save()
|
||||
|
|
@ -3809,6 +3889,13 @@ class TransferOrder(Order):
|
|||
- Mark the order as 'cancelled'
|
||||
- Delete any StockItems which have been allocated
|
||||
"""
|
||||
# Lock this order against concurrent transitions, and re-read the status
|
||||
# from the database. Without this, two simultaneous requests can both
|
||||
# observe an "open" status, and each would run the cancellation side
|
||||
# effects (duplicate events; the allocation deletion itself is
|
||||
# idempotent).
|
||||
self.status = TransferOrder.objects.select_for_update().get(pk=self.pk).status
|
||||
|
||||
if not self.can_cancel:
|
||||
return False
|
||||
|
||||
|
|
@ -4077,6 +4164,12 @@ class TransferOrderAllocation(models.Model):
|
|||
self.item: stock.models.StockItem # for type hints
|
||||
self.line: TransferOrderLineItem # for type hints
|
||||
|
||||
# Lock the stock item's row and refresh its quantity, so the branch
|
||||
# selected below (consume / split / move) is chosen using the current
|
||||
# committed quantity rather than a stale in-memory copy
|
||||
if not self.item.lock_quantity():
|
||||
raise ValidationError(_('Stock item no longer exists'))
|
||||
|
||||
# The allocation is the only thing linking this stock item to the transfer
|
||||
# As a result, we must keep the allocation present even after completion
|
||||
# This means allocations to transfer orders don't affect "available" stock
|
||||
|
|
|
|||
|
|
@ -675,6 +675,38 @@ class PurchaseOrderTest(OrderTest):
|
|||
|
||||
self.assertEqual(po.status, PurchaseOrderStatus.COMPLETE)
|
||||
|
||||
def test_po_complete_stale_instance_is_noop(self):
|
||||
"""A second completion attempt with a stale order instance must be a no-op.
|
||||
|
||||
Regression test: _action_complete() checked 'status' on the caller's
|
||||
(potentially stale) instance, so two concurrent completion requests could
|
||||
both run the completion side effects (duplicate COMPLETED events and
|
||||
duplicate pricing-update scheduling). The status is now re-read (under
|
||||
lock) from the database before the check.
|
||||
"""
|
||||
po = models.PurchaseOrder.objects.get(pk=3)
|
||||
self.assertEqual(po.status, PurchaseOrderStatus.PLACED)
|
||||
|
||||
# Two "concurrent" requests each hold their own instance of the order
|
||||
po_a = models.PurchaseOrder.objects.get(pk=po.pk)
|
||||
po_b = models.PurchaseOrder.objects.get(pk=po.pk)
|
||||
|
||||
po_a.complete_order()
|
||||
|
||||
po.refresh_from_db()
|
||||
self.assertEqual(po.status, PurchaseOrderStatus.COMPLETE)
|
||||
|
||||
# The second (stale) instance still believes the order is PLACED -
|
||||
# completion must be skipped based on the database state
|
||||
self.assertEqual(po_b.status, PurchaseOrderStatus.PLACED)
|
||||
|
||||
with mock.patch('order.models.trigger_event') as trigger:
|
||||
po_b.complete_order()
|
||||
trigger.assert_not_called()
|
||||
|
||||
po.refresh_from_db()
|
||||
self.assertEqual(po.status, PurchaseOrderStatus.COMPLETE)
|
||||
|
||||
def test_po_issue(self):
|
||||
"""Test the PurchaseOrderIssue API endpoint."""
|
||||
po = models.PurchaseOrder.objects.get(pk=2)
|
||||
|
|
@ -3828,6 +3860,38 @@ class TransferOrderTest(OrderTest):
|
|||
|
||||
self.assertEqual(to.status, TransferOrderStatus.CANCELLED)
|
||||
|
||||
def test_transfer_order_cancel_stale_instance_is_noop(self):
|
||||
"""A second cancellation attempt with a stale order instance must be a no-op.
|
||||
|
||||
Regression test: _action_cancel() checked 'can_cancel' (derived from
|
||||
'status') on the caller's (potentially stale) instance, so two concurrent
|
||||
cancellation requests could both run the cancellation side effects
|
||||
(duplicate CANCELLED events). The status is now re-read (under lock)
|
||||
from the database before the check.
|
||||
"""
|
||||
to = models.TransferOrder.objects.get(pk=1)
|
||||
self.assertEqual(to.status, TransferOrderStatus.PENDING)
|
||||
|
||||
# Two "concurrent" requests each hold their own instance of the order
|
||||
instance_a = models.TransferOrder.objects.get(pk=to.pk)
|
||||
instance_b = models.TransferOrder.objects.get(pk=to.pk)
|
||||
|
||||
instance_a.cancel_order()
|
||||
|
||||
to.refresh_from_db()
|
||||
self.assertEqual(to.status, TransferOrderStatus.CANCELLED)
|
||||
|
||||
# The second (stale) instance still believes the order is PENDING -
|
||||
# cancellation must be skipped based on the database state
|
||||
self.assertEqual(instance_b.status, TransferOrderStatus.PENDING)
|
||||
|
||||
with mock.patch('order.models.trigger_event') as trigger:
|
||||
instance_b.cancel_order()
|
||||
trigger.assert_not_called()
|
||||
|
||||
to.refresh_from_db()
|
||||
self.assertEqual(to.status, TransferOrderStatus.CANCELLED)
|
||||
|
||||
def test_transfer_order_hold(self):
|
||||
"""Test API endpoint for holdling a TransferOrder."""
|
||||
to = models.TransferOrder.objects.get(pk=1)
|
||||
|
|
@ -4155,6 +4219,60 @@ class TransferOrderTest(OrderTest):
|
|||
self.assertEqual(line.transferred, 4)
|
||||
self.assertEqual(item.quantity, 0)
|
||||
|
||||
def test_transfer_order_partial_allocation_stale_item_instance(self):
|
||||
"""A partial allocation must not fail completion if stock is reduced concurrently.
|
||||
|
||||
Regression test: transfer_quantity (and the choice between the "move" and
|
||||
"split" branches) was computed from self.item.quantity as cached in memory
|
||||
on the allocation instance. If the stock item's quantity was reduced by a
|
||||
concurrent operation after the allocation was loaded, the stale value could
|
||||
select the "split" branch when only a "move" of the entire (now-reduced)
|
||||
item is actually possible - causing splitStock() to reject the now-oversized
|
||||
split, and the whole order completion to fail. The stock item's row is now
|
||||
locked and its quantity refreshed before the branch is chosen.
|
||||
"""
|
||||
self.assignRole('transfer_order.add')
|
||||
destination = StockLocation.objects.first()
|
||||
|
||||
to = models.TransferOrder.objects.create(
|
||||
reference='TO-STALE-ITEM', description='Test TO', destination=destination
|
||||
)
|
||||
|
||||
part = Part.objects.exclude(virtual=True).first()
|
||||
|
||||
line = models.TransferOrderLineItem.objects.create(
|
||||
order=to, part=part, quantity=6
|
||||
)
|
||||
|
||||
to.issue_order()
|
||||
|
||||
item = StockItem.objects.create(part=part, quantity=10, batch='to-stale-item')
|
||||
|
||||
# A partial allocation - less than the full stock item quantity
|
||||
allocation = models.TransferOrderAllocation.objects.create(
|
||||
quantity=6, line=line, item=item
|
||||
)
|
||||
|
||||
# Load a second copy of the allocation, caching the item's original quantity
|
||||
stale_allocation = models.TransferOrderAllocation.objects.get(pk=allocation.pk)
|
||||
self.assertEqual(stale_allocation.item.quantity, 10)
|
||||
|
||||
# Reduce the available stock *after* the stale copy was loaded
|
||||
# (simulating a concurrent stock adjustment)
|
||||
item.quantity = 4
|
||||
item.save()
|
||||
|
||||
# Completing the (stale) allocation must not raise, and must transfer
|
||||
# only the quantity which is actually available
|
||||
stale_allocation.complete_allocation(None)
|
||||
|
||||
item.refresh_from_db()
|
||||
line.refresh_from_db()
|
||||
|
||||
self.assertEqual(item.location, destination)
|
||||
self.assertEqual(item.quantity, 4)
|
||||
self.assertEqual(line.transferred, 4)
|
||||
|
||||
def test_transfer_order_complete_stale_instance(self):
|
||||
"""A second completion attempt with a stale instance must be rejected.
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
"""Unit tests for the SalesOrder models."""
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from unittest import mock
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.auth.models import Group
|
||||
|
|
@ -279,6 +280,39 @@ class SalesOrderTest(InvenTreeAPITestCase):
|
|||
result = self.order.ship_order(None)
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_order_cancel_stale_instance_is_noop(self):
|
||||
"""A second cancellation attempt with a stale order instance must be a no-op.
|
||||
|
||||
Regression test: _action_cancel() checked 'status' on the caller's
|
||||
(potentially stale) instance, so two concurrent cancellation requests
|
||||
could both run the cancellation side effects (duplicate CANCELLED
|
||||
events). The status is now re-read (under lock) from the database
|
||||
before the check.
|
||||
"""
|
||||
self.allocate_stock(True)
|
||||
self.assertEqual(SalesOrderAllocation.objects.count(), 2)
|
||||
|
||||
# Two "concurrent" requests each hold their own instance of the order
|
||||
order_a = SalesOrder.objects.get(pk=self.order.pk)
|
||||
order_b = SalesOrder.objects.get(pk=self.order.pk)
|
||||
|
||||
order_a.cancel_order()
|
||||
|
||||
self.order.refresh_from_db()
|
||||
self.assertEqual(self.order.status, status.SalesOrderStatus.CANCELLED)
|
||||
self.assertEqual(SalesOrderAllocation.objects.count(), 0)
|
||||
|
||||
# The second (stale) instance still believes the order is PENDING -
|
||||
# cancellation must be skipped based on the database state
|
||||
self.assertEqual(order_b.status, status.SalesOrderStatus.PENDING)
|
||||
|
||||
with mock.patch('order.models.trigger_event') as trigger:
|
||||
order_b.cancel_order()
|
||||
trigger.assert_not_called()
|
||||
|
||||
self.order.refresh_from_db()
|
||||
self.assertEqual(self.order.status, status.SalesOrderStatus.CANCELLED)
|
||||
|
||||
def test_complete_order(self):
|
||||
"""Allocate line items, then ship the order."""
|
||||
# Assert some stuff before we run the test
|
||||
|
|
@ -368,6 +402,44 @@ class SalesOrderTest(InvenTreeAPITestCase):
|
|||
self.assertEqual(self.line.fulfilled_quantity(), 50)
|
||||
self.assertEqual(self.line.allocated_quantity(), 50)
|
||||
|
||||
def test_complete_order_stale_instance_is_noop(self):
|
||||
"""A second completion attempt with a stale order instance must be a no-op.
|
||||
|
||||
Regression test: _action_complete() checked 'status' on the caller's
|
||||
(potentially stale) instance, so two concurrent completion requests could
|
||||
both run the completion side effects (duplicate COMPLETED events and
|
||||
duplicate pricing-update scheduling). The status is now re-read (under
|
||||
lock) from the database before the check.
|
||||
"""
|
||||
self.allocate_stock(True)
|
||||
self.shipment.complete_shipment(None)
|
||||
|
||||
# Ship the order (PENDING -> SHIPPED)
|
||||
self.assertTrue(self.order.ship_order(None))
|
||||
self.order.refresh_from_db()
|
||||
self.assertEqual(self.order.status, status.SalesOrderStatus.SHIPPED)
|
||||
|
||||
# Two "concurrent" requests each hold their own instance of the order,
|
||||
# both about to transition SHIPPED -> COMPLETE
|
||||
order_a = SalesOrder.objects.get(pk=self.order.pk)
|
||||
order_b = SalesOrder.objects.get(pk=self.order.pk)
|
||||
|
||||
self.assertTrue(order_a.complete_order(None))
|
||||
|
||||
self.order.refresh_from_db()
|
||||
self.assertEqual(self.order.status, status.SalesOrderStatus.COMPLETE)
|
||||
|
||||
# The second (stale) instance still believes the order is SHIPPED -
|
||||
# completion must be skipped based on the database state
|
||||
self.assertEqual(order_b.status, status.SalesOrderStatus.SHIPPED)
|
||||
|
||||
with mock.patch('order.models.trigger_event') as trigger:
|
||||
self.assertFalse(order_b.complete_order(None))
|
||||
trigger.assert_not_called()
|
||||
|
||||
self.order.refresh_from_db()
|
||||
self.assertEqual(self.order.status, status.SalesOrderStatus.COMPLETE)
|
||||
|
||||
def test_complete_shipment_task_is_idempotent(self):
|
||||
"""Duplicate execution of the shipment completion task must not double-ship.
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue