mirror of
https://github.com/inventree/InvenTree.git
synced 2026-11-14 20:42:42 +00:00
[bug] Fix for shipment allocations (#13272)
Apply row-level locking when completing shipment allocation to avoid concurrent update issues
This commit is contained in:
1 parent
de77fc4293
commit
7787408fff
2 files changed
+162
-2
No files matched your search
@@ -2998,7 +2998,30 @@ class SalesOrderShipment(
|
||||
customer = order.customer
|
||||
|
||||
# Preselect related fields to avoid per-row database queries below
|
||||
allocations = allocations.select_related('line', 'item', 'item__part')
|
||||
allocations = list(allocations.select_related('line', 'item', 'item__part'))
|
||||
|
||||
# Lock the distinct StockItem rows (in pk order, to avoid deadlocks) and
|
||||
# re-read their quantities, so the arithmetic below operates on the
|
||||
# current committed values. Stock adjustments elsewhere lock these
|
||||
# same rows, so concurrent adjustments cannot be silently overwritten.
|
||||
locked_quantities = dict(
|
||||
stock.models.StockItem.objects
|
||||
.select_for_update()
|
||||
.filter(pk__in={allocation.item_id for allocation in allocations})
|
||||
.order_by('pk')
|
||||
.values_list('pk', 'quantity')
|
||||
)
|
||||
|
||||
# Likewise, lock the distinct SalesOrderLineItem rows and re-read their
|
||||
# 'shipped' quantities: another shipment against the same line may be
|
||||
# completed concurrently, and each must add to the latest value
|
||||
locked_shipped = dict(
|
||||
SalesOrderLineItem.objects
|
||||
.select_for_update()
|
||||
.filter(pk__in={allocation.line_id for allocation in allocations})
|
||||
.order_by('pk')
|
||||
.values_list('pk', 'shipped')
|
||||
)
|
||||
|
||||
split_items = [] # (source_item, new_item, quantity) - stock to split off
|
||||
shipped_items = [] # (target_item, quantity) - stock to mark as shipped
|
||||
@@ -3017,6 +3040,13 @@ class SalesOrderShipment(
|
||||
stock_item = seen_stock_items.get(allocation.item_id) or allocation.item
|
||||
quantity = allocation.quantity
|
||||
|
||||
if (locked_qty := locked_quantities.pop(stock_item.pk, None)) is not None:
|
||||
# First encounter: refresh the quantity from the locked database row
|
||||
stock_item.quantity = locked_qty
|
||||
elif stock_item.pk not in seen_stock_items:
|
||||
# The stock item no longer exists in the database - skip
|
||||
continue
|
||||
|
||||
seen_stock_items[stock_item.pk] = stock_item
|
||||
|
||||
if quantity < stock_item.quantity:
|
||||
@@ -3046,7 +3076,13 @@ class SalesOrderShipment(
|
||||
shipped_items.append((target_item, quantity))
|
||||
|
||||
# Increase the "shipped" quantity for the associated line
|
||||
line = seen_lines.get(allocation.line_id) or allocation.line
|
||||
line = seen_lines.get(allocation.line_id)
|
||||
|
||||
if line is None:
|
||||
# First encounter: refresh the quantity from the locked database row
|
||||
line = allocation.line
|
||||
line.shipped = locked_shipped.get(line.pk, line.shipped)
|
||||
|
||||
line.shipped += quantity
|
||||
seen_lines[line.pk] = line
|
||||
|
||||
|
||||
@@ -1342,3 +1342,127 @@ class SalesOrderAllocateStockConcurrencyTest(TransactionTestCase):
|
||||
|
||||
self.assertEqual(total_allocated, 5)
|
||||
self.assertLessEqual(total_allocated, self.stock_item.quantity)
|
||||
|
||||
|
||||
@skipUnlessDBFeature('has_select_for_update')
|
||||
class SalesOrderShipmentCompleteConcurrencyTest(TransactionTestCase):
|
||||
"""Genuine cross-transaction regression test for completing sales order shipments.
|
||||
|
||||
Uses two real threads (each with its own database connection) to reproduce
|
||||
the race: two shipments which ship stock against the *same* line item (and
|
||||
from the same StockItem) are completed at the same time. Each task locked
|
||||
only its own shipment row, read the line 'shipped' quantity (and the stock
|
||||
item quantity) from an unlocked query, adjusted them in memory, and wrote
|
||||
the absolute values back - so one of the two updates was silently lost.
|
||||
|
||||
SalesOrderShipment.complete_allocations() now locks the SalesOrderLineItem
|
||||
and StockItem rows, and re-reads their quantities, before adjusting them.
|
||||
"""
|
||||
|
||||
fixtures = ['users']
|
||||
|
||||
def setUp(self):
|
||||
"""Create a sales order with two shipments against the same line item."""
|
||||
super().setUp()
|
||||
|
||||
self.user = get_user_model().objects.get(pk=1)
|
||||
|
||||
customer = Company.objects.create(
|
||||
name='Shipment concurrency customer',
|
||||
description='Customer for shipment concurrency test',
|
||||
is_customer=True,
|
||||
)
|
||||
|
||||
part = Part.objects.create(
|
||||
name='Shipment concurrency part',
|
||||
description='Part for shipment concurrency test',
|
||||
salable=True,
|
||||
)
|
||||
|
||||
self.order = SalesOrder.objects.create(customer=customer, reference='SO-9992')
|
||||
|
||||
self.line = SalesOrderLineItem.objects.create(
|
||||
quantity=6, order=self.order, part=part
|
||||
)
|
||||
|
||||
self.stock_item = StockItem.objects.create(part=part, quantity=10)
|
||||
|
||||
self.shipments = [
|
||||
SalesOrderShipment.objects.create(order=self.order, reference=ref)
|
||||
for ref in ['A', 'B']
|
||||
]
|
||||
|
||||
for shipment in self.shipments:
|
||||
SalesOrderAllocation.objects.create(
|
||||
line=self.line, item=self.stock_item, shipment=shipment, quantity=3
|
||||
)
|
||||
|
||||
def test_concurrent_shipment_completion(self):
|
||||
"""Two shipments against the same line item, completed concurrently, are both counted."""
|
||||
import order.models
|
||||
|
||||
errors = []
|
||||
|
||||
# Hold each task after it has read (and adjusted) the quantities, but
|
||||
# before it writes them, until the other task has reached the same point.
|
||||
# Once the rows are locked correctly, the second task cannot get this far
|
||||
# until the first has committed - so the barrier simply times out.
|
||||
barrier = threading.Barrier(2, timeout=2)
|
||||
|
||||
original = order.models.bulk_create_and_fetch
|
||||
|
||||
def synced_bulk_create_and_fetch(*args, **kwargs):
|
||||
try:
|
||||
barrier.wait()
|
||||
except threading.BrokenBarrierError:
|
||||
pass
|
||||
|
||||
return original(*args, **kwargs)
|
||||
|
||||
def complete(shipment):
|
||||
try:
|
||||
order.tasks.complete_sales_order_shipment(
|
||||
shipment.pk, self.user.pk, None
|
||||
)
|
||||
except Exception as exc: # pragma: no cover - surfaced via errors list
|
||||
errors.append(exc)
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
threads = [
|
||||
threading.Thread(target=complete, args=(shipment,))
|
||||
for shipment in self.shipments
|
||||
]
|
||||
|
||||
with mock.patch.object(
|
||||
order.models, 'bulk_create_and_fetch', synced_bulk_create_and_fetch
|
||||
):
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
|
||||
for thread in threads:
|
||||
thread.join(timeout=15)
|
||||
|
||||
for thread in threads:
|
||||
self.assertFalse(thread.is_alive())
|
||||
|
||||
self.assertEqual(errors, [])
|
||||
|
||||
for shipment in self.shipments:
|
||||
shipment.refresh_from_db()
|
||||
self.assertTrue(shipment.is_complete())
|
||||
|
||||
# Both shipments are counted against the line item
|
||||
self.line.refresh_from_db()
|
||||
self.assertEqual(self.line.shipped, 6)
|
||||
|
||||
# Both quantities have been taken from the stock item
|
||||
self.stock_item.refresh_from_db()
|
||||
self.assertEqual(self.stock_item.quantity, 4)
|
||||
|
||||
shipped_items = StockItem.objects.filter(
|
||||
parent=self.stock_item, sales_order=self.order
|
||||
)
|
||||
|
||||
self.assertEqual(shipped_items.count(), 2)
|
||||
self.assertEqual(sum(item.quantity for item in shipped_items), 6)
|
||||
Reference in new issue
Block a user