services/horizon/internal/expingest/processors: Only process authoriz…

…ed trustlines when updating asset stats (#1936) The experimental ingestion pipeline has a bug in that it counts unauthorized trustlines when computing asset stats. Only authorized trustlines should be included. Another issue fixed in this commit is that the cursor for the new asset stats endpoint diverges from the existing asset stats endpoint
stellar · Nov 15, 2019 · d222515 · d222515
1 parent e402e29
commit d222515
Show file tree

Hide file tree

Showing 9 changed files with 416 additions and 48 deletions.
diff --git a/services/horizon/internal/actions/asset.go b/services/horizon/internal/actions/asset.go
@@ -39,15 +39,15 @@ func (handler AssetStatsHandler) validateAssetParams(code, issuer string, pq db2
 	}
 
 	if pq.Cursor != "" {
-		parts := strings.Split(pq.Cursor, ":")
-		if len(parts) != 2 {
+		parts := strings.SplitN(pq.Cursor, "_", 3)
+		if len(parts) != 3 {
 			return problem.MakeInvalidFieldProblem(
 				"cursor",
 				errors.New("cursor must contain exactly one colon"),
 			)
 		}
 
-		cursorCode, cursorIssuer := parts[0], parts[1]
+		cursorCode, cursorIssuer, assetType := parts[0], parts[1], parts[2]
 		if !xdr.ValidAssetCode.MatchString(cursorCode) {
 			return problem.MakeInvalidFieldProblem(
 				"cursor",
@@ -61,6 +61,14 @@ func (handler AssetStatsHandler) validateAssetParams(code, issuer string, pq db2
 				fmt.Errorf("%s is not a valid asset issuer", cursorIssuer),
 			)
 		}
+
+		if _, ok := xdr.StringToAssetType[assetType]; !ok {
+			return problem.MakeInvalidFieldProblem(
+				"cursor",
+				fmt.Errorf("%s is not a valid asset type", assetType),
+			)
+		}
+
 	}
 
 	return nil

diff --git a/services/horizon/internal/actions/asset_test.go b/services/horizon/internal/actions/asset_test.go
@@ -40,29 +40,37 @@ func TestAssetStatsValidation(t *testing.T) {
 			"not a valid asset issuer",
 		},
 		{
-			"cursor has too many colons",
+			"cursor has too many underscores",
 			map[string]string{
-				"cursor": "ABC:GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H:",
+				"cursor": "ABC_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum4_",
 			},
 			"cursor",
-			"cursor must contain exactly one colon",
+			"credit_alphanum4_ is not a valid asset type",
 		},
 		{
 			"invalid cursor code",
 			map[string]string{
-				"cursor": "tooooooooolong:GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H",
+				"cursor": "tooooooooolong_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum12",
 			},
 			"cursor",
 			"not a valid asset code",
 		},
 		{
 			"invalid cursor issuer",
 			map[string]string{
-				"cursor": "ABC:invalidissuer",
+				"cursor": "ABC_invalidissuer_credit_alphanum4",
 			},
 			"cursor",
 			"not a valid asset issuer",
 		},
+		{
+			"invalid cursor type",
+			map[string]string{
+				"cursor": "ABC_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum123",
+			},
+			"cursor",
+			"credit_alphanum123 is not a valid asset type",
+		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
 			r := makeRequest(t, testCase.queryParams, map[string]string{}, nil)
@@ -124,7 +132,7 @@ func TestAssetStats(t *testing.T) {
 			Code:   usdAssetStat.AssetCode,
 			Issuer: usdAssetStat.AssetIssuer,
 		},
-		PT:    usdAssetStat.AssetCode + ":" + usdAssetStat.AssetIssuer,
+		PT:    usdAssetStat.PagingToken(),
 		Flags: issuerFlags,
 	}
 
@@ -143,7 +151,7 @@ func TestAssetStats(t *testing.T) {
 			Code:   etherAssetStat.AssetCode,
 			Issuer: etherAssetStat.AssetIssuer,
 		},
-		PT:    etherAssetStat.AssetCode + ":" + etherAssetStat.AssetIssuer,
+		PT:    etherAssetStat.PagingToken(),
 		Flags: issuerFlags,
 	}
 
@@ -162,7 +170,7 @@ func TestAssetStats(t *testing.T) {
 			Code:   otherUSDAssetStat.AssetCode,
 			Issuer: otherUSDAssetStat.AssetIssuer,
 		},
-		PT: otherUSDAssetStat.AssetCode + ":" + otherUSDAssetStat.AssetIssuer,
+		PT: otherUSDAssetStat.PagingToken(),
 	}
 	otherUSDAssetStatResponse.Links.Toml = hal.NewLink(
 		"https://" + otherIssuer.HomeDomain + "/.well-known/stellar.toml",
@@ -183,7 +191,7 @@ func TestAssetStats(t *testing.T) {
 			Code:   eurAssetStat.AssetCode,
 			Issuer: eurAssetStat.AssetIssuer,
 		},
-		PT: eurAssetStat.AssetCode + ":" + eurAssetStat.AssetIssuer,
+		PT: eurAssetStat.PagingToken(),
 	}
 	eurAssetStatResponse.Links.Toml = hal.NewLink(
 		"https://" + otherIssuer.HomeDomain + "/.well-known/stellar.toml",
@@ -346,7 +354,7 @@ func TestAssetStatsIssuerDoesNotExist(t *testing.T) {
 			Code:   usdAssetStat.AssetCode,
 			Issuer: usdAssetStat.AssetIssuer,
 		},
-		PT: usdAssetStat.AssetCode + ":" + usdAssetStat.AssetIssuer,
+		PT: usdAssetStat.PagingToken(),
 	}
 
 	tt.Assert.Len(results, 1)

diff --git a/services/horizon/internal/db2/history/asset_stats.go b/services/horizon/internal/db2/history/asset_stats.go
@@ -104,12 +104,12 @@ func (q *Q) GetAssetStat(assetType xdr.AssetType, assetCode, assetIssuer string)
 }
 
 func parseAssetStatsCursor(cursor string) (string, string, error) {
-	parts := strings.Split(cursor, ":")
-	if len(parts) != 2 {
+	parts := strings.SplitN(cursor, "_", 3)
+	if len(parts) != 3 {
 		return "", "", fmt.Errorf("invalid asset stats cursor: %v", cursor)
 	}
 
-	code, issuer := parts[0], parts[1]
+	code, issuer, assetType := parts[0], parts[1], parts[2]
 	var issuerAccount xdr.AccountId
 	var asset xdr.Asset
 
@@ -127,6 +127,10 @@ func parseAssetStatsCursor(cursor string) (string, string, error) {
 		)
 	}
 
+	if _, ok := xdr.StringToAssetType[assetType]; !ok {
+		return "", "", errors.Errorf("invalid asset type in asset stats cursor: %v", cursor)
+	}
+
 	return code, issuer, nil
 }
 

