fix: properly handle case when perp data point comes in late

CHANGELOG.md

@@ -43,6 +43,7 @@
 - [10052](https://github.com/vegaprotocol/vega/issues/10052) - Some recent stats tables should have been `hypertables` with retention periods.
 - [10103](https://github.com/vegaprotocol/vega/issues/10103) - List ledgers `API` returns bad error when filtering by transfer type only.
 - [10120](https://github.com/vegaprotocol/vega/issues/10120) - Assure theoretical and actual funding payment calculations are consistent.
+- [10164](https://github.com/vegaprotocol/vega/issues/10164) - Properly handle edge case where an external data point is received out of order.
 - [10121](https://github.com/vegaprotocol/vega/issues/10121) - Assure `EstimatePosition` API works correctly with sparse perps data
 - [10126](https://github.com/vegaprotocol/vega/issues/10126) - Account for invalid stop orders in batch, charge default gas.
 - [10123](https://github.com/vegaprotocol/vega/issues/10123) - Ledger exports contain account types of "UNKNOWN" type

core/products/perpetual.go

@@ -39,6 +39,7 @@ var (
 	year = num.DecimalFromInt64((24 * 365 * time.Hour).Nanoseconds())
 
 	ErrDataPointAlreadyExistsAtTime = errors.New("data-point already exists at timestamp")
+	ErrDataPointIsTooOld            = errors.New("data-point is too old")
 	ErrInitialPeriodNotStarted      = errors.New("initial settlement period not started")
 )
 
@@ -286,28 +287,48 @@ func (c *cachedTWAP) insertPoint(point *dataPoint) (*num.Uint, error) {
 	return twap, nil
 }
 
+// prependPoint handles the case where the given point is either before the first point, or before the start of the period.
+func (c *cachedTWAP) prependPoint(point *dataPoint) (*num.Uint, error) {
+	first := c.points[0]
+
+	if point.t == first.t {
+		return nil, ErrDataPointAlreadyExistsAtTime
+	}
+
+	// our first point is on or before the start of the period, and the new point is before both, its too old
+	if first.t <= c.periodStart && point.t < first.t {
+		return nil, ErrDataPointIsTooOld
+	}
+
+	points := c.points[:]
+	if first.t < c.periodStart && first.t < point.t {
+		// this is the case where we have first-point < new-point < period start and we only want to keep
+		// one data point that is before the start of the period, so we throw away first-point
+		points = c.points[1:]
+	}
+
+	c.points = []*dataPoint{point}
+	c.sumProduct = num.UintZero()
+	c.setPeriod(point.t, point.t)
+	for _, p := range points {
+		c.calculate(p.t)
+		c.points = append(c.points, p)
+	}
+	return point.price.Clone(), nil
+}
+
 // addPoint takes the given point and works out where it fits against what we already have, updates the
 // running sum-product and returns the TWAP at point.t.
 func (c *cachedTWAP) addPoint(point *dataPoint) (*num.Uint, error) {
-	if len(c.points) == 0 || point.t < c.start {
-		// first point, or new point is before the start of the funding period
+	if len(c.points) == 0 {
 		c.points = []*dataPoint{point}
 		c.setPeriod(point.t, point.t)
 		c.sumProduct = num.UintZero()
 		return point.price.Clone(), nil
 	}
 
-	// point to add is before the very first point we added, a little weird but ok
-	if point.t <= c.points[0].t {
-		points := c.points[:]
-		c.points = []*dataPoint{point}
-		c.setPeriod(point.t, point.t)
-		c.sumProduct = num.UintZero()
-		for _, p := range points {
-			c.calculate(p.t)
-			c.points = append(c.points, p)
-		}
-		return point.price.Clone(), nil
+	if point.t <= c.points[0].t || point.t <= c.periodStart {
+		return c.prependPoint(point)
 	}
 
 	// new point is after the last point, just calculate the TWAP at point.t and append

core/products/perpetual_test.go

@@ -62,95 +62,69 @@ func TestPeriodicSettlement(t *testing.T) {
 	t.Run("test update perpetual", testUpdatePerpetual)
 	t.Run("test terminate trading coincides with time trigger", testTerminateTradingCoincidesTimeTrigger)
 	t.Run("test funding-payment on start boundary", TestFundingPaymentOnStartBoundary)
+	t.Run("test data point is before the first point", TestPrependPoint)
 }
 
-func TestExternalDataPointTWAPInSequence(t *testing.T) {
-	perp := testPerpetual(t)
-	defer perp.ctrl.Finish()
-
-	ctx := context.Background()
-	tstData, err := getGQLData()
-	require.NoError(t, err)
-	data := tstData.GetDataPoints()
-
-	// want to start the period from before the point with the smallest time
-	seq := math.MaxInt
-	st := data[0].t
-	for i := 0; i < len(data); i++ {
-		if data[i].t < st {
-			st = data[i].t
-		}
-		seq = num.MinV(seq, data[i].seq)
-	}
-	// leave opening auction
-	whenLeaveOpeningAuction(t, perp, st-1)
-
-	// set the first internal data-point
-	for i, dp := range data {
-		if dp.seq > seq {
-			perp.broker.EXPECT().Send(gomock.Any()).Times(2)
-			if dp.seq == 2 {
-				perp.broker.EXPECT().SendBatch(gomock.Any()).Times(1)
-			}
-			perp.perpetual.PromptSettlementCue(ctx, dp.t)
-			seq = dp.seq
-		}
-		check := func(e events.Event) {
-			de, ok := e.(*events.FundingPeriodDataPoint)
-			require.True(t, ok)
-			dep := de.Proto()
-			if dep.Twap == "0" {
-				return
-			}
-			require.Equal(t, dp.twap.String(), dep.Twap, fmt.Sprintf("IDX: %d\n%#v\n", i, dep))
-		}
-		perp.broker.EXPECT().Send(gomock.Any()).Times(1).Do(check)
-		perp.perpetual.AddTestExternalPoint(ctx, dp.price, dp.t)
+func TestRealData(t *testing.T) {
+	tcs := []struct {
+		name    string
+		reverse bool
+	}{
+		{
+			"in order",
+			false,
+		},
+		{
+			"out of order",
+			false,
+		},
 	}
-}
-
-func TestExternalDataPointTWAPOutSequence(t *testing.T) {
-	perp := testPerpetual(t)
-	defer perp.ctrl.Finish()
-
-	ctx := context.Background()
-	tstData, err := getGQLData()
-	require.NoError(t, err)
-	data := tstData.GetDataPoints()
 
-	// leave opening auction
-	whenLeaveOpeningAuction(t, perp, data[0].t-1)
+	for _, tc := range tcs {
+		t.Run(tc.name, func(tt *testing.T) {
+			perp := testPerpetual(t)
+			defer perp.ctrl.Finish()
+
+			ctx := context.Background()
+			tstData, err := getGQLData()
+			require.NoError(t, err)
+			data := tstData.GetDataPoints(false)
+
+			// want to start the period from before the point with the smallest time
+			seq := math.MaxInt
+			st := data[0].t
+			nd := data[0].t
+			for i := 0; i < len(data); i++ {
+				if data[i].t < st {
+					st = data[i].t
+				}
+				if data[i].t > nd {
+					nd = data[i].t
+				}
+				seq = num.MinV(seq, data[i].seq)
+			}
 
-	seq := data[0].seq
-	last := 0
-	for i := 0; i < len(data); i++ {
-		if data[i].seq != seq {
-			break
-		}
-		last = i
-	}
-	perp.broker.EXPECT().Send(gomock.Any()).Times(1)
-	// add the first (earliest) data-point first
-	perp.perpetual.AddTestExternalPoint(ctx, data[0].price, data[0].t)
-	// submit external data points in non-sequential order
-	for j := last; j < 0; j-- {
-		dp := data[j]
-		if dp.seq > seq {
-			// break
-			perp.broker.EXPECT().Send(gomock.Any()).Times(2)
-			perp.perpetual.PromptSettlementCue(ctx, dp.t)
-		}
-		check := func(e events.Event) {
-			de, ok := e.(*events.FundingPeriodDataPoint)
-			require.True(t, ok)
-			dep := de.Proto()
-			if dep.Twap == "0" {
-				return
+			perp.perpetual.SetSettlementListener(func(context.Context, *num.Numeric) {})
+			// leave opening auction
+			whenLeaveOpeningAuction(t, perp, st-1)
+
+			perp.broker.EXPECT().Send(gomock.Any()).AnyTimes()
+			perp.broker.EXPECT().SendBatch(gomock.Any()).AnyTimes()
+
+			// set the first internal data-point
+			for _, dp := range data {
+				if dp.seq > seq {
+					perp.perpetual.PromptSettlementCue(ctx, dp.t)
+					seq = dp.seq
+				}
+				perp.perpetual.AddTestExternalPoint(ctx, dp.price, dp.t)
+				perp.perpetual.SubmitDataPoint(ctx, num.UintZero().Add(dp.price, num.NewUint(100)), dp.t)
 			}
-			require.Equal(t, dp.twap.String(), dep.Twap, fmt.Sprintf("IDX: %d\n%#v\n", j, dep))
-		}
-		perp.broker.EXPECT().Send(gomock.Any()).Times(1).Do(check)
-		perp.perpetual.AddTestExternalPoint(ctx, dp.price, dp.t)
+			d := perp.perpetual.GetData(nd).Data.(*types.PerpetualData)
+			assert.Equal(t, "29124220000", d.ExternalTWAP)
+			assert.Equal(t, "29124220100", d.InternalTWAP)
+			assert.Equal(t, "100", d.FundingPayment)
+		})
 	}
 }
 
@@ -193,6 +167,61 @@ func testPeriodEndWithNoDataPoints(t *testing.T) {
 	assert.False(t, called)
 }
 
+func TestPrependPoint(t *testing.T) {
+	perp := testPerpetual(t)
+	defer perp.ctrl.Finish()
+
+	ctx := context.Background()
+	now := time.Unix(1000, 0)
+	whenLeaveOpeningAuction(t, perp, now.UnixNano())
+
+	perp.broker.EXPECT().Send(gomock.Any()).AnyTimes()
+
+	// we'll use this point to check that we do not lose a later point when we recalc when earlier points come in
+	err := perp.perpetual.SubmitDataPoint(ctx, num.NewUint(10), time.Unix(5000, 0).UnixNano())
+	perp.perpetual.AddTestExternalPoint(ctx, num.NewUint(9), time.Unix(5000, 0).UnixNano())
+	require.NoError(t, err)
+	require.Equal(t, "1", getFundingPayment(t, perp, time.Unix(5000, 0).UnixNano()))
+
+	// first point is after the start of the period
+	err = perp.perpetual.SubmitDataPoint(ctx, num.NewUint(10), time.Unix(2000, 0).UnixNano())
+	require.NoError(t, err)
+	require.Equal(t, "1", getFundingPayment(t, perp, time.Unix(5000, 0).UnixNano()))
+
+	// now another one comes in before this, but also after the start of the period
+	err = perp.perpetual.SubmitDataPoint(ctx, num.NewUint(50), time.Unix(1500, 0).UnixNano())
+	require.NoError(t, err)
+	require.Equal(t, "6", getFundingPayment(t, perp, time.Unix(5000, 0).UnixNano()))
+
+	// now one comes in before the start of the period
+	err = perp.perpetual.SubmitDataPoint(ctx, num.NewUint(50), time.Unix(500, 0).UnixNano())
+	require.NoError(t, err)
+	require.Equal(t, "11", getFundingPayment(t, perp, time.Unix(5000, 0).UnixNano()))
+
+	// now one comes in before this point
+	err = perp.perpetual.SubmitDataPoint(ctx, num.UintOne(), time.Unix(250, 0).UnixNano())
+	require.ErrorIs(t, err, products.ErrDataPointIsTooOld)
+	require.Equal(t, "11", getFundingPayment(t, perp, time.Unix(5000, 0).UnixNano()))
+
+	// now one comes in after the first point, but before the period start
+	err = perp.perpetual.SubmitDataPoint(ctx, num.UintOne(), time.Unix(500, 0).UnixNano())
+	require.ErrorIs(t, err, products.ErrDataPointAlreadyExistsAtTime)
+	require.Equal(t, "11", getFundingPayment(t, perp, time.Unix(5000, 0).UnixNano()))
+
+	// now one comes in after the first point, but before the period start
+	err = perp.perpetual.SubmitDataPoint(ctx, num.NewUint(50), time.Unix(750, 0).UnixNano())
+	require.NoError(t, err)
+	require.Equal(t, "11", getFundingPayment(t, perp, time.Unix(5000, 0).UnixNano()))
+
+	// now one comes that equals period start
+	err = perp.perpetual.SubmitDataPoint(ctx, num.NewUint(50), time.Unix(1000, 0).UnixNano())
+	require.NoError(t, err)
+	require.Equal(t, "11", getFundingPayment(t, perp, time.Unix(5000, 0).UnixNano()))
+
+	err = perp.perpetual.SubmitDataPoint(ctx, num.NewUint(100000), time.Unix(750, 0).UnixNano())
+	require.ErrorIs(t, err, products.ErrDataPointIsTooOld)
+}
+
 func testEqualInternalAndExternalPrices(t *testing.T) {
 	perp := testPerpetual(t)
 	defer perp.ctrl.Finish()
@@ -1578,21 +1607,25 @@ func getGQLData() (*GQL, error) {
 	if err := json.Unmarshal([]byte(testData), &ret); err != nil {
 		return nil, err
 	}
-	ret.Sort()
 	return &ret, nil
 }
 
-func (g *GQL) Sort() {
+func (g *GQL) Sort(reverse bool) {
 	// group by sequence
 	sort.SliceStable(g.Data.FundingDataPoints.Edges, func(i, j int) bool {
+		if g.Data.FundingDataPoints.Edges[i].Node.Seq == g.Data.FundingDataPoints.Edges[j].Node.Seq {
+			if reverse {
+				return g.Data.FundingDataPoints.Edges[i].Node.Timestamp.UnixNano() > g.Data.FundingDataPoints.Edges[j].Node.Timestamp.UnixNano()
+			}
+			return g.Data.FundingDataPoints.Edges[i].Node.Timestamp.UnixNano() < g.Data.FundingDataPoints.Edges[j].Node.Timestamp.UnixNano()
+		}
+
 		return g.Data.FundingDataPoints.Edges[i].Node.Seq < g.Data.FundingDataPoints.Edges[j].Node.Seq
 	})
-	for i, j := 0, len(g.Data.FundingDataPoints.Edges)-1; i < j; i, j = i+1, j-1 {
-		g.Data.FundingDataPoints.Edges[i], g.Data.FundingDataPoints.Edges[j] = g.Data.FundingDataPoints.Edges[j], g.Data.FundingDataPoints.Edges[i]
-	}
 }
 
-func (g *GQL) GetDataPoints() []DataPoint {
+func (g *GQL) GetDataPoints(reverse bool) []DataPoint {
+	g.Sort(reverse)
 	ret := make([]DataPoint, 0, len(g.Data.FundingDataPoints.Edges))
 	for _, n := range g.Data.FundingDataPoints.Edges {
 		p, _ := num.UintFromString(n.Node.Price, 10)