document-haness/docs/clean-architecture-backend-template/final/evidence/raw/079a-fileserver-cleanup-crash-budget-probe.java

package dev.caskeleton.adapter.outbound.persistence.readiness;

import static org.assertj.core.api.Assertions.assertThat;

import dev.caskeleton.adapter.outbound.persistence.fileserver.JpaCleanupQueue;
import dev.caskeleton.adapter.outbound.persistence.fileserver.JpaContentReferenceLedger;
import dev.caskeleton.adapter.outbound.persistence.fileserver.JpaFileMetadataStore;
import dev.caskeleton.adapter.outbound.persistence.fileserver.JpaFileQuotaService;
import dev.caskeleton.adapter.outbound.persistence.fileserver.JpaQuotaCommitGateway;
import dev.caskeleton.adapter.outbound.persistence.fileserver.JpaQuotaReclaimGateway;
import dev.caskeleton.adapter.outbound.persistence.fileserver.JpaRecoveryQueue;
import dev.caskeleton.adapter.outbound.persistence.fileserver.JpaStagingUploadLocator;
import dev.caskeleton.adapter.outbound.persistence.fileserver.JpaUploadSessionStore;
import dev.caskeleton.application.fileserver.api.ContentKey;
import dev.caskeleton.application.fileserver.api.FileId;
import dev.caskeleton.application.fileserver.api.FileState;
import dev.caskeleton.application.fileserver.api.StorageNamespace;
import dev.caskeleton.application.fileserver.api.UploadId;
import dev.caskeleton.application.fileserver.api.metadata.FileRecord;
import dev.caskeleton.application.fileserver.api.metadata.FileRecordDraft;
import dev.caskeleton.application.fileserver.api.metadata.FileRecordMutation;
import dev.caskeleton.application.fileserver.api.metadata.QuotaScope;
import dev.caskeleton.application.fileserver.api.metadata.UploadSession;
import dev.caskeleton.application.fileserver.api.metadata.UploadSessionDraft;
import dev.caskeleton.application.fileserver.api.transfer.UploadProtocol;
import dev.caskeleton.application.fileserver.cleanup.CleanupItem;
import dev.caskeleton.application.fileserver.cleanup.CleanupRequest;
import dev.caskeleton.application.fileserver.cleanup.CleanupType;
import dev.caskeleton.application.fileserver.recovery.ReconciliationStatus;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.UUID;
import java.util.function.Supplier;
import org.flywaydb.core.Flyway;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.support.TransactionTemplate;

/**
 * Real-PostgreSQL proof of the reclamation side of the Fileserver.
 *
 * <p>The cleanup queue, the recovery queue, and the quota ledger are the components that decide
 * when physical bytes may be destroyed and how much space a tenant is charged for. All three are
 * conditional-update designs whose correctness lives in SQL, so an in-memory fake would prove
 * nothing about them: the partial unique index, the {@code IN ('PENDING','FAILED')} claim guard,
 * and the {@code committedBytes >= :amount} floor only exist in the database.
 */
class PostgreSqlFileserverReclamationIntegrationTest {

  private static PostgreSqlReadinessSupport postgres;
  private static FileserverJpaTestContext context;
  private static JdbcTemplate jdbc;
  private static TransactionTemplate transactions;
  private static JpaFileMetadataStore files;
  private static JpaUploadSessionStore uploads;
  private static JpaFileQuotaService quota;
  private static JpaCleanupQueue cleanupQueue;
  private static JpaRecoveryQueue recoveryQueue;
  private static JpaQuotaCommitGateway quotaCommit;
  private static JpaQuotaReclaimGateway quotaReclaim;
  private static JpaContentReferenceLedger ledger;
  private static JpaStagingUploadLocator stagingLocator;

