Repository: tablelandnetwork/go-tableland
Pull request #264 (2022-09-16 14:05:10)

Implements state hash calculation every X block
# Summary
This PR implements a way of calculating the hash of a particular state of the database for every X block

# Context
As we start to onboard external validators, we must ensure we have a healthy network of validators. Having a healthy network of validators begins by making sure validators are materializing precisely the same database state. If that's not true, Tableland users are not able to trust our network.

So, this PR intends to collect a state hash metric for every X block. That metric is later published to a central telemetry server (yet to be built), and from there we can check if validators are materializing events correctly and agree on the same state for a given block.

# Implementation overview

## What's the right moment during execution to calculate the state hash?

The database is constantly changing as events come in, and each chain updates the database concurrently. So choosing a specific moment to get the database state to calculate the hash is not trivial.  We have to pick a moment where we know all validators should have the same state and any other changes cannot happen concurrently. Some thought processes that shaped our implementation:

- Since each chain processes events in its own _goroutine_, we can't calculate the global state of the database. It's not like we can freeze all _goroutines_ to calculate the database state. That means the state calculation has to be per chain.
- Inside each chain execution, blocks of events are processed serially. That means we can pick any moment between two block execution, and at that moment we know the state of the database will be frozen for all data of that chain. That's the right moment to do the calculation.

## How is the hash calculated?

The general idea is to fetch the name and the schema of all tables from a chain and all system tables, such as `registry`, `system_acl`, `system_txn_receipts` & `system_controller`. Then for each table, fetch all rows and columns. And then pass all that data to a hasher. So, we would be passing the table name, its schema, and all its data to a hasher. 

We have to make sure that the state we grab from the database is deterministic. So we have to exclude nondeterministic columns from system tables such as `created_at` and `updated_at`. For user tables, they all should have deterministic data, according to our SQL spec. No need to worry about it. And also, we have to use an `ORDER BY` clause to fetch all the data in the same order. 

## What happens after the hash is calculated?

The hash is collected and stored in a new local database called `metrics.db`. Later, another process will publish this metric to a central server. 

# Implementation details and review orientation

You can start the review by going through the bullet points. That will give a good overview of the implementation. Specifics will be commented on the `Files changed` tab.

- The hash is calculated and collected inside the event processor. Check lines [213-218](https://github.com/tablelandnetwork/go-tableland/pull/264/files#diff-046a77aeba18c13c7f18f3fdbaa1451a351c72c0ce2ee9de4d0f1a0d46193d83R213-R218)
    - Check the [`calculateHash` implementation](https://github.com/tablelandnetwork/go-tableland/pull/264/files#diff-046a77aeba18c13c7f18f3fdbaa1451a351c72c0ce2ee9de4d0f1a0d46193d83R288-R307). 
    - It makes use of a new method [StateHash](https://github.com/tablelandnetwork/go-tableland/pull/264/files#diff-6d26a4b1c786f7e0c5669cae8ec7e7af0206df2851916c9665dcedf2b167d772R168) in the `blockScope` object.
- A new package [dbhash](https://github.com/tablelandnetwork/go-tableland/pull/264/files#diff-068c52bbc04999650cf504f883632ba19447b12aee2a71e7917842d380daf2f6) was created to isolate the logic of calculating the hash. It's a fairly generic implementation that allows you to pass the queries you want to grab the state. So, the [StateHash](https://github.com/tablelandnetwork/go-tableland/pull/264/files#diff-6d26a4b1c786f7e0c5669cae8ec7e7af0206df2851916c9665dcedf2b167d772R168) defines the query it needs and delegate the hash calculation to [dbhash.DatabaseStateHash](https://github.com/tablelandnetwork/go-tableland/pull/264/files#diff-068c52bbc04999650cf504f883632ba19447b12aee2a71e7917842d380daf2f6R13)
- A new package [telemetry](https://github.com/tablelandnetwork/go-tableland/pull/264/files#diff-1daf3b94af9e46763e64ff02c31240f3c3b1421994d4d56e05769dfd4f765ba4) was created that acts as a library that allows collecting metrics such as [StateHashMetric](https://github.com/tablelandnetwork/go-tableland/pull/264/files#diff-4ab1048c1ff4b2de22e2a53e11ce2f4dd2e9c4e18949ee63541ee5854c8778b4R43), and persisting them locally.
- Metrics are persisted in another database called `metrics.db`. 


Patch:
diff --git a/cmd/api/config.go b/cmd/api/config.go
index 11ba197..0a7c81f 100644
--- a/cmd/api/config.go
+++ b/cmd/api/config.go
@@ -95,6 +95,7 @@ type ChainConfig struct {
 		StuckInterval string `default:"10m"`
 		MinBlockDepth int    `default:"5"`
 	}
+	HashCalculationStep int64 `default:"1000"`
 }
 
 func setupConfig() (*config, string) {
diff --git a/cmd/api/main.go b/cmd/api/main.go
index 133d3da..15e4202 100644
--- a/cmd/api/main.go
+++ b/cmd/api/main.go
@@ -35,6 +35,8 @@ import (
 	"github.com/textileio/go-tableland/pkg/sqlstore/impl/system"
 	"github.com/textileio/go-tableland/pkg/sqlstore/impl/user"
 	"github.com/textileio/go-tableland/pkg/tables/impl/ethereum"
+	"github.com/textileio/go-tableland/pkg/telemetry"
+	"github.com/textileio/go-tableland/pkg/telemetry/storage"
 	"github.com/textileio/go-tableland/pkg/wallet"
 )
 
@@ -96,6 +98,18 @@ func main() {
 		log.Fatal().Err(err).Msg("creating user store")
 	}
 
+	metricsDatabaseURL := fmt.Sprintf(
+		"file://%s?_busy_timeout=5000&_foreign_keys=on&_journal_mode=WAL",
+		path.Join(dirPath, "metrics.db"),
+	)
+
+	metricsStore, err := storage.New(metricsDatabaseURL)
+	if err != nil {
+		log.Fatal().Err(err).Msg("creating metrics store")
+	}
+
+	telemetry.SetMetricStore(metricsStore)
+
 	rateLimInterval, err := time.ParseDuration(config.HTTP.RateLimInterval)
 	if err != nil {
 		log.Fatal().Err(err).Msg("parsing http rate lim interval")
@@ -223,6 +237,10 @@ func main() {
 		if err := userStore.Close(); err != nil {
 			log.Error().Err(err).Msg("closing user store")
 		}
+
+		if err := metricsStore.Close(); err != nil {
+			log.Error().Err(err).Msg("closing metrics store")
+		}
 	})
 }
 
@@ -325,6 +343,7 @@ func createChainIDStack(
 	epOpts := []eventprocessor.Option{
 		eventprocessor.WithBlockFailedExecutionBackoff(blockFailedExecutionBackoff),
 		eventprocessor.WithDedupExecutedTxns(config.EventProcessor.DedupExecutedTxns),
+		eventprocessor.WithHashCalcStep(config.HashCalculationStep),
 	}
 	ep, err := epimpl.New(parser, ex, ef, config.ChainID, epOpts...)
 	if err != nil {
diff --git a/docker/deployed/staging/api/config.json b/docker/deployed/staging/api/config.json
index 55927ef..0abd406 100644
--- a/docker/deployed/staging/api/config.json
+++ b/docker/deployed/staging/api/config.json
@@ -67,7 +67,8 @@
         "CheckInterval": "10s",
         "StuckInterval": "10m",
         "MinBlockDepth": 0
-      }
+      },
+      "HashCalculationStep": 100
     },
     {
       "Name": "Optimism Goerli",
@@ -94,7 +95,8 @@
         "CheckInterval": "10s",
         "StuckInterval": "10m",
         "MinBlockDepth": 0
-      }
+      },
+      "HashCalculationStep": 100
     }
   ]
 }
