398: Statediff missing parent blocks automatically.

.github/workflows/tests.yml

@@ -5,7 +5,7 @@ on:
 
 env:
   stack-orchestrator-ref: ${{ github.event.inputs.stack-orchestrator-ref  || 'e62830c982d4dfc5f3c1c2b12c1754a7e9b538f1'}}
-  ipld-eth-db-ref: ${{ github.event.inputs.ipld-eth-db-ref  || '66cd1d9e696cfa72f9d272927f9e945905c9f093' }}
+  ipld-eth-db-ref: ${{ github.event.inputs.ipld-eth-db-ref  || '1b922dbff350bfe2a9aec5fe82079e9d855ea7ed' }}
   GOPATH: /tmp/go
 
 jobs:
@@ -53,6 +53,17 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v2
 
+      - uses: actions/checkout@v3
+        with:
+          ref: ${{ env.ipld-eth-db-ref }}
+          repository: cerc-io/ipld-eth-db
+          path: "./ipld-eth-db/"
+          fetch-depth: 0
+
+      - name: Build ipld-eth-db
+        run: |
+          docker build -f ./ipld-eth-db/Dockerfile ./ipld-eth-db/ -t cerc/ipld-eth-db:local
+
       - name: Run docker compose
         run: |
           docker-compose up -d

cmd/geth/config.go

@@ -281,7 +281,7 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) {
 			NumWorkers:              ctx.Uint(utils.StateDiffWorkersFlag.Name),
 			WaitForSync:             ctx.Bool(utils.StateDiffWaitForSync.Name),
 			BackfillCheckPastBlocks: ctx.Uint64(utils.StateDiffBackfillCheckPastBlocks.Name),
-			BackfillMaxHeadGap:      ctx.Uint64(utils.StateDiffBackfillMaxHeadGap.Name),
+			BackfillMaxDepth:        ctx.Uint64(utils.StateDiffBackfillMaxDepth.Name),
 		}
 		utils.RegisterStateDiffService(stack, eth, &cfg.Eth, p, backend)
 	}

cmd/geth/main.go

@@ -177,7 +177,7 @@ var (
 		utils.StateDiffLogStatements,
 		utils.StateDiffCopyFrom,
 		utils.StateDiffBackfillCheckPastBlocks,
-		utils.StateDiffBackfillMaxHeadGap,
+		utils.StateDiffBackfillMaxDepth,
 		configFileFlag,
 	}, utils.NetworkFlags, utils.DatabasePathFlags)
 

cmd/utils/flags.go

@@ -1125,9 +1125,9 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server.
 		Usage: "The number of blocks behind the startup statediff position to check (and fill) for gaps when head tracking.",
 		Value: 7200,
 	}
-	StateDiffBackfillMaxHeadGap = &cli.Uint64Flag{
-		Name:  "statediff.backfillmaxheadgap",
-		Usage: "The maximum gap between the current statediff and head positions that can be backfilled.",
+	StateDiffBackfillMaxDepth = &cli.Uint64Flag{
+		Name:  "statediff.backfillmaxdepth",
+		Usage: "When statediffing head, the maximum number of missing parents that can be backfilled.",
 		Value: 7200,
 	}
 )

docker-compose.yml

@@ -5,7 +5,7 @@ services:
     restart: on-failure
     depends_on:
       - ipld-eth-db
-    image: git.vdb.to/cerc-io/ipld-eth-db/ipld-eth-db:v5.0.2-alpha
+    image: cerc/ipld-eth-db:local
     environment:
       DATABASE_USER: "vdbm"
       DATABASE_NAME: "cerc_testing"

statediff/config.go

@@ -37,7 +37,7 @@ type Config struct {
 	// Whether to enable writing state diffs directly to track blockchain head
 	EnableWriteLoop bool
 	// The maximum number of blocks to backfill when tracking head.
-	BackfillMaxHeadGap uint64
+	BackfillMaxDepth uint64
 	// The maximum number of blocks behind the startup position to check for gaps.
 	BackfillCheckPastBlocks uint64
 	// Size of the worker pool
@@ -66,12 +66,33 @@ func (p *Params) ComputeWatchedAddressesLeafPaths() {
 	}
 }
 
