Real-Time Architecture Patterns

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

# Linux kernel parameters for high-connection workloads
# /etc/sysctl.conf or /etc/sysctl.d/99-realtime.conf
 
# Increase system-wide file descriptor limits
fs.file-max = 2000000
fs.nr_open = 2000000
 
# Increase socket buffer sizes (tune based on workload)
net.core.rmem_max = 16777216
net.core.wmem_max = 16777216
net.core.rmem_default = 1048576
net.core.wmem_default = 1048576
 
# TCP buffer tuning (min, default, max)
# Smaller buffers = more connections, higher latency
net.ipv4.tcp_rmem = 4096 12582912 16777216
net.ipv4.tcp_wmem = 4096 12582912 16777216
 
# Enable TCP memory management
net.ipv4.tcp_mem = 786432 1048576 1572864
 
# Increase connection tracking table
net.netfilter.nf_conntrack_max = 2000000
 
# Expand ephemeral port range
net.ipv4.ip_local_port_range = 1024 65535
 
# Enable TCP keepalive (essential for detecting dead connections)
net.ipv4.tcp_keepalive_time = 600
net.ipv4.tcp_keepalive_intvl = 60
net.ipv4.tcp_keepalive_probes = 5
 
# Increase backlog queue for incoming connections
net.core.somaxconn = 65535
net.ipv4.tcp_max_syn_backlog = 65535
 
# Enable TCP Fast Open (reduces handshake latency)
net.ipv4.tcp_fastopen = 3

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183

package main
 
import (
    "log"
    "net"
    "sync"
    "syscall"
 
    "golang.org/x/sys/unix"
)
 
// ConnectionManager handles millions of concurrent connections
// using epoll for event notification
type ConnectionManager struct {
    epollFD    int
    listener   *net.TCPListener
    clients    map[int]*ClientConnection
    clientsMu  sync.RWMutex
    eventPool  []unix.EpollEvent
}
 
// ClientConnection represents minimal state per connection
// Keeping this small is critical for memory efficiency
type ClientConnection struct {
    fd       int
    addr     string
    userID   uint64          // User identifier
    channels []uint32        // Subscribed channel IDs (compact representation)
    lastPing int64           // Unix timestamp of last ping
    sendBuf  []byte          // Outgoing message buffer (pooled)
    recvBuf  []byte          // Incoming message buffer (pooled)
}
 
func NewConnectionManager(addr string) (*ConnectionManager, error) {
    // Create epoll instance
    epollFD, err := unix.EpollCreate1(0)
    if err != nil {
        return nil, err
    }
 
    // Create TCP listener
    listener, err := net.Listen("tcp", addr)
    if err != nil {
        return nil, err
    }
 
    cm := &ConnectionManager{
        epollFD:   epollFD,
        listener:  listener.(*net.TCPListener),
        clients:   make(map[int]*ClientConnection),
        eventPool: make([]unix.EpollEvent, 10000), // Process 10k events per iteration
    }
 
    // Register listener for accept events
    listenerFD := getListenerFD(listener)
    event := &unix.EpollEvent{
        Events: unix.EPOLLIN | unix.EPOLLET, // Edge-triggered
        Fd:     int32(listenerFD),
    }
    unix.EpollCtl(epollFD, unix.EPOLL_CTL_ADD, listenerFD, event)
 
    return cm, nil
}
 
// Run is the main event loop - single-threaded, non-blocking
func (cm *ConnectionManager) Run() {
    log.Println("Starting event loop...")
    
    for {
        // Wait for events (blocks until at least one event is ready)
        numEvents, err := unix.EpollWait(cm.epollFD, cm.eventPool, -1)
        if err != nil {
            if err == unix.EINTR {
                continue // Interrupted by signal, retry
            }
            log.Printf("epoll_wait error: %v", err)
            continue
        }
 
        // Process all ready events
        for i := 0; i < numEvents; i++ {
            event := cm.eventPool[i]
            fd := int(event.Fd)
 
            if event.Events&(unix.EPOLLERR|unix.EPOLLHUP) != 0 {
                // Connection error or hangup
                cm.closeConnection(fd)
                continue
            }
 
            if event.Events&unix.EPOLLIN != 0 {
                // Data available to read
                cm.handleRead(fd)
            }
 
            if event.Events&unix.EPOLLOUT != 0 {
                // Socket ready for writing
                cm.handleWrite(fd)
            }
        }
    }
}
 
