[ETCM-266]-replaced-rate-limiter-built-on-twitter

Author: Dmitry Voronov

build.sbt

@@ -122,7 +122,6 @@ lazy val node = {
       Dependencies.cats,
       Dependencies.monix,
       Dependencies.network,
-      Dependencies.twitterUtilCollection,
       Dependencies.crypto,
       Dependencies.scopt,
       Dependencies.logging,

project/Dependencies.scala

@@ -94,8 +94,6 @@ object Dependencies {
     "org.codehaus.janino" % "janino" % "3.1.2"
   )
 
-  val twitterUtilCollection = Seq("com.twitter" %% "util-collection" % "19.1.0")
-
   val crypto = Seq("org.bouncycastle" % "bcprov-jdk15on" % "1.66")
 
   val scopt = Seq("com.github.scopt" % "scopt_2.12" % "3.7.1")

src/main/scala/io/iohk/ethereum/db/cache/SimpleLRU.scala

@@ -0,0 +1,36 @@
+package io.iohk.ethereum.db.cache
+
+import java.util
+
+/**
+  * Simplest-possible implementation of insert-time based LRU
+  *
+  * @param maxElements - capacity of the inner map, between 0 and this value.
+  * @param ttlMillis   - entry invalidation interval
+  * @tparam K - stands for the type of the keys
+  */
+class SimpleLRU[K](maxElements: Int, ttlMillis: Long) {
+
+  private[this] val inner = new util.LinkedHashMap[K, Long](maxElements, 0.75f, false) {
+    override def removeEldestEntry(old: util.Map.Entry[K, Long]): Boolean = {
+      size() > maxElements || tooOld(old.getValue)
+    }
+  }
+
+  /**
+    * @param key - will be searching and added
+    * @return
+    *   true  - if there was such entry already
+    *   false - if not
+    */
+  def checkAndRefreshEntry(key: K): Boolean = inner.synchronized {
+    val existing = Option(inner.put(key, currentTime))
+    existing.exists(!tooOld(_))
+  }
+
+  // Override this to test
+  protected def currentTime: Long = System.currentTimeMillis()
+
+  private[this] def tooOld(time: Long) = time + ttlMillis < currentTime
+
+}

src/main/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpServer.scala

@@ -25,7 +25,7 @@ import org.json4s.native.Serialization
 import org.json4s.{DefaultFormats, JInt, native}
 import scala.concurrent.duration.{FiniteDuration, _}
 
-trait JsonRpcHttpServer extends Json4sSupport with RateLimit with Logger {
+trait JsonRpcHttpServer extends Json4sSupport with Logger {
   val jsonRpcController: JsonRpcBaseController
   val jsonRpcHealthChecker: JsonRpcHealthChecker
   val config: JsonRpcHttpServerConfig
@@ -54,37 +54,38 @@ trait JsonRpcHttpServer extends Json4sSupport with RateLimit with Logger {
       }
       .result()
 
+  val rateLimit = new RateLimit(config.rateLimit)
+
   val route: Route = cors(corsSettings) {
     (path("healthcheck") & pathEndOrSingleSlash & get) {
       handleHealthcheck()
     } ~ (path("buildinfo") & pathEndOrSingleSlash & get) {
       handleBuildInfo()
     } ~ (pathEndOrSingleSlash & post) {
-      (extractClientIP & entity(as[JsonRpcRequest])) { (clientAddress, request) =>
-        handleRequest(clientAddress, request)
-      } ~ entity(as[Seq[JsonRpcRequest]]) { request =>
-        handleBatchRequest(request)
+      // TODO: maybe rate-limit this one too?
+      entity(as[JsonRpcRequest]) {
+        case statusReq if statusReq.method == FaucetJsonRpcController.Status =>
+          handleRequest(statusReq)
+        case jsonReq => rateLimit {
+          handleRequest(jsonReq)
+        }
+      // TODO: separate paths for single and multiple requests
+      // TODO: to prevent repeated body and json parsing
+      } ~ entity(as[Seq[JsonRpcRequest]]) {
+        case _ if config.rateLimit.enabled =>
+          complete(StatusCodes.MethodNotAllowed, JsonRpcError.MethodNotFound)
+        case reqSeq =>
+          complete {
+            Task
+              .traverse(reqSeq)(request => jsonRpcController.handleRequest(request))
+              .runToFuture
+          }
       }
     }
   }
 
-  def handleRequest(clientAddress: RemoteAddress, request: JsonRpcRequest): StandardRoute = {
-    //FIXME: FaucetJsonRpcController.Status should be part of a Healthcheck request or alike.
-    // As a temporary solution, it is being excluded from the Rate Limit.
-    if (config.rateLimit.enabled && request.method != FaucetJsonRpcController.Status) {
-      handleRateLimitedRequest(clientAddress, request)
-    } else complete(handleResponse(jsonRpcController.handleRequest(request)).runToFuture)
-  }
-
-  def handleRateLimitedRequest(clientAddress: RemoteAddress, request: JsonRpcRequest): StandardRoute = {
-    if (isBelowRateLimit(clientAddress))
-      complete(handleResponse(jsonRpcController.handleRequest(request)).runToFuture)
-    else {
-      log.warn(s"Request limit exceeded for ip ${clientAddress.toIP.getOrElse("unknown")}")
-      complete(
-        (StatusCodes.TooManyRequests, JsonRpcError.RateLimitError(config.rateLimit.minRequestInterval.toSeconds))
-      )
-    }
+  def handleRequest(request: JsonRpcRequest): StandardRoute = {
+    complete(handleResponse(jsonRpcController.handleRequest(request)).runToFuture)
   }
 
   private def handleResponse(f: Task[JsonRpcResponse]): Task[(StatusCode, JsonRpcResponse)] = f map { jsonRpcResponse =>
@@ -128,15 +129,6 @@ trait JsonRpcHttpServer extends Json4sSupport with RateLimit with Logger {
     )
   }
 
-  private def handleBatchRequest(requests: Seq[JsonRpcRequest]) = {
-    if (!config.rateLimit.enabled) {
-      complete {
-        Task
-          .traverse(requests)(request => jsonRpcController.handleRequest(request))
-          .runToFuture
-      }
-    } else complete(StatusCodes.MethodNotAllowed, JsonRpcError.MethodNotFound)
-  }
 }
 
 object JsonRpcHttpServer extends Logger {
@@ -160,12 +152,15 @@ object JsonRpcHttpServer extends Logger {
     }
 
   trait RateLimitConfig {
+    // TODO: Move the rateLimit.enabled setting upwards:
+    // TODO: If we don't need to limit the request rate at all - we don't have to define the other settings
     val enabled: Boolean
     val minRequestInterval: FiniteDuration
     val latestTimestampCacheSize: Int
   }
 
   object RateLimitConfig {
+    // TODO: Use pureconfig
     def apply(rateLimitConfig: TypesafeConfig): RateLimitConfig =
       new RateLimitConfig {
         override val enabled: Boolean = rateLimitConfig.getBoolean("enabled")

src/main/scala/io/iohk/ethereum/jsonrpc/server/http/RateLimit.scala

@@ -1,25 +1,42 @@
 package io.iohk.ethereum.jsonrpc.server.http
 
-import java.time.Clock
-
-import akka.http.scaladsl.model.RemoteAddress
-import com.twitter.util.LruMap
-import io.iohk.ethereum.jsonrpc.server.http.JsonRpcHttpServer.JsonRpcHttpServerConfig
-
-trait RateLimit {
-
-  val config: JsonRpcHttpServerConfig
-
-  val latestRequestTimestamps = new LruMap[RemoteAddress, Long](config.rateLimit.latestTimestampCacheSize)
-
-  val clock: Clock = Clock.systemUTC()
-
-  def isBelowRateLimit(clientAddress: RemoteAddress): Boolean = {
-    val timeMillis = clock.instant().toEpochMilli
-    val latestRequestTimestamp = latestRequestTimestamps.getOrElse(clientAddress, 0L)
+import akka.http.scaladsl.model.{RemoteAddress, StatusCodes}
+import akka.http.scaladsl.server.{Directive0, Route}
+import io.iohk.ethereum.db.cache.SimpleLRU
+import io.iohk.ethereum.jsonrpc.server.http.JsonRpcHttpServer.RateLimitConfig
+import akka.http.scaladsl.server.Directives._
+import io.iohk.ethereum.jsonrpc.JsonRpcError
+import de.heikoseeberger.akkahttpjson4s.Json4sSupport
+import io.iohk.ethereum.jsonrpc.serialization.JsonSerializers
+import org.json4s.{DefaultFormats, Formats, Serialization, native}
+
+
+class RateLimit(config: RateLimitConfig) extends Directive0 with Json4sSupport {
+
+  private implicit val serialization: Serialization = native.Serialization
+  private implicit val formats: Formats = DefaultFormats + JsonSerializers.RpcErrorJsonSerializer
+
+  // It determines whether a request needs to be blocked
+  // Such algebras prevent if-elseif-else boilerplate in the JsonRPCServer code
+  val blockingAlgebra: (RemoteAddress => Boolean) = {
+    if (config.enabled) {
+      val lru = new SimpleLRU[RemoteAddress](
+        config.latestTimestampCacheSize,
+        config.minRequestInterval.toMillis
+      )
+      lru.checkAndRefreshEntry
+    } else {
+      _ => false
+    }
+  }
 
-    val response = latestRequestTimestamp + config.rateLimit.minRequestInterval.toMillis < timeMillis
-    if (response) latestRequestTimestamps.put(clientAddress, timeMillis)
-    response
+  override def tapply(f: Unit => Route): Route = extractClientIP { ip =>
+    if (blockingAlgebra(ip)) {
+      val err = JsonRpcError.RateLimitError(config.minRequestInterval.toSeconds)
+      complete((StatusCodes.TooManyRequests, err))
+    } else {
+      f.apply( () )
+    }
   }
-}
+
+}

src/test/scala/io/iohk/ethereum/db/cache/SimpleLRUSpec.scala

@@ -0,0 +1,53 @@
+package io.iohk.ethereum.db.cache
+
+import org.scalatest.wordspec.AnyWordSpec
+
+class SimpleLRUSpec extends AnyWordSpec {
+
+  var time = 0L
+
+  private object MockedLRU extends SimpleLRU[Int](10, 100) {
+    override protected def currentTime: Long = SimpleLRUSpec.this.time
+  }
+
+  "It" should {
+
+    "Respond false with all missing entries and preserve length" in {
+      val results = (0 until 100).map { i => MockedLRU.checkAndRefreshEntry(i) }
+      assert( results.forall( _ == false ) )
+    }
+
+    "Drop the records according to maxElements" in {
+      val existing = (90 until 100).map { i => MockedLRU.checkAndRefreshEntry(i) }
+      assert( existing.forall( _ == true ) )
+
+      // maxElements guaranteed that we have no space for those
+      val absent = (0 until 10).map { i => MockedLRU.checkAndRefreshEntry(i) }
+      assert( absent.forall( _ == false ) )
+    }
+
+    "Obsolete the records according to ttlMillis" in {
+      this.time = 0L
+      var results = (0 until 5).map { i => MockedLRU.checkAndRefreshEntry(i) }
+      assert( results.forall( _ == true ) )
+
+      this.time = 50L
+      results = (5 until 10).map { i => MockedLRU.checkAndRefreshEntry(i) }
+      assert( results.forall( _ == true ) )
+
+      this.time = 150L
+      results = (0 until 5).map { i => MockedLRU.checkAndRefreshEntry(i) }
+      assert( results.forall( _ == false ) )
+
+      // those were added at 50ms and still valid
+      results = (5 until 10).map { i => MockedLRU.checkAndRefreshEntry(i) }
+      assert( results.forall( _ == true ) )
+
+      this.time = 300L
+      results = (0 until 10).map { i => MockedLRU.checkAndRefreshEntry(i) }
+      assert( results.forall( _ == false ) )
+    }
+
+  }
+
+}

src/test/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpServerSpec.scala

@@ -215,7 +215,7 @@ class JsonRpcHttpServerSpec extends AnyFlatSpec with Matchers with ScalatestRout
       status shouldEqual StatusCodes.TooManyRequests
     }
 
-    fakeClock.advanceTime(2 * serverConfigWithRateLimit.rateLimit.minRequestInterval.toMillis)
+    Thread.sleep(30)
 
     postRequest ~> Route.seal(mockJsonRpcHttpServerWithRateLimit.route) ~> check {
       status shouldEqual StatusCodes.OK
@@ -382,7 +382,7 @@ class JsonRpcHttpServerSpec extends AnyFlatSpec with Matchers with ScalatestRout
 
     val rateLimitConfig = new RateLimitConfig {
       override val enabled: Boolean = false
-      override val minRequestInterval: FiniteDuration = FiniteDuration.apply(5, TimeUnit.SECONDS)
+      override val minRequestInterval: FiniteDuration = FiniteDuration.apply(20, TimeUnit.MILLISECONDS)
       override val latestTimestampCacheSize: Int = 1024
     }
 
@@ -397,7 +397,7 @@ class JsonRpcHttpServerSpec extends AnyFlatSpec with Matchers with ScalatestRout
 
     val rateLimitEnabledConfig = new RateLimitConfig {
       override val enabled: Boolean = true
-      override val minRequestInterval: FiniteDuration = FiniteDuration.apply(5, TimeUnit.SECONDS)
+      override val minRequestInterval: FiniteDuration = FiniteDuration.apply(20, TimeUnit.MILLISECONDS)
       override val latestTimestampCacheSize: Int = 1024
     }
 
@@ -412,14 +412,12 @@ class JsonRpcHttpServerSpec extends AnyFlatSpec with Matchers with ScalatestRout
 
     val mockJsonRpcController = mock[JsonRpcController]
     val mockJsonRpcHealthChecker = mock[JsonRpcHealthChecker]
-    val fakeClock = new FakeClock
 
     val mockJsonRpcHttpServer = new FakeJsonRpcHttpServer(
       jsonRpcController = mockJsonRpcController,
       jsonRpcHealthChecker = mockJsonRpcHealthChecker,
       config = serverConfig,
       cors = serverConfig.corsAllowedOrigins,
-      testClock = fakeClock
     )
 
     val corsAllowedOrigin = HttpOrigin("http://localhost:3333")
@@ -428,15 +426,13 @@ class JsonRpcHttpServerSpec extends AnyFlatSpec with Matchers with ScalatestRout
       jsonRpcHealthChecker = mockJsonRpcHealthChecker,
       config = serverConfig,
       cors = HttpOriginMatcher(corsAllowedOrigin),
-      testClock = fakeClock
     )
 
     val mockJsonRpcHttpServerWithRateLimit = new FakeJsonRpcHttpServer(
       jsonRpcController = mockJsonRpcController,
       jsonRpcHealthChecker = mockJsonRpcHealthChecker,
       config = serverConfigWithRateLimit,
       cors = serverConfigWithRateLimit.corsAllowedOrigins,
-      testClock = fakeClock
     )
   }
 }
@@ -468,13 +464,11 @@ class FakeJsonRpcHttpServer(
     val jsonRpcHealthChecker: JsonRpcHealthChecker,
     val config: JsonRpcHttpServerConfig,
     val cors: HttpOriginMatcher,
-    val testClock: Clock
 )(implicit val actorSystem: ActorSystem)
     extends JsonRpcHttpServer
     with Logger {
   def run(): Unit = ()
   override def corsAllowedOrigins: HttpOriginMatcher = cors
-  override val clock = testClock
 }
 
 class FakeClock extends Clock {