From 5912e9d056c2819e8f3f3cf64305712dd8ae7d0c Mon Sep 17 00:00:00 2001 From: wuzilong <704541015@qq.com> Date: Sun, 1 Nov 2026 21:26:46 +0800 Subject: [PATCH] fix(pl): Read current certificate bindings when returning atomic saves. --- .../app/services/pl_quota_cert_bindings.py | 26 +++++---- backend/tests/test_pl_quota_cert_bindings.py | 54 ++++++++++++++++++- 2 files changed, 69 insertions(+), 11 deletions(-) diff --git a/backend/app/services/pl_quota_cert_bindings.py b/backend/app/services/pl_quota_cert_bindings.py index 30ce68fb3..2f2cc0aff 100644 --- a/backend/app/services/pl_quota_cert_bindings.py +++ b/backend/app/services/pl_quota_cert_bindings.py @@ -39,19 +39,25 @@ def _visible_contract_ids(db, user): return apply_crm_owner_scope(query, CrmContract, user.to_user_info(), db, user.tenant_no) -def get_cert(db, user, cert_type, cert_id): - result = certs.get_cert(db, user, cert_type, cert_id) - links = db.query(PlContractCertLink).filter(PlContractCertLink.tenant_no == user.tenant_no, +def get_cert(db, user, cert_type, cert_id, current=False): + # 保存已持有目标证锁；重复当前读不会取得新的前置证锁。 + result = certs.get_cert(db, user, cert_type, cert_id, current=current) + def rows(query): + if current: + query = query.with_for_update().populate_existing() + return query.all() + links = rows(db.query(PlContractCertLink).filter(PlContractCertLink.tenant_no == user.tenant_no, PlContractCertLink.cert_type == cert_type, PlContractCertLink.cert_id == cert_id, PlContractCertLink.del_flag == 0, PlContractCertLink.contract_id.in_(_visible_contract_ids(db, user))).order_by( - PlContractCertLink.create_time.desc(), PlContractCertLink.id).all() - products = db.query(PlContractCertProduct).filter( + PlContractCertLink.create_time.desc(), PlContractCertLink.id)) + products = rows(db.query(PlContractCertProduct).filter( PlContractCertProduct.tenant_no == user.tenant_no, PlContractCertProduct.del_flag == 0, PlContractCertProduct.contract_cert_link_id.in_([r.id for r in links])).order_by( - PlContractCertProduct.create_time.desc(), PlContractCertProduct.id).all() if links else [] - total_links = db.query(PlContractCertLink.id).filter_by(tenant_no=user.tenant_no, - cert_type=cert_type, cert_id=cert_id, del_flag=0).count() + PlContractCertProduct.create_time.desc(), PlContractCertProduct.id)) if links else [] + # 直接读取ID，避免 count() 派生表把锁读退化为一致性快照。 + total_links = len(rows(db.query(PlContractCertLink.id).filter_by(tenant_no=user.tenant_no, + cert_type=cert_type, cert_id=cert_id, del_flag=0))) result['bindingsEditable'] = total_links == len(links) result['contractBindings'] = [{ 'id': link.id, 'contractId': link.contract_id, 'linkStatus': link.link_status, @@ -80,7 +86,7 @@ def save_cert(db, user, cert_type, data, cert_id=None): present = 'contractBindings' in payload or 'contract_bindings' in payload if not present: result = certs.save_cert(db, user, cert_type, payload, cert_id) - return get_cert(db, user, cert_type, result['id']) + return get_cert(db, user, cert_type, result['id'], current=True) bindings_payload = {key: payload.pop(key) for key in ('contractBindings', 'contract_bindings') if key in payload} body = CertificateBindings(**bindings_payload) if body.contract_bindings is None: @@ -142,4 +148,4 @@ def save_cert(db, user, cert_type, data, cert_id=None): for scope in prepared: scopes.apply_contract_bindings(db, user, scope) db.flush() - return get_cert(db, user, cert_type, row.id) + return get_cert(db, user, cert_type, row.id, current=True) diff --git a/backend/tests/test_pl_quota_cert_bindings.py b/backend/tests/test_pl_quota_cert_bindings.py index 5679b517d..7ed78e93a 100644 --- a/backend/tests/test_pl_quota_cert_bindings.py +++ b/backend/tests/test_pl_quota_cert_bindings.py @@ -115,7 +115,8 @@ def test_transport_locks_all_certificates_before_contracts_and_current_reads(cas finally: event.remove(c.db, 'do_orm_execute', capture) assert events == [('cert', 'purchase', 'g1'), ('cert', 'purchase', 'g2'), - ('cert', 'transport', target), ('contract', 'c1'), ('contract', 'c2')] + ('cert', 'transport', target), ('contract', 'c1'), ('contract', 'c2'), + ('cert', 'transport', target)] # 响应当前读仅重复已持有的目标证锁。 for model in (CrmContract, CrmContractProduct, c.m.PlContractCertLink, c.m.PlTransportCertPurchaseRef): assert any(m == model and refresh for m, refresh in reads) @@ -201,3 +202,54 @@ def test_transport_refs_and_face_not_written_when_second_contract_invalid(case): assert c.db.query(c.m.PlTransportCertPurchaseRef).count() == 0 assert c.db.query(c.m.PlContractCertLink).count() == 0 assert not c.db.dirty and not c.db.new + + +@pytest.mark.parametrize('with_bindings', [True, False]) +def test_save_response_uses_current_reads_for_face_bindings_products_and_visibility(case, monkeypatch, with_bindings): + """SQLite 不模拟 RR；捕获实际 ORM 返回查询并以 MySQL 方言验证当前读意图。""" + from sqlalchemy import event + from sqlalchemy.dialects import mysql + from app.services import pl_quota_cert_bindings as svc + c = case + save(c, {'contractBindings': scopes()}) + reads = [] + returning = [False] + original = svc.get_cert + def response(*args, **kwargs): + returning[0] = True + try: + return original(*args, **kwargs) + finally: + returning[0] = False + def capture(state): + if returning[0] and state.is_select: + reads.append((state.bind_mapper.class_, str(state.statement.compile(dialect=mysql.dialect())), + state.load_options._populate_existing)) + monkeypatch.setattr(svc, 'get_cert', response) + event.listen(c.db, 'do_orm_execute', capture) + try: + result = save(c, {'contractBindings': scopes()} if with_bindings else {'remark': 'face only'}) + finally: + event.remove(c.db, 'do_orm_execute', capture) + assert len(result['contractBindings']) == 2 + for model in (c.m.PlPurchaseCert, c.m.PlContractCertLink, c.m.PlContractCertProduct): + queries = [(sql, refresh) for m, sql, refresh in reads if m == model] + assert queries, model.__name__ + assert all('FOR UPDATE' in sql and refresh for sql, refresh in queries), queries + + +def test_plain_get_does_not_acquire_write_locks(case): + from sqlalchemy import event + from app.services import pl_quota_cert_bindings as svc + c = case + save(c, {'contractBindings': scopes()}) + reads = [] + def capture(state): + if state.is_select: + reads.append(state.statement._for_update_arg is not None) + event.listen(c.db, 'do_orm_execute', capture) + try: + result = svc.get_cert(c.db, c.user, 'purchase', 'g1') + finally: + event.remove(c.db, 'do_orm_execute', capture) + assert len(result['contractBindings']) == 2 and not any(reads)