db: flag slow transactions, add opt-in commit ledger

db/interfaces.go

@@ -6,6 +6,9 @@ import (
 	"errors"
 	"math"
 	prand "math/rand"
+	"runtime"
+	"runtime/trace"
+	"strings"
 	"time"
 
 	"github.com/btcsuite/btclog/v2"
@@ -35,6 +38,14 @@ const (
 
 	// DefaultMaxRetryDelay is the default maximum delay between retries.
 	DefaultMaxRetryDelay = time.Second * 3
+
+	// slowTxnThreshold is the duration past which ExecTx flags a
+	// transaction as suspiciously long: a held write transaction starves
+	// every other writer on the single-writer lock, and a stalled begin
+	// marks the corresponding victim. One second is far above any
+	// healthy transaction on this schema and far below the 30s
+	// busy_timeout that converts a stuck writer into payment failures.
+	slowTxnThreshold = time.Second
 )
 
 // TxOptions represents a set of options one can use to control what type of
@@ -229,6 +240,26 @@ func NewTransactionExecutor[Querier any](db BatchedQuerier,
 	}
 }
 
+// execTxCallerHint returns the function name of the first caller frame
+// outside the db packages, identifying which store or domain call site owns
+// a flagged transaction. Only invoked on the slow path.
+func execTxCallerHint() string {
+	pcs := make([]uintptr, 16)
+	n := runtime.Callers(3, pcs)
+	frames := runtime.CallersFrames(pcs[:n])
+	for {
+		frame, more := frames.Next()
+		if frame.Function != "" && !strings.Contains(
+			frame.Function, "darepo-client/db",
+		) {
+			return frame.Function
+		}
+		if !more {
+			return "unknown"
+		}
+	}
+}
+
 // ExecTx is a wrapper for txBody to abstract the creation and commit of a db
 // transaction. The db transaction is embedded in a `*Queries` that txBody
 // needs to use when executing each one of the queries that need to be applied
@@ -248,6 +279,17 @@ func (t *TransactionExecutor[Q]) ExecTx(ctx context.Context,
 		return txBody(t.createQuery(tx))
 	}
 
