12349698563

Committed 16 Dec 2024 09:29AM UTC coverage: 58.55% (-0.09%) from 58.636%

Build # 12349698563

Build Type

Pull #9357

github

Committed by

GeorgeTsagk

Commit Message

contractcourt: include custom records on replayed htlc

When notifying the invoice registry for an exit hop htlc we also want to
include its custom records. The channelLink, the other caller of this
method, already populates this field. So we make sure the contest
resolver does so too.

Pull Request Pull Request #9357: contractcourt: include custom records on replayed htlc

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

262 existing lines in 24 files now uncovered.

134243 of 229278 relevant lines covered (58.55%)

19277.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.71

/lnwallet/sigpool.go

package lnwallet

import (
        "fmt"
        "sync"

        "github.com/btcsuite/btcd/btcec/v2"
        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/input"
        "github.com/lightningnetwork/lnd/lnwire"
)

const (
        // jobBuffer is a constant the represents the buffer of jobs in the two
        // main queues. This allows clients avoid necessarily blocking when
        // submitting jobs into the queue.
        jobBuffer = 100

        // TODO(roasbeef): job buffer pool?
)

// VerifyJob is a job sent to the sigPool sig pool to verify a signature
// on a transaction. The items contained in the struct are necessary and
// sufficient to verify the full signature. The passed sigHash closure function
// should be set to a function that generates the relevant sighash.
//
// TODO(roasbeef): when we move to ecschnorr, make into batch signature
// verification using bos-coster (or pip?).
type VerifyJob struct {
        // PubKey is the public key that was used to generate the purported
        // valid signature. Note that with the current channel construction,
        // this public key will likely have been tweaked using the current per
        // commitment point for a particular commitment transactions.
        PubKey *btcec.PublicKey

        // Sig is the raw signature generated using the above public key.  This
        // is the signature to be verified.
        Sig input.Signature

        // SigHash is a function closure generates the sighashes that the
        // passed signature is known to have signed.
        SigHash func() ([]byte, error)

        // HtlcIndex is the index of the HTLC from the PoV of the remote
        // party's update log.
        HtlcIndex uint64

        // Cancel is a channel that is closed by the caller if they wish to
        // cancel all pending verification jobs part of a single batch. This
        // channel is closed in the case that a single signature in a batch has
        // been returned as invalid, as there is no need to verify the remainder
        // of the signatures.
        Cancel <-chan struct{}

        // ErrResp is the channel that the result of the signature verification
        // is to be sent over. In the see that the signature is valid, a nil
        // error will be passed. Otherwise, a concrete error detailing the
        // issue will be passed. This channel MUST be buffered.
        ErrResp chan *HtlcIndexErr
}

// HtlcIndexErr is a special type of error that also includes a pointer to the
// original validation job. This error message allows us to craft more detailed
// errors at upper layers.
type HtlcIndexErr struct {
        error

        *VerifyJob
}

// SignJob is a job sent to the sigPool sig pool to generate a valid
// signature according to the passed SignDescriptor for the passed transaction.
// Jobs are intended to be sent in batches in order to parallelize the job of
// generating signatures for a new commitment transaction.
type SignJob struct {
        // SignDesc is intended to be a full populated SignDescriptor which
        // encodes the necessary material (keys, witness script, etc) required
        // to generate a valid signature for the specified input.
        SignDesc input.SignDescriptor

        // Tx is the transaction to be signed. This is required to generate the
        // proper sighash for the input to be signed.
        Tx *wire.MsgTx

        // OutputIndex is the output index of the HTLC on the commitment
        // transaction being signed.
        OutputIndex int32

        // Cancel is a channel that is closed by the caller if they wish to
        // abandon all pending sign jobs part of a single batch. This should
        // never be closed by the validator.
        Cancel <-chan struct{}

        // Resp is the channel that the response to this particular SignJob
        // will be sent over. This channel MUST be buffered.
        //
        // TODO(roasbeef): actually need to allow caller to set, need to retain
        // order mark commit sig as special
        Resp chan SignJobResp
}

// SignJobResp is the response to a sign job. Both channels are to be read in
// order to ensure no unnecessary goroutine blocking occurs. Additionally, both
// channels should be buffered.
type SignJobResp struct {
        // Sig is the generated signature for a particular SignJob In the case
        // of an error during signature generation, then this value sent will
        // be nil.
        Sig lnwire.Sig

        // Err is the error that occurred when executing the specified
        // signature job. In the case that no error occurred, this value will
        // be nil.
        Err error
}

