Skip to content

Commit 60eaacd

Browse files
committed
feat(sync): decode ERC-20/ERC-721 transfers + drop orphan logs
Two related fixes that surfaced together during testnet recovery on 2026-05-20. ## ERC-20 / ERC-721 Transfer decoder The `token_transfers` table existed but was never populated — the `indexer-handlers` crate is still a Phase-0 placeholder, so raw logs landed in `logs` and went nowhere. Scan UIs that asked for an address's token balances saw an empty list even though the underlying Transfer events were present. Inline decoder in `sync::token_decode`: - topic0 `0xddf252ad…` matches the canonical Transfer signature. - ERC-20: topics = [sig, from, to], 32-byte data carries `amount`. - ERC-721: topics = [sig, from, to, token_id], no data, amount = 1. - ERC-1155 is out of scope here (different topic0, richer encoding). Decoded transfers flow through `BlockBundle.token_transfers` and the existing block-writer transaction — they commit atomically with the block, txs, and logs the chain already wrote. False-positive risk: any contract emitting a Transfer-shaped event with three indexed args will decode as ERC-721 here. Acceptable for the visibility goal; precise registry-driven classification is the declarative-handler workstream's job. ## Orphan-log filter Sentrix's native `/chain/blocks/<n>` and `eth_getLogs` can disagree — a tx whose effects reverted gets stripped from the block tx vec but its log envelopes still come back from `eth_getLogs`. The `logs.tx_hash → transactions.hash` FK then blew the whole batch transaction and the backfill loop stalled at the first such block. Drop logs whose tx_hash isn't backed by a tx row in the same bundle (both fetch_one + ingest_one paths). Such logs are orphans on this chain by definition; preserving them would only buy us repeated FK-violation rollbacks. ## Schema note `token_transfers` has no unique constraint on (tx_hash, log_index), so the batch insert skips ON CONFLICT. The writer's atomic cursor advance prevents re-processing the same block in the steady state; reorg recovery deletes downstream rows before re-insert. Adding the unique index is a follow-up migration.
1 parent e88f529 commit 60eaacd

5 files changed

Lines changed: 250 additions & 3 deletions

File tree

crates/db/src/token_transfers.rs

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,38 @@ where
3030
Ok(())
3131
}
3232

33+
/// Batch insert. Retry-safe via `(tx_hash, log_index)` ON CONFLICT DO NOTHING
34+
/// — matches the dedup contract on the raw `logs` table.
35+
pub async fn insert_batch<'e, E>(executor: E, transfers: &[TokenTransfer]) -> DbResult<()>
36+
where
37+
E: sqlx::PgExecutor<'e>,
38+
{
39+
if transfers.is_empty() {
40+
return Ok(());
41+
}
42+
let mut qb = sqlx::QueryBuilder::new(
43+
"INSERT INTO token_transfers (block_height, tx_hash, log_index, contract, standard, \
44+
from_addr, to_addr, token_id, amount) ",
45+
);
46+
qb.push_values(transfers.iter(), |mut row, t| {
47+
row.push_bind(t.block_height)
48+
.push_bind(&t.tx_hash)
49+
.push_bind(t.log_index)
50+
.push_bind(&t.contract)
51+
.push_bind(t.standard.as_str())
52+
.push_bind(&t.from_addr)
53+
.push_bind(&t.to_addr)
54+
.push_bind(t.token_id)
55+
.push_bind(t.amount);
56+
});
57+
// No ON CONFLICT — table has no unique constraint on (tx_hash, log_index).
58+
// The writer's atomic cursor advance prevents re-processing the same
59+
// block in the steady state; reorg recovery deletes downstream rows
60+
// before re-insert. Adding a unique index here is a follow-up migration.
61+
qb.build().execute(executor).await?;
62+
Ok(())
63+
}
64+
3365
/// Paginated token-transfer history for an address — transfers where the
3466
/// address is sender OR receiver, newest-first by block height. Optional
3567
/// `standard` narrows to a specific token kind ("erc20" / "erc721" /

