Skip to content

fix: 정책 수집 배치 — 지역 행 전량 재작성 제거, 실행 거부 시 락 누수, 실패 지점 로깅 (#229) #231

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
myh7754 merged 1 commit into from
Sep 5, 2026
Merged

fix: 정책 수집 배치 — 지역 행 전량 재작성 제거, 실행 거부 시 락 누수, 실패 지점 로깅 (#229) #231

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 17 additions & 10 deletions src/main/java/com/bop/youthpick/sync/service/PolicySyncService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,16 +72,23 @@ public void startFullSyncAsync() {
policySyncLock
.tryAcquire()
.orElseThrow(() -> new CustomException(SyncErrorCode.SYNC_ALREADY_RUNNING));
batchTaskExecutor.execute(
() -> {
try {
doRunFullSync();
} catch (RuntimeException e) {
log.error("정책 수집 수동 실행 실패 — 상세는 policy_batch_history 참조", e);
} finally {
policySyncLock.release(lockToken);
}
});
try {
batchTaskExecutor.execute(
() -> {
try {
doRunFullSync();
} catch (RuntimeException e) {
log.error("정책 수집 수동 실행 실패 — 상세는 policy_batch_history 참조", e);
} finally {
policySyncLock.release(lockToken);
}
});
} catch (RuntimeException e) {
// 큐가 꽉 차 execute가 거부되면 실행 본체가 시작되지 않아 finally도 돌지 않는다 —
// 락을 여기서 풀어주지 않으면 아무도 안 도는 채로 TTL(10분) 동안 409만 나간다.
policySyncLock.release(lockToken);
throw e;
}
}

private PolicyBatchHistory doRunFullSync() {
Expand Down
120 changes: 90 additions & 30 deletions src/main/java/com/bop/youthpick/sync/service/PolicyUpsertWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,15 @@

import com.bop.youthpick.policy.entity.Policy;
import com.bop.youthpick.policy.entity.PolicyRegion;
import com.bop.youthpick.policy.entity.Region;
import com.bop.youthpick.policy.repository.PolicyRegionRepository;
import com.bop.youthpick.policy.repository.PolicyRepository;
import com.bop.youthpick.sync.dto.PolicyUpsertItem;
import com.bop.youthpick.sync.dto.PolicyWriteResult;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
Expand All @@ -30,20 +33,36 @@ public PolicyWriteResult upsertAll(List<PolicyUpsertItem> items) {
int newCount = 0;
int updatedCount = 0;
int errorCount = 0;
for (int from = 0; from < items.size(); from += CHUNK_SIZE) {
List<PolicyUpsertItem> chunk =
items.subList(from, Math.min(from + CHUNK_SIZE, items.size()));
ChunkResult result;
try {
result = transactionTemplate.execute(status -> writeChunk(chunk));
} catch (RuntimeException e) {
// 청크에 깨진 건이 섞임 — 건별 재시도로 범인만 제외하고 나머지는 살린다.
log.warn("정책 청크 저장 실패({}건) — 건별 재시도", chunk.size(), e);
result = retryOneByOne(chunk);
int processed = 0;
try {
for (int from = 0; from < items.size(); from += CHUNK_SIZE) {
List<PolicyUpsertItem> chunk =
items.subList(from, Math.min(from + CHUNK_SIZE, items.size()));
ChunkResult result;
try {
result = transactionTemplate.execute(status -> writeChunk(chunk));
} catch (RuntimeException e) {
// 청크에 깨진 건이 섞임 — 건별 재시도로 범인만 제외하고 나머지는 살린다.
log.warn("정책 청크 저장 실패({}건) — 건별 재시도", chunk.size(), e);
result = retryOneByOne(chunk);
}
newCount += result.newCount();
updatedCount += result.updatedCount();
errorCount += result.errorCount();
processed += chunk.size();
}
newCount += result.newCount();
updatedCount += result.updatedCount();
errorCount += result.errorCount();
} catch (RuntimeException e) {
// 건별 재시도까지 뚫고 올라온 예외 = DB 장애급. 앞선 청크는 이미 커밋됐는데 이력에는
// 카운트 0으로 남으므로(PolicySyncService의 catch), 어디까지 갔는지는 여기서만 알 수 있다.
// ponytail: 로그만 — 실제로 발생하면 부분 결과를 예외에 실어 이력까지 옮긴다.
log.error(
"정책 저장 중단 — {}/{}건 처리 후 실패 (신규 {} / 변경 {} / 건별실패 {})",
processed,
items.size(),
newCount,
updatedCount,
errorCount);
throw e;
}
return new PolicyWriteResult(newCount, updatedCount, errorCount);
}
Expand All @@ -61,14 +80,21 @@ private ChunkResult retryOneByOne(List<PolicyUpsertItem> chunk) {
try {
boolean isNew =
transactionTemplate.execute(
status ->
writeOne(
freshItem,
policyRepository
.findByPolicyNoIn(List.of(policyNo))
.stream()
.findFirst()
.orElse(null)));
status -> {
Policy existing =
policyRepository
.findByPolicyNoIn(List.of(policyNo))
.stream()
.findFirst()
.orElse(null);
return writeOne(
freshItem,
existing,
regionCodesByPolicyId(
existing == null
? List.of()
: List.of(existing)));
});
if (isNew) {
newCount++;
} else {
Expand All @@ -89,10 +115,12 @@ private ChunkResult writeChunk(List<PolicyUpsertItem> chunk) {
chunk.stream().map(i -> i.policy().getPolicyNo()).toList())
.stream()
.collect(Collectors.toMap(Policy::getPolicyNo, p -> p));
Map<Long, Set<String>> regionCodesByPolicyId = regionCodesByPolicyId(existingByNo.values());
int newCount = 0;
int updatedCount = 0;
for (PolicyUpsertItem item : chunk) {
if (writeOne(item, existingByNo.get(item.policy().getPolicyNo()))) {
if (writeOne(
item, existingByNo.get(item.policy().getPolicyNo()), regionCodesByPolicyId)) {
newCount++;
} else {
updatedCount++;
Expand All @@ -101,28 +129,60 @@ private ChunkResult writeChunk(List<PolicyUpsertItem> chunk) {
return new ChunkResult(newCount, updatedCount, 0);
}

/**
* 기존 정책들이 현재 갖고 있는 지역 코드 집합 (policyId → codes). 쿼리 1회로 청크 전체를 떠서 {@link #writeOne}이 지역 변경 여부를
* 판정하는 데 쓴다.
*/
private Map<Long, Set<String>> regionCodesByPolicyId(Collection<Policy> existingPolicies) {
List<Long> policyIds = existingPolicies.stream().map(Policy::getId).toList();
if (policyIds.isEmpty()) {
return Map.of();
}
return policyRegionRepository.findWithRegionByPolicyIdIn(policyIds).stream()
.collect(
Collectors.groupingBy(
policyRegion -> policyRegion.getPolicy().getId(),
Collectors.mapping(
policyRegion -> policyRegion.getRegion().getCode(),
Collectors.toSet())));
}

/** 1건 저장. 신규면 true, 변경이면 false. */
private boolean writeOne(PolicyUpsertItem item, Policy existing) {
private boolean writeOne(
PolicyUpsertItem item, Policy existing, Map<Long, Set<String>> regionCodesByPolicyId) {
Policy target;
boolean isNew = existing == null;
if (isNew) {
target = policyRepository.save(item.policy());
saveRegions(target, item.regions());
} else {
existing.updateFrom(item.policy());
// Hibernate는 flush 시 INSERT를 DELETE보다 먼저 실행하므로, 같은 지역이 유지되는
// 정책에서 (policy_id, region_code) 유니크 충돌이 난다 — 삭제를 먼저 flush한다.
policyRegionRepository.deleteByPolicy(existing);
policyRegionRepository.flush();
target = existing;
}
for (var region : item.regions()) {
policyRegionRepository.save(PolicyRegion.create(target, region));
// 지역 집합이 그대로면 손대지 않는다 — 정책 본문만 바뀐 회차에서 전국 정책 1건당
// DELETE 256 + INSERT 256이 그대로 나가던 것을 없앤다.
Set<String> currentCodes =
regionCodesByPolicyId.getOrDefault(existing.getId(), Set.of());
Set<String> nextCodes =
item.regions().stream().map(Region::getCode).collect(Collectors.toSet());
if (!currentCodes.equals(nextCodes)) {
// Hibernate는 flush 시 INSERT를 DELETE보다 먼저 실행하므로, 같은 지역이 유지되는
// 정책에서 (policy_id, region_code) 유니크 충돌이 난다 — 삭제를 먼저 flush한다.
policyRegionRepository.deleteByPolicy(existing);
policyRegionRepository.flush();
saveRegions(target, item.regions());
}
}
// 지역 행과 같은 트랜잭션에서 갱신한다 — 따로 두면 둘이 어긋나 정렬이 조용히 틀린다(#120).
target.applyRegionCoverage(item.nationwide());
return isNew;
}

private void saveRegions(Policy policy, List<Region> regions) {
for (Region region : regions) {
policyRegionRepository.save(PolicyRegion.create(policy, region));
}
}

private record ChunkResult(int newCount, int updatedCount, int errorCount) {}

/** 이번 수집에서 사라진 정책의 누락 횟수를 올린다(3회 도달 시 HIDDEN — Policy.markMissing). DB에 없는 번호는 무시. */
Expand Down
Loading