+func (p *Params) Copy() Params {
+	ret := Params{
+		IncludeBlock:    p.IncludeBlock,
+		IncludeReceipts: p.IncludeReceipts,
+		IncludeTD:       p.IncludeTD,
+		IncludeCode:     p.IncludeCode,
+	}
+	ret.WatchedAddresses = make([]common.Address, len(p.WatchedAddresses))
+	copy(ret.WatchedAddresses, p.WatchedAddresses)
+
+	return ret
+}
+
 // ParamsWithMutex allows to lock the parameters while they are being updated | read from
 type ParamsWithMutex struct {
 	Params
 	sync.RWMutex
 }
 
+// CopyParams returns a defensive copy of the Params
+func (p *ParamsWithMutex) CopyParams() Params {
+	p.RLock()
+	defer p.RUnlock()
+
+	return p.Params.Copy()
+}
+
 // Args bundles the arguments for the state diff builder
 type Args struct {
 	OldStateRoot, NewStateRoot, BlockHash common.Hash

statediff/indexer/database/dump/indexer.go

@@ -186,6 +186,7 @@ func (sdi *StateDiffIndexer) processHeader(tx *BatchTx, header *types.Header, he
 		UnclesHash:      header.UncleHash.String(),
 		Timestamp:       header.Time,
 		Coinbase:        header.Coinbase.String(),
+		Canonical:       true,
 	}
 	_, err := fmt.Fprintf(sdi.dump, "%+v\r\n", mod)
 	return headerID, err

statediff/indexer/database/file/csv_writer.go

@@ -234,7 +234,8 @@ func (csw *CSVWriter) upsertHeaderCID(header models.HeaderModel) {
 	var values []interface{}
 	values = append(values, header.BlockNumber, header.BlockHash, header.ParentHash, header.CID,
 		header.TotalDifficulty, header.NodeIDs, header.Reward, header.StateRoot, header.TxRoot,
-		header.RctRoot, header.UnclesHash, header.Bloom, strconv.FormatUint(header.Timestamp, 10), header.Coinbase)
+		header.RctRoot, header.UnclesHash, header.Bloom, strconv.FormatUint(header.Timestamp, 10), header.Coinbase,
+		header.Canonical)
 	csw.rows <- tableRow{schema.TableHeader, values}
 	metrics.IndexerMetrics.BlocksCounter.Inc(1)
 }

statediff/indexer/database/file/indexer.go

@@ -246,6 +246,7 @@ func (sdi *StateDiffIndexer) processHeader(header *types.Header, headerNode ipld
 		UnclesHash:      header.UncleHash.String(),
 		Timestamp:       header.Time,
 		Coinbase:        header.Coinbase.String(),
+		Canonical:       true,
 	})
 	return headerID
 }

statediff/indexer/database/file/sql_writer.go

@@ -140,8 +140,8 @@ const (
 	ipldInsert = "INSERT INTO ipld.blocks (block_number, key, data) VALUES ('%s', '%s', '\\x%x');\n"
 
 	headerInsert = "INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_ids, reward, " +
-		"state_root, tx_root, receipt_root, uncles_hash, bloom, timestamp, coinbase) VALUES " +
-		"('%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '\\x%x', %d, '%s');\n"
+		"state_root, tx_root, receipt_root, uncles_hash, bloom, timestamp, coinbase, canonical) VALUES " +
+		"('%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s', '\\x%x', %d, '%s', %t);\n"
 
 	uncleInsert = "INSERT INTO eth.uncle_cids (block_number, block_hash, header_id, parent_hash, cid, reward, index) VALUES " +
 		"('%s', '%s', '%s', '%s', '%s', '%s', %d);\n"
@@ -189,7 +189,7 @@ func (sqw *SQLWriter) upsertIPLDNode(blockNumber string, i ipld.IPLD) {
 func (sqw *SQLWriter) upsertHeaderCID(header models.HeaderModel) {
 	stmt := fmt.Sprintf(headerInsert, header.BlockNumber, header.BlockHash, header.ParentHash, header.CID,
 		header.TotalDifficulty, formatPostgresStringArray(header.NodeIDs), header.Reward, header.StateRoot, header.TxRoot,
-		header.RctRoot, header.UnclesHash, header.Bloom, header.Timestamp, header.Coinbase)
+		header.RctRoot, header.UnclesHash, header.Bloom, header.Timestamp, header.Coinbase, header.Canonical)
 	sqw.stmts <- []byte(stmt)
 	metrics.IndexerMetrics.BlocksCounter.Inc(1)
 }

statediff/indexer/database/sql/indexer.go

@@ -260,6 +260,7 @@ func (sdi *StateDiffIndexer) processHeader(tx *BatchTx, header *types.Header, he
 		UnclesHash:      header.UncleHash.String(),
 		Timestamp:       header.Time,
 		Coinbase:        header.Coinbase.String(),
+		Canonical:       true,
 	})
 }
 