crates/sync/src/backfill.rs

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -212,10 +212,38 @@ async fn fetch_one(
212212
.map(to_domain_log)
213213
.collect::<Result<Vec<_>, _>>()
214214
.map_err(|e| SyncError::Invalid(e.to_string()))?;
215+
216+
// 2026-05-20: Sentrix's native /chain/blocks/<n> and eth_getLogs can
217+
// disagree — a tx whose effects reverted gets stripped from the block
218+
// tx vec but its log envelopes still come back from eth_getLogs. The
219+
// `logs.tx_hash` FK then blows the whole batch. Drop logs whose
220+
// tx_hash isn't backed by a tx row in this same bundle; they'd be
221+
// orphaned anyway.
222+
use std::collections::HashSet;
223+
let tx_hash_set: HashSet<_> = dom_txs.iter().map(|t| t.hash.clone()).collect();
224+
let logs_total = dom_logs.len();
225+
let dom_logs: Vec<_> = dom_logs
226+
.into_iter()
227+
.filter(|l| tx_hash_set.contains(&l.tx_hash))
228+
.collect();
229+
if dom_logs.len() < logs_total {
230+
tracing::debug!(
231+
block = h.0,
232+
dropped = logs_total - dom_logs.len(),
233+
"backfill: dropped orphan logs (tx_hash not in block.txs)"
234+
);
235+
}
236+
237+
let dom_token_transfers: Vec<_> = dom_logs
238+
.iter()
239+
.filter_map(crate::token_decode::decode_transfer)
240+
.collect();
241+
215242
Ok(Some(BlockBundle {
216243
block: dom_block,
217244
txs: dom_txs,
218245
logs: dom_logs,
246+
token_transfers: dom_token_transfers,
219247
}))
220248
}
221249

