Merge pull request #41 from CrocSwap/better-refresher-and-knockouts

Improve liquidity refresher and knockout tracking

Author: benwolski

cache/memCache.go

@@ -20,7 +20,8 @@ type MemoryCache struct {
 	poolKnockouts        RWLockMapMap[types.PoolLocation, types.PositionLocation, *model.KnockoutSubplot]
 	userAndPoolKnockouts RWLockMapMap[chainUserAndPool, types.PositionLocation, *model.KnockoutSubplot]
 
-	knockoutSagas RWLockMap[types.BookLocation, *model.KnockoutSaga]
+	knockoutSagas      RWLockMap[types.BookLocation, *model.KnockoutSaga]
+	knockoutPivotTimes RWLockMap[types.BookLocation, int]
 
 	userTxs        RWLockMapArray[chainAndAddr, types.PoolTxEvent]
 	poolTxs        RWLockMapArray[types.PoolLocation, types.PoolTxEvent]
@@ -47,7 +48,8 @@ func New() *MemoryCache {
 		poolKnockouts:        newRwLockMapMap[types.PoolLocation, types.PositionLocation, *model.KnockoutSubplot](),
 		userAndPoolKnockouts: newRwLockMapMap[chainUserAndPool, types.PositionLocation, *model.KnockoutSubplot](),
 
-		knockoutSagas: newRwLockMap[types.BookLocation, *model.KnockoutSaga](),
+		knockoutSagas:      newRwLockMap[types.BookLocation, *model.KnockoutSaga](),
+		knockoutPivotTimes: newRwLockMap[types.BookLocation, int](),
 
 		userTxs:        newRwLockMapArray[chainAndAddr, types.PoolTxEvent](),
 		poolTxs:        newRwLockMapArray[types.PoolLocation, types.PoolTxEvent](),

cache/transactors.go

@@ -293,7 +293,7 @@ func (m *MemoryCache) MaterializePosition(loc types.PositionLocation) *model.Pos
 	return val
 }
 
-func (m *MemoryCache) MaterializeKnockoutBook(loc types.BookLocation) *model.KnockoutSaga {
+func (m *MemoryCache) MaterializeKnockoutSaga(loc types.BookLocation) *model.KnockoutSaga {
 	val, ok := m.knockoutSagas.lookup(loc)
 	if !ok {
 		val = model.NewKnockoutSaga()
@@ -305,7 +305,7 @@ func (m *MemoryCache) MaterializeKnockoutBook(loc types.BookLocation) *model.Kno
 func (m *MemoryCache) MaterializeKnockoutPos(loc types.PositionLocation) *model.KnockoutSubplot {
 	val, ok := m.liqKnockouts.lookup(loc)
 	if !ok {
-		saga := m.MaterializeKnockoutBook(loc.ToBookLoc())
+		saga := m.MaterializeKnockoutSaga(loc.ToBookLoc())
 		val = saga.ForUser(loc.User)
 		m.liqKnockouts.insert(loc, val)
 		m.userKnockouts.insert(chainAndAddr{loc.ChainId, loc.User}, loc, val)
@@ -317,3 +317,16 @@ func (m *MemoryCache) MaterializeKnockoutPos(loc types.PositionLocation) *model.
 	m.poolKoUpdates.insert(loc.PoolLocation, koAndLocPair{loc, val})
 	return val
 }
+
+func (m *MemoryCache) RetrievePivotTime(loc types.BookLocation) int {
+	pos, okay := m.knockoutPivotTimes.lookup(loc)
+	if okay {
+		return pos
+	} else {
+		return 0
+	}
+}
+
+func (m *MemoryCache) SetPivotTime(loc types.BookLocation, pivotTime int) {
+	m.knockoutPivotTimes.insert(loc, pivotTime)
+}

controller/controller.go

@@ -2,6 +2,8 @@ package controller
 
 import (
 	"log"
+	"math"
+	"math/big"
 	"time"
 
 	"github.com/CrocSwap/graphcache-go/cache"
@@ -47,7 +49,7 @@ func (c *Controller) SpinUntilLiqSync() {
 		nowTime := time.Now().Unix()
 		syncSec := c.refresher.lastRefreshSec
 		if nowTime < syncSec+REFRESH_PAUSE_SECS {
-			log.Println("Waiting for liquidity sync pause. Last refresh:", syncSec, "now:", nowTime)
+			// log.Println("Waiting for liquidity sync pause. Last refresh:", syncSec, "now:", nowTime)
 		} else {
 			return
 		}
@@ -82,6 +84,9 @@ func (c *ControllerOverNetwork) IngestBalance(b tables.Balance) {
 }
 
 func (c *ControllerOverNetwork) IngestLiqChange(l tables.LiqChange) {
+	if l.ChangeType == tables.ChangeTypeCross {
+		c.IngestKnockout(l)
+	}
 	c.applyToPosition(l)
 	c.applyToLiqCurve(l)
 	c.ctrl.history.CommitLiqChange(l)
@@ -101,7 +106,7 @@ func (c *ControllerOverNetwork) applyToPosition(l tables.LiqChange) {
 		User:              types.RequireEthAddr(l.User),
 	}
 
-	if l.PositionType == "knockout" {
+	if l.PositionType == tables.PosTypeKnockout {
 		c.applyToKnockout(l, loc)
 	} else {
 		c.applyToPassiveLiq(l, loc)
@@ -121,11 +126,63 @@ func (c *ControllerOverNetwork) applyToLiqCurve(l tables.LiqChange) {
 }
 
 func (c *ControllerOverNetwork) applyToKnockout(l tables.LiqChange, loc types.PositionLocation) {
-	if l.ChangeType == "cross" {
-		return // Cross events are handled KnockoutCross table
+	pivotLoc := loc.ToBookLoc()
+	if l.IsBid == 1 {
+		pivotLoc.LiquidityLocation.AskTick = l.BidTick
+	} else {
+		pivotLoc.LiquidityLocation.BidTick = l.AskTick
+	}
+	if l.ChangeType == tables.ChangeTypeCross {
+		c.ctrl.cache.SetPivotTime(pivotLoc, 0)
+		return
+	}
+	pivotTime := c.ctrl.cache.RetrievePivotTime(pivotLoc)
+	if l.ChangeType == tables.ChangeTypeMint && pivotTime == 0 {
+		c.ctrl.cache.SetPivotTime(pivotLoc, l.Time)
+		pivotTime = l.Time
+	}
+
+	event := model.KnockoutSagaTx{
+		TxTime:    l.Time,
+		TxHash:    l.TX,
+		PivotTime: pivotTime,
 	}
 	pos := c.ctrl.cache.MaterializeKnockoutPos(loc)
+	if l.ChangeType == tables.ChangeTypeMint {
+		pos.AppendMint(event)
+	} else if l.ChangeType == tables.ChangeTypeBurn {
+		pos.AppendBurn(event)
+	}
+
 	c.ctrl.workers.omniUpdates <- &koPosUpdateMsg{liq: l, pos: pos, loc: loc}
+
+	// Estimate position liquidity from the flows
+	if (l.ChangeType == tables.ChangeTypeMint || l.ChangeType == tables.ChangeTypeBurn || l.ChangeType == tables.ChangeTypeRecover) && l.BaseFlow != nil && l.QuoteFlow != nil {
+		liq := model.DeriveLiquidityFromConcFlow(*l.BaseFlow, *l.QuoteFlow, l.BidTick, l.AskTick)
+		if math.IsInf(liq, 0) || math.IsNaN(liq) {
+			log.Println("Invalid liq", liq, "for", l)
+			liq = 0
+		}
+		liqBigInt, _ := big.NewFloat(liq).Int(nil)
+
+		activeLiq := pos.Liq.GetActiveLiq()
+		pos.Liq.UpdateActiveLiq(*big.NewInt(0).Add(activeLiq, liqBigInt), 0)
+		afterLiq := pos.Liq.GetActiveLiq()
+		if afterLiq.Cmp(big.NewInt(0)) < 0 {
+			pos.Liq.UpdateActiveLiq(*big.NewInt(0), 0)
+			afterLiq = big.NewInt(0)
+		}
+		afterLiqFloat, _ := afterLiq.Float64()
+
+		// If it's a burn and the remaining liq is less than 10% of the liq change, set it to 0
+		if l.ChangeType == tables.ChangeTypeBurn && afterLiqFloat > 0 && math.Abs(liq)*0.10 > math.Abs(afterLiqFloat) {
+			pos.Liq.UpdateActiveLiq(*big.NewInt(0), 0)
+		}
+		if l.ChangeType == tables.ChangeTypeRecover || l.ChangeType == tables.ChangeTypeClaim {
+			pos.Liq.UpdateActiveLiq(*big.NewInt(0), 0)
+			pos.Liq.UpdatePostKOLiq(*l.PivotTime, *big.NewInt(0), 0)
+		}
+	}
 }
 
 func (c *ControllerOverNetwork) applyToPassiveLiq(l tables.LiqChange, loc types.PositionLocation) {
@@ -158,20 +215,21 @@ func (c *ControllerOverNetwork) IngestAggEvent(r tables.AggEvent) {
 	hist.NextEvent(r)
 }
 
-func (c *ControllerOverNetwork) IngestKnockout(r tables.KnockoutCross) {
-	liq := types.KnockoutTickLocation(r.Tick, r.IsBid > 0, c.knockoutTickWidth())
+func (c *ControllerOverNetwork) IngestKnockout(l tables.LiqChange) {
+	liq := types.KnockoutTickLocation(l.BidTick, l.IsBid > 0, c.knockoutTickWidth())
 	pool := types.PoolLocation{
 		ChainId: c.chainId,
-		PoolIdx: r.PoolIdx,
-		Base:    types.RequireEthAddr(r.Base),
-		Quote:   types.RequireEthAddr(r.Quote),
+		PoolIdx: l.PoolIdx,
+		Base:    types.RequireEthAddr(l.Base),
+		Quote:   types.RequireEthAddr(l.Quote),
 	}
 	loc := types.BookLocation{
 		PoolLocation:      pool,
 		LiquidityLocation: liq,
 	}
-	pos := c.ctrl.cache.MaterializeKnockoutBook(loc)
-	c.ctrl.workers.omniUpdates <- &koCrossUpdateMsg{loc: loc, pos: pos, cross: r}
+	pos := c.ctrl.cache.MaterializeKnockoutSaga(loc)
+	pos.UpdateCross(l)
+	c.ctrl.workers.omniUpdates <- &koCrossUpdateMsg{loc: loc, pos: pos, cross: l}
 }
 
 /* Called to indicate that all tables have completed the most recent sync cycle up
@@ -188,9 +246,9 @@ func (c *ControllerOverNetwork) knockoutTickWidth() int {
 }
 
 func formLiqLoc(l tables.LiqChange) types.LiquidityLocation {
-	if l.PositionType == "ambient" {
+	if l.PositionType == tables.PosTypeAmbient {
 		return types.AmbientLiquidityLocation()
-	} else if l.PositionType == "knockout" {
+	} else if l.PositionType == tables.PosTypeKnockout {
 		return types.KnockoutRangeLocation(l.BidTick, l.AskTick, l.IsBid > 0)
 	} else {
 		return types.RangeLiquidityLocation(l.BidTick, l.AskTick)
@@ -199,7 +257,7 @@ func formLiqLoc(l tables.LiqChange) types.LiquidityLocation {
 
 func (c *Controller) resyncFullCycle(time int) {
 	for poolLoc, poolPos := range c.cache.RetrieveAllPositions() {
-		if !poolPos.IsEmpty() {
+		if !poolPos.IsEmpty() || poolPos.RefreshTime == 0 {
 			c.workers.omniUpdates <- &posImpactMsg{poolLoc, poolPos, time}
 		}
 	}
@@ -208,9 +266,12 @@ func (c *Controller) resyncFullCycle(time int) {
 const REFRESH_CYCLE_TIME = 30 * 60
 
 func (c *Controller) runPeriodicRefresh() {
+	// To prevent running periodic refreshes while the startup sync is still running
+	c.SpinUntilLiqSync()
 	for {
 		time.Sleep(time.Second * REFRESH_CYCLE_TIME)
 		refreshTime := time.Now().Unix()
+		log.Println("Running full refresh at", refreshTime)
 		c.resyncFullCycle(int(refreshTime))
 	}
 }

controller/handles.go

@@ -3,13 +3,17 @@ package controller
 import (
 	"log"
 	"math/big"
+	"time"
 
 	"github.com/CrocSwap/graphcache-go/loader"
 	"github.com/CrocSwap/graphcache-go/model"
 	"github.com/CrocSwap/graphcache-go/types"
 )
 
 type IRefreshHandle interface {
+	Hash() [32]byte
+	RefreshTime() int64
+	Skippable() bool
 	RefreshQuery(query *loader.ICrocQuery)
 	LabelTag() string
 }
@@ -66,31 +70,33 @@ func (p *KnockoutAliveHandle) RefreshQuery(query *loader.ICrocQuery) {
 	pivotTime := int(tryQueryAttempt(pivotTimeFn, "pivotTimeLatest"))
 
 	if pivotTime == 0 {
-		p.pos.Liq.UpdateActiveLiq(*big.NewInt(0))
+		p.pos.Liq.UpdateActiveLiq(*big.NewInt(0), time.Now().Unix())
 
 	} else {
 		claimLoc := types.KOClaimLocation{PositionLocation: p.location, PivotTime: pivotTime}
-		liqFn := func() (*big.Int, error) { return (*query).QueryKnockoutLiq(claimLoc) }
-
-		knockoutLiq := tryQueryAttempt(liqFn, "knockoutLiq")
-		p.pos.Liq.UpdateActiveLiq(*knockoutLiq)
+		liqFn := func() (loader.KnockoutLiqResp, error) { return (*query).QueryKnockoutLiq(claimLoc) }
+		koLiqResp := tryQueryAttempt(liqFn, "knockoutLiq")
+		p.pos.Liq.UpdateActiveLiq(*koLiqResp.Liq, time.Now().Unix())
 	}
 }
 
 func (p *KnockoutPostHandle) RefreshQuery(query *loader.ICrocQuery) {
-	liqFn := func() (*big.Int, error) { return (*query).QueryKnockoutLiq(p.location) }
-	knockoutLiq := tryQueryAttempt(liqFn, "knockoutLiq")
-	p.pos.Liq.UpdatePostKOLiq(p.location.PivotTime, *knockoutLiq)
+	liqFn := func() (loader.KnockoutLiqResp, error) { return (*query).QueryKnockoutLiq(p.location) }
+	koLiqResp := tryQueryAttempt(liqFn, "knockoutLiq")
+	if koLiqResp.KnockedOut {
+		p.pos.Liq.UpdatePostKOLiq(p.location.PivotTime, *koLiqResp.Liq, time.Now().Unix())
+	}
 }
 
 func tryQueryAttempt[T any](queryFn func() (T, error), label string) T {
 	result, err := queryFn()
 	for retryCount := 0; err != nil && retryCount < N_MAX_RETRIES; retryCount += 1 {
+		log.Printf("Query attempt %d/%d failed for \"%s\" with err: \"%s\"", retryCount, N_MAX_RETRIES, label, err)
 		retryWaitRandom()
 		result, err = queryFn()
 	}
 	if err != nil {
-		log.Fatal("Unable to query liquidity for " + label)
+		log.Fatalf("Unable to query \"%s\", err: %s", label, err)
 	}
 	return result
 }
@@ -110,3 +116,57 @@ func (p *KnockoutAliveHandle) LabelTag() string {
 func (p *KnockoutPostHandle) LabelTag() string {
 	return "knockoutPost"
 }
+
+func (p *PositionRefreshHandle) RefreshTime() int64 {
+	return p.pos.RefreshTime
+}
+
+func (p *RewardsRefreshHandle) RefreshTime() int64 {
+	return p.pos.RefreshTime
+}
+
+func (p *KnockoutAliveHandle) RefreshTime() int64 {
+	return p.pos.Liq.Active.RefreshTime
+}
+
+func (p *KnockoutPostHandle) RefreshTime() int64 {
+	return p.pos.Liq.Active.RefreshTime
+}
+
+func (p *PositionRefreshHandle) Hash() [32]byte {
+	return p.location.Hash()
+}
+
+func (p *RewardsRefreshHandle) Hash() [32]byte {
+	return p.location.Hash()
+}
+
+func (p *KnockoutAliveHandle) Hash() [32]byte {
+	h := p.location.Hash()
+	// Since PositionLocation for regular positions also has an IsBid bool, it's
+	// possible for the hash of a knockout order to collide with a position of
+	// the same user. To avoid this, we increment the first byte of the hash.
+	// Sure wish Go had a non-painful way to make fields optional/nullable.
+	h[0] = byte(h[0] + 1)
+	return h
+}
+
+func (p *KnockoutPostHandle) Hash() [32]byte {
+	return p.location.Hash()
+}
+
+func (p *PositionRefreshHandle) Skippable() bool {
+	return false
+}
+
+func (p *RewardsRefreshHandle) Skippable() bool {
+	return true
+}
+
+func (p *KnockoutAliveHandle) Skippable() bool {
+	return false
+}
+
+func (p *KnockoutPostHandle) Skippable() bool {
+	return false
+}