feat: adds persistence layer for app state and job results [SBK-363]

README.md

@@ -43,18 +43,18 @@ python3 setup.py install
 
 ## Quick Usage
 
-Checkout [the example](./example.py) to see how to use the library.
+Checkout [the example](./examples/memory/main.py) to see how to use the library.
 
 To run your bot against a live network, this SDK includes a simple runner you can use via:
 
 ```sh
-$ silverback run "example:app" --network :mainnet:alchemy
+$ silverback run "examples.memory:app" --network :mainnet:alchemy
 ```
 
 ## Docker Usage
 
 ```sh
-$ docker run --volume $PWD:/home/harambe/project --volume ~/.tokenlists:/home/harambe/.tokenlists apeworx/silverback:latest run "example:app" --network :mainnet:alchemy
+$ docker run --volume $PWD:/home/harambe/project --volume ~/.tokenlists:/home/harambe/.tokenlists apeworx/silverback:latest run "examples.memory:app" --network :mainnet:alchemy
 ```
 
 ## Development

docs/userguides/development.md

@@ -61,19 +61,57 @@ Any errors you raise during this function will get captured by the client, and r
 
 ## Startup and Shutdown
 
-If you have heavier resources you want to load during startup, or otherwise perform some data collection prior to starting the bot, you can add a startup function like so:
+### Worker Events
+
+If you have heavier resources you want to load during startup, or want to initialize things like database connections, you can add a worker startup function like so:
 
 ```py
-@app.on_startup()
+@app.on_worker_startup()
 def handle_on_worker_startup(state):
+    # Connect to DB, set initial state, etc
+    ...
+
+@app.on_worker_shutdown()
+def handle_on_worker_shutdown(state):
+    # cleanup resources, close connections cleanly, etc
     ...
 ```
 
 This function comes a parameter `state` that you can use for storing the results of your startup computation or resources that you have provisioned.
-It's import to note that this is useful for ensuring that your workers (of which there can be multiple) have the resources necessary to properly handle any updates you want to make in your handler functions, such as connecting to the Telegram API, an SQL or NoSQL database connection, or something else.
-The `state` variable is also useful as this gets made available to each handler method so other stateful quantities can be maintained for other uses.
 
-TODO: Add more information about `state`
+It's import to note that this is useful for ensuring that your workers (of which there can be multiple) have the resources necessary to properly handle any updates you want to make in your handler functions, such as connecting to the Telegram API, an SQL or NoSQL database connection, or something else.  **This function will run on every worker process**.
+
+#### Worker State
+
+The `state` variable is also useful as this can be made available to each handler method so other stateful quantities can be maintained for other uses.  Each distributed worker has its own instance of state.
+
+To access the state from a handler, you must annotate `context` as a dependency like so:
+
+```py
+from typing import Annotated
+from taskiq import Context, TaskiqDepends
+
+@app.on_(chain.blocks)
+def block_handler(block, context: Annotated[Context, TaskiqDepends()]):
+    # Access state via context.state
+    ...
+```
+
+### Application Events
+
+You can also add an application startup and shutdown handler that will be **executed once upon every application startup**.  This may be useful for things like processing historical events since the application was shutdown or other one-time actions to perform at startup.
+
+```py
+@app.on_startup()
+def handle_on_startup(state):
+    # Process missed events, etc
+    ...
+
+@app.on_shutdown()
+def handle_on_startup(state):
+    # Record final state, etc
+    ...
+```
 
 ## Running your Application
 