\ No newline at end of file
diff --git a/docker/observability/grafana/provisioning/dashboards/validator-dashboard.json b/docker/observability/grafana/provisioning/dashboards/validator-dashboard.json
index 16e485d..10619f7 100644
--- a/docker/observability/grafana/provisioning/dashboards/validator-dashboard.json
+++ b/docker/observability/grafana/provisioning/dashboards/validator-dashboard.json
@@ -597,6 +597,73 @@
       "title": "Last backup",
       "type": "stat"
     },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "P1809F7CD0C75ACF3"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "thresholds"
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "dark-red",
+                "value": 18000
+              }
+            ]
+          },
+          "unit": "dtdurations"
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 5,
+        "w": 6,
+        "x": 4,
+        "y": 15
+      },
+      "id": 83,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "text": {},
+        "textMode": "auto"
+      },
+      "pluginVersion": "9.0.1",
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "P1809F7CD0C75ACF3"
+          },
+          "editorMode": "code",
+          "expr": "tableland_eventprocessor_hash_calculation_elapsed_time / 1000",
+          "legendFormat": "Chain ID: {{chain_id}}",
+          "range": true,
+          "refId": "A"
+        }
+      ],
+      "title": "Last Hash Calculation Elapsed Time",
+      "type": "stat"
+    },
     {
       "collapsed": false,
       "datasource": {
@@ -2746,13 +2813,13 @@
     "list": []
   },
   "time": {
-    "from": "now-6h",
+    "from": "now-5m",
     "to": "now"
   },
   "timepicker": {},
   "timezone": "",
   "title": "Validator",
   "uid": "2Le7qt_7z",
-  "version": 4,
+  "version": 13,
   "weekStart": ""
 }
\ No newline at end of file
diff --git a/pkg/dbhash/dbhash.go b/pkg/dbhash/dbhash.go
new file mode 100644
index 0000000..a8b2121
--- /dev/null
+++ b/pkg/dbhash/dbhash.go
@@ -0,0 +1,130 @@
+package dbhash
+
+import (
+	"context"
+	"crypto/sha1"
+	"database/sql"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+)
+
+// DatabaseStateHash calculates the hash of some state of the database according to the options passed.
+func DatabaseStateHash(ctx context.Context, tx *sql.Tx, opts ...Option) (string, error) {
+	config := DefaultConfig()
+	for _, o := range opts {
+		if err := o(config); err != nil {
+			return "", fmt.Errorf("applying provided option: %s", err)
+		}
+	}
+
+	h := sha1.New()
+	if err := databaseStateWriter(ctx, tx, h, config); err != nil {
+		return "", fmt.Errorf("database hash writer failed: %s", err)
+	}
+
+	return hex.EncodeToString(h.Sum(nil)), nil
+}
+
+func databaseStateWriter(ctx context.Context, tx *sql.Tx, writer io.Writer, c *Config) error {
+	// get all tables from db and associated schema
+	rows, err := tx.QueryContext(ctx, c.FetchSchemasQuery)
+	if err != nil {
+		return fmt.Errorf("querying schemas: %s", err)
+	}
+	defer func() {
+		_ = rows.Close()
+	}()
+
+	for rows.Next() {
+		var name, stmt string
+		if err = rows.Scan(
+			&name,
+			&stmt,
+		); err != nil {
+			return fmt.Errorf("schema scan row: %s", err)
+		}
+
+		_, _ = writer.Write([]byte(name))
+		_, _ = writer.Write([]byte(stmt))
+
+		if err := tableStateWriter(ctx, tx, name, writer, c); err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				continue
+			}
+
+			return fmt.Errorf("table state writer: %s", err)
+		}
+	}
+
+	return nil
+}
+
+func tableStateWriter(ctx context.Context, tx *sql.Tx, name string, writer io.Writer, c *Config) error {
+	tableRows, err := tx.QueryContext(ctx, c.PerTableQueryFn(name))
+	if err != nil {
+		return fmt.Errorf("querying table: %v", err)
+	}
+	defer func() {
+		_ = tableRows.Close()
+	}()
+
+	cols, err := tableRows.Columns()
+	if err != nil {
+		return fmt.Errorf("columns: %s", err)
+	}
+
+	rawBuffer := make([]sql.RawBytes, len(cols))
+	scanCallArgs := make([]interface{}, len(rawBuffer))
+	for i := range rawBuffer {
+		scanCallArgs[i] = &rawBuffer[i]
+	}
+
+	for tableRows.Next() {
+		if err := tableRows.Scan(scanCallArgs...); err != nil {
+			return fmt.Errorf("table row scan: %s", err)
+		}
+
+		for _, col := range rawBuffer {
+			_, _ = writer.Write(col)
+		}
+	}
+
+	return nil
+}
+
+// Config contains configuration parameters for tableland.
+type Config struct {
+	FetchSchemasQuery string
+	PerTableQueryFn   func(string) string
+}
+
+// DefaultConfig returns the default configuration.
+func DefaultConfig() *Config {
+	return &Config{
+		FetchSchemasQuery: "SELECT tbl_name, sql FROM sqlite_schema WHERE type='table' ORDER BY tbl_name",
+		PerTableQueryFn: func(tableName string) string {
+			return fmt.Sprintf("SELECT * FROM %s", tableName)
+		},
+	}
+}
+
+// Option modifies a configuration attribute.
+type Option func(*Config) error
+
+// WithFetchSchemasQuery limits tables that will be used for hash calculation.
+func WithFetchSchemasQuery(clause string) Option {
+	return func(c *Config) error {
+		c.FetchSchemasQuery = clause
+		return nil
+	}
+}
+
+// WithPerTableQueryFn defines a function that returns a query for a given table.
+func WithPerTableQueryFn(fn func(tableName string) string) Option {
+	return func(c *Config) error {
+		c.PerTableQueryFn = fn
+		return nil
+	}
+}
diff --git a/pkg/eventprocessor/eventprocessor.go b/pkg/eventprocessor/eventprocessor.go
index 6946b8e..7eeb51b 100644
--- a/pkg/eventprocessor/eventprocessor.go
+++ b/pkg/eventprocessor/eventprocessor.go
@@ -12,6 +12,7 @@ import (
 type Config struct {
 	BlockFailedExecutionBackoff time.Duration
 	DedupExecutedTxns           bool
+	HashCalcStep                int64
 }
 
 // DefaultConfig returns the default configuration.
@@ -19,6 +20,7 @@ func DefaultConfig() *Config {
 	return &Config{
 		BlockFailedExecutionBackoff: time.Second * 10,
 		DedupExecutedTxns:           false,
+		HashCalcStep:                100,
 	}
 }
 
@@ -50,6 +52,18 @@ func WithDedupExecutedTxns(dedupExecutedTxns bool) Option {
 	}
 }
 
