15293384865

Committed 28 May 2025 06:45AM UTC coverage: 58.362% (-10.4%) from 68.776%

Build # 15293384865

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #9866 from ellemouton/graphSQL7-nodes-tables

sqldb+graph/db: add node related tables and implement some node CRUD

Run Details

0 of 644 new or added lines in 2 files covered. (0.0%)

28264 existing lines in 453 files now uncovered.

97492 of 167046 relevant lines covered (58.36%)

1.81 hits per line

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

61.9

/batch/batch.go

package batch

import (
        "context"
        "errors"
        "sync"

        "github.com/lightningnetwork/lnd/sqldb"
)

// errSolo is a sentinel error indicating that the requester should re-run the
// operation in isolation.
var errSolo = errors.New(
        "batch function returned an error and should be re-run solo",
)

// txOpts implements the sqldb.TxOptions interface. It is used to indicate that
// the transaction can be read-only or not transaction.
type txOpts struct {
        readOnly bool
}

// ReadOnly returns true if the transaction should be read only.
//
// NOTE: This is part of the sqldb.TxOptions interface.
func (t *txOpts) ReadOnly() bool {
        return t.readOnly
}

type request[Q any] struct {
        *Request[Q]
        errChan chan error
}

type batch[Q any] struct {
        db     sqldb.BatchedTx[Q]
        start  sync.Once
        reqs   []*request[Q]
        clear  func(b *batch[Q])
        locker sync.Locker
        txOpts txOpts
}

// trigger is the entry point for the batch and ensures that run is started at
// most once.
func (b *batch[Q]) trigger(ctx context.Context) {
        b.start.Do(func() {
                b.run(ctx)
        })
}

// run executes the current batch of requests. If any individual requests fail
// alongside others they will be retried by the caller.
func (b *batch[Q]) run(ctx context.Context) {
        // Clear the batch from its scheduler, ensuring that no new requests are
        // added to this batch.
        b.clear(b)

        // If a cache lock was provided, hold it until the this method returns.
        // This is critical for ensuring external consistency of the operation,
        // so that caches don't get out of sync with the on disk state.
        if b.locker != nil {
                b.locker.Lock()
                defer b.locker.Unlock()
        }

        // Apply the batch until a subset succeeds or all of them fail. Requests
        // that fail will be retried individually.
        for len(b.reqs) > 0 {
                var failIdx = -1
                err := b.db.ExecTx(ctx, &b.txOpts, func(tx Q) error {
                        for i, req := range b.reqs {
                                err := req.Do(tx)
                                if err != nil {
                                        // If we get a serialization error, we
                                        // want the underlying SQL retry
                                        // mechanism to retry the entire batch.
                                        // Otherwise, we can succeed in an
                                        // sqldb retry and still re-execute the
                                        // failing request individually.
                                        dbErr := sqldb.MapSQLError(err)
                                        if !sqldb.IsSerializationError(dbErr) {
                                                failIdx = i

                                                return err
                                        }

                                        return dbErr
                                }
                        }
                        return nil
                }, func() {
                        for _, req := range b.reqs {
                                if req.Reset != nil {
                                        req.Reset()
                                }
                        }
                })

                // If a request's Update failed, extract it and re-run the
                // batch. The removed request will be retried individually by
                // the caller.
                if failIdx >= 0 {
                        req := b.reqs[failIdx]

                        // It's safe to shorten b.reqs here because the
                        // scheduler's batch no longer points to us.
                        b.reqs[failIdx] = b.reqs[len(b.reqs)-1]
                        b.reqs = b.reqs[:len(b.reqs)-1]

                        // Tell the submitter re-run it solo, continue with the
                        // rest of the batch.
                        req.errChan <- errSolo
                        continue
                }

                // None of the remaining requests failed, process the errors
                // using each request's OnCommit closure and return the error
                // to the requester. If no OnCommit closure is provided, simply
                // return the error directly.
                for _, req := range b.reqs {
                        if req.OnCommit != nil {
                                req.errChan <- req.OnCommit(err)
                        } else {
                                req.errChan <- err
                        }
                }

                return
        }
}

