12312390362

Committed 13 Dec 2024 08:44AM UTC coverage: 57.458% (+8.5%) from 48.92%

Build # 12312390362

Build Type

Pull #9343

github

Committed by

ellemouton

Commit Message

fn: rework the ContextGuard and add tests

In this commit, the ContextGuard struct is re-worked such that the
context that its new main WithCtx method provides is cancelled in sync
with a parent context being cancelled or with it's quit channel being
cancelled. Tests are added to assert the behaviour. In order for the
close of the quit channel to be consistent with the cancelling of the
derived context, the quit channel _must_ be contained internal to the
ContextGuard so that callers are only able to close the channel via the
exposed Quit method which will then take care to first cancel any
derived context that depend on the quit channel before returning.

Pull Request Pull Request #9343: fn: expand the ContextGuard and add tests

Run Details

101853 of 177264 relevant lines covered (57.46%)

24972.93 hits per line

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

74.26

/invoices/modification_interceptor.go

package invoices

import (
        "errors"
        "fmt"
        "sync/atomic"

        "github.com/lightningnetwork/lnd/fn/v2"
)

var (
        // ErrInterceptorClientAlreadyConnected is an error that is returned
        // when a client tries to connect to the interceptor service while
        // another client is already connected.
        ErrInterceptorClientAlreadyConnected = errors.New(
                "interceptor client already connected",
        )

        // ErrInterceptorClientDisconnected is an error that is returned when
        // the client disconnects during an interceptor session.
        ErrInterceptorClientDisconnected = errors.New(
                "interceptor client disconnected",
        )
)

// safeCallback is a wrapper around a callback function that is safe for
// concurrent access.
type safeCallback struct {
        // callback is the actual callback function that is called when an
        // invoice is intercepted. This might be nil if no client is currently
        // connected.
        callback atomic.Pointer[HtlcModifyCallback]
}

// Set atomically sets the callback function. If a callback is already set, an
// error is returned. The returned function can be used to reset the callback to
// nil once the client is done.
func (s *safeCallback) Set(callback HtlcModifyCallback) (func(), error) {
        if !s.callback.CompareAndSwap(nil, &callback) {
                return nil, ErrInterceptorClientAlreadyConnected
        }

        return func() {
                s.callback.Store(nil)
        }, nil
}

// IsConnected returns true if a client is currently connected.
func (s *safeCallback) IsConnected() bool {
        return s.callback.Load() != nil
}

// Exec executes the callback function if it is set. If the callback is not set,
// an error is returned.
func (s *safeCallback) Exec(req HtlcModifyRequest) (*HtlcModifyResponse,
        error) {

        callback := s.callback.Load()
        if callback == nil {
                return nil, ErrInterceptorClientDisconnected
        }

        return (*callback)(req)
}

// HtlcModificationInterceptor is a service that intercepts HTLCs that aim to
// settle an invoice, enabling a subscribed client to modify certain aspects of
// those HTLCs.
type HtlcModificationInterceptor struct {
        started atomic.Bool
        stopped atomic.Bool

        // callback is the wrapped client callback function that is called when
        // an invoice is intercepted. This function gives the client the ability
        // to determine how the invoice should be settled.
        callback *safeCallback

        // quit is a channel that is closed when the interceptor is stopped.
        quit chan struct{}
}

// NewHtlcModificationInterceptor creates a new HtlcModificationInterceptor.
func NewHtlcModificationInterceptor() *HtlcModificationInterceptor {
        return &HtlcModificationInterceptor{
                callback: &safeCallback{},
                quit:     make(chan struct{}),
        }
}

// Intercept generates a new intercept session for the given invoice. The call
// blocks until the client has responded to the request or an error occurs. The
// response callback is only called if a session was created in the first place,
// which is only the case if a client is registered.
func (s *HtlcModificationInterceptor) Intercept(clientRequest HtlcModifyRequest,
        responseCallback func(HtlcModifyResponse)) error {

        // If there is no client callback set we will not handle the invoice
        // further.
        if !s.callback.IsConnected() {
                log.Debugf("Not intercepting invoice with circuit key %v, no "+
                        "intercept client connected",
                        clientRequest.ExitHtlcCircuitKey)

                return nil
        }

        // We'll block until the client has responded to the request or an error
        // occurs.
        var (
                responseChan = make(chan *HtlcModifyResponse, 1)
                errChan      = make(chan error, 1)
        )

        // The callback function will block at the client's discretion. We will
        // therefore execute it in a separate goroutine. We don't need a wait
        // group because we wait for the response directly below. The caller
        // needs to make sure they don't block indefinitely, by selecting on the
        // quit channel they receive when registering the callback.
        go func() {
                log.Debugf("Waiting for client response from invoice HTLC "+
                        "interceptor session with circuit key %v",
                        clientRequest.ExitHtlcCircuitKey)

                // By this point, we've already checked that the client callback
                // is set. However, if the client disconnected since that check
                // then Exec will return an error.
                result, err := s.callback.Exec(clientRequest)
                if err != nil {
                        _ = fn.SendOrQuit(errChan, err, s.quit)

                        return
                }

                _ = fn.SendOrQuit(responseChan, result, s.quit)
        }()

        // Wait for the client to respond or an error to occur.
        select {
        case response := <-responseChan:
                log.Debugf("Received invoice HTLC interceptor response: %v",
                        response)

                responseCallback(*response)

                return nil

        case err := <-errChan:
                log.Errorf("Error from invoice HTLC interceptor session: %v",
                        err)

                return err

        case <-s.quit:
                return ErrInterceptorClientDisconnected
        }
}

