Fix result truncation, options mutation and count() row-drain issues in findBySearch

- all() now applies a default limit of 10,000 when none is set; the FTS
  service default of 10 hits silently truncated results.
- User-supplied SearchOptions are no longer mutated per subscription
  (sorts accumulated on retry/resubscribe and fluent siblings shared
  state); combining withOptions() with the individual with* methods now
  throws IllegalArgumentException.
- count()/exists() drain rows before reading the metadata trailer (the
  SDK only completes metaData() once rows are consumed) and request
  limit 0 when options are owned by the template.
- Stale-hit skip logging now fires on actual KV misses (findById maps
  DocumentNotFoundException to empty, so the previous onErrorResume was
  dead code); hydration uses flatMapSequential instead of sequential
  concatMap.
- Blocking count() throws CouchbaseQueryExecutionException instead of
  NPE on a null count.
- @Search combined with @query on one method now throws instead of
  silently preferring @Search.
- FTS integration tests skip when the cluster lacks the SEARCH
  capability; index-readiness wait loop no longer hot-loops or NPEs on
  exceptions without a message.

Signed-off-by: Emilien Bevierre <emilien.bevierre@couchbase.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: emilienbev

src/main/java/org/springframework/data/couchbase/core/ExecutableFindBySearchOperation.java

