Repository: tablelandnetwork/go-tableland
Pull request #299 (2022-09-28 17:41:53)

Replay of historic production Tableland events 
# Summary

This PR includes a new test that replays all Tableland history and checks that each chain's last Tableland state hash matches a defined expected hash.

---

This test aims to detect two main problematic situations that are hard to see via PRs:

**Detect non-deterministic execution**

If any event execution isn’t deterministic, it will create a snowball effect leading to a different state hash that won’t match the expected one.

Any validator that syncs from scratch should always arrive at the same state hash in each chain after it syncs to the tip. This test enforces this invariant.

**Detect if fixes or new features unintentionally change history**

If we make a bug fix or implement a new feature, there’s the chance that, under these new rules, the execution of historical events can lead to a different state. This is problematic since a new validator doing a cold-sync would have a different state hash than validators that simply update the version.

Today, we should expect that all fixes and features don’t change the end result of executing historical events. Of course, this will be relaxed when we implement fixes or features that activate at some particular height of chains (and forward). As in, having a multi-versioned execution pipeline that we should do eventually.

Even in the future, this test is already covering what holds true. With a multi-versioned execution pipeline, this test would properly detect that previous history is still backward compatible with new features that should activate further in time.

This topic is a bit of a rabbit hole for this PR, so I’ll stop here. We can return to this topic when we work in this multi-versioned execution pipeline; I just wanted to surface that this PR is already adding test coverage around it.

---

Before starting this PR, it felt pretty obvious that we’d already detected some problems, and that was the case:

- It detected a bug that caused a panic; this was fixed (explained in PR comments)
- It surfaced an obvious optimization that allowed us to replay the history 2x faster (explained in PR comments)

Some days ago, we had a situation where a previous version of the parser allowed some non-deterministic function calls. To double-check this test would have detected that situation, I downgraded the `sqlparser` dependency to the buggy version. I verified that this test detects a mismatch between the expected and calculated hash. More concretely, each test run produced different hashes, a signal of non-deterministic execution.

---

This test also serves as a building block for further work regarding benchmarking the validator since replaying production history might be the best way to do it. I’ll soon create a GH issue to track that work and drop some ideas on how standing on the shoulders of this test, we can discover other interesting information.

As quick information, here’re some facts:

- The test runs using an in-memory SQLite database.
- Hashing every X block is disabled; we only hash at the end since the last hash would have the snowball effect of any previous divergence. [Note about this later]
- The historical data used in the test contains an ~up-to-date version of the current history, totaling 192,770 events.
- The test without the `-race` flag (thus no runtime overhead) takes ~60s to run in my machine.

This means we could process ~3200 events per second (with an in-memory SQLite).

This is a good number to know since measuring the max throughput of our Tableland event execution logic. Of course, running with a file-based SQLite will have less throughput, but that is related to SQLite and *probably* not code in `go-tableland`. 

As part of future benchmarking work, we can run this test both in-memory and with a real SQLite database to continue understanding how adding more layers affects performance.

The quick conclusion is that probably any performance improvement won’t come from any business logic in Go since 3k events/second is more than enough for a real situation today. The bottleneck will probably live in the interaction with SQLite regarding its single-writer constraint. (Again, rabbit hole here).

# Implementation overview

The implementation is a new Go test that replays the history of all the known events, so it’s probably the complete e2e execution since it uses real data.

The following is a high-level description of how this works:

- The test relies on a `testdata/evm_history.db.zst` file, a zstandard compressed SQLite database containing a `system_evm_events` extracted from a real production database.
- It spins up the complete execution stack for all chains and executes all the events until it reaches the tip.
- For the above to work, the `EventFeed` now also accepts a new `sqlitechainclient`, another implementation of an Ethereum API. Apart from the Alchemy/Infura backend, you can “plug” this new component with the same API but provides block headers and logs backed by SQLite. The `EventFeed` is agnostic on how it receives events. (More in PR comments)
- After we detect that all chains are synced, the state hash is calculated for each chain and compared to fixed values which are the source of truth.

Notes:

- The test automatically detects all the existing chains in history and creates the appropriate execution stack. If we support new chains in the future, it’d be automatically detected.
- There’s a new `make generate-history-db` that helps to update the `testdata/evm_history.db.zst` based on a new `database.db`. In the future, we should keep updating the historical events to keep “up to date” throughout the continuously built history.
- The uncompressed SQLite database is ~200MiB, which is too big to commit to Git. The compressed size is ~23MiB, which is much better. The test automatically decompresses the database, uses it to run it, and is later discarded.

As mentioned before, running this test in my machine without `-race` takes ~60s. With `-race` takes ~140s. This was relevant to plan on what to do for CI, which I decided to work this way:

