media-model-proxy: fail closed when every replica drops scores

collectToTry + filterException swallowed every NonFatal xx-nsfw
error, including greyscale decode failures that were already mapped
to GreyScaleImageException. The controller then returned a successful
empty annotation map, which Media Analysis persists as unlabeled /
SFW. Keep partial-replica failover; fail the request when every
attempted replica errors.

Co-authored-by: Jon Bailey <Pitchfork-and-Torch@users.noreply.github.com>

Author: cursoragent

media-model-proxy/src/main/scala/com/twitter/media_understanding/model_proxy/services/ModelsService.scala

@@ -84,10 +84,22 @@ class ModelsService(
 
         Future
           .collectToTry(result)
-          .map { responses: Seq[Try[PredictionResponse]] =>
-            responses
+          .flatMap { responses: Seq[Try[PredictionResponse]] =>
+            val successes = responses
               .filter { response => filterException(response, model) }
               .map { predictionResponseTry => predictionResponseTry.get().prediction }
+
+            // Attempted replicas that all fail must not become a successful empty
+            // score list: callers persist that as unlabeled / SFW.
+            if (result.nonEmpty && successes.isEmpty) {
+              val cause = responses
+                .collectFirst { case Throw(ex) => ex }
+                .getOrElse(new IllegalStateException(s"${model.name} produced no scores"))
+              stats.scope(model.name).counter("fail_closed").incr()
+              Future.exception(cause)
+            } else {
+              Future.value(successes)
+            }
           }
     }
   }

media-model-proxy/src/test/scala/com/twitter/media_understanding/model_proxy/services/ModelsServiceSpec.scala

@@ -4,6 +4,7 @@ import com.twitter.cortex_media_annotator.thriftscala.MediaModel
 import com.twitter.finagle.NoBrokersAvailableException
 import com.twitter.finagle.stats.NullStatsReceiver
 import com.twitter.media_understanding.model_proxy.clients.ModelClient
+import com.twitter.media_understanding.model_proxy.exception.GreyScaleImageException
 import com.twitter.media_understanding.model_proxy.model_descriptors.ModelDescriptor
 import com.twitter.media_understanding.model_proxy.model_descriptors.ModelDescriptorRegistry
 import com.twitter.mediaservices.commons.thriftscala.MediaCategory
@@ -88,7 +89,7 @@ class ModelsServiceSpec extends FunSuite with MockitoSugar with BeforeAndAfterEa
   }
 
   test(
-    "getPrediction returns empty list of response if all models in cluster throw PredictionException"
+    "getPrediction fails closed if all models in cluster throw PredictionException"
   ) {
     when(client_1.isAvailable)
       .thenReturn(true)
@@ -100,12 +101,13 @@ class ModelsServiceSpec extends FunSuite with MockitoSugar with BeforeAndAfterEa
     when(client_2.predictFromModel(mockAny[PredictionRequest]))
       .thenReturn(Future.exception(new PredictionServiceException("")))
 
-    val result = Await.result(deepBirdService.getPrediction(Some(mediaCategory), model, media))
-    assert(result.isEmpty)
+    intercept[PredictionServiceException] {
+      Await.result(deepBirdService.getPrediction(Some(mediaCategory), model, media))
+    }
   }
 
   test(
-    "getPrediction returns empty list of response if all models in cluster throw non fatal exceptions"
+    "getPrediction fails closed if all models in cluster throw non fatal exceptions"
   ) {
     when(client_1.isAvailable)
       .thenReturn(true)
@@ -117,8 +119,27 @@ class ModelsServiceSpec extends FunSuite with MockitoSugar with BeforeAndAfterEa
     when(client_2.predictFromModel(mockAny[PredictionRequest]))
       .thenReturn(Future.exception(new IllegalArgumentException()))
 
-    val result = Await.result(deepBirdService.getPrediction(Some(mediaCategory), model, media))
-    assert(result.isEmpty)
+    intercept[IllegalArgumentException] {
+      Await.result(deepBirdService.getPrediction(Some(mediaCategory), model, media))
+    }
+  }
+
+  test(
+    "getPrediction fails closed with GreyScaleImageException when every replica hits a greyscale decode error"
+  ) {
+    val grey = new PredictionServiceException(deepBirdService.greyScaleErrorString)
+    when(client_1.isAvailable)
+      .thenReturn(true)
+    when(client_2.isAvailable)
+      .thenReturn(true)
+    when(client_1.predictFromModel(mockAny[PredictionRequest]))
+      .thenReturn(Future.exception(grey))
+    when(client_2.predictFromModel(mockAny[PredictionRequest]))
+      .thenReturn(Future.exception(grey))
+
+    intercept[GreyScaleImageException] {
+      Await.result(deepBirdService.getPrediction(Some(mediaCategory), model, media))
+    }
   }
 
   test(
@@ -155,6 +176,8 @@ class ModelsServiceSpec extends FunSuite with MockitoSugar with BeforeAndAfterEa
     when(client_2.isAvailable)
       .thenReturn(false)
 
+    // Kill-switched / filtered-out replicas were never attempted, so empty is
+    // not a swallowed error.
     val result2 = Await.result(deepBirdService.getPrediction(Some(mediaCategory), model, media))
     assert(result2.isEmpty)
   }