@@ -91,6 +91,9 @@ default Optional<T> first() {
 
 		/**
 		 * Get all matching elements, hydrated as entities via KV GET.
+		 * <p>
+		 * If no limit is specified via {@code withLimit(...)} or {@code withOptions(...)}, a default limit of 10,000 is
+		 * applied (the FTS service would otherwise return only its default of 10 hits).
 		 *
 		 * @return never {@literal null}.
 		 */

src/main/java/org/springframework/data/couchbase/core/ExecutableFindBySearchOperationSupport.java

@@ -120,7 +120,11 @@ public Stream<T> stream() {
 
 		@Override
 		public long count() {
-			return reactiveSupport.count().block();
+			Long count = reactiveSupport.count().block();
+			if (count == null) {
+				throw new CouchbaseQueryExecutionException("search count query did not return a count, index: " + indexName);
+			}
+			return count;
 		}
 
 		@Override

src/main/java/org/springframework/data/couchbase/core/ReactiveFindBySearchOperation.java

@@ -72,6 +72,9 @@ interface TerminatingFindBySearch<T> {
 
 		/**
 		 * Get all matching elements, hydrated as entities via KV GET.
+		 * <p>
+		 * If no limit is specified via {@code withLimit(...)} or {@code withOptions(...)}, a default limit of 10,000 is
+		 * applied (the FTS service would otherwise return only its default of 10 hits).
 		 *
 		 * @return never {@literal null}.
 		 */
@@ -120,6 +123,9 @@ interface FindBySearchWithQuery<T> extends TerminatingFindBySearch<T>, WithSearc
 
 	/**
 	 * Fluent method to specify options.
+	 * <p>
+	 * The given {@link SearchOptions} are passed to the SDK as provided (aside from collection routing) and must not be
+	 * combined with the individual {@code with*} configuration methods of this fluent API.
 	 *
 	 * @param <T> the entity type to use.
 	 */

src/main/java/org/springframework/data/couchbase/core/ReactiveFindBySearchOperationSupport.java

@@ -48,6 +48,12 @@ public class ReactiveFindBySearchOperationSupport implements ReactiveFindBySearc
 	private final ReactiveCouchbaseTemplate template;
 	private static final Logger LOG = LoggerFactory.getLogger(ReactiveFindBySearchOperationSupport.class);
 
+	/**
+	 * Limit applied when neither {@code withLimit(...)} nor {@code withOptions(...)} is used. Without an explicit size
+	 * the FTS service returns only its default of 10 hits, which would silently truncate {@code all()} results.
+	 */
+	static final int DEFAULT_LIMIT = 10_000;
+
 	public ReactiveFindBySearchOperationSupport(final ReactiveCouchbaseTemplate template) {
 		this.template = template;
 	}
@@ -223,7 +229,7 @@ public Flux<T> all() {
 				return TransactionalSupport.verifyNotInTransaction("findBySearch")
 						.thenMany(executeSearch()
 								.flatMapMany(ReactiveSearchResult::rows))
-						.concatMap(row -> hydrateRow(row))
+						.flatMapSequential(this::hydrateRow)
 						.onErrorMap(throwable -> {
 							if (throwable instanceof RuntimeException) {
 								return template.potentiallyConvertRuntimeException((RuntimeException) throwable);
@@ -240,9 +246,10 @@ public Mono<Long> count() {
 				Assert.notNull(indexName, "Index name must be specified via withIndex()");
 				Assert.notNull(searchRequest, "SearchRequest must be specified via matching()");
 
+				// rows must be drained before the SDK emits the metadata trailer, even with limit 0
 				return TransactionalSupport.verifyNotInTransaction("findBySearch")
-						.then(executeSearch())
-						.flatMap(ReactiveSearchResult::metaData)
+						.then(executeSearch(true))
+						.flatMap(result -> result.rows().then(result.metaData()))
 						.map(metaData -> metaData.metrics().totalRows())
 						.onErrorMap(throwable -> {
 							if (throwable instanceof RuntimeException) {
@@ -309,7 +316,7 @@ private Mono<SearchResult<T>> collectFullResult(ReactiveSearchResult reactiveRes
 						java.util.Map<String, SearchFacetResult> facetResults = tuple.getT3();
 
 						return Flux.fromIterable(searchRows)
-								.concatMap(row -> hydrateRow(row))
+								.flatMapSequential(this::hydrateRow)
 								.collectList()
 								.map(entities -> new SearchResult<>(entities, searchRows, metaData, facetResults));
 					});
@@ -323,19 +330,24 @@ private Mono<SearchResult<T>> collectFullResult(ReactiveSearchResult reactiveRes
 		 * misses typically indicate an out-of-sync FTS index.
 		 */
 		private Mono<T> hydrateRow(SearchRow row) {
+			// findById already maps DocumentNotFoundException to an empty Mono, so misses surface as emptiness
 			return template.findById(returnType)
 					.inScope(scope)
 					.inCollection(collection)
 					.one(row.id())
-					.onErrorResume(com.couchbase.client.core.error.DocumentNotFoundException.class, ex -> {
+					.switchIfEmpty(Mono.defer(() -> {
 						LOG.warn("Skipping stale FTS result for document id '{}': document not found in KV "
 								+ "(index '{}' may be out of sync)", row.id(), indexName);
 						return Mono.empty();
-					});
+					}));
 		}
 
 		private Mono<ReactiveSearchResult> executeSearch() {
-			SearchOptions opts = buildSearchOptions();
+			return executeSearch(false);
+		}
+
+		private Mono<ReactiveSearchResult> executeSearch(boolean metadataOnly) {
+			SearchOptions opts = buildSearchOptions(metadataOnly);
 			if (scope != null) {
 				return template.getCouchbaseClientFactory()
 						.withScope(scope).getScope().reactive()
@@ -347,8 +359,19 @@ private Mono<ReactiveSearchResult> executeSearch() {
 			}
 		}
 
-		private SearchOptions buildSearchOptions() {
-			SearchOptions opts = options != null ? options : SearchOptions.searchOptions();
+		private SearchOptions buildSearchOptions(boolean metadataOnly) {
+			if (options != null) {
+				if (hasFluentOptions()) {
+					throw new IllegalArgumentException("withOptions() cannot be combined with withConsistency(), withSort(), "
+							+ "withHighlight(), withFacets(), withFields(), withLimit() or withSkip(); "
+							+ "set those directly on the SearchOptions instead");
+				}
+				if (collection != null) {
+					options.collections(collection);
+				}
+				return options;
+			}
+			SearchOptions opts = SearchOptions.searchOptions();
 			if (scanConsistency != null) {
 				opts.scanConsistency(scanConsistency);
 			}
@@ -371,15 +394,24 @@ private SearchOptions buildSearchOptions() {
 			if (fields != null && fields.length > 0) {
 				opts.fields(fields);
 			}
-			if (limitSkip != null) {
-				if (limitSkip[0] != null) {
+			if (metadataOnly) {
+				opts.limit(0);
+			} else {
+				if (limitSkip != null && limitSkip[0] != null) {
 					opts.limit(limitSkip[0]);
+				} else {
+					opts.limit(DEFAULT_LIMIT);
 				}
-				if (limitSkip[1] != null) {
+				if (limitSkip != null && limitSkip[1] != null) {
 					opts.skip(limitSkip[1]);
 				}
 			}
 			return opts;
 		}
+
+		private boolean hasFluentOptions() {
+			return scanConsistency != null || (sort != null && sort.length > 0) || highlightStyle != null
+					|| (facets != null && !facets.isEmpty()) || (fields != null && fields.length > 0) || limitSkip != null;
+		}
 	}
 }

src/main/java/org/springframework/data/couchbase/repository/support/CouchbaseRepositoryFactory.java

@@ -168,6 +168,10 @@ public RepositoryQuery resolveQuery(final Method method, final RepositoryMetadat
 			CouchbaseQueryMethod queryMethod = new CouchbaseQueryMethod(method, metadata, factory, mappingContext);
 
 			if (queryMethod.hasSearchAnnotation()) {
+				if (queryMethod.hasN1qlAnnotation()) {
+					throw new IllegalArgumentException(
+							"Method " + method + " must not be annotated with both @Search and @Query");
+				}
 				return new SearchBasedCouchbaseQuery(queryMethod, couchbaseOperations);
 			} else if (queryMethod.hasN1qlAnnotation()) {
 				return new StringBasedCouchbaseQuery(queryMethod, couchbaseOperations,

src/main/java/org/springframework/data/couchbase/repository/support/ReactiveCouchbaseRepositoryFactory.java

@@ -157,6 +157,10 @@ public RepositoryQuery resolveQuery(Method method, RepositoryMetadata metadata,
 					mappingContext);
 
 			if (queryMethod.hasSearchAnnotation()) {
+				if (queryMethod.hasN1qlAnnotation()) {
+					throw new IllegalArgumentException(
+							"Method " + method + " must not be annotated with both @Search and @Query");
+				}
 				return new ReactiveSearchBasedCouchbaseQuery(queryMethod, couchbaseOperations);
 			} else if (queryMethod.hasN1qlAnnotation()) {
 				return new ReactiveStringBasedCouchbaseQuery(queryMethod, couchbaseOperations,

src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateFtsIntegrationTests.java

@@ -32,6 +32,7 @@
 import org.springframework.data.couchbase.core.mapping.CouchbaseMappingContext;
 import org.springframework.data.couchbase.repository.Collection;
 import org.springframework.data.couchbase.repository.Scope;
+import org.springframework.data.couchbase.util.Capabilities;
 import org.springframework.data.couchbase.util.ClusterType;
 import org.springframework.data.couchbase.util.IgnoreWhen;
 import org.springframework.data.couchbase.util.JavaIntegrationTests;
@@ -58,7 +59,7 @@
  * @author Emilien Bevierre
  * @since 6.2
  */
-@IgnoreWhen(clusterTypes = ClusterType.MOCKED)
+@IgnoreWhen(clusterTypes = ClusterType.MOCKED, missesCapabilities = Capabilities.SEARCH)
 class CouchbaseTemplateFtsIntegrationTests extends JavaIntegrationTests {
 
 	private static final String INDEX_NAME = "sd-fts-airport-idx";

src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateSearchIntegrationTests.java

@@ -27,6 +27,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.couchbase.domain.Config;
 import org.springframework.data.couchbase.domain.User;
+import org.springframework.data.couchbase.util.Capabilities;
 import org.springframework.data.couchbase.util.ClusterType;
 import org.springframework.data.couchbase.util.IgnoreWhen;
 import org.springframework.data.couchbase.util.JavaIntegrationTests;
@@ -51,7 +52,7 @@
  *
  * @since 6.2
  */
-@IgnoreWhen(clusterTypes = ClusterType.MOCKED)
+@IgnoreWhen(clusterTypes = ClusterType.MOCKED, missesCapabilities = Capabilities.SEARCH)
 @SpringJUnitConfig(Config.class)
 @DirtiesContext
 class CouchbaseTemplateSearchIntegrationTests extends JavaIntegrationTests {
@@ -341,16 +342,13 @@ private static void waitForFtsIndex(Cluster cluster, String indexName) {
 				cluster.searchQuery(indexName, SearchQuery.queryString("*"));
 				return;
 			} catch (Exception ex) {
-				if (i < maxRetries - 1 && (ex.getMessage().contains("no planPIndexes")
-						|| ex.getMessage().contains("pindex_consistency")
-						|| ex.getMessage().contains("pindex not available")
-						|| ex.getMessage().contains("index not found"))) {
-					sleepMs(2000);
-					continue;
-				}
-				if (i >= maxRetries - 1) {
+				String msg = ex.getMessage() != null ? ex.getMessage() : "";
+				boolean retryable = msg.contains("no planPIndexes") || msg.contains("pindex_consistency")
+						|| msg.contains("pindex not available") || msg.contains("index not found");
+				if (!retryable || i >= maxRetries - 1) {
 					throw new RuntimeException("FTS index " + indexName + " did not become ready in time", ex);
 				}
+				sleepMs(2000);
 			}
 		}
 	}

src/test/java/org/springframework/data/couchbase/repository/ReactiveSearchRepositoryFtsIntegrationTests.java

@@ -29,6 +29,7 @@
 import org.springframework.data.couchbase.core.mapping.CouchbaseMappingContext;
 import org.springframework.data.couchbase.repository.config.ReactiveRepositoryOperationsMapping;
 import org.springframework.data.couchbase.repository.support.ReactiveCouchbaseRepositoryFactory;
+import org.springframework.data.couchbase.util.Capabilities;
 import org.springframework.data.couchbase.util.ClusterType;
 import org.springframework.data.couchbase.util.IgnoreWhen;
 import org.springframework.data.couchbase.util.JavaIntegrationTests;
@@ -55,7 +56,7 @@
  * @author Emilien Bevierre
  * @since 6.2
  */
-@IgnoreWhen(clusterTypes = ClusterType.MOCKED)
+@IgnoreWhen(clusterTypes = ClusterType.MOCKED, missesCapabilities = Capabilities.SEARCH)
 class ReactiveSearchRepositoryFtsIntegrationTests extends JavaIntegrationTests {
 
 	private static final String INDEX_NAME = "sd-fts-airport-idx";