Function name

Tiltfile

@@ -107,13 +107,13 @@ local_resource(
 #    repo_name="localstack-charts",
 #    repo_url="https://localstack.github.io/helm-charts",
 #)
-
+#
 #local_resource(
 #    'localstack-port-forward',
 #    serve_cmd='kubectl port-forward --namespace default svc/localstack 4566:4566',
 #    resource_deps=["localstack"]
 #)
-
+#
 #k8s_yaml(configmap_from_dict("mev-inspect-export", inputs = {
 #    "services": "s3",
 #}))

cli.py

@@ -159,6 +159,22 @@ def enqueue_s3_export(block_number: int):
     export_actor.send(block_number)
 
 
+@cli.command()
+@click.argument("after_block", type=int)
+@click.argument("before_block", type=int)
+def enqueue_many_s3_exports(after_block: int, before_block: int):
+    broker = connect_broker()
+    export_actor = dramatiq.actor(
+        export_block_task,
+        broker=broker,
+        queue_name=HIGH_PRIORITY_QUEUE,
+        priority=HIGH_PRIORITY,
+    )
+    logger.info(f"Sending blocks {after_block} to {before_block} to queue")
+    for block_number in range(after_block, before_block):
+        export_actor.send(block_number)
+
+
 @cli.command()
 @click.argument("block_number", type=int)
 def s3_export(block_number: int):

mev

@@ -94,6 +94,13 @@ case "$1" in
             exit 1
         esac
   ;;
+  backfill-export)
+        after_block=$2
+        before_block=$3
+
+        echo "Sending $after_block to $before_block export to queue"
+        kubectl exec -ti deploy/mev-inspect -- poetry run enqueue-many-s3-exports $after_block $before_block
+  ;;
   enqueue-s3-export)
         block_number=$2
 

mev_inspect/s3_export.py

@@ -14,23 +14,33 @@ EXPORT_BUCKET_REGION_ENV = "EXPORT_BUCKET_REGION"
 EXPORT_AWS_ACCESS_KEY_ID_ENV = "EXPORT_AWS_ACCESS_KEY_ID"
 EXPORT_AWS_SECRET_ACCESS_KEY_ENV = "EXPORT_AWS_SECRET_ACCESS_KEY"
 
-MEV_SUMMARY_EXPORT_QUERY = """
+supported_tables = [
-    SELECT to_json(mev_summary)
+    "mev_summary",
-    FROM mev_summary
+    "arbitrages",
-WHERE
+    "liquidations",
-    block_number = :block_number
+    "sandwiches",
-    """
+    "sandwiched_swaps",
+]
 
 logger = logging.getLogger(__name__)
 
 
 def export_block(inspect_db_session, block_number: int) -> None:
-    export_bucket_name = get_export_bucket_name()
+
+    for table in supported_tables:
+
+        _export_block_by_table(inspect_db_session, block_number, table)
+
+
+def _export_block_by_table(inspect_db_session, block_number: int, table: str) -> None:
     client = get_s3_client()
-    object_key = f"mev_summary/flashbots_{block_number}.json"
+    export_bucket_name = get_export_bucket_name()
+    export_statement = _get_export_statement(table)
+
+    object_key = f"{table}/flashbots_{block_number}.json"
 
     mev_summary_json_results = inspect_db_session.execute(
-        statement=MEV_SUMMARY_EXPORT_QUERY,
+        statement=export_statement,
         params={
             "block_number": block_number,
         },
@@ -40,7 +50,7 @@ def export_block(inspect_db_session, block_number: int) -> None:
     if first_value is None:
         existing_object_size = _get_object_size(client, export_bucket_name, object_key)
         if existing_object_size is None or existing_object_size == 0:
-            logger.info(f"Skipping block {block_number} - no data")
+            logger.info(f"Skipping {table} for block {block_number} - no data")
             return
 
     mev_summary_json_fileobj = BytesIteratorIO(
@@ -56,6 +66,15 @@ def export_block(inspect_db_session, block_number: int) -> None:
     logger.info(f"Exported to {object_key}")
 
 
+def _get_export_statement(table: str) -> str:
+    return f"""
+                SELECT to_json({table})
+                FROM {table}
+                WHERE
+                    block_number = :block_number
+            """
+
+
 def _get_object_size(client, bucket: str, key: str) -> Optional[int]:
     response = client.list_objects_v2(
         Bucket=bucket,

pyproject.toml

@@ -43,6 +43,7 @@ fetch-all-prices = 'cli:fetch_all_prices'
 fetch-range = 'cli:fetch_range'
 s3-export = 'cli:s3_export'
 enqueue-s3-export = 'cli:enqueue_s3_export'
+enqueue-many-s3-exports = 'cli:enqueue_many_s3_exports'
 
 [tool.black]
 exclude = '''

worker.py

@@ -11,7 +11,9 @@ from mev_inspect.queue.middleware import (
     InspectorMiddleware,
 )
 from mev_inspect.queue.tasks import (
+    HIGH_PRIORITY,
     HIGH_PRIORITY_QUEUE,
+    LOW_PRIORITY,
     LOW_PRIORITY_QUEUE,
     export_block_task,
     inspect_many_blocks_task,
@@ -25,5 +27,9 @@ broker.add_middleware(AsyncMiddleware())
 broker.add_middleware(InspectorMiddleware(os.environ["RPC_URL"]))
 dramatiq.set_broker(broker)
 
-dramatiq.actor(inspect_many_blocks_task, queue_name=HIGH_PRIORITY_QUEUE)
+dramatiq.actor(
-dramatiq.actor(export_block_task, queue_name=LOW_PRIORITY_QUEUE)
+    inspect_many_blocks_task, queue_name=LOW_PRIORITY_QUEUE, priority=LOW_PRIORITY
+)
+dramatiq.actor(
+    export_block_task, queue_name=HIGH_PRIORITY_QUEUE, priority=HIGH_PRIORITY
+)