fix: failure in data query due to reconciler triggered by uncommitted transaction

application/src/main/java/run/halo/app/config/HaloConfiguration.java

@@ -6,6 +6,8 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.transaction.ReactiveTransactionManager;
+import org.springframework.transaction.reactive.TransactionalOperator;
 
 @Configuration(proxyBeanMethods = false)
 @EnableAsync
@@ -18,4 +20,9 @@ Jackson2ObjectMapperBuilderCustomizer objectMapperCustomizer() {
             builder.featuresToEnable(MapperFeature.ACCEPT_CASE_INSENSITIVE_ENUMS);
         };
     }
+
+    @Bean
+    TransactionalOperator transactionTemplate(ReactiveTransactionManager transactionManager) {
+        return TransactionalOperator.create(transactionManager);
+    }
 }

application/src/main/java/run/halo/app/extension/ReactiveExtensionClientImpl.java

@@ -26,7 +26,7 @@
 import org.springframework.data.domain.Sort;
 import org.springframework.data.util.Predicates;
 import org.springframework.stereotype.Component;
-import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.reactive.TransactionalOperator;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.retry.Retry;
@@ -57,6 +57,8 @@ public class ReactiveExtensionClientImpl implements ReactiveExtensionClient {
 
     private final IndexedQueryEngine indexedQueryEngine;
 
+    private final TransactionalOperator transactionalOperator;
+
     private final ConcurrentMap<GroupKind, AtomicBoolean> indexBuildingState =
         new ConcurrentHashMap<>();
 
@@ -151,7 +153,6 @@ private Mono<Unstructured> get(GroupVersionKind gvk, String name) {
     }
 
     @Override
-    @Transactional
     public <E extends Extension> Mono<E> create(E extension) {
         checkClientWritable(extension);
         return Mono.just(extension)
@@ -185,7 +186,6 @@ && hasText(extension.getMetadata().getGenerateName()))
     }
 
     @Override
-    @Transactional
     public <E extends Extension> Mono<E> update(E extension) {
         checkClientWritable(extension);
         // Refactor the atomic reference if we have a better solution.
@@ -223,7 +223,6 @@ private Mono<? extends Extension> getLatest(Extension extension) {
     }
 
     @Override
-    @Transactional
     public <E extends Extension> Mono<E> delete(E extension) {
         checkClientWritable(extension);
         // set deletionTimestamp
@@ -247,7 +246,8 @@ <E extends Extension> Mono<E> doCreate(E oldExtension, String name, byte[] data)
             var indexer = indexerFactory.getIndexer(gvk);
             return client.create(name, data)
                 .map(created -> converter.convertFrom(type, created))
-                .doOnNext(indexer::indexRecord);
+                .doOnNext(indexer::indexRecord)
+                .as(transactionalOperator::transactional);
         });
     }
 
@@ -258,7 +258,8 @@ <E extends Extension> Mono<E> doUpdate(E oldExtension, String name, Long version
             var indexer = indexerFactory.getIndexer(oldExtension.groupVersionKind());
             return client.update(name, version, data)
                 .map(updated -> converter.convertFrom(type, updated))
-                .doOnNext(indexer::updateRecord);
+                .doOnNext(indexer::updateRecord)
+                .as(transactionalOperator::transactional);
         });
     }
 

application/src/test/java/run/halo/app/extension/ReactiveExtensionClientTest.java

@@ -36,6 +36,7 @@
 import org.mockito.Spy;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.transaction.reactive.TransactionalOperator;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -63,6 +64,9 @@ class ReactiveExtensionClientTest {
     @Mock
     IndexerFactory indexerFactory;
 
+    @Mock
+    TransactionalOperator transactionalOperator;
+
     @Spy
     ObjectMapper objectMapper = JsonMapper.builder()
         .addModule(new JavaTimeModule())
@@ -76,6 +80,8 @@ void setUp() {
         lenient().when(schemeManager.get(eq(FakeExtension.class)))
             .thenReturn(fakeScheme);
         lenient().when(schemeManager.get(eq(fakeScheme.groupVersionKind()))).thenReturn(fakeScheme);
+        lenient().when(transactionalOperator.transactional(any(Mono.class)))
+            .thenAnswer(invocation -> invocation.getArgument(0));
     }
 
     FakeExtension createFakeExtension(String name, Long version) {