func (cm *ConnectionManager) handleRead(fd int) {
    cm.clientsMu.RLock()
    client, exists := cm.clients[fd]
    cm.clientsMu.RUnlock()
 
    if !exists {
        return
    }
 
    // Use pooled buffer for reading
    buf := client.recvBuf
    if buf == nil {
        buf = bufferPool.Get().([]byte)
        client.recvBuf = buf
    }
 
    // Non-blocking read - edge-triggered means we must drain the buffer
    for {
        n, err := unix.Read(fd, buf)
        if err != nil {
            if err == unix.EAGAIN || err == unix.EWOULDBLOCK {
                break // No more data available
            }
            cm.closeConnection(fd)
            return
        }
        if n == 0 {
            cm.closeConnection(fd)
            return
        }
 
        // Process the message (dispatch to worker pool for heavy processing)
        cm.processMessage(client, buf[:n])
    }
}
 
// Buffer pool for zero-allocation message handling
var bufferPool = sync.Pool{
    New: func() interface{} {
        return make([]byte, 4096) // 4KB buffers
    },
}
 
func (cm *ConnectionManager) closeConnection(fd int) {
    cm.clientsMu.Lock()
    client, exists := cm.clients[fd]
    if exists {
        // Return buffers to pool
        if client.recvBuf != nil {
            bufferPool.Put(client.recvBuf)
        }
        if client.sendBuf != nil {
            bufferPool.Put(client.sendBuf)
        }
        delete(cm.clients, fd)
    }
    cm.clientsMu.Unlock()
 
    // Deregister from epoll and close socket
    unix.EpollCtl(cm.epollFD, unix.EPOLL_CTL_DEL, fd, nil)
    unix.Close(fd)
}
 
func (cm *ConnectionManager) processMessage(client *ClientConnection, data []byte) {
    // Dispatch to worker pool for CPU-intensive processing
    // This keeps the event loop fast and non-blocking
}
 
func getListenerFD(listener net.Listener) int {
    // Platform-specific extraction of file descriptor
    return 0 // Simplified
}
 
