Fix unsorted unscored streaming multi searcher

src/main/java/it/cavallium/dbengine/database/disk/CachedIndexSearcherManager.java

@@ -35,7 +35,7 @@ public class CachedIndexSearcherManager implements IndexSearcherManager {
 
 	private static final Logger logger = LoggerFactory.getLogger(CachedIndexSearcherManager.class);
 	private final Executor SEARCH_EXECUTOR = Executors
-			.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), new ShortNamedThreadFactory("lucene-search"));
+			.newCachedThreadPool(new ShortNamedThreadFactory("lucene-search"));
 	private final SearcherFactory SEARCHER_FACTORY = new ExecutorSearcherFactory(SEARCH_EXECUTOR);
 
 	private final SnapshotsManager snapshotsManager;

src/main/java/it/cavallium/dbengine/lucene/LuceneUtils.java

@@ -350,6 +350,9 @@ public class LuceneUtils {
 		);
 	}
 
+	/**
+	 * HitsFlux must run on a blocking scheduler
+	 */
 	public static Flux<LLKeyScore> convertHits(Flux<ScoreDoc> hitsFlux,
 			List<IndexSearcher> indexSearchers,
 			String keyFieldName,

src/main/java/it/cavallium/dbengine/lucene/collector/ReactiveLeafCollector.java

@@ -2,6 +2,8 @@ package it.cavallium.dbengine.lucene.collector;
 
 import it.cavallium.dbengine.database.LLUtils;
 import it.cavallium.dbengine.lucene.searcher.LongSemaphore;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.LockSupport;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.search.CollectionTerminatedException;
@@ -42,14 +44,17 @@ public class ReactiveLeafCollector implements LeafCollector {
 
 		// Wait if no requests from downstream are found
 		try {
-			requested.acquire();
+			while (!requested.tryAcquire(1, TimeUnit.SECONDS)) {
+				if (scoreDocsSink.isCancelled()) {
+					throw new CollectionTerminatedException();
+				}
+			}
+			// Send the response
+			var scoreDoc = new ScoreDoc(leafReaderContext.docBase + i, 0, shardIndex);
+			scoreDocsSink.next(scoreDoc);
 		} catch (InterruptedException e) {
-			throw new CollectionTerminatedException();
+			throw new CompletionException(e);
 		}
 
-		// Send the response
-		var scoreDoc = new ScoreDoc(leafReaderContext.docBase + i, 0, shardIndex);
-		scoreDocsSink.next(scoreDoc);
-
 	}
 }

src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredStreamingMultiSearcher.java

@@ -77,7 +77,7 @@ public class UnsortedUnscoredStreamingMultiSearcher implements MultiSearcher {
 							try {
 								LLUtils.ensureBlocking();
 								var collectorManager = cmm.get(shardIndex);
-								shard.search(localQueryParams.query(), collectorManager);
+								shard.search(localQueryParams.query(), collectorManager.newCollector());
 							} catch (IOException e) {
 								throw new CompletionException(e);
 							}
@@ -92,13 +92,13 @@ public class UnsortedUnscoredStreamingMultiSearcher implements MultiSearcher {
 							scoreDocsSink.complete();
 						}
 					});
-					scoreDocsSink.onCancel(() -> {
+					scoreDocsSink.onDispose(() -> {
 						for (CompletableFuture<?> future : futures) {
 							future.cancel(true);
 						}
 						combinedFuture.cancel(true);
 					});
-				}, OverflowStrategy.ERROR);
+				}, OverflowStrategy.BUFFER).subscribeOn(Schedulers.boundedElastic()).publishOn(Schedulers.boundedElastic());
 
 
 				Flux<LLKeyScore> resultsFlux = LuceneUtils.convertHits(scoreDocsFlux, shards, keyFieldName, false);