Merge pull request #761 from cfromknecht/switch-persistence

Switch Persistence [ALL]: Forwarding Packages + Sphinx Replay Protection + Circuit Persistence

breacharbiter_test.go

@@ -1401,6 +1401,7 @@ func createInitChannels(revocationWindow int) (*lnwallet.LightningChannel, *lnwa
 		RemoteChanCfg:           bobCfg,
 		IdentityPub:             aliceKeyPub,
 		FundingOutpoint:         *prevOut,
+		ShortChanID:             shortChanID,
 		ChanType:                channeldb.SingleFunder,
 		IsInitiator:             true,
 		Capacity:                channelCapacity,
@@ -1417,6 +1418,7 @@ func createInitChannels(revocationWindow int) (*lnwallet.LightningChannel, *lnwa
 		RemoteChanCfg:           aliceCfg,
 		IdentityPub:             bobKeyPub,
 		FundingOutpoint:         *prevOut,
+		ShortChanID:             shortChanID,
 		ChanType:                channeldb.SingleFunder,
 		IsInitiator:             false,
 		Capacity:                channelCapacity,

config.go

@@ -170,6 +170,8 @@ type config struct {
 
 	DebugHTLC          bool `long:"debughtlc" description:"Activate the debug htlc mode. With the debug HTLC mode, all payments sent use a pre-determined R-Hash. Additionally, all HTLCs sent to a node with the debug HTLC R-Hash are immediately settled in the next available state transition."`
 	HodlHTLC           bool `long:"hodlhtlc" description:"Activate the hodl HTLC mode.  With hodl HTLC mode, all incoming HTLCs will be accepted by the receiving node, but no attempt will be made to settle the payment with the sender."`
+	UnsafeDisconnect   bool `long:"unsafe-disconnect" description:"Allows the rpcserver to intentionally disconnect from peers with open channels. USED FOR TESTING ONLY."`
+	UnsafeReplay       bool `long:"unsafe-replay" description:"Causes a link to replay the adds on its commitment txn after starting up, this enables testing of the sphinx replay logic."`
 	MaxPendingChannels int  `long:"maxpendingchannels" description:"The maximum number of incoming pending channels permitted per peer."`
 
 	Bitcoin      *chainConfig    `group:"Bitcoin" namespace:"bitcoin"`

htlcswitch/circuit.go

@@ -1,184 +1,229 @@
 package htlcswitch
 
 import (
-	"fmt"
-	"sync"
+	"encoding/binary"
+	"io"
 
-	"github.com/go-errors/errors"
+	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/lnwire"
 )
 
-// PaymentCircuit is used by the HTLC switch subsystem to determine the
-// backwards path for the settle/fail HTLC messages. A payment circuit
-// will be created once a channel link forwards the HTLC add request and
-// removed when we receive a settle/fail HTLC message.
+// EmptyCircuitKey is a default value for an outgoing circuit key returned when
+// a circuit's keystone has not been set. Note that this value is invalid for
+// use as a keystone, since the outgoing channel id can never be equal to
+// sourceHop.
+var EmptyCircuitKey CircuitKey
+
+// CircuitKey is a tuple of channel ID and HTLC ID, used to uniquely identify
+// HTLCs in a circuit. Circuits are identified primarily by the circuit key of
+// the incoming HTLC. However, a circuit may also be referenced by its outgoing
+// circuit key after the HTLC has been forwarded via the outgoing link.
+type CircuitKey = channeldb.CircuitKey
+
+// PaymentCircuit is used by the switch as placeholder between when the
+// switch makes a forwarding decision and the outgoing link determines the
+// proper HTLC ID for the local log. After the outgoing HTLC ID has been
+// determined, the half circuit will be converted into a full PaymentCircuit.
 type PaymentCircuit struct {
+	// AddRef is the forward reference of the Add update in the incoming
+	// link's forwarding package. This value is set on the htlcPacket of the
+	// returned settle/fail so that it can be removed from disk.
+	AddRef channeldb.AddRef
+
+	// Incoming is the circuit key identifying the incoming channel and htlc
+	// index from which this ADD originates.
+	Incoming CircuitKey
+
+	// Outgoing is the circuit key identifying the outgoing channel, and the
+	// HTLC index that was used to forward the ADD. It will be nil if this
+	// circuit's keystone has not been set.
+	Outgoing *CircuitKey
+
 	// PaymentHash used as unique identifier of payment.
 	PaymentHash [32]byte
 
-	// IncomingChanID identifies the channel from which add HTLC request
-	// came and to which settle/fail HTLC request will be returned back.
-	// Once the switch forwards the settle/fail message to the src the
-	// circuit is considered to be completed.
-	IncomingChanID lnwire.ShortChannelID
+	// IncomingAmount is the value of the HTLC from the incoming link.
+	IncomingAmount lnwire.MilliSatoshi
 
-	// IncomingHTLCID is the ID in the update_add_htlc message we received
-	// from the incoming channel, which will be included in any settle/fail
-	// messages we send back.
-	IncomingHTLCID uint64
-
-	// IncomingAmt is the value of the incoming HTLC. If we take this and
-	// subtract it from the OutgoingAmt, then we'll compute the total fee
-	// attached to this payment circuit.
-	IncomingAmt lnwire.MilliSatoshi
-
-	// OutgoingChanID identifies the channel to which we propagate the HTLC
-	// add update and from which we are expecting to receive HTLC
-	// settle/fail request back.
-	OutgoingChanID lnwire.ShortChannelID
-
-	// OutgoingHTLCID is the ID in the update_add_htlc message we sent to
-	// the outgoing channel.
-	OutgoingHTLCID uint64
-
-	// OutgoingAmt is the value of the outgoing HTLC. If we subtract this
-	// from the IncomingAmt, then we'll compute the total fee attached to
-	// this payment circuit.
-	OutgoingAmt lnwire.MilliSatoshi
+	// OutgoingAmount specifies the value of the HTLC leaving the switch,
+	// either as a payment or forwarded amount.
+	OutgoingAmount lnwire.MilliSatoshi
 
 	// ErrorEncrypter is used to re-encrypt the onion failure before
 	// sending it back to the originator of the payment.
 	ErrorEncrypter ErrorEncrypter
+
+	// LoadedFromDisk is set true for any circuits loaded after the circuit
+	// map is reloaded from disk.
+	//
+	// NOTE: This value is determined implicitly during a restart. It is not
+	// persisted, and should never be set outside the circuit map.
+	LoadedFromDisk bool
 }
 
-// circuitKey is a channel ID, HTLC ID tuple used as an identifying key for a
-// payment circuit. The circuit map is keyed with the identifier for the
-// outgoing HTLC
-type circuitKey struct {
-	chanID lnwire.ShortChannelID
-	htlcID uint64
+// HasKeystone returns true if an outgoing link has assigned this circuit's
+// outgoing circuit key.
+func (c *PaymentCircuit) HasKeystone() bool {
+	return c.Outgoing != nil
 }
 
-// String returns a string representation of the circuitKey.
-func (k *circuitKey) String() string {
-	return fmt.Sprintf("(Chan ID=%s, HTLC ID=%d)", k.chanID, k.htlcID)
-}
+// newPaymentCircuit initializes a payment circuit on the heap using the payment
+// hash and an in-memory htlc packet.
+func newPaymentCircuit(hash *[32]byte, pkt *htlcPacket) *PaymentCircuit {
+	var addRef channeldb.AddRef
+	if pkt.sourceRef != nil {
+		addRef = *pkt.sourceRef
+	}
 
-// CircuitMap is a data structure that implements thread safe storage of
-// circuit routing information. The switch consults a circuit map to determine
-// where to forward HTLC update messages. Each circuit is stored with its
-// outgoing HTLC as the primary key because, each offered HTLC has at most one
-// received HTLC, but there may be multiple offered or received HTLCs with the
-// same payment hash. Circuits are also indexed to provide fast lookups by
-// payment hash.
-//
-// TODO(andrew.shvv) make it persistent
-type CircuitMap struct {
-	mtx       sync.RWMutex
-	circuits  map[circuitKey]*PaymentCircuit
-	hashIndex map[[32]byte]map[PaymentCircuit]struct{}
-}
-
-// NewCircuitMap creates a new instance of the CircuitMap.
-func NewCircuitMap() *CircuitMap {
-	return &CircuitMap{
-		circuits:  make(map[circuitKey]*PaymentCircuit),
-		hashIndex: make(map[[32]byte]map[PaymentCircuit]struct{}),
+	return &PaymentCircuit{
+		AddRef: addRef,
+		Incoming: CircuitKey{
+			ChanID: pkt.incomingChanID,
+			HtlcID: pkt.incomingHTLCID,
+		},
+		PaymentHash:    *hash,
+		IncomingAmount: pkt.incomingAmount,
+		OutgoingAmount: pkt.amount,
+		ErrorEncrypter: pkt.obfuscator,
 	}
 }
 
-// LookupByHTLC looks up the payment circuit by the outgoing channel and HTLC
-// IDs. Returns nil if there is no such circuit.
-func (cm *CircuitMap) LookupByHTLC(chanID lnwire.ShortChannelID, htlcID uint64) *PaymentCircuit {
-	cm.mtx.RLock()
-
-	key := circuitKey{
-		chanID: chanID,
-		htlcID: htlcID,
+// makePaymentCircuit initalizes a payment circuit on the stack using the
+// payment hash and an in-memory htlc packet.
+func makePaymentCircuit(hash *[32]byte, pkt *htlcPacket) PaymentCircuit {
+	var addRef channeldb.AddRef
+	if pkt.sourceRef != nil {
+		addRef = *pkt.sourceRef
 	}
-	circuit := cm.circuits[key]
 
-	cm.mtx.RUnlock()
-	return circuit
+	return PaymentCircuit{
+		AddRef: addRef,
+		Incoming: CircuitKey{
+			ChanID: pkt.incomingChanID,
+			HtlcID: pkt.incomingHTLCID,
+		},
+		PaymentHash:    *hash,
+		IncomingAmount: pkt.incomingAmount,
+		OutgoingAmount: pkt.amount,
+		ErrorEncrypter: pkt.obfuscator,
+	}
 }
 
-// LookupByPaymentHash looks up and returns any payment circuits with a given
-// payment hash.
-func (cm *CircuitMap) LookupByPaymentHash(hash [32]byte) []*PaymentCircuit {
-	cm.mtx.RLock()
-
-	var circuits []*PaymentCircuit
-	if circuitSet, ok := cm.hashIndex[hash]; ok {
-		circuits = make([]*PaymentCircuit, 0, len(circuitSet))
-		for circuit := range circuitSet {
-			circuits = append(circuits, &circuit)
-		}
+// Encode writes a PaymentCircuit to the provided io.Writer.
+func (c *PaymentCircuit) Encode(w io.Writer) error {
+	if err := c.AddRef.Encode(w); err != nil {
+		return err
 	}
 
-	cm.mtx.RUnlock()
-	return circuits
+	if err := c.Incoming.Encode(w); err != nil {
+		return err
+	}
+
+	if _, err := w.Write(c.PaymentHash[:]); err != nil {
+		return err
+	}
+
+	var scratch [8]byte
+
+	binary.BigEndian.PutUint64(scratch[:], uint64(c.IncomingAmount))
+	if _, err := w.Write(scratch[:]); err != nil {
+		return err
+	}
+
+	binary.BigEndian.PutUint64(scratch[:], uint64(c.OutgoingAmount))
+	if _, err := w.Write(scratch[:]); err != nil {
+		return err
+	}
+
+	// Defaults to EncrypterTypeNone.
+	var encrypterType EncrypterType
+	if c.ErrorEncrypter != nil {
+		encrypterType = c.ErrorEncrypter.Type()
+	}
+
+	err := binary.Write(w, binary.BigEndian, encrypterType)
+	if err != nil {
+		return err
+	}
+
+	// Skip encoding of error encrypter if this half add does not have one.
+	if encrypterType == EncrypterTypeNone {
+		return nil
+	}
+
+	return c.ErrorEncrypter.Encode(w)
 }
 
-// Add adds a new active payment circuit to the CircuitMap.
-func (cm *CircuitMap) Add(circuit *PaymentCircuit) error {
-	cm.mtx.Lock()
-
-	key := circuitKey{
-		chanID: circuit.OutgoingChanID,
-		htlcID: circuit.OutgoingHTLCID,
+// Decode reads a PaymentCircuit from the provided io.Reader.
+func (c *PaymentCircuit) Decode(r io.Reader) error {
+	if err := c.AddRef.Decode(r); err != nil {
+		return err
 	}
-	cm.circuits[key] = circuit
 
-	// Add circuit to the hash index.
-	if _, ok := cm.hashIndex[circuit.PaymentHash]; !ok {
-		cm.hashIndex[circuit.PaymentHash] = make(map[PaymentCircuit]struct{})
+	if err := c.Incoming.Decode(r); err != nil {
+		return err
 	}
-	cm.hashIndex[circuit.PaymentHash][*circuit] = struct{}{}
 
-	cm.mtx.Unlock()
-	return nil
+	if _, err := io.ReadFull(r, c.PaymentHash[:]); err != nil {
+		return err
+	}
+
+	var scratch [8]byte
+
+	if _, err := io.ReadFull(r, scratch[:]); err != nil {
+		return err
+	}
+	c.IncomingAmount = lnwire.MilliSatoshi(
+		binary.BigEndian.Uint64(scratch[:]))
+
+	if _, err := io.ReadFull(r, scratch[:]); err != nil {
+		return err
+	}
+	c.OutgoingAmount = lnwire.MilliSatoshi(
+		binary.BigEndian.Uint64(scratch[:]))
+
+	// Read the encrypter type used for this circuit.
+	var encrypterType EncrypterType
+	err := binary.Read(r, binary.BigEndian, &encrypterType)
+	if err != nil {
+		return err
+	}
+
+	switch encrypterType {
+	case EncrypterTypeNone:
+		// No encrypter was provided, such as when the payment is
+		// locally initiated.
+		return nil
+
+	case EncrypterTypeSphinx:
+		// Sphinx encrypter was used as this is a forwarded HTLC.
+		c.ErrorEncrypter = NewSphinxErrorEncrypter()
+
+	case EncrypterTypeMock:
+		// Test encrypter.
+		c.ErrorEncrypter = NewMockObfuscator()
+
+	default:
+		return UnknownEncrypterType(encrypterType)
+	}
+
+	return c.ErrorEncrypter.Decode(r)
 }
 
-// Remove destroys the target circuit by removing it from the circuit map.
-func (cm *CircuitMap) Remove(chanID lnwire.ShortChannelID, htlcID uint64) error {
-	cm.mtx.Lock()
-	defer cm.mtx.Unlock()
-
-	// Look up circuit so that pointer can be matched in the hash index.
-	key := circuitKey{
-		chanID: chanID,
-		htlcID: htlcID,
-	}
-	circuit, found := cm.circuits[key]
-	if !found {
-		return errors.Errorf("Can't find circuit for HTLC %v", key)
-	}
-	delete(cm.circuits, key)
-
-	// Remove circuit from hash index.
-	circuitsWithHash, ok := cm.hashIndex[circuit.PaymentHash]
-	if !ok {
-		return errors.Errorf("Can't find circuit in hash index for HTLC %v",
-			key)
-	}
-
-	if _, ok = circuitsWithHash[*circuit]; !ok {
-		return errors.Errorf("Can't find circuit in hash index for HTLC %v",
-			key)
-	}
-
-	delete(circuitsWithHash, *circuit)
-	if len(circuitsWithHash) == 0 {
-		delete(cm.hashIndex, circuit.PaymentHash)
-	}
-	return nil
+// InKey returns the primary identifier for the circuit corresponding to the
+// incoming HTLC.
+func (c *PaymentCircuit) InKey() CircuitKey {
+	return c.Incoming
 }
 
-// pending returns number of circuits which are waiting for to be completed
-// (settle/fail responses to be received).
-func (cm *CircuitMap) pending() int {
-	cm.mtx.RLock()
-	count := len(cm.circuits)
-	cm.mtx.RUnlock()
-	return count
+// OutKey returns the keystone identifying the outgoing link and HTLC ID. If the
+// circuit hasn't been completed, this method returns an EmptyKeystone, which is
+// an invalid outgoing circuit key. Only call this method if HasKeystone returns
+// true.
+func (c *PaymentCircuit) OutKey() CircuitKey {
+	if c.Outgoing != nil {
+		return *c.Outgoing
+	}
+
+	return EmptyCircuitKey
 }

htlcswitch/circuit_map.go

@@ -0,0 +1,846 @@
+package htlcswitch
+
+import (
+	"bytes"
+	"fmt"
+	"sync"
+
+	"github.com/boltdb/bolt"
+	"github.com/go-errors/errors"
+	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/lnwire"
+)
+
+var (
+	// ErrCorruptedCircuitMap indicates that the on-disk bucketing structure
+	// has altered since the circuit map instance was initialized.
+	ErrCorruptedCircuitMap = errors.New("circuit map has been corrupted")
+
+	// ErrCircuitNotInHashIndex indicates that a particular circuit did not
+	// appear in the in-memory hash index.
+	ErrCircuitNotInHashIndex = errors.New("payment circuit not found in " +
+		"hash index")
+
+	// ErrUnknownCircuit signals that circuit could not be removed from the
+	// map because it was not found.
+	ErrUnknownCircuit = errors.New("unknown payment circuit")
+
+	// ErrCircuitClosing signals that an htlc has already closed this
+	// circuit in-memory.
+	ErrCircuitClosing = errors.New("circuit has already been closed")
+
+	// ErrDuplicateCircuit signals that this circuit was previously
+	// added.
+	ErrDuplicateCircuit = errors.New("duplicate circuit add")
+
+	// ErrUnknownKeystone signals that no circuit was found using the
+	// outgoing circuit key.
+	ErrUnknownKeystone = errors.New("unknown circuit keystone")
+
+	// ErrDuplicateKeystone signals that this circuit was previously
+	// assigned a keystone.
+	ErrDuplicateKeystone = errors.New("cannot add duplicate keystone")
+)
+
+// CircuitModifier is a common interface used by channel links to modify the
+// contents of the circuit map maintained by the switch.
+type CircuitModifier interface {
+	// OpenCircuits preemptively records a batch keystones that will mark
+	// currently pending circuits as open. These changes can be rolled back
+	// on restart if the outgoing Adds do not make it into a commitment txn.
+	OpenCircuits(...Keystone) error
+
+	// TrimOpenCircuits removes a channel's open channels with htlc indexes
+	// above `start`.
+	TrimOpenCircuits(chanID lnwire.ShortChannelID, start uint64) error
+
+	// DeleteCircuits removes the incoming circuit key to remove all
+	// persistent references to a circuit. Returns a ErrUnknownCircuit if
+	// any of the incoming keys are not known.
+	DeleteCircuits(inKeys ...CircuitKey) error
+}
+
+// CircuitFwdActions represents the forwarding decision made by the circuit map,
+// and is returned from CommitCircuits. The sequence of circuits provided to
+// CommitCircuits is split into three subsequences, allowing the caller to do an
+// in-order scan, comparing the head of each subsequence, to determine the
+// decision made by the circuit map.
+type CircuitFwdActions struct {
+	// Adds is the subsequence of circuits that were successfully committed
+	// in the circuit map.
+	Adds []*PaymentCircuit
+
+	// Drops is the subsequence of circuits for which no action should be
+	// done.
+	Drops []*PaymentCircuit
+
+	// Fails is the subsequence of circuits that should be failed back by
+	// the calling link.
+	Fails []*PaymentCircuit
+}
+
+// CircuitMap is an interface for managing the construction and teardown of
+// payment circuits used by the switch.
+type CircuitMap interface {
+	CircuitModifier
+
+	// CommitCircuits attempts to add the given circuits to the circuit
+	// map. The list of circuits is split into three distinct subsequences,
+	// corresponding to adds, drops, and fails. Adds should be forwarded to
+	// the switch, while fails should be failed back locally within the
+	// calling link.
+	CommitCircuits(circuit ...*PaymentCircuit) (*CircuitFwdActions, error)
+
+	// CloseCircuit marks the circuit identified by `outKey` as closing
+	// in-memory, which prevents duplicate settles/fails from completing an
+	// open circuit twice.
+	CloseCircuit(outKey CircuitKey) (*PaymentCircuit, error)
+
+	// FailCircuit is used by locally failed HTLCs to mark the circuit
+	// identified by `inKey` as closing in-memory, which prevents duplicate
+	// settles/fails from being accepted for the same circuit.
+	FailCircuit(inKey CircuitKey) (*PaymentCircuit, error)
+
+	// LookupCircuit queries the circuit map for the circuit identified by
+	// inKey.
+	LookupCircuit(inKey CircuitKey) *PaymentCircuit
+
+	// LookupOpenCircuit queries the circuit map for a circuit identified by
+	// its outgoing circuit key.
+	LookupOpenCircuit(outKey CircuitKey) *PaymentCircuit
+
+	// LookupByPaymentHash queries the circuit map and returns all open
+	// circuits that use the given payment hash.
+	LookupByPaymentHash(hash [32]byte) []*PaymentCircuit
+
+	// NumPending returns the total number of active circuits added by
+	// CommitCircuits.
+	NumPending() int
+
+	// NumOpen returns the number of circuits with HTLCs that have been
+	// forwarded via an outgoing link.
+	NumOpen() int
+}
+
+var (
+	// circuitAddKey is the key used to retrieve the bucket containing
+	// payment circuits. A circuit records information about how to return a
+	// packet to the source link, potentially including an error encrypter
+	// for applying this hop's encryption to the payload in the reverse
+	// direction.
+	circuitAddKey = []byte("circuit-adds")
+
+	// circuitKeystoneKey is used to retrieve the bucket containing circuit
+	// keystones, which are set in place once a forwarded packet is assigned
+	// an index on an outgoing commitment txn.
+	circuitKeystoneKey = []byte("circuit-keystones")
+)
+
+// circuitMap is a data structure that implements thread safe, persistent
+// storage of circuit routing information. The switch consults a circuit map to
+// determine where to forward returning HTLC update messages. Circuits are
+// always identifiable by their incoming CircuitKey, in addition to their
+// outgoing CircuitKey if the circuit is fully-opened.
+type circuitMap struct {
+	// db provides the persistent storage engine for the circuit map.
+	// TODO(conner): create abstraction to allow for the substitution of
+	// other persistence engines.
+	db *channeldb.DB
+
+	mtx sync.RWMutex
+
+	// pending is an in-memory mapping of all half payment circuits, and
+	// is kept in sync with the on-disk contents of the circuit map.
+	pending map[CircuitKey]*PaymentCircuit
+
+	// opened is an in-memory mapping of all full payment circuits, which is
+	// also synchronized with the persistent state of the circuit map.
+	opened map[CircuitKey]*PaymentCircuit
+
+	// closed is an in-memory set of circuits for which the switch has
+	// received a settle or fail. This precedes the actual deletion of a
+	// circuit from disk.
+	closed map[CircuitKey]struct{}
+
+	// hashIndex is a volatile index that facilitates fast queries by
+	// payment hash against the contents of circuits. This index can be
+	// reconstructed entirely from the set of persisted full circuits on
+	// startup.
+	hashIndex map[[32]byte]map[CircuitKey]struct{}
+}
+
+// NewCircuitMap creates a new instance of the circuitMap.
+func NewCircuitMap(db *channeldb.DB) (CircuitMap, error) {
+	cm := &circuitMap{
+		db: db,
+	}
+
+	// Initialize the on-disk buckets used by the circuit map.
+	if err := cm.initBuckets(); err != nil {
+		return nil, err
+	}
+
+	// Load any previously persisted circuit into back into memory.
+	if err := cm.restoreMemState(); err != nil {
+		return nil, err
+	}
+
+	// Trim any keystones that were not committed in an outgoing commit txn.
+	//
+	// NOTE: This operation will be applied to the persistent state of all
+	// active channels. Therefore, it must be called before any links are
+	// created to avoid interfering with normal operation.
+	if err := cm.trimAllOpenCircuits(); err != nil {
+		return nil, err
+	}
+
+	return cm, nil
+}
+
+// initBuckets ensures that the primary buckets used by the circuit are
+// initialized so that we can assume their existence after startup.
+func (cm *circuitMap) initBuckets() error {
+	return cm.db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists(circuitKeystoneKey)
+		if err != nil {
+			return err
+		}
+
+		_, err = tx.CreateBucketIfNotExists(circuitAddKey)
+		return err
+	})
+}
+
+// restoreMemState loads the contents of the half circuit and full circuit buckets
+// from disk and reconstructs the in-memory representation of the circuit map.
+// Afterwards, the state of the hash index is reconstructed using the recovered
+// set of full circuits.
+func (cm *circuitMap) restoreMemState() error {
+
+	var (
+		opened  = make(map[CircuitKey]*PaymentCircuit)
+		pending = make(map[CircuitKey]*PaymentCircuit)
+	)
+
+	if err := cm.db.View(func(tx *bolt.Tx) error {
+		// Restore any of the circuits persisted in the circuit bucket
+		// back into memory.
+		circuitBkt := tx.Bucket(circuitAddKey)
+		if circuitBkt == nil {
+			return ErrCorruptedCircuitMap
+		}
+
+		if err := circuitBkt.ForEach(func(_, v []byte) error {
+			circuit, err := decodeCircuit(v)
+			if err != nil {
+				return err
+			}
+
+			circuit.LoadedFromDisk = true
+			pending[circuit.Incoming] = circuit
+
+			return nil
+		}); err != nil {
+			return err
+		}
+
+		// Furthermore, load the keystone bucket and resurrect the
+		// keystones used in any open circuits.
+		keystoneBkt := tx.Bucket(circuitKeystoneKey)
+		if keystoneBkt == nil {
+			return ErrCorruptedCircuitMap
+		}
+
+		if err := keystoneBkt.ForEach(func(k, v []byte) error {
+			var (
+				inKey  CircuitKey
+				outKey = &CircuitKey{}
+			)
+
+			// Decode the incoming and outgoing circuit keys.
+			if err := inKey.SetBytes(v); err != nil {
+				return err
+			}
+			if err := outKey.SetBytes(k); err != nil {
+				return err
+			}
+
+			// Retrieve the pending circuit, set its keystone, then
+			// add it to the opened map.
+			circuit := pending[inKey]
+			circuit.Outgoing = outKey
+			opened[*outKey] = circuit
+
+			return nil
+		}); err != nil {
+			return err
+		}
+
+		return nil
+
+	}); err != nil {
+		return err
+	}
+
+	cm.pending = pending
+	cm.opened = opened
+	cm.closed = make(map[CircuitKey]struct{})
+
+	// Finally, reconstruct the hash index by running through our set of
+	// open circuits.
+	cm.hashIndex = make(map[[32]byte]map[CircuitKey]struct{})
+	for _, circuit := range opened {
+		cm.addCircuitToHashIndex(circuit)
+	}
+
+	return nil
+}
+
+// decodeCircuit reconstructs an in-memory payment circuit from a byte slice.
+// The byte slice is assumed to have been generated by the circuit's Encode
+// method.
+func decodeCircuit(v []byte) (*PaymentCircuit, error) {
+	var circuit = &PaymentCircuit{}
+
+	circuitReader := bytes.NewReader(v)
+	if err := circuit.Decode(circuitReader); err != nil {
+		return nil, err
+	}
+
+	return circuit, nil
+}
+
+// trimAllOpenCircuits reads the set of active channels from disk and trims
+// keystones for any non-pending channels. This method is intended to be called
+// on startup. Each link will also trim it's own circuits upon startup.
+//
+// NOTE: This operation will be applied to the persistent state of all active
+// channels. Therefore, it must be called before any links are created to avoid
+// interfering with normal operation.
+func (cm *circuitMap) trimAllOpenCircuits() error {
+	activeChannels, err := cm.db.FetchAllChannels()
+	if err != nil {
+		return err
+	}
+
+	for _, activeChannel := range activeChannels {
+		if activeChannel.IsPending {
+			continue
+		}
+
+		chanID := activeChannel.ShortChanID
+		start := activeChannel.LocalCommitment.LocalHtlcIndex
+		if err := cm.TrimOpenCircuits(chanID, start); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// TrimOpenCircuits removes a channel's keystones above the short chan id's
+// highest committed htlc index. This has the effect of returning those circuits
+// to a half-open state. Since opening of circuits is done in advance of
+// actually committing the Add htlcs into a commitment txn, this allows circuits
+// to be opened preemetively, since we can roll them back after any failures.
+func (cm *circuitMap) TrimOpenCircuits(chanID lnwire.ShortChannelID,
+	start uint64) error {
+
+	var trimmedOutKeys []CircuitKey
+
+	// Scan forward from the last unacked htlc id, stopping as soon as we
+	// don't find any more. Outgoing htlc id's must be assigned in order, so
+	// there should never be disjoint segments of keystones to trim.
+	cm.mtx.Lock()
+	for i := start; ; i++ {
+		outKey := CircuitKey{
+			ChanID: chanID,
+			HtlcID: i,
+		}
+
+		circuit, ok := cm.opened[outKey]
+		if !ok {
+			break
+		}
+
+		circuit.Outgoing = nil
+		delete(cm.opened, outKey)
+		trimmedOutKeys = append(trimmedOutKeys, outKey)
+		cm.removeCircuitFromHashIndex(circuit)
+	}
+	cm.mtx.Unlock()
+
+	if len(trimmedOutKeys) == 0 {
+		return nil
+	}
+
+	return cm.db.Update(func(tx *bolt.Tx) error {
+		keystoneBkt := tx.Bucket(circuitKeystoneKey)
+		if keystoneBkt == nil {
+			return ErrCorruptedCircuitMap
+		}
+
+		for _, outKey := range trimmedOutKeys {
+			err := keystoneBkt.Delete(outKey.Bytes())
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
+// LookupByHTLC looks up the payment circuit by the outgoing channel and HTLC
+// IDs. Returns nil if there is no such circuit.
+func (cm *circuitMap) LookupCircuit(inKey CircuitKey) *PaymentCircuit {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	return cm.pending[inKey]
+}
+
+// LookupOpenCircuit searches for the circuit identified by its outgoing circuit
+// key.
+func (cm *circuitMap) LookupOpenCircuit(outKey CircuitKey) *PaymentCircuit {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	return cm.opened[outKey]
+}
+
+// LookupByPaymentHash looks up and returns any payment circuits with a given
+// payment hash.
+func (cm *circuitMap) LookupByPaymentHash(hash [32]byte) []*PaymentCircuit {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	var circuits []*PaymentCircuit
+	if circuitSet, ok := cm.hashIndex[hash]; ok {
+		// Iterate over the outgoing circuit keys found with this hash,
+		// and retrieve the circuit from the opened map.
+		circuits = make([]*PaymentCircuit, 0, len(circuitSet))
+		for key := range circuitSet {
+			if circuit, ok := cm.opened[key]; ok {
+				circuits = append(circuits, circuit)
+			}
+		}
+	}
+
+	return circuits
+}
+
+// CommitCircuits accepts any number of circuits and persistently adds them to
+// the switch's circuit map. The method returns a list of circuits that had not
+// been seen prior by the switch. A link should only forward HTLCs corresponding
+// to the returned circuits to the switch.
+//
+// NOTE: This method uses batched writes to improve performance, gains will only
+// be realized if it is called concurrently from separate goroutines.
+func (cm *circuitMap) CommitCircuits(circuits ...*PaymentCircuit) (
+	*CircuitFwdActions, error) {
+
+	actions := &CircuitFwdActions{}
+
+	// If an empty list was passed, return early to avoid grabbing the lock.
+	if len(circuits) == 0 {
+		return actions, nil
+	}
+
+	// First, we reconcile the provided circuits with our set of pending
+	// circuits to construct a set of new circuits that need to be written
+	// to disk. The circuit's pointer is stored so that we only permit this
+	// exact circuit to be forwarded through the switch. If a circuit is
+	// already pending, the htlc will be reforwarded by the switch.
+	//
+	// NOTE: We track an additional addFails subsequence, which permits us
+	// to fail back all packets that weren't dropped if we encounter an
+	// error when committing the circuits.
+	cm.mtx.Lock()
+	var adds, drops, fails, addFails []*PaymentCircuit
+	for _, circuit := range circuits {
+		inKey := circuit.InKey()
+		if foundCircuit, ok := cm.pending[inKey]; ok {
+			switch {
+
+			// This circuit has a keystone, it's waiting for a
+			// response from the remote peer on the outgoing link.
+			// Drop it like it's hot, ensure duplicates get caught.
+			case foundCircuit.HasKeystone():
+				drops = append(drops, circuit)
+
+			// If no keystone is set and the switch has not been
+			// restarted, the corresponding packet should still be
+			// in the outgoing link's mailbox. It will be delivered
+			// if it comes online before the switch goes down.
+			//
+			// NOTE: Dropping here prevents a flapping, incoming
+			// link from failing a duplicate add while it is still
+			// in the server's memory mailboxes.
+			case !foundCircuit.LoadedFromDisk:
+				drops = append(drops, circuit)
+
+			// Otherwise, the in-mem packet has been lost due to a
+			// restart. It is now safe to send back a failure along
+			// the incoming link. The incoming link should be able
+			// detect and ignore duplicate packets of this type.
+			default:
+				fails = append(fails, circuit)
+				addFails = append(addFails, circuit)
+			}
+
+			continue
+		}
+
+		cm.pending[inKey] = circuit
+		adds = append(adds, circuit)
+		addFails = append(addFails, circuit)
+	}
+	cm.mtx.Unlock()
+
+	// If all circuits are dropped or failed, we are done.
+	if len(adds) == 0 {
+		actions.Drops = drops
+		actions.Fails = fails
+		return actions, nil
+	}
+
+	// Now, optimistically serialize the circuits to add.
+	var bs = make([]bytes.Buffer, len(adds))
+	for i, circuit := range adds {
+		if err := circuit.Encode(&bs[i]); err != nil {
+			actions.Drops = drops
+			actions.Fails = addFails
+			return actions, err
+		}
+	}
+
+	// Write the entire batch of circuits to the persistent circuit bucket
+	// using bolt's Batch write. This method must be called from multiple,
+	// distinct goroutines to have any impact on performance.
+	err := cm.db.Batch(func(tx *bolt.Tx) error {
+		circuitBkt := tx.Bucket(circuitAddKey)
+		if circuitBkt == nil {
+			return ErrCorruptedCircuitMap
+		}
+
+		for i, circuit := range adds {
+			inKeyBytes := circuit.InKey().Bytes()
+			circuitBytes := bs[i].Bytes()
+
+			err := circuitBkt.Put(inKeyBytes, circuitBytes)
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+
+	// Return if the write succeeded.
+	if err == nil {
+		actions.Adds = adds
+		actions.Drops = drops
+		actions.Fails = fails
+		return actions, nil
+	}
+
+	// Otherwise, rollback the circuits added to the pending set if the
+	// write failed.
+	cm.mtx.Lock()
+	for _, circuit := range adds {
+		delete(cm.pending, circuit.InKey())
+	}
+	cm.mtx.Unlock()
+
+	// Since our write failed, we will return the dropped packets and mark
+	// all other circuits as failed.
+	actions.Drops = drops
+	actions.Fails = addFails
+
+	return actions, err
+}
+
+// Keystone is a tuple binding an incoming and outgoing CircuitKey. Keystones
+// are preemptively written by an outgoing link before signing a new commitment
+// state, and cements which HTLCs we are awaiting a response from a remote peer.
+type Keystone struct {
+	InKey  CircuitKey
+	OutKey CircuitKey
+}
+
+// String returns a human readable description of the Keystone.
+func (k *Keystone) String() string {
+	return fmt.Sprintf("%s --> %s", k.InKey, k.OutKey)
+}
+
+// OpenCircuits sets the outgoing circuit key for the circuit identified by
+// inKey, persistently marking the circuit as opened. After the changes have
+// been persisted, the circuit map's in-memory indexes are updated so that this
+// circuit can be queried using LookupByKeystone or LookupByPaymentHash.
+func (cm *circuitMap) OpenCircuits(keystones ...Keystone) error {
+	if len(keystones) == 0 {
+		return nil
+	}
+
+	// Check that all keystones correspond to committed-but-unopened
+	// circuits.
+	cm.mtx.RLock()
+	openedCircuits := make([]*PaymentCircuit, 0, len(keystones))
+	for _, ks := range keystones {
+		if _, ok := cm.opened[ks.OutKey]; ok {
+			cm.mtx.RUnlock()
+			return ErrDuplicateKeystone
+		}
+
+		circuit, ok := cm.pending[ks.InKey]
+		if !ok {
+			cm.mtx.RUnlock()
+			return ErrUnknownCircuit
+		}
+
+		openedCircuits = append(openedCircuits, circuit)
+	}
+	cm.mtx.RUnlock()
+
+	err := cm.db.Update(func(tx *bolt.Tx) error {
+		// Now, load the circuit bucket to which we will write the
+		// already serialized circuit.
+		keystoneBkt := tx.Bucket(circuitKeystoneKey)
+		if keystoneBkt == nil {
+			return ErrCorruptedCircuitMap
+		}
+
+		for _, ks := range keystones {
+			outBytes := ks.OutKey.Bytes()
+			inBytes := ks.InKey.Bytes()
+			err := keystoneBkt.Put(outBytes, inBytes)
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		return err
+	}
+
+	cm.mtx.Lock()
+	for i, circuit := range openedCircuits {
+		ks := keystones[i]
+
+		// Since our persistent operation was successful, we can now
+		// modify the in memory representations. Set the outgoing
+		// circuit key on our pending circuit, add the same circuit to
+		// set of opened circuits, and add this circuit to the hash
+		// index.
+		circuit.Outgoing = &CircuitKey{}
+		*circuit.Outgoing = ks.OutKey
+
+		cm.opened[ks.OutKey] = circuit
+		cm.addCircuitToHashIndex(circuit)
+	}
+	cm.mtx.Unlock()
+
+	return nil
+}
+
+// addCirciutToHashIndex inserts a circuit into the circuit map's hash index, so
+// that it can be queried using LookupByPaymentHash.
+func (cm *circuitMap) addCircuitToHashIndex(c *PaymentCircuit) {
+	if _, ok := cm.hashIndex[c.PaymentHash]; !ok {
+		cm.hashIndex[c.PaymentHash] = make(map[CircuitKey]struct{})
+	}
+	cm.hashIndex[c.PaymentHash][c.OutKey()] = struct{}{}
+}
+
+// FailCircuit marks the circuit identified by `inKey` as closing in-memory,
+// which prevents duplicate settles/fails from completing an open circuit twice.
+func (cm *circuitMap) FailCircuit(
+	inKey CircuitKey) (*PaymentCircuit, error) {
+
+	cm.mtx.Lock()
+	defer cm.mtx.Unlock()
+
+	circuit, ok := cm.pending[inKey]
+	if !ok {
+		return nil, ErrUnknownCircuit
+	}
+
+	_, ok = cm.closed[inKey]
+	if ok {
+		return nil, ErrCircuitClosing
+	}
+
+	cm.closed[inKey] = struct{}{}
+
+	return circuit, nil
+}
+
+// CloseCircuit marks the circuit identified by `outKey` as closing
+// in-memory, which prevents duplicate settles/fails from completing an open
+// circuit twice.
+func (cm *circuitMap) CloseCircuit(
+	outKey CircuitKey) (*PaymentCircuit, error) {
+
+	cm.mtx.Lock()
+	defer cm.mtx.Unlock()
+
+	circuit, ok := cm.opened[outKey]
+	if !ok {
+		return nil, ErrUnknownCircuit
+	}
+
+	_, ok = cm.closed[circuit.Incoming]
+	if ok {
+		return nil, ErrCircuitClosing
+	}
+
+	cm.closed[circuit.Incoming] = struct{}{}
+
+	return circuit, nil
+}
+
+// DeleteCircuits destroys the target circuit by removing it from the circuit map,
+// additionally removing the circuit's keystone if the HTLC was forwarded
+// through an outgoing link. The circuit should be identified by its incoming
+// circuit key.
+func (cm *circuitMap) DeleteCircuits(inKeys ...CircuitKey) error {
+
+	var (
+		closingCircuits = make(map[CircuitKey]struct{})
+		removedCircuits = make(map[CircuitKey]*PaymentCircuit)
+	)
+
+	cm.mtx.Lock()
+	// First check that all provided keys are still known to the circuit
+	// map.
+	for _, inKey := range inKeys {
+		if _, ok := cm.pending[inKey]; !ok {
+			cm.mtx.Unlock()
+			return ErrUnknownCircuit
+		}
+	}
+
+	// If no offenders were found, remove any references to the circuit from
+	// memory, keeping track of which circuits were removed, and which ones
+	// had been marked closed. This can be used to restore these entries
+	// later if the persistent removal fails.
+	for _, inKey := range inKeys {
+		circuit := cm.pending[inKey]
+
+		delete(cm.pending, inKey)
+
+		if _, ok := cm.closed[inKey]; ok {
+			closingCircuits[inKey] = struct{}{}
+			delete(cm.closed, inKey)
+		}
+
+		if circuit.HasKeystone() {
+			delete(cm.opened, circuit.OutKey())
+			cm.removeCircuitFromHashIndex(circuit)
+		}
+
+		removedCircuits[inKey] = circuit
+	}
+	cm.mtx.Unlock()
+
+	err := cm.db.Batch(func(tx *bolt.Tx) error {
+		for _, circuit := range removedCircuits {
+			// If this htlc made it to an outgoing link, load the
+			// keystone bucket from which we will remove the
+			// outgoing circuit key.
+			if circuit.HasKeystone() {
+				keystoneBkt := tx.Bucket(circuitKeystoneKey)
+				if keystoneBkt == nil {
+					return ErrCorruptedCircuitMap
+				}
+
+				outKey := circuit.OutKey()
+
+				err := keystoneBkt.Delete(outKey.Bytes())
+				if err != nil {
+					return err
+				}
+			}
+
+			// Remove the circuit itself based on the incoming
+			// circuit key.
+			circuitBkt := tx.Bucket(circuitAddKey)
+			if circuitBkt == nil {
+				return ErrCorruptedCircuitMap
+			}
+
+			inKey := circuit.InKey()
+			if err := circuitBkt.Delete(inKey.Bytes()); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+
+	// Return if the write succeeded.
+	if err == nil {
+		return nil
+	}
+
+	// If the persistent changes failed, restore the circuit map to it's
+	// previous state.
+	cm.mtx.Lock()
+	for inKey, circuit := range removedCircuits {
+		cm.pending[inKey] = circuit
+
+		if _, ok := closingCircuits[inKey]; ok {
+			cm.closed[inKey] = struct{}{}
+		}
+
+		if circuit.HasKeystone() {
+			cm.opened[circuit.OutKey()] = circuit
+			cm.addCircuitToHashIndex(circuit)
+		}
+	}
+	cm.mtx.Unlock()
+
+	return err
+}
+
+// removeCircuitFromHashIndex removes the given circuit from the hash index,
+// pruning any unnecessary memory optimistically.
+func (cm *circuitMap) removeCircuitFromHashIndex(c *PaymentCircuit) {
+	// Locate bucket containing this circuit's payment hashes.
+	circuitsWithHash, ok := cm.hashIndex[c.PaymentHash]
+	if !ok {
+		return
+	}
+
+	outKey := c.OutKey()
+
+	// Remove this circuit from the set of circuitsWithHash.
+	delete(circuitsWithHash, outKey)
+
+	// Prune the payment hash bucket if no other entries remain.
+	if len(circuitsWithHash) == 0 {
+		delete(cm.hashIndex, c.PaymentHash)
+	}
+}
+
+// NumPending returns the number of active circuits added to the circuit map.
+func (cm *circuitMap) NumPending() int {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	return len(cm.pending)
+}
+
+// NumOpen returns the number of circuits that have been opened by way of
+// setting their keystones. This is the number of HTLCs that are waiting for a
+// settle/fail response from a remote peer.
+func (cm *circuitMap) NumOpen() int {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	return len(cm.opened)
+}

htlcswitch/interfaces.go

@@ -47,7 +47,7 @@ type ChannelLink interface {
 	//
 	// NOTE: This function MUST be non-blocking (or block as little as
 	// possible).
-	HandleSwitchPacket(*htlcPacket)
+	HandleSwitchPacket(*htlcPacket) error
 
 	// HandleChannelUpdate handles the htlc requests as settle/add/fail
 	// which sent to us from remote peer we have a channel with.
@@ -98,6 +98,10 @@ type ChannelLink interface {
 	// will use this function in forwarding decisions accordingly.
 	EligibleToForward() bool
 
+	// AttachMailBox delivers an active MailBox to the link. The MailBox may
+	// have buffered messages.
+	AttachMailBox(MailBox)
+
 	// Start/Stop are used to initiate the start/stop of the channel link
 	// functioning.
 	Start() error

htlcswitch/link_test.go

@@ -2,7 +2,10 @@ package htlcswitch
 
 import (
 	"bytes"
+	"crypto/rand"
+	"encoding/binary"
 	"fmt"
+	"io"
 	"runtime"
 	"strings"
 	"sync"
@@ -837,7 +840,7 @@ func TestUpdateForwardingPolicy(t *testing.T) {
 
 	ferr, ok := err.(*ForwardingError)
 	if !ok {
-		t.Fatalf("expected a ForwardingError, instead got: %T", err)
+		t.Fatalf("expected a ForwardingError, instead got (%T): %v", err, err)
 	}
 	switch ferr.FailureMessage.(type) {
 	case *lnwire.FailFeeInsufficient:
@@ -1050,7 +1053,11 @@ func TestChannelLinkMultiHopUnknownNextHop(t *testing.T) {
 	htlcAmt, totalTimelock, hops := generateHops(amount, testStartingHeight,
 		n.firstBobChannelLink, n.carolChannelLink)
 
-	davePub := newMockServer(t, "dave").PubKey()
+	daveServer, err := newMockServer(t, "dave", nil)
+	if err != nil {
+		t.Fatalf("unable to init dave's server: %v", err)
+	}
+	davePub := daveServer.PubKey()
 	receiver := n.bobServer
 	rhash, err := n.makePayment(n.aliceServer, n.bobServer, davePub, hops,
 		amount, htlcAmt, totalTimelock).Wait(30 * time.Second)
@@ -1412,7 +1419,14 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		},
 	}
 
-	chanID := lnwire.NewShortChanIDFromInt(4)
+	var chanIDBytes [8]byte
+	if _, err := io.ReadFull(rand.Reader, chanIDBytes[:]); err != nil {
+		return nil, nil, nil, nil, err
+	}
+
+	chanID := lnwire.NewShortChanIDFromInt(
+		binary.BigEndian.Uint64(chanIDBytes[:]))
+
 	aliceChannel, bobChannel, fCleanUp, _, err := createTestChannel(
 		alicePrivKey, bobPrivKey, chanAmt, chanAmt,
 		chanReserve, chanReserve, chanID,
@@ -1423,8 +1437,8 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 
 	var (
 		invoiceRegistry = newMockRegistry()
-		decoder         = &mockIteratorDecoder{}
-		obfuscator      = newMockObfuscator()
+		decoder         = newMockIteratorDecoder()
+		obfuscator      = NewMockObfuscator()
 		alicePeer       = &mockPeer{
 			sentMsgs: make(chan lnwire.Message, 2000),
 			quit:     make(chan struct{}),
@@ -1442,14 +1456,25 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		preimageMap: make(map[[32]byte][]byte),
 	}
 
+	aliceDb := aliceChannel.State().Db
+
+	aliceSwitch, err := New(Config{DB: aliceDb})
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+
 	t := make(chan time.Time)
 	ticker := &mockTicker{t}
 	aliceCfg := ChannelLinkConfig{
-		FwrdingPolicy:     globalPolicy,
-		Peer:              alicePeer,
-		Switch:            New(Config{}),
-		DecodeHopIterator: decoder.DecodeHopIterator,
-		DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter, lnwire.FailCode) {
+		FwrdingPolicy:      globalPolicy,
+		Peer:               alicePeer,
+		Switch:             aliceSwitch,
+		Circuits:           aliceSwitch.CircuitModifier(),
+		ForwardPackets:     aliceSwitch.ForwardPackets,
+		DecodeHopIterator:  decoder.DecodeHopIterator,
+		DecodeHopIterators: decoder.DecodeHopIterators,
+		DecodeOnionObfuscator: func(*sphinx.OnionPacket) (
+			ErrorEncrypter, lnwire.FailCode) {
 			return obfuscator, lnwire.CodeNone
 		},
 		GetLastChannelUpdate: mockGetChanUpdateMessage,
@@ -1457,10 +1482,11 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 		UpdateContractSignals: func(*contractcourt.ContractSignals) error {
 			return nil
 		},
-		Registry:    invoiceRegistry,
-		ChainEvents: &contractcourt.ChainEventSubscription{},
-		BlockEpochs: globalEpoch,
-		BatchTicker: ticker,
+		Registry:       invoiceRegistry,
+		ChainEvents:    &contractcourt.ChainEventSubscription{},
+		BlockEpochs:    globalEpoch,
+		BatchTicker:    ticker,
+		FwdPkgGCTicker: NewBatchTicker(time.NewTicker(5 * time.Second)),
 		// Make the BatchSize large enough to not
 		// trigger commit update automatically during tests.
 		BatchSize: 10000,
@@ -1468,6 +1494,9 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 
 	const startingHeight = 100
 	aliceLink := NewChannelLink(aliceCfg, aliceChannel, startingHeight)
+	mailbox := newMemoryMailBox()
+	mailbox.Start()
+	aliceLink.AttachMailBox(mailbox)
 	if err := aliceLink.Start(); err != nil {
 		return nil, nil, nil, nil, err
 	}
@@ -1659,25 +1688,27 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 
 	// We'll start the test by creating a single instance of
 	const chanAmt = btcutil.SatoshiPerBitcoin * 5
-	link, bobChannel, tmr, cleanUp, err := newSingleLinkTestHarness(chanAmt, 0)
+
+	aliceLink, bobChannel, tmr, cleanUp, err := newSingleLinkTestHarness(chanAmt, 0)
 	if err != nil {
 		t.Fatalf("unable to create link: %v", err)
 	}
 	defer cleanUp()
 
 	var (
+		carolChanID            = lnwire.NewShortChanIDFromInt(3)
 		mockBlob               [lnwire.OnionPacketSize]byte
-		aliceLink              = link.(*channelLink)
-		aliceChannel           = aliceLink.channel
-		defaultCommitFee       = aliceChannel.StateSnapshot().CommitFee
+		coreChan               = aliceLink.(*channelLink).channel
+		coreLink               = aliceLink.(*channelLink)
+		defaultCommitFee       = coreChan.StateSnapshot().CommitFee
 		aliceStartingBandwidth = aliceLink.Bandwidth()
-		aliceMsgs              = aliceLink.cfg.Peer.(*mockPeer).sentMsgs
+		aliceMsgs              = coreLink.cfg.Peer.(*mockPeer).sentMsgs
 	)
 
 	// We put Alice into HodlHTLC mode, such that she won't settle
 	// incoming HTLCs automatically.
-	aliceLink.cfg.HodlHTLC = true
-	aliceLink.cfg.DebugHTLC = true
+	coreLink.cfg.HodlHTLC = true
+	coreLink.cfg.DebugHTLC = true
 
 	estimator := &lnwallet.StaticFeeEstimator{
 		FeeRate: 24,
@@ -1705,9 +1736,22 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 		t.Fatalf("unable to create payment: %v", err)
 	}
 	addPkt := htlcPacket{
-		htlc: htlc,
+		htlc:           htlc,
+		incomingChanID: sourceHop,
+		incomingHTLCID: 0,
+		obfuscator:     NewMockObfuscator(),
+	}
+
+	circuit := makePaymentCircuit(&htlc.PaymentHash, &addPkt)
+	_, err = coreLink.cfg.Switch.commitCircuits(&circuit)
+	if err != nil {
+		t.Fatalf("unable to commit circuit: %v", err)
+	}
+
+	addPkt.circuit = &circuit
+	if err := aliceLink.HandleSwitchPacket(&addPkt); err != nil {
+		t.Fatalf("unable to handle switch packet: %v", err)
 	}
-	aliceLink.HandleSwitchPacket(&addPkt)
 	time.Sleep(time.Millisecond * 500)
 
 	// The resulting bandwidth should reflect that Alice is paying the
@@ -1733,10 +1777,9 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	}
 
 	// Lock in the HTLC.
-	if err := updateState(tmr, aliceLink, bobChannel, true); err != nil {
+	if err := updateState(tmr, coreLink, bobChannel, true); err != nil {
 		t.Fatalf("unable to update state: %v", err)
 	}
-
 	// Locking in the HTLC should not change Alice's bandwidth.
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth-htlcAmt-htlcFee)
 
@@ -1748,7 +1791,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 		t.Fatalf("unable to settle htlc: %v", err)
 	}
 	htlcSettle := &lnwire.UpdateFulfillHTLC{
-		ID:              bobIndex,
+		ID:              0,
 		PaymentPreimage: invoice.Terms.PaymentPreimage,
 	}
 	aliceLink.HandleChannelUpdate(htlcSettle)
@@ -1759,7 +1802,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth-htlcAmt-htlcFee)
 
 	// Lock in the settle.
-	if err := updateState(tmr, aliceLink, bobChannel, false); err != nil {
+	if err := updateState(tmr, coreLink, bobChannel, false); err != nil {
 		t.Fatalf("unable to update state: %v", err)
 	}
 
@@ -1773,9 +1816,22 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 		t.Fatalf("unable to create payment: %v", err)
 	}
 	addPkt = htlcPacket{
-		htlc: htlc,
+		htlc:           htlc,
+		incomingChanID: sourceHop,
+		incomingHTLCID: 1,
+		obfuscator:     NewMockObfuscator(),
+	}
+
+	circuit = makePaymentCircuit(&htlc.PaymentHash, &addPkt)
+	_, err = coreLink.cfg.Switch.commitCircuits(&circuit)
+	if err != nil {
+		t.Fatalf("unable to commit circuit: %v", err)
+	}
+
+	addPkt.circuit = &circuit
+	if err := aliceLink.HandleSwitchPacket(&addPkt); err != nil {
+		t.Fatalf("unable to handle switch packet: %v", err)
 	}
-	aliceLink.HandleSwitchPacket(&addPkt)
 	time.Sleep(time.Millisecond * 500)
 
 	// Again, Alice's bandwidth decreases by htlcAmt+htlcFee.
@@ -1787,6 +1843,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	case <-time.After(2 * time.Second):
 		t.Fatalf("did not receive message")
 	}
+
 	addHtlc, ok = msg.(*lnwire.UpdateAddHTLC)
 	if !ok {
 		t.Fatalf("expected UpdateAddHTLC, got %T", msg)
@@ -1798,7 +1855,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	}
 
 	// Lock in the HTLC, which should not affect the bandwidth.
-	if err := updateState(tmr, aliceLink, bobChannel, true); err != nil {
+	if err := updateState(tmr, coreLink, bobChannel, true); err != nil {
 		t.Fatalf("unable to update state: %v", err)
 	}
 
@@ -1812,9 +1869,10 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 		t.Fatalf("unable to fail htlc: %v", err)
 	}
 	failMsg := &lnwire.UpdateFailHTLC{
-		ID:     bobIndex,
+		ID:     1,
 		Reason: lnwire.OpaqueReason([]byte("nop")),
 	}
+
 	aliceLink.HandleChannelUpdate(failMsg)
 	time.Sleep(time.Millisecond * 500)
 
@@ -1822,7 +1880,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth-htlcAmt*2-htlcFee)
 
 	// Lock in the Fail.
-	if err := updateState(tmr, aliceLink, bobChannel, false); err != nil {
+	if err := updateState(tmr, coreLink, bobChannel, false); err != nil {
 		t.Fatalf("unable to update state: %v", err)
 	}
 
@@ -1834,7 +1892,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	// remain unchanged (but Alice will need to pay the fee for the extra
 	// HTLC).
 	htlcAmt, totalTimelock, hops := generateHops(htlcAmt, testStartingHeight,
-		aliceLink)
+		coreLink)
 	blob, err := generateRoute(hops...)
 	if err != nil {
 		t.Fatalf("unable to gen route: %v", err)
@@ -1847,11 +1905,12 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 
 	// We must add the invoice to the registry, such that Alice expects
 	// this payment.
-	err = aliceLink.cfg.Registry.(*mockInvoiceRegistry).AddInvoice(*invoice)
+	err = coreLink.cfg.Registry.(*mockInvoiceRegistry).AddInvoice(*invoice)
 	if err != nil {
 		t.Fatalf("unable to add invoice to registry: %v", err)
 	}
 
+	htlc.ID = 0
 	bobIndex, err = bobChannel.AddHTLC(htlc, nil)
 	if err != nil {
 		t.Fatalf("unable to add htlc: %v", err)
@@ -1862,58 +1921,84 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth-htlcAmt)
 
 	// Lock in the HTLC.
-	if err := updateState(tmr, aliceLink, bobChannel, false); err != nil {
+	if err := updateState(tmr, coreLink, bobChannel, false); err != nil {
 		t.Fatalf("unable to update state: %v", err)
 	}
 
 	// Since Bob is adding this HTLC, Alice only needs to pay the fee.
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth-htlcAmt-htlcFee)
+	time.Sleep(time.Millisecond * 500)
+
+	addPkt = htlcPacket{
+		htlc:           htlc,
+		incomingChanID: aliceLink.ShortChanID(),
+		incomingHTLCID: 0,
+		obfuscator:     NewMockObfuscator(),
+	}
+
+	circuit = makePaymentCircuit(&htlc.PaymentHash, &addPkt)
+	_, err = coreLink.cfg.Switch.commitCircuits(&circuit)
+	if err != nil {
+		t.Fatalf("unable to commit circuit: %v", err)
+	}
+
+	addPkt.outgoingChanID = carolChanID
+	addPkt.outgoingHTLCID = 0
+
+	err = coreLink.cfg.Switch.openCircuits(addPkt.keystone())
+	if err != nil {
+		t.Fatalf("unable to set keystone: %v", err)
+	}
 
 	// Next, we'll settle the HTLC with our knowledge of the pre-image that
 	// we eventually learn (simulating a multi-hop payment). The bandwidth
 	// of the channel should now be re-balanced to the starting point.
 	settlePkt := htlcPacket{
+		incomingChanID: aliceLink.ShortChanID(),
+		incomingHTLCID: 0,
+		circuit:        &circuit,
+		outgoingChanID: addPkt.outgoingChanID,
+		outgoingHTLCID: addPkt.outgoingHTLCID,
 		htlc: &lnwire.UpdateFulfillHTLC{
-			ID:              bobIndex,
+			ID:              0,
 			PaymentPreimage: invoice.Terms.PaymentPreimage,
 		},
+		obfuscator: NewMockObfuscator(),
 	}
 
-	aliceLink.HandleSwitchPacket(&settlePkt)
+	if err := aliceLink.HandleSwitchPacket(&settlePkt); err != nil {
+		t.Fatalf("unable to handle switch packet: %v", err)
+	}
 	time.Sleep(time.Millisecond * 500)
 
 	// Settling this HTLC gives Alice all her original bandwidth back.
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth)
 
-	// Alice wil send the Settle to Bob.
 	select {
 	case msg = <-aliceMsgs:
 	case <-time.After(2 * time.Second):
 		t.Fatalf("did not receive message")
 	}
 
-	settleHtlc, ok := msg.(*lnwire.UpdateFulfillHTLC)
+	settleMsg, ok := msg.(*lnwire.UpdateFulfillHTLC)
 	if !ok {
 		t.Fatalf("expected UpdateFulfillHTLC, got %T", msg)
 	}
-	pre := settleHtlc.PaymentPreimage
-	idx := settleHtlc.ID
-	err = bobChannel.ReceiveHTLCSettle(pre, idx)
+	err = bobChannel.ReceiveHTLCSettle(settleMsg.PaymentPreimage, settleMsg.ID)
 	if err != nil {
-		t.Fatalf("unable to receive settle: %v", err)
+		t.Fatalf("failed receiving fail htlc: %v", err)
 	}
 
-	// After a settle the link should do a state transition automatically,
-	// so we don't have to trigger it.
-	if err := handleStateUpdate(aliceLink, bobChannel); err != nil {
+	// After failing an HTLC, the link will automatically trigger
+	// a state update.
+	if err := handleStateUpdate(coreLink, bobChannel); err != nil {
 		t.Fatalf("unable to update state: %v", err)
 	}
-	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth)
 
-	// Finally, we'll test the scenario of failing an HTLC received from the
+	// Finally, we'll test the scenario of failing an HTLC received by the
 	// remote node. This should result in no perceived bandwidth changes.
 	htlcAmt, totalTimelock, hops = generateHops(htlcAmt, testStartingHeight,
-		aliceLink)
+		coreLink)
 	blob, err = generateRoute(hops...)
 	if err != nil {
 		t.Fatalf("unable to gen route: %v", err)
@@ -1922,7 +2007,8 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to create payment: %v", err)
 	}
-	if err := aliceLink.cfg.Registry.(*mockInvoiceRegistry).AddInvoice(*invoice); err != nil {
+	err = coreLink.cfg.Registry.(*mockInvoiceRegistry).AddInvoice(*invoice)
+	if err != nil {
 		t.Fatalf("unable to add invoice to registry: %v", err)
 	}
 
@@ -1940,21 +2026,49 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 
 	// No changes before the HTLC is locked in.
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth)
-	if err := updateState(tmr, aliceLink, bobChannel, false); err != nil {
+	if err := updateState(tmr, coreLink, bobChannel, false); err != nil {
 		t.Fatalf("unable to update state: %v", err)
 	}
 
 	// After lock-in, Alice will have to pay the htlc fee.
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth-htlcFee)
 
-	// Now fail this HTLC.
-	failPkt := htlcPacket{
-		incomingHTLCID: bobIndex,
-		htlc: &lnwire.UpdateFailHTLC{
-			ID: bobIndex,
-		},
+	addPkt = htlcPacket{
+		htlc:           htlc,
+		incomingChanID: aliceLink.ShortChanID(),
+		incomingHTLCID: 1,
+		obfuscator:     NewMockObfuscator(),
+	}
+
+	circuit = makePaymentCircuit(&htlc.PaymentHash, &addPkt)
+	_, err = coreLink.cfg.Switch.commitCircuits(&circuit)
+	if err != nil {
+		t.Fatalf("unable to commit circuit: %v", err)
+	}
+
+	addPkt.outgoingChanID = carolChanID
+	addPkt.outgoingHTLCID = 1
+
+	err = coreLink.cfg.Switch.openCircuits(addPkt.keystone())
+	if err != nil {
+		t.Fatalf("unable to set keystone: %v", err)
+	}
+
+	failPkt := htlcPacket{
+		incomingChanID: aliceLink.ShortChanID(),
+		incomingHTLCID: 1,
+		circuit:        &circuit,
+		outgoingChanID: addPkt.outgoingChanID,
+		outgoingHTLCID: addPkt.outgoingHTLCID,
+		htlc: &lnwire.UpdateFailHTLC{
+			ID: 1,
+		},
+		obfuscator: NewMockObfuscator(),
+	}
+
+	if err := aliceLink.HandleSwitchPacket(&failPkt); err != nil {
+		t.Fatalf("unable to handle switch packet: %v", err)
 	}
-	aliceLink.HandleSwitchPacket(&failPkt)
 	time.Sleep(time.Millisecond * 500)
 
 	// Alice should get all her bandwidth back.
@@ -1977,7 +2091,7 @@ func TestChannelLinkBandwidthConsistency(t *testing.T) {
 
 	// After failing an HTLC, the link will automatically trigger
 	// a state update.
-	if err := handleStateUpdate(aliceLink, bobChannel); err != nil {
+	if err := handleStateUpdate(coreLink, bobChannel); err != nil {
 		t.Fatalf("unable to update state: %v", err)
 	}
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth)
@@ -2015,20 +2129,27 @@ func TestChannelLinkBandwidthConsistencyOverflow(t *testing.T) {
 	}
 	feePerKw := feeRate.FeePerKWeight()
 
-	// The starting bandwidth of the channel should be exactly the amount
-	// that we created the channel between her and Bob.
-	expectedBandwidth := lnwire.NewMSatFromSatoshis(chanAmt - defaultCommitFee)
-	assertLinkBandwidth(t, aliceLink, expectedBandwidth)
-
-	addLinkHTLC := func(amt lnwire.MilliSatoshi) [32]byte {
+	var htlcID uint64
+	addLinkHTLC := func(id uint64, amt lnwire.MilliSatoshi) [32]byte {
 		invoice, htlc, err := generatePayment(amt, amt, 5, mockBlob)
 		if err != nil {
 			t.Fatalf("unable to create payment: %v", err)
 		}
-		aliceLink.HandleSwitchPacket(&htlcPacket{
-			htlc:   htlc,
-			amount: amt,
-		})
+
+		addPkt := &htlcPacket{
+			htlc:           htlc,
+			incomingHTLCID: id,
+			amount:         amt,
+			obfuscator:     NewMockObfuscator(),
+		}
+		circuit := makePaymentCircuit(&htlc.PaymentHash, addPkt)
+		_, err = coreLink.cfg.Switch.commitCircuits(&circuit)
+		if err != nil {
+			t.Fatalf("unable to commit circuit: %v", err)
+		}
+
+		addPkt.circuit = &circuit
+		aliceLink.HandleSwitchPacket(addPkt)
 		return invoice.Terms.PaymentPreimage
 	}
 
@@ -2040,10 +2161,11 @@ func TestChannelLinkBandwidthConsistencyOverflow(t *testing.T) {
 	const numHTLCs = lnwallet.MaxHTLCNumber / 2
 	var preImages [][32]byte
 	for i := 0; i < numHTLCs; i++ {
-		preImage := addLinkHTLC(htlcAmt)
+		preImage := addLinkHTLC(htlcID, htlcAmt)
 		preImages = append(preImages, preImage)
 
 		totalHtlcAmt += htlcAmt
+		htlcID++
 	}
 
 	// The HTLCs should all be sent to the remote.
@@ -2051,8 +2173,8 @@ func TestChannelLinkBandwidthConsistencyOverflow(t *testing.T) {
 	for i := 0; i < numHTLCs; i++ {
 		select {
 		case msg = <-aliceMsgs:
-		case <-time.After(2 * time.Second):
-			t.Fatalf("did not receive message")
+		case <-time.After(5 * time.Second):
+			t.Fatalf("did not receive message %d", i)
 		}
 
 		addHtlc, ok := msg.(*lnwire.UpdateAddHTLC)
@@ -2078,7 +2200,7 @@ func TestChannelLinkBandwidthConsistencyOverflow(t *testing.T) {
 	htlcFee := lnwire.NewMSatFromSatoshis(
 		feePerKw.FeeForWeight(commitWeight),
 	)
-	expectedBandwidth = aliceStartingBandwidth - totalHtlcAmt - htlcFee
+	expectedBandwidth := aliceStartingBandwidth - totalHtlcAmt - htlcFee
 	expectedBandwidth += lnwire.NewMSatFromSatoshis(defaultCommitFee)
 	assertLinkBandwidth(t, aliceLink, expectedBandwidth)
 
@@ -2094,10 +2216,11 @@ func TestChannelLinkBandwidthConsistencyOverflow(t *testing.T) {
 	// bandwidth accounting is done properly.
 	const numOverFlowHTLCs = 20
 	for i := 0; i < numOverFlowHTLCs; i++ {
-		preImage := addLinkHTLC(htlcAmt)
+		preImage := addLinkHTLC(htlcID, htlcAmt)
 		preImages = append(preImages, preImage)
 
 		totalHtlcAmt += htlcAmt
+		htlcID++
 	}
 
 	// No messages should be sent to the remote at this point.
@@ -2245,10 +2368,18 @@ func TestChannelLinkBandwidthChanReserve(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to create payment: %v", err)
 	}
-	addPkt := htlcPacket{
-		htlc: htlc,
+
+	addPkt := &htlcPacket{
+		htlc:       htlc,
+		obfuscator: NewMockObfuscator(),
 	}
-	aliceLink.HandleSwitchPacket(&addPkt)
+	circuit := makePaymentCircuit(&htlc.PaymentHash, addPkt)
+	_, err = coreLink.cfg.Switch.commitCircuits(&circuit)
+	if err != nil {
+		t.Fatalf("unable to commit circuit: %v", err)
+	}
+
+	aliceLink.HandleSwitchPacket(addPkt)
 	time.Sleep(time.Millisecond * 100)
 	assertLinkBandwidth(t, aliceLink, aliceStartingBandwidth-htlcAmt-htlcFee)
 
@@ -2834,11 +2965,11 @@ func TestChannelLinkUpdateCommitFee(t *testing.T) {
 	}
 }
 
-// TestChannelLinkRejectDuplicatePayment tests that if a link receives an
-// incoming HTLC for a payment we have already settled, then it rejects the
-// HTLC. We do this as we want to enforce the fact that invoices are only to be
-// used _once.
-func TestChannelLinkRejectDuplicatePayment(t *testing.T) {
+// TestChannelLinkAcceptDuplicatePayment tests that if a link receives an
+// incoming HTLC for a payment we have already settled, then it accepts the
+// HTLC. We do this to simplify the processing of settles after restarts or
+// failures, reducing ambiguity when a batch is only partially processed.
+func TestChannelLinkAcceptDuplicatePayment(t *testing.T) {
 	t.Parallel()
 
 	// First, we'll create our traditional three hop network. We'll only be
@@ -2891,8 +3022,8 @@ func TestChannelLinkRejectDuplicatePayment(t *testing.T) {
 	// as it's a duplicate request.
 	_, err = n.aliceServer.htlcSwitch.SendHTLC(n.bobServer.PubKey(), htlc,
 		newMockDeobfuscator())
-	if err.Error() != lnwire.CodeUnknownPaymentHash.String() {
-		t.Fatal("error haven't been received")
+	if err != nil {
+		t.Fatalf("error shouldn't have been received got: %v", err)
 	}
 }
 

htlcswitch/mailbox.go

@@ -1,23 +1,40 @@
 package htlcswitch
 
 import (
+	"container/list"
+	"errors"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/lightningnetwork/lnd/lnwire"
 )
 
-// mailBox is an interface which represents a concurrent-safe, in-order
+// ErrMailBoxShuttingDown is returned when the mailbox is interrupted by a
+// shutdown request.
+var ErrMailBoxShuttingDown = errors.New("mailbox is shutting down")
+
+// MailBox is an interface which represents a concurrent-safe, in-order
 // delivery queue for messages from the network and also from the main switch.
 // This struct servers as a buffer between incoming messages, and messages to
 // the handled by the link. Each of the mutating methods within this interface
 // should be implemented in a non-blocking manner.
-type mailBox interface {
+type MailBox interface {
 	// AddMessage appends a new message to the end of the message queue.
 	AddMessage(msg lnwire.Message) error
 
 	// AddPacket appends a new message to the end of the packet queue.
 	AddPacket(pkt *htlcPacket) error
 
+	// HasPacket queries the packets for a circuit key, this is used to drop
+	// packets bound for the switch that already have a queued response.
+	HasPacket(CircuitKey) bool
+
+	// AckPacket removes a packet from the mailboxes in-memory replay
+	// buffer. This will prevent a packet from being delivered after a link
+	// restarts if the switch has remained online.
+	AckPacket(CircuitKey) error
+
 	// MessageOutBox returns a channel that any new messages ready for
 	// delivery will be sent on.
 	MessageOutBox() chan lnwire.Message
@@ -26,6 +43,12 @@ type mailBox interface {
 	// delivery will be sent on.
 	PacketOutBox() chan *htlcPacket
 
+	// Clears any pending wire messages from the inbox.
+	ResetMessages() error
+
+	// Reset the packet head to point at the first element in the list.
+	ResetPackets() error
+
 	// Start starts the mailbox and any goroutines it needs to operate
 	// properly.
 	Start() error
@@ -34,20 +57,28 @@ type mailBox interface {
 	Stop() error
 }
 
-// memoryMailBox is an implementation of the mailBox struct backed by purely
+// memoryMailBox is an implementation of the MailBox struct backed by purely
 // in-memory queues.
 type memoryMailBox struct {
-	wireMessages []lnwire.Message
+	started uint32
+	stopped uint32
+
+	wireMessages *list.List
+	wireHead     *list.Element
 	wireMtx      sync.Mutex
 	wireCond     *sync.Cond
 
 	messageOutbox chan lnwire.Message
+	msgReset      chan chan struct{}
 
-	htlcPkts []*htlcPacket
+	htlcPkts *list.List
+	pktIndex map[CircuitKey]*list.Element
+	pktHead  *list.Element
 	pktMtx   sync.Mutex
 	pktCond  *sync.Cond
 
 	pktOutbox chan *htlcPacket
+	pktReset  chan chan struct{}
 
 	wg   sync.WaitGroup
 	quit chan struct{}
@@ -56,9 +87,14 @@ type memoryMailBox struct {
 // newMemoryMailBox creates a new instance of the memoryMailBox.
 func newMemoryMailBox() *memoryMailBox {
 	box := &memoryMailBox{
-		quit:          make(chan struct{}),
+		wireMessages:  list.New(),
+		htlcPkts:      list.New(),
 		messageOutbox: make(chan lnwire.Message),
 		pktOutbox:     make(chan *htlcPacket),
+		msgReset:      make(chan chan struct{}, 1),
+		pktReset:      make(chan chan struct{}, 1),
+		pktIndex:      make(map[CircuitKey]*list.Element),
+		quit:          make(chan struct{}),
 	}
 	box.wireCond = sync.NewCond(&box.wireMtx)
 	box.pktCond = sync.NewCond(&box.pktMtx)
@@ -66,12 +102,12 @@ func newMemoryMailBox() *memoryMailBox {
 	return box
 }
 
-// A compile time assertion to ensure that memoryMailBox meets the mailBox
+// A compile time assertion to ensure that memoryMailBox meets the MailBox
 // interface.
-var _ mailBox = (*memoryMailBox)(nil)
+var _ MailBox = (*memoryMailBox)(nil)
 
 // courierType is an enum that reflects the distinct types of messages a
-// mailBox can handle. Each type will be placed in an isolated mail box and
+// MailBox can handle. Each type will be placed in an isolated mail box and
 // will have a dedicated goroutine for delivering the messages.
 type courierType uint8
 
@@ -85,8 +121,12 @@ const (
 
 // Start starts the mailbox and any goroutines it needs to operate properly.
 //
-// NOTE: This method is part of the mailBox interface.
+// NOTE: This method is part of the MailBox interface.
 func (m *memoryMailBox) Start() error {
+	if !atomic.CompareAndSwapUint32(&m.started, 0, 1) {
+		return nil
+	}
+
 	m.wg.Add(2)
 	go m.mailCourier(wireCourier)
 	go m.mailCourier(pktCourier)
@@ -94,10 +134,90 @@ func (m *memoryMailBox) Start() error {
 	return nil
 }
 
+// ResetMessages blocks until all buffered wire messages are cleared.
+func (m *memoryMailBox) ResetMessages() error {
+	msgDone := make(chan struct{})
+	select {
+	case m.msgReset <- msgDone:
+		return m.signalUntilReset(wireCourier, msgDone)
+	case <-m.quit:
+		return ErrMailBoxShuttingDown
+	}
+}
+
+// ResetPackets blocks until the head of packets buffer is reset, causing the
+// packets to be redelivered in order.
+func (m *memoryMailBox) ResetPackets() error {
+	pktDone := make(chan struct{})
+	select {
+	case m.pktReset <- pktDone:
+		return m.signalUntilReset(pktCourier, pktDone)
+	case <-m.quit:
+		return ErrMailBoxShuttingDown
+	}
+}
+
+// signalUntilReset strobes the condition variable for the specified inbox type
+// until receiving a response that the mailbox has processed a reset.
+func (m *memoryMailBox) signalUntilReset(cType courierType,
+	done chan struct{}) error {
+
+	for {
+		switch cType {
+		case wireCourier:
+			m.wireCond.Signal()
+		case pktCourier:
+			m.pktCond.Signal()
+		}
+
+		select {
+		case <-time.After(time.Millisecond):
+			continue
+		case <-done:
+			return nil
+		case <-m.quit:
+			return ErrMailBoxShuttingDown
+		}
+	}
+}
+
+// AckPacket removes the packet identified by it's incoming circuit key from the
+// queue of packets to be delivered.
+//
+// NOTE: It is safe to call this method multiple times for the same circuit key.
+func (m *memoryMailBox) AckPacket(inKey CircuitKey) error {
+	m.pktCond.L.Lock()
+	entry, ok := m.pktIndex[inKey]
+	if !ok {
+		m.pktCond.L.Unlock()
+		return nil
+	}
+
+	m.htlcPkts.Remove(entry)
+	delete(m.pktIndex, inKey)
+	m.pktCond.L.Unlock()
+
+	return nil
+}
+
+// HasPacket queries the packets for a circuit key, this is used to drop packets
+// bound for the switch that already have a queued response.
+func (m *memoryMailBox) HasPacket(inKey CircuitKey) bool {
+	m.pktCond.L.Lock()
+	_, ok := m.pktIndex[inKey]
+	m.pktCond.L.Unlock()
+
+	return ok
+}
+
 // Stop signals the mailbox and its goroutines for a graceful shutdown.
 //
-// NOTE: This method is part of the mailBox interface.
+// NOTE: This method is part of the MailBox interface.
 func (m *memoryMailBox) Stop() error {
+	if !atomic.CompareAndSwapUint32(&m.stopped, 0, 1) {
+		return nil
+	}
+
 	close(m.quit)
 
 	m.wireCond.Signal()
@@ -121,10 +241,13 @@ func (m *memoryMailBox) mailCourier(cType courierType) {
 		switch cType {
 		case wireCourier:
 			m.wireCond.L.Lock()
-			for len(m.wireMessages) == 0 {
+			for m.wireMessages.Front() == nil {
 				m.wireCond.Wait()
 
 				select {
+				case msgDone := <-m.msgReset:
+					m.wireMessages.Init()
+					close(msgDone)
 				case <-m.quit:
 					m.wireCond.L.Unlock()
 					return
@@ -134,10 +257,13 @@ func (m *memoryMailBox) mailCourier(cType courierType) {
 
 		case pktCourier:
 			m.pktCond.L.Lock()
-			for len(m.htlcPkts) == 0 {
+			for m.pktHead == nil {
 				m.pktCond.Wait()
 
 				select {
+				case pktDone := <-m.pktReset:
+					m.pktHead = m.htlcPkts.Front()
+					close(pktDone)
 				case <-m.quit:
 					m.pktCond.L.Unlock()
 					return
@@ -155,13 +281,11 @@ func (m *memoryMailBox) mailCourier(cType courierType) {
 		)
 		switch cType {
 		case wireCourier:
-			nextMsg = m.wireMessages[0]
-			m.wireMessages[0] = nil // Set to nil to prevent GC leak.
-			m.wireMessages = m.wireMessages[1:]
+			entry := m.wireMessages.Front()
+			nextMsg = m.wireMessages.Remove(entry).(lnwire.Message)
 		case pktCourier:
-			nextPkt = m.htlcPkts[0]
-			m.htlcPkts[0] = nil // Set to nil to prevent GC leak.
-			m.htlcPkts = m.htlcPkts[1:]
+			nextPkt = m.pktHead.Value.(*htlcPacket)
+			m.pktHead = m.pktHead.Next()
 		}
 
 		// Now that we're done with the condition, we can unlock it to
@@ -173,13 +297,17 @@ func (m *memoryMailBox) mailCourier(cType courierType) {
 			m.pktCond.L.Unlock()
 		}
 
-		// With the next message obtained, we'll now select to attempt
 		// to deliver the message. If we receive a kill signal, then
 		// we'll bail out.
 		switch cType {
 		case wireCourier:
 			select {
 			case m.messageOutbox <- nextMsg:
+			case msgDone := <-m.msgReset:
+				m.wireCond.L.Lock()
+				m.wireMessages.Init()
+				m.wireCond.L.Unlock()
+				close(msgDone)
 			case <-m.quit:
 				return
 			}
@@ -187,6 +315,11 @@ func (m *memoryMailBox) mailCourier(cType courierType) {
 		case pktCourier:
 			select {
 			case m.pktOutbox <- nextPkt:
+			case pktDone := <-m.pktReset:
+				m.pktCond.L.Lock()
+				m.pktHead = m.htlcPkts.Front()
+				m.pktCond.L.Unlock()
+				close(pktDone)
 			case <-m.quit:
 				return
 			}
@@ -197,13 +330,13 @@ func (m *memoryMailBox) mailCourier(cType courierType) {
 
 // AddMessage appends a new message to the end of the message queue.
 //
-// NOTE: This method is safe for concrete use and part of the mailBox
+// NOTE: This method is safe for concrete use and part of the MailBox
 // interface.
 func (m *memoryMailBox) AddMessage(msg lnwire.Message) error {
 	// First, we'll lock the condition, and add the message to the end of
 	// the wire message inbox.
 	m.wireCond.L.Lock()
-	m.wireMessages = append(m.wireMessages, msg)
+	m.wireMessages.PushBack(msg)
 	m.wireCond.L.Unlock()
 
 	// With the message added, we signal to the mailCourier that there are
@@ -215,13 +348,22 @@ func (m *memoryMailBox) AddMessage(msg lnwire.Message) error {
 
 // AddPacket appends a new message to the end of the packet queue.
 //
-// NOTE: This method is safe for concrete use and part of the mailBox
+// NOTE: This method is safe for concrete use and part of the MailBox
 // interface.
 func (m *memoryMailBox) AddPacket(pkt *htlcPacket) error {
 	// First, we'll lock the condition, and add the packet to the end of
 	// the htlc packet inbox.
 	m.pktCond.L.Lock()
-	m.htlcPkts = append(m.htlcPkts, pkt)
+	if _, ok := m.pktIndex[pkt.inKey()]; ok {
+		m.pktCond.L.Unlock()
+		return nil
+	}
+
+	entry := m.htlcPkts.PushBack(pkt)
+	m.pktIndex[pkt.inKey()] = entry
+	if m.pktHead == nil {
+		m.pktHead = entry
+	}
 	m.pktCond.L.Unlock()
 
 	// With the packet added, we signal to the mailCourier that there are
@@ -234,7 +376,7 @@ func (m *memoryMailBox) AddPacket(pkt *htlcPacket) error {
 // MessageOutBox returns a channel that any new messages ready for delivery
 // will be sent on.
 //
-// NOTE: This method is part of the mailBox interface.
+// NOTE: This method is part of the MailBox interface.
 func (m *memoryMailBox) MessageOutBox() chan lnwire.Message {
 	return m.messageOutbox
 }
@@ -242,7 +384,7 @@ func (m *memoryMailBox) MessageOutBox() chan lnwire.Message {
 // PacketOutBox returns a channel that any new packets ready for delivery will
 // be sent on.
 //
-// NOTE: This method is part of the mailBox interface.
+// NOTE: This method is part of the MailBox interface.
 func (m *memoryMailBox) PacketOutBox() chan *htlcPacket {
 	return m.pktOutbox
 }

htlcswitch/mailbox_test.go

@@ -25,6 +25,7 @@ func TestMailBoxCouriers(t *testing.T) {
 
 	// We'll be adding 10 message of both types to the mailbox.
 	const numPackets = 10
+	const halfPackets = numPackets / 2
 
 	// We'll add a set of random packets to the mailbox.
 	sentPackets := make([]*htlcPacket, numPackets)
@@ -96,4 +97,53 @@ func TestMailBoxCouriers(t *testing.T) {
 		t.Fatalf("recvd messages mismatched: expected %v, got %v",
 			spew.Sdump(sentMessages), spew.Sdump(recvdMessages))
 	}
+
+	// Now that we've received all of the intended msgs/pkts, ack back half
+	// of the packets.
+	for _, recvdPkt := range recvdPackets[:halfPackets] {
+		mailBox.AckPacket(recvdPkt.inKey())
+	}
+
+	// With the packets drained and partially acked,  we reset the mailbox,
+	// simulating a link shutting down and then coming back up.
+	mailBox.ResetMessages()
+	mailBox.ResetPackets()
+
+	// Now, we'll use the same alternating strategy to read from our
+	// mailbox. All wire messages are dropped on startup, but any unacked
+	// packets will be replayed in the same order they were delivered
+	// initially.
+	recvdPackets2 := make([]*htlcPacket, 0, halfPackets)
+	for i := 0; i < 2*halfPackets; i++ {
+		timeout := time.After(time.Second * 5)
+		if i%2 == 0 {
+			select {
+			case <-timeout:
+				t.Fatalf("didn't recv pkt after timeout")
+			case pkt := <-mailBox.PacketOutBox():
+				recvdPackets2 = append(recvdPackets2, pkt)
+			}
+		} else {
+			select {
+			case <-mailBox.MessageOutBox():
+				t.Fatalf("should not receive wire msg after reset")
+			default:
+			}
+		}
+	}
+
+	// The number of packets we received should match the number of unacked
+	// packets left in the mailbox.
+	if halfPackets != len(recvdPackets2) {
+		t.Fatalf("expected %v packets instead got %v", halfPackets,
+			len(recvdPackets))
+	}
+
+	// Additionally, the set of packets should match exactly with the
+	// unacked packets, and we should have received the packets in the exact
+	// same ordering that we added.
+	if !reflect.DeepEqual(recvdPackets[halfPackets:], recvdPackets2) {
+		t.Fatalf("recvd packets mismatched: expected %v, got %v",
+			spew.Sdump(sentPackets), spew.Sdump(recvdPackets))
+	}
 }

htlcswitch/mock.go

@@ -4,6 +4,7 @@ import (
 	"crypto/sha256"
 	"encoding/binary"
 	"fmt"
+	"io/ioutil"
 	"sync"
 	"testing"
 	"time"
@@ -120,25 +121,48 @@ type mockServer struct {
 
 var _ Peer = (*mockServer)(nil)
 
-func newMockServer(t testing.TB, name string) *mockServer {
+func initSwitchWithDB(db *channeldb.DB) (*Switch, error) {
+	if db == nil {
+		tempPath, err := ioutil.TempDir("", "switchdb")
+		if err != nil {
+			return nil, err
+		}
+
+		db, err = channeldb.Open(tempPath)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return New(Config{
+		DB:             db,
+		SwitchPackager: channeldb.NewSwitchPackager(),
+		FwdingLog: &mockForwardingLog{
+			events: make(map[time.Time]channeldb.ForwardingEvent),
+		},
+	})
+}
+
+func newMockServer(t testing.TB, name string, db *channeldb.DB) (*mockServer, error) {
 	var id [33]byte
 	h := sha256.Sum256([]byte(name))
 	copy(id[:], h[:])
 
-	return &mockServer{
-		t:        t,
-		id:       id,
-		name:     name,
-		messages: make(chan lnwire.Message, 3000),
-		quit:     make(chan struct{}),
-		registry: newMockRegistry(),
-		htlcSwitch: New(Config{
-			FwdingLog: &mockForwardingLog{
-				events: make(map[time.Time]channeldb.ForwardingEvent),
-			},
-		}),
-		interceptorFuncs: make([]messageInterceptor, 0),
+	htlcSwitch, err := initSwitchWithDB(db)
+	if err != nil {
+		return nil, err
 	}
+
+	return &mockServer{
+		t:                t,
+		id:               id,
+		name:             name,
+		messages:         make(chan lnwire.Message, 3000),
+		quit:             make(chan struct{}),
+		registry:         newMockRegistry(),
+		htlcSwitch:       htlcSwitch,
+		interceptorFuncs: make([]messageInterceptor, 0),
+	}, nil
 }
 
 func (s *mockServer) Start() error {
@@ -196,10 +220,6 @@ type mockHopIterator struct {
 	hops []ForwardingInfo
 }
 
-func (r *mockHopIterator) OnionPacket() *sphinx.OnionPacket {
-	return nil
-}
-
 func newMockHopIterator(hops ...ForwardingInfo) HopIterator {
 	return &mockHopIterator{hops: hops}
 }
@@ -261,7 +281,8 @@ type mockObfuscator struct {
 	ogPacket *sphinx.OnionPacket
 }
 
-func newMockObfuscator() ErrorEncrypter {
+// NewMockObfuscator initializes a dummy mockObfuscator used for testing.
+func NewMockObfuscator() ErrorEncrypter {
 	return &mockObfuscator{}
 }
 
@@ -512,6 +533,10 @@ type mockChannelLink struct {
 
 	peer Peer
 
+	startMailBox bool
+
+	mailBox MailBox
+
 	packets chan *htlcPacket
 
 	eligible bool
@@ -519,6 +544,39 @@ type mockChannelLink struct {
 	htlcID uint64
 }
 
+// completeCircuit is a helper method for adding the finalized payment circuit
+// to the switch's circuit map. In testing, this should be executed after
+// receiving an htlc from the downstream packets channel.
+func (f *mockChannelLink) completeCircuit(pkt *htlcPacket) error {
+	switch htlc := pkt.htlc.(type) {
+	case *lnwire.UpdateAddHTLC:
+		pkt.outgoingChanID = f.shortChanID
+		pkt.outgoingHTLCID = f.htlcID
+		htlc.ID = f.htlcID
+
+		keystone := Keystone{pkt.inKey(), pkt.outKey()}
+		if err := f.htlcSwitch.openCircuits(keystone); err != nil {
+			return err
+		}
+
+		f.htlcID++
+
+	case *lnwire.UpdateFulfillHTLC, *lnwire.UpdateFailHTLC:
+		err := f.htlcSwitch.teardownCircuit(pkt)
+		if err != nil {
+			return err
+		}
+	}
+
+	f.mailBox.AckPacket(pkt.inKey())
+
+	return nil
+}
+
+func (f *mockChannelLink) deleteCircuit(pkt *htlcPacket) error {
+	return f.htlcSwitch.deleteCircuits(pkt.inKey())
+}
+
 func newMockChannelLink(htlcSwitch *Switch, chanID lnwire.ChannelID,
 	shortChanID lnwire.ShortChannelID, peer Peer, eligible bool,
 ) *mockChannelLink {
@@ -527,27 +585,14 @@ func newMockChannelLink(htlcSwitch *Switch, chanID lnwire.ChannelID,
 		htlcSwitch:  htlcSwitch,
 		chanID:      chanID,
 		shortChanID: shortChanID,
-		packets:     make(chan *htlcPacket, 1),
 		peer:        peer,
 		eligible:    eligible,
 	}
 }
 
-func (f *mockChannelLink) HandleSwitchPacket(packet *htlcPacket) {
-	switch htlc := packet.htlc.(type) {
-	case *lnwire.UpdateAddHTLC:
-		f.htlcSwitch.addCircuit(&PaymentCircuit{
-			PaymentHash:    htlc.PaymentHash,
-			IncomingChanID: packet.incomingChanID,
-			IncomingHTLCID: packet.incomingHTLCID,
-			OutgoingChanID: f.shortChanID,
-			OutgoingHTLCID: f.htlcID,
-			ErrorEncrypter: packet.obfuscator,
-		})
-		f.htlcID++
-	}
-
-	f.packets <- packet
+func (f *mockChannelLink) HandleSwitchPacket(pkt *htlcPacket) error {
+	f.mailBox.AddPacket(pkt)
+	return nil
 }
 
 func (f *mockChannelLink) HandleChannelUpdate(lnwire.Message) {
@@ -560,12 +605,22 @@ func (f *mockChannelLink) Stats() (uint64, lnwire.MilliSatoshi, lnwire.MilliSato
 	return 0, 0, 0
 }
 
+func (f *mockChannelLink) AttachMailBox(mailBox MailBox) {
+	f.mailBox = mailBox
+	f.packets = mailBox.PacketOutBox()
+}
+
+func (f *mockChannelLink) Start() error {
+	f.mailBox.ResetMessages()
+	f.mailBox.ResetPackets()
+	return nil
+}
+
 func (f *mockChannelLink) ChanID() lnwire.ChannelID                    { return f.chanID }
 func (f *mockChannelLink) ShortChanID() lnwire.ShortChannelID          { return f.shortChanID }
 func (f *mockChannelLink) UpdateShortChanID(sid lnwire.ShortChannelID) { f.shortChanID = sid }
 func (f *mockChannelLink) Bandwidth() lnwire.MilliSatoshi              { return 99999999 }
 func (f *mockChannelLink) Peer() Peer                                  { return f.peer }
-func (f *mockChannelLink) Start() error                                { return nil }
 func (f *mockChannelLink) Stop()                                       {}
 func (f *mockChannelLink) EligibleToForward() bool                     { return f.eligible }
 
@@ -603,6 +658,10 @@ func (i *mockInvoiceRegistry) SettleInvoice(rhash chainhash.Hash) error {
 		return fmt.Errorf("can't find mock invoice: %x", rhash[:])
 	}
 
+	if invoice.Terms.Settled {
+		return nil
+	}
+
 	invoice.Terms.Settled = true
 	i.invoices[rhash] = invoice
 

htlcswitch/packet.go

@@ -1,6 +1,7 @@
 package htlcswitch
 
 import (
+	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/lnwire"
 )
 
@@ -35,6 +36,20 @@ type htlcPacket struct {
 	// outgoing channel.
 	outgoingHTLCID uint64
 
+	// sourceRef is used by forwarded htlcPackets to locate incoming Add
+	// entry in a fwdpkg owned by the incoming link. This value can be nil
+	// if there is no such entry, e.g. switch initiated payments.
+	sourceRef *channeldb.AddRef
+
+	// destRef is used to locate a settle/fail entry in the outgoing link's
+	// fwdpkg. If sourceRef is non-nil, this reference should be to a
+	// settle/fail in response to the sourceRef.
+	destRef *channeldb.SettleFailRef
+
+	// incomingAmount is the value in milli-satoshis that arrived on an
+	// incoming link.
+	incomingAmount lnwire.MilliSatoshi
+
 	// amount is the value of the HTLC that is being created or modified.
 	amount lnwire.MilliSatoshi
 
@@ -50,10 +65,10 @@ type htlcPacket struct {
 	// encrypted with any shared secret.
 	localFailure bool
 
-	// isRouted is set to true if the incomingChanID and incomingHTLCID fields
-	// of a forwarded fail packet are already set and do not need to be looked
-	// up in the circuit map.
-	isRouted bool
+	// hasSource is set to true if the incomingChanID and incomingHTLCID
+	// fields of a forwarded fail packet are already set and do not need to
+	// be looked up in the circuit map.
+	hasSource bool
 
 	// isResolution is set to true if this packet was actually an incoming
 	// resolution message from an outside sub-system. We'll treat these as
@@ -61,4 +76,32 @@ type htlcPacket struct {
 	// encrypt all errors related to this packet as if we were the first
 	// hop.
 	isResolution bool
+
+	// circuit holds a reference to an Add's circuit which is persisted in
+	// the switch during successful forwarding.
+	circuit *PaymentCircuit
+}
+
+// inKey returns the circuit key used to identify the incoming htlc.
+func (p *htlcPacket) inKey() CircuitKey {
+	return CircuitKey{
+		ChanID: p.incomingChanID,
+		HtlcID: p.incomingHTLCID,
+	}
+}
+
+// outKey returns the circuit key used to identify the outgoing, forwarded htlc.
+func (p *htlcPacket) outKey() CircuitKey {
+	return CircuitKey{
+		ChanID: p.outgoingChanID,
+		HtlcID: p.outgoingHTLCID,
+	}
+}
+
+// keystone returns a tuple containing the incoming and outgoing circuit keys.
+func (p *htlcPacket) keystone() Keystone {
+	return Keystone{
+		InKey:  p.inKey(),
+		OutKey: p.outKey(),
+	}
 }

htlcswitch/sequencer.go

@@ -0,0 +1,128 @@
+package htlcswitch
+
+import (
+	"sync"
+
+	"github.com/boltdb/bolt"
+	"github.com/go-errors/errors"
+	"github.com/lightningnetwork/lnd/channeldb"
+)
+
+// defaultSequenceBatchSize specifies the window of sequence numbers that are
+// allocated for each write to disk made by the sequencer.
+const defaultSequenceBatchSize = 1000
+
+// Sequencer emits sequence numbers for locally initiated HTLCs. These are
+// only used internally for tracking pending payments, however they must be
+// unique in order to avoid circuit key collision in the circuit map.
+type Sequencer interface {
+	// NextID returns a unique sequence number for each invocation.
+	NextID() (uint64, error)
+}
+
+var (
+	// nextPaymentIDKey identifies the bucket that will keep track of the
+	// persistent sequence numbers for payments.
+	nextPaymentIDKey = []byte("next-payment-id-key")
+
+	// ErrSequencerCorrupted signals that the persistence engine was not
+	// initialized, or has been corrupted since startup.
+	ErrSequencerCorrupted = errors.New(
+		"sequencer database has been corrupted")
+)
+
+// persistentSequencer is a concrete implementation of IDGenerator, that uses
+// channeldb to allocate sequence numbers.
+type persistentSequencer struct {
+	db *channeldb.DB
+
+	mu sync.Mutex
+
+	nextID    uint64
+	horizonID uint64
+}
+
+// NewPersistentSequencer initializes a new sequencer using a channeldb backend.
+func NewPersistentSequencer(db *channeldb.DB) (Sequencer, error) {
+	g := &persistentSequencer{
+		db: db,
+	}
+
+	// Ensure the database bucket is created before any updates are
+	// performed.
+	if err := g.initDB(); err != nil {
+		return nil, err
+	}
+
+	return g, nil
+}
+
+// NextID returns a unique sequence number for every invocation, persisting the
+// assignment to avoid reuse.
+func (s *persistentSequencer) NextID() (uint64, error) {
+
+	// nextID will be the unique sequence number returned if no errors are
+	// encountered.
+	var nextID uint64
+
+	// If our sequence batch has not been exhausted, we can allocate the
+	// next identifier in the range.
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.nextID < s.horizonID {
+		nextID = s.nextID
+		s.nextID++
+
+		return nextID, nil
+	}
+
+	// Otherwise, our sequence batch has been exhausted. We use the last
+	// known sequence number on disk to mark the beginning of the next
+	// sequence batch, and allocate defaultSequenceBatchSize (1000) at a
+	// time.
+	//
+	// NOTE: This also will happen on the first invocation after startup,
+	// i.e. when nextID and horizonID are both 0. The next sequence batch to be
+	// allocated will start from the last known tip on disk, which is fine
+	// as we only require uniqueness of the allocated numbers.
+	var nextHorizonID uint64
+	if err := s.db.Update(func(tx *bolt.Tx) error {
+		nextIDBkt := tx.Bucket(nextPaymentIDKey)
+		if nextIDBkt == nil {
+			return ErrSequencerCorrupted
+		}
+
+		nextID = nextIDBkt.Sequence()
+		nextHorizonID = nextID + defaultSequenceBatchSize
+
+		// Cannot fail when used in Update.
+		nextIDBkt.SetSequence(nextHorizonID)
+
+		return nil
+	}); err != nil {
+		return 0, err
+	}
+
+	// Never assign index zero, to avoid collisions with the EmptyKeystone.
+	if nextID == 0 {
+		nextID++
+	}
+
+	// If our batch sequence allocation succeed, update our in-memory values
+	// so we can continue to allocate sequence numbers without hitting disk.
+	// The nextID is incremented by one in memory so the in can be used
+	// issued directly on the next invocation.
+	s.nextID = nextID + 1
+	s.horizonID = nextHorizonID
+
+	return nextID, nil
+}
+
+// initDB populates the bucket used to generate payment sequence numbers.
+func (s *persistentSequencer) initDB() error {
+	return s.db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists(nextPaymentIDKey)
+		return err
+	})
+}

htlcswitch/test_utils.go

@@ -4,14 +4,17 @@ import (
 	"bytes"
 	"crypto/rand"
 	"crypto/sha256"
+	"encoding/binary"
 	"fmt"
 	"io/ioutil"
 	"math/big"
 	"net"
 	"os"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/boltdb/bolt"
 	"github.com/btcsuite/fastsha256"
 	"github.com/go-errors/errors"
 	"github.com/lightningnetwork/lightning-onion"
@@ -53,8 +56,35 @@ var (
 		"3135609736119018462340006816851118", 10)
 )
 
-// mockGetChanUpdateMessage helper function which returns topology update of
-// the channel
+var idSeqNum uint64
+
+func genIDs() (lnwire.ChannelID, lnwire.ChannelID, lnwire.ShortChannelID,
+	lnwire.ShortChannelID) {
+
+	id := atomic.AddUint64(&idSeqNum, 2)
+
+	var scratch [8]byte
+
+	binary.BigEndian.PutUint64(scratch[:], id)
+	hash1, _ := chainhash.NewHash(bytes.Repeat(scratch[:], 4))
+
+	binary.BigEndian.PutUint64(scratch[:], id+1)
+	hash2, _ := chainhash.NewHash(bytes.Repeat(scratch[:], 4))
+
+	chanPoint1 := wire.NewOutPoint(hash1, uint32(id))
+	chanPoint2 := wire.NewOutPoint(hash2, uint32(id+1))
+
+	chanID1 := lnwire.NewChanIDFromOutPoint(chanPoint1)
+	chanID2 := lnwire.NewChanIDFromOutPoint(chanPoint2)
+
+	aliceChanID := lnwire.NewShortChanIDFromInt(id)
+	bobChanID := lnwire.NewShortChanIDFromInt(id + 1)
+
+	return chanID1, chanID2, aliceChanID, bobChanID
+}
+
+// mockGetChanUpdateMessage helper function which returns topology update
+// of the channel
 func mockGetChanUpdateMessage() (*lnwire.ChannelUpdate, error) {
 	return &lnwire.ChannelUpdate{
 		Signature: wireSig,
@@ -293,6 +323,8 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 	}
 
 	cleanUpFunc := func() {
+		dbAlice.Close()
+		dbBob.Close()
 		os.RemoveAll(bobPath)
 		os.RemoveAll(alicePath)
 	}
@@ -339,7 +371,21 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 	restore := func() (*lnwallet.LightningChannel, *lnwallet.LightningChannel,
 		error) {
 		aliceStoredChannels, err := dbAlice.FetchOpenChannels(aliceKeyPub)
-		if err != nil {
+		switch err {
+		case nil:
+		case bolt.ErrDatabaseNotOpen:
+			dbAlice, err = channeldb.Open(dbAlice.Path())
+			if err != nil {
+				return nil, nil, errors.Errorf("unable to reopen alice "+
+					"db: %v", err)
+			}
+
+			aliceStoredChannels, err = dbAlice.FetchOpenChannels(aliceKeyPub)
+			if err != nil {
+				return nil, nil, errors.Errorf("unable to fetch alice "+
+					"channel: %v", err)
+			}
+		default:
 			return nil, nil, errors.Errorf("unable to fetch alice channel: "+
 				"%v", err)
 		}
@@ -364,7 +410,21 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 		}
 
 		bobStoredChannels, err := dbBob.FetchOpenChannels(bobKeyPub)
-		if err != nil {
+		switch err {
+		case nil:
+		case bolt.ErrDatabaseNotOpen:
+			dbBob, err = channeldb.Open(dbBob.Path())
+			if err != nil {
+				return nil, nil, errors.Errorf("unable to reopen bob "+
+					"db: %v", err)
+			}
+
+			bobStoredChannels, err = dbBob.FetchOpenChannels(bobKeyPub)
+			if err != nil {
+				return nil, nil, errors.Errorf("unable to fetch bob "+
+					"channel: %v", err)
+			}
+		default:
 			return nil, nil, errors.Errorf("unable to fetch bob channel: "+
 				"%v", err)
 		}
@@ -689,8 +749,7 @@ type clusterChannels struct {
 func createClusterChannels(aliceToBob, bobToCarol btcutil.Amount) (
 	*clusterChannels, func(), func() (*clusterChannels, error), error) {
 
-	firstChanID := lnwire.NewShortChanIDFromInt(4)
-	secondChanID := lnwire.NewShortChanIDFromInt(5)
+	_, _, firstChanID, secondChanID := genIDs()
 
 	// Create lightning channels between Alice<->Bob and Bob<->Carol
 	aliceChannel, firstBobChannel, cleanAliceBob, restoreAliceBob, err :=
@@ -759,14 +818,29 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 	secondBobChannel, carolChannel *lnwallet.LightningChannel,
 	startingHeight uint32) *threeHopNetwork {
 
+	aliceDb := aliceChannel.State().Db
+	bobDb := firstBobChannel.State().Db
+	carolDb := carolChannel.State().Db
+
 	// Create three peers/servers.
-	aliceServer := newMockServer(t, "alice")
-	bobServer := newMockServer(t, "bob")
-	carolServer := newMockServer(t, "carol")
+	aliceServer, err := newMockServer(t, "alice", aliceDb)
+	if err != nil {
+		t.Fatalf("unable to create alice server: %v", err)
+	}
+	bobServer, err := newMockServer(t, "bob", bobDb)
+	if err != nil {
+		t.Fatalf("unable to create bob server: %v", err)
+	}
+	carolServer, err := newMockServer(t, "carol", carolDb)
+	if err != nil {
+		t.Fatalf("unable to create carol server: %v", err)
+	}
 
 	// Create mock decoder instead of sphinx one in order to mock the route
 	// which htlc should follow.
-	decoder := &mockIteratorDecoder{}
+	aliceDecoder := newMockIteratorDecoder()
+	bobDecoder := newMockIteratorDecoder()
+	carolDecoder := newMockIteratorDecoder()
 
 	feeEstimator := &mockFeeEstimator{
 		byteFeeIn: make(chan lnwallet.SatPerVByte),
@@ -783,7 +857,7 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 		BaseFee:       lnwire.NewMSatFromSatoshis(1),
 		TimeLockDelta: 6,
 	}
-	obfuscator := newMockObfuscator()
+	obfuscator := NewMockObfuscator()
 
 	aliceEpochChan := make(chan *chainntnfs.BlockEpoch)
 	aliceEpoch := &chainntnfs.BlockEpochEvent{
@@ -794,12 +868,14 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 	aliceTicker := time.NewTicker(50 * time.Millisecond)
 	aliceChannelLink := NewChannelLink(
 		ChannelLinkConfig{
-			FwrdingPolicy:     globalPolicy,
-			Peer:              bobServer,
-			Switch:            aliceServer.htlcSwitch,
-			DecodeHopIterator: decoder.DecodeHopIterator,
-			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter,
-				lnwire.FailCode) {
+			FwrdingPolicy:      globalPolicy,
+			Peer:               bobServer,
+			Circuits:           aliceServer.htlcSwitch.CircuitModifier(),
+			ForwardPackets:     aliceServer.htlcSwitch.ForwardPackets,
+			DecodeHopIterator:  aliceDecoder.DecodeHopIterator,
+			DecodeHopIterators: aliceDecoder.DecodeHopIterators,
+			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (
+				ErrorEncrypter, lnwire.FailCode) {
 				return obfuscator, lnwire.CodeNone
 			},
 			GetLastChannelUpdate: mockGetChanUpdateMessage,
@@ -810,10 +886,11 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			UpdateContractSignals: func(*contractcourt.ContractSignals) error {
 				return nil
 			},
-			ChainEvents: &contractcourt.ChainEventSubscription{},
-			SyncStates:  true,
-			BatchTicker: &mockTicker{aliceTicker.C},
-			BatchSize:   10,
+			ChainEvents:    &contractcourt.ChainEventSubscription{},
+			SyncStates:     true,
+			BatchTicker:    &mockTicker{aliceTicker.C},
+			FwdPkgGCTicker: &mockTicker{time.NewTicker(5 * time.Second).C},
+			BatchSize:      10,
 		},
 		aliceChannel,
 		startingHeight,
@@ -840,12 +917,14 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 	firstBobTicker := time.NewTicker(50 * time.Millisecond)
 	firstBobChannelLink := NewChannelLink(
 		ChannelLinkConfig{
-			FwrdingPolicy:     globalPolicy,
-			Peer:              aliceServer,
-			Switch:            bobServer.htlcSwitch,
-			DecodeHopIterator: decoder.DecodeHopIterator,
-			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter,
-				lnwire.FailCode) {
+			FwrdingPolicy:      globalPolicy,
+			Peer:               aliceServer,
+			Circuits:           bobServer.htlcSwitch.CircuitModifier(),
+			ForwardPackets:     bobServer.htlcSwitch.ForwardPackets,
+			DecodeHopIterator:  bobDecoder.DecodeHopIterator,
+			DecodeHopIterators: bobDecoder.DecodeHopIterators,
+			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (
+				ErrorEncrypter, lnwire.FailCode) {
 				return obfuscator, lnwire.CodeNone
 			},
 			GetLastChannelUpdate: mockGetChanUpdateMessage,
@@ -856,10 +935,11 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			UpdateContractSignals: func(*contractcourt.ContractSignals) error {
 				return nil
 			},
-			ChainEvents: &contractcourt.ChainEventSubscription{},
-			SyncStates:  true,
-			BatchTicker: &mockTicker{firstBobTicker.C},
-			BatchSize:   10,
+			ChainEvents:    &contractcourt.ChainEventSubscription{},
+			SyncStates:     true,
+			BatchTicker:    &mockTicker{firstBobTicker.C},
+			FwdPkgGCTicker: &mockTicker{time.NewTicker(5 * time.Second).C},
+			BatchSize:      10,
 		},
 		firstBobChannel,
 		startingHeight,
@@ -886,12 +966,14 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 	secondBobTicker := time.NewTicker(50 * time.Millisecond)
 	secondBobChannelLink := NewChannelLink(
 		ChannelLinkConfig{
-			FwrdingPolicy:     globalPolicy,
-			Peer:              carolServer,
-			Switch:            bobServer.htlcSwitch,
-			DecodeHopIterator: decoder.DecodeHopIterator,
-			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter,
-				lnwire.FailCode) {
+			FwrdingPolicy:      globalPolicy,
+			Peer:               carolServer,
+			Circuits:           bobServer.htlcSwitch.CircuitModifier(),
+			ForwardPackets:     bobServer.htlcSwitch.ForwardPackets,
+			DecodeHopIterator:  bobDecoder.DecodeHopIterator,
+			DecodeHopIterators: bobDecoder.DecodeHopIterators,
+			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (
+				ErrorEncrypter, lnwire.FailCode) {
 				return obfuscator, lnwire.CodeNone
 			},
 			GetLastChannelUpdate: mockGetChanUpdateMessage,
@@ -902,10 +984,11 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			UpdateContractSignals: func(*contractcourt.ContractSignals) error {
 				return nil
 			},
-			ChainEvents: &contractcourt.ChainEventSubscription{},
-			SyncStates:  true,
-			BatchTicker: &mockTicker{secondBobTicker.C},
-			BatchSize:   10,
+			ChainEvents:    &contractcourt.ChainEventSubscription{},
+			SyncStates:     true,
+			BatchTicker:    &mockTicker{secondBobTicker.C},
+			FwdPkgGCTicker: &mockTicker{time.NewTicker(5 * time.Second).C},
+			BatchSize:      10,
 		},
 		secondBobChannel,
 		startingHeight,
@@ -932,12 +1015,14 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 	carolTicker := time.NewTicker(50 * time.Millisecond)
 	carolChannelLink := NewChannelLink(
 		ChannelLinkConfig{
-			FwrdingPolicy:     globalPolicy,
-			Peer:              bobServer,
-			Switch:            carolServer.htlcSwitch,
-			DecodeHopIterator: decoder.DecodeHopIterator,
-			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (ErrorEncrypter,
-				lnwire.FailCode) {
+			FwrdingPolicy:      globalPolicy,
+			Peer:               bobServer,
+			Circuits:           carolServer.htlcSwitch.CircuitModifier(),
+			ForwardPackets:     carolServer.htlcSwitch.ForwardPackets,
+			DecodeHopIterator:  carolDecoder.DecodeHopIterator,
+			DecodeHopIterators: carolDecoder.DecodeHopIterators,
+			DecodeOnionObfuscator: func(*sphinx.OnionPacket) (
+				ErrorEncrypter, lnwire.FailCode) {
 				return obfuscator, lnwire.CodeNone
 			},
 			GetLastChannelUpdate: mockGetChanUpdateMessage,
@@ -948,10 +1033,11 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 			UpdateContractSignals: func(*contractcourt.ContractSignals) error {
 				return nil
 			},
-			ChainEvents: &contractcourt.ChainEventSubscription{},
-			SyncStates:  true,
-			BatchTicker: &mockTicker{carolTicker.C},
-			BatchSize:   10,
+			ChainEvents:    &contractcourt.ChainEventSubscription{},
+			SyncStates:     true,
+			BatchTicker:    &mockTicker{carolTicker.C},
+			FwdPkgGCTicker: &mockTicker{time.NewTicker(5 * time.Second).C},
+			BatchSize:      10,
 		},
 		carolChannel,
 		startingHeight,

lntest/node.go

@@ -239,6 +239,7 @@ func (hn *HarnessNode) start(lndError chan<- error) error {
 	hn.quit = make(chan struct{})
 
 	args := hn.cfg.genArgs()
+	args = append(args, fmt.Sprintf("--profile=%d", 9000+hn.NodeID))
 	hn.cmd = exec.Command("lnd", args...)
 
 	// Redirect stderr output to buffer
@@ -394,6 +395,12 @@ func (hn *HarnessNode) connectRPC() (*grpc.ClientConn, error) {
 	return grpc.Dial(hn.cfg.RPCAddr(), opts...)
 }
 
+// SetExtraArgs assigns the ExtraArgs field for the node's configuration. The
+// changes will take effect on restart.
+func (hn *HarnessNode) SetExtraArgs(extraArgs []string) {
+	hn.cfg.ExtraArgs = extraArgs
+}
+
 // cleanup cleans up all the temporary files created by the node's process.
 func (hn *HarnessNode) cleanup() error {
 	return os.RemoveAll(hn.cfg.BaseDir)

peer.go

@@ -379,18 +379,21 @@ func (p *peer) loadActiveChannels(chans []*channeldb.OpenChannel) error {
 		linkCfg := htlcswitch.ChannelLinkConfig{
 			Peer:                  p,
 			DecodeHopIterator:     p.server.sphinx.DecodeHopIterator,
+			DecodeHopIterators:    p.server.sphinx.DecodeHopIterators,
 			DecodeOnionObfuscator: p.server.sphinx.ExtractErrorEncrypter,
 			GetLastChannelUpdate: createGetLastUpdate(p.server.chanRouter,
 				p.PubKey(), lnChan.ShortChanID()),
-			DebugHTLC:     cfg.DebugHTLC,
-			HodlHTLC:      cfg.HodlHTLC,
-			Registry:      p.server.invoices,
-			Switch:        p.server.htlcSwitch,
-			FwrdingPolicy: *forwardingPolicy,
-			FeeEstimator:  p.server.cc.feeEstimator,
-			BlockEpochs:   blockEpoch,
-			PreimageCache: p.server.witnessBeacon,
-			ChainEvents:   chainEvents,
+			DebugHTLC:      cfg.DebugHTLC,
+			HodlHTLC:       cfg.HodlHTLC,
+			Registry:       p.server.invoices,
+			Switch:         p.server.htlcSwitch,
+			Circuits:       p.server.htlcSwitch.CircuitModifier(),
+			ForwardPackets: p.server.htlcSwitch.ForwardPackets,
+			FwrdingPolicy:  *forwardingPolicy,
+			FeeEstimator:   p.server.cc.feeEstimator,
+			BlockEpochs:    blockEpoch,
+			PreimageCache:  p.server.witnessBeacon,
+			ChainEvents:    chainEvents,
 			UpdateContractSignals: func(signals *contractcourt.ContractSignals) error {
 				return p.server.chainArb.UpdateContractSignals(
 					*chanPoint, signals,
@@ -399,7 +402,10 @@ func (p *peer) loadActiveChannels(chans []*channeldb.OpenChannel) error {
 			SyncStates: true,
 			BatchTicker: htlcswitch.NewBatchTicker(
 				time.NewTicker(50 * time.Millisecond)),
-			BatchSize: 10,
+			FwdPkgGCTicker: htlcswitch.NewBatchTicker(
+				time.NewTicker(time.Minute)),
+			BatchSize:    10,
+			UnsafeReplay: cfg.UnsafeReplay,
 		}
 		link := htlcswitch.NewChannelLink(linkCfg, lnChan,
 			uint32(currentHeight))
@@ -1020,6 +1026,7 @@ func (p *peer) writeMessage(msg lnwire.Message) error {
 // NOTE: This method MUST be run as a goroutine.
 func (p *peer) writeHandler() {
 	var exitErr error
+
 out:
 	for {
 		select {
@@ -1272,18 +1279,21 @@ out:
 			linkConfig := htlcswitch.ChannelLinkConfig{
 				Peer:                  p,
 				DecodeHopIterator:     p.server.sphinx.DecodeHopIterator,
+				DecodeHopIterators:    p.server.sphinx.DecodeHopIterators,
 				DecodeOnionObfuscator: p.server.sphinx.ExtractErrorEncrypter,
 				GetLastChannelUpdate: createGetLastUpdate(p.server.chanRouter,
 					p.PubKey(), newChanReq.channel.ShortChanID()),
-				DebugHTLC:     cfg.DebugHTLC,
-				HodlHTLC:      cfg.HodlHTLC,
-				Registry:      p.server.invoices,
-				Switch:        p.server.htlcSwitch,
-				FwrdingPolicy: p.server.cc.routingPolicy,
-				FeeEstimator:  p.server.cc.feeEstimator,
-				BlockEpochs:   blockEpoch,
-				PreimageCache: p.server.witnessBeacon,
-				ChainEvents:   chainEvents,
+				DebugHTLC:      cfg.DebugHTLC,
+				HodlHTLC:       cfg.HodlHTLC,
+				Registry:       p.server.invoices,
+				Switch:         p.server.htlcSwitch,
+				Circuits:       p.server.htlcSwitch.CircuitModifier(),
+				ForwardPackets: p.server.htlcSwitch.ForwardPackets,
+				FwrdingPolicy:  p.server.cc.routingPolicy,
+				FeeEstimator:   p.server.cc.feeEstimator,
+				BlockEpochs:    blockEpoch,
+				PreimageCache:  p.server.witnessBeacon,
+				ChainEvents:    chainEvents,
 				UpdateContractSignals: func(signals *contractcourt.ContractSignals) error {
 					return p.server.chainArb.UpdateContractSignals(
 						*chanPoint, signals,
@@ -1292,7 +1302,10 @@ out:
 				SyncStates: false,
 				BatchTicker: htlcswitch.NewBatchTicker(
 					time.NewTicker(50 * time.Millisecond)),
-				BatchSize: 10,
+				FwdPkgGCTicker: htlcswitch.NewBatchTicker(
+					time.NewTicker(time.Minute)),
+				BatchSize:    10,
+				UnsafeReplay: cfg.UnsafeReplay,
 			}
 			link := htlcswitch.NewChannelLink(linkConfig, newChan,
 				uint32(currentHeight))

rpcserver.go

@@ -650,7 +650,7 @@ func (r *rpcServer) DisconnectPeer(ctx context.Context,
 	// In order to avoid erroneously disconnecting from a peer that we have
 	// an active channel with, if we have any channels active with this
 	// peer, then we'll disallow disconnecting from them.
-	if len(nodeChannels) > 0 {
+	if len(nodeChannels) > 0 && !cfg.UnsafeDisconnect {
 		return nil, fmt.Errorf("cannot disconnect from peer(%x), "+
 			"all active channels with the peer need to be closed "+
 			"first", pubKeyBytes)

server.go

@@ -206,7 +206,8 @@ func newServer(listenAddrs []string, chanDB *channeldb.DB, cc *chainControl,
 			debugPre[:], debugHash[:])
 	}
 
-	s.htlcSwitch = htlcswitch.New(htlcswitch.Config{
+	htlcSwitch, err := htlcswitch.New(htlcswitch.Config{
+		DB:      chanDB,
 		SelfKey: s.identityPriv.PubKey(),
 		LocalChannelClose: func(pubKey []byte,
 			request *htlcswitch.ChanClose) {
@@ -230,8 +231,13 @@ func newServer(listenAddrs []string, chanDB *channeldb.DB, cc *chainControl,
 					pubKey[:], err)
 			}
 		},
-		FwdingLog: chanDB.ForwardingLog(),
+		FwdingLog:      chanDB.ForwardingLog(),
+		SwitchPackager: channeldb.NewSwitchPackager(),
 	})
+	if err != nil {
+		return nil, err
+	}
+	s.htlcSwitch = htlcSwitch
 
 	// If external IP addresses have been specified, add those to the list
 	// of this server's addresses. We need to use the cfg.net.ResolveTCPAddr

test_utils.go

@@ -2,6 +2,9 @@ package main
 
 import (
 	"bytes"
+	crand "crypto/rand"
+	"encoding/binary"
+	"io"
 	"io/ioutil"
 	"math/rand"
 	"net"
@@ -191,11 +194,21 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 		CommitSig:     bytes.Repeat([]byte{1}, 71),
 	}
 
+	var chanIDBytes [8]byte
+	if _, err := io.ReadFull(crand.Reader, chanIDBytes[:]); err != nil {
+		return nil, nil, nil, nil, err
+	}
+
+	shortChanID := lnwire.NewShortChanIDFromInt(
+		binary.BigEndian.Uint64(chanIDBytes[:]),
+	)
+
 	aliceChannelState := &channeldb.OpenChannel{
 		LocalChanCfg:            aliceCfg,
 		RemoteChanCfg:           bobCfg,
 		IdentityPub:             aliceKeyPub,
 		FundingOutpoint:         *prevOut,
+		ShortChanID:             shortChanID,
 		ChanType:                channeldb.SingleFunder,
 		IsInitiator:             true,
 		Capacity:                channelCapacity,
@@ -205,6 +218,7 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 		LocalCommitment:         aliceCommit,
 		RemoteCommitment:        aliceCommit,
 		Db:                      dbAlice,
+		Packager:                channeldb.NewChannelPackager(shortChanID),
 	}
 	bobChannelState := &channeldb.OpenChannel{
 		LocalChanCfg:            bobCfg,
@@ -220,6 +234,7 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 		LocalCommitment:         bobCommit,
 		RemoteCommitment:        bobCommit,
 		Db:                      dbBob,
+		Packager:                channeldb.NewChannelPackager(shortChanID),
 	}
 
 	addr := &net.TCPAddr{
@@ -291,7 +306,14 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 		breachArbiter: breachArbiter,
 		chainArb:      chainArb,
 	}
-	s.htlcSwitch = htlcswitch.New(htlcswitch.Config{})
+	htlcSwitch, err := htlcswitch.New(htlcswitch.Config{
+		DB:             dbAlice,
+		SwitchPackager: channeldb.NewSwitchPackager(),
+	})
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+	s.htlcSwitch = htlcSwitch
 	s.htlcSwitch.Start()
 
 	alicePeer := &peer{