lnd.xprv/contractcourt/channel_arbitrator.go
Joost Jager 86eb0a3383
cnct: log go to chain reason
This commit adds logging of the reason to go to chain for a channel.
This can help users to find out the reason why a channels forced closed.

To get all go to chain reasons, an optimization to break early is
removed. This optimization was not significant, because the normal flow
already examined all htlcs. In the exceptional case where we need to go
to chain, it does not weigh up against logging all go to chain reasons.
2019-04-08 10:34:41 +02:00

1912 lines
61 KiB
Go

package contractcourt
import (
"bytes"
"errors"
"sync"
"sync/atomic"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btcutil"
"github.com/davecgh/go-spew/spew"
"github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/channeldb"
"github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwallet"
"github.com/lightningnetwork/lnd/lnwire"
)
var (
// errAlreadyForceClosed is an error returned when we attempt to force
// close a channel that's already in the process of doing so.
errAlreadyForceClosed = errors.New("channel is already in the " +
"process of being force closed")
)
// WitnessSubscription represents an intent to be notified once new witnesses
// are discovered by various active contract resolvers. A contract resolver may
// use this to be notified of when it can satisfy an incoming contract after we
// discover the witness for an outgoing contract.
type WitnessSubscription struct {
// WitnessUpdates is a channel that newly discovered witnesses will be
// sent over.
//
// TODO(roasbeef): couple with WitnessType?
WitnessUpdates <-chan lntypes.Preimage
// CancelSubscription is a function closure that should be used by a
// client to cancel the subscription once they are no longer interested
// in receiving new updates.
CancelSubscription func()
}
// WitnessBeacon is a global beacon of witnesses. Contract resolvers will use
// this interface to lookup witnesses (preimages typically) of contracts
// they're trying to resolve, add new preimages they resolve, and finally
// receive new updates each new time a preimage is discovered.
//
// TODO(roasbeef): need to delete the pre-images once we've used them
// and have been sufficiently confirmed?
type WitnessBeacon interface {
// SubscribeUpdates returns a channel that will be sent upon *each* time
// a new preimage is discovered.
SubscribeUpdates() *WitnessSubscription
// LookupPreImage attempts to lookup a preimage in the global cache.
// True is returned for the second argument if the preimage is found.
LookupPreimage(payhash lntypes.Hash) (lntypes.Preimage, bool)
// AddPreimages adds a batch of newly discovered preimages to the global
// cache, and also signals any subscribers of the newly discovered
// witness.
AddPreimages(preimages ...lntypes.Preimage) error
}
// ChannelArbitratorConfig contains all the functionality that the
// ChannelArbitrator needs in order to properly arbitrate any contract dispute
// on chain.
type ChannelArbitratorConfig struct {
// ChanPoint is the channel point that uniquely identifies this
// channel.
ChanPoint wire.OutPoint
// ShortChanID describes the exact location of the channel within the
// chain. We'll use this to address any messages that we need to send
// to the switch during contract resolution.
ShortChanID lnwire.ShortChannelID
// BlockEpochs is an active block epoch event stream backed by an
// active ChainNotifier instance. We will use new block notifications
// sent over this channel to decide when we should go on chain to
// reclaim/redeem the funds in an HTLC sent to/from us.
BlockEpochs *chainntnfs.BlockEpochEvent
// ChainEvents is an active subscription to the chain watcher for this
// channel to be notified of any on-chain activity related to this
// channel.
ChainEvents *ChainEventSubscription
// ForceCloseChan should force close the contract that this attendant
// is watching over. We'll use this when we decide that we need to go
// to chain. It should in addition tell the switch to remove the
// corresponding link, such that we won't accept any new updates. The
// returned summary contains all items needed to eventually resolve all
// outputs on chain.
ForceCloseChan func() (*lnwallet.LocalForceCloseSummary, error)
// MarkCommitmentBroadcasted should mark the channel as the commitment
// being broadcast, and we are waiting for the commitment to confirm.
MarkCommitmentBroadcasted func() error
// MarkChannelClosed marks the channel closed in the database, with the
// passed close summary. After this method successfully returns we can
// no longer expect to receive chain events for this channel, and must
// be able to recover from a failure without getting the close event
// again.
MarkChannelClosed func(*channeldb.ChannelCloseSummary) error
// IsPendingClose is a boolean indicating whether the channel is marked
// as pending close in the database.
IsPendingClose bool
// ClosingHeight is the height at which the channel was closed. Note
// that this value is only valid if IsPendingClose is true.
ClosingHeight uint32
// CloseType is the type of the close event in case IsPendingClose is
// true. Otherwise this value is unset.
CloseType channeldb.ClosureType
// MarkChannelResolved is a function closure that serves to mark a
// channel as "fully resolved". A channel itself can be considered
// fully resolved once all active contracts have individually been
// fully resolved.
//
// TODO(roasbeef): need RPC's to combine for pendingchannels RPC
MarkChannelResolved func() error
ChainArbitratorConfig
}
// ContractReport provides a summary of a commitment tx output.
type ContractReport struct {
// Outpoint is the final output that will be swept back to the wallet.
Outpoint wire.OutPoint
// Incoming indicates whether the htlc was incoming to this channel.
Incoming bool
// Amount is the final value that will be swept in back to the wallet.
Amount btcutil.Amount
// MaturityHeight is the absolute block height that this output will
// mature at.
MaturityHeight uint32
// Stage indicates whether the htlc is in the CLTV-timeout stage (1) or
// the CSV-delay stage (2). A stage 1 htlc's maturity height will be set
// to its expiry height, while a stage 2 htlc's maturity height will be
// set to its confirmation height plus the maturity requirement.
Stage uint32
// LimboBalance is the total number of frozen coins within this
// contract.
LimboBalance btcutil.Amount
// RecoveredBalance is the total value that has been successfully swept
// back to the user's wallet.
RecoveredBalance btcutil.Amount
}
// htlcSet represents the set of active HTLCs on a given commitment
// transaction.
type htlcSet struct {
// incomingHTLCs is a map of all incoming HTLCs on our commitment
// transaction. We may potentially go onchain to claim the funds sent
// to us within this set.
incomingHTLCs map[uint64]channeldb.HTLC
// outgoingHTLCs is a map of all outgoing HTLCs on our commitment
// transaction. We may potentially go onchain to reclaim the funds that
// are currently in limbo.
outgoingHTLCs map[uint64]channeldb.HTLC
}
// newHtlcSet constructs a new HTLC set from a slice of HTLC's.
func newHtlcSet(htlcs []channeldb.HTLC) htlcSet {
outHTLCs := make(map[uint64]channeldb.HTLC)
inHTLCs := make(map[uint64]channeldb.HTLC)
for _, htlc := range htlcs {
if htlc.Incoming {
inHTLCs[htlc.HtlcIndex] = htlc
continue
}
outHTLCs[htlc.HtlcIndex] = htlc
}
return htlcSet{
incomingHTLCs: inHTLCs,
outgoingHTLCs: outHTLCs,
}
}
// ChannelArbitrator is the on-chain arbitrator for a particular channel. The
// struct will keep in sync with the current set of HTLCs on the commitment
// transaction. The job of the attendant is to go on-chain to either settle or
// cancel an HTLC as necessary iff: an HTLC times out, or we known the
// pre-image to an HTLC, but it wasn't settled by the link off-chain. The
// ChannelArbitrator will factor in an expected confirmation delta when
// broadcasting to ensure that we avoid any possibility of race conditions, and
// sweep the output(s) without contest.
type ChannelArbitrator struct {
started int32 // To be used atomically.
stopped int32 // To be used atomically.
// log is a persistent log that the attendant will use to checkpoint
// its next action, and the state of any unresolved contracts.
log ArbitratorLog
// activeHTLCs is the set of active incoming/outgoing HTLC's on the
// commitment transaction.
activeHTLCs htlcSet
// cfg contains all the functionality that the ChannelArbitrator requires
// to do its duty.
cfg ChannelArbitratorConfig
// signalUpdates is a channel that any new live signals for the channel
// we're watching over will be sent.
signalUpdates chan *signalUpdateMsg
// htlcUpdates is a channel that is sent upon with new updates from the
// active channel. Each time a new commitment state is accepted, the
// set of HTLC's on the new state should be sent across this channel.
htlcUpdates <-chan []channeldb.HTLC
// activeResolvers is a slice of any active resolvers. This is used to
// be able to signal them for shutdown in the case that we shutdown.
activeResolvers []ContractResolver
// activeResolversLock prevents simultaneous read and write to the
// resolvers slice.
activeResolversLock sync.RWMutex
// resolutionSignal is a channel that will be sent upon by contract
// resolvers once their contract has been fully resolved. With each
// send, we'll check to see if the contract is fully resolved.
resolutionSignal chan struct{}
// forceCloseReqs is a channel that requests to forcibly close the
// contract will be sent over.
forceCloseReqs chan *forceCloseReq
// state is the current state of the arbitrator. This state is examined
// upon start up to decide which actions to take.
state ArbitratorState
wg sync.WaitGroup
quit chan struct{}
}
// NewChannelArbitrator returns a new instance of a ChannelArbitrator backed by
// the passed config struct.
func NewChannelArbitrator(cfg ChannelArbitratorConfig,
startingHTLCs []channeldb.HTLC, log ArbitratorLog) *ChannelArbitrator {
return &ChannelArbitrator{
log: log,
signalUpdates: make(chan *signalUpdateMsg),
htlcUpdates: make(<-chan []channeldb.HTLC),
resolutionSignal: make(chan struct{}),
forceCloseReqs: make(chan *forceCloseReq),
activeHTLCs: newHtlcSet(startingHTLCs),
cfg: cfg,
quit: make(chan struct{}),
}
}
// Start starts all the goroutines that the ChannelArbitrator needs to operate.
func (c *ChannelArbitrator) Start() error {
if !atomic.CompareAndSwapInt32(&c.started, 0, 1) {
return nil
}
var (
err error
)
log.Debugf("Starting ChannelArbitrator(%v), htlc_set=%v",
c.cfg.ChanPoint, newLogClosure(func() string {
return spew.Sdump(c.activeHTLCs)
}),
)
// First, we'll read our last state from disk, so our internal state
// machine can act accordingly.
c.state, err = c.log.CurrentState()
if err != nil {
c.cfg.BlockEpochs.Cancel()
return err
}
log.Infof("ChannelArbitrator(%v): starting state=%v", c.cfg.ChanPoint,
c.state)
_, bestHeight, err := c.cfg.ChainIO.GetBestBlock()
if err != nil {
c.cfg.BlockEpochs.Cancel()
return err
}
// If the channel has been marked pending close in the database, and we
// haven't transitioned the state machine to StateContractClosed (or a
// succeeding state), then a state transition most likely failed. We'll
// try to recover from this by manually advancing the state by setting
// the corresponding close trigger.
trigger := chainTrigger
triggerHeight := uint32(bestHeight)
if c.cfg.IsPendingClose {
switch c.state {
case StateDefault:
fallthrough
case StateBroadcastCommit:
fallthrough
case StateCommitmentBroadcasted:
switch c.cfg.CloseType {
case channeldb.CooperativeClose:
trigger = coopCloseTrigger
case channeldb.LocalForceClose:
trigger = localCloseTrigger
case channeldb.RemoteForceClose:
trigger = remoteCloseTrigger
}
triggerHeight = c.cfg.ClosingHeight
log.Warnf("ChannelArbitrator(%v): detected stalled "+
"state=%v for closed channel, using "+
"trigger=%v", c.cfg.ChanPoint, c.state, trigger)
}
}
// We'll now attempt to advance our state forward based on the current
// on-chain state, and our set of active contracts.
startingState := c.state
nextState, _, err := c.advanceState(triggerHeight, trigger)
if err != nil {
switch err {
// If we detect that we tried to fetch resolutions, but failed,
// this channel was marked closed in the database before
// resolutions successfully written. In this case there is not
// much we can do, so we don't return the error.
case errScopeBucketNoExist:
fallthrough
case errNoResolutions:
log.Warnf("ChannelArbitrator(%v): detected closed"+
"channel with no contract resolutions written.",
c.cfg.ChanPoint)
default:
c.cfg.BlockEpochs.Cancel()
return err
}
}
// If we start and ended at the awaiting full resolution state, then
// we'll relaunch our set of unresolved contracts.
if startingState == StateWaitingFullResolution &&
nextState == StateWaitingFullResolution {
if err := c.relaunchResolvers(); err != nil {
c.cfg.BlockEpochs.Cancel()
return err
}
}
// TODO(roasbeef): cancel if breached
c.wg.Add(1)
go c.channelAttendant(bestHeight)
return nil
}
// relauchResolvers relaunches the set of resolvers for unresolved contracts in
// order to provide them with information that's not immediately available upon
// starting the ChannelArbitrator. This information should ideally be stored in
// the database, so this only serves as a intermediate work-around to prevent a
// migration.
func (c *ChannelArbitrator) relaunchResolvers() error {
// We'll now query our log to see if there are any active
// unresolved contracts. If this is the case, then we'll
// relaunch all contract resolvers.
unresolvedContracts, err := c.log.FetchUnresolvedContracts()
if err != nil {
return err
}
// Retrieve the commitment tx hash from the log.
contractResolutions, err := c.log.FetchContractResolutions()
if err != nil {
log.Errorf("unable to fetch contract resolutions: %v",
err)
return err
}
commitHash := contractResolutions.CommitHash
// Reconstruct the htlc outpoints and data from the chain action log.
// The purpose of the constructed htlc map is to supplement to resolvers
// restored from database with extra data. Ideally this data is stored
// as part of the resolver in the log. This is a workaround to prevent a
// db migration.
htlcMap := make(map[wire.OutPoint]*channeldb.HTLC)
chainActions, err := c.log.FetchChainActions()
if err != nil {
log.Errorf("unable to fetch chain actions: %v", err)
return err
}
for _, htlcs := range chainActions {
for _, htlc := range htlcs {
outpoint := wire.OutPoint{
Hash: commitHash,
Index: uint32(htlc.OutputIndex),
}
htlcMap[outpoint] = &htlc
}
}
log.Infof("ChannelArbitrator(%v): relaunching %v contract "+
"resolvers", c.cfg.ChanPoint, len(unresolvedContracts))
for _, resolver := range unresolvedContracts {
supplementResolver(resolver, htlcMap)
}
c.launchResolvers(unresolvedContracts)
return nil
}
// supplementResolver takes a resolver as it is restored from the log and fills
// in missing data from the htlcMap.
func supplementResolver(resolver ContractResolver,
htlcMap map[wire.OutPoint]*channeldb.HTLC) error {
switch r := resolver.(type) {
case *htlcSuccessResolver:
return supplementSuccessResolver(r, htlcMap)
case *htlcIncomingContestResolver:
return supplementSuccessResolver(
&r.htlcSuccessResolver, htlcMap,
)
case *htlcTimeoutResolver:
return supplementTimeoutResolver(r, htlcMap)
case *htlcOutgoingContestResolver:
return supplementTimeoutResolver(
&r.htlcTimeoutResolver, htlcMap,
)
}
return nil
}
// supplementSuccessResolver takes a htlcSuccessResolver as it is restored from
// the log and fills in missing data from the htlcMap.
func supplementSuccessResolver(r *htlcSuccessResolver,
htlcMap map[wire.OutPoint]*channeldb.HTLC) error {
res := r.htlcResolution
htlcPoint := res.HtlcPoint()
htlc, ok := htlcMap[htlcPoint]
if !ok {
return errors.New(
"htlc for success resolver unavailable",
)
}
r.htlcAmt = htlc.Amt
return nil
}
// supplementTimeoutResolver takes a htlcSuccessResolver as it is restored from
// the log and fills in missing data from the htlcMap.
func supplementTimeoutResolver(r *htlcTimeoutResolver,
htlcMap map[wire.OutPoint]*channeldb.HTLC) error {
res := r.htlcResolution
htlcPoint := res.HtlcPoint()
htlc, ok := htlcMap[htlcPoint]
if !ok {
return errors.New(
"htlc for timeout resolver unavailable",
)
}
r.htlcAmt = htlc.Amt
return nil
}
// Report returns htlc reports for the active resolvers.
func (c *ChannelArbitrator) Report() []*ContractReport {
c.activeResolversLock.RLock()
defer c.activeResolversLock.RUnlock()
var reports []*ContractReport
for _, resolver := range c.activeResolvers {
r, ok := resolver.(reportingContractResolver)
if !ok {
continue
}
if r.IsResolved() {
continue
}
report := r.report()
if report == nil {
continue
}
reports = append(reports, report)
}
return reports
}
// Stop signals the ChannelArbitrator for a graceful shutdown.
func (c *ChannelArbitrator) Stop() error {
if !atomic.CompareAndSwapInt32(&c.stopped, 0, 1) {
return nil
}
log.Debugf("Stopping ChannelArbitrator(%v)", c.cfg.ChanPoint)
if c.cfg.ChainEvents.Cancel != nil {
go c.cfg.ChainEvents.Cancel()
}
c.activeResolversLock.RLock()
for _, activeResolver := range c.activeResolvers {
activeResolver.Stop()
}
c.activeResolversLock.RUnlock()
close(c.quit)
c.wg.Wait()
return nil
}
// transitionTrigger is an enum that denotes exactly *why* a state transition
// was initiated. This is useful as depending on the initial trigger, we may
// skip certain states as those actions are expected to have already taken
// place as a result of the external trigger.
type transitionTrigger uint8
const (
// chainTrigger is a transition trigger that has been attempted due to
// changing on-chain conditions such as a block which times out HTLC's
// being attached.
chainTrigger transitionTrigger = iota
// userTrigger is a transition trigger driven by user action. Examples
// of such a trigger include a user requesting a force closure of the
// channel.
userTrigger
// remoteCloseTrigger is a transition trigger driven by the remote
// peer's commitment being confirmed.
remoteCloseTrigger
// localCloseTrigger is a transition trigger driven by our commitment
// being confirmed.
localCloseTrigger
// coopCloseTrigger is a transition trigger driven by a cooperative
// close transaction being confirmed.
coopCloseTrigger
)
// String returns a human readable string describing the passed
// transitionTrigger.
func (t transitionTrigger) String() string {
switch t {
case chainTrigger:
return "chainTrigger"
case remoteCloseTrigger:
return "remoteCloseTrigger"
case userTrigger:
return "userTrigger"
case localCloseTrigger:
return "localCloseTrigger"
case coopCloseTrigger:
return "coopCloseTrigger"
default:
return "unknown trigger"
}
}
// stateStep is a help method that examines our internal state, and attempts
// the appropriate state transition if necessary. The next state we transition
// to is returned, Additionally, if the next transition results in a commitment
// broadcast, the commitment transaction itself is returned.
func (c *ChannelArbitrator) stateStep(triggerHeight uint32,
trigger transitionTrigger) (ArbitratorState, *wire.MsgTx, error) {
var (
nextState ArbitratorState
closeTx *wire.MsgTx
)
switch c.state {
// If we're in the default state, then we'll check our set of actions
// to see if while we were down, conditions have changed.
case StateDefault:
log.Debugf("ChannelArbitrator(%v): new block (height=%v) "+
"examining active HTLC's", c.cfg.ChanPoint,
triggerHeight)
// As a new block has been connected to the end of the main
// chain, we'll check to see if we need to make any on-chain
// claims on behalf of the channel contract that we're
// arbitrating for.
chainActions := c.checkChainActions(triggerHeight, trigger)
// If there are no actions to be made, then we'll remain in the
// default state. If this isn't a self initiated event (we're
// checking due to a chain update), then we'll exit now.
if len(chainActions) == 0 && trigger == chainTrigger {
log.Tracef("ChannelArbitrator(%v): no actions for "+
"chain trigger, terminating", c.cfg.ChanPoint)
return StateDefault, closeTx, nil
}
// Otherwise, we'll log that we checked the HTLC actions as the
// commitment transaction has already been broadcast.
log.Tracef("ChannelArbitrator(%v): logging chain_actions=%v",
c.cfg.ChanPoint,
newLogClosure(func() string {
return spew.Sdump(chainActions)
}))
if err := c.log.LogChainActions(chainActions); err != nil {
return StateError, closeTx, err
}
// Depending on the type of trigger, we'll either "tunnel"
// through to a farther state, or just proceed linearly to the
// next state.
switch trigger {
// If this is a chain trigger, then we'll go straight to the
// next state, as we still need to broadcast the commitment
// transaction.
case chainTrigger:
fallthrough
case userTrigger:
nextState = StateBroadcastCommit
// If the trigger is a cooperative close being confirmed, then
// we can go straight to StateFullyResolved, as there won't be
// any contracts to resolve.
case coopCloseTrigger:
nextState = StateFullyResolved
// Otherwise, if this state advance was triggered by a
// commitment being confirmed on chain, then we'll jump
// straight to the state where the contract has already been
// closed, and we will inspect the set of unresolved contracts.
case localCloseTrigger:
log.Errorf("ChannelArbitrator(%v): unexpected local "+
"commitment confirmed while in StateDefault",
c.cfg.ChanPoint)
fallthrough
case remoteCloseTrigger:
nextState = StateContractClosed
}
// If we're in this state, then we've decided to broadcast the
// commitment transaction. We enter this state either due to an outside
// sub-system, or because an on-chain action has been triggered.
case StateBroadcastCommit:
// Under normal operation, we can only enter
// StateBroadcastCommit via a user or chain trigger. On restart,
// this state may be reexecuted after closing the channel, but
// failing to commit to StateContractClosed or
// StateFullyResolved. In that case, one of the three close
// triggers will be presented, signifying that we should skip
// rebroadcasting, and go straight to resolving the on-chain
// contract or marking the channel resolved.
switch trigger {
case localCloseTrigger, remoteCloseTrigger:
log.Infof("ChannelArbitrator(%v): detected %s "+
"close after closing channel, fast-forwarding "+
"to %s to resolve contract",
c.cfg.ChanPoint, trigger, StateContractClosed)
return StateContractClosed, closeTx, nil
case coopCloseTrigger:
log.Infof("ChannelArbitrator(%v): detected %s "+
"close after closing channel, fast-forwarding "+
"to %s to resolve contract",
c.cfg.ChanPoint, trigger, StateFullyResolved)
return StateFullyResolved, closeTx, nil
}
log.Infof("ChannelArbitrator(%v): force closing "+
"chan", c.cfg.ChanPoint)
// Now that we have all the actions decided for the set of
// HTLC's, we'll broadcast the commitment transaction, and
// signal the link to exit.
// We'll tell the switch that it should remove the link for
// this channel, in addition to fetching the force close
// summary needed to close this channel on chain.
closeSummary, err := c.cfg.ForceCloseChan()
if err != nil {
log.Errorf("ChannelArbitrator(%v): unable to "+
"force close: %v", c.cfg.ChanPoint, err)
return StateError, closeTx, err
}
closeTx = closeSummary.CloseTx
// With the close transaction in hand, broadcast the
// transaction to the network, thereby entering the post
// channel resolution state.
log.Infof("Broadcasting force close transaction, "+
"ChannelPoint(%v): %v", c.cfg.ChanPoint,
newLogClosure(func() string {
return spew.Sdump(closeTx)
}))
// At this point, we'll now broadcast the commitment
// transaction itself.
if err := c.cfg.PublishTx(closeTx); err != nil {
log.Errorf("ChannelArbitrator(%v): unable to broadcast "+
"close tx: %v", c.cfg.ChanPoint, err)
if err != lnwallet.ErrDoubleSpend {
return StateError, closeTx, err
}
}
if err := c.cfg.MarkCommitmentBroadcasted(); err != nil {
log.Errorf("ChannelArbitrator(%v): unable to "+
"mark commitment broadcasted: %v",
c.cfg.ChanPoint, err)
}
// We go to the StateCommitmentBroadcasted state, where we'll
// be waiting for the commitment to be confirmed.
nextState = StateCommitmentBroadcasted
// In this state we have broadcasted our own commitment, and will need
// to wait for a commitment (not necessarily the one we broadcasted!)
// to be confirmed.
case StateCommitmentBroadcasted:
switch trigger {
// We are waiting for a commitment to be confirmed, so any
// other trigger will be ignored.
case chainTrigger, userTrigger:
log.Infof("ChannelArbitrator(%v): noop trigger %v",
c.cfg.ChanPoint, trigger)
nextState = StateCommitmentBroadcasted
// If this state advance was triggered by any of the
// commitments being confirmed, then we'll jump to the state
// where the contract has been closed.
case localCloseTrigger, remoteCloseTrigger:
log.Infof("ChannelArbitrator(%v): trigger %v, "+
" going to StateContractClosed",
c.cfg.ChanPoint, trigger)
nextState = StateContractClosed
case coopCloseTrigger:
log.Infof("ChannelArbitrator(%v): trigger %v, "+
" going to StateFullyResolved",
c.cfg.ChanPoint, trigger)
nextState = StateFullyResolved
}
// If we're in this state, then the contract has been fully closed to
// outside sub-systems, so we'll process the prior set of on-chain
// contract actions and launch a set of resolvers.
case StateContractClosed:
// First, we'll fetch our chain actions, and both sets of
// resolutions so we can process them.
chainActions, err := c.log.FetchChainActions()
if err != nil {
log.Errorf("unable to fetch chain actions: %v", err)
return StateError, closeTx, err
}
contractResolutions, err := c.log.FetchContractResolutions()
if err != nil {
log.Errorf("unable to fetch contract resolutions: %v",
err)
return StateError, closeTx, err
}
// If the resolution is empty, then we're done here. We don't
// need to launch any resolvers, and can go straight to our
// final state.
if contractResolutions.IsEmpty() {
log.Infof("ChannelArbitrator(%v): contract "+
"resolutions empty, marking channel as fully resolved!",
c.cfg.ChanPoint)
nextState = StateFullyResolved
break
}
// If we've have broadcast the commitment transaction, we send
// our commitment output for incubation, but only if it wasn't
// trimmed. We'll need to wait for a CSV timeout before we can
// reclaim the funds.
commitRes := contractResolutions.CommitResolution
if commitRes != nil && commitRes.MaturityDelay > 0 {
log.Infof("ChannelArbitrator(%v): sending commit "+
"output for incubation", c.cfg.ChanPoint)
err = c.cfg.IncubateOutputs(
c.cfg.ChanPoint, commitRes,
nil, nil, triggerHeight,
)
if err != nil {
// TODO(roasbeef): check for AlreadyExists errors
log.Errorf("unable to incubate commitment "+
"output: %v", err)
return StateError, closeTx, err
}
}
// Now that we know we'll need to act, we'll process the htlc
// actions, wen create the structures we need to resolve all
// outstanding contracts.
htlcResolvers, pktsToSend, err := c.prepContractResolutions(
chainActions, contractResolutions, triggerHeight,
)
if err != nil {
log.Errorf("ChannelArbitrator(%v): unable to "+
"resolve contracts: %v", c.cfg.ChanPoint, err)
return StateError, closeTx, err
}
log.Debugf("ChannelArbitrator(%v): sending resolution message=%v",
c.cfg.ChanPoint,
newLogClosure(func() string {
return spew.Sdump(pktsToSend)
}))
// With the commitment broadcast, we'll then send over all
// messages we can send immediately.
err = c.cfg.DeliverResolutionMsg(pktsToSend...)
if err != nil {
// TODO(roasbeef): make sure packet sends are idempotent
log.Errorf("unable to send pkts: %v", err)
return StateError, closeTx, err
}
log.Debugf("ChannelArbitrator(%v): inserting %v contract "+
"resolvers", c.cfg.ChanPoint, len(htlcResolvers))
err = c.log.InsertUnresolvedContracts(htlcResolvers...)
if err != nil {
return StateError, closeTx, err
}
// Finally, we'll launch all the required contract resolvers.
// Once they're all resolved, we're no longer needed.
c.launchResolvers(htlcResolvers)
nextState = StateWaitingFullResolution
// This is our terminal state. We'll keep returning this state until
// all contracts are fully resolved.
case StateWaitingFullResolution:
log.Infof("ChannelArbitrator(%v): still awaiting contract "+
"resolution", c.cfg.ChanPoint)
numUnresolved, err := c.log.FetchUnresolvedContracts()
if err != nil {
return StateError, closeTx, err
}
// If we still have unresolved contracts, then we'll stay alive
// to oversee their resolution.
if len(numUnresolved) != 0 {
nextState = StateWaitingFullResolution
break
}
nextState = StateFullyResolved
// If we start as fully resolved, then we'll end as fully resolved.
case StateFullyResolved:
// To ensure that the state of the contract in persistent
// storage is properly reflected, we'll mark the contract as
// fully resolved now.
nextState = StateFullyResolved
log.Infof("ChannelPoint(%v) has been fully resolved "+
"on-chain at height=%v", c.cfg.ChanPoint, triggerHeight)
if err := c.cfg.MarkChannelResolved(); err != nil {
log.Errorf("unable to mark channel resolved: %v", err)
return StateError, closeTx, err
}
}
log.Tracef("ChannelArbitrator(%v): next_state=%v", c.cfg.ChanPoint,
nextState)
return nextState, closeTx, nil
}
// launchResolvers updates the activeResolvers list and starts the resolvers.
func (c *ChannelArbitrator) launchResolvers(resolvers []ContractResolver) {
c.activeResolversLock.Lock()
defer c.activeResolversLock.Unlock()
c.activeResolvers = resolvers
for _, contract := range resolvers {
c.wg.Add(1)
go c.resolveContract(contract)
}
}
// advanceState is the main driver of our state machine. This method is an
// iterative function which repeatedly attempts to advance the internal state
// of the channel arbitrator. The state will be advanced until we reach a
// redundant transition, meaning that the state transition is a noop. The final
// param is a callback that allows the caller to execute an arbitrary action
// after each state transition.
func (c *ChannelArbitrator) advanceState(triggerHeight uint32,
trigger transitionTrigger) (ArbitratorState, *wire.MsgTx, error) {
var (
priorState ArbitratorState
forceCloseTx *wire.MsgTx
)
// We'll continue to advance our state forward until the state we
// transition to is that same state that we started at.
for {
priorState = c.state
log.Tracef("ChannelArbitrator(%v): attempting state step with "+
"trigger=%v from state=%v", c.cfg.ChanPoint, trigger,
priorState)
nextState, closeTx, err := c.stateStep(
triggerHeight, trigger,
)
if err != nil {
log.Errorf("ChannelArbitrator(%v): unable to advance "+
"state: %v", c.cfg.ChanPoint, err)
return priorState, nil, err
}
if forceCloseTx == nil && closeTx != nil {
forceCloseTx = closeTx
}
// Our termination transition is a noop transition. If we get
// our prior state back as the next state, then we'll
// terminate.
if nextState == priorState {
log.Tracef("ChannelArbitrator(%v): terminating at "+
"state=%v", c.cfg.ChanPoint, nextState)
return nextState, forceCloseTx, nil
}
// As the prior state was successfully executed, we can now
// commit the next state. This ensures that we will re-execute
// the prior state if anything fails.
if err := c.log.CommitState(nextState); err != nil {
log.Errorf("ChannelArbitrator(%v): unable to commit "+
"next state(%v): %v", c.cfg.ChanPoint,
nextState, err)
return priorState, nil, err
}
c.state = nextState
}
}
// ChainAction is an enum that encompasses all possible on-chain actions
// we'll take for a set of HTLC's.
type ChainAction uint8
const (
// NoAction is the min chainAction type, indicating that no action
// needs to be taken for a given HTLC.
NoAction ChainAction = 0
// HtlcTimeoutAction indicates that the HTLC will timeout soon. As a
// result, we should get ready to sweep it on chain after the timeout.
HtlcTimeoutAction = 1
// HtlcClaimAction indicates that we should claim the HTLC on chain
// before its timeout period.
HtlcClaimAction = 2
// HtlcFailNowAction indicates that we should fail an outgoing HTLC
// immediately by cancelling it backwards as it has no corresponding
// output in our commitment transaction.
HtlcFailNowAction = 3
// HtlcOutgoingWatchAction indicates that we can't yet timeout this
// HTLC, but we had to go to chain on order to resolve an existing
// HTLC. In this case, we'll either: time it out once it expires, or
// will learn the pre-image if the remote party claims the output. In
// this case, well add the pre-image to our global store.
HtlcOutgoingWatchAction = 4
// HtlcIncomingWatchAction indicates that we don't yet have the
// pre-image to claim incoming HTLC, but we had to go to chain in order
// to resolve and existing HTLC. In this case, we'll either: let the
// other party time it out, or eventually learn of the pre-image, in
// which case we'll claim on chain.
HtlcIncomingWatchAction = 5
)
// String returns a human readable string describing a chain action.
func (c ChainAction) String() string {
switch c {
case NoAction:
return "NoAction"
case HtlcTimeoutAction:
return "HtlcTimeoutAction"
case HtlcClaimAction:
return "HtlcClaimAction"
case HtlcFailNowAction:
return "HtlcFailNowAction"
case HtlcOutgoingWatchAction:
return "HtlcOutgoingWatchAction"
case HtlcIncomingWatchAction:
return "HtlcIncomingWatchAction"
default:
return "<unknown action>"
}
}
// ChainActionMap is a map of a chain action, to the set of HTLC's that need to
// be acted upon for a given action type. The channel
type ChainActionMap map[ChainAction][]channeldb.HTLC
// shouldGoOnChain takes into account the absolute timeout of the HTLC, if the
// confirmation delta that we need is close, and returns a bool indicating if
// we should go on chain to claim. We do this rather than waiting up until the
// last minute as we want to ensure that when we *need* (HTLC is timed out) to
// sweep, the commitment is already confirmed.
func (c *ChannelArbitrator) shouldGoOnChain(htlcExpiry, broadcastDelta,
currentHeight uint32) bool {
// We'll calculate the broadcast cut off for this HTLC. This is the
// height that (based on our current fee estimation) we should
// broadcast in order to ensure the commitment transaction is confirmed
// before the HTLC fully expires.
broadcastCutOff := htlcExpiry - broadcastDelta
log.Tracef("ChannelArbitrator(%v): examining outgoing contract: "+
"expiry=%v, cutoff=%v, height=%v", c.cfg.ChanPoint, htlcExpiry,
broadcastCutOff, currentHeight)
// TODO(roasbeef): take into account default HTLC delta, don't need to
// broadcast immediately
// * can then batch with SINGLE | ANYONECANPAY
// We should on-chain for this HTLC, iff we're within out broadcast
// cutoff window.
return currentHeight >= broadcastCutOff
}
// checkChainActions is called for each new block connected to the end of the
// main chain. Given the new block height, this new method will examine all
// active HTLC's, and determine if we need to go on-chain to claim any of them.
// A map of action -> []htlc is returned, detailing what action (if any) should
// be performed for each HTLC. For timed out HTLC's, once the commitment has
// been sufficiently confirmed, the HTLC's should be canceled backwards. For
// redeemed HTLC's, we should send the pre-image back to the incoming link.
func (c *ChannelArbitrator) checkChainActions(height uint32,
trigger transitionTrigger) ChainActionMap {
// TODO(roasbeef): would need to lock channel? channel totem?
// * race condition if adding and we broadcast, etc
// * or would make each instance sync?
log.Debugf("ChannelArbitrator(%v): checking chain actions at "+
"height=%v", c.cfg.ChanPoint, height)
actionMap := make(ChainActionMap)
// First, we'll make an initial pass over the set of incoming and
// outgoing HTLC's to decide if we need to go on chain at all.
haveChainActions := false
for _, htlc := range c.activeHTLCs.outgoingHTLCs {
// We'll need to go on-chain for an outgoing HTLC if it was
// never resolved downstream, and it's "close" to timing out.
toChain := c.shouldGoOnChain(
htlc.RefundTimeout, c.cfg.OutgoingBroadcastDelta,
height,
)
if toChain {
log.Debugf("ChannelArbitrator(%v): go to chain for "+
"outgoing htlc %x: timeout=%v, "+
"blocks_until_expiry=%v, broadcast_delta=%v",
c.cfg.ChanPoint, htlc.RHash[:],
htlc.RefundTimeout, htlc.RefundTimeout-height,
c.cfg.OutgoingBroadcastDelta,
)
}
haveChainActions = haveChainActions || toChain
}
for _, htlc := range c.activeHTLCs.incomingHTLCs {
// We'll need to go on-chain to pull an incoming HTLC iff we
// know the pre-image and it's close to timing out. We need to
// ensure that we claim the funds that our rightfully ours
// on-chain.
if _, ok := c.cfg.PreimageDB.LookupPreimage(htlc.RHash); !ok {
continue
}
toChain := c.shouldGoOnChain(
htlc.RefundTimeout, c.cfg.IncomingBroadcastDelta,
height,
)
if toChain {
log.Debugf("ChannelArbitrator(%v): go to chain for "+
"incoming htlc %x: timeout=%v, "+
"blocks_until_expiry=%v, broadcast_delta=%v",
c.cfg.ChanPoint, htlc.RHash[:],
htlc.RefundTimeout, htlc.RefundTimeout-height,
c.cfg.IncomingBroadcastDelta,
)
}
haveChainActions = haveChainActions || toChain
}
// If we don't have any actions to make, then we'll return an empty
// action map. We only do this if this was a chain trigger though, as
// if we're going to broadcast the commitment (or the remote party) did
// we're *forced* to act on each HTLC.
if !haveChainActions && trigger == chainTrigger {
log.Tracef("ChannelArbitrator(%v): no actions to take at "+
"height=%v", c.cfg.ChanPoint, height)
return actionMap
}
// Now that we know we'll need to go on-chain, we'll examine all of our
// active outgoing HTLC's to see if we either need to: sweep them after
// a timeout (then cancel backwards), cancel them backwards
// immediately, or watch them as they're still active contracts.
for _, htlc := range c.activeHTLCs.outgoingHTLCs {
switch {
// If the HTLC is dust, then we can cancel it backwards
// immediately as there's no matching contract to arbitrate
// on-chain. We know the HTLC is dust, if the OutputIndex
// negative.
case htlc.OutputIndex < 0:
log.Tracef("ChannelArbitrator(%v): immediately "+
"failing dust htlc=%x", c.cfg.ChanPoint,
htlc.RHash[:])
actionMap[HtlcFailNowAction] = append(
actionMap[HtlcFailNowAction], htlc,
)
// If we don't need to immediately act on this HTLC, then we'll
// mark it still "live". After we broadcast, we'll monitor it
// until the HTLC times out to see if we can also redeem it
// on-chain.
case !c.shouldGoOnChain(
htlc.RefundTimeout, c.cfg.OutgoingBroadcastDelta,
height,
):
// TODO(roasbeef): also need to be able to query
// circuit map to see if HTLC hasn't been fully
// resolved
//
// * can't fail incoming until if outgoing not yet
// failed
log.Tracef("ChannelArbitrator(%v): watching chain to "+
"decide action for outgoing htlc=%x",
c.cfg.ChanPoint, htlc.RHash[:])
actionMap[HtlcOutgoingWatchAction] = append(
actionMap[HtlcOutgoingWatchAction], htlc,
)
// Otherwise, we'll update our actionMap to mark that we need
// to sweep this HTLC on-chain
default:
log.Tracef("ChannelArbitrator(%v): going on-chain to "+
"timeout htlc=%x", c.cfg.ChanPoint, htlc.RHash[:])
actionMap[HtlcTimeoutAction] = append(
actionMap[HtlcTimeoutAction], htlc,
)
}
}
// Similarly, for each incoming HTLC, now that we need to go on-chain,
// we'll either: sweep it immediately if we know the pre-image, or
// observe the output on-chain if we don't In this last, case we'll
// either learn of it eventually from the outgoing HTLC, or the sender
// will timeout the HTLC.
for _, htlc := range c.activeHTLCs.incomingHTLCs {
// If we have the pre-image, then we should go on-chain to
// redeem the HTLC immediately.
if _, ok := c.cfg.PreimageDB.LookupPreimage(htlc.RHash); ok {
log.Tracef("ChannelArbitrator(%v): preimage for "+
"htlc=%x is known!", c.cfg.ChanPoint,
htlc.RHash[:])
actionMap[HtlcClaimAction] = append(
actionMap[HtlcClaimAction], htlc,
)
continue
}
log.Tracef("ChannelArbitrator(%v): watching chain to decide "+
"action for incoming htlc=%x", c.cfg.ChanPoint,
htlc.RHash[:])
// Otherwise, we don't yet have the pre-image, but should watch
// on-chain to see if either: the remote party times out the
// HTLC, or we learn of the pre-image.
actionMap[HtlcIncomingWatchAction] = append(
actionMap[HtlcIncomingWatchAction], htlc,
)
}
return actionMap
}
// prepContractResolutions is called either int he case that we decide we need
// to go to chain, or the remote party goes to chain. Given a set of actions we
// need to take for each HTLC, this method will return a set of contract
// resolvers that will resolve the contracts on-chain if needed, and also a set
// of packets to send to the htlcswitch in order to ensure all incoming HTLC's
// are properly resolved.
func (c *ChannelArbitrator) prepContractResolutions(htlcActions ChainActionMap,
contractResolutions *ContractResolutions, height uint32,
) ([]ContractResolver, []ResolutionMsg, error) {
// There may be a class of HTLC's which we can fail back immediately,
// for those we'll prepare a slice of packets to add to our outbox. Any
// packets we need to send, will be cancels.
var (
msgsToSend []ResolutionMsg
)
incomingResolutions := contractResolutions.HtlcResolutions.IncomingHTLCs
outgoingResolutions := contractResolutions.HtlcResolutions.OutgoingHTLCs
// We'll use these two maps to quickly look up an active HTLC with its
// matching HTLC resolution.
outResolutionMap := make(map[wire.OutPoint]lnwallet.OutgoingHtlcResolution)
inResolutionMap := make(map[wire.OutPoint]lnwallet.IncomingHtlcResolution)
for i := 0; i < len(incomingResolutions); i++ {
inRes := incomingResolutions[i]
inResolutionMap[inRes.HtlcPoint()] = inRes
}
for i := 0; i < len(outgoingResolutions); i++ {
outRes := outgoingResolutions[i]
outResolutionMap[outRes.HtlcPoint()] = outRes
}
// We'll create the resolver kit that we'll be cloning for each
// resolver so they each can do their duty.
resKit := ResolverKit{
ChannelArbitratorConfig: c.cfg,
Checkpoint: func(res ContractResolver) error {
return c.log.InsertUnresolvedContracts(res)
},
}
commitHash := contractResolutions.CommitHash
failureMsg := &lnwire.FailPermanentChannelFailure{}
// For each HTLC, we'll either act immediately, meaning we'll instantly
// fail the HTLC, or we'll act only once the transaction has been
// confirmed, in which case we'll need an HTLC resolver.
var htlcResolvers []ContractResolver
for htlcAction, htlcs := range htlcActions {
switch htlcAction {
// If we can fail an HTLC immediately (an outgoing HTLC with no
// contract), then we'll assemble an HTLC fail packet to send.
case HtlcFailNowAction:
for _, htlc := range htlcs {
failMsg := ResolutionMsg{
SourceChan: c.cfg.ShortChanID,
HtlcIndex: htlc.HtlcIndex,
Failure: failureMsg,
}
msgsToSend = append(msgsToSend, failMsg)
}
// If we can claim this HTLC, we'll create an HTLC resolver to
// claim the HTLC (second-level or directly), then add the pre
case HtlcClaimAction:
for _, htlc := range htlcs {
htlcOp := wire.OutPoint{
Hash: commitHash,
Index: uint32(htlc.OutputIndex),
}
resolution, ok := inResolutionMap[htlcOp]
if !ok {
// TODO(roasbeef): panic?
log.Errorf("ChannelArbitrator(%v) unable to find "+
"incoming resolution: %v",
c.cfg.ChanPoint, htlcOp)
continue
}
resKit.Quit = make(chan struct{})
resolver := &htlcSuccessResolver{
htlcResolution: resolution,
broadcastHeight: height,
payHash: htlc.RHash,
htlcAmt: htlc.Amt,
ResolverKit: resKit,
}
htlcResolvers = append(htlcResolvers, resolver)
}
// If we can timeout the HTLC directly, then we'll create the
// proper resolver to do so, who will then cancel the packet
// backwards.
case HtlcTimeoutAction:
for _, htlc := range htlcs {
htlcOp := wire.OutPoint{
Hash: commitHash,
Index: uint32(htlc.OutputIndex),
}
resolution, ok := outResolutionMap[htlcOp]
if !ok {
log.Errorf("ChannelArbitrator(%v) unable to find "+
"outgoing resolution: %v", c.cfg.ChanPoint, htlcOp)
continue
}
resKit.Quit = make(chan struct{})
resolver := &htlcTimeoutResolver{
htlcResolution: resolution,
broadcastHeight: height,
htlcIndex: htlc.HtlcIndex,
htlcAmt: htlc.Amt,
ResolverKit: resKit,
}
htlcResolvers = append(htlcResolvers, resolver)
}
// If this is an incoming HTLC, but we can't act yet, then
// we'll create an incoming resolver to redeem the HTLC if we
// learn of the pre-image, or let the remote party time out.
case HtlcIncomingWatchAction:
for _, htlc := range htlcs {
htlcOp := wire.OutPoint{
Hash: commitHash,
Index: uint32(htlc.OutputIndex),
}
// TODO(roasbeef): need to handle incoming dust...
// TODO(roasbeef): can't be negative!!!
resolution, ok := inResolutionMap[htlcOp]
if !ok {
log.Errorf("ChannelArbitrator(%v) unable to find "+
"incoming resolution: %v",
c.cfg.ChanPoint, htlcOp)
continue
}
resKit.Quit = make(chan struct{})
resolver := &htlcIncomingContestResolver{
htlcExpiry: htlc.RefundTimeout,
htlcSuccessResolver: htlcSuccessResolver{
htlcResolution: resolution,
broadcastHeight: height,
payHash: htlc.RHash,
htlcAmt: htlc.Amt,
ResolverKit: resKit,
},
}
htlcResolvers = append(htlcResolvers, resolver)
}
// Finally, if this is an outgoing HTLC we've sent, then we'll
// launch a resolver to watch for the pre-image (and settle
// backwards), or just timeout.
case HtlcOutgoingWatchAction:
for _, htlc := range htlcs {
htlcOp := wire.OutPoint{
Hash: commitHash,
Index: uint32(htlc.OutputIndex),
}
resolution, ok := outResolutionMap[htlcOp]
if !ok {
log.Errorf("ChannelArbitrator(%v) unable to find "+
"outgoing resolution: %v",
c.cfg.ChanPoint, htlcOp)
continue
}
resKit.Quit = make(chan struct{})
resolver := &htlcOutgoingContestResolver{
htlcTimeoutResolver: htlcTimeoutResolver{
htlcResolution: resolution,
broadcastHeight: height,
htlcIndex: htlc.HtlcIndex,
htlcAmt: htlc.Amt,
ResolverKit: resKit,
},
}
htlcResolvers = append(htlcResolvers, resolver)
}
}
}
// Finally, if this is was a unilateral closure, then we'll also create
// a resolver to sweep our commitment output (but only if it wasn't
// trimmed).
if contractResolutions.CommitResolution != nil {
resKit.Quit = make(chan struct{})
resolver := &commitSweepResolver{
commitResolution: *contractResolutions.CommitResolution,
broadcastHeight: height,
chanPoint: c.cfg.ChanPoint,
ResolverKit: resKit,
}
htlcResolvers = append(htlcResolvers, resolver)
}
return htlcResolvers, msgsToSend, nil
}
// replaceResolver replaces a in the list of active resolvers. If the resolver
// to be replaced is not found, it returns an error.
func (c *ChannelArbitrator) replaceResolver(oldResolver,
newResolver ContractResolver) error {
c.activeResolversLock.Lock()
defer c.activeResolversLock.Unlock()
oldKey := oldResolver.ResolverKey()
for i, r := range c.activeResolvers {
if bytes.Equal(r.ResolverKey(), oldKey) {
c.activeResolvers[i] = newResolver
return nil
}
}
return errors.New("resolver to be replaced not found")
}
// resolveContract is a goroutine tasked with fully resolving an unresolved
// contract. Either the initial contract will be resolved after a single step,
// or the contract will itself create another contract to be resolved. In
// either case, one the contract has been fully resolved, we'll signal back to
// the main goroutine so it can properly keep track of the set of unresolved
// contracts.
//
// NOTE: This MUST be run as a goroutine.
func (c *ChannelArbitrator) resolveContract(currentContract ContractResolver) {
defer c.wg.Done()
log.Debugf("ChannelArbitrator(%v): attempting to resolve %T",
c.cfg.ChanPoint, currentContract)
// Until the contract is fully resolved, we'll continue to iteratively
// resolve the contract one step at a time.
for !currentContract.IsResolved() {
log.Debugf("ChannelArbitrator(%v): contract %T not yet resolved",
c.cfg.ChanPoint, currentContract)
select {
// If we've been signalled to quit, then we'll exit early.
case <-c.quit:
return
default:
// Otherwise, we'll attempt to resolve the current
// contract.
nextContract, err := currentContract.Resolve()
if err != nil {
log.Errorf("ChannelArbitrator(%v): unable to "+
"progress resolver: %v",
c.cfg.ChanPoint, err)
return
}
switch {
// If this contract produced another, then this means
// the current contract was only able to be partially
// resolved in this step. So we'll not a contract swap
// within our logs: the new contract will take the
// place of the old one.
case nextContract != nil:
log.Debugf("ChannelArbitrator(%v): swapping "+
"out contract %T for %T ",
c.cfg.ChanPoint, currentContract,
nextContract)
// Swap contract in log.
err := c.log.SwapContract(
currentContract, nextContract,
)
if err != nil {
log.Errorf("unable to add recurse "+
"contract: %v", err)
}
// Swap contract in resolvers list. This is to
// make sure that reports are queried from the
// new resolver.
err = c.replaceResolver(
currentContract, nextContract,
)
if err != nil {
log.Errorf("unable to replace "+
"contract: %v", err)
}
// As this contract produced another, we'll
// re-assign, so we can continue our resolution
// loop.
currentContract = nextContract
// If this contract is actually fully resolved, then
// we'll mark it as such within the database.
case currentContract.IsResolved():
log.Debugf("ChannelArbitrator(%v): marking "+
"contract %T fully resolved",
c.cfg.ChanPoint, currentContract)
err := c.log.ResolveContract(currentContract)
if err != nil {
log.Errorf("unable to resolve contract: %v",
err)
}
// Now that the contract has been resolved,
// well signal to the main goroutine.
select {
case c.resolutionSignal <- struct{}{}:
case <-c.quit:
return
}
}
}
}
}
// signalUpdateMsg is a struct that carries fresh signals to the
// ChannelArbitrator. We need to receive a message like this each time the
// channel becomes active, as it's internal state may change.
type signalUpdateMsg struct {
// newSignals is the set of new active signals to be sent to the
// arbitrator.
newSignals *ContractSignals
// doneChan is a channel that will be closed on the arbitrator has
// attached the new signals.
doneChan chan struct{}
}
// UpdateContractSignals updates the set of signals the ChannelArbitrator needs
// to receive from a channel in real-time in order to keep in sync with the
// latest state of the contract.
func (c *ChannelArbitrator) UpdateContractSignals(newSignals *ContractSignals) {
done := make(chan struct{})
select {
case c.signalUpdates <- &signalUpdateMsg{
newSignals: newSignals,
doneChan: done,
}:
case <-c.quit:
}
select {
case <-done:
case <-c.quit:
}
}
// channelAttendant is the primary goroutine that acts at the judicial
// arbitrator between our channel state, the remote channel peer, and the
// blockchain Our judge). This goroutine will ensure that we faithfully execute
// all clauses of our contract in the case that we need to go on-chain for a
// dispute. Currently, two such conditions warrant our intervention: when an
// outgoing HTLC is about to timeout, and when we know the pre-image for an
// incoming HTLC, but it hasn't yet been settled off-chain. In these cases,
// we'll: broadcast our commitment, cancel/settle any HTLC's backwards after
// sufficient confirmation, and finally send our set of outputs to the UTXO
// Nursery for incubation, and ultimate sweeping.
//
// NOTE: This MUST be run as a goroutine.
func (c *ChannelArbitrator) channelAttendant(bestHeight int32) {
// TODO(roasbeef): tell top chain arb we're done
defer func() {
c.cfg.BlockEpochs.Cancel()
c.wg.Done()
}()
for {
select {
// A new block has arrived, we'll examine all the active HTLC's
// to see if any of them have expired, and also update our
// track of the best current height.
case blockEpoch, ok := <-c.cfg.BlockEpochs.Epochs:
if !ok {
return
}
bestHeight = blockEpoch.Height
// If we're not in the default state, then we can
// ignore this signal as we're waiting for contract
// resolution.
if c.state != StateDefault {
continue
}
// Now that a new block has arrived, we'll attempt to
// advance our state forward.
nextState, _, err := c.advanceState(
uint32(bestHeight), chainTrigger,
)
if err != nil {
log.Errorf("unable to advance state: %v", err)
}
// If as a result of this trigger, the contract is
// fully resolved, then well exit.
if nextState == StateFullyResolved {
return
}
// A new signal update was just sent. This indicates that the
// channel under watch is now live, and may modify its internal
// state, so we'll get the most up to date signals to we can
// properly do our job.
case signalUpdate := <-c.signalUpdates:
log.Tracef("ChannelArbitrator(%v) got new signal "+
"update!", c.cfg.ChanPoint)
// First, we'll update our set of signals.
c.htlcUpdates = signalUpdate.newSignals.HtlcUpdates
c.cfg.ShortChanID = signalUpdate.newSignals.ShortChanID
// Now that the signals have been updated, we'll now
// close the done channel to signal to the caller we've
// registered the new contracts.
close(signalUpdate.doneChan)
// A new set of HTLC's has been added or removed from the
// commitment transaction. So we'll update our activeHTLCs map
// accordingly.
case newStateHTLCs := <-c.htlcUpdates:
// We'll wipe out our old set of HTLC's and instead
// monitor only the HTLC's that are still active on the
// current commitment state.
c.activeHTLCs = newHtlcSet(newStateHTLCs)
log.Tracef("ChannelArbitrator(%v): fresh set of "+
"htlcs=%v", c.cfg.ChanPoint,
newLogClosure(func() string {
return spew.Sdump(c.activeHTLCs)
}),
)
// We've cooperatively closed the channel, so we're no longer
// needed. We'll mark the channel as resolved and exit.
case closeInfo := <-c.cfg.ChainEvents.CooperativeClosure:
log.Infof("ChannelArbitrator(%v) marking channel "+
"cooperatively closed", c.cfg.ChanPoint)
err := c.cfg.MarkChannelClosed(
closeInfo.ChannelCloseSummary,
)
if err != nil {
log.Errorf("unable to mark channel closed: "+
"%v", err)
return
}
// We'll now advance our state machine until it reaches
// a terminal state, and the channel is marked resolved.
_, _, err = c.advanceState(
closeInfo.CloseHeight, coopCloseTrigger,
)
if err != nil {
log.Errorf("unable to advance state: %v", err)
return
}
// We have broadcasted our commitment, and it is now confirmed
// on-chain.
case closeInfo := <-c.cfg.ChainEvents.LocalUnilateralClosure:
log.Infof("ChannelArbitrator(%v): local on-chain "+
"channel close", c.cfg.ChanPoint)
if c.state != StateCommitmentBroadcasted {
log.Errorf("ChannelArbitrator(%v): unexpected "+
"local on-chain channel close",
c.cfg.ChanPoint)
}
closeTx := closeInfo.CloseTx
contractRes := &ContractResolutions{
CommitHash: closeTx.TxHash(),
CommitResolution: closeInfo.CommitResolution,
HtlcResolutions: *closeInfo.HtlcResolutions,
}
// When processing a unilateral close event, we'll
// transition to the ContractClosed state. We'll log
// out the set of resolutions such that they are
// available to fetch in that state.
err := c.log.LogContractResolutions(contractRes)
if err != nil {
log.Errorf("unable to write resolutions: %v",
err)
return
}
// After the set of resolutions are successfully
// logged, we can safely close the channel. After this
// succeeds we won't be getting chain events anymore,
// so we must make sure we can recover on restart after
// it is marked closed. If the next state transition
// fails, we'll start up in the prior state again, and
// we won't be longer getting chain events. In this
// case we must manually re-trigger the state
// transition into StateContractClosed based on the
// close status of the channel.
err = c.cfg.MarkChannelClosed(
closeInfo.ChannelCloseSummary,
)
if err != nil {
log.Errorf("unable to mark "+
"channel closed: %v", err)
return
}
// We'll now advance our state machine until it reaches
// a terminal state.
_, _, err = c.advanceState(
uint32(closeInfo.SpendingHeight),
localCloseTrigger,
)
if err != nil {
log.Errorf("unable to advance state: %v", err)
}
// The remote party has broadcast the commitment on-chain.
// We'll examine our state to determine if we need to act at
// all.
case uniClosure := <-c.cfg.ChainEvents.RemoteUnilateralClosure:
log.Infof("ChannelArbitrator(%v): remote party has "+
"closed channel out on-chain", c.cfg.ChanPoint)
// If we don't have a self output, and there are no
// active HTLC's, then we can immediately mark the
// contract as fully resolved and exit.
contractRes := &ContractResolutions{
CommitHash: *uniClosure.SpenderTxHash,
CommitResolution: uniClosure.CommitResolution,
HtlcResolutions: *uniClosure.HtlcResolutions,
}
// As we're now acting upon an event triggered by the
// broadcast of the remote commitment transaction,
// we'll swap out our active HTLC set with the set
// present on their commitment.
c.activeHTLCs = newHtlcSet(uniClosure.RemoteCommit.Htlcs)
// When processing a unilateral close event, we'll
// transition to the ContractClosed state. We'll log
// out the set of resolutions such that they are
// available to fetch in that state.
err := c.log.LogContractResolutions(contractRes)
if err != nil {
log.Errorf("unable to write resolutions: %v",
err)
return
}
// After the set of resolutions are successfully
// logged, we can safely close the channel. After this
// succeeds we won't be getting chain events anymore,
// so we must make sure we can recover on restart after
// it is marked closed. If the next state transition
// fails, we'll start up in the prior state again, and
// we won't be longer getting chain events. In this
// case we must manually re-trigger the state
// transition into StateContractClosed based on the
// close status of the channel.
closeSummary := &uniClosure.ChannelCloseSummary
err = c.cfg.MarkChannelClosed(closeSummary)
if err != nil {
log.Errorf("unable to mark channel closed: %v",
err)
return
}
// We'll now advance our state machine until it reaches
// a terminal state.
_, _, err = c.advanceState(
uint32(uniClosure.SpendingHeight),
remoteCloseTrigger,
)
if err != nil {
log.Errorf("unable to advance state: %v", err)
}
// A new contract has just been resolved, we'll now check our
// log to see if all contracts have been resolved. If so, then
// we can exit as the contract is fully resolved.
case <-c.resolutionSignal:
log.Infof("ChannelArbitrator(%v): a contract has been "+
"fully resolved!", c.cfg.ChanPoint)
nextState, _, err := c.advanceState(
uint32(bestHeight), chainTrigger,
)
if err != nil {
log.Errorf("unable to advance state: %v", err)
}
// If we don't have anything further to do after
// advancing our state, then we'll exit.
if nextState == StateFullyResolved {
log.Infof("ChannelArbitrator(%v): all "+
"contracts fully resolved, exiting",
c.cfg.ChanPoint)
return
}
// We've just received a request to forcibly close out the
// channel. We'll
case closeReq := <-c.forceCloseReqs:
if c.state != StateDefault {
select {
case closeReq.closeTx <- nil:
case <-c.quit:
}
select {
case closeReq.errResp <- errAlreadyForceClosed:
case <-c.quit:
}
continue
}
nextState, closeTx, err := c.advanceState(
uint32(bestHeight), userTrigger,
)
if err != nil {
log.Errorf("unable to advance state: %v", err)
}
select {
case closeReq.closeTx <- closeTx:
case <-c.quit:
return
}
select {
case closeReq.errResp <- err:
case <-c.quit:
return
}
// If we don't have anything further to do after
// advancing our state, then we'll exit.
if nextState == StateFullyResolved {
log.Infof("ChannelArbitrator(%v): all "+
"contracts resolved, exiting",
c.cfg.ChanPoint)
return
}
case <-c.quit:
return
}
}
}