// RegisterInterceptor sets the client callback function that will be called
// when an invoice is intercepted. If a callback is already set, an error is
// returned. The returned function must be used to reset the callback to nil
// once the client is done or disconnects.
func (s *HtlcModificationInterceptor) RegisterInterceptor(
        callback HtlcModifyCallback) (func(), <-chan struct{}, error) {

        done, err := s.callback.Set(callback)
        return done, s.quit, err
}

// Start starts the service.
func (s *HtlcModificationInterceptor) Start() error {
        log.Info("HtlcModificationInterceptor starting...")

        if !s.started.CompareAndSwap(false, true) {
                return fmt.Errorf("HtlcModificationInterceptor started more" +
                        "than once")
        }

        log.Debugf("HtlcModificationInterceptor started")

        return nil
}

// Stop stops the service.
func (s *HtlcModificationInterceptor) Stop() error {
        log.Info("HtlcModificationInterceptor stopping...")

        if !s.stopped.CompareAndSwap(false, true) {
                return fmt.Errorf("HtlcModificationInterceptor stopped more" +
                        "than once")
        }

        close(s.quit)

        log.Debug("HtlcModificationInterceptor stopped")

        return nil
}

// Ensure that HtlcModificationInterceptor implements the HtlcInterceptor and
// HtlcModifier interfaces.
var _ HtlcInterceptor = (*HtlcModificationInterceptor)(nil)
var _ HtlcModifier = (*HtlcModificationInterceptor)(nil)

