Merge pull request #286 from vegaprotocol/fix-difftool

fix: implement pagination for accounts and deposits

CHANGELOG.md

@@ -81,6 +81,7 @@
 - [173](https://github.com/vegaprotocol/vegatools/issues/173) - Add `MarketTracker` to checkpoint parser
 - [262](https://github.com/vegaprotocol/vegatools/issues/262) - Update vega dependency, ignore slippage factors in market and last block in epoch
 - [279](https://github.com/vegaprotocol/vegatools/issues/279) - Update to work with new datanode API for querying live orders / liquidity provisions
+- [286](https://github.com/vegaprotocol/vegatools/pull/286) - Add support for pagination in the accounts and delegations queries to data-node
 
 ## 0.41.1
 *2021-08-31*

difftool/diff/datanode_client.go

@@ -207,15 +207,36 @@ func (dnc *dataNodeClient) Collect() (*Result, error) {
 }
 
 func (dnc *dataNodeClient) listAccounts() ([]*dn.AccountBalance, error) {
-	accResp, err := dnc.datanode.ListAccounts(context.Background(), &dn.ListAccountsRequest{})
-	if err != nil {
-		return nil, err
-	}
-	accounts := make([]*dn.AccountBalance, 0, len(accResp.Accounts.Edges))
-	for _, ae := range accResp.Accounts.Edges {
-		accounts = append(accounts, ae.Node)
+	result := []*dn.AccountBalance{}
+
+	lastRecordCursor := ""
+	for {
+		requestInput := dn.ListAccountsRequest{}
+		if len(lastRecordCursor) > 0 {
+			requestInput = dn.ListAccountsRequest{
+				Pagination: &dn.Pagination{
+					After: &lastRecordCursor,
+				},
+			}
+		}
+
+		accResp, err := dnc.datanode.ListAccounts(context.Background(), &requestInput)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, ae := range accResp.Accounts.Edges {
+			result = append(result, ae.Node)
+		}
+
+		if accResp.Accounts == nil || accResp.Accounts.PageInfo == nil || !accResp.Accounts.PageInfo.HasNextPage {
+			break
+		}
+
+		lastRecordCursor = accResp.Accounts.PageInfo.EndCursor
 	}
-	return accounts, nil
+
+	return result, nil
 }
 
 func (dnc *dataNodeClient) listOrders() ([]*vega.Order, error) {
@@ -291,15 +312,36 @@ func (dnc *dataNodeClient) getVegaTime() (int64, error) {
 }
 
 func (dnc *dataNodeClient) listDelegations() ([]*vega.Delegation, error) {
-	delegationResp, err := dnc.datanode.ListDelegations(context.Background(), &dn.ListDelegationsRequest{})
-	if err != nil {
-		return nil, err
-	}
-	delegations := make([]*vega.Delegation, 0, len(delegationResp.Delegations.Edges))
-	for _, d := range delegationResp.Delegations.Edges {
-		delegations = append(delegations, d.Node)
+	result := []*vega.Delegation{}
+
+	lastRecordCursor := ""
+	for {
+		requestInput := dn.ListDelegationsRequest{}
+		if len(lastRecordCursor) > 0 {
+			requestInput = dn.ListDelegationsRequest{
+				Pagination: &dn.Pagination{
+					After: &lastRecordCursor,
+				},
+			}
+		}
+
+		delegationResp, err := dnc.datanode.ListDelegations(context.Background(), &requestInput)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, d := range delegationResp.Delegations.Edges {
+			result = append(result, d.Node)
+		}
+
+		if delegationResp.Delegations == nil || delegationResp.Delegations.PageInfo == nil || !delegationResp.Delegations.PageInfo.HasNextPage {
+			break
+		}
+
+		lastRecordCursor = delegationResp.Delegations.PageInfo.EndCursor
 	}
-	return delegations, nil
+
+	return result, nil
 }
 
 func (dnc *dataNodeClient) getEpoch() (*vega.Epoch, error) {
@@ -421,7 +463,7 @@ func (dnc *dataNodeClient) listLiquidityProvisions(market string) ([]*vega.Liqui
 	}
 	lps := make([]*vega.LiquidityProvision, 0, len(resp.LiquidityProvisions.Edges))
 	for _, lpe := range resp.LiquidityProvisions.Edges {
-		if lpe.Node.Status == vega.LiquidityProvision_STATUS_PENDING ||  lpe.Node.Status == vega.LiquidityProvision_STATUS_ACTIVE || lpe.Node.Status == vega.LiquidityProvision_STATUS_UNDEPLOYED {
+		if lpe.Node.Status == vega.LiquidityProvision_STATUS_PENDING || lpe.Node.Status == vega.LiquidityProvision_STATUS_ACTIVE || lpe.Node.Status == vega.LiquidityProvision_STATUS_UNDEPLOYED {
 			lps = append(lps, lpe.Node)
 		}
 	}

difftool/diff/diff_report.go

@@ -521,7 +521,22 @@ func diffTransfers(coreSnapshot *Result, dn *Result) Status {
 
 	for i, a := range core {
 		d := datanode[i]
-		if a.String() != d.String() {
+		coreTransferTimestamp := a.Timestamp
+		dnTransferTimestamp := d.Timestamp
+
+		// We ignore the difference in the timestamp, as the transfer timestamp
+		// is updated when the network is started from the checkpoint.
+		a.Timestamp = 0
+		d.Timestamp = 0
+
+		transfetMatch := a.String() == d.String()
+
+		// restore the original value as copy is not trivial due to Mutex
+		// inside of the proto message
+		a.Timestamp = coreTransferTimestamp
+		d.Timestamp = dnTransferTimestamp
+
+		if !transfetMatch {
 			return getValueMismatchStatus("transfers", core, datanode)
 		}
 	}