feat: duplicates limit evaluation

tests/integration/snapshots/snap_test_snapshots.py

@@ -104,7 +104,7 @@
         "tx_b": "0x112ac55e0122204165ab94bad060ffcee026e4db572f74b3325d56bd0950ade1",
     },
     {
-        "filter": None,
+        "filter": "limited_collisions_per_address",
         "tx_a": "0x775232180c49821d4208b3c5470d6367de5b96810c79ae0993aeb98ada762ee8",
         "tx_b": "0x1ea1709059406a15686edef98de051fdfb5e854cc0991687b9573cba9005b021",
     },
@@ -130,6 +130,17 @@
     },
 ]
 
+snapshots["test_tod_attack_miner_evaluation_duplicates_limit representatives"] = [
+    (
+        (
+            "0x775232180c49821d4208b3c5470d6367de5b96810c79ae0993aeb98ada762ee8",
+            "0x1ea1709059406a15686edef98de051fdfb5e854cc0991687b9573cba9005b021",
+        ),
+        False,
+        set([]),
+    )
+]
+
 snapshots[
     "test_tod_attack_miner_evaluation_indirect_dependencies indirect dependencies"
 ] = [

tests/integration/test_snapshots.py

@@ -92,7 +92,8 @@ def test_tod_attack_miner_evaluation(
     miner.fetch(block_range.start, block_range.end)
     miner.find_collisions()
     results = miner.evaluate_candidates(
-        get_filters_except_duplicate_limits(3), evaluation_candidates
+        get_filters_except_duplicate_limits(3) + get_filters_duplicate_limits(10),
+        evaluation_candidates,
     )
 
     snapshot.assert_match(results, "evaluation results")
@@ -113,3 +114,22 @@ def test_tod_attack_miner_evaluation_indirect_dependencies(
     )
 
     snapshot.assert_match(results, "indirect dependencies")
+
+
+@pytest.mark.vcr
+def test_tod_attack_miner_evaluation_duplicates_limit(
+    postgresql: Connection, snapshot: PyTestSnapshotTest
+):
+    block_range = BlockRange(19895500, 19895504)
+
+    miner = Miner(RPC(test_provider_url), DB(postgresql))
+
+    miner.fetch(block_range.start, block_range.end)
+    miner.find_collisions()
+    results = miner.get_limit_representatives(
+        get_filters_except_duplicate_limits(3),
+        get_filters_duplicate_limits(10),
+        evaluation_candidates,
+    )
+
+    snapshot.assert_match(results, "representatives")

tod_attack_miner/cli.py

@@ -27,12 +27,6 @@ def main():
     parser.add_argument("--archive-node-provider", default="http://localhost:8124/eth")
     parser.add_argument("--from-block", default=19895500)
     parser.add_argument("--to-block", default=19895504)
-    parser.add_argument(
-        "--window-size",
-        type=int,
-        default=None,
-        help="If passed, filter TOD candidates that are {window-size} or more blocks apart",
-    )
     parser.add_argument(
         "--reset-db",
         action="store_true",
@@ -54,6 +48,11 @@ def main():
         action="store_true",
         help="When evaluating candidates, stop before the indirect dependencies filter and output indirect dependency paths",
     )
+    parser.add_argument(
+        "--extract-limit-representatives",
+        action="store_true",
+        help="When evaluating candidates, stop before the duplicate limit and find non-filtered evaluation candidates that represent the candidates filtered because of duplication",
+    )
     parser.add_argument("--postgres-user", type=str, default="postgres")
     parser.add_argument("--postgres-password", type=str, default="password")
     parser.add_argument("--postgres-host", type=str, default="localhost")
@@ -70,6 +69,7 @@ def main():
     evaluate_candidates_csv: Path | None = args.evaluate_candidates_csv
     evaluation_results_csv: Path = args.evaluation_result_csv
     extract_indirect_dependencies: bool = args.extract_indirect_dependencies
+    extract_limit_repesentatives: bool = args.extract_limit_representatives
 
     with psycopg.connect(
         f"user={args.postgres_user} password={args.postgres_password} host={args.postgres_host} port={args.postgres_port}"
@@ -94,6 +94,19 @@ def main():
                     csv_writer = csv.writer(results_csv_file)
                     csv_writer.writerow(("tx_a", "tx_b", "dependency_path"))
                     csv_writer.writerows(results)
+                elif extract_limit_repesentatives:
+                    filters = get_filters_except_duplicate_limits(25)
+                    duplicate_filters = get_filters_duplicate_limits(10)
+                    results = miner.get_limit_representatives(
+                        filters, duplicate_filters, candidates
+                    )
+                    rows = [
+                        (tx_a, tx_b, str(x), "|".join([f"{a}-{b}" for a, b in y]))
+                        for (tx_a, tx_b), x, y in results
+                    ]
+                    csv_writer = csv.writer(results_csv_file)
+                    csv_writer.writerow(("tx_a", "tx_b", "covered", "representatives"))
+                    csv_writer.writerows(rows)
                 else:
                     filters = get_filters_except_duplicate_limits(
                         25

tod_attack_miner/db/filters.py

@@ -1,4 +1,5 @@
-from typing import Callable, Iterable
+from collections import defaultdict
+from typing import Callable, Iterable, Mapping
 import psycopg
 import psycopg.sql
 from tod_attack_miner.db.db import DB
@@ -139,6 +140,28 @@ def get_evaluation_indirect_dependencies_quick(
         return cursor.execute(sql).fetchall()
 
 
+def get_remaining_evaluation_candidate_collisions(
+    db: DB,
+) -> Mapping[tuple[str, str], set[tuple[str, str]]]:
+    """Return a mapping from evaluation candidates to collisions (type, key)"""
+
+    sql = """
+        SELECT c.tx_write_hash, c.tx_access_hash, type, key
+        FROM collisions c
+        INNER JOIN evaluation_candidates ec
+        ON c.tx_write_hash = ec.tx_write_hash AND c.tx_access_hash = ec.tx_access_hash
+        WHERE filtered_by = ''
+    """
+    with db._con.cursor() as cursor:
+        collisions: list[tuple[str, str, str, str]] = cursor.execute(sql).fetchall()
+        mapping: dict[tuple[str, str], set[tuple[str, str]]] = defaultdict(set)
+        for tx_a, tx_b, type, key in collisions:
+            mapping[(tx_a, tx_b)].add((type, key))
+
+        # copy so we don't have a defaultdict outside, which is error-prone
+        return {**mapping}
+
+
 def create_limit_collisions_per_address(limit: int):
     def limit_collisions_per_address(db: DB):
         sql = f"""

tod_attack_miner/miner/miner.py

@@ -1,6 +1,9 @@
 from typing import Callable, Iterable, Sequence
 from tod_attack_miner.db.db import DB, Candidate, EvaluationCandidate
-from tod_attack_miner.db.filters import get_evaluation_indirect_dependencies_quick
+from tod_attack_miner.db.filters import (
+    get_evaluation_indirect_dependencies_quick,
+    get_remaining_evaluation_candidate_collisions,
+)
 from tod_attack_miner.fetcher.fetcher import BlockRange, fetch_block_range
 from tod_attack_miner.rpc.rpc import RPC
 
@@ -56,6 +59,42 @@ def get_indirect_dependencies(
         indirect_dependencies = get_evaluation_indirect_dependencies_quick(self.db)
         return sorted(indirect_dependencies)
 
+    def get_limit_representatives(
+        self,
+        filters: Filters,
+        duplicate_filters: Filters,
+        candidates: Iterable[tuple[str, str]],
+    ) -> Iterable[tuple[tuple[str, str], bool, Iterable[tuple[str, str]]]]:
+        self.evaluate_candidates(filters, candidates)
+        collisions = get_remaining_evaluation_candidate_collisions(self.db)
+        for name, filter_duplicates in duplicate_filters:
+            filter_duplicates(self.db)
+            self.db.update_filtered_evaluation_candidates(name)
+        remaining_candidates = self.db.get_evaluation_candidates()
+        covered_collisions: dict[tuple[str, str], tuple[str, str]] = {}
+        for candidate in remaining_candidates:
+            tx_a, tx_b = candidate["tx_a"], candidate["tx_b"]
+            if (tx_a, tx_b) not in collisions:
+                continue
+            # only include non-filtered candidates for the covered collisions
+            if candidate["filter"] is not None:
+                continue
+            for coll in collisions[(tx_a, tx_b)]:
+                covered_collisions[coll] = (tx_a, tx_b)
+
+        result = []
+        for candidate, colls in collisions.items():
+            completeley_represented = True
+            representatives: set[tuple[str, str]] = set()
+            for coll in colls:
+                if coll in covered_collisions:
+                    representatives.add(covered_collisions[coll])
+                else:
+                    completeley_represented = False
+            result.append((candidate, completeley_represented, representatives))
+
+        return sorted(result)
+
     def count_candidates(self) -> int:
         return self.db.count_candidates()