From b4005eff3939380b4d907835e0a89184b868cc64 Mon Sep 17 00:00:00 2001 From: Ian Norden Date: Thu, 12 Mar 2020 13:51:58 -0500 Subject: [PATCH] db cleaners to eth and btc --- pkg/super_node/backfiller.go | 7 +- pkg/super_node/btc/cleaner.go | 132 ++++++++++++++++ pkg/super_node/constructors.go | 11 ++ pkg/super_node/eth/cleaner.go | 226 ++++++++++++++++++++++++++++ pkg/super_node/shared/chain_type.go | 4 +- pkg/super_node/shared/data_type.go | 94 ++++++++++++ pkg/super_node/shared/intefaces.go | 5 + pkg/watcher/eth/repository.go | 12 +- 8 files changed, 478 insertions(+), 13 deletions(-) create mode 100644 pkg/super_node/btc/cleaner.go create mode 100644 pkg/super_node/eth/cleaner.go create mode 100644 pkg/super_node/shared/data_type.go diff --git a/pkg/super_node/backfiller.go b/pkg/super_node/backfiller.go index de8a78d1..ace79700 100644 --- a/pkg/super_node/backfiller.go +++ b/pkg/super_node/backfiller.go @@ -30,7 +30,7 @@ import ( const ( DefaultMaxBatchSize uint64 = 100 - defaultMaxBatchNumber int64 = 10 + DefaultMaxBatchNumber int64 = 10 ) // BackFillInterface for filling in gaps in the super node @@ -146,8 +146,7 @@ func (bfs *BackFillService) fillGaps(startingBlock, endingBlock uint64) error { log.Infof("going to fill in %s gap from %d to %d", bfs.chain.String(), startingBlock, endingBlock) errChan := make(chan error) done := make(chan bool) - err := bfs.backFill(startingBlock, endingBlock, errChan, done) - if err != nil { + if err := bfs.backFill(startingBlock, endingBlock, errChan, done); err != nil { return err } for { @@ -184,7 +183,7 @@ func (bfs *BackFillService) backFill(startingBlock, endingBlock uint64, errChan for _, blockHeights := range blockRangeBins { // if we have reached our limit of active goroutines // wait for one to finish before starting the next - if atomic.AddInt64(&activeCount, 1) > defaultMaxBatchNumber { + if atomic.AddInt64(&activeCount, 1) > DefaultMaxBatchNumber { // this blocks until a process signals it has finished <-forwardDone } diff --git a/pkg/super_node/btc/cleaner.go b/pkg/super_node/btc/cleaner.go new file mode 100644 index 00000000..a2ed3408 --- /dev/null +++ b/pkg/super_node/btc/cleaner.go @@ -0,0 +1,132 @@ +// VulcanizeDB +// Copyright © 2019 Vulcanize + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. + +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package btc + +import ( + "fmt" + + "github.com/jmoiron/sqlx" + "github.com/sirupsen/logrus" + + "github.com/vulcanize/vulcanizedb/pkg/postgres" + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" +) + +// Cleaner satisfies the shared.Cleaner interface fo bitcoin +type Cleaner struct { + db *postgres.DB +} + +// NewCleaner returns a new Cleaner struct that satisfies the shared.Cleaner interface +func NewCleaner(db *postgres.DB) *Cleaner { + return &Cleaner{ + db: db, + } +} + +// Clean removes the specified data from the db within the provided block range +func (c *Cleaner) Clean(rngs [][2]uint64, t shared.DataType) error { + tx, err := c.db.Beginx() + if err != nil { + return err + } + for _, rng := range rngs { + if err := c.clean(tx, rng, t); err != nil { + if err := tx.Rollback(); err != nil { + logrus.Error(err) + } + return err + } + } + return tx.Commit() +} + +func (c *Cleaner) clean(tx *sqlx.Tx, rng [2]uint64, t shared.DataType) error { + switch t { + case shared.Full: + return c.cleanFull(tx, rng) + case shared.Headers: + if err := c.cleanTransactionIPLDs(tx, rng); err != nil { + return err + } + if err := c.cleanHeaderIPLDs(tx, rng); err != nil { + return err + } + return c.cleanHeaderMetaData(tx, rng) + case shared.Transactions: + if err := c.cleanTransactionIPLDs(tx, rng); err != nil { + return err + } + return c.cleanTransactionMetaData(tx, rng) + default: + return fmt.Errorf("btc cleaner unrecognized type: %s", t.String()) + } + return nil +} + +func (c *Cleaner) cleanFull(tx *sqlx.Tx, rng [2]uint64) error { + // Clear all of the indexed iplds + pgStr := `DELETE FROM public.blocks A + USING btc.transaction_cids B, btc.header_cids C + WHERE (A.key = B.cid OR A.key = C.cid) + AND B.header_id = C.id + AND C.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + if err != nil { + return err + } + // Clear all the header_cids, this will cascade delete the rest of the index metadata + pgStr = `DELETE FROM eth.header_cids + WHERE block_number BETWEEN $1 AND $2` + _, err = tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanTransactionIPLDs(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM public.blocks A + USING btc.transaction_cids B, btc.header_cids C + WHERE A.key = B.cid + AND B.header_id = C.id + AND C.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanTransactionMetaData(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM btc.transaction_cids A + USING btc.header_cids B + WHERE A.header_id = B.id + AND B.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanHeaderIPLDs(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM public.blocks A + USING btc.header_cids B + WHERE A.key = B.cid + AND B.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanHeaderMetaData(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM btc.header_cids + WHERE block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} diff --git a/pkg/super_node/constructors.go b/pkg/super_node/constructors.go index 5a2888f9..8f0ccf6e 100644 --- a/pkg/super_node/constructors.go +++ b/pkg/super_node/constructors.go @@ -166,3 +166,14 @@ func NewPublicAPI(chain shared.ChainType, db *postgres.DB, ipfsPath string) (rpc return rpc.API{}, fmt.Errorf("invalid chain %s for public api constructor", chain.String()) } } + +// NewCleaner constructs a Cleaner for the provided chain type +func NewCleaner(chain shared.ChainType, db *postgres.DB) (shared.Cleaner, error) { + switch chain { + case shared.Ethereum: + return eth.NewCleaner(db), nil + // TODO: support BTC + default: + return nil, fmt.Errorf("invalid chain %s for publisher constructor", chain.String()) + } +} diff --git a/pkg/super_node/eth/cleaner.go b/pkg/super_node/eth/cleaner.go new file mode 100644 index 00000000..d66cbf48 --- /dev/null +++ b/pkg/super_node/eth/cleaner.go @@ -0,0 +1,226 @@ +// VulcanizeDB +// Copyright © 2019 Vulcanize + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. + +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package eth + +import ( + "fmt" + + "github.com/jmoiron/sqlx" + "github.com/sirupsen/logrus" + + "github.com/vulcanize/vulcanizedb/pkg/postgres" + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" +) + +// Cleaner satisfies the shared.Cleaner interface fo ethereum +type Cleaner struct { + db *postgres.DB +} + +// NewCleaner returns a new Cleaner struct that satisfies the shared.Cleaner interface +func NewCleaner(db *postgres.DB) *Cleaner { + return &Cleaner{ + db: db, + } +} + +// Clean removes the specified data from the db within the provided block range +func (c *Cleaner) Clean(rngs [][2]uint64, t shared.DataType) error { + tx, err := c.db.Beginx() + if err != nil { + return err + } + for _, rng := range rngs { + if err := c.clean(tx, rng, t); err != nil { + if err := tx.Rollback(); err != nil { + logrus.Error(err) + } + return err + } + } + return tx.Commit() +} + +func (c *Cleaner) clean(tx *sqlx.Tx, rng [2]uint64, t shared.DataType) error { + switch t { + case shared.Full: + return c.cleanFull(tx, rng) + case shared.Headers: + if err := c.cleanStorageIPLDs(tx, rng); err != nil { + return err + } + if err := c.cleanStateIPLDs(tx, rng); err != nil { + return err + } + if err := c.cleanReceiptIPLDs(tx, rng); err != nil { + return err + } + if err := c.cleanTransactionIPLDs(tx, rng); err != nil { + return err + } + if err := c.cleanHeaderIPLDs(tx, rng); err != nil { + return err + } + return c.cleanHeaderMetaData(tx, rng) + case shared.Transactions: + if err := c.cleanReceiptIPLDs(tx, rng); err != nil { + return err + } + if err := c.cleanTransactionIPLDs(tx, rng); err != nil { + return err + } + return c.cleanTransactionMetaData(tx, rng) + case shared.Receipts: + if err := c.cleanReceiptIPLDs(tx, rng); err != nil { + return err + } + return c.cleanReceiptMetaData(tx, rng) + case shared.State: + if err := c.cleanStorageIPLDs(tx, rng); err != nil { + return err + } + if err := c.cleanStateIPLDs(tx, rng); err != nil { + return err + } + return c.cleanStateMetaData(tx, rng) + case shared.Storage: + if err := c.cleanStorageIPLDs(tx, rng); err != nil { + return err + } + return c.cleanStorageMetaData(tx, rng) + default: + return fmt.Errorf("eth cleaner unrecognized type: %s", t.String()) + } + return nil +} + +func (c *Cleaner) cleanFull(tx *sqlx.Tx, rng [2]uint64) error { + // Clear all of the indexed iplds + pgStr := `DELETE FROM public.blocks A + USING eth.storage_cids B, eth.state_cids C, eth.receipt_cids D, eth.transaction_cids E, eth.header_cids F + WHERE (A.key = B.cid OR A.key = C.cid OR A.key = D.cid OR A.key = E.cid OR A.key = F.cid) + AND B.state_id = C.id + AND C.header_id = F.id + AND D.tx_id = E.id + AND E.header_id = F.id + AND F.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + if err != nil { + return err + } + // Clear all the header_cids, this will cascade delete the rest of the index metadata + pgStr = `DELETE FROM eth.header_cids + WHERE block_number BETWEEN $1 AND $2` + _, err = tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanStorageIPLDs(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM public.blocks A + USING eth.storage_cids B, eth.state_cids C, eth.header_cids D + WHERE A.key = B.cid + AND B.state_id = C.id + AND C.header_id = D.id + AND D.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanStorageMetaData(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM eth.storage_cids A + USING eth.state_cids B, eth.header_cids C + WHERE A.state_id = B.id + AND B.header_id = C.id + AND C.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanStateIPLDs(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM public.blocks A + USING eth.state_cids B, eth.header_cids C + WHERE A.key = B.cid + AND B.header_cid = C.id + AND C.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanStateMetaData(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM eth.state_cids A + USING eth.header_cids B + WHERE A.header_id = B.id + AND B.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanReceiptIPLDs(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM public.blocks A + USING eth.receipt_cids B, eth.transaction_cids C, eth.header_cids D + WHERE A.key = B.cid + AND B.tx_id = C.id + AND C.header_id = D.id + AND D.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanReceiptMetaData(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM eth.receipt_cids A + USING eth.transaction_cids B, eth.header_cids C + WHERE A.tx_id = B.id + AND B.header_id = C.id + AND C.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanTransactionIPLDs(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM public.blocks A + USING eth.transaction_cids B, eth.header_cids C + WHERE A.key = B.cid + AND B.header_id = C.id + AND C.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanTransactionMetaData(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM eth.transaction_cids A + USING eth.header_cids B + WHERE A.header_id = B.id + AND B.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanHeaderIPLDs(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM public.blocks A + USING eth.header_cids B + WHERE A.key = B.cid + AND B.block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} + +func (c *Cleaner) cleanHeaderMetaData(tx *sqlx.Tx, rng [2]uint64) error { + pgStr := `DELETE FROM eth.header_cids + WHERE block_number BETWEEN $1 AND $2` + _, err := tx.Exec(pgStr, rng[0], rng[1]) + return err +} diff --git a/pkg/super_node/shared/chain_type.go b/pkg/super_node/shared/chain_type.go index 90faf946..83445192 100644 --- a/pkg/super_node/shared/chain_type.go +++ b/pkg/super_node/shared/chain_type.go @@ -25,7 +25,7 @@ import ( type ChainType int const ( - Unknown ChainType = iota + UnknownChain ChainType = iota Ethereum Bitcoin Omni @@ -66,6 +66,6 @@ func NewChainType(name string) (ChainType, error) { case "omni": return Omni, nil default: - return Unknown, errors.New("invalid name for chain") + return UnknownChain, errors.New("invalid name for chain") } } diff --git a/pkg/super_node/shared/data_type.go b/pkg/super_node/shared/data_type.go new file mode 100644 index 00000000..0c219c6a --- /dev/null +++ b/pkg/super_node/shared/data_type.go @@ -0,0 +1,94 @@ +// VulcanizeDB +// Copyright © 2019 Vulcanize + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. + +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package shared + +import ( + "fmt" + "strings" +) + +// DataType is an enum to loosely represent type of chain data +type DataType int + +const ( + UnknownDataType DataType = iota - 1 + Full + Headers + Transactions + Receipts + State + Storage +) + +// String() method to resolve ReSyncType enum +func (r DataType) String() string { + switch r { + case Full: + return "full" + case Headers: + return "headers" + case Transactions: + return "transactions" + case Receipts: + return "receipts" + case State: + return "state" + case Storage: + return "storage" + default: + return "unknown" + } +} + +// GenerateResyncTypeFromString +func GenerateResyncTypeFromString(str string) (DataType, error) { + switch strings.ToLower(str) { + case "full", "f": + return Full, nil + case "headers", "header", "h": + return Headers, nil + case "transactions", "transaction", "trxs", "txs", "trx", "tx", "t": + return Transactions, nil + case "receipts", "receipt", "rcts", "rct", "r": + return Receipts, nil + case "state": + return State, nil + case "storage": + return Storage, nil + default: + return UnknownDataType, fmt.Errorf("unrecognized resync type: %s", str) + } +} + +func SupportedResyncType(d DataType) bool { + switch d { + case Full: + return true + case Headers: + return false + case Transactions: + return false + case Receipts: + return false + case State: + return false + case Storage: + return false + default: + return false + } +} diff --git a/pkg/super_node/shared/intefaces.go b/pkg/super_node/shared/intefaces.go index f62bba76..72940422 100644 --- a/pkg/super_node/shared/intefaces.go +++ b/pkg/super_node/shared/intefaces.go @@ -74,6 +74,11 @@ type DagPutter interface { DagPut(raw interface{}) ([]string, error) } +// Cleaner is for cleaning out data from the cache within the given ranges +type Cleaner interface { + Clean(rngs [][2]uint64, t DataType) error +} + // SubscriptionSettings is the interface every subscription filter type needs to satisfy, no matter the chain // Further specifics of the underlying filter type depend on the internal needs of the types // which satisfy the ResponseFilterer and CIDRetriever interfaces for a specific chain diff --git a/pkg/watcher/eth/repository.go b/pkg/watcher/eth/repository.go index cde4255d..1791697e 100644 --- a/pkg/watcher/eth/repository.go +++ b/pkg/watcher/eth/repository.go @@ -82,13 +82,11 @@ func (r *Repository) QueueData(payload super_node.SubscriptionPayload) error { return err } -// GetQueueData grabs payload data from the queue table so that it can -// be forwarded to the ready tables -// this is used to make sure we enter data into the tables that triggers act on in sequential order -// even if we receive data out-of-order -// it returns the new index -// delete the data it retrieves so as to clear the queue -// periodically vacuum's the table to free up space from the deleted rows +// GetQueueData grabs payload data from the queue table so that it can be readied +// Used ensure we enter data into the tables that triggers act on in sequential order, even if we receive data out-of-order +// Returns the queued data, the new index, and err +// Deletes from the queue the data it retrieves +// Periodically vacuum's the table to free up space from the deleted rows func (r *Repository) GetQueueData(height int64) (super_node.SubscriptionPayload, int64, error) { pgStr := `DELETE FROM eth.queued_data WHERE height = $1