Dealing with duplicate hashes
If there is a mismatch between when the originator VASP appends the transaction hash to the travel rule message they have already sent and when you call txNotify, you might end up with two travel rule messages where the transaction hash is the same for both, but the status is different. Usually, one would be INCOMPLETE while the other is SENT, ACK or ACCEPTED.
If you want to clean up these INCOMPLETE ones, you can follow the steps below:
- Make sure you are consuming the
notification.transactionUpdatedwebhook on your side to create a database of incoming TR messages. - Create a script that identifies and extracts duplicate transaction hashes from that database.
- Use a similar script to the ones below to confirm that they are duplicates. You can either call the API to fetch the details for the records or directly use the data in your transactionUpdated database to compare the fields.
- If duplicates are found, cancel them by calling the txCancel API.
Compare using API
#!/usr/bin/env python3
import os
import sys
import json
import time
import argparse
import logging
import requests
TARGET_OK_STATUSES = {"SENT","ACK","ACCEPTED"}
def setup_logging(verbose=False):
level = logging.DEBUG if verbose else logging.INFO
logging.basicConfig(
level=level,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
def env(name, default=None):
val = os.environ.get(name)
return val if val is not None and val != "" else default
API_BASE = env("NOTABENE_API_BASE", "https://api.notabene.dev")
VASP_DID = env("NOTABENE_VASP_DID", "did:ethr:0xd4bd902ec78578f33a20ff601504d2ab324cfab9")
TOKEN = env("NOTABENE_API_TOKEN")
def require_token():
if not TOKEN:
raise SystemExit("NOTABENE_API_TOKEN is not set. Export it before running.")
def api_get(path, params):
url = f"{API_BASE}{path}"
headers = {"Authorization": f"Bearer {TOKEN}"}
logging.debug("GET %s params=%s", url, params)
r = requests.get(url, headers=headers, params=params, timeout=30)
if r.status_code != 200:
logging.error("GET %s failed: %s %s", url, r.status_code, r.text)
raise RuntimeError(f"GET {url} failed: {r.status_code} {r.text}")
return r.json()
def api_post(path, body):
url = f"{API_BASE}{path}"
headers = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}
logging.debug("POST %s body=%s", url, body)
r = requests.post(url, headers=headers, json=body, timeout=30)
return r.status_code, r.text
def paginate_tx_list(q_hash, results_per_page=50, page_start=0, max_pages=50, sleep_sec=0.0):
"""Yield all transactions matching a given q hash, across pages."""
page = page_start
total_pages = None
while True:
params = {
"vaspDID": VASP_DID,
"resultsPerPage": results_per_page,
"page": page,
"q": q_hash,
"sort": "updatedAt:DESC",
}
data = api_get("/tx/list", params)
txs = data.get("transactions") or []
logging.info("Fetched %d transactions for hash %s (page %d)", len(txs), q_hash, page)
for t in txs:
yield t
if total_pages is None:
total_pages = int(data.get("paginateInfo", {}).get("totalPages", 1))
page += 1
if page >= total_pages or page - page_start >= max_pages:
break
if sleep_sec > 0:
time.sleep(sleep_sec)
def extract_fields(tx):
asset = (tx.get("transactionAsset") or "").strip()
amount = str(tx.get("transactionAmount") or "").strip()
bc = tx.get("transactionBlockchainInfo") or {}
txhash = (bc.get("txHash") or "").strip()
dest = (bc.get("destination") or "").strip()
status = (tx.get("status") or "").strip().upper()
tid = tx.get("id")
updated_at = tx.get("updatedAt")
return {
"id": tid,
"status": status,
"asset": asset,
"amount": amount,
"txHash": txhash,
"destination": dest,
"updatedAt": updated_at,
"raw": tx,
}
def same_key(a, b):
return (
a["asset"] == b["asset"] and
a["amount"] == b["amount"] and
a["txHash"] == b["txHash"] and
a["destination"] == b["destination"]
)
def cancel_tx(incomplete_id, other_id, dry_run=False):
body = {
"id": incomplete_id,
"reason": f"This is a duplicate of {other_id}",
}
if dry_run:
logging.info("[DRY-RUN] Would POST /tx/cancel %s", body)
return 200, "DRY-RUN"
status, text = api_post("/tx/cancel", body)
if 200 <= status < 300:
logging.info("[OK] Cancelled %s (duplicate of %s)", incomplete_id, other_id)
else:
logging.error("[ERROR] Cancel %s failed: HTTP %s %s", incomplete_id, status, text)
return status, text
def process_hash(q_hash, dry_run=False):
logging.info("Processing hash %s", q_hash)
txs = [extract_fields(t) for t in paginate_tx_list(q_hash)]
txs = [t for t in txs if t["txHash"] == q_hash]
logging.debug("After filtering exact txHash, %d remain", len(txs))
if len(txs) <= 1:
logging.info("Hash %s: %d match(es) -> nothing to do.", q_hash, len(txs))
return
incompletes = [t for t in txs if t["status"] == "INCOMPLETE"]
others = [t for t in txs if t["status"] in TARGET_OK_STATUSES]
logging.info("Hash %s: %d INCOMPLETE, %d others", q_hash, len(incompletes), len(others))
others_sorted = sorted(others, key=lambda x: (x["updatedAt"] or ""), reverse=True)
if not incompletes or not others_sorted:
logging.info("Hash %s: no INCOMPLETE/eligible counterpart. Skipping.", q_hash)
return
for inc in incompletes:
counterpart = next((o for o in others_sorted if same_key(inc, o)), None)
if counterpart:
logging.debug("INCOMPLETE %s matches counterpart %s", inc["id"], counterpart["id"])
cancel_tx(inc["id"], counterpart["id"], dry_run=dry_run)
else:
logging.info("Hash %s: INCOMPLETE %s has no exact matching counterpart.", q_hash, inc['id'])
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--hashes", nargs="*", help="List of blockchain tx hashes to check")
ap.add_argument("--hash-file", help="File with one hash per line")
ap.add_argument("--dry-run", action="store_true", help="Don't call txCancel; just log actions")
ap.add_argument("-v","--verbose", action="store_true", help="Verbose (debug) logging")
args = ap.parse_args()
setup_logging(args.verbose)
require_token()
hashes = []
if args.hashes:
hashes.extend(args.hashes)
if args.hash_file:
with open(args.hash_file, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line:
hashes.append(line)
if not hashes:
logging.error("Provide --hashes ... or --hash-file path")
sys.exit(1)
for h in hashes:
try:
process_hash(h, dry_run=args.dry_run)
except Exception as e:
logging.exception("Error processing hash %s", h)
if __name__ == "__main__":
main()
Run command:
$env:NOTABENE_API_TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6IklkYzY4WTU5NU9xVDZuaTA3SHYzYiJ9..."
python hash_check_and_cancel_with_logging.py --hash-file hashes.txt --dry-run
Compare using database
#!/usr/bin/env python3
"""
Given:
- A CSV "database" containing transactions with fields similar to:
id, status, transactionAsset, transactionAmount, transactionBlockchainInfo.txHash,
transactionBlockchainInfo.destination, updatedAt (names can vary)
- A list of blockchain hashes (via --hashes or --hash-file)
For each hash:
- Look up rows in the CSV where txHash == <hash>
- If more than one row matches:
* Compare (transactionAsset, transactionAmount, txHash, destination)
* If an INCOMPLETE row has an exact counterpart with status in {"SENT","ACK","ACCEPTED"},
call txCancel on the INCOMPLETE row with reason "This is a duplicate of <OTHER_ID>".
Features:
- Column auto-detection with overrides
- Case-insensitive comparison for destination
- Logging and --dry-run
- Env vars:
NOTABENE_API_TOKEN (Bearer token; required unless --dry-run)
NOTABENE_API_BASE (default: https://api.notabene.dev)
Usage:
python hash_check_from_db_and_cancel.py --csv path/to/db.csv --hashes 0xabc 0xdef --dry-run -v
python hash_check_from_db_and_cancel.py --csv path/to/db.csv --hash-file hashes.txt
Column overrides (if your headers differ):
--db-id idcol --db-status statuscol --db-asset assetcol --db-amount amountcol
--db-hash hashcol --db-destination destcol --db-updated updatedAtCol
"""
import os
import sys
import argparse
import logging
import json
import pandas as pd
import requests
TARGET_OK_STATUSES = {"SENT","ACK","ACCEPTED"}
def setup_logging(verbose=False):
level = logging.DEBUG if verbose else logging.INFO
logging.basicConfig(
level=level,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
def env(name, default=None):
val = os.environ.get(name)
return val if val is not None and val != "" else default
API_BASE = env("NOTABENE_API_BASE", "https://api.notabene.dev")
TOKEN = env("NOTABENE_API_TOKEN")
def require_token(dry_run=False):
if dry_run:
return
if not TOKEN:
raise SystemExit("NOTABENE_API_TOKEN is not set. Export it or use --dry-run.")
def api_post(path, body):
url = f"{API_BASE}{path}"
headers = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}
logging.debug("POST %s body=%s", url, body)
r = requests.post(url, headers=headers, json=body, timeout=30)
return r.status_code, r.text
# ---------- Column handling ----------
def pick_col(cands, cols):
"""Pick a column name from 'cols' given a list of candidate names or substrings (case-insensitive)."""
lower_map = {c.lower(): c for c in cols}
# exact matches
for c in cands:
if c.lower() in lower_map:
return lower_map[c.lower()]
# substring matches
for col in cols:
lc = col.lower()
for c in cands:
if c.lower() in lc:
return col
return None
def infer_columns(df, overrides):
cols = list(df.columns)
# support dotted paths in CSV (e.g., transactionBlockchainInfo.txHash)
id_col = overrides.get("id") or pick_col(["id","tx_id","transaction_id","uuid"], cols)
status_col = overrides.get("status") or pick_col(["status","tx_status","transaction_status"], cols)
asset_col = overrides.get("asset") or pick_col(["transactionAsset","asset","symbol","currency"], cols)
amount_col = overrides.get("amount") or pick_col(["transactionAmount","amount","value"], cols)
hash_col = overrides.get("hash") or pick_col(["transactionblockchaininfo.txhash","txhash","hash","blockchain_hash"], cols)
dest_col = overrides.get("destination") or pick_col(["transactionblockchaininfo.destination","destination","to","beneficiary","dest_address"], cols)
updated_col= overrides.get("updated") or pick_col(["updatedat","updated_at","timestamp","time","created_at","createdat"], cols)
required = {
"id": id_col,
"status": status_col,
"asset": asset_col,
"amount": amount_col,
"hash": hash_col,
"destination": dest_col,
}
missing = [k for k,v in required.items() if v is None]
if missing:
raise SystemExit(f"Could not infer required columns {missing}. Available columns: {cols}. "
"Use --db-* overrides to specify exact names.")
return {
"id": id_col,
"status": status_col,
"asset": asset_col,
"amount": amount_col,
"hash": hash_col,
"destination": dest_col,
"updated": updated_col, # optional
}
# ---------- Normalization & compare ----------
def norm_dest(x):
return str(x).strip().lower()
def same_key(a, b):
return (
a["asset"] == b["asset"] and
str(a["amount"]) == str(b["amount"]) and
a["txHash"] == b["txHash"] and
norm_dest(a["destination"]) == norm_dest(b["destination"])
)
def cancel_tx(incomplete_id, other_id, dry_run=False):
body = {"id": incomplete_id, "reason": f"This is a duplicate of {other_id}"}
if dry_run:
logging.info("[DRY-RUN] Would POST /tx/cancel %s", body)
return 200, "DRY-RUN"
status, text = api_post("/tx/cancel", body)
if 200 <= status < 300:
logging.info("[OK] Cancelled %s (duplicate of %s)", incomplete_id, other_id)
else:
logging.error("[ERROR] Cancel %s failed: HTTP %s %s", incomplete_id, status, text)
return status, text
def extract_row(row, cols):
return {
"id": row[cols["id"]],
"status": str(row[cols["status"]]).strip().upper(),
"asset": str(row[cols["asset"]]).strip(),
"amount": row[cols["amount"]],
"txHash": str(row[cols["hash"]]).strip(),
"destination": str(row[cols["destination"]]).strip(),
"updatedAt": str(row[cols["updated"]]) if cols["updated"] and cols["updated"] in row else None,
}
def process_hash_from_df(df, q_hash, cols, dry_run=False):
logging.info("Processing hash %s", q_hash)
# exact filter on txHash
sub = df[df[cols["hash"]].astype(str).str.strip() == str(q_hash).strip()].copy()
logging.info("Hash %s: %d row(s) found in DB", q_hash, len(sub))
if len(sub) <= 1:
logging.info("Hash %s: nothing to do (<=1 match).", q_hash)
return
records = [extract_row(r, cols) for _, r in sub.iterrows()]
incompletes = [r for r in records if r["status"] == "INCOMPLETE"]
others = [r for r in records if r["status"] in TARGET_OK_STATUSES]
logging.info("Hash %s: %d INCOMPLETE, %d others", q_hash, len(incompletes), len(others))
if not incompletes or not others:
logging.info("Hash %s: no actionable pairs.", q_hash)
return
# Sort others by updatedAt desc (string compare is fine for ISO; otherwise falls back to None at end)
others_sorted = sorted(others, key=lambda x: (x["updatedAt"] or ""), reverse=True)
for inc in incompletes:
counterpart = next((o for o in others_sorted if same_key(inc, o)), None)
if counterpart:
logging.debug("INCOMPLETE %s matches counterpart %s", inc["id"], counterpart["id"])
cancel_tx(inc["id"], counterpart["id"], dry_run=dry_run)
else:
logging.info("Hash %s: INCOMPLETE %s has no exact matching counterpart.", q_hash, inc["id"])
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--csv", required=True, help="Path to the database CSV")
ap.add_argument("--hashes", nargs="*", help="List of blockchain tx hashes to check")
ap.add_argument("--hash-file", help="File with one hash per line")
ap.add_argument("--dry-run", action="store_true", help="Don't call txCancel; just log actions")
ap.add_argument("-v","--verbose", action="store_true", help="Verbose (debug) logging")
# Overrides
ap.add_argument("--db-id", dest="db_id")
ap.add_argument("--db-status", dest="db_status")
ap.add_argument("--db-asset", dest="db_asset")
ap.add_argument("--db-amount", dest="db_amount")
ap.add_argument("--db-hash", dest="db_hash")
ap.add_argument("--db-destination", dest="db_destination")
ap.add_argument("--db-updated", dest="db_updated")
args = ap.parse_args()
setup_logging(args.verbose)
require_token(args.dry_run)
# Load CSV
df = pd.read_csv(args.csv)
logging.info("Loaded DB CSV with %d rows and columns: %s", len(df), list(df.columns))
overrides = {
"id": args.db_id,
"status": args.db_status,
"asset": args.db_asset,
"amount": args.db_amount,
"hash": args.db_hash,
"destination": args.db_destination,
"updated": args.db_updated,
}
cols = infer_columns(df, overrides)
logging.info("Using columns: %s", cols)
# Collect hashes
hashes = []
if args.hashes:
hashes.extend(args.hashes)
if args.hash_file:
with open(args.hash_file, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line:
hashes.append(line)
if not hashes:
logging.error("Provide --hashes ... or --hash-file path")
sys.exit(1)
# Process each hash
for h in hashes:
try:
process_hash_from_df(df, h, cols, dry_run=args.dry_run)
except Exception as e:
logging.exception("Error processing hash %s", h)
if __name__ == "__main__":
main()
Run command:
# Set your API token (or use --dry-run to test safely)
$env:NOTABENE_API_TOKEN = "<your-bearer-token>"
# Dry run (no API calls)
python .\hash_check_from_db_and_cancel.py `
--csv "D:\path\to\webhookDB.csv" `
--hash-file .\hashes.txt `
--dry-run -v
# Real run
python .\hash_check_from_db_and_cancel.py `
--csv "D:\path\to\webhookDB.csv" `
--hash-file .\hashes.txt -v
Samle DB:
id transaction_asset transaction_amount beneficiary_vasp_did originator_vasp_did origin_address destination_address tx_hash status created_at updated_at
71683b7e-41ff-4b89-8a2c-eceddf41cf7e ETH-ETH 54325342 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0x4b9d27c388baf53c10d8dc4fdc1f2e0a7f3bb0e9 0x26F30441866De260879eEA444381F2eC7179c876 0x8f0D9acBdf8Dbeea67af639CbC995a9767e14488 0x26F30441866De260879eEA444381F2eC7179c876 INCOMPLETE 2025-08-20 2:40:12 2025-08-20 2:40:12
a61cf8af-57ea-4886-baa6-27e18b5ff54e ETH-ETH 54325342 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0x4b9d27c388baf53c10d8dc4fdc1f2e0a7f3bb0e9 0x26F30441866De260879eEA444381F2eC7179c876 0x8f0D9acBdf8Dbeea67af639CbC995a9767e14488 0x26F30441866De260879eEA444381F2eC7179c876 SENT 2025-08-21 2:40:12 2025-08-21 2:40:12
644f5ae8-d73f-48de-8e80-7760398fdb72 ETH-ETH 11850000 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0x3f86c2b9d14a7c5590b36bd4e2f1a9d1b2c3a4d5 0xC04ebAe1B29BA89cC168b079B03caa5496915670 0x2dF9b935C44057Ac240634c7536511d8Aa03028D 0x2cf7543e4fcdf6f140e66bfd725ba6d939e6c34c3b2ef847e2d5e11671a0d233 INCOMPELTE 2025-08-22 2:40:12 2025-08-22 2:40:12
cee0c5d5-9f03-499b-8f87-b587d215d204 ETH-ETH 11850000 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0x3f86c2b9d14a7c5590b36bd4e2f1a9d1b2c3a4d5 0xC04ebAe1B29BA89cC168b079B03caa5496915670 0x2dF9b935C44057Ac240634c7536511d8Aa03028D 0x2cf7543e4fcdf6f140e66bfd725ba6d939e6c34c3b2ef847e2d5e11671a0d233 ACCEPTED 2025-08-23 2:40:12 2025-08-23 2:40:12
f3e35bb6-941a-4aaf-a4dc-9d84f27e9a45 ETH-ETH 432412 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0xd7db47597122b46af37bc50e664915acaf345dc6 0xb23360CCDd9Ed1b15D45E5d3824Bb409C8D7c460 0xC4e65AD18a6a0b7B33Ab99B203084B40fDd5aEe1 0xc724fe9d0b492a42d052d1ae27e477e1228faa362412220f6702ded1cf3fe2c2 INCOMPLETE 2025-08-24 2:40:12 2025-08-24 2:40:12
d59485c3-875c-4ca7-99ef-f2012b585716 ETH-ETH 432412 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0xd7db47597122b46af37bc50e664915acaf345dc6 0xb23360CCDd9Ed1b15D45E5d3824Bb409C8D7c460 0xC4e65AD18a6a0b7B33Ab99B203084B40fDd5aEe1 0xc724fe9d0b492a42d052d1ae27e477e1228faa362412220f6702ded1cf3fe2c2 ACK 2025-08-25 2:40:12 2025-08-25 2:40:12
28cdc207-c8a0-455f-8e67-d2a95f123d37 ETH-ETH 543256436 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0x7f4dd2f2fddbb296369e3ac5b1e603e269ef23c1 0x4838B106FCe9647Bdf1E7877BF73cE8B0BAD5f97 0x4C1746A800D224393fE2470C70A35717eD4eA5F1 0x1AB4973a48dc892Cd9971ECE8e01DcC7688f8F23 SENT 2025-08-26 2:40:12 2025-08-26 2:40:12
acaac9ef-1155-40e7-975a-e59b4b3b5e55 ETH-ETH 321454 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0xc9603188cf059d7f11135b96a91432e6d79cd491 0x4675C7e5BaAFBFFbca748158bEcBA61ef3b0a263 0xdAC17F958D2ee523a2206206994597C13D831ec7 0x49d071eb4ca633e640b0f42c98a7224867994624247461ee4e30acb79abc084f ACK 2025-08-27 2:40:12 2025-08-27 2:40:12
eea01971-a069-4a1e-b22a-89dd135b4cb2 ETH-ETH 7432 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0x3a69821892b1c3754838ac0bf72aeeda96feae4d 0x1AB4973a48dc892Cd9971ECE8e01DcC7688f8F23 0x1ADE49c7D296847c65c952b8B4859D81739DBa4e 0x4d1bb5279206b8a558a6880d8972ab98ee8f44fb19e772cc6cb17e311aaea693 SENT 2025-08-28 2:40:12 2025-08-28 2:40:12
42a5311a-9b2a-4dbb-86e6-8b0e90de122d ETH-ETH 3215732 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0x181caf505ba4c3766ddebf9dd68bda4429b74b41 0x1AB4973a48dc892Cd9971ECE8e01DcC7688f8F23 0x49FEd3C7eB5756f6a359E2e3531734a2c0A16B22 0x49d071eb4ca633e640b0f42c98a7224867994624247461ee4e30acb79abc084f ACCEPTED 2025-08-29 2:40:12 2025-08-29 2:40:12
4af46b00-9475-4786-90ef-48a96e475966 ETH-ETH 72432 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0xdf5b2f9895feaec204cde9fdc0757ce49a501008 0x1A6ccF888DE9300eCCB70D3D33c4560e3fa5cD62 0x46c943CB8228D62c19D9DA300E8e6DBD5ac79859 0x1b6e44698fc46074b1838b07996b16aee2bdde49eeabe85d58bdd689c8fc17d8 SENT 2025-08-30 2:40:12 2025-08-30 2:40:12
e8482267-bd22-417c-894f-818424ba2873 ETH-ETH 54316 did:ethr:0x8e7b5a6f2f3e7d1b4c8a129fbd45e2a11a2b9c77 did:ethr:0xdf5b2f9895feaec204cde9fdc0757ce49a501008 0x3d03673aB0Bbfb5d86E2bd733469c18E0BBf433e 0xCC78Ee4037A0b7e03125512D7778B6d66C7A4f82 0xcccde479948c4d87bc61095a39725e8128c7fd0bda479833377862d9fbcadff9 ACCEPTED 2025-08-31 2:40:12 2025-08-31 2:40:12Updated 8 days ago
Did this page help you?
