Remove hits we could not fill in time

4 files changed, 65 insertions, 15 deletions

diff --git a/container-search/src/main/java/com/yahoo/search/federation/FederationSearcher.java b/container-search/src/main/java/com/yahoo/search/federation/FederationSearcher.java
index 621f0fbe090..67898dc0289 100644
--- a/container-search/src/main/java/com/yahoo/search/federation/FederationSearcher.java
+++ b/container-search/src/main/java/com/yahoo/search/federation/FederationSearcher.java
@@ -2,6 +2,7 @@
 package com.yahoo.search.federation;
 
 import com.google.inject.Inject;
+import com.yahoo.collections.Pair;
 import com.yahoo.component.ComponentId;
 import com.yahoo.component.ComponentSpecification;
 import com.yahoo.component.chain.Chain;
@@ -391,7 +392,6 @@ public class FederationSearcher extends ForkingSearcher {
 
     @Override
     public void fill(Result result, String summaryClass, Execution execution) {
-        List<FutureResult> filledResults = new ArrayList<>();
         UniqueExecutionsToResults uniqueExecutionsToResults = new UniqueExecutionsToResults();
         addResultsToFill(result.hits(), result, summaryClass, uniqueExecutionsToResults);
         Set<Entry<Chain<Searcher>, Map<Query, Result>>> resultsForAllChains = 
@@ -402,6 +402,7 @@ public class FederationSearcher extends ForkingSearcher {
             numberOfCallsToFillNeeded += resultsToFillForAChain.getValue().size();
         }
 
+        List<Pair<Result, FutureResult>> futureFilledResults = new ArrayList<>();
         for (Entry<Chain<Searcher>, Map<Query, Result>> resultsToFillForAChain : resultsForAllChains) {
             Chain<Searcher> chain = resultsToFillForAChain.getKey();
             Execution chainExecution = (chain == null) ? execution : new Execution(chain, execution.context());
@@ -413,12 +414,24 @@ public class FederationSearcher extends ForkingSearcher {
                     propagateErrors(resultToFill, result);
                 } else {
                     AsyncExecution asyncFill = new AsyncExecution(chainExecution);
-                    filledResults.add(asyncFill.fill(resultToFill, summaryClass));
+                    futureFilledResults.add(new Pair<>(resultToFill, asyncFill.fill(resultToFill, summaryClass)));
                 }
             }
         }
-        for (FutureResult filledResult : filledResults) {
-            propagateErrors(filledResult.get(result.getQuery().getTimeLeft(), TimeUnit.MILLISECONDS), result);
+        for (Pair<Result, FutureResult> futureFilledResult : futureFilledResults) {
+            // futureFilledResult is a pair of a result to be filled and the future in which that same result is filled
+            Optional<Result> filledResult = futureFilledResult.getSecond().getIfAvailable(result.getQuery().getTimeLeft(), TimeUnit.MILLISECONDS);
+            if (filledResult.isPresent()) { // fill completed
+                propagateErrors(filledResult.get(), result);
+            }
+            else { // fill timed out: Remove these hits as they are incomplete and may cause a race when accessed later
+                result.hits().addError(futureFilledResult.getSecond().createTimeoutError());
+                for (Iterator<Hit> i = futureFilledResult.getFirst().hits().unorderedDeepIterator(); i.hasNext(); ) {
+                    // Note that some of these hits may be filled, but as the fill thread may still be working on them
+                    // and we do not synchronize with it we need to discard all
+                    Hit removed = result.hits().remove(i.next().getId());
+                }
+            }
         }
     }
 
@@ -435,7 +448,7 @@ public class FederationSearcher extends ForkingSearcher {
                 addResultsToFill((HitGroup) hit, result, summaryClass, uniqueExecutionsToResults);
             } else {
                 if ( ! hit.isFilled(summaryClass))
-                    getSearchChainGroup(hit,result,uniqueExecutionsToResults).hits().add(hit);
+                    getSearchChainGroup(hit, result, uniqueExecutionsToResults).hits().add(hit);
             }
         }
     }
@@ -445,7 +458,7 @@ public class FederationSearcher extends ForkingSearcher {
         Chain<Searcher> chain = (Chain<Searcher>) hit.getSearcherSpecificMetaData(this);
         Query query = hit.getQuery() !=null ? hit.getQuery() : result.getQuery();
 
-        return uniqueExecutionsToResults.get(chain,query);
+        return uniqueExecutionsToResults.get(chain, query);
     }
 
     /**
@@ -633,7 +646,7 @@ public class FederationSearcher extends ForkingSearcher {
             Result resultsToFillForAChainAndQuery = resultsToFillForAChain.get(query);
             if (resultsToFillForAChainAndQuery == null) {
                 resultsToFillForAChainAndQuery = new Result(query);
-                resultsToFillForAChain.put(query,resultsToFillForAChainAndQuery);
+                resultsToFillForAChain.put(query, resultsToFillForAChainAndQuery);
             }
 
             return resultsToFillForAChainAndQuery;
diff --git a/container-search/src/main/java/com/yahoo/search/searchchain/AsyncExecution.java b/container-search/src/main/java/com/yahoo/search/searchchain/AsyncExecution.java
index 739337add14..8706567c48d 100644
--- a/container-search/src/main/java/com/yahoo/search/searchchain/AsyncExecution.java
+++ b/container-search/src/main/java/com/yahoo/search/searchchain/AsyncExecution.java
@@ -117,11 +117,11 @@ public class AsyncExecution {
      *
      * @see com.yahoo.search.searchchain.Execution
      */