@@ -273,12 +301,28 @@ pub async fn ingest_one(
273301
.collect::<Result<Vec<_>, _>>()
274302
.map_err(|e| SyncError::Invalid(e.to_string()))?;
275303

304+
// Same orphan-log filter as fetch_one — keep the FK invariant on the
305+
// tail path so the live chain doesn't stall the indexer the way the
306+
// backfill batch did.
307+
use std::collections::HashSet;
308+
let tx_hash_set: HashSet<_> = dom_txs.iter().map(|t| t.hash.clone()).collect();
309+
let dom_logs: Vec<_> = dom_logs
310+
.into_iter()
311+
.filter(|l| tx_hash_set.contains(&l.tx_hash))
312+
.collect();
313+
314+
let dom_token_transfers: Vec<_> = dom_logs
315+
.iter()
316+
.filter_map(crate::token_decode::decode_transfer)
317+
.collect();
318+
276319
write_block(
277320
pool,
278321
BlockBundle {
279322
block: dom_block,
280323
txs: dom_txs,
281324
logs: dom_logs,
325+
token_transfers: dom_token_transfers,
282326
},
283327
analytics,
284328
)

crates/sync/src/block_writer.rs

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,8 @@
1414
use crate::cursor::write_cursor;
1515
use crate::{SyncError, SyncResult};
1616
use indexer_analytics::{AnalyticsHandle, RawTxRow};
17-
use indexer_db::{PgPool, blocks, logs, transactions};
18-
use indexer_domain::{Block, Log, Transaction};
17+
use indexer_db::{PgPool, blocks, logs, token_transfers, transactions};
18+
use indexer_domain::{Block, Log, TokenTransfer, Transaction};
1919

2020
/// Page size for batch inserts. Postgres protocol caps bind params at
2121
/// ~65k per query; the widest table (transactions, 15 cols) tops out at
@@ -32,6 +32,10 @@ pub struct BlockBundle {
3232
pub txs: Vec<Transaction>,
3333
/// All logs emitted during the block's txs, ordered by `log_index`.
3434
pub logs: Vec<Log>,
35+
/// Decoded ERC-20 / ERC-721 transfers from this block's logs. Sync
36+
/// layer fills via `token_decode::decode_transfer`; empty for blocks
37+
/// with no qualifying events.
38+
pub token_transfers: Vec<TokenTransfer>,
3539
}
3640

3741
/// Write a block bundle + advance the chain-wide cursor in one transaction.
@@ -49,14 +53,17 @@ pub async fn write_block(
4953
let mut tx = pool.begin().await.map_err(SyncError::from)?;
5054

5155
// Order matters: blocks first (FK target), then transactions (FK target
52-
// for logs), then logs.
56+
// for logs), then logs, then derived token_transfers.
5357
blocks::insert(&mut *tx, &b.block).await?;
5458
for t in &b.txs {
5559
transactions::insert(&mut *tx, t).await?;
5660
}
5761
for l in &b.logs {
5862
logs::insert(&mut *tx, l).await?;
5963
}
64+
for tt in &b.token_transfers {
65+
token_transfers::insert(&mut *tx, tt).await?;
66+
}
6067

6168
// Cursor advance shares the transaction so it lands or rolls back with
6269
// the data. `now_ts` = the block's chain timestamp so cursor staleness
@@ -146,6 +153,11 @@ pub async fn batch_write_blocks(
146153
all_txs.sort_by_key(|t| (t.block_height, t.tx_index));
147154
let mut all_logs: Vec<Log> = bundles.iter().flat_map(|b| b.logs.clone()).collect();
148155
all_logs.sort_by_key(|l| (l.block_height, l.log_index));
156+
let mut all_transfers: Vec<TokenTransfer> = bundles
157+
.iter()
158+
.flat_map(|b| b.token_transfers.clone())
159+
.collect();
160+
all_transfers.sort_by_key(|t| (t.block_height, t.log_index));
149161

150162
let mut tx = pool.begin().await.map_err(SyncError::from)?;
151163
for chunk in all_blocks.chunks(BATCH_INSERT_CHUNK) {
@@ -157,6 +169,9 @@ pub async fn batch_write_blocks(
157169
for chunk in all_logs.chunks(BATCH_INSERT_CHUNK) {
158170
logs::insert_batch(&mut *tx, chunk).await?;
159171
}
172+
for chunk in all_transfers.chunks(BATCH_INSERT_CHUNK) {
173+
token_transfers::insert_batch(&mut *tx, chunk).await?;
174+
}
160175
write_cursor(&mut *tx, max_height, cursor_ts).await?;
161176
tx.commit().await.map_err(SyncError::from)?;
162177

crates/sync/src/lib.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ pub mod cursor;
2828
pub mod reorg;
2929
pub mod single_flight;
3030
pub mod tail;
31+
pub mod token_decode;
3132

3233
mod convert;
3334

crates/sync/src/token_decode.rs

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
//! ERC-20 / ERC-721 Transfer event decoder.
2+
//!
3+
//! Sentrix indexer-handlers crate is still a placeholder (Phase 0), so the
4+
//! `token_transfers` table sat empty even though raw logs were captured.
5+
//! Until the declarative-handler framework lands, decode the well-known
6+
//! Transfer signatures inline here so scan UIs can resolve token balances.
7+
//!
8+
//! ERC-1155 has a different topic0 (`TransferSingle` / `TransferBatch`) and
9+
//! a richer encoding; out of scope for this pass.
10+
11+
use alloy_primitives::U256;
12+
use indexer_domain::{Log, TokenStandard, TokenTransfer, Wei};
13+
14+
/// `keccak256("Transfer(address,address,uint256)")` — same selector for
15+
/// ERC-20 amount transfers and ERC-721 token-id transfers. The two are
16+
/// distinguished by topic count + data shape.
17+
const TRANSFER_TOPIC0: &str = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef";
18+
19+
/// Try to decode a log as an ERC-20 or ERC-721 Transfer event. Returns
20+
/// `None` for anything else — caller drops it.
21+
///
22+
/// Decoding rules:
23+
/// - topic0 must equal the canonical Transfer selector.
24+
/// - topic1, topic2 are 32-byte-padded from/to addresses (indexed).
25+
/// - ERC-20: topic3 absent, data = 32-byte amount.
26+
/// - ERC-721: topic3 present (indexed token_id), data empty, amount = 1.
27+
pub fn decode_transfer(log: &Log) -> Option<TokenTransfer> {
28+
if log.topic0.as_deref() != Some(TRANSFER_TOPIC0) {
29+
return None;
30+
}
31+
let from_topic = log.topic1.as_deref()?;
32+
let to_topic = log.topic2.as_deref()?;
33+
let from_addr = topic_to_address(from_topic)?;
34+
let to_addr = topic_to_address(to_topic)?;
35+
36+
let (standard, token_id, amount) = match log.topic3.as_deref() {
37+
Some(id_topic) => {
38+
// ERC-721: token_id in topic3, data empty (or padding only).
39+
let token_id = topic_to_u256(id_topic)?;
40+
(
41+
TokenStandard::Erc721,
42+
Some(Wei(token_id)),
43+
Wei(U256::from(1u64)),
44+
)
45+
}
46+
None => {
47+
// ERC-20: amount in data (must be exactly 32 bytes).
48+
let data_str = log.data.as_deref()?;
49+
let amount = data_to_u256(data_str)?;
50+
(TokenStandard::Erc20, None, Wei(amount))
51+
}
52+
};
53+
54+
Some(TokenTransfer {
55+
id: None,
56+
block_height: log.block_height,
57+
tx_hash: log.tx_hash.clone(),
58+
log_index: log.log_index,
59+
contract: log.address.clone(),
60+
standard,
61+
from_addr,
62+
to_addr,
63+
token_id,
64+
amount,
65+
})
66+
}
67+
68+
/// Last 20 bytes of a 32-byte topic → `0x`-prefixed lowercase address.
69+
fn topic_to_address(topic: &str) -> Option<String> {
70+
let hex = topic.trim_start_matches("0x");
71+
if hex.len() != 64 {
72+
return None;
73+
}
74+
Some(format!("0x{}", &hex[24..]))
75+
}
76+
77+
/// 32-byte topic → U256.
78+
fn topic_to_u256(topic: &str) -> Option<U256> {
79+
let hex = topic.trim_start_matches("0x");
80+
if hex.len() != 64 {
81+
return None;
82+
}
83+
let bytes = hex::decode(hex).ok()?;
84+
Some(U256::from_be_slice(&bytes))
85+
}
86+
87+
/// 32-byte data field → U256. ERC-20 Transfer always has data length 32.
88+
fn data_to_u256(data: &str) -> Option<U256> {
89+
let hex = data.trim_start_matches("0x");
90+
if hex.len() != 64 {
91+
return None;
92+
}
93+
let bytes = hex::decode(hex).ok()?;
94+
Some(U256::from_be_slice(&bytes))
95+
}
96+
97+
#[cfg(test)]
98+
mod tests {
99+
use super::*;
100+
use indexer_domain::{BlockHeight, LogIndex};
101+
102+
fn base_log() -> Log {
103+
Log {
104+
block_height: BlockHeight(1),
105+
tx_hash: "0xabc".into(),
106+
log_index: LogIndex(0),
107+
address: "0xcontract".into(),
108+
topic0: Some(TRANSFER_TOPIC0.into()),
109+
topic1: Some(
110+
"0x000000000000000000000000aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into(),
111+
),
112+
topic2: Some(
113+
"0x000000000000000000000000bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".into(),
114+
),
115+
topic3: None,
116+
data: Some("0x0000000000000000000000000000000000000000000000000000000000000064".into()),
117+
}
118+
}
119+
120+
#[test]
121+
fn decodes_erc20_transfer() {
122+
let t = decode_transfer(&base_log()).expect("erc20 should decode");
123+
assert_eq!(t.standard, TokenStandard::Erc20);
124+
assert_eq!(t.from_addr, "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa");
125+
assert_eq!(t.to_addr, "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
126+
assert_eq!(t.amount.0, U256::from(100u64));
127+
assert!(t.token_id.is_none());
128+
}
129+
130+
#[test]
131+
fn decodes_erc721_transfer() {
132+
let mut log = base_log();
133+
log.topic3 =
134+
Some("0x0000000000000000000000000000000000000000000000000000000000000007".into());
135+
log.data = Some("0x".into());
136+
let t = decode_transfer(&log).expect("erc721 should decode");
137+
assert_eq!(t.standard, TokenStandard::Erc721);
138+
assert_eq!(t.token_id.unwrap().0, U256::from(7u64));
139+
assert_eq!(t.amount.0, U256::from(1u64));
140+
}
141+
142+
#[test]
143+
fn skips_non_transfer() {
144+
let mut log = base_log();
145+
log.topic0 = Some("0xdeadbeef".into());
146+
assert!(decode_transfer(&log).is_none());
147+
}
148+
149+
#[test]
150+
fn skips_malformed_topic() {
151+
let mut log = base_log();
152+
log.topic1 = Some("0xshort".into());
153+
assert!(decode_transfer(&log).is_none());
154+
}
155+
}

0 commit comments

Comments
 (0)