services/horizon: Add Liquidity Pools to asset stats (#3887)

This commit adds Liquidity Pools to asset stats.

Assets stats should include all data from all ledger entries so the sum of
balances equals the sum of circulating supply of an issued asset.

protocols/horizon/main.go

@@ -166,16 +166,18 @@ type AssetStat struct {
 	} `json:"_links"`
 
 	base.Asset
-	PT                   string            `json:"paging_token"`
-	Accounts             AssetStatAccounts `json:"accounts"`
-	NumClaimableBalances int32             `json:"num_claimable_balances"`
+	PT string `json:"paging_token"`
+	// Action needed in release: horizon-v3.0.0: deprecated field
+	NumAccounts          int32 `json:"num_accounts"`
+	NumClaimableBalances int32 `json:"num_claimable_balances"`
+	NumLiquidityPools    int32 `json:"num_liquidity_pools"`
 	// Action needed in release: horizon-v3.0.0: deprecated field
 	Amount                  string            `json:"amount"`
-	Balances                AssetStatBalances `json:"balances"`
+	Accounts                AssetStatAccounts `json:"accounts"`
 	ClaimableBalancesAmount string            `json:"claimable_balances_amount"`
-	// Action needed in release: horizon-v3.0.0: deprecated field
-	NumAccounts int32        `json:"num_accounts"`
-	Flags       AccountFlags `json:"flags"`
+	LiquidityPoolsAmount    string            `json:"liquidity_pools_amount"`
+	Balances                AssetStatBalances `json:"balances"`
+	Flags                   AccountFlags      `json:"flags"`
 }
 
 // PagingToken implementation for hal.Pageable

services/horizon/internal/actions/asset.go

