donghyeon.kang/document-haness
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
75bed382c8fd87a32aad33cb26ac5c107a2255ac
document-haness/docs/clean-architecture-backend-template/final/evidence/raw/079-fileserver-reachability-quota-schema-contract.txt
T
DongHyeonkaandClaude Opus 5 b2963105a8 docs(keycloak-session-store): import the session-storage lab as a new project 
The keycloak project ended with four open questions that design could not
settle. A two-VM lab was built to answer them by measurement, and this is
that material: 26 experiments, 125 raw command outputs, 22 browser captures.

Follows the import procedure in README.md.

  source/     the originating repository verbatim — 78 documents, 28 SVGs,
              8 manifests, plus .source-revision recording the commit
  final/      the SSOT
    document.md   729 lines written from the 29 experiment documents, not
                  concatenated: what was predicted, what was measured, and
                  where the measurement itself was wrong
    evidence/raw    125 outputs, flattened to <experiment>__<file> because
                    the originals collided (01-baseline.txt appeared three
                    times) and the audit only globs the top level
    evidence/meta   one per raw file; command and exitCode are null and the
                    README says why rather than inventing them
    evidence/browser  22 captures
    assets/       three diagrams through techviz
    .techviz/     their VizSpecs

A separate project rather than an addition to keycloak: the B-layer answers
that project's four questions, but the A, C and D layers are about cluster
failure, SSO and operations, and one document.md should hold one subject.
The four question records there can point here through 관계.

Recorded rather than papered over: only three of the 28 diagrams were
remade. The repository forbids hand-drawn SVG and forbids titles inside the
canvas; all 28 originals carry both, so converting them is redrawing, not
reformatting. They stay in source/ and the gap is written into the document.

verify-pipeline.py passes. audit-records.py reports no issues.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-04 22:51:59 +09:00

724 lines
33 KiB
Plaintext
Raw Blame History

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**
Reference in New Issue View Git Blame Copy Permalink