Fix the pending problem when requesting RSS endpoint (#64)

This PR fixes the problem and refine the logic of getting file bytes.

After upgrading to Halo 2.20.16, requesting RSS endpoint may cause entirely pending issue. Please see the following concrete logs:

```java
2025-03-11T01:04:39.536+08:00 ERROR 7 --- [loomBoundedElastic-361] a.w.r.e.AbstractErrorWebExceptionHandler : [756c052a-61]  500 Server Error for HTTP GET "/rss.xml"

com.google.common.util.concurrent.UncheckedExecutionException: reactor.core.Exceptions$ReactiveException: java.lang.InterruptedException
	at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2085) ~[guava-33.3.1-jre.jar:na]
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
	*__checkpoint ⇢ run.halo.app.security.InitializeRedirectionWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.app.infra.webfilter.LocaleChangeWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.app.security.device.DeviceSessionFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.cache.page.PageCacheWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityWebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ AuthorizationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ExceptionTranslationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ LogoutWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ ServerRequestCacheWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityWebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ AnonymousAuthenticationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityWebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.social.login.security.SocialAuthenticatorFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityWebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ AuthenticationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ AuthenticationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ AuthenticationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ MapOAuth2AuthenticationFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ AuthenticationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityWebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢  [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.social.login.security.SocialAuthorizationRequestRedirectWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityWebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ ReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ CsrfWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ CorsWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ HttpHeaderWriterWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ SecurityWebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.seo.tools.AdvancedRedirectWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.moments.rss.OldRssRouteRedirectionFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.comment.widget.captcha.CommentCaptchaFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.seo.tools.CrawlRecordFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ cc.ryanc.staticpages.endpoint.RewriteOnNotFoundFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ run.halo.app.infra.webfilter.AdditionalWebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.web.filter.reactive.ServerWebExchangeContextFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ HTTP GET "/rss.xml" [ExceptionHandlingWebHandler]
Original Stack Trace:
		at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2085) ~[guava-33.3.1-jre.jar:na]
		at com.google.common.cache.LocalCache.get(LocalCache.java:4017) ~[guava-33.3.1-jre.jar:na]
		at com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4898) ~[guava-33.3.1-jre.jar:na]
		at run.halo.feed.RssCacheManager.lambda$get$2(RssCacheManager.java:29) ~[na:na]
		at reactor.core.publisher.MonoCallable$MonoCallableSubscription.request(MonoCallable.java:137) ~[reactor-core-3.7.3.jar:3.7.3]
		at reactor.core.publisher.MonoCacheTime$CoordinatorSubscriber.onSubscribe(MonoCacheTime.java:293) ~[reactor-core-3.7.3.jar:3.7.3]
		at reactor.core.publisher.MonoCallable.subscribe(MonoCallable.java:48) ~[reactor-core-3.7.3.jar:3.7.3]
		at reactor.core.publisher.MonoCacheTime.subscribeOrReturn(MonoCacheTime.java:143) ~[reactor-core-3.7.3.jar:3.7.3]
		at reactor.core.publisher.Mono.subscribe(Mono.java:4560) ~[reactor-core-3.7.3.jar:3.7.3]
		at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:126) ~[reactor-core-3.7.3.jar:3.7.3]
		at reactor.core.scheduler.BoundedElasticThreadPerTaskScheduler$SchedulerTask.run(BoundedElasticThreadPerTaskScheduler.java:1013) ~[reactor-core-3.7.3.jar:3.7.3]
		at java.base/java.lang.VirtualThread.run(Unknown Source) ~[na:na]
Caused by: reactor.core.Exceptions$ReactiveException: java.lang.InterruptedException
	at reactor.core.Exceptions$ReactiveException.fillInStackTrace(Exceptions.java:736) ~[reactor-core-3.7.3.jar:3.7.3]
	at java.base/java.lang.Throwable.<init>(Unknown Source) ~[na:na]
	at java.base/java.lang.Throwable.<init>(Unknown Source) ~[na:na]
	at java.base/java.lang.Exception.<init>(Unknown Source) ~[na:na]
	at java.base/java.lang.RuntimeException.<init>(Unknown Source) ~[na:na]
	at reactor.core.Exceptions$ReactiveException.<init>(Exceptions.java:726) ~[reactor-core-3.7.3.jar:3.7.3]
	at reactor.core.Exceptions.propagate(Exceptions.java:410) ~[reactor-core-3.7.3.jar:3.7.3]
	at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:96) ~[reactor-core-3.7.3.jar:3.7.3]
	at reactor.core.publisher.Mono.block(Mono.java:1779) ~[reactor-core-3.7.3.jar:3.7.3]
	at run.halo.feed.RssCacheManager.lambda$get$1(RssCacheManager.java:32) ~[na:na]
	at com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4903) ~[guava-33.3.1-jre.jar:na]
	at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3574) ~[guava-33.3.1-jre.jar:na]
	at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2316) ~[guava-33.3.1-jre.jar:na]
	at com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2189) ~[guava-33.3.1-jre.jar:na]
	at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2079) ~[guava-33.3.1-jre.jar:na]
	at com.google.common.cache.LocalCache.get(LocalCache.java:4017) ~[guava-33.3.1-jre.jar:na]
	at com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4898) ~[guava-33.3.1-jre.jar:na]
	at run.halo.feed.RssCacheManager.lambda$get$2(RssCacheManager.java:29) ~[na:na]
	at reactor.core.publisher.MonoCallable$MonoCallableSubscription.request(MonoCallable.java:137) ~[reactor-core-3.7.3.jar:3.7.3]
	at reactor.core.publisher.MonoCacheTime$CoordinatorSubscriber.onSubscribe(MonoCacheTime.java:293) ~[reactor-core-3.7.3.jar:3.7.3]
	at reactor.core.publisher.MonoCallable.subscribe(MonoCallable.java:48) ~[reactor-core-3.7.3.jar:3.7.3]
	at reactor.core.publisher.MonoCacheTime.subscribeOrReturn(MonoCacheTime.java:143) ~[reactor-core-3.7.3.jar:3.7.3]
	at reactor.core.publisher.Mono.subscribe(Mono.java:4560) ~[reactor-core-3.7.3.jar:3.7.3]
	at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:126) ~[reactor-core-3.7.3.jar:3.7.3]
	at reactor.core.scheduler.BoundedElasticThreadPerTaskScheduler$SchedulerTask.run(BoundedElasticThreadPerTaskScheduler.java:1013) ~[reactor-core-3.7.3.jar:3.7.3]
	at java.base/java.lang.VirtualThread.run(Unknown Source) ~[na:na]
Caused by: java.lang.InterruptedException: null
	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(Unknown Source) ~[na:na]
	at java.base/java.util.concurrent.CountDownLatch.await(Unknown Source) ~[na:na]
	at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:91) ~[reactor-core-3.7.3.jar:3.7.3]
	... 18 common frames omitted
```

/kind bug

```release-note
None
```

Author: JohnNiang

app/build.gradle

@@ -15,14 +15,15 @@ jar {
 dependencies {
     implementation project(':api')
     compileOnly 'run.halo.app:api'
-    implementation 'org.dom4j:dom4j:2.1.3'
+    implementation 'org.dom4j:dom4j:2.1.4'
     implementation('org.apache.commons:commons-text:1.10.0') {
         // Because the transitive dependency already exists in halo.
         exclude group: 'org.apache.commons', module: 'commons-lang3'
     }
 
     testImplementation 'run.halo.app:api'
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
+    testImplementation 'io.projectreactor:reactor-test'
 }
 
 tasks.register('copyUI', Copy) {

app/src/main/java/run/halo/feed/RssCacheManager.java

@@ -3,6 +3,7 @@
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import java.time.Duration;
+import java.util.concurrent.ExecutionException;
 import java.util.function.Predicate;
 import lombok.RequiredArgsConstructor;
 import org.springframework.context.event.EventListener;
@@ -18,21 +19,20 @@
 @RequiredArgsConstructor
 public class RssCacheManager {
     private final Duration expireMinutes = Duration.ofMinutes(60);
-    private final Cache<String, String> cache = CacheBuilder.newBuilder()
+    private final Cache<String, Mono<String>> cache = CacheBuilder.newBuilder()
         .expireAfterWrite(expireMinutes)
         .build();
 
     private final SystemInfoGetter systemInfoGetter;
     private final ReactiveSettingFetcher settingFetcher;
 
     public Mono<String> get(String requestPath, Mono<RSS2> loader) {
-        return Mono.fromCallable(() -> cache.get(requestPath,
-                () -> generateRssXml(requestPath, loader)
-                    .doOnNext(xml -> cache.put(requestPath, xml))
-                    .block()
-            ))
-            .cache()
-            .subscribeOn(Schedulers.boundedElastic());
+        try {
+            return cache.get(requestPath, () -> generateRssXml(requestPath, loader).cache())
+                .subscribeOn(Schedulers.boundedElastic());
+        } catch (ExecutionException e) {
+            return Mono.error(e);
+        }
     }
 
     private Mono<String> generateRssXml(String requestPath, Mono<RSS2> loader) {
@@ -51,10 +51,8 @@ private Mono<String> generateRssXml(String requestPath, Mono<RSS2> loader) {
             .doOnNext(prop -> builder.withExtractRssTags(prop.getRssExtraTags()));
 
         return Mono.when(rssMono, generatorMono, extractTagsMono)
-            // toXmlString is a blocking operation
-            .then(Mono.fromCallable(builder::toXmlString)
-                .subscribeOn(Schedulers.boundedElastic())
-            );
+            .thenReturn(builder)
+            .flatMap(RssXmlBuilder::toXmlString);
     }
 
     @EventListener(PluginConfigUpdatedEvent.class)

app/src/main/java/run/halo/feed/RssXmlBuilder.java

@@ -2,6 +2,7 @@
 
 import com.google.common.base.Throwables;
 import java.io.StringReader;
+import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.time.Instant;
@@ -20,10 +21,14 @@
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.client.reactive.ReactorClientHttpConnector;
 import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
 import org.springframework.util.CollectionUtils;
 import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClientException;
 import org.springframework.web.util.UriComponentsBuilder;
 import org.springframework.web.util.UriUtils;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.netty.http.client.HttpClient;
 import run.halo.feed.telemetry.TelemetryEndpoint;
 
@@ -82,7 +87,7 @@ RssXmlBuilder withExternalUrl(String externalUrl) {
         return this;
     }
 
-    public String toXmlString() {
+    public Mono<String> toXmlString() {
         Document document = DocumentHelper.createDocument();
 
         Element root = DocumentHelper.createElement("rss");
@@ -136,9 +141,10 @@ public String toXmlString() {
         }
 
         var items = rss2.getItems();
-        createItemElementsToChannel(channel, items);
 
-        return document.asXML();
+        return createItemElementsToChannel(channel, items)
+            .thenReturn(document)
+            .map(Document::asXML);
     }
 
     private void copyAttributesAndChildren(Element target, Element source) {
@@ -171,14 +177,16 @@ private Element parseXmlString(String xml) throws DocumentException {
         }
     }
 
-    private void createItemElementsToChannel(Element channel, List<RSS2.Item> items) {
+    private Mono<Void> createItemElementsToChannel(Element channel, List<RSS2.Item> items) {
         if (CollectionUtils.isEmpty(items)) {
-            return;
+            return Mono.empty();
         }
-        items.forEach(item -> createItemElementToChannel(channel, item));
+        return Flux.fromIterable(items)
+            .flatMap(item -> createItemElementToChannel(channel, item))
+            .then();
     }
 
-    private void createItemElementToChannel(Element channel, RSS2.Item item) {
+    private Mono<Void> createItemElementToChannel(Element channel, RSS2.Item item) {
         Element itemElement = channel.addElement("item");
         itemElement.addElement("title")
             .addCDATA(XmlCharUtils.removeInvalidXmlChar(item.getTitle()));
@@ -205,18 +213,19 @@ private void createItemElementToChannel(Element channel, RSS2.Item item) {
                 .addText(item.getAuthor());
         }
 
+        Mono<Void> handleEnclosure = Mono.empty();
         if (StringUtils.isNotBlank(item.getEnclosureUrl())) {
             var enclosureElement = itemElement.addElement("enclosure")
                 .addAttribute("url", item.getEnclosureUrl())
                 .addAttribute("type", item.getEnclosureType());
-
             var enclosureLength = item.getEnclosureLength();
+            enclosureElement.addAttribute("length", enclosureLength);
             if (StringUtils.isBlank(enclosureLength)) {
                 // https://www.rssboard.org/rss-validator/docs/error/MissingAttribute.html
-                var fileBytes = getFileSizeBytes(item.getEnclosureUrl());
-                enclosureLength = String.valueOf(fileBytes);
+                handleEnclosure = getFileSizeBytes(item.getEnclosureUrl())
+                    .doOnNext(fileBytes -> enclosureElement.addAttribute("length", String.valueOf(fileBytes)))
+                    .then();
             }
-            enclosureElement.addAttribute("length", enclosureLength);
         }
 
         nullSafeList(item.getCategories())
@@ -264,6 +273,7 @@ private void createItemElementToChannel(Element channel, RSS2.Item item) {
                 }
             }
         });
+        return handleEnclosure;
     }
 
     private String getDescriptionWithTelemetry(RSS2.Item item) {
@@ -298,23 +308,47 @@ static String instantToString(Instant instant) {
             .format(DateTimeFormatter.RFC_1123_DATE_TIME);
     }
 
+
     @NonNull
-    private Long getFileSizeBytes(String url) {
+    Mono<Long> getFileSizeBytes(String url) {
         return webClient.get()
-            .uri(url)
+            .uri(URI.create(url))
             .header(HttpHeaders.USER_AGENT, UA)
             // https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Range
             .header(HttpHeaders.RANGE, "bytes=0-0")
+            .headers(headers -> {
+                if (StringUtils.isNotBlank(externalUrl)) {
+                    // For Referrer anti-hotlinking
+                    headers.set(HttpHeaders.REFERER, externalUrl);
+                }
+            })
             .retrieve()
             .toBodilessEntity()
             .map(HttpEntity::getHeaders)
-            .mapNotNull(headers -> headers.getFirst(HttpHeaders.CONTENT_LENGTH))
-            .map(Long::parseLong)
+            .mapNotNull(headers -> headers.getFirst(HttpHeaders.CONTENT_RANGE))
+            .mapNotNull(RssXmlBuilder::parseLengthOfContentRange)
             .doOnError(e -> log.debug("Failed to get file size from url: {}", url,
                 Throwables.getRootCause(e))
             )
             .onErrorReturn(0L)
-            .blockOptional()
-            .orElse(0L);
+            .defaultIfEmpty(0L);
     }
+
+    @Nullable
+    static Long parseLengthOfContentRange(@Nullable String contentRange) {
+        if (contentRange == null) {
+            return null;
+        }
+        // Refer to https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Range#syntax
+        var range = contentRange.split("/");
+        if (range.length != 2) {
+            return null;
+        }
+        var lengthStr = range[1];
+        if ("*".equals(lengthStr)) {
+            return null;
+        }
+        return Long.parseLong(range[1]);
+    }
+
 }

app/src/main/java/run/halo/feed/provider/AbstractPostRssProvider.java

@@ -126,8 +126,7 @@ protected Flux<PostWithContent> listPostsByFunc(Function<BasicProp, Flux<Post>>
 
     protected Mono<User> fetchUser(String username) {
         return client.fetch(User.class, username)
-            .switchIfEmpty(Mono.error(new ServerWebInputException("User not found")))
-            .subscribeOn(Schedulers.boundedElastic());
+            .switchIfEmpty(Mono.error(new ServerWebInputException("User not found")));
     }
 
     private Flux<String> fetchCategoryDisplayName(List<String> categoryNames) {
@@ -137,8 +136,7 @@ private Flux<String> fetchCategoryDisplayName(List<String> categoryNames) {
         return client.listAll(Category.class, ListOptions.builder()
                 .fieldQuery(QueryFactory.in("metadata.name", categoryNames))
                 .build(), ExtensionUtil.defaultSort())
-            .map(category -> category.getSpec().getDisplayName())
-            .subscribeOn(Schedulers.boundedElastic());
+            .map(category -> category.getSpec().getDisplayName());
     }
 
     @Data

app/src/test/java/run/halo/feed/RSS2Test.java

@@ -6,6 +6,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import org.junit.jupiter.api.Test;
+import reactor.test.StepVerifier;
 
 class RSS2Test {
 
@@ -80,7 +81,10 @@ void toXmlString() {
             	</channel>
             </rss>
             """.formatted(lastBuildDate);
-        assertThat(rssXml).isEqualToIgnoringWhitespace(expected);
+
+        StepVerifier.create(rssXml)
+            .assertNext(xml -> assertThat(xml).isEqualToIgnoringWhitespace(expected))
+            .verifyComplete();
     }
 
     @Test
@@ -132,7 +136,9 @@ void extractRssTagsTest() {
                 </channel>
              </rss>
             """.formatted(lastBuildDate);
-        assertThat(rssXml).isEqualToIgnoringWhitespace(expected);
+        StepVerifier.create(rssXml)
+            .assertNext(xml -> assertThat(xml).isEqualToIgnoringWhitespace(expected))
+            .verifyComplete();
     }
 
     @Test
@@ -188,6 +194,8 @@ void invalidCharTest() {
             	</channel>
             </rss>
             """.formatted(lastBuildDate);
-        assertThat(rssXml).isEqualToIgnoringWhitespace(expected);
+        StepVerifier.create(rssXml)
+            .assertNext(xml -> assertThat(xml).isEqualToIgnoringWhitespace(expected))
+            .verifyComplete();
     }
 }

app/src/test/java/run/halo/feed/RssXmlBuilderTest.java

@@ -0,0 +1,28 @@
+package run.halo.feed;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+import static run.halo.feed.RssXmlBuilder.parseLengthOfContentRange;
+
+import java.util.stream.Stream;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class RssXmlBuilderTest {
+
+    static Stream<Arguments> testParseLengthOfContentRange() {
+        return Stream.of(
+            arguments("bytes 0-0/123", 123L),
+            arguments("bytes */567", 567L),
+            arguments("bytes 0-0", null),
+            arguments("bytes 0-0/*", null)
+        );
+    }
+
+    @ParameterizedTest
+    @MethodSource
+    void testParseLengthOfContentRange(String contentRange, Long expected) {
+        assertEquals(expected, parseLengthOfContentRange(contentRange));
+    }
+}

app/src/test/resources/logback.xml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <include resource="org/springframework/boot/logging/logback/defaults.xml"/>
+    <include resource="org/springframework/boot/logging/logback/console-appender.xml" />
+    <root level="INFO">
+        <appender-ref ref="CONSOLE" />
+    </root>
+    <logger name="run.halo.feed" level="DEBUG"/>
+</configuration>