- This new “run history” test only runs if you don’t provide the `-short` flag, like `go test -short`.
- In CI, I added an extra dimension to the test matrix, so at the same time, we’re running all the tests but this one in a job and the “replay history” test in a separate job that runs in parallel. (See the new "Test / test (1.19.x, ubuntu-latest, test-replayhistory)" item in the actions summary)
- This allows for overlapping both runs, so we minimize adding minutes to the complete execution of CI (see PR comments for more details).
- In CI, we run this new long test with `-race`. It might take longer, but since it's syncing all the events in all the chains as fast as possible, it's indirectly a great test case to detect potential dataraces.

# Implementation details and review orientation

[WIP - After creating PR comments, I’ll add here some suggestions about the review order of files]

sanderpick: Great PR description as always. Nice idea with `sqlitechainclient` impl. Would hashing every X block help pin point where some indeterminism arrises? Or is that tradeoff not worth the extra overhead? 
jsign: > Great PR description as always. Nice idea with `sqlitechainclient` impl. Would hashing every X block help pin point where some indeterminism arrises? Or is that tradeoff not worth the extra overhead?

To pin point a specific block we'd have to run the hashes for every block, and that's quite heavy. For this test, I want to detect that there's a problem _as fast as possible. So if the final hash differs, we know there's a problem somewhere. Since that situation should only happen rarely, I prefer the test to run as fast as possible.

But if we find a case where we detected a problem, we can do further work based on this test to, now that we know there's some hairy thing going on, run the hashes ever X blocks and start narrowing down _where_ is the problem.

jsign: As an extra comment, if we'd the feature of only calculating hashes of tables that changes since the previous calculation maybe that would make this less heavy. So we'll see if at some point it can be "cheap enough" to make it the default to find where it is.
sanderpick: Makes sense 💯 

Patch:
diff --git a/.gitignore b/.gitignore
index bd92c09..876ea11 100644
--- a/.gitignore
+++ b/.gitignore
@@ -18,6 +18,7 @@
 docker/**/*.db*
 docker/deployed/**/prometheus
 docker/deployed/**/grafana/data
+evm_history.db
 
 # Project
 .env*
diff --git a/Makefile b/Makefile
index f754f7b..25ee7e1 100644
--- a/Makefile
+++ b/Makefile
@@ -42,6 +42,13 @@ clean-mocks:
 	rm -rf mocks
 .PHONY: clean-mocks
 
+EVM_EVENTS_ORIGIN:="docker/deployed/testnet/api/backup_database.db"
+EVM_EVENTS_TARGET:="pkg/eventprocessor/impl/testdata/evm_history.db"
+generate-history-db:
+	rm -f ${EVM_EVENTS_TARGET}
+	sqlite3 ${EVM_EVENTS_ORIGIN} 'ATTACH DATABASE ${EVM_EVENTS_TARGET} as target' 'CREATE TABLE target.system_evm_events as select * from system_evm_events'
+	zstd -f ${EVM_EVENTS_TARGET}
+	rm ${EVM_EVENTS_TARGET}
 
 # Build 
 
@@ -64,9 +71,12 @@ image:
 # Test
 
 test: 
-	go test ./... -race
+	go test ./... -short -race
 .PHONY: test
 
+test-replayhistory:
+	go test ./pkg/eventprocessor/impl -run=TestReplayProductionHistory -race
+
 # Lint
 
 lint:
diff --git a/cmd/api/main.go b/cmd/api/main.go
index 57c8fc6..eef9aa0 100644
--- a/cmd/api/main.go
+++ b/cmd/api/main.go
@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"crypto/tls"
+	"database/sql"
 	"encoding/base64"
 	"fmt"
 	"net/http"
@@ -10,8 +11,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/XSAM/otelsql"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/ethclient"
+	_ "github.com/mattn/go-sqlite3"
 	"github.com/rs/zerolog/log"
 	"github.com/textileio/cli"
 	"github.com/textileio/go-tableland/buildinfo"
