Initial commit: bositao cloud (jnpf-java-cloud-v6x)

刘光辉

4 小时以前 06f314545ac3514d7ce28939920866ae8188d31c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
package jnpf.dmsService;
 
import jnpf.dmsEntity.storage.approval.DmsApprovedBatch;
import jnpf.dmsMapper.DmsApprovedBatchMapper;
import jnpf.dmsService.impl.DmsWenjianBanbenServiceImpl;
import jnpf.exception.DataException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.beans.factory.annotation.Value;
import java.util.UUID;
 
/** Persistent retry orchestration. Never hold a database transaction across a file RPC. */
@Slf4j
@Service
@RequiredArgsConstructor
public class DmsApprovedFileRetryService {
    private final DmsApprovedBatchMapper receipts;
    private final DmsApprovedFileAcceptanceService acceptance;
    private final DmsApprovedContentService files;
    private final DmsApprovedFileActivationService activation;
    private final DmsWenjianBanbenServiceImpl business;
    private final DmsApprovedDisableProcessor disables;
    private final DmsInvalidationProperties invalidation;
 
    @Value("${dms.approved-files.retry-backoff-enabled:true}")
    private boolean retryBackoffEnabled = true;
 
    public void process(DmsApprovedBatch candidate) {
        if (DmsApprovedDisableProcessor.supports(candidate.getAuditType()) && !disables.isEnabled()) return;
        if ("shixiao_shenqing".equals(candidate.getAuditType())
                && !invalidation.allows(candidate.getTenantId(), candidate.getId())) return;
        String token = UUID.randomUUID().toString();
        if (receipts.claim(candidate.getTenantId(), candidate.getId(), token, retryBackoffEnabled) != 1) return;
        DmsApprovedBatch claimed = null;
        try {
            claimed = receipts.find(candidate.getTenantId(), candidate.getTaskId(), candidate.getAuditType());
            if (claimed == null || !token.equals(claimed.getLeaseToken())) throw new DataException("登记租约归属不匹配");
            if (DmsApprovedDisableProcessor.supports(claimed.getAuditType())) {
                disables.process(claimed);
            } else if ("PREPARING".equals(claimed.getStatus()) || "REGISTERING".equals(claimed.getStatus())) {
                DmsApprovedBatch staged = acceptance.stage(claimed.getTenantId(), claimed.getFlowId(), claimed.getTaskId(),
                        claimed.getAuditType(), claimed.getActorId(), claimed.getApprovedAt());
                staged.setLeaseToken(token);
                if ("REGISTERING".equals(staged.getStatus())) {
                    DmsApprovedBatch accepted = files.register(staged);
                    if (accepted == null) throw new DataException("文件登记未返回进度");
                    advance(accepted);
                } else advance(staged);
            } else advance(claimed);
            receipts.finish(candidate.getTenantId(), candidate.getId(), token, null, 0);
        } catch (RuntimeException failure) {
            String error = failure.getMessage() == null ? failure.getClass().getSimpleName() : failure.getMessage();
            // Keep errors bounded; no file payloads or credentials are persisted here.
            error = error.substring(0, Math.min(error.length(), 1000));
            try {
                receipts.finish(candidate.getTenantId(), candidate.getId(), token, error,
                        retryBackoffEnabled ? retryDelay(claimed == null ? 1 : claimed.getAttempts()) : 0);
            } catch (RuntimeException persistenceFailure) {
                failure.addSuppressed(persistenceFailure); // Expired lease recovers a failed receipt write.
            }
            log.error("批准文件处理失败，已保留进度待自动重试 tenant={} batch={}", candidate.getTenantId(), candidate.getId(), failure);
            throw failure;
        }
    }
 
    static int retryDelay(int attempts) {
        return Math.min(3600, 60 * (1 << Math.min(6, Math.max(0, attempts - 1))));
    }
 
    public void advance(DmsApprovedBatch receipt) {
        if (DmsApprovedDisableProcessor.supports(receipt.getAuditType()))
            throw new DataException("停用任务必须由统一重试入口分派");
        DmsApprovedBatch activated;
        try {
            activated = activation.activate(receipt.getTenantId(), receipt.getId());
        } catch (RuntimeException failure) {
            receipts.fail(receipt.getTenantId(), receipt.getId());
            throw failure;
        }
        if ("EFFECTIVE".equals(activated.getStatus())) business.applyApproved(activated);
        if ("FAILED".equals(activated.getStatus())) throw new DataException("该批准文件生效失败，等待自动恢复");
    }
}