Add max error for exiting

README.md

@@ -38,7 +38,7 @@ Please apply the SQLs to your database.
 
 ```
 resolvers += Resolver.bintrayRepo("givers", "maven")
-libraryDependencies += "givers.moonlight" %% "play-moonlight" % "0.2.0"
+libraryDependencies += "givers.moonlight" %% "play-moonlight" % "0.3.0"
 ```
 
 The artifacts are hosted here: https://bintray.com/givers/maven/play-moonlight
@@ -92,12 +92,18 @@ Create a module with defined `WorkerSpec`:
 ```
 package modules
 
-import givers.moonlight.Moonlight
+import givers.moonlight.{Config, Moonlight}
 import play.api.{Configuration, Environment}
 
 class MoonlightModule extends play.api.inject.Module {
   def bindings(environment: Environment, configuration: Configuration)  = Seq(
-    bind[Moonlight].toInstance(new Moonlight(SimpleWorkerSpec))
+    bind[Moonlight].toInstance(new Moonlight(
+      // When 3 errors occurs, Moonlight will exit. This is for avoiding being stuck in error repetitively.
+      // For example, in our case, on Heroku, when Moonlight runs on a bad machine, it will stop by itself and be
+      // started on a good machine.
+      config = Config(maxErrorCountToKillOpt = Some(3),
+      workers = Seq(SimpleWorkerSpec)
+    ))
   )
 }
 ```

build.sbt

