[wip] Adds a Merkle Tree library and "snapshotting" of tables' rows

.github/workflows/test.yml

@@ -14,11 +14,11 @@ jobs:
   test:
     strategy:
       matrix:
-        go-version: [1.18.x, 1.19.x]
+        go-version: [1.19.x, 1.20.x]
         os: [ubuntu-latest]
         make-cmd: [test, test-replayhistory]
         exclude:
-          - go-version: 1.18.x
+          - go-version: 1.19x
             make-cmd: test-replayhistory
     runs-on: ${{ matrix.os }}
     steps:

.golangci.yml

@@ -59,5 +59,5 @@ run:
   timeout: 30m
 
   skip-dirs:
-    - "pkg/sqlstore/impl/system/internal/db"
+    - "pkg/sqlstore/impl/system/db"
     - "internal/router/controllers/apiv1"

Makefile

@@ -81,7 +81,7 @@ lint:
 .PHONY: lint
 
 # OpenAPI
-SPEC_URL=https://raw.githubusercontent.com/tablelandnetwork/docs/main/specs/validator/tableland-openapi-spec.yaml
+SPEC_URL=https://raw.githubusercontent.com/tablelandnetwork/docs/bcalza/merkle/specs/validator/tableland-openapi-spec.yaml
 APIV1=${PWD}/internal/router/controllers/apiv1
 gen-api-v1:
 	mkdir -p ${APIV1}

cmd/api/config.go

@@ -122,6 +122,16 @@ type ChainConfig struct {
 		MinBlockDepth int    `default:"5"`
 	}
 	HashCalculationStep int64 `default:"1000"`
