fix(pl): Read current certificate bindings when returning atomic saves.

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)

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)