避免联通实名回调因内部长度限制永久失效
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 8m9s
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 8m9s
将实名事件身份收敛为受 Outbox 上限约束的稳定摘要,并允许相同 CUCC 回调原子认领失败终态后重试。 Constraint: 运营商回调固定返回成功,公共 Outbox event_id 上限为 64 字符 Rejected: 扩大数据库字段 | 根因是不受控事件 ID,且无法解决旧 failed 回调被吞 Confidence: high Scope-risk: moderate Directive: 新增稳定事件 ID 时必须服从持久化长度上限;失败回调恢复必须使用条件更新原子认领 Tested: gofmt;git diff --cached --check Not-tested: 按用户要求未运行自动化测试、构建或测试环境重放
This commit is contained in:
@@ -10,7 +10,9 @@
|
|||||||
|
|
||||||
- 合法 ICCID 只按 19/20 位对应列精确查询;不复制旧代码的 20 位截 19 位,不跨列降级,也不任取多匹配卡。
|
- 合法 ICCID 只按 19/20 位对应列精确查询;不复制旧代码的 20 位截 19 位,不跨列降级,也不任取多匹配卡。
|
||||||
- 使用 `ICCID + dateChanged` 的安全摘要作为语义幂等键;解析失败使用正文摘要,重复、冲突和中断恢复沿用 Integration Log 租约规则。
|
- 使用 `ICCID + dateChanged` 的安全摘要作为语义幂等键;解析失败使用正文摘要,重复、冲突和中断恢复沿用 Integration Log 租约规则。
|
||||||
|
- 内部失败终态收到相同正文重推时,原子恢复为 `pending` 并递增尝试次数;成功等其他终态仍保持幂等跳过。
|
||||||
- 唯一命中后调用公共 `ApplyCardObservation(verified=true)`,实名事实变化时由公共用例写 Outbox,并尽力提前完成同卡实名观测序列。
|
- 唯一命中后调用公共 `ApplyCardObservation(verified=true)`,实名事实变化时由公共用例写 Outbox,并尽力提前完成同卡实名观测序列。
|
||||||
|
- 卡实名 Outbox 事件 ID 使用受长度上限约束的稳定摘要,避免上游观测 ID 过长导致业务事务回滚。
|
||||||
- 不调用旧 `inner_callback`、第三方推送、旧平台登录、`ModifyDate` 或 Gateway 二次确认;`dateChanged` 只作为上游变更时间和幂等语义留痕,不直接改写本地业务时间。
|
- 不调用旧 `inner_callback`、第三方推送、旧平台登录、`ModifyDate` 或 Gateway 二次确认;`dateChanged` 只作为上游变更时间和幂等语义留痕,不直接改写本地业务时间。
|
||||||
- 无论开关状态、报文结果或内部处理结果,均返回入口时间对应的 HTTP 200 固定 JSON 应答,避免运营商不可控重推。
|
- 无论开关状态、报文结果或内部处理结果,均返回入口时间对应的 HTTP 200 固定 JSON 应答,避免运营商不可控重推。
|
||||||
|
|
||||||
|
|||||||
@@ -3,6 +3,8 @@ package cardobservation
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"crypto/sha256"
|
||||||
|
"encoding/hex"
|
||||||
"strconv"
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -98,7 +100,7 @@ func (s *Service) ApplyCardObservation(ctx context.Context, observation domain.R
|
|||||||
return errors.New(errors.CodeConflict, "卡实名状态已被其他请求更新")
|
return errors.New(errors.CodeConflict, "卡实名状态已被其他请求更新")
|
||||||
}
|
}
|
||||||
if decision.StatusChanged {
|
if decision.StatusChanged {
|
||||||
eventID := "card-realname:" + strconv.FormatUint(uint64(card.ID), 10) + ":" + observation.Metadata.ObservationID + ":changed"
|
eventID := realnameChangedEventID(card.ID, observation.Metadata.ObservationID)
|
||||||
if err := s.eventWriter.AppendRealname(ctx, tx, RealnameChangedEvent{
|
if err := s.eventWriter.AppendRealname(ctx, tx, RealnameChangedEvent{
|
||||||
EventID: eventID, CardID: card.ID, BeforeStatus: card.RealNameStatus, AfterStatus: decision.AfterStatus,
|
EventID: eventID, CardID: card.ID, BeforeStatus: card.RealNameStatus, AfterStatus: decision.AfterStatus,
|
||||||
FirstVerified: decision.FirstVerified, ObservedAt: observation.Metadata.ObservedAt,
|
FirstVerified: decision.FirstVerified, ObservedAt: observation.Metadata.ObservedAt,
|
||||||
@@ -118,3 +120,10 @@ func (s *Service) ApplyCardObservation(ctx context.Context, observation domain.R
|
|||||||
}
|
}
|
||||||
return decision, nil
|
return decision, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func realnameChangedEventID(cardID uint, observationID string) string {
|
||||||
|
prefix := "card-realname:"
|
||||||
|
digest := sha256.Sum256([]byte(strconv.FormatUint(uint64(cardID), 10) + ":" + observationID + ":changed"))
|
||||||
|
// 使用稳定摘要保留幂等语义,同时严格服从公共 Outbox 的字段长度上限。
|
||||||
|
return prefix + hex.EncodeToString(digest[:])[:constants.OutboxEventIDMaxLength-len(prefix)]
|
||||||
|
}
|
||||||
|
|||||||
@@ -78,7 +78,7 @@ func (h *CUCCRealnameHandler) process(ctx context.Context, body []byte, contentT
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if !created {
|
if !created {
|
||||||
claimed, claimErr := h.claimPending(ctx, log)
|
claimed, claimErr := h.claimRetryable(ctx, log)
|
||||||
if claimErr != nil || !claimed {
|
if claimErr != nil || !claimed {
|
||||||
return claimErr
|
return claimErr
|
||||||
}
|
}
|
||||||
@@ -129,7 +129,7 @@ func (h *CUCCRealnameHandler) recordConflict(ctx context.Context, body []byte, c
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if !created {
|
if !created {
|
||||||
claimed, claimErr := h.claimPending(ctx, log)
|
claimed, claimErr := h.claimRetryable(ctx, log)
|
||||||
if claimErr != nil || !claimed {
|
if claimErr != nil || !claimed {
|
||||||
return claimErr
|
return claimErr
|
||||||
}
|
}
|
||||||
@@ -137,11 +137,18 @@ func (h *CUCCRealnameHandler) recordConflict(ctx context.Context, body []byte, c
|
|||||||
return h.complete(ctx, log.IntegrationID, constants.IntegrationResultConflict, false, "同一实名语义对应不同载荷")
|
return h.complete(ctx, log.IntegrationID, constants.IntegrationResultConflict, false, "同一实名语义对应不同载荷")
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *CUCCRealnameHandler) claimPending(ctx context.Context, log *model.IntegrationLog) (bool, error) {
|
func (h *CUCCRealnameHandler) claimRetryable(ctx context.Context, log *model.IntegrationLog) (bool, error) {
|
||||||
if log == nil || log.Result != constants.IntegrationResultPending {
|
if log == nil {
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
switch log.Result {
|
||||||
|
case constants.IntegrationResultPending:
|
||||||
return h.integration.ClaimExpiredInboundPending(ctx, log.IntegrationID, constants.IntegrationInboundProcessingLease)
|
return h.integration.ClaimExpiredInboundPending(ctx, log.IntegrationID, constants.IntegrationInboundProcessingLease)
|
||||||
|
case constants.IntegrationResultFailed:
|
||||||
|
return h.integration.ClaimFailedInbound(ctx, log.IntegrationID)
|
||||||
|
default:
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *CUCCRealnameHandler) complete(ctx context.Context, integrationID, result string, changed bool, reason string) error {
|
func (h *CUCCRealnameHandler) complete(ctx context.Context, integrationID, result string, changed bool, reason string) error {
|
||||||
|
|||||||
@@ -256,6 +256,29 @@ func (r *Repository) ClaimExpiredInboundPending(ctx context.Context, integration
|
|||||||
return result.RowsAffected == 1, nil
|
return result.RowsAffected == 1, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ClaimFailedInbound 原子认领失败的入站回调并恢复为待处理状态。
|
||||||
|
func (r *Repository) ClaimFailedInbound(ctx context.Context, integrationID string) (bool, error) {
|
||||||
|
if r == nil || r.db == nil {
|
||||||
|
return false, pkgerrors.New(pkgerrors.CodeInvalidStatus, "Integration Log 数据库未配置")
|
||||||
|
}
|
||||||
|
if strings.TrimSpace(integrationID) == "" {
|
||||||
|
return false, pkgerrors.New(pkgerrors.CodeInvalidParam, "入站 Integration Log 恢复参数无效")
|
||||||
|
}
|
||||||
|
now := r.now().UTC()
|
||||||
|
result := r.db.WithContext(ctx).Model(&model.IntegrationLog{}).
|
||||||
|
Where("integration_id = ? AND direction = ? AND result = ?", integrationID, constants.IntegrationDirectionInbound, constants.IntegrationResultFailed).
|
||||||
|
Updates(map[string]any{
|
||||||
|
"result": constants.IntegrationResultPending, "started_at": now,
|
||||||
|
"attempt": gorm.Expr("attempt + 1"), "updated_at": now,
|
||||||
|
"http_status": nil, "provider_code": nil, "provider_message": nil,
|
||||||
|
"response_summary": nil, "duration_ms": 0, "state_changed": false, "recovery_strategy": nil,
|
||||||
|
})
|
||||||
|
if result.Error != nil {
|
||||||
|
return false, pkgerrors.Wrap(pkgerrors.CodeDatabaseError, result.Error, "认领失败入站 Integration Log 失败")
|
||||||
|
}
|
||||||
|
return result.RowsAffected == 1, nil
|
||||||
|
}
|
||||||
|
|
||||||
func validateAttempt(input Attempt) error {
|
func validateAttempt(input Attempt) error {
|
||||||
if input.Provider == "" || input.Operation == "" {
|
if input.Provider == "" || input.Operation == "" {
|
||||||
return pkgerrors.New(pkgerrors.CodeInvalidParam, "Integration Log 提供方和操作不能为空")
|
return pkgerrors.New(pkgerrors.CodeInvalidParam, "Integration Log 提供方和操作不能为空")
|
||||||
|
|||||||
@@ -14,6 +14,8 @@ const (
|
|||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
// OutboxEventIDMaxLength 是公共 Outbox 稳定事件ID的数据库长度上限。
|
||||||
|
OutboxEventIDMaxLength = 64
|
||||||
// OutboxDefaultMaxRetries 是公共 Outbox 默认最大重试次数。
|
// OutboxDefaultMaxRetries 是公共 Outbox 默认最大重试次数。
|
||||||
OutboxDefaultMaxRetries = 10
|
OutboxDefaultMaxRetries = 10
|
||||||
// OutboxDefaultLeaseDuration 是 Relay 默认租约时长。
|
// OutboxDefaultLeaseDuration 是 Relay 默认租约时长。
|
||||||
|
|||||||
Reference in New Issue
Block a user