Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 16 additions & 2 deletions internal/auth/scim_tokens.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"errors"
"fmt"
"log/slog"
"sync"
"time"

"windshift/internal/cacheutil"
Expand All @@ -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
Expand Down
14 changes: 14 additions & 0 deletions internal/auth/session.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (
"log/slog"
"net"
"net/http"
"sync"
"time"

"windshift/internal/database"
Expand Down Expand Up @@ -54,13 +55,26 @@ 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
// fail closed.
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"`
Expand Down
14 changes: 14 additions & 0 deletions internal/auth/tokens.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"errors"
"fmt"
"log/slog"
"sync"
"time"

"windshift/internal/cacheutil"
Expand Down Expand Up @@ -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
Expand Down
81 changes: 55 additions & 26 deletions internal/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"os"
"strconv"
"strings"
"sync"
"time"

"windshift/internal/aitools"
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -322,6 +331,7 @@ func (s *Server) initialize() error {
cfg.Auth.SessionValidationCacheTTL,
primarySessionCacheMB,
)
s.sessionManager = sessionManager

effectivePort := cfg.Port
if cfg.AllowedPort != "" {
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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()
}
Expand Down Expand Up @@ -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 {
Expand Down
14 changes: 14 additions & 0 deletions internal/services/item_cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"encoding/json"
"fmt"
"log/slog"
"sync"
"sync/atomic"
"time"

Expand All @@ -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
Expand All @@ -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
Expand Down
12 changes: 10 additions & 2 deletions internal/services/permission_cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@ type PermissionService struct {
cacheCommitMu sync.RWMutex
cacheGeneration atomic.Uint64
workspaceAccess *workspaceAccessCache
closeOnce sync.Once
closeErr error

hits int64
misses int64
Expand Down Expand Up @@ -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
}
Loading