[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

@@ -49,8 +49,14 @@ type Client struct {
 
 	// Ensure stop only executes one time.
 	stopDaemon sync.Once
+
+	// include HealthCheckableImpl to track the health of the daemon.
+	daemontypes.HealthCheckableImpl
 }
 
+// Ensure Client implements the HealthCheckable interface.
+var _ daemontypes.HealthCheckable = (*Client)(nil)
+
 func newClient() *Client {
 	client := &Client{
 		tickers: []*time.Ticker{},
@@ -59,14 +65,15 @@ func newClient() *Client {
 
 	// Set the client's daemonStartup state to indicate that the daemon has not finished starting up.
 	client.daemonStartup.Add(1)
+	client.InitializeHealthStatus(constants.PricefeedDaemonModuleName, &libtime.TimeProviderImpl{})
 	return client
 }
 
 // newTickerWithStop creates a new ticker and a channel for iteratively looping through a subtask with a stop signal
 // 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 +98,12 @@ func (c *Client) Stop() {
 		}
 
 		c.runningSubtasksWaitGroup.Wait()
+
+		// When the daemon stops, toggle it into an unhealthy state.
+		c.RecordUpdateFailure(
+			&libtime.TimeProviderImpl{},
+			fmt.Errorf("%v stopped", constants.PricefeedDaemonModuleName),
+		)
 	})
 }
 
@@ -249,6 +262,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

@@ -11,6 +11,7 @@ import (
 	daemonserver "github.com/dydxprotocol/v4-chain/protocol/daemons/server"
 	pricefeed_types "github.com/dydxprotocol/v4-chain/protocol/daemons/server/types/pricefeed"
 	daemontypes "github.com/dydxprotocol/v4-chain/protocol/daemons/types"
+	libtime "github.com/dydxprotocol/v4-chain/protocol/lib/time"
 	"github.com/dydxprotocol/v4-chain/protocol/testutil/appoptions"
 	grpc_util "github.com/dydxprotocol/v4-chain/protocol/testutil/grpc"
 	pricetypes "github.com/dydxprotocol/v4-chain/protocol/x/prices/types"
@@ -51,6 +52,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 +249,14 @@ 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(&libtime.TimeProviderImpl{}),
+				"pricefeed-daemon is initializing",
+			)
+
 			if tc.expectedError == nil {
 				require.NoError(t, err)
 			} else {
@@ -619,6 +629,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 +732,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(
+				"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.
+			timeProvider := &libtime.TimeProviderImpl{}
+			if tc.expectedError == nil {
+				require.NoError(t, client.HealthCheck(timeProvider))
+			} else {
+				require.ErrorContains(t, client.HealthCheck(timeProvider), 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,10 +2,12 @@ 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"
 	daemontypes "github.com/dydxprotocol/v4-chain/protocol/daemons/types"
+	libtime "github.com/dydxprotocol/v4-chain/protocol/lib/time"
 	pricetypes "github.com/dydxprotocol/v4-chain/protocol/x/prices/types"
 	"net/http"
 	"time"
@@ -35,6 +37,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 +79,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,
@@ -87,8 +91,17 @@ func (s *SubTaskRunnerImpl) StartPriceUpdater(
 		select {
 		case <-ticker.C:
 			err := RunPriceUpdaterTaskLoop(ctx, exchangeToMarketPrices, priceFeedServiceClient, logger)
-			if err != nil {
-				panic(err)
+
+			if err == nil {
+				// Record update success for the daemon health check.
+				c.RecordUpdateSuccess(&libtime.TimeProviderImpl{})
+			} else {
+				logger.Error("Failed to run price updater task loop for price daemon", constants.ErrorLogKey, err)
+				// Record update failure for the daemon health check.
+				c.RecordUpdateFailure(
+					&libtime.TimeProviderImpl{},
+					errors.Wrap(err, "failed to run price updater task loop for price daemon"),
+				)
 			}
 
 		case <-stop:
@@ -265,8 +278,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 +306,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")
+)