diff --git a/src/main/java/com/depromeet/piki/extractor/extraction/DefaultProductLinkExtractor.java b/src/main/java/com/depromeet/piki/extractor/extraction/DefaultProductLinkExtractor.java index 410e84b..8744c44 100644 --- a/src/main/java/com/depromeet/piki/extractor/extraction/DefaultProductLinkExtractor.java +++ b/src/main/java/com/depromeet/piki/extractor/extraction/DefaultProductLinkExtractor.java @@ -4,6 +4,7 @@ import com.depromeet.piki.extractor.domain.ProductSnapshot; import com.depromeet.piki.extractor.domain.ProductSnapshotException; import com.depromeet.piki.extractor.extraction.http.PageFetchException; +import com.depromeet.piki.extractor.extraction.http.SiteConcurrencyLimiter; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Component; @@ -20,11 +21,12 @@ public class DefaultProductLinkExtractor implements LinkExtractionStrategy { private final PageFetcher pageFetcher; private final HtmlSnapshotPipeline htmlSnapshotPipeline; + private final SiteConcurrencyLimiter siteLimiter; @Override public ProductSnapshot extract(ProductLink link, boolean authorized, String model) { long fetchStart = System.nanoTime(); - PageContent page = pageFetcher.fetch(link); + PageContent page = siteLimiter.call(link, () -> pageFetcher.fetch(link)); long fetchMs = (System.nanoTime() - fetchStart) / 1_000_000; try { diff --git a/src/main/java/com/depromeet/piki/extractor/extraction/http/SiteBusyException.java b/src/main/java/com/depromeet/piki/extractor/extraction/http/SiteBusyException.java new file mode 100644 index 0000000..dd69e6c --- /dev/null +++ b/src/main/java/com/depromeet/piki/extractor/extraction/http/SiteBusyException.java @@ -0,0 +1,20 @@ +package com.depromeet.piki.extractor.extraction.http; + +import com.depromeet.piki.extractor.common.exception.ExtractionErrorCode; +import com.depromeet.piki.extractor.common.exception.ExtractionException; +import java.time.Duration; + +/** + * 같은 사이트의 동시 fetch 자리가 나지 않은 실패. 일시 실패다. {@link PageFetchException} 이 아니라서 헤드리스로 + * 넘어가지 않는다. 넘기면 같은 사이트를 렌더 서버 IP 로 다시 요청해 상한이 무력해진다. + */ +public final class SiteBusyException extends ExtractionException { + + private SiteBusyException(String message) { + super(message, ExtractionErrorCode.UPSTREAM_ERROR, false, null); + } + + static SiteBusyException waited(Duration maxWait) { + return new SiteBusyException("같은 사이트의 동시 fetch 자리를 " + maxWait + " 동안 얻지 못했다."); + } +} diff --git a/src/main/java/com/depromeet/piki/extractor/extraction/http/SiteConcurrencyLimiter.java b/src/main/java/com/depromeet/piki/extractor/extraction/http/SiteConcurrencyLimiter.java new file mode 100644 index 0000000..624940b --- /dev/null +++ b/src/main/java/com/depromeet/piki/extractor/extraction/http/SiteConcurrencyLimiter.java @@ -0,0 +1,65 @@ +package com.depromeet.piki.extractor.extraction.http; + +import com.depromeet.piki.extractor.domain.ProductLink; +import com.google.common.net.InternetDomainName; +import com.google.common.util.concurrent.Striped; +import java.time.Duration; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +/** + * 몰 요청은 전부 고정 IP 하나에서 나가 몰의 IP 단위 제한에 걸리면 그 몰 전체가 막힌다. 같은 사이트로 동시에 나가는 + * 정적 fetch 수를 묶는다(#119). 사이트는 공개 접미사 목록 기준 등록 도메인이라 www·m 이 한 사이트다. + */ +@Component +public class SiteConcurrencyLimiter { + + /** 사이트 키가 늘어도 메모리가 자라지 않게 줄무늬로 나눈다. 다른 사이트가 같은 줄무늬를 나눌 확률은 무시할 만하다. */ + private static final int STRIPES = 1024; + + private final Striped semaphores; + private final Duration maxWait; + + @Autowired + public SiteConcurrencyLimiter(SiteLimitProperties properties) { + this(properties.maxConcurrent(), properties.maxWait()); + } + + public SiteConcurrencyLimiter(int maxConcurrent, Duration maxWait) { + this.semaphores = Striped.lazyWeakSemaphore(STRIPES, maxConcurrent); + this.maxWait = maxWait; + } + + public T call(ProductLink link, Supplier fetch) { + Semaphore semaphore = semaphores.get(siteOf(link.value().getHost())); + acquire(semaphore); + try { + return fetch.get(); + } finally { + semaphore.release(); + } + } + + private void acquire(Semaphore semaphore) { + try { + if (!semaphore.tryAcquire(maxWait.toMillis(), TimeUnit.MILLISECONDS)) { + throw SiteBusyException.waited(maxWait); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw SiteBusyException.waited(maxWait); + } + } + + static String siteOf(String host) { + try { + InternetDomainName name = InternetDomainName.from(host); + return name.isUnderPublicSuffix() ? name.topPrivateDomain().toString() : host; + } catch (IllegalArgumentException e) { + return host; + } + } +} diff --git a/src/main/java/com/depromeet/piki/extractor/extraction/http/SiteLimitProperties.java b/src/main/java/com/depromeet/piki/extractor/extraction/http/SiteLimitProperties.java new file mode 100644 index 0000000..7eb5a44 --- /dev/null +++ b/src/main/java/com/depromeet/piki/extractor/extraction/http/SiteLimitProperties.java @@ -0,0 +1,17 @@ +package com.depromeet.piki.extractor.extraction.http; + +import java.time.Duration; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.bind.DefaultValue; + +/** + * 사이트별 동시 fetch 상한({@link SiteConcurrencyLimiter}). + * + * @param maxWait 자리를 기다리는 최대 시간. 링크 추출 기한(호출자 read 타임아웃 55초) 안에 fetch·LLM 몫을 남긴다. + */ +@ConfigurationProperties("fetch.site-limit") +public record SiteLimitProperties( + @DefaultValue("2") int maxConcurrent, + @DefaultValue("15s") Duration maxWait +) { +} diff --git a/src/test/java/com/depromeet/piki/extractor/extraction/DefaultProductLinkExtractorTest.java b/src/test/java/com/depromeet/piki/extractor/extraction/DefaultProductLinkExtractorTest.java index bd332be..075a22d 100644 --- a/src/test/java/com/depromeet/piki/extractor/extraction/DefaultProductLinkExtractorTest.java +++ b/src/test/java/com/depromeet/piki/extractor/extraction/DefaultProductLinkExtractorTest.java @@ -11,9 +11,11 @@ import com.depromeet.piki.extractor.extraction.gemini.GeminiApiException; import com.depromeet.piki.extractor.extraction.gemini.GeminiExtractionResult; import com.depromeet.piki.extractor.extraction.http.PageFetchException; +import com.depromeet.piki.extractor.extraction.http.SiteConcurrencyLimiter; import com.depromeet.piki.extractor.extraction.structured.StructuredDataExtractor; import com.depromeet.piki.extractor.support.StubGeminiClient; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.time.Duration; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import tools.jackson.databind.ObjectMapper; @@ -43,7 +45,8 @@ private DefaultProductLinkExtractor extractorFetching(String html) { new StructuredDataExtractor(new ObjectMapper()), new GeminiHtmlExtractor(stubGemini), new SimpleMeterRegistry() - ) + ), + new SiteConcurrencyLimiter(2, Duration.ofSeconds(15)) ); } diff --git a/src/test/java/com/depromeet/piki/extractor/extraction/ProductLinkExtractE2ETest.java b/src/test/java/com/depromeet/piki/extractor/extraction/ProductLinkExtractE2ETest.java index 2e2a39c..2d37a2d 100644 --- a/src/test/java/com/depromeet/piki/extractor/extraction/ProductLinkExtractE2ETest.java +++ b/src/test/java/com/depromeet/piki/extractor/extraction/ProductLinkExtractE2ETest.java @@ -9,9 +9,11 @@ import com.depromeet.piki.extractor.extraction.http.PageFetchHttpClientConfig; import com.depromeet.piki.extractor.extraction.http.RequestScopedCookieStore; import com.depromeet.piki.extractor.extraction.http.RequestScopedDnsResolver; +import com.depromeet.piki.extractor.extraction.http.SiteConcurrencyLimiter; import com.depromeet.piki.extractor.extraction.structured.StructuredDataExtractor; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import io.micrometer.observation.ObservationRegistry; +import java.time.Duration; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; @@ -66,7 +68,8 @@ class ProductLinkExtractE2ETest { new StructuredDataExtractor(objectMapper), new GeminiHtmlExtractor(httpClient), new SimpleMeterRegistry() - ) + ), + new SiteConcurrencyLimiter(2, Duration.ofSeconds(15)) ); @Test diff --git a/src/test/java/com/depromeet/piki/extractor/extraction/http/SiteConcurrencyLimiterTest.java b/src/test/java/com/depromeet/piki/extractor/extraction/http/SiteConcurrencyLimiterTest.java new file mode 100644 index 0000000..01d780c --- /dev/null +++ b/src/test/java/com/depromeet/piki/extractor/extraction/http/SiteConcurrencyLimiterTest.java @@ -0,0 +1,106 @@ +package com.depromeet.piki.extractor.extraction.http; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import com.depromeet.piki.extractor.common.exception.ExtractionErrorCode; +import com.depromeet.piki.extractor.domain.ProductLink; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +class SiteConcurrencyLimiterTest { + + @ParameterizedTest + @CsvSource({ + "www.musinsa.com, musinsa.com", + "m.musinsa.com, musinsa.com", + "smartstore.naver.com, naver.com", + "naver.me, naver.me", + "shop.example.co.kr, example.co.kr", + "localhost, localhost", + "10.0.1.5, 10.0.1.5", + }) + @DisplayName("사이트는 공개 접미사 목록 기준 등록 도메인이고 그 밖의 host 는 그대로다") + void siteOfGroupsByRegistrableDomain(String host, String site) { + assertEquals(site, SiteConcurrencyLimiter.siteOf(host)); + } + + @Test + @DisplayName("같은 사이트의 자리가 차 있으면 기다리다 헤드리스로 넘어가지 않는 일시 실패로 끝낸다") + void sameSiteOverLimitFailsTransiently() throws Exception { + SiteConcurrencyLimiter limiter = new SiteConcurrencyLimiter(1, Duration.ofMillis(100)); + CountDownLatch holding = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) { + Future first = executor.submit(() -> limiter.call(link("https://www.musinsa.com/p/1"), () -> { + holding.countDown(); + await(release); + return "first"; + })); + holding.await(5, TimeUnit.SECONDS); + + SiteBusyException e = assertThrows( + SiteBusyException.class, + () -> limiter.call(link("https://m.musinsa.com/p/2"), () -> "second") + ); + release.countDown(); + + assertEquals("first", first.get(5, TimeUnit.SECONDS)); + assertEquals(ExtractionErrorCode.UPSTREAM_ERROR, e.code()); + assertFalse(e.permanent()); + } + } + + @Test + @DisplayName("다른 사이트는 같은 사이트의 자리와 무관하게 바로 진행한다") + void otherSiteIsNotBlocked() throws Exception { + SiteConcurrencyLimiter limiter = new SiteConcurrencyLimiter(1, Duration.ofMillis(100)); + CountDownLatch holding = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) { + executor.submit(() -> limiter.call(link("https://www.musinsa.com/p/1"), () -> { + holding.countDown(); + await(release); + return "first"; + })); + holding.await(5, TimeUnit.SECONDS); + + assertEquals("other", limiter.call(link("https://www.29cm.co.kr/p/1"), () -> "other")); + release.countDown(); + } + } + + @Test + @DisplayName("fetch 가 실패해도 자리를 돌려준다") + void releasesOnFailure() { + SiteConcurrencyLimiter limiter = new SiteConcurrencyLimiter(1, Duration.ofMillis(100)); + ProductLink link = link("https://www.musinsa.com/p/1"); + + assertThrows(IllegalStateException.class, () -> limiter.call(link, () -> { + throw new IllegalStateException("fetch 실패"); + })); + + assertEquals("next", limiter.call(link, () -> "next")); + } + + private static ProductLink link(String url) { + return ProductLink.parse(url); + } + + private static void await(CountDownLatch latch) { + try { + latch.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +}