-
Notifications
You must be signed in to change notification settings - Fork 0
/
pricefeed.go
130 lines (111 loc) · 3.76 KB
/
pricefeed.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
package server
import (
"context"
"time"
errorsmod "cosmossdk.io/errors"
"github.com/jinxprotocol/v4-chain/protocol/daemons/server/types"
gometrics "github.com/armon/go-metrics"
"github.com/jinxprotocol/v4-chain/protocol/daemons/constants"
"github.com/jinxprotocol/v4-chain/protocol/daemons/pricefeed/api"
pricefeedmetrics "github.com/jinxprotocol/v4-chain/protocol/daemons/pricefeed/metrics"
pricefeedtypes "github.com/jinxprotocol/v4-chain/protocol/daemons/server/types/pricefeed"
daemontypes "github.com/jinxprotocol/v4-chain/protocol/daemons/types"
"github.com/jinxprotocol/v4-chain/protocol/lib/metrics"
"github.com/cosmos/cosmos-sdk/telemetry"
)
// PriceFeedServer defines the fields required for price updates.
type PriceFeedServer struct {
marketToExchange *pricefeedtypes.MarketToExchangePrices
}
// WithPriceFeedMarketToExchangePrices sets the `MarketToExchangePrices` field.
// This is used by the price feed service to communicate price updates
// to the main application.
func (server *Server) WithPriceFeedMarketToExchangePrices(
marketToExchange *pricefeedtypes.MarketToExchangePrices,
) *Server {
server.marketToExchange = marketToExchange
return server
}
// UpdateMarketPrices updates prices from exchanges for each market provided.
func (s *Server) UpdateMarketPrices(
ctx context.Context,
req *api.UpdateMarketPricesRequest,
) (
response *api.UpdateMarketPricesResponse,
err error,
) {
// Measure latency in ingesting and handling gRPC price update.
defer telemetry.ModuleMeasureSince(
metrics.PricefeedServer,
time.Now(),
metrics.PricefeedServerUpdatePrices,
metrics.Latency,
)
// This panic is an unexpected condition because we initialize the market price cache in app initialization before
// starting the server or daemons.
if s.marketToExchange == nil {
panic(
errorsmod.Wrapf(
daemontypes.ErrServerNotInitializedCorrectly,
"MarketToExchange not initialized",
),
)
}
if err = validateMarketPricesUpdatesMessage(req); err != nil {
// Log if failure occurs during an update.
s.logger.Error("Failed to validate price update message", "error", err)
return nil, err
}
s.marketToExchange.UpdatePrices(req.MarketPriceUpdates)
// Capture valid responses in metrics.
s.reportValidResponse(types.PricefeedDaemonServiceName)
return &api.UpdateMarketPricesResponse{}, nil
}
// validateMarketPricesUpdatesMessage validates a `UpdateMarketPricesRequest`.
func validateMarketPricesUpdatesMessage(req *api.UpdateMarketPricesRequest) error {
if len(req.MarketPriceUpdates) == 0 {
return daemontypes.ErrPriceFeedMarketPriceUpdateEmpty
}
for _, mpu := range req.MarketPriceUpdates {
if err := validateMarketPriceUpdate(mpu); err != nil {
// Measure failure per market in validation.
telemetry.IncrCounterWithLabels(
[]string{
metrics.PricefeedServer,
metrics.PricefeedServerValidatePrices,
metrics.Error,
},
1,
[]gometrics.Label{
pricefeedmetrics.GetLabelForMarketId(mpu.MarketId),
},
)
return err
}
}
return nil
}
// validateMarketPriceUpdate validates a single `MarketPriceUpdate`.
func validateMarketPriceUpdate(mpu *api.MarketPriceUpdate) error {
for _, ep := range mpu.ExchangePrices {
if ep.Price == constants.DefaultPrice {
return generateSdkErrorForExchangePrice(
daemontypes.ErrPriceFeedInvalidPrice,
ep,
mpu.MarketId,
)
}
if ep.LastUpdateTime == nil {
return generateSdkErrorForExchangePrice(
daemontypes.ErrPriceFeedLastUpdateTimeNotSet,
ep,
mpu.MarketId,
)
}
}
return nil
}
// generateSdkErrorForExchangePrice generates an error for an invalid `ExchangePrice`.
func generateSdkErrorForExchangePrice(err error, ep *api.ExchangePrice, marketId uint32) error {
return errorsmod.Wrapf(err, "ExchangePrice: %v and MarketId: %d", ep, marketId)
}