  @BeforeAll
  static void startAndMigratePostgreSql() {
    PostgreSqlReadinessSupport.assertDockerAvailable();
    postgres = PostgreSqlReadinessSupport.start(8, 2_000);
    migrate("classpath:db/migration/postgresql", "flyway_schema_history");
    migrateIndependent(
        "classpath:db/migration/jpa/core", "flyway_jpa_core_history", "explicit-jpa-core-adoption");
    migrateIndependent(
        "classpath:db/migration/jpa/fileserver",
        "flyway_jpa_fileserver_history",
        "explicit-jpa-fileserver-adoption");

    jdbc = new JdbcTemplate(postgres.dataSource());
    context = new FileserverJpaTestContext(postgres.dataSource(), Clock.systemUTC());
    transactions = context.transactions();
    files = context.bean(JpaFileMetadataStore.class);
    uploads = context.bean(JpaUploadSessionStore.class);
    quota = context.bean(JpaFileQuotaService.class);
    cleanupQueue = context.bean(JpaCleanupQueue.class);
    recoveryQueue = context.bean(JpaRecoveryQueue.class);
    quotaCommit = context.bean(JpaQuotaCommitGateway.class);
    quotaReclaim = context.bean(JpaQuotaReclaimGateway.class);
    ledger = context.bean(JpaContentReferenceLedger.class);
    stagingLocator = context.bean(JpaStagingUploadLocator.class);
  }

  @AfterAll
  static void stopPostgreSql() {
    if (context != null) {
      context.close();
    }
    if (postgres != null) {
      postgres.close();
    }
  }

  @BeforeEach
  void truncate() {
    jdbc.execute(
        "TRUNCATE fs_recovery_item, fs_cleanup_item, fs_quota_reservation,"
            + " fs_verification_result, fs_upload_session, fs_file CASCADE");
  }

  @Test
  void aStagingCleanupSurvivesTheRoundTripWithItsUploadIdentity() {
    FileRecord record = insertRecord();
    UploadId uploadId = UploadId.of(UUID.randomUUID());
    inTransaction(
        () -> {
          cleanupQueue.enqueue(
              CleanupRequest.forStaging(CleanupType.CANCELLED_STAGING, record.fileId(), uploadId));
          return null;
        });

    List<CleanupItem> claimed = inTransaction(() -> cleanupQueue.claimDue(Instant.now(), 10));

    assertThat(claimed)
        .singleElement()
        .satisfies(
            item -> {
              assertThat(item.request().uploadId()).contains(uploadId);
              assertThat(item.request().contentKey()).isEmpty();
              assertThat(item.type()).isEqualTo(CleanupType.CANCELLED_STAGING);
            });
  }

  @Test
  void aClaimedItemIsNotHandedToASecondWorker() {
    FileRecord record = insertRecord();
    inTransaction(
        () -> {
          cleanupQueue.enqueue(
              CleanupRequest.forContent(
                  CleanupType.DELETED_READY_CONTENT, record.fileId(), contentKey()));
          return null;
        });

    List<CleanupItem> first = inTransaction(() -> cleanupQueue.claimDue(Instant.now(), 10));
    List<CleanupItem> second = inTransaction(() -> cleanupQueue.claimDue(Instant.now(), 10));

    assertThat(first).hasSize(1);
    assertThat(second).isEmpty();
  }

  @Test
  void aFailedItemBecomesDueAgainOnlyAfterItsBackoff() {
    FileRecord record = insertRecord();
    inTransaction(
        () -> {
          cleanupQueue.enqueue(
              CleanupRequest.forContent(
                  CleanupType.DELETED_READY_CONTENT, record.fileId(), contentKey()));
          return null;
        });
    CleanupItem item = inTransaction(() -> cleanupQueue.claimDue(Instant.now(), 10)).get(0);
    Instant retryAt = Instant.now().plus(Duration.ofMinutes(5));
    inTransaction(
        () -> {
          cleanupQueue.markFailed(item, "STORAGE_UNAVAILABLE", retryAt);
          return null;
        });

    assertThat(inTransaction(() -> cleanupQueue.claimDue(Instant.now(), 10))).isEmpty();
    assertThat(inTransaction(() -> cleanupQueue.claimDue(retryAt.plusSeconds(1), 10))).hasSize(1);
  }

