From 8bddaeea782e2937d4f9c5ae864e16c0436de516 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Tue, 5 May 2026 15:16:30 +0200 Subject: [PATCH 01/11] Remove batching from event based decryption triggers Context is issue #698 --- .../keyperimpl/shutterservice/newblock.go | 21 ++++++++----------- 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/rolling-shutter/keyperimpl/shutterservice/newblock.go b/rolling-shutter/keyperimpl/shutterservice/newblock.go index 8b8dfa46..344c0215 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), + 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 } From 676cca3cb018287b6cd0e5445f450fba91cbc807 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Mon, 18 May 2026 17:15:01 +0200 Subject: [PATCH 02/11] Add integration tests for changed batching This adds (generated) integration tests and a `mise` test task for ensuring the changed batching behavior when responding to event based decryption triggers. Context is #698 --- .tool-versions | 18 +- mise-test-setup/mise-tasks/gen-keyper-configs | 3 + mise-test-setup/mise-tasks/init-keyper-dbs | 117 +++++-- .../mise-tasks/test-event-decryption | 322 ++++++++++++++++++ mise-test-setup/mise-tasks/utils.py | 38 +++ .../shutterservice/newblock_test.go | 37 +- .../keyperimpl/shutterservice/setup_test.go | 17 + rolling-shutter/p2pmsg/gossip.pb.go | 2 +- rolling-shutter/shmsg/shmsg.pb.go | 2 +- 9 files changed, 501 insertions(+), 55 deletions(-) create mode 100755 mise-test-setup/mise-tasks/test-event-decryption diff --git a/.tool-versions b/.tool-versions index a50ce8eb..ea59406b 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/gen-keyper-configs b/mise-test-setup/mise-tasks/gen-keyper-configs index 27961297..a9f91a0f 100755 --- a/mise-test-setup/mise-tasks/gen-keyper-configs +++ b/mise-test-setup/mise-tasks/gen-keyper-configs @@ -67,6 +67,9 @@ for index in range(int(os.environ["NUM_KEYPERS"])): 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")) + 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}") diff --git a/mise-test-setup/mise-tasks/init-keyper-dbs b/mise-test-setup/mise-tasks/init-keyper-dbs index c38453a0..df45896f 100755 --- a/mise-test-setup/mise-tasks/init-keyper-dbs +++ b/mise-test-setup/mise-tasks/init-keyper-dbs @@ -3,6 +3,7 @@ #MISE depends=["up-db", "gen-keyper-configs"] import os +from pathlib import Path import utils @@ -47,39 +48,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/test-event-decryption b/mise-test-setup/mise-tasks/test-event-decryption new file mode 100755 index 00000000..9016d848 --- /dev/null +++ b/mise-test-setup/mise-tasks/test-event-decryption @@ -0,0 +1,322 @@ +#!/usr/bin/env -S uv run --script +#MISE description="E2E test: event-based decryption triggers without batching" +#MISE depends=["wait-for-initial-dkg", "deploy"] + +import json +import os +import secrets +import time +from pathlib import Path + +import utils + +# --------------------------------------------------------------------------- +# RLP helpers — minimal subset needed for EventTriggerDefinition encoding +# --------------------------------------------------------------------------- + +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) that matches logs from + `contract_addr` where topic[topic_index] == topic_value (32 bytes). + + Mirrors Go's EventTriggerDefinition.MarshalBytes(). + + ValuePredicate uses a custom EncodeRLP that produces a flat list + [Op, *int_args, *byte_args] — NOT [Op, [IntArgs], [ByteArgs]]. + For BytesEq (Op=5) there are 0 int args and 1 byte arg, so the list is [5, topic_value]. + """ + log_value_ref = _rlp_list([ + _rlp_uint(0), # Dynamic = false (bool encoded as uint 0) + _rlp_uint(topic_index), + ]) + # Flat list: [Op, byteArg] (no IntArgs for BytesEq, one ByteArg) + value_predicate = _rlp_list([ + _rlp_uint(5), # Op = BytesEq + _rlp_bytes(topic_value), # the single byte argument, directly in the list + ]) + log_predicate = _rlp_list([log_value_ref, value_predicate]) + definition = _rlp_list([ + _rlp_bytes(contract_addr), + _rlp_list([log_predicate]), + ]) + return bytes([0x02]) + definition + + +# --------------------------------------------------------------------------- +# Contract deployment + interaction via cast inside the ethereum container +# --------------------------------------------------------------------------- + +RPC = "http://127.0.0.1:8545" + + +def cast(*args: str, capture: bool = True) -> str: + """Run a cast subcommand inside the running ethereum container.""" + result = utils.run( + ["docker", "compose", "exec", "-T", "ethereum", "cast", *args], + capture_output=capture, + ) + return result.stdout.strip() if capture else "" + + +def cast_send(*args: str) -> dict: + """Broadcast a transaction and return the parsed JSON receipt.""" + raw = cast("send", "--json", "--rpc-url", RPC, + "--private-key", os.environ["DEPLOY_KEY"], *args) + return json.loads(raw) + + +def deploy_contract(bytecode_hex: str) -> str: + """Deploy a contract from hex bytecode; returns the checksummed contract address.""" + receipt = cast_send("--create", bytecode_hex) + 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)}") + return addr + + +def call_register(registry_addr: str, keyper_config_index: int, + identity_prefix_hex: str, trigger_def_hex: str, ttl: int) -> None: + cast_send( + registry_addr, + "register(uint64,bytes32,bytes,uint64)", + str(keyper_config_index), + f"0x{identity_prefix_hex}", + f"0x{trigger_def_hex}", + str(ttl), + ) + + +def call_trigger(helper_addr: str, topic2_hex: str) -> None: + zero32 = "0x" + "0" * 64 + cast_send( + helper_addr, + "trigger(uint64,bytes32,bytes32,bytes32)", + "0", + f"0x{topic2_hex}", + zero32, + zero32, + ) + + +# --------------------------------------------------------------------------- +# DB helpers +# --------------------------------------------------------------------------- + + +def query_db(sql: str, keyper_index: int = 0) -> str: + return utils.query_keyper_db(keyper_index, sql) + + +def wait_for_event_decryption_key( + keyper_config_index: int, + identity_prefix_hex: str, + poll_interval: float = 1.0, + timeout: float = 120.0, +) -> str: + """ + Poll until the decryption key for the event trigger with the given identity_prefix + is available. Returns the hex-encoded identity (epoch_id). + """ + deadline = time.time() + timeout + while True: + if time.time() > deadline: + raise SystemExit( + f"Timeout waiting for decryption key for " + f"identity_prefix={identity_prefix_hex}, eon={keyper_config_index}" + ) + + identity = query_db( + f"SELECT encode(identity, 'hex') FROM event_trigger_registered_event " + f"WHERE eon = {keyper_config_index} " + f"AND encode(identity_prefix, 'hex') = '{identity_prefix_hex}' " + "LIMIT 1" + ) + if not identity: + time.sleep(poll_interval) + continue + + exists = query_db( + "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: identity_prefix={identity_prefix_hex} → epoch_id={identity}") + return identity + + time.sleep(poll_interval) + + +def assert_no_batching( + keyper_config_index: int, + num_triggers: int, + poll_interval: float = 1.0, + timeout: float = 30.0, +) -> None: + """ + Assert that decryption_signatures contains `num_triggers` distinct identities_hash + values for the given eon. With no batching each DecryptionTrigger carries exactly + one identity, so its identities_hash = keccak256(identity). With batching all N + identities would share one hash. + """ + deadline = time.time() + timeout + while True: + count_str = query_db( + 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) + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +def get_registry_address(data_dir: Path) -> str: + broadcast = json.loads( + (data_dir / "contracts" / "broadcast" + / utils.DEPLOYMENT_SCRIPTS["service"] + / os.environ["ETHEREUM_CHAIN_ID"] + / "run-latest.json").read_text() + ) + addr = utils.get_uups_proxy_address(broadcast, "ShutterEventTriggerRegistryV1") + if not addr: + raise SystemExit("ShutterEventTriggerRegistryV1 proxy not found in deployment artifacts") + return addr + + +def get_helper_bytecode(data_dir: Path) -> str: + artifact = json.loads( + (data_dir / "contracts" / "out" + / "EventTriggerTestHelper.sol" + / "EventTriggerTestHelper.json").read_text() + ) + return artifact["bytecode"]["object"].removeprefix("0x") + + +def main() -> None: + 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") + + data_dir = Path(os.environ["DATA_DIR"]) + poll_interval = float(os.environ.get("DECRYPTION_KEY_POLL_INTERVAL", "1")) + num_triggers = 3 + + # --- Resolve eon and keyper_config_index ----------------------------------- + # ShutterEventTriggerRegistryV1.register() takes the *keyper_config_index*, + # not the eon number. We obtain both from the eons table and use accordingly. + eon_str = query_db("SELECT max(eon) FROM eons") + if not eon_str: + raise SystemExit("No eon found in keyper DB — has DKG completed?") + eon_number = int(eon_str) + + keyper_config_index_str = query_db( + f"SELECT keyper_config_index FROM eons WHERE eon = {eon_number} LIMIT 1" + ) + if not keyper_config_index_str: + raise SystemExit(f"No keyper_config_index found for eon={eon_number}") + keyper_config_index = int(keyper_config_index_str) + + print(f"Using eon={eon_number}, keyper_config_index={keyper_config_index}") + + # --- Deploy EventTriggerTestHelper ---------------------------------------- + print("Deploying EventTriggerTestHelper…") + helper_bytecode = get_helper_bytecode(data_dir) + helper_addr = deploy_contract(helper_bytecode) + print(f" EventTriggerTestHelper deployed at {helper_addr}") + + # --- Build the trigger definition ----------------------------------------- + # Match Trigger(…, topic2=trigger_topic, …) events emitted by helper_addr. + # topic[2] is the second indexed parameter of the Trigger event (bytes32). + trigger_topic = secrets.token_bytes(32) + trigger_topic_hex = trigger_topic.hex() + + helper_addr_bytes = bytes.fromhex(helper_addr.removeprefix("0x").zfill(40)) + trigger_def = encode_trigger_def( + contract_addr=helper_addr_bytes, + topic_index=2, # topic[2] = topic2 (bytes32 indexed) + topic_value=trigger_topic, + ) + trigger_def_hex = trigger_def.hex() + print(f" EventTriggerDefinition: 0x{trigger_def_hex}") + + # --- Generate identity prefixes (one per trigger) ------------------------- + identity_prefixes = [secrets.token_bytes(32).hex() for _ in range(num_triggers)] + + # --- Register N event triggers -------------------------------------------- + registry_addr = get_registry_address(data_dir) + ttl = 10_000 # blocks until expiry + + print(f"Registering {num_triggers} event triggers on registry {registry_addr}…") + for ip_hex in identity_prefixes: + call_register(registry_addr, keyper_config_index, ip_hex, trigger_def_hex, ttl) + print(f" Registered trigger with identity_prefix=0x{ip_hex[:8]}…") + + # Give keypers time to sync the EventTriggerRegistered events before we fire. + # With ETHEREUM_BLOCK_TIME=1 a couple of seconds is sufficient. + 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 3 registered definitions will match ----- + print(f"Firing Trigger event with topic2=0x{trigger_topic_hex[:8]}…") + call_trigger(helper_addr, trigger_topic_hex) + + # --- Wait for decryption keys --------------------------------------------- + print("Waiting for decryption keys…") + for ip_hex in identity_prefixes: + wait_for_event_decryption_key(keyper_config_index, ip_hex, poll_interval) + + # --- Assert no batching --------------------------------------------------- + print("Asserting no-batching (each trigger must produce a distinct identities_hash)…") + assert_no_batching(keyper_config_index, num_triggers) + + print( + f"\nAll {num_triggers} event-based decryption triggers handled without batching. ✓" + ) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/mise-test-setup/mise-tasks/utils.py b/mise-test-setup/mise-tasks/utils.py index 589107ea..ef07fe9e 100644 --- a/mise-test-setup/mise-tasks/utils.py +++ b/mise-test-setup/mise-tasks/utils.py @@ -115,3 +115,41 @@ 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/rolling-shutter/keyperimpl/shutterservice/newblock_test.go b/rolling-shutter/keyperimpl/shutterservice/newblock_test.go index 14fbd040..3b1d4f37 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 093ad8da..50261c54 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/p2pmsg/gossip.pb.go b/rolling-shutter/p2pmsg/gossip.pb.go index 0fc0f065..978e05d2 100644 --- a/rolling-shutter/p2pmsg/gossip.pb.go +++ b/rolling-shutter/p2pmsg/gossip.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.5 -// protoc v5.29.3 +// protoc v7.34.1 // source: gossip.proto package p2pmsg diff --git a/rolling-shutter/shmsg/shmsg.pb.go b/rolling-shutter/shmsg/shmsg.pb.go index 8ffa518e..1a1f446f 100644 --- a/rolling-shutter/shmsg/shmsg.pb.go +++ b/rolling-shutter/shmsg/shmsg.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.5 -// protoc v5.29.3 +// protoc v7.34.1 // source: shmsg.proto package shmsg From 20a364aa851f878ee17a868b549c6b74063a7ab2 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Mon, 18 May 2026 17:53:47 +0200 Subject: [PATCH 03/11] Fix linter error ActivationBlockNumber can not be negative, so no overflow risk here. --- rolling-shutter/keyperimpl/shutterservice/newblock.go | 2 +- rolling-shutter/p2pmsg/gossip.pb.go | 2 +- rolling-shutter/shmsg/shmsg.pb.go | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/rolling-shutter/keyperimpl/shutterservice/newblock.go b/rolling-shutter/keyperimpl/shutterservice/newblock.go index 344c0215..296ff050 100644 --- a/rolling-shutter/keyperimpl/shutterservice/newblock.go +++ b/rolling-shutter/keyperimpl/shutterservice/newblock.go @@ -274,7 +274,7 @@ func (kpr *Keyper) prepareEventBasedTriggers(ctx context.Context) ([]epochkghand for _, firedTrigger := range firedTriggers { identityPreimage := []identitypreimage.IdentityPreimage{firedTrigger.Identity} decryptionTrigger := epochkghandler.DecryptionTrigger{ - BlockNumber: uint64(eonStruct.ActivationBlockNumber), + BlockNumber: uint64(eonStruct.ActivationBlockNumber), //nolint:gosec IdentityPreimages: identityPreimage, } decryptionTriggers = append(decryptionTriggers, decryptionTrigger) diff --git a/rolling-shutter/p2pmsg/gossip.pb.go b/rolling-shutter/p2pmsg/gossip.pb.go index 978e05d2..0fc0f065 100644 --- a/rolling-shutter/p2pmsg/gossip.pb.go +++ b/rolling-shutter/p2pmsg/gossip.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.5 -// protoc v7.34.1 +// protoc v5.29.3 // source: gossip.proto package p2pmsg diff --git a/rolling-shutter/shmsg/shmsg.pb.go b/rolling-shutter/shmsg/shmsg.pb.go index 1a1f446f..8ffa518e 100644 --- a/rolling-shutter/shmsg/shmsg.pb.go +++ b/rolling-shutter/shmsg/shmsg.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.5 -// protoc v7.34.1 +// protoc v5.29.3 // source: shmsg.proto package shmsg From f417598fea060c80bd314c68c9a4ad922c6a618f Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Thu, 21 May 2026 11:18:52 +0200 Subject: [PATCH 04/11] Remove superfluous db helper Also apply `black` on `test-event-decryption`. --- .../mise-tasks/test-event-decryption | 142 +++++++++++------- 1 file changed, 88 insertions(+), 54 deletions(-) diff --git a/mise-test-setup/mise-tasks/test-event-decryption b/mise-test-setup/mise-tasks/test-event-decryption index 9016d848..e935a740 100755 --- a/mise-test-setup/mise-tasks/test-event-decryption +++ b/mise-test-setup/mise-tasks/test-event-decryption @@ -1,6 +1,6 @@ #!/usr/bin/env -S uv run --script -#MISE description="E2E test: event-based decryption triggers without batching" -#MISE depends=["wait-for-initial-dkg", "deploy"] +# MISE description="E2E test: event-based decryption triggers without batching" +# MISE depends=["wait-for-initial-dkg", "deploy"] import json import os @@ -14,6 +14,7 @@ import utils # RLP helpers — minimal subset needed for EventTriggerDefinition encoding # --------------------------------------------------------------------------- + def _rlp_bytes(data: bytes) -> bytes: if len(data) == 1 and data[0] < 0x80: return data @@ -22,7 +23,7 @@ def _rlp_bytes(data: bytes) -> bytes: 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 + return bytes([0xB7 + len(lb)]) + lb + data def _rlp_uint(n: int) -> bytes: @@ -35,12 +36,14 @@ def _rlp_uint(n: int) -> bytes: def _rlp_list(items: list[bytes]) -> bytes: payload = b"".join(items) if len(payload) <= 55: - return bytes([0xc0 + len(payload)]) + payload + 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 + return bytes([0xF7 + len(lb)]) + lb + payload -def encode_trigger_def(contract_addr: bytes, topic_index: int, topic_value: bytes) -> bytes: +def encode_trigger_def( + contract_addr: bytes, topic_index: int, topic_value: bytes +) -> bytes: """ Encode an EventTriggerDefinition (version 0x02 + RLP) that matches logs from `contract_addr` where topic[topic_index] == topic_value (32 bytes). @@ -51,20 +54,26 @@ def encode_trigger_def(contract_addr: bytes, topic_index: int, topic_value: byte [Op, *int_args, *byte_args] — NOT [Op, [IntArgs], [ByteArgs]]. For BytesEq (Op=5) there are 0 int args and 1 byte arg, so the list is [5, topic_value]. """ - log_value_ref = _rlp_list([ - _rlp_uint(0), # Dynamic = false (bool encoded as uint 0) - _rlp_uint(topic_index), - ]) + log_value_ref = _rlp_list( + [ + _rlp_uint(0), # Dynamic = false (bool encoded as uint 0) + _rlp_uint(topic_index), + ] + ) # Flat list: [Op, byteArg] (no IntArgs for BytesEq, one ByteArg) - value_predicate = _rlp_list([ - _rlp_uint(5), # Op = BytesEq - _rlp_bytes(topic_value), # the single byte argument, directly in the list - ]) + value_predicate = _rlp_list( + [ + _rlp_uint(5), # Op = BytesEq + _rlp_bytes(topic_value), # the single byte argument, directly in the list + ] + ) log_predicate = _rlp_list([log_value_ref, value_predicate]) - definition = _rlp_list([ - _rlp_bytes(contract_addr), - _rlp_list([log_predicate]), - ]) + definition = _rlp_list( + [ + _rlp_bytes(contract_addr), + _rlp_list([log_predicate]), + ] + ) return bytes([0x02]) + definition @@ -86,8 +95,15 @@ def cast(*args: str, capture: bool = True) -> str: def cast_send(*args: str) -> dict: """Broadcast a transaction and return the parsed JSON receipt.""" - raw = cast("send", "--json", "--rpc-url", RPC, - "--private-key", os.environ["DEPLOY_KEY"], *args) + raw = cast( + "send", + "--json", + "--rpc-url", + RPC, + "--private-key", + os.environ["DEPLOY_KEY"], + *args, + ) return json.loads(raw) @@ -96,12 +112,19 @@ def deploy_contract(bytecode_hex: str) -> str: receipt = cast_send("--create", bytecode_hex) 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)}") + raise SystemExit( + f"Could not find contractAddress in receipt:\n{json.dumps(receipt, indent=2)}" + ) return addr -def call_register(registry_addr: str, keyper_config_index: int, - identity_prefix_hex: str, trigger_def_hex: str, ttl: int) -> None: +def call_register( + registry_addr: str, + keyper_config_index: int, + identity_prefix_hex: str, + trigger_def_hex: str, + ttl: int, +) -> None: cast_send( registry_addr, "register(uint64,bytes32,bytes,uint64)", @@ -124,15 +147,6 @@ def call_trigger(helper_addr: str, topic2_hex: str) -> None: ) -# --------------------------------------------------------------------------- -# DB helpers -# --------------------------------------------------------------------------- - - -def query_db(sql: str, keyper_index: int = 0) -> str: - return utils.query_keyper_db(keyper_index, sql) - - def wait_for_event_decryption_key( keyper_config_index: int, identity_prefix_hex: str, @@ -151,25 +165,29 @@ def wait_for_event_decryption_key( f"identity_prefix={identity_prefix_hex}, eon={keyper_config_index}" ) - identity = query_db( + 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_hex}' " - "LIMIT 1" + "LIMIT 1", ) if not identity: time.sleep(poll_interval) continue - exists = query_db( + 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: identity_prefix={identity_prefix_hex} → epoch_id={identity}") + print( + f" Decryption key ready: identity_prefix={identity_prefix_hex} → epoch_id={identity}" + ) return identity time.sleep(poll_interval) @@ -189,9 +207,10 @@ def assert_no_batching( """ deadline = time.time() + timeout while True: - count_str = query_db( + count_str = utils.query_keyper_db( + 0, f"SELECT COUNT(DISTINCT identities_hash) FROM decryption_signatures " - f"WHERE eon = {keyper_config_index}" + f"WHERE eon = {keyper_config_index}", ) count = int(count_str) if count_str else 0 if count >= num_triggers: @@ -213,30 +232,43 @@ def assert_no_batching( # Main # --------------------------------------------------------------------------- + def get_registry_address(data_dir: Path) -> str: broadcast = json.loads( - (data_dir / "contracts" / "broadcast" - / utils.DEPLOYMENT_SCRIPTS["service"] - / os.environ["ETHEREUM_CHAIN_ID"] - / "run-latest.json").read_text() + ( + data_dir + / "contracts" + / "broadcast" + / utils.DEPLOYMENT_SCRIPTS["service"] + / os.environ["ETHEREUM_CHAIN_ID"] + / "run-latest.json" + ).read_text() ) addr = utils.get_uups_proxy_address(broadcast, "ShutterEventTriggerRegistryV1") if not addr: - raise SystemExit("ShutterEventTriggerRegistryV1 proxy not found in deployment artifacts") + raise SystemExit( + "ShutterEventTriggerRegistryV1 proxy not found in deployment artifacts" + ) return addr def get_helper_bytecode(data_dir: Path) -> str: artifact = json.loads( - (data_dir / "contracts" / "out" - / "EventTriggerTestHelper.sol" - / "EventTriggerTestHelper.json").read_text() + ( + data_dir + / "contracts" + / "out" + / "EventTriggerTestHelper.sol" + / "EventTriggerTestHelper.json" + ).read_text() ) return artifact["bytecode"]["object"].removeprefix("0x") def main() -> None: - deployment_type = utils.resolve_deployment_type(os.environ.get("DEPLOYMENT_TYPE", "")) + 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") @@ -247,13 +279,13 @@ def main() -> None: # --- Resolve eon and keyper_config_index ----------------------------------- # ShutterEventTriggerRegistryV1.register() takes the *keyper_config_index*, # not the eon number. We obtain both from the eons table and use accordingly. - eon_str = query_db("SELECT max(eon) FROM eons") + eon_str = utils.query_keyper_db(0, "SELECT max(eon) FROM eons") if not eon_str: raise SystemExit("No eon found in keyper DB — has DKG completed?") eon_number = int(eon_str) - keyper_config_index_str = query_db( - f"SELECT keyper_config_index FROM eons WHERE eon = {eon_number} LIMIT 1" + keyper_config_index_str = utils.query_keyper_db( + 0, f"SELECT keyper_config_index FROM eons WHERE eon = {eon_number} LIMIT 1" ) if not keyper_config_index_str: raise SystemExit(f"No keyper_config_index found for eon={eon_number}") @@ -276,7 +308,7 @@ def main() -> None: helper_addr_bytes = bytes.fromhex(helper_addr.removeprefix("0x").zfill(40)) trigger_def = encode_trigger_def( contract_addr=helper_addr_bytes, - topic_index=2, # topic[2] = topic2 (bytes32 indexed) + topic_index=2, # topic[2] = topic2 (bytes32 indexed) topic_value=trigger_topic, ) trigger_def_hex = trigger_def.hex() @@ -310,7 +342,9 @@ def main() -> None: wait_for_event_decryption_key(keyper_config_index, ip_hex, poll_interval) # --- Assert no batching --------------------------------------------------- - print("Asserting no-batching (each trigger must produce a distinct identities_hash)…") + print( + "Asserting no-batching (each trigger must produce a distinct identities_hash)…" + ) assert_no_batching(keyper_config_index, num_triggers) print( @@ -319,4 +353,4 @@ def main() -> None: if __name__ == "__main__": - main() \ No newline at end of file + main() From 427955e081f59b13de366a83c6a111245a7c7234 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Tue, 26 May 2026 13:43:41 +0200 Subject: [PATCH 05/11] Move helper deployment to new mise task --- mise-test-setup/mise-tasks/deploy-test-helper | 59 +++++++++++++++++++ .../mise-tasks/test-event-decryption | 34 +++-------- 2 files changed, 66 insertions(+), 27 deletions(-) create mode 100755 mise-test-setup/mise-tasks/deploy-test-helper diff --git a/mise-test-setup/mise-tasks/deploy-test-helper b/mise-test-setup/mise-tasks/deploy-test-helper new file mode 100755 index 00000000..8c8821e1 --- /dev/null +++ b/mise-test-setup/mise-tasks/deploy-test-helper @@ -0,0 +1,59 @@ +#!/usr/bin/env -S uv run --script +#MISE description="Deploy EventTriggerTestHelper (idempotent, address cached in DATA_DIR)" +#MISE depends=["deploy"] + +import json +import os +from pathlib import Path + +import utils + +RPC = "http://127.0.0.1:8545" + +data_dir = Path(os.environ["DATA_DIR"]) +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}") + raise SystemExit(0) + +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}") diff --git a/mise-test-setup/mise-tasks/test-event-decryption b/mise-test-setup/mise-tasks/test-event-decryption index e935a740..5d093217 100755 --- a/mise-test-setup/mise-tasks/test-event-decryption +++ b/mise-test-setup/mise-tasks/test-event-decryption @@ -1,6 +1,6 @@ #!/usr/bin/env -S uv run --script # MISE description="E2E test: event-based decryption triggers without batching" -# MISE depends=["wait-for-initial-dkg", "deploy"] +# MISE depends=["wait-for-initial-dkg", "deploy-test-helper"] import json import os @@ -107,16 +107,6 @@ def cast_send(*args: str) -> dict: return json.loads(raw) -def deploy_contract(bytecode_hex: str) -> str: - """Deploy a contract from hex bytecode; returns the checksummed contract address.""" - receipt = cast_send("--create", bytecode_hex) - 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)}" - ) - return addr - def call_register( registry_addr: str, @@ -252,17 +242,9 @@ def get_registry_address(data_dir: Path) -> str: return addr -def get_helper_bytecode(data_dir: Path) -> str: - artifact = json.loads( - ( - data_dir - / "contracts" - / "out" - / "EventTriggerTestHelper.sol" - / "EventTriggerTestHelper.json" - ).read_text() - ) - return artifact["bytecode"]["object"].removeprefix("0x") +def get_helper_address(data_dir: Path) -> str: + addr_path = data_dir / "test-helper-addr.json" + return json.loads(addr_path.read_text())["address"] def main() -> None: @@ -293,11 +275,9 @@ def main() -> None: print(f"Using eon={eon_number}, keyper_config_index={keyper_config_index}") - # --- Deploy EventTriggerTestHelper ---------------------------------------- - print("Deploying EventTriggerTestHelper…") - helper_bytecode = get_helper_bytecode(data_dir) - helper_addr = deploy_contract(helper_bytecode) - print(f" EventTriggerTestHelper deployed at {helper_addr}") + # --- Read EventTriggerTestHelper address (deployed by deploy-test-helper) --- + helper_addr = get_helper_address(data_dir) + print(f"Using EventTriggerTestHelper at {helper_addr}") # --- Build the trigger definition ----------------------------------------- # Match Trigger(…, topic2=trigger_topic, …) events emitted by helper_addr. From e0f0c4405652da1fcc5a9101cbd5d946c70269f3 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Wed, 27 May 2026 10:07:30 +0200 Subject: [PATCH 06/11] PR comment fixes - deploy test helper in `deploy` - untangled test and setup: - explicit `submit-event-registration` task - explicit `trigger-event` task - explicit `wait-for-event-decryption-key` --- mise-test-setup/mise-tasks/deploy | 87 ++++- mise-test-setup/mise-tasks/deploy-test-helper | 59 --- .../mise-tasks/submit-event-registration | 171 +++++++++ .../mise-tasks/test-event-decryption | 341 +++--------------- mise-test-setup/mise-tasks/trigger-event | 63 ++++ mise-test-setup/mise-tasks/utils.py | 7 + .../mise-tasks/wait-for-event-decryption-key | 55 +++ 7 files changed, 422 insertions(+), 361 deletions(-) delete mode 100755 mise-test-setup/mise-tasks/deploy-test-helper create mode 100755 mise-test-setup/mise-tasks/submit-event-registration create mode 100755 mise-test-setup/mise-tasks/trigger-event create mode 100755 mise-test-setup/mise-tasks/wait-for-event-decryption-key diff --git a/mise-test-setup/mise-tasks/deploy b/mise-test-setup/mise-tasks/deploy index 3d117b2b..a69ec8e8 100755 --- a/mise-test-setup/mise-tasks/deploy +++ b/mise-test-setup/mise-tasks/deploy @@ -2,11 +2,14 @@ #MISE description="Deploy contracts if not already deployed" #MISE depends=["up-ethereum"] -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,6 +33,54 @@ 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] @@ -39,19 +90,21 @@ run_path = data_dir / "contracts" / "broadcast" / deploy_script / chain_id / "ru 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/deploy-test-helper b/mise-test-setup/mise-tasks/deploy-test-helper deleted file mode 100755 index 8c8821e1..00000000 --- a/mise-test-setup/mise-tasks/deploy-test-helper +++ /dev/null @@ -1,59 +0,0 @@ -#!/usr/bin/env -S uv run --script -#MISE description="Deploy EventTriggerTestHelper (idempotent, address cached in DATA_DIR)" -#MISE depends=["deploy"] - -import json -import os -from pathlib import Path - -import utils - -RPC = "http://127.0.0.1:8545" - -data_dir = Path(os.environ["DATA_DIR"]) -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}") - raise SystemExit(0) - -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}") 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 00000000..a18d43c3 --- /dev/null +++ b/mise-test-setup/mise-tasks/submit-event-registration @@ -0,0 +1,171 @@ +#!/usr/bin/env -S uv run --script +#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)" + +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/test-event-decryption b/mise-test-setup/mise-tasks/test-event-decryption index 5d093217..ab79b597 100755 --- a/mise-test-setup/mise-tasks/test-event-decryption +++ b/mise-test-setup/mise-tasks/test-event-decryption @@ -1,186 +1,16 @@ #!/usr/bin/env -S uv run --script -# MISE description="E2E test: event-based decryption triggers without batching" -# MISE depends=["wait-for-initial-dkg", "deploy-test-helper"] +#MISE description="E2E test: event-based decryption triggers without batching" +#MISE depends=["wait-for-initial-dkg"] import json import os import secrets import time -from pathlib import Path import utils -# --------------------------------------------------------------------------- -# RLP helpers — minimal subset needed for EventTriggerDefinition encoding -# --------------------------------------------------------------------------- - - -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) that matches logs from - `contract_addr` where topic[topic_index] == topic_value (32 bytes). - - Mirrors Go's EventTriggerDefinition.MarshalBytes(). - - ValuePredicate uses a custom EncodeRLP that produces a flat list - [Op, *int_args, *byte_args] — NOT [Op, [IntArgs], [ByteArgs]]. - For BytesEq (Op=5) there are 0 int args and 1 byte arg, so the list is [5, topic_value]. - """ - log_value_ref = _rlp_list( - [ - _rlp_uint(0), # Dynamic = false (bool encoded as uint 0) - _rlp_uint(topic_index), - ] - ) - # Flat list: [Op, byteArg] (no IntArgs for BytesEq, one ByteArg) - value_predicate = _rlp_list( - [ - _rlp_uint(5), # Op = BytesEq - _rlp_bytes(topic_value), # the single byte argument, directly in the list - ] - ) - log_predicate = _rlp_list([log_value_ref, value_predicate]) - definition = _rlp_list( - [ - _rlp_bytes(contract_addr), - _rlp_list([log_predicate]), - ] - ) - return bytes([0x02]) + definition - - -# --------------------------------------------------------------------------- -# Contract deployment + interaction via cast inside the ethereum container -# --------------------------------------------------------------------------- - -RPC = "http://127.0.0.1:8545" - - -def cast(*args: str, capture: bool = True) -> str: - """Run a cast subcommand inside the running ethereum container.""" - result = utils.run( - ["docker", "compose", "exec", "-T", "ethereum", "cast", *args], - capture_output=capture, - ) - return result.stdout.strip() if capture else "" - - -def cast_send(*args: str) -> dict: - """Broadcast a transaction and return the parsed JSON receipt.""" - raw = cast( - "send", - "--json", - "--rpc-url", - RPC, - "--private-key", - os.environ["DEPLOY_KEY"], - *args, - ) - return json.loads(raw) - - - -def call_register( - registry_addr: str, - keyper_config_index: int, - identity_prefix_hex: str, - trigger_def_hex: str, - ttl: int, -) -> None: - cast_send( - registry_addr, - "register(uint64,bytes32,bytes,uint64)", - str(keyper_config_index), - f"0x{identity_prefix_hex}", - f"0x{trigger_def_hex}", - str(ttl), - ) - - -def call_trigger(helper_addr: str, topic2_hex: str) -> None: - zero32 = "0x" + "0" * 64 - cast_send( - helper_addr, - "trigger(uint64,bytes32,bytes32,bytes32)", - "0", - f"0x{topic2_hex}", - zero32, - zero32, - ) - - -def wait_for_event_decryption_key( - keyper_config_index: int, - identity_prefix_hex: str, - poll_interval: float = 1.0, - timeout: float = 120.0, -) -> str: - """ - Poll until the decryption key for the event trigger with the given identity_prefix - is available. Returns the hex-encoded identity (epoch_id). - """ - deadline = time.time() + timeout - while True: - if time.time() > deadline: - raise SystemExit( - f"Timeout waiting for decryption key for " - f"identity_prefix={identity_prefix_hex}, eon={keyper_config_index}" - ) - - 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_hex}' " - "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: identity_prefix={identity_prefix_hex} → epoch_id={identity}" - ) - return identity - - time.sleep(poll_interval) +NUM_TRIGGERS = 3 +POLL_INTERVAL = float(os.environ.get("DECRYPTION_KEY_POLL_INTERVAL", "1")) def assert_no_batching( @@ -189,12 +19,7 @@ def assert_no_batching( poll_interval: float = 1.0, timeout: float = 30.0, ) -> None: - """ - Assert that decryption_signatures contains `num_triggers` distinct identities_hash - values for the given eon. With no batching each DecryptionTrigger carries exactly - one identity, so its identities_hash = keccak256(identity). With batching all N - identities would share one hash. - """ + """Assert each DecryptionTrigger carries exactly one identity (no batching).""" deadline = time.time() + timeout while True: count_str = utils.query_keyper_db( @@ -218,119 +43,65 @@ def assert_no_batching( time.sleep(poll_interval) -# --------------------------------------------------------------------------- -# Main -# --------------------------------------------------------------------------- - - -def get_registry_address(data_dir: Path) -> str: - broadcast = json.loads( - ( - data_dir - / "contracts" - / "broadcast" - / utils.DEPLOYMENT_SCRIPTS["service"] - / os.environ["ETHEREUM_CHAIN_ID"] - / "run-latest.json" - ).read_text() - ) - addr = utils.get_uups_proxy_address(broadcast, "ShutterEventTriggerRegistryV1") - if not addr: - raise SystemExit( - "ShutterEventTriggerRegistryV1 proxy not found in deployment artifacts" - ) - return addr - - -def get_helper_address(data_dir: Path) -> str: - addr_path = data_dir / "test-helper-addr.json" - return json.loads(addr_path.read_text())["address"] - - -def main() -> None: - 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") - - data_dir = Path(os.environ["DATA_DIR"]) - poll_interval = float(os.environ.get("DECRYPTION_KEY_POLL_INTERVAL", "1")) - num_triggers = 3 +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") - # --- Resolve eon and keyper_config_index ----------------------------------- - # ShutterEventTriggerRegistryV1.register() takes the *keyper_config_index*, - # not the eon number. We obtain both from the eons table and use accordingly. - eon_str = utils.query_keyper_db(0, "SELECT max(eon) FROM eons") - if not eon_str: - raise SystemExit("No eon found in keyper DB — has DKG completed?") - eon_number = int(eon_str) +# 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") - keyper_config_index_str = utils.query_keyper_db( - 0, f"SELECT keyper_config_index FROM eons WHERE eon = {eon_number} LIMIT 1" +# 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, + ], ) - if not keyper_config_index_str: - raise SystemExit(f"No keyper_config_index found for eon={eon_number}") - keyper_config_index = int(keyper_config_index_str) - - print(f"Using eon={eon_number}, keyper_config_index={keyper_config_index}") - - # --- Read EventTriggerTestHelper address (deployed by deploy-test-helper) --- - helper_addr = get_helper_address(data_dir) - print(f"Using EventTriggerTestHelper at {helper_addr}") - - # --- Build the trigger definition ----------------------------------------- - # Match Trigger(…, topic2=trigger_topic, …) events emitted by helper_addr. - # topic[2] is the second indexed parameter of the Trigger event (bytes32). - trigger_topic = secrets.token_bytes(32) - trigger_topic_hex = trigger_topic.hex() - - helper_addr_bytes = bytes.fromhex(helper_addr.removeprefix("0x").zfill(40)) - trigger_def = encode_trigger_def( - contract_addr=helper_addr_bytes, - topic_index=2, # topic[2] = topic2 (bytes32 indexed) - topic_value=trigger_topic, + 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]}…" ) - trigger_def_hex = trigger_def.hex() - print(f" EventTriggerDefinition: 0x{trigger_def_hex}") - - # --- Generate identity prefixes (one per trigger) ------------------------- - identity_prefixes = [secrets.token_bytes(32).hex() for _ in range(num_triggers)] - - # --- Register N event triggers -------------------------------------------- - registry_addr = get_registry_address(data_dir) - ttl = 10_000 # blocks until expiry - - print(f"Registering {num_triggers} event triggers on registry {registry_addr}…") - for ip_hex in identity_prefixes: - call_register(registry_addr, keyper_config_index, ip_hex, trigger_def_hex, ttl) - print(f" Registered trigger with identity_prefix=0x{ip_hex[:8]}…") - - # Give keypers time to sync the EventTriggerRegistered events before we fire. - # With ETHEREUM_BLOCK_TIME=1 a couple of seconds is sufficient. - 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 3 registered definitions will match ----- - print(f"Firing Trigger event with topic2=0x{trigger_topic_hex[:8]}…") - call_trigger(helper_addr, trigger_topic_hex) +keyper_config_index = registrations[0]["keyper_config_index"] - # --- Wait for decryption keys --------------------------------------------- - print("Waiting for decryption keys…") - for ip_hex in identity_prefixes: - wait_for_event_decryption_key(keyper_config_index, ip_hex, poll_interval) +# 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) - # --- Assert no batching --------------------------------------------------- - print( - "Asserting no-batching (each trigger must produce a distinct identities_hash)…" - ) - assert_no_batching(keyper_config_index, num_triggers) +# 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]) - print( - f"\nAll {num_triggers} event-based decryption triggers handled without batching. ✓" +# 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) -if __name__ == "__main__": - main() +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 00000000..5364f551 --- /dev/null +++ b/mise-test-setup/mise-tasks/trigger-event @@ -0,0 +1,63 @@ +#!/usr/bin/env -S uv run --script +#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)" + +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/utils.py b/mise-test-setup/mise-tasks/utils.py index ef07fe9e..4f5d1ab9 100644 --- a/mise-test-setup/mise-tasks/utils.py +++ b/mise-test-setup/mise-tasks/utils.py @@ -22,6 +22,13 @@ 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 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 00000000..a908659b --- /dev/null +++ b/mise-test-setup/mise-tasks/wait-for-event-decryption-key @@ -0,0 +1,55 @@ +#!/usr/bin/env -S uv run --script +#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" + +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) From 82ae21da0e8ff538fc903ed783b9f46965f06536 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Wed, 27 May 2026 10:25:48 +0200 Subject: [PATCH 07/11] Apply black & isort --- .../mise-tasks/add-initial-keyper-set | 3 +- mise-test-setup/mise-tasks/add-keyper-set | 11 ++- mise-test-setup/mise-tasks/build-image | 2 + mise-test-setup/mise-tasks/clean | 2 + mise-test-setup/mise-tasks/deploy | 6 +- mise-test-setup/mise-tasks/down | 2 + mise-test-setup/mise-tasks/gen-compose | 5 +- mise-test-setup/mise-tasks/gen-keyper-configs | 82 +++++++++++++++---- mise-test-setup/mise-tasks/init-chain-nodes | 11 ++- mise-test-setup/mise-tasks/init-chain-seed | 3 +- mise-test-setup/mise-tasks/init-keyper-dbs | 4 +- mise-test-setup/mise-tasks/patch-genesis | 6 +- .../mise-tasks/submit-event-registration | 7 +- .../mise-tasks/submit-identity-registration | 6 +- mise-test-setup/mise-tasks/test-decryption | 3 +- .../mise-tasks/test-event-decryption | 6 +- mise-test-setup/mise-tasks/trigger-event | 2 + mise-test-setup/mise-tasks/up | 4 +- mise-test-setup/mise-tasks/up-db | 3 +- mise-test-setup/mise-tasks/up-ethereum | 3 +- mise-test-setup/mise-tasks/utils.py | 16 ++-- .../mise-tasks/wait-for-decryption-key | 3 +- mise-test-setup/mise-tasks/wait-for-dkg | 3 +- .../mise-tasks/wait-for-event-decryption-key | 2 + .../mise-tasks/wait-for-initial-dkg | 2 + 25 files changed, 148 insertions(+), 49 deletions(-) diff --git a/mise-test-setup/mise-tasks/add-initial-keyper-set b/mise-test-setup/mise-tasks/add-initial-keyper-set index b7b84ae8..9d7ec099 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 ec599106..dd0ef2f1 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 69bc9554..e1a91f67 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 d15c60ad..04efec2f 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 a69ec8e8..d95eaa58 100755 --- a/mise-test-setup/mise-tasks/deploy +++ b/mise-test-setup/mise-tasks/deploy @@ -1,6 +1,8 @@ #!/usr/bin/env -S uv run --script +# fmt: off #MISE description="Deploy contracts if not already deployed" #MISE depends=["up-ethereum"] +# fmt: on import json import os @@ -86,7 +88,9 @@ 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") diff --git a/mise-test-setup/mise-tasks/down b/mise-test-setup/mise-tasks/down index 03da7cb8..b7ae230c 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 6c3f3078..71ff71e8 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 a9f91a0f..bfcf92b9 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,25 +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")) - event_trigger_registry = utils.get_uups_proxy_address(deployment_run, "ShutterEventTriggerRegistryV1") + 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) + 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 61d42f7c..ffc29044 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 27cf8848..60a827ca 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 df45896f..baf8ba23 100755 --- a/mise-test-setup/mise-tasks/init-keyper-dbs +++ b/mise-test-setup/mise-tasks/init-keyper-dbs @@ -1,6 +1,8 @@ #!/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 @@ -81,7 +83,7 @@ def drop_db(database_name: str) -> None: "-U", "postgres", "-c", - f"DROP DATABASE IF EXISTS \"{database_name}\"", + f'DROP DATABASE IF EXISTS "{database_name}"', ] ) diff --git a/mise-test-setup/mise-tasks/patch-genesis b/mise-test-setup/mise-tasks/patch-genesis index 1bd6856d..72af2d1b 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 index a18d43c3..6cb68594 100755 --- a/mise-test-setup/mise-tasks/submit-event-registration +++ b/mise-test-setup/mise-tasks/submit-event-registration @@ -1,4 +1,5 @@ #!/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)" @@ -7,6 +8,7 @@ #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 @@ -15,7 +17,6 @@ from pathlib import Path import utils - # --------------------------------------------------------------------------- # RLP helpers # --------------------------------------------------------------------------- @@ -89,7 +90,9 @@ broadcast = json.loads( ) registry_addr = utils.get_uups_proxy_address(broadcast, "ShutterEventTriggerRegistryV1") if not registry_addr: - raise SystemExit("ShutterEventTriggerRegistryV1 proxy not found in deployment artifacts") + 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: diff --git a/mise-test-setup/mise-tasks/submit-identity-registration b/mise-test-setup/mise-tasks/submit-identity-registration index 892caa7c..17a17d07 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 4b7c474e..d357f55b 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 index ab79b597..a2621ddd 100755 --- a/mise-test-setup/mise-tasks/test-event-decryption +++ b/mise-test-setup/mise-tasks/test-event-decryption @@ -1,6 +1,8 @@ #!/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 @@ -97,9 +99,7 @@ for reg in registrations: ) # Assert that no batching occurred. -print( - "Asserting no-batching (each trigger must produce a distinct identities_hash)…" -) +print("Asserting no-batching (each trigger must produce a distinct identities_hash)…") assert_no_batching(keyper_config_index, NUM_TRIGGERS, POLL_INTERVAL) print( diff --git a/mise-test-setup/mise-tasks/trigger-event b/mise-test-setup/mise-tasks/trigger-event index 5364f551..881be777 100755 --- a/mise-test-setup/mise-tasks/trigger-event +++ b/mise-test-setup/mise-tasks/trigger-event @@ -1,8 +1,10 @@ #!/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 diff --git a/mise-test-setup/mise-tasks/up b/mise-test-setup/mise-tasks/up index 3cda696a..6e8299a2 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 3596d2d3..5d0f3063 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 74e06e25..02555e9b 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 4f5d1ab9..1f388640 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", @@ -24,9 +23,7 @@ def run( 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 - ) + return subprocess.run(command, check=True, text=True, stdout=subprocess.PIPE) def wait_for_service_health(service: str, *, timeout_seconds: float = 30.0) -> None: @@ -70,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]: @@ -151,9 +153,7 @@ def get_uups_proxy_address( continue if candidate.get("contractName") != "ERC1967Proxy": continue - input_data = ( - candidate.get("transaction", {}).get("input") or "" - ).lower() + 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: diff --git a/mise-test-setup/mise-tasks/wait-for-decryption-key b/mise-test-setup/mise-tasks/wait-for-decryption-key index 20af1ea5..dd000978 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 b0699526..78c4a021 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 index a908659b..eafb7dc2 100755 --- a/mise-test-setup/mise-tasks/wait-for-event-decryption-key +++ b/mise-test-setup/mise-tasks/wait-for-event-decryption-key @@ -1,7 +1,9 @@ #!/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 diff --git a/mise-test-setup/mise-tasks/wait-for-initial-dkg b/mise-test-setup/mise-tasks/wait-for-initial-dkg index f56da942..f1751efd 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 From f5836c33898868884fd93458b540bef0ab483167 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Mon, 29 Jun 2026 15:03:25 +0200 Subject: [PATCH 08/11] B4 optimization - New() precomputes the union of contract addresses, the union of event topic0 hashes, and a (address, topic) -> *EventType dispatch table. - syncAllInRange() now issues a single FilterLogs covering every registered (address, topic), then dispatches each returned log to its EventType via the precomputed table. Logs that match the query but have no registered handler (cross-product hits) are skipped. - Removed the per-event-type fan-out (syncSingleInRange) along with the sync.Mutex and golang.org/x/sync/errgroup dependencies. For the snapshot keyper (the only current caller via chainobserver), this drops per-block eth_getLogs requests from N (one per event type) to 1. Behavior is preserved: the same logs reach Next() in the same block + log-index order. --- .../medley/eventsyncer/eventsyncer.go | 118 ++++++++++-------- 1 file changed, 66 insertions(+), 52 deletions(-) diff --git a/rolling-shutter/medley/eventsyncer/eventsyncer.go b/rolling-shutter/medley/eventsyncer/eventsyncer.go index c72b3582..9c0a4e9c 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,6 @@ import ( "github.com/ethereum/go-ethereum/ethclient" "github.com/jackc/pgx/v4" "github.com/pkg/errors" - "golang.org/x/sync/errgroup" "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/db" "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/retry" @@ -84,6 +82,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 +98,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 +110,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 +136,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), } @@ -208,73 +239,56 @@ func (s *EventSyncer) sync(ctx context.Context) error { } } -// 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 len(s.Events) == 0 { + return nil, nil } - if err := errorgroup.Wait(); err != nil { - return nil, err - } - - 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 } From 88fe40a609b53a6e773b86023d719f4320194463 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Mon, 29 Jun 2026 16:14:11 +0200 Subject: [PATCH 09/11] S2 optimization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit keyperimpl/shutterservice/eventprocessor.go — new interface - FilterCriteria struct carries Addresses []common.Address and Topics0 []common.Hash (empty Topics0 means "any topic" for that processor). - EventProcessor interface replaces the old FetchEvents(ctx, start, end) with FilterCriteria(ctx, start, end) + ParseEvents(ctx, start, end, logs). keyperimpl/shutterservice/multieventsyncer.go — combined fetch - gatherFilterCriteria asks each processor for its criteria, builds the unioned address list and topic[0] list, captures per-processor scopes for routing. - syncRange now issues at most one FilterLogs(addresses, [topic0Union]) per range (skipped entirely if no processor wants any addresses), then routes the matching subset of logs to each processor's ParseEvents. If any processor declines to constrain topic[0], the combined query drops topic filtering. keyperimpl/shutterservice/eventtriggerregisteredprocessor.go - Constructor now takes the contract address and parses the bindings' ABI to extract the EventTriggerRegistered topic. Returns an error from construction. - FilterCriteria returns {[address], [topic]}. - ParseEvents calls Contract.ParseEventTriggerRegistered(log) (abigen helper) per log. keyperimpl/shutterservice/triggerprocessor.go - New activeTriggers helper reads + parses active trigger definitions from the DB. - FilterCriteria unions the addresses and topic[0] hashes across all active triggers; sets anyTopic when a trigger leaves topic[0] unconstrained. - ParseEvents iterates each trigger over the pre-filtered logs and emits TriggerEvents for ones that pass def.Match and aren't expired. keyperimpl/shutterservice/keyper.go - Updated NewEventTriggerRegisteredEventProcessor call to pass the address and handle the new error return. Steady-state RPC count per block range with N active triggers drops from 1 (FilterEventTriggerRegistered) + N (per-trigger FilterLogs) to 1 combined FilterLogs. The HeaderByNumber(end) is still there (that's S3, not in scope). Note: there's one DB cost trade-off worth flagging — GetActiveEventTriggerRegisteredEvents is now queried twice per range inside TriggerProcessor (once in FilterCriteria, once in ParseEvents). S8 in the doc covers the cache that would eliminate this. --- .../shutterservice/eventprocessor.go | 27 +++- .../eventtriggerregisteredprocessor.go | 52 ++++--- .../keyperimpl/shutterservice/keyper.go | 6 +- .../shutterservice/multieventsyncer.go | 134 +++++++++++++++-- .../shutterservice/triggerprocessor.go | 135 ++++++++++++------ 5 files changed, 278 insertions(+), 76 deletions(-) diff --git a/rolling-shutter/keyperimpl/shutterservice/eventprocessor.go b/rolling-shutter/keyperimpl/shutterservice/eventprocessor.go index 9ee8225a..5b540751 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 0f1b42d7..a4822723 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 4d5960c4..cf41dc5d 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 1c27cdd2..8714136d 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/triggerprocessor.go b/rolling-shutter/keyperimpl/shutterservice/triggerprocessor.go index bc0d1782..525a3436 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 { From 60486ad0fe0ba03021c6576c74ec68f2d491ce25 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Thu, 2 Jul 2026 21:54:49 +0200 Subject: [PATCH 10/11] C1: drive EventSyncer off SubscribeNewHead Replaces the 2s eth_blockNumber poll with a SubscribeNewHead subscription. On startup we still ask HeaderByNumber(nil) once to know the initial tip so historical catchup can drain before any live head arrives; from then on the subscription is the only signal for new work. If the transport rejects Subscribe (e.g. HTTP-only ethclient), the syncer falls back to the previous polling loop. Steady state per-minute eth_blockNumber calls drop from ~30 to 0. --- .../medley/eventsyncer/eventsyncer.go | 113 ++++++++++++++---- 1 file changed, 90 insertions(+), 23 deletions(-) diff --git a/rolling-shutter/medley/eventsyncer/eventsyncer.go b/rolling-shutter/medley/eventsyncer/eventsyncer.go index 9c0a4e9c..aef4bbc6 100644 --- a/rolling-shutter/medley/eventsyncer/eventsyncer.go +++ b/rolling-shutter/medley/eventsyncer/eventsyncer.go @@ -15,6 +15,7 @@ import ( "github.com/ethereum/go-ethereum/ethclient" "github.com/jackc/pgx/v4" "github.com/pkg/errors" + "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" @@ -196,47 +197,113 @@ 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. It issues From 862ac2bd344accef85cf4f95305bf493d2f06264 Mon Sep 17 00:00:00 2001 From: Konrad Feldmeier Date: Thu, 2 Jul 2026 21:55:18 +0200 Subject: [PATCH 11/11] C4: switch sequencer + validator syncers to Watch* subscriptions Both syncers gain a Start(ctx, runner) method that runs a one-off Sync for historical catchup and then subscribes via WatchTransactionSubmitted / WatchUpdated. A goroutine drains the subscription, persisting each event individually (or logging Removed for reorged blocks). Sync itself is retained for the catchup path and for tests but is no longer invoked per-new-block. processNewBlock loses its two per-block Sync calls entirely; steady state per-block eth_getLogs from the two syncers drops from 2 to 0. Keyper.Start now registers both syncers alongside the other services via runner.StartService so their lifetimes match the process. --- rolling-shutter/keyperimpl/gnosis/keyper.go | 36 +++----- rolling-shutter/keyperimpl/gnosis/newblock.go | 14 ++- .../keyperimpl/gnosis/sequencersyncer.go | 79 +++++++++++++++++ .../keyperimpl/gnosis/validatorsyncer.go | 86 +++++++++++++++++++ 4 files changed, 184 insertions(+), 31 deletions(-) diff --git a/rolling-shutter/keyperimpl/gnosis/keyper.go b/rolling-shutter/keyperimpl/gnosis/keyper.go index 70d8fbd8..2e0d802a 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 8a1af240..7f5729af 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 c9c3313c..0878007f 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 7dadb2ab..2e5aa2f6 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)