@@ -14,7 +14,7 @@ libraryDependencies ++= Seq(
 
 organization := "givers.moonlight"
 name := "play-moonlight"
-version := "0.2.1"
+version := "0.3.0"
 parallelExecution in Test := false
 
 publishMavenStyle := true

src/main/scala/givers/moonlight/Main.scala

@@ -2,7 +2,7 @@ package givers.moonlight
 
 import java.time.Instant
 import java.util.concurrent.TimeUnit
-import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
 
 import com.google.inject.Inject
 import play.api._
@@ -12,7 +12,11 @@ import scala.concurrent.duration.Duration
 import scala.concurrent.{Await, ExecutionContext, Future}
 
 
-class Moonlight(val workers: WorkerSpec*)
+case class Config(
+  maxErrorCountToKillOpt: Option[Int]
+)
+
+class Moonlight(val config: Config, val workers: Seq[WorkerSpec])
 
 object Main {
   private[this] val logger = Logger(this.getClass)
@@ -55,10 +59,12 @@ class Main @Inject()(
   private[this] val DEFAULT_FUTURE_TIMEOUT = Duration.apply(5, TimeUnit.MINUTES)
   private[this] val logger = Logger(this.getClass)
 
+  val errorCount = new AtomicInteger(0)
+
   var sleep: Long => Unit = Thread.sleep
   val running = new AtomicBoolean(true)
 
-  def run(args: Array[String]): Unit = {
+  def run(args: Array[String]): Unit = try {
     running.set(true)
 
     Runtime.getRuntime.addShutdownHook(new Thread() {
@@ -71,8 +77,12 @@ class Main @Inject()(
     while (running.get()) {
       runOneJob(running)
     }
-
+  } catch { case e: Throwable =>
+    logger.error("Error captured in Main.run(). Exit", e)
+    System.exit(1) // force terminating all hanging threads. This prevents a hang when there's an exception.
+  } finally {
     logger.info("Exit")
+    System.exit(0) // force terminating all hanging threads.
   }
 
   def getWorker(jobType: String): Worker[_] = {
@@ -108,6 +118,15 @@ class Main @Inject()(
           } catch {
             case e: InterruptedException => throw e
             case e: Throwable =>
+              errorCount.incrementAndGet()
+
+              moonlight.config.maxErrorCountToKillOpt.foreach { maxErrorCountToKill =>
+                if (maxErrorCountToKill <= errorCount.get) {
+                  logger.warn(s"Too many errors (maxErrorCountToKill = $maxErrorCountToKill, currentErrorCount = ${errorCount.get}). Exit")
+                  running.set(false)
+                }
+              }
+
               await(backgroundJobService.fail(job.id, e))
               logger.error(s"Error occurred while running ${runnable.getClass.getSimpleName} (id=${job.id}, type=${job.jobType}, params=${job.paramsInJsonString}.", e)
               logger.info(s"Finished ${runnable.getClass.getSimpleName} (id=${job.id}) with the above error")

src/test/scala/givers.moonlight/MainIntegrationSpec.scala

@@ -23,7 +23,7 @@ object MainIntegrationSpec extends BaseSpec {
       )))
       .bindings(new Module {
         override def bindings(environment: Environment, configuration: Configuration) = Seq(
-          bind[Moonlight].toInstance(new Moonlight(SimpleWorker))
+          bind[Moonlight].toInstance(new Moonlight(Config(maxErrorCountToKillOpt = None), Seq(SimpleWorker)))
         )
       })
       .in(Mode.Test)

src/test/scala/givers.moonlight/MainSpec.scala

@@ -17,7 +17,8 @@ object MainSpec extends BaseSpec {
   val tests = Tests {
     val injector = mock[Injector]
     val app = mock[Application]
-    val moonlight = new Moonlight(SimpleWorker, AmbiguousWorker)
+    val config = Config(maxErrorCountToKillOpt = Some(10))
+    val moonlight = new Moonlight(config, Seq(SimpleWorker, AmbiguousWorker))
     val backgroundJobService = mock[BackgroundJobService]
     val main = new Main(app, moonlight, backgroundJobService)
     main.sleep = { _ => () }
@@ -106,6 +107,28 @@ object MainSpec extends BaseSpec {
         verifyNoMoreInteractions(backgroundJobService)
       }
 
+      "Fail too many times" - {
+        when(worker.run(any())).thenAnswer(new Answer[Unit] {
+          override def answer(invocation: InvocationOnMock) = throw new Exception("FakeError")
+        })
+        when(backgroundJobService.get()).thenReturn(Future(Some(job)))
+
+        0.to(9).foreach { _ =>
+          main.runOneJob(running)
+        }
+        assert(running.get())
+
+        main.runOneJob(running)
+        assert(!running.get())
+
+        verify(worker, times(10)).run(job)
+        verify(backgroundJobService, times(10)).updateTimeoutJobs()
+        verify(backgroundJobService, times(10)).get()
+        verify(backgroundJobService, times(10)).start(job.id, 1)
+        verify(backgroundJobService, times(10)).fail(eq(job.id), argThat { e: Throwable => e.getMessage == "FakeError" })
+        verifyNoMoreInteractions(backgroundJobService)
+      }
+
       "No job" - {
         when(backgroundJobService.get()).thenReturn(Future(None))
 

test-project/app/worker/MoonlightModule.scala

@@ -1,10 +1,10 @@
 package worker
 
-import givers.moonlight.Moonlight
+import givers.moonlight.{Config, Moonlight}
 import play.api.{Configuration, Environment}
 
 class MoonlightModule extends play.api.inject.Module {
   def bindings(environment: Environment, configuration: Configuration)  = Seq(
-    bind[Moonlight].toInstance(new Moonlight(SimpleWorkerSpec))
+    bind[Moonlight].toInstance(new Moonlight(Config(maxErrorCountToKillOpt = Some(3)), Seq(SimpleWorkerSpec)))
   )
 }

test-project/app/worker/SimpleWorkerSpec.scala

@@ -25,5 +25,6 @@ class SimpleWorker @Inject()() extends Worker[SimpleWorkerSpec.Job] {
   def run(param: SimpleWorkerSpec.Job, job: BackgroundJob): Unit = {
     println(s"Process data: ${param.data}")
     Thread.sleep(10000)
+    throw new Exception("Fake error")
   }
 }

test-project/conf/logback.xml

@@ -0,0 +1,13 @@
+<!-- https://www.playframework.com/documentation/latest/SettingsLogger -->
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%level %date{HH:mm:ss} %logger{15} - %message%n%xException{10}</pattern>
+        </encoder>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT" />
+    </root>
+</configuration>
+