filer: a listing over a hard link no longer deadlocks a bounded SQL pool (#11118)

* filer: give the SQL stores' key-value reads their own connections

A listing holds the connection its rows are on for the whole iteration, and
FilerStoreWrapper calls maybeReadHardLink -> KvGet from inside that iteration,
so a hard-linked entry needs a second connection while the first is still busy.
Out of one bounded pool that is a deadlock: the listings fill the pool and then
wait for a connection none of them will release, and the wrapper's
context.WithoutCancel leaves the waiters without a deadline, so the filer stays
wedged rather than erroring.

The sqlite store shows it at its sharpest -- it allows a single connection, so
one listing over one hard-linked entry never returns. On postgres with
connection_max_open = 50, 60 concurrent listings over hard-linked entries made
no progress at all.

Key-value reads now run on their own pool, carved out of connection_max_open
rather than added to it, so the operator's cap still bounds what the store opens
against the database. An unbounded pool keeps a single pool: nothing can wait
there. sqlite's single connection becomes two, one per pool, and its writes get
a busy timeout so a write that meets the reader waits instead of failing.

Claude-Session: https://claude.ai/code/session_018DWwctzD4T2DmnczPRM47t

* sqlite: keep both pools on one database, whatever the dbFile spells

A dbFile that already carries URI options got a second "?" appended, which the
driver reads as part of the preceding option value, and a bare :memory: is
private to each connection, so the key-value pool would open its own empty
database and every key-value operation would fail on a missing filemeta.

Claude-Session: https://claude.ai/code/session_018DWwctzD4T2DmnczPRM47t

* sqlite: assert the busy timeout on the in-memory DSN too

Claude-Session: https://claude.ai/code/session_018DWwctzD4T2DmnczPRM47t

weed/filer/abstract_sql/abstract_sql_store.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -29,6 +30,7 @@ type SqlGenerator interface {
 type AbstractSqlStore struct {
 	SqlGenerator
 	DB                     *sql.DB
+	KvDB                   *sql.DB
 	SupportBucketTable     bool
 	dbs                    map[string]bool
 	dbsLock                sync.Mutex
@@ -97,6 +99,49 @@ func (store *AbstractSqlStore) RollbackTransaction(ctx context.Context) error {
 	return nil
 }
 
+// A listing holds its connection for the whole row iteration while its callback
+// reads a hard link through KvGet, so both cannot come out of one bounded pool:
+// the listings fill it and then wait for a connection none of them will release.
+// Give the key-value reads their own slice of connection_max_open. A cap of 1 is
+// the exception -- it has to become 2, or a single listing cannot finish.
+func splitPoolForKv(maxOpen int) (mainOpen, kvOpen int) {
+	if maxOpen <= 0 {
+		return maxOpen, 0
+	}
+	kvOpen = min(max(maxOpen/4, 1), maxKvPoolSize)
+	return max(maxOpen-kvOpen, 1), kvOpen
+}
+
+const maxKvPoolSize = 8
+
+// UseConnectionPools sizes the store's pool and, when it is bounded, opens the
+// separate pool the key-value reads run on.
+func (store *AbstractSqlStore) UseConnectionPools(db *sql.DB, openKv func() (*sql.DB, error), maxIdle, maxOpen, maxLifetimeSeconds int) error {
+
+	lifetime := time.Duration(maxLifetimeSeconds) * time.Second
+	mainOpen, kvOpen := splitPoolForKv(maxOpen)
+
+	db.SetMaxIdleConns(maxIdle)
+	db.SetMaxOpenConns(mainOpen)
+	db.SetConnMaxLifetime(lifetime)
+	store.DB = db
+
+	if kvOpen == 0 {
+		return nil
+	}
+
+	kvDB, err := openKv()
+	if err != nil {
+		return err
+	}
+	kvDB.SetMaxIdleConns(kvOpen)
+	kvDB.SetMaxOpenConns(kvOpen)
+	kvDB.SetConnMaxLifetime(lifetime)
+	store.KvDB = kvDB
+
+	return nil
+}
+
 func (store *AbstractSqlStore) getTxOrDB(ctx context.Context, fullpath util.FullPath, isForChildren bool) (txOrDB TxOrDB, bucket string, shortPath util.FullPath, err error) {
 
 	shortPath = fullpath
@@ -400,6 +445,9 @@ func (store *AbstractSqlStore) ListDirectoryEntries(ctx context.Context, dirPath
 
 func (store *AbstractSqlStore) Shutdown() {
 	store.DB.Close()
+	if store.KvDB != nil {
+		store.KvDB.Close()
+	}
 }
 
 func isValidBucket(bucket string) bool {

weed/filer/abstract_sql/abstract_sql_store_kv.go

@@ -12,12 +12,21 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
+// Key-value reads run outside the listing pool: a listing calls KvGet from
+// inside its own callback, still holding the connection its rows are on.
+func (store *AbstractSqlStore) kvDB(ctx context.Context) TxOrDB {
+	if tx, ok := ctx.Value("tx").(*sql.Tx); ok {
+		return tx
+	}
+	if store.KvDB != nil {
+		return store.KvDB
+	}
+	return store.DB
+}
+
 func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
 
-	db, _, _, err := store.getTxOrDB(ctx, "", false)
-	if err != nil {
-		return fmt.Errorf("findDB: %w", err)
-	}
+	db := store.kvDB(ctx)
 
 	dirStr, dirHash, name := GenDirAndName(key)
 
@@ -49,10 +58,7 @@ func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []by
 
 func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
 
-	db, _, _, err := store.getTxOrDB(ctx, "", false)
-	if err != nil {
-		return nil, fmt.Errorf("findDB: %w", err)
-	}
+	db := store.kvDB(ctx)
 
 	dirStr, dirHash, name := GenDirAndName(key)
 	row := db.QueryRowContext(ctx, store.GetSqlFind(DEFAULT_TABLE), dirHash, name, dirStr)
@@ -72,10 +78,7 @@ func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []b
 
 func (store *AbstractSqlStore) KvDelete(ctx context.Context, key []byte) (err error) {
 
-	db, _, _, err := store.getTxOrDB(ctx, "", false)
-	if err != nil {
-		return fmt.Errorf("findDB: %w", err)
-	}
+	db := store.kvDB(ctx)
 
 	dirStr, dirHash, name := GenDirAndName(key)
 

weed/filer/abstract_sql/abstract_sql_store_test.go

@@ -51,3 +51,23 @@ func TestGetTxOrDBRealBucket(t *testing.T) {
 		t.Errorf("shortPath = %q, want /dir/file.txt", shortPath)
 	}
 }
+
+func TestSplitPoolForKv(t *testing.T) {
+	cases := []struct {
+		maxOpen, mainOpen, kvOpen int
+	}{
+		{0, 0, 0},   // unbounded: nothing waits, one pool is enough
+		{-1, -1, 0}, // same, however the operator spelled it
+		{50, 42, 8},
+		{16, 12, 4},
+		{4, 3, 1},
+		{2, 1, 1},
+		{1, 1, 1}, // a listing needs two, so a cap of 1 has to become 2
+	}
+	for _, c := range cases {
+		mainOpen, kvOpen := splitPoolForKv(c.maxOpen)
+		if mainOpen != c.mainOpen || kvOpen != c.kvOpen {
+			t.Errorf("splitPoolForKv(%d) = %d, %d, want %d, %d", c.maxOpen, mainOpen, kvOpen, c.mainOpen, c.kvOpen)
+		}
+	}
+}

weed/filer/mysql/mysql_store.go

@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"os"
 	"strings"
-	"time"
 
 	"github.com/go-sql-driver/mysql"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -151,10 +150,11 @@ func (store *MysqlStore) initialize(dsn string, upsertQuery string, enableUpsert
 		return fmt.Errorf("can not create mysql connector for %s error:%w", maskedDSN(cfg), err)
 	}
 
-	store.DB = sql.OpenDB(connector)
-	store.DB.SetMaxIdleConns(maxIdle)
-	store.DB.SetMaxOpenConns(maxOpen)
-	store.DB.SetConnMaxLifetime(time.Duration(maxLifetimeSeconds) * time.Second)
+	if err = store.UseConnectionPools(sql.OpenDB(connector), func() (*sql.DB, error) {
+		return sql.OpenDB(connector), nil
+	}, maxIdle, maxOpen, maxLifetimeSeconds); err != nil {
+		return err
+	}
 
 	if err = store.DB.Ping(); err != nil {
 		return fmt.Errorf("connect to %s error:%v", maskedDSN(cfg), err)

weed/filer/mysql2/mysql2_store.go

@@ -5,7 +5,6 @@ import (
 	"database/sql"
 	"fmt"
 	"strings"
-	"time"
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -84,19 +83,19 @@ func (store *MysqlStore2) initialize(createTable, upsertQuery string, enableUpse
 		adaptedSqlUrl += "&interpolateParams=true"
 	}
 
-	var dbErr error
-	store.DB, dbErr = sql.Open("mysql", sqlUrl)
+	db, dbErr := sql.Open("mysql", sqlUrl)
 	if dbErr != nil {
-		if store.DB != nil {
-			store.DB.Close()
+		if db != nil {
+			db.Close()
 		}
-		store.DB = nil
 		return fmt.Errorf("can not connect to %s error:%w", adaptedSqlUrl, dbErr)
 	}
 
-	store.DB.SetMaxIdleConns(maxIdle)
-	store.DB.SetMaxOpenConns(maxOpen)
-	store.DB.SetConnMaxLifetime(time.Duration(maxLifetimeSeconds) * time.Second)
+	if err = store.UseConnectionPools(db, func() (*sql.DB, error) {
+		return sql.Open("mysql", sqlUrl)
+	}, maxIdle, maxOpen, maxLifetimeSeconds); err != nil {
+		return err
+	}
 
 	if err = store.DB.Ping(); err != nil {
 		return fmt.Errorf("connect to %s error:%v", adaptedSqlUrl, err)

weed/filer/postgres/postgres_store.go

@@ -8,6 +8,7 @@
 package postgres
 
 import (
+	"database/sql"
 	"strconv"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -119,7 +120,11 @@ func (store *PostgresStore) initialize(upsertQuery string, enableUpsert bool, us
 	if openErr != nil {
 		return openErr
 	}
-	store.DB = db
+	if err = store.UseConnectionPools(db, func() (*sql.DB, error) {
+		return OpenPGXDB(sqlUrl, adaptedSqlUrl, pgbouncerCompatible, maxIdle, maxOpen, maxLifetimeSeconds)
+	}, maxIdle, maxOpen, maxLifetimeSeconds); err != nil {
+		return err
+	}
 
 	ConfigureListOrdering(store.DB, gen)
 

weed/filer/postgres2/postgres2_store.go

@@ -9,6 +9,7 @@ package postgres2
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"strconv"
 
@@ -128,7 +129,11 @@ func (store *PostgresStore2) initialize(createTable, upsertQuery string, enableU
 	if openErr != nil {
 		return openErr
 	}
-	store.DB = db
+	if err = store.UseConnectionPools(db, func() (*sql.DB, error) {
+		return postgres.OpenPGXDB(sqlUrl, adaptedSqlUrl, pgbouncerCompatible, maxIdle, maxOpen, maxLifetimeSeconds)
+	}, maxIdle, maxOpen, maxLifetimeSeconds); err != nil {
+		return err
+	}
 
 	if err = store.CreateTable(context.Background(), abstract_sql.DEFAULT_TABLE); err != nil {
 		return fmt.Errorf("init table %s: %v", abstract_sql.DEFAULT_TABLE, err)

weed/filer/sqlite/sqlite_store.go

@@ -8,6 +8,8 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"strings"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/filer/abstract_sql"
@@ -49,6 +51,21 @@ func (store *SqliteStore) Initialize(configuration util.Configuration, prefix st
 	)
 }
 
+// sqliteDSN keeps the store's two pools on one database and lets a writer that
+// meets the other pool's reader wait for it instead of failing outright. A bare
+// :memory: is private to each connection, so it has to be named and shared.
+func sqliteDSN(dbFile string) string {
+	dsn := dbFile
+	if dsn == ":memory:" {
+		dsn = fmt.Sprintf("file:seaweedfs%d?mode=memory&cache=shared", time.Now().UnixNano())
+	}
+	separator := "?"
+	if strings.Contains(dsn, "?") {
+		separator = "&"
+	}
+	return dsn + separator + "_pragma=busy_timeout(10000)"
+}
+
 func (store *SqliteStore) initialize(dbFile, createTable, upsertQuery string) (err error) {
 
 	store.SupportBucketTable = true
@@ -58,21 +75,25 @@ func (store *SqliteStore) initialize(dbFile, createTable, upsertQuery string) (e
 		UpsertQueryTemplate:    upsertQuery,
 	}
 
-	var dbErr error
-	store.DB, dbErr = sql.Open("sqlite", dbFile)
+	dsn := sqliteDSN(dbFile)
+
+	db, dbErr := sql.Open("sqlite", dsn)
 	if dbErr != nil {
-		if store.DB != nil {
-			store.DB.Close()
-			store.DB = nil
+		if db != nil {
+			db.Close()
 		}
 		return fmt.Errorf("can not connect to %s error:%v", dbFile, dbErr)
 	}
 
-	if err = store.DB.Ping(); err != nil {
+	if err = db.Ping(); err != nil {
 		return fmt.Errorf("connect to %s error:%v", dbFile, err)
 	}
 
-	store.DB.SetMaxOpenConns(1)
+	if err = store.UseConnectionPools(db, func() (*sql.DB, error) {
+		return sql.Open("sqlite", dsn)
+	}, 1, 1, 0); err != nil {
+		return err
+	}
 
 	if err = store.CreateTable(context.Background(), abstract_sql.DEFAULT_TABLE); err != nil {
 		return fmt.Errorf("init table %s: %v", abstract_sql.DEFAULT_TABLE, err)

weed/filer/sqlite/sqlite_store_test.go

@@ -0,0 +1,92 @@
+//go:build (linux || darwin || windows) && sqlite
+
+package sqlite
+
+import (
+	"context"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+const (
+	createTableSql = `CREATE TABLE IF NOT EXISTS "%s" (dirhash BIGINT, name VARCHAR(1000), directory TEXT, meta BLOB, PRIMARY KEY (dirhash, name)) WITHOUT ROWID;`
+	upsertQuerySql = `INSERT INTO "%s"(dirhash,name,directory,meta)VALUES(?,?,?,?) ON CONFLICT(dirhash,name) DO UPDATE SET directory=excluded.directory, meta=excluded.meta;`
+)
+
+// A listing whose callback reads a hard link needs a second connection while it
+// still holds the one its rows are on, and the store allows exactly one.
+func TestListDirectoryOverHardLinkDoesNotDeadlock(t *testing.T) {
+	store := &SqliteStore{}
+	if err := store.initialize(filepath.Join(t.TempDir(), "filer.db"), createTableSql, upsertQuerySql); err != nil {
+		t.Fatal(err)
+	}
+	defer store.Shutdown()
+
+	wrapper := filer.NewFilerStoreWrapper(store)
+	ctx := context.Background()
+	entry := &filer.Entry{
+		FullPath:   util.FullPath("/buckets/hlbucket/f0001"),
+		Attr:       filer.Attr{Mode: 0644, Mtime: time.Now(), Crtime: time.Now()},
+		HardLinkId: filer.HardLinkId("hardlink-00000001"),
+	}
+	if err := wrapper.InsertEntry(ctx, entry); err != nil {
+		t.Fatal(err)
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := wrapper.ListDirectoryEntries(ctx, util.FullPath("/buckets/hlbucket"), "", true, 10,
+			func(*filer.Entry) (bool, error) { return true, nil })
+		done <- err
+	}()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Errorf("list: %v", err)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatalf("one listing over one hard-linked entry never returned: %+v", store.DB.Stats())
+	}
+}
+
+func TestSqliteDSN(t *testing.T) {
+	if got, want := sqliteDSN("/data/filer.db"), "/data/filer.db?_pragma=busy_timeout(10000)"; got != want {
+		t.Errorf("sqliteDSN = %q, want %q", got, want)
+	}
+	if got, want := sqliteDSN("file:/data/filer.db?cache=shared"), "file:/data/filer.db?cache=shared&_pragma=busy_timeout(10000)"; got != want {
+		t.Errorf("sqliteDSN with options = %q, want %q", got, want)
+	}
+	memory := sqliteDSN(":memory:")
+	if !strings.HasPrefix(memory, "file:seaweedfs") || !strings.Contains(memory, "mode=memory&cache=shared") ||
+		!strings.HasSuffix(memory, "&_pragma=busy_timeout(10000)") {
+		t.Errorf("sqliteDSN(:memory:) = %q, want a named shared-memory URI carrying the busy timeout", memory)
+	}
+}
+
+// Both pools have to open the same database: the table is created on one and
+// the key-value operations run on the other.
+func TestInMemoryStoreSharesOneDatabase(t *testing.T) {
+	store := &SqliteStore{}
+	if err := store.initialize(":memory:", createTableSql, upsertQuerySql); err != nil {
+		t.Fatal(err)
+	}
+	defer store.Shutdown()
+
+	ctx := context.Background()
+	key := []byte("hardlink-00000001")
+	if err := store.KvPut(ctx, key, []byte("value")); err != nil {
+		t.Fatalf("KvPut: %v", err)
+	}
+	value, err := store.KvGet(ctx, key)
+	if err != nil {
+		t.Fatalf("KvGet: %v", err)
+	}
+	if string(value) != "value" {
+		t.Errorf("KvGet = %q, want value", value)
+	}
+}