statediff/indexer/database/sql/interfaces.go

@@ -49,6 +49,7 @@ type Statements interface {
 	MaxHeaderStm() string
 	ExistsHeaderStm() string
 	InsertHeaderStm() string
+	SetCanonicalHeaderStm() string
 	InsertUncleStm() string
 	InsertTxStm() string
 	InsertRctStm() string

statediff/indexer/database/sql/postgres/database.go

@@ -62,6 +62,12 @@ func (db *DB) InsertHeaderStm() string {
 	return schema.TableHeader.ToInsertStatement(db.upsert)
 }
 
+// SetCanonicalHeaderStm satisfies the sql.Statements interface
+// Stm == Statement
+func (db *DB) SetCanonicalHeaderStm() string {
+	return fmt.Sprintf("UPDATE %s SET canonical = false WHERE block_number = $1::BIGINT AND block_hash <> $2::TEXT AND canonical = true", schema.TableHeader.Name)
+}
+
 // InsertUncleStm satisfies the sql.Statements interface
 func (db *DB) InsertUncleStm() string {
 	return schema.TableUncle.ToInsertStatement(db.upsert)

statediff/indexer/database/sql/writer.go

@@ -92,6 +92,7 @@ func (w *Writer) maxHeader() (*models.HeaderModel, error) {
 		&model.Bloom,
 		&model.Timestamp,
 		&model.Coinbase,
+		&model.Canonical,
 	)
 	model.BlockNumber = strconv.FormatUint(number, 10)
 	model.TotalDifficulty = strconv.FormatUint(td, 10)
@@ -120,11 +121,22 @@ func (w *Writer) upsertHeaderCID(tx Tx, header models.HeaderModel) error {
 		header.UnclesHash,
 		header.Bloom,
 		header.Timestamp,
-		header.Coinbase)
+		header.Coinbase,
+		header.Canonical,
+	)
 	if err != nil {
 		return insertError{"eth.header_cids", err, w.db.InsertHeaderStm(), header}
 	}
 	metrics.IndexerMetrics.BlocksCounter.Inc(1)
+
+	_, err = tx.Exec(w.db.Context(), w.db.SetCanonicalHeaderStm(),
+		header.BlockNumber,
+		header.BlockHash,
+	)
+	if err != nil {
+		return insertError{"eth.header_cids", err, w.db.SetCanonicalHeaderStm(), header}
+	}
+
 	return nil
 }
 

statediff/indexer/models/models.go

@@ -41,6 +41,7 @@ type HeaderModel struct {
 	Bloom           []byte         `db:"bloom"`
 	Timestamp       uint64         `db:"timestamp"`
 	Coinbase        string         `db:"coinbase"`
+	Canonical       bool           `db:"canonical"`
 }
 
 // UncleModel is the db model for eth.uncle_cids

statediff/indexer/shared/schema/schema.go

@@ -54,6 +54,7 @@ var TableHeader = Table{
 		{Name: "bloom", Type: Dbytea},
 		{Name: "timestamp", Type: Dnumeric},
 		{Name: "coinbase", Type: Dvarchar},
+		{Name: "canonical", Type: Dboolean},
 	},
 	UpsertClause: OnConflict("block_number", "block_hash").Set(
 		"parent_hash",
@@ -68,6 +69,7 @@ var TableHeader = Table{
 		"bloom",
 		"timestamp",
 		"coinbase",
+		"canonical",
 	)}
 
 var TableStateNode = Table{