StackSaga Cassandra 恢复目录缓存与容量计算指南 (Directory Caching & Calculations Guide)

es_days_by_year                             ← 第 1 层 (Tier 1): 哪一日历天存在待处理工作？
  └── es_recovery_windows_by_day            ← 第 2 层 (Tier 2): 该日内的哪一个 UTC 分钟窗口？
        └── es_instances_by_recovery_window ← 第 3 层 (Tier 3): 哪个 Pod 实例在该窗口内写入了数据？
              └── es_buckets_by_instance    ← 第 4 层 (Tier 4): 该 Pod 创建了哪些分桶 (Bucket) 分区？
                    └── es_recovery_transactions_by_instance ← 第 5 层 (Tier 5): 实际的事务指针记录。

package org.stacksaga.cassandra.recovery;

import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

@Service
@RequiredArgsConstructor
public class ReactiveRecoveryService {

    private final RecoveryDirectoryTracker tracker;
    private final CassandraRecoveryDao recoveryDao;

    /**
     * 注册瞬态失败的重试事务
     */
    public Mono<Void> registerRetry(String domain, String transactionId, byte[] payload) {
        // 步骤 1: 非阻塞方式极速分配槽位 (~5 纳秒)
        return tracker.allocateReactive(domain, RecoveryType.RETRY)
            .flatMap(allocation -> {
                // 步骤 2: 基于延迟确认标志位进行分支判断
                if (allocation.requiresParentDirectoryInsert()) {
                    // 未确认状态: 并发写入第 1–4 层，DB 成功后置为已确认，随后追加第 5 层
                    return Mono.when(
                        recoveryDao.insertTier1DaysByYear(allocation.targetDate()),
                        recoveryDao.insertTier2WindowsByDay(allocation.targetDate(), allocation.targetMinuteOfDay()),
                        recoveryDao.insertTier3InstanceMarker(allocation.targetDate(), allocation.targetMinuteOfDay()),
                        recoveryDao.insertTier4BucketIndex(allocation.targetDate(), allocation.targetMinuteOfDay(), allocation.bucketIndex())
                    )
                    // 一旦 Cassandra 返回成功，立即在内存中翻转确认标志!
                    .doOnSuccess(unused -> tracker.markBucketConfirmed(domain, RecoveryType.RETRY, allocation))
                    .then(recoveryDao.insertTier5RetryTransaction(allocation, transactionId, payload));
                } else {
                    // 快速通道 (已确认状态): 零阻塞等待，仅写入第 5 层
                    return recoveryDao.insertTier5RetryTransaction(allocation, transactionId, payload);
                }
            });
    }

    /**
     * 在事务启动时注册崩溃守护看门狗记录
     */
    public Mono<String> registerRestoreWatchdog(String domain, String transactionId) {
        return tracker.allocateReactive(domain, RecoveryType.RESTORE)
            .flatMap(allocation -> {
                if (allocation.requiresParentDirectoryInsert()) {
                    return Mono.when(
                        recoveryDao.insertTier1DaysByYear(allocation.targetDate()),
                        recoveryDao.insertTier2WindowsByDay(allocation.targetDate(), allocation.targetMinuteOfDay()),
                        recoveryDao.insertTier3InstanceMarker(allocation.targetDate(), allocation.targetMinuteOfDay()),
                        recoveryDao.insertTier4BucketIndex(allocation.targetDate(), allocation.targetMinuteOfDay(), allocation.bucketIndex())
                    )
                    .doOnSuccess(unused -> tracker.markBucketConfirmed(domain, RecoveryType.RESTORE, allocation))
                    .then(recoveryDao.insertTier5RestoreWatchdog(allocation, transactionId))
                    .thenReturn(buildWatchdogPath(allocation, transactionId));
                } else {
                    return recoveryDao.insertTier5RestoreWatchdog(allocation, transactionId)
                            .thenReturn(buildWatchdogPath(allocation, transactionId));
                }
            });
    }

    private String buildWatchdogPath(RecoveryAllocationResult alloc, String txId) {
        return alloc.targetDate() + "/" + alloc.targetMinuteOfDay() + "/" + alloc.bucketIndex() + "/" + txId;
    }
}

@Scheduled(fixedDelay = 300_000) // 每 5 分钟执行一次
public void pruneStaleRecoveryWindows() {
    LocalDateTime nowUtc = LocalDateTime.now(ZoneOffset.UTC);
    int currentMinuteOfDay = nowUtc.getHour() * 60 + nowUtc.getMinute();

    // 清理所有已经成为过去式的时间窗口状态
    tracker.pruneExpiredWindows(nowUtc.toLocalDate(), currentMinuteOfDay);
}