diff --git a/go.mod b/go.mod index 0759ddd5f..010ba7c3e 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/c2h5oh/datasize v0.0.0-20200825124411-48ed595a09d2 github.com/golang/protobuf v1.5.2 github.com/google/btree v1.0.1 + github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d github.com/holiman/uint256 v1.2.0 github.com/ledgerwatch/log/v3 v3.3.0 diff --git a/go.sum b/go.sum index 5b2de2668..d2ae2e86e 100644 --- a/go.sum +++ b/go.sum @@ -25,8 +25,11 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ= github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= +github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= +github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk= github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= @@ -53,11 +56,17 @@ github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/ github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU= github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 h1:+9834+KizmvFV7pXQGSXQTsaWhq2GjuNUt0aUU0YBYw= +github.com/grpc-ecosystem/go-grpc-middleware v1.3.0/go.mod h1:z0ButlSOZa5vEBq9m2m2hlwIgKw+rp3sdCBRoJY+30Y= github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw= github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d h1:dg1dEPuWpEqDnvIw251EVy4zlP8gWbsGj4BsUKCRpYs= github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/holiman/uint256 v1.2.0 h1:gpSYcPLWGv4sG43I2mVLiDZCNDh/EpGjSk8tmtxitHM= github.com/holiman/uint256 v1.2.0/go.mod h1:y4ga/t+u+Xwd7CpDgZESaRcWy0I7XMlTMA25ApIH5Jw= +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/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= +github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= github.com/ledgerwatch/log/v3 v3.3.0 h1:k8N/3NQLILr8CKCMyza261vLFKU7VA+nMNNb0wVyQSc= github.com/ledgerwatch/log/v3 v3.3.0/go.mod h1:J58eOHHrIYHxl7LKkRsb/0YibKwtLfauUryl5SLRGm0= github.com/ledgerwatch/secp256k1 v0.0.0-20210626115225-cd5cd00ed72d h1:/IKMrJdfRsoYNc36PXqP4xMH3vhW/8IQyBKGQbKZUno= @@ -71,11 +80,16 @@ github.com/mattn/go-isatty v0.0.13 h1:qdl+GuBjcsKKDco5BsxPJlId98mSWNKqYA+Co0SC1y github.com/mattn/go-isatty v0.0.13/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU= github.com/mschoch/smat v0.2.0 h1:8imxQsjDm8yFEAVBe7azKmKSgzSkZXDuKkSq9374khM= github.com/mschoch/smat v0.2.0/go.mod h1:kc9mz7DoBKqDyiRL7VZN8KvXQMWeTaVnttLRXOlotKw= +github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= +github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ= +github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= @@ -87,10 +101,15 @@ github.com/valyala/fastrand v1.0.0 h1:LUKT9aKer2dVQNUi3waewTbKV+7H17kvWFNKs2Obdk github.com/valyala/fastrand v1.0.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ= github.com/valyala/histogram v1.1.2 h1:vOk5VrGjMBIoPR5k6wA8vBaC8toeJ8XO0yfRjFEc1h8= github.com/valyala/histogram v1.1.2/go.mod h1:CZAr6gK9dbD7hYx2s8WSPh0p5x5wETjC+2b3PJVtEdg= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI= +go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE= go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= +go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= @@ -100,6 +119,7 @@ golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU= golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0 h1:RM4zey1++hCTbCVQfnWeKs9/IEsaBLA8vTkd0WVtmH4= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -109,8 +129,10 @@ golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73r golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/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-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110 h1:qWPm9rbaAMKs8Bq/9LRpbMqxWRVUAQwMI9fVrssnTfw= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= @@ -119,13 +141,17 @@ golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c h1:F1jZWGFhYfh0Ci55sIpILtKKK8p3i2/krTr0H1rg74I= @@ -140,8 +166,10 @@ golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= -golang.org/x/tools v0.0.0-20200815165600-90abf76919f3 h1:0aScV/0rLmANzEYIhjCOi2pTvDyhZNduBUMD2q3iqs4= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200815165600-90abf76919f3/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a h1:CB3a9Nez8M13wwlr/E2YtwoU+qYHKfC+JrDa45RXXoQ= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= 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= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -151,6 +179,7 @@ google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9Ywl google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc= +google.golang.org/genproto v0.0.0-20200423170343-7949de9c1215/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c= google.golang.org/genproto v0.0.0-20200513103714-09dca8ec2884/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c= google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 h1:+kGHl1aib/qcwaRi1CbqBZ1rk19r85MNUf8HaBghugY= google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo= @@ -158,6 +187,7 @@ google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZi google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg= google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY= google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk= +google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3IjizoKk= google.golang.org/grpc v1.33.1/go.mod h1:fr5YgcSWrqhRRxogOsw7RzIpsmvOZ6IcH4kBYTpR3n0= google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU= google.golang.org/grpc v1.39.1 h1:f37vZbBVTiJ6jKG5mWz8ySOBxNqy6ViPgyhSdVnxF3E= diff --git a/gointerfaces/txpool/txpool.pb.go b/gointerfaces/txpool/txpool.pb.go index 040ae6bbf..19b0afd41 100644 --- a/gointerfaces/txpool/txpool.pb.go +++ b/gointerfaces/txpool/txpool.pb.go @@ -83,8 +83,9 @@ func (ImportResult) EnumDescriptor() ([]byte, []int) { type AllReply_Type int32 const ( - AllReply_PENDING AllReply_Type = 0 // All currently processable transactions - AllReply_QUEUED AllReply_Type = 1 // Queued but non-processable transactions + AllReply_PENDING AllReply_Type = 0 // All currently processable transactions + AllReply_QUEUED AllReply_Type = 1 // Queued but non-processable transactions + AllReply_BASE_FEE AllReply_Type = 2 // BaseFee not enough baseFee non-processable transactions ) // Enum value maps for AllReply_Type. @@ -92,10 +93,12 @@ var ( AllReply_Type_name = map[int32]string{ 0: "PENDING", 1: "QUEUED", + 2: "BASE_FEE", } AllReply_Type_value = map[string]int32{ - "PENDING": 0, - "QUEUED": 1, + "PENDING": 0, + "QUEUED": 1, + "BASE_FEE": 2, } ) @@ -584,6 +587,7 @@ type StatusReply struct { PendingCount uint32 `protobuf:"varint,1,opt,name=pendingCount,proto3" json:"pendingCount,omitempty"` QueuedCount uint32 `protobuf:"varint,2,opt,name=queuedCount,proto3" json:"queuedCount,omitempty"` + BaseFeeCount uint32 `protobuf:"varint,3,opt,name=baseFeeCount,proto3" json:"baseFeeCount,omitempty"` } func (x *StatusReply) Reset() { @@ -632,6 +636,13 @@ func (x *StatusReply) GetQueuedCount() uint32 { return 0 } +func (x *StatusReply) GetBaseFeeCount() uint32 { + if x != nil { + return x.BaseFeeCount + } + return 0 +} + type AllReply_Tx struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache @@ -724,7 +735,7 @@ var file_txpool_txpool_proto_rawDesc = []byte{ 0x65, 0x73, 0x74, 0x22, 0x24, 0x0a, 0x0a, 0x4f, 0x6e, 0x41, 0x64, 0x64, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x16, 0x0a, 0x06, 0x72, 0x70, 0x6c, 0x54, 0x78, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0c, 0x52, 0x06, 0x72, 0x70, 0x6c, 0x54, 0x78, 0x73, 0x22, 0x0c, 0x0a, 0x0a, 0x41, 0x6c, 0x6c, - 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0xb1, 0x01, 0x0a, 0x08, 0x41, 0x6c, 0x6c, 0x52, + 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0xbf, 0x01, 0x0a, 0x08, 0x41, 0x6c, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x25, 0x0a, 0x03, 0x74, 0x78, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x41, 0x6c, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x2e, 0x54, 0x78, 0x52, 0x03, 0x74, 0x78, 0x73, 0x1a, 0x5d, 0x0a, 0x02, 0x54, @@ -733,48 +744,51 @@ var file_txpool_txpool_proto_rawDesc = []byte{ 0x79, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x65, 0x6e, 0x64, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x73, 0x65, 0x6e, 0x64, 0x65, 0x72, 0x12, 0x14, 0x0a, 0x05, 0x72, 0x6c, 0x70, 0x54, 0x78, 0x18, 0x03, 0x20, - 0x01, 0x28, 0x0c, 0x52, 0x05, 0x72, 0x6c, 0x70, 0x54, 0x78, 0x22, 0x1f, 0x0a, 0x04, 0x54, 0x79, + 0x01, 0x28, 0x0c, 0x52, 0x05, 0x72, 0x6c, 0x70, 0x54, 0x78, 0x22, 0x2d, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x50, 0x45, 0x4e, 0x44, 0x49, 0x4e, 0x47, 0x10, 0x00, 0x12, - 0x0a, 0x0a, 0x06, 0x51, 0x55, 0x45, 0x55, 0x45, 0x44, 0x10, 0x01, 0x22, 0x0f, 0x0a, 0x0d, 0x53, - 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x53, 0x0a, 0x0b, - 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x22, 0x0a, 0x0c, 0x70, - 0x65, 0x6e, 0x64, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, - 0x0d, 0x52, 0x0c, 0x70, 0x65, 0x6e, 0x64, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, - 0x20, 0x0a, 0x0b, 0x71, 0x75, 0x65, 0x75, 0x65, 0x64, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x02, - 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0b, 0x71, 0x75, 0x65, 0x75, 0x65, 0x64, 0x43, 0x6f, 0x75, 0x6e, - 0x74, 0x2a, 0x6c, 0x0a, 0x0c, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x75, 0x6c, - 0x74, 0x12, 0x0b, 0x0a, 0x07, 0x53, 0x55, 0x43, 0x43, 0x45, 0x53, 0x53, 0x10, 0x00, 0x12, 0x12, - 0x0a, 0x0e, 0x41, 0x4c, 0x52, 0x45, 0x41, 0x44, 0x59, 0x5f, 0x45, 0x58, 0x49, 0x53, 0x54, 0x53, - 0x10, 0x01, 0x12, 0x0f, 0x0a, 0x0b, 0x46, 0x45, 0x45, 0x5f, 0x54, 0x4f, 0x4f, 0x5f, 0x4c, 0x4f, - 0x57, 0x10, 0x02, 0x12, 0x09, 0x0a, 0x05, 0x53, 0x54, 0x41, 0x4c, 0x45, 0x10, 0x03, 0x12, 0x0b, - 0x0a, 0x07, 0x49, 0x4e, 0x56, 0x41, 0x4c, 0x49, 0x44, 0x10, 0x04, 0x12, 0x12, 0x0a, 0x0e, 0x49, - 0x4e, 0x54, 0x45, 0x52, 0x4e, 0x41, 0x4c, 0x5f, 0x45, 0x52, 0x52, 0x4f, 0x52, 0x10, 0x05, 0x32, - 0x80, 0x03, 0x0a, 0x06, 0x54, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x12, 0x36, 0x0a, 0x07, 0x56, 0x65, - 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, - 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x13, 0x2e, - 0x74, 0x79, 0x70, 0x65, 0x73, 0x2e, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x70, - 0x6c, 0x79, 0x12, 0x31, 0x0a, 0x0b, 0x46, 0x69, 0x6e, 0x64, 0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, - 0x6e, 0x12, 0x10, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x54, 0x78, 0x48, 0x61, 0x73, - 0x68, 0x65, 0x73, 0x1a, 0x10, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x54, 0x78, 0x48, - 0x61, 0x73, 0x68, 0x65, 0x73, 0x12, 0x2b, 0x0a, 0x03, 0x41, 0x64, 0x64, 0x12, 0x12, 0x2e, 0x74, - 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x41, 0x64, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, - 0x1a, 0x10, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x41, 0x64, 0x64, 0x52, 0x65, 0x70, - 0x6c, 0x79, 0x12, 0x46, 0x0a, 0x0c, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, - 0x6e, 0x73, 0x12, 0x1b, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x54, 0x72, 0x61, 0x6e, - 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, - 0x19, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x61, 0x63, - 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x2b, 0x0a, 0x03, 0x41, 0x6c, - 0x6c, 0x12, 0x12, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x41, 0x6c, 0x6c, 0x52, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x10, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x41, - 0x6c, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x33, 0x0a, 0x05, 0x4f, 0x6e, 0x41, 0x64, 0x64, - 0x12, 0x14, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x4f, 0x6e, 0x41, 0x64, 0x64, 0x52, - 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, - 0x4f, 0x6e, 0x41, 0x64, 0x64, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x30, 0x01, 0x12, 0x34, 0x0a, 0x06, - 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x15, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, - 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, - 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x70, - 0x6c, 0x79, 0x42, 0x11, 0x5a, 0x0f, 0x2e, 0x2f, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x3b, 0x74, - 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x0a, 0x0a, 0x06, 0x51, 0x55, 0x45, 0x55, 0x45, 0x44, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x42, + 0x41, 0x53, 0x45, 0x5f, 0x46, 0x45, 0x45, 0x10, 0x02, 0x22, 0x0f, 0x0a, 0x0d, 0x53, 0x74, 0x61, + 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x77, 0x0a, 0x0b, 0x53, 0x74, + 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x22, 0x0a, 0x0c, 0x70, 0x65, 0x6e, + 0x64, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, + 0x0c, 0x70, 0x65, 0x6e, 0x64, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x20, 0x0a, + 0x0b, 0x71, 0x75, 0x65, 0x75, 0x65, 0x64, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x02, 0x20, 0x01, + 0x28, 0x0d, 0x52, 0x0b, 0x71, 0x75, 0x65, 0x75, 0x65, 0x64, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, + 0x22, 0x0a, 0x0c, 0x62, 0x61, 0x73, 0x65, 0x46, 0x65, 0x65, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x18, + 0x03, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0c, 0x62, 0x61, 0x73, 0x65, 0x46, 0x65, 0x65, 0x43, 0x6f, + 0x75, 0x6e, 0x74, 0x2a, 0x6c, 0x0a, 0x0c, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, + 0x75, 0x6c, 0x74, 0x12, 0x0b, 0x0a, 0x07, 0x53, 0x55, 0x43, 0x43, 0x45, 0x53, 0x53, 0x10, 0x00, + 0x12, 0x12, 0x0a, 0x0e, 0x41, 0x4c, 0x52, 0x45, 0x41, 0x44, 0x59, 0x5f, 0x45, 0x58, 0x49, 0x53, + 0x54, 0x53, 0x10, 0x01, 0x12, 0x0f, 0x0a, 0x0b, 0x46, 0x45, 0x45, 0x5f, 0x54, 0x4f, 0x4f, 0x5f, + 0x4c, 0x4f, 0x57, 0x10, 0x02, 0x12, 0x09, 0x0a, 0x05, 0x53, 0x54, 0x41, 0x4c, 0x45, 0x10, 0x03, + 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x4e, 0x56, 0x41, 0x4c, 0x49, 0x44, 0x10, 0x04, 0x12, 0x12, 0x0a, + 0x0e, 0x49, 0x4e, 0x54, 0x45, 0x52, 0x4e, 0x41, 0x4c, 0x5f, 0x45, 0x52, 0x52, 0x4f, 0x52, 0x10, + 0x05, 0x32, 0x80, 0x03, 0x0a, 0x06, 0x54, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x12, 0x36, 0x0a, 0x07, + 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, + 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, + 0x13, 0x2e, 0x74, 0x79, 0x70, 0x65, 0x73, 0x2e, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x52, + 0x65, 0x70, 0x6c, 0x79, 0x12, 0x31, 0x0a, 0x0b, 0x46, 0x69, 0x6e, 0x64, 0x55, 0x6e, 0x6b, 0x6e, + 0x6f, 0x77, 0x6e, 0x12, 0x10, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x54, 0x78, 0x48, + 0x61, 0x73, 0x68, 0x65, 0x73, 0x1a, 0x10, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x54, + 0x78, 0x48, 0x61, 0x73, 0x68, 0x65, 0x73, 0x12, 0x2b, 0x0a, 0x03, 0x41, 0x64, 0x64, 0x12, 0x12, + 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x41, 0x64, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, + 0x73, 0x74, 0x1a, 0x10, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x41, 0x64, 0x64, 0x52, + 0x65, 0x70, 0x6c, 0x79, 0x12, 0x46, 0x0a, 0x0c, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x61, 0x63, 0x74, + 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1b, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x54, 0x72, + 0x61, 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x1a, 0x19, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x54, 0x72, 0x61, 0x6e, 0x73, + 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x2b, 0x0a, 0x03, + 0x41, 0x6c, 0x6c, 0x12, 0x12, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x41, 0x6c, 0x6c, + 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x10, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, + 0x2e, 0x41, 0x6c, 0x6c, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x33, 0x0a, 0x05, 0x4f, 0x6e, 0x41, + 0x64, 0x64, 0x12, 0x14, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x4f, 0x6e, 0x41, 0x64, + 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, + 0x6c, 0x2e, 0x4f, 0x6e, 0x41, 0x64, 0x64, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x30, 0x01, 0x12, 0x34, + 0x0a, 0x06, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x15, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, + 0x6c, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, + 0x13, 0x2e, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, + 0x65, 0x70, 0x6c, 0x79, 0x42, 0x11, 0x5a, 0x0f, 0x2e, 0x2f, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, + 0x3b, 0x74, 0x78, 0x70, 0x6f, 0x6f, 0x6c, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/gointerfaces/txpool/txpool_grpc.pb.go b/gointerfaces/txpool/txpool_grpc.pb.go index 1156b3ec6..4c1d4b06a 100644 --- a/gointerfaces/txpool/txpool_grpc.pb.go +++ b/gointerfaces/txpool/txpool_grpc.pb.go @@ -3,12 +3,13 @@ package txpool import ( - context "context" - types "github.com/ledgerwatch/erigon-lib/gointerfaces/types" - grpc "google.golang.org/grpc" - codes "google.golang.org/grpc/codes" - status "google.golang.org/grpc/status" - emptypb "google.golang.org/protobuf/types/known/emptypb" + "context" + + "github.com/ledgerwatch/erigon-lib/gointerfaces/types" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/emptypb" ) // This is a compile-time assertion to ensure that this generated file diff --git a/interfaces/txpool/txpool.proto b/interfaces/txpool/txpool.proto index 73bc99fdb..a687cd155 100644 --- a/interfaces/txpool/txpool.proto +++ b/interfaces/txpool/txpool.proto @@ -35,6 +35,7 @@ message AllReply { enum Type { PENDING = 0; // All currently processable transactions QUEUED = 1; // Queued but non-processable transactions + BASE_FEE = 2; // BaseFee not enough baseFee non-processable transactions } message Tx { Type type = 1; @@ -48,6 +49,7 @@ message StatusRequest {} message StatusReply { uint32 pendingCount = 1; uint32 queuedCount = 2; + uint32 baseFeeCount = 3; } service Txpool { diff --git a/txpool/fetch.go b/txpool/fetch.go index ca3aa0e78..dcff4f170 100644 --- a/txpool/fetch.go +++ b/txpool/fetch.go @@ -310,7 +310,7 @@ func (f *Fetch) handleInboundMessage(ctx context.Context, req *sentry.InboundMes if len(txs.txs) == 0 { return nil } - f.pool.OnNewRemoteTxs(ctx, txs) + f.pool.AddRemoteTxs(ctx, txs) default: //defer log.Info("dropped", "id", req.Id) } diff --git a/txpool/grpc_server.go b/txpool/grpc_server.go new file mode 100644 index 000000000..611519106 --- /dev/null +++ b/txpool/grpc_server.go @@ -0,0 +1,302 @@ +package txpool + +import ( + "context" + "fmt" + "net" + "sync" + "time" + + grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware" + grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery" + "github.com/ledgerwatch/erigon-lib/gointerfaces" + txpool_proto "github.com/ledgerwatch/erigon-lib/gointerfaces/txpool" + types2 "github.com/ledgerwatch/erigon-lib/gointerfaces/types" + "github.com/ledgerwatch/erigon-lib/kv" + "github.com/ledgerwatch/log/v3" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" + "google.golang.org/protobuf/types/known/emptypb" +) + +// TxPoolAPIVersion +var TxPoolAPIVersion = &types2.VersionReply{Major: 1, Minor: 0, Patch: 0} + +type txPool interface { + GetRlp(tx kv.Tx, hash []byte) ([]byte, error) + AddLocals(ctx context.Context, newTxs TxSlots, tx kv.Tx) ([]DiscardReason, error) + DeprecatedForEach(_ context.Context, f func(rlp, sender []byte, t SubPoolType), tx kv.Tx) error + CountContent() (int, int, int) + IdHashKnown(tx kv.Tx, hash []byte) (bool, error) +} + +type GrpcServer struct { + txpool_proto.UnimplementedTxpoolServer + ctx context.Context + txPool txPool + db kv.RoDB + NewSlotsStreams *NewSlotsStreams +} + +func NewGrpcServer(ctx context.Context, txPool txPool, db kv.RoDB) *GrpcServer { + return &GrpcServer{ctx: ctx, txPool: txPool, db: db, NewSlotsStreams: &NewSlotsStreams{}} +} + +func (s *GrpcServer) Version(context.Context, *emptypb.Empty) (*types2.VersionReply, error) { + return TxPoolAPIVersion, nil +} +func convertSubPoolType(t SubPoolType) txpool_proto.AllReply_Type { + switch t { + case PendingSubPool: + return txpool_proto.AllReply_PENDING + case BaseFeeSubPool: + return txpool_proto.AllReply_PENDING + case QueuedSubPool: + return txpool_proto.AllReply_QUEUED + default: + panic("unknown") + } +} +func (s *GrpcServer) All(ctx context.Context, _ *txpool_proto.AllRequest) (*txpool_proto.AllReply, error) { + tx, err := s.db.BeginRo(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback() + reply := &txpool_proto.AllReply{} + reply.Txs = make([]*txpool_proto.AllReply_Tx, 0, 32) + if err := s.txPool.DeprecatedForEach(ctx, func(rlp, sender []byte, t SubPoolType) { + reply.Txs = append(reply.Txs, &txpool_proto.AllReply_Tx{ + Sender: sender, + Type: convertSubPoolType(t), + RlpTx: rlp, + }) + }, tx); err != nil { + return nil, err + } + return reply, nil +} + +func (s *GrpcServer) FindUnknown(ctx context.Context, in *txpool_proto.TxHashes) (*txpool_proto.TxHashes, error) { + return nil, fmt.Errorf("unimplemented") + /* + var underpriced int + for i := range in.Hashes { + h := gointerfaces.ConvertH256ToHash(in.Hashes[i]) + if s.txPool.Has(h) { + continue + } + if s.underpriced.Contains(h) { + underpriced++ + continue + } + reply.Hashes = append(reply.Hashes, in.Hashes[i]) + } + txAnnounceInMeter.Mark(int64(len(in.Hashes))) + txAnnounceKnownMeter.Mark(int64(len(in.Hashes) - len(reply.Hashes))) + txAnnounceUnderpricedMeter.Mark(int64(underpriced)) + */ +} + +func (s *GrpcServer) Add(ctx context.Context, in *txpool_proto.AddRequest) (*txpool_proto.AddReply, error) { + tx, err := s.db.BeginRo(context.Background()) + if err != nil { + return nil, err + } + defer tx.Rollback() + + var slots TxSlots + slots.Resize(uint(len(in.RlpTxs))) + parseCtx := NewTxParseContext() + parseCtx.Reject(func(hash []byte) bool { + known, _ := s.txPool.IdHashKnown(tx, hash) + return known + }) + for i := range in.RlpTxs { + slots.txs[i] = &TxSlot{} + slots.isLocal[i] = true + if _, err := parseCtx.ParseTransaction(in.RlpTxs[i], 0, slots.txs[i], slots.senders.At(i)); err != nil { + log.Warn("stream.Recv", "err", err) + continue + } + } + + reply := &txpool_proto.AddReply{Imported: make([]txpool_proto.ImportResult, len(in.RlpTxs)), Errors: make([]string, len(in.RlpTxs))} + discardReasons, err := s.txPool.AddLocals(ctx, slots, tx) + if err != nil { + return nil, err + } + //TODO: concept of discardReasons not really implemented yet + _ = discardReasons + /* + for i, err := range discardReasons { + if err == nil { + continue + } + + reply.Errors[i] = err.Error() + + // Track a few interesting failure types + switch err { + case Success: // Noop, but need to handle to not count these + + //case core.ErrAlreadyKnown: + // reply.Imported[i] = txpool_proto.ImportResult_ALREADY_EXISTS + //case core.ErrUnderpriced, core.ErrReplaceUnderpriced: + // reply.Imported[i] = txpool_proto.ImportResult_FEE_TOO_LOW + //case core.ErrInvalidSender, core.ErrGasLimit, core.ErrNegativeValue, core.ErrOversizedData: + // reply.Imported[i] = txpool_proto.ImportResult_INVALID + default: + reply.Imported[i] = txpool_proto.ImportResult_INTERNAL_ERROR + } + } + */ + return reply, nil +} + +func (s *GrpcServer) OnAdd(req *txpool_proto.OnAddRequest, stream txpool_proto.Txpool_OnAddServer) error { + //txpool.Loop does send messages to this streams + remove := s.NewSlotsStreams.Add(stream) + defer remove() + select { + case <-stream.Context().Done(): + return stream.Context().Err() + case <-s.ctx.Done(): + return s.ctx.Err() + } +} + +func (s *GrpcServer) Transactions(ctx context.Context, in *txpool_proto.TransactionsRequest) (*txpool_proto.TransactionsReply, error) { + tx, err := s.db.BeginRo(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback() + + reply := &txpool_proto.TransactionsReply{RlpTxs: make([][]byte, len(in.Hashes))} + for i := range in.Hashes { + h := gointerfaces.ConvertH256ToHash(in.Hashes[i]) + txnRlp, err := s.txPool.GetRlp(tx, h[:]) + if err != nil { + return nil, err + } + reply.RlpTxs[i] = txnRlp + } + + return reply, nil +} + +func (s *GrpcServer) Status(_ context.Context, _ *txpool_proto.StatusRequest) (*txpool_proto.StatusReply, error) { + pending, baseFee, queued := s.txPool.CountContent() + return &txpool_proto.StatusReply{ + PendingCount: uint32(pending), + QueuedCount: uint32(queued), + BaseFeeCount: uint32(baseFee), + }, nil +} + +// NewSlotsStreams - it's safe to use this class as non-pointer +type NewSlotsStreams struct { + chans map[uint]txpool_proto.Txpool_OnAddServer + mu sync.Mutex + id uint +} + +func (s *NewSlotsStreams) Add(stream txpool_proto.Txpool_OnAddServer) (remove func()) { + s.mu.Lock() + defer s.mu.Unlock() + if s.chans == nil { + s.chans = make(map[uint]txpool_proto.Txpool_OnAddServer) + } + s.id++ + id := s.id + s.chans[id] = stream + return func() { s.remove(id) } +} + +func (s *NewSlotsStreams) Broadcast(reply *txpool_proto.OnAddReply) { + s.mu.Lock() + defer s.mu.Unlock() + for id, stream := range s.chans { + err := stream.Send(reply) + if err != nil { + log.Debug("failed send to mined block stream", "err", err) + select { + case <-stream.Context().Done(): + delete(s.chans, id) + default: + } + } + } +} + +func (s *NewSlotsStreams) remove(id uint) { + s.mu.Lock() + defer s.mu.Unlock() + _, ok := s.chans[id] + if !ok { // double-unsubscribe support + return + } + delete(s.chans, id) +} + +func StartGrpc(txPoolServer txpool_proto.TxpoolServer, miningServer txpool_proto.MiningServer, addr string, creds *credentials.TransportCredentials) (*grpc.Server, error) { + lis, err := net.Listen("tcp", addr) + if err != nil { + return nil, fmt.Errorf("could not create listener: %w, addr=%s", err, addr) + } + + var ( + streamInterceptors []grpc.StreamServerInterceptor + unaryInterceptors []grpc.UnaryServerInterceptor + ) + streamInterceptors = append(streamInterceptors, grpc_recovery.StreamServerInterceptor()) + unaryInterceptors = append(unaryInterceptors, grpc_recovery.UnaryServerInterceptor()) + + //if metrics.Enabled { + // streamInterceptors = append(streamInterceptors, grpc_prometheus.StreamServerInterceptor) + // unaryInterceptors = append(unaryInterceptors, grpc_prometheus.UnaryServerInterceptor) + //} + + var grpcServer *grpc.Server + //cpus := uint32(runtime.GOMAXPROCS(-1)) + opts := []grpc.ServerOption{ + //grpc.NumStreamWorkers(cpus), // reduce amount of goroutines + grpc.WriteBufferSize(1024), // reduce buffers to save mem + grpc.ReadBufferSize(1024), + grpc.MaxConcurrentStreams(kv.ReadersLimit - 128), // to force clients reduce concurrency level + // Don't drop the connection, settings accordign to this comment on GitHub + // https://github.com/grpc/grpc-go/issues/3171#issuecomment-552796779 + grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ + MinTime: 10 * time.Second, + PermitWithoutStream: true, + }), + grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(streamInterceptors...)), + grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(unaryInterceptors...)), + } + if creds == nil { + // no specific opts + } else { + opts = append(opts, grpc.Creds(*creds)) + } + grpcServer = grpc.NewServer(opts...) + if txPoolServer != nil { + txpool_proto.RegisterTxpoolServer(grpcServer, txPoolServer) + } + if miningServer != nil { + txpool_proto.RegisterMiningServer(grpcServer, miningServer) + } + + //if metrics.Enabled { + // grpc_prometheus.Register(grpcServer) + //} + + go func() { + if err := grpcServer.Serve(lis); err != nil { + log.Error("private RPC server fail", "err", err) + } + }() + log.Info("Started gRPC server", "on", addr) + return grpcServer, nil +} diff --git a/txpool/mocks_test.go b/txpool/mocks_test.go index d655c48a6..b72de3f56 100644 --- a/txpool/mocks_test.go +++ b/txpool/mocks_test.go @@ -22,6 +22,9 @@ var _ Pool = &PoolMock{} // AddNewGoodPeerFunc: func(peerID PeerID) { // panic("mock out the AddNewGoodPeer method") // }, +// AddRemoteTxsFunc: func(ctx context.Context, newTxs TxSlots) { +// panic("mock out the AddRemoteTxs method") +// }, // GetRlpFunc: func(tx kv.Tx, hash []byte) ([]byte, error) { // panic("mock out the GetRlp method") // }, @@ -31,9 +34,6 @@ var _ Pool = &PoolMock{} // OnNewBlockFunc: func(stateChanges map[string]senderInfo, unwindTxs TxSlots, minedTxs TxSlots, baseFee uint64, blockHeight uint64, blockHash [32]byte) error { // panic("mock out the OnNewBlock method") // }, -// OnNewTxsFunc: func(ctx context.Context, newTxs TxSlots) { -// panic("mock out the OnNewRemoteTxs method") -// }, // StartedFunc: func() bool { // panic("mock out the Started method") // }, @@ -47,6 +47,9 @@ type PoolMock struct { // AddNewGoodPeerFunc mocks the AddNewGoodPeer method. AddNewGoodPeerFunc func(peerID PeerID) + // AddRemoteTxsFunc mocks the AddRemoteTxs method. + AddRemoteTxsFunc func(ctx context.Context, newTxs TxSlots) + // GetRlpFunc mocks the GetRlp method. GetRlpFunc func(tx kv.Tx, hash []byte) ([]byte, error) @@ -56,9 +59,6 @@ type PoolMock struct { // OnNewBlockFunc mocks the OnNewBlock method. OnNewBlockFunc func(stateChanges map[string]senderInfo, unwindTxs TxSlots, minedTxs TxSlots, baseFee uint64, blockHeight uint64, blockHash [32]byte) error - // OnNewTxsFunc mocks the OnNewRemoteTxs method. - OnNewTxsFunc func(ctx context.Context, newTxs TxSlots) - // StartedFunc mocks the Started method. StartedFunc func() bool @@ -69,6 +69,13 @@ type PoolMock struct { // PeerID is the peerID argument value. PeerID PeerID } + // AddRemoteTxs holds details about calls to the AddRemoteTxs method. + AddRemoteTxs []struct { + // Ctx is the ctx argument value. + Ctx context.Context + // NewTxs is the newTxs argument value. + NewTxs TxSlots + } // GetRlp holds details about calls to the GetRlp method. GetRlp []struct { // Tx is the tx argument value. @@ -98,22 +105,15 @@ type PoolMock struct { // BlockHash is the blockHash argument value. BlockHash [32]byte } - // OnNewRemoteTxs holds details about calls to the OnNewRemoteTxs method. - OnNewTxs []struct { - // Ctx is the ctx argument value. - Ctx context.Context - // NewTxs is the newTxs argument value. - NewTxs TxSlots - } // Started holds details about calls to the Started method. Started []struct { } } lockAddNewGoodPeer sync.RWMutex + lockAddRemoteTxs sync.RWMutex lockGetRlp sync.RWMutex lockIdHashKnown sync.RWMutex lockOnNewBlock sync.RWMutex - lockOnNewTxs sync.RWMutex lockStarted sync.RWMutex } @@ -148,6 +148,41 @@ func (mock *PoolMock) AddNewGoodPeerCalls() []struct { return calls } +// AddRemoteTxs calls AddRemoteTxsFunc. +func (mock *PoolMock) AddRemoteTxs(ctx context.Context, newTxs TxSlots) { + callInfo := struct { + Ctx context.Context + NewTxs TxSlots + }{ + Ctx: ctx, + NewTxs: newTxs, + } + mock.lockAddRemoteTxs.Lock() + mock.calls.AddRemoteTxs = append(mock.calls.AddRemoteTxs, callInfo) + mock.lockAddRemoteTxs.Unlock() + if mock.AddRemoteTxsFunc == nil { + return + } + mock.AddRemoteTxsFunc(ctx, newTxs) +} + +// AddRemoteTxsCalls gets all the calls that were made to AddRemoteTxs. +// Check the length with: +// len(mockedPool.AddRemoteTxsCalls()) +func (mock *PoolMock) AddRemoteTxsCalls() []struct { + Ctx context.Context + NewTxs TxSlots +} { + var calls []struct { + Ctx context.Context + NewTxs TxSlots + } + mock.lockAddRemoteTxs.RLock() + calls = mock.calls.AddRemoteTxs + mock.lockAddRemoteTxs.RUnlock() + return calls +} + // GetRlp calls GetRlpFunc. func (mock *PoolMock) GetRlp(tx kv.Tx, hash []byte) ([]byte, error) { callInfo := struct { @@ -280,41 +315,6 @@ func (mock *PoolMock) OnNewBlockCalls() []struct { return calls } -// OnNewRemoteTxs calls OnNewTxsFunc. -func (mock *PoolMock) OnNewRemoteTxs(ctx context.Context, newTxs TxSlots) { - callInfo := struct { - Ctx context.Context - NewTxs TxSlots - }{ - Ctx: ctx, - NewTxs: newTxs, - } - mock.lockOnNewTxs.Lock() - mock.calls.OnNewTxs = append(mock.calls.OnNewTxs, callInfo) - mock.lockOnNewTxs.Unlock() - if mock.OnNewTxsFunc == nil { - return - } - mock.OnNewTxsFunc(ctx, newTxs) -} - -// OnNewTxsCalls gets all the calls that were made to OnNewRemoteTxs. -// Check the length with: -// len(mockedPool.OnNewTxsCalls()) -func (mock *PoolMock) OnNewTxsCalls() []struct { - Ctx context.Context - NewTxs TxSlots -} { - var calls []struct { - Ctx context.Context - NewTxs TxSlots - } - mock.lockOnNewTxs.RLock() - calls = mock.calls.OnNewTxs - mock.lockOnNewTxs.RUnlock() - return calls -} - // Started calls StartedFunc. func (mock *PoolMock) Started() bool { callInfo := struct { diff --git a/txpool/pool.go b/txpool/pool.go index a74f02154..d60da753d 100644 --- a/txpool/pool.go +++ b/txpool/pool.go @@ -33,6 +33,7 @@ import ( "github.com/google/btree" "github.com/hashicorp/golang-lru/simplelru" "github.com/holiman/uint256" + proto_txpool "github.com/ledgerwatch/erigon-lib/gointerfaces/txpool" "github.com/ledgerwatch/erigon-lib/kv" "github.com/ledgerwatch/erigon-lib/kv/mdbx" "github.com/ledgerwatch/log/v3" @@ -72,7 +73,7 @@ type Pool interface { IdHashKnown(tx kv.Tx, hash []byte) (bool, error) Started() bool GetRlp(tx kv.Tx, hash []byte) ([]byte, error) - OnNewRemoteTxs(ctx context.Context, newTxs TxSlots) + AddRemoteTxs(ctx context.Context, newTxs TxSlots) OnNewBlock(stateChanges map[string]senderInfo, unwindTxs, minedTxs TxSlots, baseFee, blockHeight uint64, blockHash [32]byte) error AddNewGoodPeer(peerID PeerID) @@ -96,6 +97,19 @@ const ( IsLocal = 0b00001 ) +type DiscardReason uint8 + +const ( + //TODO: all below codes are not fixed yet. Need add them to discardLocked func. Need save discard reasons to LRU or DB. + Success DiscardReason = 1 + AlreadyKnown DiscardReason = 2 + UnderPriced DiscardReason = 3 + FeeTooLow DiscardReason = 4 + OversizedData DiscardReason = 5 + InvalidSender DiscardReason = 6 + NegativeValue DiscardReason = 7 +) + // metaTx holds transaction and some metadata type metaTx struct { subPool SubPoolMarker @@ -489,11 +503,13 @@ type TxPool struct { // track isLocal flag of already mined transactions. used at unwind. localsHistory *simplelru.LRU db kv.RwDB + coreDB kv.RoDB // fields for transaction propagation recentlyConnectedPeers *recentlyConnectedPeers newTxs chan Hashes deletedTxs []*metaTx + discardReasons []DiscardReason senders *sendersBatch txNonce2Tx *ByNonce // senderID => (sorted map of tx nonce => *metaTx) @@ -508,7 +524,7 @@ type TxPool struct { cfg Config } -func New(newTxs chan Hashes, db kv.RwDB, cfg Config) (*TxPool, error) { +func New(newTxs chan Hashes, db kv.RwDB, coreDB kv.RoDB, cfg Config) (*TxPool, error) { localsHistory, err := simplelru.NewLRU(1024, nil) if err != nil { return nil, err @@ -525,6 +541,7 @@ func New(newTxs chan Hashes, db kv.RwDB, cfg Config) (*TxPool, error) { newTxs: newTxs, senders: newSendersCache(), db: db, + coreDB: coreDB, cfg: cfg, senderID: 1, unprocessedRemoteTxs: &TxSlots{}, @@ -608,6 +625,9 @@ func (p *TxPool) AppendRemoteHashes(buf []byte) []byte { } buf = append(buf, hash...) } + for hash := range p.unprocessedRemoteByHash { + buf = append(buf, hash...) + } return buf } func (p *TxPool) AppendAllHashes(buf []byte) []byte { @@ -654,12 +674,17 @@ func (p *TxPool) Best(n uint16, txs *TxSlots, tx kv.Tx) error { txs.txs[i] = best[i].Tx txs.isLocal[i] = best[i].subPool&IsLocal > 0 + found := false for addr, senderID := range p.senders.senderIDs { // TODO: do we need inverted index here? if best[i].Tx.senderID == senderID { copy(txs.senders.At(i), addr) + found = true break } } + if found { + continue + } binary.BigEndian.PutUint64(encID, best[i].Tx.senderID) v, err := tx.GetOne(kv.PoolSenderIDToAdress, encID) @@ -673,7 +698,63 @@ func (p *TxPool) Best(n uint16, txs *TxSlots, tx kv.Tx) error { } return nil } -func (p *TxPool) OnNewRemoteTxs(_ context.Context, newTxs TxSlots) { + +func (p *TxPool) CountContent() (int, int, int) { + p.lock.RLock() + defer p.lock.RUnlock() + return p.pending.Len(), p.baseFee.Len(), p.queued.Len() +} + +//Deprecated need switch to streaming-like +func (p *TxPool) DeprecatedForEach(_ context.Context, f func(rlp, sender []byte, t SubPoolType), tx kv.Tx) error { + p.lock.RLock() + defer p.lock.RUnlock() + + encID := make([]byte, 8) + p.txNonce2Tx.tree.Ascend(func(i btree.Item) bool { + mt := i.(*sortByNonce).metaTx + slot := mt.Tx + slotRlp := slot.rlp + if slot.rlp == nil { + v, err := tx.GetOne(kv.PoolTransaction, slot.idHash[:]) + if err != nil { + log.Error("get tx from db", "err", err) + return false + } + if v == nil { + log.Error("tx not found in db") + return false + } + slotRlp = v[8:] + } + + var sender []byte + found := false + for addr, senderID := range p.senders.senderIDs { // TODO: do we need inverted index here? + if slot.senderID == senderID { + sender = []byte(addr) + found = true + break + } + } + if !found { + binary.BigEndian.PutUint64(encID, slot.senderID) + v, err := tx.GetOne(kv.PoolSenderIDToAdress, encID) + if err != nil { + log.Error("get sender from db", "err", err) + return false + } + if v == nil { + log.Error("sender not found in db") + return false + } + } + f(slotRlp, sender, mt.currentSubPool) + return true + }) + return nil +} +func (p *TxPool) AddRemoteTxs(_ context.Context, newTxs TxSlots) { p.lock.Lock() defer p.lock.Unlock() for i := range newTxs.txs { @@ -684,8 +765,59 @@ func (p *TxPool) OnNewRemoteTxs(_ context.Context, newTxs TxSlots) { p.unprocessedRemoteTxs.Append(newTxs.txs[i], newTxs.senders.At(i), newTxs.isLocal[i]) } } +func (p *TxPool) AddLocals(ctx context.Context, newTxs TxSlots, tx kv.Tx) ([]DiscardReason, error) { + p.lock.Lock() + defer p.lock.Unlock() + discardReasonsIndex := len(p.discardReasons) -func (p *TxPool) processRemoteTxs(ctx context.Context, coreDB kv.RoDB) error { + for i := range newTxs.isLocal { + newTxs.isLocal[i] = true + } + + cacheMisses, err := p.senders.onNewTxs(tx, newTxs) + if err != nil { + return nil, err + } + if len(cacheMisses) > 0 { + if err := p.coreDB.View(ctx, func(tx kv.Tx) error { return p.senders.loadFromCore(tx, cacheMisses) }); err != nil { + return nil, err + } + } + if err := newTxs.Valid(); err != nil { + return nil, err + } + + protocolBaseFee, currentBaseFee := p.protocolBaseFee.Load(), p.currentBaseFee.Load() + if protocolBaseFee == 0 || currentBaseFee == 0 { + return nil, fmt.Errorf("non-zero base fee: %d,%d", protocolBaseFee, currentBaseFee) + } + if err := onNewTxs(tx, p.senders, newTxs, protocolBaseFee, currentBaseFee, p.pending, p.baseFee, p.queued, p.txNonce2Tx, p.byHash, p.discardLocked); err != nil { + return nil, err + } + + // notify about all non-dropped txs + notifyNewTxs := make(Hashes, 0, 32*len(newTxs.txs)) + for i := range newTxs.txs { + _, ok := p.byHash[string(newTxs.txs[i].idHash[:])] + if !ok { + continue + } + notifyNewTxs = append(notifyNewTxs, newTxs.txs[i].idHash[:]...) + } + if len(notifyNewTxs) > 0 { + select { + case p.newTxs <- notifyNewTxs: + default: + } + } + return p.copyDiscardReasons(discardReasonsIndex), nil +} +func (p *TxPool) copyDiscardReasons(from int) []DiscardReason { + cpy := make([]DiscardReason, len(p.discardReasons)-from) + copy(cpy, p.discardReasons[from:]) + return cpy +} +func (p *TxPool) processRemoteTxs(ctx context.Context) error { p.lock.RLock() l := len(p.unprocessedRemoteTxs.txs) p.lock.RUnlock() @@ -709,7 +841,7 @@ func (p *TxPool) processRemoteTxs(ctx context.Context, coreDB kv.RoDB) error { return err } if len(cacheMisses) > 0 { - if err := coreDB.View(ctx, func(tx kv.Tx) error { return p.senders.loadFromCore(tx, cacheMisses) }); err != nil { + if err := p.coreDB.View(ctx, func(tx kv.Tx) error { return p.senders.loadFromCore(tx, cacheMisses) }); err != nil { return err } } @@ -768,7 +900,7 @@ func onNewTxs(tx kv.Tx, senders *sendersBatch, newTxs TxSlots, protocolBaseFee, queued.EnforceInvariants() promote(pending, baseFee, queued, discard) - pending.EnforceInvariants() + pending.EnforceInvariants() //TODO: find way to enforce invariants once. promote - now expect invariants - but can it work without them? return nil } @@ -1353,14 +1485,14 @@ func (p *WorstQueue) Pop() interface{} { return item } -// BroadcastLoop - does: +// MainLoop - does: // send pending byHash to p2p: // - new byHash // - all pooled byHash to recently connected peers // - all local pooled byHash to random peers periodically // promote/demote transactions // reorgs -func BroadcastLoop(ctx context.Context, db kv.RwDB, coreDB kv.RoDB, p *TxPool, newTxs chan Hashes, send *Send) { +func MainLoop(ctx context.Context, db kv.RwDB, coreDB kv.RoDB, p *TxPool, newTxs chan Hashes, send *Send, newTxSlotsStreams *NewSlotsStreams) { //db.Update(ctx, func(tx kv.RwTx) error { return tx.ClearBucket(kv.PooledSender) }) if err := db.Update(ctx, func(tx kv.RwTx) error { return coreDB.View(ctx, func(coreTx kv.Tx) error { @@ -1401,7 +1533,7 @@ func BroadcastLoop(ctx context.Context, db kv.RwDB, coreDB kv.RoDB, p *TxPool, n log.Error("log stats", "err", err) } case <-processRemoteTxsEvery.C: - if err := p.processRemoteTxs(ctx, coreDB); err != nil { + if err := p.processRemoteTxs(ctx); err != nil { log.Error("process batch remote txs", "err", err) } case <-commitEvery.C: @@ -1429,6 +1561,21 @@ func BroadcastLoop(ctx context.Context, db kv.RwDB, coreDB kv.RoDB, p *TxPool, n send.BroadcastLocalPooledTxs(localTxHashes) send.BroadcastRemotePooledTxs(remoteTxHashes) + + if err := db.View(ctx, func(tx kv.Tx) error { + slotsRlp := make([][]byte, 0, h.Len()) + for i := 0; i < h.Len(); i++ { + slotRlp, err := p.GetRlp(tx, h.At(i)) + if err != nil { + return err + } + slotsRlp = append(slotsRlp, slotRlp) + } + newTxSlotsStreams.Broadcast(&proto_txpool.OnAddReply{RplTxs: slotsRlp}) + return nil + }); err != nil { + log.Error("send new slots by grpc", "err", err) + } case <-syncToNewPeersEvery.C: // new peer newPeers := p.recentlyConnectedPeers.GetAndClean() if len(newPeers) == 0 { @@ -1568,6 +1715,7 @@ func (p *TxPool) flushLocked(tx kv.RwTx) (evicted uint64, err error) { // DB will stay consitant but some in-memory structures may be alread cleaned, and retry will not work // failed write transaction must not create side-effects p.deletedTxs = p.deletedTxs[:0] + p.discardReasons = p.discardReasons[:0] return evicted, nil } diff --git a/txpool/pool_fuzz_test.go b/txpool/pool_fuzz_test.go index fcf168e7c..358f8a492 100644 --- a/txpool/pool_fuzz_test.go +++ b/txpool/pool_fuzz_test.go @@ -23,7 +23,7 @@ import ( // golang.org/s/draft-fuzzing-design //gotip doc testing //gotip doc testing.F -//gotip doc testing.F.OnNewRemoteTxs +//gotip doc testing.F.AddRemoteTxs //gotip doc testing.F.Fuzz // gotip test -trimpath -v -fuzz=Fuzz -fuzztime=10s ./txpool @@ -532,8 +532,8 @@ func FuzzOnNewBlocks12(f *testing.F) { checkNotify(TxSlots{}, txs3, "fork2 mined") // add some remote txs from p2p - pool.OnNewRemoteTxs(context.Background(), p2pReceived) - err = pool.processRemoteTxs(context.Background(), nil) + pool.AddRemoteTxs(context.Background(), p2pReceived) + err = pool.processRemoteTxs(context.Background()) assert.NoError(err) check(p2pReceived, TxSlots{}, "p2pmsg1") checkNotify(p2pReceived, TxSlots{}, "p2pmsg1")