extractor 사이트별 동시 요청 상한 추가

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 {

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 + " 동안 얻지 못했다.");
+    }
+}

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<Semaphore> 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> T call(ProductLink link, Supplier<T> 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;
+        }
+    }
+}

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
+) {
+}

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))
         );
     }
 

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

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<String> 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();
+        }
+    }
+}