  @Test
  void anItemThatKeepsFailingIsAbandonedRatherThanRetriedForever() {
    FileRecord record = insertRecord();
    inTransaction(
        () -> {
          cleanupQueue.enqueue(
              CleanupRequest.forContent(
                  CleanupType.DELETED_READY_CONTENT, record.fileId(), contentKey()));
          return null;
        });

    Instant due = Instant.now();
    for (int attempt = 0; attempt < JpaCleanupQueue.MAXIMUM_ATTEMPTS; attempt++) {
      List<CleanupItem> claimed = inTransaction(() -> cleanupQueue.claimDue(due, 10));
      if (claimed.isEmpty()) {
        break;
      }
      inTransaction(
          () -> {
            cleanupQueue.markFailed(claimed.get(0), "STILL_FAILING", due);
            return null;
          });
    }

    assertThat(inTransaction(() -> cleanupQueue.claimDue(due.plusSeconds(1), 10))).isEmpty();
    assertThat(statusCounts("fs_cleanup_item", "ABANDONED")).isEqualTo(1);
  }

  @Test
  void aDoneItemIsNeverClaimedAgain() {
    FileRecord record = insertRecord();
    inTransaction(
        () -> {
          cleanupQueue.enqueue(
              CleanupRequest.forContent(
                  CleanupType.DELETED_READY_CONTENT, record.fileId(), contentKey()));
          return null;
        });
    CleanupItem item = inTransaction(() -> cleanupQueue.claimDue(Instant.now(), 10)).get(0);

    inTransaction(
        () -> {
          cleanupQueue.markDone(item);
          return null;
        });

    assertThat(inTransaction(() -> cleanupQueue.claimDue(Instant.now().plusSeconds(600), 10)))
        .isEmpty();
  }

  @Test
  void thesameFileReportedTwiceHoldsOneOpenRecoveryItem() {
    FileRecord record = insertRecord();

    inTransaction(
        () -> {
          recoveryQueue.enqueue(record.fileId(), "READY_DIGEST_MISMATCH");
          recoveryQueue.enqueue(record.fileId(), "READY_SIZE_MISMATCH");
          return null;
        });

    assertThat(inTransaction(() -> recoveryQueue.pending(10))).containsExactly(record.fileId());
    assertThat(rowCount("fs_recovery_item")).isEqualTo(1);
    assertThat(jdbc.queryForObject("SELECT reason_code FROM fs_recovery_item", String.class))
        .isEqualTo("READY_SIZE_MISMATCH");
  }

  @Test
  void aResolvedRecoveryItemLeavesThePendingListButKeepsItsOutcome() {
    FileRecord record = insertRecord();
    inTransaction(
        () -> {
          recoveryQueue.enqueue(record.fileId(), "PUBLISH_EVIDENCE_INCOMPLETE");
          return null;
        });

    inTransaction(
        () -> {
          recoveryQueue.resolve(record.fileId(), ReconciliationStatus.QUARANTINE_REQUIRED);
          return null;
        });

    assertThat(inTransaction(() -> recoveryQueue.pending(10))).isEmpty();
    assertThat(statusCounts("fs_recovery_item", "QUARANTINE_REQUIRED")).isEqualTo(1);
  }

  @Test
  void aResolvedFileCanBeRaisedAgainLater() {
    FileRecord record = insertRecord();
    inTransaction(
        () -> {
          recoveryQueue.enqueue(record.fileId(), "FIRST");
          recoveryQueue.resolve(record.fileId(), ReconciliationStatus.UNRESOLVED);
          recoveryQueue.enqueue(record.fileId(), "SECOND");
          return null;
        });

    assertThat(inTransaction(() -> recoveryQueue.pending(10))).containsExactly(record.fileId());
    assertThat(rowCount("fs_recovery_item")).isEqualTo(2);
  }

  @Test
  void committingAnUploadMovesReservedBytesToCommitted() {
    FileRecord record = insertRecord();
    UploadSession session = insertSession(record.fileId());
    QuotaScope scope = QuotaScope.ofNamespace(record.namespace().value());
    inTransaction(() -> quota.reserve(scope, 1_000, Duration.ofHours(1)));

    inTransaction(
        () -> {
          quotaCommit.commit(session, 600);
          return null;
        });

    assertThat(inTransaction(() -> quota.reservedBytes(scope))).isZero();
    assertThat(inTransaction(() -> quota.committedBytes(scope))).isEqualTo(600);
  }

