14227452255

Committed 02 Apr 2025 07:01PM UTC coverage: 69.045% (+10.4%) from 58.637%

Build # 14227452255

Build Type

Pull #9356

github

Committed by

web-flow

Commit Message

Merge 12f979a13 into 6a3845b79

Pull Request Pull Request #9356: lnrpc: add incoming/outgoing channel ids filter to forwarding history request

Run Details

43 of 61 new or added lines in 3 files covered. (70.49%)

36 existing lines in 11 files now uncovered.

133471 of 193310 relevant lines covered (69.05%)

22169.88 hits per line

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

92.35

/channeldb/forwarding_log.go

package channeldb

import (
        "bytes"
        "io"
        "sort"
        "time"

        "github.com/btcsuite/btcwallet/walletdb"
        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lnwire"
)

var (
        // forwardingLogBucket is the bucket that we'll use to store the
        // forwarding log. The forwarding log contains a time series database
        // of the forwarding history of a lightning daemon. Each key within the
        // bucket is a timestamp (in nano seconds since the unix epoch), and
        // the value a slice of a forwarding event for that timestamp.
        forwardingLogBucket = []byte("circuit-fwd-log")
)

const (
        // forwardingEventSize is the size of a forwarding event. The breakdown
        // is as follows:
        //
        //  * 8 byte incoming chan ID || 8 byte outgoing chan ID || 8 byte value in
        //    || 8 byte value out
        //
        // From the value in and value out, callers can easily compute the
        // total fee extract from a forwarding event.
        forwardingEventSize = 32

        // MaxResponseEvents is the max number of forwarding events that will
        // be returned by a single query response. This size was selected to
        // safely remain under gRPC's 4MiB message size response limit. As each
        // full forwarding event (including the timestamp) is 40 bytes, we can
        // safely return 50k entries in a single response.
        MaxResponseEvents = 50000
)

// ForwardingLog returns an instance of the ForwardingLog object backed by the
// target database instance.
func (d *DB) ForwardingLog() *ForwardingLog {
        return &ForwardingLog{
                db: d,
        }
}

// ForwardingLog is a time series database that logs the fulfillment of payment
// circuits by a lightning network daemon. The log contains a series of
// forwarding events which map a timestamp to a forwarding event. A forwarding
// event describes which channels were used to create+settle a circuit, and the
// amount involved. Subtracting the outgoing amount from the incoming amount
// reveals the fee charged for the forwarding service.
type ForwardingLog struct {
        db *DB
}

// ForwardingEvent is an event in the forwarding log's time series. Each
// forwarding event logs the creation and tear-down of a payment circuit. A
// circuit is created once an incoming HTLC has been fully forwarded, and
// destroyed once the payment has been settled.
type ForwardingEvent struct {
        // Timestamp is the settlement time of this payment circuit.
        Timestamp time.Time

        // IncomingChanID is the incoming channel ID of the payment circuit.
        IncomingChanID lnwire.ShortChannelID

        // OutgoingChanID is the outgoing channel ID of the payment circuit.
        OutgoingChanID lnwire.ShortChannelID

        // AmtIn is the amount of the incoming HTLC. Subtracting this from the
        // outgoing amount gives the total fees of this payment circuit.
        AmtIn lnwire.MilliSatoshi

        // AmtOut is the amount of the outgoing HTLC. Subtracting the incoming
        // amount from this gives the total fees for this payment circuit.
        AmtOut lnwire.MilliSatoshi
}

// encodeForwardingEvent writes out the target forwarding event to the passed
// io.Writer, using the expected DB format. Note that the timestamp isn't
// serialized as this will be the key value within the bucket.
func encodeForwardingEvent(w io.Writer, f *ForwardingEvent) error {
        return WriteElements(
                w, f.IncomingChanID, f.OutgoingChanID, f.AmtIn, f.AmtOut,
        )
}

