lotus/chain/index/msgindex.go

554 lines
13 KiB
Go
Raw Normal View History

2023-03-11 14:49:07 +00:00
package index
import (
"context"
"database/sql"
"errors"
"io/fs"
"os"
"path"
2023-03-11 17:11:08 +00:00
"sync"
2023-03-11 14:49:07 +00:00
"time"
2023-03-12 13:30:05 +00:00
"github.com/ipfs/go-cid"
2023-03-11 14:49:07 +00:00
logging "github.com/ipfs/go-log/v2"
_ "github.com/mattn/go-sqlite3"
"golang.org/x/xerrors"
2023-03-11 16:21:16 +00:00
"github.com/filecoin-project/go-state-types/abi"
2023-03-12 13:30:05 +00:00
2023-03-11 14:49:07 +00:00
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
)
2023-03-11 17:11:08 +00:00
var log = logging.Logger("msgindex")
var dbName = "msgindex.db"
var dbDefs = []string{
`CREATE TABLE IF NOT EXISTS messages (
cid VARCHAR(80) PRIMARY KEY ON CONFLICT REPLACE,
tipset_cid VARCHAR(80) NOT NULL,
epoch INTEGER NOT NULL
)`,
`CREATE INDEX IF NOT EXISTS tipset_cids ON messages (tipset_cid)
`,
`CREATE TABLE IF NOT EXISTS _meta (
version UINT64 NOT NULL UNIQUE
)`,
`INSERT OR IGNORE INTO _meta (version) VALUES (1)`,
}
var dbPragmas = []string{}
const (
// prepared stmts
dbqGetMessageInfo = "SELECT tipset_cid, epoch FROM messages WHERE cid = ?"
dbqInsertMessage = "INSERT INTO messages VALUES (?, ?, ?)"
dbqDeleteTipsetMessages = "DELETE FROM messages WHERE tipset_cid = ?"
// reconciliation
dbqCountMessages = "SELECT COUNT(*) FROM messages"
dbqMinEpoch = "SELECT MIN(epoch) FROM messages"
dbqCountTipsetMessages = "SELECT COUNT(*) FROM messages WHERE tipset_cid = ?"
dbqDeleteMessagesByEpoch = "DELETE FROM messages WHERE epoch >= ?"
)
2023-03-11 17:11:08 +00:00
// coalescer configuration (TODO: use observer instead)
2023-03-13 12:55:26 +00:00
// these are exposed to make tests snappy
var (
2023-03-13 12:55:26 +00:00
CoalesceMinDelay = time.Second
CoalesceMaxDelay = 15 * time.Second
CoalesceMergeInterval = time.Second
2023-03-11 17:11:08 +00:00
)
2023-03-11 15:55:48 +00:00
// chain store interface; we could use store.ChainStore directly,
// but this simplifies unit testing.
type ChainStore interface {
SubscribeHeadChanges(f store.ReorgNotifee)
MessagesForTipset(ctx context.Context, ts *types.TipSet) ([]types.ChainMsg, error)
2023-03-11 19:26:11 +00:00
GetHeaviestTipSet() *types.TipSet
GetTipSetFromKey(ctx context.Context, tsk types.TipSetKey) (*types.TipSet, error)
2023-03-11 15:55:48 +00:00
}
var _ ChainStore = (*store.ChainStore)(nil)
2023-03-11 14:49:07 +00:00
type msgIndex struct {
2023-03-11 15:55:48 +00:00
cs ChainStore
2023-03-11 14:49:07 +00:00
2023-03-11 16:21:16 +00:00
db *sql.DB
selectMsgStmt *sql.Stmt
2023-03-11 17:11:08 +00:00
insertMsgStmt *sql.Stmt
2023-03-11 16:21:16 +00:00
deleteTipSetStmt *sql.Stmt
2023-03-11 14:49:07 +00:00
2023-03-11 17:11:08 +00:00
sema chan struct{}
mx sync.Mutex
pend []headChange
2023-03-11 14:49:07 +00:00
2023-03-11 20:09:31 +00:00
cancel func()
workers sync.WaitGroup
closeLk sync.RWMutex
closed bool
2023-03-11 17:11:08 +00:00
}
2023-03-11 14:49:07 +00:00
2023-03-11 17:11:08 +00:00
var _ MsgIndex = (*msgIndex)(nil)
2023-03-11 14:49:07 +00:00
2023-03-11 17:11:08 +00:00
type headChange struct {
rev []*types.TipSet
app []*types.TipSet
}
2023-03-11 14:49:07 +00:00
2023-03-13 08:42:48 +00:00
func NewMsgIndex(lctx context.Context, basePath string, cs ChainStore) (MsgIndex, error) {
2023-03-11 14:49:07 +00:00
var (
dbPath string
exists bool
2023-03-11 14:49:07 +00:00
err error
)
err = os.MkdirAll(basePath, 0755)
if err != nil {
return nil, xerrors.Errorf("error creating msgindex base directory: %w", err)
}
dbPath = path.Join(basePath, dbName)
_, err = os.Stat(dbPath)
switch {
case err == nil:
exists = true
2023-03-11 14:49:07 +00:00
case errors.Is(err, fs.ErrNotExist):
case err != nil:
return nil, xerrors.Errorf("error stating msgindex database: %w", err)
}
db, err := sql.Open("sqlite3", dbPath)
if err != nil {
// TODO [nice to have]: automaticaly delete corrupt databases
// but for now we can just error and let the operator delete.
return nil, xerrors.Errorf("error opening msgindex database: %w", err)
}
if err := prepareDB(db); err != nil {
return nil, xerrors.Errorf("error creating msgindex database: %w", err)
}
2023-03-11 19:26:11 +00:00
// TODO we may consider populating the index when first creating the db
if exists {
if err := reconcileIndex(db, cs); err != nil {
2023-03-11 14:49:07 +00:00
return nil, xerrors.Errorf("error reconciling msgindex database: %w", err)
}
}
2023-03-13 08:42:48 +00:00
ctx, cancel := context.WithCancel(lctx)
2023-03-11 17:11:08 +00:00
msgIndex := &msgIndex{
db: db,
cs: cs,
sema: make(chan struct{}, 1),
cancel: cancel,
}
2023-03-11 14:49:07 +00:00
err = msgIndex.prepareStatements()
if err != nil {
2023-03-13 03:54:32 +00:00
if err := db.Close(); err != nil {
log.Errorf("error closing msgindex database: %s", err)
2023-03-11 14:49:07 +00:00
}
return nil, xerrors.Errorf("error preparing msgindex database statements: %w", err)
}
rnf := store.WrapHeadChangeCoalescer(
msgIndex.onHeadChange,
2023-03-13 12:55:26 +00:00
CoalesceMinDelay,
CoalesceMaxDelay,
CoalesceMergeInterval,
2023-03-11 14:49:07 +00:00
)
cs.SubscribeHeadChanges(rnf)
2023-03-11 20:09:31 +00:00
msgIndex.workers.Add(1)
2023-03-11 17:11:08 +00:00
go msgIndex.background(ctx)
2023-03-11 14:49:07 +00:00
return msgIndex, nil
}
func PopulateAfterSnapshot(lctx context.Context, basePath string, cs ChainStore) error {
err := os.MkdirAll(basePath, 0755)
if err != nil {
return xerrors.Errorf("error creating msgindex base directory: %w", err)
}
dbPath := path.Join(basePath, dbName)
// if a database already exists, we try to delete it and create a new one
if _, err := os.Stat(dbPath); err == nil {
if err = os.Remove(dbPath); err != nil {
return xerrors.Errorf("msgindex already exists at %s and can't be deleted", dbPath)
}
}
db, err := sql.Open("sqlite3", dbPath)
if err != nil {
return xerrors.Errorf("error opening msgindex database: %w", err)
}
2023-03-25 11:17:54 +00:00
defer func() {
if err := db.Close(); err != nil {
log.Errorf("error closing msgindex database: %s", err)
}
}()
if err := prepareDB(db); err != nil {
return xerrors.Errorf("error creating msgindex database: %w", err)
}
tx, err := db.Begin()
if err != nil {
return xerrors.Errorf("error when starting transaction: %w", err)
}
rollback := func() {
if err := tx.Rollback(); err != nil {
log.Errorf("error in rollback: %s", err)
}
}
insertStmt, err := tx.Prepare(dbqInsertMessage)
if err != nil {
2023-03-28 15:08:53 +00:00
rollback()
2023-03-25 11:17:54 +00:00
return xerrors.Errorf("error preparing insertStmt: %w", err)
}
curTs := cs.GetHeaviestTipSet()
startHeight := curTs.Height()
for curTs != nil {
tscid, err := curTs.Key().Cid()
if err != nil {
rollback()
return xerrors.Errorf("error computing tipset cid: %w", err)
}
tskey := tscid.String()
epoch := int64(curTs.Height())
msgs, err := cs.MessagesForTipset(lctx, curTs)
if err != nil {
log.Infof("stopping import after %d tipsets", startHeight-curTs.Height())
break
}
for _, msg := range msgs {
key := msg.Cid().String()
if _, err := insertStmt.Exec(key, tskey, epoch); err != nil {
rollback()
return xerrors.Errorf("error inserting message: %w", err)
}
}
curTs, err = cs.GetTipSetFromKey(lctx, curTs.Parents())
if err != nil {
rollback()
return xerrors.Errorf("error walking chain: %w", err)
}
}
err = tx.Commit()
if err != nil {
2023-03-25 11:17:54 +00:00
return xerrors.Errorf("error committing transaction: %w", err)
}
return nil
}
2023-03-11 14:49:07 +00:00
// init utilities
func prepareDB(db *sql.DB) error {
for _, stmt := range dbDefs {
if _, err := db.Exec(stmt); err != nil {
return xerrors.Errorf("error executing sql statement '%s': %w", stmt, err)
}
}
for _, stmt := range dbPragmas {
if _, err := db.Exec(stmt); err != nil {
return xerrors.Errorf("error executing sql statement '%s': %w", stmt, err)
}
2023-03-11 16:21:16 +00:00
}
return nil
2023-03-11 14:49:07 +00:00
}
2023-03-11 15:55:48 +00:00
func reconcileIndex(db *sql.DB, cs ChainStore) error {
2023-03-11 19:26:11 +00:00
// Invariant: after reconciliation, every tipset in the index is in the current chain; ie either
// the chain head or reachable by walking the chain.
// Algorithm:
2023-03-13 03:57:26 +00:00
// 1. Count messages in index; if none, trivially reconciled.
2023-03-11 19:26:11 +00:00
// TODO we may consider populating the index in that case
// 2. Find the minimum tipset in the index; this will mark the end of the reconciliation walk
// 3. Walk from current tipset until we find a tipset in the index.
// 4. Delete (revert!) all tipsets above the found tipset.
// 5. If the walk ends in the boundary epoch, then delete everything.
//
row := db.QueryRow(dbqCountMessages)
2023-03-11 19:26:11 +00:00
var result int64
if err := row.Scan(&result); err != nil {
return xerrors.Errorf("error counting messages: %w", err)
}
if result == 0 {
return nil
}
row = db.QueryRow(dbqMinEpoch)
2023-03-11 19:26:11 +00:00
if err := row.Scan(&result); err != nil {
return xerrors.Errorf("error finding boundary epoch: %w", err)
}
boundaryEpoch := abi.ChainEpoch(result)
countMsgsStmt, err := db.Prepare(dbqCountTipsetMessages)
2023-03-11 19:26:11 +00:00
if err != nil {
return xerrors.Errorf("error preparing statement: %w", err)
}
curTs := cs.GetHeaviestTipSet()
for curTs != nil && curTs.Height() >= boundaryEpoch {
tsCid, err := curTs.Key().Cid()
if err != nil {
return xerrors.Errorf("error computing tipset cid: %w", err)
}
key := tsCid.String()
row = countMsgsStmt.QueryRow(key)
if err := row.Scan(&result); err != nil {
return xerrors.Errorf("error counting messages: %w", err)
}
if result > 0 {
// found it!
boundaryEpoch = curTs.Height() + 1
break
}
// walk up
parents := curTs.Parents()
curTs, err = cs.GetTipSetFromKey(context.TODO(), parents)
if err != nil {
return xerrors.Errorf("error walking chain: %w", err)
}
}
// delete everything above the minEpoch
if _, err = db.Exec(dbqDeleteMessagesByEpoch, int64(boundaryEpoch)); err != nil {
2023-03-11 19:26:11 +00:00
return xerrors.Errorf("error deleting stale reorged out message: %w", err)
}
return nil
2023-03-11 14:49:07 +00:00
}
func (x *msgIndex) prepareStatements() error {
stmt, err := x.db.Prepare(dbqGetMessageInfo)
2023-03-11 16:21:16 +00:00
if err != nil {
2023-03-12 11:21:03 +00:00
return xerrors.Errorf("prepare selectMsgStmt: %w", err)
2023-03-11 16:21:16 +00:00
}
x.selectMsgStmt = stmt
stmt, err = x.db.Prepare(dbqInsertMessage)
2023-03-11 17:11:08 +00:00
if err != nil {
2023-03-12 11:21:03 +00:00
return xerrors.Errorf("prepare insertMsgStmt: %w", err)
2023-03-11 17:11:08 +00:00
}
x.insertMsgStmt = stmt
stmt, err = x.db.Prepare(dbqDeleteTipsetMessages)
2023-03-11 16:21:16 +00:00
if err != nil {
2023-03-12 11:21:03 +00:00
return xerrors.Errorf("prepare deleteTipSetStmt: %w", err)
2023-03-11 16:21:16 +00:00
}
x.deleteTipSetStmt = stmt
return nil
2023-03-11 14:49:07 +00:00
}
// head change notifee
func (x *msgIndex) onHeadChange(rev, app []*types.TipSet) error {
2023-03-11 20:09:31 +00:00
x.closeLk.RLock()
defer x.closeLk.RUnlock()
if x.closed {
return nil
}
2023-03-11 17:11:08 +00:00
// do it in the background to avoid blocking head change processing
x.mx.Lock()
x.pend = append(x.pend, headChange{rev: rev, app: app})
pendLen := len(x.pend)
2023-03-11 17:11:08 +00:00
x.mx.Unlock()
// complain loudly if this is building backlog
if pendLen > 10 {
log.Warnf("message index head change processing is building backlog: %d pending head changes", pendLen)
}
2023-03-11 17:11:08 +00:00
select {
case x.sema <- struct{}{}:
default:
}
return nil
}
func (x *msgIndex) background(ctx context.Context) {
2023-03-11 20:09:31 +00:00
defer x.workers.Done()
2023-03-11 17:11:08 +00:00
for {
select {
case <-x.sema:
err := x.processHeadChanges(ctx)
if err != nil {
// we can't rely on an inconsistent index, so shut it down.
log.Errorf("error processing head change notifications: %s; shutting down message index", err)
if err2 := x.Close(); err2 != nil {
log.Errorf("error shutting down index: %s", err2)
}
2023-03-11 17:11:08 +00:00
}
case <-ctx.Done():
return
}
}
}
func (x *msgIndex) processHeadChanges(ctx context.Context) error {
x.mx.Lock()
pend := x.pend
x.pend = nil
x.mx.Unlock()
2023-03-12 11:35:50 +00:00
tx, err := x.db.Begin()
2023-03-11 17:11:08 +00:00
if err != nil {
return xerrors.Errorf("error creating transaction: %w", err)
}
for _, hc := range pend {
for _, ts := range hc.rev {
2023-03-12 11:35:50 +00:00
if err := x.doRevert(ctx, tx, ts); err != nil {
if err2 := tx.Rollback(); err2 != nil {
log.Errorf("error rolling back transaction: %s", err2)
2023-03-13 09:41:46 +00:00
}
2023-03-11 17:11:08 +00:00
return xerrors.Errorf("error reverting %s: %w", ts, err)
}
}
for _, ts := range hc.app {
2023-03-12 11:35:50 +00:00
if err := x.doApply(ctx, tx, ts); err != nil {
if err2 := tx.Rollback(); err2 != nil {
log.Errorf("error rolling back transaction: %s", err2)
2023-03-13 09:41:46 +00:00
}
2023-03-11 17:11:08 +00:00
return xerrors.Errorf("error applying %s: %w", ts, err)
}
}
}
2023-03-12 11:35:50 +00:00
return tx.Commit()
2023-03-11 17:11:08 +00:00
}
2023-03-12 11:35:50 +00:00
func (x *msgIndex) doRevert(ctx context.Context, tx *sql.Tx, ts *types.TipSet) error {
2023-03-11 17:11:08 +00:00
tskey, err := ts.Key().Cid()
if err != nil {
return xerrors.Errorf("error computing tipset cid: %w", err)
}
key := tskey.String()
2023-03-12 11:35:50 +00:00
_, err = tx.Stmt(x.deleteTipSetStmt).Exec(key)
2023-03-11 17:11:08 +00:00
return err
}
2023-03-12 11:35:50 +00:00
func (x *msgIndex) doApply(ctx context.Context, tx *sql.Tx, ts *types.TipSet) error {
2023-03-11 17:11:08 +00:00
tscid, err := ts.Key().Cid()
if err != nil {
return xerrors.Errorf("error computing tipset cid: %w", err)
}
tskey := tscid.String()
epoch := int64(ts.Height())
2023-03-11 17:11:08 +00:00
msgs, err := x.cs.MessagesForTipset(ctx, ts)
if err != nil {
return xerrors.Errorf("error retrieving messages for tipset %s: %w", ts, err)
2023-03-11 17:11:08 +00:00
}
insertStmt := tx.Stmt(x.insertMsgStmt)
for _, msg := range msgs {
key := msg.Cid().String()
if _, err := insertStmt.Exec(key, tskey, epoch); err != nil {
return xerrors.Errorf("error inserting message: %w", err)
2023-03-11 17:11:08 +00:00
}
}
return nil
2023-03-11 14:49:07 +00:00
}
// interface
func (x *msgIndex) GetMsgInfo(ctx context.Context, m cid.Cid) (MsgInfo, error) {
2023-03-11 20:09:31 +00:00
x.closeLk.RLock()
defer x.closeLk.RUnlock()
if x.closed {
return MsgInfo{}, ErrClosed
}
2023-03-11 16:21:16 +00:00
var (
tipset string
epoch int64
)
key := m.String()
row := x.selectMsgStmt.QueryRow(key)
err := row.Scan(&tipset, &epoch)
2023-03-11 16:21:16 +00:00
switch {
case err == sql.ErrNoRows:
return MsgInfo{}, ErrNotFound
case err != nil:
return MsgInfo{}, xerrors.Errorf("error querying msgindex database: %w", err)
}
tipsetCid, err := cid.Decode(tipset)
if err != nil {
return MsgInfo{}, xerrors.Errorf("error decoding tipset cid: %w", err)
}
return MsgInfo{
Message: m,
2023-03-12 11:21:03 +00:00
TipSet: tipsetCid,
2023-03-11 16:21:16 +00:00
Epoch: abi.ChainEpoch(epoch),
}, nil
2023-03-11 14:49:07 +00:00
}
func (x *msgIndex) Close() error {
2023-03-11 20:09:31 +00:00
x.closeLk.Lock()
defer x.closeLk.Unlock()
if x.closed {
return nil
}
x.closed = true
x.cancel()
x.workers.Wait()
return x.db.Close()
2023-03-11 14:49:07 +00:00
}
2023-03-13 12:55:26 +00:00
// informal apis for itests; not exposed in the main interface
func (x *msgIndex) CountMessages() (int64, error) {
x.closeLk.RLock()
defer x.closeLk.RUnlock()
if x.closed {
return 0, ErrClosed
}
var result int64
row := x.db.QueryRow(dbqCountMessages)
err := row.Scan(&result)
return result, err
}