diff --git a/channeldb/channel.go b/channeldb/channel.go index 135565c7c49..023d052c2cf 100644 --- a/channeldb/channel.go +++ b/channeldb/channel.go @@ -2450,7 +2450,7 @@ func (c *ChannelStateDB) AdvanceCommitChainTail(channel *OpenChannel, err = chanBucket.Put(remoteUnsignedLocalUpdatesKey, b2.Bytes()) if err != nil { return fmt.Errorf("unable to restore remote unsigned "+ - "local updates: %v", err) + "local updates: %w", err) } newRemoteCommit = &newCommit.Commitment diff --git a/config.go b/config.go index 5b69e943adb..1fe31a2f23e 100644 --- a/config.go +++ b/config.go @@ -1987,7 +1987,7 @@ func (c *Config) ImplementationConfig( RestRegistrar: rpcImpl, ExternalValidator: rpcImpl, DatabaseBuilder: NewDefaultDatabaseBuilder( - c, ltndLog, + c, ltndLog, interceptor.ShutdownChannel(), ), WalletConfigBuilder: rpcImpl, ChainControlBuilder: rpcImpl, @@ -1996,10 +1996,12 @@ func (c *Config) ImplementationConfig( defaultImpl := NewDefaultWalletImpl(c, ltndLog, interceptor, false) return &ImplementationCfg{ - GrpcRegistrar: defaultImpl, - RestRegistrar: defaultImpl, - ExternalValidator: defaultImpl, - DatabaseBuilder: NewDefaultDatabaseBuilder(c, ltndLog), + GrpcRegistrar: defaultImpl, + RestRegistrar: defaultImpl, + ExternalValidator: defaultImpl, + DatabaseBuilder: NewDefaultDatabaseBuilder( + c, ltndLog, interceptor.ShutdownChannel(), + ), WalletConfigBuilder: defaultImpl, ChainControlBuilder: defaultImpl, } diff --git a/config_builder.go b/config_builder.go index 25ec8401b57..3e6d9514e33 100644 --- a/config_builder.go +++ b/config_builder.go @@ -979,16 +979,24 @@ type DatabaseInstances struct { type DefaultDatabaseBuilder struct { cfg *Config logger btclog.Logger + + // quit is closed once the daemon starts shutting down. The SQL backed + // kv stores use it to abort an in-flight transaction retry loop, so + // that a transaction which keeps hitting serialization errors can't + // delay shutdown for the length of its retry budget. + quit <-chan struct{} } // NewDefaultDatabaseBuilder returns a new instance of the default database -// builder. -func NewDefaultDatabaseBuilder(cfg *Config, - logger btclog.Logger) *DefaultDatabaseBuilder { +// builder. The passed quit channel should be closed once the daemon starts +// shutting down, and may be nil in contexts where no such signal exists. +func NewDefaultDatabaseBuilder(cfg *Config, logger btclog.Logger, + quit <-chan struct{}) *DefaultDatabaseBuilder { return &DefaultDatabaseBuilder{ cfg: cfg, logger: logger, + quit: quit, } } @@ -1011,7 +1019,8 @@ func (d *DefaultDatabaseBuilder) BuildDatabase( startOpenTime := time.Now() databaseBackends, err := cfg.DB.GetBackends( - ctx, cfg.graphDatabaseDir(), cfg.networkDir, filepath.Join( + ctx, d.quit, cfg.graphDatabaseDir(), cfg.networkDir, + filepath.Join( cfg.Watchtower.TowerDir, BitcoinChainName, lncfg.NormalizeNetwork(cfg.ActiveNetParams.Name), ), cfg.WtClient.Active, cfg.Watchtower.Active, d.logger, diff --git a/docs/release-notes/release-notes-0.22.0.md b/docs/release-notes/release-notes-0.22.0.md index 750a20069cd..011e2f5a79c 100644 --- a/docs/release-notes/release-notes-0.22.0.md +++ b/docs/release-notes/release-notes-0.22.0.md @@ -55,6 +55,23 @@ the reported network statistics such as total network capacity, channel count and max out degree. +* [Fixed a bug](https://github.com/lightningnetwork/lnd/pull/10996) where a + Postgres serialization failure hit while persisting an incoming revocation + was reported to the remote peer as an `invalid revocation` protocol error, + which prompted the peer to force close the channel. Local database errors now + fail the link silently: no error is sent on the wire and no force close is + triggered, we simply disconnect and let the channel reestablish flow resync + the state. The cooperative close path no longer relays raw internal error text + to the peer either. + + The `kvdb` retry loop is also far more patient, bounding retries by a two + minute time budget rather than by a fixed count of 50 attempts. This applies + to SQLite as well as Postgres, since a busy SQLite database is classified as + the same kind of retriable error, so waiting out lock contention can now take + up to two minutes instead of roughly 46 seconds. Retries abort immediately + once the daemon starts shutting down, so the longer budget never delays + shutdown. + # New Features ## Functional Enhancements diff --git a/htlcswitch/link.go b/htlcswitch/link.go index 9e3adf0bbc2..3f7ad9167af 100644 --- a/htlcswitch/link.go +++ b/htlcswitch/link.go @@ -1286,11 +1286,15 @@ func (l *channelLink) handleChanSyncErr(err error) { default: } + // None of the cases above match a database error, so an error from our + // own database ends up in the unspecified branch. We must not ask the + // peer to recover a channel that is perfectly fine, we just recycle the + // connection and sync again. l.failf( - LinkFailureError{ + linkFailureForDBErr(err, LinkFailureError{ code: ErrRecoveryError, FailureAction: LinkFailureForceNone, - }, + }), "unable to synchronize channel states: %v", err, ) } @@ -1995,10 +1999,15 @@ func (l *channelLink) updateCommitTxOrFail(ctx context.Context) bool { return false // Any other error is treated results in an Error message being sent to - // the peer. + // the peer, unless it was our own database that let us down. default: - l.failf(LinkFailureError{code: ErrInternalError}, - "unable to update commitment: %v", err) + l.failf( + linkFailureForDBErr(err, LinkFailureError{ + code: ErrInternalError, + }), + "unable to update commitment: %v", err, + ) + return false } @@ -3040,8 +3049,13 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg) { fwdPkg.ID(), decodeReqs, reforward, ) if sphinxErr != nil { - l.failf(LinkFailureError{code: ErrInternalError}, - "unable to decode hop iterators: %v", sphinxErr) + l.failf( + linkFailureForDBErr(sphinxErr, LinkFailureError{ + code: ErrInternalError, + }), + "unable to decode hop iterators: %v", sphinxErr, + ) + return } @@ -3193,9 +3207,13 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg) { heightNow, pld, ) if err != nil { - l.failf(LinkFailureError{ - code: ErrInternalError, - }, "%v", err) + l.failf( + linkFailureForDBErr( + err, LinkFailureError{ + code: ErrInternalError, + }, + ), "%v", err, + ) return } @@ -3367,8 +3385,13 @@ func (l *channelLink) processRemoteAdds(fwdPkg *channeldb.FwdPkg) { if fwdPkg.State == channeldb.FwdStateLockedIn { err := l.channel.SetFwdFilter(fwdPkg.Height, fwdPkg.FwdFilter) if err != nil { - l.failf(LinkFailureError{code: ErrInternalError}, - "unable to set fwd filter: %v", err) + l.failf( + linkFailureForDBErr(err, LinkFailureError{ + code: ErrInternalError, + }), + "unable to set fwd filter: %v", err, + ) + return } } @@ -3798,6 +3821,26 @@ func (l *channelLink) failf(linkErr LinkFailureError, format string, l.log.Errorf("failing link: %s with error: %v", reason, linkErr) + // A database error fails the link quietly, without so much as a message + // to the peer, so on its own it looks just like a peer that keeps + // flapping. Once we've seen a few of them in short order, say so + // plainly, since at that point the database itself is the story. + // + // NOTE: We deliberately don't log this at the critical level. In lnd a + // critical log requests a daemon shutdown, and tearing the node down + // over a contended database would be a worse outcome than the failure + // we're reporting. The kvdb layer avoids critical logs for the same + // class of error, see catchPanic in kvdb/sqlbase. + if linkErr.code == ErrInternalDBError { + failures := linkDBFailures.record(time.Now()) + if failures >= dbFailureEscalation { + l.log.Errorf("Failed %v links within %v because of "+ + "local database errors, the database may be "+ + "unhealthy: %v", failures, dbFailureWindow, + reason) + } + } + // Set failed, such that we won't process any more updates, and notify // the peer about the failure. l.failed = true @@ -4035,11 +4078,15 @@ func (l *channelLink) resumeLink(ctx context.Context) error { l.failf(LinkFailureError{code: ErrCircuitError}, "temporary circuit error: %v", err) - // A non-nil error was encountered, send an Error message to - // the peer. + // A non-nil error was encountered, send an Error message to the peer, + // unless it was our own database that let us down. default: - l.failf(LinkFailureError{code: ErrInternalError}, - "unable to resolve fwd pkgs: %v", err) + l.failf( + linkFailureForDBErr(err, LinkFailureError{ + code: ErrInternalError, + }), + "unable to resolve fwd pkgs: %v", err, + ) } return err @@ -4112,8 +4159,12 @@ func (l *channelLink) processRemoteUpdateAddHTLC( // event that we know the preimage. index, err := l.channel.ReceiveHTLC(msg) if err != nil { - l.failf(LinkFailureError{code: ErrInvalidUpdate}, - "unable to handle upstream add HTLC: %v", err) + l.failf( + linkFailureForDBErr(err, LinkFailureError{ + code: ErrInvalidUpdate, + }), + "unable to handle upstream add HTLC: %v", err, + ) return err } @@ -4151,12 +4202,20 @@ func (l *channelLink) processRemoteUpdateFulfillHTLC( return err } + // NOTE: The wrapper below is inert today, because ReceiveHTLCSettle only + // touches in-memory state and so can never return a database error. It + // is here so that this site doesn't get missed if that changes. Note + // that the calculus is different here than on the other paths: this + // failure force closes on purpose, because a peer that reveals a bad + // preimage has to be taken on-chain. Anyone adding a database write to + // ReceiveHTLCSettle needs to make sure a genuine bad preimage still + // reaches the chain rather than being classified as our own fault. if err := l.channel.ReceiveHTLCSettle(pre, idx); err != nil { l.failf( - LinkFailureError{ + linkFailureForDBErr(err, LinkFailureError{ code: ErrInvalidUpdate, FailureAction: LinkFailureForceClose, - }, + }), "unable to handle upstream settle HTLC: %v", err, ) @@ -4244,8 +4303,12 @@ func (l *channelLink) processRemoteUpdateFailMalformedHTLC( // usual HTLC fail message. err := l.channel.ReceiveFailHTLC(msg.ID, b.Bytes()) if err != nil { - l.failf(LinkFailureError{code: ErrInvalidUpdate}, - "unable to handle upstream fail HTLC: %v", err) + l.failf( + linkFailureForDBErr(err, LinkFailureError{ + code: ErrInvalidUpdate, + }), + "unable to handle upstream fail HTLC: %v", err, + ) return err } @@ -4285,8 +4348,12 @@ func (l *channelLink) processRemoteUpdateFailHTLC( idx := msg.ID err := l.channel.ReceiveFailHTLC(idx, msg.Reason[:]) if err != nil { - l.failf(LinkFailureError{code: ErrInvalidUpdate}, - "unable to handle upstream fail HTLC: %v", err) + l.failf( + linkFailureForDBErr(err, LinkFailureError{ + code: ErrInvalidUpdate, + }), + "unable to handle upstream fail HTLC: %v", err, + ) return err } @@ -4310,7 +4377,9 @@ func (l *channelLink) processRemoteCommitSig(ctx context.Context, err := l.cfg.PreimageCache.AddPreimages(l.uncommittedPreimages...) if err != nil { l.failf( - LinkFailureError{code: ErrInternalError}, + linkFailureForDBErr(err, LinkFailureError{ + code: ErrInternalError, + }), "unable to add preimages=%v to cache: %v", l.uncommittedPreimages, err, ) @@ -4354,11 +4423,11 @@ func (l *channelLink) processRemoteCommitSig(ctx context.Context, sendData = []byte(err.Error()) } l.failf( - LinkFailureError{ + linkFailureForDBErr(err, LinkFailureError{ code: ErrInvalidCommitment, FailureAction: LinkFailureForceClose, SendData: sendData, - }, + }), "ChannelPoint(%v): unable to accept new "+ "commitment: %v", l.channel.ChannelPoint(), err, @@ -4383,11 +4452,11 @@ func (l *channelLink) processRemoteCommitSig(ctx context.Context, // resolve itself in case our db was just busy not accepting new // transactions. l.failf( - LinkFailureError{ + linkFailureForDBErr(err, LinkFailureError{ code: ErrInternalError, Warning: true, FailureAction: LinkFailureDisconnect, - }, + }), "ChannelPoint(%v): unable to accept new "+ "commitment: %v", l.channel.ChannelPoint(), err, @@ -4482,11 +4551,16 @@ func (l *channelLink) processRemoteRevokeAndAck(ctx context.Context, fwdPkg, remoteHTLCs, err := l.channel.ReceiveRevocation(msg) if err != nil { // TODO(halseth): force close? + // + // NOTE: If the revocation could not be persisted because our + // own database is busy, then we must not blame the peer for it. + // We only recycle the connection in that case and let the + // channel reestablish flow sort the state out. l.failf( - LinkFailureError{ + linkFailureForDBErr(err, LinkFailureError{ code: ErrInvalidRevocation, FailureAction: LinkFailureDisconnect, - }, + }), "unable to accept revocation: %v", err, ) @@ -4521,9 +4595,12 @@ func (l *channelLink) processRemoteRevokeAndAck(ctx context.Context, &chanID, state.RemoteCommitment.CommitHeight-1, ) if err != nil { - l.failf(LinkFailureError{ - code: ErrInternalError, - }, "unable to queue breach backup: %v", err) + l.failf( + linkFailureForDBErr(err, LinkFailureError{ + code: ErrInternalError, + }), + "unable to queue breach backup: %v", err, + ) return err } @@ -4604,8 +4681,13 @@ func (l *channelLink) processRemoteUpdateFee(msg *lnwire.UpdateFee) error { // We received fee update from peer. If we are the initiator we will // fail the channel, if not we will apply the update. if err := l.channel.ReceiveUpdateFee(fee); err != nil { - l.failf(LinkFailureError{code: ErrInvalidUpdate}, - "error receiving fee update: %v", err) + l.failf( + linkFailureForDBErr(err, LinkFailureError{ + code: ErrInvalidUpdate, + }), + "error receiving fee update: %v", err, + ) + return err } diff --git a/htlcswitch/link_test.go b/htlcswitch/link_test.go index 28760897551..1fc1d5f6b07 100644 --- a/htlcswitch/link_test.go +++ b/htlcswitch/link_test.go @@ -6,6 +6,7 @@ import ( "crypto/rand" "crypto/sha256" "encoding/binary" + "errors" "fmt" "io" prand "math/rand" @@ -39,6 +40,7 @@ import ( "github.com/lightningnetwork/lnd/lnwallet" "github.com/lightningnetwork/lnd/lnwallet/chainfee" "github.com/lightningnetwork/lnd/lnwire" + "github.com/lightningnetwork/lnd/sqldb" "github.com/lightningnetwork/lnd/ticker" "github.com/stretchr/testify/require" ) @@ -5784,6 +5786,178 @@ func (m *mockFailLoadFwdPkgStore) LoadFwdPkgs( return nil, fmt.Errorf("failing LoadFwdPkgs") } +// mockFailAdvanceTailStore wraps a real channel state store and overrides only +// AdvanceCommitChainTail. This lets us inject a failure on the exact write that +// persists an incoming revocation, without touching the rest of the store. +type mockFailAdvanceTailStore struct { + cstate.Store + + // failErr is the error that AdvanceCommitChainTail returns. + failErr error +} + +// AdvanceCommitChainTail fails the write that persists an incoming revocation. +func (m *mockFailAdvanceTailStore) AdvanceCommitChainTail(*cstate.OpenChannel, + *cstate.FwdPkg, []cstate.LogUpdate, uint32, uint32) error { + + return m.failErr +} + +// TestChannelLinkFailRevocationDBError tests that a failure to persist an +// incoming revocation because of a local database problem fails the link +// without blaming our peer. Reporting such a failure on the wire is what made +// peers force close the channel in issue #10995, even though a reconnect and a +// channel reestablish would have resolved the state cleanly. +func TestChannelLinkFailRevocationDBError(t *testing.T) { + t.Parallel() + + // serializationErr mimics the error postgres hands us when a + // transaction couldn't be serialized against the other concurrent + // transactions. + serializationErr := sqldb.MapSQLError(errors.New("ERROR: could not " + + "serialize access due to read/write dependencies among " + + "transactions (SQLSTATE 40001)")) + + tests := []struct { + name string + + // failErr is the error the channel state store returns when the + // link tries to persist the incoming revocation. + failErr error + + // expCode is the failure code we expect the link to fail with. + expCode errorCode + + // expSendToPeer is whether we expect the link failure to be + // reported to our peer on the wire. + expSendToPeer bool + }{ + { + name: "serialization error", + failErr: fmt.Errorf("unable to restore remote "+ + "unsigned local updates: %w", serializationErr), + expCode: ErrInternalDBError, + expSendToPeer: false, + }, + { + name: "retries exceeded", + failErr: fmt.Errorf("%w: %w", sqldb.ErrRetriesExceeded, + serializationErr), + expCode: ErrInternalDBError, + expSendToPeer: false, + }, + { + name: "non-db error still blames the peer", + failErr: errors.New("revocation key mismatch"), + expCode: ErrInvalidRevocation, + expSendToPeer: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + assertRevocationDBFailure( + t, test.failErr, test.expCode, + test.expSendToPeer, + ) + }) + } +} + +// assertRevocationDBFailure drives a full commitment dance up to the point +// where Alice receives a revocation from Bob, with Alice's channel state store +// rigged to fail that write, and then asserts on the resulting link failure. +func assertRevocationDBFailure(t *testing.T, failErr error, expCode errorCode, + expSendToPeer bool) { + + t.Helper() + + const chanAmt = btcutil.SatoshiPerBitcoin * 5 + harness, err := newSingleLinkTestHarness(t, chanAmt, 0) + require.NoError(t, err) + + //nolint:forcetypeassert + coreLink := harness.aliceLink.(*channelLink) + + // Rig Alice's channel state store so that persisting an incoming + // revocation fails with the error under test. + state := coreLink.channel.State() + state.Db = &mockFailAdvanceTailStore{ + Store: state.Db, + failErr: failErr, + } + + linkErrors := make(chan LinkFailureError, 1) + coreLink.cfg.OnChannelFailure = func(_ lnwire.ChannelID, + _ lnwire.ShortChannelID, linkErr LinkFailureError) { + + linkErrors <- linkErr + } + + require.NoError(t, harness.start()) + + //nolint:forcetypeassert + aliceMsgs := coreLink.cfg.Peer.(*mockPeer).sentMsgs + ctx := linkTestContext{ + t: t, + aliceSwitch: harness.aliceSwitch, + aliceLink: harness.aliceLink, + aliceMsgs: aliceMsgs, + bobChannel: harness.bobChannel, + } + + // Bob adds an HTLC and signs for it, which makes Alice revoke her + // current state and sign a new one in return. + htlc := generateHtlc(t, coreLink, 0) + ctx.sendHtlcBobToAlice(htlc) + ctx.sendCommitSigBobToAlice(1) + ctx.receiveRevAndAckAliceToBob() + ctx.receiveCommitSigAliceToBob(1) + + // Now let Bob revoke his old state. Alice will try to persist that + // revocation, which is the write we rigged to fail. + ctx.sendRevAndAckBobToAlice() + + var linkErr LinkFailureError + select { + case linkErr = <-linkErrors: + case <-time.After(15 * time.Second): + t.Fatalf("link did not fail") + } + + require.Equal(t, expCode, linkErr.code) + require.Equal(t, expSendToPeer, linkErr.ShouldSendToPeer()) + + // We never want a local database problem to cost us the channel, so + // neither a force close nor a permanent failure is acceptable here. We + // do want the connection recycled, so that the reestablish flow can + // resync the state. + require.NotEqual(t, LinkFailureForceClose, linkErr.FailureAction) + require.Equal(t, LinkFailureDisconnect, linkErr.FailureAction) + require.False(t, linkErr.PermanentFailure) + + // Whatever else happens, the link itself must never put an error or a + // warning on the wire. Whether the failure is reported to the peer at + // all is the peer's decision, driven by ShouldSendToPeer above. + for { + select { + case msg := <-aliceMsgs: + switch msg.(type) { + case *lnwire.Error, *lnwire.Warning: + t.Fatalf("link put %T on the wire", msg) + } + + continue + + case <-time.After(100 * time.Millisecond): + } + + break + } +} + // TestChannelLinkFail tests that we will fail the channel, and force close the // channel in certain situations. func TestChannelLinkFail(t *testing.T) { diff --git a/htlcswitch/linkfailure.go b/htlcswitch/linkfailure.go index a2ce7305f38..00ee083d540 100644 --- a/htlcswitch/linkfailure.go +++ b/htlcswitch/linkfailure.go @@ -1,6 +1,12 @@ package htlcswitch -import "errors" +import ( + "errors" + "sync" + "time" + + "github.com/lightningnetwork/lnd/sqldb" +) var ( // ErrLinkShuttingDown signals that the link is shutting down. @@ -57,6 +63,12 @@ const ( // time, or that an update has been sent/received while the channel is // quiesced. ErrStfuViolation + + // ErrInternalDBError indicates that we were unable to process a message + // from our peer because our own database is in trouble. This is a local + // infrastructure failure and not peer misbehavior, so we never report + // it to the peer and we never force close because of it. + ErrInternalDBError ) // LinkFailureAction is an enum-like type that describes the action that should @@ -130,6 +142,8 @@ func (e LinkFailureError) Error() string { return "non-fatal circuit map error" case ErrStfuViolation: return "quiescence protocol executed improperly" + case ErrInternalDBError: + return "internal database error" default: return "unknown error" } @@ -159,3 +173,82 @@ func (e LinkFailureError) ShouldSendToPeer() bool { return false } } + +// dbErrLinkFailure is the failure we use whenever a message from our peer +// could not be processed because of a local database error. We disconnect +// instead of failing the channel: once the connection is re-established, the +// channel reestablish flow resyncs both sides and the state transition that we +// couldn't persist is simply retried. +var dbErrLinkFailure = LinkFailureError{ + code: ErrInternalDBError, + FailureAction: LinkFailureDisconnect, +} + +// linkFailureForDBErr returns the link failure that should be used to fail the +// link given the error that was hit while processing a message from our peer. +// If the error was caused by our own database rather than by the peer, we +// return a failure that neither reports anything to the peer nor force closes +// the channel. Otherwise the passed default failure is returned unchanged. +// +// This exists because a local database problem used to be translated into an +// lnwire.Error on the wire, which some peers answer by force closing the +// channel. Losing a channel to a transient database hiccup is never the right +// trade, see https://github.com/lightningnetwork/lnd/issues/10995. +func linkFailureForDBErr(err error, + defaultFailure LinkFailureError) LinkFailureError { + + if !sqldb.IsInternalDBError(err) { + return defaultFailure + } + + return dbErrLinkFailure +} + +const ( + // dbFailureEscalation is the number of database caused link failures we + // tolerate within dbFailureWindow before we start logging about the + // health of the database itself rather than about the individual links. + dbFailureEscalation = 3 + + // dbFailureWindow is how far apart two database caused link failures + // can be before we stop considering them related. + dbFailureWindow = 5 * time.Minute +) + +// linkDBFailures counts the database caused link failures across all links of +// this daemon. Failing a link because of a database error is deliberately quiet +// on the wire, so without this the only trace of a sick database would be one +// error line per failed link, which reads exactly like a peer that keeps +// flapping. This exists to tell those two apart for an operator. +var linkDBFailures dbFailureTracker + +// dbFailureTracker counts how often links have recently been failed because of +// a local database error. +type dbFailureTracker struct { + mu sync.Mutex + + // count is the number of failures seen so far within the current + // window. + count int + + // lastSeen is when we recorded the most recent failure. + lastSeen time.Time +} + +// record notes another database caused link failure that happened at the given +// time, and returns the number of failures seen within the current window, +// including this one. Failures that are further apart than dbFailureWindow are +// treated as unrelated, and start a fresh window. +func (d *dbFailureTracker) record(now time.Time) int { + d.mu.Lock() + defer d.mu.Unlock() + + if !d.lastSeen.IsZero() && now.Sub(d.lastSeen) > dbFailureWindow { + d.count = 0 + } + + d.count++ + d.lastSeen = now + + return d.count +} diff --git a/htlcswitch/linkfailure_test.go b/htlcswitch/linkfailure_test.go new file mode 100644 index 00000000000..e99efd5dfd6 --- /dev/null +++ b/htlcswitch/linkfailure_test.go @@ -0,0 +1,151 @@ +package htlcswitch + +import ( + "errors" + "fmt" + "testing" + "time" + + "github.com/lightningnetwork/lnd/sqldb" + "github.com/stretchr/testify/require" +) + +// TestLinkFailureForDBErr tests that errors originating from our own database +// are mapped onto a link failure that is neither reported to our peer nor +// causes a force close, while all other errors keep the failure the caller +// asked for. +func TestLinkFailureForDBErr(t *testing.T) { + t.Parallel() + + // serializationErr mimics the error postgres hands us when a + // transaction couldn't be serialized against the other concurrent + // transactions. + serializationErr := sqldb.MapSQLError(errors.New("ERROR: could not " + + "serialize access due to read/write dependencies among " + + "transactions (SQLSTATE 40001)")) + + // The default failure is the one the revocation path used to always + // use. It both reports an error to the peer and asks for a disconnect. + defaultFailure := LinkFailureError{ + code: ErrInvalidRevocation, + FailureAction: LinkFailureDisconnect, + } + + tests := []struct { + name string + err error + + // expCode is the error code we expect the resulting failure to + // carry. + expCode errorCode + + // expSendToPeer is whether we expect the resulting failure to + // be reported to our peer on the wire. + expSendToPeer bool + }{ + { + name: "peer error is left alone", + err: errors.New("revocation key mismatch"), + expCode: ErrInvalidRevocation, + expSendToPeer: true, + }, + { + name: "serialization error", + err: serializationErr, + expCode: ErrInternalDBError, + expSendToPeer: false, + }, + { + name: "wrapped serialization error", + err: fmt.Errorf("unable to restore remote unsigned "+ + "local updates: %w", serializationErr), + expCode: ErrInternalDBError, + expSendToPeer: false, + }, + { + name: "retries exceeded", + err: sqldb.ErrRetriesExceeded, + expCode: ErrInternalDBError, + expSendToPeer: false, + }, + { + name: "wrapped retries exceeded", + err: fmt.Errorf("unable to accept revocation: %w", + fmt.Errorf("%w: %w", sqldb.ErrRetriesExceeded, + serializationErr)), + expCode: ErrInternalDBError, + expSendToPeer: false, + }, + { + name: "canceled retry", + err: fmt.Errorf("%w: %w", sqldb.ErrRetryCanceled, + serializationErr), + expCode: ErrInternalDBError, + expSendToPeer: false, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + failure := linkFailureForDBErr(test.err, defaultFailure) + + require.Equal(t, test.expCode, failure.code) + require.Equal( + t, test.expSendToPeer, + failure.ShouldSendToPeer(), + ) + + // No matter the error, we never want to force close + // here, and we always want the connection recycled. + require.NotEqual( + t, LinkFailureForceClose, + failure.FailureAction, + ) + require.Equal( + t, LinkFailureDisconnect, failure.FailureAction, + ) + require.False(t, failure.PermanentFailure) + }) + } +} + +// TestDBErrLinkFailureIsSilent asserts that the failure we use for database +// errors is never reported to our peer and never force closes the channel. +func TestDBErrLinkFailureIsSilent(t *testing.T) { + t.Parallel() + + require.False(t, dbErrLinkFailure.ShouldSendToPeer()) + require.False(t, dbErrLinkFailure.PermanentFailure) + require.Equal( + t, LinkFailureDisconnect, dbErrLinkFailure.FailureAction, + ) + require.Equal(t, "internal database error", dbErrLinkFailure.Error()) +} + +// TestDBFailureTracker tests that we only escalate once several database caused +// link failures happen close enough together, and that a quiet spell resets the +// count. +func TestDBFailureTracker(t *testing.T) { + t.Parallel() + + var tracker dbFailureTracker + now := time.Now() + + // A burst of failures within the window accumulates. + require.Equal(t, 1, tracker.record(now)) + require.Equal(t, 2, tracker.record(now.Add(time.Second))) + require.Equal(t, 3, tracker.record(now.Add(2*time.Second))) + + // A failure that arrives after the window has passed is unrelated, so + // we start counting from scratch. Note that this must land below the + // escalation threshold, otherwise a healthy node that hits one such + // error a month would eventually escalate. + late := now.Add(2*time.Second + dbFailureWindow + time.Second) + require.Equal(t, 1, tracker.record(late)) + require.Less(t, 1, dbFailureEscalation) + + // Failures within the window of that one accumulate again. + require.Equal(t, 2, tracker.record(late.Add(time.Second))) +} diff --git a/kvdb/go.mod b/kvdb/go.mod index d932a0dd1b9..c577b8729d1 100644 --- a/kvdb/go.mod +++ b/kvdb/go.mod @@ -35,7 +35,6 @@ require ( github.com/containerd/continuity v0.3.0 // indirect github.com/coreos/go-semver v0.3.0 // indirect github.com/coreos/go-systemd/v22 v22.5.0 // indirect - github.com/distribution/reference v0.6.0 // indirect github.com/docker/cli v28.1.1+incompatible // indirect github.com/docker/docker v28.1.1+incompatible // indirect github.com/docker/go-connections v0.4.0 // indirect @@ -58,12 +57,8 @@ require ( github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-multierror v1.1.1 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect - github.com/jackc/chunkreader/v2 v2.0.1 // indirect - github.com/jackc/pgconn v1.14.3 // indirect github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 // indirect - github.com/jackc/pgio v1.0.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect - github.com/jackc/pgproto3/v2 v2.3.3 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/jonboulle/clockwork v0.2.2 // indirect @@ -144,3 +139,5 @@ require ( ) go 1.25.11 + +replace github.com/lightningnetwork/lnd/sqldb => ../sqldb diff --git a/kvdb/go.sum b/kvdb/go.sum index d977fce26dd..769f4ebcbff 100644 --- a/kvdb/go.sum +++ b/kvdb/go.sum @@ -169,21 +169,10 @@ github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= -github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk= -github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8= -github.com/jackc/chunkreader/v2 v2.0.1/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk= -github.com/jackc/pgconn v1.14.3 h1:bVoTr12EGANZz66nZPkMInAV/KHD2TxH9npjXXgiB3w= -github.com/jackc/pgconn v1.14.3/go.mod h1:RZbme4uasqzybK2RK5c65VsHxoyaml09lx3tXOcO/VM= github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 h1:Dj0L5fhJ9F82ZJyVOmBx6msDp/kfd1t9GRfny/mfJA0= github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds= -github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE= -github.com/jackc/pgio v1.0.0/go.mod h1:oP+2QK2wFfUWgr+gxjoBH9KGBb31Eio69xUb0w5bYf8= -github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65 h1:DadwsjnMwFjfWc9y5Wi/+Zz7xoE5ALHsRQlOctkOiHc= -github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65/go.mod h1:5R2h2EEX+qri8jOWMbJCtaPWkrrNc7OHwsp2TCqp7ak= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= -github.com/jackc/pgproto3/v2 v2.3.3 h1:1HLSx5H+tXR9pW3in3zaztoEwQYRC9SQaYUHjTSUOag= -github.com/jackc/pgproto3/v2 v2.3.3/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw= @@ -214,8 +203,6 @@ github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/lightningnetwork/lnd/healthcheck v1.2.4 h1:lLPLac+p/TllByxGSlkCwkJlkddqMP5UCoawCj3mgFQ= github.com/lightningnetwork/lnd/healthcheck v1.2.4/go.mod h1:G7Tst2tVvWo7cx6mSBEToQC5L1XOGxzZTPB29g9Rv2I= -github.com/lightningnetwork/lnd/sqldb v1.0.6 h1:LJdDSVdN33bVBIefsaJlPW9PDAm6GrXlyFucmzSJ3Ts= -github.com/lightningnetwork/lnd/sqldb v1.0.6/go.mod h1:OG09zL/PHPaBJefp4HsPz2YLUJ+zIQHbpgCtLnOx8I4= github.com/lightningnetwork/lnd/ticker v1.1.0 h1:ShoBiRP3pIxZHaETndfQ5kEe+S4NdAY1hiX7YbZ4QE4= github.com/lightningnetwork/lnd/ticker v1.1.0/go.mod h1:ubqbSVCn6RlE0LazXuBr7/Zi6QT0uQo++OgIRBxQUrk= github.com/lightningnetwork/lnd/tor v1.0.0 h1:wvEc7I+Y7IOtPglVP3cVBbYhiVhc7uTd7cMF9gQRzwA= diff --git a/kvdb/postgres/config.go b/kvdb/postgres/config.go index 5ea06430782..ad14c439554 100644 --- a/kvdb/postgres/config.go +++ b/kvdb/postgres/config.go @@ -10,4 +10,13 @@ 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."` + + // Quit is an optional channel that is closed once the daemon starts + // shutting down. It is used to abort an in-flight transaction retry + // loop, so that a transaction which keeps hitting serialization errors + // can't delay shutdown for the length of the retry budget. + // + // NOTE: This is injected at runtime and is deliberately not a command + // line flag. + Quit <-chan struct{} } diff --git a/kvdb/postgres/db.go b/kvdb/postgres/db.go index 6aca0276d2d..e3218012f67 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, + Quit: config.Quit, } } diff --git a/kvdb/sqlbase/db.go b/kvdb/sqlbase/db.go index 8ff7f979aff..e5f3dcb9328 100644 --- a/kvdb/sqlbase/db.go +++ b/kvdb/sqlbase/db.go @@ -24,7 +24,29 @@ const ( // DefaultNumTxRetries is the default number of times we'll retry a // transaction if it fails with an error that permits transaction // repetition. + // + // NOTE: This is no longer the primary bound of the retry loop, see + // DefaultTxRetryBudget below. It is kept around because it is part of + // the exported API of this package. DefaultNumTxRetries = 50 + + // DefaultTxRetryBudget is the total amount of wall clock time we're + // willing to spend retrying a single kv transaction that keeps failing + // with a serialization error. + // + // This backend holds the channel state, so surfacing a serialization + // error to the caller is expensive: the link that was trying to persist + // a state transition has to be failed and the connection to the peer + // recycled. Blocking for a while longer is strictly cheaper than that, + // which is why we bound the retry loop by a generous amount of elapsed + // time instead of by a fixed attempt count. With the backoff capped at + // one second, the old count of 50 attempts gave up after roughly 46 + // seconds, which is not much of a contention burst to ride out. + // + // Note that the retry loop also aborts immediately once the quit + // channel of the backend is closed, so this budget can never delay + // shutdown. + DefaultTxRetryBudget = 2 * time.Minute ) // Config holds a set of configuration options of a sql database connection. @@ -62,6 +84,13 @@ type Config struct { // NOTE: Temporary, should be removed when all parts of the LND code // are more resilient against concurrent db access.. WithTxLevelLock bool + + // Quit is an optional channel that is closed once the daemon starts + // shutting down. We only use it to abort an in-flight transaction retry + // loop, never to abort a query that is already running, so that the + // writes we still make while shutting down are given a fair chance to + // land. + Quit <-chan struct{} } // db holds a reference to the sql db connection. @@ -90,6 +119,10 @@ type db struct { // lock is the global write lock that ensures single writer. This is // only used if cfg.WithTxLevelLock is set. lock sync.RWMutex + + // quit is closed once the daemon starts shutting down. See the + // documentation of Config.Quit for the details. + quit <-chan struct{} } // Enforce db implements the walletdb.DB interface. @@ -151,6 +184,7 @@ func NewSqlBackend(ctx context.Context, cfg *Config) (*db, error) { db: dbConn, table: table, prefix: cfg.TableNamePrefix, + quit: cfg.Quit, }, nil } @@ -262,9 +296,16 @@ func (db *db) executeTransaction(f func(tx walletdb.ReadWriteTx) error, return attemptRollback(kvTx) } - return sqldb.ExecuteSQLTransactionWithRetry( + // We deliberately leave the attempt count unbounded here and rely on + // the elapsed time budget instead, so that a long lived contention + // burst doesn't cost us a channel link. The quit channel keeps that + // budget from holding up shutdown. + return sqldb.ExecuteSQLTransactionWithRetryConfig( db.ctx, makeTx, rollbackTx, execTxBody, onBackoff, - DefaultNumTxRetries, + sqldb.RetryConfig{ + MaxElapsed: DefaultTxRetryBudget, + Quit: db.quit, + }, ) } diff --git a/kvdb/sqlite/config.go b/kvdb/sqlite/config.go index b79bc36e637..cc8549cfd6c 100644 --- a/kvdb/sqlite/config.go +++ b/kvdb/sqlite/config.go @@ -10,4 +10,13 @@ type Config struct { BusyTimeout time.Duration `long:"busytimeout" description:"The maximum amount of time to wait for a database connection to become available for a query."` MaxConnections int `long:"maxconnections" description:"The maximum number of open connections to the database. Set to zero for unlimited."` PragmaOptions []string `long:"pragmaoptions" description:"A list of pragma options to set on a database connection. For example, 'auto_vacuum=incremental'. Note that the flag must be specified multiple times if multiple options are to be set."` + + // Quit is an optional channel that is closed once the daemon starts + // shutting down. It is used to abort an in-flight transaction retry + // loop, so that a transaction which keeps hitting serialization errors + // can't delay shutdown for the length of the retry budget. + // + // NOTE: This is injected at runtime and is deliberately not a command + // line flag. + Quit <-chan struct{} } diff --git a/kvdb/sqlite/db.go b/kvdb/sqlite/db.go index 07a6d07cf34..607a9644924 100644 --- a/kvdb/sqlite/db.go +++ b/kvdb/sqlite/db.go @@ -85,6 +85,7 @@ func NewSqliteBackend(ctx context.Context, cfg *Config, dbPath, fileName, Dsn: dsn, Timeout: cfg.Timeout, TableNamePrefix: prefix, + Quit: cfg.Quit, } return sqlbase.NewSqlBackend(ctx, sqlCfg) diff --git a/lncfg/db.go b/lncfg/db.go index 4a8680b386e..ada08337c38 100644 --- a/lncfg/db.go +++ b/lncfg/db.go @@ -284,7 +284,13 @@ func GetSqliteConfigKVDB(cfg *sqldb.SqliteConfig) *sqlite.Config { } // GetBackends returns a set of kvdb.Backends as set in the DB config. -func (db *DB) GetBackends(ctx context.Context, chanDBPath, +// +// The quit channel is closed once the daemon starts shutting down. The SQL +// backed kv stores use it to abort a transaction retry loop, so that a +// transaction which keeps hitting serialization errors can't delay shutdown for +// the length of its retry budget. It may be nil, in which case the retry loops +// are only bound by their own budgets. +func (db *DB) GetBackends(ctx context.Context, quit <-chan struct{}, chanDBPath, walletDBPath, towerServerDBPath string, towerClientEnabled, towerServerEnabled bool, logger btclog.Logger) (*DatabaseBackends, error) { @@ -406,12 +412,14 @@ func (db *DB) GetBackends(ctx context.Context, chanDBPath, // This is a temporary measure until we migrate all kvdb SQL // users to native SQL. postgresConfig := GetPostgresConfigKVDB(db.Postgres) + postgresConfig.Quit = quit // Create a separate config for channeldb with the global lock // setting if configured. postgresConfigChannelDB := GetPostgresConfigKVDB(db.Postgres) postgresConfigChannelDB.WithGlobalLock = db.Postgres. ChannelDBWithGlobalLock + postgresConfigChannelDB.Quit = quit postgresBackend, err := kvdb.Open( kvdb.PostgresBackendName, ctx, @@ -530,6 +538,7 @@ func (db *DB) GetBackends(ctx context.Context, chanDBPath, // This is a temporary measure until we migrate all kvdb SQL // users to native SQL. sqliteConfig := GetSqliteConfigKVDB(db.Sqlite) + sqliteConfig.Quit = quit // Note that for sqlite, we put kv tables for the channel.db, // wtclient.db and sphinxreplay.db all in the channel.sqlite db. diff --git a/peer/brontide.go b/peer/brontide.go index e9c258df88a..65103af373f 100644 --- a/peer/brontide.go +++ b/peer/brontide.go @@ -55,6 +55,7 @@ import ( "github.com/lightningnetwork/lnd/protofsm" "github.com/lightningnetwork/lnd/queue" "github.com/lightningnetwork/lnd/routing/route" + "github.com/lightningnetwork/lnd/sqldb" "github.com/lightningnetwork/lnd/subscribe" "github.com/lightningnetwork/lnd/ticker" "github.com/lightningnetwork/lnd/tlv" @@ -5284,11 +5285,28 @@ func (p *Brontide) handleCloseMsg(msg *closeMsg) { p.log.Errorf("Unable to respond to remote close msg: %v", err) + // If our own database is what let us down here, then there's + // nothing to tell the peer: the channel is fine and the close + // can be retried once we're healthy again. We only recycle the + // connection. + if sqldb.IsInternalDBError(err) { + p.Disconnect(fmt.Errorf("unable to respond to close "+ + "msg: %w", err)) + + return + } + + // We never put the raw error text on the wire, as it can carry + // details of our internal state that the peer has no business + // seeing. errMsg := &lnwire.Error{ ChanID: msg.cid, - Data: lnwire.ErrorData(err.Error()), + Data: lnwire.ErrorData( + "close failed due to internal error", + ), } p.queueMsg(errMsg, nil) + return } diff --git a/sqldb/go.mod b/sqldb/go.mod index 224be39a981..e037dd7821f 100644 --- a/sqldb/go.mod +++ b/sqldb/go.mod @@ -2,6 +2,7 @@ module github.com/lightningnetwork/lnd/sqldb require ( github.com/btcsuite/btclog/v2 v2.0.1-0.20250602222548-9967d19bb084 + github.com/btcsuite/btcwallet/walletdb v1.6.0 github.com/davecgh/go-spew v1.1.1 github.com/golang-migrate/migrate/v4 v4.17.0 github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 diff --git a/sqldb/go.sum b/sqldb/go.sum index 0218c9d2902..f8bb062b0d8 100644 --- a/sqldb/go.sum +++ b/sqldb/go.sum @@ -10,6 +10,8 @@ github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c h1:4HxD1lBUGUddhzg github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c/go.mod h1:w7xnGOhwT3lmrS4H3b/D1XAXxvh+tbhUm8xeHN2y3TQ= github.com/btcsuite/btclog/v2 v2.0.1-0.20250602222548-9967d19bb084 h1:y3bvkt8ki0KX35eUEU8XShRHusz1S+55QwXUTmxn888= github.com/btcsuite/btclog/v2 v2.0.1-0.20250602222548-9967d19bb084/go.mod h1:XItGUfVOxotJL8kkuk2Hj3EVow5KCugXl3wWfQ6K0AE= +github.com/btcsuite/btcwallet/walletdb v1.6.0 h1:Yund5XbdqFxNW7+R2Sxs02bMC5fMrmORj4GN8MV55no= +github.com/btcsuite/btcwallet/walletdb v1.6.0/go.mod h1:q9xif0Csp52GVb3l252BbHCuyiCnuEbrPWu/HAsvaYc= github.com/cenkalti/backoff/v4 v4.1.3 h1:cFAlzYUlVYDysBEH2T5hyJZMh3+5+WCBvSnK6Q8UtC4= github.com/cenkalti/backoff/v4 v4.1.3/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw= github.com/containerd/continuity v0.3.0 h1:nisirsYROK15TAMVukJOUyGJjz4BNQJBVsNvAXZJ/eg= @@ -126,6 +128,8 @@ github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +go.etcd.io/bbolt v1.3.11 h1:yGEzV1wPz2yVCLsD8ZAiGHhHVlczyC9d1rP43/VCRJ0= +go.etcd.io/bbolt v1.3.11/go.mod h1:dksAq7YMXoljX0xu6VF5DMZGbhYYoLUalEiSySYAS4I= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU= diff --git a/sqldb/interfaces.go b/sqldb/interfaces.go index 12ce63a1bb9..d398f8bf19c 100644 --- a/sqldb/interfaces.go +++ b/sqldb/interfaces.go @@ -121,8 +121,9 @@ type BatchedQuerier interface { // executor. This can be used to do things like retry a transaction due to an // error a certain amount of times. type txExecutorOptions struct { - numRetries int - retryDelay time.Duration + numRetries int + retryDelay time.Duration + retryBudget time.Duration } // defaultTxExecutorOptions returns the default options for the transaction @@ -160,6 +161,16 @@ func WithTxRetryDelay(delay time.Duration) TxExecutorOption { } } +// WithTxRetryBudget is a functional option that allows us to bound the total +// amount of wall clock time that may be spent retrying a single transaction. A +// value of zero, which is the default, means that only the retry count bounds +// the retry loop. +func WithTxRetryBudget(budget time.Duration) TxExecutorOption { + return func(o *txExecutorOptions) { + o.retryBudget = budget + } +} + // TransactionExecutor is a generic struct that abstracts away from the type of // query a type needs to run under a database transaction, and also the set of // options for that transaction. The QueryCreator is used to create a query @@ -244,16 +255,97 @@ type RollbackTx func(tx Tx) error // the delay before the next retry. type OnBackoff func(retry int, delay time.Duration) +// RetryConfig bounds how long the transaction retry loop is willing to keep +// retrying a transaction that repeatedly fails with a serialization error. Two +// independent budgets are available, and the loop stops as soon as either of +// them is exhausted, as soon as the context of the caller is canceled, or as +// soon as the Quit channel below is closed. +type RetryConfig struct { + // MaxRetries is the maximum number of times a transaction will be + // attempted before we give up. A value of zero means that the number of + // attempts is unbounded, in which case only MaxElapsed, the context and + // Quit bound the loop. + MaxRetries int + + // MaxElapsed is the maximum amount of wall clock time that may be spent + // across all attempts of a single transaction. A value of zero means + // that the elapsed time is unbounded, in which case only MaxRetries, the + // context and Quit bound the loop. + MaxElapsed time.Duration + + // Quit is an optional channel that aborts the retry loop as soon as it + // is closed. Callers that allow a generous time budget should wire this + // to their shutdown signal, so that a transaction that keeps conflicting + // can never hold up shutdown for the length of the budget. + // + // NOTE: A nil channel blocks forever in a select, so leaving this unset + // simply means that only the budgets and the context bound the loop. + Quit <-chan struct{} +} + +// NumRetriesConfig returns a retry budget that is bounded by the given number +// of attempts only. +func NumRetriesConfig(numRetries int) RetryConfig { + return RetryConfig{ + MaxRetries: numRetries, + } +} + +// exhausted returns true if either of the configured budgets has been used up +// after the given number of attempts and elapsed wall clock time. +func (r RetryConfig) exhausted(attempts int, elapsed time.Duration) bool { + // If neither budget was set, then we fall back to the default attempt + // count. We do this so that a zero value config can never put us into a + // truly unbounded loop by accident. + if r.MaxRetries <= 0 && r.MaxElapsed <= 0 { + return attempts >= DefaultNumTxRetries + } + + if r.MaxRetries > 0 && attempts >= r.MaxRetries { + return true + } + + if r.MaxElapsed > 0 && elapsed >= r.MaxElapsed { + return true + } + + return false +} + // ExecuteSQLTransactionWithRetry is a helper function that executes a // transaction with retry logic. It will retry the transaction if it fails with // a serialization error. The function will return an error if the transaction // fails with a non-retryable error, the context is cancelled or the number of // retries is exceeded. +// +// NOTE: This is retained for backwards compatibility with the callers that +// predate RetryConfig. New code should call +// ExecuteSQLTransactionWithRetryConfig directly, which also allows bounding the +// retries by elapsed time and aborting them on shutdown. func ExecuteSQLTransactionWithRetry(ctx context.Context, makeTx MakeTx, rollbackTx RollbackTx, txBody TxBody, onBackoff OnBackoff, numRetries int) error { - waitBeforeRetry := func(attemptNumber int) bool { + return ExecuteSQLTransactionWithRetryConfig( + ctx, makeTx, rollbackTx, txBody, onBackoff, + NumRetriesConfig(numRetries), + ) +} + +// ExecuteSQLTransactionWithRetryConfig is a helper function that executes a +// transaction with retry logic. It will retry the transaction if it fails with +// a serialization error. The function will return an error if the transaction +// fails with a non-retryable error, the context is cancelled, the quit channel +// is closed or the retry budget is exhausted. +func ExecuteSQLTransactionWithRetryConfig(ctx context.Context, makeTx MakeTx, + rollbackTx RollbackTx, txBody TxBody, onBackoff OnBackoff, + retryCfg RetryConfig) error { + + // waitBeforeRetry blocks for a randomized, exponentially increasing + // backoff before the next attempt is made. It returns a non-nil error + // if we should stop retrying, either because the context of the caller + // was canceled or because we were asked to quit while we were waiting. + waitBeforeRetry := func(attemptNumber int, dbErr error) error { retryDelay := randRetryDelay( DefaultRetryDelay, DefaultMaxRetryDelay, attemptNumber, ) @@ -264,31 +356,37 @@ func ExecuteSQLTransactionWithRetry(ctx context.Context, makeTx MakeTx, // Before we try again, we'll wait with a random backoff based // on the retry delay. case <-time.After(retryDelay): - return true + return nil - // If the daemon is shutting down, then we'll exit early. + // If the daemon is shutting down, then we'll exit early. We + // label the error we return here, so that a caller can tell an + // interrupted retry loop apart from a transaction that truly + // failed. The original database error is still wrapped, so that + // both errors.Is and errors.As keep working on the result. case <-ctx.Done(): - return false + return fmt.Errorf("%w: %w", ErrRetryCanceled, dbErr) + + // Same, but for callers whose shutdown signal is a plain + // channel rather than a context. + case <-retryCfg.Quit: + return fmt.Errorf("%w: %w", ErrRetryCanceled, dbErr) } } - for i := 0; i < numRetries; i++ { + // attemptTx runs a single attempt of the transaction. The first return + // value is true if the attempt failed with a serialization error and is + // therefore worth retrying, in which case the returned error is the + // serialization error that caused the retry. + attemptTx := func() (bool, error) { tx, err := makeTx() if err != nil { dbErr := MapSQLError(err) log.Tracef("Failed to makeTx: err=%v, dbErr=%v", err, dbErr) - if IsSerializationError(dbErr) { - // Nothing to roll back here, since we haven't - // even get a transaction yet. We'll just wait - // and try again. - if waitBeforeRetry(i) { - continue - } - } - - return dbErr + // Nothing to roll back here, since we haven't even got + // a transaction yet. + return IsSerializationError(dbErr), dbErr } // Rollback is safe to call even if the tx is already closed, @@ -300,50 +398,69 @@ func ExecuteSQLTransactionWithRetry(ctx context.Context, makeTx MakeTx, if bodyErr := txBody(tx); bodyErr != nil { log.Tracef("Error in txBody: %v", bodyErr) - // Roll back the transaction, then attempt a random - // backoff and try again if the error was a - // serialization error. + // Roll back the transaction, then signal a retry if the + // error was a serialization error. if err := rollbackTx(tx); err != nil { - return MapSQLError(err) + return false, MapSQLError(err) } dbErr := MapSQLError(bodyErr) - if IsSerializationError(dbErr) { - if waitBeforeRetry(i) { - continue - } - } - return dbErr + return IsSerializationError(dbErr), dbErr } // Commit transaction. if commitErr := tx.Commit(); commitErr != nil { log.Tracef("Failed to commit tx: %v", commitErr) - // Roll back the transaction, then attempt a random - // backoff and try again if the error was a - // serialization error. + // Roll back the transaction, then signal a retry if the + // error was a serialization error. if err := rollbackTx(tx); err != nil { - return MapSQLError(err) + return false, MapSQLError(err) } dbErr := MapSQLError(commitErr) - if IsSerializationError(dbErr) { - if waitBeforeRetry(i) { - continue - } - } - return dbErr + return IsSerializationError(dbErr), dbErr } - return nil + return false, nil + } + + var ( + startTime = time.Now() + attempts int + lastErr error + ) + + for { + retry, err := attemptTx() + if !retry { + return err + } + + lastErr = err + attempts++ + + // We check the budget before we back off, so that we never + // sleep for an attempt that we're not going to make anyway. + if retryCfg.exhausted(attempts, time.Since(startTime)) { + break + } + + if waitErr := waitBeforeRetry(attempts-1, err); waitErr != nil { + return waitErr + } } // If we get to this point, then we weren't able to successfully commit - // a tx given the max number of retries. - return ErrRetriesExceeded + // a tx within the retry budget. We attach the last error we saw, so + // that the actual reason for the failure isn't lost. + // + // NOTE: lastErr is always non-nil here, since we only ever leave the + // loop above after an attempt that asked to be retried. + return fmt.Errorf("%w (attempts=%v, elapsed=%v): %w", + ErrRetriesExceeded, attempts, time.Since(startTime), lastErr) } // ExecTx is a wrapper for txBody to abstract the creation and commit of a db @@ -385,9 +502,11 @@ func (t *TransactionExecutor[Q]) ExecTx(ctx context.Context, return nil } - return ExecuteSQLTransactionWithRetry( - ctx, makeTx, rollbackTx, execTxBody, onBackoff, - t.opts.numRetries, + return ExecuteSQLTransactionWithRetryConfig( + ctx, makeTx, rollbackTx, execTxBody, onBackoff, RetryConfig{ + MaxRetries: t.opts.numRetries, + MaxElapsed: t.opts.retryBudget, + }, ) } diff --git a/sqldb/interfaces_test.go b/sqldb/interfaces_test.go new file mode 100644 index 00000000000..1caeb4a5ff2 --- /dev/null +++ b/sqldb/interfaces_test.go @@ -0,0 +1,285 @@ +package sqldb + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// mockTx is a minimal Tx implementation used to drive the transaction retry +// loop without an actual database behind it. +type mockTx struct{} + +// Commit always succeeds. +func (m *mockTx) Commit() error { + return nil +} + +// Rollback always succeeds. +func (m *mockTx) Rollback() error { + return nil +} + +// retryHarness bundles the callbacks the retry loop needs, along with a counter +// for the number of attempts that were made. +type retryHarness struct { + attempts int + + // bodyErr is returned by the transaction body on every attempt. If it + // is nil, the transaction is considered successful. + bodyErr error +} + +// run executes the retry loop with the harness' callbacks. +func (h *retryHarness) run(ctx context.Context, + retryCfg RetryConfig) error { + + makeTx := func() (Tx, error) { + return &mockTx{}, nil + } + + txBody := func(Tx) error { + h.attempts++ + + return h.bodyErr + } + + rollbackTx := func(Tx) error { + return nil + } + + onBackoff := func(int, time.Duration) {} + + return ExecuteSQLTransactionWithRetryConfig( + ctx, makeTx, rollbackTx, txBody, onBackoff, retryCfg, + ) +} + +// TestRetryConfigExhausted tests the two independent budgets that bound the +// transaction retry loop. +func TestRetryConfigExhausted(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + cfg RetryConfig + attempts int + elapsed time.Duration + exp bool + }{ + { + name: "zero value falls back to default count", + cfg: RetryConfig{}, + attempts: DefaultNumTxRetries - 1, + exp: false, + }, + { + name: "zero value is bounded by default count", + cfg: RetryConfig{}, + attempts: DefaultNumTxRetries, + exp: true, + }, + { + name: "count budget not yet used up", + cfg: RetryConfig{MaxRetries: 3}, + attempts: 2, + exp: false, + }, + { + name: "count budget used up", + cfg: RetryConfig{MaxRetries: 3}, + attempts: 3, + exp: true, + }, + { + name: "time budget ignores the attempt count", + cfg: RetryConfig{MaxElapsed: time.Minute}, + attempts: 10_000, + elapsed: time.Second, + exp: false, + }, + { + name: "time budget used up", + cfg: RetryConfig{MaxElapsed: time.Minute}, + attempts: 1, + elapsed: time.Minute, + exp: true, + }, + { + name: "either budget stops the loop", + cfg: RetryConfig{ + MaxRetries: 3, + MaxElapsed: time.Minute, + }, + attempts: 3, + exp: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + require.Equal(t, test.exp, test.cfg.exhausted( + test.attempts, test.elapsed, + )) + }) + } +} + +// TestExecuteSQLTransactionWithRetrySuccess tests that a transaction that +// commits on the first attempt isn't retried. +func TestExecuteSQLTransactionWithRetrySuccess(t *testing.T) { + t.Parallel() + + h := &retryHarness{} + err := h.run(t.Context(), NumRetriesConfig(5)) + require.NoError(t, err) + require.Equal(t, 1, h.attempts) +} + +// TestExecuteSQLTransactionWithRetryNonRetryable tests that an error that isn't +// a serialization error is returned as-is, without any retries. +func TestExecuteSQLTransactionWithRetryNonRetryable(t *testing.T) { + t.Parallel() + + bodyErr := errors.New("not a serialization error") + h := &retryHarness{bodyErr: bodyErr} + + err := h.run(t.Context(), NumRetriesConfig(5)) + require.ErrorIs(t, err, bodyErr) + require.Equal(t, 1, h.attempts) + require.False(t, IsInternalDBError(err)) +} + +// TestExecuteSQLTransactionWithRetryExceeded tests that once the attempt budget +// is used up, we return a labeled error that still carries the underlying +// database error along for diagnostics. +func TestExecuteSQLTransactionWithRetryExceeded(t *testing.T) { + t.Parallel() + + h := &retryHarness{bodyErr: serializationErr()} + + err := h.run(t.Context(), NumRetriesConfig(3)) + require.Equal(t, 3, h.attempts) + + // The error must be recognizable both as an exhausted retry loop and as + // the serialization error that caused it. + require.ErrorIs(t, err, ErrRetriesExceeded) + require.True(t, IsSerializationError(err)) + require.True(t, IsInternalDBError(err)) + + // The raw postgres text alone is not a useful error message, so we make + // sure our own label is part of it as well. + require.Contains(t, err.Error(), ErrRetriesExceeded.Error()) + require.Contains(t, err.Error(), "SQLSTATE 40001") +} + +// TestExecuteSQLTransactionWithRetryCanceled tests that a retry loop that is +// interrupted by a canceled context returns a labeled error instead of the bare +// database error. +func TestExecuteSQLTransactionWithRetryCanceled(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + h := &retryHarness{bodyErr: serializationErr()} + + // The attempt budget is generous here, we expect the canceled context + // to stop the loop after the very first backoff. + err := h.run(ctx, NumRetriesConfig(1000)) + require.Equal(t, 1, h.attempts) + + require.ErrorIs(t, err, ErrRetryCanceled) + require.True(t, IsSerializationError(err)) + require.True(t, IsInternalDBError(err)) + + // The bare serialization error used to leak out of here unlabeled, + // which made it look like a protocol failure to the callers. + require.Contains(t, err.Error(), ErrRetryCanceled.Error()) + require.Contains(t, err.Error(), "SQLSTATE 40001") +} + +// TestExecuteSQLTransactionWithRetryBudget tests that a purely time based retry +// budget keeps retrying past any fixed attempt count, and still terminates. +func TestExecuteSQLTransactionWithRetryBudget(t *testing.T) { + t.Parallel() + + const budget = 500 * time.Millisecond + + h := &retryHarness{bodyErr: serializationErr()} + + start := time.Now() + err := h.run(t.Context(), RetryConfig{MaxElapsed: budget}) + elapsed := time.Since(start) + + require.ErrorIs(t, err, ErrRetriesExceeded) + require.True(t, IsInternalDBError(err)) + + // We should have spent at least the budget retrying, and we should not + // have overshot it by more than a single capped backoff. + require.GreaterOrEqual(t, elapsed, budget) + require.Less(t, elapsed, budget+2*DefaultMaxRetryDelay) + + // More than one attempt must have been made, otherwise the budget + // wasn't actually driving the loop. + require.Greater(t, h.attempts, 1) +} + +// TestExecuteSQLTransactionWithRetryQuit tests that closing the quit channel +// aborts the retry loop immediately, even though the retry budget is nowhere +// near used up. This is what keeps a generous time budget from holding up +// shutdown. +func TestExecuteSQLTransactionWithRetryQuit(t *testing.T) { + t.Parallel() + + quit := make(chan struct{}) + close(quit) + + h := &retryHarness{bodyErr: serializationErr()} + + start := time.Now() + err := h.run(t.Context(), RetryConfig{ + MaxElapsed: time.Hour, + Quit: quit, + }) + elapsed := time.Since(start) + + // Exactly one attempt is made, and we don't wait out a single backoff. + require.Equal(t, 1, h.attempts) + require.Less(t, elapsed, time.Second) + + require.ErrorIs(t, err, ErrRetryCanceled) + require.True(t, IsInternalDBError(err)) +} + +// TestExecuteSQLTransactionWithRetryCompat tests that the backwards compatible +// entry point still bounds the loop by the passed attempt count. +func TestExecuteSQLTransactionWithRetryCompat(t *testing.T) { + t.Parallel() + + h := &retryHarness{bodyErr: serializationErr()} + + makeTx := func() (Tx, error) { + return &mockTx{}, nil + } + txBody := func(Tx) error { + h.attempts++ + + return h.bodyErr + } + rollbackTx := func(Tx) error { + return nil + } + + err := ExecuteSQLTransactionWithRetry( + t.Context(), makeTx, rollbackTx, txBody, + func(int, time.Duration) {}, 2, + ) + require.Equal(t, 2, h.attempts) + require.ErrorIs(t, err, ErrRetriesExceeded) +} diff --git a/sqldb/sqlerrors_common.go b/sqldb/sqlerrors_common.go new file mode 100644 index 00000000000..39a161f6c06 --- /dev/null +++ b/sqldb/sqlerrors_common.go @@ -0,0 +1,92 @@ +package sqldb + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "syscall" + + "github.com/btcsuite/btcwallet/walletdb" +) + +var ( + // ErrRetryCanceled is returned when the transaction retry loop was + // interrupted before it could either complete or exhaust its retry + // budget. This happens when the context of the caller is canceled, or + // the caller's quit channel is closed, while we're waiting to retry a + // transaction that failed with a serialization error. + ErrRetryCanceled = errors.New("db tx retry canceled") +) + +// IsInternalDBError returns true if the passed error signals trouble with our +// local database infrastructure rather than a problem with the data that was +// read or written. +// +// Callers use this to tell apart errors that are our own fault (a busy or +// unavailable database) from errors that are caused by the input they were +// handed. This distinction matters on the channel state machine paths: a +// database hiccup must never be reported to our channel peer as a protocol +// violation, since some peers respond to such a report by force closing the +// channel. +// +// The predicate is deliberately conservative. It only matches errors that +// unambiguously mean local infrastructure trouble, so that a genuine protocol +// violation is never mistaken for one and silently swallowed. +// +// NOTE: This covers the bbolt backed kv stores as well as the SQL ones. The +// walletdb layer converts the two bbolt errors that can surface on a write path +// into the sentinels matched below, so there is no need to match on bbolt +// itself here. See convertErr in btcwallet's walletdb/bdb package. +func IsInternalDBError(err error) bool { + if err == nil { + return false + } + + switch { + // The transaction couldn't be serialized against the other concurrent + // transactions, and the retry machinery gave up on it. + case IsSerializationError(err): + return true + + // The retry machinery exhausted its budget while retrying a + // serialization error. + case errors.Is(err, ErrRetriesExceeded): + return true + + // The retry machinery was interrupted before it could complete. + case errors.Is(err, ErrRetryCanceled): + return true + + // The connection to the database is gone, or the database driver + // decided the connection was no longer usable. + case errors.Is(err, sql.ErrConnDone), + errors.Is(err, sql.ErrTxDone), + errors.Is(err, driver.ErrBadConn): + + return true + + // The query was canceled or timed out. This is either a shutdown, or a + // database that is too slow to answer within the configured timeout. + // Neither is the fault of our peer. + case errors.Is(err, context.Canceled), + errors.Is(err, context.DeadlineExceeded): + + return true + + // The kv store was closed out from under us, or it was opened + // read-only. Both mean our own database is unusable, and neither says + // anything about the data we were trying to write. + case errors.Is(err, walletdb.ErrDbNotOpen), + errors.Is(err, walletdb.ErrTxNotWritable): + + return true + + // We ran out of disk space. This arrives wrapped in an *os.PathError + // from the file backed stores, which errors.Is unwraps for us. + case errors.Is(err, syscall.ENOSPC): + return true + } + + return false +} diff --git a/sqldb/sqlerrors_test.go b/sqldb/sqlerrors_test.go new file mode 100644 index 00000000000..9bd3e7dd8c6 --- /dev/null +++ b/sqldb/sqlerrors_test.go @@ -0,0 +1,141 @@ +package sqldb + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "fmt" + "os" + "syscall" + "testing" + + "github.com/btcsuite/btcwallet/walletdb" + "github.com/stretchr/testify/require" +) + +// serializationErr returns an error that looks like the one postgres hands us +// when a transaction couldn't be serialized against the other concurrent +// transactions. +func serializationErr() error { + return MapSQLError(errors.New("ERROR: could not serialize access due " + + "to read/write dependencies among transactions (SQLSTATE " + + "40001)")) +} + +// TestIsInternalDBError tests that we correctly tell errors caused by our own +// database infrastructure apart from errors caused by the data we were handed. +func TestIsInternalDBError(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + exp bool + }{ + { + name: "nil error", + err: nil, + exp: false, + }, + { + name: "unrelated error", + err: errors.New("revocation key mismatch"), + exp: false, + }, + { + name: "unique constraint violation", + err: &ErrSQLUniqueConstraintViolation{ + DBError: errors.New("duplicate key"), + }, + exp: false, + }, + { + name: "serialization error", + err: serializationErr(), + exp: true, + }, + { + name: "wrapped serialization error", + err: fmt.Errorf("unable to restore remote unsigned "+ + "local updates: %w", serializationErr()), + exp: true, + }, + { + name: "retries exceeded", + err: ErrRetriesExceeded, + exp: true, + }, + { + name: "wrapped retries exceeded", + err: fmt.Errorf("%w: %w", ErrRetriesExceeded, + serializationErr()), + exp: true, + }, + { + name: "canceled retry", + err: fmt.Errorf("%w: %w", ErrRetryCanceled, + serializationErr()), + exp: true, + }, + { + name: "connection done", + err: fmt.Errorf("query failed: %w", sql.ErrConnDone), + exp: true, + }, + { + name: "tx done", + err: fmt.Errorf("commit failed: %w", sql.ErrTxDone), + exp: true, + }, + { + name: "bad conn", + err: fmt.Errorf("query failed: %w", driver.ErrBadConn), + exp: true, + }, + { + name: "kv store not open", + err: fmt.Errorf("unable to fetch chan bucket: %w", + walletdb.ErrDbNotOpen), + exp: true, + }, + { + name: "kv tx not writable", + err: walletdb.ErrTxNotWritable, + exp: true, + }, + { + name: "kv bucket not found is not infra", + err: walletdb.ErrBucketNotFound, + exp: false, + }, + { + name: "out of disk space", + err: fmt.Errorf("write failed: %w", &os.PathError{ + Op: "write", + Path: "/lnd/channel.db", + Err: syscall.ENOSPC, + }), + exp: true, + }, + { + name: "context canceled", + err: fmt.Errorf("query failed: %w", + context.Canceled), + exp: true, + }, + { + name: "context deadline exceeded", + err: context.DeadlineExceeded, + exp: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + require.Equal(t, test.exp, IsInternalDBError(test.err)) + }) + } +}