Merge pull request #611 from oasisprotocol/ptrus/feature/statecheck-e…

…xempt

statecheck: exempt accounts that are stale (to be requeried)

storage/migrations/03_statecheck_snapshots.up.sql

@@ -1,16 +1,19 @@
-BEGIN;
+-- Removed in https://github.com/oasisprotocol/nexus/pull/611
+-- From there on statecheck manually handles the snapshots schemas.
 
--- Schema for creating snapshot copies of the key tables.
-CREATE SCHEMA IF NOT EXISTS snapshot;
-GRANT USAGE ON SCHEMA snapshot TO PUBLIC;
+-- BEGIN;
 
--- Heights at which the key tables have been "snapshotted" (i.e. copied from
--- table chain.X to snapshot.X) as part of `tests/statecheck`.
-CREATE TABLE snapshot.snapshotted_heights
-(
-  analyzer TEXT NOT NULL,
-  height BIGINT PRIMARY KEY,
-  snapshot_time TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
-);
+-- -- Schema for creating snapshot copies of the key tables.
+-- CREATE SCHEMA IF NOT EXISTS snapshot;
+-- GRANT USAGE ON SCHEMA snapshot TO PUBLIC;
 
-COMMIT;
+-- -- Heights at which the key tables have been "snapshotted" (i.e. copied from
+-- -- table chain.X to snapshot.X) as part of `tests/statecheck`.
+-- CREATE TABLE snapshot.snapshotted_heights
+-- (
+--   analyzer TEXT NOT NULL,
+--   height BIGINT PRIMARY KEY,
+--   snapshot_time TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
+-- );
+
+-- COMMIT;

tests/statecheck/consensus_test.go

@@ -211,7 +211,7 @@ func validateEntities(t *testing.T, genesis *registryAPI.Genesis, target *postgr
 	}
 
 	entityRows, err := target.Query(ctx,
-		`SELECT id FROM snapshot.entities`,
+		`SELECT id FROM snapshot_consensus.entities`,
 	)
 	require.NoError(t, err)
 
