Skip to content

Commit

Permalink
Merge branch 'main' into security-update
Browse files Browse the repository at this point in the history
  • Loading branch information
mt-polygon-technology authored Mar 13, 2024
2 parents 2dec1d5 + 9ed8d9a commit f7ff8e3
Show file tree
Hide file tree
Showing 12 changed files with 1,435 additions and 280 deletions.
3 changes: 3 additions & 0 deletions config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,9 @@ type L1Config struct {
Timeout types.Duration `mapstructure:"Timeout"`
RetryPeriod types.Duration `mapstructure:"RetryPeriod"`
BlockBatchSize uint `mapstructure:"BlockBatchSize"`

// GenesisBlock represents the block number where PolygonValidium contract is deployed on L1
GenesisBlock uint64 `mapstructure:"GenesisBlock"`
}

// Load loads the configuration baseed on the cli context
Expand Down
1 change: 1 addition & 0 deletions config/default.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ DataCommitteeAddress = "0x68B1D87F95878fE05B998F19b66F4baba5De1aed"
Timeout = "1m"
RetryPeriod = "5s"
BlockBatchSize = "64"
GenesisBlock = "0"
[Log]
Environment = "development" # "production" or "development"
Expand Down
179 changes: 137 additions & 42 deletions db/db.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,16 @@ var (
// DB defines functions that a DB instance should implement
type DB interface {
BeginStateTransaction(ctx context.Context) (Tx, error)
Exists(ctx context.Context, key common.Hash) bool

StoreLastProcessedBlock(ctx context.Context, task string, block uint64, dbTx sqlx.ExecerContext) error
GetLastProcessedBlock(ctx context.Context, task string) (uint64, error)

StoreUnresolvedBatchKeys(ctx context.Context, bks []types.BatchKey, dbTx sqlx.ExecerContext) error
GetUnresolvedBatchKeys(ctx context.Context) ([]types.BatchKey, error)
DeleteUnresolvedBatchKeys(ctx context.Context, bks []types.BatchKey, dbTx sqlx.ExecerContext) error

Exists(ctx context.Context, key common.Hash) bool
GetOffChainData(ctx context.Context, key common.Hash, dbTx sqlx.QueryerContext) (types.ArgBytes, error)
StoreLastProcessedBlock(ctx context.Context, task string, block uint64, dbTx sqlx.ExecerContext) error
StoreOffChainData(ctx context.Context, od []types.OffChainData, dbTx sqlx.ExecerContext) error
}

Expand Down Expand Up @@ -50,47 +56,107 @@ func (db *pgDB) BeginStateTransaction(ctx context.Context) (Tx, error) {
return db.pg.BeginTxx(ctx, nil)
}

// StoreOffChainData stores and array of key values in the Db
func (db *pgDB) StoreOffChainData(ctx context.Context, od []types.OffChainData, dbTx sqlx.ExecerContext) error {
const storeOffChainDataSQL = `
INSERT INTO data_node.offchain_data (key, value)
// StoreLastProcessedBlock stores a record of a block processed by the synchronizer for named task
func (db *pgDB) StoreLastProcessedBlock(ctx context.Context, task string, block uint64, dbTx sqlx.ExecerContext) error {
const storeLastProcessedBlockSQL = `
INSERT INTO data_node.sync_tasks (task, block)
VALUES ($1, $2)
ON CONFLICT (key) DO NOTHING;
ON CONFLICT (task) DO UPDATE
SET block = EXCLUDED.block, processed = NOW();
`

for _, d := range od {
if _, err := dbTx.ExecContext(
ctx, storeOffChainDataSQL,
d.Key.Hex(),
common.Bytes2Hex(d.Value),
); err != nil {
return err
}
if _, err := db.execer(dbTx).ExecContext(ctx, storeLastProcessedBlockSQL, task, block); err != nil {
return err
}

return nil
}

// GetOffChainData returns the value identified by the key
func (db *pgDB) GetOffChainData(ctx context.Context, key common.Hash, dbTx sqlx.QueryerContext) (types.ArgBytes, error) {
const getOffchainDataSQL = `
SELECT value
FROM data_node.offchain_data
WHERE key = $1 LIMIT 1;
`
// GetLastProcessedBlock returns the latest block successfully processed by the synchronizer for named task
func (db *pgDB) GetLastProcessedBlock(ctx context.Context, task string) (uint64, error) {
const getLastProcessedBlockSQL = "SELECT block FROM data_node.sync_tasks WHERE task = $1;"

var (
hexValue string
lastBlock uint64
)

if err := dbTx.QueryRowxContext(ctx, getOffchainDataSQL, key.Hex()).Scan(&hexValue); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrStateNotSynchronized
if err := db.pg.QueryRowContext(ctx, getLastProcessedBlockSQL, task).Scan(&lastBlock); err != nil {
return 0, err
}

return lastBlock, nil
}

// StoreUnresolvedBatchKeys stores unresolved batch keys in the database
func (db *pgDB) StoreUnresolvedBatchKeys(ctx context.Context, bks []types.BatchKey, dbTx sqlx.ExecerContext) error {
const storeUnresolvedBatchesSQL = `
INSERT INTO data_node.unresolved_batches (num, hash)
VALUES ($1, $2)
ON CONFLICT (num, hash) DO NOTHING;
`

execer := db.execer(dbTx)
for _, bk := range bks {
if _, err := execer.ExecContext(
ctx, storeUnresolvedBatchesSQL,
bk.Number,
bk.Hash.Hex(),
); err != nil {
return err
}
}

return nil
}

// GetUnresolvedBatchKeys returns the unresolved batch keys from the database
func (db *pgDB) GetUnresolvedBatchKeys(ctx context.Context) ([]types.BatchKey, error) {
const getUnresolvedBatchKeysSQL = "SELECT num, hash FROM data_node.unresolved_batches;"

rows, err := db.pg.QueryxContext(ctx, getUnresolvedBatchKeysSQL)
if err != nil {
return nil, err
}

return common.FromHex(hexValue), nil
defer rows.Close()

var bks []types.BatchKey
for rows.Next() {
bk := struct {
Number uint64 `db:"num"`
Hash string `db:"hash"`
}{}
if err = rows.StructScan(&bk); err != nil {
return nil, err
}

bks = append(bks, types.BatchKey{
Number: bk.Number,
Hash: common.HexToHash(bk.Hash),
})
}

return bks, nil
}

// DeleteUnresolvedBatchKeys deletes the unresolved batch keys from the database
func (db *pgDB) DeleteUnresolvedBatchKeys(ctx context.Context, bks []types.BatchKey, dbTx sqlx.ExecerContext) error {
const deleteUnresolvedBatchKeysSQL = `
DELETE FROM data_node.unresolved_batches
WHERE num = $1 AND hash = $2;
`

for _, bk := range bks {
if _, err := db.execer(dbTx).ExecContext(
ctx, deleteUnresolvedBatchKeysSQL,
bk.Number,
bk.Hash.Hex(),
); err != nil {
return err
}
}

return nil
}

// Exists checks if a key exists in offchain data table
Expand All @@ -108,33 +174,62 @@ func (db *pgDB) Exists(ctx context.Context, key common.Hash) bool {
return count > 0
}

// StoreLastProcessedBlock stores a record of a block processed by the synchronizer for named task
func (db *pgDB) StoreLastProcessedBlock(ctx context.Context, task string, block uint64, dbTx sqlx.ExecerContext) error {
const storeLastProcessedBlockSQL = `
INSERT INTO data_node.sync_tasks (task, block)
// StoreOffChainData stores and array of key values in the Db
func (db *pgDB) StoreOffChainData(ctx context.Context, od []types.OffChainData, dbTx sqlx.ExecerContext) error {
const storeOffChainDataSQL = `
INSERT INTO data_node.offchain_data (key, value)
VALUES ($1, $2)
ON CONFLICT (task) DO UPDATE
SET block = EXCLUDED.block, processed = NOW();
ON CONFLICT (key) DO NOTHING;
`

if _, err := dbTx.ExecContext(ctx, storeLastProcessedBlockSQL, task, block); err != nil {
return err
execer := db.execer(dbTx)
for _, d := range od {
if _, err := execer.ExecContext(
ctx, storeOffChainDataSQL,
d.Key.Hex(),
common.Bytes2Hex(d.Value),
); err != nil {
return err
}
}

return nil
}

// GetLastProcessedBlock returns the latest block successfully processed by the synchronizer for named task
func (db *pgDB) GetLastProcessedBlock(ctx context.Context, task string) (uint64, error) {
const getLastProcessedBlockSQL = "SELECT block FROM data_node.sync_tasks WHERE task = $1;"
// GetOffChainData returns the value identified by the key
func (db *pgDB) GetOffChainData(ctx context.Context, key common.Hash, dbTx sqlx.QueryerContext) (types.ArgBytes, error) {
const getOffchainDataSQL = `
SELECT value
FROM data_node.offchain_data
WHERE key = $1 LIMIT 1;
`

var (
lastBlock uint64
hexValue string
)

if err := db.pg.QueryRowContext(ctx, getLastProcessedBlockSQL, task).Scan(&lastBlock); err != nil {
return 0, err
if err := db.querier(dbTx).QueryRowxContext(ctx, getOffchainDataSQL, key.Hex()).Scan(&hexValue); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrStateNotSynchronized
}
return nil, err
}

return lastBlock, nil
return common.FromHex(hexValue), nil
}

func (db *pgDB) execer(dbTx sqlx.ExecerContext) sqlx.ExecerContext {
if dbTx != nil {
return dbTx
}

return db.pg
}

func (db *pgDB) querier(dbTx sqlx.QueryerContext) sqlx.QueryerContext {
if dbTx != nil {
return dbTx
}

return db.pg
}
Loading

0 comments on commit f7ff8e3

Please sign in to comment.