State table verification

.gitignore

@@ -5,11 +5,13 @@
 /.vscode/*.sql
 /.vscode/settings.json
 /.vscode/launch.json
+/captive-core*
 .idea
 debug
 .bundle
 *.swp
 *.crt
 *.csr
 *.key
+*.txt
 stellar-etl

cmd/export_ledger_entry_changes.go

@@ -1,6 +1,7 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 	"math"
 	"os"
@@ -11,8 +12,11 @@ import (
 	"github.com/stellar/stellar-etl/internal/input"
 	"github.com/stellar/stellar-etl/internal/transform"
 	"github.com/stellar/stellar-etl/internal/utils"
+	"github.com/stellar/stellar-etl/internal/utils/verify"
 )
 
+const verifyBatchSize = 50000
+
 var exportLedgerEntryChangesCmd = &cobra.Command{
 	Use:   "export_ledger_entry_changes",
 	Short: "This command exports the changes in accounts, offers, trustlines and liquidity pools.",
@@ -29,10 +33,11 @@ be exported.`,
 		endNum, strictExport, isTest, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
 		env := utils.GetEnvironmentDetails(isTest)
-
+		archive, _ := utils.CreateHistoryArchiveClient(env.ArchiveURLs)
 		execPath, configPath, startNum, batchSize, outputFolder := utils.MustCoreFlags(cmd.Flags(), cmdLogger)
 		exportAccounts, exportOffers, exportTrustlines, exportPools, exportBalances := utils.MustExportTypeFlags(cmd.Flags(), cmdLogger)
 		gcsBucket, gcpCredentials := utils.MustGcsFlags(cmd.Flags(), cmdLogger)
+		ctx := context.Background()
 
 		err := os.MkdirAll(outputFolder, os.ModePerm)
 		if err != nil {
@@ -71,6 +76,7 @@ be exported.`,
 			endNum = math.MaxInt32
 		}
 
+		verifyOutputs := make(map[uint32]transform.TransformedOutput, endNum)
 		changeChan := make(chan input.ChangeBatch)
 		closeChan := make(chan int)
 		go input.StreamChanges(core, startNum, endNum, batchSize, changeChan, closeChan, env, cmdLogger)
@@ -91,77 +97,120 @@ be exported.`,
 					"trustlines":         {},
 					"liquidity_pools":    {},
 				}
+
 				for entryType, changes := range batch.Changes {
 					switch entryType {
 					case xdr.LedgerEntryTypeAccount:
 						for _, change := range changes {
+							entry, _, _, _ := utils.ExtractEntryFromChange(change)
 							if changed, err := change.AccountChangedExceptSigners(); err != nil {
 								cmdLogger.LogError(fmt.Errorf("unable to identify changed accounts: %v", err))
 								continue
 							} else if changed {
 								acc, err := transform.TransformAccount(change)
 								if err != nil {
-									entry, _, _, _ := utils.ExtractEntryFromChange(change)
 									cmdLogger.LogError(fmt.Errorf("error transforming account entry last updated at %d: %s", entry.LastModifiedLedgerSeq, err))
 									continue
 								}
 								transformedOutputs["accounts"] = append(transformedOutputs["accounts"], acc)
+
+								if ok, actualLedger := utils.LedgerIsCheckpoint(entry.LastModifiedLedgerSeq); ok {
+									x := verifyOutputs[actualLedger]
+									x.Accounts = append(x.Accounts, acc)
+									verifyOutputs[actualLedger] = x
+								}
 							}
 							if change.AccountSignersChanged() {
 								signers, err := transform.TransformSigners(change)
 								if err != nil {
-									entry, _, _, _ := utils.ExtractEntryFromChange(change)
 									cmdLogger.LogError(fmt.Errorf("error transforming account signers from %d :%s", entry.LastModifiedLedgerSeq, err))
 									continue
 								}
 								for _, s := range signers {
 									transformedOutputs["signers"] = append(transformedOutputs["signers"], s)
+
+									if ok, actualLedger := utils.LedgerIsCheckpoint(entry.LastModifiedLedgerSeq); ok {
+										x := verifyOutputs[actualLedger]
+										x.Signers = append(x.Signers, s)
+										verifyOutputs[actualLedger] = x
+									}
 								}
 							}
 						}
 					case xdr.LedgerEntryTypeClaimableBalance:
 						for _, change := range changes {
+							entry, _, _, _ := utils.ExtractEntryFromChange(change)
 							balance, err := transform.TransformClaimableBalance(change)
 							if err != nil {
-								entry, _, _, _ := utils.ExtractEntryFromChange(change)
 								cmdLogger.LogError(fmt.Errorf("error transforming balance entry last updated at %d: %s", entry.LastModifiedLedgerSeq, err))
 								continue
 							}
 							transformedOutputs["claimable_balances"] = append(transformedOutputs["claimable_balances"], balance)
+
+							if ok, actualLedger := utils.LedgerIsCheckpoint(entry.LastModifiedLedgerSeq); ok {
+								x := verifyOutputs[actualLedger]
+								x.Claimable_balances = append(x.Claimable_balances, balance)
+								verifyOutputs[actualLedger] = x
+							}
 						}
 					case xdr.LedgerEntryTypeOffer:
 						for _, change := range changes {
+							entry, _, _, _ := utils.ExtractEntryFromChange(change)
 							offer, err := transform.TransformOffer(change)
 							if err != nil {
-								entry, _, _, _ := utils.ExtractEntryFromChange(change)
 								cmdLogger.LogError(fmt.Errorf("error transforming offer entry last updated at %d: %s", entry.LastModifiedLedgerSeq, err))
 								continue
 							}
 							transformedOutputs["offers"] = append(transformedOutputs["offers"], offer)
+
+							if ok, actualLedger := utils.LedgerIsCheckpoint(entry.LastModifiedLedgerSeq); ok {
+								x := verifyOutputs[actualLedger]
+								x.Offers = append(x.Offers, offer)
+								verifyOutputs[actualLedger] = x
+							}
 						}
 					case xdr.LedgerEntryTypeTrustline:
 						for _, change := range changes {
+							entry, _, _, _ := utils.ExtractEntryFromChange(change)
 							trust, err := transform.TransformTrustline(change)
 							if err != nil {
-								entry, _, _, _ := utils.ExtractEntryFromChange(change)
 								cmdLogger.LogError(fmt.Errorf("error transforming trustline entry last updated at %d: %s", entry.LastModifiedLedgerSeq, err))
 								continue
 							}
 							transformedOutputs["trustlines"] = append(transformedOutputs["trustlines"], trust)
+
+							if ok, actualLedger := utils.LedgerIsCheckpoint(entry.LastModifiedLedgerSeq); ok {
+								x := verifyOutputs[actualLedger]
+								x.Trustlines = append(x.Trustlines, trust)
+								verifyOutputs[actualLedger] = x
+							}
 						}
 					case xdr.LedgerEntryTypeLiquidityPool:
 						for _, change := range changes {
+							entry, _, _, _ := utils.ExtractEntryFromChange(change)
 							pool, err := transform.TransformPool(change)
 							if err != nil {
-								entry, _, _, _ := utils.ExtractEntryFromChange(change)
 								cmdLogger.LogError(fmt.Errorf("error transforming liquidity pool entry last updated at %d: %s", entry.LastModifiedLedgerSeq, err))
 								continue
 							}
 							transformedOutputs["liquidity_pools"] = append(transformedOutputs["liquidity_pools"], pool)
+
+							if ok, actualLedger := utils.LedgerIsCheckpoint(entry.LastModifiedLedgerSeq); ok {
+								x := verifyOutputs[actualLedger]
+								x.Liquidity_pools = append(x.Liquidity_pools, pool)
+								verifyOutputs[actualLedger] = x
+							}
 						}
 					}
 				}
 
+				for checkpointLedgers := range verifyOutputs {
+					_, err := verify.VerifyState(ctx, verifyOutputs[checkpointLedgers], archive, checkpointLedgers, verifyBatchSize)
+					if err != nil {
+						panic(err)
+					}
+				}
+
 				err := exportTransformedData(batch.BatchStart, batch.BatchEnd, outputFolder, transformedOutputs, gcpCredentials, gcsBucket, extra)
 				if err != nil {
 					cmdLogger.LogError(err)

internal/transform/account.go

@@ -9,7 +9,7 @@ import (
 	"github.com/stellar/stellar-etl/internal/utils"
 )
 
-//TransformAccount converts an account from the history archive ingestion system into a form suitable for BigQuery
+// TransformAccount converts an account from the history archive ingestion system into a form suitable for BigQuery
 func TransformAccount(ledgerChange ingest.Change) (AccountOutput, error) {
 	ledgerEntry, changeType, outputDeleted, err := utils.ExtractEntryFromChange(ledgerChange)
 	if err != nil {
@@ -77,27 +77,34 @@ func TransformAccount(ledgerChange ingest.Change) (AccountOutput, error) {
 	outputLastModifiedLedger := uint32(ledgerEntry.LastModifiedLedgerSeq)
 
 	transformedAccount := AccountOutput{
-		AccountID:            outputID,
-		Balance:              utils.ConvertStroopValueToReal(outputBalance),
-		BuyingLiabilities:    utils.ConvertStroopValueToReal(outputBuyingLiabilities),
-		SellingLiabilities:   utils.ConvertStroopValueToReal(outputSellingLiabilities),
-		SequenceNumber:       outputSequenceNumber,
-		SequenceLedger:       zero.IntFrom(int64(outputSequenceLedger)),
-		SequenceTime:         zero.IntFrom(int64(outputSequenceTime)),
-		NumSubentries:        outputNumSubentries,
-		InflationDestination: outputInflationDest,
-		Flags:                outputFlags,
-		HomeDomain:           outputHomeDomain,
-		MasterWeight:         outputMasterWeight,
-		ThresholdLow:         outputThreshLow,
-		ThresholdMedium:      outputThreshMed,
-		ThresholdHigh:        outputThreshHigh,
-		LastModifiedLedger:   outputLastModifiedLedger,
-		Sponsor:              ledgerEntrySponsorToNullString(ledgerEntry),
-		NumSponsored:         uint32(accountEntry.NumSponsored()),
-		NumSponsoring:        uint32(accountEntry.NumSponsoring()),
-		LedgerEntryChange:    uint32(changeType),
-		Deleted:              outputDeleted,
+		AccountID:             outputID,
+		RawBalance:            outputBalance,
+		RawBuyingLiabilities:  outputBuyingLiabilities,
+		RawSellingLiabilities: outputSellingLiabilities,
+		Balance:               utils.ConvertStroopValueToReal(outputBalance),
+		BuyingLiabilities:     utils.ConvertStroopValueToReal(outputBuyingLiabilities),
+		SellingLiabilities:    utils.ConvertStroopValueToReal(outputSellingLiabilities),
+		SequenceNumber:        outputSequenceNumber,
+		SequenceLedger:        zero.IntFrom(int64(outputSequenceLedger)),
+		SequenceTime:          zero.IntFrom(int64(outputSequenceTime)),
+		NumSubentries:         outputNumSubentries,
+		InflationDestination:  outputInflationDest,
+		Flags:                 outputFlags,
+		HomeDomain:            outputHomeDomain,
+		MasterWeight:          outputMasterWeight,
+		ThresholdLow:          outputThreshLow,
+		ThresholdMedium:       outputThreshMed,
+		ThresholdHigh:         outputThreshHigh,
+		LastModifiedLedger:    outputLastModifiedLedger,
+		Sponsor:               ledgerEntrySponsorToNullString(ledgerEntry),
+		NumSponsored:          uint32(accountEntry.NumSponsored()),
+		NumSponsoring:         uint32(accountEntry.NumSponsoring()),
+		LedgerEntryChange:     uint32(changeType),
+		Deleted:               outputDeleted,
+		RawMasterWeight:       accountEntry.MasterKeyWeight(),
+		RawThresholdLow:       accountEntry.ThresholdLow(),
+		RawThresholdMedium:    accountEntry.ThresholdMedium(),
+		RawThresholdHigh:      accountEntry.ThresholdHigh(),
 	}
 	return transformedAccount, nil
 }

internal/transform/account_test.go

@@ -151,24 +151,31 @@ func makeAccountTestInput() ingest.Change {
 
 func makeAccountTestOutput() AccountOutput {
 	return AccountOutput{
-		AccountID:            testAccount1Address,
-		Balance:              1.0959979,
-		BuyingLiabilities:    0.0001,
-		SellingLiabilities:   0.00015,
-		SequenceNumber:       117801117454198833,
-		NumSubentries:        141,
-		InflationDestination: testAccount2Address,
-		Flags:                4,
-		HomeDomain:           "examplehome.com",
-		MasterWeight:         2,
-		ThresholdLow:         1,
-		ThresholdMedium:      3,
-		ThresholdHigh:        5,
-		Sponsor:              null.StringFrom(testAccount3Address),
-		NumSponsored:         3,
-		NumSponsoring:        1,
-		LastModifiedLedger:   30705278,
-		LedgerEntryChange:    2,
-		Deleted:              true,
+		AccountID:             testAccount1Address,
+		RawBalance:            10959979,
+		RawBuyingLiabilities:  1000,
+		RawSellingLiabilities: 1500,
+		Balance:               1.0959979,
+		BuyingLiabilities:     0.0001,
+		SellingLiabilities:    0.00015,
+		SequenceNumber:        117801117454198833,
+		NumSubentries:         141,
+		InflationDestination:  testAccount2Address,
+		Flags:                 4,
+		HomeDomain:            "examplehome.com",
+		RawMasterWeight:       2,
+		RawThresholdLow:       1,
+		RawThresholdMedium:    3,
+		RawThresholdHigh:      5,
+		MasterWeight:          2,
+		ThresholdLow:          1,
+		ThresholdMedium:       3,
+		ThresholdHigh:         5,
+		Sponsor:               null.StringFrom(testAccount3Address),
+		NumSponsored:          3,
+		NumSponsoring:         1,
+		LastModifiedLedger:    30705278,
+		LedgerEntryChange:     2,
+		Deleted:               true,
 	}
 }

internal/transform/claimable_balance.go

@@ -20,7 +20,7 @@ func transformClaimants(claimants []xdr.Claimant) []Claimant {
 	return transformed
 }
 
-//TransformClaimableBalance converts a claimable balance from the history archive ingestion system into a form suitable for BigQuery
+// TransformClaimableBalance converts a claimable balance from the history archive ingestion system into a form suitable for BigQuery
 func TransformClaimableBalance(ledgerChange ingest.Change) (ClaimableBalanceOutput, error) {
 	ledgerEntry, changeType, outputDeleted, err := utils.ExtractEntryFromChange(ledgerChange)
 	if err != nil {
@@ -47,11 +47,13 @@ func TransformClaimableBalance(ledgerChange ingest.Change) (ClaimableBalanceOutp
 
 	transformed := ClaimableBalanceOutput{
 		BalanceID:          balanceID,
+		Asset:              balanceEntry.Asset,
 		AssetCode:          outputAsset.AssetCode,
 		AssetIssuer:        outputAsset.AssetIssuer,
 		AssetType:          outputAsset.AssetType,
 		Claimants:          outputClaimants,
 		AssetAmount:        float64(outputAmount) / 1.0e7,
+		RawAssetAmount:     outputAmount,
 		Sponsor:            ledgerEntrySponsorToNullString(ledgerEntry),
 		LastModifiedLedger: outputLastModifiedLedger,
 		LedgerEntryChange:  uint32(changeType),

internal/transform/claimable_balance_test.go

@@ -98,10 +98,12 @@ func makeClaimableBalanceTestOutput() ClaimableBalanceOutput {
 				},
 			},
 		},
+		Asset:              xdr.MustNewCreditAsset("\x01\x02\x03\x04\x05\x06\a\b\t", "GBT4YAEGJQ5YSFUMNKX6BPBUOCPNAIOFAVZOF6MIME2CECBMEIUXFZZN"),
 		AssetIssuer:        "GBT4YAEGJQ5YSFUMNKX6BPBUOCPNAIOFAVZOF6MIME2CECBMEIUXFZZN",
 		AssetType:          "credit_alphanum12",
 		AssetCode:          "\x01\x02\x03\x04\x05\x06\a\b\t",
 		AssetAmount:        999,
+		RawAssetAmount:     9990000000,
 		Sponsor:            null.StringFrom("GAAQEAYEAUDAOCAJAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABO3W"),
 		Flags:              10,
 		LastModifiedLedger: 30705278,

internal/transform/liquidity_pool.go

@@ -8,7 +8,7 @@ import (
 	"github.com/stellar/stellar-etl/internal/utils"
 )
 
-//TransformPool converts an liquidity pool ledger change entry into a form suitable for BigQuery
+// TransformPool converts an liquidity pool ledger change entry into a form suitable for BigQuery
 func TransformPool(ledgerChange ingest.Change) (PoolOutput, error) {
 	ledgerEntry, changeType, outputDeleted, err := utils.ExtractEntryFromChange(ledgerChange)
 	if err != nil {
@@ -46,21 +46,27 @@ func TransformPool(ledgerChange ingest.Change) (PoolOutput, error) {
 
 	transformedPool := PoolOutput{
 		PoolID:             PoolIDToString(lp.LiquidityPoolId),
+		RawPoolType:        lp.Body.Type,
 		PoolType:           poolType,
 		PoolFee:            uint32(cp.Params.Fee),
 		TrustlineCount:     uint64(cp.PoolSharesTrustLineCount),
 		PoolShareCount:     utils.ConvertStroopValueToReal(cp.TotalPoolShares),
+		AssetA:             cp.Params.AssetA,
 		AssetAType:         assetAType,
 		AssetACode:         assetACode,
 		AssetAIssuer:       assetAIssuer,
 		AssetAReserve:      utils.ConvertStroopValueToReal(cp.ReserveA),
+		AssetB:             cp.Params.AssetB,
 		AssetBType:         assetBType,
 		AssetBCode:         assetBCode,
 		AssetBIssuer:       assetBIssuer,
 		AssetBReserve:      utils.ConvertStroopValueToReal(cp.ReserveB),
 		LastModifiedLedger: uint32(ledgerEntry.LastModifiedLedgerSeq),
 		LedgerEntryChange:  uint32(changeType),
 		Deleted:            outputDeleted,
+		RawAssetAReserve:   cp.ReserveA,
+		RawAssetBReserve:   cp.ReserveB,
+		RawPoolShareCount:  cp.TotalPoolShares,
 	}
 	return transformedPool, nil
 }

internal/transform/liquidity_pool_test.go

@@ -107,5 +107,11 @@ func makePoolTestOutput() PoolOutput {
 		LastModifiedLedger: 30705278,
 		LedgerEntryChange:  2,
 		Deleted:            true,
+		AssetA:             xdr.MustNewNativeAsset(),
+		AssetB:             xdr.MustNewCreditAsset("USSD", "GBVVRXLMNCJQW3IDDXC3X6XCH35B5Q7QXNMMFPENSOGUPQO7WO7HGZPA"),
+		RawAssetAReserve:   105,
+		RawAssetBReserve:   10,
+		RawPoolShareCount:  35,
+		RawPoolType:        0,
 	}
 }

internal/transform/offer.go

@@ -71,13 +71,16 @@ func TransformOffer(ledgerChange ingest.Change) (OfferOutput, error) {
 	transformedOffer := OfferOutput{
 		SellerID:           outputSellerID,
 		OfferID:            outputOfferID,
+		SellingAsset:       offerEntry.Selling,
+		BuyingAsset:        offerEntry.Buying,
 		SellingAssetType:   outputSellingAsset.AssetType,
 		SellingAssetCode:   outputSellingAsset.AssetCode,
 		SellingAssetIssuer: outputSellingAsset.AssetIssuer,
 		BuyingAssetType:    outputBuyingAsset.AssetType,
 		BuyingAssetCode:    outputBuyingAsset.AssetCode,
 		BuyingAssetIssuer:  outputBuyingAsset.AssetIssuer,
 		Amount:             utils.ConvertStroopValueToReal(outputAmount),
+		RawAmount:          outputAmount,
 		PriceN:             outputPriceN,
 		PriceD:             outputPriceD,
 		Price:              outputPrice,