Feature/async multi processor

cli/__init__.py

@@ -17,7 +17,7 @@ def get_version():
 
 
 @click.group()
-@click.version_option(version=get_version())
+# @click.version_option(version=get_version())
 @click.pass_context
 def cli(ctx):
     pass

cli/reorg.py

@@ -11,10 +11,11 @@
 from indexer.utils.exception_recorder import ExceptionRecorder
 from indexer.utils.logging_utils import configure_logging, configure_signals
 from indexer.utils.provider import get_provider_from_uri
+from indexer.utils.reorg import check_reorg
 from indexer.utils.rpc_utils import pick_random_provider_uri
 from indexer.utils.thread_local_proxy import ThreadLocalProxy
 
-exception_recorder = ExceptionRecorder()
+# exception_recorder = ExceptionRecorder()
 
 
 @click.command(context_settings=dict(help_option_names=["-h", "--help"]))
@@ -46,18 +47,6 @@
     envvar="POSTGRES_URL",
     help="The required postgres connection url." "e.g. postgresql+psycopg2://postgres:admin@127.0.0.1:5432/ethereum",
 )
-@click.option(
-    "-v",
-    "--db-version",
-    default="head",
-    show_default=True,
-    type=str,
-    envvar="DB_VERSION",
-    help="The database version to initialize the database. using the alembic script's revision ID to "
-    "specify a version."
-    " e.g. head, indicates the latest version."
-    "or base, indicates the empty database without any table.",
-)
 @click.option(
     "-b",
     "--batch-size",
@@ -77,6 +66,7 @@
 )
 @click.option(
     "--block-number",
+    default=None,
     show_default=True,
     type=int,
     envvar="BLOCK_NUMBER",
@@ -85,12 +75,20 @@
 @click.option(
     "-r",
     "--ranges",
-    default=1000,
+    default=10,
     show_default=True,
     type=int,
     envvar="RANGES",
     help="Specify the range limit for data fixing.",
 )
+@click.option(
+    "--check-ranges",
+    default=None,
+    show_default=True,
+    type=int,
+    envvar="CHECK_RANGES",
+    help="Specify the range for block continuous checking.",
+)
 @click.option(
     "--log-file",
     default=None,
@@ -109,14 +107,6 @@
     envvar="MULTI_CALL_ENABLE",
 )
 @click.option("--cache", default=None, show_default=True, type=str, envvar="CACHE", help="Cache")
-@click.option(
-    "--auto-upgrade-db",
-    default=True,
-    show_default=True,
-    type=bool,
-    envvar="AUTO_UPGRADE_DB",
-    help="Whether to automatically run database migration scripts to update the database to the latest version.",
-)
 @click.option(
     "--log-level",
     default="INFO",
@@ -131,14 +121,13 @@ def reorg(
     postgres_url,
     block_number,
     ranges,
+    check_ranges,
     batch_size,
     debug_batch_size,
-    db_version="head",
     multicall=True,
     log_file=None,
     cache=None,
     config_file=None,
-    auto_upgrade_db=True,
     log_level="INFO",
 ):
     configure_logging(log_level=log_level, log_file=log_file)
@@ -151,9 +140,9 @@ def reorg(
 
     # build postgresql service
     if postgres_url:
-        service = PostgreSQLService(postgres_url, db_version=db_version, init_schema=auto_upgrade_db)
-        config = {"db_service": service}
-        exception_recorder.init_pg_service(service)
+        service = PostgreSQLService(postgres_url)
+        config = {"db_service": postgres_url}
+        # exception_recorder.init_pg_service(service)
     else:
         logging.error("No postgres url provided. Exception recorder will not be useful.")
         exit(1)
@@ -177,9 +166,9 @@ def reorg(
     output_types = list(generate_output_types(entity_types))
 
     job_scheduler = ReorgScheduler(
-        batch_web3_provider=ThreadLocalProxy(lambda: get_provider_from_uri(provider_uri, batch=True)),
-        batch_web3_debug_provider=ThreadLocalProxy(lambda: get_provider_from_uri(debug_provider_uri, batch=True)),
-        item_exporters=PostgresItemExporter(config["db_service"]),
+        web3_provider_uri=provider_uri,
+        web3_debug_provider_uri=debug_provider_uri,
+        item_exporters=PostgresItemExporter(service_url=postgres_url),
         batch_size=batch_size,
         debug_batch_size=debug_batch_size,
         required_output_types=output_types,
@@ -192,23 +181,29 @@ def reorg(
         batch_web3_provider=ThreadLocalProxy(lambda: get_provider_from_uri(provider_uri, batch=False)),
         job_scheduler=job_scheduler,
         ranges=ranges,
-        config=config,
+        service=service,
     )
 
-    job = None
+    current_block = controller.get_current_block_number()
+    if check_ranges:
+        check_begin = current_block - check_ranges
+        check_reorg(service, check_begin)
+    else:
+        check_reorg(service)
+
     while True:
-        if job:
-            controller.action(
-                job_id=job.job_id,
-                block_number=job.last_fixed_block_number - 1,
-                remains=job.remain_process,
-            )
-        else:
+        if block_number:
             controller.action(block_number=block_number)
-
-        job = controller.wake_up_next_job()
-        if job:
-            logging.info(f"Waking up uncompleted job: {job.job_id}.")
         else:
-            logging.info("No more uncompleted jobs to wake-up, reorg process will terminate.")
-            break
+            job = controller.wake_up_next_job()
+            if job:
+                logging.info(f"Waking up uncompleted job: {job.job_id}.")
+
+                controller.action(
+                    job_id=job.job_id,
+                    block_number=job.last_fixed_block_number - 1,
+                    remains=job.remain_process,
+                )
+            else:
+                logging.info("No more uncompleted jobs to wake-up, reorg process will terminate.")
+                break

cli/stream.py

@@ -183,13 +183,22 @@ def wrapper(*args, **kwargs):
     envvar="BLOCK_BATCH_SIZE",
     help="How many blocks to batch in single sync round",
 )
+@click.option(
+    "-P",
+    "--max-processors",
+    default=1,
+    show_default=True,
+    type=int,
+    help="How many sync round to concurrently execute.",
+    envvar="MAX_PROCESSOR",
+)
 @click.option(
     "-w",
     "--max-workers",
     default=5,
     show_default=True,
     type=int,
-    help="The number of workers",
+    help="The number of workers during a request to rpc.",
     envvar="MAX_WORKERS",
 )
 @click.option(
@@ -325,6 +334,7 @@ def stream(
     batch_size=10,
     debug_batch_size=1,
     block_batch_size=1,
+    max_processors=1,
     max_workers=5,
     log_file=None,
     pid_file=None,
@@ -346,6 +356,12 @@ def stream(
     debug_provider_uri = pick_random_provider_uri(debug_provider_uri)
     logging.getLogger("ROOT").info("Using provider " + provider_uri)
     logging.getLogger("ROOT").info("Using debug provider " + debug_provider_uri)
+    logging.getLogger("ROOT").info(
+        f"Indexer will run in {'multi' if max_processors > 1 else 'single'}-process mode "
+        f"{'with ' if max_processors > 1 else ''}"
+        f"{max_processors if max_processors > 1 else ''}"
+        f" {'processor' if max_processors > 1 else ''} "
+    )
 
     # parameter logic checking
     if source_path:
@@ -361,7 +377,7 @@ def stream(
 
     if postgres_url:
         service = PostgreSQLService(postgres_url, db_version=db_version, init_schema=auto_upgrade_db)
-        config["db_service"] = service
+        config["db_service"] = postgres_url
         exception_recorder.init_pg_service(service)
     else:
         logging.getLogger("ROOT").warning("No postgres url provided. Exception recorder will not be useful.")
@@ -397,14 +413,14 @@ def stream(
     output_types = list(
         set(generate_dataclass_type_list_from_parameter(output_types, "output") + output_types_by_entity_type)
     )
+    output_types.sort(key=lambda x: x.type())
 
     if source_path and source_path.startswith("postgresql://"):
         source_types = generate_dataclass_type_list_from_parameter(source_types, "source")
 
     job_scheduler = JobScheduler(
-        batch_web3_provider=ThreadLocalProxy(lambda: get_provider_from_uri(provider_uri, batch=True)),
-        batch_web3_debug_provider=ThreadLocalProxy(lambda: get_provider_from_uri(debug_provider_uri, batch=True)),
-        item_exporters=create_item_exporters(output, config),
+        web3_provider_uri=provider_uri,
+        web3_debug_provider_uri=debug_provider_uri,
         batch_size=batch_size,
         debug_batch_size=debug_batch_size,
         max_workers=max_workers,
@@ -414,12 +430,16 @@ def stream(
         cache=cache,
         auto_reorg=auto_reorg,
         multicall=multicall,
+        multiprocess=max_processors > 1,
         force_filter_mode=force_filter_mode,
     )
 
     controller = StreamController(
         batch_web3_provider=ThreadLocalProxy(lambda: get_provider_from_uri(provider_uri, batch=False)),
-        job_scheduler=job_scheduler,
+        max_processors=max_processors,
+        scheduled_jobs=job_scheduler.get_scheduled_jobs(),
+        item_exporters=create_item_exporters(output, config),
+        required_output_types=output_types,
         sync_recorder=create_recorder(sync_recorder, config),
         limit_reader=create_limit_reader(
             source_path, ThreadLocalProxy(lambda: get_provider_from_uri(provider_uri, batch=False))
@@ -435,3 +455,5 @@ def stream(
         period_seconds=period_seconds,
         pid_file=pid_file,
     )
+
+    controller.shutdown()

common/models/failure_records.py

@@ -0,0 +1,16 @@
+from sqlalchemy import Column
+from sqlalchemy.dialects.postgresql import BIGINT, JSON, TIMESTAMP, VARCHAR
+
+from common.models import HemeraModel
+
+
+class FailureRecords(HemeraModel):
+    __tablename__ = "failure_records"
+    record_id = Column(BIGINT, primary_key=True, autoincrement=True)
+    mission_sign = Column(VARCHAR)
+    output_types = Column(VARCHAR)
+    start_block_number = Column(BIGINT)
+    end_block_number = Column(BIGINT)
+    exception_stage = Column(VARCHAR)
+    exception = Column(JSON)
+    crash_time = Column(TIMESTAMP)