+	MerkleTree          struct {
+		Enabled bool `default:"false"`
+		// We aim to have a new root calculated every 30 min.
+		// That means that the step should be configured to LeavesSnapshottingStep = 30*60/chain_avg_block_time_in_seconds.
+		// e.g. In Ethereum, chain_avg_block_time_in_seconds = 12s, so LeavesSnapshottingStep for Ethereum is 30*60/12 = 150.
+		LeavesSnapshottingStep int64 `default:"1000"`
+		// We aim to have a new root calculated every 30 min. Setting the default to half of that.
+		PublishingInterval   string `default:"15m"`
+		RootRegistryContract string `default:""`
+	}
 }
 
 func setupConfig() (*config, string) {

cmd/api/main.go

@@ -31,6 +31,7 @@ import (
 	epimpl "github.com/textileio/go-tableland/pkg/eventprocessor/impl"
 	executor "github.com/textileio/go-tableland/pkg/eventprocessor/impl/executor/impl"
 	"github.com/textileio/go-tableland/pkg/logging"
+
 	"github.com/textileio/go-tableland/pkg/metrics"
 	nonceimpl "github.com/textileio/go-tableland/pkg/nonce/impl"
 	"github.com/textileio/go-tableland/pkg/parsing"
@@ -47,6 +48,9 @@ import (
 	"github.com/textileio/go-tableland/pkg/telemetry/storage"
 	"github.com/textileio/go-tableland/pkg/wallet"
 	"go.opentelemetry.io/otel/attribute"
+
+	merklepublisher "github.com/textileio/go-tableland/pkg/merkletree/publisher"
+	merklepublisherimpl "github.com/textileio/go-tableland/pkg/merkletree/publisher/impl"
 )
 
 type moduleCloser func(ctx context.Context) error
@@ -83,10 +87,18 @@ func main() {
 		log.Fatal().Err(err).Msg("creating parser")
 	}
 
+	// Wiring
+	treeDatabaseURL := path.Join(dirPath, "trees.db")
+	treeStore, err := merklepublisherimpl.NewMerkleTreeStore(treeDatabaseURL)
+	if err != nil {
+		log.Fatal().Err(err).Msg("creating new merkle tree store")
+	}
+
 	// Chain stacks.
 	chainStacks, closeChainStacks, err := createChainStacks(
 		databaseURL,
 		parser,
+		treeStore,
 		config.Chains,
 		config.TableConstraints,
 		config.Analytics.FetchExtraBlockInfo)
@@ -105,7 +117,7 @@ func main() {
 	}
 
 	// HTTP API server.
-	closeHTTPServer, err := createAPIServer(config.HTTP, config.Gateway, parser, userStore, chainStacks)
+	closeHTTPServer, err := createAPIServer(config.HTTP, config.Gateway, parser, userStore, treeStore, chainStacks)
 	if err != nil {
 		log.Fatal().Err(err).Msg("creating HTTP server")
 	}
@@ -164,6 +176,7 @@ func createChainIDStack(
 	dbURI string,
 	executorsDB *sql.DB,
 	parser parsing.SQLValidator,
+	treeStore *merklepublisherimpl.MerkleTreeStore,
 	tableConstraints TableConstraints,
 	fetchExtraBlockInfo bool,
 ) (chains.ChainStack, error) {
@@ -268,6 +281,30 @@ func createChainIDStack(
 	if err := ep.Start(); err != nil {
 		return chains.ChainStack{}, fmt.Errorf("starting event processor: %s", err)
 	}
+
+	// starts Merkle Tree Publisher
+	var merkleRootPublisher *merklepublisher.MerkleRootPublisher
+	if config.MerkleTree.Enabled {
+		scAddress := common.HexToAddress(config.MerkleTree.RootRegistryContract)
+		rootRegistry, err := merklepublisherimpl.NewMerkleRootRegistryEthereum(conn, scAddress, wallet, tracker)
+		if err != nil {
+			return chains.ChainStack{}, fmt.Errorf("creating merkle root registry: %s", err)
+		}
+
+		merkleTreePublishingInterval, err := time.ParseDuration(config.MerkleTree.PublishingInterval)
+		if err != nil {
+			return chains.ChainStack{}, fmt.Errorf("parsing merkle tree publishing interval: %s", err)
+		}
+
+		merkleRootPublisher = merklepublisher.NewMerkleRootPublisher(
+			merklepublisherimpl.NewLeavesStore(systemStore),
+			treeStore,
+			rootRegistry,
+			merkleTreePublishingInterval,
+		)
+		merkleRootPublisher.Start()
+	}
+
 	return chains.ChainStack{
 		Store:          systemStore,
 		Registry:       registry,
@@ -284,6 +321,11 @@ func createChainIDStack(
 			if err := systemStore.Close(); err != nil {
 				return fmt.Errorf("closing system store for chain_id %d: %s", config.ChainID, err)
 			}
+
+			if config.MerkleTree.Enabled {
+				merkleRootPublisher.Close()
+			}
+
 			return nil
 		},
 	}, nil
@@ -414,6 +456,7 @@ func createParser(queryConstraints QueryConstraints) (parsing.SQLValidator, erro
 func createChainStacks(
 	databaseURL string,
 	parser parsing.SQLValidator,
+	treeStore *merklepublisherimpl.MerkleTreeStore,
 	chainsConfig []ChainConfig,
 	tableConstraintsConfig TableConstraints,
 	fetchExtraBlockInfo bool,
@@ -440,6 +483,7 @@ func createChainStacks(
 			databaseURL,
 			executorsDB,
 			parser,
+			treeStore,
 			tableConstraintsConfig,
 			fetchExtraBlockInfo)
 		if err != nil {
@@ -480,6 +524,7 @@ func createAPIServer(
 	gatewayConfig GatewayConfig,
 	parser parsing.SQLValidator,
 	userStore *user.UserStore,
+	treeStore *merklepublisherimpl.MerkleTreeStore,
 	chainStacks map[tableland.ChainID]chains.ChainStack,
 ) (moduleCloser, error) {
 	instrUserStore, err := sqlstoreimpl.NewInstrumentedUserStore(userStore)
@@ -519,6 +564,7 @@ func createAPIServer(
 	router, err := router.ConfiguredRouter(
 		mesaService,
 		systemService,
+		treeStore,
 		httpConfig.MaxRequestPerInterval,
 		rateLimInterval,
 		supportedChainIDs,

docker/deployed/staging/api/config.json

@@ -76,7 +76,13 @@
         "StuckInterval": "10m",
         "MinBlockDepth": 0
       },
-      "HashCalculationStep": 100
+      "HashCalculationStep": 100,
+      "MerkleTree": {
+        "Enabled" : true,
+        "LeavesSnapshottingStep" : 5,
+        "PublishingInterval" : "5m",
+        "RootRegistryContract" : "0x8065b18fDF17E6180614308bCFb798E877A4c291"
+      }
     }
   ]
 }

go.mod

@@ -1,6 +1,6 @@
 module github.com/textileio/go-tableland
 
-go 1.18
+go 1.19
 
 require (
 	cloud.google.com/go/bigquery v1.45.0
@@ -24,12 +24,14 @@ require (
 	github.com/stretchr/testify v1.8.2
 	github.com/tablelandnetwork/sqlparser v0.0.0-20221230162331-b318f234cefd
 	github.com/textileio/cli v1.0.2
+	go.etcd.io/bbolt v1.3.6
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.37.0
 	go.opentelemetry.io/otel v1.14.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.37.0
 	go.opentelemetry.io/otel/metric v0.37.0
 	go.opentelemetry.io/otel/sdk/metric v0.37.0
 	go.uber.org/atomic v1.10.0
+	golang.org/x/crypto v0.6.0
 	golang.org/x/sync v0.1.0
 )
 
@@ -103,7 +105,6 @@ require (
 	go.opentelemetry.io/otel/trace v1.14.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	go.uber.org/zap v1.19.0 // indirect
-	golang.org/x/crypto v0.6.0 // indirect
 	golang.org/x/net v0.7.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20221014153046-6fdb5e3db783 // indirect
 	golang.org/x/sys v0.5.0 // indirect

go.sum

@@ -1244,6 +1244,7 @@ gitlab.com/nyarla/go-crypt v0.0.0-20160106005555-d9a5dc2b789b/go.mod h1:T3BPAOm2
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
+go.etcd.io/bbolt v1.3.6 h1:/ecaJf0sk1l4l6V4awd65v2C3ILy7MSj+s/x1ADCIMU=
 go.etcd.io/bbolt v1.3.6/go.mod h1:qXsaaIqmgQH0T+OPdb99Bf+PKfBBQVAdyD6TY9G8XM4=
 go.etcd.io/etcd v0.5.0-alpha.5.0.20200910180754-dd1b699fc489/go.mod h1:yVHk9ub3CSBatqGNg7GRmsnfLWtoW60w4eDYfh7vHDg=
 go.etcd.io/etcd/api/v3 v3.5.0/go.mod h1:cbVKeC6lCfl7j/8jBhAK6aIYO9XOjdptoxU/nLQcPvs=

internal/router/controllers/apiv1/api_proof.go

@@ -0,0 +1,19 @@
+/*
+ * Tableland Validator - OpenAPI 3.0
+ *
+ * In Tableland, Validators are the execution unit/actors of the protocol. They have the following responsibilities: - Listen to on-chain events to materialize Tableland-compliant SQL queries in a database engine (currently, SQLite by default). - Serve read-queries (e.g: SELECT * FROM foo_69_1) to the external world. - Serve state queries (e.g. list tables, get receipts, etc) to the external world.  In the 1.0.0 release of the Tableland Validator API, we've switched to a design first approach! You can now help us improve the API whether it's by making changes to the definition itself or to the code. That way, with time, we can improve the API in general, and expose some of the new features in OAS3.
+ *
+ * API version: 1.0.0
+ * Contact: [email protected]
+ * Generated by: Swagger Codegen (https://github.com/swagger-api/swagger-codegen.git)
+ */
+package apiv1
+
+import (
+	"net/http"
+)
+
+func QueryProof(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Content-Type", "application/json; charset=UTF-8")
+	w.WriteHeader(http.StatusOK)
+}

internal/router/controllers/apiv1/model_proof.go

@@ -0,0 +1,15 @@
+/*
+ * Tableland Validator - OpenAPI 3.0
+ *
+ * In Tableland, Validators are the execution unit/actors of the protocol. They have the following responsibilities: - Listen to on-chain events to materialize Tableland-compliant SQL queries in a database engine (currently, SQLite by default). - Serve read-queries (e.g: SELECT * FROM foo_69_1) to the external world. - Serve state queries (e.g. list tables, get receipts, etc) to the external world.  In the 1.0.0 release of the Tableland Validator API, we've switched to a design first approach! You can now help us improve the API whether it's by making changes to the definition itself or to the code. That way, with time, we can improve the API in general, and expose some of the new features in OAS3.
+ *
+ * API version: 1.0.0
+ * Contact: [email protected]
+ * Generated by: Swagger Codegen (https://github.com/swagger-api/swagger-codegen.git)
+ */
+package apiv1
+
+type Proof struct {
+
+	Proof []string `json:"proof,omitempty"`
+}

internal/router/controllers/apiv1/routers.go

@@ -62,6 +62,13 @@ var routes = Routes{
 		Health,
 	},
 
+	Route{
+		"QueryProof",
+		strings.ToUpper("Get"),
+		"/api/v1/proof/{chainId}/{tableId}/{row}",
+		QueryProof,
+	},
+
 	Route{
 		"QueryByStatement",
 		strings.ToUpper("Get"),

internal/router/controllers/controller.go

@@ -2,8 +2,10 @@ package controllers
 
 import (
 	"context"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"math/big"
 	"net/http"
 	"strconv"
 	"strings"
@@ -20,6 +22,7 @@ import (
 	"github.com/textileio/go-tableland/internal/system"
 	"github.com/textileio/go-tableland/internal/tableland"
 	"github.com/textileio/go-tableland/pkg/errors"
+	"github.com/textileio/go-tableland/pkg/merkletree"
 	"github.com/textileio/go-tableland/pkg/tables"
 	"github.com/textileio/go-tableland/pkg/telemetry"
 )
@@ -29,15 +32,22 @@ type SQLRunner interface {
 	RunReadQuery(ctx context.Context, stmt string) (*tableland.TableData, error)
 }
 
+// MerkleTreeGetter defines the API for fetching a merkle tree.
+type MerkleTreeGetter interface {
+	Get(chainID int64, tableID *big.Int, blockNumber int64) (*merkletree.MerkleTree, error)
+}
+
 // Controller defines the HTTP handlers for interacting with user tables.
 type Controller struct {
+	treeStore     MerkleTreeGetter
 	runner        SQLRunner
 	systemService system.SystemService
 }
 
 // NewController creates a new Controller.
-func NewController(runner SQLRunner, svc system.SystemService) *Controller {
+func NewController(runner SQLRunner, svc system.SystemService, treeStore MerkleTreeGetter) *Controller {
 	return &Controller{
+		treeStore:     treeStore,
 		runner:        runner,
 		systemService: svc,
 	}
@@ -253,8 +263,15 @@ func (c *Controller) GetReceiptByTransactionHash(rw http.ResponseWriter, r *http
 	receipt, exists, err := c.systemService.GetReceiptByTransactionHash(ctx, txnHash)
 	if err != nil {
 		rw.Header().Set("Content-Type", "application/json")
-		rw.WriteHeader(http.StatusBadRequest)
 		log.Ctx(ctx).Error().Err(err).Msg("get receipt by transaction hash")
+
+		if strings.Contains(err.Error(), "database table is locked") ||
+			strings.Contains(err.Error(), "database schema is locked") {
+			rw.WriteHeader(http.StatusLocked)
+		} else {
+			rw.WriteHeader(http.StatusBadRequest)
+		}
+
 		_ = json.NewEncoder(rw).Encode(errors.ServiceError{Message: "Get receipt by transaction hash failed"})
 		return
 	}
@@ -551,6 +568,42 @@ func (c *Controller) GetTableQuery(rw http.ResponseWriter, r *http.Request) {
 	_, _ = rw.Write(formatted)
 }
 
+// GetProof handles the GET /proof/{chainId}/{tableId}/{row} call.
+func (c *Controller) GetProof(rw http.ResponseWriter, r *http.Request) {
+	rw.Header().Set("Content-Type", "application/json")
+
+	vars := mux.Vars(r)
+	chainID, tableID, encodedRow := vars["chainId"], vars["tableId"], vars["row"]
+
+	row, err := hex.DecodeString(encodedRow)
+	if err != nil {
+		return
+	}
+
+	chainIDInt, err := strconv.ParseInt(chainID, 10, 0)
+	if err != nil {
+		return
+	}
+
+	tableIDInt, err := strconv.ParseInt(tableID, 10, 0)
+	if err != nil {
+		return
+	}
+
+	tree, err := c.treeStore.Get(chainIDInt, big.NewInt(tableIDInt), 0)
+	if err != nil {
+		return
+	}
+
+	found, proof := tree.GetProof(row)
+	if !found {
+		rw.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	_ = json.NewEncoder(rw).Encode(apiv1.Proof{Proof: proof.Hex()})
+}
+
 func (c *Controller) runReadRequest(
 	ctx context.Context,
 	stm string,