1	package batch
2
3	import (
4	"context"
5	"errors"
6	"sync"
7
8	"github.com/lightningnetwork/lnd/sqldb"
9	)
10
11	// errSolo is a sentinel error indicating that the requester should re-run the
12	// operation in isolation.
13	var errSolo = errors.New(
14	"batch function returned an error and should be re-run solo",
15	)
16
17	// txOpts implements the sqldb.TxOptions interface. It is used to indicate that
18	// the transaction can be read-only or not transaction.
19	type txOpts struct {
20	readOnly bool
21	}
22
23	// ReadOnly returns true if the transaction should be read only.
24	//
25	// NOTE: This is part of the sqldb.TxOptions interface.
26	func (t *txOpts) ReadOnly() bool {	3✔
27	return t.readOnly	3✔
28	}	3✔
29
30	type request[Q any] struct {
31	*Request[Q]
32	errChan chan error
33	}
34
35	type batch[Q any] struct {
36	db sqldb.BatchedTx[Q]
37	start sync.Once
38	reqs []*request[Q]
39	clear func(b *batch[Q])
40	locker sync.Locker
41	txOpts txOpts
42	}
43
44	// trigger is the entry point for the batch and ensures that run is started at
45	// most once.
46	func (b *batch[Q]) trigger(ctx context.Context) {	3✔
47	b.start.Do(func() {	6✔
48	b.run(ctx)	3✔
49	})	3✔
50	}
51
52	// run executes the current batch of requests. If any individual requests fail
53	// alongside others they will be retried by the caller.
54	func (b *batch[Q]) run(ctx context.Context) {	3✔
55	// Clear the batch from its scheduler, ensuring that no new requests are	3✔
56	// added to this batch.	3✔
57	b.clear(b)	3✔
58		3✔
59	// If a cache lock was provided, hold it until the this method returns.	3✔
60	// This is critical for ensuring external consistency of the operation,	3✔
61	// so that caches don't get out of sync with the on disk state.	3✔
62	if b.locker != nil {	6✔
63	b.locker.Lock()	3✔
64	defer b.locker.Unlock()	3✔
65	}	3✔
66
67	// Apply the batch until a subset succeeds or all of them fail. Requests
68	// that fail will be retried individually.
69	for len(b.reqs) > 0 {	6✔
70	var failIdx = -1	3✔
71	err := b.db.ExecTx(ctx, &b.txOpts, func(tx Q) error {	6✔
72	for i, req := range b.reqs {	6✔
73	err := req.Do(tx)	3✔
74	if err != nil {	3✔
UNCOV 75	// If we get a serialization error, we	×
UNCOV 76	// want the underlying SQL retry	×
UNCOV 77	// mechanism to retry the entire batch.	×
UNCOV 78	// Otherwise, we can succeed in an	×
UNCOV 79	// sqldb retry and still re-execute the	×
UNCOV 80	// failing request individually.	×
UNCOV 81	dbErr := sqldb.MapSQLError(err)	×
UNCOV 82	if !sqldb.IsSerializationError(dbErr) {	×
UNCOV 83	failIdx = i	×
UNCOV 84		×
UNCOV 85	return err	×
UNCOV 86	}	×
87
UNCOV 88	return dbErr	×
89	}
90	}
91	return nil	3✔
92	}, func() {	3✔
93	for _, req := range b.reqs {	6✔
94	if req.Reset != nil {	6✔
95	req.Reset()	3✔
96	}	3✔
97	}
98	})
99
100	// If a request's Update failed, extract it and re-run the
101	// batch. The removed request will be retried individually by
102	// the caller.
103	if failIdx >= 0 {	3✔
UNCOV 104	req := b.reqs[failIdx]	×
UNCOV 105		×
UNCOV 106	// It's safe to shorten b.reqs here because the	×
UNCOV 107	// scheduler's batch no longer points to us.	×
UNCOV 108	b.reqs[failIdx] = b.reqs[len(b.reqs)-1]	×
UNCOV 109	b.reqs = b.reqs[:len(b.reqs)-1]	×
UNCOV 110		×
UNCOV 111	// Tell the submitter re-run it solo, continue with the	×
UNCOV 112	// rest of the batch.	×
UNCOV 113	req.errChan <- errSolo	×
UNCOV 114	continue	×
115	}
116
117	// None of the remaining requests failed, process the errors
118	// using each request's OnCommit closure and return the error
119	// to the requester. If no OnCommit closure is provided, simply
120	// return the error directly.
121	for _, req := range b.reqs {	6✔
122	if req.OnCommit != nil {	6✔
123	req.errChan <- req.OnCommit(err)	3✔
124	} else {	6✔
125	req.errChan <- err	3✔
126	}	3✔
127	}
128
129	return	3✔
130	}
131	}

lightningnetwork / lnd / 15293384865

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