diff --git a/services/horizon/internal/db2/history/asset_stats_test.go b/services/horizon/internal/db2/history/asset_stats_test.go
@@ -241,28 +241,33 @@ func TestGetAssetStatsCursorValidation(t *testing.T) {
 		expectedError string
 	}{
 		{
-			"cursor does not use colon as serpator",
+			"cursor does not use underscore as serpator",
 			"usdc-GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H",
 			"invalid asset stats cursor",
 		},
 		{
-			"cursor has no colon",
+			"cursor has no underscore",
 			"usdc",
 			"invalid asset stats cursor",
 		},
 		{
-			"cursor has too many colons",
-			"usdc:GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H:",
-			"invalid asset stats cursor",
+			"cursor has too many underscores",
+			"usdc_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum4_",
+			"invalid asset type in asset stats cursor",
 		},
 		{
 			"issuer in cursor is invalid",
-			"usd:abcdefghijklmnopqrstuv",
+			"usd_abcdefghijklmnopqrstuv_credit_alphanum4",
 			"invalid issuer in asset stats cursor",
 		},
+		{
+			"asset type in cursor is invalid",
+			"usd_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum",
+			"invalid asset type in asset stats cursor",
+		},
 		{
 			"asset code in cursor is too long",
-			"abcdefghijklmnopqrstuv:GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H",
+			"abcdefghijklmnopqrstuv_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum12",
 			"invalid asset stats cursor",
 		},
 	} {
@@ -376,7 +381,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"no filter with cursor",
 			"",
 			"",
-			"ABC:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"ABC_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"asc",
 			[]ExpAssetStat{
 				etherAssetStat,
@@ -389,7 +394,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"no filter with cursor descending",
 			"",
 			"",
-			"ZZZ:GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H",
+			"ZZZ_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum4",
 			"desc",
 			[]ExpAssetStat{
 				usdAssetStat,
@@ -402,7 +407,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"no filter with cursor and offset",
 			"",
 			"",
-			"ETHER:GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H",
+			"ETHER_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum12",
 			"asc",
 			[]ExpAssetStat{
 				eurAssetStat,
@@ -414,7 +419,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"no filter with cursor and offset descending",
 			"",
 			"",
-			"EUR:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"EUR_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"desc",
 			[]ExpAssetStat{
 				etherAssetStat,
@@ -424,7 +429,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"no filter with cursor and offset descending including eur",
 			"",
 			"",
-			"EUR:GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H",
+			"EUR_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum4",
 			"desc",
 			[]ExpAssetStat{
 				eurAssetStat,
@@ -446,7 +451,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"filter on code with cursor",
 			"USD",
 			"",
-			"USD:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"USD_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"asc",
 			[]ExpAssetStat{
 				usdAssetStat,
@@ -456,7 +461,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"filter on code with cursor descending",
 			"USD",
 			"",
-			"USD:GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H",
+			"USD_GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H_credit_alphanum4",
 			"desc",
 			[]ExpAssetStat{
 				otherUSDAssetStat,
@@ -477,7 +482,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"filter on issuer with cursor",
 			"",
 			"GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
-			"EUR:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"EUR_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"asc",
 			[]ExpAssetStat{
 				otherUSDAssetStat,
@@ -487,7 +492,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"filter on issuer with cursor descending",
 			"",
 			"GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
-			"USD:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"USD_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"desc",
 			[]ExpAssetStat{
 				eurAssetStat,
@@ -505,7 +510,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"filter on non existant code with cursor",
 			"BTC",
 			"",
-			"BTC:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"BTC_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"asc",
 			nil,
 		},
@@ -521,7 +526,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"filter on non existant issuer with cursor",
 			"",
 			"GAEIHD6U4WSBHJGA2HPWOQ3OQEFQ3Y7QZE2DR76YKZNKPW5YDLYW4UGF",
-			"AAA:GAEIHD6U4WSBHJGA2HPWOQ3OQEFQ3Y7QZE2DR76YKZNKPW5YDLYW4UGF",
+			"AAA_GAEIHD6U4WSBHJGA2HPWOQ3OQEFQ3Y7QZE2DR76YKZNKPW5YDLYW4UGF_credit_alphanum4",
 			"asc",
 			nil,
 		},
@@ -537,7 +542,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"filter on non existant code and non existant issuer with cursor",
 			"BTC",
 			"GAEIHD6U4WSBHJGA2HPWOQ3OQEFQ3Y7QZE2DR76YKZNKPW5YDLYW4UGF",
-			"AAA:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"AAA_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"asc",
 			nil,
 		},
@@ -555,7 +560,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"filter on both code and issuer with cursor",
 			"USD",
 			"GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
-			"USC:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"USC_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"asc",
 			[]ExpAssetStat{
 				otherUSDAssetStat,
@@ -565,7 +570,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"filter on both code and issuer with cursor descending",
 			"USD",
 			"GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
-			"USE:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"USE_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"desc",
 			[]ExpAssetStat{
 				otherUSDAssetStat,
@@ -575,7 +580,7 @@ func TestGetAssetStatsFiltersAndCursor(t *testing.T) {
 			"cursor negates filter",
 			"USD",
 			"GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
-			"USD:GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2",
+			"USD_GA5WBPYA5Y4WAEHXWR2UKO2UO4BUGHUQ74EUPKON2QHV4WRHOIRNKKH2_credit_alphanum4",
 			"asc",
 			nil,
 		},

diff --git a/services/horizon/internal/db2/history/main.go b/services/horizon/internal/db2/history/main.go
@@ -242,7 +242,12 @@ type ExpAssetStat struct {
 
 // PagingToken returns a cursor for this asset stat
 func (e ExpAssetStat) PagingToken() string {
-	return fmt.Sprintf("%s:%s", e.AssetCode, e.AssetIssuer)
+	return fmt.Sprintf(
+		"%s_%s_%s",
+		e.AssetCode,
+		e.AssetIssuer,
+		xdr.AssetTypeToString[e.AssetType],
+	)
 }
 
 // QAssetStats defines exp_asset_stats related queries.

diff --git a/services/horizon/internal/expingest/processors/asset_stats_set.go b/services/horizon/internal/expingest/processors/asset_stats_set.go
@@ -22,7 +22,12 @@ type assetStatValue struct {
 type AssetStatSet map[assetStatKey]*assetStatValue
 
 // Add updates the set with a trustline entry from a history archive snapshot
+// if the trustline is authorized
 func (s AssetStatSet) Add(trustLine xdr.TrustLineEntry) error {
+	if !xdr.TrustLineFlags(trustLine.Flags).IsAuthorized() {
+		return nil
+	}
+
 	var key assetStatKey
 	if err := trustLine.Asset.Extract(&key.assetType, &key.assetCode, &key.assetIssuer); err != nil {
 		return errors.Wrap(err, "could not extract asset info from trustline")