修复门禁:增加外部操作原子派发与崩溃保护

This commit is contained in:
wangzhibiao committed 2026-09-21 14:57:21 +08:00
1 parent 71bde5e674
commit 3349675183
6 files changed
+193 -9

No files matched your search

@@ -38,17 +38,21 @@ public class TrainExternalOperation extends BaseEntity
public static final String TYPE_GRANT = "GRANT";
public static final String TYPE_REVOKE = "REVOKE";
public static final String TYPE_STATUS_QUERY = "STATUS_QUERY";
public static final String OP_TYPE_GRANT = TYPE_GRANT;
public static final String OP_TYPE_REVOKE = TYPE_REVOKE;
public static final String STATUS_CREATED = "CREATED";
public static final String STATUS_DISPATCHING = "DISPATCHING";
public static final String STATUS_REQUESTED = "REQUESTED";
public static final String STATUS_ACCEPTED = "ACCEPTED";
public static final String STATUS_PENDING = "PENDING";
public static final String STATUS_RETRY_WAIT = "RETRY_WAIT";
public static final String STATUS_AMBIGUOUS = "AMBIGUOUS";
public static final String STATUS_SUCCESS = "SUCCESS";
public static final String STATUS_FAILED = "FAILED";
public static final String STATUS_TIMEOUT = "TIMEOUT";
/** 操作状态(CREATED, REQUESTED, ACCEPTED, PENDING, RETRY_WAIT, SUCCESS, FAILED, TIMEOUT) */
/** 操作状态(CREATED, DISPATCHING, REQUESTED, ACCEPTED, PENDING, RETRY_WAIT, AMBIGUOUS, SUCCESS, FAILED, TIMEOUT) */
@Excel(name = "操作状态")
private String operationStatus;
@@ -148,6 +148,25 @@ public class DmsOperationPoller {
public int pollPendingOperations() {
Instant now = clock.instant();
Date threshold = Date.from(now);
// BLK-03 / 崩溃恢复语义:长时间处于 DISPATCHING 的操作必须进入 AMBIGUOUS,严禁自动重新 dispatch
try {
Date staleThreshold = Date.from(now.minusSeconds(60));
List<TrainExternalOperation> staleOps = externalOperationService.selectStaleDispatchingOperations(staleThreshold);
if (staleOps != null && !staleOps.isEmpty()) {
for (TrainExternalOperation stale : staleOps) {
externalOperationService.transitStatus(stale.getOperationId(), TrainExternalOperation.STATUS_AMBIGUOUS,
"AMBIGUOUS_EXTERNAL_MUTATION", "Operation timed out while in DISPATCHING state; external mutation outcome unknown", "DmsPollerCrashCleanup");
if (stale.getDoorAuthId() != null && dormAuthService != null) {
dormAuthService.updateAuthStatus(stale.getDoorAuthId(), TrainDormAuth.AUTH_STATUS_ERROR, stale.getOperationId(), "DmsPollerCrashCleanup");
}
log.warn("Recovered stale DISPATCHING operation {} to AMBIGUOUS (crash protection guard)", stale.getOperationId());
}
}
} catch (Exception e) {
log.error("Error during stale dispatching cleanup: {}", e.getMessage());
}
List<TrainExternalOperation> pendingList = externalOperationService.listPendingPollOperations("DMS", threshold);
if (pendingList == null || pendingList.isEmpty()) {
@@ -342,25 +361,25 @@ public class DmsOperationPoller {
if (TrainExternalOperation.STATUS_SUCCESS.equals(terminalStatus)) {
if (TrainExternalOperation.TYPE_GRANT.equals(opType)) {
int updated = dormAuthService.updateAuthStatusByPoller(
op.getDoorAuthId(), TrainDormAuth.AUTH_STATUS_ACTIVE, op.getOperationId(), null, "DmsPoller");
op.getDoorAuthId(), TrainDormAuth.AUTH_STATUS_ACTIVE, op.getOperationId(), opType, null, "DmsPoller");
if (updated > 0) {
log.info("DoorAuthorization {} advanced to ACTIVE by Poller operation {}", op.getDoorAuthId(), op.getOperationId());
} else {
log.warn("DoorAuthorization {} update to ACTIVE skipped (Stale operation guard): opId={}", op.getDoorAuthId(), op.getOperationId());
log.warn("DoorAuthorization {} update to ACTIVE skipped (Stale/Compatibility guard): opId={}", op.getDoorAuthId(), op.getOperationId());
}
} else if (TrainExternalOperation.TYPE_REVOKE.equals(opType)) {
int updated = dormAuthService.updateAuthStatusByPoller(
op.getDoorAuthId(), TrainDormAuth.AUTH_STATUS_REVOKED, op.getOperationId(), null, "DmsPoller");
op.getDoorAuthId(), TrainDormAuth.AUTH_STATUS_REVOKED, op.getOperationId(), opType, null, "DmsPoller");
if (updated > 0) {
log.info("DoorAuthorization {} advanced to REVOKED by Poller operation {}", op.getDoorAuthId(), op.getOperationId());
} else {
log.warn("DoorAuthorization {} update to REVOKED skipped (Stale operation guard): opId={}", op.getDoorAuthId(), op.getOperationId());
log.warn("DoorAuthorization {} update to REVOKED skipped (Stale/Compatibility guard): opId={}", op.getDoorAuthId(), op.getOperationId());
}
}
} else {
// 外部设备执行失败或超时:推进为 ERROR 状态,防止伪成功
dormAuthService.updateAuthStatusByPoller(
op.getDoorAuthId(), TrainDormAuth.AUTH_STATUS_ERROR, op.getOperationId(), errorMsg, "DmsPoller");
op.getDoorAuthId(), TrainDormAuth.AUTH_STATUS_ERROR, op.getOperationId(), opType, errorMsg, "DmsPoller");
}
}
@@ -398,6 +417,15 @@ public class DmsOperationPoller {
return;
}
// DB级原子 Claim 抢占派发权 (BLK-03)
if (externalOperationService != null) {
boolean claimed = externalOperationService.claimOperationForDispatch(op.getOperationId());
if (!claimed) {
log.info("Operation {} already claimed by concurrent dispatcher, skipping poller compensation", op.getOperationId());
return;
}
}
try {
DmsPasswordDeleteResponse resp = credentialGateway.revokeOnlinePassword(externalKeyId);
String extOpIdStr = resp.getOperationId() != null ? String.valueOf(resp.getOperationId()) : null;
@@ -407,8 +435,18 @@ public class DmsOperationPoller {
externalOperationService.transitStatus(op.getOperationId(), TrainExternalOperation.STATUS_ACCEPTED, null, null, "DmsPollerCrashCompensation");
}
log.info("DmsPoller successfully compensated CREATED revoke operation {}: extOpId={}", op.getOperationId(), extOpIdStr);
} catch (DmsBusinessException e) {
handlePollError(op, now, ERROR_CATEGORY_VENDOR_BIZ, sanitize(e.getMessage()), true);
} catch (Exception e) {
handlePollError(op, now, ERROR_CATEGORY_NETWORK, sanitize(e.getMessage()), false);
// 外部请求可能已发出的网络传输/超时异常,进入 AMBIGUOUS,禁止自动重试!(BLK-01)
log.error("Ambiguous error during poller compensation for op {}: {}", op.getOperationId(), sanitize(e.getMessage()));
if (externalOperationService != null) {
externalOperationService.transitStatus(op.getOperationId(), TrainExternalOperation.STATUS_AMBIGUOUS,
"AMBIGUOUS_EXTERNAL_MUTATION", sanitize(e.getMessage()), "DmsPollerCrashCompensation");
}
if (dormAuthService != null) {
dormAuthService.updateAuthStatus(op.getDoorAuthId(), TrainDormAuth.AUTH_STATUS_ERROR, op.getOperationId(), "DmsPollerCrashCompensation");
}
}
}
@@ -100,4 +100,37 @@ public interface TrainExternalOperationMapper
@Param("externalHardwareId") String externalHardwareId,
@Param("externalKeyId") String externalKeyId,
@Param("updateBy") String updateBy);
/**
* DB级原子Claim操作派发权 (BLK-03)
* 仅当处于 CREATED 状态时原子推进为 DISPATCHING
*
* @param operationId 本地流水ID
* @return 影响行数 (恰好为1表示成功获取唯一派发权)
*/
public int claimOperationForDispatch(@Param("operationId") Long operationId);
/**
* 将超时停留在 DISPATCHING 状态的悬挂流水标记为 AMBIGUOUS (BLK-01 / BLK-03)
*
* @param staleThreshold 超时阈值时间
* @return 影响行数
*/
public int markStaleDispatchingAsAmbiguous(@Param("staleThreshold") Date staleThreshold);
/**
* 查询所有停留在 DISPATCHING 且已超时的悬挂操作
*
* @param staleThreshold 超时阈值时间
* @return 超时操作列表
*/
public List<TrainExternalOperation> selectStaleDispatchingOperations(@Param("staleThreshold") Date staleThreshold);
/**
* 查询指定门禁授权下最新的 GRANT 外部操作
*
* @param doorAuthId 门禁授权ID
* @return 最新 GRANT 操作流水
*/
public TrainExternalOperation selectLatestGrantOperationByDoorAuthId(@Param("doorAuthId") Long doorAuthId);
}
@@ -128,4 +128,36 @@ public interface IExternalOperationService
String errorCode,
String errorMessage,
String operator);
/**
* DB级原子抢占派发权 (BLK-03)
*
* @param operationId 本地流水ID
* @return true表示成功获得唯一派发权
*/
public boolean claimOperationForDispatch(Long operationId);
/**
* 将长时间处于 DISPATCHING 的悬挂操作置为 AMBIGUOUS (BLK-01 / BLK-03)
*
* @param staleThreshold 超时阈值
* @return 影响行数
*/
public int markStaleDispatchingAsAmbiguous(Date staleThreshold);
/**
* 查询超时的 DISPATCHING 操作列表
*
* @param staleThreshold 超时阈值
* @return 超时操作列表
*/
public List<TrainExternalOperation> selectStaleDispatchingOperations(Date staleThreshold);
/**
* 获取指定门禁授权下最新的 GRANT 外部操作
*
* @param doorAuthId 门禁授权ID
* @return 最新 GRANT 操作流水
*/
public TrainExternalOperation selectLatestGrantOperationByDoorAuthId(Long doorAuthId);
}
@@ -31,7 +31,7 @@ public class ExternalOperationServiceImpl implements IExternalOperationService
));
private static final Set<String> TERMINAL_STATUSES = new HashSet<>(Arrays.asList(
"SUCCESS", "FAILED", "TIMEOUT"
"SUCCESS", "FAILED", "TIMEOUT", "AMBIGUOUS"
));
@Autowired
@@ -292,4 +292,47 @@ public class ExternalOperationServiceImpl implements IExternalOperationService
operationMapper.updateExternalOperationStatus(op);
return op;
}
@Override
@Transactional(rollbackFor = Exception.class)
public boolean claimOperationForDispatch(Long operationId)
{
if (operationId == null)
{
return false;
}
int rows = operationMapper.claimOperationForDispatch(operationId);
return rows == 1;
}
@Override
@Transactional(rollbackFor = Exception.class)
public int markStaleDispatchingAsAmbiguous(Date staleThreshold)
{
if (staleThreshold == null)
{
return 0;
}
return operationMapper.markStaleDispatchingAsAmbiguous(staleThreshold);
}
@Override
public List<TrainExternalOperation> selectStaleDispatchingOperations(Date staleThreshold)
{
if (staleThreshold == null)
{
return Collections.emptyList();
}
return operationMapper.selectStaleDispatchingOperations(staleThreshold);
}
@Override
public TrainExternalOperation selectLatestGrantOperationByDoorAuthId(Long doorAuthId)
{
if (doorAuthId == null)
{
return null;
}
return operationMapper.selectLatestGrantOperationByDoorAuthId(doorAuthId);
}
}
@@ -64,7 +64,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<select id="selectPendingPollOperations" resultMap="TrainExternalOperationResult">
<include refid="selectTrainExternalOperationVo"/>
WHERE operation_status IN ('CREATED', 'ACCEPTED', 'PENDING', 'RETRY_WAIT')
WHERE ((operation_status = 'CREATED' AND operation_type = 'REVOKE')
OR operation_status IN ('ACCEPTED', 'PENDING', 'RETRY_WAIT'))
AND (next_poll_time IS NULL OR next_poll_time &lt;= #{maxPollTime})
<if test="vendorCode != null and vendorCode != ''">
AND vendor_code = #{vendorCode}
@@ -171,4 +172,37 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
WHERE operation_id = #{operationId}
</update>
<update id="claimOperationForDispatch">
UPDATE train_external_operation
SET operation_status = 'DISPATCHING',
update_time = CURRENT_TIMESTAMP
WHERE operation_id = #{operationId}
AND operation_status = 'CREATED'
</update>
<update id="markStaleDispatchingAsAmbiguous">
UPDATE train_external_operation
SET operation_status = 'AMBIGUOUS',
last_error_code = 'AMBIGUOUS_EXTERNAL_MUTATION',
last_error_message = 'Operation timed out while in DISPATCHING state; external mutation outcome unknown',
update_time = CURRENT_TIMESTAMP
WHERE operation_status = 'DISPATCHING'
AND update_time &lt;= #{staleThreshold}
</update>
<select id="selectStaleDispatchingOperations" resultMap="TrainExternalOperationResult">
<include refid="selectTrainExternalOperationVo"/>
WHERE operation_status = 'DISPATCHING'
AND update_time &lt;= #{staleThreshold}
ORDER BY operation_id ASC
</select>
<select id="selectLatestGrantOperationByDoorAuthId" resultMap="TrainExternalOperationResult">
<include refid="selectTrainExternalOperationVo"/>
WHERE door_auth_id = #{doorAuthId}
AND operation_type = 'GRANT'
ORDER BY operation_id DESC
LIMIT 1
</select>
</mapper>