+// WithHashCalcStep determines the pace of state hash calculations.
+// The hash will be calculated for every block equal or greater to the next multiple of the step.
+func WithHashCalcStep(step int64) Option {
+	return func(c *Config) error {
+		if step < 1 {
+			return fmt.Errorf("step cannot be less than 1")
+		}
+		c.HashCalcStep = step
+		return nil
+	}
+}
+
 // EventProcessor processes events from a smart-contract.
 type EventProcessor interface {
 	Start() error
diff --git a/pkg/eventprocessor/impl/eventprocessor.go b/pkg/eventprocessor/impl/eventprocessor.go
index c632bbb..cffac82 100644
--- a/pkg/eventprocessor/impl/eventprocessor.go
+++ b/pkg/eventprocessor/impl/eventprocessor.go
@@ -14,6 +14,7 @@ import (
 	"github.com/textileio/go-tableland/pkg/eventprocessor/eventfeed"
 	"github.com/textileio/go-tableland/pkg/eventprocessor/impl/executor"
 	"github.com/textileio/go-tableland/pkg/parsing"
+	"github.com/textileio/go-tableland/pkg/telemetry"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric/instrument/syncint64"
 	"go.uber.org/atomic"
@@ -37,18 +38,21 @@ type EventProcessor struct {
 	config   *eventprocessor.Config
 	chainID  tableland.ChainID
 
+	nextHashCalcBlockNumber int64
+
 	lock           sync.Mutex
 	daemonCtx      context.Context
 	daemonCancel   context.CancelFunc
 	daemonCanceled chan struct{}
 
 	// Metrics
-	mBaseLabels            []attribute.KeyValue
-	mExecutionRound        atomic.Int64
-	mLastProcessedHeight   atomic.Int64
-	mBlockExecutionLatency syncint64.Histogram
-	mEventExecutionCounter syncint64.Counter
-	mTxnExecutionLatency   syncint64.Histogram
+	mBaseLabels                 []attribute.KeyValue
+	mExecutionRound             atomic.Int64
+	mLastProcessedHeight        atomic.Int64
+	mBlockExecutionLatency      syncint64.Histogram
+	mEventExecutionCounter      syncint64.Counter
+	mTxnExecutionLatency        syncint64.Histogram
+	mHashCalculationElapsedTime atomic.Int64
 }
 
 // New returns a new EventProcessor.
@@ -139,6 +143,7 @@ func (ep *EventProcessor) startDaemon() error {
 		return fmt.Errorf("get last executed block number: %s", err)
 	}
 	ep.mLastProcessedHeight.Store(fromHeight)
+	ep.nextHashCalcBlockNumber = nextMultipleOf(fromHeight, ep.config.HashCalcStep)
 
 	// We fire an EventFeed asking for new events from the last processing height.
 	// Notice that if the client calls StopSync(...) it will cancel fp.daemonCtx
@@ -206,6 +211,13 @@ func (ep *EventProcessor) executeBlock(ctx context.Context, block eventfeed.Bloc
 		}
 	}()
 
+	if block.BlockNumber >= ep.nextHashCalcBlockNumber {
+		if err := ep.calculateHash(ctx, bs); err != nil {
+			return fmt.Errorf("calculate hash: %s", err)
+		}
+		ep.nextHashCalcBlockNumber = nextMultipleOf(block.BlockNumber, ep.config.HashCalcStep)
+	}
+
 	receipts := make([]eventprocessor.Receipt, 0, len(block.Txns))
 	for idxInBlock, txnEvents := range block.Txns {
 		if ep.config.DedupExecutedTxns {
@@ -273,3 +285,30 @@ func (ep *EventProcessor) executeBlock(ctx context.Context, block eventfeed.Bloc
 
 	return nil
 }
+
+func (ep *EventProcessor) calculateHash(ctx context.Context, bs executor.BlockScope) error {
+	startTime := time.Now()
+	stateHash, err := bs.StateHash(ctx, ep.chainID)
+	if err != nil {
+		return fmt.Errorf("calculating hash for current block: %s", err)
+	}
+	elapsedTime := time.Since(startTime).Milliseconds()
+	ep.log.Info().
+		Str("hash", stateHash.Hash()).
+		Int64("block_number", stateHash.BlockNumber()).
+		Int64("chain_id", stateHash.ChainID()).
+		Int64("elapsed_time", elapsedTime).
+		Msg("state hash")
+
+	ep.mHashCalculationElapsedTime.Store(elapsedTime)
+
+	if err := telemetry.Collect(ctx, stateHash); err != nil {
+		return fmt.Errorf("calculating hash for current block: %s", err)
+	}
+
+	return nil
+}
+
+func nextMultipleOf(x, y int64) int64 {
+	return y * ((x + y) / y)
+}
diff --git a/pkg/eventprocessor/impl/executor/executor.go b/pkg/eventprocessor/impl/executor/executor.go
index 5efa868..a37a171 100644
--- a/pkg/eventprocessor/impl/executor/executor.go
+++ b/pkg/eventprocessor/impl/executor/executor.go
@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/textileio/go-tableland/internal/tableland"
 	"github.com/textileio/go-tableland/pkg/eventprocessor"
 	"github.com/textileio/go-tableland/pkg/eventprocessor/eventfeed"
 	"github.com/textileio/go-tableland/pkg/tables"
@@ -41,6 +42,9 @@ type BlockScope interface {
 	// TxnReceiptExists return true if the provided transaction hash was already processed, and false otherwise.
 	TxnReceiptExists(ctx context.Context, txnHash common.Hash) (bool, error)
 
+	// StateHash calculates the hash of some state of the database.
+	StateHash(ctx context.Context, chainID tableland.ChainID) (StateHash, error)
+
 	// Commit commits all the changes that happened in  previously successful ExecuteTxnEvents(...) calls.
 	Commit() error
 
@@ -56,3 +60,34 @@ type TxnExecutionResult struct {
 	Error         *string
 	ErrorEventIdx *int
 }
+
+// StateHash represents the state of the database at given block number for a particular chain id.
+type StateHash struct {
+	chainID     tableland.ChainID
+	blockNumber int64
+	hash        string
+}
+
+// NewStateHash creates a new state hash.
+func NewStateHash(chainID tableland.ChainID, blockNumber int64, hash string) StateHash {
+	return StateHash{
+		chainID:     chainID,
+		blockNumber: blockNumber,
+		hash:        hash,
+	}
+}
+
+// ChainID returns chain id.
+func (h StateHash) ChainID() int64 {
+	return int64(h.chainID)
+}
+
+// BlockNumber returns the block number.
+func (h StateHash) BlockNumber() int64 {
+	return h.blockNumber
+}
+
+// Hash returns the hash.
+func (h StateHash) Hash() string {
+	return h.hash
+}
diff --git a/pkg/eventprocessor/impl/executor/impl/blockscope.go b/pkg/eventprocessor/impl/executor/impl/blockscope.go
index e9ad3ef..2445378 100644
--- a/pkg/eventprocessor/impl/executor/impl/blockscope.go
+++ b/pkg/eventprocessor/impl/executor/impl/blockscope.go
@@ -10,6 +10,7 @@ import (
 	"github.com/rs/zerolog"
 	logger "github.com/rs/zerolog/log"
 	"github.com/textileio/go-tableland/internal/tableland"
+	"github.com/textileio/go-tableland/pkg/dbhash"
 	"github.com/textileio/go-tableland/pkg/eventprocessor"
 	"github.com/textileio/go-tableland/pkg/eventprocessor/eventfeed"
 	"github.com/textileio/go-tableland/pkg/eventprocessor/impl/executor"
@@ -30,6 +31,7 @@ type blockScope struct {
 type scopeVars struct {
 	ChainID          tableland.ChainID
 	MaxTableRowCount int
+	BlockNumber      int64
 }
 
 func newBlockScope(
@@ -37,13 +39,12 @@ func newBlockScope(
 	scopeVars scopeVars,
 	parser parsing.SQLValidator,
 	acl tableland.ACL,
-	blockNum int64,
 	closed func(),
 ) *blockScope {
 	log := logger.With().
 		Str("component", "blockscope").
 		Int64("chain_id", int64(scopeVars.ChainID)).
-		Int64("block_number", blockNum).
+		Int64("block_number", scopeVars.BlockNumber).
 		Logger()
 
 	return &blockScope{
@@ -162,6 +163,54 @@ func (bs *blockScope) TxnReceiptExists(ctx context.Context, txnHash common.Hash)
 	return true, nil
 }
 
+func (bs *blockScope) StateHash(ctx context.Context, chainID tableland.ChainID) (executor.StateHash, error) {
+	hash, err := dbhash.DatabaseStateHash(ctx, bs.txn, []dbhash.Option{
+		dbhash.WithFetchSchemasQuery(
+			fmt.Sprintf(`SELECT tbl_name, sql 
+				FROM sqlite_schema
+			    WHERE name NOT LIKE 'sqlite_%%'  
+				AND name LIKE '%%\_%d\_%%' ESCAPE '\'
+				AND type = 'table'
+				UNION ALL
+				SELECT tbl_name, sql 
+				FROM sqlite_schema
+				WHERE name in ('registry', 'system_acl', 'system_controller', 'system_txn_receipts')
+				ORDER BY tbl_name;`, chainID),
+		),
+		dbhash.WithPerTableQueryFn(func(tableName string) string {
+			switch tableName {
+			case "registry":
+				return fmt.Sprintf(`SELECT id, chain_id, controller, prefix, structure 
+							FROM registry 
+							WHERE chain_id = %d 
+							ORDER BY id`, chainID)
+			case "system_acl":
+				return fmt.Sprintf(`SELECT chain_id, table_id, controller, privileges 
+							FROM system_acl 
+							WHERE chain_id = %d 
+							ORDER BY table_id`, chainID)
+			case "system_controller":
+				return fmt.Sprintf(`SELECT chain_id, table_id, controller 
+							FROM system_controller 
+							WHERE chain_id = %d
+							ORDER BY table_id`, chainID)
+			case "system_txn_receipts":
+				return fmt.Sprintf(`SELECT chain_id, block_number, index_in_block, txn_hash, error, table_id 
+							FROM system_txn_receipts 
+							WHERE chain_id = %d 
+							ORDER BY table_id, block_number, index_in_block`, chainID)
+			default:
+				return fmt.Sprintf("SELECT * FROM %s ORDER BY rowid", tableName)
+			}
+		}),
+	}...)
+	if err != nil {
+		return executor.StateHash{}, fmt.Errorf("database state hash: %s", err)
+	}
+
+	return executor.NewStateHash(chainID, bs.scopeVars.BlockNumber, hash), nil
+}
+
 // Close closes gracefully the block scope.
 // Clients should *always* `defer Close()` when opening batches.
 func (bs *blockScope) Close() error {
diff --git a/pkg/eventprocessor/impl/executor/impl/executor.go b/pkg/eventprocessor/impl/executor/impl/executor.go
index 479a33a..057dc5c 100644
--- a/pkg/eventprocessor/impl/executor/impl/executor.go
+++ b/pkg/eventprocessor/impl/executor/impl/executor.go
@@ -106,8 +106,12 @@ func (ex *Executor) NewBlockScope(ctx context.Context, newBlockNum int64) (execu
 		return nil, fmt.Errorf("latest executed block %d isn't smaller than new block %d", lastBlockNum, newBlockNum)
 	}
 
-	scopeVars := scopeVars{ChainID: ex.chainID, MaxTableRowCount: ex.maxTableRowCount}
-	bs := newBlockScope(txn, scopeVars, ex.parser, ex.acl, newBlockNum, func() { ex.chBlockScope <- struct{}{} })
+	scopeVars := scopeVars{
+		ChainID:          ex.chainID,
+		MaxTableRowCount: ex.maxTableRowCount,
+		BlockNumber:      newBlockNum,
+	}
+	bs := newBlockScope(txn, scopeVars, ex.parser, ex.acl, func() { ex.chBlockScope <- struct{}{} })
 
 	return bs, nil
 }
diff --git a/pkg/eventprocessor/impl/metrics.go b/pkg/eventprocessor/impl/metrics.go
index 47cf67e..865436a 100644
--- a/pkg/eventprocessor/impl/metrics.go
+++ b/pkg/eventprocessor/impl/metrics.go
@@ -23,10 +23,17 @@ func (ep *EventProcessor) initMetrics(chainID tableland.ChainID) error {
 	if err != nil {
 		return fmt.Errorf("creating last processed height gauge: %s", err)
 	}
-	err = meter.RegisterCallback([]instrument.Asynchronous{mExecutionRound, mLastProcessedHeight},
+	mHashCalculationElapsedTime, err := meter.AsyncInt64().Gauge("tableland.eventprocessor.hash.calculation.elapsed.time")
+	if err != nil {
+		return fmt.Errorf("creating hash calculation elapsed time gauge: %s", err)
+	}
+	err = meter.RegisterCallback([]instrument.Asynchronous{
+		mExecutionRound, mLastProcessedHeight, mHashCalculationElapsedTime,
+	},
 		func(ctx context.Context) {
 			mExecutionRound.Observe(ctx, ep.mExecutionRound.Load(), ep.mBaseLabels...)
 			mLastProcessedHeight.Observe(ctx, ep.mLastProcessedHeight.Load(), ep.mBaseLabels...)
+			mHashCalculationElapsedTime.Observe(ctx, ep.mHashCalculationElapsedTime.Load(), ep.mBaseLabels...)
 		})
 	if err != nil {
 		return fmt.Errorf("registering async metric callback: %s", err)
diff --git a/pkg/telemetry/metrics.go b/pkg/telemetry/metrics.go
new file mode 100644
index 0000000..2ed8272
--- /dev/null
+++ b/pkg/telemetry/metrics.go
@@ -0,0 +1,49 @@
+package telemetry
+
+import (
+	"encoding/json"
+	"time"
+
+	"github.com/pkg/errors"
+)
+
+// MetricType defines the metric type.
+type MetricType int
+
+const (
+	// StateHashType is the type for the StateHashMetric.
+	StateHashType MetricType = iota
+)
+
+// Metric defines a metric.
+type Metric struct {
+	Version   int
+	Timestamp time.Time
+	Type      MetricType
+	Payload   interface{}
+}
+
+// Serialize serializes the metric.
+func (m Metric) Serialize() ([]byte, error) {
+	b, err := json.Marshal(m.Payload)
+	if err != nil {
+		return []byte(nil), errors.Errorf("marshal: %s", err)
+	}
+
+	return b, nil
+}
+
+// StateHash defines how data is accessed to create a StateHashMetric.
+type StateHash interface {
+	ChainID() int64
+	BlockNumber() int64
+	Hash() string
+}
+
+// StateHashMetric defines a state hash metric.
+type StateHashMetric struct {
+	Version     int64  `json:"version"`
+	ChainID     int64  `json:"chain_id"`
+	BlockNumber int64  `json:"block_number"`
+	Hash        string `json:"hash"`
+}
diff --git a/pkg/telemetry/storage/db.go b/pkg/telemetry/storage/db.go
new file mode 100644
index 0000000..00f5fc4
--- /dev/null
+++ b/pkg/telemetry/storage/db.go
@@ -0,0 +1,108 @@
+package storage
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/XSAM/otelsql"
+	"github.com/golang-migrate/migrate/v4"
+	_ "github.com/golang-migrate/migrate/v4/database/sqlite3" // migration for sqlite3
+	bindata "github.com/golang-migrate/migrate/v4/source/go_bindata"
+
+	"github.com/rs/zerolog"
+	logger "github.com/rs/zerolog/log"
+	"github.com/textileio/go-tableland/pkg/telemetry"
+	"github.com/textileio/go-tableland/pkg/telemetry/storage/migrations"
+	"go.opentelemetry.io/otel/attribute"
+)
+
+// TelemetryDatabase implements the MetricStore interface and provides storage for a metric.
+type TelemetryDatabase struct {
+	log   zerolog.Logger
+	sqlDB *sql.DB
+}
+
+// New returns a new TelemetryDatabase backed by database/sql.
+func New(dbURI string) (*TelemetryDatabase, error) {
+	sqlDB, err := otelsql.Open("sqlite3", dbURI, otelsql.WithAttributes(
+		attribute.String("name", "telemetrydb"),
+	))
+	if err != nil {
+		return nil, fmt.Errorf("connecting to db: %s", err)
+	}
+	sqlDB.SetMaxIdleConns(0)
+	if err := otelsql.RegisterDBStatsMetrics(sqlDB, otelsql.WithAttributes(
+		attribute.String("name", "telemetrydb"),
+	)); err != nil {
+		return nil, fmt.Errorf("registering dbstats: %s", err)
+	}
+
+	log := logger.With().
+		Str("component", "telemetrydb").
+		Logger()
+
+	db := &TelemetryDatabase{
+		log:   log,
+		sqlDB: sqlDB,
+	}
+
+	as := bindata.Resource(migrations.AssetNames(), migrations.Asset)
+	if err := db.executeMigration(dbURI, as); err != nil {
+		return nil, fmt.Errorf("initializing db connection: %s", err)
+	}
+
+	return db, nil
+}
+
+// StoreMetric persists a metric.
+func (db *TelemetryDatabase) StoreMetric(ctx context.Context, metric telemetry.Metric) error {
+	payloadJSON, err := metric.Serialize()
+	if err != nil {
+		return fmt.Errorf("marshal json: %s", err)
+	}
+
+	_, err = db.sqlDB.ExecContext(ctx,
+		`INSERT INTO system_metrics ("version", "timestamp", "type", "payload", "published") VALUES (?1, ?2, ?3, ?4, ?5)`,
+		metric.Version, metric.Timestamp.UnixMilli(), metric.Type, payloadJSON, 0,
+	)
+	if err != nil {
+		return fmt.Errorf("insert into system_metrics: %s", err)
+	}
+
+	return nil
+}
+
+// Close closes the database.
+func (db *TelemetryDatabase) Close() error {
+	if err := db.sqlDB.Close(); err != nil {
+		return fmt.Errorf("close: %s", err)
+	}
+
+	return nil
+}
+
+// executeMigration run db migrations and return a ready to use connection to the SQLite database.
+func (db *TelemetryDatabase) executeMigration(dbURI string, as *bindata.AssetSource) error {
+	d, err := bindata.WithInstance(as)
+	if err != nil {
+		return fmt.Errorf("creating source driver: %s", err)
+	}
+
+	m, err := migrate.NewWithSourceInstance("go-bindata", d, "sqlite3://"+dbURI)
+	if err != nil {
+		return fmt.Errorf("creating migration: %s", err)
+	}
+	version, dirty, err := m.Version()
+	db.log.Info().
+		Uint("dbVersion", version).
+		Bool("dirty", dirty).
+		Err(err).
+		Msg("database migration executed")
+
+	if err := m.Up(); err != nil && err != migrate.ErrNoChange {
+		return fmt.Errorf("running migration up: %s", err)
+	}
+
+	return nil
+}
diff --git a/pkg/telemetry/storage/migrations/001_system_metrics.up.sql b/pkg/telemetry/storage/migrations/001_system_metrics.up.sql
new file mode 100644
index 0000000..d47c5ff
--- /dev/null
+++ b/pkg/telemetry/storage/migrations/001_system_metrics.up.sql
@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS system_metrics(
+    version INTERGER NOT NULL,
+    timestamp INTEGER NOT NULL,
+    type INTEGER NOT NULL,
+    payload TEXT NOT NULL,
+    published INTEGER NOT NULL
+);
\ No newline at end of file
diff --git a/pkg/telemetry/storage/migrations/002_system_metrics.down.sql b/pkg/telemetry/storage/migrations/002_system_metrics.down.sql
new file mode 100644
index 0000000..045dfe5
--- /dev/null
+++ b/pkg/telemetry/storage/migrations/002_system_metrics.down.sql
@@ -0,0 +1,1 @@
+DROP TABLE system_metrics;
\ No newline at end of file
diff --git a/pkg/telemetry/storage/migrations/migrations.go b/pkg/telemetry/storage/migrations/migrations.go
new file mode 100644
index 0000000..42c1eb5
--- /dev/null
+++ b/pkg/telemetry/storage/migrations/migrations.go
@@ -0,0 +1,268 @@
+// Code generated by go-bindata. (@generated) DO NOT EDIT.
+
+ //Package migrations generated by go-bindata.// sources:
+// migrations/001_system_metrics.up.sql
+// migrations/002_system_metrics.down.sql
+package migrations
+
+import (
+	"bytes"
+	"compress/gzip"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+func bindataRead(data []byte, name string) ([]byte, error) {
+	gz, err := gzip.NewReader(bytes.NewBuffer(data))
+	if err != nil {
+		return nil, fmt.Errorf("read %q: %v", name, err)
+	}
+
+	var buf bytes.Buffer
+	_, err = io.Copy(&buf, gz)
+	clErr := gz.Close()
+
+	if err != nil {
+		return nil, fmt.Errorf("read %q: %v", name, err)
+	}
+	if clErr != nil {
+		return nil, err
+	}
+
+	return buf.Bytes(), nil
+}
+
+type asset struct {
+	bytes []byte
+	info  os.FileInfo
+}
+
+type bindataFileInfo struct {
+	name    string
+	size    int64
+	mode    os.FileMode
+	modTime time.Time
+}
+
+// Name return file name
+func (fi bindataFileInfo) Name() string {
+	return fi.name
+}
+
+// Size return file size
+func (fi bindataFileInfo) Size() int64 {
+	return fi.size
+}
+
+// Mode return file mode
+func (fi bindataFileInfo) Mode() os.FileMode {
+	return fi.mode
+}
+
+// ModTime return file modify time
+func (fi bindataFileInfo) ModTime() time.Time {
+	return fi.modTime
+}
+
+// IsDir return file whether a directory
+func (fi bindataFileInfo) IsDir() bool {
+	return fi.mode&os.ModeDir != 0
+}
+
+// Sys return file is sys mode
+func (fi bindataFileInfo) Sys() interface{} {
+	return nil
+}
+
+var __001_system_metricsUpSql = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\x74\xcb\x4d\x0a\xc2\x30\x10\x47\xf1\x7d\x4f\xf1\x5f\x2a\x78\x03\x57\x55\x46\x29\x84\x08\xe9\x08\xdd\x49\xb4\x03\x06\x1a\x1b\x3a\x51\xc8\xed\x85\x2c\xfd\x58\xff\xde\xdb\x3b\x6a\x99\xc0\xed\xce\x10\xba\x03\xec\x89\x41\x43\xd7\x73\x0f\x2d\x9a\x25\x5e\xa2\xe4\x25\xdc\x74\xd5\x00\xc0\x4b\x16\x0d\xf3\x03\x9d\x65\x72\x47\x72\xb5\xb7\x67\x63\x36\x95\x73\x88\xa2\xd9\xc7\x54\x83\x1f\x5e\x92\xfc\xa1\xe4\xcb\x34\xfb\x11\x4c\x03\x7f\xd2\xf3\x3a\x05\xbd\xcb\xf8\xb5\x36\xeb\xed\x3b\x00\x00\xff\xff\x27\x39\xba\xc5\xc1\x00\x00\x00")
+
+func _001_system_metricsUpSqlBytes() ([]byte, error) {
+	return bindataRead(
+		__001_system_metricsUpSql,
+		"001_system_metrics.up.sql",
+	)
+}
+
+func _001_system_metricsUpSql() (*asset, error) {
+	bytes, err := _001_system_metricsUpSqlBytes()
+	if err != nil {
+		return nil, err
+	}
+
+	info := bindataFileInfo{name: "001_system_metrics.up.sql", size: 193, mode: os.FileMode(420), modTime: time.Unix(1663781121, 0)}
+	a := &asset{bytes: bytes, info: info}
+	return a, nil
+}
+
+var __002_system_metricsDownSql = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\x72\x09\xf2\x0f\x50\x08\x71\x74\xf2\x71\x55\x28\xae\x2c\x2e\x49\xcd\x8d\xcf\x4d\x2d\x29\xca\x4c\x2e\xb6\x06\x04\x00\x00\xff\xff\x3b\xb1\x10\x44\x1a\x00\x00\x00")
+
+func _002_system_metricsDownSqlBytes() ([]byte, error) {
+	return bindataRead(
+		__002_system_metricsDownSql,
+		"002_system_metrics.down.sql",
+	)
+}
+
+func _002_system_metricsDownSql() (*asset, error) {
+	bytes, err := _002_system_metricsDownSqlBytes()
+	if err != nil {
+		return nil, err
+	}
+
+	info := bindataFileInfo{name: "002_system_metrics.down.sql", size: 26, mode: os.FileMode(420), modTime: time.Unix(1663766762, 0)}
+	a := &asset{bytes: bytes, info: info}
+	return a, nil
+}
+
+// Asset loads and returns the asset for the given name.
+// It returns an error if the asset could not be found or
+// could not be loaded.
+func Asset(name string) ([]byte, error) {
+	cannonicalName := strings.Replace(name, "\\", "/", -1)
+	if f, ok := _bindata[cannonicalName]; ok {
+		a, err := f()
+		if err != nil {
+			return nil, fmt.Errorf("Asset %s can't read by error: %v", name, err)
+		}
+		return a.bytes, nil
+	}
+	return nil, fmt.Errorf("Asset %s not found", name)
+}
+
+// MustAsset is like Asset but panics when Asset would return an error.
+// It simplifies safe initialization of global variables.
+func MustAsset(name string) []byte {
+	a, err := Asset(name)
+	if err != nil {
+		panic("asset: Asset(" + name + "): " + err.Error())
+	}
+
+	return a
+}
+
+// AssetInfo loads and returns the asset info for the given name.
+// It returns an error if the asset could not be found or
+// could not be loaded.
+func AssetInfo(name string) (os.FileInfo, error) {
+	cannonicalName := strings.Replace(name, "\\", "/", -1)
+	if f, ok := _bindata[cannonicalName]; ok {
+		a, err := f()
+		if err != nil {
+			return nil, fmt.Errorf("AssetInfo %s can't read by error: %v", name, err)
+		}
+		return a.info, nil
+	}
+	return nil, fmt.Errorf("AssetInfo %s not found", name)
+}
+
+// AssetNames returns the names of the assets.
+func AssetNames() []string {
+	names := make([]string, 0, len(_bindata))
+	for name := range _bindata {
+		names = append(names, name)
+	}
+	return names
+}
+
+// _bindata is a table, holding each asset generator, mapped to its name.
+var _bindata = map[string]func() (*asset, error){
+	"001_system_metrics.up.sql":   _001_system_metricsUpSql,
+	"002_system_metrics.down.sql": _002_system_metricsDownSql,
+}
+
+// AssetDir returns the file names below a certain
+// directory embedded in the file by go-bindata.
+// For example if you run go-bindata on data/... and data contains the
+// following hierarchy:
+//     data/
+//       foo.txt
+//       img/
+//         a.png
+//         b.png
+// then AssetDir("data") would return []string{"foo.txt", "img"}
+// AssetDir("data/img") would return []string{"a.png", "b.png"}
+// AssetDir("foo.txt") and AssetDir("notexist") would return an error
+// AssetDir("") will return []string{"data"}.
+func AssetDir(name string) ([]string, error) {
+	node := _bintree
+	if len(name) != 0 {
+		cannonicalName := strings.Replace(name, "\\", "/", -1)
+		pathList := strings.Split(cannonicalName, "/")
+		for _, p := range pathList {
+			node = node.Children[p]
+			if node == nil {
+				return nil, fmt.Errorf("Asset %s not found", name)
+			}
+		}
+	}
+	if node.Func != nil {
+		return nil, fmt.Errorf("Asset %s not found", name)
+	}
+	rv := make([]string, 0, len(node.Children))
+	for childName := range node.Children {
+		rv = append(rv, childName)
+	}
+	return rv, nil
+}
+
+type bintree struct {
+	Func     func() (*asset, error)
+	Children map[string]*bintree
+}
+
+var _bintree = &bintree{nil, map[string]*bintree{
+	"001_system_metrics.up.sql":   &bintree{_001_system_metricsUpSql, map[string]*bintree{}},
+	"002_system_metrics.down.sql": &bintree{_002_system_metricsDownSql, map[string]*bintree{}},
+}}
+
+// RestoreAsset restores an asset under the given directory
+func RestoreAsset(dir, name string) error {
+	data, err := Asset(name)
+	if err != nil {
+		return err
+	}
+	info, err := AssetInfo(name)
+	if err != nil {
+		return err
+	}
+	err = os.MkdirAll(_filePath(dir, filepath.Dir(name)), os.FileMode(0755))
+	if err != nil {
+		return err
+	}
+	err = ioutil.WriteFile(_filePath(dir, name), data, info.Mode())
+	if err != nil {
+		return err
+	}
+	err = os.Chtimes(_filePath(dir, name), info.ModTime(), info.ModTime())
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// RestoreAssets restores an asset under the given directory recursively
+func RestoreAssets(dir, name string) error {
+	children, err := AssetDir(name)
+	// File
+	if err != nil {
+		return RestoreAsset(dir, name)
+	}
+	// Dir
+	for _, child := range children {
+		err = RestoreAssets(dir, filepath.Join(name, child))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func _filePath(dir, name string) string {
+	cannonicalName := strings.Replace(name, "\\", "/", -1)
+	return filepath.Join(append([]string{dir}, strings.Split(cannonicalName, "/")...)...)
+}
diff --git a/pkg/telemetry/telemetry.go b/pkg/telemetry/telemetry.go
new file mode 100644
index 0000000..f4bec94
--- /dev/null
+++ b/pkg/telemetry/telemetry.go
@@ -0,0 +1,71 @@
+package telemetry
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	logger "github.com/rs/zerolog/log"
+)
+
+var (
+	metricStore MetricStore
+	log         zerolog.Logger
+
+	mu   = &sync.Mutex{}
+	once sync.Once
+)
+
+func init() {
+	log = logger.With().
+		Str("component", "telemetry").
+		Logger()
+}
+
+// MetricStore specifies the methods for persisting a metric.
+type MetricStore interface {
+	StoreMetric(context.Context, Metric) error
+	Close() error
+}
+
+// SetMetricStore sets the store implementation.
+// Only the first call will have an effect. If Collect is called without setting a MetricStore, it will be a noop.
+func SetMetricStore(s MetricStore) {
+	once.Do(func() {
+		metricStore = s
+	})
+}
+
+// Collect collects the metric by persisting locally for later publication.
+// If Collect is called before setting the metric store, it will simply log the metric without persisting it.
+func Collect(ctx context.Context, metric interface{}) error {
+	mu.Lock()
+	defer mu.Unlock()
+	if metricStore == nil {
+		log.Warn().Msg("no metric store was set")
+		return nil
+	}
+
+	switch v := metric.(type) {
+	case StateHash:
+		if err := metricStore.StoreMetric(ctx, Metric{
+			Version:   1,
+			Timestamp: time.Now().UTC(),
+			Type:      StateHashType,
+			Payload: StateHashMetric{
+				Version:     1,
+				ChainID:     v.ChainID(),
+				BlockNumber: v.BlockNumber(),
+				Hash:        v.Hash(),
+			},
+		}); err != nil {
+			return errors.Errorf("store metric: %s", err)
+		}
+		return nil
+	default:
+		return fmt.Errorf("unknown metric type %T", v)
+	}
+}


Test patch:
diff --git a/docker/deployed/testnet/api/config.json b/docker/deployed/testnet/api/config.json
index 31ff2c5..eed6166 100644
--- a/docker/deployed/testnet/api/config.json
+++ b/docker/deployed/testnet/api/config.json
@@ -68,7 +68,8 @@
                 "CheckInterval": "10s",
                 "StuckInterval": "5m",
                 "MinBlockDepth": 0
-            }
+            },
+            "HashCalculationStep": 1800
         },
         {
             "Name": "Ethereum Goerli",
@@ -95,7 +96,8 @@
                 "CheckInterval": "15s",
                 "StuckInterval": "10m",
                 "MinBlockDepth": 1
-            }
+            },
+            "HashCalculationStep": 150
         },
         {
             "Name": "Polygon Mumbai",
@@ -122,7 +124,8 @@
                 "CheckInterval": "15s",
                 "StuckInterval": "5m",
                 "MinBlockDepth": 1
-            }
+            },
+            "HashCalculationStep": 360
         },
         {
             "Name": "Ethereum Mainnet",
@@ -149,7 +152,8 @@
                 "CheckInterval": "20s",
                 "StuckInterval": "10m",
                 "MinBlockDepth": 1
-            }
+            },
+            "HashCalculationStep": 150
         },
         {
             "Name": "Polygon Mainnet",
@@ -176,7 +180,8 @@
                 "CheckInterval": "15s",
                 "StuckInterval": "10m",
                 "MinBlockDepth": 1
-            }
+            },
+            "HashCalculationStep": 360
         },
         {
             "Name": "Optimism Mainnet",
@@ -203,7 +208,8 @@
                 "CheckInterval": "15s",
                 "StuckInterval": "10m",
                 "MinBlockDepth": 0
-            }
+            },
+            "HashCalculationStep": 1800
         },
         {
             "Name": "Arbitrum Goerli",
@@ -230,7 +236,8 @@
                 "CheckInterval": "15s",
                 "StuckInterval": "10m",
                 "MinBlockDepth": 0
-            }
+            },
+            "HashCalculationStep": 360
         },
         {
             "Name": "Optimism Goerli",
@@ -257,7 +264,8 @@
                 "CheckInterval": "10s",
                 "StuckInterval": "5m",
                 "MinBlockDepth": 0
-            }
+            },
+            "HashCalculationStep": 1800
         }
     ]
 }
