[CORE-666] - Robust health check

protocol/daemons/liquidation/client/sub_task_runner.go

@@ -0,0 +1,79 @@
+package client
+
+import (
+	"context"
+	"github.com/cosmos/cosmos-sdk/telemetry"
+	"github.com/dydxprotocol/v4-chain/protocol/daemons/flags"
+	"github.com/dydxprotocol/v4-chain/protocol/daemons/liquidation/api"
+	"github.com/dydxprotocol/v4-chain/protocol/lib/metrics"
+	clobtypes "github.com/dydxprotocol/v4-chain/protocol/x/clob/types"
+	satypes "github.com/dydxprotocol/v4-chain/protocol/x/subaccounts/types"
+	"time"
+)
+
+// SubTaskRunner provides an interface that encapsulates the liquidations daemon logic to gather and report
+// potentially liquidatable subaccount ids. This interface is used to mock the daemon logic in tests.
+type SubTaskRunner interface {
+	RunLiquidationDaemonTaskLoop(
+		ctx context.Context,
+		liqFlags flags.LiquidationFlags,
+		subaccountQueryClient satypes.QueryClient,
+		clobQueryClient clobtypes.QueryClient,
+		liquidationServiceClient api.LiquidationServiceClient,
+	) error
+}
+
+type SubTaskRunnerImpl struct{}
+
+// Ensure SubTaskRunnerImpl implements the SubTaskRunner interface.
+var _ SubTaskRunner = (*SubTaskRunnerImpl)(nil)
+
+// RunLiquidationDaemonTaskLoop contains the logic to communicate with various gRPC services
+// to find the liquidatable subaccount ids.
+func (s *SubTaskRunnerImpl) RunLiquidationDaemonTaskLoop(
+	ctx context.Context,
+	liqFlags flags.LiquidationFlags,
+	subaccountQueryClient satypes.QueryClient,
+	clobQueryClient clobtypes.QueryClient,
+	liquidationServiceClient api.LiquidationServiceClient,
+) error {
+	defer telemetry.ModuleMeasureSince(
+		metrics.LiquidationDaemon,
+		time.Now(),
+		metrics.MainTaskLoop,
+		metrics.Latency,
+	)
+
+	// 1. Fetch all subaccounts from query service.
+	subaccounts, err := GetAllSubaccounts(
+		ctx,
+		subaccountQueryClient,
+		liqFlags.SubaccountPageLimit,
+	)
+	if err != nil {
+		return err
+	}
+
+	// 2. Check collateralization statuses of subaccounts with at least one open position.
+	liquidatableSubaccountIds, err := GetLiquidatableSubaccountIds(
+		ctx,
+		clobQueryClient,
+		liqFlags,
+		subaccounts,
+	)
+	if err != nil {
+		return err
+	}
+
+	// 3. Send the list of liquidatable subaccount ids to the daemon server.
+	err = SendLiquidatableSubaccountIds(
+		ctx,
+		liquidationServiceClient,
+		liquidatableSubaccountIds,
+	)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

protocol/daemons/pricefeed/client/client.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	sdkerrors "cosmossdk.io/errors"
 	"errors"
 	"fmt"
 	"sync"
@@ -49,12 +50,46 @@ type Client struct {
 
 	// Ensure stop only executes one time.
 	stopDaemon sync.Once
+
+	// healthLock is used to synchronize access to the healthError field.
+	healthLock sync.Mutex
+
+	// healthError is used to track daemon health. For a healthy daemon, this field is nil. For an unhealthy daemon,
+	// this field is set to the error that caused the daemon to become unhealthy.
+	healthError error
+}
+
+// Ensure Client implements the HealthCheckable interface.
+var _ daemontypes.HealthCheckable = (*Client)(nil)
+
+// HealthCheck returns an error if the daemon is unhealthy.
+// The daemon is considered unhealthy if it is not producing non-empty price updates. We expect a short period of time
+// to pass on daemon startup where the daemon is unhealthy as the price cache warms up.
+// This method is synchronized by isHealthy, which is an atomic.Bool.
+func (c *Client) HealthCheck(_ context.Context) error {
+	c.healthLock.Lock()
+	defer c.healthLock.Unlock()
+
+	if c.healthError == nil {
+		return nil
+	}
+	return sdkerrors.Wrap(c.healthError, "price deamon unhealthy")
+}
+
+// setHealth sets the health of the daemon. Setting the health with a nil err indicates the daemon is healthy. If the
+// daemon is unhealthy, the err parameter will be used to track the reason. This method is synchronized by healthLock.
+func (c *Client) setHealth(err error) {
+	c.healthLock.Lock()
+	defer c.healthLock.Unlock()
+
+	c.healthError = err
 }
 
 func newClient() *Client {
 	client := &Client{
-		tickers: []*time.Ticker{},
-		stops:   []chan bool{},
+		tickers:     []*time.Ticker{},
+		stops:       []chan bool{},
+		healthError: errors.New("daemon uninitialized"),
 	}
 
 	// Set the client's daemonStartup state to indicate that the daemon has not finished starting up.
@@ -66,7 +101,7 @@ func newClient() *Client {
 // for any subtask kicked off by the client. The ticker and channel are tracked in order to properly clean up and send
 // all needed stop signals when the daemon is stopped.
 // Note: this method is not synchronized. It is expected to be called from the client's `StartNewClient` method before
-// `client.CompleteStartup`.
+// the daemonStartup waitgroup signals.
 func (c *Client) newTickerWithStop(intervalMs int) (*time.Ticker, <-chan bool) {
 	ticker := time.NewTicker(time.Duration(intervalMs) * time.Millisecond)
 	c.tickers = append(c.tickers, ticker)
@@ -91,6 +126,7 @@ func (c *Client) Stop() {
 		}
 
 		c.runningSubtasksWaitGroup.Wait()
+		c.setHealth(errors.New("daemon stopped"))
 	})
 }
 
@@ -249,6 +285,7 @@ func (c *Client) start(ctx context.Context,
 
 	pricefeedClient := api.NewPriceFeedServiceClient(daemonConn)
 	subTaskRunner.StartPriceUpdater(
+		c,
 		ctx,
 		priceUpdaterTicker,
 		priceUpdaterStop,

protocol/daemons/pricefeed/client/client_test.go

@@ -51,6 +51,7 @@ type FakeSubTaskRunner struct {
 
 // StartPriceUpdater replaces `client.StartPriceUpdater` and advances `UpdaterCallCount` by one.
 func (f *FakeSubTaskRunner) StartPriceUpdater(
+	c *Client,
 	ctx context.Context,
 	ticker *time.Ticker,
 	stop <-chan bool,
@@ -247,6 +248,10 @@ func TestStart_InvalidConfig(t *testing.T) {
 				&faketaskRunner,
 			)
 
+			// Expect daemon is not healthy on startup. Daemon becomes healthy after the first successful market
+			// update.
+			require.ErrorContains(t, client.HealthCheck(grpc_util.Ctx), "daemon uninitialized")
+
 			if tc.expectedError == nil {
 				require.NoError(t, err)
 			} else {
@@ -619,6 +624,7 @@ func TestPriceUpdater_Mixed(t *testing.T) {
 		},
 		"No exchange market prices, does not call `UpdateMarketPrices`": {
 			exchangeAndMarketPrices: []*client.ExchangeIdMarketPriceTimestamp{},
+			priceUpdateError:        types.ErrEmptyMarketPriceUpdate,
 		},
 		"One market for one exchange": {
 			exchangeAndMarketPrices: []*client.ExchangeIdMarketPriceTimestamp{
@@ -721,6 +727,89 @@ func TestPriceUpdater_Mixed(t *testing.T) {
 	}
 }
 
+// singleTickTickerAndStop creates a ticker that ticks once before the stop channel is signaled. In order for this
+// ticker to be effective, it needs to be consumed within 100ms.
+func singleTickTickerAndStop() (*time.Ticker, chan bool) {
+	// Create a ticker with a duration long enough that we do not expect to see a tick within the timeframe
+	// of a normal unit test.
+	ticker := time.NewTicker(10 * time.Minute)
+	// Override the ticker's channel with a new channel we can insert into directly, and add a single tick.
+	newChan := make(chan time.Time, 1)
+	newChan <- time.Now()
+	ticker.C = newChan
+
+	stop := make(chan bool, 1)
+
+	// After 100ms, stop the ticker and signal the stop channel.
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		ticker.Stop()
+		stop <- true
+	}()
+
+	return ticker, stop
+}
+
+func TestHealthCheck_Mixed(t *testing.T) {
+	tests := map[string]struct {
+		updateMarketPricesError error
+		expectedError           error
+	}{
+		"No error - daemon healthy": {
+			updateMarketPricesError: nil,
+			expectedError:           nil,
+		},
+		"Error - daemon unhealthy": {
+			updateMarketPricesError: fmt.Errorf("failed to update market prices"),
+			expectedError: fmt.Errorf(
+				"price deamon unhealthy: failed to run price updater task loop for price daemon: " +
+					"failed to update market prices",
+			),
+		},
+	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			// Setup.
+			// Create `ExchangeIdMarketPriceTimestamp` and populate it with market-price updates.
+			etmp, err := types.NewExchangeToMarketPrices([]types.ExchangeId{constants.ExchangeId1})
+			require.NoError(t, err)
+			etmp.UpdatePrice(constants.ExchangeId1, constants.Market9_TimeT_Price1)
+
+			// Create a mock `PriceFeedServiceClient`.
+			mockPriceFeedClient := generateMockQueryClient()
+
+			// Mock the `UpdateMarketPrices` call to return an error if specified.
+			mockPriceFeedClient.On("UpdateMarketPrices", grpc_util.Ctx, mock.Anything).
+				Return(nil, tc.updateMarketPricesError).Once()
+
+			ticker, stop := singleTickTickerAndStop()
+			client := newClient()
+
+			// Act.
+			// Run the price updater for a single tick with a successful update. Expect the daemon to toggle to a
+			// healthy state.
+			subTaskRunnerImpl.StartPriceUpdater(
+				client,
+				grpc_util.Ctx,
+				ticker,
+				stop,
+				etmp,
+				mockPriceFeedClient,
+				log.NewNopLogger(),
+			)
+			// Assert.
+			if tc.expectedError == nil {
+				require.NoError(t, client.HealthCheck(grpc_util.Ctx))
+			} else {
+				require.ErrorContains(t, client.HealthCheck(grpc_util.Ctx), tc.expectedError.Error())
+			}
+
+			// Cleanup.
+			close(stop)
+		})
+	}
+}
+
 // TestMarketUpdater_Mixed tests the `RunMarketParamUpdaterTaskLoop` function invokes the grpc
 // query to the prices query client and that if the query succeeds, the config is updated.
 func TestMarketUpdater_Mixed(t *testing.T) {

protocol/daemons/pricefeed/client/sub_task_runner.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"cosmossdk.io/errors"
 	"github.com/dydxprotocol/v4-chain/protocol/daemons/pricefeed/client/constants"
 	"github.com/dydxprotocol/v4-chain/protocol/daemons/pricefeed/client/price_encoder"
 	"github.com/dydxprotocol/v4-chain/protocol/daemons/pricefeed/client/price_fetcher"
@@ -35,6 +36,7 @@ var _ SubTaskRunner = (*SubTaskRunnerImpl)(nil)
 // SubTaskRunner is the interface for running pricefeed client task functions.
 type SubTaskRunner interface {
 	StartPriceUpdater(
+		c *Client,
 		ctx context.Context,
 		ticker *time.Ticker,
 		stop <-chan bool,
@@ -76,6 +78,7 @@ type SubTaskRunner interface {
 // StartPriceUpdater runs in the daemon's main goroutine and does not need access to the daemon's wait group
 // to signal task completion.
 func (s *SubTaskRunnerImpl) StartPriceUpdater(
+	c *Client,
 	ctx context.Context,
 	ticker *time.Ticker,
 	stop <-chan bool,
@@ -88,9 +91,11 @@ func (s *SubTaskRunnerImpl) StartPriceUpdater(
 		case <-ticker.C:
 			err := RunPriceUpdaterTaskLoop(ctx, exchangeToMarketPrices, priceFeedServiceClient, logger)
 			if err != nil {
-				panic(err)
+				logger.Error("Failed to run price updater task loop for price daemon", constants.ErrorLogKey, err)
 			}
 
+			c.setHealth(errors.Wrap(err, "failed to run price updater task loop for price daemon"))
+
 		case <-stop:
 			return
 		}
@@ -265,8 +270,10 @@ func RunPriceUpdaterTaskLoop(
 		metrics.Latency,
 	)
 
-	// On startup the length of request will likely be 0. However, sending a request of length 0
-	// is a fatal error.
+	// On startup the length of request will likely be 0. Even so, we return an error here because this
+	// is unexpected behavior once the daemon reaches a steady state. The daemon health check process should
+	// be robust enough to ignore temporarily unhealthy daemons.
+	// Sending a request of length 0, however, causes a panic.
 	// panic: rpc error: code = Unknown desc = Market price update has length of 0.
 	if len(request.MarketPriceUpdates) > 0 {
 		_, err := priceFeedServiceClient.UpdateMarketPrices(ctx, request)
@@ -291,6 +298,7 @@ func RunPriceUpdaterTaskLoop(
 			metrics.PriceUpdaterZeroPrices,
 			metrics.Count,
 		)
+		return types.ErrEmptyMarketPriceUpdate
 	}
 
 	return nil

protocol/daemons/pricefeed/client/types/errors.go

@@ -0,0 +1,9 @@
+package types
+
+import (
+	"errors"
+)
+
+var (
+	ErrEmptyMarketPriceUpdate = errors.New("Market price update has length of 0")
+)