+	// Under an active runtime trace (e.g. an arktest --trace run), mark
+	// the whole transaction as a region and tag it with the owning call
+	// site so `go tool trace` shows exactly which goroutine sat inside a
+	// transaction during a stall. StartRegion is a few nanoseconds when
+	// tracing is off, and the caller walk is gated behind IsEnabled.
+	if trace.IsEnabled() {
+		region := trace.StartRegion(ctx, "db.ExecTx")
+		trace.Log(ctx, "db.execTxCaller", execTxCallerHint())
+		defer region.End()
+	}
+
 	waitBeforeRetry := func(attemptNumber int) {
 		retryDelay := t.opts.randRetryDelay(attemptNumber)
 
@@ -263,8 +305,19 @@ func (t *TransactionExecutor[Q]) ExecTx(ctx context.Context,
 	}
 
 	for i := 0; i < t.opts.numRetries; i++ {
-		// Create the db transaction.
+		// Create the db transaction. A slow begin means this
+		// transaction was the victim of whoever held the writer; a
+		// slow body/commit (below) identifies the holder itself.
+		beginStart := time.Now()
 		tx, err := t.BatchedQuerier.BeginTx(ctx, txOptions)
+		if wait := time.Since(beginStart); wait >= slowTxnThreshold {
+			t.log.WarnS(ctx, "Transaction begin stalled on the "+
+				"database lock", nil,
+				"wait", wait,
+				"readonly", txOptions.ReadOnly(),
+				"caller", execTxCallerHint(),
+			)
+		}
 		if err != nil {
 			dbErr := MapSQLError(err)
 			if IsSerializationOrDeadlockError(dbErr) {
@@ -300,7 +353,26 @@ func (t *TransactionExecutor[Q]) ExecTx(ctx context.Context,
 			_ = tx.Rollback()
 		}()
 
+		// A long-held write transaction starves every other writer on
+		// the connection's single-writer lock, so flag the holder with
+		// its call site. The check is two clock reads on the fast
+		// path; the caller walk runs only when the threshold trips.
+		holdStart := time.Now()
+		warnIfHeldLong := func() {
+			held := time.Since(holdStart)
+			if held < slowTxnThreshold {
+				return
+			}
+
+			t.log.WarnS(ctx, "Transaction held long", nil,
+				"held", held,
+				"readonly", txOptions.ReadOnly(),
+				"caller", execTxCallerHint(),
+			)
+		}
+
 		if err := txBody(t.createQuery(tx)); err != nil {
+			warnIfHeldLong()
 			dbErr := MapSQLError(err)
 			if IsSerializationOrDeadlockError(dbErr) {
 				// Roll back the transaction, then pop back up
@@ -327,7 +399,9 @@ func (t *TransactionExecutor[Q]) ExecTx(ctx context.Context,
 		}
 
 		// Commit transaction.
-		if err = tx.Commit(); err != nil {
+		err = tx.Commit()
+		warnIfHeldLong()
+		if err != nil {
 			dbErr := MapSQLError(err)
 			if IsSerializationOrDeadlockError(dbErr) {
 				// Roll back the transaction, then pop back up
@@ -344,6 +418,12 @@ func (t *TransactionExecutor[Q]) ExecTx(ctx context.Context,
 			return dbErr
 		}
 
+		// Attribute the committed transaction to its owning call site
+		// when the accounting ledger is armed.
+		if txnAccountingEnabled.Load() {
+			recordTxnCommit(txOptions.ReadOnly())
+		}
+
 		return nil
 	}
 

db/txn_count.go

@@ -0,0 +1,98 @@
+package db
+
+import (
+	"sort"
+	"sync"
+	"sync/atomic"
+)
+
+// txnAccountKey identifies one transaction accounting bucket: the domain
+// call site that owns the transaction plus whether it ran read-only.
+type txnAccountKey struct {
+	caller   string
+	readOnly bool
+}
+
+var (
+	// txnAccountingEnabled gates the per-commit accounting hook in
+	// ExecTx. The gate costs one atomic load per commit when disabled,
+	// so production pays nothing unless a harness opts in.
+	txnAccountingEnabled atomic.Bool
+
+	// txnAccountingMu guards txnAccountingCounts.
+	txnAccountingMu sync.Mutex
+
+	// txnAccountingCounts accumulates committed-transaction counts keyed
+	// by owning call site.
+	txnAccountingCounts = make(map[txnAccountKey]uint64)
+)
+
+// TxnCommitCount reports the number of committed transactions attributed to
+// one call site.
+type TxnCommitCount struct {
+	// Caller is the fully qualified function name of the first frame
+	// outside the db packages that started the transaction.
+	Caller string
+
+	// ReadOnly reports whether the bucket's transactions ran read-only.
+	// On SQLite only read-write commits pay the single-writer lock and
+	// the fsync, so the split separates durability load from read load.
+	ReadOnly bool
+
+	// Count is the number of committed transactions in this bucket.
+	Count uint64
+}
+
+// EnableTxnAccounting turns on per-call-site commit accounting. It exists
+// for benchmarks and tests that need an empirical transaction ledger; the
+// gate is process-global and stays on until process exit.
+func EnableTxnAccounting() {
+	txnAccountingEnabled.Store(true)
+}
+
+// ResetTxnAccounting clears all accumulated accounting buckets, typically
+// to scope counting to a workload phase, for example after setup but before
+// the measured payments begin.
+func ResetTxnAccounting() {
+	txnAccountingMu.Lock()
+	defer txnAccountingMu.Unlock()
+
+	txnAccountingCounts = make(map[txnAccountKey]uint64)
+}
+
+// TxnAccountingSnapshot returns the accumulated commit counts sorted by
+// descending count.
+func TxnAccountingSnapshot() []TxnCommitCount {
+	txnAccountingMu.Lock()
+	defer txnAccountingMu.Unlock()
+
+	counts := make([]TxnCommitCount, 0, len(txnAccountingCounts))
+	for key, count := range txnAccountingCounts {
+		counts = append(counts, TxnCommitCount{
+			Caller:   key.caller,
+			ReadOnly: key.readOnly,
+			Count:    count,
+		})
+	}
+
+	sort.Slice(counts, func(i, j int) bool {
+		return counts[i].Count > counts[j].Count
+	})
+
+	return counts
+}
+
+// recordTxnCommit attributes one committed transaction to its owning call
+// site. Only invoked when accounting is enabled; the caller walk costs about
+// a microsecond, well below the cost of the commit it accounts for.
+func recordTxnCommit(readOnly bool) {
+	key := txnAccountKey{
+		caller:   execTxCallerHint(),
+		readOnly: readOnly,
+	}
+
+	txnAccountingMu.Lock()
+	defer txnAccountingMu.Unlock()
+
+	txnAccountingCounts[key]++
+}

db/txn_count_test.go

@@ -0,0 +1,61 @@
+package db
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	"github.com/lightninglabs/darepo-client/db/sqlc"
+	"github.com/stretchr/testify/require"
+)
+
+// TestTxnAccountingCountsCommits verifies that enabling transaction
+// accounting attributes committed read and write transactions to caller
+// buckets, splits them by the read-only flag, and that a reset clears the
+// ledger. The test is intentionally not parallel: the accounting table is
+// process-global, so overlapping package tests would bleed into the counts.
+func TestTxnAccountingCountsCommits(t *testing.T) {
+	store := NewTestSqliteDB(t)
+
+	executor := NewTransactionExecutor(
+		store.BaseDB,
+		func(tx *sql.Tx) *sqlc.Queries {
+			return store.BaseDB.Queries.WithTx(tx)
+		},
+		store.log,
+	)
+
+	EnableTxnAccounting()
+	ResetTxnAccounting()
+
+	ctx := context.Background()
+	noop := func(*sqlc.Queries) error { return nil }
+
+	for i := 0; i < 2; i++ {
+		require.NoError(
+			t,
+			executor.ExecTx(
+				ctx, WriteTxOption(), noop,
+			),
+		)
+	}
+	require.NoError(t, executor.ExecTx(ctx, ReadTxOption(), noop))
+
+	snapshot := TxnAccountingSnapshot()
+	require.Len(t, snapshot, 2)
+
+	var writes, reads uint64
+	for _, bucket := range snapshot {
+		require.NotEmpty(t, bucket.Caller)
+		if bucket.ReadOnly {
+			reads += bucket.Count
+		} else {
+			writes += bucket.Count
+		}
+	}
+	require.Equal(t, uint64(2), writes)
+	require.Equal(t, uint64(1), reads)
+
+	ResetTxnAccounting()
+	require.Empty(t, TxnAccountingSnapshot())
+}