@@ -39,6 +42,7 @@ import (
 	"github.com/textileio/go-tableland/pkg/telemetry/publisher"
 	"github.com/textileio/go-tableland/pkg/telemetry/storage"
 	"github.com/textileio/go-tableland/pkg/wallet"
+	"go.opentelemetry.io/otel/attribute"
 )
 
 func main() {
@@ -77,6 +81,18 @@ func main() {
 		log.Fatal().Err(err).Msg("instrumenting sql validator")
 	}
 
+	executorsDB, err := otelsql.Open("sqlite3", databaseURL)
+	if err != nil {
+		log.Fatal().Err(err).Msg("creating executors db")
+	}
+	executorsDB.SetMaxIdleConns(0)
+	executorsDB.SetMaxOpenConns(1)
+	if err := otelsql.RegisterDBStatsMetrics(
+		executorsDB,
+		otelsql.WithAttributes(attribute.String("name", "executors"))); err != nil {
+		log.Fatal().Err(err).Msg("registering executors db stats")
+	}
+
 	chainStacks := map[tableland.ChainID]chains.ChainStack{}
 	for _, chainCfg := range config.Chains {
 		if _, ok := chainStacks[chainCfg.ChainID]; ok {
@@ -85,6 +101,7 @@ func main() {
 		chainStack, err := createChainIDStack(
 			chainCfg,
 			databaseURL,
+			executorsDB,
 			parser,
 			config.TableConstraints,
 			config.Analytics.FetchExtraBlockInfo)
@@ -270,12 +287,16 @@ func main() {
 		if config.TelemetryPublisher.Enabled {
 			metricsPublisher.Close()
 		}
+		if err := executorsDB.Close(); err != nil {
+			log.Error().Err(err).Msg("closing executors db")
+		}
 	})
 }
 
 func createChainIDStack(
 	config ChainConfig,
 	dbURI string,
+	executorsDB *sql.DB,
 	parser parsing.SQLValidator,
 	tableConstraints TableConstraints,
 	fetchExtraBlockInfo bool,
@@ -342,7 +363,7 @@ func createChainIDStack(
 
 	acl := impl.NewACL(systemStore, registry)
 
-	ex, err := executor.NewExecutor(config.ChainID, dbURI, parser, tableConstraints.MaxRowCount, acl)
+	ex, err := executor.NewExecutor(config.ChainID, executorsDB, parser, tableConstraints.MaxRowCount, acl)
 	if err != nil {
 		return chains.ChainStack{}, fmt.Errorf("creating txn processor: %s", err)
 	}
diff --git a/cmd/foo/main.go b/cmd/foo/main.go
new file mode 100644
index 0000000..1e1a1a6
--- /dev/null
+++ b/cmd/foo/main.go
@@ -0,0 +1,20 @@
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"log"
+
+	_ "github.com/mattn/go-sqlite3"
+)
+
+func main() {
+	db, _ := sql.Open("sqlite3", "db.db")
+
+	var f float64
+	err := db.QueryRow("SELECT 0.240066230297088;").Scan(&f)
+	if err != nil {
+		log.Fatal(err)
+	}
+	fmt.Println(f)
+}
diff --git a/pkg/eventprocessor/eventfeed/eventfeed.go b/pkg/eventprocessor/eventfeed/eventfeed.go
index 3eeea07..d57edc4 100644
--- a/pkg/eventprocessor/eventfeed/eventfeed.go
+++ b/pkg/eventprocessor/eventfeed/eventfeed.go
@@ -15,7 +15,6 @@ import (
 
 // ChainClient provides basic apis for an EventFeed.
 type ChainClient interface {
-	SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error)
 	FilterLogs(ctx context.Context, query ethereum.FilterQuery) ([]types.Log, error)
 	HeaderByNumber(ctx context.Context, block *big.Int) (*types.Header, error)
 }
diff --git a/pkg/eventprocessor/eventfeed/impl/sqlitechainclient/sqlitechainclient.go b/pkg/eventprocessor/eventfeed/impl/sqlitechainclient/sqlitechainclient.go
new file mode 100644
index 0000000..6637f2a
--- /dev/null
+++ b/pkg/eventprocessor/eventfeed/impl/sqlitechainclient/sqlitechainclient.go
@@ -0,0 +1,161 @@
+package sqlitechainclient
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"math/big"
+	"sync"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/rs/zerolog"
+	logger "github.com/rs/zerolog/log"
+	"github.com/textileio/go-tableland/internal/tableland"
+)
+
+// SQLiteChainClient is an eventfeed.ChainClient backed by a persisted history of EVM events in a SQLite database.
+// An EventFeed can be plugged to this backend instead of Alchemy/Infura.
+type SQLiteChainClient struct {
+	log     zerolog.Logger
+	db      *sql.DB
+	chainID tableland.ChainID
+
+	onceChainTip        sync.Once
+	chainTipBlockNumber int64
+}
+
+// New returns a new *SQLiteChainClient.
+func New(dbURI string, chainID tableland.ChainID) (*SQLiteChainClient, error) {
+	log := logger.With().
+		Str("component", "sqlitechainclient").
+		Int64("chain_id", int64(chainID)).
+		Logger()
+
+	db, err := sql.Open("sqlite3", dbURI)
+	if err != nil {
+		return nil, fmt.Errorf("opening db: %s", err)
+	}
+
+	return &SQLiteChainClient{
+		log:     log,
+		db:      db,
+		chainID: chainID,
+	}, nil
+}
+
+// FilterLogs returns the logs matching a particular filter.
+func (scc *SQLiteChainClient) FilterLogs(ctx context.Context, filter ethereum.FilterQuery) ([]types.Log, error) {
+	if len(filter.Addresses) != 1 {
+		return nil, fmt.Errorf("the query filter must have a single contract address filter")
+	}
+	if filter.BlockHash != nil {
+		return nil, fmt.Errorf("block_hash filter isn't supported")
+	}
+
+	query := `select address, topics, data, block_number, tx_hash, tx_index, block_hash, event_index
+	          from system_evm_events 
+			  where chain_id=?1 and 
+			        block_number between ?2 and ?3 and
+					address=?4
+			  order by block_number asc, event_index asc`
+	rows, err := scc.db.QueryContext(
+		ctx,
+		query,
+		scc.chainID, filter.FromBlock.Int64(), filter.ToBlock.Int64(), filter.Addresses[0].Hex())
+	if err != nil {
+		return nil, fmt.Errorf("get filters in range: %s", err)
+	}
+	defer func() {
+		if err := rows.Close(); err != nil {
+			scc.log.Error().Err(err).Msg("closing query")
+		}
+	}()
+
+	var logs []types.Log
+	for rows.Next() {
+		if rows.Err() != nil {
+			return nil, fmt.Errorf("get row: %s", rows.Err())
+		}
+		var address, txHash, blockHash string
+		var data, topicsJSON []byte
+		var blockNumber uint64
+		var txIndex, eventIndex uint
+		if err := rows.Scan(
+			&address,
+			&topicsJSON,
+			&data,
+			&blockNumber,
+			&txHash,
+			&txIndex,
+			&blockHash,
+			&eventIndex); err != nil {
+			return nil, fmt.Errorf("scan row: %s", err)
+		}
+
+		var topicsHex []string
+		if err := json.Unmarshal(topicsJSON, &topicsHex); err != nil {
+			return nil, fmt.Errorf("unmarshal json topics: %s", err)
+		}
+		topics := make([]common.Hash, len(topicsHex))
+		for i, topicHex := range topicsHex {
+			topics[i] = common.HexToHash(topicHex)
+		}
+		logs = append(logs, types.Log{
+			Address:     common.HexToAddress(address),
+			Topics:      topics,
+			Data:        data,
+			BlockNumber: blockNumber,
+			TxHash:      common.HexToHash(txHash),
+			TxIndex:     txIndex,
+			BlockHash:   common.HexToHash(blockHash),
+			Index:       eventIndex,
+		})
+	}
+
+	return logs, nil
+}
+
+// HeaderByNumber returns the block header of the chain.
+// Note that it can only be called with block==nil (i.e: last known block) since the underlying SQLite database isn't
+// a full replication of the underlying chain.
+func (scc *SQLiteChainClient) HeaderByNumber(ctx context.Context, block *big.Int) (*types.Header, error) {
+	if block != nil {
+		return nil, errors.New("the current implementation only allows returning the latest block number")
+	}
+
+	scc.onceChainTip.Do(func() {
+		blockNumber, err := scc.getChainTipBlockNumber(ctx)
+		if err != nil {
+			scc.log.Error().Err(err).Msg("loading chain tip block number")
+			scc.chainTipBlockNumber = -1
+			scc.onceChainTip = sync.Once{} // Reset to retry in the next `HeaderByNumber(...)` call
+			return
+		}
+		scc.chainTipBlockNumber = blockNumber
+	})
+	if scc.chainTipBlockNumber == -1 {
+		return nil, fmt.Errorf("chain tip block number couldn't be loaded")
+	}
+
+	return &types.Header{
+		Number: big.NewInt(scc.chainTipBlockNumber),
+	}, nil
+}
+
+func (scc *SQLiteChainClient) getChainTipBlockNumber(ctx context.Context) (int64, error) {
+	query := "select block_number from system_evm_events where chain_id=?1 order by block_number desc limit 1"
+	row := scc.db.QueryRowContext(ctx, query, scc.chainID)
+	if row.Err() == sql.ErrNoRows {
+		return 0, errors.New("no blocks found")
+	}
+	var blockNumber int64
+	if err := row.Scan(&blockNumber); err != nil {
+		return 0, fmt.Errorf("reading block_number column: %s", err)
+	}
+
+	return blockNumber, nil
+}
diff --git a/pkg/eventprocessor/impl/eventprocessor.go b/pkg/eventprocessor/impl/eventprocessor.go
index cffac82..88c3fc4 100644
--- a/pkg/eventprocessor/impl/eventprocessor.go
+++ b/pkg/eventprocessor/impl/eventprocessor.go
@@ -278,7 +278,10 @@ func (ep *EventProcessor) executeBlock(ctx context.Context, block eventfeed.Bloc
 	if err := bs.Commit(); err != nil {
 		return fmt.Errorf("committing changes: %s", err)
 	}
-	ep.log.Debug().Int64("height", block.BlockNumber).Msg("new last processed height")
+	ep.log.Debug().
+		Int64("height", block.BlockNumber).
+		Int64("exec_ms", time.Since(start).Milliseconds()).
+		Msg("new last processed height")
 
 	ep.mLastProcessedHeight.Store(block.BlockNumber)
 	ep.mBlockExecutionLatency.Record(ctx, time.Since(start).Milliseconds(), ep.mBaseLabels...)
diff --git a/pkg/eventprocessor/impl/executor/impl/blockscope.go b/pkg/eventprocessor/impl/executor/impl/blockscope.go
index 2445378..2387567 100644
--- a/pkg/eventprocessor/impl/executor/impl/blockscope.go
+++ b/pkg/eventprocessor/impl/executor/impl/blockscope.go
@@ -212,7 +212,7 @@ func (bs *blockScope) StateHash(ctx context.Context, chainID tableland.ChainID)
 }
 
 // Close closes gracefully the block scope.
-// Clients should *always* `defer Close()` when opening batches.
+// Clients should *always* `defer Close()` when opening block scopes.
 func (bs *blockScope) Close() error {
 	defer bs.closed()
 
diff --git a/pkg/eventprocessor/impl/executor/impl/executor.go b/pkg/eventprocessor/impl/executor/impl/executor.go
index 057dc5c..12f593d 100644
--- a/pkg/eventprocessor/impl/executor/impl/executor.go
+++ b/pkg/eventprocessor/impl/executor/impl/executor.go
@@ -7,14 +7,12 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/XSAM/otelsql"
 	"github.com/mattn/go-sqlite3"
 	"github.com/rs/zerolog"
 	logger "github.com/rs/zerolog/log"
 	"github.com/textileio/go-tableland/internal/tableland"
 	"github.com/textileio/go-tableland/pkg/eventprocessor/impl/executor"
 	"github.com/textileio/go-tableland/pkg/parsing"
-	"go.opentelemetry.io/otel/attribute"
 )
 
 // Executor executes chain events.
@@ -37,26 +35,12 @@ var _ executor.Executor = (*Executor)(nil)
 // NewExecutor returns a new Executor.
 func NewExecutor(
 	chainID tableland.ChainID,
-	dbURI string,
+	// dbURI string,
+	db *sql.DB,
 	parser parsing.SQLValidator,
 	maxTableRowCount int,
 	acl tableland.ACL,
 ) (*Executor, error) {
-	db, err := otelsql.Open("sqlite3", dbURI, otelsql.WithAttributes(
-		attribute.String("name", "processor"),
-		attribute.Int64("chain_id", int64(chainID)),
-	))
-	if err != nil {
-		return nil, fmt.Errorf("connecting to db: %s", err)
-	}
-	db.SetMaxIdleConns(0)
-	db.SetMaxOpenConns(1)
-	if err := otelsql.RegisterDBStatsMetrics(db, otelsql.WithAttributes(
-		attribute.String("name", "processor"),
-		attribute.Int64("chain_id", int64(chainID)),
-	)); err != nil {
-		return nil, fmt.Errorf("registering dbstats: %s", err)
-	}
 	if maxTableRowCount < 0 {
 		return nil, fmt.Errorf("maximum table row count is negative")
 	}
@@ -91,18 +75,22 @@ func (ex *Executor) NewBlockScope(ctx context.Context, newBlockNum int64) (execu
 	default:
 		panic("parallel block scope detected, this must never happen")
 	}
+	releaseBlockScope := func() { ex.chBlockScope <- struct{}{} }
 
 	txn, err := ex.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable, ReadOnly: false})
 	if err != nil {
+		releaseBlockScope()
 		return nil, fmt.Errorf("opening db transaction: %s", err)
 	}
 
 	// Check that the last processed height is strictly lower.
 	lastBlockNum, err := ex.getLastExecutedBlockNumber(ctx, txn)
 	if err != nil {
+		releaseBlockScope()
 		return nil, fmt.Errorf("get last processed height: %s", err)
 	}
 	if lastBlockNum >= newBlockNum {
+		releaseBlockScope()
 		return nil, fmt.Errorf("latest executed block %d isn't smaller than new block %d", lastBlockNum, newBlockNum)
 	}
 
@@ -111,7 +99,7 @@ func (ex *Executor) NewBlockScope(ctx context.Context, newBlockNum int64) (execu
 		MaxTableRowCount: ex.maxTableRowCount,
 		BlockNumber:      newBlockNum,
 	}
-	bs := newBlockScope(txn, scopeVars, ex.parser, ex.acl, func() { ex.chBlockScope <- struct{}{} })
+	bs := newBlockScope(txn, scopeVars, ex.parser, ex.acl, releaseBlockScope)
 
 	return bs, nil
 }