// decodeForwardingEvent attempts to decode the raw bytes of a serialized
// forwarding event into the target ForwardingEvent. Note that the timestamp
// won't be decoded, as the caller is expected to set this due to the bucket
// structure of the forwarding log.
func decodeForwardingEvent(r io.Reader, f *ForwardingEvent) error {
        return ReadElements(
                r, &f.IncomingChanID, &f.OutgoingChanID, &f.AmtIn, &f.AmtOut,
        )
}

// AddForwardingEvents adds a series of forwarding events to the database.
// Before inserting, the set of events will be sorted according to their
// timestamp. This ensures that all writes to disk are sequential.
func (f *ForwardingLog) AddForwardingEvents(events []ForwardingEvent) error {
        // Before we create the database transaction, we'll ensure that the set
        // of forwarding events are properly sorted according to their
        // timestamp and that no duplicate timestamps exist to avoid collisions
        // in the key we are going to store the events under.
        makeUniqueTimestamps(events)

        var timestamp [8]byte

        return kvdb.Batch(f.db.Backend, func(tx kvdb.RwTx) error {
                // First, we'll fetch the bucket that stores our time series
                // log.
                logBucket, err := tx.CreateTopLevelBucket(
                        forwardingLogBucket,
                )
                if err != nil {
                        return err
                }

                // With the bucket obtained, we can now begin to write out the
                // series of events.
                for _, event := range events {
                        err := storeEvent(logBucket, event, timestamp[:])
                        if err != nil {
                                return err
                        }
                }

                return nil
        })
}

// storeEvent tries to store a forwarding event into the given bucket by trying
// to avoid collisions. If a key for the event timestamp already exists in the
// database, the timestamp is incremented in nanosecond intervals until a "free"
// slot is found.
func storeEvent(bucket walletdb.ReadWriteBucket, event ForwardingEvent,
        timestampScratchSpace []byte) error {

        // First, we'll serialize this timestamp into our
        // timestamp buffer.
        byteOrder.PutUint64(
                timestampScratchSpace, uint64(event.Timestamp.UnixNano()),
        )

        // Next we'll loop until we find a "free" slot in the bucket to store
        // the event under. This should almost never happen unless we're running
        // on a system that has a very bad system clock that doesn't properly
        // resolve to nanosecond scale. We try up to 100 times (which would come
        // to a maximum shift of 0.1 microsecond which is acceptable for most
        // use cases). If we don't find a free slot, we just give up and let
        // the collision happen. Something must be wrong with the data in that
        // case, even on a very fast machine forwarding payments _will_ take a
        // few microseconds at least so we should find a nanosecond slot
        // somewhere.
        const maxTries = 100
        tries := 0
        for tries < maxTries {
                val := bucket.Get(timestampScratchSpace)
                if val == nil {
                        break
                }

                // Collision, try the next nanosecond timestamp.
                nextNano := event.Timestamp.UnixNano() + 1
                event.Timestamp = time.Unix(0, nextNano)
                byteOrder.PutUint64(timestampScratchSpace, uint64(nextNano))
                tries++
        }

        // With the key encoded, we'll then encode the event
        // into our buffer, then write it out to disk.
        var eventBytes [forwardingEventSize]byte
        eventBuf := bytes.NewBuffer(eventBytes[0:0:forwardingEventSize])
        err := encodeForwardingEvent(eventBuf, &event)
        if err != nil {
                return err
        }
        return bucket.Put(timestampScratchSpace, eventBuf.Bytes())
}

// ForwardingEventQuery represents a query to the forwarding log payment
// circuit time series database. The query allows a caller to retrieve all
// records for a particular time slice, offset in that time slice, limiting the
// total number of responses returned.
type ForwardingEventQuery struct {
        // StartTime is the start time of the time slice.
        StartTime time.Time

        // EndTime is the end time of the time slice.
        EndTime time.Time

        // IndexOffset is the offset within the time slice to start at. This
        // can be used to start the response at a particular record.
        IndexOffset uint32

        // NumMaxEvents is the max number of events to return.
        NumMaxEvents uint32

        // IncomingChanIds is the list of channels to filter HTLCs being
        // received from a particular channel.
        // If the list is empty, then it is ignored.
        IncomingChanIDs fn.Set[uint64]

        // OutgoingChanIds is the list of channels to filter HTLCs being
        // forwarded to a particular channel.
        // If the list is empty, then it is ignored.
        OutgoingChanIDs fn.Set[uint64]
}

