cleanup!

collector/node_conn_bloxroute.go

@@ -21,7 +21,6 @@ import (
 const (
 	initialBackoffSec = 5
 	maxBackoffSec     = 120
-	srcTag            = "blx"
 )
 
 // options - via https://docs.bloxroute.com/introduction/cloud-api-ips
@@ -40,7 +39,7 @@ type BlxNodeConnection struct {
 
 func NewBlxNodeConnection(log *zap.SugaredLogger, blxAuthHeader string, txC chan TxIn) *BlxNodeConnection {
 	return &BlxNodeConnection{
-		log:           log.With("src", srcTag),
+		log:           log.With("src", common.BloxrouteTag),
 		blxAuthHeader: blxAuthHeader,
 		txC:           txC,
 		backoffSec:    initialBackoffSec,
@@ -127,6 +126,6 @@ func (nc *BlxNodeConnection) connect() {
 			continue
 		}
 
-		nc.txC <- TxIn{time.Now().UTC(), &tx, srcTag}
+		nc.txC <- TxIn{time.Now().UTC(), &tx, common.BloxrouteTag}
 	}
 }

common/consts.go

@@ -2,6 +2,8 @@ package common
 
 import "strings"
 
+const BloxrouteTag = "blx"
+
 func TxSourcName(uri string) string {
 	sourceAlias := SourceAliasesFromEnv()
 	if alias, ok := sourceAlias[uri]; ok {

scripts/analyze-source-stats/analyzer.go

@@ -10,12 +10,10 @@ import (
 )
 
 const (
-	referenceSource    = "local"
-	referenceSource2   = "apool"
-	referenceBloxroute = "blx"
+	referenceLocalSource = "local"
 )
 
-var bucketsMS = []int64{1, 5, 10, 50, 100, 250, 500, 1000, 2000} // note: 0 would be equal timestamps
+var bucketsMS = []int64{1, 5, 50, 100, 250, 500, 1000, 2000} // note: 0 would be equal timestamps
 
 func prettyInt(i int) string {
 	return printer.Sprintf("%d", i)
@@ -46,9 +44,6 @@ type Analyzer struct {
 	timeFirst      time.Time
 	timeLast       time.Time
 	duration       time.Duration
-
-	bloxrouteTxBeforeLocal         map[int64]int64 // [bucket_ms] = count
-	nbloxrouteSeenBeforeLocalTotal int64
 }
 
 func NewAnalyzer(transactions map[string]map[string]int64) *Analyzer {
@@ -57,15 +52,14 @@ func NewAnalyzer(transactions map[string]map[string]int64) *Analyzer {
 		nTransactionsPerSource: make(map[string]int64),
 		nUniqueTxPerSource:     make(map[string]int64),
 		nNotSeenLocalPerSource: make(map[string]int64),
-		bloxrouteTxBeforeLocal: make(map[int64]int64),
 	}
 
 	a.init()
 	return a
 }
 
 // Init does some efficient initial data analysis and preparation for later use
-func (a *Analyzer) init() { //nolint:gocognit
+func (a *Analyzer) init() {
 	// unique tx
 	a.nUniqueTx = len(a.txs)
 
@@ -75,14 +69,14 @@ func (a *Analyzer) init() { //nolint:gocognit
 		a.nAllTx += len(sources)
 
 		// number of unique tx -- special case for local+apool
-		if len(sources) == 2 {
-			if sources[referenceSource] != 0 && sources[referenceSource2] != 0 {
-				a.nUniqueTxPerSource[referenceSource] += 1
-				a.nTxSeenBySingleSource += 1
-			}
-		}
-
-		if sources[referenceSource] == 0 {
+		// if len(sources) == 2 {
+		// 	if sources[referenceSource] != 0 && sources[referenceSource2] != 0 {
+		// 		a.nUniqueTxPerSource[referenceSource] += 1
+		// 		a.nTxSeenBySingleSource += 1
+		// 	}
+		// }
+
+		if sources[referenceLocalSource] == 0 {
 			a.nOverallNotSeenLocal += 1
 		}
 
@@ -95,7 +89,7 @@ func (a *Analyzer) init() { //nolint:gocognit
 			}
 
 			// remember if this transaction was not seen by the reference source
-			if sources[referenceSource] == 0 {
+			if sources[referenceLocalSource] == 0 {
 				a.nNotSeenLocalPerSource[src] += 1
 			}
 
@@ -122,39 +116,42 @@ func (a *Analyzer) init() { //nolint:gocognit
 		a.sources = append(a.sources, src)
 	}
 	sort.Strings(a.sources)
-
-	// bloxroute specific analysis
-	a.initBlx()
 }
 
-func (a *Analyzer) initBlx() {
+func (a *Analyzer) benchmarkSourceVsLocal(src, ref string) (srcFirstBuckets map[int64]int64, totalFirstBySrc, totalSeenByBoth int) {
+	srcFirstBuckets = make(map[int64]int64) // [bucket_ms] = count
+
 	// How much earlier were transactions received by blx vs. the local node?
 	for _, sources := range a.txs {
 		if len(sources) == 1 {
 			continue
 		}
 
-		// ensure seen by both local and blx
-		if _, seenByBlx := sources[referenceBloxroute]; !seenByBlx {
+		// ensure tx was seen by both source and reference nodes
+		if _, seenBySrc := sources[src]; !seenBySrc {
 			continue
 		}
-		if _, seenLocally := sources[referenceSource]; !seenLocally {
+		if _, seenByRef := sources[ref]; !seenByRef {
 			continue
 		}
 
-		blxTS := sources[referenceBloxroute]
-		refTS := sources[referenceSource]
-		diff := blxTS - refTS
+		totalSeenByBoth += 1
+
+		srcTS := sources[src]
+		localTS := sources[ref]
+		diff := srcTS - localTS
 
 		if diff > 0 {
-			a.nbloxrouteSeenBeforeLocalTotal += 1
+			totalFirstBySrc += 1
 			for _, thresholdMS := range bucketsMS {
 				if diff >= thresholdMS {
-					a.bloxrouteTxBeforeLocal[thresholdMS] += 1
+					srcFirstBuckets[thresholdMS] += 1
 				}
 			}
 		}
 	}
+
+	return srcFirstBuckets, totalFirstBySrc, totalSeenByBoth
 }
 
 func (a *Analyzer) Print() {
@@ -165,8 +162,13 @@ func (a *Analyzer) Print() {
 	fmt.Println("")
 	fmt.Printf("Sources: %s \n", strings.Join(a.sources, ", "))
 	fmt.Println("")
-	fmt.Printf("- Unique transactions: %8s \n", prettyInt(a.nUniqueTx))
-	fmt.Printf("- All transactions:    %8s \n", prettyInt(a.nAllTx))
+	fmt.Printf("- Transactions: %8s \n", prettyInt(a.nAllTx))
+	fmt.Printf("- Unique txs:   %8s \n", prettyInt(a.nUniqueTx))
+
+	fmt.Println("")
+	fmt.Println("-------------")
+	fmt.Println("Overall stats")
+	fmt.Println("-------------")
 
 	fmt.Println("")
 	fmt.Printf("All transactions received: %s \n", prettyInt(a.nAllTx))
@@ -188,17 +190,34 @@ func (a *Analyzer) Print() {
 	fmt.Println("")
 	fmt.Printf("Transactions not seen by local node: %s / %s (%s)\n", prettyInt64(a.nOverallNotSeenLocal), prettyInt(a.nUniqueTx), common.Int64DiffPercentFmt(a.nOverallNotSeenLocal, int64(a.nUniqueTx)))
 	for _, src := range a.sources {
-		if a.nTransactionsPerSource[src] > 0 && src != referenceSource {
+		if a.nTransactionsPerSource[src] > 0 && src != referenceLocalSource {
 			cnt := a.nNotSeenLocalPerSource[src]
 			fmt.Printf("- %-8s %10s\n", src, prettyInt64(cnt))
 		}
 	}
 
+	// latency analysis for various sources:
 	fmt.Println("")
-	fmt.Printf("Bloxroute transactions received before local node: %s / %s (%s) \n", prettyInt64(a.nbloxrouteSeenBeforeLocalTotal), prettyInt(a.nUniqueTx), common.Int64DiffPercentFmt(a.nbloxrouteSeenBeforeLocalTotal, int64(a.nUniqueTx)))
-	for _, bucketMS := range bucketsMS {
-		s := fmt.Sprintf("%d ms", bucketMS)
-		cnt := a.bloxrouteTxBeforeLocal[bucketMS]
-		fmt.Printf(" - %-8s %8s \n", s, prettyInt64(cnt))
+	fmt.Println("------------------")
+	fmt.Println("Latency comparison")
+	fmt.Println("------------------")
+	latencyComps := []struct{ src, ref string }{
+		{common.BloxrouteTag, referenceLocalSource},
+		{"apool2", referenceLocalSource},
+		{"apool2", common.BloxrouteTag},
+		{common.BloxrouteTag, "apool2"},
+	}
+
+	for _, comp := range latencyComps {
+		srcFirstBuckets, totalFirstBySrc, _ := a.benchmarkSourceVsLocal(comp.src, comp.ref)
+
+		fmt.Println("")
+		// fmt.Printf("%s transactions received before %s: %s / %s (%s) \n", comp.src, comp.ref, prettyInt64(int64(totalFirstBySrc)), prettyInt64(int64(totalSeenByBoth)), common.Int64DiffPercentFmt(int64(totalFirstBySrc), int64(totalSeenByBoth)))
+		fmt.Printf("%s transactions received before %s: %s \n", comp.src, comp.ref, prettyInt64(int64(totalFirstBySrc)))
+		for _, bucketMS := range bucketsMS {
+			s := fmt.Sprintf("%d ms", bucketMS)
+			cnt := srcFirstBuckets[bucketMS]
+			fmt.Printf(" - %-8s %8s \n", s, prettyInt64(cnt))
+		}
 	}
 }

scripts/analyze-source-stats/main.go

@@ -9,29 +9,7 @@ Input: CSV file(s) with the following format:
 
 Output (currently):
 
-	2023-08-30T20:34:47.253Z        INFO    Processed all input files       {"files": 22, "txTotal": "627,891", "memUsedMiB": "594"}
-	2023-08-30T20:34:47.648Z        INFO    Overall tx count        {"infura": "578,606", "alchemy": "568,790", "ws://localhost:8546": "593,046", "blx": "419,725"}
-	2023-08-30T20:34:47.696Z        INFO    Unique tx count {"blx": "22,403", "ws://localhost:8546": "9,962", "alchemy": "2,940", "infura": "4,658", "unique": "39,963 / 627,891"}
-	2023-08-30T20:34:47.816Z        INFO    Not seen by local node  {"blx": "23,895", "infura": "9,167", "alchemy": "7,039", "notSeenByRef": "34,845 / 627,891"}
-
-	Total unique tx: 627,891
-
-	Transactions received:
-	- alchemy: 			   568,790
-	- blx:				   419,725
-	- infura: 			   578,606
-	- ws://localhost:8546: 593,046
-
-	Unique tx (sole sender):
-	- alchemy: 				2,940
-	- blx: 					22,403
-	- infura: 				4,658
-	- ws://localhost:8546: 	9,962
-
-	Transactions not seen by local node: 34,845 / 627,891
-	- alchemy: 	7,039
-	- blx: 		23,895
-	- infura: 	9,167
+
 
 more insight ideas?
 - who sent first