// SigPool is a struct that is meant to allow the current channel state
// machine to parallelize all signature generation and verification. This
// struct is needed as _each_ HTLC when creating a commitment transaction
// requires a signature, and similarly a receiver of a new commitment must
// verify all the HTLC signatures included within the CommitSig message. A pool
// of workers will be maintained by the sigPool. Batches of jobs (either
// to sign or verify) can be sent to the pool of workers which will
// asynchronously perform the specified job.
type SigPool struct {
        started sync.Once
        stopped sync.Once

        signer input.Signer

        verifyJobs chan VerifyJob
        signJobs   chan SignJob

        wg   sync.WaitGroup
        quit chan struct{}

        numWorkers int
}

// NewSigPool creates a new signature pool with the specified number of
// workers. The recommended parameter for the number of works is the number of
// physical CPU cores available on the target machine.
func NewSigPool(numWorkers int, signer input.Signer) *SigPool {
        return &SigPool{
                signer:     signer,
                numWorkers: numWorkers,
                verifyJobs: make(chan VerifyJob, jobBuffer),
                signJobs:   make(chan SignJob, jobBuffer),
                quit:       make(chan struct{}),
        }
}

// Start starts of all goroutines that the sigPool sig pool needs to
// carry out its duties.
func (s *SigPool) Start() error {
        s.started.Do(func() {
                walletLog.Info("SigPool starting")
                for i := 0; i < s.numWorkers; i++ {
                        s.wg.Add(1)
                        go s.poolWorker()
                }
        })
        return nil
}

// Stop signals any active workers carrying out jobs to exit so the sigPool can
// gracefully shutdown.
func (s *SigPool) Stop() error {
        s.stopped.Do(func() {
                close(s.quit)
                s.wg.Wait()
        })
        return nil
}

// poolWorker is the main worker goroutine within the sigPool sig pool.
// Individual batches are distributed amongst each of the active workers. The
// workers then execute the task based on the type of job, and return the
// result back to caller.
func (s *SigPool) poolWorker() {
        defer s.wg.Done()

        for {
                select {

                // We've just received a new signature job. Given the items
                // contained within the message, we'll craft a signature and
                // send the result along with a possible error back to the
                // caller.
                case sigMsg := <-s.signJobs:
                        rawSig, err := s.signer.SignOutputRaw(
                                sigMsg.Tx, &sigMsg.SignDesc,
                        )
                        if err != nil {
                                select {
                                case sigMsg.Resp <- SignJobResp{
                                        Sig: lnwire.Sig{},
                                        Err: err,
                                }:
                                        continue
                                case <-sigMsg.Cancel:
                                        continue
                                case <-s.quit:
                                        return
                                }
                        }

                        // Use the sig mapper to go from the input.Signature
                        // into the serialized lnwire.Sig that we'll send
                        // across the wire.
                        sig, err := lnwire.NewSigFromSignature(rawSig)

                        select {
                        case sigMsg.Resp <- SignJobResp{
                                Sig: sig,
                                Err: err,
                        }:
                        case <-sigMsg.Cancel:
                                continue
                        case <-s.quit:
                                return
                        }

                // We've just received a new verification job from the outside
                // world. We'll attempt to construct the sighash, parse the
                // signature, and finally verify the signature.
                case verifyMsg := <-s.verifyJobs:
                        sigHash, err := verifyMsg.SigHash()
                        if err != nil {
                                select {
                                case verifyMsg.ErrResp <- &HtlcIndexErr{
                                        error:     err,
                                        VerifyJob: &verifyMsg,
                                }:
                                        continue
                                case <-verifyMsg.Cancel:
                                        continue
                                }
                        }

                        rawSig := verifyMsg.Sig

                        if !rawSig.Verify(sigHash, verifyMsg.PubKey) {
                                err := fmt.Errorf("invalid signature "+
                                        "sighash: %x, sig: %x", sigHash,
                                        rawSig.Serialize())

                                select {
                                case verifyMsg.ErrResp <- &HtlcIndexErr{
                                        error:     err,
                                        VerifyJob: &verifyMsg,
                                }:
                                case <-verifyMsg.Cancel:
                                case <-s.quit:
                                        return
                                }
                        } else {
                                select {
                                case verifyMsg.ErrResp <- nil:
                                case <-verifyMsg.Cancel:
                                case <-s.quit:
                                        return
                                }
                        }

                // The sigPool sig pool is exiting, so we will as well.
                case <-s.quit:
                        return
                }
        }
}