  @Test
  void committingWithoutALiveReservationStillRecordsTheDurableUsage() {
    FileRecord record = insertRecord();
    UploadSession session = insertSession(record.fileId());
    QuotaScope scope = QuotaScope.ofNamespace(record.namespace().value());

    inTransaction(
        () -> {
          quotaCommit.commit(session, 450);
          return null;
        });

    assertThat(inTransaction(() -> quota.committedBytes(scope))).isEqualTo(450);
  }

  @Test
  void releasingAnUploadGivesTheReservedCapacityBack() {
    FileRecord record = insertRecord();
    UploadSession session = insertSession(record.fileId());
    QuotaScope scope = QuotaScope.ofNamespace(record.namespace().value());
    inTransaction(() -> quota.reserve(scope, 2_000, Duration.ofHours(1)));

    inTransaction(
        () -> {
          quotaCommit.release(session);
          return null;
        });

    assertThat(inTransaction(() -> quota.reservedBytes(scope))).isZero();
    assertThat(inTransaction(() -> quota.committedBytes(scope))).isZero();
  }

  @Test
  void reclaimingDrawsCommittedBytesDownAcrossRows() {
    FileRecord record = insertRecord();
    UploadSession session = insertSession(record.fileId());
    QuotaScope scope = QuotaScope.ofNamespace(record.namespace().value());
    inTransaction(
        () -> {
          quotaCommit.commit(session, 300);
          quotaCommit.commit(session, 700);
          return null;
        });

    inTransaction(
        () -> {
          quotaReclaim.reclaim(scope, 800);
          return null;
        });

    assertThat(inTransaction(() -> quota.committedBytes(scope))).isEqualTo(200);
  }

  @Test
  void reclaimingMoreThanIsRecordedStopsAtZeroRatherThanGoingNegative() {
    FileRecord record = insertRecord();
    UploadSession session = insertSession(record.fileId());
    QuotaScope scope = QuotaScope.ofNamespace(record.namespace().value());
    inTransaction(
        () -> {
          quotaCommit.commit(session, 100);
          return null;
        });

    inTransaction(
        () -> {
          quotaReclaim.reclaim(scope, 5_000);
          return null;
        });

    assertThat(inTransaction(() -> quota.committedBytes(scope))).isZero();
  }

  @Test
  void theLedgerReportsAKeyAsReferencedOnlyWhileARecordNamesIt() {
    ContentKey key = contentKey();
    assertThat(inTransaction(() -> ledger.isReferenced(key))).isFalse();

    FileRecord record = insertRecord();
    inTransaction(() -> publish(record, key));

    assertThat(inTransaction(() -> ledger.isReferenced(key))).isTrue();
  }

  @Test
  void theStagingLocatorReturnsTheNewestSessionForAFile() {
    FileRecord record = insertRecord();
    UploadSession older = insertSession(record.fileId());
    UploadSession newer = insertSession(record.fileId());

    Optional<UploadId> located = inTransaction(() -> stagingLocator.locate(record.fileId()));

    assertThat(located).isPresent();
    assertThat(located.get()).isIn(older.uploadId(), newer.uploadId());
    assertThat(inTransaction(() -> stagingLocator.locate(FileId.of(UUID.randomUUID())))).isEmpty();
  }

  private FileRecord insertRecord() {
    return inTransaction(
        () ->
            files.insert(
                new FileRecordDraft(
                    FileId.of(UUID.randomUUID()),
                    StorageNamespace.of("tenant-a"),
                    "report.bin",
                    Optional.of("application/octet-stream"),
                    OptionalLong.of(1_000))));
  }

