revision=a24ece9cf797f7ea647e33bf846b115208ed1ba5

=== production quota aggregate readers ===
src/adapter/outbound/persistence-jpa/src/main/java/dev/caskeleton/adapter/outbound/persistence/fileserver/JpaFileQuotaService.java:85:    return reservations.sumReservedBytes(scope.type(), scope.value(), clock.instant());
src/adapter/outbound/persistence-jpa/src/main/java/dev/caskeleton/adapter/outbound/persistence/fileserver/JpaFileQuotaService.java:90:    return reservations.sumCommittedBytes(scope.type(), scope.value());
src/adapter/outbound/persistence-jpa/src/main/java/dev/caskeleton/adapter/outbound/persistence/fileserver/repository/FileserverQuotaRepository.java:75:  long sumReservedBytes(
src/adapter/outbound/persistence-jpa/src/main/java/dev/caskeleton/adapter/outbound/persistence/fileserver/repository/FileserverQuotaRepository.java:87:  long sumCommittedBytes(

=== quota ceiling config/symbol search ===
src/application-core/src/main/java/dev/caskeleton/application/fileserver/api/error/QuotaExceededException.java:3:/** A quota scope reservation, commit, or concurrency limit rejected the request. */

=== DefaultTransferAdmissionController ===
     1	package dev.caskeleton.application.fileserver.quota;
     2	
     3	import dev.caskeleton.application.fileserver.api.error.FileTooLargeException;
     4	import dev.caskeleton.application.fileserver.api.error.FileserverErrorCode;
     5	import dev.caskeleton.application.fileserver.api.error.FileserverFailureContext;
     6	import dev.caskeleton.application.fileserver.api.error.QuotaExceededException;
     7	import dev.caskeleton.application.fileserver.api.error.StorageFullException;
     8	import dev.caskeleton.application.fileserver.api.error.TransferAdmissionRejectedException;
     9	import dev.caskeleton.application.fileserver.api.metadata.QuotaScope;
    10	import java.util.Map;
    11	import java.util.OptionalDouble;
    12	import java.util.concurrent.ConcurrentHashMap;
    13	import java.util.concurrent.Semaphore;
    14	import java.util.concurrent.atomic.AtomicBoolean;
    15	
    16	/**
    17	 * In-process admission control backed by bounded semaphores.
    18	 *
    19	 * <p>The failure vocabulary is deliberately distinct so a client can tell a permanent policy denial
    20	 * from a transient one: an exhausted pool is {@code STORAGE_FULL}, a scope over its ceiling is
    21	 * {@code QUOTA_EXCEEDED}, and momentary permit exhaustion is a retryable admission rejection.
    22	 */
    23	public final class DefaultTransferAdmissionController implements TransferAdmissionController {
    24	
    25	  private final TransferAdmissionProperties properties;
    26	  private final StorageUsageProbe usageProbe;
    27	  private final long maximumFileSize;
    28	  private final Semaphore instanceUploads;
    29	  private final Semaphore directDownloads;
    30	  private final Map<String, Semaphore> scopeUploads = new ConcurrentHashMap<>();
    31	
    32	  public DefaultTransferAdmissionController(
    33	      TransferAdmissionProperties properties, StorageUsageProbe usageProbe, long maximumFileSize) {
    34	    if (maximumFileSize <= 0) {
    35	      throw new IllegalArgumentException("maximumFileSize must be positive");
    36	    }
    37	    this.properties = properties;
    38	    this.usageProbe = usageProbe;
    39	    this.maximumFileSize = maximumFileSize;
    40	    this.instanceUploads = new Semaphore(properties.instanceUploadPermits());
    41	    this.directDownloads = new Semaphore(properties.directDownloadPermits());
    42	  }
    43	
    44	  @Override
    45	  public TransferPermit acquireUpload(QuotaScope scope, long requestedBytes) {
    46	    requireWithinFileSizePolicy(requestedBytes);
    47	    requireBelowHardHighWater();
    48	    Semaphore scopePermits = scopePermits(scope);
    49	    if (!scopePermits.tryAcquire()) {
    50	      throw new QuotaExceededException(
    51	          "scope upload concurrency is exhausted",
    52	          FileserverFailureContext.of(FileserverErrorCode.QUOTA_EXCEEDED, true));
    53	    }
    54	    if (!instanceUploads.tryAcquire()) {
    55	      scopePermits.release();
    56	      throw new TransferAdmissionRejectedException(
    57	          "instance upload permits are exhausted",
    58	          FileserverFailureContext.of(FileserverErrorCode.TRANSFER_ADMISSION_REJECTED, true));
    59	    }
    60	    return new SemaphorePermit(scopePermits, instanceUploads);
    61	  }
    62	
    63	  @Override
    64	  public TransferPermit acquireDirectDownload(QuotaScope scope) {
    65	    if (!directDownloads.tryAcquire()) {
    66	      throw new TransferAdmissionRejectedException(
    67	          "instance direct download permits are exhausted",
    68	          FileserverFailureContext.of(FileserverErrorCode.TRANSFER_ADMISSION_REJECTED, true));
    69	    }
    70	    return new SemaphorePermit(directDownloads);
    71	  }
    72	
    73	  /**
    74	   * True when the pool crossed the soft mark.
    75	   *
    76	   * <p>Callers use this to throttle or defer large uploads while still accepting small ones.
    77	   */
    78	  public boolean isAboveSoftHighWater() {
    79	    OptionalDouble used = usageProbe.usedFraction();
    80	    return used.isPresent() && used.getAsDouble() >= properties.softHighWater();
    81	  }
    82	
    83	  private void requireWithinFileSizePolicy(long requestedBytes) {
    84	    if (requestedBytes > maximumFileSize) {
    85	      throw new FileTooLargeException(
    86	          "requested upload exceeds the configured maximum file size",
    87	          FileserverFailureContext.of(FileserverErrorCode.FILE_TOO_LARGE, false));
    88	    }
    89	  }
    90	
    91	  private void requireBelowHardHighWater() {
    92	    OptionalDouble used = usageProbe.usedFraction();
    93	    if (used.isPresent() && used.getAsDouble() >= properties.hardHighWater()) {
    94	      throw new StorageFullException(
    95	          "storage pool crossed its hard high-water mark",
    96	          FileserverFailureContext.of(FileserverErrorCode.STORAGE_FULL, false));
    97	    }
    98	  }
    99	
   100	  private Semaphore scopePermits(QuotaScope scope) {
   101	    return scopeUploads.computeIfAbsent(
   102	        scope.canonicalKey(), ignored -> new Semaphore(properties.scopeUploadPermits()));
   103	  }
   104	
   105	  /** Releases each held semaphore exactly once, however many times {@code close} is called. */
   106	  private static final class SemaphorePermit implements TransferPermit {
   107	
   108	    private final Semaphore[] held;
   109	    private final AtomicBoolean released = new AtomicBoolean();
   110	
   111	    private SemaphorePermit(Semaphore... held) {
   112	      this.held = held.clone();
   113	    }
   114	
   115	    @Override
   116	    public void close() {
   117	      if (released.compareAndSet(false, true)) {
   118	        for (Semaphore semaphore : held) {
   119	          semaphore.release();
   120	        }
   121	      }
   122	    }
   123	
   124	    @Override
   125	    public boolean isHeld() {
   126	      return !released.get();
   127	    }
   128	  }
   129	}

=== FileserverPlatformSettings quota ===
  /** Transfer admission control and storage high-water marks. */
  public record Quota(
      @DefaultValue("16") int instanceUploadPermits,
      @DefaultValue("4") int scopeUploadPermits,
      @DefaultValue("64") int directDownloadPermits,
      @DefaultValue("0.70") double softHighWater,
      @DefaultValue("0.85") double hardHighWater) {

    public Quota {
      requirePositive(instanceUploadPermits, "quota.instance-upload-permits");
      requirePositive(scopeUploadPermits, "quota.scope-upload-permits");
      requirePositive(directDownloadPermits, "quota.direct-download-permits");
      requireFraction(softHighWater, "quota.soft-high-water");
      requireFraction(hardHighWater, "quota.hard-high-water");
      if (softHighWater >= hardHighWater) {
        throw new IllegalStateException(
            PREFIX + ".quota.soft-high-water must be below quota.hard-high-water");
      }
      if (scopeUploadPermits > instanceUploadPermits) {
        throw new IllegalStateException(
            PREFIX + ".quota.scope-upload-permits must not exceed quota.instance-upload-permits");
      }
    }
  }

=== JpaFileQuotaService ===
     1	package dev.caskeleton.adapter.outbound.persistence.fileserver;
     2	
     3	import dev.caskeleton.adapter.outbound.persistence.fileserver.entity.QuotaReservationEntity;
     4	import dev.caskeleton.adapter.outbound.persistence.fileserver.repository.FileserverQuotaRepository;
     5	import dev.caskeleton.application.fileserver.api.error.FileserverErrorCode;
     6	import dev.caskeleton.application.fileserver.api.error.FileserverFailureContext;
     7	import dev.caskeleton.application.fileserver.api.error.QuotaExceededException;
     8	import dev.caskeleton.application.fileserver.api.metadata.FileQuotaService;
     9	import dev.caskeleton.application.fileserver.api.metadata.QuotaReservation;
    10	import dev.caskeleton.application.fileserver.api.metadata.QuotaReservationStatus;
    11	import dev.caskeleton.application.fileserver.api.metadata.QuotaScope;
    12	import java.time.Clock;
    13	import java.time.Duration;
    14	import java.time.Instant;
    15	import java.util.UUID;
    16	import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
    17	import org.springframework.stereotype.Repository;
    18	
    19	/**
    20	 * JPA-backed {@link FileQuotaService}.
    21	 *
    22	 * <p>Reservation, extension, commit, and release are conditional statements, so a reservation that
    23	 * already expired or was released can never be extended or committed.
    24	 */
    25	@Repository
    26	@ConditionalOnProperty(prefix = "app.fileserver-platform", name = "enabled", havingValue = "true")
    27	public class JpaFileQuotaService implements FileQuotaService {
    28	
    29	  private final FileserverQuotaRepository reservations;
    30	  private final Clock clock;
    31	
    32	  public JpaFileQuotaService(FileserverQuotaRepository reservations, Clock clock) {
    33	    this.reservations = reservations;
    34	    this.clock = clock;
    35	  }
    36	
    37	  @Override
    38	  public QuotaReservation reserve(QuotaScope scope, long expectedBytes, Duration ttl) {
    39	    if (expectedBytes < 0) {
    40	      throw new IllegalArgumentException("expectedBytes must not be negative");
    41	    }
    42	    Instant now = clock.instant();
    43	    QuotaReservationEntity entity =
    44	        new QuotaReservationEntity(
    45	            UUID.randomUUID(),
    46	            scope.type(),
    47	            scope.value(),
    48	            expectedBytes,
    49	            now.plus(ttl),
    50	            QuotaReservationStatus.RESERVED.name(),
    51	            now);
    52	    return FileEntityMapper.toReservation(reservations.save(entity));
    53	  }
    54	
    55	  @Override
    56	  public void extend(QuotaReservation reservation, long additionalBytes) {
    57	    if (additionalBytes < 0) {
    58	      throw new IllegalArgumentException("additionalBytes must not be negative");
    59	    }
    60	    int updated =
    61	        reservations.extend(reservation.reservationId(), additionalBytes, clock.instant());
    62	    if (updated == 0) {
    63	      throw quotaConflict("reservation is no longer extendable");
    64	    }
    65	  }
    66	
    67	  @Override
    68	  public void commit(QuotaReservation reservation, long actualBytes) {
    69	    if (actualBytes < 0) {
    70	      throw new IllegalArgumentException("actualBytes must not be negative");
    71	    }
    72	    int updated = reservations.commit(reservation.reservationId(), actualBytes, clock.instant());
    73	    if (updated == 0) {
    74	      throw quotaConflict("reservation is no longer committable");
    75	    }
    76	  }
    77	
    78	  @Override
    79	  public void release(QuotaReservation reservation) {
    80	    reservations.release(reservation.reservationId(), clock.instant());
    81	  }
    82	
    83	  /** Bytes currently reserved but not yet committed for a scope. */
    84	  public long reservedBytes(QuotaScope scope) {
    85	    return reservations.sumReservedBytes(scope.type(), scope.value(), clock.instant());
    86	  }
    87	
    88	  /** Bytes durably committed for a scope. */
    89	  public long committedBytes(QuotaScope scope) {
    90	    return reservations.sumCommittedBytes(scope.type(), scope.value());
    91	  }
    92	
    93	  private QuotaExceededException quotaConflict(String message) {
    94	    return new QuotaExceededException(
    95	        message, FileserverFailureContext.of(FileserverErrorCode.QUOTA_EXCEEDED, false));
    96	  }
    97	}

=== quota repository ===
     1	package dev.caskeleton.adapter.outbound.persistence.fileserver.repository;
     2	
     3	import dev.caskeleton.adapter.outbound.persistence.fileserver.entity.QuotaReservationEntity;
     4	import java.time.Instant;
     5	import java.util.List;
     6	import java.util.UUID;
     7	import org.springframework.data.domain.Limit;
     8	import org.springframework.data.jpa.repository.JpaRepository;
     9	import org.springframework.data.jpa.repository.Modifying;
    10	import org.springframework.data.jpa.repository.Query;
    11	import org.springframework.data.repository.query.Param;
    12	
    13	/**
    14	 * Conditional quota reservation statements.
    15	 *
    16	 * <p>Extend, commit, and release all require the reservation to still be {@code RESERVED} at the
    17	 * expected version, so a reservation reclaimed by expiry cannot be resurrected.
    18	 */
    19	public interface FileserverQuotaRepository extends JpaRepository<QuotaReservationEntity, UUID> {
    20	
    21	  @Modifying(clearAutomatically = true, flushAutomatically = true)
    22	  @Query(
    23	      """
    24	      update QuotaReservationEntity q
    25	         set q.reservedBytes = q.reservedBytes + :additionalBytes,
    26	             q.version = q.version + 1,
    27	             q.updatedAt = :now
    28	       where q.reservationId = :reservationId
    29	         and q.status = 'RESERVED'
    30	         and q.expiresAt > :now
    31	      """)
    32	  int extend(
    33	      @Param("reservationId") UUID reservationId,
    34	      @Param("additionalBytes") long additionalBytes,
    35	      @Param("now") Instant now);
    36	
    37	  @Modifying(clearAutomatically = true, flushAutomatically = true)
    38	  @Query(
    39	      """
    40	      update QuotaReservationEntity q
    41	         set q.status = 'COMMITTED',
    42	             q.committedBytes = :actualBytes,
    43	             q.reservedBytes = 0,
    44	             q.version = q.version + 1,
    45	             q.updatedAt = :now
    46	       where q.reservationId = :reservationId
    47	         and q.status = 'RESERVED'
    48	      """)
    49	  int commit(
    50	      @Param("reservationId") UUID reservationId,
    51	      @Param("actualBytes") long actualBytes,
    52	      @Param("now") Instant now);
    53	
    54	  @Modifying(clearAutomatically = true, flushAutomatically = true)
    55	  @Query(
    56	      """
    57	      update QuotaReservationEntity q
    58	         set q.status = 'RELEASED',
    59	             q.reservedBytes = 0,
    60	             q.version = q.version + 1,
    61	             q.updatedAt = :now
    62	       where q.reservationId = :reservationId
    63	         and q.status = 'RESERVED'
    64	      """)
    65	  int release(@Param("reservationId") UUID reservationId, @Param("now") Instant now);
    66	
    67	  @Query(
    68	      """
    69	      select coalesce(sum(q.reservedBytes), 0) from QuotaReservationEntity q
    70	       where q.scopeType = :scopeType
    71	         and q.scopeValue = :scopeValue
    72	         and q.status = 'RESERVED'
    73	         and q.expiresAt > :now
    74	      """)
    75	  long sumReservedBytes(
    76	      @Param("scopeType") String scopeType,
    77	      @Param("scopeValue") String scopeValue,
    78	      @Param("now") Instant now);
    79	
    80	  @Query(
    81	      """
    82	      select coalesce(sum(q.committedBytes), 0) from QuotaReservationEntity q
    83	       where q.scopeType = :scopeType
    84	         and q.scopeValue = :scopeValue
    85	         and q.status = 'COMMITTED'
    86	      """)
    87	  long sumCommittedBytes(
    88	      @Param("scopeType") String scopeType, @Param("scopeValue") String scopeValue);
    89	
    90	  /** Live reservations for a scope, oldest first. */
    91	  @Query(
    92	      """
    93	      select q from QuotaReservationEntity q
    94	       where q.scopeType = :scopeType
    95	         and q.scopeValue = :scopeValue
    96	         and q.status = 'RESERVED'
    97	         and q.expiresAt > :now
    98	       order by q.createdAt asc
    99	      """)
   100	  List<QuotaReservationEntity> findActiveReservations(
   101	      @Param("scopeType") String scopeType,
   102	      @Param("scopeValue") String scopeValue,
   103	      @Param("now") Instant now,
   104	      Limit limit);
   105	
   106	  /** Committed rows for a scope that still carry bytes, newest first. */
   107	  @Query(
   108	      """
   109	      select q from QuotaReservationEntity q
   110	       where q.scopeType = :scopeType
   111	         and q.scopeValue = :scopeValue
   112	         and q.status = 'COMMITTED'
   113	         and q.committedBytes > 0
   114	       order by q.updatedAt desc
   115	      """)
   116	  List<QuotaReservationEntity> findCommittedWithBytes(
   117	      @Param("scopeType") String scopeType, @Param("scopeValue") String scopeValue, Limit limit);
   118	
   119	  /**
   120	   * Gives back part of a committed row.
   121	   *
   122	   * <p>The guard is what makes concurrent reclaims safe: a row that another reclaim already drew
   123	   * down below {@code amount} updates zero rows, and the caller moves to the next row instead of
   124	   * driving the ledger negative.
   125	   */
   126	  @Modifying(clearAutomatically = true, flushAutomatically = true)
   127	  @Query(
   128	      """
   129	      update QuotaReservationEntity q
   130	         set q.committedBytes = q.committedBytes - :amount,
   131	             q.version = q.version + 1,
   132	             q.updatedAt = :now
   133	       where q.reservationId = :reservationId
   134	         and q.committedBytes >= :amount
   135	      """)
   136	  int reduceCommitted(
   137	      @Param("reservationId") UUID reservationId,
   138	      @Param("amount") long amount,
   139	      @Param("now") Instant now);
   140	}

=== quota reclaim gateway ===
     1	package dev.caskeleton.adapter.outbound.persistence.fileserver;
     2	
     3	import dev.caskeleton.adapter.outbound.persistence.fileserver.entity.QuotaReservationEntity;
     4	import dev.caskeleton.adapter.outbound.persistence.fileserver.repository.FileserverQuotaRepository;
     5	import dev.caskeleton.application.fileserver.api.metadata.QuotaScope;
     6	import dev.caskeleton.application.fileserver.cleanup.QuotaReclaimGateway;
     7	import java.time.Clock;
     8	import java.time.Instant;
     9	import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
    10	import org.springframework.data.domain.Limit;
    11	import org.springframework.stereotype.Repository;
    12	
    13	/**
    14	 * Returns reclaimed bytes to a scope's committed total.
    15	 *
    16	 * <p>Committed usage is spread over many rows, so a reclaim is drawn down newest-first across them
    17	 * until the amount is satisfied. Newest-first matters: the most recently committed rows are the
    18	 * ones a delete is most likely to correspond to, and drawing from them keeps historical rows from
    19	 * being hollowed out by unrelated deletions.
    20	 *
    21	 * <p>Each draw-down is conditional on the row still holding at least that many bytes, so two
    22	 * cleanup workers reclaiming at once cannot push the ledger negative — the loser simply moves to
    23	 * the next row.
    24	 *
    25	 * <p>A remainder that no row can absorb is dropped rather than carried. The ledger's floor is zero:
    26	 * a scope cannot owe negative bytes, and a reclaim that outruns the recorded total means the total
    27	 * was already understated, which a negative balance would not fix.
    28	 */
    29	@Repository
    30	@ConditionalOnProperty(prefix = "app.fileserver-platform", name = "enabled", havingValue = "true")
    31	public class JpaQuotaReclaimGateway implements QuotaReclaimGateway {
    32	
    33	  private static final Limit RECLAIM_PAGE = Limit.of(64);
    34	
    35	  private final FileserverQuotaRepository reservations;
    36	  private final Clock clock;
    37	
    38	  public JpaQuotaReclaimGateway(FileserverQuotaRepository reservations, Clock clock) {
    39	    this.reservations = reservations;
    40	    this.clock = clock;
    41	  }
    42	
    43	  @Override
    44	  public void reclaim(QuotaScope scope, long bytes) {
    45	    if (bytes < 0) {
    46	      throw new IllegalArgumentException("bytes must not be negative");
    47	    }
    48	    if (bytes == 0) {
    49	      return;
    50	    }
    51	    Instant now = clock.instant();
    52	    long outstanding = bytes;
    53	    for (QuotaReservationEntity committed :
    54	        reservations.findCommittedWithBytes(scope.type(), scope.value(), RECLAIM_PAGE)) {
    55	      if (outstanding == 0) {
    56	        return;
    57	      }
    58	      long draw = Math.min(outstanding, committed.getCommittedBytes());
    59	      if (reservations.reduceCommitted(committed.getReservationId(), draw, now) == 1) {
    60	        outstanding -= draw;
    61	      }
    62	    }
    63	  }
    64	}

=== recovery upsert ===
     1	package dev.caskeleton.adapter.outbound.persistence.fileserver;
     2	
     3	import dev.caskeleton.adapter.outbound.persistence.fileserver.entity.RecoveryItemEntity;
     4	import dev.caskeleton.adapter.outbound.persistence.fileserver.repository.FileserverRecoveryRepository;
     5	import dev.caskeleton.application.fileserver.api.FileId;
     6	import dev.caskeleton.application.fileserver.recovery.ReconciliationStatus;
     7	import dev.caskeleton.application.fileserver.recovery.RecoveryQueue;
     8	import java.time.Clock;
     9	import java.time.Instant;
    10	import java.util.List;
    11	import java.util.UUID;
    12	import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
    13	import org.springframework.data.domain.Limit;
    14	import org.springframework.stereotype.Repository;
    15	
    16	/**
    17	 * Durable recovery queue over {@code fs_recovery_item}.
    18	 *
    19	 * <p>An enqueue is an upsert: the same file reported twice updates the open item rather than adding
    20	 * a second one. Reconciliation runs on a schedule and re-raises whatever it still cannot settle, so
    21	 * an append-only queue would grow one row per sweep per unresolved file and bury the distinct
    22	 * problems under repetitions of the same one.
    23	 *
    24	 * <p>Resolution keeps the outcome rather than deleting the row. {@code UNRESOLVED} and {@code
    25	 * QUARANTINE_REQUIRED} are the two answers a human has to act on, and both are worthless if the
    26	 * record of what the system concluded disappears with the item.
    27	 */
    28	@Repository
    29	@ConditionalOnProperty(prefix = "app.fileserver-platform", name = "enabled", havingValue = "true")
    30	public class JpaRecoveryQueue implements RecoveryQueue {
    31	
    32	  private static final String STATUS_PENDING = "PENDING";
    33	
    34	  private final FileserverRecoveryRepository items;
    35	  private final Clock clock;
    36	
    37	  public JpaRecoveryQueue(FileserverRecoveryRepository items, Clock clock) {
    38	    this.items = items;
    39	    this.clock = clock;
    40	  }
    41	
    42	  @Override
    43	  public void enqueue(FileId fileId, String reasonCode) {
    44	    Instant now = clock.instant();
    45	    if (items.refreshPending(fileId.value(), reasonCode, now) > 0) {
    46	      return;
    47	    }
    48	    items.save(
    49	        new RecoveryItemEntity(UUID.randomUUID(), fileId.value(), reasonCode, STATUS_PENDING, now));
    50	  }
    51	
    52	  @Override
    53	  public List<FileId> pending(int limit) {
    54	    if (limit < 1) {
    55	      throw new IllegalArgumentException("limit must be positive");
    56	    }
    57	    return items.findPending(Limit.of(limit)).stream()
    58	        .map(RecoveryItemEntity::getFileId)
    59	        .map(FileId::of)
    60	        .toList();
    61	  }
    62	
    63	  @Override
    64	  public void resolve(FileId fileId, ReconciliationStatus status) {
    65	    items.resolvePending(fileId.value(), status.name(), clock.instant());
    66	  }
    67	}
     1	package dev.caskeleton.adapter.outbound.persistence.fileserver.repository;
     2	
     3	import dev.caskeleton.adapter.outbound.persistence.fileserver.entity.RecoveryItemEntity;
     4	import java.time.Instant;
     5	import java.util.List;
     6	import java.util.UUID;
     7	import org.springframework.data.domain.Limit;
     8	import org.springframework.data.jpa.repository.JpaRepository;
     9	import org.springframework.data.jpa.repository.Modifying;
    10	import org.springframework.data.jpa.repository.Query;
    11	import org.springframework.data.repository.query.Param;
    12	
    13	/** Durable list of files whose physical and logical state could not be reconciled automatically. */
    14	public interface FileserverRecoveryRepository extends JpaRepository<RecoveryItemEntity, UUID> {
    15	
    16	  @Query(
    17	      """
    18	      select r from RecoveryItemEntity r
    19	       where r.status = 'PENDING'
    20	       order by r.createdAt asc
    21	      """)
    22	  List<RecoveryItemEntity> findPending(Limit limit);
    23	
    24	  /**
    25	   * Re-raises an open item instead of adding a second one.
    26	   *
    27	   * <p>Reconciliation is retried on a schedule, so the same file reaches the queue repeatedly. One
    28	   * open item per file keeps the queue a worklist rather than a failure log; the newest reason wins
    29	   * because it describes the most recent evidence.
    30	   */
    31	  @Modifying(clearAutomatically = true, flushAutomatically = true)
    32	  @Query(
    33	      """
    34	      update RecoveryItemEntity r
    35	         set r.reasonCode = :reasonCode,
    36	             r.attempt = r.attempt + 1,
    37	             r.updatedAt = :now
    38	       where r.fileId = :fileId
    39	         and r.status = 'PENDING'
    40	      """)
    41	  int refreshPending(
    42	      @Param("fileId") UUID fileId,
    43	      @Param("reasonCode") String reasonCode,
    44	      @Param("now") Instant now);
    45	
    46	  @Modifying(clearAutomatically = true, flushAutomatically = true)
    47	  @Query(
    48	      """
    49	      update RecoveryItemEntity r
    50	         set r.status = :status,
    51	             r.updatedAt = :now
    52	       where r.fileId = :fileId
    53	         and r.status = 'PENDING'
    54	      """)
    55	  int resolvePending(
    56	      @Param("fileId") UUID fileId, @Param("status") String status, @Param("now") Instant now);
    57	}

=== schema revision and activation ===
     1	package dev.caskeleton.adapter.outbound.persistence.fileserver;
     2	
     3	import org.springframework.jdbc.core.JdbcOperations;
     4	
     5	/**
     6	 * Proves the Fileserver schema stream was applied and promoted before the capability serves a
     7	 * request.
     8	 *
     9	 * <p>The stream is operator-applied, like every other optional capability stream: the application
    10	 * migrates {@code db/migration/postgresql} only, and {@code db/migration/jpa/fileserver} is applied
    11	 * and promoted to {@code ACTIVE} deliberately. Until that happens the {@code fs_*} tables either do
    12	 * not exist or are not sanctioned for use.
    13	 *
    14	 * <p>The check runs once, at startup, rather than per operation. The sibling capabilities verify on
    15	 * every call because they are low-frequency; a file download is not, and a registry round trip on
    16	 * the metadata read path would be paid by every byte served. Startup is also the honest place for
    17	 * it — an unpromoted stream is a deployment state, not a per-request condition.
    18	 *
    19	 * <p>Failing here rather than at the first upload is the point. The alternative is a raw "relation
    20	 * fs_file does not exist" surfacing as a 500 to whoever happened to upload first.
    21	 */
    22	public final class FileserverSchemaActivation {
    23	
    24	  static final String CAPABILITY_ID = "jpa-fileserver-metadata-v1";
    25	
    26	  private static final String ACTIVE_CAPABILITY_SQL =
    27	      """
    28	      select count(*)
    29	        from capability_schema_registry
    30	       where capability_id = 'jpa-fileserver-metadata-v1'
    31	         and core_epoch = 1
    32	         and feature_revision >= 2
    33	         and lifecycle_state = 'ACTIVE'
    34	      """;
    35	
    36	  private final JdbcOperations jdbc;
    37	
    38	  public FileserverSchemaActivation(JdbcOperations jdbc) {
    39	    this.jdbc = jdbc;
    40	  }
    41	
    42	  /**
    43	   * Fails closed unless the stream is applied and promoted.
    44	   *
    45	   * <p>An unreadable registry is treated as "not promoted" rather than "assume fine": the registry
    46	   * table itself is created by the core stream, so its absence means the prerequisite chain was
    47	   * never established.
    48	   */
    49	  public void requireActive() {
    50	    Integer active;
    51	    try {
    52	      active = jdbc.queryForObject(ACTIVE_CAPABILITY_SQL, Integer.class);
    53	    } catch (RuntimeException unreadable) {
    54	      throw new IllegalStateException(
    55	          CAPABILITY_ID
    56	              + " could not be verified: the capability schema registry is unreadable, so the "
    57	              + "Fileserver schema stream cannot be confirmed as applied",
    58	          unreadable);
    59	    }
    60	    if (active == null || active != 1) {
    61	      throw new IllegalStateException(
    62	          CAPABILITY_ID
    63	              + " is not ACTIVE at core epoch 1 revision 2. Apply db/migration/jpa/fileserver "
    64	              + "against history table flyway_jpa_fileserver_history and promote the capability "
    65	              + "before enabling app.fileserver-platform.enabled");
    66	    }
    67	  }
    68	}
--- src/adapter/outbound/persistence-jpa/src/main/resources/db/migration/jpa/fileserver/V1__create_fileserver_metadata.sql
13:          AND lifecycle_state = 'ACTIVE'
187:    feature_revision,
188:    lifecycle_state
--- src/adapter/outbound/persistence-jpa/src/main/resources/db/migration/jpa/fileserver/V2__fileserver_recovery_and_staging_cleanup.sql
16:          AND feature_revision >= 1
18:        RAISE EXCEPTION 'fileserver recovery schema requires fileserver metadata revision 1';
61:   SET feature_revision = 2
63:   AND feature_revision < 2;
--- src/adapter/outbound/persistence-jpa/src/main/resources/db/migration/jpa/fileserver/V3__fileserver_fenced_cleanup_lease.sql
19:    ADD COLUMN IF NOT EXISTS claim_token   uuid,
23:COMMENT ON COLUMN fs_cleanup_item.claim_token IS
--- src/adapter/outbound/persistence-jpa/src/main/resources/db/migration/jpa/fileserver/V4__fileserver_upload_terminal_state.sql
24:          AND feature_revision >= 2
26:        RAISE EXCEPTION 'fileserver upload terminal state requires fileserver metadata revision 2';
32:    ADD COLUMN IF NOT EXISTS lifecycle_state varchar(16) NOT NULL DEFAULT 'ACTIVE';
35:    DROP CONSTRAINT IF EXISTS ck_fs_upload_lifecycle_state;
38:    ADD CONSTRAINT ck_fs_upload_lifecycle_state
39:        CHECK (lifecycle_state IN ('ACTIVE', 'TERMINAL'));
41:COMMENT ON COLUMN fs_upload_session.lifecycle_state IS
49:    WHERE lifecycle_state = 'TERMINAL';

=== quota contract docs ===
### 9. Quota settlement is FIFO within a scope

Nothing links a reservation row to the upload that took it, and the design deliberately reclaims
stragglers by TTL and the `STALE_QUOTA_RESERVATION` cleanup type rather than threading a reservation
id through the upload session. `QuotaCommitGateway` therefore settles the oldest live reservation in
the file's namespace.

Which row closes does not change any quota decision: enforcement sums reserved and committed bytes
per scope and never reads an individual row. Concurrent uploads of different sizes can leave the
reserved total transiently high or low, and it converges as each settles. Durable usage with no live
reservation behind it — an upload that outlived its TTL — is still recorded, because a ledger that
silently under-counts is worse than one that is briefly imprecise.

Pinned by `PostgreSqlFileserverReclamationIntegrationTest`.
Expected: FAIL because admission control is not implemented.

- [ ] **Step 3: Implement reservation and bounded permits**

Use DB conditional updates for quota bytes and JVM semaphores for per-instance transfer concurrency. A create request with unknown length reserves the configured initial chunk; append extends the reservation before writing additional bytes. On cancellation or failure, release the reservation in `finally` or cleanup recovery.

```java
public interface TransferAdmissionController {
    TransferPermit acquireUpload(QuotaScope scope, long requestedBytes);
    TransferPermit acquireDirectDownload(QuotaScope scope);
}
```

A hard storage high-water condition maps to `StorageFullException`; scope limit maps to `QuotaExceededException`; temporary permit exhaustion maps to `TransferAdmissionRejectedException` with `retryable=true`.

- [ ] **Step 4: Run quota and concurrency tests**