@@ -101,6 +139,34 @@ If you configure your application to use a signer, and that signer signs anythin
 Always test your applications throughly before deploying.
 ```
 
+### Distributed Execution
+
+Using only the `silverback run ...` command in a defualt configuration executes everything in one process and the job queue is completely in-memory with a shared state.  In some high volume environments, you may want to deploy your Silverback application in a distributed configuration.
+
+The primary components are the client and workers.  The client handles Silverback events (blocks and contract event logs) and creates jobs for the workers to process in an asynchronous manner.
+
+For this to work, you must configure a [TaskIQ broker](https://taskiq-python.github.io/guide/architecture-overview.html#broker) capable of distributed processing.  For instance, with [`taskiq_redis`](https://github.com/taskiq-python/taskiq-redis) you could do something like this for the client:
+
+```bash
+export SILVERBACK_BROKER_CLASS="taskiq_redis:ListQueueBroker"
+export SILVERBACK_BROKER_URI="redis://127.0.0.1:6379"
+
+silverback run "examples.redis.main:app" \
+    --network :mainnet:alchemy \
+    --runner "silverback.runner:WebsocketRunner"
+```
+
+And then the worker process with 2 worker subprocesses:
+
+```bash
+export SILVERBACK_BROKER_CLASS="taskiq_redis:ListQueueBroker"
+export SILVERBACK_BROKER_URI="redis://127.0.0.1:6379"
+
+silverback worker -w2 "examples.redis.main:app"
+```
+
+This will run one client and 2 workers and all queue data will be go through Redis.
+
 ## Testing your Application
 
 TODO: Add backtesting mode w/ `silverback test`

examples/memory/__init__.py

@@ -0,0 +1,3 @@
+from .main import app
+
+__all__ = ["app"]

examples/memory/main.py

@@ -0,0 +1,74 @@
+from typing import Annotated
+
+from ape import chain
+from ape.api import BlockAPI
+from ape.types import ContractLog
+from ape_tokens import tokens  # type: ignore[import]
+from taskiq import Context, TaskiqDepends, TaskiqState
+
+from silverback import CircuitBreaker, SilverbackApp, SilverbackStartupState
+
+# Do this to initialize your app
+app = SilverbackApp()
+
+# NOTE: Don't do any networking until after initializing app
+USDC = tokens["USDC"]
+YFI = tokens["YFI"]
+
+
+@app.on_startup()
+def app_startup(startup_state: SilverbackStartupState):
+    return {"message": "Starting...", "block_number": startup_state.last_block_seen}
+
+
+@app.on_client_startup()
+def client_startup(state):
+    return {"message": "Client started."}
+
+
+# Can handle some initialization on startup, like models or network connections
+@app.on_worker_startup()
+def worker_startup(state: TaskiqState):
+    state.block_count = 0
+    # state.db = MyDB()
+    return {"message": "Worker started."}
+
+
+# This is how we trigger off of new blocks
+@app.on_(chain.blocks)
+# context must be a type annotated kwarg to be provided to the task
+def exec_block(block: BlockAPI, context: Annotated[Context, TaskiqDepends()]):
+    context.state.block_count += 1
+    return len(block.transactions)
+
+
+# This is how we trigger off of events
+# Set new_block_timeout to adjust the expected block time.
+@app.on_(USDC.Transfer, start_block=18588777, new_block_timeout=25)
+# NOTE: Typing isn't required
+def exec_event1(log):
+    if log.log_index % 7 == 3:
+        # If you ever want the app to shutdown under some scenario, call this exception
+        raise CircuitBreaker("Oopsie!")
+    return {"amount": log.amount}
+
+
+@app.on_(YFI.Approval)
+# Any handler function can be async too
+async def exec_event2(log: ContractLog):
+    return log.amount
+
+
+# Just in case you need to release some resources or something
+@app.on_worker_shutdown()
+def worker_shutdown(state):
+    return {
+        "message": f"Worker stopped after handling {state.block_count} blocks.",
+        "block_count": state.block_count,
+    }
+
+
+# A final job to execute on Silverback shutdown
+@app.on_shutdown()
+def app_shutdown(state):
+    return {"message": "Stopping..."}

examples/redis/__init__.py

@@ -0,0 +1,3 @@
+from .main import app
+
+__all__ = ["app"]

examples/redis/main.py

@@ -0,0 +1,75 @@
+from typing import Annotated
+
+from ape import chain
+from ape.api import BlockAPI
+from ape.types import ContractLog
+from ape_tokens import tokens  # type: ignore[import]
+from taskiq import Context, TaskiqDepends, TaskiqState
+
+from silverback import CircuitBreaker, SilverbackApp, SilverbackStartupState
+
+# Do this to initialize your app
+app = SilverbackApp()
+
+# NOTE: Don't do any networking until after initializing app
+USDC = tokens["USDC"]
+YFI = tokens["YFI"]
+
+
+# Can handle some stuff on startup, like loading a heavy model or something
+@app.on_startup()
+def app_startup(startup_state: SilverbackStartupState):
+    return {"message": "Starting...", "block_number": startup_state.last_block_seen}
+
+
+@app.on_client_startup()
+def client_startup(state):
+    return {"message": "Client started."}
+
+
+@app.on_worker_startup()
+def worker_startup(state: TaskiqState):
+    state.block_count = 0
+    # state.db = MyDB()
+    return {"message": "Worker started."}
+
+
+# This is how we trigger off of new blocks
+@app.on_(chain.blocks)
+# context must be a type annotated kwarg to be provided to the task
+def exec_block(block: BlockAPI, context: Annotated[Context, TaskiqDepends()]):
+    context.state.block_count += 1
+    return len(block.transactions)
+
+
+# This is how we trigger off of events
+# Set new_block_timeout to adjust the expected block time.
+@app.on_(USDC.Transfer, start_block=18588777, new_block_timeout=25)
+# NOTE: Typing isn't required
+def exec_event1(log):
+    if log.log_index % 7 == 3:
+        # If you ever want the app to shutdown under some scenario, call this exception
+        raise CircuitBreaker("Oopsie!")
+    return {"amount": log.amount}
+
+
+@app.on_(YFI.Approval)
+# Any handler function can be async too
+async def exec_event2(log: ContractLog):
+    return log.amount
+
+
+# Just in case you need to release some resources or something
+@app.on_worker_shutdown()
+def worker_shutdown(state):
+    block_count = state.block_count if hasattr(state, "block_count") else 0
+    return {
+        "message": f"Worker stopped after handling {block_count} blocks.",
+        "block_count": block_count,
+    }
+
+
+# A final job to execute on Silverback shutdown
+@app.on_shutdown()
+def app_shutdown(state):
+    return {"message": "Stopping..."}

silverback/__init__.py

@@ -1,8 +1,10 @@
 from .application import SilverbackApp
 from .exceptions import CircuitBreaker, SilverbackException
+from .types import SilverbackStartupState
 
 __all__ = [
     "CircuitBreaker",
     "SilverbackApp",
     "SilverbackException",
+    "SilverbackStartupState",
 ]

silverback/_cli.py

@@ -1,8 +1,12 @@
 import asyncio
 import os
+from concurrent.futures import ThreadPoolExecutor
 
 import click
 from ape.cli import AccountAliasPromptChoice, ape_cli_context, network_option, verbosity_option
+from taskiq import AsyncBroker
+from taskiq.cli.worker.run import shutdown_broker
+from taskiq.receiver import Receiver
 
 from silverback._importer import import_from_string
 from silverback.runner import PollingRunner
@@ -40,7 +44,27 @@ def _network_callback(ctx, param, val):
     return val
 
 
-@cli.command()
+async def run_worker(broker: AsyncBroker, worker_count=2, shutdown_timeout=90):
+    try:
+        tasks = []
+        with ThreadPoolExecutor(max_workers=worker_count) as pool:
+            for _ in range(worker_count):
+                receiver = Receiver(
+                    broker=broker,
+                    executor=pool,
+                    validate_params=True,
+                    max_async_tasks=1,
+                    max_prefetch=0,
+                )
+                broker.is_worker_process = True
+                tasks.append(receiver.listen())
+
+            await asyncio.gather(*tasks)
+    finally:
+        await shutdown_broker(broker, shutdown_timeout)
+
+
+@cli.command(help="Run Silverback application client")
 @ape_cli_context()
 @verbosity_option()
 @network_option(default=None, callback=_network_callback)
@@ -57,3 +81,17 @@ def run(cli_ctx, network, account, runner, max_exceptions, path):
         app = import_from_string(path)
         runner = runner(app, max_exceptions=max_exceptions)
         asyncio.run(runner.run())
+
+
+@cli.command(help="Run Silverback application task workers")
+@ape_cli_context()
+@verbosity_option()
+@network_option(default=None, callback=_network_callback)
+@click.option("--account", type=AccountAliasPromptChoice(), callback=_account_callback)
+@click.option("-w", "--workers", type=int, default=2)
+@click.option("-x", "--max-exceptions", type=int, default=3)
+@click.option("-s", "--shutdown_timeout", type=int, default=90)
+@click.argument("path")
+def worker(cli_ctx, network, account, workers, max_exceptions, shutdown_timeout, path):
+    app = import_from_string(path)
+    asyncio.run(run_worker(app.broker, worker_count=workers, shutdown_timeout=shutdown_timeout))