From adf49c1d55a306840de9b852806a27e07d576b9c Mon Sep 17 00:00:00 2001 From: aman035 Date: Fri, 21 Aug 2026 14:08:16 +0530 Subject: [PATCH 1/2] fix: remove stale chains outside the read lock and close per-chain databases (F-2026-18797) --- universalClient/chains/chains.go | 74 +++++++++++++-- universalClient/chains/chains_test.go | 132 +++++++++++++++++++++++++- 2 files changed, 195 insertions(+), 11 deletions(-) diff --git a/universalClient/chains/chains.go b/universalClient/chains/chains.go index bb0b102b..b8ae8ebc 100644 --- a/universalClient/chains/chains.go +++ b/universalClient/chains/chains.go @@ -29,8 +29,15 @@ type Chains struct { // Chain client management chains map[string]common.ChainClient // key: CAIP-2 chain ID chainConfigs map[string]*uregistrytypes.ChainConfig // key: CAIP-2 chain ID - chainsMu sync.RWMutex - pushChainID string // Push chain ID (always present) + // Handle opened for each live chain, kept so removal can close it. Every + // getChainDB call opens a new pool, so a handle dropped without closing keeps + // its file descriptors until the process exits. + chainDBs map[string]*db.DB // key: CAIP-2 chain ID + chainsMu sync.RWMutex + pushChainID string // Push chain ID (always present) + + // Database opener, swapped in tests to observe handle lifecycle. + openDB func(dir, filename string, migrateSchema bool) (*db.DB, error) // Background control muRunning sync.Mutex @@ -58,6 +65,8 @@ func NewChains( logger: logger.With().Str("component", "chains").Logger(), chains: make(map[string]common.ChainClient), chainConfigs: make(map[string]*uregistrytypes.ChainConfig), + chainDBs: make(map[string]*db.DB), + openDB: db.OpenFileDB, pushChainID: cfg.PushChainID, } } @@ -199,19 +208,33 @@ func (c *Chains) fetchAndUpdate(parent context.Context) error { } } - // Remove stale chains (never remove Push chain) + c.removeStaleChains(seenChains) + + return nil +} + +// removeStaleChains drops chains the registry no longer lists, never the Push chain. +// +// The ids are collected under the read lock and removed after releasing it. +// removeChain takes the write lock and sync.RWMutex is not reentrant, so removing +// from inside the loop would park the refresh goroutine forever while it still +// holds the read lock, taking every later reader of the registry down with it. +func (c *Chains) removeStaleChains(seenChains map[string]bool) { c.chainsMu.RLock() + var stale []string for chainID := range c.chains { if chainID != c.pushChainID && !seenChains[chainID] { - c.logger.Info().Str("chain", chainID).Msg("removing chain no longer in config") - if err := c.removeChain(chainID); err != nil { - c.logger.Error().Err(err).Str("chain", chainID).Msg("failed to remove chain") - } + stale = append(stale, chainID) } } c.chainsMu.RUnlock() - return nil + for _, chainID := range stale { + c.logger.Info().Str("chain", chainID).Msg("removing chain no longer in config") + if err := c.removeChain(chainID); err != nil { + c.logger.Error().Err(err).Str("chain", chainID).Msg("failed to remove chain") + } + } } // chainAction represents the action to take for a chain config @@ -271,6 +294,19 @@ func (c *Chains) addChain(ctx context.Context, cfg *uregistrytypes.ChainConfig) return fmt.Errorf("failed to get database for chain %s: %w", cfg.Chain, err) } + // Ownership passes to the registry only once the client is live. Until then + // close it on the way out, or a chain that cannot start leaks a handle on + // every refresh tick for as long as the misconfiguration lasts. + adopted := false + defer func() { + if adopted { + return + } + if cerr := chainDB.Close(); cerr != nil { + c.logger.Warn().Err(cerr).Str("chain", cfg.Chain).Msg("failed to close database after unsuccessful chain add") + } + }() + // Get chain-specific config chainConfig := c.config.GetChainConfig(cfg.Chain) @@ -298,7 +334,9 @@ func (c *Chains) addChain(ctx context.Context, cfg *uregistrytypes.ChainConfig) c.chainsMu.Lock() c.chains[cfg.Chain] = client c.chainConfigs[cfg.Chain] = cfg + c.chainDBs[cfg.Chain] = chainDB c.chainsMu.Unlock() + adopted = true c.logger.Info(). Str("chain", cfg.Chain). @@ -324,6 +362,14 @@ func (c *Chains) removeChain(chainID string) error { Msg("error stopping chain client during removal") } + // After Stop, so nothing is still reading through it. + if database, ok := c.chainDBs[chainID]; ok { + if err := database.Close(); err != nil { + c.logger.Error().Err(err).Str("chain", chainID).Msg("error closing chain database during removal") + } + delete(c.chainDBs, chainID) + } + delete(c.chains, chainID) delete(c.chainConfigs, chainID) @@ -350,9 +396,19 @@ func (c *Chains) StopAll() { } } + for chainID, database := range c.chainDBs { + if err := database.Close(); err != nil { + c.logger.Error(). + Err(err). + Str("chain", chainID). + Msg("error closing chain database") + } + } + // Clear the registry c.chains = make(map[string]common.ChainClient) c.chainConfigs = make(map[string]*uregistrytypes.ChainConfig) + c.chainDBs = make(map[string]*db.DB) } // GetClient returns the chain client for the specified chain ID @@ -413,7 +469,7 @@ func (c *Chains) getChainDB(chainID string) (*db.DB, error) { // Derive database base directory from NodeHome baseDir := filepath.Join(c.config.NodeHome, config.DatabasesSubdir) - database, err := db.OpenFileDB(baseDir, dbFilename, true) + database, err := c.openDB(baseDir, dbFilename, true) if err != nil { return nil, fmt.Errorf("failed to create database for chain %s: %w", chainID, err) } diff --git a/universalClient/chains/chains_test.go b/universalClient/chains/chains_test.go index 8fcb4706..ac89ad38 100644 --- a/universalClient/chains/chains_test.go +++ b/universalClient/chains/chains_test.go @@ -12,6 +12,7 @@ import ( "github.com/pushchain/push-chain-node/universalClient/chains/common" "github.com/pushchain/push-chain-node/universalClient/config" + "github.com/pushchain/push-chain-node/universalClient/db" uregistrytypes "github.com/pushchain/push-chain-node/x/uregistry/types" ) @@ -1665,8 +1666,8 @@ func TestNewChains_ConfigPreserved(t *testing.T) { t.Run("preserves all config fields", func(t *testing.T) { logger := zerolog.Nop() cfg := &config.Config{ - PushChainID: "push:1", - NodeHome: "/tmp/test", + PushChainID: "push:1", + NodeHome: "/tmp/test", ConfigRefreshIntervalSeconds: 30, } @@ -1739,3 +1740,130 @@ func TestDetermineChainAction_PushChainID(t *testing.T) { assert.Equal(t, chainActionAdd, action) }) } + +// dbIsOpen reports whether the handle still answers queries. A closed *db.DB +// errors on use, which is how these tests tell a released handle from a leaked one. +func dbIsOpen(t *testing.T, database *db.DB) bool { + t.Helper() + sqlDB, err := database.Client().DB() + if err != nil { + return false + } + return sqlDB.Ping() == nil +} + +// A chain that drops out of the registry is removed under the write lock, so the +// stale sweep must not still be holding the read lock when it calls removeChain. +// sync.RWMutex is not reentrant: doing so parks the refresh goroutine forever +// and every later reader of the registry blocks behind it. +func TestFetchAndUpdate_StaleRemovalDoesNotDeadlock(t *testing.T) { + c := newTestChains() + c.chains["eip155:1"] = &mockChainClient{} + c.chainConfigs["eip155:1"] = &uregistrytypes.ChainConfig{Chain: "eip155:1"} + + // Drive the stale sweep directly: the chain is absent from seenChains, which + // is what a delisted chain looks like on the next config fetch. + done := make(chan struct{}) + go func() { + defer close(done) + c.removeStaleChains(map[string]bool{c.pushChainID: true}) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("stale removal deadlocked: removeChain was called while the read lock was held") + } + + // The registry must be usable afterwards, not left with a held lock. + acquired := make(chan struct{}) + go func() { + c.chainsMu.Lock() + c.chainsMu.Unlock() + close(acquired) + }() + select { + case <-acquired: + case <-time.After(5 * time.Second): + t.Fatal("chainsMu still held after the stale sweep") + } + + _, err := c.GetClient("eip155:1") + assert.Error(t, err, "the delisted chain should be gone") +} + +// Every getChainDB call opens a fresh pool, so a handle that is dropped rather +// than closed keeps its descriptors for the life of the process. A chain that +// cannot start is retried on every refresh tick, which turns that into growth. +func TestAddChain_ClosesDatabaseWhenTheChainCannotStart(t *testing.T) { + c := newTestChains() + c.config.NodeHome = t.TempDir() + + // An unsupported VM type fails after the database has been opened. + cfg := &uregistrytypes.ChainConfig{ + Chain: "eip155:99", + VmType: uregistrytypes.VmType(9999), + Enabled: &uregistrytypes.ChainEnabled{IsInboundEnabled: true}, + } + + // Capture every handle addChain opens so we can assert each was released. + var opened []*db.DB + realOpen := c.openDB + c.openDB = func(dir, filename string, migrate bool) (*db.DB, error) { + database, err := realOpen(dir, filename, migrate) + if err == nil { + opened = append(opened, database) + } + return database, err + } + + for i := 0; i < 5; i++ { // five refresh ticks with the same broken config + err := c.addChain(context.Background(), cfg) + require.Error(t, err) + } + + require.Len(t, opened, 5, "each attempt opens its own handle") + for i, database := range opened { + assert.False(t, dbIsOpen(t, database), + "handle from attempt %d leaked; a persistent misconfiguration would grow one per tick", i) + } + assert.NotContains(t, c.chains, "eip155:99") +} + +// Removal has to release the handle too, not just drop the map entry. +func TestRemoveChain_ClosesTheDatabase(t *testing.T) { + c := newTestChains() + database, err := db.OpenFileDB(t.TempDir(), "eip155_1.db", true) + require.NoError(t, err) + + c.chains["eip155:1"] = &mockChainClient{} + c.chainConfigs["eip155:1"] = &uregistrytypes.ChainConfig{Chain: "eip155:1"} + c.chainDBs["eip155:1"] = database + require.True(t, dbIsOpen(t, database)) + + require.NoError(t, c.removeChain("eip155:1")) + + assert.False(t, dbIsOpen(t, database), "removal must close the handle") + assert.NotContains(t, c.chainDBs, "eip155:1") +} + +func TestStopAll_ClosesEveryDatabase(t *testing.T) { + c := newTestChains() + dir := t.TempDir() + + var opened []*db.DB + for _, id := range []string{"eip155:1", "eip155:2"} { + database, err := db.OpenFileDB(dir, sanitizeChainID(id)+".db", true) + require.NoError(t, err) + c.chains[id] = &mockChainClient{} + c.chainDBs[id] = database + opened = append(opened, database) + } + + c.StopAll() + + for i, database := range opened { + assert.False(t, dbIsOpen(t, database), "handle %d must be closed", i) + } + assert.Empty(t, c.chainDBs) +} From cf261c4a19ed6911e62a7ed1ac4bd8092794c843 Mon Sep 17 00:00:00 2001 From: aman035 Date: Fri, 21 Aug 2026 14:19:59 +0530 Subject: [PATCH 2/2] fix: apply the same database ownership rule to the push chain path (F-2026-18797) --- universalClient/chains/chains.go | 14 +++++++++ universalClient/chains/chains_test.go | 41 +++++++++++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/universalClient/chains/chains.go b/universalClient/chains/chains.go index b8ae8ebc..7fdda147 100644 --- a/universalClient/chains/chains.go +++ b/universalClient/chains/chains.go @@ -502,6 +502,18 @@ func (c *Chains) ensurePushChain(ctx context.Context) error { return fmt.Errorf("failed to get database for push chain: %w", err) } + // Same ownership rule as addChain: the registry adopts the handle only once + // the client is live, and closes it on the way out of every other path. + adopted := false + defer func() { + if adopted { + return + } + if cerr := pushDB.Close(); cerr != nil { + c.logger.Warn().Err(cerr).Str("chain", c.pushChainID).Msg("failed to close database after unsuccessful push chain add") + } + }() + // Create a minimal chain config for push chain // Push chain doesn't need gateway or other configs pushConfig := &uregistrytypes.ChainConfig{ @@ -538,7 +550,9 @@ func (c *Chains) ensurePushChain(ctx context.Context) error { c.chainsMu.Lock() c.chains[c.pushChainID] = client c.chainConfigs[c.pushChainID] = pushConfig + c.chainDBs[c.pushChainID] = pushDB c.chainsMu.Unlock() + adopted = true c.logger.Info(). Str("chain", c.pushChainID). diff --git a/universalClient/chains/chains_test.go b/universalClient/chains/chains_test.go index ac89ad38..3d169fff 100644 --- a/universalClient/chains/chains_test.go +++ b/universalClient/chains/chains_test.go @@ -1867,3 +1867,44 @@ func TestStopAll_ClosesEveryDatabase(t *testing.T) { } assert.Empty(t, c.chainDBs) } + +// ensurePushChain opens its own handle rather than going through addChain, so the +// same ownership rule has to hold there: released on failure, and registered on +// success so shutdown can close it. +func TestEnsurePushChain_HandleOwnership(t *testing.T) { + t.Run("failure to construct the client releases the handle", func(t *testing.T) { + c := newTestChains() + c.config.NodeHome = t.TempDir() + c.pushCore = nil // push.NewClient rejects a nil core + + var opened []*db.DB + realOpen := c.openDB + c.openDB = func(dir, filename string, migrate bool) (*db.DB, error) { + database, err := realOpen(dir, filename, migrate) + if err == nil { + opened = append(opened, database) + } + return database, err + } + + err := c.ensurePushChain(context.Background()) + require.Error(t, err) + require.Len(t, opened, 1) + assert.False(t, dbIsOpen(t, opened[0]), "handle must be released when the push client cannot be built") + assert.NotContains(t, c.chainDBs, c.pushChainID) + }) + + // On success the handle must be registered, or shutdown silently leaves the + // push chain's database open. + t.Run("shutdown closes a registered push handle", func(t *testing.T) { + c := newTestChains() + database, err := db.OpenFileDB(t.TempDir(), "push.db", true) + require.NoError(t, err) + + c.chains[c.pushChainID] = &mockChainClient{} + c.chainDBs[c.pushChainID] = database + + c.StopAll() + assert.False(t, dbIsOpen(t, database), "push chain handle must be closed on shutdown") + }) +}