1	package invoices
2
3	import (
4	"errors"
5	"fmt"
6	"sync/atomic"
7
8	"github.com/lightningnetwork/lnd/fn/v2"
9	)
10
11	var (
12	// ErrInterceptorClientAlreadyConnected is an error that is returned
13	// when a client tries to connect to the interceptor service while
14	// another client is already connected.
15	ErrInterceptorClientAlreadyConnected = errors.New(
16	"interceptor client already connected",
17	)
18
19	// ErrInterceptorClientDisconnected is an error that is returned when
20	// the client disconnects during an interceptor session.
21	ErrInterceptorClientDisconnected = errors.New(
22	"interceptor client disconnected",
23	)
24	)
25
26	// safeCallback is a wrapper around a callback function that is safe for
27	// concurrent access.
28	type safeCallback struct {
29	// callback is the actual callback function that is called when an
30	// invoice is intercepted. This might be nil if no client is currently
31	// connected.
32	callback atomic.Pointer[HtlcModifyCallback]
33	}
34
35	// Set atomically sets the callback function. If a callback is already set, an
36	// error is returned. The returned function can be used to reset the callback to
37	// nil once the client is done.
38	func (s *safeCallback) Set(callback HtlcModifyCallback) (func(), error) {	3✔
39	if !s.callback.CompareAndSwap(nil, &callback) {	4✔
40	return nil, ErrInterceptorClientAlreadyConnected	1✔
41	}	1✔
42
43	return func() {	4✔
44	s.callback.Store(nil)	2✔
45	}, nil	2✔
46	}
47
48	// IsConnected returns true if a client is currently connected.
49	func (s *safeCallback) IsConnected() bool {	3✔
50	return s.callback.Load() != nil	3✔
51	}	3✔
52
53	// Exec executes the callback function if it is set. If the callback is not set,
54	// an error is returned.
55	func (s safeCallback) Exec(req HtlcModifyRequest) (HtlcModifyResponse,
56	error) {	2✔
57		2✔
58	callback := s.callback.Load()	2✔
59	if callback == nil {	2✔
60	return nil, ErrInterceptorClientDisconnected	×
61	}	×
62
63	return (*callback)(req)	2✔
64	}
65
66	// HtlcModificationInterceptor is a service that intercepts HTLCs that aim to
67	// settle an invoice, enabling a subscribed client to modify certain aspects of
68	// those HTLCs.
69	type HtlcModificationInterceptor struct {
70	started atomic.Bool
71	stopped atomic.Bool
72
73	// callback is the wrapped client callback function that is called when
74	// an invoice is intercepted. This function gives the client the ability
75	// to determine how the invoice should be settled.
76	callback *safeCallback
77
78	// quit is a channel that is closed when the interceptor is stopped.
79	quit chan struct{}
80	}
81
82	// NewHtlcModificationInterceptor creates a new HtlcModificationInterceptor.
83	func NewHtlcModificationInterceptor() *HtlcModificationInterceptor {	1✔
84	return &HtlcModificationInterceptor{	1✔
85	callback: &safeCallback{},	1✔
86	quit: make(chan struct{}),	1✔
87	}	1✔
88	}	1✔
89
90	// Intercept generates a new intercept session for the given invoice. The call
91	// blocks until the client has responded to the request or an error occurs. The
92	// response callback is only called if a session was created in the first place,
93	// which is only the case if a client is registered.
94	func (s *HtlcModificationInterceptor) Intercept(clientRequest HtlcModifyRequest,
95	responseCallback func(HtlcModifyResponse)) error {	3✔
96		3✔
97	// If there is no client callback set we will not handle the invoice	3✔
98	// further.	3✔
99	if !s.callback.IsConnected() {	4✔
100	log.Debugf("Not intercepting invoice with circuit key %v, no "+	1✔
101	"intercept client connected",	1✔
102	clientRequest.ExitHtlcCircuitKey)	1✔
103		1✔
104	return nil	1✔
105	}	1✔
106
107	// We'll block until the client has responded to the request or an error
108	// occurs.
109	var (	2✔
110	responseChan = make(chan *HtlcModifyResponse, 1)	2✔
111	errChan = make(chan error, 1)	2✔
112	)	2✔
113		2✔
114	// The callback function will block at the client's discretion. We will	2✔
115	// therefore execute it in a separate goroutine. We don't need a wait	2✔
116	// group because we wait for the response directly below. The caller	2✔
117	// needs to make sure they don't block indefinitely, by selecting on the	2✔
118	// quit channel they receive when registering the callback.	2✔
119	go func() {	4✔
120	log.Debugf("Waiting for client response from invoice HTLC "+	2✔
121	"interceptor session with circuit key %v",	2✔
122	clientRequest.ExitHtlcCircuitKey)	2✔
123		2✔
124	// By this point, we've already checked that the client callback	2✔
125	// is set. However, if the client disconnected since that check	2✔
126	// then Exec will return an error.	2✔
127	result, err := s.callback.Exec(clientRequest)	2✔
128	if err != nil {	3✔
129	_ = fn.SendOrQuit(errChan, err, s.quit)	1✔
130		1✔
131	return	1✔
132	}	1✔
133
134	_ = fn.SendOrQuit(responseChan, result, s.quit)	1✔
135	}()
136
137	// Wait for the client to respond or an error to occur.
138	select {	2✔
139	case response := <-responseChan:	1✔
140	log.Debugf("Received invoice HTLC interceptor response: %v",	1✔
141	response)	1✔
142		1✔
143	responseCallback(*response)	1✔
144		1✔
145	return nil	1✔
146
147	case err := <-errChan:	1✔
148	log.Errorf("Error from invoice HTLC interceptor session: %v",	1✔
149	err)	1✔
150		1✔
151	return err	1✔
152
153	case <-s.quit:	×
154	return ErrInterceptorClientDisconnected	×
155	}
156	}
157
158	// RegisterInterceptor sets the client callback function that will be called
159	// when an invoice is intercepted. If a callback is already set, an error is
160	// returned. The returned function must be used to reset the callback to nil
161	// once the client is done or disconnects.
162	func (s *HtlcModificationInterceptor) RegisterInterceptor(
163	callback HtlcModifyCallback) (func(), <-chan struct{}, error) {	3✔
164		3✔
165	done, err := s.callback.Set(callback)	3✔
166	return done, s.quit, err	3✔
167	}	3✔
168
169	// Start starts the service.
170	func (s *HtlcModificationInterceptor) Start() error {	×
171	log.Info("HtlcModificationInterceptor starting...")	×
172		×
173	if !s.started.CompareAndSwap(false, true) {	×
174	return fmt.Errorf("HtlcModificationInterceptor started more" +	×
175	"than once")	×
176	}	×
177
178	log.Debugf("HtlcModificationInterceptor started")	×
179		×
180	return nil	×
181	}
182
183	// Stop stops the service.
184	func (s *HtlcModificationInterceptor) Stop() error {	×
185	log.Info("HtlcModificationInterceptor stopping...")	×
186		×
187	if !s.stopped.CompareAndSwap(false, true) {	×
188	return fmt.Errorf("HtlcModificationInterceptor stopped more" +	×
189	"than once")	×
190	}	×
191
192	close(s.quit)	×
193		×
194	log.Debug("HtlcModificationInterceptor stopped")	×
195		×
196	return nil	×
197	}
198
199	// Ensure that HtlcModificationInterceptor implements the HtlcInterceptor and
200	// HtlcModifier interfaces.
201	var _ HtlcInterceptor = (*HtlcModificationInterceptor)(nil)
202	var _ HtlcModifier = (*HtlcModificationInterceptor)(nil)

lightningnetwork / lnd / 12312390362

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