@@ -155,12 +155,15 @@ func (handler AssetStatsHandler) GetResourcePage(
 	for _, record := range assetStats {
 		var assetStatResponse horizon.AssetStat
 
-		resourceadapter.PopulateAssetStat(
+		err := resourceadapter.PopulateAssetStat(
 			ctx,
 			&assetStatResponse,
 			record,
 			issuerAccounts[record.AssetIssuer],
 		)
+		if err != nil {
+			return nil, err
+		}
 		response = append(response, assetStatResponse)
 	}
 

services/horizon/internal/actions/asset_test.go

@@ -130,12 +130,14 @@ func TestAssetStats(t *testing.T) {
 			AuthorizedToMaintainLiabilities: 3,
 			Unauthorized:                    4,
 			ClaimableBalances:               1,
+			LiquidityPools:                  5,
 		},
 		Balances: history.ExpAssetStatBalances{
 			Authorized:                      "1",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
 			ClaimableBalances:               "10",
+			LiquidityPools:                  "20",
 		},
 		Amount:      "1",
 		NumAccounts: 2,
@@ -147,12 +149,14 @@ func TestAssetStats(t *testing.T) {
 			Unauthorized:                    usdAssetStat.Accounts.Unauthorized,
 		},
 		NumClaimableBalances: usdAssetStat.Accounts.ClaimableBalances,
+		NumLiquidityPools:    usdAssetStat.Accounts.LiquidityPools,
 		Balances: horizon.AssetStatBalances{
 			Authorized:                      "0.0000001",
 			AuthorizedToMaintainLiabilities: "0.0000002",
 			Unauthorized:                    "0.0000003",
 		},
 		ClaimableBalancesAmount: "0.0000010",
+		LiquidityPoolsAmount:    "0.0000020",
 		Amount:                  "0.0000001",
 		NumAccounts:             usdAssetStat.NumAccounts,
 		Asset: base.Asset{
@@ -179,6 +183,7 @@ func TestAssetStats(t *testing.T) {
 			AuthorizedToMaintainLiabilities: "46",
 			Unauthorized:                    "92",
 			ClaimableBalances:               "0",
+			LiquidityPools:                  "0",
 		},
 		Amount:      "23",
 		NumAccounts: 1,
@@ -196,6 +201,7 @@ func TestAssetStats(t *testing.T) {
 			Unauthorized:                    "0.0000092",
 		},
 		ClaimableBalancesAmount: "0.0000000",
+		LiquidityPoolsAmount:    "0.0000000",
 		Amount:                  "0.0000023",
 		NumAccounts:             etherAssetStat.NumAccounts,
 		Asset: base.Asset{
@@ -222,6 +228,7 @@ func TestAssetStats(t *testing.T) {
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
 			ClaimableBalances:               "0",
+			LiquidityPools:                  "0",
 		},
 		Amount:      "1",
 		NumAccounts: 2,
@@ -239,6 +246,7 @@ func TestAssetStats(t *testing.T) {
 			Unauthorized:                    "0.0000003",
 		},
 		ClaimableBalancesAmount: "0.0000000",
+		LiquidityPoolsAmount:    "0.0000000",
 		Amount:                  "0.0000001",
 		NumAccounts:             otherUSDAssetStat.NumAccounts,
 		Asset: base.Asset{
@@ -267,6 +275,7 @@ func TestAssetStats(t *testing.T) {
 			AuthorizedToMaintainLiabilities: "222",
 			Unauthorized:                    "333",
 			ClaimableBalances:               "0",
+			LiquidityPools:                  "0",
 		},
 		Amount:      "111",
 		NumAccounts: 3,
@@ -284,6 +293,7 @@ func TestAssetStats(t *testing.T) {
 			Unauthorized:                    "0.0000333",
 		},
 		ClaimableBalancesAmount: "0.0000000",
+		LiquidityPoolsAmount:    "0.0000000",
 		Amount:                  "0.0000111",
 		NumAccounts:             eurAssetStat.NumAccounts,
 		Asset: base.Asset{
@@ -467,6 +477,7 @@ func TestAssetStatsIssuerDoesNotExist(t *testing.T) {
 			Unauthorized:                    "0.0000003",
 		},
 		ClaimableBalancesAmount: "0.0000000",
+		LiquidityPoolsAmount:    "0.0000000",
 		Amount:                  "0.0000001",
 		NumAccounts:             usdAssetStat.NumAccounts,
 		Asset: base.Asset{

services/horizon/internal/db2/history/asset_stats_test.go

@@ -30,6 +30,8 @@ func TestInsertAssetStats(t *testing.T) {
 				Authorized:                      "1",
 				AuthorizedToMaintainLiabilities: "2",
 				Unauthorized:                    "3",
+				ClaimableBalances:               "4",
+				LiquidityPools:                  "5",
 			},
 			Amount:      "1",
 			NumAccounts: 2,
@@ -47,6 +49,8 @@ func TestInsertAssetStats(t *testing.T) {
 				Authorized:                      "23",
 				AuthorizedToMaintainLiabilities: "2",
 				Unauthorized:                    "3",
+				ClaimableBalances:               "4",
+				LiquidityPools:                  "5",
 			},
 			Amount:      "23",
 			NumAccounts: 1,
@@ -81,6 +85,8 @@ func TestInsertAssetStat(t *testing.T) {
 				Authorized:                      "1",
 				AuthorizedToMaintainLiabilities: "2",
 				Unauthorized:                    "3",
+				ClaimableBalances:               "4",
+				LiquidityPools:                  "5",
 			},
 			Amount:      "1",
 			NumAccounts: 2,
@@ -98,6 +104,8 @@ func TestInsertAssetStat(t *testing.T) {
 				Authorized:                      "23",
 				AuthorizedToMaintainLiabilities: "2",
 				Unauthorized:                    "3",
+				ClaimableBalances:               "4",
+				LiquidityPools:                  "5",
 			},
 			Amount:      "23",
 			NumAccounts: 1,
@@ -134,6 +142,8 @@ func TestInsertAssetStatAlreadyExistsError(t *testing.T) {
 			Authorized:                      "1",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
+			ClaimableBalances:               "4",
+			LiquidityPools:                  "5",
 		},
 		Amount:      "1",
 		NumAccounts: 2,
@@ -179,6 +189,8 @@ func TestUpdateAssetStatDoesNotExistsError(t *testing.T) {
 			Authorized:                      "1",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
+			ClaimableBalances:               "4",
+			LiquidityPools:                  "5",
 		},
 		Amount:      "1",
 		NumAccounts: 2,
@@ -212,6 +224,8 @@ func TestUpdateStat(t *testing.T) {
 			Authorized:                      "1",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
+			ClaimableBalances:               "4",
+			LiquidityPools:                  "5",
 		},
 		Amount:      "1",
 		NumAccounts: 2,
@@ -256,6 +270,8 @@ func TestGetAssetStatDoesNotExist(t *testing.T) {
 			Authorized:                      "1",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
+			ClaimableBalances:               "4",
+			LiquidityPools:                  "5",
 		},
 		Amount:      "1",
 		NumAccounts: 2,
@@ -285,6 +301,8 @@ func TestRemoveAssetStat(t *testing.T) {
 			Authorized:                      "1",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
+			ClaimableBalances:               "4",
+			LiquidityPools:                  "5",
 		},
 		Amount:      "1",
 		NumAccounts: 2,
@@ -419,6 +437,8 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			Authorized:                      "1",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
+			ClaimableBalances:               "0",
+			LiquidityPools:                  "0",
 		},
 		Amount:      "1",
 		NumAccounts: 2,
@@ -436,6 +456,8 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			Authorized:                      "23",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
+			ClaimableBalances:               "0",
+			LiquidityPools:                  "0",
 		},
 		Amount:      "23",
 		NumAccounts: 1,
@@ -453,6 +475,8 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			Authorized:                      "1",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
+			ClaimableBalances:               "0",
+			LiquidityPools:                  "0",
 		},
 		Amount:      "1",
 		NumAccounts: 2,
@@ -470,6 +494,8 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			Authorized:                      "111",
 			AuthorizedToMaintainLiabilities: "2",
 			Unauthorized:                    "3",
+			ClaimableBalances:               "1",
+			LiquidityPools:                  "2",
 		},
 		Amount:      "111",
 		NumAccounts: 3,

services/horizon/internal/db2/history/main.go

@@ -371,6 +371,7 @@ type ExpAssetStatAccounts struct {
 	Authorized                      int32 `json:"authorized"`
 	AuthorizedToMaintainLiabilities int32 `json:"authorized_to_maintain_liabilities"`
 	ClaimableBalances               int32 `json:"claimable_balances"`
+	LiquidityPools                  int32 `json:"liquidity_pools"`
 	Unauthorized                    int32 `json:"unauthorized"`
 }
 
@@ -392,6 +393,7 @@ func (a ExpAssetStatAccounts) Add(b ExpAssetStatAccounts) ExpAssetStatAccounts {
 		Authorized:                      a.Authorized + b.Authorized,
 		AuthorizedToMaintainLiabilities: a.AuthorizedToMaintainLiabilities + b.AuthorizedToMaintainLiabilities,
 		ClaimableBalances:               a.ClaimableBalances + b.ClaimableBalances,
+		LiquidityPools:                  a.LiquidityPools + b.LiquidityPools,
 		Unauthorized:                    a.Unauthorized + b.Unauthorized,
 	}
 }
@@ -401,10 +403,12 @@ func (a ExpAssetStatAccounts) IsZero() bool {
 }
 
 // ExpAssetStatBalances represents the summarized balances for a single Asset
+// Note: the string representation is in stroops!
 type ExpAssetStatBalances struct {
 	Authorized                      string `json:"authorized"`
 	AuthorizedToMaintainLiabilities string `json:"authorized_to_maintain_liabilities"`
 	ClaimableBalances               string `json:"claimable_balances"`
+	LiquidityPools                  string `json:"liquidity_pools"`
 	Unauthorized                    string `json:"unauthorized"`
 }
 
@@ -418,7 +422,29 @@ func (e *ExpAssetStatBalances) Scan(src interface{}) error {
 		return errors.New("Type assertion .([]byte) failed.")
 	}
 
-	return json.Unmarshal(source, &e)
+	err := json.Unmarshal(source, &e)
+	if err != nil {
+		return err
+	}
+
+	// Sets zero values for empty balances
+	if e.Authorized == "" {
+		e.Authorized = "0"
+	}
+	if e.AuthorizedToMaintainLiabilities == "" {
+		e.AuthorizedToMaintainLiabilities = "0"
+	}
+	if e.ClaimableBalances == "" {
+		e.ClaimableBalances = "0"
+	}
+	if e.LiquidityPools == "" {
+		e.LiquidityPools = "0"
+	}
+	if e.Unauthorized == "" {
+		e.Unauthorized = "0"
+	}
+
+	return nil
 }
 
 // QAssetStats defines exp_asset_stats related queries.

services/horizon/internal/ingest/processors/asset_stats_processor.go

@@ -41,7 +41,9 @@ func (p *AssetStatsProcessor) reset() {
 }
 
 func (p *AssetStatsProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
-	if change.Type != xdr.LedgerEntryTypeClaimableBalance && change.Type != xdr.LedgerEntryTypeTrustline {
+	if change.Type != xdr.LedgerEntryTypeLiquidityPool &&
+		change.Type != xdr.LedgerEntryTypeClaimableBalance &&
+		change.Type != xdr.LedgerEntryTypeTrustline {
 		return nil
 	}
 	if p.useLedgerEntryCache {
@@ -52,6 +54,8 @@ func (p *AssetStatsProcessor) ProcessChange(ctx context.Context, change ingest.C
 	}
 
 	switch change.Type {
+	case xdr.LedgerEntryTypeLiquidityPool:
+		return p.assetStatSet.AddLiquidityPool(change)
 	case xdr.LedgerEntryTypeClaimableBalance:
 		return p.assetStatSet.AddClaimableBalance(change)
 	case xdr.LedgerEntryTypeTrustline:
@@ -90,6 +94,8 @@ func (p *AssetStatsProcessor) Commit(ctx context.Context) error {
 	for _, change := range changes {
 		var err error
 		switch change.Type {
+		case xdr.LedgerEntryTypeLiquidityPool:
+			err = p.assetStatSet.AddLiquidityPool(change)
 		case xdr.LedgerEntryTypeClaimableBalance:
 			err = p.assetStatSet.AddClaimableBalance(change)
 		case xdr.LedgerEntryTypeTrustline:
@@ -144,7 +150,14 @@ func (p *AssetStatsProcessor) Commit(ctx context.Context) error {
 				))
 			} else if delta.Accounts.ClaimableBalances < 0 {
 				return ingest.NewStateError(errors.Errorf(
-					"Claimable balance accounts negative but DB entry does not exist for asset: %s %s %s",
+					"Claimable balance negative but DB entry does not exist for asset: %s %s %s",
+					delta.AssetType,
+					delta.AssetCode,
+					delta.AssetIssuer,
+				))
+			} else if delta.Accounts.LiquidityPools < 0 {
+				return ingest.NewStateError(errors.Errorf(
+					"Liquidity pools negative but DB entry does not exist for asset: %s %s %s",
 					delta.AssetType,
 					delta.AssetCode,
 					delta.AssetIssuer,