Merge pull request #358 from spacemeshos/register-challenges-in-batches

Register challenges in batches

.github/workflows/ci.yml

@@ -57,7 +57,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: 
+        os:
           - ubuntu-latest
           - [self-hosted, linux, arm64]
           - macos-latest
@@ -79,6 +79,7 @@ jobs:
         env:
           GOTESTSUM_FORMAT: standard-verbose
           GOTESTSUM_JUNITFILE: unit-tests.xml
+          TEST_FLAGS: -short
         run: make test
       - name: Publish Test Report
         uses: mikepenz/action-junit-report@v3

Makefile

@@ -5,6 +5,9 @@ PROJECT := poet
 BRANCH := $(shell git rev-parse --abbrev-ref HEAD)
 VERSION ?= $(shell git describe --tags)
 
+# Flags appended to `go test` command in `make test`
+TEST_FLAGS ?=
+
 GOLANGCI_LINT_VERSION := v1.52.0
 STATICCHECK_VERSION := v0.4.3
 GOTESTSUM_VERSION := v1.10.0
@@ -91,7 +94,7 @@ all: build
 .PHONY: all
 
 test:
-	$(GOTESTSUM) -- -race -timeout 5m -p 1 ./...
+	$(GOTESTSUM) -- -race -timeout 5m -p 1 $(TEST_FLAGS) ./...
 .PHONY: test
 
 install: install-buf install-protoc $(GOVULNCHECK) $(GOLINES)

config/config.go

@@ -31,6 +31,8 @@ const (
 	defaultTreeFileBufferSize       = 4096
 	defaultEstimatedLeavesPerSecond = 78000
 	defaultMaxRoundMembers          = 1 << 32
+	defaultSubmitFlushInterval      = 100 * time.Millisecond
+	defaultMaxSubmitBatchSize       = 1000
 )
 
 var (
@@ -89,6 +91,8 @@ func DefaultConfig() *Config {
 			TreeFileBufferSize:       defaultTreeFileBufferSize,
 			EstimatedLeavesPerSecond: defaultEstimatedLeavesPerSecond,
 			MaxRoundMembers:          defaultMaxRoundMembers,
+			MaxSubmitBatchSize:       defaultMaxSubmitBatchSize,
+			SubmitFlushInterval:      defaultSubmitFlushInterval,
 		},
 	}
 }

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/google/uuid v1.3.1
 	github.com/grpc-ecosystem/go-grpc-middleware v1.4.0
-	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
+	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.0-rc.0
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.17.0
 	github.com/jessevdk/go-flags v1.5.0
 	github.com/natefinch/atomic v1.0.1
