修复:避免首事件间隙锁死锁

This commit is contained in:
焦龙言
2026-07-26 14:24:55 +08:00
parent 6c2a5fa344
commit 317a0fbf68
4 changed files with 54 additions and 14 deletions
@@ -93,16 +93,21 @@ class InventoryDao:
event: BizPalletEvent, event: BizPalletEvent,
) -> BizPalletEvent: ) -> BizPalletEvent:
"""在调用方持有库存行锁时,为托盘追加下一序号事件。""" """在调用方持有库存行锁时,为托盘追加下一序号事件。"""
sequence_result = await db.execute( last_sequence = 0
select(BizPalletEvent.event_seq) if inventory.last_event_id is not None:
.where(BizPalletEvent.pallet_id == inventory.pallet_id) sequence_result = await db.execute(
.order_by(BizPalletEvent.event_seq.desc()) select(BizPalletEvent.event_seq)
.limit(1) .where(
.with_for_update() BizPalletEvent.event_id == inventory.last_event_id,
) BizPalletEvent.pallet_id == inventory.pallet_id,
last_sequence = sequence_result.scalar_one_or_none() )
.with_for_update()
)
last_sequence = sequence_result.scalar_one_or_none()
if last_sequence is None:
raise RuntimeError('托盘库存事件指针不一致')
event.pallet_id = inventory.pallet_id event.pallet_id = inventory.pallet_id
event.event_seq = (last_sequence or 0) + 1 event.event_seq = last_sequence + 1
db.add(event) db.add(event)
await db.flush() await db.flush()
inventory.last_event_id = event.event_id inventory.last_event_id = event.event_id
@@ -31,6 +31,8 @@ def _validate_page_query_date_range(value: BaseModel) -> BaseModel:
end_value = value.end_time end_value = value.end_time
begin_date = _parse_page_query_date(begin_value) if begin_value else None begin_date = _parse_page_query_date(begin_value) if begin_value else None
end_date = _parse_page_query_date(end_value) if end_value else None end_date = _parse_page_query_date(end_value) if end_value else None
if end_date and end_date.date() == datetime.max.date():
raise ValueError('结束日期必须早于 9999-12-31')
if begin_date and end_date and begin_date > end_date: if begin_date and end_date and begin_date > end_date:
raise ValueError('开始日期不能晚于结束日期') raise ValueError('开始日期不能晚于结束日期')
return value return value
@@ -284,12 +284,12 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
self.assertIn("biz_pallet_event.idempotency_key = 'event-key'", sql) self.assertIn("biz_pallet_event.idempotency_key = 'event-key'", sql)
self.assertNotIn('FOR UPDATE', sql) self.assertNotIn('FOR UPDATE', sql)
async def test_add_event_allocates_next_sequence_and_updates_inventory_pointer(self) -> None: async def test_add_event_reads_exact_previous_pointer_and_updates_inventory(self) -> None:
sequence_result = MagicMock() sequence_result = MagicMock()
sequence_result.scalar_one.return_value = 4 sequence_result.scalar_one.return_value = 4
sequence_result.scalar_one_or_none.return_value = 4 sequence_result.scalar_one_or_none.return_value = 4
self.db.execute.return_value = sequence_result self.db.execute.return_value = sequence_result
inventory = BizPalletInventory(pallet_id=7, last_event_id=None) inventory = BizPalletInventory(pallet_id=7, last_event_id=40)
event = BizPalletEvent(pallet_id=99, event_seq=99, idempotency_key='event-key') event = BizPalletEvent(pallet_id=99, event_seq=99, idempotency_key='event-key')
flush_count = 0 flush_count = 0
@@ -311,15 +311,17 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
self.assertEqual(2, self.db.flush.await_count) self.assertEqual(2, self.db.flush.await_count)
statement = self.db.execute.await_args.args[0] statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.selected_columns[0].compare(BizPalletEvent.event_seq.property.columns[0])) self.assertTrue(statement.selected_columns[0].compare(BizPalletEvent.event_seq.property.columns[0]))
self.assertTrue(statement.get_final_froms()[0].compare(BizPalletEvent.__table__))
for dialect in (mysql.dialect(), postgresql.dialect()): for dialect in (mysql.dialect(), postgresql.dialect()):
sql = compile_sql(statement, dialect) sql = compile_sql(statement, dialect)
self.assertNotIn('max(', sql.lower()) self.assertNotIn('max(', sql.lower())
self.assertIn('biz_pallet_event.event_id = 40', sql)
self.assertIn('biz_pallet_event.pallet_id = 7', sql) self.assertIn('biz_pallet_event.pallet_id = 7', sql)
self.assertIn('ORDER BY biz_pallet_event.event_seq DESC', sql)
self.assertIn('LIMIT 1', sql)
self.assertIn('FOR UPDATE', sql) self.assertIn('FOR UPDATE', sql)
self.assertNotIn('ORDER BY', sql)
self.assertNotIn('LIMIT', sql)
async def test_add_first_event_starts_sequence_at_one(self) -> None: async def test_add_first_event_starts_at_one_without_query(self) -> None:
sequence_result = MagicMock() sequence_result = MagicMock()
sequence_result.scalar_one.return_value = None sequence_result.scalar_one.return_value = None
sequence_result.scalar_one_or_none.return_value = None sequence_result.scalar_one_or_none.return_value = None
@@ -331,6 +333,25 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
self.assertEqual(1, event.event_seq) self.assertEqual(1, event.event_seq)
self.assertEqual(51, inventory.last_event_id) self.assertEqual(51, inventory.last_event_id)
self.db.execute.assert_not_awaited()
self.db.add.assert_called_once_with(event)
self.assertEqual(2, self.db.flush.await_count)
async def test_add_event_rejects_missing_previous_pointer_without_write(self) -> None:
sequence_result = MagicMock()
sequence_result.scalar_one_or_none.return_value = None
self.db.execute.return_value = sequence_result
inventory = BizPalletInventory(pallet_id=8, last_event_id=50)
event = BizPalletEvent(pallet_id=8, idempotency_key='broken-pointer')
with self.assertRaisesRegex(RuntimeError, '托盘库存事件指针不一致'):
await InventoryDao.add_event(self.db, inventory, event)
self.db.execute.assert_awaited_once()
self.db.add.assert_not_called()
self.db.flush.assert_not_awaited()
self.assertEqual(50, inventory.last_event_id)
self.assertIsNone(event.event_seq)
class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
@@ -809,6 +809,18 @@ class BusinessVoContractTest(unittest.TestCase):
with self.subTest(model=model_class.__name__, case='inverted'), self.assertRaises(ValidationError): with self.subTest(model=model_class.__name__, case='inverted'), self.assertRaises(ValidationError):
model_class.model_validate({'beginTime': '2026-07-03', 'endTime': '2026-07-02'}) model_class.model_validate({'beginTime': '2026-07-03', 'endTime': '2026-07-02'})
def test_operation_page_query_rejects_overflowing_end_but_allows_same_begin(self) -> None:
for model_class in OPERATION_PAGE_QUERY_MODELS:
with self.subTest(model=model_class.__name__, field='beginTime'):
model = model_class.model_validate({'beginTime': '9999-12-31'})
self.assertEqual('9999-12-31', model.begin_time)
self.assertIsNone(model.end_time)
with (
self.subTest(model=model_class.__name__, field='endTime'),
self.assertRaises(ValidationError),
):
model_class.model_validate({'endTime': '9999-12-31'})
def test_pallet_page_size_is_capped_at_100(self) -> None: def test_pallet_page_size_is_capped_at_100(self) -> None:
self.assertEqual(100, PalletPageQueryModel.model_validate({'pageSize': 100}).page_size) self.assertEqual(100, PalletPageQueryModel.model_validate({'pageSize': 100}).page_size)
with self.assertRaises(ValidationError): with self.assertRaises(ValidationError):