// ForwardingLogTimeSlice is the response to a forwarding query. It includes
// the original query, the set  events that match the query, and an integer
// which represents the offset index of the last item in the set of returned
// events. This integer allows callers to resume their query using this offset
// in the event that the query's response exceeds the max number of returnable
// events.
type ForwardingLogTimeSlice struct {
        ForwardingEventQuery

        // ForwardingEvents is the set of events in our time series that answer
        // the query embedded above.
        ForwardingEvents []ForwardingEvent

        // LastIndexOffset is the index of the last element in the set of
        // returned ForwardingEvents above. Callers can use this to resume
        // their query in the event that the time slice has too many events to
        // fit into a single response.
        LastIndexOffset uint32
}

// Query allows a caller to query the forwarding event time series for a
// particular time slice. The caller can control the precise time as well as
// the number of events to be returned.
//
// TODO(roasbeef): rename?
func (f *ForwardingLog) Query(q ForwardingEventQuery) (ForwardingLogTimeSlice, error) {
        var resp ForwardingLogTimeSlice

        // If the user provided an index offset, then we'll not know how many
        // records we need to skip. We'll also keep track of the record offset
        // as that's part of the final return value.
        recordsToSkip := q.IndexOffset
        recordOffset := q.IndexOffset

        err := kvdb.View(f.db, func(tx kvdb.RTx) error {
                // If the bucket wasn't found, then there aren't any events to
                // be returned.
                logBucket := tx.ReadBucket(forwardingLogBucket)
                if logBucket == nil {
                        return ErrNoForwardingEvents
                }

                // We'll be using a cursor to seek into the database, so we'll
                // populate byte slices that represent the start of the key
                // space we're interested in, and the end.
                var startTime, endTime [8]byte
                byteOrder.PutUint64(startTime[:], uint64(q.StartTime.UnixNano()))
                byteOrder.PutUint64(endTime[:], uint64(q.EndTime.UnixNano()))

                // If we know that a set of log events exists, then we'll begin
                // our seek through the log in order to satisfy the query.
                // We'll continue until either we reach the end of the range,
                // or reach our max number of events.
                logCursor := logBucket.ReadCursor()
                timestamp, events := logCursor.Seek(startTime[:])
                for ; timestamp != nil && bytes.Compare(timestamp, endTime[:]) <= 0; timestamp, events = logCursor.Next() {
                        // If our current return payload exceeds the max number
                        // of events, then we'll exit now.
                        if uint32(len(resp.ForwardingEvents)) >= q.NumMaxEvents {
                                return nil
                        }

                        // If no incoming or outgoing channel IDs were provided
                        // and we're not yet past the user defined offset, then
                        // we'll continue to seek forward.
                        if recordsToSkip > 0 &&
                                q.IncomingChanIDs.IsEmpty() &&
                                q.OutgoingChanIDs.IsEmpty() {
                                recordsToSkip--
                                continue
                        }

                        currentTime := time.Unix(
                                0, int64(byteOrder.Uint64(timestamp)),
                        )

                        // At this point, we've skipped enough records to start
                        // to collate our query. For each record, we'll
                        // increment the final record offset so the querier can
                        // utilize pagination to seek further.
                        readBuf := bytes.NewReader(events)
                        for readBuf.Len() != 0 {
                                var event ForwardingEvent
                                err := decodeForwardingEvent(readBuf, &event)
                                if err != nil {
                                        return err
                                }

                                // Check if the incoming channel ID matches the
                                // filter criteria.
                                // Either no filtering is applied (IsEmpty), or
                                // the ID is explicitly included.
                                incomingMatch := q.IncomingChanIDs.IsEmpty() ||
                                        q.IncomingChanIDs.Contains(
                                                event.IncomingChanID.ToUint64(),
                                        )

                                // Check if the outgoing channel ID matches the
                                // filter criteria.
                                // Either no filtering is applied (IsEmpty), or
                                // the ID is explicitly included.
                                outgoingMatch := q.OutgoingChanIDs.IsEmpty() ||
                                        q.OutgoingChanIDs.Contains(
                                                event.OutgoingChanID.ToUint64(),
                                        )

                                // If both conditions are met, then we'll add
                                // the event to our return payload.
                                if incomingMatch && outgoingMatch {
                                        // If we're not yet past the user
                                        // defined offset , then we'll continue
                                        // to seek forward.
                                        if recordsToSkip > 0 {
                                                recordsToSkip--
                                                continue
                                        }

                                        event.Timestamp = currentTime
                                        resp.ForwardingEvents = append(
                                                resp.ForwardingEvents,
                                                event,
                                        )
                                        recordOffset++
                                }
                        }
                }

                return nil
        }, func() {
                resp = ForwardingLogTimeSlice{
                        ForwardingEventQuery: q,
                }
        })
        if err != nil && err != ErrNoForwardingEvents {
                return ForwardingLogTimeSlice{}, err
        }

        resp.LastIndexOffset = recordOffset

        return resp, nil
}