@@ -230,7 +230,7 @@ func validateEntities(t *testing.T, genesis *registryAPI.Genesis, target *postgr
 		//
 		// Registry backend `StateToGenesis` returns the union of these nodes.
 		nodeRowsFromEntity, err := target.Query(ctx,
-			`SELECT node_id FROM snapshot.claimed_nodes WHERE entity_id = $1`,
+			`SELECT node_id FROM snapshot_consensus.claimed_nodes WHERE entity_id = $1`,
 			e.ID)
 		assert.NoError(t, err)
 		for nodeRowsFromEntity.Next() {
@@ -243,7 +243,7 @@ func validateEntities(t *testing.T, genesis *registryAPI.Genesis, target *postgr
 		}
 
 		nodeRowsFromNode, err := target.Query(ctx,
-			`SELECT id FROM snapshot.nodes WHERE entity_id = $1`,
+			`SELECT id FROM snapshot_consensus.nodes WHERE entity_id = $1`,
 			e.ID)
 		assert.NoError(t, err)
 		for nodeRowsFromNode.Next() {
@@ -331,7 +331,7 @@ func validateNodes(t *testing.T, genesis *registryAPI.Genesis, source consensusA
 			tls_pubkey, p2p_pubkey,
 			vrf_pubkey, roles, software_version
 		FROM
-			snapshot.nodes
+			snapshot_consensus.nodes
 		WHERE
 			roles LIKE '%validator%'
 	`)
@@ -426,7 +426,7 @@ func validateRuntimes(t *testing.T, genesis *registryAPI.Genesis, target *postgr
 	}
 
 	runtimeRows, err := target.Query(ctx,
-		`SELECT id, suspended, kind, tee_hardware, COALESCE(key_manager, 'none') FROM snapshot.runtimes`,
+		`SELECT id, suspended, kind, tee_hardware, COALESCE(key_manager, 'none') FROM snapshot_consensus.runtimes`,
 	)
 	require.NoError(t, err)
 
@@ -481,7 +481,7 @@ func validateAccounts(t *testing.T, genesis *stakingAPI.Genesis, target *postgre
 
 	acctRows, err := target.Query(ctx,
 		`SELECT address, nonce, general_balance, escrow_balance_active, escrow_balance_debonding
-				FROM snapshot.accounts`,
+				FROM snapshot_consensus.accounts`,
 	)
 	require.NoError(t, err)
 	actualAccts := make(map[string]bool)
@@ -509,7 +509,7 @@ func validateAccounts(t *testing.T, genesis *stakingAPI.Genesis, target *postgre
 		actualAllowances := make(map[string]uint64)
 		allowanceRows, err := target.Query(ctx, `
 			SELECT beneficiary, allowance
-				FROM snapshot.allowances
+				FROM snapshot_consensus.allowances
 				WHERE owner = $1
 			`,
 			a.Address,
@@ -623,7 +623,7 @@ func validateProposals(t *testing.T, genesis *governanceAPI.Genesis, target *pos
 		SELECT id, submitter, state, executed, deposit,
 				handler, cp_target_version, rhp_target_version, rcp_target_version, upgrade_epoch, cancels,
 				created_at, closes_at, invalid_votes
-		FROM snapshot.proposals`,
+		FROM snapshot_consensus.proposals`,
 	)
 	require.NoError(t, err)
 
@@ -681,7 +681,7 @@ func validateVotes(t *testing.T, genesis *governanceAPI.Genesis, target *postgre
 		}
 	}
 
-	voteRows, err := target.Query(ctx, `SELECT proposal, voter, vote FROM snapshot.votes`)
+	voteRows, err := target.Query(ctx, `SELECT proposal, voter, vote FROM snapshot_consensus.votes`)
 	require.NoError(t, err)
 
 	actualVotes := make(map[string]TestVote)

tests/statecheck/runtime_test.go

@@ -2,6 +2,7 @@ package statecheck
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"testing"
 
@@ -11,6 +12,7 @@ import (
 	sdkConfig "github.com/oasisprotocol/oasis-sdk/client-sdk/go/config"
 	sdkTypes "github.com/oasisprotocol/oasis-sdk/client-sdk/go/types"
 
+	"github.com/oasisprotocol/nexus/analyzer/runtime/evm"
 	common "github.com/oasisprotocol/nexus/common"
 )
 
@@ -35,6 +37,8 @@ func testRuntimeAccounts(t *testing.T, runtime common.Runtime) {
 		t.Skip("skipping test since healthcheck tests are not enabled")
 	}
 
+	schema := fmt.Sprintf("snapshot_%s", runtime)
+
 	t.Log("Initializing data stores...")
 
 	ctx := context.Background()
@@ -63,39 +67,74 @@ func testRuntimeAccounts(t *testing.T, runtime common.Runtime) {
 	}
 	t.Logf("Fetched %d account addresses", len(addresses))
 
-	acctRows, err := postgresClient.Query(ctx,
-		`SELECT account_address, balance, symbol
-			FROM snapshot.runtime_sdk_balances
-			WHERE runtime=$1`, runtime,
-	)
+	// Fetch addresses for which native balance is known to be stale.
+	// These are balances that we expect them to be invalid, so exempt them from the following test.
+	exemptRows, err := postgresClient.Query(ctx, fmt.Sprintf(`
+		SELECT account_address
+		FROM %s.stale_balances
+		WHERE runtime=$1 AND token_address=$2
+	`, schema), runtime, evm.NativeRuntimeTokenAddress)
+	require.NoError(t, err)
+	exemptAccs := make(map[string]bool)
+	defer exemptRows.Close()
+	for exemptRows.Next() {
+		var addr string
+		err = exemptRows.Scan(&addr)
+		require.NoError(t, err)
+		exemptAccs[addr] = true
+	}
+	t.Logf("Found %d exempted accounts (skipped for the test)", len(exemptAccs))
+
+	// Fetch account balances.
+	acctRows, err := postgresClient.Query(ctx, fmt.Sprintf(`
+		SELECT account_address, balance, symbol
+		FROM %s.runtime_sdk_balances
+		WHERE runtime=$1
+	`, schema), runtime)
 	require.NoError(t, err)
+	defer acctRows.Close()
 	actualAccts := make(map[string]bool)
 	var allBalances uint64
 	var balanceDiscrepancies uint64
+	var notExpectedFound uint64
+
+	// Check that the account balances are accurate.
 	for acctRows.Next() {
 		var a TestRuntimeAccount
 		err = acctRows.Scan(
 			&a.Address,
 			&a.Balance,
 			&a.Symbol,
 		)
-		assert.NoError(t, err)
+		require.NoError(t, err)
 
 		// Check that the account exists.
 		var actualAddr sdkTypes.Address
 		err = actualAddr.UnmarshalText([]byte(a.Address))
-		assert.NoError(t, err)
+		require.NoError(t, err)
+
+		// Skip accounts that are exempted.
+		if _, ok := exemptAccs[a.Address]; ok {
+			actualAccts[a.Address] = true
+			continue
+		}
 
 		_, ok := expectedAccts[actualAddr]
 		if !ok {
-			t.Logf("address %s found, but not expected", a.Address)
-			t.Fail()
+			// If the account is not expected (missing from oasis-node) it should have a zero balance.
+			if a.Balance.Int64() != 0 {
+				notExpectedFound++
+				t.Logf("Unexpected address '%s' found, reported balance (Nexus): %d", a.Address, a.Balance.Int64())
+				t.Fail()
+			} else {
+				allBalances++
+			}
 			continue
 		}
 
 		// Check that the account balance is accurate.
 		balances, err := oasisRuntimeClient.Accounts.Balances(ctx, uint64(height), actualAddr)
-		assert.NoError(t, err)
+		require.NoError(t, err)
 		for denom, amount := range balances.Balances {
 			if stringifyDenomination(denom, sdkPT) == a.Symbol {
 				allBalances++
@@ -106,15 +145,36 @@ func testRuntimeAccounts(t *testing.T, runtime common.Runtime) {
 		}
 		actualAccts[a.Address] = true
 	}
-	t.Logf("Number of discrepancies in account balances: %d (out of: %d)", balanceDiscrepancies, allBalances)
 
+	// Check addresses that are expected, but were not present in the snapshot.
+	var expectedNotFound uint64
+	var expectedNotFoundNonZero uint64
 	for _, addr := range addresses {
 		_, ok := actualAccts[addr.String()]
 		if !ok {
-			t.Logf("address %s expected, but not found", addr.String())
+			t.Logf("Expected address '%s' not found", addr.String())
+			expectedNotFound++
 			t.Fail()
+			balances, err := oasisRuntimeClient.Accounts.Balances(ctx, uint64(height), addr)
+			assert.NoError(t, err)
+			for denom, amount := range balances.Balances {
+				if stringifyDenomination(denom, sdkPT) == nativeTokenSymbol(sdkPT) {
+					t.Logf("Balance: %s", amount)
+					if amount.ToBigInt().Int64() > 0 {
+						expectedNotFoundNonZero++
+					}
+				}
+			}
 		}
 	}
+
+	// Report results.
+	t.Logf(`Number of discrepancies in account balances: %d (out of: %d).
+ 		- Does not include accounts that are only listed in Nexus or on the chain.
+		- Exempted accounts because of known stale balance in Nexus: %d
+	`, balanceDiscrepancies, allBalances, len(exemptAccs))
+	t.Logf("Number of unexpected addresses found in Nexus: %d", notExpectedFound)
+	t.Logf("Number of expected addresses not found in Nexus: %d (with non-zero balance: %d)", expectedNotFound, expectedNotFoundNonZero)
 }
 
 func nativeTokenSymbol(sdkPT *sdkConfig.ParaTime) string {

tests/statecheck/util.go

@@ -42,23 +42,40 @@ func newSdkConnection(ctx context.Context) (connection.Connection, error) {
 func snapshotBackends(target *postgres.Client, analyzer string, tables []string) (int64, error) {
 	ctx := context.Background()
 
+	schema := fmt.Sprintf("snapshot_%s", analyzer)
+
 	batch := &storage.QueryBatch{}
-	batch.Queue(`CREATE SCHEMA IF NOT EXISTS snapshot;`)
+	batch.Queue(fmt.Sprintf(`CREATE SCHEMA IF NOT EXISTS %s;`, schema))
 	for _, t := range tables {
 		batch.Queue(fmt.Sprintf(`
-			DROP TABLE IF EXISTS snapshot.%s CASCADE;
-		`, t))
+			DROP TABLE IF EXISTS %s.%s CASCADE;
+		`, schema, t))
 		batch.Queue(fmt.Sprintf(`
-			CREATE TABLE snapshot.%[1]s AS TABLE chain.%[1]s;
-		`, t))
+			CREATE TABLE %s.%[2]s AS TABLE chain.%[2]s;
+		`, schema, t))
 	}
-	batch.Queue(`
-		INSERT INTO snapshot.snapshotted_heights (analyzer, height)
-			SELECT analyzer, height
-			FROM analysis.processed_blocks WHERE analyzer=$1 AND processed_time IS NOT NULL
-			ORDER BY height DESC LIMIT 1
-			ON CONFLICT DO NOTHING;
-	`, analyzer)
+
+	// Store the height at which the snapshot was taken.
+	batch.Queue(fmt.Sprintf(`
+		CREATE TEMP TABLE %s_meta (height UINT63 NOT NULL);
+	`, schema))
+	batch.Queue(fmt.Sprintf(`
+		INSERT INTO %s_meta (height)
+			SELECT height
+			FROM analysis.processed_blocks
+			WHERE analyzer=$1 AND processed_time IS NOT NULL
+			ORDER BY height DESC LIMIT 1;
+	`, schema), analyzer)
+
+	// Store balances that are known to be stale - so that the statecheck tests can avoid comparing those.
+	batch.Queue(fmt.Sprintf(`
+		DROP TABLE IF EXISTS %s.stale_balances CASCADE;
+	`, schema))
+	batch.Queue(fmt.Sprintf(`
+		CREATE TABLE %s.stale_balances AS
+		SELECT * FROM analysis.evm_token_balances
+		WHERE last_download_round IS NULL OR last_download_round < last_mutate_round;
+	`, schema))
 
 	// Create the snapshot using a high level of isolation; we don't want another
 	// tx to be able to modify the tables while this is running, creating a snapshot that
@@ -67,12 +84,12 @@ func snapshotBackends(target *postgres.Client, analyzer string, tables []string)
 		return 0, err
 	}
 
+	// Report snapshotted height.
 	var snapshotHeight int64
-	if err := target.QueryRow(ctx, `
-		SELECT height from snapshot.snapshotted_heights
-			WHERE analyzer=$1
+	if err := target.QueryRow(ctx, fmt.Sprintf(`
+		SELECT height from %s_meta
 			ORDER BY height DESC LIMIT 1;
-	`, analyzer).Scan(&snapshotHeight); err != nil {
+	`, schema)).Scan(&snapshotHeight); err != nil {
 		return 0, err
 	}