\ No newline at end of file
diff --git a/pkg/dbhash/dbhash_test.go b/pkg/dbhash/dbhash_test.go
new file mode 100644
index 0000000..28e85a4
--- /dev/null
+++ b/pkg/dbhash/dbhash_test.go
@@ -0,0 +1,150 @@
+package dbhash
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha1"
+	"database/sql"
+	"encoding/hex"
+	"fmt"
+	"testing"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/require"
+	"github.com/textileio/go-tableland/tests"
+)
+
+func TestDatabaseHash(t *testing.T) {
+	testCases := []testCase{
+		tc0(),
+		tc1(),
+		tc2(),
+	}
+
+	for i, tc := range testCases {
+		t.Run(fmt.Sprint(i), func(t *testing.T) {
+			dbURI := tests.Sqlite3URI()
+
+			db, err := sql.Open("sqlite3", dbURI)
+			require.NoError(t, err)
+
+			_, err = db.Exec(tc.dbSeed)
+			require.NoError(t, err)
+
+			tx, err := db.BeginTx(context.Background(), nil)
+			defer func() {
+				_ = tx.Commit()
+			}()
+			require.NoError(t, err)
+
+			hash, err := DatabaseStateHash(context.Background(), tx, tc.opts...)
+			require.NoError(t, err)
+			require.Equal(t, tc.exp, hash)
+		})
+	}
+}
+
+type testCase struct {
+	dbSeed string
+	opts   []Option
+	exp    string
+}
+
+func tc0() testCase {
+	return testCase{
+		dbSeed: `CREATE TABLE a (a int);
+				CREATE TABLE b (c int, d text);
+				INSERT INTO a VALUES (123);
+				INSERT INTO a VALUES (456);
+				INSERT INTO b VALUES (10, "ten");`,
+		opts: []Option{
+			WithFetchSchemasQuery("SELECT tbl_name, sql FROM sqlite_schema WHERE type = 'table'"),
+		},
+		exp: func() string {
+			var buf bytes.Buffer
+			buf.Write([]byte("a"))
+			buf.Write([]byte("CREATE TABLE a (a int)"))
+			buf.Write([]byte("123"))
+			buf.Write([]byte("456"))
+			buf.Write([]byte("b"))
+			buf.Write([]byte("CREATE TABLE b (c int, d text)"))
+			buf.Write([]byte("10"))
+			buf.Write([]byte("ten"))
+
+			h := sha1.New()
+			h.Write(buf.Bytes())
+
+			return hex.EncodeToString(h.Sum(nil))
+		}(),
+	}
+}
+
+// Similar to tc0. Changes seed order but get the same hash because of ORDER BY clause.
+func tc1() testCase {
+	return testCase{
+		dbSeed: `CREATE TABLE b (c int, d text);
+				CREATE TABLE a (a int);
+				INSERT INTO b VALUES (10, "ten");
+				INSERT INTO a VALUES (123);
+				INSERT INTO a VALUES (456);
+				`,
+		opts: []Option{
+			WithFetchSchemasQuery("SELECT tbl_name, sql FROM sqlite_schema WHERE type = 'table' ORDER BY tbl_name"),
+		},
+		exp: func() string {
+			var buf bytes.Buffer
+			buf.Write([]byte("a"))
+			buf.Write([]byte("CREATE TABLE a (a int)"))
+			buf.Write([]byte("123"))
+			buf.Write([]byte("456"))
+			buf.Write([]byte("b"))
+			buf.Write([]byte("CREATE TABLE b (c int, d text)"))
+			buf.Write([]byte("10"))
+			buf.Write([]byte("ten"))
+
+			h := sha1.New()
+			h.Write(buf.Bytes())
+
+			return hex.EncodeToString(h.Sum(nil))
+		}(),
+	}
+}
+
+// Adds a per table query function.
+func tc2() testCase {
+	return testCase{
+		dbSeed: `CREATE TABLE a (a int);
+				CREATE TABLE b (c int, d text);
+				INSERT INTO a VALUES (123);
+				INSERT INTO a VALUES (456);
+				INSERT INTO b VALUES (10, "ten");`,
+		opts: []Option{
+			WithFetchSchemasQuery("SELECT tbl_name, sql FROM sqlite_schema WHERE type = 'table'"),
+			WithPerTableQueryFn(func(s string) string {
+				// We create a query specific for table b
+				switch s {
+				case "b":
+					return fmt.Sprintf("SELECT d FROM %s", s) // we are only selecting the d column
+				default:
+					return fmt.Sprintf("SELECT * FROM %s", s)
+				}
+			}),
+		},
+		exp: func() string {
+			var buf bytes.Buffer
+			buf.Write([]byte("a"))
+			buf.Write([]byte("CREATE TABLE a (a int)"))
+			buf.Write([]byte("123"))
+			buf.Write([]byte("456"))
+			buf.Write([]byte("b"))
+			buf.Write([]byte("CREATE TABLE b (c int, d text)"))
+			// buf.Write([]byte("10")) this column value is not part of the output
+			buf.Write([]byte("ten"))
+
+			h := sha1.New()
+			h.Write(buf.Bytes())
+
+			return hex.EncodeToString(h.Sum(nil))
+		}(),
+	}
+}
diff --git a/pkg/telemetry/storage/db_test.go b/pkg/telemetry/storage/db_test.go
new file mode 100644
index 0000000..cf333b4
--- /dev/null
+++ b/pkg/telemetry/storage/db_test.go
@@ -0,0 +1,52 @@
+package storage
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/textileio/go-tableland/pkg/telemetry"
+	"github.com/textileio/go-tableland/tests"
+)
+
+func TestCollectSqliteStore(t *testing.T) {
+	t.Run("state hash", func(t *testing.T) {
+		dbURI := tests.Sqlite3URI()
+		s, err := New(dbURI)
+		require.NoError(t, err)
+		telemetry.SetMetricStore(s)
+
+		err = telemetry.Collect(context.Background(), stateHash{})
+		require.NoError(t, err)
+
+		var version, timestamp, published int
+		var payload string
+		var typ telemetry.MetricType
+		row := s.sqlDB.QueryRowContext(context.Background(), "SELECT * FROM system_metrics LIMIT 1")
+		require.NoError(t, row.Scan(&version, &timestamp, &typ, &payload, &published))
+
+		require.Equal(t, 0, published)
+		require.Equal(t, telemetry.StateHashType, typ)
+
+		var stateHashMetric telemetry.StateHashMetric
+		require.NoError(t, json.Unmarshal([]byte(payload), &stateHashMetric))
+		require.Equal(t, stateHash{}.ChainID(), stateHashMetric.ChainID)
+		require.Equal(t, stateHash{}.BlockNumber(), stateHashMetric.BlockNumber)
+		require.Equal(t, stateHash{}.Hash(), stateHashMetric.Hash)
+	})
+}
+
+type stateHash struct{}
+
+func (h stateHash) ChainID() int64 {
+	return 1
+}
+
+func (h stateHash) BlockNumber() int64 {
+	return 1
+}
+
+func (h stateHash) Hash() string {
+	return "abcdefgh"
+}
diff --git a/pkg/telemetry/telemetry_test.go b/pkg/telemetry/telemetry_test.go
new file mode 100644
index 0000000..f312094
--- /dev/null
+++ b/pkg/telemetry/telemetry_test.go
@@ -0,0 +1,61 @@
+package telemetry
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCollectWithoutStore(t *testing.T) {
+	metricStore = nil
+	require.NoError(t, Collect(context.Background(), stateHash{}))
+}
+
+func TestCollectMockedtStore(t *testing.T) {
+	t.Run("state hash", func(t *testing.T) {
+		s := &store{}
+		metricStore = s
+
+		require.False(t, s.called)
+		err := Collect(context.Background(), stateHash{})
+		require.NoError(t, err)
+		require.True(t, s.called)
+	})
+}
+
+func TestCollectUnknownMetric(t *testing.T) {
+	s := &store{}
+	metricStore = s
+
+	err := Collect(context.Background(), struct{}{})
+	require.Error(t, err)
+	require.ErrorContains(t, err, "unknown metric")
+}
+
+type stateHash struct{}
+
+func (h stateHash) ChainID() int64 {
+	return 1
+}
+
+func (h stateHash) BlockNumber() int64 {
+	return 1
+}
+
+func (h stateHash) Hash() string {
+	return "abcdefgh"
+}
+
+type store struct {
+	called bool
+}
+
+func (db *store) StoreMetric(_ context.Context, _ Metric) error {
+	db.called = true
+	return nil
+}
+
+func (db *store) Close() error {
+	return nil
+}