lotus/chain/events/events.go

191 lines
4.5 KiB
Go
Raw Normal View History

2019-09-05 07:40:50 +00:00
package events
2019-09-03 17:45:55 +00:00
import (
2019-09-18 11:01:52 +00:00
"context"
2019-09-03 17:45:55 +00:00
"sync"
2019-09-18 11:01:52 +00:00
"time"
2019-09-03 17:45:55 +00:00
2020-02-08 02:18:32 +00:00
"github.com/filecoin-project/specs-actors/actors/abi"
2019-09-18 11:01:52 +00:00
"github.com/ipfs/go-cid"
logging "github.com/ipfs/go-log/v2"
2019-09-04 16:09:08 +00:00
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
2019-09-03 17:45:55 +00:00
)
2019-09-05 07:40:50 +00:00
var log = logging.Logger("events")
// HeightHandler `curH`-`ts.Height` = `confidence`
2020-02-08 02:18:32 +00:00
type HeightHandler func(ctx context.Context, ts *types.TipSet, curH abi.ChainEpoch) error
2019-11-05 14:03:59 +00:00
type RevertHandler func(ctx context.Context, ts *types.TipSet) error
2019-09-03 17:45:55 +00:00
2019-09-04 16:09:08 +00:00
type heightHandler struct {
2019-09-03 17:45:55 +00:00
confidence int
2019-12-04 12:41:22 +00:00
called bool
2019-09-03 17:45:55 +00:00
2019-09-04 16:09:08 +00:00
handle HeightHandler
revert RevertHandler
2019-09-03 17:45:55 +00:00
}
type eventAPI interface {
ChainNotify(context.Context) (<-chan []*api.HeadChange, error)
2019-09-18 11:01:52 +00:00
ChainGetBlockMessages(context.Context, cid.Cid) (*api.BlockMessages, error)
ChainGetTipSetByHeight(context.Context, abi.ChainEpoch, types.TipSetKey) (*types.TipSet, error)
StateGetReceipt(context.Context, cid.Cid, types.TipSetKey) (*types.MessageReceipt, error)
ChainGetTipSet(context.Context, types.TipSetKey) (*types.TipSet, error)
2019-11-19 21:27:25 +00:00
StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error) // optional / for CalledMsg
2019-09-03 17:45:55 +00:00
}
type Events struct {
api eventAPI
2019-09-03 17:45:55 +00:00
tsc *tipSetCache
lk sync.Mutex
2019-09-18 11:01:52 +00:00
ready sync.WaitGroup
readyOnce sync.Once
heightEvents
2019-09-04 16:09:08 +00:00
calledEvents
2019-09-03 17:45:55 +00:00
}
func NewEvents(ctx context.Context, api eventAPI) *Events {
2019-09-03 17:45:55 +00:00
gcConfidence := 2 * build.ForkLengthThreshold
tsc := newTSCache(gcConfidence, api.ChainGetTipSetByHeight)
2019-09-04 16:09:08 +00:00
2019-09-03 17:45:55 +00:00
e := &Events{
2019-09-18 11:01:52 +00:00
api: api,
2019-09-03 17:45:55 +00:00
2019-09-04 16:09:08 +00:00
tsc: tsc,
2019-09-03 17:45:55 +00:00
heightEvents: heightEvents{
tsc: tsc,
2019-11-05 14:03:59 +00:00
ctx: ctx,
gcConfidence: gcConfidence,
heightTriggers: map[uint64]*heightHandler{},
2020-02-08 02:18:32 +00:00
htTriggerHeights: map[abi.ChainEpoch][]uint64{},
htHeights: map[abi.ChainEpoch][]uint64{},
},
2019-09-03 17:45:55 +00:00
2019-09-04 16:09:08 +00:00
calledEvents: calledEvents{
2019-09-18 11:01:52 +00:00
cs: api,
tsc: tsc,
2019-11-05 14:03:59 +00:00
ctx: ctx,
gcConfidence: uint64(gcConfidence),
2019-09-04 16:09:08 +00:00
confQueue: map[triggerH]map[msgH][]*queuedEvent{},
revertQueue: map[msgH][]triggerH{},
triggers: map[triggerID]*callHandler{},
matchers: map[triggerID][]MatchFunc{},
timeouts: map[abi.ChainEpoch]map[triggerID]int{},
2019-09-04 16:09:08 +00:00
},
2019-09-03 17:45:55 +00:00
}
2019-09-18 11:01:52 +00:00
e.ready.Add(1)
go e.listenHeadChanges(ctx)
2019-09-18 11:01:52 +00:00
e.ready.Wait()
2019-09-03 17:45:55 +00:00
2019-09-03 17:59:32 +00:00
// TODO: cleanup/gc goroutine
2019-09-03 17:45:55 +00:00
return e
}
func (e *Events) listenHeadChanges(ctx context.Context) {
for {
if err := e.listenHeadChangesOnce(ctx); err != nil {
log.Errorf("listen head changes errored: %s", err)
} else {
log.Warn("listenHeadChanges quit")
}
2019-09-18 11:01:52 +00:00
if ctx.Err() != nil {
log.Warnf("not restarting listenHeadChanges: context error: %s", ctx.Err())
return
}
time.Sleep(time.Second)
log.Info("restarting listenHeadChanges")
}
2019-09-18 11:01:52 +00:00
}
func (e *Events) listenHeadChangesOnce(ctx context.Context) error {
2019-09-27 11:37:44 +00:00
ctx, cancel := context.WithCancel(ctx)
defer cancel()
2019-09-18 11:01:52 +00:00
notifs, err := e.api.ChainNotify(ctx)
if err != nil {
// TODO: retry
return xerrors.Errorf("listenHeadChanges ChainNotify call failed: %w", err)
2019-09-18 11:01:52 +00:00
}
cur, ok := <-notifs // TODO: timeout?
if !ok {
return xerrors.Errorf("notification channel closed")
2019-09-18 11:01:52 +00:00
}
if len(cur) != 1 {
return xerrors.Errorf("unexpected initial head notification length: %d", len(cur))
2019-09-18 11:01:52 +00:00
}
if cur[0].Type != store.HCCurrent {
return xerrors.Errorf("expected first head notification type to be 'current', was '%s'", cur[0].Type)
2019-09-18 11:01:52 +00:00
}
if err := e.tsc.add(cur[0].Val); err != nil {
log.Warn("tsc.add: adding current tipset failed: %w", err)
2019-09-18 11:01:52 +00:00
}
e.readyOnce.Do(func() {
e.at = cur[0].Val.Height()
2019-09-18 11:01:52 +00:00
e.ready.Done()
})
for notif := range notifs {
var rev, app []*types.TipSet
for _, notif := range notif {
switch notif.Type {
case store.HCRevert:
rev = append(rev, notif.Val)
case store.HCApply:
app = append(app, notif.Val)
default:
log.Warnf("unexpected head change notification type: '%s'", notif.Type)
}
}
if err := e.headChange(rev, app); err != nil {
log.Warnf("headChange failed: %s", err)
}
// sync with fake chainstore (for tests)
2020-05-15 09:17:13 +00:00
if fcs, ok := e.api.(interface{ notifDone() }); ok {
fcs.notifDone()
}
2019-09-18 11:01:52 +00:00
}
return nil
2019-09-18 11:01:52 +00:00
}
2019-09-03 17:45:55 +00:00
func (e *Events) headChange(rev, app []*types.TipSet) error {
2019-09-03 17:59:32 +00:00
if len(app) == 0 {
return xerrors.New("events.headChange expected at least one applied tipset")
}
2019-09-03 17:45:55 +00:00
e.lk.Lock()
defer e.lk.Unlock()
2019-09-03 17:59:32 +00:00
if err := e.headChangeAt(rev, app); err != nil {
return err
2019-09-03 17:45:55 +00:00
}
2019-09-03 17:59:32 +00:00
return e.headChangeCalled(rev, app)
}