diff --git a/channeldb/channel.go b/channeldb/channel.go index 135565c7c49..b8f677b6db2 100644 --- a/channeldb/channel.go +++ b/channeldb/channel.go @@ -1545,11 +1545,29 @@ func syncNewChannel(tx kvdb.RwTx, c *OpenChannel, addrs []net.Addr, return err } - // If a LinkNode for this identity public key already exists, - // then we can exit early. + // If a LinkNode for this identity public key already exists, then we + // don't want to clobber the state that has accumulated for it (extra + // addresses, last seen time, etc). We do however re-write the existing + // record verbatim instead of skipping the write entirely. + // + // This idempotent re-put is what makes this transaction conflict with a + // concurrent transaction that prunes the very same link node (see + // ChannelStateDB.pruneLinkNode and MarkChanFullyClosed). Those prune + // paths read the peer's set of open channels and delete the link node + // when that set is empty. Were we to skip the write here, then under + // snapshot isolation both transactions could commit: the pruner would + // not see our new channel, and we would not see its deletion, leaving + // an open channel behind with no link node. By always touching the link + // node row, one of the two transactions is instead aborted with a + // retryable serialization error. nodePub := c.IdentityPub.SerializeCompressed() - if nodeInfoBucket.Get(nodePub) != nil { - return nil + if existing := nodeInfoBucket.Get(nodePub); existing != nil { + // The returned slice may point directly into the database's + // memory, so we copy it before handing it back to Put. + linkNodeBytes := make([]byte, len(existing)) + copy(linkNodeBytes, existing) + + return nodeInfoBucket.Put(nodePub, linkNodeBytes) } // Next, we need to establish a (possibly) new LinkNode relationship diff --git a/channeldb/db.go b/channeldb/db.go index 3ed623eb177..076232a4e2b 100644 --- a/channeldb/db.go +++ b/channeldb/db.go @@ -1516,10 +1516,17 @@ func (c *ChannelStateDB) MarkChanFullyClosed(chanPoint *wire.OutPoint) error { return nil } - // If there are no open channels with this peer, prune the - // link node. We do this within the same transaction to avoid - // a race condition where a new channel could be opened - // between this check and the deletion. + // If there are no open channels with this peer, prune the link + // node. We do this within the same transaction as the read + // above so that a channel that is opened concurrently cannot + // slip in between the check and the deletion. + // + // NOTE: This is only safe because syncNewChannel always writes + // the peer's link node row, even when that row already exists. + // The write turns what would otherwise be a pair of + // transactions with disjoint write sets (write skew) into a + // same-row conflict, which the database reports as a retryable + // serialization failure. log.Infof("Pruning link node %x with zero open "+ "channels from database", remotePub.SerializeCompressed()) @@ -1541,13 +1548,17 @@ func (c *ChannelStateDB) MarkChanFullyClosed(chanPoint *wire.OutPoint) error { // channels exist. It will double-check within a write transaction to avoid a // race condition where a channel could be opened between the initial check // and the deletion. +// +// NOTE: The double-check below only rules out a concurrent channel open +// because syncNewChannel always writes the peer's link node row, even when +// that row already exists. Without that write the two transactions would have +// disjoint write sets and could both commit under snapshot isolation +// (REPEATABLE READ), leaving an open channel with no link node. See the +// comment in syncNewChannel for the full argument. func (c *ChannelStateDB) pruneLinkNode(remotePub *btcec.PublicKey) error { return kvdb.Update(c.backend, func(tx kvdb.RwTx) error { // Double-check for open channels to avoid deleting a link node // if a channel was opened since the caller's initial check. - // - // NOTE: This avoids a race condition where a channel could be - // opened between the initial check and the deletion. openChannels, err := c.fetchOpenChannels(tx, remotePub) if err != nil { return err diff --git a/channeldb/nodes_test.go b/channeldb/nodes_test.go index 413e0bc3910..7dd8233e5cb 100644 --- a/channeldb/nodes_test.go +++ b/channeldb/nodes_test.go @@ -372,3 +372,140 @@ func TestCreateLinkNodes(t *testing.T) { require.Equal(t, wire.MainNet, fetchedNode4.Network, "node4 should have correct network") } + +// linkNodeWriteCounterTx wraps a kvdb.RwTx so that all writes to the top-level +// link node bucket are counted. +type linkNodeWriteCounterTx struct { + kvdb.RwTx + + writes *int +} + +// CreateTopLevelBucket returns a write counting bucket if the requested bucket +// is the link node bucket, and otherwise defers to the wrapped transaction. +func (t *linkNodeWriteCounterTx) CreateTopLevelBucket( + key []byte) (kvdb.RwBucket, error) { + + bucket, err := t.RwTx.CreateTopLevelBucket(key) + if err != nil || !bytes.Equal(key, nodeInfoBucket) { + return bucket, err + } + + return &linkNodeWriteCounterBucket{ + RwBucket: bucket, + writes: t.writes, + }, nil +} + +// linkNodeWriteCounterBucket wraps a kvdb.RwBucket and counts the number of +// values written to it. +type linkNodeWriteCounterBucket struct { + kvdb.RwBucket + + writes *int +} + +// Put counts the write before deferring to the wrapped bucket. +func (b *linkNodeWriteCounterBucket) Put(key, value []byte) error { + *b.writes++ + + return b.RwBucket.Put(key, value) +} + +// TestSyncNewChannelWritesLinkNode tests that syncNewChannel always writes the +// peer's link node row, even when a link node for that peer already exists. The +// unconditional write is what causes a channel open to conflict with a +// concurrent link node prune under snapshot isolation. The write must however +// be a verbatim re-write, so that any state that has accumulated for the link +// node (such as extra addresses) is left untouched. +func TestSyncNewChannelWritesLinkNode(t *testing.T) { + t.Parallel() + + fullDB, err := MakeTestDB(t) + require.NoError(t, err, "unable to make test database") + + cdb := fullDB.ChannelStateDB() + + channel := createTestChannelState(t, cdb) + pub := channel.IdentityPub + + addr1, err := net.ResolveTCPAddr("tcp", "10.0.0.1:9000") + require.NoError(t, err, "unable to create test addr") + addr2, err := net.ResolveTCPAddr("tcp", "10.0.0.2:9000") + require.NoError(t, err, "unable to create test addr") + + syncChannel := func(addrs ...net.Addr) int { + var writes int + err := kvdb.Update(cdb.backend, func(tx kvdb.RwTx) error { + countTx := &linkNodeWriteCounterTx{ + RwTx: tx, + writes: &writes, + } + + return syncNewChannel( + countTx, channel, addrs, cdb.backend, + ) + }, func() { + writes = 0 + }) + require.NoError(t, err, "unable to sync channel") + + return writes + } + + // rawLinkNode returns the bytes that are stored for the peer's link + // node, so that the record can be compared byte for byte. + rawLinkNode := func() []byte { + var raw []byte + err := kvdb.View(cdb.backend, func(tx kvdb.RTx) error { + bucket := tx.ReadBucket(nodeInfoBucket) + require.NotNil(t, bucket) + + value := bucket.Get(pub.SerializeCompressed()) + raw = make([]byte, len(value)) + copy(raw, value) + + return nil + }, func() { + raw = nil + }) + require.NoError(t, err, "unable to read link node") + + return raw + } + + // The first sync creates the link node from scratch, which obviously + // writes the link node row. + require.Equal(t, 1, syncChannel(addr1)) + + linkNode, err := cdb.linkNodeDB.FetchLinkNode(pub) + require.NoError(t, err, "unable to fetch link node") + require.Len(t, linkNode.Addresses, 1) + + // Accumulate some extra state for the link node that a naive re-write + // of the link node would clobber. + updated := NewLinkNode( + cdb.linkNodeDB, linkNode.Network, pub, addr1, addr2, + ) + require.NoError(t, updated.Sync()) + + before := rawLinkNode() + require.NotEmpty(t, before) + + // A second sync (of another channel with the same peer) must still + // write the link node row, but it must leave the existing record + // exactly as it was. + channel.FundingOutpoint.Index++ + require.Equal(t, 1, syncChannel(addr1)) + + // The stored record must be byte for byte what it was before the sync. + // Nothing about the link node may be re-derived or re-serialized here, + // since the only reason for the write is the row conflict it creates. + require.Equal(t, before, rawLinkNode()) + + linkNode, err = cdb.linkNodeDB.FetchLinkNode(pub) + require.NoError(t, err, "unable to fetch link node") + require.Len(t, linkNode.Addresses, 2) + require.Equal(t, addr1.String(), linkNode.Addresses[0].String()) + require.Equal(t, addr2.String(), linkNode.Addresses[1].String()) +} diff --git a/docs/postgres.md b/docs/postgres.md index 89b16ebcf26..3194d4940a9 100644 --- a/docs/postgres.md +++ b/docs/postgres.md @@ -52,6 +52,56 @@ resource exhaustion in case LND experiencing high concurrent load: * `db.postgres.channeldb-with-global-lock=false` to run the channeldb_kv table with a single writer (default is false). +## Transaction isolation + +`lnd` opens read-write transactions at the `SERIALIZABLE` isolation level by +default. Read-only transactions are opened at `REPEATABLE READ`, which in +Postgres is snapshot isolation: the transaction reads from a single consistent +snapshot, taken when its first statement runs. Reads therefore acquire no +`SIRead` predicate locks and take no part in Postgres' serializable snapshot +isolation conflict graph, which substantially cuts the number of `40001` +serialization failures that `lnd` and its concurrent writers have to retry +through. + +Operators should be aware that some of `lnd`'s read transactions are long +lived. Loading the graph cache at startup and each `GraphSession` used for +pathfinding hold a read transaction open for their full duration, and each +holds its snapshot for that whole time. Two consequences follow. First, +Postgres cannot vacuum row versions that are still visible to an open snapshot, +so a very slow or stuck read transaction delays cleanup and can bloat tables. +Second, such a session sits in the `idle in transaction` state whenever `lnd` +is computing between queries, so if `idle_in_transaction_session_timeout` is +configured it must be generous enough to cover a full pathfinding pass or +Postgres will terminate the transaction mid-flight. The same caution applies to +`statement_timeout` for the individual queries these transactions run. + +The default for read-write transactions can be relaxed with: + +* `db.postgres.tx-isolation=repeatable-read` to also run read-write + transactions at `REPEATABLE READ` (default is `serializable`). + +**This option is experimental and opt-in.** Under `SERIALIZABLE`, Postgres +detects and aborts any pair of transactions whose interleaving is not +equivalent to running them one after the other. That safety costs a lot of +aborted transactions on a busy node, each of which `lnd` has to retry. + +`REPEATABLE READ` on Postgres is snapshot isolation. It still rules out dirty +reads, non-repeatable reads, phantom reads and lost updates: a transaction that +writes a row that another in-flight transaction has already written is aborted +with a serialization failure. The one anomaly it permits is *write skew*, where +two transactions each read what the other writes but write to disjoint sets of +rows, so neither of them conflicts and both are allowed to commit. + +The write paths that were known to be exposed to write skew have been changed +to either conflict on a shared row or to serialize in process, which is what +makes this option safe enough to offer at all. It nonetheless stays off by +default until it has accumulated soak time on real nodes. If you turn it on and +observe database inconsistencies, set it back to `serializable` and open an +issue. + +The option has no effect on SQLite, which only ever admits a single writer and +so is always effectively serializable. + ## Important note about replication In case a replication architecture is planned, streaming replication should be avoided, as the master does not verify the replica is indeed identical, but it will only forward the edits queue, and let the slave catch up autonomously; synchronous mode, albeit slower, is paramount for `lnd` data integrity across the copies, as it will finalize writes only after the slave confirmed successful replication. diff --git a/docs/release-notes/release-notes-0.22.0.md b/docs/release-notes/release-notes-0.22.0.md index 750a20069cd..b2370618867 100644 --- a/docs/release-notes/release-notes-0.22.0.md +++ b/docs/release-notes/release-notes-0.22.0.md @@ -55,6 +55,13 @@ the reported network statistics such as total network capacity, channel count and max out degree. +* [Fixed a bug](https://github.com/lightningnetwork/lnd/pull/10998) in the + watchtower client where a retried `AckUpdate` transaction could commit an + acknowledgment everywhere except in the acked-update index it belongs in, + after which the client would consider a state backed up that the tower had + never been told about. Only the SQL backends could retry a transaction, so + `bbolt` was never affected. + # New Features ## Functional Enhancements @@ -96,6 +103,23 @@ ## Performance Improvements +* [Read-only Postgres transactions now run at `REPEATABLE READ` instead of + `SERIALIZABLE`](https://github.com/lightningnetwork/lnd/pull/10997). In + Postgres that is snapshot isolation: a read-only transaction still reads from + a single consistent snapshot for its whole lifetime, taken when its first + statement runs. That snapshot is no longer guaranteed to correspond to a + serial ordering of the writers running alongside it, which is acceptable + because `lnd`'s read paths only consume a point-in-time view and never + depended on being ordered against writers in other transactions. In exchange, + such a transaction takes no part in Postgres' serializable snapshot isolation + conflict graph: it acquires no `SIRead` predicate locks, is not itself subject + to SSI serialization failures, and can no longer cause a concurrent writer to + be aborted as a pivot. Since `lnd` is very read heavy, this removes a large + amount of needless abort pressure. Read-write transactions are unaffected and + remain `SERIALIZABLE`, and the SQLite backend is untouched. See + [docs/postgres.md](../postgres.md) for the operator-facing note on long-lived + read transactions. + ## Deprecations # Technical and Architectural Updates @@ -138,6 +162,63 @@ ## Database +* [Four database write paths were hardened against snapshot + isolation](https://github.com/lightningnetwork/lnd/pull/10998), preparing for + read-write Postgres transactions to move from `SERIALIZABLE` to `REPEATABLE + READ`, the way [read-only transactions already + did](https://github.com/lightningnetwork/lnd/pull/10997). Under snapshot + isolation a pair of transactions that each read what the other writes, but + whose write sets don't overlap, both commit rather than one of them being + aborted, so each of these paths was changed to conflict on a shared row or to + serialize in process instead: + + * A channel open now always writes the peer's link node row, so that it can't + race a link node prune that runs when the peer's last channel is closed. + + * `PruneGraphNodes` now takes the cache mutex in both graph stores, like every + other graph mutator does, so that a node prune can't interleave with a + channel edge being added for that node. + + * Bucket creation in the SQL kvdb backends is now phrased as an upsert, so + that two transactions racing to create the same bucket see a retryable + serialization failure rather than a unique constraint violation, which is + not retried. + + * The watchtower client now evaluates a session for closability when it acks + an update for a channel that has already been closed, and the channel close + and ack paths conflict on a shared row. This also fixes a pre-existing leak + where a session that acked its first update for a channel only after that + channel was closed would never be marked closable, and so would hold on to + the tower's storage forever. + +* [Read-write Postgres transactions can now optionally be run at `REPEATABLE + READ`](https://github.com/lightningnetwork/lnd/pull/10999) via the new + `db.postgres.tx-isolation` option, which accepts `serializable` (the default) + and `repeatable-read`. This is the final piece of the work that + [moved read-only transactions to `REPEATABLE + READ`](https://github.com/lightningnetwork/lnd/pull/10997) and then [hardened + the write paths that snapshot isolation + exposes](https://github.com/lightningnetwork/lnd/pull/10998); both of those + are prerequisites for it. + + Under `SERIALIZABLE`, Postgres aborts any pair of transactions whose + interleaving isn't equivalent to running them one after the other, and on a + busy node that costs a lot of retries. `REPEATABLE READ` is snapshot + isolation, which still rules out dirty reads, non-repeatable reads, phantom + reads and lost updates, and leaves only write skew on the table. The write + paths known to be exposed to write skew were hardened in the PR above. + + **The option is experimental and stays off by default** until it has + accumulated soak time on real nodes. See `docs/postgres.md` before enabling + it. + +* [`KVStore.DeleteNode` now takes the graph store's cache + mutex](https://github.com/lightningnetwork/lnd/pull/10999) like every other + graph mutator, so that a node deletion can't interleave with a channel edge + being added for that node. The method is currently only reachable from tests, + so this isn't a live bug, but it's the same shape as the `PruneGraphNodes` + fix above. + ## Code Health ## Tooling and Documentation diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 7ca125d23ed..4311036a8b6 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -1125,6 +1125,20 @@ func (c *KVStore) DeleteNode(_ context.Context, v lnwire.GossipVersion, return ErrVersionNotSupportedForKVDB } + // Like every other mutator of the graph, we take the cache mutex before + // opening the write transaction. Beyond guarding the caches, this mutex + // is also the in-process serialization point against the batched + // channel edge insertion path: addChannelEdge reads a node's row + // without writing it, so a node deletion that ran concurrently with it + // could remove a node that the edge being added still references, + // leaving a dangling edge behind. Holding the mutex for the duration of + // the transaction rules that interleaving out. + // + // NOTE: The lock ordering here is cacheMu -> DB, which all other + // callers respect. + c.cacheMu.Lock() + defer c.cacheMu.Unlock() + // TODO(roasbeef): ensure dangling edges are removed... return kvdb.Update(c.db, func(tx kvdb.RwTx) error { nodes := tx.ReadWriteBucket(nodeBucket) @@ -1632,6 +1646,20 @@ func (c *KVStore) PruneGraph(_ context.Context, spentOutputs []*wire.OutPoint, // that we only maintain a graph of reachable nodes. In the event that a pruned // node gains more channels, it will be re-added back to the graph. func (c *KVStore) PruneGraphNodes(_ context.Context) ([]route.Vertex, error) { + // Like every other mutator of the graph, we take the cache mutex before + // opening the write transaction. Beyond guarding the caches, this mutex + // is also the in-process serialization point against the batched + // channel edge insertion path: addChannelEdge reads a node's row + // without writing it, so a node prune that ran concurrently with it + // could delete a node that the edge being added still references, + // leaving a dangling edge behind. Holding the mutex for the duration of + // the transaction rules that interleaving out. + // + // NOTE: The lock ordering here is cacheMu -> DB, which all other + // callers respect. + c.cacheMu.Lock() + defer c.cacheMu.Unlock() + var prunedNodes []route.Vertex err := kvdb.Update(c.db, func(tx kvdb.RwTx) error { nodes := tx.ReadWriteBucket(nodeBucket) diff --git a/graph/db/kv_store_test.go b/graph/db/kv_store_test.go new file mode 100644 index 00000000000..e15a6f8001b --- /dev/null +++ b/graph/db/kv_store_test.go @@ -0,0 +1,122 @@ +//go:build !test_db_sqlite && !test_db_postgres + +package graphdb + +import ( + "context" + "testing" + "time" + + "github.com/lightningnetwork/lnd/kvdb" + "github.com/lightningnetwork/lnd/lnwire" + "github.com/stretchr/testify/require" +) + +// TestPruneGraphNodesTakesCacheMutex asserts that PruneGraphNodes acquires the +// store's cache mutex before it opens its write transaction, just like every +// other mutator of the graph does. +// +// Beyond guarding the caches, that mutex is also the in-process serialization +// point against the batched channel edge insertion path. That path reads a +// node's row without writing it, so a node prune that ran concurrently with it +// could delete a node that the edge being added still references, leaving a +// dangling edge behind. +func TestPruneGraphNodesTakesCacheMutex(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + backend, backendCleanup, err := kvdb.GetTestBackend(t.TempDir(), "cgr") + require.NoError(t, err) + t.Cleanup(backendCleanup) + + store, err := NewKVStore(backend) + require.NoError(t, err) + + // The prune walks the graph starting from the source node, so we need + // one to be set for the prune to succeed. + sourceNode := createTestVertex(t, lnwire.GossipVersion1) + require.NoError(t, store.SetSourceNode(ctx, sourceNode)) + + // With the cache mutex held, a prune must not be able to make any + // progress. + store.cacheMu.Lock() + + pruneErr := make(chan error, 1) + go func() { + _, err := store.PruneGraphNodes(ctx) + pruneErr <- err + }() + + select { + case <-pruneErr: + t.Fatal("PruneGraphNodes did not wait for the cache mutex") + + case <-time.After(250 * time.Millisecond): + } + + // Once we release the mutex, the prune should be able to run to + // completion. + store.cacheMu.Unlock() + + select { + case err := <-pruneErr: + require.NoError(t, err) + + case <-time.After(time.Minute): + t.Fatal("PruneGraphNodes did not complete") + } +} + +// TestDeleteNodeTakesCacheMutex asserts that DeleteNode acquires the store's +// cache mutex before it opens its write transaction. It removes a node from the +// graph just like PruneGraphNodes does, so it needs the same exclusion against +// the batched channel edge insertion path. +func TestDeleteNodeTakesCacheMutex(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + backend, backendCleanup, err := kvdb.GetTestBackend(t.TempDir(), "cgr") + require.NoError(t, err) + t.Cleanup(backendCleanup) + + store, err := NewKVStore(backend) + require.NoError(t, err) + + // Add the node we're about to delete. This has to happen before we take + // the mutex below, since the node insertion path goes through the batch + // scheduler, which takes the very same mutex. + node := createTestVertex(t, lnwire.GossipVersion1) + require.NoError(t, store.AddNode(ctx, node)) + + // With the cache mutex held, a delete must not be able to make any + // progress. + store.cacheMu.Lock() + + deleteErr := make(chan error, 1) + go func() { + deleteErr <- store.DeleteNode( + ctx, lnwire.GossipVersion1, node.PubKeyBytes, + ) + }() + + select { + case <-deleteErr: + t.Fatal("DeleteNode did not wait for the cache mutex") + + case <-time.After(250 * time.Millisecond): + } + + // Once we release the mutex, the delete should be able to run to + // completion. + store.cacheMu.Unlock() + + select { + case err := <-deleteErr: + require.NoError(t, err) + + case <-time.After(time.Minute): + t.Fatal("DeleteNode did not complete") + } +} diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 1476515171b..b57fc4c704c 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -3234,6 +3234,16 @@ func (s *SQLStore) forEachChanInSCIDList(ctx context.Context, db SQLQueries, func (s *SQLStore) PruneGraphNodes(ctx context.Context) ( []route.Vertex, error) { + // Like every other mutator of the graph, we take the cache mutex before + // opening the write transaction. See the comment on the KV store's + // implementation of this method for why a node prune in particular must + // not be allowed to interleave with a channel edge being added. + // + // NOTE: The lock ordering here is cacheMu -> DB, which all other + // callers respect. + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + var prunedNodes []route.Vertex err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { var err error diff --git a/kvdb/postgres/config.go b/kvdb/postgres/config.go index 5ea06430782..29d31ddd494 100644 --- a/kvdb/postgres/config.go +++ b/kvdb/postgres/config.go @@ -10,4 +10,8 @@ type Config struct { Timeout time.Duration `long:"timeout" description:"Database connection timeout. Set to zero to disable."` MaxConnections int `long:"maxconnections" description:"The maximum number of open connections to the database. Set to zero for unlimited."` WithGlobalLock bool `long:"withgloballock" description:"Use a global lock to ensure a single writer."` + + // WriteTxRepeatableRead indicates that read-write transactions should + // be opened at REPEATABLE READ instead of SERIALIZABLE. + WriteTxRepeatableRead bool } diff --git a/kvdb/postgres/db.go b/kvdb/postgres/db.go index 6aca0276d2d..602f50b7600 100644 --- a/kvdb/postgres/db.go +++ b/kvdb/postgres/db.go @@ -27,6 +27,7 @@ func newSQLBaseConfig(config *Config, prefix string) *sqlbase.Config { TableNamePrefix: prefix, SQLiteCmdReplacements: sqliteCmdReplacements, WithTxLevelLock: config.WithGlobalLock, + WriteTxRepeatableRead: config.WriteTxRepeatableRead, } } diff --git a/kvdb/sqlbase/db.go b/kvdb/sqlbase/db.go index 8ff7f979aff..38c226cc788 100644 --- a/kvdb/sqlbase/db.go +++ b/kvdb/sqlbase/db.go @@ -25,6 +25,11 @@ const ( // transaction if it fails with an error that permits transaction // repetition. DefaultNumTxRetries = 50 + + // postgresDriverName is the name of the registered SQL driver that is + // used whenever the shared SQL backend is backed by Postgres. It is + // used to select behavior that only applies to Postgres. + postgresDriverName = "pgx" ) // Config holds a set of configuration options of a sql database connection. @@ -62,6 +67,11 @@ type Config struct { // NOTE: Temporary, should be removed when all parts of the LND code // are more resilient against concurrent db access.. WithTxLevelLock bool + + // WriteTxRepeatableRead indicates that read-write transactions should + // be opened at REPEATABLE READ instead of SERIALIZABLE. This only has + // an effect on Postgres. + WriteTxRepeatableRead bool } // db holds a reference to the sql db connection. @@ -95,6 +105,13 @@ type db struct { // Enforce db implements the walletdb.DB interface. var _ walletdb.DB = (*db)(nil) +// isPostgres returns true if the backend is backed by a Postgres database. The +// shared SQL backend is also used for SQLite, so any Postgres specific behavior +// needs to be gated on this. +func (d *db) isPostgres() bool { + return d.cfg.DriverName == postgresDriverName +} + var ( // dbConns is a global set of database connections. dbConns *dbConnSet diff --git a/kvdb/sqlbase/readwrite_bucket.go b/kvdb/sqlbase/readwrite_bucket.go index f8913723f86..d7e8b4b17aa 100644 --- a/kvdb/sqlbase/readwrite_bucket.go +++ b/kvdb/sqlbase/readwrite_bucket.go @@ -134,19 +134,12 @@ func (b *readWriteBucket) NestedReadWriteBucket( return newReadWriteBucket(b.tx, &id) } -// CreateBucket creates and returns a new nested bucket with the given key. -// Returns ErrBucketExists if the bucket already exists, ErrBucketNameRequired -// if the key is empty, or ErrIncompatibleValue if the key value is otherwise -// invalid for the particular database implementation. Other errors are -// possible depending on the implementation. -func (b *readWriteBucket) CreateBucket(key []byte) ( - walletdb.ReadWriteBucket, error) { - - if len(key) == 0 { - return nil, walletdb.ErrBucketNameRequired - } - - // Check to see if the bucket already exists. +// createBucket returns the id of the bucket with the given key, creating the +// bucket first if it doesn't exist yet. The first returned boolean signals +// whether the row was already there before this call, and the second signals +// whether that row holds a value instead of a bucket. +func (b *readWriteBucket) createBucket(key []byte) (int64, bool, bool, error) { + // Check to see if the key is already taken. var ( value *[]byte id int64 @@ -156,31 +149,95 @@ func (b *readWriteBucket) CreateBucket(key []byte) ( " AND key=$1", key, ) defer cancel() - err := row.Scan(&id, &value) + err := row.Scan(&id, &value) switch { - case err == sql.ErrNoRows: + case err == nil: + return id, true, value != nil, nil + + case !errors.Is(err, sql.ErrNoRows): + return 0, false, false, err + } + + // The key isn't taken as far as this transaction can see, so we go + // ahead and create the bucket. The database generates the id of the new + // bucket for us. + // + // Note that we deliberately don't use a bare insert here. Another + // transaction may be creating the very same bucket concurrently, and if + // it commits first then a bare insert leaves us with a unique + // constraint violation. That error maps to + // ErrSQLUniqueConstraintViolation, which the transaction retry loop + // does not consider retryable, so it would surface as a hard failure to + // the caller. Phrasing the insert as an upsert instead means the + // database reports the conflict as a serialization failure, which is + // retryable: on the retry the select above finds the winner's row. + // + // The conflict target has to match the partial unique index that + // applies to the row being inserted. There is one index for top level + // rows (_unp, on key where parent_id IS NULL) and one for nested + // rows (
_up, on (parent_id, key) where parent_id IS NOT NULL). + if b.id == nil { + row, cancel = b.tx.QueryRow( + "INSERT INTO "+b.table+" (key) VALUES($1) "+ + "ON CONFLICT (key) WHERE parent_id IS NULL "+ + "DO UPDATE SET key=$1 "+ + "RETURNING id, value", key, + ) + } else { + row, cancel = b.tx.QueryRow( + "INSERT INTO "+b.table+" (key, parent_id) "+ + "VALUES($1, $2) "+ + "ON CONFLICT (key, parent_id) "+ + "WHERE parent_id IS NOT NULL "+ + "DO UPDATE SET key=$1 "+ + "RETURNING id, value", key, b.id, + ) + } + defer cancel() - case err == nil && value == nil: - return nil, walletdb.ErrBucketExists + err = row.Scan(&id, &value) + if err != nil { + return 0, false, false, err + } - case err == nil && value != nil: - return nil, walletdb.ErrIncompatibleValue + // If the row we got back holds a value, then we collided with a value + // that was written concurrently, and the key can't be used for a + // bucket. + if value != nil { + return id, true, true, nil + } - case err != nil: - return nil, err + // At this point the row is ours: the select above proved that no such + // row was visible to this transaction, and any row inserted + // concurrently would have made the upsert fail with a serialization + // error under both of the isolation levels we run write transactions at + // (serializable and repeatable read). + return id, false, false, nil +} + +// CreateBucket creates and returns a new nested bucket with the given key. +// Returns ErrBucketExists if the bucket already exists, ErrBucketNameRequired +// if the key is empty, or ErrIncompatibleValue if the key value is otherwise +// invalid for the particular database implementation. Other errors are +// possible depending on the implementation. +func (b *readWriteBucket) CreateBucket(key []byte) ( + walletdb.ReadWriteBucket, error) { + + if len(key) == 0 { + return nil, walletdb.ErrBucketNameRequired } - // Bucket does not yet exist, so create it. Postgres will generate a - // bucket id for the new bucket. - row, cancel = b.tx.QueryRow( - "INSERT INTO "+b.table+" (parent_id, key) "+ - "VALUES($1, $2) RETURNING id", b.id, key, - ) - defer cancel() - err = row.Scan(&id) - if err != nil { + id, existed, isValue, err := b.createBucket(key) + switch { + case err != nil: return nil, err + + case isValue: + return nil, walletdb.ErrIncompatibleValue + + case existed: + return nil, walletdb.ErrBucketExists } return newReadWriteBucket(b.tx, &id), nil @@ -198,37 +255,13 @@ func (b *readWriteBucket) CreateBucketIfNotExists(key []byte) ( return nil, walletdb.ErrBucketNameRequired } - // Check to see if the bucket already exists. - var ( - value *[]byte - id int64 - ) - row, cancel := b.tx.QueryRow( - "SELECT id,value FROM "+b.table+" WHERE "+parentSelector(b.id)+ - " AND key=$1", key, - ) - defer cancel() - err := row.Scan(&id, &value) - + id, _, isValue, err := b.createBucket(key) switch { - // Bucket does not yet exist, so create it now. Postgres will generate a - // bucket id for the new bucket. - case err == sql.ErrNoRows: - row, cancel := b.tx.QueryRow( - "INSERT INTO "+b.table+" (parent_id, key) "+ - "VALUES($1, $2) RETURNING id", b.id, key, - ) - defer cancel() - err := row.Scan(&id) - if err != nil { - return nil, err - } - - case err == nil && value != nil: - return nil, walletdb.ErrIncompatibleValue - case err != nil: return nil, err + + case isValue: + return nil, walletdb.ErrIncompatibleValue } return newReadWriteBucket(b.tx, &id), nil diff --git a/kvdb/sqlbase/readwrite_bucket_postgres_test.go b/kvdb/sqlbase/readwrite_bucket_postgres_test.go new file mode 100644 index 00000000000..b117eeb211c --- /dev/null +++ b/kvdb/sqlbase/readwrite_bucket_postgres_test.go @@ -0,0 +1,342 @@ +//go:build kvdb_postgres + +package sqlbase + +import ( + "context" + "database/sql" + "fmt" + "sync" + "testing" + "time" + + "github.com/btcsuite/btcwallet/walletdb" + "github.com/lightningnetwork/lnd/sqldb" + "github.com/stretchr/testify/require" +) + +const ( + // numRacers is the number of transactions that concurrently attempt to + // create the same bucket. + numRacers = 8 + + // uniqueViolationCode is the SQLSTATE of a unique constraint violation, + // which the transaction retry loop does not retry. + uniqueViolationCode = "SQLSTATE 23505" + + // serializationFailureCode is the SQLSTATE of a serialization failure, + // which the transaction retry loop does retry. + serializationFailureCode = "SQLSTATE 40001" +) + +// TestPostgresConcurrentBucketCreation asserts that transactions racing to +// create the very same bucket all end up succeeding, with only a single row +// created for the bucket. The bucket creation path used to do a select followed +// by a bare insert, which leaves the loser of such a race with a unique +// constraint violation. That error is not retried by the transaction retry +// loop, so it would surface as a hard failure to the caller. +func TestPostgresConcurrentBucketCreation(t *testing.T) { + backend := newPostgresTestBackend(t) + + // raceCreate runs the given bucket creation function in numRacers + // concurrent transactions, none of which start creating before every + // last one of them has its transaction open. It returns the error of + // each of the transactions. + raceCreate := func(create func(walletdb.ReadWriteTx) error) []error { + var ( + ready sync.WaitGroup + arrived = make([]sync.Once, numRacers) + errs = make([]error, numRacers) + wg sync.WaitGroup + ) + ready.Add(numRacers) + + for i := 0; i < numRacers; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + + errs[i] = backend.Update( + func(tx walletdb.ReadWriteTx) error { + // Only stall on the very first + // attempt, a retry must not + // wait for the others. + arrived[i].Do(ready.Done) + ready.Wait() + + return create(tx) + }, func() {}, + ) + }(i) + } + + wg.Wait() + + return errs + } + + // countRows returns the number of rows with the given key that either + // are or aren't top level rows. + countRows := func(t *testing.T, key string, topLevel bool) int { + conn, err := sql.Open( + postgresDriverName, + fmt.Sprintf(testPgDsnTemplate, testPgPort), + ) + require.NoError(t, err) + defer conn.Close() + + parent := "parent_id IS NOT NULL" + if topLevel { + parent = "parent_id IS NULL" + } + + var count int + row := conn.QueryRow( + "SELECT count(*) FROM "+backend.table+" WHERE key=$1 "+ + "AND "+parent, key, + ) + require.NoError(t, row.Scan(&count)) + + return count + } + + // A racing CreateBucketIfNotExists must succeed everywhere, both for + // the top level bucket and for the nested one. + t.Run("create bucket if not exists", func(t *testing.T) { + top, nested := "apple", "banana" + + errs := raceCreate(func(tx walletdb.ReadWriteTx) error { + bkt, err := tx.CreateTopLevelBucket([]byte(top)) + if err != nil { + return err + } + + _, err = bkt.CreateBucketIfNotExists([]byte(nested)) + + return err + }) + + for _, err := range errs { + require.NoError(t, err) + } + + require.Equal(t, 1, countRows(t, top, true)) + require.Equal(t, 1, countRows(t, nested, false)) + }) + + // A racing CreateBucket must hand the bucket to exactly one caller and + // report ErrBucketExists to all the others. Crucially, the losers must + // not see a unique constraint violation. + t.Run("create bucket", func(t *testing.T) { + top, nested := "cherry", "date" + + errs := raceCreate(func(tx walletdb.ReadWriteTx) error { + bkt, err := tx.CreateTopLevelBucket([]byte(top)) + if err != nil { + return err + } + + _, err = bkt.CreateBucket([]byte(nested)) + + return err + }) + + var created int + for _, err := range errs { + if err == nil { + created++ + + continue + } + + require.ErrorIs(t, err, walletdb.ErrBucketExists) + } + + require.Equal(t, 1, created) + require.Equal(t, 1, countRows(t, top, true)) + require.Equal(t, 1, countRows(t, nested, false)) + }) +} + +// TestPostgresBucketCreationConflict asserts that the statement used to create +// a bucket reports a concurrent creation of that same bucket as a retryable +// serialization failure and not as a unique constraint violation, at both of +// the isolation levels that write transactions may be run at. The bare insert +// that this statement replaced is exercised alongside it to show why it can't +// be used once write transactions move to repeatable read. +func TestPostgresBucketCreationConflict(t *testing.T) { + backend := newPostgresTestBackend(t) + + conn, err := sql.Open( + postgresDriverName, fmt.Sprintf(testPgDsnTemplate, testPgPort), + ) + require.NoError(t, err) + defer conn.Close() + + ctx := context.Background() + table := backend.table + + // Create a bucket that the nested test cases can be parented to. + var parentID int64 + row := conn.QueryRowContext( + ctx, "INSERT INTO "+table+" (key) VALUES('parent') "+ + "RETURNING id", + ) + require.NoError(t, row.Scan(&parentID)) + + // race opens two transactions at the given isolation level, has both of + // them take their snapshot, and then has both of them run the given + // statement for the same key. The error of the transaction that loses + // the race is returned. + race := func(t *testing.T, level sql.IsolationLevel, stmt string, + args ...interface{}) error { + + opts := &sql.TxOptions{Isolation: level} + + tx1, err := conn.BeginTx(ctx, opts) + require.NoError(t, err) + defer tx1.Rollback() //nolint:errcheck + + tx2, err := conn.BeginTx(ctx, opts) + require.NoError(t, err) + defer tx2.Rollback() //nolint:errcheck + + // A snapshot is only taken once the first statement of a + // transaction runs, so we make sure that both transactions have + // one that predates the inserts below. + var count int + for _, tx := range []*sql.Tx{tx1, tx2} { + row := tx.QueryRowContext( + ctx, "SELECT count(*) FROM "+table+ + " WHERE key=$1", args[0], + ) + require.NoError(t, row.Scan(&count)) + require.Zero(t, count) + } + + _, err = tx1.ExecContext(ctx, stmt, args...) + require.NoError(t, err) + + // The second transaction blocks on the first one until it + // commits, so it has to be run separately. + loser := make(chan error, 1) + go func() { + _, err := tx2.ExecContext(ctx, stmt, args...) + loser <- err + }() + + select { + case err := <-loser: + t.Fatalf("second insert did not block: %v", err) + + case <-time.After(250 * time.Millisecond): + } + + require.NoError(t, tx1.Commit()) + + select { + case err := <-loser: + require.Error(t, err) + + return err + + case <-time.After(time.Minute): + t.Fatal("second insert never returned") + + return nil + } + } + + levels := []struct { + name string + level sql.IsolationLevel + }{ + { + name: "serializable", + level: sql.LevelSerializable, + }, + { + name: "repeatable_read", + level: sql.LevelRepeatableRead, + }, + } + + // The statements below are the top level and the nested flavour of both + // the old and the new bucket creation statement. Each has to line up + // with the partial unique index that covers the row it inserts: + //
_unp for top level rows and
_up for nested ones. + stmts := []struct { + name string + legacy bool + stmt string + args []interface{} + }{ + { + name: "upsert_top_level", + stmt: "INSERT INTO " + table + " (key) VALUES($1) " + + "ON CONFLICT (key) WHERE parent_id IS NULL " + + "DO UPDATE SET key=$1 RETURNING id, value", + }, + { + name: "upsert_nested", + stmt: "INSERT INTO " + table + " (key, parent_id) " + + "VALUES($1, $2) ON CONFLICT (key, parent_id) " + + "WHERE parent_id IS NOT NULL " + + "DO UPDATE SET key=$1 RETURNING id, value", + args: []interface{}{parentID}, + }, + { + name: "legacy_top_level", + legacy: true, + stmt: "INSERT INTO " + table + " (parent_id, key) " + + "VALUES(NULL, $1) RETURNING id", + }, + { + name: "legacy_nested", + legacy: true, + stmt: "INSERT INTO " + table + " (parent_id, key) " + + "VALUES($2, $1) RETURNING id", + args: []interface{}{parentID}, + }, + } + + for _, level := range levels { + for _, stmt := range stmts { + name := level.name + "/" + stmt.name + t.Run(name, func(t *testing.T) { + args := append( + []interface{}{"key-" + name}, + stmt.args..., + ) + err := race(t, level.level, stmt.stmt, args...) + + // The bare insert is only kept around to + // demonstrate the failure mode that the upsert + // avoids under snapshot isolation. + if stmt.legacy && + level.level == sql.LevelRepeatableRead { + + require.Contains( + t, err.Error(), + uniqueViolationCode, + ) + + return + } + + // Everything else must fail in a way that the + // transaction retry loop knows how to handle. + require.Contains( + t, err.Error(), serializationFailureCode, + ) + + var serErr *sqldb.ErrSerializationError + require.ErrorAs( + t, sqldb.MapSQLError(err), &serErr, + "want serialization failure, got %v", + err, + ) + }) + } + } +} diff --git a/kvdb/sqlbase/readwrite_tx.go b/kvdb/sqlbase/readwrite_tx.go index ec761931adc..848623cdc9d 100644 --- a/kvdb/sqlbase/readwrite_tx.go +++ b/kvdb/sqlbase/readwrite_tx.go @@ -25,6 +25,59 @@ type readWriteTx struct { locker sync.Locker } +// txIsolationLevel returns the isolation level that a transaction against the +// given database should be opened with. +// +// Read-only transactions on Postgres are opened at REPEATABLE READ, which in +// Postgres is snapshot isolation: the transaction reads from a single +// consistent snapshot for its whole lifetime, taken when its first statement +// runs rather than at BEGIN. +// +// That is a real, if modest, weakening. Snapshot isolation is not +// serializability, so the reader is no longer guaranteed to observe a state +// that corresponds to some serial ordering of the writers running alongside it. +// The read-only transaction anomaly described by Fekete and O'Neil is once +// again permitted. We accept that here because our read paths only ever consume +// a point-in-time view of the database, much like bbolt's View transactions do, +// and never depended on being ordered against writers in other transactions. A +// read that feeds a later write in a separate transaction was never protected +// across that boundary at any isolation level. +// +// In exchange, a read-only REPEATABLE READ transaction takes no part in +// Postgres' serializable snapshot isolation conflict graph. It acquires no +// SIRead predicate locks, is not itself subject to SSI serialization failures, +// and can no longer cause a concurrent writer to be aborted as a pivot. Since +// lnd is extremely read heavy, this removes a large amount of needless abort +// pressure from the system. +// +// Read-write transactions run at SERIALIZABLE by default, since that is the +// level the kvdb abstraction has always promised for writers, but they can be +// moved to REPEATABLE READ with the db.postgres.tx-isolation option. REPEATABLE +// READ on Postgres is snapshot isolation, which still rules out dirty reads, +// non-repeatable reads, phantom reads and lost updates: a transaction that +// writes a row another in-flight transaction has already written is aborted +// with a serialization failure. The one anomaly class that remains is write +// skew, where two transactions each read what the other writes but write +// disjoint sets of rows, so neither conflicts and both commit. Every write path +// that was known to be exposed to that has been hardened to either touch a +// shared row or to serialize in process, which is what makes offering this at +// all defensible. It nonetheless stays opt-in until it has accumulated soak +// time on real nodes. +// +// SQLite is always effectively serializable because it only ever admits a +// single writer, so there is nothing to gain there and we leave it alone. +func txIsolationLevel(db *db, readOnly bool) sql.IsolationLevel { + if !db.isPostgres() { + return sql.LevelSerializable + } + + if readOnly || db.cfg.WriteTxRepeatableRead { + return sql.LevelRepeatableRead + } + + return sql.LevelSerializable +} + // newReadWriteTx creates an rw transaction using a connection from the // specified pool. func newReadWriteTx(db *db, readOnly bool) (*readWriteTx, error) { @@ -50,7 +103,7 @@ func newReadWriteTx(db *db, readOnly bool) (*readWriteTx, error) { context.Background(), &sql.TxOptions{ ReadOnly: readOnly, - Isolation: sql.LevelSerializable, + Isolation: txIsolationLevel(db, readOnly), }, ) if err != nil { diff --git a/kvdb/sqlbase/readwrite_tx_postgres_test.go b/kvdb/sqlbase/readwrite_tx_postgres_test.go new file mode 100644 index 00000000000..441057d4ed6 --- /dev/null +++ b/kvdb/sqlbase/readwrite_tx_postgres_test.go @@ -0,0 +1,209 @@ +//go:build kvdb_postgres + +package sqlbase + +import ( + "context" + "fmt" + "io" + "path/filepath" + "testing" + "time" + + "github.com/btcsuite/btcwallet/walletdb" + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + "github.com/stretchr/testify/require" +) + +const ( + // testPgPort is the port that the embedded Postgres instance used by + // this package listens on. A dedicated port is used here so that this + // test binary can run alongside the one of the kvdb/postgres package, + // which brings up its own instance. + testPgPort = 9877 + + // testPgDsnTemplate is the connection string template for the embedded + // Postgres instance above. + testPgDsnTemplate = "postgres://postgres:postgres@localhost:%d/" + + "postgres?sslmode=disable" + + // testPgMaxConnections is the maximum number of connections that the + // embedded Postgres instance accepts. + testPgMaxConnections = 20 +) + +// newPostgresTestBackend spins up an embedded Postgres instance and returns a +// SQL backend that is connected to it. +func newPostgresTestBackend(t *testing.T) *db { + t.Helper() + + Init(testPgMaxConnections) + + // Keep all of the state of the embedded instance contained in a + // temporary directory that is removed once the test completes. + runtimePath := t.TempDir() + + pg := embeddedpostgres.NewDatabase( + embeddedpostgres.DefaultConfig(). + Port(testPgPort). + RuntimePath(runtimePath). + DataPath(filepath.Join(runtimePath, "data")). + Logger(io.Discard). + StartParameters(map[string]string{ + "max_connections": fmt.Sprintf( + "%d", testPgMaxConnections, + ), + }), + ) + require.NoError(t, pg.Start()) + t.Cleanup(func() { + require.NoError(t, pg.Stop()) + }) + + backend, err := NewSqlBackend(context.Background(), &Config{ + DriverName: postgresDriverName, + Dsn: fmt.Sprintf(testPgDsnTemplate, testPgPort), + Timeout: time.Minute, + Schema: "public", + TableNamePrefix: "test", + SQLiteCmdReplacements: SQLiteCmdReplacements{ + "BLOB": "BYTEA", + "INTEGER PRIMARY KEY": "BIGSERIAL PRIMARY KEY", + }, + }) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, backend.Close()) + }) + + return backend +} + +// TestPostgresTxIsolationLevel asserts that the isolation level that Postgres +// itself reports for a transaction matches what we expect: read-only +// transactions run at repeatable read while read-write transactions remain +// serializable unless the opt-in knob moves them to repeatable read as well. +// We also assert the read-only flag that Postgres reports, so that dropping it +// from the tx options would be caught here as well. +func TestPostgresTxIsolationLevel(t *testing.T) { + backend := newPostgresTestBackend(t) + + tests := []struct { + name string + readOnly bool + rrWrites bool + expected string + expectedFlag string + }{ + { + name: "read-only", + readOnly: true, + expected: "repeatable read", + expectedFlag: "on", + }, + { + name: "read-write", + readOnly: false, + expected: "serializable", + expectedFlag: "off", + }, + { + name: "read-only, rr writes", + readOnly: true, + rrWrites: true, + expected: "repeatable read", + expectedFlag: "on", + }, + { + name: "read-write, rr writes", + readOnly: false, + rrWrites: true, + expected: "repeatable read", + expectedFlag: "off", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + // The config isn't consulted anywhere else while a + // transaction is being opened, and these sub tests are + // not run in parallel, so it's safe to flip the knob in + // place rather than to bring up a second backend. + backend.cfg.WriteTxRepeatableRead = test.rrWrites + + tx, err := newReadWriteTx(backend, test.readOnly) + require.NoError(t, err) + defer func() { + require.NoError(t, tx.Rollback()) + }() + + var level string + row := tx.tx.QueryRow("SHOW transaction_isolation") + require.NoError(t, row.Scan(&level)) + require.Equal(t, test.expected, level) + + var readOnly string + row = tx.tx.QueryRow("SHOW transaction_read_only") + require.NoError(t, row.Scan(&readOnly)) + require.Equal(t, test.expectedFlag, readOnly) + }) + } +} + +// TestPostgresReadTxSnapshotStability asserts the property that the relaxed +// isolation level is chosen for, rather than just the knob itself: a read-only +// transaction keeps reading from the snapshot it started with, even after a +// concurrent writer on a different connection has committed over the same key. +func TestPostgresReadTxSnapshotStability(t *testing.T) { + backend := newPostgresTestBackend(t) + + var ( + bucketKey = []byte("snapshot") + key = []byte("key") + before = []byte("before") + after = []byte("after") + ) + + // Seed the key with its original value. + err := backend.Update(func(tx walletdb.ReadWriteTx) error { + bucket, err := tx.CreateTopLevelBucket(bucketKey) + if err != nil { + return err + } + + return bucket.Put(key, before) + }, func() {}) + require.NoError(t, err) + + readTx, err := newReadWriteTx(backend, true) + require.NoError(t, err) + defer func() { + require.NoError(t, readTx.Rollback()) + }() + + // Take the first read. Note that this is what pins the snapshot, since + // Postgres only acquires it once the first statement of a repeatable + // read transaction runs, not at BEGIN. + readBucket := readTx.ReadBucket(bucketKey) + require.NotNil(t, readBucket) + require.Equal(t, before, readBucket.Get(key)) + + // Now overwrite the key and commit, using a separate connection from + // the pool. + err = backend.Update(func(tx walletdb.ReadWriteTx) error { + return tx.ReadWriteBucket(bucketKey).Put(key, after) + }, func() {}) + require.NoError(t, err) + + // The write is visible to anyone starting fresh. + err = backend.View(func(tx walletdb.ReadTx) error { + require.Equal(t, after, tx.ReadBucket(bucketKey).Get(key)) + + return nil + }, func() {}) + require.NoError(t, err) + + // The long lived read transaction, however, must still observe the + // value that its snapshot was taken at. + require.Equal(t, before, readTx.ReadBucket(bucketKey).Get(key)) +} diff --git a/kvdb/sqlbase/readwrite_tx_test.go b/kvdb/sqlbase/readwrite_tx_test.go new file mode 100644 index 00000000000..b369fa4d591 --- /dev/null +++ b/kvdb/sqlbase/readwrite_tx_test.go @@ -0,0 +1,114 @@ +//go:build kvdb_postgres || (kvdb_sqlite && !(windows && (arm || 386)) && !(linux && (ppc64 || mips || mipsle || mips64))) + +package sqlbase + +import ( + "database/sql" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestTxIsolationLevel tests that the isolation level of a transaction is only +// relaxed for read-only transactions on Postgres, and for read-write +// transactions on Postgres once the opt-in knob is set. Every other combination +// must remain fully serializable. +func TestTxIsolationLevel(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + driverName string + readOnly bool + rrWrites bool + expected sql.IsolationLevel + }{ + { + name: "postgres read-only", + driverName: "pgx", + readOnly: true, + expected: sql.LevelRepeatableRead, + }, + { + name: "postgres read-write", + driverName: "pgx", + readOnly: false, + expected: sql.LevelSerializable, + }, + { + name: "postgres read-only, rr writes", + driverName: "pgx", + readOnly: true, + rrWrites: true, + expected: sql.LevelRepeatableRead, + }, + { + name: "postgres read-write, rr writes", + driverName: "pgx", + readOnly: false, + rrWrites: true, + expected: sql.LevelRepeatableRead, + }, + { + name: "sqlite read-only", + driverName: "sqlite", + readOnly: true, + expected: sql.LevelSerializable, + }, + { + name: "sqlite read-write", + driverName: "sqlite", + readOnly: false, + expected: sql.LevelSerializable, + }, + { + name: "sqlite read-write, rr writes", + driverName: "sqlite", + readOnly: false, + rrWrites: true, + expected: sql.LevelSerializable, + }, + + // Anything we don't positively recognize as Postgres must fall + // back to the strictest level. In particular "postgres" is not + // the driver name we register, so it must not opt in here, + // with or without the write isolation knob. + { + name: "unset driver read-only", + driverName: "", + readOnly: true, + expected: sql.LevelSerializable, + }, + { + name: "unknown driver read-only", + driverName: "postgres", + readOnly: true, + expected: sql.LevelSerializable, + }, + { + name: "unknown driver read-write, rr writes", + driverName: "postgres", + readOnly: false, + rrWrites: true, + expected: sql.LevelSerializable, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + db := &db{ + cfg: &Config{ + DriverName: test.driverName, + WriteTxRepeatableRead: test.rrWrites, + }, + } + + require.Equal( + t, test.expected, + txIsolationLevel(db, test.readOnly), + ) + }) + } +} diff --git a/lncfg/db.go b/lncfg/db.go index 4a8680b386e..7c919252499 100644 --- a/lncfg/db.go +++ b/lncfg/db.go @@ -122,7 +122,11 @@ func DefaultDB() *DB { // behavior until the wallet subsystem is upgraded to // a native sql schema. WalletDBWithGlobalLock: true, - QueryConfig: *sqldb.DefaultPostgresConfig(), + // Read-write transactions stay at SERIALIZABLE until + // the relaxed level has had enough soak time on real + // nodes. + TxIsolation: sqldb.TxIsolationSerializable, + QueryConfig: *sqldb.DefaultPostgresConfig(), }, Sqlite: &sqldb.SqliteConfig{ MaxConnections: sqldb.DefaultSqliteMaxConns, @@ -267,9 +271,10 @@ type DatabaseBackends struct { // postgres.Config. func GetPostgresConfigKVDB(cfg *sqldb.PostgresConfig) *postgres.Config { return &postgres.Config{ - Dsn: cfg.Dsn, - Timeout: cfg.Timeout, - MaxConnections: cfg.MaxConnections, + Dsn: cfg.Dsn, + Timeout: cfg.Timeout, + MaxConnections: cfg.MaxConnections, + WriteTxRepeatableRead: cfg.WriteTxRepeatableRead(), } } diff --git a/lncfg/db_test.go b/lncfg/db_test.go index 37ad4abe0c5..8b6649f53b3 100644 --- a/lncfg/db_test.go +++ b/lncfg/db_test.go @@ -3,8 +3,10 @@ package lncfg_test import ( "testing" + "github.com/jessevdk/go-flags" "github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/lncfg" + "github.com/lightningnetwork/lnd/sqldb" "github.com/stretchr/testify/require" ) @@ -21,4 +23,88 @@ func TestDBDefaultConfig(t *testing.T) { // Implicitly, the following fields are default to false. require.False(t, defaultConfig.Bolt.AutoCompact) require.True(t, defaultConfig.Bolt.NoFreelistSync) + + // Read-write transactions must stay at SERIALIZABLE unless the user + // explicitly opts into the relaxed level. + require.Equal( + t, sqldb.TxIsolationSerializable, + defaultConfig.Postgres.TxIsolation, + ) + require.False(t, defaultConfig.Postgres.WriteTxRepeatableRead()) +} + +// TestDBPostgresTxIsolation tests that the db.postgres.tx-isolation option is +// parsed by the flag parser, is rejected when it holds a value we don't +// support, and reaches the kvdb Postgres config. +func TestDBPostgresTxIsolation(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + arg string + parses bool + expected sqldb.TxIsolation + expectedRR bool + }{ + { + name: "no value given", + parses: true, + expected: sqldb.TxIsolationSerializable, + }, + { + name: "serializable", + arg: "--db.postgres.tx-isolation=serializable", + parses: true, + expected: sqldb.TxIsolationSerializable, + }, + { + name: "repeatable read", + arg: "--db.postgres.tx-isolation=repeatable-read", + parses: true, + expected: sqldb.TxIsolationRepeatableRead, + expectedRR: true, + }, + { + name: "garbage", + arg: "--db.postgres.tx-isolation=snapshot", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + cfg := struct { + DB *lncfg.DB `group:"db" namespace:"db"` + }{ + DB: lncfg.DefaultDB(), + } + + var args []string + if test.arg != "" { + args = append(args, test.arg) + } + + parser := flags.NewParser(&cfg, flags.None) + _, err := parser.ParseArgs(args) + if !test.parses { + require.Error(t, err) + return + } + require.NoError(t, err) + + require.Equal( + t, test.expected, cfg.DB.Postgres.TxIsolation, + ) + + // The value must also survive the trip into the kvdb + // flavored Postgres config, which is what the kvdb SQL + // backends are handed. + kvCfg := lncfg.GetPostgresConfigKVDB(cfg.DB.Postgres) + require.Equal( + t, test.expectedRR, + kvCfg.WriteTxRepeatableRead, + ) + }) + } } diff --git a/sample-lnd.conf b/sample-lnd.conf index f881c1174e9..c23480476b0 100644 --- a/sample-lnd.conf +++ b/sample-lnd.conf @@ -1717,11 +1717,20 @@ ; db.postgres.channeldb-with-global-lock=false -; Use a global lock for wallet database access. This is a temporary workaround +; Use a global lock for wallet database access. This is a temporary workaround ; until the wallet subsystem is upgraded to a native sql schema. ; db.postgres.walletdb-with-global-lock=true +; The isolation level that read-write database transactions are opened with. +; Read-only transactions always run at repeatable read and are not affected by +; this setting. EXPERIMENTAL: 'repeatable-read' lowers the abort rate under +; concurrency, but permits write skew, which only the write paths that were +; explicitly hardened against it are known to be safe from. See docs/postgres.md +; before enabling it. Valid values are 'serializable' and 'repeatable-read'. +; db.postgres.tx-isolation=serializable + + ; The maximum number of elements to use in a native-SQL batch query IN clause. ; db.postgres.query.max-batch-size=5000 diff --git a/sqldb/config.go b/sqldb/config.go index bf2a8deb896..5cfe2e2aa4e 100644 --- a/sqldb/config.go +++ b/sqldb/config.go @@ -74,6 +74,22 @@ func (p *SqliteConfig) Validate() error { return nil } +// TxIsolation is the isolation level that read-write transactions against a +// Postgres backend are opened with. Read-only transactions are always opened at +// REPEATABLE READ and are not affected by this setting. +type TxIsolation string + +const ( + // TxIsolationSerializable opens read-write transactions at + // SERIALIZABLE. This is the level lnd has always used and remains the + // default. + TxIsolationSerializable TxIsolation = "serializable" + + // TxIsolationRepeatableRead opens read-write transactions at REPEATABLE + // READ. This is an experimental, opt-in setting. + TxIsolationRepeatableRead TxIsolation = "repeatable-read" +) + // PostgresConfig holds the postgres database configuration. // //nolint:ll @@ -84,6 +100,7 @@ type PostgresConfig struct { SkipMigrations bool `long:"skipmigrations" description:"Skip applying migrations on startup."` ChannelDBWithGlobalLock bool `long:"channeldb-with-global-lock" description:"Use a global lock for channeldb access. This ensures only a single writer at a time but reduces concurrency. This is a temporary workaround until the revocation log is migrated to a native sql schema."` WalletDBWithGlobalLock bool `long:"walletdb-with-global-lock" description:"Use a global lock for wallet database access. This ensures only a single writer at a time but reduces concurrency. This is a temporary workaround until the wallet subsystem is upgraded to a native sql schema."` + TxIsolation TxIsolation `long:"tx-isolation" description:"The isolation level that read-write database transactions are opened with. Read-only transactions always run at repeatable read and are not affected by this setting. EXPERIMENTAL: 'repeatable-read' lowers the abort rate under concurrency, but permits write skew, which only the write paths that were explicitly hardened against it are known to be safe from. See docs/postgres.md before enabling it." choice:"serializable" choice:"repeatable-read"` QueryConfig `group:"query" namespace:"query"` } @@ -99,9 +116,26 @@ func (p *PostgresConfig) Validate() error { return fmt.Errorf("invalid DSN: %w", err) } + // An empty value is allowed here and means that the default, + // SERIALIZABLE, is used. + switch p.TxIsolation { + case "", TxIsolationSerializable, TxIsolationRepeatableRead: + + default: + return fmt.Errorf("invalid tx isolation level '%s': must be "+ + "either '%s' or '%s'", p.TxIsolation, + TxIsolationSerializable, TxIsolationRepeatableRead) + } + if err := p.QueryConfig.Validate(false); err != nil { return fmt.Errorf("invalid query config: %w", err) } return nil } + +// WriteTxRepeatableRead returns true if read-write transactions should be +// opened at REPEATABLE READ instead of SERIALIZABLE. +func (p *PostgresConfig) WriteTxRepeatableRead() bool { + return p.TxIsolation == TxIsolationRepeatableRead +} diff --git a/sqldb/config_test.go b/sqldb/config_test.go new file mode 100644 index 00000000000..c65b66bbe11 --- /dev/null +++ b/sqldb/config_test.go @@ -0,0 +1,84 @@ +package sqldb + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// TestPostgresConfigTxIsolation asserts that the tx isolation option only +// accepts the two levels we support, and that only the relaxed one asks for +// read-write transactions at repeatable read. +func TestPostgresConfigTxIsolation(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + txIsolation TxIsolation + valid bool + rrWrites bool + }{ + { + name: "unset", + txIsolation: "", + valid: true, + }, + { + name: "serializable", + txIsolation: TxIsolationSerializable, + valid: true, + }, + { + name: "repeatable read", + txIsolation: TxIsolationRepeatableRead, + valid: true, + rrWrites: true, + }, + { + name: "garbage", + txIsolation: "not-an-isolation-level", + }, + { + // The Postgres spelling of the level is not the one we + // accept, since our own options are dash separated. + name: "postgres spelling", + txIsolation: "repeatable read", + }, + { + name: "wrong case", + txIsolation: "SERIALIZABLE", + }, + { + // We deliberately don't expose the weaker levels that + // Postgres itself supports. + name: "read committed", + txIsolation: "read-committed", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + cfg := &PostgresConfig{ + Dsn: "postgres://lnd@localhost/lnd", + TxIsolation: test.txIsolation, + QueryConfig: *DefaultPostgresConfig(), + } + + err := cfg.Validate() + if !test.valid { + require.ErrorContains( + t, err, "invalid tx isolation level", + ) + + return + } + + require.NoError(t, err) + require.Equal( + t, test.rrWrites, cfg.WriteTxRepeatableRead(), + ) + }) + } +} diff --git a/sqldb/interfaces.go b/sqldb/interfaces.go index 12ce63a1bb9..2ada6538e73 100644 --- a/sqldb/interfaces.go +++ b/sqldb/interfaces.go @@ -32,6 +32,21 @@ const ( DefaultMaxRetryDelay = time.Second ) +// BackendType is an enum that represents the type of database backend we're +// using. +type BackendType uint8 + +const ( + // BackendTypeUnknown indicates we're using an unknown backend. + BackendTypeUnknown BackendType = iota + + // BackendTypeSqlite indicates we're using a SQLite backend. + BackendTypeSqlite + + // BackendTypePostgres indicates we're using a Postgres backend. + BackendTypePostgres +) + // TxOptions represents a set of options one can use to control what type of // database transaction is created. Transaction can be either read or write. type TxOptions interface { @@ -409,16 +424,87 @@ type BaseDB struct { *sql.DB *sqlc.Queries + + // BackendType defines the type of database backend the database is. + BackendType BackendType + + // WriteTxRepeatableRead indicates that read-write transactions should + // be opened at REPEATABLE READ instead of SERIALIZABLE. This only has + // an effect on Postgres. + WriteTxRepeatableRead bool +} + +// Backend returns the type of the database backend used. +func (s *BaseDB) Backend() BackendType { + return s.BackendType } // BeginTx wraps the normal sql specific BeginTx method with the TxOptions // interface. This interface is then mapped to the concrete sql tx options // struct. func (s *BaseDB) BeginTx(ctx context.Context, opts TxOptions) (*sql.Tx, error) { + readOnly := opts.ReadOnly() + sqlOptions := sql.TxOptions{ - Isolation: sql.LevelSerializable, - ReadOnly: opts.ReadOnly(), + Isolation: txIsolationLevel( + s.BackendType, readOnly, s.WriteTxRepeatableRead, + ), + ReadOnly: readOnly, } return s.DB.BeginTx(ctx, &sqlOptions) } + +// txIsolationLevel returns the isolation level that a transaction against the +// given backend should be opened with. +// +// Read-only transactions on Postgres are opened at REPEATABLE READ, which in +// Postgres is snapshot isolation: the transaction reads from a single +// consistent snapshot for its whole lifetime, taken when its first statement +// runs rather than at BEGIN. +// +// That is a real, if modest, weakening. Snapshot isolation is not +// serializability, so the reader is no longer guaranteed to observe a state +// that corresponds to some serial ordering of the writers running alongside it. +// The read-only transaction anomaly described by Fekete and O'Neil is once +// again permitted. We accept that here because our read paths only ever consume +// a point-in-time view of the database and never depended on being ordered +// against writers in other transactions. A read that feeds a later write in a +// separate transaction was never protected across that boundary at any +// isolation level. +// +// In exchange, a read-only REPEATABLE READ transaction takes no part in +// Postgres' serializable snapshot isolation conflict graph. It acquires no +// SIRead predicate locks, is not itself subject to SSI serialization failures, +// and can no longer cause a concurrent writer to be aborted as a pivot. Since +// lnd is extremely read heavy, this removes a large amount of needless abort +// pressure from the system. +// +// Read-write transactions run at SERIALIZABLE by default, but can be moved to +// REPEATABLE READ with the db.postgres.tx-isolation option. REPEATABLE READ on +// Postgres is snapshot isolation, which still rules out dirty reads, +// non-repeatable reads, phantom reads and lost updates: a transaction that +// writes a row another in-flight transaction has already written is aborted +// with a serialization failure. The one anomaly class that remains is write +// skew, where two transactions each read what the other writes but write +// disjoint sets of rows, so neither conflicts and both commit. Every write path +// that was known to be exposed to that has been hardened to either touch a +// shared row or to serialize in process, which is what makes offering this at +// all defensible. It nonetheless stays opt-in until it has accumulated soak +// time on real nodes. +// +// SQLite is always effectively serializable because it only ever admits a +// single writer, so there is nothing to gain there and we leave it alone. +func txIsolationLevel(backend BackendType, readOnly, + writeTxRepeatableRead bool) sql.IsolationLevel { + + if backend != BackendTypePostgres { + return sql.LevelSerializable + } + + if readOnly || writeTxRepeatableRead { + return sql.LevelRepeatableRead + } + + return sql.LevelSerializable +} diff --git a/sqldb/interfaces_test.go b/sqldb/interfaces_test.go new file mode 100644 index 00000000000..207e65e7c5f --- /dev/null +++ b/sqldb/interfaces_test.go @@ -0,0 +1,193 @@ +package sqldb + +import ( + "database/sql" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestTxIsolationLevel tests that the isolation level of a transaction is only +// relaxed for read-only transactions on Postgres, and for read-write +// transactions on Postgres once the opt-in knob is set. Every other combination +// must remain fully serializable. +func TestTxIsolationLevel(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + backend BackendType + readOnly bool + rrWrites bool + expected sql.IsolationLevel + }{ + { + name: "postgres read-only", + backend: BackendTypePostgres, + readOnly: true, + expected: sql.LevelRepeatableRead, + }, + { + name: "postgres read-write", + backend: BackendTypePostgres, + readOnly: false, + expected: sql.LevelSerializable, + }, + { + name: "postgres read-only, rr writes", + backend: BackendTypePostgres, + readOnly: true, + rrWrites: true, + expected: sql.LevelRepeatableRead, + }, + { + name: "postgres read-write, rr writes", + backend: BackendTypePostgres, + readOnly: false, + rrWrites: true, + expected: sql.LevelRepeatableRead, + }, + { + name: "sqlite read-only", + backend: BackendTypeSqlite, + readOnly: true, + expected: sql.LevelSerializable, + }, + { + name: "sqlite read-write", + backend: BackendTypeSqlite, + readOnly: false, + expected: sql.LevelSerializable, + }, + { + name: "sqlite read-write, rr writes", + backend: BackendTypeSqlite, + readOnly: false, + rrWrites: true, + expected: sql.LevelSerializable, + }, + { + name: "unknown read-only", + backend: BackendTypeUnknown, + readOnly: true, + expected: sql.LevelSerializable, + }, + { + name: "unknown read-write, rr writes", + backend: BackendTypeUnknown, + readOnly: false, + rrWrites: true, + expected: sql.LevelSerializable, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + require.Equal( + t, test.expected, txIsolationLevel( + test.backend, test.readOnly, + test.rrWrites, + ), + ) + }) + } +} + +// TestBeginTxIsolationLevel asserts that the isolation level that the database +// itself reports for a transaction opened through BeginTx matches what we +// expect. On Postgres, read-only transactions run at repeatable read while +// read-write transactions remain serializable unless the opt-in knob moves +// them to repeatable read as well. We also assert the read-only flag that +// Postgres reports, so that dropping it from the tx options would be caught +// here as well. +func TestBeginTxIsolationLevel(t *testing.T) { + t.Parallel() + ctx := t.Context() + + db := NewTestDB(t).GetBaseDB() + + // The knob defaults to off, which is what the first two cases below + // assert. + require.False(t, db.WriteTxRepeatableRead) + + tests := []struct { + name string + opts TxOptions + rrWrites bool + expected string + expectedFlag string + }{ + { + name: "read-only", + opts: ReadTxOpt(), + expected: "repeatable read", + expectedFlag: "on", + }, + { + name: "read-write", + opts: WriteTxOpt(), + expected: "serializable", + expectedFlag: "off", + }, + { + name: "read-only, rr writes", + opts: ReadTxOpt(), + rrWrites: true, + expected: "repeatable read", + expectedFlag: "on", + }, + { + name: "read-write, rr writes", + opts: WriteTxOpt(), + rrWrites: true, + expected: "repeatable read", + expectedFlag: "off", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + // Nothing else reads this field while a transaction is + // being opened, and these sub tests are not run in + // parallel, so it's safe to flip the knob in place + // rather than to bring up a second database. + db.WriteTxRepeatableRead = test.rrWrites + + tx, err := db.BeginTx(ctx, test.opts) + + // SQLite has no notion of a transaction isolation + // level and its driver ignores the one we request + // outright. All we can assert there is that opening + // the transaction still succeeds, which is what would + // break if the driver ever started rejecting the + // levels we ask for. + require.NoError(t, err) + defer func() { + require.NoError(t, tx.Rollback()) + }() + + if isSQLite { + require.Equal(t, BackendTypeSqlite, db.Backend()) + return + } + + require.Equal(t, BackendTypePostgres, db.Backend()) + + var level string + row := tx.QueryRowContext( + ctx, "SHOW transaction_isolation", + ) + require.NoError(t, row.Scan(&level)) + require.Equal(t, test.expected, level) + + var readOnly string + row = tx.QueryRowContext( + ctx, "SHOW transaction_read_only", + ) + require.NoError(t, row.Scan(&readOnly)) + require.Equal(t, test.expectedFlag, readOnly) + }) + } +} diff --git a/sqldb/postgres.go b/sqldb/postgres.go index 70dba82a1a0..30cf3576dd9 100644 --- a/sqldb/postgres.go +++ b/sqldb/postgres.go @@ -136,8 +136,10 @@ func NewPostgresStore(cfg *PostgresConfig) (*PostgresStore, error) { return &PostgresStore{ cfg: cfg, BaseDB: &BaseDB{ - DB: db, - Queries: queries, + DB: db, + Queries: queries, + BackendType: BackendTypePostgres, + WriteTxRepeatableRead: cfg.WriteTxRepeatableRead(), }, }, nil } diff --git a/sqldb/sqlerrors.go b/sqldb/sqlerrors.go index 5f8d998b53d..a9a03694581 100644 --- a/sqldb/sqlerrors.go +++ b/sqldb/sqlerrors.go @@ -113,21 +113,21 @@ func parsePostgresError(pqErr *pgconn.PgError) error { // Unable to serialize the transaction, so we'll need to try again. case pgerrcode.SerializationFailure: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), } // In failed SQL transaction because we didn't catch a previous // serialization error, so return this one as a serialization error. case pgerrcode.InFailedSQLTransaction: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), } // Deadlock detedted because of a serialization error, so return this // one as a serialization error. case pgerrcode.DeadlockDetected: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), } default: diff --git a/sqldb/sqlerrors_no_sqlite.go b/sqldb/sqlerrors_no_sqlite.go index 9e85c98fb98..0d1b7a71d1c 100644 --- a/sqldb/sqlerrors_no_sqlite.go +++ b/sqldb/sqlerrors_no_sqlite.go @@ -43,7 +43,21 @@ func parsePostgresError(pqErr *pgconn.PgError) error { // Unable to serialize the transaction, so we'll need to try again. case pgerrcode.SerializationFailure: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), + } + + // In failed SQL transaction because we didn't catch a previous + // serialization error, so return this one as a serialization error. + case pgerrcode.InFailedSQLTransaction: + return &ErrSerializationError{ + DBError: withPostgresDetail(pqErr), + } + + // Deadlock detedted because of a serialization error, so return this + // one as a serialization error. + case pgerrcode.DeadlockDetected: + return &ErrSerializationError{ + DBError: withPostgresDetail(pqErr), } default: diff --git a/sqldb/sqlerrors_postgres.go b/sqldb/sqlerrors_postgres.go new file mode 100644 index 00000000000..4fb319c6118 --- /dev/null +++ b/sqldb/sqlerrors_postgres.go @@ -0,0 +1,34 @@ +package sqldb + +import ( + "fmt" + + "github.com/jackc/pgx/v5/pgconn" +) + +// withPostgresDetail returns an error that also carries the Detail field of the +// given Postgres error, if it has one. The error string that the driver builds +// leaves that field out, and for a transaction conflict it is the only thing +// that says which kind of conflict was hit: an abort that only serializable +// isolation raises spells out the reason code of the pivot that was cancelled, +// whereas a plain write-write conflict just reports a concurrent update. Being +// able to tell the two apart is what says how much of the abort pressure a +// deployment sees would go away by relaxing the isolation level. +// +// NOTE: This must only ever be applied to the errors that report a transaction +// conflict. For any other error, and for a constraint violation in particular, +// the detail spells out the offending column values, which for our schemas +// means things like payment session keys, invoice hashes and raw key/value +// bucket keys. None of that belongs in a log line, let alone in an error that +// may travel back over an RPC. +// +// This lives in its own file, free of build tags, so that the two build +// specific copies of the error parsing below it share a single definition of +// it. +func withPostgresDetail(pqErr *pgconn.PgError) error { + if pqErr.Detail == "" { + return pqErr + } + + return fmt.Errorf("%w (detail: %s)", pqErr, pqErr.Detail) +} diff --git a/sqldb/sqlite.go b/sqldb/sqlite.go index 1ed26810d02..c32797046e2 100644 --- a/sqldb/sqlite.go +++ b/sqldb/sqlite.go @@ -144,8 +144,9 @@ func NewSqliteStore(cfg *SqliteConfig, dbPath string) (*SqliteStore, error) { s := &SqliteStore{ cfg: cfg, BaseDB: &BaseDB{ - DB: db, - Queries: queries, + DB: db, + Queries: queries, + BackendType: BackendTypeSqlite, }, } diff --git a/sqldb/v2/config.go b/sqldb/v2/config.go index 9e6c3c015bd..c78294e2838 100644 --- a/sqldb/v2/config.go +++ b/sqldb/v2/config.go @@ -96,6 +96,22 @@ func (p *SqliteConfig) Validate() error { return nil } +// TxIsolation is the isolation level that read-write transactions against a +// Postgres backend are opened with. Read-only transactions are always opened at +// REPEATABLE READ and are not affected by this setting. +type TxIsolation string + +const ( + // TxIsolationSerializable opens read-write transactions at + // SERIALIZABLE. This is the level lnd has always used and remains the + // default. + TxIsolationSerializable TxIsolation = "serializable" + + // TxIsolationRepeatableRead opens read-write transactions at REPEATABLE + // READ. This is an experimental, opt-in setting. + TxIsolationRepeatableRead TxIsolation = "repeatable-read" +) + // PostgresConfig holds the postgres database configuration. // //nolint:ll @@ -108,6 +124,7 @@ type PostgresConfig struct { ConnMaxIdleTime time.Duration `long:"connmaxidletime" description:"Max amount of time a connection can be idle for before it is closed. Valid time units are {s, m, h}."` RequireSSL bool `long:"requiressl" description:"Whether to require using SSL (mode: require) when connecting to the server."` SkipMigrations bool `long:"skipmigrations" description:"Skip applying migrations on startup."` + TxIsolation TxIsolation `long:"tx-isolation" description:"The isolation level that read-write database transactions are opened with. Read-only transactions always run at repeatable read and are not affected by this setting. EXPERIMENTAL: 'repeatable-read' lowers the abort rate under concurrency, but permits write skew, which only the write paths that were explicitly hardened against it are known to be safe from. See docs/postgres.md before enabling it." choice:"serializable" choice:"repeatable-read"` QueryConfig `group:"query" namespace:"query"` } @@ -123,9 +140,26 @@ func (p *PostgresConfig) Validate() error { return fmt.Errorf("invalid DSN: %w", err) } + // An empty value is allowed here and means that the default, + // SERIALIZABLE, is used. + switch p.TxIsolation { + case "", TxIsolationSerializable, TxIsolationRepeatableRead: + + default: + return fmt.Errorf("invalid tx isolation level '%s': must be "+ + "either '%s' or '%s'", p.TxIsolation, + TxIsolationSerializable, TxIsolationRepeatableRead) + } + if err := p.QueryConfig.Validate(false); err != nil { return fmt.Errorf("invalid query config: %w", err) } return nil } + +// WriteTxRepeatableRead returns true if read-write transactions should be +// opened at REPEATABLE READ instead of SERIALIZABLE. +func (p *PostgresConfig) WriteTxRepeatableRead() bool { + return p.TxIsolation == TxIsolationRepeatableRead +} diff --git a/sqldb/v2/config_test.go b/sqldb/v2/config_test.go index c6e9d211717..7e9b3df8d81 100644 --- a/sqldb/v2/config_test.go +++ b/sqldb/v2/config_test.go @@ -86,3 +86,80 @@ func TestSqliteConfigMaxIdleConns(t *testing.T) { }) } } + +// TestPostgresConfigTxIsolation asserts that the tx isolation option only +// accepts the two levels we support, and that only the relaxed one asks for +// read-write transactions at repeatable read. +func TestPostgresConfigTxIsolation(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + txIsolation TxIsolation + valid bool + rrWrites bool + }{ + { + name: "unset", + txIsolation: "", + valid: true, + }, + { + name: "serializable", + txIsolation: TxIsolationSerializable, + valid: true, + }, + { + name: "repeatable read", + txIsolation: TxIsolationRepeatableRead, + valid: true, + rrWrites: true, + }, + { + name: "garbage", + txIsolation: "not-an-isolation-level", + }, + { + // The Postgres spelling of the level is not the one we + // accept, since our own options are dash separated. + name: "postgres spelling", + txIsolation: "repeatable read", + }, + { + name: "wrong case", + txIsolation: "SERIALIZABLE", + }, + { + // We deliberately don't expose the weaker levels that + // Postgres itself supports. + name: "read committed", + txIsolation: "read-committed", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + cfg := &PostgresConfig{ + Dsn: "postgres://lnd@localhost/lnd", + TxIsolation: test.txIsolation, + QueryConfig: *DefaultPostgresConfig(), + } + + err := cfg.Validate() + if !test.valid { + require.ErrorContains( + t, err, "invalid tx isolation level", + ) + + return + } + + require.NoError(t, err) + require.Equal( + t, test.rrWrites, cfg.WriteTxRepeatableRead(), + ) + }) + } +} diff --git a/sqldb/v2/interfaces.go b/sqldb/v2/interfaces.go index 0bf6da88194..9d9bbb71511 100644 --- a/sqldb/v2/interfaces.go +++ b/sqldb/v2/interfaces.go @@ -446,20 +446,83 @@ type BaseDB struct { // SkipMigrations can be set to true to skip running any migrations // during the iinitialization of the database. SkipMigrations bool + + // WriteTxRepeatableRead indicates that read-write transactions should + // be opened at REPEATABLE READ instead of SERIALIZABLE. This only has + // an effect on Postgres. + WriteTxRepeatableRead bool } // BeginTx wraps the normal sql specific BeginTx method with the TxOptions // interface. This interface is then mapped to the concrete sql tx options // struct. func (s *BaseDB) BeginTx(ctx context.Context, opts TxOptions) (*sql.Tx, error) { + readOnly := opts.ReadOnly() + sqlOptions := sql.TxOptions{ - Isolation: sql.LevelSerializable, - ReadOnly: opts.ReadOnly(), + Isolation: txIsolationLevel( + s.BackendType, readOnly, s.WriteTxRepeatableRead, + ), + ReadOnly: readOnly, } return s.DB.BeginTx(ctx, &sqlOptions) } +// txIsolationLevel returns the isolation level that a transaction against the +// given backend should be opened with. +// +// Read-only transactions on Postgres are opened at REPEATABLE READ, which in +// Postgres is snapshot isolation: the transaction reads from a single +// consistent snapshot for its whole lifetime, taken when its first statement +// runs rather than at BEGIN. +// +// That is a real, if modest, weakening. Snapshot isolation is not +// serializability, so the reader is no longer guaranteed to observe a state +// that corresponds to some serial ordering of the writers running alongside it. +// The read-only transaction anomaly described by Fekete and O'Neil is once +// again permitted. We accept that here because our read paths only ever consume +// a point-in-time view of the database and never depended on being ordered +// against writers in other transactions. A read that feeds a later write in a +// separate transaction was never protected across that boundary at any +// isolation level. +// +// In exchange, a read-only REPEATABLE READ transaction takes no part in +// Postgres' serializable snapshot isolation conflict graph. It acquires no +// SIRead predicate locks, is not itself subject to SSI serialization failures, +// and can no longer cause a concurrent writer to be aborted as a pivot. Since +// lnd is extremely read heavy, this removes a large amount of needless abort +// pressure from the system. +// +// Read-write transactions run at SERIALIZABLE by default, but can be moved to +// REPEATABLE READ with the db.postgres.tx-isolation option. REPEATABLE READ on +// Postgres is snapshot isolation, which still rules out dirty reads, +// non-repeatable reads, phantom reads and lost updates: a transaction that +// writes a row another in-flight transaction has already written is aborted +// with a serialization failure. The one anomaly class that remains is write +// skew, where two transactions each read what the other writes but write +// disjoint sets of rows, so neither conflicts and both commit. Every write path +// that was known to be exposed to that has been hardened to either touch a +// shared row or to serialize in process, which is what makes offering this at +// all defensible. It nonetheless stays opt-in until it has accumulated soak +// time on real nodes. +// +// SQLite is always effectively serializable because it only ever admits a +// single writer, so there is nothing to gain there and we leave it alone. +func txIsolationLevel(backend BackendType, readOnly, + writeTxRepeatableRead bool) sql.IsolationLevel { + + if backend != BackendTypePostgres { + return sql.LevelSerializable + } + + if readOnly || writeTxRepeatableRead { + return sql.LevelRepeatableRead + } + + return sql.LevelSerializable +} + // Backend returns the type of the database backend used. func (s *BaseDB) Backend() BackendType { return s.BackendType diff --git a/sqldb/v2/interfaces_db_test.go b/sqldb/v2/interfaces_db_test.go new file mode 100644 index 00000000000..d8dd860897e --- /dev/null +++ b/sqldb/v2/interfaces_db_test.go @@ -0,0 +1,104 @@ +//go:build !js && !(windows && (arm || 386)) && !(linux && (ppc64 || mips || mipsle || mips64)) && !(netbsd || openbsd) + +package sqldb + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// TestBeginTxIsolationLevel asserts that the isolation level that the database +// itself reports for a transaction opened through BeginTx matches what we +// expect. On Postgres, read-only transactions run at repeatable read while +// read-write transactions remain serializable unless the opt-in knob moves +// them to repeatable read as well. We also assert the read-only flag that +// Postgres reports, so that dropping it from the tx options would be caught +// here as well. +func TestBeginTxIsolationLevel(t *testing.T) { + t.Parallel() + ctx := t.Context() + + db := NewTestDB(t, nil).GetBaseDB() + + // The knob defaults to off, which is what the first two cases below + // assert. + require.False(t, db.WriteTxRepeatableRead) + + tests := []struct { + name string + opts TxOptions + rrWrites bool + expected string + expectedFlag string + }{ + { + name: "read-only", + opts: ReadTxOpt(), + expected: "repeatable read", + expectedFlag: "on", + }, + { + name: "read-write", + opts: WriteTxOpt(), + expected: "serializable", + expectedFlag: "off", + }, + { + name: "read-only, rr writes", + opts: ReadTxOpt(), + rrWrites: true, + expected: "repeatable read", + expectedFlag: "on", + }, + { + name: "read-write, rr writes", + opts: WriteTxOpt(), + rrWrites: true, + expected: "repeatable read", + expectedFlag: "off", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + // Nothing else reads this field while a transaction is + // being opened, and these sub tests are not run in + // parallel, so it's safe to flip the knob in place + // rather than to bring up a second database. + db.WriteTxRepeatableRead = test.rrWrites + + tx, err := db.BeginTx(ctx, test.opts) + + // SQLite has no notion of a transaction isolation + // level and its driver ignores the one we request + // outright. All we can assert there is that opening + // the transaction still succeeds, which is what would + // break if the driver ever started rejecting the + // levels we ask for. + require.NoError(t, err) + defer func() { + require.NoError(t, tx.Rollback()) + }() + + if db.Backend() != BackendTypePostgres { + require.Equal(t, BackendTypeSqlite, db.Backend()) + return + } + + var level string + row := tx.QueryRowContext( + ctx, "SHOW transaction_isolation", + ) + require.NoError(t, row.Scan(&level)) + require.Equal(t, test.expected, level) + + var readOnly string + row = tx.QueryRowContext( + ctx, "SHOW transaction_read_only", + ) + require.NoError(t, row.Scan(&readOnly)) + require.Equal(t, test.expectedFlag, readOnly) + }) + } +} diff --git a/sqldb/v2/interfaces_test.go b/sqldb/v2/interfaces_test.go index 024b367465c..de68c33f492 100644 --- a/sqldb/v2/interfaces_test.go +++ b/sqldb/v2/interfaces_test.go @@ -46,3 +46,91 @@ func TestTransactionExecutorBackend(t *testing.T) { require.Equal(t, BackendTypePostgres, executor.Backend()) } + +// TestTxIsolationLevel tests that the isolation level of a transaction is only +// relaxed for read-only transactions on Postgres, and for read-write +// transactions on Postgres once the opt-in knob is set. Every other combination +// must remain fully serializable. +func TestTxIsolationLevel(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + backend BackendType + readOnly bool + rrWrites bool + expected sql.IsolationLevel + }{ + { + name: "postgres read-only", + backend: BackendTypePostgres, + readOnly: true, + expected: sql.LevelRepeatableRead, + }, + { + name: "postgres read-write", + backend: BackendTypePostgres, + readOnly: false, + expected: sql.LevelSerializable, + }, + { + name: "postgres read-only, rr writes", + backend: BackendTypePostgres, + readOnly: true, + rrWrites: true, + expected: sql.LevelRepeatableRead, + }, + { + name: "postgres read-write, rr writes", + backend: BackendTypePostgres, + readOnly: false, + rrWrites: true, + expected: sql.LevelRepeatableRead, + }, + { + name: "sqlite read-only", + backend: BackendTypeSqlite, + readOnly: true, + expected: sql.LevelSerializable, + }, + { + name: "sqlite read-write", + backend: BackendTypeSqlite, + readOnly: false, + expected: sql.LevelSerializable, + }, + { + name: "sqlite read-write, rr writes", + backend: BackendTypeSqlite, + readOnly: false, + rrWrites: true, + expected: sql.LevelSerializable, + }, + { + name: "unknown read-only", + backend: BackendTypeUnknown, + readOnly: true, + expected: sql.LevelSerializable, + }, + { + name: "unknown read-write, rr writes", + backend: BackendTypeUnknown, + readOnly: false, + rrWrites: true, + expected: sql.LevelSerializable, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + require.Equal( + t, test.expected, txIsolationLevel( + test.backend, test.readOnly, + test.rrWrites, + ), + ) + }) + } +} diff --git a/sqldb/v2/postgres.go b/sqldb/v2/postgres.go index 833c7445ab1..a3d09d04413 100644 --- a/sqldb/v2/postgres.go +++ b/sqldb/v2/postgres.go @@ -193,12 +193,15 @@ func NewPostgresStore(cfg *PostgresConfig) (*PostgresStore, error) { db.SetConnMaxLifetime(connMaxLifetime) db.SetConnMaxIdleTime(connMaxIdleTime) + rrWrites := effectiveCfg.WriteTxRepeatableRead() + return &PostgresStore{ cfg: &effectiveCfg, BaseDB: &BaseDB{ - DB: db, - BackendType: BackendTypePostgres, - SkipMigrations: effectiveCfg.SkipMigrations, + DB: db, + BackendType: BackendTypePostgres, + SkipMigrations: effectiveCfg.SkipMigrations, + WriteTxRepeatableRead: rrWrites, }, }, nil } diff --git a/sqldb/v2/sqlerrors.go b/sqldb/v2/sqlerrors.go index 8f132766ef3..e28abf59a38 100644 --- a/sqldb/v2/sqlerrors.go +++ b/sqldb/v2/sqlerrors.go @@ -134,21 +134,21 @@ func parsePostgresError(pqErr *pgconn.PgError) error { // Unable to serialize the transaction, so we'll need to try again. case pgerrcode.SerializationFailure: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), } // In failed SQL transaction because we didn't catch a previous // serialization error, so return this one as a serialization error. case pgerrcode.InFailedSQLTransaction: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), } // Deadlock detedted because of a serialization error, so return this // one as a serialization error. case pgerrcode.DeadlockDetected: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), } // Handle schema error. diff --git a/sqldb/v2/sqlerrors_no_sqlite.go b/sqldb/v2/sqlerrors_no_sqlite.go index bcc548d79c6..99f410dd929 100644 --- a/sqldb/v2/sqlerrors_no_sqlite.go +++ b/sqldb/v2/sqlerrors_no_sqlite.go @@ -43,21 +43,21 @@ func parsePostgresError(pqErr *pgconn.PgError) error { // Unable to serialize the transaction, so we'll need to try again. case pgerrcode.SerializationFailure: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), } // In failed SQL transaction because we didn't catch a previous // serialization error, so return this one as a serialization error. case pgerrcode.InFailedSQLTransaction: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), } // Deadlock detected because of a serialization error, so return this // one as a serialization error. case pgerrcode.DeadlockDetected: return &ErrSerializationError{ - DBError: pqErr, + DBError: withPostgresDetail(pqErr), } // Handle schema error. diff --git a/sqldb/v2/sqlerrors_postgres.go b/sqldb/v2/sqlerrors_postgres.go new file mode 100644 index 00000000000..4fb319c6118 --- /dev/null +++ b/sqldb/v2/sqlerrors_postgres.go @@ -0,0 +1,34 @@ +package sqldb + +import ( + "fmt" + + "github.com/jackc/pgx/v5/pgconn" +) + +// withPostgresDetail returns an error that also carries the Detail field of the +// given Postgres error, if it has one. The error string that the driver builds +// leaves that field out, and for a transaction conflict it is the only thing +// that says which kind of conflict was hit: an abort that only serializable +// isolation raises spells out the reason code of the pivot that was cancelled, +// whereas a plain write-write conflict just reports a concurrent update. Being +// able to tell the two apart is what says how much of the abort pressure a +// deployment sees would go away by relaxing the isolation level. +// +// NOTE: This must only ever be applied to the errors that report a transaction +// conflict. For any other error, and for a constraint violation in particular, +// the detail spells out the offending column values, which for our schemas +// means things like payment session keys, invoice hashes and raw key/value +// bucket keys. None of that belongs in a log line, let alone in an error that +// may travel back over an RPC. +// +// This lives in its own file, free of build tags, so that the two build +// specific copies of the error parsing below it share a single definition of +// it. +func withPostgresDetail(pqErr *pgconn.PgError) error { + if pqErr.Detail == "" { + return pqErr + } + + return fmt.Errorf("%w (detail: %s)", pqErr, pqErr.Detail) +} diff --git a/watchtower/wtdb/client_db.go b/watchtower/wtdb/client_db.go index 9fa03a3c17a..717e9bac2d7 100644 --- a/watchtower/wtdb/client_db.go +++ b/watchtower/wtdb/client_db.go @@ -1717,12 +1717,26 @@ func (c *ClientDB) DeleteSession(id SessionID) error { // sessions that are now considered closable due to the close of this channel. // The details for this channel will be deleted from the DB if there are no more // sessions in the DB that contain updates for this channel. +// +// This method, together with AckUpdate, maintains the following invariant: +// every session that has acked updates for a channel is evaluated for +// closability at least once after that channel has been marked closed. Without +// it a session may hold on to the tower's storage forever, since a session is +// only ever deleted once it has been found closable. The two halves of the +// invariant are: +// +// 1. This method evaluates every session that had acked an update for the +// channel by the time it runs. +// +// 2. AckUpdate evaluates its own session whenever it acks an update for a +// channel that is already closed, which covers the sessions that this +// method could not have known about. func (c *ClientDB) MarkChannelClosed(chanID lnwire.ChannelID, blockHeight uint32) ([]SessionID, error) { var closableSessions []SessionID err := kvdb.Update(c.db, func(tx kvdb.RwTx) error { - sessionsBkt := tx.ReadBucket(cSessionBkt) + sessionsBkt := tx.ReadWriteBucket(cSessionBkt) if sessionsBkt == nil { return ErrUninitializedDB } @@ -1768,6 +1782,21 @@ func (c *ClientDB) MarkChannelClosed(chanID lnwire.ChannelID, return err } + // The set of sessions we're about to iterate over is only read + // here, while AckUpdate adds to it the first time a session + // acks an update for this channel. Under snapshot isolation + // that's write skew: a concurrent AckUpdate wouldn't see the + // closed height we just wrote, we wouldn't see the session it + // added, and both of us would commit having each assumed that + // the other would evaluate that session. We therefore write the + // channel's db-ID row, which AckUpdate writes as well whenever + // it adds a session here, so that one of the two transactions + // is instead aborted with a retryable serialization error. + err = touchKey(chanDetails, cChanDBID) + if err != nil { + return err + } + // Now iterate through all the sessions of the channel to check // if any of them are closeable. return chanSessIDsBkt.ForEach(func(sessDBID, _ []byte) error { @@ -1784,6 +1813,27 @@ func (c *ClientDB) MarkChannelClosed(chanID lnwire.ChannelID, return err } + // The closability of a session is decided from the + // session's own state, which CommitUpdate and AckUpdate + // both change, and both of them write the session's + // body row when they do. So for the same reason as + // above, we write that row here to make this evaluation + // collide with any concurrent change of the session it + // is evaluating. Otherwise a session could be marked + // closable off a view of it that a concurrent + // transaction has already invalidated, and a session + // that still has updates to make good on could end up + // being deleted. + sessBkt := sessionsBkt.NestedReadWriteBucket(sID[:]) + if sessBkt == nil { + return ErrSessionNotFound + } + + err = touchKey(sessBkt, cSessionBody) + if err != nil { + return err + } + isClosable, err := isSessionClosable( sessionsBkt, chanDetailsBkt, chanIDIndexBkt, sID, @@ -2054,10 +2104,34 @@ func (c *ClientDB) CommitUpdate(id *SessionID, // AckUpdate persists an acknowledgment for a given (session, seqnum) pair. This // removes the update from the set of committed updates, and validates the // lastApplied value returned from the tower. +// +// If the channel that the acked update belongs to has already been closed, then +// the session is evaluated for closability here, since MarkChannelClosed has +// already made its pass over that channel's sessions. See the comment on +// MarkChannelClosed for the invariant that the two of them maintain together. func (c *ClientDB) AckUpdate(id *SessionID, seqNum uint16, lastApplied uint16) error { - return kvdb.Update(c.db, func(tx kvdb.RwTx) error { + // RangeIndex.Add mutates the in-memory range index as part of the + // transaction below, but that mutation is not rolled back along with + // the transaction. Were we to retry on top of a range index that a + // rolled back attempt had already mutated, the retry would find the + // height to be covered already, compute no changes to apply, and the + // ack would never make it to disk even though everything else in the + // transaction did. So we keep track of the range index that the + // transaction dirtied and drop it before each retry, which forces it to + // be read back from the database. + var dirtyChan *lnwire.ChannelID + evictDirty := func() { + if dirtyChan == nil { + return + } + + c.evictRangeIndex(*id, *dirtyChan) + dirtyChan = nil + } + + err := kvdb.Update(c.db, func(tx kvdb.RwTx) error { sessions := tx.ReadWriteBucket(cSessionBkt) if sessions == nil { return ErrUninitializedDB @@ -2181,37 +2255,39 @@ func (c *ClientDB) AckUpdate(id *SessionID, seqNum uint16, } // In the rare chance that this session only has rogue - // updates, we check here if the count is equal to the - // MaxUpdate of the session. If it is, then we mark the - // session as closable. - if rogueCount != uint64(session.Policy.MaxUpdates) { - return nil - } - - // Before we mark the session as closable, we do a - // sanity check to ensure that this session has no - // acked-update index. - sessionAckRanges := sessionBkt.NestedReadBucket( - cSessionAckRangeIndex, - ) - if sessionAckRanges != nil { - return fmt.Errorf("session(%s) has an "+ - "acked ranges index but has a rogue "+ - "count indicating saturation", - session.ID) - } - - closableSessBkt := tx.ReadWriteBucket( - cClosableSessionsBkt, - ) - if closableSessBkt == nil { - return ErrUninitializedDB + // updates, the count reaching the MaxUpdates of the + // session is what exhausts it. Before we go on to + // consider such a session closable, we do a sanity + // check to ensure that it has no acked-update index. + if rogueCount == uint64(session.Policy.MaxUpdates) { + sessionAckRanges := sessionBkt.NestedReadBucket( + cSessionAckRangeIndex, + ) + if sessionAckRanges != nil { + return fmt.Errorf("session(%s) has an "+ + "acked ranges index but has a "+ + "rogue count indicating "+ + "saturation", session.ID) + } } + // This ack may well have been the one that made the + // session closable, either because the rogue count just + // saturated it or because it was the last un-acked + // update of a session whose channels have all been + // closed already. The channel this update was for is + // gone from the DB, so there is no close height to + // attribute the session to and we use a zero height, + // which is what this branch has always done. var height [4]byte byteOrder.PutUint32(height[:], 0) - return closableSessBkt.Put(dbSessIDBytes, height[:]) + c.maybeMarkSessionClosable( + tx, sessions, chanDetailsBkt, id, + dbSessIDBytes, height[:], + ) + + return nil } else if err != nil { return err } @@ -2233,19 +2309,85 @@ func (c *ClientDB) AckUpdate(id *SessionID, seqNum uint16, return ErrChannelNotRegistered } - err = putChannelToSessionMapping(chanDetails, dbSessionID) + isNewSession, err := putChannelToSessionMapping( + chanDetails, dbSessionID, + ) if err != nil { return err } + // Adding this session to the channel's set of sessions means + // that MarkChannelClosed now has one more session to evaluate + // for closability when this channel is closed. That set is only + // ever read there, so under snapshot isolation a close of this + // channel that runs concurrently with us would not see the row + // we just wrote, and we would not see the closed height that it + // wrote. Both transactions would commit and this session would + // never be evaluated for this channel, which is how a session + // ends up leaking: if this was the last open channel of the + // session, then nothing else will ever mark it closable. + // + // To rule that out, we write a row that MarkChannelClosed + // writes as well, which makes the two transactions collide and + // one of them retry. + if isNewSession { + err = touchKey(chanDetails, cChanDBID) + if err != nil { + return err + } + } + // Get the range index for the given session-channel pair. index, err := c.getRangeIndex(tx, *id, chanID) if err != nil { return err } - return index.Add(height, rangesBkt) - }, func() {}) + // From here on the in-memory copy of this range index no longer + // matches what is on disk unless this transaction commits, so + // it has to be dropped if we end up retrying. + dirtyChan = &chanID + + err = index.Add(height, rangesBkt) + if err != nil { + return err + } + + // The channel may already have been closed by the time this ack + // reaches us, either because the ack genuinely came in late or + // because we lost the race described above and are now running + // for a second time. Either way MarkChannelClosed has already + // made its pass over the channel's sessions without this + // session being part of it, so we have to evaluate this session + // ourselves to uphold the invariant that every session with + // acked updates for a closed channel is checked for closability + // at least once after that channel was closed. + closedHeight := chanDetails.Get(cChanClosedHeight) + if len(closedHeight) == 0 { + return nil + } + + // The height is written to another bucket, so it must not alias + // the database's memory. + heightCopy := make([]byte, len(closedHeight)) + copy(heightCopy, closedHeight) + + c.maybeMarkSessionClosable( + tx, sessions, chanDetailsBkt, id, dbSessIDBytes, + heightCopy, + ) + + return nil + }, evictDirty) + if err != nil { + // The transaction gave up for good, so whatever it left behind + // in the in-memory range index has to go as well. + evictDirty() + + return err + } + + return nil } // GetDBQueue returns a BackupID Queue instance under the given namespace. @@ -2374,23 +2516,121 @@ func (c *ClientDB) DeleteCommittedUpdates(id *SessionID) error { } // putChannelToSessionMapping adds the given session ID to a channel's -// cChanSessions bucket. +// cChanSessions bucket. The returned boolean indicates whether the session was +// newly added to the channel's set of sessions. func putChannelToSessionMapping(chanDetails kvdb.RwBucket, - dbSessID uint64) error { + dbSessID uint64) (bool, error) { chanSessIDsBkt, err := chanDetails.CreateBucketIfNotExists( cChanSessions, ) if err != nil { - return err + return false, err } b, err := writeBigSize(dbSessID) + if err != nil { + return false, err + } + + isNew := chanSessIDsBkt.Get(b) == nil + + return isNew, chanSessIDsBkt.Put(b, []byte{1}) +} + +// touchKey re-writes the value stored under the given key with the very same +// value. +// +// A write like this is a no-op as far as the contents of the database are +// concerned, but it is not a no-op as far as the database's concurrency control +// is concerned: it turns a row that a transaction would otherwise only read +// into part of that transaction's write set. Two transactions that touch the +// same row then collide, and the loser is aborted with a retryable +// serialization error, instead of both of them committing on top of a view of +// the world that the other one has already invalidated. +// +// This is only needed for the SQL backends running at snapshot isolation. +// Bolt has a single writer, so there is nothing to collide with there. +func touchKey(bkt kvdb.RwBucket, key []byte) error { + value := bkt.Get(key) + if value == nil { + return fmt.Errorf("no value found for key %x", key) + } + + // The slice handed back by Get may point straight into the database's + // own memory, so we copy it before writing it back. + valueCopy := make([]byte, len(value)) + copy(valueCopy, value) + + return bkt.Put(key, valueCopy) +} + +// maybeMarkSessionClosable evaluates whether the given session has become +// closable and, if it has, records it in the closable sessions bucket under the +// given height. A failure to do so is logged rather than returned. +// +// The evaluation is deliberately best effort. It runs in the same transaction +// as an update acknowledgment, and that acknowledgment must be persisted no +// matter what: the tower has the update either way, and the session queue +// treats a failed AckUpdate as fatal, so a session that can't be evaluated for +// closability would take its queue down with it and stay down across the retry. +// Failing to evaluate a session only means it isn't reclaimed as early as it +// could have been, which is the behaviour we had before it was evaluated here +// at all. +func (c *ClientDB) maybeMarkSessionClosable(tx kvdb.RwTx, sessionsBkt, + chanDetailsBkt kvdb.RBucket, id *SessionID, dbSessIDBytes, + height []byte) { + + err := markSessionClosable( + tx, sessionsBkt, chanDetailsBkt, id, dbSessIDBytes, height, + ) + if err != nil { + log.Errorf("Could not determine if session %s has become "+ + "closable: %v", id, err) + } +} + +// evictRangeIndex drops the in-memory range index of the given session-channel +// pair, so that the next caller that needs it reads it back from the database. +func (c *ClientDB) evictRangeIndex(sID SessionID, chanID lnwire.ChannelID) { + c.ackedRangeIndexMu.Lock() + defer c.ackedRangeIndexMu.Unlock() + + delete(c.ackedRangeIndex[sID], chanID) +} + +// markSessionClosable evaluates whether the given session has become closable +// and, if it has, records it in the closable sessions bucket under the given +// height. +// +// NOTE: unlike MarkChannelClosed, this doesn't hand the session back to the +// caller, so the session is only picked up by the closable session handler on +// the next startup. +func markSessionClosable(tx kvdb.RwTx, sessionsBkt, chanDetailsBkt kvdb.RBucket, + id *SessionID, dbSessIDBytes, height []byte) error { + + chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt) + if chanIDIndexBkt == nil { + return ErrUninitializedDB + } + + isClosable, err := isSessionClosable( + sessionsBkt, chanDetailsBkt, chanIDIndexBkt, id, + ) if err != nil { return err } - return chanSessIDsBkt.Put(b, []byte{1}) + if !isClosable { + return nil + } + + closableSessBkt := tx.ReadWriteBucket(cClosableSessionsBkt) + if closableSessBkt == nil { + return ErrUninitializedDB + } + + return closableSessBkt.Put(dbSessIDBytes, height) } // getClientSessionBody loads the body of a ClientSession from the sessions diff --git a/watchtower/wtdb/client_db_internal_test.go b/watchtower/wtdb/client_db_internal_test.go new file mode 100644 index 00000000000..7ed18c22393 --- /dev/null +++ b/watchtower/wtdb/client_db_internal_test.go @@ -0,0 +1,298 @@ +package wtdb + +import ( + "net" + "sync" + "testing" + + "github.com/btcsuite/btcd/btcec/v2" + "github.com/lightningnetwork/lnd/kvdb" + "github.com/lightningnetwork/lnd/lnwire" + "github.com/lightningnetwork/lnd/watchtower/blob" + "github.com/lightningnetwork/lnd/watchtower/wtpolicy" + "github.com/stretchr/testify/require" +) + +// newTestClientDB opens a bolt backed client DB for testing. +func newTestClientDB(t *testing.T) *ClientDB { + t.Helper() + + backend, err := NewBoltBackendCreator( + true, t.TempDir(), "wtclient.db", + )(&kvdb.BoltConfig{DBTimeout: kvdb.DefaultDBTimeout}) + require.NoError(t, err) + + db, err := OpenClientDB(backend) + require.NoError(t, err) + + t.Cleanup(func() { + require.NoError(t, db.Close()) + }) + + return db +} + +// newTestClientDBOnTestBackend opens a client DB on whichever kvdb backend the +// current build selects. Unlike the bolt backed helper above, this gives the +// SQL backends a chance to run the test, which is where concurrent +// transactions actually interleave. +func newTestClientDBOnTestBackend(t *testing.T) *ClientDB { + t.Helper() + + backend, cleanup, err := kvdb.GetTestBackend(t.TempDir(), "wtclient") + require.NoError(t, err) + t.Cleanup(cleanup) + + db, err := OpenClientDB(backend) + require.NoError(t, err) + + t.Cleanup(func() { + require.NoError(t, db.Close()) + }) + + return db +} + +// newTestSession registers a tower and a session with the given max updates +// against it, and returns the session. +func newTestSession(t *testing.T, db *ClientDB, + maxUpdates uint16) *ClientSession { + + t.Helper() + + privKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + tower, err := db.CreateTower(&lnwire.NetAddress{ + IdentityKey: privKey.PubKey(), + Address: &net.TCPAddr{IP: []byte{0x01, 0x00, 0x00, 0x00}}, + }) + require.NoError(t, err) + + const blobType = blob.TypeAltruistCommit + + keyIndex, err := db.NextSessionKeyIndex(tower.ID, blobType, false) + require.NoError(t, err) + + sessionPriv, err := btcec.NewPrivateKey() + require.NoError(t, err) + + session := &ClientSession{ + ID: NewSessionIDFromPubKey(sessionPriv.PubKey()), + ClientSessionBody: ClientSessionBody{ + TowerID: tower.ID, + KeyIndex: keyIndex, + Policy: wtpolicy.Policy{ + TxPolicy: wtpolicy.TxPolicy{ + BlobType: blobType, + }, + MaxUpdates: maxUpdates, + }, + RewardPkScript: []byte{0x01, 0x02, 0x03}, + }, + } + require.NoError(t, db.CreateClientSession(session)) + + return session +} + +// TestAckUpdateRangeIndexEviction asserts that the in-memory range index of a +// session-channel pair can be dropped, so that it is read back from the +// database the next time it is needed. +// +// AckUpdate relies on this: RangeIndex.Add mutates the in-memory index as part +// of AckUpdate's transaction, but that mutation is not rolled back along with +// the transaction. An attempt that is retried on top of the mutated index would +// find the height covered already, apply nothing to the database, and leave the +// ack persisted nowhere while the rest of the transaction committed. +func TestAckUpdateRangeIndexEviction(t *testing.T) { + t.Parallel() + + db := newTestClientDB(t) + session := newTestSession(t, db, 5) + + var chanID lnwire.ChannelID + copy(chanID[:], []byte{0x01, 0x02, 0x03}) + require.NoError(t, db.RegisterChannel(chanID, []byte{0x01})) + + // Commit and ack a single update, which leaves the range index for this + // session-channel pair both on disk and in memory. + const ackedHeight = 5 + update := &CommittedUpdate{ + SeqNum: 1, + CommittedUpdateBody: CommittedUpdateBody{ + BackupID: BackupID{ + ChanID: chanID, + CommitHeight: ackedHeight, + }, + EncryptedBlob: []byte{0x01, 0x02, 0x03}, + }, + } + _, err := db.CommitUpdate(&session.ID, update) + require.NoError(t, err) + require.NoError(t, db.AckUpdate(&session.ID, 1, 1)) + + // readFromDisk reads the range index of the pair straight from the + // database, bypassing the in-memory copy of it entirely. + readFromDisk := func() *RangeIndex { + var index *RangeIndex + err := kvdb.View(db.db, func(tx kvdb.RTx) error { + rangesBkt, err := getRangesReadBucket( + tx, session.ID, chanID, + ) + if err != nil { + return err + } + + index, err = readRangeIndex(rangesBkt) + + return err + }, func() {}) + require.NoError(t, err) + + return index + } + + // Both copies agree at this point. + index, err := db.getRangeIndex(nil, session.ID, chanID) + require.NoError(t, err) + require.True(t, index.IsInIndex(ackedHeight)) + require.True(t, readFromDisk().IsInIndex(ackedHeight)) + + // Now mutate the in-memory index without touching the database, which + // is the state that an attempt of AckUpdate that was rolled back leaves + // behind. + const rolledBackHeight = 9 + require.NoError(t, index.Add(rolledBackHeight, nil)) + require.True(t, index.IsInIndex(rolledBackHeight)) + require.False(t, readFromDisk().IsInIndex(rolledBackHeight)) + + // Evicting the index is what makes the next read of it agree with the + // database again. + db.evictRangeIndex(session.ID, chanID) + + index, err = db.getRangeIndex(nil, session.ID, chanID) + require.NoError(t, err) + require.True(t, index.IsInIndex(ackedHeight)) + require.False(t, index.IsInIndex(rolledBackHeight)) + + // With the stale height gone, acking an update at that height actually + // makes it to disk. Were the index still holding on to it, the ack + // would be a no-op against the database. + update = &CommittedUpdate{ + SeqNum: 2, + CommittedUpdateBody: CommittedUpdateBody{ + BackupID: BackupID{ + ChanID: chanID, + CommitHeight: rolledBackHeight, + }, + EncryptedBlob: []byte{0x04, 0x05, 0x06}, + }, + } + _, err = db.CommitUpdate(&session.ID, update) + require.NoError(t, err) + require.NoError(t, db.AckUpdate(&session.ID, 2, 2)) + + require.True(t, readFromDisk().IsInIndex(rolledBackHeight)) +} + +// TestAckUpdateRacesChannelClose asserts that a session that acks its first +// update for a channel at the very same time as that channel is being marked +// closed still ends up being evaluated for closability, no matter which of the +// two transactions gets there first. +// +// The two halves of that invariant live in different transactions: +// MarkChannelClosed evaluates the sessions it can see in the channel's set of +// sessions, and AckUpdate evaluates its own session when it finds the channel +// already closed. What ties them together is that they both write the channel's +// db-ID row, so the database can't let both of them commit on a view of the +// world the other one has already invalidated. +func TestAckUpdateRacesChannelClose(t *testing.T) { + t.Parallel() + + db := newTestClientDBOnTestBackend(t) + + var chanID lnwire.ChannelID + copy(chanID[:], []byte{0x0a, 0x0b, 0x0c}) + require.NoError(t, db.RegisterChannel(chanID, []byte{0x01})) + + // The first session acks an update for the channel up front, which is + // what keeps the channel's details around once it is closed. It is far + // from exhausted, so it never becomes closable itself. + keeper := newTestSession(t, db, 5) + keeperUpdate := &CommittedUpdate{ + SeqNum: 1, + CommittedUpdateBody: CommittedUpdateBody{ + BackupID: BackupID{ + ChanID: chanID, + CommitHeight: 1, + }, + EncryptedBlob: []byte{0x01}, + }, + } + _, err := db.CommitUpdate(&keeper.ID, keeperUpdate) + require.NoError(t, err) + require.NoError(t, db.AckUpdate(&keeper.ID, 1, 1)) + + // The racing session has a single update, so acking it both adds the + // session to the channel's set of sessions for the very first time and + // exhausts the session. + racer := newTestSession(t, db, 1) + racerUpdate := &CommittedUpdate{ + SeqNum: 1, + CommittedUpdateBody: CommittedUpdateBody{ + BackupID: BackupID{ + ChanID: chanID, + CommitHeight: 2, + }, + EncryptedBlob: []byte{0x02}, + }, + } + _, err = db.CommitUpdate(&racer.ID, racerUpdate) + require.NoError(t, err) + + // Now run the ack and the channel close at the same time. + const closeHeight = 100 + + var ( + wg sync.WaitGroup + ackErr, closeErr error + closedByMarkClose []SessionID + ) + + wg.Add(2) + go func() { + defer wg.Done() + + ackErr = db.AckUpdate(&racer.ID, 1, 1) + }() + go func() { + defer wg.Done() + + closedByMarkClose, closeErr = db.MarkChannelClosed( + chanID, closeHeight, + ) + }() + wg.Wait() + + require.NoError(t, ackErr) + require.NoError(t, closeErr) + + // Whichever of the two won, the racing session must have been found + // closable by exactly one of them, and the height it is recorded under + // is the height the channel closed at either way. + closable, err := db.ListClosableSessions() + require.NoError(t, err) + require.Contains(t, closable, racer.ID) + require.EqualValues(t, closeHeight, closable[racer.ID]) + + // The session that is still far from exhausted must not have been + // swept up along with it. + require.NotContains(t, closable, keeper.ID) + + // If the close was the one that saw the session, it hands it back to + // its caller as well. + if len(closedByMarkClose) > 0 { + require.Equal(t, []SessionID{racer.ID}, closedByMarkClose) + } +} diff --git a/watchtower/wtdb/client_db_test.go b/watchtower/wtdb/client_db_test.go index dd6479b74ee..e910dbc253e 100644 --- a/watchtower/wtdb/client_db_test.go +++ b/watchtower/wtdb/client_db_test.go @@ -1099,6 +1099,119 @@ func testMarkChannelClosed(h *clientDBHarness) { h.deleteSession(session2.ID, nil) } +// testAckAfterChannelClose asserts that a session that acks its first update +// for a channel only after that channel has been marked as closed is still +// evaluated for closability. MarkChannelClosed can't have known about such a +// session, since the session only joins the channel's set of sessions once it +// acks an update for it, so it is up to AckUpdate to evaluate the session. +// Without that, the session would never be marked closable and would hold on to +// the tower's storage forever. +func testAckAfterChannelClose(h *clientDBHarness) { + tower := h.newTower() + + // Create the channel that both of the sessions below will have updates + // for. + chanID := randChannelID(h.t) + h.registerChan(chanID, nil, nil) + + // The first session acks an update for the channel right away, which is + // what keeps the channel's details around once it is closed. + session1 := h.randSession(h.t, tower.ID, 5) + h.insertSession(session1, nil) + + update := randCommittedUpdateForChannel(h.t, chanID, 1) + lastApplied := h.commitUpdate(&session1.ID, update, nil) + h.ackUpdate(&session1.ID, 1, lastApplied, nil) + + // The second session only ever has this one update, so acking it will + // exhaust the session. + session2 := h.randSession(h.t, tower.ID, 1) + h.insertSession(session2, nil) + + update = randCommittedUpdateForChannel(h.t, chanID, 1) + lastApplied = h.commitUpdate(&session2.ID, update, nil) + + // Close the channel before the second session gets to ack its update. + // Only the first session is known to the channel at this point, and it + // isn't closable since it is not yet exhausted. + const closeHeight = 100 + sl := h.markChannelClosed(chanID, closeHeight, nil) + require.Empty(h.t, sl) + require.Empty(h.t, h.listClosableSessions(nil)) + + // Now let the second session ack its update. That both adds the session + // to the channel's set of sessions and exhausts the session, and since + // the only channel it has updates for is closed, the session is now + // closable. + h.ackUpdate(&session2.ID, 1, lastApplied, nil) + + closable := h.listClosableSessions(nil) + require.InDeltaMapValues(h.t, closable, map[wtdb.SessionID]uint32{ + session2.ID: closeHeight, + }, 0) + + // A closable session is one that may be deleted, so the tower storage + // this session occupies can now actually be reclaimed. + h.deleteSession(session2.ID, nil) + + // The first session is still not closable, since it is not exhausted. + require.Empty(h.t, h.listClosableSessions(nil)) + h.deleteSession(session1.ID, wtdb.ErrSessionNotClosable) +} + +// testRogueAckAfterChannelClose asserts that a session that is exhausted by a +// rogue ack, one for a channel whose details are already gone from the DB, is +// still evaluated for closability even though that rogue ack didn't saturate +// the session's rogue update count on its own. +func testRogueAckAfterChannelClose(h *clientDBHarness) { + tower := h.newTower() + + // Two updates are all it takes to exhaust this session. + session := h.randSession(h.t, tower.ID, 2) + h.insertSession(session, nil) + + // The first update is acked normally, so the session ends up in the + // channel's set of sessions and the channel's details survive its + // close. + chanID1 := randChannelID(h.t) + h.registerChan(chanID1, nil, nil) + + update := randCommittedUpdateForChannel(h.t, chanID1, 1) + lastApplied := h.commitUpdate(&session.ID, update, nil) + h.ackUpdate(&session.ID, 1, lastApplied, nil) + + const closeHeight = 100 + sl := h.markChannelClosed(chanID1, closeHeight, nil) + require.Empty(h.t, sl) + + // The second update is committed for another channel, but that channel + // is closed before the update is acked. Since no session ever acked an + // update for it, closing it removes its details from the DB entirely. + chanID2 := randChannelID(h.t) + h.registerChan(chanID2, nil, nil) + + update = randCommittedUpdateForChannel(h.t, chanID2, 2) + lastApplied = h.commitUpdate(&session.ID, update, nil) + + sl = h.markChannelClosed(chanID2, closeHeight, nil) + require.Empty(h.t, sl) + require.Empty(h.t, h.listClosableSessions(nil)) + + // Acking that second update is a rogue ack: the channel it belongs to + // is no longer known. It only brings the rogue count to one out of the + // session's two updates, so it doesn't saturate the session on its own, + // but it does exhaust the session, and every other channel the session + // has acked updates for is closed. The session is therefore closable. + h.ackUpdate(&session.ID, 2, lastApplied, nil) + + closable := h.listClosableSessions(nil) + require.InDeltaMapValues(h.t, closable, map[wtdb.SessionID]uint32{ + session.ID: 0, + }, 0) + + h.deleteSession(session.ID, nil) +} + // testAckUpdate asserts the behavior of AckUpdate. func testAckUpdate(h *clientDBHarness) { const blobType = blob.TypeAltruistCommit @@ -1313,6 +1426,14 @@ func TestClientDB(t *testing.T) { name: "mark channel closed", run: testMarkChannelClosed, }, + { + name: "ack after channel close", + run: testAckAfterChannelClose, + }, + { + name: "rogue ack after channel close", + run: testRogueAckAfterChannelClose, + }, { name: "rogue updates", run: testRogueUpdates,