diff --git a/.tool-versions b/.tool-versions index a50ce8eb8..ea59406bb 100644 --- a/.tool-versions +++ b/.tool-versions @@ -1,10 +1,10 @@ -circleci 0.1.31425 -golang 1.24.5 +circleci 0.1.31425 +golang 1.26.3 golangci-lint 1.64.5 -nodejs 18.17.0 -postgres 14.2 -pre-commit 4.1.0 -protoc 29.3 -shfmt 3.10.0 -solidity 0.8.9 -sqlc 1.28.0 +nodejs 18.17.0 +postgres 18.3 +pre-commit 4.1.0 +protoc 29.3 +shfmt 3.10.0 +solidity 0.8.9 +sqlc latest diff --git a/mise-test-setup/mise-tasks/add-initial-keyper-set b/mise-test-setup/mise-tasks/add-initial-keyper-set index b7b84ae8f..9d7ec0997 100755 --- a/mise-test-setup/mise-tasks/add-initial-keyper-set +++ b/mise-test-setup/mise-tasks/add-initial-keyper-set @@ -1,6 +1,8 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Add the initial keyper set on-chain" #MISE depends=["gen-keyper-configs"] +# fmt: on import json import os @@ -8,7 +10,6 @@ from pathlib import Path import utils - deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) data_dir = Path(os.environ["DATA_DIR"]) run_path = ( diff --git a/mise-test-setup/mise-tasks/add-keyper-set b/mise-test-setup/mise-tasks/add-keyper-set index ec5991066..dd0ef2f18 100755 --- a/mise-test-setup/mise-tasks/add-keyper-set +++ b/mise-test-setup/mise-tasks/add-keyper-set @@ -1,8 +1,10 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Add a keyper set on-chain" #MISE depends=["gen-keyper-configs"] #USAGE flag "--indices " help="Comma-separated keyper indices" #USAGE flag "--threshold " help="Keyper set threshold" +# fmt: on import json import os @@ -10,7 +12,6 @@ from pathlib import Path import utils - deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) data_dir = Path(os.environ["DATA_DIR"]) indices = utils.parse_indices(os.environ["usage_indices"]) @@ -24,8 +25,12 @@ run_path = ( / "run-latest.json" ) deployment_run = json.loads(run_path.read_text()) -keyper_set_manager = utils.get_created_contract_address(deployment_run, "KeyperSetManager") -key_broadcast_contract = utils.get_created_contract_address(deployment_run, "KeyBroadcastContract") +keyper_set_manager = utils.get_created_contract_address( + deployment_run, "KeyperSetManager" +) +key_broadcast_contract = utils.get_created_contract_address( + deployment_run, "KeyBroadcastContract" +) keyper_addresses = ",".join( utils.keyper_address(data_dir / f"keyper-{index}.toml") for index in indices ) diff --git a/mise-test-setup/mise-tasks/build-image b/mise-test-setup/mise-tasks/build-image index 69bc95549..e1a91f675 100755 --- a/mise-test-setup/mise-tasks/build-image +++ b/mise-test-setup/mise-tasks/build-image @@ -1,5 +1,7 @@ #!/usr/bin/env bash +# fmt: off #MISE description="Build the rolling-shutter image" +# fmt: on docker build \ -f ../docker/build-src/rolling-shutter/Dockerfile \ diff --git a/mise-test-setup/mise-tasks/clean b/mise-test-setup/mise-tasks/clean index d15c60ada..04efec2f5 100755 --- a/mise-test-setup/mise-tasks/clean +++ b/mise-test-setup/mise-tasks/clean @@ -1,6 +1,8 @@ #!/usr/bin/env bash +# fmt: off #MISE description="Delete all generated files" #MISE depends=["down"] +# fmt: on rm -rf "${DATA_DIR}" rm -rf "${GENERATED_DIR}" diff --git a/mise-test-setup/mise-tasks/deploy b/mise-test-setup/mise-tasks/deploy index 3d117b2ba..d95eaa58a 100755 --- a/mise-test-setup/mise-tasks/deploy +++ b/mise-test-setup/mise-tasks/deploy @@ -1,12 +1,17 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Deploy contracts if not already deployed" #MISE depends=["up-ethereum"] +# fmt: on -import os import json +import os from pathlib import Path + import utils +RPC = "http://127.0.0.1:8545" + REQUIRED_CONTRACTS: dict[str, list[str]] = { "gnosis": [ "KeyBroadcastContract", @@ -30,28 +35,80 @@ def deployment_is_complete(path: Path, deployment_type: str) -> bool: return True +def deploy_test_helper(data_dir: Path) -> None: + addr_path = data_dir / "test-helper-addr.json" + if addr_path.exists(): + addr = json.loads(addr_path.read_text())["address"] + print(f"EventTriggerTestHelper already deployed at {addr}") + return + + artifact = json.loads( + ( + data_dir + / "contracts" + / "out" + / "EventTriggerTestHelper.sol" + / "EventTriggerTestHelper.json" + ).read_text() + ) + bytecode_hex = artifact["bytecode"]["object"].removeprefix("0x") + + result = utils.run( + [ + "docker", + "compose", + "exec", + "-T", + "ethereum", + "cast", + "send", + "--json", + "--rpc-url", + RPC, + "--private-key", + os.environ["DEPLOY_KEY"], + "--create", + bytecode_hex, + ], + capture_output=True, + ) + receipt = json.loads(result.stdout.strip()) + addr = receipt.get("contractAddress") or receipt.get("creates") + if not addr: + raise SystemExit( + f"Could not find contractAddress in receipt:\n{json.dumps(receipt, indent=2)}" + ) + + addr_path.write_text(json.dumps({"address": addr})) + print(f"EventTriggerTestHelper deployed at {addr}") + + deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) deploy_script = utils.DEPLOYMENT_SCRIPTS[deployment_type] data_dir = Path(os.environ["DATA_DIR"]) chain_id = os.environ["ETHEREUM_CHAIN_ID"] -run_path = data_dir / "contracts" / "broadcast" / deploy_script / chain_id / "run-latest.json" +run_path = ( + data_dir / "contracts" / "broadcast" / deploy_script / chain_id / "run-latest.json" +) if deployment_is_complete(run_path, deployment_type): print(f"Deployment already present at {run_path}; skipping") - raise SystemExit(0) - -utils.run( - [ - "docker", - "compose", - "run", - "--rm", - "--env", - f"DEPLOY_KEY={os.environ['DEPLOY_KEY']}", - "contracts", - "script", - f"/app/script/{deploy_script}:Deploy", - "--broadcast", - ] -) +else: + utils.run( + [ + "docker", + "compose", + "run", + "--rm", + "--env", + f"DEPLOY_KEY={os.environ['DEPLOY_KEY']}", + "contracts", + "script", + f"/app/script/{deploy_script}:Deploy", + "--broadcast", + ] + ) + +if deployment_type == "service": + deploy_test_helper(data_dir) diff --git a/mise-test-setup/mise-tasks/down b/mise-test-setup/mise-tasks/down index 03da7cb8b..b7ae230ca 100755 --- a/mise-test-setup/mise-tasks/down +++ b/mise-test-setup/mise-tasks/down @@ -1,5 +1,7 @@ #!/usr/bin/env bash +# fmt: off #MISE description="Stop the docker3 stack" +# fmt: on if [ -f "${GENERATED_DIR}/compose.yml" ]; then docker compose down || true diff --git a/mise-test-setup/mise-tasks/gen-compose b/mise-test-setup/mise-tasks/gen-compose index 6c3f3078a..71ff71e88 100755 --- a/mise-test-setup/mise-tasks/gen-compose +++ b/mise-test-setup/mise-tasks/gen-compose @@ -1,15 +1,16 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Render docker compose files based on DEPLOYMENT_TYPE" # /// script # dependencies = ["jinja2"] # /// +# fmt: on import os from pathlib import Path -from jinja2 import Environment, FileSystemLoader import utils - +from jinja2 import Environment, FileSystemLoader deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) generated_dir = Path(os.environ["GENERATED_DIR"]) diff --git a/mise-test-setup/mise-tasks/gen-keyper-configs b/mise-test-setup/mise-tasks/gen-keyper-configs index 27961297e..bfcf92b91 100755 --- a/mise-test-setup/mise-tasks/gen-keyper-configs +++ b/mise-test-setup/mise-tasks/gen-keyper-configs @@ -1,12 +1,14 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Ensure config files for all keypers" #MISE depends=["deploy"] # /// script # dependencies = ["tomlkit"] # /// +# fmt: on -import os import json +import os from pathlib import Path import tomlkit @@ -52,22 +54,73 @@ for index in range(int(os.environ["NUM_KEYPERS"])): ) document = tomlkit.parse(config_path.read_text()) - utils.set_toml_path(document, ["DatabaseURL"], f"postgresql://postgres@db:5432/keyper-{index}") - utils.set_toml_path(document, ["Shuttermint", "ShuttermintURL"], f"http://chain-{index}:{os.environ['SHUTTERMINT_RPC_PORT']}") - utils.set_toml_path(document, ["Shuttermint", "DKGPhaseLength"], int(os.environ["SHUTTERMINT_DKG_PHASE_LENGTH"])) - utils.set_toml_path(document, ["Shuttermint", "StartBlockDelta"], int(os.environ["SHUTTERMINT_START_BLOCK_DELTA"])) - utils.set_toml_path(document, ["P2P", "CustomBootstrapAddresses"], bootstrap_addresses) - utils.set_toml_path(document, ["Chain", "Contracts", "KeyBroadcastContract"], utils.get_created_contract_address(deployment_run, "KeyBroadcastContract")) - utils.set_toml_path(document, ["Chain", "Contracts", "KeyperSetManager"], utils.get_created_contract_address(deployment_run, "KeyperSetManager")) + utils.set_toml_path( + document, ["DatabaseURL"], f"postgresql://postgres@db:5432/keyper-{index}" + ) + utils.set_toml_path( + document, + ["Shuttermint", "ShuttermintURL"], + f"http://chain-{index}:{os.environ['SHUTTERMINT_RPC_PORT']}", + ) + utils.set_toml_path( + document, + ["Shuttermint", "DKGPhaseLength"], + int(os.environ["SHUTTERMINT_DKG_PHASE_LENGTH"]), + ) + utils.set_toml_path( + document, + ["Shuttermint", "StartBlockDelta"], + int(os.environ["SHUTTERMINT_START_BLOCK_DELTA"]), + ) + utils.set_toml_path( + document, ["P2P", "CustomBootstrapAddresses"], bootstrap_addresses + ) + utils.set_toml_path( + document, + ["Chain", "Contracts", "KeyBroadcastContract"], + utils.get_created_contract_address(deployment_run, "KeyBroadcastContract"), + ) + utils.set_toml_path( + document, + ["Chain", "Contracts", "KeyperSetManager"], + utils.get_created_contract_address(deployment_run, "KeyperSetManager"), + ) if deployment_type == "gnosis": - utils.set_toml_path(document, ["Gnosis", "Node", "EthereumURL"], "ws://ethereum:8545") - utils.set_toml_path(document, ["Chain", "Contracts", "Sequencer"], utils.get_created_contract_address(deployment_run, "Sequencer")) - utils.set_toml_path(document, ["Chain", "Contracts", "ValidatorRegistry"], utils.get_created_contract_address(deployment_run, "ValidatorRegistry")) + utils.set_toml_path( + document, ["Gnosis", "Node", "EthereumURL"], "ws://ethereum:8545" + ) + utils.set_toml_path( + document, + ["Chain", "Contracts", "Sequencer"], + utils.get_created_contract_address(deployment_run, "Sequencer"), + ) + utils.set_toml_path( + document, + ["Chain", "Contracts", "ValidatorRegistry"], + utils.get_created_contract_address(deployment_run, "ValidatorRegistry"), + ) elif deployment_type == "service": - utils.set_toml_path(document, ["Chain", "Node", "EthereumURL"], "ws://ethereum:8545") - utils.set_toml_path(document, ["Chain", "Contracts", "ShutterRegistry"], utils.get_created_contract_address(deployment_run, "ShutterRegistry")) + utils.set_toml_path( + document, ["Chain", "Node", "EthereumURL"], "ws://ethereum:8545" + ) + utils.set_toml_path( + document, + ["Chain", "Contracts", "ShutterRegistry"], + utils.get_created_contract_address(deployment_run, "ShutterRegistry"), + ) + event_trigger_registry = utils.get_uups_proxy_address( + deployment_run, "ShutterEventTriggerRegistryV1" + ) + if event_trigger_registry: + utils.set_toml_path( + document, + ["Chain", "Contracts", "ShutterEventTriggerRegistry"], + event_trigger_registry, + ) else: - raise SystemExit(f"Unsupported DEPLOYMENT_TYPE for config generation: {deployment_type}") + raise SystemExit( + f"Unsupported DEPLOYMENT_TYPE for config generation: {deployment_type}" + ) config_path.write_text(tomlkit.dumps(document)) diff --git a/mise-test-setup/mise-tasks/init-chain-nodes b/mise-test-setup/mise-tasks/init-chain-nodes index 61d42f7cd..ffc290445 100755 --- a/mise-test-setup/mise-tasks/init-chain-nodes +++ b/mise-test-setup/mise-tasks/init-chain-nodes @@ -1,9 +1,11 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Initialize shuttermint chain nodes for all keypers" #MISE depends=["init-chain-seed", "gen-keyper-configs"] # /// script # dependencies = ["tomlkit"] # /// +# fmt: on import os from pathlib import Path @@ -11,14 +13,15 @@ from pathlib import Path import tomlkit import utils - data_dir = Path(os.environ["DATA_DIR"]) seed_node_id_path = data_dir / "chain-seed" / "config" / "node_key.json.id" seed_node_id = seed_node_id_path.read_text().strip() for index in range(int(os.environ["NUM_KEYPERS"])): config_path = data_dir / f"chain-{index}" / "config" / "config.toml" - validator_pubkey_path = data_dir / f"chain-{index}" / "config" / "priv_validator_pubkey.hex" + validator_pubkey_path = ( + data_dir / f"chain-{index}" / "config" / "priv_validator_pubkey.hex" + ) keyper_config_path = data_dir / f"keyper-{index}.toml" if not config_path.exists(): @@ -56,5 +59,7 @@ for index in range(int(os.environ["NUM_KEYPERS"])): validator_pubkey = validator_pubkey_path.read_text().strip() keyper_document = tomlkit.parse(keyper_config_path.read_text()) - utils.set_toml_path(keyper_document, ["Shuttermint", "ValidatorPublicKey"], validator_pubkey) + utils.set_toml_path( + keyper_document, ["Shuttermint", "ValidatorPublicKey"], validator_pubkey + ) keyper_config_path.write_text(tomlkit.dumps(keyper_document)) diff --git a/mise-test-setup/mise-tasks/init-chain-seed b/mise-test-setup/mise-tasks/init-chain-seed index 27cf88486..60a827ca4 100755 --- a/mise-test-setup/mise-tasks/init-chain-seed +++ b/mise-test-setup/mise-tasks/init-chain-seed @@ -1,10 +1,12 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Initialize the shuttermint seed node" #MISE depends=["gen-compose"] #MISE outputs=["{{env.DATA_DIR}}/chain-seed/config/config.toml"] # /// script # dependencies = ["tomlkit"] # /// +# fmt: on import os from pathlib import Path @@ -12,7 +14,6 @@ from pathlib import Path import tomlkit import utils - data_dir = Path(os.environ["DATA_DIR"]) config_path = data_dir / "chain-seed" / "config" / "config.toml" diff --git a/mise-test-setup/mise-tasks/init-keyper-dbs b/mise-test-setup/mise-tasks/init-keyper-dbs index c38453a00..baf8ba23d 100755 --- a/mise-test-setup/mise-tasks/init-keyper-dbs +++ b/mise-test-setup/mise-tasks/init-keyper-dbs @@ -1,8 +1,11 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Initialize dbs for all keypers" #MISE depends=["up-db", "gen-keyper-configs"] +# fmt: on import os +from pathlib import Path import utils @@ -47,39 +50,101 @@ def public_table_count(database_name: str) -> int: return int(result.stdout.strip() or "0") +def db_ethereum_address(database_name: str) -> str: + result = utils.run( + [ + "docker", + "compose", + "exec", + "-T", + "db", + "psql", + "-U", + "postgres", + "-d", + database_name, + "-tAc", + "SELECT value FROM meta_inf WHERE key = 'ethereum address'", + ], + capture_output=True, + ) + return result.stdout.strip() + + +def drop_db(database_name: str) -> None: + utils.run( + [ + "docker", + "compose", + "exec", + "-T", + "db", + "psql", + "-U", + "postgres", + "-c", + f'DROP DATABASE IF EXISTS "{database_name}"', + ] + ) + + +def create_db(database_name: str) -> None: + utils.run( + [ + "docker", + "compose", + "exec", + "-T", + "db", + "createdb", + "-U", + "postgres", + database_name, + ] + ) + + +def init_db(index: int) -> None: + utils.run( + [ + "docker", + "compose", + "run", + "-T", + "--no-deps", + "--rm", + f"keyper-{index}", + utils.KEYPER_SUBCOMMANDS[deployment_type], + "initdb", + "--config", + "/config.toml", + ] + ) + + deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) +data_dir = Path(os.environ["DATA_DIR"]) for index in range(int(os.environ["NUM_KEYPERS"])): database_name = f"keyper-{index}" + config_path = data_dir / f"keyper-{index}.toml" if not db_exists(database_name): - utils.run( - [ - "docker", - "compose", - "exec", - "-T", - "db", - "createdb", - "-U", - "postgres", - database_name, - ] - ) + create_db(database_name) + init_db(index) + continue if public_table_count(database_name) == 0: - utils.run( - [ - "docker", - "compose", - "run", - "-T", - "--no-deps", - "--rm", - f"keyper-{index}", - utils.KEYPER_SUBCOMMANDS[deployment_type], - "initdb", - "--config", - "/config.toml", - ] + init_db(index) + continue + + # DB exists with tables. Re-initialize if the stored address doesn't match the config. + config_addr = utils.keyper_address(config_path) + db_addr = db_ethereum_address(database_name) + if db_addr.lower() != config_addr.lower(): + print( + f"keyper-{index}: DB address {db_addr} != config address {config_addr}; re-initializing" ) + drop_db(database_name) + create_db(database_name) + init_db(index) diff --git a/mise-test-setup/mise-tasks/patch-genesis b/mise-test-setup/mise-tasks/patch-genesis index 1bd6856dc..72af2d1b5 100755 --- a/mise-test-setup/mise-tasks/patch-genesis +++ b/mise-test-setup/mise-tasks/patch-genesis @@ -1,6 +1,8 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Patch and distribute the shuttermint genesis" #MISE depends=["init-chain-nodes"] +# fmt: on import json import os @@ -10,7 +12,9 @@ import utils data_dir = Path(os.environ["DATA_DIR"]) num_keypers = int(os.environ["NUM_KEYPERS"]) -initial_keyper_set_indices = utils.parse_indices(os.environ["INITIAL_KEYPER_SET_INDICES"]) +initial_keyper_set_indices = utils.parse_indices( + os.environ["INITIAL_KEYPER_SET_INDICES"] +) source_path = data_dir / "chain-seed" / "config" / "genesis.json" genesis = json.loads(source_path.read_text()) diff --git a/mise-test-setup/mise-tasks/submit-event-registration b/mise-test-setup/mise-tasks/submit-event-registration new file mode 100755 index 000000000..6cb685948 --- /dev/null +++ b/mise-test-setup/mise-tasks/submit-event-registration @@ -0,0 +1,174 @@ +#!/usr/bin/env -S uv run --script +# fmt: off +#MISE description="Register an event-based decryption trigger" +#MISE depends=["deploy"] +#USAGE flag "--keyper-config-index " help="Keyper config index (auto-detected if omitted)" +#USAGE flag "--identity-prefix " help="32-byte identity prefix hex (random if omitted)" +#USAGE flag "--contract-addr " help="Contract address to watch (defaults to EventTriggerTestHelper)" +#USAGE flag "--topic-index " help="Topic index to match (default: 2)" +#USAGE flag "--topic-value " help="32-byte topic value hex to match (random if omitted)" +#USAGE flag "--ttl " help="TTL in blocks (default: 10000)" +# fmt: on + +import json +import os +import secrets +from pathlib import Path + +import utils + +# --------------------------------------------------------------------------- +# RLP helpers +# --------------------------------------------------------------------------- + + +def _rlp_bytes(data: bytes) -> bytes: + if len(data) == 1 and data[0] < 0x80: + return data + if len(data) == 0: + return b"\x80" + if len(data) <= 55: + return bytes([0x80 + len(data)]) + data + lb = len(data).to_bytes((len(data).bit_length() + 7) // 8, "big") + return bytes([0xB7 + len(lb)]) + lb + data + + +def _rlp_uint(n: int) -> bytes: + if n == 0: + return b"\x80" + raw = n.to_bytes((n.bit_length() + 7) // 8, "big") + return _rlp_bytes(raw) + + +def _rlp_list(items: list[bytes]) -> bytes: + payload = b"".join(items) + if len(payload) <= 55: + return bytes([0xC0 + len(payload)]) + payload + lb = len(payload).to_bytes((len(payload).bit_length() + 7) // 8, "big") + return bytes([0xF7 + len(lb)]) + lb + payload + + +def encode_trigger_def( + contract_addr: bytes, topic_index: int, topic_value: bytes +) -> bytes: + """ + Encode an EventTriggerDefinition (version 0x02 + RLP). + + Matches logs from contract_addr where topic[topic_index] == topic_value. + ValuePredicate uses a flat list [Op, *int_args, *byte_args]. + For BytesEq (Op=5): [5, topic_value]. + """ + log_value_ref = _rlp_list([_rlp_uint(0), _rlp_uint(topic_index)]) + value_predicate = _rlp_list([_rlp_uint(5), _rlp_bytes(topic_value)]) + log_predicate = _rlp_list([log_value_ref, value_predicate]) + definition = _rlp_list([_rlp_bytes(contract_addr), _rlp_list([log_predicate])]) + return bytes([0x02]) + definition + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +RPC = "http://127.0.0.1:8545" + +deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) +if deployment_type != "service": + raise SystemExit("submit-event-registration only supports DEPLOYMENT_TYPE=service") + +data_dir = Path(os.environ["DATA_DIR"]) +chain_id = os.environ["ETHEREUM_CHAIN_ID"] + +broadcast = json.loads( + ( + data_dir + / "contracts" + / "broadcast" + / utils.DEPLOYMENT_SCRIPTS["service"] + / chain_id + / "run-latest.json" + ).read_text() +) +registry_addr = utils.get_uups_proxy_address(broadcast, "ShutterEventTriggerRegistryV1") +if not registry_addr: + raise SystemExit( + "ShutterEventTriggerRegistryV1 proxy not found in deployment artifacts" + ) + +keyper_config_index_str = os.environ.get("usage_keyper_config_index", "").strip() +if not keyper_config_index_str: + eon_str = utils.query_keyper_db(0, "SELECT max(eon) FROM eons") + if not eon_str: + raise SystemExit("No eon found in keyper DB") + keyper_config_index_str = utils.query_keyper_db( + 0, f"SELECT keyper_config_index FROM eons WHERE eon = {int(eon_str)} LIMIT 1" + ) + if not keyper_config_index_str: + raise SystemExit(f"No keyper_config_index found for eon={eon_str}") +keyper_config_index = int(keyper_config_index_str) + +identity_prefix = os.environ.get("usage_identity_prefix", "").strip() +if not identity_prefix: + identity_prefix = secrets.token_hex(32) +identity_prefix = identity_prefix.removeprefix("0x") + +contract_addr_str = os.environ.get("usage_contract_addr", "").strip() +if not contract_addr_str: + addr_path = data_dir / "test-helper-addr.json" + if not addr_path.exists(): + raise SystemExit( + "No --contract-addr given and EventTriggerTestHelper not deployed " + "(run: mise run deploy)" + ) + contract_addr_str = json.loads(addr_path.read_text())["address"] + +topic_index_str = os.environ.get("usage_topic_index", "").strip() +topic_index = int(topic_index_str) if topic_index_str else 2 + +topic_value_str = os.environ.get("usage_topic_value", "").strip() +if not topic_value_str: + topic_value_str = secrets.token_hex(32) +topic_value_bytes = bytes.fromhex(topic_value_str.removeprefix("0x")) +if len(topic_value_bytes) != 32: + raise SystemExit("--topic-value must be exactly 32 bytes (64 hex chars)") + +ttl_str = os.environ.get("usage_ttl", "").strip() +ttl = int(ttl_str) if ttl_str else 10_000 + +contract_addr_bytes = bytes.fromhex(contract_addr_str.removeprefix("0x").zfill(40)) +trigger_def = encode_trigger_def(contract_addr_bytes, topic_index, topic_value_bytes) +trigger_def_hex = trigger_def.hex() + +utils.run( + [ + "docker", + "compose", + "exec", + "-T", + "ethereum", + "cast", + "send", + "--rpc-url", + RPC, + "--private-key", + os.environ["DEPLOY_KEY"], + registry_addr, + "register(uint64,bytes32,bytes,uint64)", + str(keyper_config_index), + f"0x{identity_prefix}", + f"0x{trigger_def_hex}", + str(ttl), + ], +) + +print( + json.dumps( + { + "keyper_config_index": keyper_config_index, + "identity_prefix": identity_prefix, + "contract_addr": contract_addr_str, + "topic_index": topic_index, + "topic_value": topic_value_str.removeprefix("0x"), + "ttl": ttl, + } + ) +) diff --git a/mise-test-setup/mise-tasks/submit-identity-registration b/mise-test-setup/mise-tasks/submit-identity-registration index 892caa7c3..17a17d07f 100755 --- a/mise-test-setup/mise-tasks/submit-identity-registration +++ b/mise-test-setup/mise-tasks/submit-identity-registration @@ -1,9 +1,11 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Submit an identity registration transaction" #MISE depends=["deploy"] #USAGE flag "--eon " help="Target eon" #USAGE flag "--identity-prefix " help="Identity prefix hex string" #USAGE flag "--timestamp " help="Trigger timestamp" +# fmt: on import json import os @@ -27,7 +29,9 @@ def deployment_run_path(data_dir: Path, deployment_type: str) -> Path: deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) if deployment_type != "service": - raise SystemExit("submit-identity-registration only supports DEPLOYMENT_TYPE=service") + raise SystemExit( + "submit-identity-registration only supports DEPLOYMENT_TYPE=service" + ) data_dir = Path(os.environ["DATA_DIR"]) deployment_run = json.loads(deployment_run_path(data_dir, deployment_type).read_text()) diff --git a/mise-test-setup/mise-tasks/test-decryption b/mise-test-setup/mise-tasks/test-decryption index 4b7c474e9..d357f55b7 100755 --- a/mise-test-setup/mise-tasks/test-decryption +++ b/mise-test-setup/mise-tasks/test-decryption @@ -1,11 +1,12 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Submit a decryption trigger and wait for the key" +# fmt: on import json import utils - result = utils.run( ["mise", "run", "submit-identity-registration"], capture_output=True, diff --git a/mise-test-setup/mise-tasks/test-event-decryption b/mise-test-setup/mise-tasks/test-event-decryption new file mode 100755 index 000000000..a2621ddd7 --- /dev/null +++ b/mise-test-setup/mise-tasks/test-event-decryption @@ -0,0 +1,107 @@ +#!/usr/bin/env -S uv run --script +# fmt: off +#MISE description="E2E test: event-based decryption triggers without batching" +#MISE depends=["wait-for-initial-dkg"] +# fmt: on + +import json +import os +import secrets +import time + +import utils + +NUM_TRIGGERS = 3 +POLL_INTERVAL = float(os.environ.get("DECRYPTION_KEY_POLL_INTERVAL", "1")) + + +def assert_no_batching( + keyper_config_index: int, + num_triggers: int, + poll_interval: float = 1.0, + timeout: float = 30.0, +) -> None: + """Assert each DecryptionTrigger carries exactly one identity (no batching).""" + deadline = time.time() + timeout + while True: + count_str = utils.query_keyper_db( + 0, + f"SELECT COUNT(DISTINCT identities_hash) FROM decryption_signatures " + f"WHERE eon = {keyper_config_index}", + ) + count = int(count_str) if count_str else 0 + if count >= num_triggers: + print( + f"No-batching assertion passed: " + f"{count} distinct identities_hash for {num_triggers} triggers" + ) + return + if time.time() > deadline: + raise SystemExit( + f"No-batching assertion failed: expected {num_triggers} distinct " + f"identities_hash in decryption_signatures for eon={keyper_config_index}, " + f"got {count}" + ) + time.sleep(poll_interval) + + +deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) +if deployment_type != "service": + raise SystemExit("test-event-decryption only supports DEPLOYMENT_TYPE=service") + +# Shared topic value — all registrations match the same event when it fires. +trigger_topic_hex = secrets.token_hex(32) +print(f"Using shared topic2=0x{trigger_topic_hex[:8]}… for {NUM_TRIGGERS} triggers") + +# Register NUM_TRIGGERS event triggers, each with a distinct identity prefix. +registrations: list[dict] = [] +for i in range(NUM_TRIGGERS): + result = utils.run_subtask( + [ + "mise", + "run", + "submit-event-registration", + "--topic-value", + trigger_topic_hex, + ], + ) + ctx = json.loads(result.stdout.strip().splitlines()[-1]) + registrations.append(ctx) + print( + f" [{i + 1}/{NUM_TRIGGERS}] Registered: " + f"identity_prefix=0x{ctx['identity_prefix'][:8]}…" + ) + +keyper_config_index = registrations[0]["keyper_config_index"] + +# Give keypers time to sync the EventTriggerRegistered events before we fire. +sync_wait = max(3, int(os.environ.get("ACTIVATION_DELTA", "10"))) +print(f"Waiting {sync_wait}s for keypers to sync trigger registrations…") +time.sleep(sync_wait) + +# Fire one Trigger event — all NUM_TRIGGERS registered definitions will match. +print(f"Firing Trigger event with topic2=0x{trigger_topic_hex[:8]}…") +utils.run_subtask(["mise", "run", "trigger-event", "--topic2", trigger_topic_hex]) + +# Wait for each decryption key. +print("Waiting for decryption keys…") +for reg in registrations: + utils.run_subtask( + [ + "mise", + "run", + "wait-for-event-decryption-key", + "--keyper-config-index", + str(reg["keyper_config_index"]), + "--identity-prefix", + reg["identity_prefix"], + ] + ) + +# Assert that no batching occurred. +print("Asserting no-batching (each trigger must produce a distinct identities_hash)…") +assert_no_batching(keyper_config_index, NUM_TRIGGERS, POLL_INTERVAL) + +print( + f"\nAll {NUM_TRIGGERS} event-based decryption triggers handled without batching. ✓" +) diff --git a/mise-test-setup/mise-tasks/trigger-event b/mise-test-setup/mise-tasks/trigger-event new file mode 100755 index 000000000..881be777e --- /dev/null +++ b/mise-test-setup/mise-tasks/trigger-event @@ -0,0 +1,65 @@ +#!/usr/bin/env -S uv run --script +# fmt: off +#MISE description="Fire an EventTriggerTestHelper Trigger event on-chain" +#MISE depends=["deploy"] +#USAGE flag "--topic2 " help="32-byte topic2 value hex (random if omitted)" +#USAGE flag "--helper-addr " help="EventTriggerTestHelper address (defaults to deployed address)" +# fmt: on + +import json +import os +import secrets +from pathlib import Path + +import utils + +RPC = "http://127.0.0.1:8545" + +deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) +if deployment_type != "service": + raise SystemExit("trigger-event only supports DEPLOYMENT_TYPE=service") + +data_dir = Path(os.environ["DATA_DIR"]) + +helper_addr = os.environ.get("usage_helper_addr", "").strip() +if not helper_addr: + addr_path = data_dir / "test-helper-addr.json" + if not addr_path.exists(): + raise SystemExit( + "No --helper-addr given and EventTriggerTestHelper not deployed " + "(run: mise run deploy)" + ) + helper_addr = json.loads(addr_path.read_text())["address"] + +topic2_str = os.environ.get("usage_topic2", "").strip() +if not topic2_str: + topic2_str = secrets.token_hex(32) +topic2_str = topic2_str.removeprefix("0x") +if len(bytes.fromhex(topic2_str)) != 32: + raise SystemExit("--topic2 must be exactly 32 bytes (64 hex chars)") + +zero32 = "0x" + "0" * 64 + +utils.run( + [ + "docker", + "compose", + "exec", + "-T", + "ethereum", + "cast", + "send", + "--rpc-url", + RPC, + "--private-key", + os.environ["DEPLOY_KEY"], + helper_addr, + "trigger(uint64,bytes32,bytes32,bytes32)", + "0", + f"0x{topic2_str}", + zero32, + zero32, + ], +) + +print(f"Fired Trigger event: topic2=0x{topic2_str}") diff --git a/mise-test-setup/mise-tasks/up b/mise-test-setup/mise-tasks/up index 3cda696a3..6e8299a2a 100755 --- a/mise-test-setup/mise-tasks/up +++ b/mise-test-setup/mise-tasks/up @@ -1,10 +1,12 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Start the implemented docker3 stack" #MISE depends=["patch-genesis", "init-keyper-dbs"] +# fmt: on import os -import utils +import utils utils.run(["docker", "compose", "up", "-d"]) diff --git a/mise-test-setup/mise-tasks/up-db b/mise-test-setup/mise-tasks/up-db index 3596d2d36..5d0f3063d 100755 --- a/mise-test-setup/mise-tasks/up-db +++ b/mise-test-setup/mise-tasks/up-db @@ -1,9 +1,10 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Run db and wait for healthcheck" #MISE depends=["gen-compose"] +# fmt: on import utils - utils.run(["docker", "compose", "up", "-d", "db"]) utils.wait_for_service_health("db") diff --git a/mise-test-setup/mise-tasks/up-ethereum b/mise-test-setup/mise-tasks/up-ethereum index 74e06e257..02555e9b2 100755 --- a/mise-test-setup/mise-tasks/up-ethereum +++ b/mise-test-setup/mise-tasks/up-ethereum @@ -1,9 +1,10 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Run Ethereum and wait for RPC readiness" #MISE depends=["gen-compose"] +# fmt: on import utils - utils.run(["docker", "compose", "up", "-d", "ethereum"]) utils.wait_for_service_health("ethereum") diff --git a/mise-test-setup/mise-tasks/utils.py b/mise-test-setup/mise-tasks/utils.py index 589107eae..1f3886403 100644 --- a/mise-test-setup/mise-tasks/utils.py +++ b/mise-test-setup/mise-tasks/utils.py @@ -4,7 +4,6 @@ import time from pathlib import Path - DEPLOYMENT_SCRIPTS: dict[str, str] = { "gnosis": "Deploy.gnosh.s.sol", "service": "Deploy.service.s.sol", @@ -22,6 +21,11 @@ def run( return subprocess.run(command, check=True, text=True, capture_output=capture_output) +def run_subtask(command: list[str]) -> subprocess.CompletedProcess[str]: + """Run a subtask capturing stdout for JSON parsing while letting stderr flow through.""" + return subprocess.run(command, check=True, text=True, stdout=subprocess.PIPE) + + def wait_for_service_health(service: str, *, timeout_seconds: float = 30.0) -> None: container_id = run( ["docker", "compose", "ps", "-q", service], capture_output=True @@ -63,7 +67,12 @@ def set_toml_path(document, parts: list[str], value) -> None: def keyper_address(config_path: Path) -> str: - return config_path.read_text().splitlines()[0].removeprefix("# Ethereum address: ").strip() + return ( + config_path.read_text() + .splitlines()[0] + .removeprefix("# Ethereum address: ") + .strip() + ) def parse_indices(indices: str) -> list[int]: @@ -115,3 +124,39 @@ def get_created_contract_address( if isinstance(address, str) and address: return address return None + + +def get_uups_proxy_address( + deployment_run: dict[str, object], impl_contract_name: str +) -> str | None: + """Return the ERC1967Proxy address deployed for the given UUPS implementation. + + The deployment script deploys the implementation first, then an ERC1967Proxy + whose constructor input encodes the implementation address. We identify the + correct proxy by searching for the ERC1967Proxy CREATE that immediately follows + the implementation and whose input data contains the implementation address. + """ + transactions = deployment_run.get("transactions") + if not isinstance(transactions, list): + return None + for i, tx in enumerate(transactions): + if not isinstance(tx, dict): + continue + if tx.get("contractName") != impl_contract_name: + continue + impl_addr = (tx.get("contractAddress") or "").lower().replace("0x", "") + if not impl_addr: + continue + for j in range(i + 1, len(transactions)): + candidate = transactions[j] + if not isinstance(candidate, dict): + continue + if candidate.get("contractName") != "ERC1967Proxy": + continue + input_data = (candidate.get("transaction", {}).get("input") or "").lower() + if impl_addr in input_data: + proxy_addr = candidate.get("contractAddress") + if isinstance(proxy_addr, str) and proxy_addr: + return proxy_addr + break + return None diff --git a/mise-test-setup/mise-tasks/wait-for-decryption-key b/mise-test-setup/mise-tasks/wait-for-decryption-key index 20af1ea5f..dd0009783 100755 --- a/mise-test-setup/mise-tasks/wait-for-decryption-key +++ b/mise-test-setup/mise-tasks/wait-for-decryption-key @@ -1,14 +1,15 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Wait for a decryption key" #USAGE flag "--eon " help="Target eon" #USAGE flag "--identity-prefix " help="Identity prefix hex string" +# fmt: on import os import time import utils - requested_eon = os.environ.get("usage_eon", "").strip() identity_prefix = os.environ.get("usage_identity_prefix", "").strip() poll_interval = float(os.environ["DECRYPTION_KEY_POLL_INTERVAL"]) diff --git a/mise-test-setup/mise-tasks/wait-for-dkg b/mise-test-setup/mise-tasks/wait-for-dkg index b06995267..78c4a021c 100755 --- a/mise-test-setup/mise-tasks/wait-for-dkg +++ b/mise-test-setup/mise-tasks/wait-for-dkg @@ -1,14 +1,15 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Wait for DKG completion" #USAGE flag "--keyper-set-index " help="Wait for a successful DKG for this keyper set index" #USAGE flag "--eon " help="Wait for completion of this eon" +# fmt: on import os import time import utils - keyper_set_index = os.environ.get("usage_keyper_set_index", "").strip() eon = os.environ.get("usage_eon", "").strip() poll_interval = float(os.environ["DKG_RESULT_POLL_INTERVAL"]) diff --git a/mise-test-setup/mise-tasks/wait-for-event-decryption-key b/mise-test-setup/mise-tasks/wait-for-event-decryption-key new file mode 100755 index 000000000..eafb7dc22 --- /dev/null +++ b/mise-test-setup/mise-tasks/wait-for-event-decryption-key @@ -0,0 +1,57 @@ +#!/usr/bin/env -S uv run --script +# fmt: off +#MISE description="Wait for a decryption key generated from an event-based trigger" +#USAGE flag "--keyper-config-index " help="Keyper config index" +#USAGE flag "--identity-prefix " help="32-byte identity prefix hex" +# fmt: on + +import os +import time + +import utils + +keyper_config_index_str = os.environ.get("usage_keyper_config_index", "").strip() +identity_prefix = os.environ.get("usage_identity_prefix", "").strip() +if not keyper_config_index_str or not identity_prefix: + raise SystemExit("Provide both --keyper-config-index and --identity-prefix") + +keyper_config_index = int(keyper_config_index_str) +identity_prefix = identity_prefix.removeprefix("0x") +poll_interval = float(os.environ.get("DECRYPTION_KEY_POLL_INTERVAL", "1")) + +deadline = time.time() + float(os.environ.get("DECRYPTION_KEY_TIMEOUT", "120")) +while True: + if time.time() > deadline: + raise SystemExit( + f"Timeout waiting for decryption key: " + f"keyper_config_index={keyper_config_index}, " + f"identity_prefix={identity_prefix}" + ) + + identity = utils.query_keyper_db( + 0, + f"SELECT encode(identity, 'hex') FROM event_trigger_registered_event " + f"WHERE eon = {keyper_config_index} " + f"AND encode(identity_prefix, 'hex') = '{identity_prefix}' " + "LIMIT 1", + ) + if not identity: + time.sleep(poll_interval) + continue + + exists = utils.query_keyper_db( + 0, + "SELECT EXISTS (" + " SELECT 1 FROM decryption_key " + f" WHERE eon = {keyper_config_index} " + f" AND encode(epoch_id, 'hex') = '{identity}'" + ")", + ) + if exists == "t": + print( + f"Decryption key ready: " + f"identity_prefix={identity_prefix} → epoch_id={identity}" + ) + raise SystemExit(0) + + time.sleep(poll_interval) diff --git a/mise-test-setup/mise-tasks/wait-for-initial-dkg b/mise-test-setup/mise-tasks/wait-for-initial-dkg index f56da9423..f1751efd2 100755 --- a/mise-test-setup/mise-tasks/wait-for-initial-dkg +++ b/mise-test-setup/mise-tasks/wait-for-initial-dkg @@ -1,6 +1,8 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Wait for the initial DKG to succeed" #MISE depends=["up", "add-initial-keyper-set"] +# fmt: on import utils diff --git a/rolling-shutter/keyperimpl/gnosis/keyper.go b/rolling-shutter/keyperimpl/gnosis/keyper.go index 70d8fbd8f..2e0d802a5 100644 --- a/rolling-shutter/keyperimpl/gnosis/keyper.go +++ b/rolling-shutter/keyperimpl/gnosis/keyper.go @@ -166,7 +166,17 @@ func (kpr *Keyper) Start(ctx context.Context, runner service.Runner) error { } runner.Go(func() error { return kpr.processInputs(ctx) }) - return runner.StartService(kpr.core, kpr.chainSyncClient, kpr.slotTicker, kpr.eonKeyPublisher) + services := []service.Service{ + kpr.core, + kpr.chainSyncClient, + kpr.slotTicker, + kpr.eonKeyPublisher, + kpr.validatorSyncer, + } + if kpr.sequencerSyncer != nil { + services = append(services, kpr.sequencerSyncer) + } + return runner.StartService(services...) } func NewKeyper(kpr *Keyper, messagingMiddleware *MessagingMiddleware) (*keyper.KeyperCore, error) { @@ -217,17 +227,8 @@ func (kpr *Keyper) initSequencerSyncer(ctx context.Context) error { SyncStartBlockNumber: kpr.config.Gnosis.SyncStartBlockNumber, } - // Perform an initial sync now because it might take some time and doing so during regular - // slot processing might hold up things - latestHeader, err := client.HeaderByNumber(ctx, nil) - if err != nil { - return errors.Wrap(err, "failed to get latest block header") - } - err = kpr.sequencerSyncer.Sync(ctx, latestHeader) - if err != nil { - return err - } - + // Catchup + live subscription lifecycle is owned by SequencerSyncer.Start + // (see runner.StartService call in Keyper.Start). return nil } @@ -257,16 +258,7 @@ func (kpr *Keyper) initValidatorSyncer(ctx context.Context) error { EnableAggregateValidatorRegistrationV1: kpr.config.Gnosis.EnableAggregateValidatorRegistrationV1, } - // Perform an initial sync now because it might take some time and doing so during regular - // slot processing might hold up things - latestHeader, err := validatorSyncerClient.HeaderByNumber(ctx, nil) - if err != nil { - return errors.Wrap(err, "failed to get latest block header") - } - err = kpr.validatorSyncer.Sync(ctx, latestHeader) - if err != nil { - return err - } + // Catchup + live subscription lifecycle is owned by ValidatorSyncer.Start. return nil } diff --git a/rolling-shutter/keyperimpl/gnosis/newblock.go b/rolling-shutter/keyperimpl/gnosis/newblock.go index 8a1af2400..7f5729af0 100644 --- a/rolling-shutter/keyperimpl/gnosis/newblock.go +++ b/rolling-shutter/keyperimpl/gnosis/newblock.go @@ -8,15 +8,11 @@ import ( ) func (kpr *Keyper) processNewBlock(ctx context.Context, ev *syncevent.LatestBlock) error { - if kpr.sequencerSyncer != nil { - if err := kpr.sequencerSyncer.Sync(ctx, ev.Header); err != nil { - return err - } - } - err := kpr.validatorSyncer.Sync(ctx, ev.Header) - if err != nil { - return err - } + // The sequencer + validator syncers used to fetch events via + // eth_getLogs on every new block. That per-block poll has been replaced + // by WatchTransactionSubmitted / WatchUpdated subscriptions set up in + // their Start methods (C4). processNewBlock now only drives the + // slot-timed decryption trigger. slot := medley.BlockTimestampToSlot( ev.Header.Time, kpr.config.Gnosis.GenesisSlotTimestamp, diff --git a/rolling-shutter/keyperimpl/gnosis/sequencersyncer.go b/rolling-shutter/keyperimpl/gnosis/sequencersyncer.go index c9c3313c1..0878007fb 100644 --- a/rolling-shutter/keyperimpl/gnosis/sequencersyncer.go +++ b/rolling-shutter/keyperimpl/gnosis/sequencersyncer.go @@ -18,6 +18,7 @@ import ( "github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/gnosis/database" "github.com/shutter-network/rolling-shutter/rolling-shutter/medley" + "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service" "github.com/shutter-network/rolling-shutter/rolling-shutter/shdb" ) @@ -113,9 +114,87 @@ func (s *SequencerSyncer) handlePotentialReorg(ctx context.Context, header *type return nil } +// Start implements service.Service. It performs a one-off historical catch-up +// via Sync and then switches to a WatchTransactionSubmitted subscription for +// live events, avoiding the per-new-block eth_getLogs poll the old +// Sync-per-block pattern issued. +func (s *SequencerSyncer) Start(ctx context.Context, runner service.Runner) error { + latestHeader, err := s.ExecutionClient.HeaderByNumber(ctx, nil) + if err != nil { + return errors.Wrap(err, "failed to get latest block header for sequencer catchup") + } + if err := s.Sync(ctx, latestHeader); err != nil { + return errors.Wrap(err, "sequencer syncer catchup failed") + } + + startBlock := latestHeader.Number.Uint64() + 1 + sink := make(chan *sequencerBindings.SequencerTransactionSubmitted, 64) + sub, err := s.Contract.WatchTransactionSubmitted(&bind.WatchOpts{ + Context: ctx, + Start: &startBlock, + }, sink) + if err != nil { + return errors.Wrap(err, "failed to subscribe to TransactionSubmitted events") + } + + runner.Go(func() error { + defer sub.Unsubscribe() + for { + select { + case ev := <-sink: + if err := s.handleWatchedEvent(ctx, ev); err != nil { + return errors.Wrap(err, "failed to handle watched sequencer event") + } + case err := <-sub.Err(): + if err == nil { + return nil + } + return errors.Wrap(err, "sequencer TransactionSubmitted subscription ended") + case <-ctx.Done(): + return ctx.Err() + } + } + }) + return nil +} + +// handleWatchedEvent persists a single event delivered by the live +// TransactionSubmitted subscription. For reorged (Removed) events it deletes +// the previously inserted row. +func (s *SequencerSyncer) handleWatchedEvent(ctx context.Context, ev *sequencerBindings.SequencerTransactionSubmitted) error { + if ev.Raw.Removed { + queries := database.New(s.DBPool) + return queries.DeleteTransactionSubmittedEventsFromBlockNumber(ctx, int64(ev.Raw.BlockNumber)) + } + + filtered := s.filterEvents([]*sequencerBindings.SequencerTransactionSubmitted{ev}) + if len(filtered) == 0 { + return nil + } + header, err := s.ExecutionClient.HeaderByHash(ctx, ev.Raw.BlockHash) + if err != nil { + return errors.Wrap(err, "failed to get block header for watched sequencer event") + } + return s.DBPool.BeginFunc(ctx, func(tx pgx.Tx) error { + if err := s.insertTransactionSubmittedEvents(ctx, tx, filtered); err != nil { + return err + } + slot := medley.BlockTimestampToSlot(header.Time, s.GenesisSlotTimestamp, s.SecondsPerSlot) + return database.New(tx).SetTransactionSubmittedEventsSyncedUntil(ctx, database.SetTransactionSubmittedEventsSyncedUntilParams{ + BlockNumber: int64(ev.Raw.BlockNumber), + BlockHash: ev.Raw.BlockHash.Bytes(), + Slot: int64(slot), + }) + }) +} + // Sync fetches transaction submitted events from the sequencer contract and inserts them into the // database. It starts at the end point of the previous call to sync (or 0 if it is the first call) // and ends at the given block number. +// +// Deprecated: Sync is retained for historical catchup and tests; per-new-block +// invocation from processNewBlock has been replaced by the subscription set up +// by Start. func (s *SequencerSyncer) Sync(ctx context.Context, header *types.Header) error { if err := s.handlePotentialReorg(ctx, header); err != nil { return err diff --git a/rolling-shutter/keyperimpl/gnosis/validatorsyncer.go b/rolling-shutter/keyperimpl/gnosis/validatorsyncer.go index 7dadb2ab7..2e5aa2f6e 100644 --- a/rolling-shutter/keyperimpl/gnosis/validatorsyncer.go +++ b/rolling-shutter/keyperimpl/gnosis/validatorsyncer.go @@ -21,6 +21,7 @@ import ( "github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/gnosis/database" "github.com/shutter-network/rolling-shutter/rolling-shutter/medley" "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/beaconapiclient" + "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service" "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/validatorregistry" ) @@ -38,6 +39,91 @@ type ValidatorSyncer struct { EnableAggregateValidatorRegistrationV1 bool } +// Start implements service.Service. It performs a one-off historical catch-up +// via Sync and then switches to a WatchUpdated subscription for live events, +// avoiding the per-new-block eth_getLogs poll the old Sync-per-block pattern +// issued. +func (v *ValidatorSyncer) Start(ctx context.Context, runner service.Runner) error { + latestHeader, err := v.ExecutionClient.HeaderByNumber(ctx, nil) + if err != nil { + return errors.Wrap(err, "failed to get latest block header for validator catchup") + } + if err := v.Sync(ctx, latestHeader); err != nil { + return errors.Wrap(err, "validator syncer catchup failed") + } + + startBlock := latestHeader.Number.Uint64() + 1 + sink := make(chan *validatorRegistryBindings.ValidatorregistryUpdated, 64) + sub, err := v.Contract.WatchUpdated(&bind.WatchOpts{ + Context: ctx, + Start: &startBlock, + }, sink) + if err != nil { + return errors.Wrap(err, "failed to subscribe to validator registry Updated events") + } + + runner.Go(func() error { + defer sub.Unsubscribe() + for { + select { + case ev := <-sink: + if err := v.handleWatchedEvent(ctx, ev); err != nil { + return errors.Wrap(err, "failed to handle watched validator event") + } + case err := <-sub.Err(): + if err == nil { + return nil + } + return errors.Wrap(err, "validator Updated subscription ended") + case <-ctx.Done(): + return ctx.Err() + } + } + }) + return nil +} + +// handleWatchedEvent persists a single event delivered by the live Updated +// subscription. Reorged (Removed) events are logged; validator registrations +// have no schema-level delete-from-block yet, so a proper reorg pathway is +// left for a follow-up alongside a matching sqlc query. +func (v *ValidatorSyncer) handleWatchedEvent(ctx context.Context, ev *validatorRegistryBindings.ValidatorregistryUpdated) error { + if ev.Raw.Removed { + log.Warn(). + Uint64("block-number", ev.Raw.BlockNumber). + Hex("block-hash", ev.Raw.BlockHash.Bytes()). + Msg("validator registry event marked Removed by subscription; not yet implemented") + return nil + } + + filtered, err := v.filterEvents(ctx, []*validatorRegistryBindings.ValidatorregistryUpdated{ev}) + if err != nil { + return err + } + if len(filtered) == 0 { + return nil + } + header, err := v.ExecutionClient.HeaderByHash(ctx, ev.Raw.BlockHash) + if err != nil { + return errors.Wrap(err, "failed to get block header for watched validator event") + } + db := database.New(v.DBPool) + return v.DBPool.BeginFunc(ctx, func(tx pgx.Tx) error { + if err := v.insertEvents(ctx, tx, filtered); err != nil { + return err + } + return db.SetValidatorRegistrationsSyncedUntil(ctx, database.SetValidatorRegistrationsSyncedUntilParams{ + BlockNumber: int64(ev.Raw.BlockNumber), + BlockHash: header.Hash().Bytes(), + }) + }) +} + +// Sync catches up validator registration events for a range. Retained for the +// initial startup catchup (invoked from Start) and for tests; the per-new-block +// polling call site has been replaced by the WatchUpdated subscription. +// +// Deprecated: use Start for lifetime-managed subscription-based syncing. func (v *ValidatorSyncer) Sync(ctx context.Context, header *types.Header) error { db := database.New(v.DBPool) syncedUntil, err := db.GetValidatorRegistrationsSyncedUntil(ctx) diff --git a/rolling-shutter/keyperimpl/shutterservice/eventprocessor.go b/rolling-shutter/keyperimpl/shutterservice/eventprocessor.go index 9ee8225af..5b5407512 100644 --- a/rolling-shutter/keyperimpl/shutterservice/eventprocessor.go +++ b/rolling-shutter/keyperimpl/shutterservice/eventprocessor.go @@ -3,16 +3,37 @@ package shutterservice import ( "context" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" "github.com/jackc/pgx/v4" ) +// FilterCriteria describes the set of contract addresses and event topic[0] +// hashes that an EventProcessor wants logs for. The MultiEventSyncer unions +// these across all processors and issues a single FilterLogs query per range. +// +// An empty Topics0 means the processor wants every log emitted by its +// Addresses, regardless of topic. The union query then drops topic filtering +// entirely. Avoid this unless strictly necessary. +type FilterCriteria struct { + Addresses []common.Address + Topics0 []common.Hash +} + // EventProcessor defines the interface that event processors for MultiEventSyncer must implement. type EventProcessor interface { // GetProcessorName returns a unique name for this processor. GetProcessorName() string - // FetchEvents retrieves events in the given block range (inclusive). - FetchEvents(ctx context.Context, start, end uint64) ([]Event, error) - // ProcessEvents processes the fetched events and stores them in the database. + // FilterCriteria returns the addresses and topic[0] hashes this processor + // wants logs for over the given block range. May be recomputed per range + // since the criteria can depend on state (e.g. active triggers from the db). + FilterCriteria(ctx context.Context, start, end uint64) (FilterCriteria, error) + // ParseEvents inspects the pre-filtered logs and turns them into Events. + // The provided logs are guaranteed to match this processor's FilterCriteria + // from the same range; processors do not need to re-check (address, + // topic[0]) but may apply additional predicates. + ParseEvents(ctx context.Context, start, end uint64, logs []types.Log) ([]Event, error) + // ProcessEvents processes the parsed events and stores them in the database. ProcessEvents(ctx context.Context, tx pgx.Tx, events []Event) error // RollbackEvents removes events with block numbers greater than the specified block number. RollbackEvents(ctx context.Context, tx pgx.Tx, toBlock int64) error diff --git a/rolling-shutter/keyperimpl/shutterservice/eventtriggerregisteredprocessor.go b/rolling-shutter/keyperimpl/shutterservice/eventtriggerregisteredprocessor.go index 0f1b42d71..a48227235 100644 --- a/rolling-shutter/keyperimpl/shutterservice/eventtriggerregisteredprocessor.go +++ b/rolling-shutter/keyperimpl/shutterservice/eventtriggerregisteredprocessor.go @@ -5,7 +5,8 @@ import ( "context" "math" - "github.com/ethereum/go-ethereum/accounts/abi/bind" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/jackc/pgx/v4" "github.com/jackc/pgx/v4/pgxpool" @@ -21,40 +22,53 @@ import ( // in the ShutterRegistry contract. type EventTriggerRegisteredEventProcessor struct { Contract *triggerRegistryV1Bindings.Shuttereventtriggerregistryv1 + Address common.Address DBPool *pgxpool.Pool + + topic common.Hash } func NewEventTriggerRegisteredEventProcessor( contract *triggerRegistryV1Bindings.Shuttereventtriggerregistryv1, + address common.Address, dbPool *pgxpool.Pool, -) *EventTriggerRegisteredEventProcessor { +) (*EventTriggerRegisteredEventProcessor, error) { + parsedABI, err := triggerRegistryV1Bindings.Shuttereventtriggerregistryv1MetaData.GetAbi() + if err != nil { + return nil, errors.Wrap(err, "failed to parse trigger registry ABI") + } + ev, ok := parsedABI.Events["EventTriggerRegistered"] + if !ok { + return nil, errors.New("EventTriggerRegistered event not present in trigger registry ABI") + } return &EventTriggerRegisteredEventProcessor{ Contract: contract, + Address: address, DBPool: dbPool, - } + topic: ev.ID, + }, nil } func (p *EventTriggerRegisteredEventProcessor) GetProcessorName() string { return "event_trigger_registered" } -func (p *EventTriggerRegisteredEventProcessor) FetchEvents(ctx context.Context, start, end uint64) ([]Event, error) { - opts := bind.FilterOpts{ - Start: start, - End: &end, - Context: ctx, - } - it, err := p.Contract.FilterEventTriggerRegistered(&opts, []uint64{}) - if err != nil { - return nil, errors.Wrap(err, "failed to query identity registered events") - } +func (p *EventTriggerRegisteredEventProcessor) FilterCriteria(_ context.Context, _, _ uint64) (FilterCriteria, error) { + return FilterCriteria{ + Addresses: []common.Address{p.Address}, + Topics0: []common.Hash{p.topic}, + }, nil +} - var events []Event - for it.Next() { - events = append(events, it.Event) - } - if it.Error() != nil { - return nil, errors.Wrap(it.Error(), "failed to iterate identity registered events") +func (p *EventTriggerRegisteredEventProcessor) ParseEvents(_ context.Context, _, _ uint64, logs []types.Log) ([]Event, error) { + events := make([]Event, 0, len(logs)) + for i := range logs { + l := logs[i] + ev, err := p.Contract.ParseEventTriggerRegistered(l) + if err != nil { + return nil, errors.Wrap(err, "failed to parse EventTriggerRegistered log") + } + events = append(events, ev) } return events, nil } diff --git a/rolling-shutter/keyperimpl/shutterservice/keyper.go b/rolling-shutter/keyperimpl/shutterservice/keyper.go index 4d5960c41..cf41dc5d6 100644 --- a/rolling-shutter/keyperimpl/shutterservice/keyper.go +++ b/rolling-shutter/keyperimpl/shutterservice/keyper.go @@ -205,10 +205,14 @@ func (kpr *Keyper) initMultiEventSyncer(ctx context.Context) error { if err != nil { return fmt.Errorf("failed to create ShutterRegistry contract instance: %w", err) } - eventTriggerRegisteredProcessor := NewEventTriggerRegisteredEventProcessor( + eventTriggerRegisteredProcessor, err := NewEventTriggerRegisteredEventProcessor( eventTriggerRegistryContract, + kpr.config.Chain.Contracts.ShutterEventTriggerRegistry, kpr.dbpool, ) + if err != nil { + return fmt.Errorf("failed to initialize event trigger registered processor: %w", err) + } triggerClient, err := ethclient.DialContext(ctx, kpr.config.Chain.Node.EthereumURL) if err != nil { diff --git a/rolling-shutter/keyperimpl/shutterservice/multieventsyncer.go b/rolling-shutter/keyperimpl/shutterservice/multieventsyncer.go index 1c27cdd2e..8714136d2 100644 --- a/rolling-shutter/keyperimpl/shutterservice/multieventsyncer.go +++ b/rolling-shutter/keyperimpl/shutterservice/multieventsyncer.go @@ -5,6 +5,8 @@ import ( "context" "math/big" + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" "github.com/jackc/pgx/v4" @@ -102,37 +104,67 @@ func (s *MultiEventSyncer) Sync(ctx context.Context, header *types.Header) error return nil } +// processorScope holds the per-processor address/topic filter so we can route +// the unified FilterLogs response back to each processor without re-checking +// criteria. +type processorScope struct { + processor EventProcessor + addresses map[common.Address]struct{} + topics0 map[common.Hash]struct{} + anyTopic bool +} + func (s *MultiEventSyncer) syncRange(ctx context.Context, start, end uint64) (int, error) { header, err := s.ExecutionClient.HeaderByNumber(ctx, new(big.Int).SetUint64(end)) if err != nil { return 0, errors.Wrap(err, "failed to get execution block header") } + scopes, addressUnion, topicUnion, anyTopic, err := s.gatherFilterCriteria(ctx, start, end) + if err != nil { + return 0, err + } + + var logs []types.Log + if len(addressUnion) > 0 { + query := ethereum.FilterQuery{ + FromBlock: new(big.Int).SetUint64(start), + ToBlock: new(big.Int).SetUint64(end), + Addresses: addressUnion, + } + if !anyTopic && len(topicUnion) > 0 { + query.Topics = [][]common.Hash{topicUnion} + } + logs, err = s.ExecutionClient.FilterLogs(ctx, query) + if err != nil { + return 0, errors.Wrap(err, "failed to fetch logs for combined event query") + } + } + allEvents := make(map[string][]Event) numEvents := 0 - for name, processor := range s.Processors { - events, err := processor.FetchEvents(ctx, start, end) + for _, scope := range scopes { + scoped := filterLogsForScope(logs, scope) + events, err := scope.processor.ParseEvents(ctx, start, end, scoped) if err != nil { - return 0, errors.Wrapf(err, "failed to fetch events for processor %s in range [%d, %d]", name, start, end) + return 0, errors.Wrapf(err, "failed to parse events for processor %s in range [%d, %d]", + scope.processor.GetProcessorName(), start, end) } - allEvents[name] = events + allEvents[scope.processor.GetProcessorName()] = events numEvents += len(events) } err = s.DBPool.BeginFunc(ctx, func(tx pgx.Tx) error { - for name, processor := range s.Processors { - events := allEvents[name] - err := processor.ProcessEvents(ctx, tx, events) - if err != nil { - return errors.Wrapf(err, "failed to process events for processor %s", name) + for _, scope := range scopes { + events := allEvents[scope.processor.GetProcessorName()] + if err := scope.processor.ProcessEvents(ctx, tx, events); err != nil { + return errors.Wrapf(err, "failed to process events for processor %s", + scope.processor.GetProcessorName()) } } - - err := s.setSyncStatus(ctx, tx, int64(end), header.Hash().Bytes()) - if err != nil { + if err := s.setSyncStatus(ctx, tx, int64(end), header.Hash().Bytes()); err != nil { return errors.Wrap(err, "failed to update global sync status") } - return nil }) if err != nil { @@ -142,6 +174,82 @@ func (s *MultiEventSyncer) syncRange(ctx context.Context, start, end uint64) (in return numEvents, nil } +// gatherFilterCriteria asks each processor for its filter criteria and +// returns: +// - per-processor scopes used to route logs back after the combined fetch, +// - the deduplicated union of addresses for the FilterQuery, +// - the deduplicated union of topic[0] hashes, and +// - anyTopic=true if any processor declined to constrain topic[0] (in which +// case the combined query must not filter on topics). +func (s *MultiEventSyncer) gatherFilterCriteria( + ctx context.Context, start, end uint64, +) ([]processorScope, []common.Address, []common.Hash, bool, error) { + scopes := make([]processorScope, 0, len(s.Processors)) + addressSet := map[common.Address]struct{}{} + topicSet := map[common.Hash]struct{}{} + anyTopic := false + for name, processor := range s.Processors { + crit, err := processor.FilterCriteria(ctx, start, end) + if err != nil { + return nil, nil, nil, false, errors.Wrapf(err, + "failed to gather filter criteria for processor %s", name) + } + scope := processorScope{ + processor: processor, + addresses: map[common.Address]struct{}{}, + topics0: map[common.Hash]struct{}{}, + anyTopic: len(crit.Topics0) == 0, + } + for _, addr := range crit.Addresses { + scope.addresses[addr] = struct{}{} + addressSet[addr] = struct{}{} + } + for _, topic := range crit.Topics0 { + scope.topics0[topic] = struct{}{} + topicSet[topic] = struct{}{} + } + if scope.anyTopic && len(crit.Addresses) > 0 { + anyTopic = true + } + scopes = append(scopes, scope) + } + + addresses := make([]common.Address, 0, len(addressSet)) + for a := range addressSet { + addresses = append(addresses, a) + } + topics := make([]common.Hash, 0, len(topicSet)) + for t := range topicSet { + topics = append(topics, t) + } + return scopes, addresses, topics, anyTopic, nil +} + +// filterLogsForScope returns the subset of logs that fall within a processor's +// (address, topic[0]) criteria. +func filterLogsForScope(logs []types.Log, scope processorScope) []types.Log { + if len(scope.addresses) == 0 { + return nil + } + filtered := make([]types.Log, 0, len(logs)) + for i := range logs { + l := &logs[i] + if _, ok := scope.addresses[l.Address]; !ok { + continue + } + if !scope.anyTopic { + if len(l.Topics) == 0 { + continue + } + if _, ok := scope.topics0[l.Topics[0]]; !ok { + continue + } + } + filtered = append(filtered, *l) + } + return filtered +} + func (s *MultiEventSyncer) getSyncStatus(ctx context.Context) (*SyncStatus, error) { queries := database.New(s.DBPool) status, err := queries.GetMultiEventSyncStatus(ctx) diff --git a/rolling-shutter/keyperimpl/shutterservice/newblock.go b/rolling-shutter/keyperimpl/shutterservice/newblock.go index 8b8dfa460..296ff0502 100644 --- a/rolling-shutter/keyperimpl/shutterservice/newblock.go +++ b/rolling-shutter/keyperimpl/shutterservice/newblock.go @@ -218,7 +218,8 @@ func (kpr *Keyper) createTriggersFromIdentityRegisteredEvents( if identityPreimages[event.Eon] == nil { identityPreimages[event.Eon] = make([]identitypreimage.IdentityPreimage, 0) } - identityPreimages[event.Eon] = append(identityPreimages[event.Eon], identitypreimage.IdentityPreimage(event.Identity)) + identityPreimages[event.Eon] = append(identityPreimages[event.Eon], + identitypreimage.IdentityPreimage(event.Identity)) if _, exists := lastEonBlock[event.Eon]; !exists { lastEonBlock[event.Eon] = eon.ActivationBlockNumber @@ -269,19 +270,15 @@ func (kpr *Keyper) prepareEventBasedTriggers(ctx context.Context) ([]epochkghand continue } - identities := []identitypreimage.IdentityPreimage{} + // issue 698: here we "unbatch" all event based decryption triggers and create one broker event per identity for _, firedTrigger := range firedTriggers { - identities = append(identities, firedTrigger.Identity) + identityPreimage := []identitypreimage.IdentityPreimage{firedTrigger.Identity} + decryptionTrigger := epochkghandler.DecryptionTrigger{ + BlockNumber: uint64(eonStruct.ActivationBlockNumber), //nolint:gosec + IdentityPreimages: identityPreimage, + } + decryptionTriggers = append(decryptionTriggers, decryptionTrigger) } - - sortedIdentityPreimages := sortIdentityPreimages(identities) - - decryptionTrigger := epochkghandler.DecryptionTrigger{ - BlockNumber: uint64(eonStruct.ActivationBlockNumber), - IdentityPreimages: sortedIdentityPreimages, - } - - decryptionTriggers = append(decryptionTriggers, decryptionTrigger) } return decryptionTriggers, nil } diff --git a/rolling-shutter/keyperimpl/shutterservice/newblock_test.go b/rolling-shutter/keyperimpl/shutterservice/newblock_test.go index 14fbd040d..3b1d4f37e 100644 --- a/rolling-shutter/keyperimpl/shutterservice/newblock_test.go +++ b/rolling-shutter/keyperimpl/shutterservice/newblock_test.go @@ -1,7 +1,6 @@ package shutterservice import ( - "bytes" "context" "database/sql" "math" @@ -111,7 +110,9 @@ func TestProcessBlockSuccess(t *testing.T) { select { case ev := <-decryptionTriggerChannel: assert.Equal(t, ev.Value.BlockNumber, activationBlockNumberUint64) - assert.DeepEqual(t, ev.Value.IdentityPreimages, []identitypreimage.IdentityPreimage{identity}) + assert.DeepEqual(t, ev.Value.IdentityPreimages, []identitypreimage.IdentityPreimage{ + identity, + }) case <-time.After(2 * time.Second): t.Fatal("expected decryption trigger") } @@ -455,10 +456,7 @@ func setupEventBasedOrderingTest( ) (*Keyper, *servicedatabase.Queries, int64) { t.Helper() - const keyperIndex = uint64(1) - testsetup.InitializeEon(ctx, t, dbpool, config, keyperIndex) - - privateKey, sender, err := generateRandomAccount() + privateKey, _, err := generateRandomAccount() assert.NilError(t, err) kpr := &Keyper{ @@ -471,12 +469,16 @@ func setupEventBasedOrderingTest( }, }, } - _ = sender + + // Register kpr's address as the keyper-at-index-1 so resolveDecryptableEon + // finds it in the keyper set. + const keyperIndex = uint64(1) + testsetup.InitializeEon(ctx, t, dbpool, &kprAddressTestConfig{addr: kpr.config.GetAddress()}, keyperIndex) return kpr, servicedatabase.New(dbpool), 1 } -func TestFiredTriggersProducesOrderedShares(t *testing.T) { +func TestFiredTriggersProducesUnbatchedDecryptionTriggers(t *testing.T) { if testing.Short() { t.Skip("skipping integration test") } @@ -531,15 +533,14 @@ func TestFiredTriggersProducesOrderedShares(t *testing.T) { triggers, err := kpr.prepareEventBasedTriggers(ctx) assert.NilError(t, err) - assert.Equal(t, len(triggers), 1) - assert.Equal(t, len(triggers[0].IdentityPreimages), len(inserted)) - for i := 1; i < len(triggers[0].IdentityPreimages); i++ { - assert.Assert(t, bytes.Compare( - triggers[0].IdentityPreimages[i-1], - triggers[0].IdentityPreimages[i], - ) < 0) + + // Each fired trigger must produce its own DecryptionTrigger with exactly one identity (no batching). + assert.Equal(t, len(triggers), len(inserted)) + for _, trigger := range triggers { + assert.Equal(t, len(trigger.IdentityPreimages), 1) } + // Verify the first trigger can still be used to construct valid key shares. coreDB := corekeyperdatabase.New(dbpool) triggerBlockNumber := triggers[0].BlockNumber if triggerBlockNumber > math.MaxInt64 { @@ -551,14 +552,16 @@ func TestFiredTriggersProducesOrderedShares(t *testing.T) { keyShareHandler := &epochkghandler.KeyShareHandler{ InstanceID: config.GetInstanceID(), - KeyperAddress: config.GetAddress(), + KeyperAddress: kpr.config.GetAddress(), MaxNumKeysPerMessage: config.GetMaxNumKeysPerMessage(), DBPool: dbpool, } msg, err := keyShareHandler.ConstructDecryptionKeyShares(ctx, triggerEon, triggers[0].IdentityPreimages) assert.NilError(t, err) - validator := epochkghandler.NewDecryptionKeyShareHandler(config, dbpool) + validator := epochkghandler.NewDecryptionKeyShareHandler(&kprAddressTestConfig{ + addr: kpr.config.GetAddress(), + }, dbpool) res, err := validator.ValidateMessage(ctx, msg) assert.Equal(t, res, pubsub.ValidationAccept) assert.NilError(t, err) diff --git a/rolling-shutter/keyperimpl/shutterservice/setup_test.go b/rolling-shutter/keyperimpl/shutterservice/setup_test.go index 093ad8da8..50261c547 100644 --- a/rolling-shutter/keyperimpl/shutterservice/setup_test.go +++ b/rolling-shutter/keyperimpl/shutterservice/setup_test.go @@ -33,3 +33,20 @@ func (c *TestConfig) GetMaxNumKeysPerMessage() uint64 { } var _ testsetup.TestConfig = &TestConfig{} + +// kprAddressTestConfig is a TestConfig that substitutes a specific address +// while delegating all other fields to the global config. Used to register +// a dynamically-generated keyper address in InitializeEon. +type kprAddressTestConfig struct { + addr common.Address +} + +func (c *kprAddressTestConfig) GetAddress() common.Address { return c.addr } +func (c *kprAddressTestConfig) GetInstanceID() uint64 { return config.GetInstanceID() } +func (c *kprAddressTestConfig) GetEon() uint64 { return config.GetEon() } +func (c *kprAddressTestConfig) GetCollatorKey() *ecdsa.PrivateKey { return nil } +func (c *kprAddressTestConfig) GetMaxNumKeysPerMessage() uint64 { + return config.GetMaxNumKeysPerMessage() +} + +var _ testsetup.TestConfig = &kprAddressTestConfig{} diff --git a/rolling-shutter/keyperimpl/shutterservice/triggerprocessor.go b/rolling-shutter/keyperimpl/shutterservice/triggerprocessor.go index bc0d17822..525a3436b 100644 --- a/rolling-shutter/keyperimpl/shutterservice/triggerprocessor.go +++ b/rolling-shutter/keyperimpl/shutterservice/triggerprocessor.go @@ -3,8 +3,8 @@ package shutterservice import ( "context" "fmt" - "math/big" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" "github.com/jackc/pgx/v4" @@ -40,57 +40,85 @@ func (tp *TriggerProcessor) GetProcessorName() string { return "trigger" } -func (tp *TriggerProcessor) FetchEvents(ctx context.Context, start, end uint64) ([]Event, error) { - queries := database.New(tp.DBPool) - // Consider event triggers that have not fired yet and have not expired at the start block. - // They might have expired at the end block though which will be checked later. - triggerRegisteredEvents, err := queries.GetActiveEventTriggerRegisteredEvents(ctx, int64(start)) +// FilterCriteria walks the active registered triggers and unions their +// contract addresses and topic[0] hashes. If any active trigger leaves +// topic[0] unconstrained, Topics0 is returned empty so the combined query +// fetches every log emitted by the listed addresses. +func (tp *TriggerProcessor) FilterCriteria(ctx context.Context, start, _ uint64) (FilterCriteria, error) { + triggers, err := tp.activeTriggers(ctx, start) if err != nil { - return nil, errors.Wrap(err, "failed to get event trigger registered events") + return FilterCriteria{}, err } - var events []Event - for _, triggerRegisteredEvent := range triggerRegisteredEvents { - triggerLog := log.With(). - Int64("block-number", triggerRegisteredEvent.BlockNumber). - Hex("block-hash", triggerRegisteredEvent.BlockHash). - Int64("tx-index", triggerRegisteredEvent.TxIndex). - Int64("log-index", triggerRegisteredEvent.LogIndex). - Hex("identity-prefix", triggerRegisteredEvent.IdentityPrefix). - Str("sender", triggerRegisteredEvent.Sender). - Hex("definition", triggerRegisteredEvent.Definition). - Int64("expiration-block-number", triggerRegisteredEvent.ExpirationBlockNumber). - Logger() + addresses := map[common.Address]struct{}{} + topics := map[common.Hash]struct{}{} + anyTopic := false - trigger := EventTriggerDefinition{} - err := trigger.UnmarshalBytes(triggerRegisteredEvent.Definition) + for _, t := range triggers { + q, err := t.def.ToFilterQuery() if err != nil { - // This is not supposed to happen as only valid triggers are inserted into the database. - triggerLog.Error().Err(err).Msg("ignoring invalid trigger definition in database") + // Should not occur for validated triggers, but skip rather than + // poison the whole criteria union. + log.Error().Err(err).Msg("failed to build filter query for active trigger; skipping") continue } - - filterQuery, err := trigger.ToFilterQuery() - if err != nil { - // This is not supposed to happen as only valid triggers are inserted into the database - // and valid triggers should always have a valid filter query. - triggerLog.Error().Err(err).Msg("failed to create filter query for trigger") + for _, a := range q.Addresses { + addresses[a] = struct{}{} + } + if len(q.Topics) == 0 || len(q.Topics[0]) == 0 { + anyTopic = true continue } - filterQuery.FromBlock = new(big.Int).SetUint64(start) - filterQuery.ToBlock = new(big.Int).SetUint64(end) + for _, h := range q.Topics[0] { + topics[h] = struct{}{} + } + } - logs, err := tp.ExecutionClient.FilterLogs(ctx, filterQuery) - if err != nil { - return nil, errors.Wrapf(err, "failed to filter logs for event trigger") + crit := FilterCriteria{ + Addresses: make([]common.Address, 0, len(addresses)), + } + for a := range addresses { + crit.Addresses = append(crit.Addresses, a) + } + if !anyTopic { + crit.Topics0 = make([]common.Hash, 0, len(topics)) + for h := range topics { + crit.Topics0 = append(crit.Topics0, h) } + } + return crit, nil +} - for _, eventLog := range logs { - // Check that the trigger has not expired at the time of the event. - if eventLog.BlockNumber > uint64(triggerRegisteredEvent.ExpirationBlockNumber) { +// ParseEvents inspects each log against every active trigger and emits a +// TriggerEvent for each (log, trigger) pair that matches. +func (tp *TriggerProcessor) ParseEvents(ctx context.Context, start, _ uint64, logs []types.Log) ([]Event, error) { + if len(logs) == 0 { + return nil, nil + } + triggers, err := tp.activeTriggers(ctx, start) + if err != nil { + return nil, err + } + + var events []Event + for _, t := range triggers { + triggerLog := log.With(). + Int64("block-number", t.event.BlockNumber). + Hex("block-hash", t.event.BlockHash). + Int64("tx-index", t.event.TxIndex). + Int64("log-index", t.event.LogIndex). + Hex("identity-prefix", t.event.IdentityPrefix). + Str("sender", t.event.Sender). + Hex("definition", t.event.Definition). + Int64("expiration-block-number", t.event.ExpirationBlockNumber). + Logger() + + for i := range logs { + eventLog := logs[i] + if eventLog.BlockNumber > uint64(t.event.ExpirationBlockNumber) { continue } - match, err := trigger.Match(&eventLog) + match, err := t.def.Match(&eventLog) if err != nil { triggerLog.Error().Err(err).Msg("failed to match trigger with event log") continue @@ -103,14 +131,41 @@ func (tp *TriggerProcessor) FetchEvents(ctx context.Context, start, end uint64) } events = append(events, &TriggerEvent{ Log: eventLog, - EventTriggerRegisteredEvent: triggerRegisteredEvent, + EventTriggerRegisteredEvent: t.event, }) } } - return events, nil } +// activeTrigger pairs a stored EventTriggerRegisteredEvent row with its parsed +// definition. +type activeTrigger struct { + event database.EventTriggerRegisteredEvent + def EventTriggerDefinition +} + +func (tp *TriggerProcessor) activeTriggers(ctx context.Context, start uint64) ([]activeTrigger, error) { + queries := database.New(tp.DBPool) + rows, err := queries.GetActiveEventTriggerRegisteredEvents(ctx, int64(start)) + if err != nil { + return nil, errors.Wrap(err, "failed to get event trigger registered events") + } + active := make([]activeTrigger, 0, len(rows)) + for _, row := range rows { + def := EventTriggerDefinition{} + if err := def.UnmarshalBytes(row.Definition); err != nil { + log.Error().Err(err). + Int64("block-number", row.BlockNumber). + Hex("definition", row.Definition). + Msg("ignoring invalid trigger definition in database") + continue + } + active = append(active, activeTrigger{event: row, def: def}) + } + return active, nil +} + func (tp *TriggerProcessor) ProcessEvents(ctx context.Context, tx pgx.Tx, events []Event) error { queries := database.New(tx) for _, untypedEvent := range events { diff --git a/rolling-shutter/medley/eventsyncer/eventsyncer.go b/rolling-shutter/medley/eventsyncer/eventsyncer.go index c72b35820..aef4bbc61 100644 --- a/rolling-shutter/medley/eventsyncer/eventsyncer.go +++ b/rolling-shutter/medley/eventsyncer/eventsyncer.go @@ -5,7 +5,6 @@ import ( "math/big" "reflect" "sort" - "sync" "time" "github.com/ethereum/go-ethereum" @@ -16,7 +15,7 @@ import ( "github.com/ethereum/go-ethereum/ethclient" "github.com/jackc/pgx/v4" "github.com/pkg/errors" - "golang.org/x/sync/errgroup" + "github.com/rs/zerolog/log" "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/db" "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/retry" @@ -84,6 +83,13 @@ type EventSyncUpdate struct { LogIndex uint64 } +// dispatchKey identifies an EventType by its contract address and the first +// topic of the matching log. +type dispatchKey struct { + address common.Address + topic common.Hash +} + // EventSyncer watches the blockchain for events of given types and yields them in order. type EventSyncer struct { Client *ethclient.Client @@ -93,6 +99,10 @@ type EventSyncer struct { FromBlock uint64 FromLogIndex uint64 + addresses []common.Address + topics []common.Hash + dispatch map[dispatchKey]*EventType + started bool logChannel chan logChannelItem } @@ -101,6 +111,24 @@ type EventSyncer struct { // log index. The types of events to filter for are specified as a set of EventTypes. The finality // offset is the number of blocks we trail behind the current block to be safe from reorgs. func New(client *ethclient.Client, finalityOffset uint64, events []*EventType, fromBlock uint64, fromLogIndex uint64) *EventSyncer { + addressSet := map[common.Address]struct{}{} + topicSet := map[common.Hash]struct{}{} + dispatch := map[dispatchKey]*EventType{} + for _, ev := range events { + topic := ev.ABI.Events[ev.Name].ID + addressSet[ev.Address] = struct{}{} + topicSet[topic] = struct{}{} + dispatch[dispatchKey{address: ev.Address, topic: topic}] = ev + } + addresses := make([]common.Address, 0, len(addressSet)) + for a := range addressSet { + addresses = append(addresses, a) + } + topics := make([]common.Hash, 0, len(topicSet)) + for t := range topicSet { + topics = append(topics, t) + } + return &EventSyncer{ Client: client, FinalityOffset: finalityOffset, @@ -109,6 +137,10 @@ func New(client *ethclient.Client, finalityOffset uint64, events []*EventType, f FromBlock: fromBlock, FromLogIndex: fromLogIndex, + addresses: addresses, + topics: topics, + dispatch: dispatch, + started: false, logChannel: make(chan logChannelItem, outputChannelCapacity), } @@ -165,116 +197,165 @@ func (s *EventSyncer) Start(ctx context.Context, runner service.Runner) error { return nil } -// sync continuously searches for events. +// sync continuously searches for events. It prefers a new-head subscription +// (one eth_subscribe at startup) over the periodic eth_blockNumber poll and +// falls back to polling when the transport does not support subscriptions +// (typically HTTP-only ethclients). func (s *EventSyncer) sync(ctx context.Context) error { fromBlock := s.FromBlock + + headCh := make(chan *types.Header, 1) + sub, err := s.Client.SubscribeNewHead(ctx, headCh) + if err != nil { + log.Debug().Err(err).Msg("SubscribeNewHead unavailable, falling back to polling") + return s.syncByPolling(ctx, fromBlock) + } + defer sub.Unsubscribe() + + return s.syncBySubscription(ctx, fromBlock, headCh, sub) +} + +// syncBySubscription drives the sync loop off a SubscribeNewHead +// subscription instead of polling BlockNumber. On startup it also queries +// HeaderByNumber(nil) once so the initial historical range can be drained +// before the first live head arrives; from then on the subscription is the +// only signal for new work. +func (s *EventSyncer) syncBySubscription( + ctx context.Context, + fromBlock uint64, + headCh <-chan *types.Header, + sub ethereum.Subscription, +) error { + // Kick off with the current tip so we cover the historical gap between + // FromBlock and the head, without waiting for a new block to arrive. + tip, err := retry.FunctionCall(ctx, func(ctx context.Context) (*types.Header, error) { + return s.Client.HeaderByNumber(ctx, nil) + }) + if err != nil { + return errors.Wrap(err, "failed to query current head") + } + fromBlock, err = s.advanceUpTo(ctx, fromBlock, tip.Number.Uint64()) + if err != nil { + return err + } + + for { + select { + case head, ok := <-headCh: + if !ok { + return errors.New("event syncer head channel closed") + } + fromBlock, err = s.advanceUpTo(ctx, fromBlock, head.Number.Uint64()) + if err != nil { + return err + } + case err := <-sub.Err(): + return errors.Wrap(err, "event syncer new-head subscription failed") + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// syncByPolling is the pre-subscription loop, retained for transports that +// cannot subscribe (HTTP ethclient). +func (s *EventSyncer) syncByPolling(ctx context.Context, fromBlock uint64) error { for { currentBlock, err := retry.FunctionCall(ctx, s.Client.BlockNumber) if err != nil { return errors.Wrap(err, "failed to query current block number") } - toBlock := fromBlock + pageSizeBlocks - 1 - var maxToBlock uint64 - if currentBlock >= s.FinalityOffset { - maxToBlock = currentBlock - s.FinalityOffset - } else { - maxToBlock = 0 - } - if toBlock > maxToBlock { - toBlock = maxToBlock + fromBlock, err = s.advanceUpTo(ctx, fromBlock, currentBlock) + if err != nil { + return err } - // if there's no new blocks, wait some time and try again - if toBlock < fromBlock { - select { - case <-time.After(blockPollInterval): - continue - case <-ctx.Done(): - return ctx.Err() - } + select { + case <-time.After(blockPollInterval): + continue + case <-ctx.Done(): + return ctx.Err() } + } +} +// advanceUpTo advances the syncer up to (currentBlock - FinalityOffset), +// consuming as many pageSizeBlocks-wide ranges as fit. Returns the new +// fromBlock. When the current chain is not yet ahead of the finality offset, +// nothing is fetched. +func (s *EventSyncer) advanceUpTo(ctx context.Context, fromBlock, currentBlock uint64) (uint64, error) { + var maxToBlock uint64 + if currentBlock >= s.FinalityOffset { + maxToBlock = currentBlock - s.FinalityOffset + } + for fromBlock <= maxToBlock { + toBlock := fromBlock + pageSizeBlocks - 1 + if toBlock > maxToBlock { + toBlock = maxToBlock + } logItems, err := s.syncAllInRange(ctx, fromBlock, toBlock) if err != nil { - return err + return fromBlock, err } - err = s.sendLogItemsToChannel(ctx, logItems, toBlock) - if err != nil { - return err + if err := s.sendLogItemsToChannel(ctx, logItems, toBlock); err != nil { + return fromBlock, err } - fromBlock = toBlock + 1 } + return fromBlock, nil } -// syncAllInRange returns all events found in the given block range. +// syncAllInRange returns all events found in the given block range. It issues +// a single FilterLogs query covering every registered address and topic, then +// dispatches each returned log to its matching EventType. func (s *EventSyncer) syncAllInRange(ctx context.Context, fromBlock uint64, toBlock uint64) ([]logChannelItem, error) { - logs := []logChannelItem{} - mu := sync.Mutex{} - - errorgroup, errorctx := errgroup.WithContext(ctx) - for _, event := range s.Events { - ev := event - errorgroup.Go(func() error { - logsSingle, err := s.syncSingleInRange(errorctx, ev, fromBlock, toBlock) - if err != nil { - return err - } - - mu.Lock() - defer mu.Unlock() - logs = append(logs, logsSingle...) - return nil - }) - } - if err := errorgroup.Wait(); err != nil { - return nil, err + if len(s.Events) == 0 { + return nil, nil } - - sort.Slice(logs, func(i, j int) bool { - bi := logs[i].log.BlockNumber - bj := logs[j].log.BlockNumber - if bi < bj { - return true - } - if bi == bj { - li := logs[i].log.Index - lj := logs[j].log.Index - return li < lj - } - return false - }) - - return logs, nil -} - -// syncSingleInRange returns the events matching the given type in the given block range. -func (s *EventSyncer) syncSingleInRange(ctx context.Context, event *EventType, fromBlock uint64, toBlock uint64) ([]logChannelItem, error) { - topic := event.ABI.Events[event.Name].ID query := ethereum.FilterQuery{ - BlockHash: nil, FromBlock: new(big.Int).SetUint64(fromBlock), ToBlock: new(big.Int).SetUint64(toBlock), - Addresses: []common.Address{event.Address}, - Topics: [][]common.Hash{{topic}}, + Addresses: s.addresses, + Topics: [][]common.Hash{s.topics}, } logs, err := retry.FunctionCall(ctx, func(ctx context.Context) ([]types.Log, error) { return s.Client.FilterLogs(ctx, query) }) if err != nil { - return nil, errors.New("failed to filter event logs") + return nil, errors.Wrap(err, "failed to filter event logs") } - items := []logChannelItem{} + items := make([]logChannelItem, 0, len(logs)) for i := range logs { + l := &logs[i] + if len(l.Topics) == 0 { + continue + } + ev, ok := s.dispatch[dispatchKey{address: l.Address, topic: l.Topics[0]}] + if !ok { + // The (address, topic) combination has no registered handler. This + // can happen when an address listed for one event type also emits a + // topic that belongs to another address; we simply skip it. + continue + } items = append(items, logChannelItem{ - log: &logs[i], - blockNumber: logs[i].BlockNumber, - eventType: event, + log: l, + blockNumber: l.BlockNumber, + eventType: ev, }) } + + sort.Slice(items, func(i, j int) bool { + bi := items[i].log.BlockNumber + bj := items[j].log.BlockNumber + if bi != bj { + return bi < bj + } + return items[i].log.Index < items[j].log.Index + }) + return items, nil }