From 2007b349081b26cf6c8ee6ccda4a5825d6e887aa Mon Sep 17 00:00:00 2001 From: Mohammed Sohail Date: Tue, 2 Apr 2024 16:11:30 +0800 Subject: [PATCH] feat: add badger blocks db --- .gitignore | 1 + cmd/tracker/main.go | 49 +++++++++---- config.toml | 9 ++- go.mod | 11 +++ go.sum | 33 +++++++++ internal/batcher/batcher.go | 6 -- internal/chain/block.go | 14 ++++ internal/chain/chain.go | 2 +- internal/db/.gitkeep | 0 internal/db/blocks.go | 125 ++++++++++++++++++++++++++++++++ internal/db/db.go | 96 ++++++++++++++++++++++++ internal/processor/processor.go | 31 +++++--- internal/queue/queue.go | 13 ---- internal/stats/stats.go | 2 +- internal/syncer/historical.go | 67 ++++++++++++----- internal/syncer/syncer.go | 51 +++++++------ revive.toml | 44 +++++++++++ 17 files changed, 461 insertions(+), 93 deletions(-) create mode 100644 .gitignore delete mode 100644 internal/batcher/batcher.go delete mode 100644 internal/db/.gitkeep create mode 100644 internal/db/blocks.go create mode 100644 internal/db/db.go delete mode 100644 internal/queue/queue.go create mode 100644 revive.toml diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0207909 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +celo_tracker_blocks_db \ No newline at end of file diff --git a/cmd/tracker/main.go b/cmd/tracker/main.go index 1a1a843..7a66ba9 100644 --- a/cmd/tracker/main.go +++ b/cmd/tracker/main.go @@ -13,10 +13,11 @@ import ( "github.com/celo-org/celo-blockchain/core/types" "github.com/ef-ds/deque/v2" - "github.com/grassrootseconomics/celo-events/internal/chain" - "github.com/grassrootseconomics/celo-events/internal/processor" - "github.com/grassrootseconomics/celo-events/internal/stats" - "github.com/grassrootseconomics/celo-events/internal/syncer" + "github.com/grassrootseconomics/celo-tracker/internal/chain" + "github.com/grassrootseconomics/celo-tracker/internal/db" + "github.com/grassrootseconomics/celo-tracker/internal/processor" + "github.com/grassrootseconomics/celo-tracker/internal/stats" + "github.com/grassrootseconomics/celo-tracker/internal/syncer" "github.com/knadh/koanf/v2" ) @@ -59,49 +60,65 @@ func main() { - RealtimeSyncer - BlockProcessor */ - stats := stats.NewStats(lo) + stats := stats.New(lo) - chain, err := chain.NewChainProvider(chain.ChainOpts{ + chain, err := chain.New(chain.ChainOpts{ RPCEndpoint: ko.MustString("chain.rpc_endpoint"), TestNet: ko.Bool("chain.testnet"), Logg: lo, }) if err != nil { lo.Error("could not initialize chain client", "error", err) + os.Exit(1) } - chainSyncer, err := syncer.NewSyncer(syncer.SyncerOpts{ + db, err := db.New(db.DBOpts{ + Logg: lo, + }) + if err != nil { + lo.Error("could not initialize blocks db", "error", err) + os.Exit(1) + } + + chainSyncer, err := syncer.New(syncer.SyncerOpts{ WebSocketEndpoint: ko.MustString("chain.ws_endpoint"), BatchQueue: &batchQueue, BlocksQueue: &blocksQueue, Chain: chain, Logg: lo, Stats: stats, + DB: db, + InitialLowerBound: uint64(ko.MustInt64("chain.start_block")), }) if err != nil { lo.Error("could not initialize chain syncer", "error", err) + os.Exit(1) + } + if err := chainSyncer.BootstrapHistoricalSyncer(); err != nil { + lo.Error("could not bootstrap historical syncer", "error", err) + os.Exit(1) } - // chainSyncer.BootstrapHistoricalSyncer() blockProcessor := processor.NewProcessor(processor.ProcessorOpts{ Chain: chain, BlocksQueue: &blocksQueue, Logg: lo, Stats: stats, + DB: db, }) + wg.Add(1) + go func() { + defer wg.Done() + chainSyncer.StartHistoricalSyncer(ctx) + }() + wg.Add(1) go func() { defer wg.Done() chainSyncer.StartRealtimeSyncer(ctx) }() - // wg.Add(1) - // go func() { - // defer wg.Done() - // chainSyncer.StartHistoricalSyncer(ctx) - // }() - wg.Add(1) go func() { defer wg.Done() @@ -109,12 +126,16 @@ func main() { }() <-ctx.Done() + lo.Info("shutdown signal received") shutdownCtx, cancel := context.WithTimeout(context.Background(), defaultGracefulShutdownPeriod) wg.Add(1) go func() { defer wg.Done() blockProcessor.Stop() + if err := db.Close(); err != nil { + lo.Error("error closing db", "error", err) + } }() go func() { diff --git a/config.toml b/config.toml index 7eb00e5..265db5b 100644 --- a/config.toml +++ b/config.toml @@ -8,8 +8,9 @@ go_process = true address = ":5001" [chain] +start_block = 24873500 graphql_endpoint = "" -ws_endpoint = "wss://ws.celo.grassecon.net" -rpc_endpoint = "https://1rpc.io/4aU926thTN5bM1BJV/celo" -testnet = false -registry_address = "" \ No newline at end of file +ws_endpoint = "wss://ws.celo.grassecon.net" +rpc_endpoint = "https://celo.grassecon.net" +testnet = false +registry_address = "" diff --git a/go.mod b/go.mod index 8ce1979..4d62002 100644 --- a/go.mod +++ b/go.mod @@ -31,11 +31,19 @@ require ( github.com/cespare/xxhash/v2 v2.2.0 // indirect github.com/deckarep/golang-set v1.8.0 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect + github.com/dgraph-io/badger/v4 v4.2.0 // indirect + github.com/dgraph-io/ristretto v0.1.1 // indirect + github.com/dustin/go-humanize v1.0.0 // indirect github.com/fsnotify/fsnotify v1.6.0 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/go-stack/stack v1.8.1 // indirect github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/golang/glog v1.0.0 // indirect + github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7 // indirect + github.com/golang/protobuf v1.5.2 // indirect github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect + github.com/google/flatbuffers v1.12.1 // indirect github.com/gorilla/websocket v1.5.0 // indirect github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect github.com/hdevalence/ed25519consensus v0.0.0-20201207055737-7fde80a9d5ff // indirect @@ -43,6 +51,7 @@ require ( github.com/holiman/uint256 v1.2.4 // indirect github.com/huin/goupnp v1.0.3 // indirect github.com/jackpal/go-nat-pmp v1.0.2 // indirect + github.com/klauspost/compress v1.12.3 // indirect github.com/knadh/koanf/maps v0.1.1 // indirect github.com/lmittmann/tint v1.0.4 // indirect github.com/mattn/go-runewidth v0.0.14 // indirect @@ -58,6 +67,7 @@ require ( github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 // indirect github.com/tklauser/go-sysconf v0.3.10 // indirect github.com/tklauser/numcpus v0.4.0 // indirect + go.opencensus.io v0.22.5 // indirect golang.org/x/crypto v0.19.0 // indirect golang.org/x/net v0.17.0 // indirect golang.org/x/sync v0.3.0 // indirect @@ -65,6 +75,7 @@ require ( golang.org/x/text v0.14.0 // indirect golang.org/x/time v0.5.0 // indirect golang.org/x/xerrors v0.0.0-20220517211312-f3a8303e98df // indirect + google.golang.org/protobuf v1.28.1 // indirect gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) diff --git a/go.sum b/go.sum index 911cedf..5fb60d7 100644 --- a/go.sum +++ b/go.sum @@ -138,12 +138,19 @@ github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1/go.mod h1:hyedUtir6IdtD/7lIxGeC github.com/decred/dcrd/lru v1.0.0/go.mod h1:mxKOwFd7lFjN2GZYsiz/ecgqR6kkYAl+0pz0tEMk218= github.com/deepmap/oapi-codegen v1.6.0/go.mod h1:ryDa9AgbELGeB+YEXE1dR53yAjHwFvE9iAUlWl9Al3M= github.com/deepmap/oapi-codegen v1.8.2/go.mod h1:YLgSKSDv/bZQB7N4ws6luhozi3cEdRktEqrX88CvjIw= +github.com/dgraph-io/badger/v4 v4.2.0 h1:kJrlajbXXL9DFTNuhhu9yCx7JJa4qpYWxtE8BzuWsEs= +github.com/dgraph-io/badger/v4 v4.2.0/go.mod h1:qfCqhPoWDFJRx1gp5QwwyGo8xk1lbHUxvK9nK0OGAak= +github.com/dgraph-io/ristretto v0.1.1 h1:6CWw5tJNgpegArSHpNHJKldNeq03FQCwYvfMVWajOK8= +github.com/dgraph-io/ristretto v0.1.1/go.mod h1:S1GPSBCYCIhmVNfcth17y2zZtQT6wzkzgwUve0VDWWA= github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ= github.com/dgryski/go-bitstream v0.0.0-20180413035011-3522498ce2c8/go.mod h1:VMaSuZ+SZcx/wljOQKvp5srsbCiKDEb6K2wC4+PiBmQ= +github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw= github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no= github.com/dlclark/regexp2 v1.2.0/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc= github.com/docker/docker v1.6.1/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= github.com/dop251/goja v0.0.0-20200721192441-a695b0cdd498/go.mod h1:Mw6PkjjMXWbTj+nnj4s3QPXq1jaT0s5pC0iFD4+BOAA= +github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo= +github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= github.com/eclipse/paho.mqtt.golang v1.2.0/go.mod h1:H9keYFcgq3Qr5OUJm/JZI/i6U7joQ8SYLhZwfeOo6Ts= github.com/ef-ds/deque/v2 v2.0.2 h1:GQtDK1boBMu/qsNbSLQsqzwNptaioxZI39X3UxT5ALA= github.com/ef-ds/deque/v2 v2.0.2/go.mod h1:hoZy4VooWLhRT4uS+sSCilfgBQUNptJU2FGqr08a5sc= @@ -186,10 +193,15 @@ github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1/go.mod h1:oJDH3BJKyqBA2TXFhDs github.com/gofrs/uuid v3.3.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0/go.mod h1:E/TSTwGwJL78qG/PmXZO1EjYhfJinVAhrmmHX6Z8B9k= github.com/golang/geo v0.0.0-20190916061304-5b978397cfec/go.mod h1:QZ0nwyI2jOfgRAoBvP+ab5aRr7c9x7lhGEJrKvBwjWI= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= +github.com/golang/glog v1.0.0 h1:nfP3RFugxnNRyKgeWd4oI1nYvXpxrx8ck8ZrcizshdQ= +github.com/golang/glog v1.0.0/go.mod h1:EWib/APOK0SL3dFbYqvxE3UYd8E6s1ouQ7iEp/0LWV4= github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= +github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7 h1:5ZkaAPbicIKTF2I64qf5Fh8Aa83Q/dnOafMYV0OMwjA= github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A= github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A= @@ -204,6 +216,9 @@ github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:W github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0= github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw= +github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= @@ -213,12 +228,15 @@ github.com/golangci/lint-1 v0.0.0-20181222135242-d2cdd8c08219/go.mod h1:/X8TswGS github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/flatbuffers v1.11.0/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8= +github.com/google/flatbuffers v1.12.1 h1:MVlul7pQNoDzWRLTw5imwYsl+usrS1TXG2H4jg6ImGw= +github.com/google/flatbuffers v1.12.1/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8= github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M= github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/gofuzz v1.1.1-0.20200604201612-c04b05f3adfa/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= @@ -294,9 +312,12 @@ github.com/kamikazechaser/common v0.2.0 h1:bqi5UaMTDm/wtZlJEvQDNhsLVJP4Beg+HKWeQ github.com/kamikazechaser/common v0.2.0/go.mod h1:I1LEc8+W+g/KHZWARc1gMhuSa2STbQgfL4Hao6I/ZwY= github.com/karalabe/usb v0.0.0-20190919080040-51dc0efba356/go.mod h1:Od972xHfMJowv7NGVDiWVxk2zxnWgjLlJzE+F4F7AGU= github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4= github.com/klauspost/compress v1.4.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A= +github.com/klauspost/compress v1.12.3 h1:G5AfA94pHPysR56qqrkO2pxEexdDzrpFJ6yt/VqWxVU= +github.com/klauspost/compress v1.12.3/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg= github.com/klauspost/cpuid v0.0.0-20170728055534-ae7887de9fa5/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek= github.com/klauspost/crc32 v0.0.0-20161016154125-cb6bfca970f6/go.mod h1:+ZoRqAPRLkC4NPOvfYeR5KNOrY6TD+/sAC3HXPZgDYg= github.com/klauspost/pgzip v1.0.2-0.20170402124221-0bf5dcad4ada/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs= @@ -461,10 +482,13 @@ github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPU github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ= github.com/willf/bitset v1.1.3/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4= github.com/xlab/treeprint v0.0.0-20180616005107-d6fb6747feb6/go.mod h1:ce1O1j6UtZfjr22oyGxGLbauSBp2YVXpARAosm7dHBg= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU= go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8= go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= +go.opencensus.io v0.22.5 h1:dntmOdLpSpHlVqbW5Eay97DelsZHe+55D+xC6i0dDS0= +go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk= go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= @@ -507,6 +531,7 @@ golang.org/x/mobile v0.0.0-20190719004257-d2bd2a29d028/go.mod h1:E/iHnbuqvinMTCc golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc= golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY= golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/net v0.0.0-20180719180050-a680a1efc54d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -525,6 +550,7 @@ golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLL golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20200813134508-3edf25e44fcc/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= @@ -591,6 +617,7 @@ golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20221010170243-090e33056c14/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y= golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= @@ -636,6 +663,8 @@ golang.org/x/tools v0.0.0-20191125144606-a911d9008d1f/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20191216173652-a0e659d51361/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= golang.org/x/tools v0.0.0-20191227053925-7b8e75db28f4/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= golang.org/x/tools v0.0.0-20200108203644-89082a384178/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -686,6 +715,10 @@ google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQ google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE= google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo= google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w= +google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/batcher/batcher.go b/internal/batcher/batcher.go deleted file mode 100644 index b2e3bc1..0000000 --- a/internal/batcher/batcher.go +++ /dev/null @@ -1,6 +0,0 @@ -package batcher - -type ( - Batcher struct { - } -) diff --git a/internal/chain/block.go b/internal/chain/block.go index bb6da6b..cf385a4 100644 --- a/internal/chain/block.go +++ b/internal/chain/block.go @@ -39,3 +39,17 @@ func (c *Chain) GetBlock(ctx context.Context, blockNumber uint64) (types.Block, return block, nil } + +func (c *Chain) GetLatestBlock(ctx context.Context) (uint64, error) { + var ( + latestBlock big.Int + ) + + latestBlockCall := eth.BlockNumber().Returns(&latestBlock) + + if err := c.provider.Client.CallCtx(ctx, latestBlockCall); err != nil { + return 0, err + } + + return latestBlock.Uint64(), nil +} diff --git a/internal/chain/chain.go b/internal/chain/chain.go index 24ac854..9433121 100644 --- a/internal/chain/chain.go +++ b/internal/chain/chain.go @@ -19,7 +19,7 @@ type ( } ) -func NewChainProvider(o ChainOpts) (*Chain, error) { +func New(o ChainOpts) (*Chain, error) { providerOpts := celoutils.ProviderOpts{ RpcEndpoint: o.RPCEndpoint, ChainId: celoutils.MainnetChainId, diff --git a/internal/db/.gitkeep b/internal/db/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/internal/db/blocks.go b/internal/db/blocks.go new file mode 100644 index 0000000..f8f1c92 --- /dev/null +++ b/internal/db/blocks.go @@ -0,0 +1,125 @@ +package db + +import ( + "bytes" + + "github.com/bits-and-blooms/bitset" + "github.com/dgraph-io/badger/v4" +) + +func (d *DB) SetLowerBound(v uint64) error { + return d.setUint64(lowerBoundKey, v) +} + +func (d *DB) GetLowerBound() (uint64, error) { + v, err := d.get(lowerBoundKey) + if err != nil { + return 0, err + } + return unmarshalUint64(v), nil +} + +func (d *DB) SetUpperBound(v uint64) error { + return d.setUint64(upperBoundKey, v) +} + +func (d *DB) GetUpperBound() (uint64, error) { + v, err := d.get(upperBoundKey) + if err != nil { + return 0, err + } + return unmarshalUint64(v), nil +} + +func (d *DB) SetValue(v uint64) error { + return d.setUint64AsKey(v) +} + +func (d *DB) GetMissingValuesBitSet(lowerBound uint64, upperBound uint64) (*bitset.BitSet, error) { + var ( + b bitset.BitSet + ) + + err := d.db.View(func(txn *badger.Txn) error { + var ( + lowerRaw = marshalUint64(lowerBound) + upperRaw = marshalUint64(upperBound) + ) + + for i := lowerBound; i <= upperBound; i++ { + b.Set(uint(i)) + } + + opts := badger.DefaultIteratorOptions + opts.PrefetchValues = false + + iter := txn.NewIterator(opts) + defer iter.Close() + + for iter.Seek(lowerRaw); iter.Valid(); iter.Next() { + k := iter.Item().Key() + + if bytes.Compare(k, upperRaw) > 0 { + return nil + } + + b.Clear(uint(unmarshalUint64(k))) + } + + return nil + }) + if err != nil { + return nil, err + } + + return &b, nil +} + +func (d *DB) Cleanup() error { + var ( + safeToDeleteKeys [][]byte + ) + + err := d.db.View(func(txn *badger.Txn) error { + lowerBound, err := d.get(lowerBoundKey) + if err != nil { + return err + } + + lowerBound = marshalUint64(unmarshalUint64(lowerBound) - 1) + + opts := badger.DefaultIteratorOptions + opts.PrefetchValues = false + + it := txn.NewIterator(opts) + defer it.Close() + + for it.Rewind(); it.Valid(); it.Next() { + k := it.Item().Key() + + if bytes.Compare(k, lowerBound) > 0 { + return nil + } + + safeToDeleteKeys = append(safeToDeleteKeys, it.Item().KeyCopy(nil)) + } + + return nil + }) + if err != nil { + return err + } + + wb := d.db.NewWriteBatch() + for _, k := range safeToDeleteKeys { + if err := wb.Delete(k); err != nil { + return nil + } + } + + if err := wb.Flush(); err != nil { + return err + } + + return nil +} diff --git a/internal/db/db.go b/internal/db/db.go new file mode 100644 index 0000000..86681c9 --- /dev/null +++ b/internal/db/db.go @@ -0,0 +1,96 @@ +package db + +import ( + "encoding/binary" + "log/slog" + + "github.com/dgraph-io/badger/v4" +) + +type ( + DBOpts struct { + Logg *slog.Logger + } + + DB struct { + db *badger.DB + logg *slog.Logger + } +) + +const ( + dbFolderName = "celo_tracker_blocks_db" + + upperBoundKey = "upper" + lowerBoundKey = "lower" +) + +var ( + sortableOrder = binary.BigEndian +) + +func New(o DBOpts) (*DB, error) { + opts := badger.DefaultOptions(dbFolderName) + opts.Logger = nil + db, err := badger.Open(opts) + if err != nil { + return nil, err + } + + return &DB{ + db: db, + logg: o.Logg, + }, nil +} + +func (d *DB) Close() error { + return d.db.Close() +} + +func (d *DB) get(k string) ([]byte, error) { + var v []byte + err := d.db.View(func(txn *badger.Txn) error { + item, err := txn.Get([]byte(k)) + if err != nil { + return err + } + v, err = item.ValueCopy(nil) + return err + }) + + if err != nil { + return nil, err + } + + return v, nil +} + +func (d *DB) setUint64(k string, v uint64) error { + err := d.db.Update(func(txn *badger.Txn) error { + return txn.Set([]byte(k), marshalUint64(v)) + }) + if err != nil { + return err + } + return nil +} + +func (d *DB) setUint64AsKey(v uint64) error { + err := d.db.Update(func(txn *badger.Txn) error { + return txn.Set(marshalUint64(v), nil) + }) + if err != nil { + return err + } + return nil +} + +func unmarshalUint64(b []byte) uint64 { + return sortableOrder.Uint64(b) +} + +func marshalUint64(v uint64) []byte { + b := make([]byte, 8) + sortableOrder.PutUint64(b, v) + return b +} diff --git a/internal/processor/processor.go b/internal/processor/processor.go index 81341be..ac8aceb 100644 --- a/internal/processor/processor.go +++ b/internal/processor/processor.go @@ -5,12 +5,12 @@ import ( "log/slog" "github.com/alitto/pond" - "github.com/celo-org/celo-blockchain/common" "github.com/celo-org/celo-blockchain/core/types" "github.com/ef-ds/deque/v2" - "github.com/grassrootseconomics/celo-events/internal/chain" - "github.com/grassrootseconomics/celo-events/internal/pool" - "github.com/grassrootseconomics/celo-events/internal/stats" + "github.com/grassrootseconomics/celo-tracker/internal/chain" + "github.com/grassrootseconomics/celo-tracker/internal/db" + "github.com/grassrootseconomics/celo-tracker/internal/pool" + "github.com/grassrootseconomics/celo-tracker/internal/stats" ) type ( @@ -19,6 +19,7 @@ type ( BlocksQueue *deque.Deque[types.Block] Logg *slog.Logger Stats *stats.Stats + DB *db.DB } Processor struct { @@ -27,6 +28,7 @@ type ( blocksQueue *deque.Deque[types.Block] logg *slog.Logger stats *stats.Stats + db *db.DB } ) @@ -37,17 +39,19 @@ func NewProcessor(o ProcessorOpts) *Processor { blocksQueue: o.BlocksQueue, logg: o.Logg, stats: o.Stats, + db: o.DB, } } -func (p *Processor) Start(ctx context.Context) error { +func (p *Processor) Start(ctx context.Context) { for { select { case <-ctx.Done(): p.logg.Info("block processor shutting down") - return nil + p.Stop() + return default: - for p.blocksQueue.Len() > 0 { + if p.blocksQueue.Len() > 0 { v, _ := p.blocksQueue.PopFront() p.pool.Submit(func() { if err := p.processBlock(v); err != nil { @@ -65,8 +69,9 @@ func (p *Processor) Stop() { func (p *Processor) processBlock(block types.Block) error { ctx := context.Background() + blockNumber := block.NumberU64() - transactionsResp, err := p.chain.GetTransactions(ctx, block) + _, err := p.chain.GetTransactions(ctx, block) if err != nil { return err } @@ -76,12 +81,16 @@ func (p *Processor) processBlock(block types.Block) error { return err } - for i, receipt := range receiptsResp { + for _, receipt := range receiptsResp { if receipt.Status < 1 { - p.logg.Warn("reverted receipt", "tx_hash", receipt.TxHash, "input_data", common.Bytes2Hex(transactionsResp[i].Data())) + // } - p.logg.Info("successful receipt", "tx_hash", receipt.TxHash, "status", receipt.Status) } + if err := p.db.SetValue(blockNumber); err != nil { + return err + } + p.logg.Debug("successfully processed block", "block", blockNumber) + return nil } diff --git a/internal/queue/queue.go b/internal/queue/queue.go deleted file mode 100644 index fdfb360..0000000 --- a/internal/queue/queue.go +++ /dev/null @@ -1,13 +0,0 @@ -package queue - -import ( - "github.com/celo-org/celo-blockchain/core/types" - "github.com/ef-ds/deque/v2" -) - -type ( - Queue struct { - BlocksQueue *deque.Deque[types.Block] - BatchQueue *deque.Deque[uint64] - } -) diff --git a/internal/stats/stats.go b/internal/stats/stats.go index f663ff4..819f20d 100644 --- a/internal/stats/stats.go +++ b/internal/stats/stats.go @@ -22,7 +22,7 @@ type ( } ) -func NewStats(logg *slog.Logger) *Stats { +func New(logg *slog.Logger) *Stats { return &Stats{ logg: logg, } diff --git a/internal/syncer/historical.go b/internal/syncer/historical.go index cfa586c..88a9d67 100644 --- a/internal/syncer/historical.go +++ b/internal/syncer/historical.go @@ -3,42 +3,71 @@ package syncer import ( "context" "fmt" + + "github.com/dgraph-io/badger/v4" ) -func (s *Syncer) BootstrapHistoricalSyncer() { - // logg here - for i, e := s.db.NextSet(0); e; i, e = s.db.NextSet(i + 1) { - if i > 0 { - s.batchQueue.PushBack(uint64(i)) +const ( + blockBatchSize = 100 +) + +func (s *Syncer) BootstrapHistoricalSyncer() error { + v, err := s.db.GetLowerBound() + if err != nil { + if err == badger.ErrKeyNotFound { + if err := s.db.SetLowerBound(s.initialLowerBound); err != nil { + return err + } + v = s.initialLowerBound + } else { + return err } } + + latestBlock, err := s.chain.GetLatestBlock(context.Background()) + if err != nil { + return err + } + if err := s.db.SetUpperBound(latestBlock); err != nil { + return err + } + + missingBlocks, err := s.db.GetMissingValuesBitSet(v, latestBlock) + if err != nil { + return err + } + missingBlocksCount := missingBlocks.Count() + s.logg.Info("bootstrapping historical syncer", "missing_blocks", missingBlocksCount, "lower_bound", v, "upper_bound", latestBlock) + + buffer := make([]uint, missingBlocksCount) + missingBlocks.NextSetMany(0, buffer) + for _, v := range buffer { + s.batchQueue.PushFront(uint64(v)) + } + + return nil } func (s *Syncer) StartHistoricalSyncer(ctx context.Context) error { + s.logg.Info("starting historical syncer", "batch_size", blockBatchSize) for { select { case <-ctx.Done(): s.logg.Info("historical syncer shutting down") return nil default: - for s.batchQueue.Len() > 0 { + if s.batchQueue.Len() > 0 { var ( currentIterLen = s.batchQueue.Len() - batch []uint64 ) - if currentIterLen < blockBatchSize { - batch = make([]uint64, currentIterLen) - for i := 0; i < currentIterLen; i++ { - v, _ := s.batchQueue.PopFront() - batch[i] = v - } - } else { - batch = make([]uint64, blockBatchSize) - for i := 0; i < blockBatchSize; i++ { - v, _ := s.batchQueue.PopFront() - batch[i] = v - } + if currentIterLen > blockBatchSize { + currentIterLen = blockBatchSize + } + batch := make([]uint64, currentIterLen) + for i := 0; i < currentIterLen; i++ { + v, _ := s.batchQueue.PopFront() + batch[i] = v } blocks, err := s.chain.GetBlocks(context.Background(), batch) diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index 4db20ad..c7286e8 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -1,18 +1,15 @@ package syncer import ( + "errors" "log/slog" - "github.com/bits-and-blooms/bitset" "github.com/celo-org/celo-blockchain/core/types" "github.com/celo-org/celo-blockchain/ethclient" "github.com/ef-ds/deque/v2" - "github.com/grassrootseconomics/celo-events/internal/chain" - "github.com/grassrootseconomics/celo-events/internal/stats" -) - -const ( - blockBatchSize = 100 + "github.com/grassrootseconomics/celo-tracker/internal/chain" + "github.com/grassrootseconomics/celo-tracker/internal/db" + "github.com/grassrootseconomics/celo-tracker/internal/stats" ) type ( @@ -23,34 +20,40 @@ type ( Chain *chain.Chain Logg *slog.Logger Stats *stats.Stats - // replace with db - Db *bitset.BitSet + DB *db.DB + InitialLowerBound uint64 } Syncer struct { - batchQueue *deque.Deque[uint64] - blocksQueue *deque.Deque[types.Block] - chain *chain.Chain - logg *slog.Logger - stats *stats.Stats - ethClient *ethclient.Client - db *bitset.BitSet + batchQueue *deque.Deque[uint64] + blocksQueue *deque.Deque[types.Block] + chain *chain.Chain + logg *slog.Logger + stats *stats.Stats + ethClient *ethclient.Client + db *db.DB + initialLowerBound uint64 } ) -func NewSyncer(o SyncerOpts) (*Syncer, error) { +func New(o SyncerOpts) (*Syncer, error) { + if o.InitialLowerBound == 0 { + return nil, errors.New("initial lower bound not set") + } + ethClient, err := ethclient.Dial(o.WebSocketEndpoint) if err != nil { return nil, err } return &Syncer{ - batchQueue: o.BatchQueue, - blocksQueue: o.BlocksQueue, - chain: o.Chain, - logg: o.Logg, - stats: o.Stats, - ethClient: ethClient, - db: o.Db, + batchQueue: o.BatchQueue, + blocksQueue: o.BlocksQueue, + chain: o.Chain, + logg: o.Logg, + stats: o.Stats, + ethClient: ethClient, + db: o.DB, + initialLowerBound: o.InitialLowerBound, }, nil } diff --git a/revive.toml b/revive.toml new file mode 100644 index 0000000..500fd87 --- /dev/null +++ b/revive.toml @@ -0,0 +1,44 @@ +ignoreGeneratedHeader = true +severity = "warning" +confidence = 0.8 +errorCode = 0 +warningCode = 0 + +[rule.blank-imports] +[rule.context-as-argument] +[rule.context-keys-type] +[rule.dot-imports] +[rule.error-return] +[rule.error-strings] +[rule.error-naming] +[rule.increment-decrement] +[rule.var-naming] +[rule.var-declaration] +[rule.range] +[rule.receiver-naming] +[rule.time-naming] +[rule.unexported-return] +[rule.indent-error-flow] +[rule.errorf] +[rule.empty-block] +[rule.superfluous-else] +[rule.unused-parameter] +[rule.unreachable-code] +[rule.redefines-builtin-id] + + +[rule.defer] +arguments = [["loop", "method-call", "recover", "return", "immediate-recover"]] + +[rule.string-of-int] + +[rule.atomic] +[rule.call-to-gc] +[rule.constant-logical-expr] +[rule.identical-branches] +[rule.modifies-parameter] +[rule.modifies-value-receiver] +[rule.range-val-address] +[rule.range-val-in-closure] +[rule.unconditional-recursion] +[rule.waitgroup-by-value]