func main() {
    cm, err := NewConnectionManager(":8080")
    if err != nil {
        log.Fatal(err)
    }
    cm.Run()
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

package main
 
import (
    "crypto/sha256"
    "encoding/binary"
    "encoding/json"
    "net/http"
    "sync/atomic"
)
 
// ConnectionRouter determines which connection server a client should use
type ConnectionRouter struct {
    servers       []ServerInfo
    healthChecker *HealthChecker
    algorithm     RoutingAlgorithm
}
 
type ServerInfo struct {
    URL             string
    CurrentLoad     int64  // Current connection count
    MaxCapacity     int64  // Maximum connections this server should accept
    HealthScore     int    // 0-100, higher is better
    DrainMode       bool   // True if server is being drained for maintenance
}
 
type RoutingAlgorithm int
 
const (
    LeastConnections RoutingAlgorithm = iota
    ConsistentHashing
    WeightedRandom
)
 
// GetConnectionServer returns the best server for a new connection
func (r *ConnectionRouter) GetConnectionServer(userID string, guildID string) *ServerInfo {
    healthyServers := r.filterHealthyServers()
    if len(healthyServers) == 0 {
        return nil
    }
 
    switch r.algorithm {
    case ConsistentHashing:
        // Use consistent hashing for guild-based routing
        // All users in same guild connect to same server for efficiency
        return r.consistentHashSelect(guildID, healthyServers)
    
    case LeastConnections:
        // Find server with lowest load relative to capacity
        return r.leastConnectionsSelect(healthyServers)
    
    case WeightedRandom:
        // Random selection weighted by available capacity
        return r.weightedRandomSelect(healthyServers)
    }
 
    return healthyServers[0]
}
 
func (r *ConnectionRouter) filterHealthyServers() []*ServerInfo {
    var healthy []*ServerInfo
    for i := range r.servers {
        server := &r.servers[i]
        if server.HealthScore >= 50 && 
           !server.DrainMode && 
           atomic.LoadInt64(&server.CurrentLoad) < server.MaxCapacity {
            healthy = append(healthy, server)
        }
    }
    return healthy
}
 
func (r *ConnectionRouter) consistentHashSelect(key string, servers []*ServerInfo) *ServerInfo {
    // Simple consistent hashing - in production, use a ring with virtual nodes
    hash := sha256.Sum256([]byte(key))
    hashValue := binary.BigEndian.Uint64(hash[:8])
    index := hashValue % uint64(len(servers))
    return servers[index]
}
 
func (r *ConnectionRouter) leastConnectionsSelect(servers []*ServerInfo) *ServerInfo {
    var best *ServerInfo
    var bestRatio float64 = 2.0 // Higher than possible load ratio
 
    for _, server := range servers {
        currentLoad := float64(atomic.LoadInt64(&server.CurrentLoad))
        ratio := currentLoad / float64(server.MaxCapacity)
        
        if ratio < bestRatio {
            bestRatio = ratio
            best = server
        }
    }
    return best
}
 
// HTTP handler for connection routing
func (r *ConnectionRouter) HandleConnectRequest(w http.ResponseWriter, req *http.Request) {
    userID := req.Header.Get("X-User-ID")
    guildID := req.URL.Query().Get("guild_id")
 
    server := r.GetConnectionServer(userID, guildID)
    if server == nil {
        http.Error(w, "No healthy servers available", http.StatusServiceUnavailable)
        return
    }
 
    // Return WebSocket URL for client to connect directly
    response := map[string]string{
        "websocket_url": server.URL,
    }
    json.NewEncoder(w).Encode(response)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

package state
 
import (
    "context"
    "encoding/json"
    "sync"
    "time"
 
    "github.com/go-redis/redis/v8"
)
 
// ConnectionState represents minimal in-memory state per connection
// This struct should be as small as possible - every byte is multiplied by connection count
type ConnectionState struct {
    ConnectionID  uint64    // 8 bytes
    UserID        uint64    // 8 bytes
    DeviceID      uint32    // 4 bytes
    Permissions   uint32    // 4 bytes - bitfield of permissions
    Subscriptions []uint32  // Variable - channel IDs, not full objects
    ConnectedAt   int64     // 8 bytes - unix timestamp
    LastActivity  int64     // 8 bytes - for idle detection
}
 
// Full state stored in Redis - loaded on demand
type FullUserState struct {
    UserID          uint64
    Username        string
    AvatarURL       string
    Permissions     map[string][]string // Channel -> Permissions
    Preferences     UserPreferences
    RateLimitTokens int
    PresenceStatus  string
    CustomStatus    string
}
 
type UserPreferences struct {
    NotificationsEnabled bool
    Theme               string
    Language            string
}
 
// StateManager handles hybrid local/remote state
type StateManager struct {
    redis       *redis.Client
    localCache  sync.Map // map[uint64]*cachedState
    ttl         time.Duration
}
 
type cachedState struct {
    state     *FullUserState
    expiresAt time.Time
}
 
func NewStateManager(redisClient *redis.Client) *StateManager {
    return &StateManager{
        redis: redisClient,
        ttl:   time.Minute * 5,
    }
}
 
// GetFullState retrieves full user state with local caching
func (sm *StateManager) GetFullState(ctx context.Context, userID uint64) (*FullUserState, error) {
    // Check local cache first
    if cached, ok := sm.localCache.Load(userID); ok {
        cs := cached.(*cachedState)
        if time.Now().Before(cs.expiresAt) {
            return cs.state, nil
        }
        // Expired, delete and fetch fresh
        sm.localCache.Delete(userID)
    }
 
    // Fetch from Redis
    key := fmt.Sprintf("user:state:%d", userID)
    data, err := sm.redis.Get(ctx, key).Bytes()
    if err != nil {
        if err == redis.Nil {
            return nil, ErrUserNotFound
        }
        return nil, err
    }
 
    var state FullUserState
    if err := json.Unmarshal(data, &state); err != nil {
        return nil, err
    }
 
    // Cache locally
    sm.localCache.Store(userID, &cachedState{
        state:     &state,
        expiresAt: time.Now().Add(sm.ttl),
    })
 
    return &state, nil
}
 
// InvalidateLocalCache removes user from local cache
// Called when receiving cache invalidation event from Redis pub/sub
func (sm *StateManager) InvalidateLocalCache(userID uint64) {
    sm.localCache.Delete(userID)
}
 
// SubscribeToCacheInvalidation listens for cache invalidation events
func (sm *StateManager) SubscribeToCacheInvalidation(ctx context.Context) {
    pubsub := sm.redis.Subscribe(ctx, "cache:invalidate:user")
    defer pubsub.Close()
 
    for {
        select {
        case <-ctx.Done():
            return
        case msg := <-pubsub.Channel():
            // Message payload is user ID
            var userID uint64
            if _, err := fmt.Sscanf(msg.Payload, "%d", &userID); err == nil {
                sm.InvalidateLocalCache(userID)
            }
        }
    }
}
 
// PublishCacheInvalidation notifies all servers to invalidate cached state
func (sm *StateManager) PublishCacheInvalidation(ctx context.Context, userID uint64) error {
    return sm.redis.Publish(ctx, "cache:invalidate:user", userID).Err()
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173

package keepalive
 
import (
    "sync"
    "time"
 
    "github.com/gorilla/websocket"
)
 
type KeepaliveManager struct {
    connections  sync.Map // map[connectionID]*trackedConnection
    pingInterval time.Duration
    pongTimeout  time.Duration
    idleTimeout  time.Duration
    ticker       *time.Ticker
    stopCh       chan struct{}
}
 
type trackedConnection struct {
    conn         *websocket.Conn
    connectionID string
    lastPong     time.Time
    lastActivity time.Time
    pingPending  bool
    mu           sync.Mutex
}
 
func NewKeepaliveManager(pingInterval, pongTimeout, idleTimeout time.Duration) *KeepaliveManager {
    return &KeepaliveManager{
        pingInterval: pingInterval, // e.g., 30 seconds
        pongTimeout:  pongTimeout,  // e.g., 10 seconds
        idleTimeout:  idleTimeout,  // e.g., 5 minutes
        ticker:       time.NewTicker(pingInterval),
        stopCh:       make(chan struct{}),
    }
}
 
func (km *KeepaliveManager) Track(connID string, conn *websocket.Conn) {
    tc := &trackedConnection{
        conn:         conn,
        connectionID: connID,
        lastPong:     time.Now(),
        lastActivity: time.Now(),
    }
 
    // Set up pong handler
    conn.SetPongHandler(func(appData string) error {
        tc.mu.Lock()
        tc.lastPong = time.Now()
        tc.pingPending = false
        tc.mu.Unlock()
        return nil
    })
 
    km.connections.Store(connID, tc)
}
 
func (km *KeepaliveManager) RecordActivity(connID string) {
    if val, ok := km.connections.Load(connID); ok {
        tc := val.(*trackedConnection)
        tc.mu.Lock()
        tc.lastActivity = time.Now()
        tc.mu.Unlock()
    }
}
 
func (km *KeepaliveManager) Untrack(connID string) {
    km.connections.Delete(connID)
}
 
// Run starts the keepalive check loop
func (km *KeepaliveManager) Run(onTimeout func(connID string)) {
    for {
        select {
        case <-km.stopCh:
            return
        case <-km.ticker.C:
            km.checkConnections(onTimeout)
        }
    }
}
 
func (km *KeepaliveManager) checkConnections(onTimeout func(connID string)) {
    now := time.Now()
    var toClose []string
 
    km.connections.Range(func(key, value interface{}) bool {
        connID := key.(string)
        tc := value.(*trackedConnection)
 
        tc.mu.Lock()
        defer tc.mu.Unlock()
 
        // Check for pong timeout (ping was sent but no pong received)
        if tc.pingPending && now.Sub(tc.lastPong) > km.pongTimeout {
            toClose = append(toClose, connID)
            return true
        }
 
        // Check for idle timeout
        if now.Sub(tc.lastActivity) > km.idleTimeout {
            toClose = append(toClose, connID)
            return true
        }
 
        // Send ping if interval has passed
        if !tc.pingPending && now.Sub(tc.lastPong) > km.pingInterval {
            if err := tc.conn.WriteControl(
                websocket.PingMessage, 
                []byte{}, 
                now.Add(time.Second),
            ); err != nil {
                toClose = append(toClose, connID)
            } else {
                tc.pingPending = true
            }
        }
 
        return true
    })
 
    // Close dead connections
    for _, connID := range toClose {
        if val, ok := km.connections.Load(connID); ok {
            tc := val.(*trackedConnection)
            tc.conn.Close()
            km.connections.Delete(connID)
            onTimeout(connID)
        }
    }
}
 
func (km *KeepaliveManager) Stop() {
    close(km.stopCh)
    km.ticker.Stop()
}
 
// Metrics returns current connection health statistics
func (km *KeepaliveManager) Metrics() KeepaliveMetrics {
    var total, healthy, pendingPong, nearIdle int
 
    km.connections.Range(func(key, value interface{}) bool {
        total++
        tc := value.(*trackedConnection)
        tc.mu.Lock()
        defer tc.mu.Unlock()
 
        if time.Since(tc.lastPong) < km.pingInterval*2 {
            healthy++
        }
        if tc.pingPending {
            pendingPong++
        }
        if time.Since(tc.lastActivity) > km.idleTimeout/2 {
            nearIdle++
        }
        return true
    })
 
    return KeepaliveMetrics{
        TotalConnections:   total,
        HealthyConnections: healthy,
        PendingPong:        pendingPong,
        NearIdleTimeout:    nearIdle,
    }
}
 
type KeepaliveMetrics struct {
    TotalConnections   int
    HealthyConnections int
    PendingPong        int
    NearIdleTimeout    int
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

package main
 
import (
    "context"
    "encoding/json"
    "log"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
)
 
type GracefulShutdownManager struct {
    server           *ConnectionServer
    draining         bool
    shutdownComplete chan struct{}
    mu               sync.RWMutex
}
 
type ReconnectMessage struct {
    Type         string `json:"type"`
    Reason       string `json:"reason"`
    ReconnectURL string `json:"reconnect_url"`
    ReconnectIn  int    `json:"reconnect_in_seconds"`
}
 
func (gsm *GracefulShutdownManager) HandleSignals() {
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT, syscall.SIGUSR1)
 
    for sig := range sigCh {
        switch sig {
        case syscall.SIGTERM, syscall.SIGINT:
            log.Println("Received shutdown signal, initiating graceful drain...")
            gsm.initiateGracefulShutdown()
        case syscall.SIGUSR1:
            log.Println("Received SIGUSR1, initiating drain without shutdown...")
            gsm.initiateDrain()
        }
    }
}
 
func (gsm *GracefulShutdownManager) initiateGracefulShutdown() {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
    defer cancel()
 
    // Phase 1: Enter drain mode (stop accepting new connections)
    gsm.mu.Lock()
    gsm.draining = true
    gsm.mu.Unlock()
 
    log.Printf("Phase 1: Entered drain mode, stopping new connections")
    gsm.server.StopAcceptingNewConnections()
 
    // Phase 2: Notify all connected clients to reconnect elsewhere
    reconnectURL := gsm.server.GetAlternateServerURL()
    message := ReconnectMessage{
        Type:         "reconnect",
        Reason:       "server_maintenance",
        ReconnectURL: reconnectURL,
        ReconnectIn:  30, // Client should reconnect within 30 seconds
    }
    messageBytes, _ := json.Marshal(message)
 
    log.Printf("Phase 2: Notifying %d connections to reconnect", gsm.server.ConnectionCount())
    gsm.server.BroadcastToAll(messageBytes)
 
    // Phase 3: Wait for voluntary disconnections
    waitDuration := 45 * time.Second
    checkInterval := time.Second
    deadline := time.Now().Add(waitDuration)
 
    log.Printf("Phase 3: Waiting %v for voluntary disconnections", waitDuration)
    for time.Now().Before(deadline) {
        remaining := gsm.server.ConnectionCount()
        if remaining == 0 {
            log.Println("All connections closed voluntarily")
            break
        }
        log.Printf("Waiting... %d connections remaining", remaining)
        time.Sleep(checkInterval)
    }
 
    // Phase 4: Force close remaining connections
    remaining := gsm.server.ConnectionCount()
    if remaining > 0 {
        log.Printf("Phase 4: Force closing %d remaining connections", remaining)
        gsm.server.ForceCloseAllConnections("server_shutdown")
    }
 
    // Phase 5: Wait for all goroutines to finish
    log.Println("Phase 5: Waiting for cleanup...")
    gsm.server.WaitForShutdown(ctx)
 
    log.Println("Graceful shutdown complete")
    close(gsm.shutdownComplete)
}
 
func (gsm *GracefulShutdownManager) initiateDrain() {
    gsm.mu.Lock()
    gsm.draining = true
    gsm.mu.Unlock()
    gsm.server.StopAcceptingNewConnections()
    log.Println("Server is now in drain mode - no new connections accepted")
}
 
func (gsm *GracefulShutdownManager) IsDraining() bool {
    gsm.mu.RLock()
    defer gsm.mu.RUnlock()
    return gsm.draining
}
 
func (gsm *GracefulShutdownManager) WaitForShutdown() {
    <-gsm.shutdownComplete
}
 
// ConnectionServer stub for illustration
type ConnectionServer struct {
    // ... connection management
}
 
func (s *ConnectionServer) StopAcceptingNewConnections()                { /* ... */ }
func (s *ConnectionServer) GetAlternateServerURL() string               { return "wss://alt-server.example.com" }
func (s *ConnectionServer) ConnectionCount() int                        { return 0 }
func (s *ConnectionServer) BroadcastToAll(msg []byte)                   { /* ... */ }
func (s *ConnectionServer) ForceCloseAllConnections(reason string)      { /* ... */ }
func (s *ConnectionServer) WaitForShutdown(ctx context.Context)         { /* ... */ }