@@ -38,6 +38,7 @@ require (
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
+	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.0-rc.3 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/minio/sha256-simd v1.0.1 // indirect

server/server.go

@@ -12,8 +12,9 @@ import (
 
 	"github.com/google/uuid"
 	grpcmw "github.com/grpc-ecosystem/go-grpc-middleware"
-	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
 	proxy "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
+	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
@@ -100,13 +101,20 @@ func (s *Server) Start(ctx context.Context) error {
 
 	logger := logging.FromContext(ctx)
 
+	// grpc metrics
+	metrics := grpc_prometheus.NewServerMetrics(
+		grpc_prometheus.WithServerHandlingTimeHistogram(
+			grpc_prometheus.WithHistogramBuckets(prometheus.ExponentialBuckets(0.001, 2, 16)),
+		),
+	)
+
 	// Initialize and register the implementation of gRPC interface
 	var grpcServer *grpc.Server
 	var proxyRegstr []func(context.Context, *proxy.ServeMux, string, []grpc.DialOption) error
 	options := []grpc.ServerOption{
 		grpc.UnaryInterceptor(grpcmw.ChainUnaryServer(
 			loggerInterceptor(logger),
-			grpc_prometheus.UnaryServerInterceptor,
+			metrics.UnaryServerInterceptor(),
 		)),
 		// XXX: this is done to prevent routers from cleaning up our connections (e.g aws load balances..)
 		// TODO: these parameters work for now but we might need to revisit or add them as configuration
@@ -145,7 +153,8 @@ func (s *Server) Start(ctx context.Context) error {
 	proxyRegstr = append(proxyRegstr, api.RegisterPoetServiceHandlerFromEndpoint)
 
 	reflection.Register(grpcServer)
-	grpc_prometheus.Register(grpcServer)
+	metrics.InitializeMetrics(grpcServer)
+	prometheus.Register(metrics)
 
 	// Start the gRPC server listening for HTTP/2 connections.
 	serverGroup.Go(func() error {

server/server_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zapcore"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -20,6 +21,7 @@ import (
 
 	"github.com/spacemeshos/poet/config"
 	"github.com/spacemeshos/poet/hash"
+	"github.com/spacemeshos/poet/logging"
 	"github.com/spacemeshos/poet/prover"
 	api "github.com/spacemeshos/poet/release/proto/go/rpc/api/v1"
 	"github.com/spacemeshos/poet/server"
@@ -400,3 +402,73 @@ func TestGettingInitialPowParams(t *testing.T) {
 	cancel()
 	req.NoError(eg.Wait())
 }
+
+func TestLoadSubmits(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping load test in short mode")
+	}
+	req := require.New(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ctx = logging.NewContext(ctx, logging.New(zapcore.InfoLevel, "log.log", false))
+
+	cfg := config.DefaultConfig()
+	cfg.PoetDir = t.TempDir()
+	cfg.Service.EpochDuration = time.Minute * 2
+	cfg.Service.PhaseShift = time.Minute
+	cfg.RawRPCListener = randomHost
+	cfg, err := config.SetupConfig(cfg)
+	req.NoError(err)
+
+	srv, err := server.New(context.Background(), *cfg)
+	req.NoError(err)
+
+	concurrentSubmits := 10000
+	// spawn clients
+	clients := make([]api.PoetServiceClient, 0, concurrentSubmits)
+	for i := 0; i < concurrentSubmits; i++ {
+		conn, err := grpc.Dial(
+			srv.GrpcAddr().String(),
+			grpc.WithTransportCredentials(insecure.NewCredentials()))
+		req.NoError(err)
+		t.Cleanup(func() { assert.NoError(t, conn.Close()) })
+		clients = append(clients, api.NewPoetServiceClient(conn))
+	}
+
+	// Submit challenges
+	var egSrv errgroup.Group
+	egSrv.Go(func() error {
+		return srv.Start(ctx)
+	})
+	start := time.Now()
+	var eg errgroup.Group
+
+	for _, client := range clients {
+		client := client
+		eg.Go(func() error {
+			pubKey, privKey, err := ed25519.GenerateKey(rand.Reader)
+			req.NoError(err)
+
+			challenge := make([]byte, 32)
+			_, err = rand.Read(challenge)
+			req.NoError(err)
+
+			powParams, err := client.PowParams(context.Background(), &api.PowParamsRequest{})
+			require.NoError(t, err)
+
+			signature := ed25519.Sign(privKey, challenge)
+			_, err = client.Submit(context.Background(), &api.SubmitRequest{
+				Challenge: challenge,
+				Pubkey:    pubKey,
+				PowParams: powParams.PowParams,
+				Signature: signature,
+			})
+			req.NoError(err)
+			return nil
+		})
+	}
+	eg.Wait()
+	t.Logf("submitting %d challenges took %v", concurrentSubmits, time.Since(start))
+	cancel()
+	egSrv.Wait()
+}

service/pow_verifier.go

@@ -87,8 +87,12 @@ func (v *powVerifiers) SetParams(p PowParams) {
 	v.mutex.Lock()
 	defer v.mutex.Unlock()
 
-	v.previous = v.current
-	v.current.SetParams(p)
+	v.previous = &powVerifier{
+		params: v.current.Params(),
+	}
+	v.current = &powVerifier{
+		params: p,
+	}
 }
 
 func (v *powVerifiers) Params() PowParams {