Merge pull request #65 from flashbots/backoff-retry

Add middleware to retry with backoff

.dockerignore

@@ -0,0 +1,2 @@
+tests
+cache

.pre-commit-config.yaml

@@ -18,3 +18,4 @@ repos:
       - id: 'mypy'
         additional_dependencies:
           - 'pydantic'
+          - 'types-requests'

Dockerfile

@@ -18,4 +18,4 @@ COPY . /app
 # easter eggs 😝
 RUN echo "PS1='🕵️:\[\033[1;36m\]\h \[\033[1;34m\]\W\[\033[0;35m\]\[\033[1;36m\]$ \[\033[0m\]'" >> ~/.bashrc
 
-CMD ["/bin/bash"]
+CMD [ "python", "run.py"]

Tiltfile

@@ -11,7 +11,7 @@ k8s_yaml(secret_from_dict("mev-inspect-db-credentials", inputs = {
     "password": "password",
 }))
 
-docker_build('mev-inspect', '.',
+docker_build('mev-inspect-py', '.',
     live_update=[
         sync('.', '/app'),
         run('cd /app && poetry install',

k8s/app.yaml

@@ -16,9 +16,8 @@ spec:
     spec:
       containers:
       - name: mev-inspect
-        image: mev-inspect:latest
-        command: [ "/bin/bash", "-c", "--" ]
-        args: [ "while true; do sleep 30; done;" ]
+        image: mev-inspect-py
+        command: [ "python", "run.py"]
         env:
         - name: POSTGRES_USER
           valueFrom:
@@ -30,6 +29,8 @@ spec:
             secretKeyRef:
               name: mev-inspect-db-credentials
               key: password
+        - name: POSTGRES_HOST
+          value: postgresql
         livenessProbe:
           exec:
             command:

mev_inspect/db.py

@@ -7,9 +7,9 @@ from sqlalchemy.orm import sessionmaker
 def get_sqlalchemy_database_uri():
     username = os.getenv("POSTGRES_USER")
     password = os.getenv("POSTGRES_PASSWORD")
-    server = "postgresql"
+    host = os.getenv("POSTGRES_HOST")
     db_name = "mev_inspect"
-    return f"postgresql://{username}:{password}@{server}/{db_name}"
+    return f"postgresql://{username}:{password}@{host}/{db_name}"
 
 
 def get_engine():

mev_inspect/retry.py

@@ -0,0 +1,59 @@
+import time
+from typing import (
+    Any,
+    Callable,
+    Collection,
+    Type,
+)
+
+from requests.exceptions import (
+    ConnectionError,
+    HTTPError,
+    Timeout,
+    TooManyRedirects,
+)
+from web3 import Web3
+from web3.middleware.exception_retry_request import check_if_retry_on_failure
+from web3.types import (
+    RPCEndpoint,
+    RPCResponse,
+)
+
+
+def exception_retry_with_backoff_middleware(
+    make_request: Callable[[RPCEndpoint, Any], RPCResponse],
+    web3: Web3,  # pylint: disable=unused-argument
+    errors: Collection[Type[BaseException]],
+    retries: int = 5,
+    backoff_time_seconds: float = 0.1,
+) -> Callable[[RPCEndpoint, Any], RPCResponse]:
+    """
+    Creates middleware that retries failed HTTP requests. Is a default
+    middleware for HTTPProvider.
+    """
+
+    def middleware(method: RPCEndpoint, params: Any) -> RPCResponse:
+        if check_if_retry_on_failure(method):
+            for i in range(retries):
+                try:
+                    return make_request(method, params)
+                # https://github.com/python/mypy/issues/5349
+                except errors:  # type: ignore
+                    if i < retries - 1:
+                        time.sleep(backoff_time_seconds)
+                        continue
+                    else:
+                        raise
+            return None
+        else:
+            return make_request(method, params)
+
+    return middleware
+
+
+def http_retry_with_backoff_request_middleware(
+    make_request: Callable[[RPCEndpoint, Any], Any], web3: Web3
+) -> Callable[[RPCEndpoint, Any], Any]:
+    return exception_retry_with_backoff_middleware(
+        make_request, web3, (ConnectionError, HTTPError, Timeout, TooManyRedirects)
+    )

pyproject.toml

@@ -32,7 +32,6 @@ build-backend = "poetry.core.masonry.api"
 lint = 'scripts.poetry.dev_tools:lint'
 test = 'scripts.poetry.dev_tools:test'
 isort = 'scripts.poetry.dev_tools:isort'
-mypy = 'scripts.poetry.dev_tools:mypy'
 black = 'scripts.poetry.dev_tools:black'
 pre_commit = 'scripts.poetry.dev_tools:pre_commit'
 start = 'scripts.poetry.docker:start'

run.py

@@ -0,0 +1,5 @@
+import time
+
+if __name__ == "__main__":
+    while True:
+        time.sleep(30)

scripts/inspect_block.py

@@ -22,6 +22,7 @@ from mev_inspect.crud.swaps import delete_swaps_for_block, write_swaps
 from mev_inspect.db import get_session
 from mev_inspect.miner_payments import get_miner_payments
 from mev_inspect.swaps import get_swaps
+from mev_inspect.retry import http_retry_with_backoff_request_middleware
 
 
 @click.group()
@@ -34,7 +35,7 @@ def cli():
 @click.argument("rpc")
 @click.option("--cache/--no-cache", default=True)
 def inspect_block(block_number: int, rpc: str, cache: bool):
-    base_provider = Web3.HTTPProvider(rpc)
+    base_provider = _get_base_provider(rpc)
     w3 = Web3(base_provider)
 
     if not cache:
@@ -49,7 +50,7 @@ def inspect_block(block_number: int, rpc: str, cache: bool):
 @click.argument("rpc")
 @click.option("--cache/--no-cache", default=True)
 def inspect_many_blocks(after_block: int, before_block: int, rpc: str, cache: bool):
-    base_provider = Web3.HTTPProvider(rpc)
+    base_provider = _get_base_provider(rpc)
     w3 = Web3(base_provider)
 
     if not cache:
@@ -156,5 +157,16 @@ def get_stats(classified_traces) -> dict:
     return stats
 
 
+def _get_base_provider(rpc: str) -> Web3.HTTPProvider:
+    base_provider = Web3.HTTPProvider(rpc)
+    base_provider.middlewares.remove("http_retry_request")
+    base_provider.middlewares.add(
+        http_retry_with_backoff_request_middleware,
+        "http_retry_with_backoff",
+    )
+
+    return base_provider
+
+
 if __name__ == "__main__":
     cli()