  private FileRecord publish(FileRecord record, ContentKey key) {
    FileRecord uploading =
        files.transition(
            record.fileId(),
            record.version(),
            FileState.CREATED,
            FileState.UPLOADING,
            FileRecordMutation.none());
    FileRecord uploaded =
        files.transition(
            uploading.fileId(),
            uploading.version(),
            FileState.UPLOADING,
            FileState.UPLOADED,
            FileRecordMutation.uploaded(10, "b".repeat(64)));
    FileRecord verifying =
        files.transition(
            uploaded.fileId(),
            uploaded.version(),
            FileState.UPLOADED,
            FileState.VERIFYING,
            FileRecordMutation.none());
    return files.transition(
        verifying.fileId(),
        verifying.version(),
        FileState.VERIFYING,
        FileState.READY,
        FileRecordMutation.publishAt(
            key, 10, "b".repeat(64), "\"" + "b".repeat(64) + "\"", Instant.now()));
  }

  private UploadSession insertSession(FileId fileId) {
    return inTransaction(
        () ->
            uploads.create(
                new UploadSessionDraft(
                    UploadId.of(UUID.randomUUID()),
                    fileId,
                    UploadProtocol.RAW,
                    OptionalLong.of(1_000),
                    Instant.now().plus(Duration.ofHours(1)))));
  }

  private static ContentKey contentKey() {
    String flat = UUID.randomUUID().toString().replace("-", "");
    return ContentKey.of(flat.substring(0, 2) + '/' + flat.substring(2, 4) + '/' + flat);
  }

  private static <T> T inTransaction(Supplier<T> action) {
    return transactions.execute(status -> action.get());
  }

  private static int rowCount(String table) {
    Integer count = jdbc.queryForObject("SELECT count(*) FROM " + table, Integer.class);
    return count == null ? 0 : count;
  }

  private static int statusCounts(String table, String status) {
    Integer count =
        jdbc.queryForObject(
            "SELECT count(*) FROM " + table + " WHERE status = ?", Integer.class, status);
    return count == null ? 0 : count;
  }

  private static void migrate(String location, String historyTable) {
    Flyway.configure()
        .dataSource(postgres.dataSource())
        .locations(location)
        .table(historyTable)
        .baselineOnMigrate(false)
        .outOfOrder(false)
        .load()
        .migrate();
  }

  private static void migrateIndependent(
      String location, String historyTable, String baselineDescription) {
    Flyway flyway =
        Flyway.configure()
            .dataSource(postgres.dataSource())
            .locations(location)
            .table(historyTable)
            .baselineVersion("0")
            .baselineDescription(baselineDescription)
            .baselineOnMigrate(false)
            .outOfOrder(false)
            .load();
    flyway.baseline();
    flyway.migrate();
  }

  @Test
  void analysisProbeCrashReclaimsBypassMaximumAttempts() {
    FileRecord record = insertRecord();
    inTransaction(
        () -> {
          cleanupQueue.enqueue(
              CleanupRequest.forContent(
                  CleanupType.DELETED_READY_CONTENT, record.fileId(), contentKey()));
          return null;
        });

    Instant base = Instant.now();
    for (int crash = 0; crash < JpaCleanupQueue.MAXIMUM_ATTEMPTS + 1; crash++) {
      Instant claimAt = base.plus(Duration.ofHours(crash));
      List<CleanupItem> claimed = inTransaction(() -> cleanupQueue.claimDue(claimAt, 10));
      assertThat(claimed).hasSize(1);
      int reclaimed =
          inTransaction(() -> cleanupQueue.reclaimExpiredClaims(claimAt.plus(Duration.ofMinutes(11)), 10));
      assertThat(reclaimed).isEqualTo(1);
    }

    Integer attempt =
        jdbc.queryForObject("SELECT attempt FROM fs_cleanup_item", Integer.class);
    String status =
        jdbc.queryForObject("SELECT status FROM fs_cleanup_item", String.class);
    String error =
        jdbc.queryForObject("SELECT last_error_code FROM fs_cleanup_item", String.class);
    System.out.println("fileserverCleanupCrash.maxAttempts=" + JpaCleanupQueue.MAXIMUM_ATTEMPTS);
    System.out.println("fileserverCleanupCrash.actualAttempt=" + attempt);
    System.out.println("fileserverCleanupCrash.status=" + status);
    System.out.println("fileserverCleanupCrash.lastError=" + error);
    assertThat(attempt).isEqualTo(JpaCleanupQueue.MAXIMUM_ATTEMPTS + 1);
    assertThat(status).isEqualTo("FAILED");
  }

}