// SubmitSignBatch submits a batch of signature jobs to the sigPool.  The
// response and cancel channels for each of the SignJob's are expected to be
// fully populated, as the response for each job will be sent over the
// response channel within the job itself.
func (s *SigPool) SubmitSignBatch(signJobs []SignJob) {
        for _, job := range signJobs {
                select {
                case s.signJobs <- job:
                case <-job.Cancel:
                        // TODO(roasbeef): return error?
                case <-s.quit:
                        return
                }
        }
}

// SubmitVerifyBatch submits a batch of verification jobs to the sigPool. For
// each job submitted, an error will be passed into the returned channel
// denoting if signature verification was valid or not. The passed cancelChan
// allows the caller to cancel all pending jobs in the case that they wish to
// bail early.
func (s *SigPool) SubmitVerifyBatch(verifyJobs []VerifyJob,
        cancelChan chan struct{}) <-chan *HtlcIndexErr {

        errChan := make(chan *HtlcIndexErr, len(verifyJobs))

        for _, job := range verifyJobs {
                job.Cancel = cancelChan
                job.ErrResp = errChan

                select {
                case s.verifyJobs <- job:
                case <-job.Cancel:
                        return errChan
                }
        }

        return errChan
}

1	package lnwallet
2
3	import (
4	"fmt"
5	"sync"
6
7	"github.com/btcsuite/btcd/btcec/v2"
8	"github.com/btcsuite/btcd/wire"
9	"github.com/lightningnetwork/lnd/input"
10	"github.com/lightningnetwork/lnd/lnwire"
11	)
12
13	const (
14	// jobBuffer is a constant the represents the buffer of jobs in the two
15	// main queues. This allows clients avoid necessarily blocking when
16	// submitting jobs into the queue.
17	jobBuffer = 100
18
19	// TODO(roasbeef): job buffer pool?
20	)
21
22	// VerifyJob is a job sent to the sigPool sig pool to verify a signature
23	// on a transaction. The items contained in the struct are necessary and
24	// sufficient to verify the full signature. The passed sigHash closure function
25	// should be set to a function that generates the relevant sighash.
26	//
27	// TODO(roasbeef): when we move to ecschnorr, make into batch signature
28	// verification using bos-coster (or pip?).
29	type VerifyJob struct {
30	// PubKey is the public key that was used to generate the purported
31	// valid signature. Note that with the current channel construction,
32	// this public key will likely have been tweaked using the current per
33	// commitment point for a particular commitment transactions.
34	PubKey *btcec.PublicKey
35
36	// Sig is the raw signature generated using the above public key. This
37	// is the signature to be verified.
38	Sig input.Signature
39
40	// SigHash is a function closure generates the sighashes that the
41	// passed signature is known to have signed.
42	SigHash func() ([]byte, error)
43
44	// HtlcIndex is the index of the HTLC from the PoV of the remote
45	// party's update log.
46	HtlcIndex uint64
47
48	// Cancel is a channel that is closed by the caller if they wish to
49	// cancel all pending verification jobs part of a single batch. This
50	// channel is closed in the case that a single signature in a batch has
51	// been returned as invalid, as there is no need to verify the remainder
52	// of the signatures.
53	Cancel <-chan struct{}
54
55	// ErrResp is the channel that the result of the signature verification
56	// is to be sent over. In the see that the signature is valid, a nil
57	// error will be passed. Otherwise, a concrete error detailing the
58	// issue will be passed. This channel MUST be buffered.
59	ErrResp chan *HtlcIndexErr
60	}
61
62	// HtlcIndexErr is a special type of error that also includes a pointer to the
63	// original validation job. This error message allows us to craft more detailed
64	// errors at upper layers.
65	type HtlcIndexErr struct {
66	error
67
68	*VerifyJob
69	}
70
71	// SignJob is a job sent to the sigPool sig pool to generate a valid
72	// signature according to the passed SignDescriptor for the passed transaction.
73	// Jobs are intended to be sent in batches in order to parallelize the job of
74	// generating signatures for a new commitment transaction.
75	type SignJob struct {
76	// SignDesc is intended to be a full populated SignDescriptor which
77	// encodes the necessary material (keys, witness script, etc) required
78	// to generate a valid signature for the specified input.
79	SignDesc input.SignDescriptor
80
81	// Tx is the transaction to be signed. This is required to generate the
82	// proper sighash for the input to be signed.
83	Tx *wire.MsgTx
84
85	// OutputIndex is the output index of the HTLC on the commitment
86	// transaction being signed.
87	OutputIndex int32
88
89	// Cancel is a channel that is closed by the caller if they wish to
90	// abandon all pending sign jobs part of a single batch. This should
91	// never be closed by the validator.
92	Cancel <-chan struct{}
93
94	// Resp is the channel that the response to this particular SignJob
95	// will be sent over. This channel MUST be buffered.
96	//
97	// TODO(roasbeef): actually need to allow caller to set, need to retain
98	// order mark commit sig as special
99	Resp chan SignJobResp
100	}
101
102	// SignJobResp is the response to a sign job. Both channels are to be read in
103	// order to ensure no unnecessary goroutine blocking occurs. Additionally, both
104	// channels should be buffered.
105	type SignJobResp struct {
106	// Sig is the generated signature for a particular SignJob In the case
107	// of an error during signature generation, then this value sent will
108	// be nil.
109	Sig lnwire.Sig
110
111	// Err is the error that occurred when executing the specified
112	// signature job. In the case that no error occurred, this value will
113	// be nil.
114	Err error
115	}
116
117	// SigPool is a struct that is meant to allow the current channel state
118	// machine to parallelize all signature generation and verification. This
119	// struct is needed as _each_ HTLC when creating a commitment transaction
120	// requires a signature, and similarly a receiver of a new commitment must
121	// verify all the HTLC signatures included within the CommitSig message. A pool
122	// of workers will be maintained by the sigPool. Batches of jobs (either
123	// to sign or verify) can be sent to the pool of workers which will
124	// asynchronously perform the specified job.
125	type SigPool struct {
126	started sync.Once
127	stopped sync.Once
128
129	signer input.Signer
130
131	verifyJobs chan VerifyJob
132	signJobs chan SignJob
133
134	wg sync.WaitGroup
135	quit chan struct{}
136
137	numWorkers int
138	}
139
140	// NewSigPool creates a new signature pool with the specified number of
141	// workers. The recommended parameter for the number of works is the number of
142	// physical CPU cores available on the target machine.
143	func NewSigPool(numWorkers int, signer input.Signer) *SigPool {	631✔
144	return &SigPool{	631✔
145	signer: signer,	631✔
146	numWorkers: numWorkers,	631✔
147	verifyJobs: make(chan VerifyJob, jobBuffer),	631✔
148	signJobs: make(chan SignJob, jobBuffer),	631✔
149	quit: make(chan struct{}),	631✔
150	}	631✔
151	}	631✔
152
153	// Start starts of all goroutines that the sigPool sig pool needs to
154	// carry out its duties.
155	func (s *SigPool) Start() error {	631✔
156	s.started.Do(func() {	1,262✔
157	walletLog.Info("SigPool starting")	631✔
158	for i := 0; i < s.numWorkers; i++ {	1,940✔
159	s.wg.Add(1)	1,309✔
160	go s.poolWorker()	1,309✔
161	}	1,309✔
162	})
163	return nil	631✔
164	}
165
166	// Stop signals any active workers carrying out jobs to exit so the sigPool can
167	// gracefully shutdown.
168	func (s *SigPool) Stop() error {	405✔
169	s.stopped.Do(func() {	810✔
170	close(s.quit)	405✔
171	s.wg.Wait()	405✔
172	})	405✔
173	return nil	405✔
174	}
175
176	// poolWorker is the main worker goroutine within the sigPool sig pool.
177	// Individual batches are distributed amongst each of the active workers. The
178	// workers then execute the task based on the type of job, and return the
179	// result back to caller.
180	func (s *SigPool) poolWorker() {	1,309✔
181	defer s.wg.Done()	1,309✔
182		1,309✔
183	for {	10,525✔
184	select {	9,216✔
185
186	// We've just received a new signature job. Given the items
187	// contained within the message, we'll craft a signature and
188	// send the result along with a possible error back to the
189	// caller.
190	case sigMsg := <-s.signJobs:	3,984✔
191	rawSig, err := s.signer.SignOutputRaw(	3,984✔
192	sigMsg.Tx, &sigMsg.SignDesc,	3,984✔
193	)	3,984✔
194	if err != nil {	3,984✔
195	select {	×
196	case sigMsg.Resp <- SignJobResp{
197	Sig: lnwire.Sig{},
198	Err: err,
199	}:	×
200	continue	×
201	case <-sigMsg.Cancel:	×
202	continue	×
203	case <-s.quit:	×
204	return	×
205	}
206	}
207
208	// Use the sig mapper to go from the input.Signature
209	// into the serialized lnwire.Sig that we'll send
210	// across the wire.
211	sig, err := lnwire.NewSigFromSignature(rawSig)	3,984✔
212		3,984✔
213	select {	3,984✔
214	case sigMsg.Resp <- SignJobResp{
215	Sig: sig,
216	Err: err,
217	}:	3,983✔
UNCOV 218	case <-sigMsg.Cancel:	×
UNCOV 219	continue	×
220	case <-s.quit:	1✔
221	return	1✔
222	}
223
224	// We've just received a new verification job from the outside
225	// world. We'll attempt to construct the sighash, parse the
226	// signature, and finally verify the signature.
227	case verifyMsg := <-s.verifyJobs:	3,930✔
228	sigHash, err := verifyMsg.SigHash()	3,930✔
229	if err != nil {	3,930✔
230	select {	×
231	case verifyMsg.ErrResp <- &HtlcIndexErr{
232	error: err,
233	VerifyJob: &verifyMsg,
234	}:	×
235	continue	×
236	case <-verifyMsg.Cancel:	×
237	continue	×
238	}
239	}
240
241	rawSig := verifyMsg.Sig	3,930✔
242		3,930✔
243	if !rawSig.Verify(sigHash, verifyMsg.PubKey) {	3,930✔
244	err := fmt.Errorf("invalid signature "+	×
245	"sighash: %x, sig: %x", sigHash,	×
246	rawSig.Serialize())	×
247		×
248	select {	×
249	case verifyMsg.ErrResp <- &HtlcIndexErr{
250	error: err,
251	VerifyJob: &verifyMsg,
252	}:	×
253	case <-verifyMsg.Cancel:	×
254	case <-s.quit:	×
255	return	×
256	}
257	} else {	3,930✔
258	select {	3,930✔
259	case verifyMsg.ErrResp <- nil:	3,930✔
260	case <-verifyMsg.Cancel:	×
261	case <-s.quit:	×
262	return	×
263	}
264	}
265
266	// The sigPool sig pool is exiting, so we will as well.
267	case <-s.quit:	404✔
268	return	404✔
269	}
270	}
271	}
272
273	// SubmitSignBatch submits a batch of signature jobs to the sigPool. The
274	// response and cancel channels for each of the SignJob's are expected to be
275	// fully populated, as the response for each job will be sent over the
276	// response channel within the job itself.
277	func (s *SigPool) SubmitSignBatch(signJobs []SignJob) {	2,055✔
278	for _, job := range signJobs {	6,048✔
279	select {	3,993✔
280	case s.signJobs <- job:	3,993✔
281	case <-job.Cancel:	×
282	// TODO(roasbeef): return error?
283	case <-s.quit:	×
284	return	×
285	}
286	}
287	}
288
289	// SubmitVerifyBatch submits a batch of verification jobs to the sigPool. For
290	// each job submitted, an error will be passed into the returned channel
291	// denoting if signature verification was valid or not. The passed cancelChan
292	// allows the caller to cancel all pending jobs in the case that they wish to
293	// bail early.
294	func (s *SigPool) SubmitVerifyBatch(verifyJobs []VerifyJob,
295	cancelChan chan struct{}) <-chan *HtlcIndexErr {	2,027✔
296		2,027✔
297	errChan := make(chan *HtlcIndexErr, len(verifyJobs))	2,027✔
298		2,027✔
299	for _, job := range verifyJobs {	5,957✔
300	job.Cancel = cancelChan	3,930✔
301	job.ErrResp = errChan	3,930✔
302		3,930✔
303	select {	3,930✔
304	case s.verifyJobs <- job:	3,930✔
305	case <-job.Cancel:	×
306	return errChan	×
307	}
308	}
309
310	return errChan	2,027✔
311	}

lightningnetwork / lnd / 12349698563

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous