2019-09-05 07:40:50 +00:00
|
|
|
package events
|
2019-09-04 16:09:08 +00:00
|
|
|
|
|
|
|
import (
|
2019-09-18 11:01:52 +00:00
|
|
|
"context"
|
2019-09-04 16:09:08 +00:00
|
|
|
"math"
|
|
|
|
"sync"
|
|
|
|
|
2020-02-08 02:18:32 +00:00
|
|
|
"github.com/filecoin-project/specs-actors/actors/abi"
|
2019-09-05 07:40:50 +00:00
|
|
|
"github.com/ipfs/go-cid"
|
2019-11-24 16:35:50 +00:00
|
|
|
"golang.org/x/xerrors"
|
2019-09-05 07:40:50 +00:00
|
|
|
|
2019-11-24 16:35:50 +00:00
|
|
|
"github.com/filecoin-project/lotus/chain/store"
|
2019-10-18 04:47:41 +00:00
|
|
|
"github.com/filecoin-project/lotus/chain/types"
|
2019-09-04 16:09:08 +00:00
|
|
|
)
|
|
|
|
|
2020-02-08 02:18:32 +00:00
|
|
|
const NoTimeout = math.MaxInt64
|
2019-09-05 07:58:50 +00:00
|
|
|
|
2019-09-05 07:31:16 +00:00
|
|
|
type triggerId = uint64
|
|
|
|
|
|
|
|
// msgH is the block height at which a message was present / event has happened
|
2020-02-08 02:18:32 +00:00
|
|
|
type msgH = abi.ChainEpoch
|
2019-09-05 07:31:16 +00:00
|
|
|
|
|
|
|
// triggerH is the block height at which the listener will be notified about the
|
|
|
|
// message (msgH+confidence)
|
2020-02-08 02:18:32 +00:00
|
|
|
type triggerH = abi.ChainEpoch
|
2019-09-04 16:09:08 +00:00
|
|
|
|
|
|
|
// `ts` is the tipset, in which the `msg` is included.
|
|
|
|
// `curH`-`ts.Height` = `confidence`
|
2020-02-08 02:18:32 +00:00
|
|
|
type CalledHandler func(msg *types.Message, rec *types.MessageReceipt, ts *types.TipSet, curH abi.ChainEpoch) (more bool, err error)
|
2019-09-04 16:09:08 +00:00
|
|
|
|
2019-09-04 20:22:25 +00:00
|
|
|
// CheckFunc is used for atomicity guarantees. If the condition the callbacks
|
|
|
|
// wait for has already happened in tipset `ts`
|
|
|
|
//
|
|
|
|
// If `done` is true, timeout won't be triggered
|
|
|
|
// If `more` is false, no messages will be sent to CalledHandler (RevertHandler
|
|
|
|
// may still be called)
|
|
|
|
type CheckFunc func(ts *types.TipSet) (done bool, more bool, err error)
|
2019-09-04 16:09:08 +00:00
|
|
|
|
2019-11-19 15:51:12 +00:00
|
|
|
type MatchFunc func(msg *types.Message) (bool, error)
|
|
|
|
|
2019-09-04 16:09:08 +00:00
|
|
|
type callHandler struct {
|
|
|
|
confidence int
|
2020-02-08 02:18:32 +00:00
|
|
|
timeout abi.ChainEpoch
|
2019-09-04 16:09:08 +00:00
|
|
|
|
2019-09-04 19:41:34 +00:00
|
|
|
disabled bool // TODO: GC after gcConfidence reached
|
|
|
|
|
2019-09-04 16:09:08 +00:00
|
|
|
handle CalledHandler
|
|
|
|
revert RevertHandler
|
|
|
|
}
|
|
|
|
|
|
|
|
type queuedEvent struct {
|
2019-09-05 07:31:16 +00:00
|
|
|
trigger triggerId
|
2019-09-04 16:09:08 +00:00
|
|
|
|
2020-02-08 02:18:32 +00:00
|
|
|
h abi.ChainEpoch
|
2019-09-04 16:09:08 +00:00
|
|
|
msg *types.Message
|
2019-09-04 18:56:06 +00:00
|
|
|
|
|
|
|
called bool
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
type calledEvents struct {
|
2019-09-18 11:01:52 +00:00
|
|
|
cs eventApi
|
2019-09-05 07:36:11 +00:00
|
|
|
tsc *tipSetCache
|
2019-11-05 14:03:59 +00:00
|
|
|
ctx context.Context
|
2019-09-05 07:36:11 +00:00
|
|
|
gcConfidence uint64
|
2019-09-04 16:09:08 +00:00
|
|
|
|
|
|
|
lk sync.Mutex
|
|
|
|
|
2019-09-05 07:31:16 +00:00
|
|
|
ctr triggerId
|
2019-09-04 16:09:08 +00:00
|
|
|
|
2019-11-19 15:51:12 +00:00
|
|
|
triggers map[triggerId]*callHandler
|
|
|
|
matchers map[triggerId][]MatchFunc
|
2019-09-04 19:41:34 +00:00
|
|
|
|
2019-09-04 16:09:08 +00:00
|
|
|
// maps block heights to events
|
|
|
|
// [triggerH][msgH][event]
|
2019-09-05 07:31:16 +00:00
|
|
|
confQueue map[triggerH]map[msgH][]*queuedEvent
|
2019-09-04 16:09:08 +00:00
|
|
|
|
|
|
|
// [msgH][triggerH]
|
2019-09-05 07:31:16 +00:00
|
|
|
revertQueue map[msgH][]triggerH
|
2019-09-04 16:09:08 +00:00
|
|
|
|
2019-09-04 19:41:34 +00:00
|
|
|
// [timeoutH+confidence][triggerId]{calls}
|
2020-02-08 02:18:32 +00:00
|
|
|
timeouts map[abi.ChainEpoch]map[triggerId]int
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (e *calledEvents) headChangeCalled(rev, app []*types.TipSet) error {
|
|
|
|
for _, ts := range rev {
|
|
|
|
e.handleReverts(ts)
|
|
|
|
}
|
|
|
|
|
2019-11-19 22:07:58 +00:00
|
|
|
for _, ts := range app {
|
2019-09-04 16:09:08 +00:00
|
|
|
// called triggers
|
|
|
|
|
2019-09-04 20:01:06 +00:00
|
|
|
e.checkNewCalls(ts)
|
2019-09-04 16:09:08 +00:00
|
|
|
e.applyWithConfidence(ts)
|
2019-09-04 19:41:34 +00:00
|
|
|
e.applyTimeouts(ts)
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (e *calledEvents) handleReverts(ts *types.TipSet) {
|
|
|
|
reverts, ok := e.revertQueue[ts.Height()]
|
|
|
|
if !ok {
|
|
|
|
return // nothing to do
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, triggerH := range reverts {
|
|
|
|
toRevert := e.confQueue[triggerH][ts.Height()]
|
|
|
|
for _, event := range toRevert {
|
2019-09-04 18:56:06 +00:00
|
|
|
if !event.called {
|
|
|
|
continue // event wasn't apply()-ied yet
|
|
|
|
}
|
|
|
|
|
2019-09-04 16:09:08 +00:00
|
|
|
trigger := e.triggers[event.trigger]
|
2019-09-04 18:56:06 +00:00
|
|
|
|
2019-11-05 14:03:59 +00:00
|
|
|
if err := trigger.revert(e.ctx, ts); err != nil {
|
2019-09-04 16:09:08 +00:00
|
|
|
log.Errorf("reverting chain trigger (call %s.%d() @H %d, called @ %d) failed: %s", event.msg.To, event.msg.Method, ts.Height(), triggerH, err)
|
|
|
|
}
|
|
|
|
}
|
2019-09-04 18:56:06 +00:00
|
|
|
delete(e.confQueue[triggerH], ts.Height())
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
2019-09-04 18:56:06 +00:00
|
|
|
delete(e.revertQueue, ts.Height())
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
|
2019-09-04 20:01:06 +00:00
|
|
|
func (e *calledEvents) checkNewCalls(ts *types.TipSet) {
|
|
|
|
e.messagesForTs(ts, func(msg *types.Message) {
|
2019-11-19 15:51:12 +00:00
|
|
|
// TODO: provide receipts
|
|
|
|
|
|
|
|
for tid, matchFns := range e.matchers {
|
|
|
|
var matched bool
|
|
|
|
for _, matchFn := range matchFns {
|
|
|
|
ok, err := matchFn(msg)
|
|
|
|
if err != nil {
|
2020-02-23 20:00:47 +00:00
|
|
|
log.Warnf("event matcher failed: %s", err)
|
2019-11-19 15:51:12 +00:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
matched = ok
|
|
|
|
|
|
|
|
if matched {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
2019-09-04 16:09:08 +00:00
|
|
|
|
2019-11-19 15:51:12 +00:00
|
|
|
if matched {
|
|
|
|
e.queueForConfidence(tid, msg, ts)
|
|
|
|
break
|
|
|
|
}
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
func (e *calledEvents) queueForConfidence(triggerId uint64, msg *types.Message, ts *types.TipSet) {
|
|
|
|
trigger := e.triggers[triggerId]
|
2019-11-07 09:02:29 +00:00
|
|
|
|
|
|
|
// messages are not applied in the tipset they are included in
|
|
|
|
appliedH := ts.Height() + 1
|
|
|
|
|
2020-02-08 02:18:32 +00:00
|
|
|
triggerH := appliedH + abi.ChainEpoch(trigger.confidence)
|
2019-09-04 16:09:08 +00:00
|
|
|
|
|
|
|
byOrigH, ok := e.confQueue[triggerH]
|
|
|
|
if !ok {
|
2020-02-08 02:18:32 +00:00
|
|
|
byOrigH = map[abi.ChainEpoch][]*queuedEvent{}
|
2019-09-04 16:09:08 +00:00
|
|
|
e.confQueue[triggerH] = byOrigH
|
|
|
|
}
|
|
|
|
|
2019-11-07 09:02:29 +00:00
|
|
|
byOrigH[appliedH] = append(byOrigH[appliedH], &queuedEvent{
|
2019-09-04 16:09:08 +00:00
|
|
|
trigger: triggerId,
|
2019-11-07 09:02:29 +00:00
|
|
|
h: appliedH,
|
2019-09-04 16:09:08 +00:00
|
|
|
msg: msg,
|
|
|
|
})
|
2019-09-04 18:56:06 +00:00
|
|
|
|
2019-11-07 09:02:29 +00:00
|
|
|
e.revertQueue[appliedH] = append(e.revertQueue[appliedH], triggerH)
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (e *calledEvents) applyWithConfidence(ts *types.TipSet) {
|
|
|
|
byOrigH, ok := e.confQueue[ts.Height()]
|
|
|
|
if !ok {
|
|
|
|
return // no triggers at thin height
|
|
|
|
}
|
|
|
|
|
|
|
|
for origH, events := range byOrigH {
|
|
|
|
triggerTs, err := e.tsc.get(origH)
|
|
|
|
if err != nil {
|
|
|
|
log.Errorf("events: applyWithConfidence didn't find tipset for event; wanted %d; current %d", origH, ts.Height())
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, event := range events {
|
2019-12-04 12:58:22 +00:00
|
|
|
if event.called {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2019-09-04 16:09:08 +00:00
|
|
|
trigger := e.triggers[event.trigger]
|
2019-09-04 19:41:34 +00:00
|
|
|
if trigger.disabled {
|
|
|
|
continue
|
|
|
|
}
|
2019-09-04 18:56:06 +00:00
|
|
|
|
2020-02-11 23:29:45 +00:00
|
|
|
rec, err := e.cs.StateGetReceipt(e.ctx, event.msg.Cid(), ts.Key())
|
2019-11-19 21:27:25 +00:00
|
|
|
if err != nil {
|
|
|
|
log.Error(err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
more, err := trigger.handle(event.msg, rec, triggerTs, ts.Height())
|
2019-09-04 19:41:34 +00:00
|
|
|
if err != nil {
|
2019-09-04 16:09:08 +00:00
|
|
|
log.Errorf("chain trigger (call %s.%d() @H %d, called @ %d) failed: %s", event.msg.To, event.msg.Method, origH, ts.Height(), err)
|
|
|
|
continue // don't revert failed calls
|
|
|
|
}
|
|
|
|
|
2019-09-04 18:56:06 +00:00
|
|
|
event.called = true
|
2019-09-04 19:41:34 +00:00
|
|
|
|
|
|
|
touts, ok := e.timeouts[trigger.timeout]
|
|
|
|
if ok {
|
|
|
|
touts[event.trigger]++
|
|
|
|
}
|
|
|
|
|
|
|
|
trigger.disabled = !more
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-09-04 19:41:34 +00:00
|
|
|
func (e *calledEvents) applyTimeouts(ts *types.TipSet) {
|
|
|
|
triggers, ok := e.timeouts[ts.Height()]
|
|
|
|
if !ok {
|
|
|
|
return // nothing to do
|
|
|
|
}
|
|
|
|
|
|
|
|
for triggerId, calls := range triggers {
|
|
|
|
if calls > 0 {
|
|
|
|
continue // don't timeout if the method was called
|
|
|
|
}
|
|
|
|
trigger := e.triggers[triggerId]
|
|
|
|
if trigger.disabled {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2020-02-08 02:18:32 +00:00
|
|
|
timeoutTs, err := e.tsc.get(ts.Height() - abi.ChainEpoch(trigger.confidence))
|
2019-09-04 19:41:34 +00:00
|
|
|
if err != nil {
|
2020-02-08 02:18:32 +00:00
|
|
|
log.Errorf("events: applyTimeouts didn't find tipset for event; wanted %d; current %d", ts.Height()-abi.ChainEpoch(trigger.confidence), ts.Height())
|
2019-09-04 19:41:34 +00:00
|
|
|
}
|
|
|
|
|
2019-11-19 21:27:25 +00:00
|
|
|
more, err := trigger.handle(nil, nil, timeoutTs, ts.Height())
|
2019-09-04 19:41:34 +00:00
|
|
|
if err != nil {
|
|
|
|
log.Errorf("chain trigger (call @H %d, called @ %d) failed: %s", timeoutTs.Height(), ts.Height(), err)
|
|
|
|
continue // don't revert failed calls
|
|
|
|
}
|
|
|
|
|
|
|
|
trigger.disabled = !more // allows messages after timeout
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-09-04 20:01:06 +00:00
|
|
|
func (e *calledEvents) messagesForTs(ts *types.TipSet, consume func(*types.Message)) {
|
2019-09-04 16:09:08 +00:00
|
|
|
seen := map[cid.Cid]struct{}{}
|
|
|
|
|
|
|
|
for _, tsb := range ts.Blocks() {
|
2019-09-18 11:01:52 +00:00
|
|
|
|
2019-09-18 11:45:52 +00:00
|
|
|
msgs, err := e.cs.ChainGetBlockMessages(context.TODO(), tsb.Cid())
|
2019-09-04 16:09:08 +00:00
|
|
|
if err != nil {
|
2019-09-04 20:01:06 +00:00
|
|
|
log.Errorf("messagesForTs MessagesForBlock failed (ts.H=%d, Bcid:%s, B.Mcid:%s): %s", ts.Height(), tsb.Cid(), tsb.Messages, err)
|
|
|
|
// this is quite bad, but probably better than missing all the other updates
|
|
|
|
continue
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
|
2019-09-18 11:01:52 +00:00
|
|
|
for _, m := range msgs.BlsMessages {
|
2019-09-04 16:09:08 +00:00
|
|
|
_, ok := seen[m.Cid()]
|
|
|
|
if ok {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
seen[m.Cid()] = struct{}{}
|
|
|
|
|
|
|
|
consume(m)
|
|
|
|
}
|
|
|
|
|
2019-09-18 11:01:52 +00:00
|
|
|
for _, m := range msgs.SecpkMessages {
|
2019-09-04 16:09:08 +00:00
|
|
|
_, ok := seen[m.Message.Cid()]
|
|
|
|
if ok {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
seen[m.Message.Cid()] = struct{}{}
|
|
|
|
|
|
|
|
consume(&m.Message)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-09-05 08:27:08 +00:00
|
|
|
// Called registers a callbacks which are triggered when a specified method is
|
|
|
|
// called on an actor, or a timeout is reached.
|
|
|
|
//
|
|
|
|
// * `CheckFunc` callback is invoked immediately with a recent tipset, it
|
|
|
|
// returns two booleans - `done`, and `more`.
|
|
|
|
//
|
|
|
|
// * `done` should be true when some on-chain action we are waiting for has
|
|
|
|
// happened. When `done` is set to true, timeout trigger is disabled.
|
|
|
|
//
|
|
|
|
// * `more` should be false when we don't want to receive new notifications
|
|
|
|
// through CalledHandler. Note that notifications may still be delivered to
|
|
|
|
// RevertHandler
|
|
|
|
//
|
|
|
|
// * `CalledHandler` is called when the specified event was observed on-chain,
|
|
|
|
// and a confidence threshold was reached, or the specified `timeout` height
|
|
|
|
// was reached with no events observed. When this callback is invoked on a
|
|
|
|
// timeout, `msg` is set to nil. This callback returns a boolean specifying
|
|
|
|
// whether further notifications should be sent, like `more` return param
|
|
|
|
// from `CheckFunc` above.
|
|
|
|
//
|
|
|
|
// * `RevertHandler` is called after apply handler, when we drop the tipset
|
|
|
|
// containing the message. The tipset passed as the argument is the tipset
|
|
|
|
// that is being dropped. Note that the message dropped may be re-applied
|
|
|
|
// in a different tipset in small amount of time.
|
2020-02-08 02:18:32 +00:00
|
|
|
func (e *calledEvents) Called(check CheckFunc, hnd CalledHandler, rev RevertHandler, confidence int, timeout abi.ChainEpoch, mf MatchFunc) error {
|
2019-09-04 16:09:08 +00:00
|
|
|
e.lk.Lock()
|
|
|
|
defer e.lk.Unlock()
|
|
|
|
|
2019-11-24 16:35:50 +00:00
|
|
|
ts := e.tsc.best()
|
|
|
|
done, more, err := check(ts)
|
2019-09-04 16:09:08 +00:00
|
|
|
if err != nil {
|
2019-11-24 16:35:50 +00:00
|
|
|
return xerrors.Errorf("called check error (h: %d): %w", ts.Height(), err)
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
if done {
|
2019-09-05 07:58:50 +00:00
|
|
|
timeout = NoTimeout
|
2019-09-04 16:09:08 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
id := e.ctr
|
|
|
|
e.ctr++
|
|
|
|
|
2019-09-04 19:41:34 +00:00
|
|
|
e.triggers[id] = &callHandler{
|
2019-09-04 16:09:08 +00:00
|
|
|
confidence: confidence,
|
2020-02-08 02:18:32 +00:00
|
|
|
timeout: timeout + abi.ChainEpoch(confidence),
|
2019-09-04 16:09:08 +00:00
|
|
|
|
2019-09-04 20:22:25 +00:00
|
|
|
disabled: !more,
|
|
|
|
|
2019-09-04 16:09:08 +00:00
|
|
|
handle: hnd,
|
|
|
|
revert: rev,
|
|
|
|
}
|
|
|
|
|
2019-11-19 15:51:12 +00:00
|
|
|
e.matchers[id] = append(e.matchers[id], mf)
|
2019-09-04 16:09:08 +00:00
|
|
|
|
2019-09-05 07:58:50 +00:00
|
|
|
if timeout != NoTimeout {
|
2020-02-08 02:18:32 +00:00
|
|
|
if e.timeouts[timeout+abi.ChainEpoch(confidence)] == nil {
|
|
|
|
e.timeouts[timeout+abi.ChainEpoch(confidence)] = map[uint64]int{}
|
2019-09-04 19:41:34 +00:00
|
|
|
}
|
2020-02-08 02:18:32 +00:00
|
|
|
e.timeouts[timeout+abi.ChainEpoch(confidence)][id] = 0
|
2019-09-04 19:41:34 +00:00
|
|
|
}
|
|
|
|
|
2019-09-04 16:09:08 +00:00
|
|
|
return nil
|
|
|
|
}
|
2019-11-19 21:27:25 +00:00
|
|
|
|
2020-02-08 02:18:32 +00:00
|
|
|
func (e *calledEvents) CalledMsg(ctx context.Context, hnd CalledHandler, rev RevertHandler, confidence int, timeout abi.ChainEpoch, msg store.ChainMsg) error {
|
2019-11-24 16:35:50 +00:00
|
|
|
return e.Called(e.CheckMsg(ctx, msg, hnd), hnd, rev, confidence, timeout, e.MatchMsg(msg.VMMessage()))
|
2019-11-19 21:27:25 +00:00
|
|
|
}
|