// makeUniqueTimestamps takes a slice of forwarding events, sorts it by the
// event timestamps and then makes sure there are no duplicates in the
// timestamps. If duplicates are found, some of the timestamps are increased on
// the nanosecond scale until only unique values remain. This is a fix to
// address the problem that in some environments (looking at you, Windows) the
// system clock has such a bad resolution that two serial invocations of
// time.Now() might return the same timestamp, even if some time has elapsed
// between the calls.
func makeUniqueTimestamps(events []ForwardingEvent) {
        sort.Slice(events, func(i, j int) bool {
                return events[i].Timestamp.Before(events[j].Timestamp)
        })

        // Now that we know the events are sorted by timestamp, we can go
        // through the list and fix all duplicates until only unique values
        // remain.
        for outer := 0; outer < len(events)-1; outer++ {
                current := events[outer].Timestamp.UnixNano()
                next := events[outer+1].Timestamp.UnixNano()

                // We initially sorted the slice. So if the current is now
                // greater or equal to the next one, it's either because it's a
                // duplicate or because we increased the current in the last
                // iteration.
                if current >= next {
                        next = current + 1
                        events[outer+1].Timestamp = time.Unix(0, next)
                }
        }
}

1	package channeldb
2
3	import (
4	"bytes"
5	"io"
6	"sort"
7	"time"
8
9	"github.com/btcsuite/btcwallet/walletdb"
10	"github.com/lightningnetwork/lnd/fn/v2"
11	"github.com/lightningnetwork/lnd/kvdb"
12	"github.com/lightningnetwork/lnd/lnwire"
13	)
14
15	var (
16	// forwardingLogBucket is the bucket that we'll use to store the
17	// forwarding log. The forwarding log contains a time series database
18	// of the forwarding history of a lightning daemon. Each key within the
19	// bucket is a timestamp (in nano seconds since the unix epoch), and
20	// the value a slice of a forwarding event for that timestamp.
21	forwardingLogBucket = []byte("circuit-fwd-log")
22	)
23
24	const (
25	// forwardingEventSize is the size of a forwarding event. The breakdown
26	// is as follows:
27	//
28	// * 8 byte incoming chan ID \|\| 8 byte outgoing chan ID \|\| 8 byte value in
29	// \|\| 8 byte value out
30	//
31	// From the value in and value out, callers can easily compute the
32	// total fee extract from a forwarding event.
33	forwardingEventSize = 32
34
35	// MaxResponseEvents is the max number of forwarding events that will
36	// be returned by a single query response. This size was selected to
37	// safely remain under gRPC's 4MiB message size response limit. As each
38	// full forwarding event (including the timestamp) is 40 bytes, we can
39	// safely return 50k entries in a single response.
40	MaxResponseEvents = 50000
41	)
42
43	// ForwardingLog returns an instance of the ForwardingLog object backed by the
44	// target database instance.
45	func (d DB) ForwardingLog() ForwardingLog {	3✔
46	return &ForwardingLog{	3✔
47	db: d,	3✔
48	}	3✔
49	}	3✔
50
51	// ForwardingLog is a time series database that logs the fulfillment of payment
52	// circuits by a lightning network daemon. The log contains a series of
53	// forwarding events which map a timestamp to a forwarding event. A forwarding
54	// event describes which channels were used to create+settle a circuit, and the
55	// amount involved. Subtracting the outgoing amount from the incoming amount
56	// reveals the fee charged for the forwarding service.
57	type ForwardingLog struct {
58	db *DB
59	}
60
61	// ForwardingEvent is an event in the forwarding log's time series. Each
62	// forwarding event logs the creation and tear-down of a payment circuit. A
63	// circuit is created once an incoming HTLC has been fully forwarded, and
64	// destroyed once the payment has been settled.
65	type ForwardingEvent struct {
66	// Timestamp is the settlement time of this payment circuit.
67	Timestamp time.Time
68
69	// IncomingChanID is the incoming channel ID of the payment circuit.
70	IncomingChanID lnwire.ShortChannelID
71
72	// OutgoingChanID is the outgoing channel ID of the payment circuit.
73	OutgoingChanID lnwire.ShortChannelID
74
75	// AmtIn is the amount of the incoming HTLC. Subtracting this from the
76	// outgoing amount gives the total fees of this payment circuit.
77	AmtIn lnwire.MilliSatoshi
78
79	// AmtOut is the amount of the outgoing HTLC. Subtracting the incoming
80	// amount from this gives the total fees for this payment circuit.
81	AmtOut lnwire.MilliSatoshi
82	}
83
84	// encodeForwardingEvent writes out the target forwarding event to the passed
85	// io.Writer, using the expected DB format. Note that the timestamp isn't
86	// serialized as this will be the key value within the bucket.
87	func encodeForwardingEvent(w io.Writer, f *ForwardingEvent) error {	383✔
88	return WriteElements(	383✔
89	w, f.IncomingChanID, f.OutgoingChanID, f.AmtIn, f.AmtOut,	383✔
90	)	383✔
91	}	383✔
92
93	// decodeForwardingEvent attempts to decode the raw bytes of a serialized
94	// forwarding event into the target ForwardingEvent. Note that the timestamp
95	// won't be decoded, as the caller is expected to set this due to the bucket
96	// structure of the forwarding log.
97	func decodeForwardingEvent(r io.Reader, f *ForwardingEvent) error {	283✔
98	return ReadElements(	283✔
99	r, &f.IncomingChanID, &f.OutgoingChanID, &f.AmtIn, &f.AmtOut,	283✔
100	)	283✔
101	}	283✔
102
103	// AddForwardingEvents adds a series of forwarding events to the database.
104	// Before inserting, the set of events will be sorted according to their
105	// timestamp. This ensures that all writes to disk are sequential.
106	func (f *ForwardingLog) AddForwardingEvents(events []ForwardingEvent) error {	10✔
107	// Before we create the database transaction, we'll ensure that the set	10✔
108	// of forwarding events are properly sorted according to their	10✔
109	// timestamp and that no duplicate timestamps exist to avoid collisions	10✔
110	// in the key we are going to store the events under.	10✔
111	makeUniqueTimestamps(events)	10✔
112		10✔
113	var timestamp [8]byte	10✔
114		10✔
115	return kvdb.Batch(f.db.Backend, func(tx kvdb.RwTx) error {	20✔
116	// First, we'll fetch the bucket that stores our time series	10✔
117	// log.	10✔
118	logBucket, err := tx.CreateTopLevelBucket(	10✔
119	forwardingLogBucket,	10✔
120	)	10✔
121	if err != nil {	10✔
122	return err	×
123	}	×
124
125	// With the bucket obtained, we can now begin to write out the
126	// series of events.
127	for _, event := range events {	393✔
128	err := storeEvent(logBucket, event, timestamp[:])	383✔
129	if err != nil {	383✔
130	return err	×
131	}	×
132	}
133
134	return nil	10✔
135	})
136	}
137
138	// storeEvent tries to store a forwarding event into the given bucket by trying
139	// to avoid collisions. If a key for the event timestamp already exists in the
140	// database, the timestamp is incremented in nanosecond intervals until a "free"
141	// slot is found.
142	func storeEvent(bucket walletdb.ReadWriteBucket, event ForwardingEvent,
143	timestampScratchSpace []byte) error {	383✔
144		383✔
145	// First, we'll serialize this timestamp into our	383✔
146	// timestamp buffer.	383✔
147	byteOrder.PutUint64(	383✔
148	timestampScratchSpace, uint64(event.Timestamp.UnixNano()),	383✔
149	)	383✔
150		383✔
151	// Next we'll loop until we find a "free" slot in the bucket to store	383✔
152	// the event under. This should almost never happen unless we're running	383✔
153	// on a system that has a very bad system clock that doesn't properly	383✔
154	// resolve to nanosecond scale. We try up to 100 times (which would come	383✔
155	// to a maximum shift of 0.1 microsecond which is acceptable for most	383✔
156	// use cases). If we don't find a free slot, we just give up and let	383✔
157	// the collision happen. Something must be wrong with the data in that	383✔
158	// case, even on a very fast machine forwarding payments _will_ take a	383✔
159	// few microseconds at least so we should find a nanosecond slot	383✔
160	// somewhere.	383✔
161	const maxTries = 100	383✔
162	tries := 0	383✔
163	for tries < maxTries {	1,166✔
164	val := bucket.Get(timestampScratchSpace)	783✔
165	if val == nil {	1,166✔
166	break	383✔
167	}
168
169	// Collision, try the next nanosecond timestamp.
170	nextNano := event.Timestamp.UnixNano() + 1	400✔
171	event.Timestamp = time.Unix(0, nextNano)	400✔
172	byteOrder.PutUint64(timestampScratchSpace, uint64(nextNano))	400✔
173	tries++	400✔
174	}
175
176	// With the key encoded, we'll then encode the event
177	// into our buffer, then write it out to disk.
178	var eventBytes [forwardingEventSize]byte	383✔
179	eventBuf := bytes.NewBuffer(eventBytes[0:0:forwardingEventSize])	383✔
180	err := encodeForwardingEvent(eventBuf, &event)	383✔
181	if err != nil {	383✔
182	return err	×
183	}	×
184	return bucket.Put(timestampScratchSpace, eventBuf.Bytes())	383✔
185	}
186
187	// ForwardingEventQuery represents a query to the forwarding log payment
188	// circuit time series database. The query allows a caller to retrieve all
189	// records for a particular time slice, offset in that time slice, limiting the
190	// total number of responses returned.
191	type ForwardingEventQuery struct {
192	// StartTime is the start time of the time slice.
193	StartTime time.Time
194
195	// EndTime is the end time of the time slice.
196	EndTime time.Time
197
198	// IndexOffset is the offset within the time slice to start at. This
199	// can be used to start the response at a particular record.
200	IndexOffset uint32
201
202	// NumMaxEvents is the max number of events to return.
203	NumMaxEvents uint32
204
205	// IncomingChanIds is the list of channels to filter HTLCs being
206	// received from a particular channel.
207	// If the list is empty, then it is ignored.
208	IncomingChanIDs fn.Set[uint64]
209
210	// OutgoingChanIds is the list of channels to filter HTLCs being
211	// forwarded to a particular channel.
212	// If the list is empty, then it is ignored.
213	OutgoingChanIDs fn.Set[uint64]
214	}
215
216	// ForwardingLogTimeSlice is the response to a forwarding query. It includes
217	// the original query, the set events that match the query, and an integer
218	// which represents the offset index of the last item in the set of returned
219	// events. This integer allows callers to resume their query using this offset
220	// in the event that the query's response exceeds the max number of returnable
221	// events.
222	type ForwardingLogTimeSlice struct {
223	ForwardingEventQuery
224
225	// ForwardingEvents is the set of events in our time series that answer
226	// the query embedded above.
227	ForwardingEvents []ForwardingEvent
228
229	// LastIndexOffset is the index of the last element in the set of
230	// returned ForwardingEvents above. Callers can use this to resume
231	// their query in the event that the time slice has too many events to
232	// fit into a single response.
233	LastIndexOffset uint32
234	}
235
236	// Query allows a caller to query the forwarding event time series for a
237	// particular time slice. The caller can control the precise time as well as
238	// the number of events to be returned.
239	//
240	// TODO(roasbeef): rename?
241	func (f *ForwardingLog) Query(q ForwardingEventQuery) (ForwardingLogTimeSlice, error) {	10✔
242	var resp ForwardingLogTimeSlice	10✔
243		10✔
244	// If the user provided an index offset, then we'll not know how many	10✔
245	// records we need to skip. We'll also keep track of the record offset	10✔
246	// as that's part of the final return value.	10✔
247	recordsToSkip := q.IndexOffset	10✔
248	recordOffset := q.IndexOffset	10✔
249		10✔
250	err := kvdb.View(f.db, func(tx kvdb.RTx) error {	20✔
251	// If the bucket wasn't found, then there aren't any events to	10✔
252	// be returned.	10✔
253	logBucket := tx.ReadBucket(forwardingLogBucket)	10✔
254	if logBucket == nil {	10✔
255	return ErrNoForwardingEvents	×
256	}	×
257
258	// We'll be using a cursor to seek into the database, so we'll
259	// populate byte slices that represent the start of the key
260	// space we're interested in, and the end.
261	var startTime, endTime [8]byte	10✔
262	byteOrder.PutUint64(startTime[:], uint64(q.StartTime.UnixNano()))	10✔
263	byteOrder.PutUint64(endTime[:], uint64(q.EndTime.UnixNano()))	10✔
264		10✔
265	// If we know that a set of log events exists, then we'll begin	10✔
266	// our seek through the log in order to satisfy the query.	10✔
267	// We'll continue until either we reach the end of the range,	10✔
268	// or reach our max number of events.	10✔
269	logCursor := logBucket.ReadCursor()	10✔
270	timestamp, events := logCursor.Seek(startTime[:])	10✔
271	for ; timestamp != nil && bytes.Compare(timestamp, endTime[:]) <= 0; timestamp, events = logCursor.Next() {	305✔
272	// If our current return payload exceeds the max number	295✔
273	// of events, then we'll exit now.	295✔
274	if uint32(len(resp.ForwardingEvents)) >= q.NumMaxEvents {	297✔
275	return nil	2✔
276	}	2✔
277
278	// If no incoming or outgoing channel IDs were provided
279	// and we're not yet past the user defined offset, then
280	// we'll continue to seek forward.
281	if recordsToSkip > 0 &&	293✔
282	q.IncomingChanIDs.IsEmpty() &&	293✔
283	q.OutgoingChanIDs.IsEmpty() {	306✔
284	recordsToSkip--	13✔
285	continue	13✔
286	}
287
288	currentTime := time.Unix(	283✔
289	0, int64(byteOrder.Uint64(timestamp)),	283✔
290	)	283✔
291		283✔
292	// At this point, we've skipped enough records to start	283✔
293	// to collate our query. For each record, we'll	283✔
294	// increment the final record offset so the querier can	283✔
295	// utilize pagination to seek further.	283✔
296	readBuf := bytes.NewReader(events)	283✔
297	for readBuf.Len() != 0 {	566✔
298	var event ForwardingEvent	283✔
299	err := decodeForwardingEvent(readBuf, &event)	283✔
300	if err != nil {	283✔
301	return err	×
302	}	×
303
304	// Check if the incoming channel ID matches the
305	// filter criteria.
306	// Either no filtering is applied (IsEmpty), or
307	// the ID is explicitly included.
308	incomingMatch := q.IncomingChanIDs.IsEmpty() \|\|	283✔
309	q.IncomingChanIDs.Contains(	283✔
310	event.IncomingChanID.ToUint64(),	283✔
311	)	283✔
312		283✔
313	// Check if the outgoing channel ID matches the	283✔
314	// filter criteria.	283✔
315	// Either no filtering is applied (IsEmpty), or	283✔
316	// the ID is explicitly included.	283✔
317	outgoingMatch := q.OutgoingChanIDs.IsEmpty() \|\|	283✔
318	q.OutgoingChanIDs.Contains(	283✔
319	event.OutgoingChanID.ToUint64(),	283✔
320	)	283✔
321		283✔
322	// If both conditions are met, then we'll add	283✔
323	// the event to our return payload.	283✔
324	if incomingMatch && outgoingMatch {	560✔
325	// If we're not yet past the user	277✔
326	// defined offset , then we'll continue	277✔
327	// to seek forward.	277✔
328	if recordsToSkip > 0 {	277✔
NEW 329	recordsToSkip--	×
NEW 330	continue	×
331	}
332
333	event.Timestamp = currentTime	277✔
334	resp.ForwardingEvents = append(	277✔
335	resp.ForwardingEvents,	277✔
336	event,	277✔
337	)	277✔
338	recordOffset++	277✔
339	}
340	}
341	}
342
343	return nil	8✔
344	}, func() {	10✔
345	resp = ForwardingLogTimeSlice{	10✔
346	ForwardingEventQuery: q,	10✔
347	}	10✔
348	})	10✔
349	if err != nil && err != ErrNoForwardingEvents {	10✔
350	return ForwardingLogTimeSlice{}, err	×
351	}	×
352
353	resp.LastIndexOffset = recordOffset	10✔
354		10✔
355	return resp, nil	10✔
356	}
357
358	// makeUniqueTimestamps takes a slice of forwarding events, sorts it by the
359	// event timestamps and then makes sure there are no duplicates in the
360	// timestamps. If duplicates are found, some of the timestamps are increased on
361	// the nanosecond scale until only unique values remain. This is a fix to
362	// address the problem that in some environments (looking at you, Windows) the
363	// system clock has such a bad resolution that two serial invocations of
364	// time.Now() might return the same timestamp, even if some time has elapsed
365	// between the calls.
366	func makeUniqueTimestamps(events []ForwardingEvent) {	11✔
367	sort.Slice(events, func(i, j int) bool {	437✔
368	return events[i].Timestamp.Before(events[j].Timestamp)	426✔
369	})	426✔
370
371	// Now that we know the events are sorted by timestamp, we can go
372	// through the list and fix all duplicates until only unique values
373	// remain.
374	for outer := 0; outer < len(events)-1; outer++ {	394✔
375	current := events[outer].Timestamp.UnixNano()	383✔
376	next := events[outer+1].Timestamp.UnixNano()	383✔
377		383✔
378	// We initially sorted the slice. So if the current is now	383✔
379	// greater or equal to the next one, it's either because it's a	383✔
380	// duplicate or because we increased the current in the last	383✔
381	// iteration.	383✔
382	if current >= next {	388✔
383	next = current + 1	5✔
384	events[outer+1].Timestamp = time.Unix(0, next)	5✔
385	}	5✔
386	}
387	}

lightningnetwork / lnd / 14227452255

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