-    public FutureResult search(final Query query) {
+    public FutureResult search(Query query) {
         return getFutureResult(() -> execution.search(query), query);
     }
 
-    public FutureResult searchAndFill(final Query query) {
+    public FutureResult searchAndFill(Query query) {
         return getFutureResult(() -> {
             Result result = execution.search(query);
             execution.fill(result, query.getPresentation().getSummary());
@@ -138,7 +138,7 @@ public class AsyncExecution {
      *
      * @see com.yahoo.search.searchchain.Execution
      */
-    public FutureResult fill(final Result result, final String summaryClass) {
+    public FutureResult fill(Result result, String summaryClass) {
         return getFutureResult(() -> {
             execution.fill(result, summaryClass);
             return result;
diff --git a/container-search/src/main/java/com/yahoo/search/searchchain/FutureResult.java b/container-search/src/main/java/com/yahoo/search/searchchain/FutureResult.java
index ec87305da3b..0c74d331332 100644
--- a/container-search/src/main/java/com/yahoo/search/searchchain/FutureResult.java
+++ b/container-search/src/main/java/com/yahoo/search/searchchain/FutureResult.java
@@ -97,7 +97,7 @@ public class FutureResult extends FutureTask<Result> {
                                                         e.getCause());
     }
 
-    ErrorMessage createTimeoutError() {
+    public ErrorMessage createTimeoutError() {
         return ErrorMessage.createTimeout("Error executing '" + execution + "': " + " Chain timed out.");
     }
 
diff --git a/container-search/src/test/java/com/yahoo/search/federation/test/FederationSearcherTest.java b/container-search/src/test/java/com/yahoo/search/federation/test/FederationSearcherTest.java
index f9b37729956..16333b149d9 100644
--- a/container-search/src/test/java/com/yahoo/search/federation/test/FederationSearcherTest.java
+++ b/container-search/src/test/java/com/yahoo/search/federation/test/FederationSearcherTest.java
@@ -70,6 +70,29 @@ public class FederationSearcherTest {
 
     }
 
+    private static class TimeoutInFillSearcher extends Searcher {
+
+        private Hit createHit(String id) {
+            Hit hit = new Hit(id);
+            hit.setFillable();
+            return hit;
+        }
+
+        @Override
+        public Result search(Query query, Execution execution) {
+            Result result = execution.search(query);
+            result.hits().add(createHit("timeout1"));
+            result.hits().add(createHit("timeout2"));
+            return result;
+        }
+
+        @Override
+        public void fill(Result result, String summaryClass, Execution execution) {
+            try { Thread.sleep(500); } catch (InterruptedException e) {}
+        }
+
+    }
+
     private static class ModifyQueryAndAddHitSearcher extends AddHitSearcher {
 
         private final String marker;
@@ -138,13 +161,26 @@ public class FederationSearcherTest {
 
         Result result = tester.search();
         tester.fill(result);
-        for (Iterator<Hit> i = result.hits().deepIterator(); i.hasNext();) {
-            Hit h = i.next();
-            assertFilled(h);
-        }
         assertEquals(3, result.hits().getConcreteSize());
+        for (Iterator<Hit> i = result.hits().deepIterator(); i.hasNext();)
+            assertFilled(i.next());
     }
 
+    @Test
+    public void require_that_hits_that_time_out_in_fill_are_removed() {
+        FederationTester tester = new FederationTester();
+        tester.addSearchChain("chain1", new AddHitSearcher());
+        tester.addSearchChain("chain2", new TimeoutInFillSearcher());
+
+        Query query = new Query();
+        query.setTimeout(2);
+        Result result = tester.search(query);
+        tester.fill(result);
+        assertEquals(1, result.hits().getConcreteSize());
+        for (Iterator<Hit> i = result.hits().deepIterator(); i.hasNext();)
+            assertFilled(i.next());
+        assertEquals("Timed out", result.hits().getError().getMessage());
+    }
 
     @Test
     public void require_that_optional_search_chains_does_not_delay_federation() {
@@ -191,6 +227,7 @@ public class FederationSearcherTest {
         return nonErrorHits;
     }
     private static void assertFilled(Hit hit) {
+        if (hit.isMeta()) return;
         assertTrue((Boolean)hit.getField(hasBeenFilled));
     }