diff --git a/cmd/p2p/sensor/sensor.go b/cmd/p2p/sensor/sensor.go index cacd1e029..e96703ed9 100644 --- a/cmd/p2p/sensor/sensor.go +++ b/cmd/p2p/sensor/sensor.go @@ -42,6 +42,7 @@ type ( TrustedNodesFile string ProjectID string DatabaseID string + ClickHouseDSN string SensorID string MaxPeers int MaxDatabaseConcurrency int @@ -203,6 +204,12 @@ var SensorCmd = &cobra.Command{ if err != nil { return err } + // Drain buffered writes and release the connection on shutdown. + defer func() { + if cerr := db.Close(); cerr != nil { + log.Error().Err(cerr).Msg("Failed to close database") + } + }() // Fetch the latest block which will be used later when crafting the status // message. This call will only be made once and stored in the head field @@ -483,6 +490,20 @@ func newDatabase(ctx context.Context) (database.Database, error) { ShouldWritePeers: inputSensorParams.ShouldWritePeers, TTL: inputSensorParams.TTL, }), nil + case "clickhouse": + return database.NewClickHouse(ctx, database.ClickHouseOptions{ + DSN: inputSensorParams.ClickHouseDSN, + SensorID: inputSensorParams.SensorID, + ChainID: inputSensorParams.NetworkID, + MaxConcurrency: inputSensorParams.MaxDatabaseConcurrency, + ShouldWriteBlocks: inputSensorParams.ShouldWriteBlocks, + ShouldWriteBlockEvents: inputSensorParams.ShouldWriteBlockEvents, + ShouldWriteFirstBlockEvent: inputSensorParams.ShouldWriteFirstBlockEvent, + ShouldWriteTransactions: inputSensorParams.ShouldWriteTransactions, + ShouldWriteTransactionEvents: inputSensorParams.ShouldWriteTransactionEvents, + ShouldWriteFirstTransactionEvent: inputSensorParams.ShouldWriteFirstTransactionEvent, + ShouldWritePeers: inputSensorParams.ShouldWritePeers, + }), nil case "json": return database.NewJSONDatabase(database.JSONDatabaseOptions{ SensorID: inputSensorParams.SensorID, @@ -508,6 +529,8 @@ func init() { flag.MarkFlagsRequired(SensorCmd, "network-id") f.StringVarP(&inputSensorParams.ProjectID, "project-id", "p", "", "GCP project ID") f.StringVarP(&inputSensorParams.DatabaseID, "database-id", "d", "", "datastore database ID") + f.StringVar(&inputSensorParams.ClickHouseDSN, "clickhouse-dsn", "", + "ClickHouse DSN, e.g. clickhouse://user:pass@host:9000/sensor (used with --database=clickhouse)") f.StringVarP(&inputSensorParams.SensorID, "sensor-id", "s", "", "sensor ID when writing block/tx events") flag.MarkFlagsRequired(SensorCmd, "sensor-id") f.IntVarP(&inputSensorParams.MaxPeers, "max-peers", "m", 2000, "maximum number of peers to connect to") @@ -564,6 +587,7 @@ will result in less chance of missing data but can significantly increase memory f.StringVar(&inputSensorParams.Database, "database", "none", `which database to persist data to, options are: - datastore (GCP Datastore) + - clickhouse (ClickHouse, see --clickhouse-dsn) - json (output to stdout) - none (no persistence)`) f.BoolVar(&inputSensorParams.NoDiscovery, "no-discovery", false, "disable P2P peer discovery") diff --git a/doc/polycli_p2p_sensor.md b/doc/polycli_p2p_sensor.md index 5cdb07148..320835bc1 100644 --- a/doc/polycli_p2p_sensor.md +++ b/doc/polycli_p2p_sensor.md @@ -152,8 +152,10 @@ polycli p2p sensor amoy-nodes.json \ --broadcast-txs broadcast full transactions to peers --broadcast-workers int number of concurrent broadcast workers (default 4) --cache-only-validated-blocks only cache and serve blocks signed by a known validator (unknown-signer blocks are still recorded to the database); has no effect without --validate-block-signer (default true) + --clickhouse-dsn string ClickHouse DSN, e.g. clickhouse://user:pass@host:9000/sensor (used with --database=clickhouse) --database string which database to persist data to, options are: - datastore (GCP Datastore) + - clickhouse (ClickHouse, see --clickhouse-dsn) - json (output to stdout) - none (no persistence) (default "none") -d, --database-id string datastore database ID diff --git a/go.mod b/go.mod index 870374a86..2029f7933 100644 --- a/go.mod +++ b/go.mod @@ -38,15 +38,16 @@ require ( require github.com/alecthomas/participle/v2 v2.1.4 require ( + github.com/ClickHouse/ch-go v0.73.0 // indirect github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251213223233-751f36331c62 // indirect + github.com/andybalholm/brotli v1.2.1 // indirect github.com/chromedp/sysutil v1.1.0 // indirect github.com/containerd/errdefs v1.0.0 // indirect github.com/containerd/errdefs/pkg v0.3.0 // indirect - github.com/containerd/log v0.1.0 // indirect github.com/crate-crypto/go-eth-kzg v1.5.0 // indirect github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 // indirect github.com/distribution/reference v0.6.0 // indirect - github.com/docker/go-connections v0.4.0 // indirect + github.com/docker/go-connections v0.7.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/emicklei/dot v1.6.2 // indirect github.com/ethereum/c-kzg-4844/v2 v2.1.6 // indirect @@ -54,6 +55,8 @@ require ( github.com/ferranbt/fastssz v0.1.4 // indirect github.com/fjl/jsonw v0.1.0 // indirect github.com/gdamore/encoding v1.0.1 // indirect + github.com/go-faster/city v1.0.1 // indirect + github.com/go-faster/errors v0.7.1 // indirect github.com/go-json-experiment/json v0.0.0-20260214004413-d219187c3433 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/gobwas/httphead v0.1.0 // indirect @@ -66,11 +69,15 @@ require ( github.com/morikuni/aec v1.0.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/paulmach/orb v0.13.0 // indirect + github.com/pierrec/lz4/v4 v4.1.27 // indirect github.com/pion/dtls/v3 v3.1.2 // indirect github.com/pion/logging v0.2.4 // indirect github.com/pion/stun/v3 v3.1.2 // indirect github.com/pion/transport/v4 v4.0.1 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect + github.com/segmentio/asm v1.2.1 // indirect + github.com/shopspring/decimal v1.4.0 // indirect github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect github.com/spf13/afero v1.15.0 // indirect github.com/spf13/cast v1.10.0 // indirect @@ -119,7 +126,7 @@ require ( github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/ipfs/go-cid v0.4.1 // indirect github.com/jackpal/go-nat-pmp v1.0.2 // indirect - github.com/klauspost/compress v1.18.0 // indirect + github.com/klauspost/compress v1.18.6 // indirect github.com/klauspost/cpuid/v2 v2.2.8 // indirect github.com/kr/pretty v0.3.1 // indirect github.com/kr/text v0.2.0 // indirect @@ -148,8 +155,8 @@ require ( github.com/shirou/gopsutil v3.21.11+incompatible // indirect github.com/spaolacci/murmur3 v1.1.0 // indirect github.com/supranational/blst v0.3.16 // indirect - github.com/tklauser/go-sysconf v0.3.14 // indirect - github.com/tklauser/numcpus v0.9.0 // indirect + github.com/tklauser/go-sysconf v0.4.0 // indirect + github.com/tklauser/numcpus v0.12.0 // indirect github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect go.opencensus.io v0.24.0 // indirect @@ -169,6 +176,7 @@ require ( require ( cloud.google.com/go/kms v1.31.0 github.com/0xPolygon/cdk-contracts-tooling v0.0.1 + github.com/ClickHouse/clickhouse-go/v2 v2.47.0 github.com/btcsuite/btcd/btcutil v1.2.0 github.com/chromedp/cdproto v0.0.0-20260321001828-e3e3800016bc github.com/chromedp/chromedp v0.15.1 @@ -221,8 +229,8 @@ require ( github.com/yusufpapurcu/wmi v1.2.4 // indirect go.etcd.io/bbolt v1.4.0-alpha.0.0.20240404170359-43604f3112c5 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 // indirect - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 // indirect - go.opentelemetry.io/otel v1.43.0 // indirect - go.opentelemetry.io/otel/metric v1.43.0 // indirect - go.opentelemetry.io/otel/trace v1.43.0 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 // indirect + go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect ) diff --git a/go.sum b/go.sum index 16c37e3a8..8b4790440 100644 --- a/go.sum +++ b/go.sum @@ -19,9 +19,13 @@ github.com/0xPolygon/cdk-contracts-tooling v0.0.1 h1:2HH8KpO1CZRl1zHfn0IYwJhPA7l github.com/0xPolygon/cdk-contracts-tooling v0.0.1/go.mod h1:mFlcEjsm2YBBsu8atHJ3zyVnwM+Z/fMXpVmIJge+WVU= github.com/0xPolygon/cdk-rpc v0.0.0-20250213125803-179882ad6229 h1:6YhqNQVcXkoxqs5zQVg1bREuoeKvwpffpfoL8QQT+u4= github.com/0xPolygon/cdk-rpc v0.0.0-20250213125803-179882ad6229/go.mod h1:2scWqMMufrQXu7TikDgQ3BsyaKoX8qP26D6E262vSOg= -github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0= -github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= +github.com/ClickHouse/ch-go v0.73.0 h1:jsHiGRbQ3sz+gekvDFJF29LWDo5dzbJm5s1h8TWVP2M= +github.com/ClickHouse/ch-go v0.73.0/go.mod h1:wkFIxrqlXeRJ9cn3r5Fz5Qen9jl5aTMPuGZeuJpANNY= +github.com/ClickHouse/clickhouse-go/v2 v2.47.0 h1:ZDAzrnKSOPTIsm4tdUNfrii2yc8dk4SVRLC77BR7Z5Q= +github.com/ClickHouse/clickhouse-go/v2 v2.47.0/go.mod h1:sPj7C7UYQ2MWHcfX+4eGN6nwnCqwUKfgO6PcwKpd6K8= github.com/DataDog/zstd v1.5.6 h1:LbEglqepa/ipmmQJUDnSsfvA8e8IStVcGaFWDuxvGOY= github.com/DataDog/zstd v1.5.6/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw= github.com/FactomProject/basen v0.0.0-20150613233007-fe3947df716e h1:ahyvB3q25YnZWly5Gq1ekg6jcmWaGj/vG/MhF4aisoc= @@ -43,6 +47,8 @@ github.com/alecthomas/repr v0.4.0 h1:GhI2A8MACjfegCPVq9f1FLvIBS+DrQ2KQBFZP1iFzXc github.com/alecthomas/repr v0.4.0/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4= github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah4HI848JfFxHt+iPb26b4zyfspmqY0/8= github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM= +github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro= +github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bits-and-blooms/bitset v1.24.4 h1:95H15Og1clikBrKr/DuzMXkQzECs1M6hhoGXLwLQOZE= @@ -155,8 +161,8 @@ github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5Qvfr github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= github.com/docker/docker v28.5.2+incompatible h1:DBX0Y0zAjZbSrm1uzOkdr1onVghKaftjlSWt4AFexzM= github.com/docker/docker v28.5.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= -github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ= -github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec= +github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c= +github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q= github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= @@ -212,6 +218,10 @@ github.com/go-echarts/go-echarts/v2 v2.7.2 h1:lhypL1CekgqaLHM5V7fBPfaYGfimJ9dGyl github.com/go-echarts/go-echarts/v2 v2.7.2/go.mod h1:Z+spPygZRIEyqod69r0WMnkN5RV3MwhYDtw601w3G8w= github.com/go-errors/errors v1.5.1 h1:ZwEMSLRCapFLflTpT7NKaAc7ukJ8ZPEjzlxt8rPN8bk= github.com/go-errors/errors v1.5.1/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= +github.com/go-faster/city v1.0.1 h1:4WAxSZ3V2Ws4QRDrscLEDcibJY8uf41H6AhXDrNDcGw= +github.com/go-faster/city v1.0.1/go.mod h1:jKcUJId49qdW3L1qKHH/3wPeUstCVpVSXTM6vO3VcTw= +github.com/go-faster/errors v0.7.1 h1:MkJTnDoEdi9pDabt1dpWf7AA8/BaSYZqibYyhZ20AYg= +github.com/go-faster/errors v0.7.1/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo= github.com/go-json-experiment/json v0.0.0-20260214004413-d219187c3433 h1:vymEbVwYFP/L05h5TKQxvkXoKxNvTpjxYKdF1Nlwuao= github.com/go-json-experiment/json v0.0.0-20260214004413-d219187c3433/go.mod h1:tphK2c80bpPhMOI4v6bIc2xWywPfbqi1Z06+RcrMkDg= github.com/go-kit/log v0.2.1 h1:MRVx0/zhvdseW+Gza6N9rVzU/IVzaeE1SFI4raAhmBU= @@ -341,8 +351,8 @@ github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlT 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.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= -github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= +github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.2.8 h1:+StwCXwm9PdpiEkPyzBXIy+M9KUb4ODm0Zarf1kS5BM= github.com/klauspost/cpuid/v2 v2.2.8/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -389,8 +399,8 @@ github.com/moby/sys/atomicwriter v0.1.0 h1:kw5D/EqkBwsBFi0ss9v1VG3wIkVhzGvLklJ+w github.com/moby/sys/atomicwriter v0.1.0/go.mod h1:Ul8oqv2ZMNHOceF643P6FKPXeCmYtlQMvpizfsSoaWs= github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU= github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko= -github.com/moby/term v0.0.0-20221205130635-1aeaba878587 h1:HfkjXDfhgVaN5rmueG8cL8KKeFNecRCXFhaJ2qZ5SKA= -github.com/moby/term v0.0.0-20221205130635-1aeaba878587/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y= +github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= +github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= github.com/montanaflynn/stats v0.10.0 h1:8GQhct8Qup299/6gFmW3K3xRvPrdpmdjVCWRsZ5bnes= github.com/montanaflynn/stats v0.10.0/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow= github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= @@ -443,12 +453,16 @@ github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsq github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde h1:x0TT0RDC7UhAVbbWWBzr41ElhJx5tXPWkIHA2HWPRuw= github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0= +github.com/paulmach/orb v0.13.0 h1:r7n7mQGGF+cj/CbcivEj9J3HGK+XR+yXnvzRdq9saIw= +github.com/paulmach/orb v0.13.0/go.mod h1:6scRWINywA2Jf05dcjOfLfxrUIMECvTSG2MVbRLxu/k= github.com/pelletier/go-toml/v2 v2.4.3 h1:GTRvJQutkOSftxIFD5xw9aepkYNuPWmVJpffdDPYVpY= github.com/pelletier/go-toml/v2 v2.4.3/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM= github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0= github.com/petermattis/goid v0.0.0-20240813172612-4fcff4a6cae7 h1:Dx7Ovyv/SFnMFw3fD4oEoeorXc6saIiQ23LrGLth0Gw= github.com/petermattis/goid v0.0.0-20240813172612-4fcff4a6cae7/go.mod h1:pxMtw7cyUw6B2bRH0ZBANSPg+AoSud1I1iyJHI69jH4= +github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk= +github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4= github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= github.com/pion/dtls/v3 v3.1.2 h1:gqEdOUXLtCGW+afsBLO0LtDD8GnuBBjEy6HRtyofZTc= @@ -500,10 +514,14 @@ github.com/sasha-s/go-deadlock v0.3.5 h1:tNCOEEDG6tBqrNDOX35j/7hL5FcFViG6awUGROb github.com/sasha-s/go-deadlock v0.3.5/go.mod h1:bugP6EGbdGYObIlx7pUZtWqlvo8k9H6vCBBsiChJQ5U= github.com/schollz/progressbar/v3 v3.19.1 h1:iv8BgwOvdML/S3p84uBpy/IMigv4U9594vPZYa2EdrU= github.com/schollz/progressbar/v3 v3.19.1/go.mod h1:LFL7jqimKxfhero4K1eCkUr/6R39AgQeiPCJtlTWIW8= +github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= +github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI= github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= -github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= -github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= +github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= +github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw= github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8/go.mod h1:3n1Cwaq1E1/1lhQhtRK2ts/ZwZEhjcQeJQ1RuC6Q/8U= github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI= @@ -538,10 +556,10 @@ github.com/supranational/blst v0.3.16 h1:bTDadT+3fK497EvLdWRQEjiGnUtzJ7jjIUMF0jq github.com/supranational/blst v0.3.16/go.mod h1:jZJtfjgudtNl4en1tzwPIV3KjUnQUvG3/j+w+fVonLw= github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d h1:vfofYNRScrDdvS342BElfbETmL1Aiz3i2t0zfRj16Hs= github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d/go.mod h1:RRCYJbIwD5jmqPI9XoAFR0OcDxqUctll6zUj/+B4S48= -github.com/tklauser/go-sysconf v0.3.14 h1:g5vzr9iPFFz24v2KZXs/pvpvh8/V9Fw6vQK5ZZb78yU= -github.com/tklauser/go-sysconf v0.3.14/go.mod h1:1ym4lWMLUOhuBOPGtRcJm7tEGX4SCYNEEEtghGG/8uY= -github.com/tklauser/numcpus v0.9.0 h1:lmyCHtANi8aRUgkckBgoDk1nHCux3n2cgkJLXdQGPDo= -github.com/tklauser/numcpus v0.9.0/go.mod h1:SN6Nq1O3VychhC1npsWostA+oW+VOQTxZrS604NSRyI= +github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU= +github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI= +github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4= +github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg= github.com/tyler-smith/go-bip32 v1.0.0 h1:sDR9juArbUgX+bO/iblgZnMPeWY1KZMUC2AFUJdv5KE= github.com/tyler-smith/go-bip32 v1.0.0/go.mod h1:onot+eHknzV4BVPwrzqY5OoVpyCvnwD7lMawL5aQupE= github.com/tyler-smith/go-bip39 v1.1.0 h1:5eUemwrMargf3BSLRRCalXT93Ns6pQJIjYQN2nyfOP8= @@ -559,6 +577,8 @@ github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y= github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 h1:gEOO8jv9F4OT7lGCjxCBTO/36wtF6j2nSip77qHd4x4= github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM= +github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= +github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= @@ -574,22 +594,22 @@ go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 h1:yI1/OhfEPy7J9eoa6Sj051C7n5dvpj0QX8g4sRchg04= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0/go.mod h1:NoUCKYWK+3ecatC4HjkRktREheMeEtrXoQxrqYFeHSc= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 h1:OyrsyzuttWTSur2qN/Lm0m2a8yqyIjUVBZcxFPuXq2o= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0/go.mod h1:C2NGBr+kAB4bk3xtMXfZ94gqFDtg/GkI7e9zqGh5Beg= -go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= -go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.41.0 h1:ao6Oe+wSebTlQ1OEht7jlYTzQKE+pnx/iNywFvTbuuI= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.41.0/go.mod h1:u3T6vz0gh/NVzgDgiwkgLxpsSF6PaPmo2il0apGJbls= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.41.0 h1:inYW9ZhgqiDqh6BioM7DVHHzEGVq76Db5897WLGZ5Go= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.41.0/go.mod h1:Izur+Wt8gClgMJqO/cZ8wdeeMryJ/xxiOVgFSSfpDTY= -go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= -go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= -go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= -go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= -go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= -go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= -go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= -go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= diff --git a/p2p/conns.go b/p2p/conns.go index 3716264b2..7ebac420c 100644 --- a/p2p/conns.go +++ b/p2p/conns.go @@ -66,6 +66,12 @@ type Conns struct { // txs caches transactions for serving to peers and duplicate detection txs *ds.LRU[common.Hash, *types.Transaction] + // announcedTxs tracks tx hashes seen in any inbound announcement, so the + // first-seen transaction event is recorded exactly once (the tx analog of + // the blocks cache's first-seen check). Separate from txs so it isn't + // affected by the fetch/serve dedup. + announcedTxs *ds.LRU[common.Hash, struct{}] + // knownTxsBloom stores bloom filter options for per-peer known tx tracking knownTxsBloom ds.BloomSetOptions // knownBlocksMax stores the maximum size for per-peer known block caches @@ -143,6 +149,7 @@ func NewConns(opts ConnsOptions) *Conns { conns: make(map[string]*conn), blocks: ds.NewLRU[common.Hash, BlockCache](opts.BlocksCache), txs: ds.NewLRU[common.Hash, *types.Transaction](opts.TxsCache), + announcedTxs: ds.NewLRU[common.Hash, struct{}](opts.TxsCache), knownTxsBloom: opts.KnownTxsBloom, knownBlocksMax: opts.KnownBlocksMax, oldest: oldest, @@ -623,6 +630,14 @@ func (c *Conns) FilterUnknownTxHashes(hashes []common.Hash) []common.Hash { return c.txs.FilterMissing(hashes) } +// MarkTxSeen records that a tx hash was seen in an inbound announcement and +// reports whether this is the first time (across all peers). Used to write the +// first-seen transaction event exactly once. +func (c *Conns) MarkTxSeen(hash common.Hash) (firstSeen bool) { + existed := c.announcedTxs.Update(hash, func(v struct{}) struct{} { return v }) + return !existed +} + // Blocks returns the global blocks cache. func (c *Conns) Blocks() *ds.LRU[common.Hash, BlockCache] { return c.blocks diff --git a/p2p/database/clickhouse.go b/p2p/database/clickhouse.go new file mode 100644 index 000000000..c6ac21e54 --- /dev/null +++ b/p2p/database/clickhouse.go @@ -0,0 +1,591 @@ +package database + +import ( + "context" + "fmt" + "math/big" + "sync" + "sync/atomic" + "time" + + "github.com/0xPolygon/polygon-cli/util" + "github.com/ClickHouse/clickhouse-go/v2" + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/eth/protocols/eth" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/rs/zerolog/log" +) + +// Default batching parameters. Write* calls only enqueue rows; a background +// goroutine flushes them in batches, since ClickHouse prefers large, infrequent +// inserts. Buffers are fixed-size and drop-on-full so a slow database can never +// stall the sensor or exhaust memory. +const ( + chFlushInterval = 1 * time.Second + chFlushTimeout = 30 * time.Second + // chMaxFlushAttempts bounds retries of a failed batch insert. Each retry is + // immediate with a fresh connection, recovering from stale-connection and + // transient errors without delaying shutdown. + chMaxFlushAttempts = 3 + chBlockBatch = 5000 + chBlockEventBatch = 50000 + chTxBatch = 20000 + chTxEventBatch = 50000 + chPeerBatch = 2000 +) + +// ClickHouse implements the Database interface backed by a ClickHouse cluster. +// The table definitions this writer targets (and the block_first_seen +// materialized view) live in the sensor-network-tools repo +// (clickhouse_schema.sql), not this repo. +type ClickHouse struct { + conn driver.Conn + sensorID string + chainID *big.Int + maxConcurrency int + shouldWriteBlocks bool + shouldWriteBlockEvents bool + shouldWriteFirstBlockEvent bool + shouldWriteTransactions bool + shouldWriteTransactionEvents bool + shouldWriteFirstTransactionEvent bool + shouldWritePeers bool + + blocks *rowBatcher[chBlock] + blockEvt *rowBatcher[chEvent] + txs *rowBatcher[chTx] + txEvt *rowBatcher[chEvent] + peers *rowBatcher[chPeer] + + // cancel stops the batcher goroutines; wg tracks them so Close can wait for + // their final drain flush before the connection is closed. + cancel context.CancelFunc + wg sync.WaitGroup +} + +// ClickHouseOptions is used when creating a NewClickHouse. +type ClickHouseOptions struct { + DSN string + SensorID string + ChainID uint64 + MaxConcurrency int + ShouldWriteBlocks bool + ShouldWriteBlockEvents bool + ShouldWriteFirstBlockEvent bool + ShouldWriteTransactions bool + ShouldWriteTransactionEvents bool + ShouldWriteFirstTransactionEvent bool + ShouldWritePeers bool +} + +// NewClickHouse connects to ClickHouse, verifies connectivity, and starts the +// background batch flushers. Callers should defer Close to drain buffered rows +// on shutdown. If the connection cannot be established the returned Database +// no-ops all writes (mirroring the Datastore backend) so the sensor keeps running. +func NewClickHouse(ctx context.Context, opts ClickHouseOptions) Database { + c := &ClickHouse{ + sensorID: opts.SensorID, + chainID: new(big.Int).SetUint64(opts.ChainID), + maxConcurrency: opts.MaxConcurrency, + shouldWriteBlocks: opts.ShouldWriteBlocks, + shouldWriteBlockEvents: opts.ShouldWriteBlockEvents, + shouldWriteFirstBlockEvent: opts.ShouldWriteFirstBlockEvent, + shouldWriteTransactions: opts.ShouldWriteTransactions, + shouldWriteTransactionEvents: opts.ShouldWriteTransactionEvents, + shouldWriteFirstTransactionEvent: opts.ShouldWriteFirstTransactionEvent, + shouldWritePeers: opts.ShouldWritePeers, + } + + conn, err := connectClickHouse(ctx, opts.DSN) + if err != nil { + log.Error().Err(err).Msg("Could not initialize ClickHouse connection") + return c + } + c.conn = conn + + // Derive a cancellable context so Close can stop the batchers independently + // of the parent context. + bctx, cancel := context.WithCancel(ctx) + c.cancel = cancel + c.startBatchers(bctx) + + return c +} + +// Close stops the batcher goroutines, waits for their final drain flush to +// complete, and closes the connection. It is safe to call on a no-op instance +// (connection never established). +func (c *ClickHouse) Close() error { + if c.cancel != nil { + c.cancel() + } + c.wg.Wait() + if c.conn != nil { + return c.conn.Close() + } + return nil +} + +// connectClickHouse parses the DSN, opens a connection, and verifies +// connectivity with a ping. +func connectClickHouse(ctx context.Context, dsn string) (driver.Conn, error) { + chOpts, err := clickhouse.ParseDSN(dsn) + if err != nil { + return nil, fmt.Errorf("could not parse ClickHouse DSN: %w", err) + } + + // Writer-friendly defaults when the DSN omits them: LZ4 compression is a + // network win on the wide blocks table, and the pool is sized for the + // concurrent per-table flushers plus the occasional query. + if chOpts.Compression == nil { + chOpts.Compression = &clickhouse.Compression{Method: clickhouse.CompressionLZ4} + } + if chOpts.MaxIdleConns == 0 { + chOpts.MaxIdleConns = 10 + } + if chOpts.MaxOpenConns == 0 { + chOpts.MaxOpenConns = 20 + } + + conn, err := clickhouse.Open(chOpts) + if err != nil { + return nil, fmt.Errorf("could not connect to ClickHouse: %w", err) + } + + if err := conn.Ping(ctx); err != nil { + return nil, fmt.Errorf("could not ping ClickHouse: %w", err) + } + return conn, nil +} + +// startBatchers creates the background row batchers, one per target table. Each +// batcher's append closure maps a row to its column values; the batch/flush/error +// handling lives in newInsertBatcher. +func (c *ClickHouse) startBatchers(ctx context.Context) { + c.blocks = newInsertBatcher(ctx, c, "blocks", chBlockBatch, + "INSERT INTO blocks (hash, number, parent_hash, block_time, coinbase, signer, difficulty, total_difficulty, gas_used, gas_limit, base_fee, tx_count, uncle_count, uncle_hash, state_root, tx_root, receipt_root, logs_bloom, extra_data, mix_digest, nonce, sensor_id, ingested_at, is_parent)", + func(b driver.Batch, r chBlock) error { + return b.Append(r.hash, r.number, r.parentHash, r.blockTime, r.coinbase, r.signer, r.difficulty, r.totalDifficulty, r.gasUsed, r.gasLimit, r.baseFee, r.txCount, r.uncleCount, r.uncleHash, r.stateRoot, r.txRoot, r.receiptRoot, r.logsBloom, r.extraData, r.mixDigest, r.nonce, c.sensorID, r.ingestedAt, r.isParent) + }) + // block_events and transaction_events share the same row shape and column + // order, so both batchers use the same append function. + appendEvent := func(b driver.Batch, r chEvent) error { + return b.Append(r.hash, c.sensorID, r.peerID, r.seenAt) + } + c.blockEvt = newInsertBatcher(ctx, c, "block_events", chBlockEventBatch, + "INSERT INTO block_events (block_hash, sensor_id, peer_id, seen_at)", + appendEvent) + c.txs = newInsertBatcher(ctx, c, "transactions", chTxBatch, + "INSERT INTO transactions (hash, from_address, to_address, value, gas, gas_price, gas_fee_cap, gas_tip_cap, nonce, tx_type, sensor_id, ingested_at)", + func(b driver.Batch, r chTx) error { + return b.Append(r.hash, r.from, r.to, r.value, r.gas, r.gasPrice, r.gasFeeCap, r.gasTipCap, r.nonce, r.txType, c.sensorID, r.ingestedAt) + }) + c.txEvt = newInsertBatcher(ctx, c, "transaction_events", chTxEventBatch, + "INSERT INTO transaction_events (tx_hash, sensor_id, peer_id, seen_at)", + appendEvent) + c.peers = newInsertBatcher(ctx, c, "peers", chPeerBatch, + "INSERT INTO peers (peer_id, name, url, caps, last_seen_by, time_last_seen)", + func(b driver.Batch, r chPeer) error { + return b.Append(r.peerID, r.name, r.url, r.caps, c.sensorID, r.timeLastSeen) + }) +} + +// newInsertBatcher wraps newRowBatcher with the common flush behaviour: prepare +// the INSERT, append rows, send, retrying transient failures. Only appendRow +// varies per table. +func newInsertBatcher[T any](ctx context.Context, c *ClickHouse, name string, maxRows int, query string, appendRow func(driver.Batch, T) error) *rowBatcher[T] { + return newRowBatcher(ctx, &c.wg, name, maxRows, func(rows []T) error { + var err error + for attempt := 1; attempt <= chMaxFlushAttempts; attempt++ { + if err = flushBatch(c.conn, query, rows, appendRow); err == nil { + return nil + } + if attempt < chMaxFlushAttempts { + log.Warn().Err(err).Str("table", name).Int("attempt", attempt).Int("rows", len(rows)). + Msg("ClickHouse batch insert failed; retrying") + } + } + return err + }) +} + +// flushBatch prepares, fills, and sends a single INSERT. It runs on a detached, +// time-bounded context so a flush triggered during shutdown (parent context +// already cancelled) still completes. +func flushBatch[T any](conn driver.Conn, query string, rows []T, appendRow func(driver.Batch, T) error) error { + ctx, cancel := context.WithTimeout(context.Background(), chFlushTimeout) + defer cancel() + + b, err := conn.PrepareBatch(ctx, query) + if err != nil { + return fmt.Errorf("prepare batch: %w", err) + } + for _, r := range rows { + if err := appendRow(b, r); err != nil { + return fmt.Errorf("append row: %w", err) + } + } + if err := b.Send(); err != nil { + return fmt.Errorf("send batch: %w", err) + } + return nil +} + +// --- row types ------------------------------------------------------------- + +type chBlock struct { + hash string + number uint64 + parentHash string + blockTime time.Time + coinbase string + signer string + difficulty uint64 + totalDifficulty *big.Int + gasUsed uint64 + gasLimit uint64 + baseFee uint64 + txCount uint32 + uncleCount uint16 + uncleHash string + stateRoot string + txRoot string + receiptRoot string + logsBloom []byte + extraData []byte + mixDigest string + nonce uint64 + ingestedAt time.Time + isParent bool +} + +type chEvent struct { + hash string + peerID string + seenAt time.Time +} + +type chTx struct { + hash string + from string + to string + value *big.Int + gas uint64 + gasPrice *big.Int + gasFeeCap *big.Int + gasTipCap *big.Int + nonce uint64 + txType uint8 + ingestedAt time.Time +} + +type chPeer struct { + peerID string + name string + url string + caps []string + timeLastSeen time.Time +} + +// --- Database interface ---------------------------------------------------- + +func (c *ClickHouse) WriteBlock(ctx context.Context, peer *enode.Node, block *types.Block, td *big.Int, tfs time.Time) { + if c.conn == nil { + return + } + if c.shouldWriteBlockEvents && peer != nil { + c.blockEvt.add(chEvent{hash: block.Hash().Hex(), peerID: peer.URLv4(), seenAt: tfs}) + } + if c.shouldWriteBlocks { + c.blocks.add(newChBlock(block.Header(), td, tfs, len(block.Transactions()), len(block.Uncles()), false)) + } + if c.shouldWriteTransactions { + c.writeTxs(block.Transactions(), tfs) + } +} + +func (c *ClickHouse) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool) { + if c.conn == nil || !c.shouldWriteBlocks { + return + } + // A header carries no tx/uncle counts, so they are written as 0; the + // full-block (NewBlock) path writes a separate row with the real counts. + // isParent marks headers fetched as ancestors during backfill. + for _, h := range headers { + c.blocks.add(newChBlock(h, big.NewInt(0), tfs, 0, 0, isParent)) + } +} + +func (c *ClickHouse) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash common.Hash, tfs time.Time) { + if c.conn == nil || !c.shouldWriteTransactions { + return + } + // The block row is written from the header path; here we only persist the + // transactions carried in the body (no read-modify-write on blocks). + txs, err := body.Transactions.Items() + if err != nil { + log.Error().Err(err).Str("hash", hash.Hex()).Msg("Failed to decode transactions from block body") + return + } + c.writeTxs(txs, tfs) +} + +func (c *ClickHouse) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { + if c.conn == nil || peer == nil { + return + } + peerID := peer.URLv4() + for _, hash := range hashes { + c.blockEvt.add(chEvent{hash: hash.Hex(), peerID: peerID, seenAt: tfs}) + } +} + +// WriteBlockHashFirstSeen is a no-op: ClickHouse derives earliest first-seen at +// query time from the block_events stream (see the block_first_seen +// materialized view), so no per-block stamp is needed. +func (c *ClickHouse) WriteBlockHashFirstSeen(ctx context.Context, peer *enode.Node, hash common.Hash, tfsh time.Time) { +} + +func (c *ClickHouse) WriteTransactionEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { + if c.conn == nil || peer == nil { + return + } + peerID := peer.URLv4() + for _, hash := range hashes { + c.txEvt.add(chEvent{hash: hash.Hex(), peerID: peerID, seenAt: tfs}) + } +} + +func (c *ClickHouse) WriteTransactions(ctx context.Context, peer *enode.Node, txs []*types.Transaction, tfs time.Time) { + if c.conn == nil || !c.shouldWriteTransactions { + return + } + c.writeTxs(txs, tfs) +} + +func (c *ClickHouse) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.Time) { + if c.conn == nil || !c.shouldWritePeers { + return + } + for _, peer := range peers { + c.peers.add(chPeer{ + peerID: peer.ID().String(), + name: peer.Fullname(), + url: peer.Node().URLv4(), + caps: peer.Info().Caps, + timeLastSeen: tls, + }) + } +} + +// HasBlock reports whether the block already exists. Called once per new block +// (not per event), so an indexed point lookup is cheap. Without a connection it +// reports true so the sensor never attempts a backfill it could not persist. +func (c *ClickHouse) HasBlock(ctx context.Context, hash common.Hash) bool { + if c.conn == nil { + return true + } + var exists uint8 + err := c.conn.QueryRow(ctx, "SELECT 1 FROM blocks WHERE hash = ? LIMIT 1", hash.Hex()).Scan(&exists) + return err == nil && exists == 1 +} + +func (c *ClickHouse) NodeList(ctx context.Context, limit int) ([]string, error) { + if c.conn == nil { + return []string{}, nil + } + rows, err := c.conn.Query(ctx, + "SELECT peer_id FROM block_events GROUP BY peer_id ORDER BY max(seen_at) DESC LIMIT ?", limit) + if err != nil { + return nil, fmt.Errorf("query node list: %w", err) + } + defer func() { + if err := rows.Close(); err != nil { + log.Error().Err(err).Msg("Failed to close ClickHouse rows") + } + }() + + nodelist := []string{} + for rows.Next() { + var peerID string + if err := rows.Scan(&peerID); err != nil { + log.Error().Err(err).Msg("Failed to scan peer_id") + continue + } + nodelist = append(nodelist, peerID) + } + if err := rows.Err(); err != nil { + return nodelist, fmt.Errorf("iterate node list: %w", err) + } + return nodelist, nil +} + +func (c *ClickHouse) MaxConcurrentWrites() int { return c.maxConcurrency } +func (c *ClickHouse) ShouldWriteBlocks() bool { return c.shouldWriteBlocks } +func (c *ClickHouse) ShouldWriteBlockEvents() bool { return c.shouldWriteBlockEvents } +func (c *ClickHouse) ShouldWriteFirstBlockEvent() bool { return c.shouldWriteFirstBlockEvent } +func (c *ClickHouse) ShouldWriteTransactions() bool { return c.shouldWriteTransactions } +func (c *ClickHouse) ShouldWriteTransactionEvents() bool { return c.shouldWriteTransactionEvents } +func (c *ClickHouse) ShouldWriteFirstTransactionEvent() bool { + return c.shouldWriteFirstTransactionEvent +} +func (c *ClickHouse) ShouldWritePeers() bool { return c.shouldWritePeers } + +// --- helpers --------------------------------------------------------------- + +// newChBlock maps a header (plus data not carried on the header itself) to a +// blocks-table row. +func newChBlock(h *types.Header, td *big.Int, tfs time.Time, txCount, uncleCount int, isParent bool) chBlock { + baseFee := uint64(0) + if h.BaseFee != nil { + baseFee = h.BaseFee.Uint64() + } + if td == nil { + td = big.NewInt(0) + } + // Recover the block signer from the header seal so signer-based analytics + // don't have to ecrecover on every query. Left empty when it can't be recovered. + var signer string + if sig, err := util.Ecrecover(h); err == nil { + signer = common.BytesToAddress(sig).Hex() + } + return chBlock{ + hash: h.Hash().Hex(), + number: h.Number.Uint64(), + parentHash: h.ParentHash.Hex(), + blockTime: time.Unix(int64(h.Time), 0).UTC(), + coinbase: h.Coinbase.Hex(), + signer: signer, + difficulty: h.Difficulty.Uint64(), + totalDifficulty: new(big.Int).Set(td), + gasUsed: h.GasUsed, + gasLimit: h.GasLimit, + baseFee: baseFee, + txCount: uint32(txCount), + uncleCount: uint16(uncleCount), + uncleHash: h.UncleHash.Hex(), + stateRoot: h.Root.Hex(), + txRoot: h.TxHash.Hex(), + receiptRoot: h.ReceiptHash.Hex(), + logsBloom: h.Bloom.Bytes(), + extraData: h.Extra, + mixDigest: h.MixDigest.Hex(), + nonce: h.Nonce.Uint64(), + ingestedAt: tfs, + isParent: isParent, + } +} + +func (c *ClickHouse) writeTxs(txs []*types.Transaction, tfs time.Time) { + for _, tx := range txs { + var from, to string + chainID := tx.ChainId() + if chainID == nil || chainID.Sign() <= 0 { + chainID = c.chainID + } + if addr, err := types.Sender(types.LatestSignerForChainID(chainID), tx); err == nil { + from = addr.Hex() + } + if tx.To() != nil { + to = tx.To().Hex() + } + c.txs.add(chTx{ + hash: tx.Hash().Hex(), + from: from, + to: to, + value: new(big.Int).Set(tx.Value()), + gas: tx.Gas(), + gasPrice: new(big.Int).Set(tx.GasPrice()), + gasFeeCap: new(big.Int).Set(tx.GasFeeCap()), + gasTipCap: new(big.Int).Set(tx.GasTipCap()), + nonce: tx.Nonce(), + txType: tx.Type(), + ingestedAt: tfs, + }) + } +} + +// --- batching -------------------------------------------------------------- + +// rowBatcher buffers rows and flushes them in bulk when the buffer reaches +// maxRows or on a fixed interval. add is non-blocking: rows are dropped and +// counted when the fixed-size buffer is full, so a slow database never stalls +// the sensor hot path. +type rowBatcher[T any] struct { + name string + in chan T + maxRows int + flush func([]T) error + dropped atomic.Uint64 +} + +func newRowBatcher[T any](ctx context.Context, wg *sync.WaitGroup, name string, maxRows int, flush func([]T) error) *rowBatcher[T] { + b := &rowBatcher[T]{ + name: name, + in: make(chan T, maxRows*2), + maxRows: maxRows, + flush: flush, + } + wg.Add(1) + go func() { + defer wg.Done() + b.loop(ctx) + }() + return b +} + +func (b *rowBatcher[T]) add(v T) { + select { + case b.in <- v: + default: + b.dropped.Add(1) + } +} + +func (b *rowBatcher[T]) loop(ctx context.Context) { + ticker := time.NewTicker(chFlushInterval) + defer ticker.Stop() + + buf := make([]T, 0, b.maxRows) + doFlush := func() { + if len(buf) == 0 { + return + } + if err := b.flush(buf); err != nil { + log.Error().Err(err).Str("table", b.name).Int("rows", len(buf)).Msg("ClickHouse batch insert failed") + } + buf = buf[:0] + } + + for { + select { + case <-ctx.Done(): + // Drain any buffered rows before exiting. + for { + select { + case v := <-b.in: + buf = append(buf, v) + if len(buf) >= b.maxRows { + doFlush() + } + default: + doFlush() + return + } + } + case v := <-b.in: + buf = append(buf, v) + if len(buf) >= b.maxRows { + doFlush() + } + case <-ticker.C: + doFlush() + if d := b.dropped.Swap(0); d > 0 { + log.Warn().Uint64("dropped", d).Str("table", b.name).Msg("ClickHouse batcher dropped rows (buffer full)") + } + } + } +} diff --git a/p2p/database/clickhouse_test.go b/p2p/database/clickhouse_test.go new file mode 100644 index 000000000..a49a02304 --- /dev/null +++ b/p2p/database/clickhouse_test.go @@ -0,0 +1,135 @@ +package database + +import ( + "context" + "math/big" + "os" + "testing" + "time" + + "github.com/ClickHouse/clickhouse-go/v2" + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + "github.com/ethereum/go-ethereum/consensus/clique" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" +) + +// TestClickHouseWrites exercises the ClickHouse backend end-to-end against a +// real server. It is skipped unless POLYCLI_TEST_CLICKHOUSE_DSN is set, e.g. +// +// POLYCLI_TEST_CLICKHOUSE_DSN=clickhouse://localhost:19000/sensor go test ./p2p/database/ -run TestClickHouseWrites -v +// +// The target database must already have the schema from +// sensor-network-tools/clickhouse_schema.sql applied. +func TestClickHouseWrites(t *testing.T) { + dsn := os.Getenv("POLYCLI_TEST_CLICKHOUSE_DSN") + if dsn == "" { + t.Skip("POLYCLI_TEST_CLICKHOUSE_DSN not set; skipping ClickHouse integration test") + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + db := NewClickHouse(ctx, ClickHouseOptions{ + DSN: dsn, + SensorID: "test-sensor", + ChainID: 137, + MaxConcurrency: 10, + ShouldWriteBlocks: true, + ShouldWriteBlockEvents: true, + ShouldWriteTransactions: true, + ShouldWriteTransactionEvents: true, + ShouldWritePeers: true, + }) + + now := time.Now().UTC() + priv, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("generate key: %v", err) + } + wantSigner := crypto.PubkeyToAddress(priv.PublicKey) + header := &types.Header{ + Number: big.NewInt(42), + Time: uint64(now.Unix()), + Difficulty: big.NewInt(7), + GasLimit: 30_000_000, + GasUsed: 21_000, + BaseFee: big.NewInt(1_000_000_000), + Extra: make([]byte, crypto.SignatureLength), + } + sig, err := crypto.Sign(clique.SealHash(header).Bytes(), priv) + if err != nil { + t.Fatalf("sign header: %v", err) + } + copy(header.Extra[len(header.Extra)-crypto.SignatureLength:], sig) + block := types.NewBlockWithHeader(header) + + tx := types.NewTx(&types.LegacyTx{ + Nonce: 1, + GasPrice: big.NewInt(2_000_000_000), + Gas: 21_000, + Value: big.NewInt(1), + }) + + db.WriteBlock(ctx, nil, block, big.NewInt(100), now) + db.WriteTransactions(ctx, nil, []*types.Transaction{tx}, now) + db.WritePeers(ctx, nil, now) // empty peer slice is fine; exercises the path + + // Close drains the buffered rows synchronously before we verify them. + if cerr := db.Close(); cerr != nil { + t.Fatalf("close db: %v", cerr) + } + + conn, err := clickhouse.Open(mustParseDSN(t, dsn)) + if err != nil { + t.Fatalf("open verify conn: %v", err) + } + defer func() { + if err := conn.Close(); err != nil { + t.Errorf("close verify conn: %v", err) + } + }() + + checkCount(t, conn, "blocks", block.Hash().Hex()) + checkCount(t, conn, "transactions", tx.Hash().Hex()) + + // Verify the round-tripped block fields. + var ( + number uint64 + gasUsed uint64 + baseFee uint64 + signer string + ) + row := conn.QueryRow(context.Background(), + "SELECT number, gas_used, base_fee, signer FROM blocks WHERE hash = ? LIMIT 1", block.Hash().Hex()) + if err := row.Scan(&number, &gasUsed, &baseFee, &signer); err != nil { + t.Fatalf("scan block: %v", err) + } + if number != 42 || gasUsed != 21_000 || baseFee != 1_000_000_000 { + t.Fatalf("unexpected block fields: number=%d gas_used=%d base_fee=%d", number, gasUsed, baseFee) + } + if signer != wantSigner.Hex() { + t.Fatalf("signer mismatch: want %s got %s", wantSigner.Hex(), signer) + } +} + +func mustParseDSN(t *testing.T, dsn string) *clickhouse.Options { + t.Helper() + opts, err := clickhouse.ParseDSN(dsn) + if err != nil { + t.Fatalf("parse dsn: %v", err) + } + return opts +} + +func checkCount(t *testing.T, conn driver.Conn, table, hash string) { + t.Helper() + var count uint64 + // #nosec G202 -- table is a test-only constant, not user input + if err := conn.QueryRow(context.Background(), + "SELECT count() FROM "+table+" WHERE hash = ?", hash).Scan(&count); err != nil { + t.Fatalf("count %s: %v", table, err) + } + if count == 0 { + t.Fatalf("expected a row in %s for hash %s, got none", table, hash) + } +} diff --git a/p2p/database/database.go b/p2p/database/database.go index d31e7cdb0..0453e1d5b 100644 --- a/p2p/database/database.go +++ b/p2p/database/database.go @@ -24,23 +24,36 @@ type Database interface { // returns true. WriteBlockHeaders(context.Context, []*types.Header, time.Time, bool) - // WriteBlockHashes will write the block hashes if ShouldWriteBlockEvents - // returns true. - WriteBlockHashes(context.Context, *enode.Node, []common.Hash, time.Time) + // WriteBlockEvents appends an inbound block event (peer, hash, time) for + // each hash — one per peer we received the announcement from. The caller + // decides which hashes to pass (every announcement for the full per-peer + // stream, or just the first-seen ones); the backend only appends. + WriteBlockEvents(context.Context, *enode.Node, []common.Hash, time.Time) - // WriteBlockHashFirstSeen writes a partial block entry with just the hash - // first seen time if the block doesn't exist yet. + // WriteBlockHashFirstSeen records the earliest sighting of a block hash on + // the block entity itself (Datastore's TimeFirstSeenHash). Backends that + // derive first-seen from the event stream (e.g. ClickHouse) treat it as a + // no-op. WriteBlockHashFirstSeen(context.Context, *enode.Node, common.Hash, time.Time) - // WriteBlockBody will write the block bodies if ShouldWriteBlocks returns - // true. + // WriteBlockBody writes the transactions carried in the block body (the block + // row itself comes from WriteBlock/WriteBlockHeaders). Backends with a + // separate transactions table (e.g. ClickHouse) gate this on + // ShouldWriteTransactions; the Datastore backend gates on ShouldWriteBlocks + // because it links the transactions and uncles onto the block entity. WriteBlockBody(context.Context, *eth.BlockBody, common.Hash, time.Time) - // WriteTransactions will write the both the transaction and transaction - // event to the database if ShouldWriteTransactions and - // ShouldWriteTransactionEvents return true, respectively. + // WriteTransactions writes the transaction bodies if ShouldWriteTransactions + // returns true. Transaction events are recorded separately via + // WriteTransactionEvents at announcement time. WriteTransactions(context.Context, *enode.Node, []*types.Transaction, time.Time) + // WriteTransactionEvents appends an inbound transaction event (peer, hash, + // time) for each hash — the tx mirror of WriteBlockEvents. The caller + // decides which hashes to pass (every announcement for the full per-peer + // stream, or just the first-seen ones); the backend only appends. + WriteTransactionEvents(context.Context, *enode.Node, []common.Hash, time.Time) + // WritePeers will write the connected peers to the database. WritePeers(context.Context, []*p2p.Peer, time.Time) @@ -59,4 +72,10 @@ type Database interface { // NodeList will return a list of enode URLs. NodeList(ctx context.Context, limit int) ([]string, error) + + // Close flushes any buffered writes and releases the underlying database + // connection. It blocks until in-flight writes have drained (or their + // shutdown flush times out). Implementations with no resources to release + // return nil. + Close() error } diff --git a/p2p/database/datastore.go b/p2p/database/datastore.go index 4d4af9579..79d67a25a 100644 --- a/p2p/database/datastore.go +++ b/p2p/database/datastore.go @@ -224,9 +224,9 @@ func (d *Datastore) WriteBlockBody(ctx context.Context, body *eth.BlockBody, has }) } -// WriteBlockHashes will write the block events to datastore. -func (d *Datastore) WriteBlockHashes(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { - if d.client == nil || !d.ShouldWriteBlockEvents() || len(hashes) == 0 { +// WriteBlockEvents appends an inbound block event per hash for the given peer. +func (d *Datastore) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { + if d.client == nil || peer == nil || len(hashes) == 0 { return } @@ -235,11 +235,10 @@ func (d *Datastore) WriteBlockHashes(ctx context.Context, peer *enode.Node, hash }) } -// WriteBlockHashFirstSeen writes a partial block entry with just the hash -// first seen time if the block doesn't exist yet. If it exists, updates the -// TimeFirstSeenHash if the new time is earlier. +// WriteBlockHashFirstSeen stamps the block entity's earliest hash-first-seen +// time. The block event itself is written separately via WriteBlockEvents. func (d *Datastore) WriteBlockHashFirstSeen(ctx context.Context, peer *enode.Node, hash common.Hash, tfsh time.Time) { - if d.client == nil || (!d.ShouldWriteBlocks() && !d.shouldWriteFirstBlockEvent) { + if d.client == nil || !d.ShouldWriteBlocks() { return } @@ -250,16 +249,6 @@ func (d *Datastore) WriteBlockHashFirstSeen(ctx context.Context, peer *enode.Nod // writeBlockHashFirstSeen performs the actual transaction to write or update the block hash first seen time. func (d *Datastore) writeBlockHashFirstSeen(ctx context.Context, peer *enode.Node, hash common.Hash, tfsh time.Time) { - // Write block event if flag enabled and block events are disabled (mutually exclusive). - // Cache check in protocol.go already verified first-seen. - if d.shouldWriteFirstBlockEvent && !d.ShouldWriteBlockEvents() && peer != nil { - d.writeEvent(peer, BlockEventsKind, hash, BlocksKind, tfsh) - } - - if !d.shouldWriteBlocks { - return - } - key := datastore.NameKey(BlocksKind, hash.Hex(), nil) _, err := d.client.RunInTransaction(ctx, func(tx *datastore.Transaction) error { @@ -291,28 +280,28 @@ func (d *Datastore) writeBlockHashFirstSeen(ctx context.Context, peer *enode.Nod } } -// WriteTransactions will write the transactions and transaction events to datastore. +// WriteTransactions writes the transaction bodies to datastore. Transaction +// events are recorded separately via WriteTransactionEvents. func (d *Datastore) WriteTransactions(ctx context.Context, peer *enode.Node, txs []*types.Transaction, tfs time.Time) { - if d.client == nil { + if d.client == nil || !d.ShouldWriteTransactions() { return } - if d.ShouldWriteTransactions() { - d.runAsync(func() { - d.writeTransactions(ctx, txs, tfs) - }) - } - - if d.ShouldWriteTransactionEvents() { - hashes := make([]common.Hash, 0, len(txs)) - for _, tx := range txs { - hashes = append(hashes, tx.Hash()) - } + d.runAsync(func() { + d.writeTransactions(ctx, txs, tfs) + }) +} - d.runAsync(func() { - d.writeEvents(ctx, peer, TransactionEventsKind, hashes, TransactionsKind, tfs) - }) +// WriteTransactionEvents appends an inbound transaction event per hash for the +// given peer. +func (d *Datastore) WriteTransactionEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { + if d.client == nil || peer == nil || len(hashes) == 0 { + return } + + d.runAsync(func() { + d.writeEvents(ctx, peer, TransactionEventsKind, hashes, TransactionsKind, tfs) + }) } // WritePeers writes the connected peers to datastore. @@ -714,3 +703,17 @@ func (d *Datastore) NodeList(ctx context.Context, limit int) ([]string, error) { return nodelist, nil } + +// Close waits for in-flight async writes to finish, then releases the +// underlying Datastore client. Acquiring every slot in the jobs semaphore +// guarantees no write is in progress, since datastore.Client.Close must not be +// called concurrently with pending operations. +func (d *Datastore) Close() error { + if d.client == nil { + return nil + } + for range cap(d.jobs) { + d.jobs <- struct{}{} + } + return d.client.Close() +} diff --git a/p2p/database/json.go b/p2p/database/json.go index 633ce0c79..b8a796d6e 100644 --- a/p2p/database/json.go +++ b/p2p/database/json.go @@ -217,9 +217,9 @@ func (j *JSONDatabase) WriteBlockHeaders(ctx context.Context, headers []*types.H } } -// WriteBlockHashes writes the block events as JSON. -func (j *JSONDatabase) WriteBlockHashes(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { - if !j.ShouldWriteBlockEvents() || len(hashes) == 0 || peer == nil { +// WriteBlockEvents writes the block events as JSON. +func (j *JSONDatabase) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { + if peer == nil || len(hashes) == 0 { return } @@ -272,7 +272,6 @@ func (j *JSONDatabase) WriteBlockBody(ctx context.Context, body *eth.BlockBody, // WriteTransactions writes the transactions and transaction events as JSON. func (j *JSONDatabase) WriteTransactions(_ context.Context, peer *enode.Node, txs []*types.Transaction, tfs time.Time) { j.writeTxs(txs, tfs) - j.writeTxEvents(peer, txs, tfs) } func (j *JSONDatabase) writeTxs(txs []*types.Transaction, tfs time.Time) { @@ -315,17 +314,17 @@ func (j *JSONDatabase) writeTxs(txs []*types.Transaction, tfs time.Time) { } } -func (j *JSONDatabase) writeTxEvents(peer *enode.Node, txs []*types.Transaction, tfs time.Time) { - if !j.ShouldWriteTransactionEvents() || peer == nil { +func (j *JSONDatabase) WriteTransactionEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { + if peer == nil || len(hashes) == 0 { return } - for _, tx := range txs { + for _, hash := range hashes { event := JSONTransactionEvent{ Type: "transaction_event", SensorID: j.sensorID, PeerID: peer.URLv4(), - Hash: tx.Hash().Hex(), + Hash: hash.Hex(), Timestamp: tfs, } @@ -403,3 +402,8 @@ func (j *JSONDatabase) ShouldWritePeers() bool { func (j *JSONDatabase) NodeList(ctx context.Context, limit int) ([]string, error) { return []string{}, nil } + +// Close does nothing; records are written to stdout unbuffered. +func (j *JSONDatabase) Close() error { + return nil +} diff --git a/p2p/database/nodb.go b/p2p/database/nodb.go index 6a5337968..ae9f79459 100644 --- a/p2p/database/nodb.go +++ b/p2p/database/nodb.go @@ -29,8 +29,8 @@ func (n *nodb) WriteBlock(ctx context.Context, peer *enode.Node, block *types.Bl func (n *nodb) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool) { } -// WriteBlockHashes does nothing. -func (n *nodb) WriteBlockHashes(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { +// WriteBlockEvents does nothing. +func (n *nodb) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { } // WriteBlockHashFirstSeen does nothing. @@ -45,6 +45,10 @@ func (n *nodb) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash com func (n *nodb) WriteTransactions(ctx context.Context, peer *enode.Node, txs []*types.Transaction, tfs time.Time) { } +// WriteTransactionEvents does nothing. +func (n *nodb) WriteTransactionEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { +} + // WritePeers does nothing. func (n *nodb) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.Time) { } @@ -98,3 +102,8 @@ func (n *nodb) ShouldWritePeers() bool { func (n *nodb) NodeList(ctx context.Context, limit int) ([]string, error) { return []string{}, nil } + +// Close does nothing. +func (n *nodb) Close() error { + return nil +} diff --git a/p2p/protocol.go b/p2p/protocol.go index 2aec66e64..6dcb2e5e9 100644 --- a/p2p/protocol.go +++ b/p2p/protocol.go @@ -211,7 +211,7 @@ func NewEthProtocol(version uint, opts EthProtocolOptions) ethp2p.Protocol { case eth.NewBlockMsg: err = c.handleNewBlock(ctx, msg) case eth.NewPooledTransactionHashesMsg: - err = c.handleNewPooledTransactionHashes(version, msg) + err = c.handleNewPooledTransactionHashes(ctx, version, msg) case eth.GetPooledTransactionsMsg: err = c.handleGetPooledTransactions(msg) case eth.PooledTransactionsMsg: @@ -518,6 +518,20 @@ func (c *conn) getParentBlock(ctx context.Context, header *types.Header) error { return c.getBlockData(header.ParentHash, cache, true) } +// eventHashes selects which announced hashes to record as inbound events: every +// announcement for the full per-peer stream, only the first-seen hashes for +// first-seen mode, or none if neither is enabled. +func eventHashes(all, firstSeen []common.Hash, full, firstOnly bool) []common.Hash { + switch { + case full: + return all + case firstOnly: + return firstSeen + default: + return nil + } +} + func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error { var packet NewBlockHashesPacket if err := msg.Decode(&packet); err != nil { @@ -528,13 +542,15 @@ func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error { c.countMsgReceived(packet.Name(), float64(len(packet))) - // Collect unique hashes (and their numbers) for the database write and, - // when signer validation is disabled, for immediate rebroadcast. + // allHashes is every announced hash (the full per-peer event stream); + // uniqueHashes are the first-seen ones (for first-seen events and rebroadcast). + allHashes := make([]common.Hash, 0, len(packet)) uniqueHashes := make([]common.Hash, 0, len(packet)) uniqueNumbers := make([]uint64, 0, len(packet)) for _, entry := range packet { hash := entry.Hash + allHashes = append(allHashes, hash) // Update latest block info atomically if this block is newer c.latestBlock.Update(func(current latestBlock) (latestBlock, bool) { @@ -577,13 +593,16 @@ func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error { } } - // Write only unique hashes to the database. + // Record inbound block events: the full per-peer stream (every peer that + // announced) or only the first-seen hashes, per the flags. + c.db.WriteBlockEvents(ctx, c.node, + eventHashes(allHashes, uniqueHashes, c.db.ShouldWriteBlockEvents(), c.db.ShouldWriteFirstBlockEvent()), tfs) + + // Only newly-seen hashes are rebroadcast. if len(uniqueHashes) == 0 { return nil } - c.db.WriteBlockHashes(ctx, c.node, uniqueHashes, tfs) - // When signer validation is enabled, defer the hash rebroadcast to // handleBlockHeaders so the fetched header can be validated first. Body // requests (getBlockData above) are intentionally not gated. When @@ -1338,7 +1357,7 @@ func (c *conn) handleGetPooledTransactions(msg ethp2p.Msg) error { return ethp2p.Send(c.rw, eth.PooledTransactionsMsg, response) } -func (c *conn) handleNewPooledTransactionHashes(version uint, msg ethp2p.Msg) error { +func (c *conn) handleNewPooledTransactionHashes(ctx context.Context, version uint, msg ethp2p.Msg) error { var hashes []common.Hash var name string @@ -1363,7 +1382,26 @@ func (c *conn) handleNewPooledTransactionHashes(version uint, msg ethp2p.Msg) er c.countMsgReceived(name, float64(len(hashes))) - if !c.db.ShouldWriteTransactions() && !c.db.ShouldWriteTransactionEvents() { + if !c.db.ShouldWriteTransactions() && !c.db.ShouldWriteTransactionEvents() && !c.db.ShouldWriteFirstTransactionEvent() { + return nil + } + + // Record inbound tx events: the full per-peer stream (every announcement) + // or only the first-seen hashes, per the flags. Done independently of the + // fetch dedup below so re-announcements from other peers are still recorded. + if c.db.ShouldWriteTransactionEvents() || c.db.ShouldWriteFirstTransactionEvent() { + firstSeen := make([]common.Hash, 0, len(hashes)) + for _, hash := range hashes { + if c.conns.MarkTxSeen(hash) { + firstSeen = append(firstSeen, hash) + } + } + c.db.WriteTransactionEvents(ctx, c.node, + eventHashes(hashes, firstSeen, c.db.ShouldWriteTransactionEvents(), c.db.ShouldWriteFirstTransactionEvent()), time.Now()) + } + + // Transaction bodies are only fetched when we're persisting them. + if !c.db.ShouldWriteTransactions() { return nil } diff --git a/p2p/protocol_events_test.go b/p2p/protocol_events_test.go new file mode 100644 index 000000000..ee9ed5691 --- /dev/null +++ b/p2p/protocol_events_test.go @@ -0,0 +1,127 @@ +package p2p + +import ( + "context" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/eth/protocols/eth" + ethp2p "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/enode" + + "github.com/0xPolygon/polygon-cli/p2p/database" +) + +// recordingDB is a Database that counts the hashes passed to the event writers +// and reports configurable Should* flags. Unwritten methods fall through to the +// no-op backend. +type recordingDB struct { + database.Database + blockEvents, txEvents int + fullBlock, firstBlock, fullTx, firstTx bool +} + +func (r *recordingDB) WriteBlockEvents(_ context.Context, _ *enode.Node, hashes []common.Hash, _ time.Time) { + r.blockEvents += len(hashes) +} + +func (r *recordingDB) WriteTransactionEvents(_ context.Context, _ *enode.Node, hashes []common.Hash, _ time.Time) { + r.txEvents += len(hashes) +} + +func (r *recordingDB) ShouldWriteBlockEvents() bool { return r.fullBlock } +func (r *recordingDB) ShouldWriteFirstBlockEvent() bool { return r.firstBlock } +func (r *recordingDB) ShouldWriteTransactionEvents() bool { return r.fullTx } +func (r *recordingDB) ShouldWriteFirstTransactionEvent() bool { return r.firstTx } +func (r *recordingDB) ShouldWriteTransactions() bool { return false } + +func makeTxAnnounce(t *testing.T, hash common.Hash) ethp2p.Msg { + t.Helper() + return encodeMsg(t, eth.NewPooledTransactionHashesMsg, ð.NewPooledTransactionHashesPacket{ + Types: []byte{0}, + Sizes: []uint32{100}, + Hashes: []common.Hash{hash}, + }) +} + +func TestEventHashes(t *testing.T) { + all := []common.Hash{{1}, {2}} + first := []common.Hash{{1}} + if got := eventHashes(all, first, true, false); len(got) != 2 { + t.Errorf("full: got %d hashes, want 2", len(got)) + } + if got := eventHashes(all, first, false, true); len(got) != 1 { + t.Errorf("first: got %d hashes, want 1", len(got)) + } + if got := eventHashes(all, first, false, false); got != nil { + t.Errorf("neither: got %v, want nil", got) + } + if got := eventHashes(all, first, true, true); len(got) != 2 { + t.Errorf("full takes precedence: got %d hashes, want 2", len(got)) + } +} + +// TestBlockEventFullVsFirst announces the same hash twice (two peers). The full +// per-peer stream records every announcement; first-seen mode records it once; +// neither records nothing. +func TestBlockEventFullVsFirst(t *testing.T) { + conns := sharedTestConns(t, false) + ctx := context.Background() + cases := []struct { + name string + full, first bool + want int + }{ + {"full", true, false, 2}, + {"first", false, true, 1}, + {"neither", false, false, 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rec := &recordingDB{Database: database.NoDatabase(), fullBlock: tc.full, firstBlock: tc.first} + c := newTestConn(&recordRW{}, conns) + c.db = rec + hash := common.BytesToHash([]byte("blkevt-" + tc.name)) + for i := 0; i < 2; i++ { + if err := c.handleNewBlockHashes(ctx, makeAnnounce(t, hash, 200)); err != nil { + t.Fatalf("announce %d: %v", i, err) + } + } + if rec.blockEvents != tc.want { + t.Fatalf("%s: recorded %d block events, want %d", tc.name, rec.blockEvents, tc.want) + } + }) + } +} + +// TestTransactionEventFullVsFirst is the tx mirror of TestBlockEventFullVsFirst. +func TestTransactionEventFullVsFirst(t *testing.T) { + conns := sharedTestConns(t, false) + ctx := context.Background() + cases := []struct { + name string + full, first bool + want int + }{ + {"full", true, false, 2}, + {"first", false, true, 1}, + {"neither", false, false, 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rec := &recordingDB{Database: database.NoDatabase(), fullTx: tc.full, firstTx: tc.first} + c := newTestConn(&recordRW{}, conns) + c.db = rec + hash := common.BytesToHash([]byte("txevt-" + tc.name)) + for i := 0; i < 2; i++ { + if err := c.handleNewPooledTransactionHashes(ctx, 68, makeTxAnnounce(t, hash)); err != nil { + t.Fatalf("announce %d: %v", i, err) + } + } + if rec.txEvents != tc.want { + t.Fatalf("%s: recorded %d tx events, want %d", tc.name, rec.txEvents, tc.want) + } + }) + } +}