From 9f8a9a36f1c89a2d331890d175a38ba452165e1d Mon Sep 17 00:00:00 2001 From: myh Date: Sat, 5 Sep 2026 15:15:33 +0900 Subject: [PATCH] =?UTF-8?q?fix:=20=EC=A0=95=EC=B1=85=20=EC=88=98=EC=A7=91?= =?UTF-8?q?=20=EB=B0=B0=EC=B9=98=20=E2=80=94=20=EC=A7=80=EC=97=AD=20?= =?UTF-8?q?=ED=96=89=20=EC=A0=84=EB=9F=89=20=EC=9E=AC=EC=9E=91=EC=84=B1=20?= =?UTF-8?q?=EC=A0=9C=EA=B1=B0,=20=EC=8B=A4=ED=96=89=20=EA=B1=B0=EB=B6=80?= =?UTF-8?q?=20=EC=8B=9C=20=EB=9D=BD=20=EB=88=84=EC=88=98,=20=EC=8B=A4?= =?UTF-8?q?=ED=8C=A8=20=EC=A7=80=EC=A0=90=20=EB=A1=9C=EA=B9=85=20(#229)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - PolicyUpsertWriter: 기존 지역 코드 집합을 청크당 1쿼리로 떠서 새 집합과 비교하고, 같으면 policy_regions 삭제·삽입을 건너뛴다. 정책 본문만 바뀐 회차에서 전국 정책 1건당 DELETE 284 + INSERT 284가 그대로 나가던 것을 없앤다. 스냅샷 델타 쓰기(#73)의 의도를 지역 행까지 확장한 것. - PolicySyncService: batchTaskExecutor.execute()가 RejectedExecutionException으로 거부되면 실행 본체가 시작되지 않아 finally의 락 해제가 돌지 않았다. 거부 시 즉시 락을 풀고 예외를 다시 던진다. 그대로 두면 아무도 안 도는 채로 TTL 10분간 409만 나간다. - PolicyUpsertWriter: 건별 재시도까지 뚫린 예외는 앞선 청크가 이미 커밋된 상태인데 이력에는 카운트 0으로만 남아 진단이 불가능했다. 중단 시 처리/전체 건수와 신규·변경· 건별실패 카운트를 error 로그로 남긴다. 테스트 263건 통과 (sync / policy / admin.sync). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01ESRTprV1GLdFjGrymsQMY4 --- .../sync/service/PolicySyncService.java | 27 ++-- .../sync/service/PolicyUpsertWriter.java | 120 +++++++++++++----- 2 files changed, 107 insertions(+), 40 deletions(-) diff --git a/src/main/java/com/bop/youthpick/sync/service/PolicySyncService.java b/src/main/java/com/bop/youthpick/sync/service/PolicySyncService.java index 444d40f..f907db0 100644 --- a/src/main/java/com/bop/youthpick/sync/service/PolicySyncService.java +++ b/src/main/java/com/bop/youthpick/sync/service/PolicySyncService.java @@ -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() { diff --git a/src/main/java/com/bop/youthpick/sync/service/PolicyUpsertWriter.java b/src/main/java/com/bop/youthpick/sync/service/PolicyUpsertWriter.java index 872e3b5..9ff435e 100644 --- a/src/main/java/com/bop/youthpick/sync/service/PolicyUpsertWriter.java +++ b/src/main/java/com/bop/youthpick/sync/service/PolicyUpsertWriter.java @@ -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; @@ -30,20 +33,36 @@ public PolicyWriteResult upsertAll(List items) { int newCount = 0; int updatedCount = 0; int errorCount = 0; - for (int from = 0; from < items.size(); from += CHUNK_SIZE) { - List 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 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); } @@ -61,14 +80,21 @@ private ChunkResult retryOneByOne(List 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 { @@ -89,10 +115,12 @@ private ChunkResult writeChunk(List chunk) { chunk.stream().map(i -> i.policy().getPolicyNo()).toList()) .stream() .collect(Collectors.toMap(Policy::getPolicyNo, p -> p)); + Map> 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++; @@ -101,28 +129,60 @@ private ChunkResult writeChunk(List chunk) { return new ChunkResult(newCount, updatedCount, 0); } + /** + * 기존 정책들이 현재 갖고 있는 지역 코드 집합 (policyId → codes). 쿼리 1회로 청크 전체를 떠서 {@link #writeOne}이 지역 변경 여부를 + * 판정하는 데 쓴다. + */ + private Map> regionCodesByPolicyId(Collection existingPolicies) { + List 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> 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 currentCodes = + regionCodesByPolicyId.getOrDefault(existing.getId(), Set.of()); + Set 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 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에 없는 번호는 무시. */