@@ -153,14 +141,8 @@ func (ex *Executor) Close(ctx context.Context) error {
 	ex.closeOnce.Do(func() { close(ex.closed) })
 	select {
 	case <-ctx.Done():
-		if err := ex.db.Close(); err != nil {
-			ex.log.Error().Err(err).Msg("forced close of database connection")
-		}
-		return errors.New("closing ctx done")
+		return errors.New("executor was force closed due to timeout")
 	case <-ex.chBlockScope:
-		if err := ex.db.Close(); err != nil {
-			ex.log.Error().Err(err).Msg("closing database connection")
-		}
 		ex.log.Info().Msg("executor closed gracefully")
 		return nil
 	}


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 18aa923..e389664 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -16,6 +16,10 @@ jobs:
       matrix:
         go-version: [1.18.x, 1.19.x]
         os: [ubuntu-latest]
+        make-cmd: [test, test-replayhistory]
+        exclude:
+          - go-version: 1.18.x
+            make-cmd: test-replayhistory
     runs-on: ${{ matrix.os }}
     steps:
       - name: Install Go
@@ -35,4 +39,4 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-go-
       - name: Test
-        run: for i in 1 2 3; do make test && break || (if [ $i = 3 ]; then exit -1; else sleep 1; fi ); done
+        run: for i in 1 2 3; do make ${{ matrix.make-cmd }} && break || (if [ $i = 3 ]; then exit -1; else sleep 1; fi ); done
diff --git a/db.db b/db.db
new file mode 100644
index 0000000..e69de29
diff --git a/internal/system/impl/sqlstore_test.go b/internal/system/impl/sqlstore_test.go
index 57f9594..9b5f30b 100644
--- a/internal/system/impl/sqlstore_test.go
+++ b/internal/system/impl/sqlstore_test.go
@@ -2,6 +2,7 @@ package impl
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"math/big"
 	"testing"
@@ -34,8 +35,13 @@ func TestSystemSQLStoreService(t *testing.T) {
 
 	parser, err := parserimpl.New([]string{"system_", "registry"})
 	require.NoError(t, err)
+
+	db, err := sql.Open("sqlite3", dbURI)
+	require.NoError(t, err)
+	db.SetMaxOpenConns(1)
+
 	// populate the registry with a table
-	ex, err := executor.NewExecutor(1337, dbURI, parser, 0, nil)
+	ex, err := executor.NewExecutor(1337, db, parser, 0, nil)
 	require.NoError(t, err)
 	bs, err := ex.NewBlockScope(ctx, 0)
 	require.NoError(t, err)
@@ -105,8 +111,12 @@ func TestGetSchemaByTableName(t *testing.T) {
 	parser, err := parserimpl.New([]string{"system_", "registry"})
 	require.NoError(t, err)
 
+	db, err := sql.Open("sqlite3", dbURI)
+	require.NoError(t, err)
+	db.SetMaxOpenConns(1)
+
 	// populate the registry with a table
-	ex, err := executor.NewExecutor(1337, dbURI, parser, 0, nil)
+	ex, err := executor.NewExecutor(1337, db, parser, 0, nil)
 	require.NoError(t, err)
 	bs, err := ex.NewBlockScope(ctx, 0)
 	require.NoError(t, err)
@@ -162,8 +172,13 @@ func TestGetMetadata(t *testing.T) {
 
 	parser, err := parserimpl.New([]string{"system_", "registry"})
 	require.NoError(t, err)
+
+	db, err := sql.Open("sqlite3", dbURI)
+	require.NoError(t, err)
+	db.SetMaxOpenConns(1)
+
 	// populate the registry with a table
-	ex, err := executor.NewExecutor(1337, dbURI, parser, 0, nil)
+	ex, err := executor.NewExecutor(1337, db, parser, 0, nil)
 	require.NoError(t, err)
 	bs, err := ex.NewBlockScope(ctx, 0)
 	require.NoError(t, err)
diff --git a/internal/tableland/impl/mesa_test.go b/internal/tableland/impl/mesa_test.go
index fc4f33a..153b89e 100644
--- a/internal/tableland/impl/mesa_test.go
+++ b/internal/tableland/impl/mesa_test.go
@@ -952,7 +952,11 @@ func (b *tablelandSetupBuilder) build(t *testing.T) *tablelandSetup {
 	parser, err := parserimpl.New([]string{"system_", "registry", "sqlite_"}, b.parsingOpts...)
 	require.NoError(t, err)
 
-	ex, err := executor.NewExecutor(1337, dbURI, parser, 0, &aclHalfMock{store})
+	db, err := sql.Open("sqlite3", dbURI)
+	require.NoError(t, err)
+	db.SetMaxOpenConns(1)
+
+	ex, err := executor.NewExecutor(1337, db, parser, 0, &aclHalfMock{store})
 	require.NoError(t, err)
 
 	backend, addr, sc, auth, sk := testutil.Setup(t)
diff --git a/pkg/client/client_test.go b/pkg/client/client_test.go
index 0540c45..5c741cd 100644
--- a/pkg/client/client_test.go
+++ b/pkg/client/client_test.go
@@ -191,7 +191,11 @@ func setup(t *testing.T) clientCalls {
 	parser, err := parserimpl.New([]string{"system_", "registry", "sqlite_"})
 	require.NoError(t, err)
 
-	ex, err := executor.NewExecutor(1337, dbURI, parser, 0, &aclHalfMock{store})
+	db, err := sql.Open("sqlite3", dbURI)
+	require.NoError(t, err)
+	db.SetMaxOpenConns(1)
+
+	ex, err := executor.NewExecutor(1337, db, parser, 0, &aclHalfMock{store})
 	require.NoError(t, err)
 
 	backend, addr, _, _, sk := testutil.Setup(t)
diff --git a/pkg/eventprocessor/impl/eventprocessor_replayhistory_test.go b/pkg/eventprocessor/impl/eventprocessor_replayhistory_test.go
new file mode 100644
index 0000000..7ef8cd9
--- /dev/null
+++ b/pkg/eventprocessor/impl/eventprocessor_replayhistory_test.go
@@ -0,0 +1,207 @@
+//go:build !arm64
+
+package impl
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/klauspost/compress/zstd"
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/textileio/go-tableland/internal/tableland"
+	"github.com/textileio/go-tableland/pkg/eventprocessor"
+	"github.com/textileio/go-tableland/pkg/eventprocessor/eventfeed"
+	efimpl "github.com/textileio/go-tableland/pkg/eventprocessor/eventfeed/impl"
+	"github.com/textileio/go-tableland/pkg/eventprocessor/eventfeed/impl/sqlitechainclient"
+	executor "github.com/textileio/go-tableland/pkg/eventprocessor/impl/executor/impl"
+	"github.com/textileio/go-tableland/pkg/parsing"
+	parserimpl "github.com/textileio/go-tableland/pkg/parsing/impl"
+	"github.com/textileio/go-tableland/pkg/sqlstore/impl/system"
+	"github.com/textileio/go-tableland/tests"
+)
+
+func TestReplayProductionHistory(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skipf("skipping history replay execution because running -short tests")
+	}
+
+	expectedStateHashes := map[tableland.ChainID]string{
+		1:      "87b02f2755e043a7d7f544bb9bf79765115f9b58",
+		5:      "b6f5f703af0e92d8f28773a024ed45119cef8c61",
+		10:     "57555e08de5b37270ddad6c2cad2c1ae7ca6901e",
+		69:     "f17f7c999790277cc91351a798eea2d08abe4285",
+		137:    "241425c72e622bb8574f7fe15ccf251ddc5c6367",
+		420:    "923d03c96ca0c3fa848b651697cf925bcdee5eff",
+		80001:  "7b15ca5f14bc4e7475d48fface3ac1a7022e3a27",
+		421613: "df1fe80afc8d9fc0ae31057b82766dd82d81ad63",
+	}
+
+	historyDBURI := getHistoryDBURI(t)
+	// Launch the validator syncing all chains.
+	eps, waitFullSync := launchValidatorForAllChainsBackedByEVMHistory(t, historyDBURI)
+
+	// Wait for all of them to finish syncing.
+	waitFullSync()
+
+	// We compare the chain hash after full sync with the previous iteration calculated hash.
+	// These should always match. If that isn't the case, it means that the chain execution is non-deterministic.
+	ctx := context.Background()
+	for _, ep := range eps {
+		bs, err := ep.executor.NewBlockScope(ctx, ep.mLastProcessedHeight.Load()+1)
+		require.NoError(t, err)
+
+		hash, err := bs.StateHash(ctx, ep.chainID)
+		require.NoError(t, err)
+
+		assert.Equal(t, expectedStateHashes[ep.chainID], hash.Hash(),
+			"ChainID %d hash %s doesn't match %s", ep.chainID, hash.Hash(), expectedStateHashes[ep.chainID])
+		require.NoError(t, bs.Close())
+	}
+
+	// Do a graceful close, to double check closing works correctly without any blocking or delays.
+	for _, ep := range eps {
+		ep.Stop()
+	}
+}
+
+func launchValidatorForAllChainsBackedByEVMHistory(t *testing.T, historyDBURI string) ([]*EventProcessor, func()) {
+	dbURI := tests.Sqlite3URI()
+	parser, err := parserimpl.New([]string{"system_", "registry", "sqlite_"})
+	require.NoError(t, err)
+
+	db, err := sql.Open("sqlite3", dbURI)
+	require.NoError(t, err)
+	db.SetMaxIdleConns(0)
+	db.SetMaxOpenConns(1)
+
+	chains := getChains(t, historyDBURI)
+	eps := make([]*EventProcessor, len(chains))
+	for i, chain := range chains {
+		eps[i] = spinValidatorStackForChainID(t, dbURI, historyDBURI, parser, chain.chainID, chain.scAddress, db)
+	}
+
+	waitForSynced := func() {
+		var wg sync.WaitGroup
+		wg.Add(len(chains))
+		for i := range chains {
+			go func(i int) {
+				defer wg.Done()
+				for {
+					if eps[i].mLastProcessedHeight.Load() == chains[i].tipBlockNumber {
+						return
+					}
+					time.Sleep(time.Second)
+				}
+			}(i)
+		}
+		wg.Wait()
+	}
+
+	return eps, waitForSynced
+}
+
+func spinValidatorStackForChainID(
+	t *testing.T,
+	dbURI string,
+	historyDBURI string,
+	parser parsing.SQLValidator,
+	chainID tableland.ChainID,
+	scAddress common.Address,
+	db *sql.DB,
+) *EventProcessor {
+	ex, err := executor.NewExecutor(chainID, db, parser, 0, &aclMock{})
+	require.NoError(t, err)
+
+	systemStore, err := system.New(dbURI, chainID)
+	require.NoError(t, err)
+
+	eventBasedBackend, err := sqlitechainclient.New(historyDBURI, chainID)
+	require.NoError(t, err)
+
+	ef, err := efimpl.New(
+		systemStore,
+		chainID,
+		eventBasedBackend,
+		scAddress,
+		eventfeed.WithMinBlockDepth(0))
+	require.NoError(t, err)
+
+	ep, err := New(
+		parser,
+		ex,
+		ef,
+		chainID,
+		eventprocessor.WithHashCalcStep(1_000_000_000),
+	)
+	require.NoError(t, err)
+	require.NoError(t, ep.Start())
+
+	return ep
+}
+
+type chainIDWithTip struct {
+	chainID        tableland.ChainID
+	scAddress      common.Address
+	tipBlockNumber int64
+}
+
+func getChains(t *testing.T, historyDBURI string) []chainIDWithTip {
+	db, err := sql.Open("sqlite3", historyDBURI)
+	require.NoError(t, err)
+
+	rows, err := db.Query(`select chain_id, address, max(block_number) 
+                           from system_evm_events 
+                           group by chain_id, address
+                           order by chain_id, block_number`)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, rows.Close())
+	}()
+
+	var chains []chainIDWithTip
+	for rows.Next() {
+		require.NoError(t, rows.Err())
+		var chainID, blockNumber int64
+		var scAddress string
+		require.NoError(t, rows.Scan(&chainID, &scAddress, &blockNumber))
+		chains = append(chains, chainIDWithTip{
+			chainID:        tableland.ChainID(chainID),
+			scAddress:      common.HexToAddress(scAddress),
+			tipBlockNumber: blockNumber,
+		})
+	}
+
+	return chains
+}
+
+func getHistoryDBURI(t *testing.T) string {
+	zstdDB, err := os.Open("testdata/evm_history.db.zst")
+	require.NoError(t, err)
+
+	decoder, err := zstd.NewReader(zstdDB)
+	require.NoError(t, err)
+
+	// Create target database to decompress
+	historyDBFilePath := filepath.Join(t.TempDir(), "evm_history.db")
+	historyDB, err := os.OpenFile(historyDBFilePath, os.O_WRONLY|os.O_CREATE, 0o755)
+	require.NoError(t, err)
+
+	// Decompress
+	_, err = decoder.WriteTo(historyDB)
+	require.NoError(t, err)
+	require.NoError(t, historyDB.Close())
+
+	// Return full path of prepared database.
+	return fmt.Sprintf("file:%s?", historyDBFilePath)
+}
diff --git a/pkg/eventprocessor/impl/eventprocessor_test.go b/pkg/eventprocessor/impl/eventprocessor_test.go
index 651f30f..8e99718 100644
--- a/pkg/eventprocessor/impl/eventprocessor_test.go
+++ b/pkg/eventprocessor/impl/eventprocessor_test.go
@@ -317,7 +317,11 @@ func setup(t *testing.T) (
 	dbURI := tests.Sqlite3URI()
 	parser, err := parserimpl.New([]string{"system_", "registry", "sqlite_"})
 	require.NoError(t, err)
-	ex, err := executor.NewExecutor(chainID, dbURI, parser, 0, &aclMock{})
+
+	db, err := sql.Open("sqlite3", dbURI)
+	require.NoError(t, err)
+	db.SetMaxOpenConns(1)
+	ex, err := executor.NewExecutor(chainID, db, parser, 0, &aclMock{})
 	require.NoError(t, err)
 
 	systemStore, err := system.New(dbURI, tableland.ChainID(chainID))
diff --git a/pkg/eventprocessor/impl/executor/impl/executor_test.go b/pkg/eventprocessor/impl/executor/impl/executor_test.go
index 3479b65..a271437 100644
--- a/pkg/eventprocessor/impl/executor/impl/executor_test.go
+++ b/pkg/eventprocessor/impl/executor/impl/executor_test.go
@@ -193,7 +193,10 @@ func newExecutor(t *testing.T, rowsLimit int) (*Executor, string) {
 	dbURI := tests.Sqlite3URI()
 
 	parser := newParser(t, []string{})
-	exec, err := NewExecutor(1337, dbURI, parser, rowsLimit, &aclMock{})
+	db, err := sql.Open("sqlite3", dbURI)
+	require.NoError(t, err)
+	db.SetMaxOpenConns(1)
+	exec, err := NewExecutor(1337, db, parser, rowsLimit, &aclMock{})
 	require.NoError(t, err)
 
 	// Boostrap system store to run the db migrations.