tx type in summary

README.md

@@ -47,6 +47,7 @@ $ clickhouse local -q "DESCRIBE TABLE 'transactions.parquet';"
 timestamp               Nullable(DateTime64(3))
 hash                    Nullable(String)
 chainId                 Nullable(String)
+txType                  Nullable(Int64)
 from                    Nullable(String)
 to                      Nullable(String)
 value                   Nullable(String)
@@ -69,7 +70,7 @@ rawTx                   Nullable(String)
 Same as parquet, but without `rawTx`:
 
 ```
-timestamp_ms,hash,chain_id,from,to,value,nonce,gas,gas_price,gas_tip_cap,gas_fee_cap,data_size,data_4bytes,sources,included_at_block_height,included_block_timestamp_ms,inclusion_delay_ms
+timestamp_ms,hash,chain_id,from,to,value,nonce,gas,gas_price,gas_tip_cap,gas_fee_cap,data_size,data_4bytes,sources,included_at_block_height,included_block_timestamp_ms,inclusion_delay_ms,tx_type
 ```
 
 ---
@@ -100,6 +101,9 @@ We recommend to use [ClickHouse local](https://clickhouse.com/docs/en/operations
 # count rows
 $ clickhouse local -q "SELECT count(*) FROM 'transactions.parquet' LIMIT 1;"
 
+# count by transaction type
+$ clickhouse local -q "SELECT txType, COUNT(txType) FROM 'transactions.parquet' GROUP BY txType;"
+
 # show hash+rawTx from first entry
 $ clickhouse local -q "SELECT hash,hex(rawTx) FROM 'transactions.parquet' LIMIT 1;"
 

collector/tx_processor.go

@@ -114,6 +114,10 @@ func (p *TxProcessor) Start() {
 		// send tx to receivers before processing it
 		// this will reduce the latency for the receivers but may lead to receivers getting the same tx multiple times
 		// or getting txs that are incorrect
+		if txIn.Tx == nil {
+			p.log.Errorf("nil tx from source %s", txIn.Source)
+			continue
+		}
 		go p.sendTxToReceivers(txIn)
 		p.processTx(txIn)
 	}

common/analyzer.go

@@ -30,6 +30,10 @@ type Analyzer2 struct {
 	nIncluded           int64
 	nNotIncluded        int64
 
+	txTypes              []int64
+	nTransactionsPerType map[int64]int64
+	txBytesPerType       map[int64]int64
+
 	// landed vs non-landed transactions
 	nTxOnChainBySource    map[string]int64
 	nTxNotOnChainBySource map[string]int64
@@ -57,8 +61,11 @@ func NewAnalyzer2(opts Analyzer2Opts) *Analyzer2 {
 		nTxOnChainBySource:     make(map[string]int64),
 		nTxNotOnChainBySource:  make(map[string]int64),
 		nTxExclusiveIncluded:   make(map[string]map[bool]int64), // [source][isIncluded]count
+		nTransactionsPerType:   make(map[int64]int64),
+		txBytesPerType:         make(map[int64]int64),
 	}
 
+	// Now add all transactions to analyzer cache that were not included before received
 	for _, tx := range opts.Transactions {
 		if tx.WasIncludedBeforeReceived() {
 			continue
@@ -67,6 +74,7 @@ func NewAnalyzer2(opts Analyzer2Opts) *Analyzer2 {
 		a.Transactions[strings.ToLower(tx.Hash)] = tx
 	}
 
+	// Run the analyzer
 	a.init()
 	return a
 }
@@ -83,6 +91,11 @@ func (a *Analyzer2) init() {
 			a.nIncluded += 1
 		}
 
+		// Count transactions per type
+		a.nTransactionsPerType[tx.TxType] += 1
+		a.txBytesPerType[tx.TxType] += int64(len(tx.RawTx)) / 2 // not sure this is correct, the end result seems low for blob transactions...
+
+		// Go over sources
 		for _, src := range tx.Sources {
 			// Count overall tx / source
 			a.nTransactionsPerSource[src] += 1
@@ -129,6 +142,12 @@ func (a *Analyzer2) init() {
 		a.sources = append(a.sources, src)
 	}
 	sort.Strings(a.sources)
+
+	// get sorted list of txTypes
+	for txType := range a.nTransactionsPerType {
+		a.txTypes = append(a.txTypes, txType)
+	}
+	sort.Slice(a.txTypes, func(i, j int) bool { return a.txTypes[i] < a.txTypes[j] })
 }
 
 // latencyComp returns arrays of latency differences for the node that was faster
@@ -240,10 +259,33 @@ func (a *Analyzer2) Sprint() string {
 	out += fmt.Sprintln("-----------------")
 	out += fmt.Sprintln("")
 
-	// Add per-source tx stats
-	var buff bytes.Buffer
+	// TxType count
+	buff := bytes.Buffer{}
 	table := tablewriter.NewWriter(&buff)
 	SetupMarkdownTableWriter(table)
+	table.SetHeader([]string{"Tx Type", "Count"})
+	// table.SetHeader([]string{"Tx Type", "Count", "Size Total", "Size Avg"})
+	for txType := range a.txTypes {
+		count := a.nTransactionsPerType[int64(txType)]
+		table.Append([]string{
+			fmt.Sprint(txType),
+			Printer.Sprintf("%10d (%5s)", count, Int64DiffPercentFmt(count, a.nUniqueTransactions, 1)),
+			// HumanBytes(uint64(a.txBytesPerType[int64(txType)])),
+			// HumanBytes(uint64(a.txBytesPerType[int64(txType)] / count)),
+		})
+	}
+	table.Render()
+	out += buff.String()
+
+	// Add per-source tx stats
+	out += fmt.Sprintln("")
+	out += fmt.Sprintln("------------")
+	out += fmt.Sprintln("Source Stats")
+	out += fmt.Sprintln("------------")
+	out += fmt.Sprintln("")
+	buff = bytes.Buffer{}
+	table = tablewriter.NewWriter(&buff)
+	SetupMarkdownTableWriter(table)
 	table.SetHeader([]string{"Source", "Transactions", "Included on-chain", "Not included"})
 	for _, src := range a.sources {
 		nTx := a.nTransactionsPerSource[src]

website/templates/index_root.html

@@ -15,6 +15,6 @@ <h2>Ethereum Mainnet</h2>
 <hr>
 <p>
 <p><small>The data is dedicated to the public domain under the <a href="https://creativecommons.org/publicdomain/zero/1.0">CC-0 license</a>.</small></p>
-<p><small>Brought to you by <a href="https://flashbots.net">Flashbots</a> in collaboration with <a href="https://bloxroute.com/">bloXroute</a>, <a href="https://www.chainbound.io/">Chainbound</a>, <a href="https://www.edennetwork.io/">Eden</a>, <a href="https://www.infura.io/">Infura</a> and <a href="https://mempool.guru">Mempool Guru</a>.</p>
+<p><small>Brought to you by <a href="https://flashbots.net">Flashbots</a> in collaboration with <a href="https://bloxroute.com/">bloXroute</a>, <a href="https://www.chainbound.io/">Chainbound</a>, <a href="https://www.edennetwork.io/">Eden</a> and <a href="https://mempool.guru">Mempool Guru</a>.</p>
 </p>
 {{ end }}