Fix Monero scan position tracking to work per-account

- Each crypto processor now scans only its assigned Monero account
- Scan position is tracked independently per processor
- Fixed indentation in scan loop that was broken during refactoring
- Updated tests to match new per-account scanning behavior
- Resolves issue where payments on accounts other than 0 weren't updating scan position
This commit is contained in:
Russell Ballestrini 2025-10-01 13:23:32 -04:00
parent 78ea2c0c13
commit da18fe2323
2 changed files with 250 additions and 216 deletions

View file

@ -3446,234 +3446,231 @@ def scan_wallet_for_double_or_late_payments(request, settings):
if coin_type == "XMR":
try:
# Get the minimum scan position across all processors to determine what's "new"
processor_positions = []
for p in coin_processors:
pos = _get_scan_position_from_semaphore(
p.last_scan_semaphore, coin_type
)
processor_positions.append(
(p.id, p.shop_id, p.last_scan_semaphore, pos)
)
log.processing_cycle(
f"Processor {p.id} (shop {p.shop_id}) scan position: {p.last_scan_semaphore} → height {pos}"
)
min_scan_position = min(pos for _, _, _, pos in processor_positions)
log.processing_cycle(
f"Using minimum scan position {min_scan_position} across processors",
len(processor_positions),
)
# Build bounded query for XMR transfers
query_params = {
"in": True,
"out": False,
"pending": True,
"failed": False,
"pool": True,
"all_accounts": True, # Scan all accounts, not just account 0
}
# Add height bounds if we have a valid scan position
if min_scan_position > 0:
query_params["filter_by_height"] = True
query_params["min_height"] = min_scan_position
log.processing_cycle(
f"Scanning for transfers with min_height >= {min_scan_position}"
)
else:
log.processing_cycle("Scanning ALL transfers (no height filter)")
# Get transfers from the wallet using bounded query
log.processing_cycle(
f"Calling get_transfers with params: {query_params}"
)
result = client._call("get_transfers", query_params)
all_transfers = []
for transfer_type in ["in", "pending", "pool"]:
if transfer_type in result:
transfers_in_type = result[transfer_type]
all_transfers.extend(transfers_in_type)
# Process each processor's account separately
for processor in coin_processors:
# Get account index from wallet_label (for XMR it's the account index)
try:
account_index = int(processor.wallet_label)
except (ValueError, TypeError):
log.processing_cycle(
f"Found transfers in '{transfer_type}' category",
len(transfers_in_type),
)
if not all_transfers:
log.processing_cycle(
f"No {coin_type} transfers found in wallet scan"
)
continue
# Process all transfers returned by bounded query
# RPC already filtered by height, so all transfers are "new"
new_transfers = all_transfers
max_height = max(
(tx.get("height", 0) for tx in all_transfers), default=0
)
log.processing_cycle(
f"Found incoming {coin_type} transfers newer than scan position {min_scan_position}, max height: {max_height}",
len(new_transfers),
)
# Log details of each transfer for debugging
for i, tx in enumerate(new_transfers):
subaddr = tx.get("subaddr_index", {})
log.processing_cycle(
f"Transfer {i+1}: height={tx.get('height', 'unknown')}, "
f"amount={tx.get('amount', 0) / 1e12:.12f} XMR, "
f"confirmations={tx.get('confirmations', 0)}, "
f"subaddr={subaddr.get('major', 0)}.{subaddr.get('minor', 'unknown')}, "
f"txid={tx.get('txid', 'unknown')[:16]}..."
)
# Process new transfers and match to payments
late_payments_found = 0
transfers_checked = 0
for tx in new_transfers:
transfers_checked += 1
subaddr = tx.get("subaddr_index", {})
if not subaddr:
log.processing_cycle(
f"Transfer {transfers_checked}: skipping - no subaddr_index"
f"Processor {processor.id} has invalid wallet_label for XMR: {processor.wallet_label}"
)
continue
account_idx = subaddr.get("major", 0)
subaddr_idx = subaddr.get("minor")
if subaddr_idx is None:
log.processing_cycle(
f"Transfer {transfers_checked}: skipping - no minor subaddr"
)
continue
# Look up payment by subaddress
payment = (
db.query(CryptoPayment)
.options(
sa.orm.joinedload(CryptoPayment.user),
sa.orm.joinedload(CryptoPayment.shop),
)
.filter(
CryptoPayment.coin_type == coin_type,
CryptoPayment.account_index == account_idx,
CryptoPayment.subaddress_index == subaddr_idx,
)
.first()
# Get this processor's scan position
scan_position = _get_scan_position_from_semaphore(
processor.last_scan_semaphore, coin_type
)
log.processing_cycle(
f"Processor {processor.id} (shop {processor.shop_id}, account {account_index}) scan position: {processor.last_scan_semaphore} → height {scan_position}"
)
if payment:
log.payment_info(
payment,
f"Transfer {transfers_checked} to subaddr {account_idx}.{subaddr_idx}: FOUND payment record (status {payment.status})",
# Build bounded query for this specific account
query_params = {
"in": True,
"out": False,
"pending": True,
"failed": False,
"pool": True,
"account_index": account_index, # Scan only this account
}
# Add height bounds if we have a valid scan position
if scan_position > 0:
query_params["filter_by_height"] = True
query_params["min_height"] = scan_position
log.processing_cycle(
f"Scanning account {account_index} for transfers with min_height >= {scan_position}"
)
else:
log.processing_cycle(f"Scanning ALL transfers for account {account_index} (no height filter)")
# Get transfers from the wallet using bounded query
log.processing_cycle(
f"Calling get_transfers with params: {query_params}"
)
result = client._call("get_transfers", query_params)
all_transfers = []
for transfer_type in ["in", "pending", "pool"]:
if transfer_type in result:
transfers_in_type = result[transfer_type]
all_transfers.extend(transfers_in_type)
log.processing_cycle(
f"Found transfers in '{transfer_type}' category",
len(transfers_in_type),
)
if not all_transfers:
log.processing_cycle(
f"Transfer {transfers_checked} to subaddr {account_idx}.{subaddr_idx}: NO payment record"
f"No {coin_type} transfers found for account {account_index}"
)
continue
# Process all transfers returned by bounded query
# RPC already filtered by height, so all transfers are "new"
new_transfers = all_transfers
max_height = max(
(tx.get("height", 0) for tx in all_transfers), default=0
)
log.processing_cycle(
f"Found incoming {coin_type} transfers for account {account_index} newer than scan position {scan_position}, max height: {max_height}",
len(new_transfers),
)
# Log details of each transfer for debugging
for i, tx in enumerate(new_transfers):
subaddr = tx.get("subaddr_index", {})
log.processing_cycle(
f"Transfer {i+1}: height={tx.get('height', 'unknown')}, "
f"amount={tx.get('amount', 0) / 1e12:.12f} XMR, "
f"confirmations={tx.get('confirmations', 0)}, "
f"subaddr={subaddr.get('major', 0)}.{subaddr.get('minor', 'unknown')}, "
f"txid={tx.get('txid', 'unknown')[:16]}..."
)
if payment:
should_process = _should_process_late_payment(payment, tx)
log.payment_info(
payment, f"should_process_late_payment: {should_process}"
)
if should_process:
late_payments_found += 1
# Process new transfers and match to payments
late_payments_found = 0
transfers_checked = 0
for tx in new_transfers:
transfers_checked += 1
subaddr = tx.get("subaddr_index", {})
if not subaddr:
log.processing_cycle(
f"Transfer {transfers_checked}: skipping - no subaddr_index"
)
continue
# Check if this is a duplicate payment (original already has funds)
if payment.received_amount > 0:
account_idx = subaddr.get("major", 0)
subaddr_idx = subaddr.get("minor")
if subaddr_idx is None:
log.processing_cycle(
f"Transfer {transfers_checked}: skipping - no minor subaddr"
)
continue
# Look up payment by subaddress
payment = (
db.query(CryptoPayment)
.options(
sa.orm.joinedload(CryptoPayment.user),
sa.orm.joinedload(CryptoPayment.shop),
)
.filter(
CryptoPayment.coin_type == coin_type,
CryptoPayment.account_index == account_idx,
CryptoPayment.subaddress_index == subaddr_idx,
)
.first()
)
if payment:
log.payment_info(
payment,
f"DUPLICATE payment detected for {payment.status} quote: "
f"{tx.get('amount', 0) / 1e12:.12f} XMR, "
f"{tx.get('confirmations', 0)} confirmations, height {tx.get('height', 0)}",
)
# Check if duplicate payment record already exists for this transaction
existing_duplicate = (
db.query(CryptoPayment)
.filter(
CryptoPayment.coin_type == coin_type,
CryptoPayment.account_index == account_idx,
CryptoPayment.subaddress_index == subaddr_idx,
CryptoPayment.status
== CryptoPayment.STATUS_DOUBLEPAY_REFUNDED,
CryptoPayment.tx_hashes.contains(
f'"{tx.get("txid")}"'
),
)
.first()
)
if existing_duplicate:
log.payment_info(
existing_duplicate,
f"Duplicate payment record already exists for txid {tx.get('txid')}, processing refund",
)
duplicate_payment = existing_duplicate
else:
# Create new payment object for each duplicate transaction
duplicate_payment = _create_duplicate_payment(
payment, tx, coin_type, db
)
db.add(duplicate_payment)
db.flush() # Get the ID assigned
log.payment_info(
duplicate_payment,
"Created duplicate payment for refund processing",
)
# Process the duplicate payment through refund pipeline
incoming = [tx]
process_payment(
request, duplicate_payment, incoming, client=client
f"Transfer {transfers_checked} to subaddr {account_idx}.{subaddr_idx}: FOUND payment record (status {payment.status})",
)
else:
# First payment to this address - process normally
# Determine if this is a late payment or double payment
if payment.status in [
CryptoPayment.STATUS_EXPIRED,
CryptoPayment.STATUS_CANCELLED,
]:
payment_type = "late payment"
else:
payment_type = "double payment"
log.payment_info(
payment,
f"Found {payment_type} to {payment.status} quote: "
f"{tx.get('amount', 0) / 1e12:.12f} XMR, "
f"{tx.get('confirmations', 0)} confirmations, height {tx.get('height', 0)}",
)
incoming = [tx]
process_payment(request, payment, incoming, client=client)
# Update scan positions for all processors of this coin type
if max_height > 0: # Only update if we found confirmed transfers
for processor in coin_processors:
old_position = _get_scan_position_from_semaphore(
processor.last_scan_semaphore, coin_type
)
if old_position < max_height:
new_semaphore = _format_scan_semaphore(
coin_type, max_height
)
processor.last_scan_semaphore = new_semaphore
db.add(processor)
log.processing_cycle(
f"Updated scan position for {processor.coin_type} processor {processor.id}: "
f"{old_position}{max_height}"
f"Transfer {transfers_checked} to subaddr {account_idx}.{subaddr_idx}: NO payment record"
)
if payment:
should_process = _should_process_late_payment(payment, tx)
log.payment_info(
payment, f"should_process_late_payment: {should_process}"
)
if should_process:
late_payments_found += 1
# Check if this is a duplicate payment (original already has funds)
if payment.received_amount > 0:
log.payment_info(
payment,
f"DUPLICATE payment detected for {payment.status} quote: "
f"{tx.get('amount', 0) / 1e12:.12f} XMR, "
f"{tx.get('confirmations', 0)} confirmations, height {tx.get('height', 0)}",
)
# Check if duplicate payment record already exists for this transaction
existing_duplicate = (
db.query(CryptoPayment)
.filter(
CryptoPayment.coin_type == coin_type,
CryptoPayment.account_index == account_idx,
CryptoPayment.subaddress_index == subaddr_idx,
CryptoPayment.status
== CryptoPayment.STATUS_DOUBLEPAY_REFUNDED,
CryptoPayment.tx_hashes.contains(
f'"{tx.get("txid")}"'
),
)
.first()
)
if existing_duplicate:
log.payment_info(
existing_duplicate,
f"Duplicate payment record already exists for txid {tx.get('txid')}, processing refund",
)
duplicate_payment = existing_duplicate
else:
# Create new payment object for each duplicate transaction
duplicate_payment = _create_duplicate_payment(
payment, tx, coin_type, db
)
db.add(duplicate_payment)
db.flush() # Get the ID assigned
log.payment_info(
duplicate_payment,
"Created duplicate payment for refund processing",
)
# Process the duplicate payment through refund pipeline
incoming = [tx]
process_payment(
request, duplicate_payment, incoming, client=client
)
else:
# First payment to this address - process normally
# Determine if this is a late payment or double payment
if payment.status in [
CryptoPayment.STATUS_EXPIRED,
CryptoPayment.STATUS_CANCELLED,
]:
payment_type = "late payment"
else:
payment_type = "double payment"
log.payment_info(
payment,
f"Found {payment_type} to {payment.status} quote: "
f"{tx.get('amount', 0) / 1e12:.12f} XMR, "
f"{tx.get('confirmations', 0)} confirmations, height {tx.get('height', 0)}",
)
incoming = [tx]
process_payment(request, payment, incoming, client=client)
# Update scan position for THIS processor only
if max_height > 0 and max_height > scan_position:
new_semaphore = _format_scan_semaphore(coin_type, max_height)
processor.last_scan_semaphore = new_semaphore
db.add(processor)
log.processing_cycle(
f"Updated scan position for {processor.coin_type} processor {processor.id} (account {account_index}): "
f"{scan_position}{max_height}"
)
log.processing_cycle(
f"Account {account_index} scan completed, found late payments",
late_payments_found,
f"scanned {len(new_transfers)} transfers",
)
log.processing_cycle(
f"Wallet scan completed in ~{(time.time() - scan_start_time):.1f} seconds, found late payments",
late_payments_found,
f"scanned {len(new_transfers)} transfers",
f"XMR wallet scan completed in ~{(time.time() - scan_start_time):.1f} seconds for all accounts"
)
except Exception as e:

View file

@ -1722,9 +1722,12 @@ class PassiveMonitoringTests(unittest.TestCase):
self, mock_process, mock_get_client
):
"""Test scanning XMR wallet finds expired payment transfers."""
# Mock crypto processor
# Mock crypto processor with proper wallet_label for XMR (account index)
mock_processor = MagicMock()
mock_processor.id = "processor_123"
mock_processor.shop_id = "shop_123"
mock_processor.coin_type = "XMR"
mock_processor.wallet_label = "1" # XMR uses account index as wallet_label
mock_processor.last_scan_semaphore = None # No previous scan
# Mock CryptoProcessor query
@ -1739,7 +1742,8 @@ class PassiveMonitoringTests(unittest.TestCase):
"txid": "test_tx_123",
"amount": 1000000000000, # 1 XMR
"confirmations": 10,
"subaddr_index": {"major": 0, "minor": 42},
"subaddr_index": {"major": 1, "minor": 42},
"height": 3512193,
}
],
"pending": [],
@ -1752,7 +1756,12 @@ class PassiveMonitoringTests(unittest.TestCase):
mock_payment.id = "payment_123"
mock_payment.status = CryptoPayment.STATUS_EXPIRED
mock_payment.coin_type = "XMR"
mock_payment.account_index = 1
mock_payment.subaddress_index = 42
mock_payment.received_amount = 0 # No previous funds received
mock_payment.tx_hashes = "[]"
mock_payment.user = None
mock_payment.shop = None
# Configure query chain - need to handle both CryptoProcessor and CryptoPayment queries
def query_side_effect(model):
@ -1760,15 +1769,18 @@ class PassiveMonitoringTests(unittest.TestCase):
return mock_processor_query
else: # CryptoPayment query
mock_payment_query = MagicMock()
mock_payment_query.filter.return_value.first.return_value = mock_payment
mock_options = MagicMock()
mock_options.filter.return_value.first.return_value = mock_payment
mock_payment_query.options.return_value = mock_options
return mock_payment_query
self.mock_dbsession.query.side_effect = query_side_effect
self.mock_dbsession.add = MagicMock()
# Run scan
scan_wallet_for_double_or_late_payments(self.mock_request, self.mock_settings)
# Verify client was called to get transfers
# Verify client was called to get transfers with account_index for per-account scanning
mock_client._call.assert_called_once_with(
"get_transfers",
{
@ -1777,6 +1789,7 @@ class PassiveMonitoringTests(unittest.TestCase):
"pending": True,
"failed": False,
"pool": True,
"account_index": 1, # Now includes account_index for per-account scanning
},
)
@ -1791,6 +1804,10 @@ class PassiveMonitoringTests(unittest.TestCase):
self.assertEqual(len(call_args[0][2]), 1) # One transfer
self.assertEqual(call_args[1]["client"], mock_client)
# Verify scan position was updated for this processor
self.assertEqual(mock_processor.last_scan_semaphore, "height:3512193")
self.mock_dbsession.add.assert_called_with(mock_processor)
@patch("make_post_sell.lib.crypto_watcher.get_crypto_client")
@patch("make_post_sell.lib.crypto_watcher.process_payment")
def test_scan_wallet_skips_non_expired_payments(
@ -1851,9 +1868,12 @@ class PassiveMonitoringTests(unittest.TestCase):
self, mock_process, mock_get_client
):
"""Test scanning wallet processes cancelled payments with transfers."""
# Mock crypto processor
# Mock crypto processor with proper wallet_label for XMR (account index)
mock_processor = MagicMock()
mock_processor.id = "processor_456"
mock_processor.shop_id = "shop_456"
mock_processor.coin_type = "XMR"
mock_processor.wallet_label = "0" # XMR uses account index as wallet_label
mock_processor.last_scan_semaphore = None
# Mock CryptoProcessor query
@ -1869,6 +1889,7 @@ class PassiveMonitoringTests(unittest.TestCase):
"amount": 500000000000, # 0.5 XMR
"confirmations": 5,
"subaddr_index": {"major": 0, "minor": 99},
"height": 3512194,
}
],
"pending": [],
@ -1881,7 +1902,12 @@ class PassiveMonitoringTests(unittest.TestCase):
mock_payment.id = "payment_456"
mock_payment.status = CryptoPayment.STATUS_CANCELLED
mock_payment.coin_type = "XMR"
mock_payment.account_index = 0
mock_payment.subaddress_index = 99
mock_payment.received_amount = 0 # No previous funds received
mock_payment.tx_hashes = "[]"
mock_payment.user = None
mock_payment.shop = None
# Configure query chain
def query_side_effect(model):
@ -1889,24 +1915,34 @@ class PassiveMonitoringTests(unittest.TestCase):
return mock_processor_query
else:
mock_payment_query = MagicMock()
mock_payment_query.filter.return_value.first.return_value = mock_payment
mock_options = MagicMock()
mock_options.filter.return_value.first.return_value = mock_payment
mock_payment_query.options.return_value = mock_options
return mock_payment_query
self.mock_dbsession.query.side_effect = query_side_effect
self.mock_dbsession.add = MagicMock()
# Run scan
scan_wallet_for_double_or_late_payments(self.mock_request, self.mock_settings)
# Verify process_payment was called for cancelled payment
mock_process.assert_called_once()
# Verify scan position was updated
self.assertEqual(mock_processor.last_scan_semaphore, "height:3512194")
self.mock_dbsession.add.assert_called_with(mock_processor)
@patch("make_post_sell.lib.crypto_watcher.get_crypto_client")
@patch("make_post_sell.lib.crypto_watcher.process_payment")
def test_scan_wallet_handles_rpc_errors(self, mock_process, mock_get_client):
"""Test scanning wallet handles RPC errors gracefully."""
# Mock crypto processor
# Mock crypto processor with proper wallet_label for XMR (account index)
mock_processor = MagicMock()
mock_processor.id = "processor_err"
mock_processor.shop_id = "shop_err"
mock_processor.coin_type = "XMR"
mock_processor.wallet_label = "0" # XMR uses account index as wallet_label
mock_processor.last_scan_semaphore = None
# Mock CryptoProcessor query
@ -2010,7 +2046,7 @@ class PassiveMonitoringIntegrationTests(unittest.TestCase):
sweep_to_address="cold_storage_address",
)
processor.enabled = True
processor.wallet_label = "test_wallet"
processor.wallet_label = "0" # XMR uses account index as wallet_label
dbsession.add(processor)
dbsession.flush()
@ -2075,6 +2111,7 @@ class PassiveMonitoringIntegrationTests(unittest.TestCase):
"major": 0,
"minor": 42,
}, # Matches expired payment
"height": 3512195,
}
],
"pending": [],