From b6f2574b532f2afc6c971a04bb3ddc29133bf5ac Mon Sep 17 00:00:00 2001 From: Optic00 Date: Mon, 7 Sep 2026 23:07:14 +0200 Subject: [PATCH] fix: close server-owned caches and stop loops on startup failure --- internal/auth/scim_tokens.go | 18 +++++- internal/auth/session.go | 14 +++++ internal/auth/tokens.go | 14 +++++ internal/server/server.go | 81 ++++++++++++++++++--------- internal/services/item_cache.go | 14 +++++ internal/services/permission_cache.go | 12 +++- 6 files changed, 123 insertions(+), 30 deletions(-) diff --git a/internal/auth/scim_tokens.go b/internal/auth/scim_tokens.go index bd9b96766..c6d4df9a3 100644 --- a/internal/auth/scim_tokens.go +++ b/internal/auth/scim_tokens.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "log/slog" + "sync" "time" "windshift/internal/cacheutil" @@ -30,8 +31,21 @@ type scimTokenCacheEntry struct { // SCIMTokenManager handles SCIM token operations type SCIMTokenManager struct { - db database.Database - cache *bigcache.BigCache + db database.Database + cache *bigcache.BigCache + closeOnce sync.Once + closeErr error +} + +// Close stops the validation cache's background worker without closing the +// shared database. It may be called more than once, including with no cache. +func (tm *SCIMTokenManager) Close() error { + tm.closeOnce.Do(func() { + if tm.cache != nil { + tm.closeErr = tm.cache.Close() + } + }) + return tm.closeErr } // NewSCIMTokenManager creates a new SCIM token manager diff --git a/internal/auth/session.go b/internal/auth/session.go index 80d2720ad..c14c15a02 100644 --- a/internal/auth/session.go +++ b/internal/auth/session.go @@ -10,6 +10,7 @@ import ( "log/slog" "net" "net/http" + "sync" "time" "windshift/internal/database" @@ -54,6 +55,8 @@ type SessionManager struct { db database.Database opaqueKey []byte sessionValidation *sessionValidator + closeOnce sync.Once + closeErr error // ipBinding is the resolved SESSION_IP_BINDING mode (config.SessionIPBinding*) // that session validation applies to a client-IP change. An unknown or // zero value is treated as strict so managers built without config.Load @@ -61,6 +64,17 @@ type SessionManager struct { ipBinding string } +// Close stops the session validation cache's background worker without closing +// the shared database. It is safe with caching disabled and on repeated calls. +func (sm *SessionManager) Close() error { + sm.closeOnce.Do(func() { + if sm.sessionValidation != nil && sm.sessionValidation.cache != nil { + sm.closeErr = sm.sessionValidation.cache.Close() + } + }) + return sm.closeErr +} + // Session represents an active user session type Session struct { ID int `json:"id"` diff --git a/internal/auth/tokens.go b/internal/auth/tokens.go index a06d0948d..0b9e2416a 100644 --- a/internal/auth/tokens.go +++ b/internal/auth/tokens.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "log/slog" + "sync" "time" "windshift/internal/cacheutil" @@ -57,6 +58,19 @@ type TokenManager struct { db database.Database tokenTracker TokenUsageRecorder cache *bigcache.BigCache + closeOnce sync.Once + closeErr error +} + +// Close stops the validation cache's background worker. It does not close the +// shared database or token tracker, and may be called more than once. +func (tm *TokenManager) Close() error { + tm.closeOnce.Do(func() { + if tm.cache != nil { + tm.closeErr = tm.cache.Close() + } + }) + return tm.closeErr } // NewTokenManager creates a new token manager diff --git a/internal/server/server.go b/internal/server/server.go index 6437cbfe4..be9e33202 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -18,6 +18,7 @@ import ( "os" "strconv" "strings" + "sync" "time" "windshift/internal/aitools" @@ -100,6 +101,12 @@ type Server struct { db database.Database listener net.Listener + permissionService *services.PermissionService + sessionManager *auth.SessionManager + tokenManager *auth.TokenManager + scimTokenManager *auth.SCIMTokenManager + itemCache *services.ItemCacheService + ldapHandler *handlers.LDAPHandler notificationManager *handlers.NotificationManager notificationService *services.NotificationService @@ -156,9 +163,10 @@ type Server struct { publicBoardLimiter *middleware.RateLimiter userConcurrency *middleware.UserConcurrencyLimiter - actualPort int - started bool - shuttingDown bool + actualPort int + started bool + shuttingDown bool + backgroundStopOnce sync.Once } // New creates a new Server instance with the given configuration. @@ -285,6 +293,7 @@ func (s *Server) initialize() error { if err != nil { return fmt.Errorf("failed to initialize permission service: %w", err) } + s.permissionService = permService // Shared channel service used by ChannelHandler, WebhookHandler, // FormHandler, RequestTypeHandler, and AssetReportHandler for the @@ -322,6 +331,7 @@ func (s *Server) initialize() error { cfg.Auth.SessionValidationCacheTTL, primarySessionCacheMB, ) + s.sessionManager = sessionManager effectivePort := cfg.Port if cfg.AllowedPort != "" { @@ -379,6 +389,7 @@ func (s *Server) initialize() error { apiTokenCacheMB, _ := config.SplitSSHCacheBudget(s.memoryBudget.APITokenCacheMB, cfg.SSH.Enabled) tokenManager := auth.NewTokenManager(s.db, s.tokenTracker, apiTokenCacheMB) + s.tokenManager = tokenManager if cleaned, cleanupErr := tokenManager.CleanupExpiredTokens(); cleanupErr != nil { slog.Warn("failed to cleanup expired api tokens on startup", "error", cleanupErr) } else if cleaned > 0 { @@ -520,6 +531,7 @@ func (s *Server) initialize() error { transitionMatrixService := services.NewTransitionMatrixService(s.db) bulkOperationMetrics := services.NewBulkOperationMetrics() itemHandler := handlers.NewItemHandler(s.db, permService, s.activityTracker, s.notificationService, s.memoryBudget.ItemCacheMB) + s.itemCache = itemHandler.ItemCacheService() itemHandler.SetDBRequestTimeout(s.config.DB.RequestTimeout) customFieldHandler := handlers.NewCustomFieldHandler(s.db) workspaceHandler := handlers.NewWorkspaceHandler(s.db, permService, s.activityTracker, workspaceKeyCache, authorizationCacheInvalidator) @@ -592,6 +604,7 @@ func (s *Server) initialize() error { agentHandler := handlers.NewAgentHandler(s.db, permService) scimTokenManager := auth.NewSCIMTokenManager(s.db, s.memoryBudget.SCIMTokenCacheMB) + s.scimTokenManager = scimTokenManager scimAuthMiddleware := middleware.NewSCIMAuthMiddleware(scimTokenManager) scimHandler := handlers.NewSCIMHandler( repository.NewSCIMRepository(s.db), @@ -1979,29 +1992,7 @@ func (s *Server) Shutdown(ctx context.Context) error { s.databasePoolMonitor.Stop() } - // Stop schedulers first - use safeClose helper to avoid panics on already-closed channels - safeClose := func(ch chan struct{}) { - if ch != nil { - defer func() { recover() }() //nolint:errcheck // Intentionally ignoring recover() return; used to suppress panics from closing already-closed channels - close(ch) - } - } - - // Close, but do NOT nil, the stop channels: background schedulers select - // on these fields in a loop, so the nil-write races with their reads (and - // a select on a nil channel blocks forever, leaking the goroutine). - // Double-close safety comes from safeClose's recover, not from nil-ing. - safeClose(s.scmSyncStopChan) - safeClose(s.issueSyncStopChan) - safeClose(s.magicLinkStopChan) - - if s.cleanupTicker != nil { - // Stop, but do NOT nil: runActivityCleanup selects on cleanupTicker.C - // in a loop and the nil-write races with that read. - s.cleanupTicker.Stop() - } - safeClose(s.cleanupStopChan) - safeClose(s.jiraHostStopChan) + s.stopBackgroundLoops() if s.notificationScheduler != nil { slog.Info("stopping notification scheduler") @@ -2141,6 +2132,9 @@ func isAPIPath(p string) bool { // cleanup releases all resources. func (s *Server) cleanup() { + // New also calls cleanup after a partially completed initialize. Signal any + // loops already started there, even when Shutdown was never reachable. + s.stopBackgroundLoops() if s.databasePoolMonitor != nil { s.databasePoolMonitor.Stop() } @@ -2228,12 +2222,47 @@ func (s *Server) cleanup() { _ = s.tokenTracker.Close() } + // These caches are owned by this HTTP server, including on partial startup. + // Close them after consumers have stopped, before releasing the shared DB. + if s.permissionService != nil { + _ = s.permissionService.Close() + } + if s.sessionManager != nil { + _ = s.sessionManager.Close() + } + if s.tokenManager != nil { + _ = s.tokenManager.Close() + } + if s.scimTokenManager != nil { + _ = s.scimTokenManager.Close() + } + if s.itemCache != nil { + _ = s.itemCache.Close() + } + // Close database if s.db != nil { _ = s.db.Close() } } +func (s *Server) stopBackgroundLoops() { + s.backgroundStopOnce.Do(func() { + // Do not nil these fields: workers read them concurrently. + if s.cleanupTicker != nil { + s.cleanupTicker.Stop() + } + for _, ch := range []chan struct{}{ + s.scmSyncStopChan, s.issueSyncStopChan, s.magicLinkStopChan, + s.cleanupStopChan, s.jiraHostStopChan, + } { + if ch != nil { + close(ch) + } + } + }) +} + // RegisterDatabasePool makes a process-local auxiliary SQL pool visible to // admin diagnostics and threshold monitoring. func (s *Server) RegisterDatabasePool(name string, db database.Database) error { diff --git a/internal/services/item_cache.go b/internal/services/item_cache.go index 3e9862cdf..fb8e39544 100644 --- a/internal/services/item_cache.go +++ b/internal/services/item_cache.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "log/slog" + "sync" "sync/atomic" "time" @@ -28,6 +29,8 @@ type ItemHierarchyCache struct { type ItemCacheService struct { hierarchyCache *bigcache.BigCache db database.Database + closeOnce sync.Once + closeErr error // Cache statistics hierarchyHits int64 @@ -38,6 +41,17 @@ type ItemCacheService struct { config ItemCacheConfig } +// Close stops the hierarchy cache's background worker without closing the +// shared database. It may be called more than once. +func (ics *ItemCacheService) Close() error { + ics.closeOnce.Do(func() { + if ics.hierarchyCache != nil { + ics.closeErr = ics.hierarchyCache.Close() + } + }) + return ics.closeErr +} + // ItemCacheConfig represents configuration for the item cache type ItemCacheConfig struct { HierarchyTTL time.Duration `json:"hierarchy_ttl"` // Default: 30min diff --git a/internal/services/permission_cache.go b/internal/services/permission_cache.go index 637ecf618..04dbbe15f 100644 --- a/internal/services/permission_cache.go +++ b/internal/services/permission_cache.go @@ -26,6 +26,8 @@ type PermissionService struct { cacheCommitMu sync.RWMutex cacheGeneration atomic.Uint64 workspaceAccess *workspaceAccessCache + closeOnce sync.Once + closeErr error hits int64 misses int64 @@ -968,7 +970,13 @@ func (ps *PermissionService) getRecentlyActiveUsers(duration time.Duration) ([]i return scanIntColumn(rows) } -// Close gracefully shuts down the permission service +// Close stops the cache worker without closing the shared database. +// Repeated calls are safe, including when no cache was initialized. func (ps *PermissionService) Close() error { - return ps.cache.Close() + ps.closeOnce.Do(func() { + if ps.cache != nil { + ps.closeErr = ps.cache.Close() + } + }) + return ps.closeErr }