From bf60c6ce065c9191cae2facb8ddce8d1d06b3828 Mon Sep 17 00:00:00 2001 From: Manav Darji Date: Tue, 19 Jul 2022 16:22:41 +0530 Subject: [PATCH] consensus/bor: add mock heimdall rest-server, tests for timeout and cancellation (#458) * add: mock heimdall rest-server, tests for timeout and cancellation * fix: use mux for route handling --- consensus/bor/bor_test.go | 49 +--- consensus/bor/heimdall/client_test.go | 313 ++++++++++++++++++++++++++ 2 files changed, 314 insertions(+), 48 deletions(-) diff --git a/consensus/bor/bor_test.go b/consensus/bor/bor_test.go index 1e447ba872..0e756d3577 100644 --- a/consensus/bor/bor_test.go +++ b/consensus/bor/bor_test.go @@ -1,15 +1,13 @@ package bor import ( - "context" "math/big" "testing" "github.com/stretchr/testify/require" "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/common/hexutil" - "github.com/ethereum/go-ethereum/consensus/bor/heimdall" //nolint:typecheck + "github.com/ethereum/go-ethereum/common/hexutil" //nolint:typecheck "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/state" @@ -142,48 +140,3 @@ func TestEncodeSigHeaderJaipur(t *testing.T) { hash = SealHash(h, ¶ms.BorConfig{JaipurBlock: 10}) require.Equal(t, hash, hashWithoutBaseFee) } - -// TestCheckpoint can be used for to fetch checkpoint -// count and checkpoint for debugging purpose. -// Also, this is kept only for local use. -func TestCheckpoint(t *testing.T) { - t.Skip() - t.Parallel() - - ctx := context.Background() - - // TODO: For testing, add heimdall url here - h := heimdall.NewHeimdallClient("http://localhost:1317") - - count, err := h.FetchCheckpointCount(ctx) - if err != nil { - t.Error(err) - } - - t.Log("Count:", count) - - checkpoint1, err := h.FetchCheckpoint(ctx, count) - if err != nil { - t.Error(err) - } - - t.Log("Checkpoint1:", checkpoint1) - - checkpoint2, err := h.FetchCheckpoint(ctx, 10000) - if err != nil { - t.Error(err) - } - - t.Log("Checkpoint2:", checkpoint2) - - checkpoint3, err := h.FetchCheckpoint(ctx, -1) - if err != nil { - t.Error(err) - } - - t.Log("Checkpoint3:", checkpoint3) - - if checkpoint3.RootHash != checkpoint1.RootHash { - t.Fatal("Invalid root hash") - } -} diff --git a/consensus/bor/heimdall/client_test.go b/consensus/bor/heimdall/client_test.go index c423a6f659..f9ed0028b1 100644 --- a/consensus/bor/heimdall/client_test.go +++ b/consensus/bor/heimdall/client_test.go @@ -1,9 +1,322 @@ package heimdall import ( + "context" + "encoding/json" + "errors" + "fmt" + "math/big" + "net" + "net/http" + "sync" + "sync/atomic" "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/consensus/bor/heimdall/checkpoint" + + "github.com/stretchr/testify/require" ) +var maxPortCheck int32 = 100 + +// HttpHandlerFake defines the handler functions required to serve +// requests to the mock heimdal server for specific functions. Add more handlers +// according to requirements. +type HttpHandlerFake struct { + handleFetchCheckpoint http.HandlerFunc +} + +func (h *HttpHandlerFake) GetCheckpointHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + h.handleFetchCheckpoint.ServeHTTP(w, r) + } +} + +func CreateMockHeimdallServer(wg *sync.WaitGroup, port int32, handler *HttpHandlerFake) (*http.Server, error) { + // Create a new server mux + mux := http.NewServeMux() + + // Create a route for fetching latest checkpoint + mux.HandleFunc("/checkpoints/latest", func(w http.ResponseWriter, r *http.Request) { + handler.GetCheckpointHandler()(w, r) + }) + + // Add other routes as per requirement + + // Create the server with given port and mux + srv := &http.Server{ + Addr: fmt.Sprintf("localhost:%d", port), + Handler: mux, + } + + go func() { + defer wg.Done() + + // always returns error. ErrServerClosed on graceful close + if err := srv.ListenAndServe(); err != http.ErrServerClosed { + fmt.Printf("error in server.ListenAndServe(): %v", err) + } + }() + + return srv, nil +} + +// TestFetchCheckpointFromMockHeimdall tests the heimdall client side logic +// to fetch checkpoints (latest for the scope of test) from a mock heimdall server. +// It can be used for debugging purpose (like response fields, marshalling/unmarshalling, etc). +func TestFetchCheckpointFromMockHeimdall(t *testing.T) { + t.Parallel() + + // Create a wait group for sending across the mock server + wg := &sync.WaitGroup{} + wg.Add(1) + + // Initialize the fake handler and add a fake checkpoint handler function + handler := &HttpHandlerFake{} + handler.handleFetchCheckpoint = func(w http.ResponseWriter, _ *http.Request) { + err := json.NewEncoder(w).Encode(checkpoint.CheckpointResponse{ + Height: "0", + Result: checkpoint.Checkpoint{ + Proposer: common.Address{}, + StartBlock: big.NewInt(0), + EndBlock: big.NewInt(512), + RootHash: common.Hash{}, + BorChainID: "15001", + Timestamp: 0, + }, + }) + + if err != nil { + w.WriteHeader(500) // Return 500 Internal Server Error. + } + } + + // Fetch available port starting from 50000 + port, err := findAvailablePort(50000, 0) + require.NoError(t, err, "expect no error in finding available port") + + // Create mock heimdall server and pass handler instance for setting up the routes + srv, err := CreateMockHeimdallServer(wg, port, handler) + require.NoError(t, err, "expect no error in starting mock heimdall server") + + // Create a new heimdall client and use same port for connection + client := NewHeimdallClient(fmt.Sprintf("http://localhost:%d", port)) + _, err = client.FetchCheckpoint(context.Background(), -1) + require.NoError(t, err, "expect no error in fetching checkpoint") + + // Shutdown the server + err = srv.Shutdown(context.TODO()) + require.NoError(t, err, "expect no error in shutting down mock heimdall server") + + // Wait for `wg.Done()` to be called in the mock server's routine. + wg.Wait() +} + +// TestFetchShutdown tests the heimdall client side logic for context timeout and +// interrupt handling while fetching checkpoints (latest for the scope of test) +// from a mock heimdall server. +func TestFetchShutdown(t *testing.T) { + t.Parallel() + + // Create a wait group for sending across the mock server + wg := &sync.WaitGroup{} + wg.Add(1) + + // Initialize the fake handler and add a fake checkpoint handler function + handler := &HttpHandlerFake{} + + // Case1 - Testing context timeout: Create delay in serving requests for simulating timeout. Add delay slightly + // greater than `retryDelay`. This should cause the request to timeout and trigger shutdown + // due to `ctx.Done()`. Expect context timeout error. + handler.handleFetchCheckpoint = func(w http.ResponseWriter, _ *http.Request) { + time.Sleep(6 * time.Second) + + err := json.NewEncoder(w).Encode(checkpoint.CheckpointResponse{ + Height: "0", + Result: checkpoint.Checkpoint{ + Proposer: common.Address{}, + StartBlock: big.NewInt(0), + EndBlock: big.NewInt(512), + RootHash: common.Hash{}, + BorChainID: "15001", + Timestamp: 0, + }, + }) + + if err != nil { + w.WriteHeader(500) // Return 500 Internal Server Error. + } + } + + // Fetch available port starting from 50000 + port, err := findAvailablePort(50000, 0) + require.NoError(t, err, "expect no error in finding available port") + + // Create mock heimdall server and pass handler instance for setting up the routes + srv, err := CreateMockHeimdallServer(wg, port, handler) + require.NoError(t, err, "expect no error in starting mock heimdall server") + + // Create a new heimdall client and use same port for connection + client := NewHeimdallClient(fmt.Sprintf("http://localhost:%d", port)) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + + // Expect this to fail due to timeout + _, err = client.FetchCheckpoint(ctx, -1) + require.Equal(t, "context deadline exceeded", err.Error(), "expect the function error to be a context deadline exeeded error") + require.Equal(t, "context deadline exceeded", ctx.Err().Error(), "expect the ctx error to be a context deadline exeeded error") + + cancel() + + // Case2 - Testing context cancellation. Pass a context with timeout to the request and + // cancel it before timeout. This should cause the request to timeout and trigger shutdown + // due to `ctx.Done()`. Expect context cancellation error. + handler.handleFetchCheckpoint = func(w http.ResponseWriter, _ *http.Request) { + time.Sleep(10 * time.Millisecond) + w.WriteHeader(500) // Return 500 Internal Server Error. + } + + ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second) // Use some high value for timeout + + // Cancel the context after a delay until we make request + go func(cancel context.CancelFunc) { + time.Sleep(10 * time.Millisecond) + cancel() + }(cancel) + + // Expect this to fail due to cancellation + _, err = client.FetchCheckpoint(ctx, -1) + require.Equal(t, "context canceled", err.Error(), "expect the function error to be a context cancelled error") + require.Equal(t, "context canceled", ctx.Err().Error(), "expect the ctx error to be a context cancelled error") + + // Case3 - Testing interrupt: Closing the `closeCh` in heimdall client simulating interrupt. This + // should cause the request to fail and throw an error due to `<-closeCh` in fetchWithRetry. + // Expect shutdown detected error. + handler.handleFetchCheckpoint = func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(500) // Return 500 Internal Server Error. + } + + // Close the channel after a delay until we make request + go func() { + time.Sleep(1 * time.Second) + close(client.closeCh) + }() + + // Expect this to fail due to shutdown + _, err = client.FetchCheckpoint(context.Background(), -1) + require.Equal(t, ErrShutdownDetected.Error(), err.Error(), "expect the function error to be a shutdown detected error") + + // Shutdown the server + err = srv.Shutdown(context.TODO()) + require.NoError(t, err, "expect no error in shutting down mock heimdall server") + + // Wait for `wg.Done()` to be called in the mock server's routine. + wg.Wait() +} + +// findAvailablePort returns the next available port starting from `from` +func findAvailablePort(from int32, count int32) (int32, error) { + if count == maxPortCheck { + return 0, fmt.Errorf("no available port found") + } + + port := atomic.AddInt32(&from, 1) + addr := fmt.Sprintf("localhost:%d", port) + + count++ + + lis, err := net.Listen("tcp", addr) + if err == nil { + lis.Close() + return port, nil + } else { + return findAvailablePort(from, count) + } +} + +// TestContext includes bunch of simple tests to verify the working of timeout +// based context and cancellation. +func TestContext(t *testing.T) { + t.Parallel() + + ctx, cancel1 := context.WithTimeout(context.Background(), 1*time.Second) + + // Case1: Done is not yet closed, so Err returns nil. + require.NoError(t, ctx.Err(), "expect nil error") + + wg := &sync.WaitGroup{} + + // Case2: Check if timeout is being handled + wg.Add(1) + + go func(ctx context.Context, wg *sync.WaitGroup) { + defer wg.Done() + select { + case <-ctx.Done(): + // Expect context deadline exceeded error + require.Equal(t, "context deadline exceeded", ctx.Err().Error(), "expect the ctx error to be a context deadline exceeded error") + case <-time.After(2 * time.Second): + // Case for safely exiting the tests + return + } + }(ctx, wg) + + // Case3: Check normal case + ctx, cancel2 := context.WithTimeout(context.Background(), 3*time.Second) + + wg.Add(1) + + errCh := make(chan error, 1) + + go func(ctx context.Context, wg *sync.WaitGroup) { + defer wg.Done() + select { + case <-ctx.Done(): + // Expect this to never occur, throw explicit error + errCh <- errors.New("unexpectecd call to `ctx.Done()`") + case <-time.After(2 * time.Second): + // Case for safely exiting the tests + errCh <- nil + return + } + }(ctx, wg) + + if err := <-errCh; err != nil { + t.Fatalf("err: %v", err) + } + + // Case4: Check if cancellation is being handled + ctx, cancel3 := context.WithTimeout(context.Background(), 1*time.Second) + + wg.Add(1) + + go func(cancel context.CancelFunc) { + time.Sleep(500 * time.Millisecond) + cancel() + }(cancel3) + + go func(ctx context.Context, wg *sync.WaitGroup) { + defer wg.Done() + select { + case <-ctx.Done(): + // Expect context canceled error + require.Equal(t, "context canceled", ctx.Err().Error(), "expect the ctx error to be a context canceled error") + case <-time.After(2 * time.Second): + // Case for safely exiting the tests + return + } + }(ctx, wg) + + // Wait for all tests to pass + wg.Wait() + + // Cancel all remaining contexts + cancel1() + cancel2() +} + func TestSpanURL(t *testing.T) { t.Parallel()