StackSaga Cassandra Recovery Directory Caching & Calculations Guide

es_days_by_year                             ← Tier 1: Which calendar days have pending work?
  └── es_recovery_windows_by_day            ← Tier 2: Which UTC minute windows on that day?
        └── es_instances_by_recovery_window ← Tier 3: Which pod instances wrote work in that window?
              └── es_buckets_by_instance    ← Tier 4: Which bucket partitions did that pod create?
                    └── es_recovery_transactions_by_instance ← Tier 5: The actual transaction pointers.

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;

    /**
     * Registers a transient retry failure.
     */
    public Mono<Void> registerRetry(String domain, String transactionId, byte[] payload) {
        // Step 1: Allocate slot non-blockingly (~5 nanoseconds)
        return tracker.allocateReactive(domain, RecoveryType.RETRY)
            .flatMap(allocation -> {
                // Step 2: Branch based on deferred confirmation check
                if (allocation.requiresParentDirectoryInsert()) {
                    // Unconfirmed: Upsert Tiers 1-4, mark confirmed on DB success, then append Tier 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())
                    )
                    // Once Cassandra confirms success, flip confirmed flag in memory!
                    .doOnSuccess(unused -> tracker.markBucketConfirmed(domain, RecoveryType.RETRY, allocation))
                    .then(recoveryDao.insertTier5RetryTransaction(allocation, transactionId, payload));
                } else {
                    // Fast Path (Already confirmed): Append to Tier 5 ONLY with zero waiting
                    return recoveryDao.insertTier5RetryTransaction(allocation, transactionId, payload);
                }
            });
    }

    /**
     * Registers a dead-man's switch watchdog at transaction start.
     */
    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) // Runs every 5 minutes
public void pruneStaleRecoveryWindows() {
    LocalDateTime nowUtc = LocalDateTime.now(ZoneOffset.UTC);
    int currentMinuteOfDay = nowUtc.getHour() * 60 + nowUtc.getMinute();

    // Purges any window tracking state strictly in the past
    tracker.pruneExpiredWindows(nowUtc.toLocalDate(), currentMinuteOfDay);
}