diff --git a/CONFIGURATION.md b/CONFIGURATION.md index fc8423702..e14688791 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -133,6 +133,62 @@ } ``` +## Multi-Node Lavalink Configuration + +Quaver supports connecting to multiple Lavalink nodes for improved reliability, load distribution, and region-based routing. To enable multi-node mode, replace the `lavalink` object with a `nodes` array: + +```json +{ + "lavalink": { + "nodes": [ + { + "host": "sg-lavalink.example.com", + "port": 2333, + "password": "youshallnotpass", + "secure": true, + "region": "singapore", + "reconnect": { + "delay": 3000, + "tries": 5 + } + }, + { + "host": "us-lavalink.example.com", + "port": 2333, + "password": "youshallnotpass", + "secure": true, + "region": "us-central", + "reconnect": { + "delay": 3000, + "tries": 5 + } + }, + { + "host": "eu-lavalink.example.com", + "port": 2333, + "password": "youshallnotpass", + "secure": true, + "region": "rotterdam", + "reconnect": { + "delay": 3000, + "tries": 5 + } + } + ] + } +} +``` + +### Multi-Node Features + +- **Ping-Based Region Affinity**: Quaver learns which Lavalink nodes provide the best latency for different Discord media regions over time, automatically routing new connections to the optimal node +- **Intelligent Load Balancing**: Uses penalty-based selection considering CPU load, active players, and frame statistics +- **Automatic Failover**: If a node becomes unavailable, new players route to healthy nodes +- **Session Recovery**: Players can be restored after restarts, maintaining node affinity when possible + +> **Note**: The `region` field in each node configuration is an internal identifier for affinity tracking. You can set it to any value you prefer (e.g., `"sg1"`, `"us-east"`, `"europe"`). It's not matched against Discord's voice regions. + + | Config Item Path | Description | Required | Version Added | |-------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------------------------|---------------| | `token` | Your bot token. You can get it from the [Discord Developer Portal](https://discord.com/developers/applications). | ✅ | | @@ -152,12 +208,20 @@ | `managers` | The [user IDs](https://support.discord.com/hc/en-us/articles/206346498-Where-can-I-find-my-User-Server-Message-ID-) that are given manager-level permissions on Quaver. | ✅ | | | `database.protocol` | The database protocol. At this time, only `sqlite` is supported. | ✅ | | | `database.path` | The database path. For `sqlite`, this is relative to your Quaver directory containing `dist`, `locales`, etc. | ✅ | | -| `lavalink.host` | The Lavalink instance host address. | ✅ | | -| `lavalink.port` | The Lavalink instance port. | ✅ | | -| `lavalink.password` | The Lavalink instance password. | ✅ | | -| `lavalink.secure` | Whether the Lavalink instance uses a secure connection. | ❌ | | -| `lavalink.reconnect.delay` | The delay in milliseconds between Lavalink reconnect attempts. | ❌ | | -| `lavalink.reconnect.tries` | The number of times to attempt to reconnect to Lavalink. | ❌ | | +| `lavalink.host` | The Lavalink instance host address. **Single-node mode only.** Use `lavalink.nodes` for multi-node configuration. | ✅ (single-node)
❌ (multi-node) | | +| `lavalink.port` | The Lavalink instance port. **Single-node mode only.** | ✅ (single-node)
❌ (multi-node) | | +| `lavalink.password` | The Lavalink instance password. **Single-node mode only.** | ✅ (single-node)
❌ (multi-node) | | +| `lavalink.secure` | Whether the Lavalink instance uses a secure connection. **Single-node mode only.** | ❌ | | +| `lavalink.reconnect.delay` | The delay in milliseconds between Lavalink reconnect attempts. **Single-node mode only.** | ❌ | | +| `lavalink.reconnect.tries` | The number of times to attempt to reconnect to Lavalink. **Single-node mode only.** | ❌ | | +| `lavalink.nodes` | Array of Lavalink node configurations for multi-node mode. When specified, enables intelligent load balancing and region-based routing across multiple Lavalink servers. **Multi-node mode only.** Mutually exclusive with single-node fields (`host`, `port`, `password`, etc.). | ✅ (multi-node)
❌ (single-node) | `8.0.0` | +| `lavalink.nodes[].host` | The Lavalink node host address. | ✅ (multi-node) | `8.0.0` | +| `lavalink.nodes[].port` | The Lavalink node port. | ✅ (multi-node) | `8.0.0` | +| `lavalink.nodes[].password` | The Lavalink node password. | ✅ (multi-node) | `8.0.0` | +| `lavalink.nodes[].secure` | Whether the Lavalink node uses a secure connection. | ❌ | `8.0.0` | +| `lavalink.nodes[].region` | Discord voice region this node serves (e.g., `singapore`, `us-central`, `rotterdam`). Players in voice channels with matching `rtcRegion` will route to nodes configured for that region. If multiple nodes serve the same region, load balancing determines which node is used. Must match region IDs from Discord's `/voice/regions` API. | ❌ | `8.0.0` | +| `lavalink.nodes[].reconnect.delay` | The delay in milliseconds between reconnect attempts for this node. | ❌ | `8.0.0` | +| `lavalink.nodes[].reconnect.tries` | The number of times to attempt to reconnect to this node. | ❌ | `8.0.0` | | `features.autolyrics` | Auto Lyrics feature: Allows users to toggle Quaver automatically sending lyrics for the current song. | ✅ | `6.7.0` | | `features.autolyrics.enabled` | Whether the feature is enabled. | ✅ | `6.7.0` | | `features.autolyrics.whitelist` | Whether the feature requires guilds to be whitelisted. You will be able to whitelist guilds through the terminal. | ✅ (if feature is enabled)
❌ (if feature is disabled) | `6.7.0` | diff --git a/src/commands/chatInputCommands/lyrics.ts b/src/commands/chatInputCommands/lyrics.ts index 967a41b17..b86c697b0 100644 --- a/src/commands/chatInputCommands/lyrics.ts +++ b/src/commands/chatInputCommands/lyrics.ts @@ -64,9 +64,6 @@ export default new ChatInputCommandHandler() const response = await interaction.client.music.rest.execute({ path: `/v4/sessions/${player.api.session.id}/players/${interaction.guildId}/track/lyrics`, method: 'GET', - headers: { - Authorization: `Bearer ${settings.lavalink.password}`, - }, }); json = (await response.json()) as LavaLyricsResponse; lyrics = formatLavaLyricsResponse(json, player); diff --git a/src/events/music/trackStart.ts b/src/events/music/trackStart.ts index 03415f5fb..3e53a63d9 100644 --- a/src/events/music/trackStart.ts +++ b/src/events/music/trackStart.ts @@ -236,9 +236,6 @@ export default { const response = await queue.player.client.music.rest.execute({ path: `/v4/sessions/${queue.player.api.session.id}/players/${guild.id}/track/lyrics`, method: 'GET', - headers: { - Authorization: `Bearer ${settings.lavalink.password}`, - }, }); json = (await response.json()) as LavaLyricsResponse; lyrics = formatLavaLyricsResponse(json, queue.player); diff --git a/src/lib/QuaverClient.ts b/src/lib/QuaverClient.ts index 7f4fb0e21..d1d2905a0 100644 --- a/src/lib/QuaverClient.ts +++ b/src/lib/QuaverClient.ts @@ -1,6 +1,8 @@ import { getAbsoluteFileURL } from '@zptxdev/zptx-lib'; import { Client, GatewayDispatchEvents } from 'discord.js'; import { readdirSync } from 'node:fs'; +import { dirname, resolve } from 'node:path'; +import { fileURLToPath } from 'node:url'; import type { Server } from 'socket.io'; import { MessageOptionsBuilderType } from '.'; import type { EventHandler } from './builders'; @@ -9,12 +11,12 @@ import { InteractionHandler, type InteractionHandlerMapsFlat, } from './interactions'; -import { ConnectionHealthMonitor, QuaverNode } from './music'; +import { ConnectionHealthMonitor, QuaverNode, QuaverCluster, RegionAffinity } from './music'; import { settings } from './util'; export class QuaverClient extends Client { io?: Server; - music?: QuaverNode; + music?: QuaverNode | QuaverCluster; interactionHandler: InteractionHandler; connectionHealth: ConnectionHealthMonitor; private lastMediaUnstable: boolean = false; @@ -31,39 +33,101 @@ export class QuaverClient extends Client { } connectToMusicNode(): void { - this.music = new QuaverNode( - { - info: { - host: settings.lavalink.host, - port: settings.lavalink.port, - auth: settings.lavalink.password, - tls: !!settings.lavalink.secure, - }, - ws: { - reconnecting: { - delay: settings.lavalink.reconnect.delay ?? 3000, - tries: settings.lavalink.reconnect.tries ?? 5, + const config = settings.lavalink; + + // Detect configuration type and instantiate appropriate class + if ('nodes' in config) { + // Multi-node configuration: use QuaverCluster + const nodes = config.nodes.map( + ( + node, + ): { + info: { host: string; port: number; auth: string; tls: boolean }; + ws: { reconnecting: { delay: number; tries: number } }; + region: string; + } => ({ + info: { + host: node.host, + port: node.port, + auth: node.password, + tls: !!node.secure, + }, + ws: { + reconnecting: { + delay: node.reconnect?.delay ?? 3000, + tries: node.reconnect?.tries ?? 5, + }, + }, + region: node.region, + }), + ); + + // Create RegionAffinity for ping-based node selection + const __dirname = dirname(fileURLToPath(import.meta.url)); + const databaseUri = settings.database + ? `${settings.database.protocol}://${resolve( + __dirname, + '..', + '..', + settings.database.path, + ).replace(/\\/g, '/')}` + : `sqlite://${resolve(__dirname, '..', '..', 'database.sqlite').replace(/\\/g, '/')}`; + const regionAffinity = new RegionAffinity(databaseUri); + + // Update ConnectionHealthMonitor with RegionAffinity + this.connectionHealth.setRegionAffinity(regionAffinity); + + this.music = new QuaverCluster( + { + nodes, + discord: { + sendGatewayCommand: (id, payload): void => + this.guilds.cache.get(id)?.shard?.send(payload), }, }, - discord: { - sendGatewayCommand: (id, payload): void => - this.guilds.cache.get(id)?.shard?.send(payload), + this, + regionAffinity, + ); + } else { + // Single-node configuration: use QuaverNode + this.music = new QuaverNode( + { + info: { + host: config.host, + port: config.port, + auth: config.password, + tls: !!config.secure, + }, + ws: { + reconnecting: { + delay: config.reconnect?.delay ?? 3000, + tries: config.reconnect?.tries ?? 5, + }, + }, + discord: { + sendGatewayCommand: (id, payload): void => + this.guilds.cache.get(id)?.shard?.send(payload), + }, }, - }, - this, - ); + this, + ); + } + + // Setup voice update handlers (work with both Node and Cluster) this.ws.on( GatewayDispatchEvents.VoiceServerUpdate, async (payload): Promise => { // Capture media server endpoint for health monitoring - this.connectionHealth.updateMediaEndpoint(payload.endpoint ?? null); - return this.music.players.handleVoiceUpdate(payload); + // Get the node ID from the player manager + const nodeId = this.music!.players.getNodeIdForGuild(payload.guild_id); + this.connectionHealth.updateMediaEndpoint(payload.endpoint ?? null, nodeId); + return this.music!.players.handleVoiceUpdate(payload); }, ); this.ws.on( GatewayDispatchEvents.VoiceStateUpdate, async (payload): Promise => - this.music.players.handleVoiceUpdate(payload), + this.music!.players.handleVoiceUpdate(payload), ); } @@ -141,7 +205,7 @@ export class QuaverClient extends Client { const player = await this.music?.players.fetch(guild.id); if (player?.voice.connected && player.queue.channel) { // Send warning message to the player's bound text channel - await player.sendMessage(guild.locale('MUSIC.PLAYER.CONNECTION_UNSTABLE'), { + await player.sendMessage(g.locale('MUSIC.PLAYER.CONNECTION_UNSTABLE'), { type: MessageOptionsBuilderType.Warning, }); } diff --git a/src/lib/music/ClusterPlayerManager.ts b/src/lib/music/ClusterPlayerManager.ts new file mode 100644 index 000000000..a8ee25843 --- /dev/null +++ b/src/lib/music/ClusterPlayerManager.ts @@ -0,0 +1,229 @@ +import type { QuaverNode } from './QuaverNode'; +import type { QuaverPlayer, QuaverPlayerJSON } from './QuaverPlayer'; +import type { FetchOptions, PlayerManager } from 'lavaclient'; +import type { VoiceServerUpdate, VoiceStateUpdate } from 'lavaclient/dist/playerVoice'; +import type { Identifiable } from 'lavaclient/dist/tools'; +import type { QuaverCluster } from './QuaverCluster'; +import type { Guild } from 'discord.js'; + +/** + * ClusterPlayerManager routes player operations to the appropriate QuaverNode + * based on region affinity and load balancing. + */ +export class ClusterPlayerManager implements PlayerManager { + readonly cluster: QuaverCluster; + private guildNodeMap: Map = new Map(); + + constructor(cluster: QuaverCluster) { + this.cluster = cluster; + } + + /** + * Get combined cache of all players across all nodes + */ + get cache(): Map> { + const combined = new Map>(); + for (const node of this.cluster.nodes.values()) { + for (const [guildId, player] of node.players.cache) { + combined.set(guildId, player as QuaverPlayer); + } + } + return combined; + } + + /** + * Check if a player exists for the guild + */ + has(guild: Identifiable): boolean { + const guildId = typeof guild === 'string' ? guild : guild.id; + return this.getNodeForGuild(guildId)?.players.has(guildId) ?? false; + } + + /** + * Resolve a player for the guild + */ + resolve(guild: Identifiable): QuaverPlayer | undefined { + const guildId = typeof guild === 'string' ? guild : guild.id; + const node = this.getNodeForGuild(guildId); + return node?.players.resolve(guildId) as QuaverPlayer | undefined; + } + + /** + * Get the node ID for a guild, verifying the player still exists + */ + getNodeIdForGuild(guildId: string): string | null { + const nodeId = this.guildNodeMap.get(guildId); + if (nodeId) { + // Verify player still exists + const node = this.cluster.nodes.get(nodeId); + if (node && node.players.has(guildId)) { + return nodeId; + } + // Clean up stale entry + this.guildNodeMap.delete(guildId); + } + + // Fallback: search all nodes directly + // This handles cases like restart or direct player creation + for (const [id, node] of this.cluster.nodes.entries()) { + if (node.players.has(guildId)) { + // Update our tracking + this.guildNodeMap.set(guildId, id); + return id; + } + } + + return null; + } + + /** + * Fetch players from all nodes + */ + fetch(cache?: boolean): Promise[]>; + fetch(guild: Identifiable, options?: FetchOptions): Promise | undefined>; + async fetch( + guildOrCache?: Identifiable | boolean, + options?: FetchOptions, + ): Promise[] | QuaverPlayer | undefined> { + if (typeof guildOrCache === 'boolean' || guildOrCache === undefined) { + // Fetch all players + const allPlayers: QuaverPlayer[] = []; + for (const node of this.cluster.nodes.values()) { + const players = await node.players.fetch(guildOrCache as boolean | undefined); + allPlayers.push(...(players as QuaverPlayer[])); + } + return allPlayers; + } + + // Fetch specific guild's player + const guildId = typeof guildOrCache === 'string' ? guildOrCache : guildOrCache.id; + const node = this.getNodeForGuild(guildId); + return (await node?.players.fetch(guildId, options)) as QuaverPlayer | undefined; + } + + /** + * Create a player on the best available node + */ + create(guild: Guild): QuaverPlayer { + const guildId = guild.id; + + // Check if player already exists + const existing = this.resolve(guildId); + if (existing) return existing; + + // Select best node based on voice region if available + const voiceChannel = guild.members.me?.voice?.channel; + const region = voiceChannel?.rtcRegion ?? null; + const node = this.cluster.getNodeForRegion(region); + if (!node) { + throw new Error('No available Lavalink nodes'); + } + + // Create player on selected node + const player = node.players.create(guild) as QuaverPlayer; + + // Track which node this guild is on + const nodeId = Array.from(this.cluster.nodes.entries()) + .find(([, n]: [string, QuaverNode]): boolean => n === node)?.[0]; + if (nodeId) { + this.guildNodeMap.set(guildId, nodeId); + } + + return player; + } + + /** + * Create a player from JSON data (for restoration after restart) + */ + async createFromJSON( + guild: Guild, + data: QuaverPlayerJSON, + resumed = false, + ): Promise> { + // Check if player already exists on a node + let node = this.getNodeForGuild(guild.id); + + // If no existing node, select best node based on voice region + if (!node) { + const voiceChannel = guild.members.me?.voice?.channel; + const region = voiceChannel?.rtcRegion ?? null; + node = this.cluster.getNodeForRegion(region); + if (!node) { + throw new Error('No available Lavalink nodes'); + } + + // Track which node this guild is on + const nodeId = Array.from(this.cluster.nodes.entries()) + .find(([, n]: [string, QuaverNode]): boolean => n === node)?.[0]; + if (nodeId) { + this.guildNodeMap.set(guild.id, nodeId); + } + } + + // Delegate to the node's player manager + return node.players.createFromJSON(guild, data, resumed) as Promise>; + } + + /** + * Destroy player(s) + */ + destroy(guild: Identifiable, force?: boolean): Promise; + destroy(): Promise; + async destroy(guild?: Identifiable, force?: boolean): Promise { + if (!guild) { + // Destroy all players + let count = 0; + for (const node of this.cluster.nodes.values()) { + count += await node.players.destroy(); + } + this.guildNodeMap.clear(); + return count; + } + + const guildId = typeof guild === 'string' ? guild : guild.id; + const node = this.getNodeForGuild(guildId); + const result = await node?.players.destroy(guildId, force) ?? false; + + if (result) { + this.guildNodeMap.delete(guildId); + } + + return result; + } + + /** + * Handle voice updates (route to appropriate node) + */ + async handleVoiceUpdate(update: VoiceStateUpdate | VoiceServerUpdate): Promise { + const guildId = update.guild_id; + const node = this.getNodeForGuild(guildId); + + if (!node) { + // If no node assigned yet, this might be the first voice update + // Just return false, player will be created later + return false; + } + + return node.players.handleVoiceUpdate(update); + } + + /** + * Get the node that a guild's player is on + */ + private getNodeForGuild(guildId: string): QuaverNode | undefined { + const nodeId = this.guildNodeMap.get(guildId); + if (nodeId) { + return this.cluster.nodes.get(nodeId); + } + + // Check if player exists on any node + for (const [id, node] of this.cluster.nodes) { + if (node.players.has(guildId)) { + this.guildNodeMap.set(guildId, id); + return node; + } + } + + return undefined; + } +} diff --git a/src/lib/music/ConnectionHealthMonitor.ts b/src/lib/music/ConnectionHealthMonitor.ts index b340027b2..29175cde7 100644 --- a/src/lib/music/ConnectionHealthMonitor.ts +++ b/src/lib/music/ConnectionHealthMonitor.ts @@ -1,6 +1,7 @@ import type { QuaverClient } from '#src/lib'; import { logger } from '#src/lib/logger'; import { settings } from '../util'; +import type { RegionAffinity } from './RegionAffinity'; export type GatewayHealthData = { averagePing: number; @@ -38,6 +39,7 @@ type ConnectionHealthConfig = { export class ConnectionHealthMonitor { private client: QuaverClient; private config: ConnectionHealthConfig; + private regionAffinity: RegionAffinity | null; // Gateway health tracking private pingSamples: number[] = []; @@ -48,6 +50,7 @@ export class ConnectionHealthMonitor { // Media server health tracking private mediaEndpoint: string | null = null; + private currentNodeId: string | null = null; private mediaLatencySamples: (number | null)[] = []; private mediaConsecutiveFailures = 0; private mediaCheckInterval?: ReturnType; @@ -80,13 +83,26 @@ export class ConnectionHealthMonitor { this.emitGatewayHealthUpdate(); }; - constructor(client: QuaverClient) { + constructor(client: QuaverClient, regionAffinity: RegionAffinity | null = null) { this.client = client; + this.regionAffinity = regionAffinity; this.config = this.loadConfig(); this.initializeGatewayMonitoring(); this.initializeMediaMonitoring(); } + /** + * Extracts the region prefix from a Discord media endpoint URL. + * Example: 'c-sin13-f16265ef.discord.media' -> 'c-sin' + */ + private static extractRegionPrefix(endpoint: string): string { + // Remove protocol if present + const cleaned = endpoint.replace(/^https?:\/\//, ''); + // Extract the region prefix (e.g., 'c-sin' from 'c-sin13-f16265ef.discord.media') + const match = cleaned.match(/^([a-z]+-[a-z]+)/); + return match ? match[1] : 'unknown'; + } + private loadConfig(): ConnectionHealthConfig { const userConfig = (settings as Record) .connectionHealth as ConnectionHealthConfig | undefined; @@ -108,6 +124,14 @@ export class ConnectionHealthMonitor { }; } + /** + * Sets the RegionAffinity instance for tracking ping-based node selection. + * This is called after initialization when multi-node configuration is detected. + */ + public setRegionAffinity(regionAffinity: RegionAffinity | null): void { + this.regionAffinity = regionAffinity; + } + private initializeGatewayMonitoring(): void { // Register bound event listeners so they can be removed during cleanup this.client.ws.on('ready' as never, this.onReady); @@ -173,6 +197,15 @@ export class ConnectionHealthMonitor { logger.debug( `Media server health check OK: ${this.mediaEndpoint} (${latency}ms)`, ); + + // Update region affinity if enabled + if (this.regionAffinity && this.currentNodeId && this.mediaEndpoint) { + const regionPrefix = ConnectionHealthMonitor.extractRegionPrefix(this.mediaEndpoint); + await this.regionAffinity.upsertAffinity(this.currentNodeId, regionPrefix, latency); + logger.debug( + `Updated region affinity: node=${this.currentNodeId}, region=${regionPrefix}, ping=${latency}ms`, + ); + } } else { this.mediaConsecutiveFailures++; logger.warn( @@ -200,10 +233,11 @@ export class ConnectionHealthMonitor { this.emitMediaHealthUpdate(); } - public updateMediaEndpoint(endpoint: string | null): void { + public updateMediaEndpoint(endpoint: string | null, nodeId: string | null = null): void { if (endpoint === null && this.mediaEndpoint !== null) { logger.info('Media server endpoint cleared, stopping health checks'); this.mediaEndpoint = null; + this.currentNodeId = null; this.mediaConsecutiveFailures = 0; this.mediaLatencySamples = []; this.lastMediaCheckTimestamp = Date.now(); @@ -212,8 +246,9 @@ export class ConnectionHealthMonitor { } if (endpoint && endpoint !== this.mediaEndpoint) { - logger.info(`Media server endpoint updated: ${endpoint}`); + logger.info(`Media server endpoint updated: ${endpoint} (node: ${nodeId ?? 'unknown'})`); this.mediaEndpoint = endpoint; + this.currentNodeId = nodeId; this.mediaConsecutiveFailures = 0; this.mediaLatencySamples = []; // Trigger an immediate health check diff --git a/src/lib/music/Penalties.ts b/src/lib/music/Penalties.ts new file mode 100644 index 000000000..70b669e3e --- /dev/null +++ b/src/lib/music/Penalties.ts @@ -0,0 +1,134 @@ +import type { QuaverNode } from './QuaverNode'; + +/** + * Node statistics interface (from Lavalink stats event) + */ +interface NodeStats { + players: number; + playingPlayers: number; + uptime: number; + memory: { + free: number; + used: number; + allocated: number; + reservable: number; + }; + cpu: { + cores: number; + systemLoad: number; + lavalinkLoad: number; + }; + frameStats?: { + sent: number; + nulled: number; + deficit: number; + }; +} + +export interface PenaltyProvider { + /** + * Calculates the penalty count for the given node. + */ + calculate: (node: QuaverNode) => number; +} + +/** + * Penalty calculation courtesy of: + * https://github.com/duncte123/lavalink-client/blob/main/src/main/kotlin/dev/arbjerg/lavalink/internal/loadbalancing/Penalties.kt + * + * Calculates load penalties for Lavalink nodes to enable intelligent load balancing. + * Lower penalty = better node for new players. + */ +export const Penalties = { + /** + * Calculate the total penalty for a node based on various metrics. + * + * @param node - The QuaverNode to calculate penalties for + * @returns Total penalty score (lower is better) + */ + calculate(node: QuaverNode): number { + // Access stats via type assertion since it's not in the type definitions + // but exists at runtime when the node receives stats from Lavalink + const stats = (node as { stats?: NodeStats }).stats; + + if (!stats) { + // No stats available yet, return high penalty + return Number.MAX_SAFE_INTEGER; + } + + let penalty = 0; + + // CPU penalty + penalty += this.calculateCpuPenalty(stats.cpu); + + // Player penalty (each player adds to the load) + penalty += stats.players; + + // Null frame penalty (frames that couldn't be provided) + if (stats.frameStats?.nulled) { + penalty += stats.frameStats.nulled * 2; + } + + // Deficit frame penalty (frames that were late) + if (stats.frameStats?.deficit) { + penalty += stats.frameStats.deficit * 1.5; + } + + return penalty; + }, + + /** + * Calculate CPU penalty based on system load and Lavalink process load. + * + * @param cpu - CPU stats from node + * @returns CPU penalty value + */ + calculateCpuPenalty(cpu: { systemLoad: number; lavalinkLoad: number }): number { + const systemLoad = cpu.systemLoad; + const lavalinkLoad = cpu.lavalinkLoad; + + // If system load exceeds 50%, apply exponential penalty + let cpuPenalty = 0; + if (systemLoad > 0.5) { + // Exponential penalty as system load increases + cpuPenalty += Math.pow(1.05, 100 * systemLoad) * 10 - 10; + } + + // Lavalink process load contributes more heavily + // If Lavalink itself is loaded, that's more directly impactful + if (lavalinkLoad > 0.5) { + cpuPenalty += Math.pow(1.05, 100 * lavalinkLoad) * 10 - 10; + } + + // Give Lavalink load higher weight + cpuPenalty += lavalinkLoad * 11 - 10; + + return cpuPenalty; + }, + + /** + * Find the node with the lowest penalty from the provided list. + * + * @param nodes - Array of QuaverNodes to evaluate + * @returns The node with the lowest penalty, or undefined if empty + */ + findBestNode(nodes: QuaverNode[]): QuaverNode | undefined { + if (nodes.length === 0) return undefined; + if (nodes.length === 1) return nodes[0]; + + // Initialize to first node to ensure we always return a node + // even if all nodes have equal (maximum) penalty during startup + let bestNode: QuaverNode = nodes[0]; + let lowestPenalty = Number.MAX_SAFE_INTEGER; + + for (const node of nodes) { + const penalty = this.calculate(node); + if (penalty < lowestPenalty) { + lowestPenalty = penalty; + bestNode = node; + } + } + + return bestNode; + }, +}; diff --git a/src/lib/music/QuaverCluster.ts b/src/lib/music/QuaverCluster.ts new file mode 100644 index 000000000..c257bf36a --- /dev/null +++ b/src/lib/music/QuaverCluster.ts @@ -0,0 +1,362 @@ +import type { QuaverClient } from '#src/lib'; +import { TypedEmitter } from 'tiny-typed-emitter'; +import type { NodeOptions, NodeEvents } from 'lavaclient'; +import { QuaverNode } from './QuaverNode'; +import { ClusterPlayerManager } from './ClusterPlayerManager'; +import { Penalties } from './Penalties'; +import type { RegionAffinity } from './RegionAffinity'; +import { settings } from '../util'; +import { logger } from '../logger'; + +export interface QuaverClusterNodeOptions { + info: NodeOptions['info']; + ws?: NodeOptions['ws']; + rest?: NodeOptions['rest']; + region: string; +} + +export interface QuaverClusterOptions { + nodes: QuaverClusterNodeOptions[]; + discord: NodeOptions['discord']; +} + +/** + * QuaverCluster manages multiple QuaverNode instances for multi-region Lavalink connectivity. + * Provides region-aware node selection with automatic fallback to available nodes. + */ +export class QuaverCluster extends TypedEmitter { + readonly client: QuaverClient; + readonly nodes: Map; + readonly regionMap: Map; + readonly players: ClusterPlayerManager; + private regionAffinity: RegionAffinity | null; + private pruneInterval?: ReturnType; + private affinityCache: Map = new Map(); + + constructor(options: QuaverClusterOptions, client: QuaverClient, regionAffinity: RegionAffinity | null = null) { + super(); + this.client = client; + this.nodes = new Map(); + this.regionMap = new Map(); + this.regionAffinity = regionAffinity; + + // Create QuaverNode instances for each configured node + options.nodes.forEach((nodeConfig, index): void => { + const nodeId = `node-${index}`; + const node = new QuaverNode( + { + info: nodeConfig.info, + discord: options.discord, + ws: nodeConfig.ws, + rest: nodeConfig.rest, + }, + client, + ); + + this.nodes.set(nodeId, node); + + // Build region mapping: region -> [nodeId1, nodeId2, ...] + const nodeIds = this.regionMap.get(nodeConfig.region) || []; + nodeIds.push(nodeId); + this.regionMap.set(nodeConfig.region, nodeIds); + }); + + // Create cluster player manager that routes operations across nodes + this.players = new ClusterPlayerManager(this); + + // Set up periodic pruning and cache refresh for affinity data if enabled + if (this.regionAffinity && settings.regionAffinity?.enabled) { + const staleAfterMs = settings.regionAffinity.staleAfterMs ?? 300000; + const refreshSeconds = settings.regionAffinity.refreshSeconds ?? 30; + + // Initial cache refresh + void this.refreshAffinityCache(); + + this.pruneInterval = setInterval((): void => { + // Run operations sequentially to avoid race conditions + void (async (): Promise => { + try { + // Prune stale entries first + await this.regionAffinity?.pruneStaleEntries(staleAfterMs); + // Then refresh cache + await this.refreshAffinityCache(); + } catch (err) { + logger.error({ message: 'Failed to refresh affinity data', label: 'QuaverCluster', error: err }); + } + })(); + }, refreshSeconds * 1000); + + logger.info('Region affinity pruning and caching scheduled'); + } + } + + /** + * Refreshes the in-memory cache of affinity data for synchronous access. + */ + private async refreshAffinityCache(): Promise { + if (!this.regionAffinity) return; + + try { + const staleAfterMs = settings.regionAffinity?.staleAfterMs ?? 300000; + const allNodes = await this.regionAffinity.getAllNodes(staleAfterMs); + + // Build new cache (don't clear in-place to avoid race conditions) + const newCache = new Map(); + + // Populate new cache with compound keys: nodeId:regionPrefix + for (const { nodeId, regionPrefix, data } of allNodes) { + const key = `${nodeId}:${regionPrefix}`; + newCache.set(key, { + nodeId, + regionPrefix, + avgPing: data.avgPing, + lastUpdated: data.lastUpdated, + }); + } + + // Atomically replace the cache reference + this.affinityCache = newCache; + + logger.debug(`Affinity cache refreshed with ${allNodes.length} entries`); + } catch (error) { + logger.warn({ message: 'Failed to refresh affinity cache', label: 'QuaverCluster', error }); + } + } + + /** + * Gets the best node for a given Discord voice region using affinity-based or penalty-based selection. + * Priority: + * 1. Affinity-based selection (if enabled and data available) + * 2. Region-based penalty selection + * 3. Global penalty-based load balancing + */ + getNodeForRegion(region?: string | null): QuaverNode | undefined { + // Try affinity-based selection first if enabled and region is specified + if (region && this.regionAffinity && settings.regionAffinity?.enabled) { + const affinityNode = this.selectNodeByAffinity(region); + if (affinityNode) { + // Find the node ID for logging + let nodeId = 'unknown'; + for (const [id, node] of this.nodes.entries()) { + if (node === affinityNode) { + nodeId = id; + break; + } + } + logger.debug(`Selected node by affinity: ${nodeId} for region: ${region}`); + return affinityNode; + } + } + + // If no region specified, use penalty-based load balancing across all nodes + if (!region) { + return this.getNextAvailableNode(); + } + + // Find nodes that serve this region + const nodeIds = this.regionMap.get(region); + if (nodeIds && nodeIds.length > 0) { + // Collect all ready nodes for this region + const readyNodes: QuaverNode[] = []; + for (const nodeId of nodeIds) { + const node = this.nodes.get(nodeId); + if (node && this.isNodeReady(node)) { + readyNodes.push(node); + } + } + + // Use penalty-based selection to find the best node + if (readyNodes.length > 0) { + return Penalties.findBestNode(readyNodes); + } + } + + // Fallback to any available node + return this.getNextAvailableNode(); + } + + /** + * Selects a node based on region affinity data (ping measurements) using cached data. + * Only considers nodes that serve the specified target region. + * @param targetRegion - The Lavalink configured region to filter nodes by (e.g., "singapore") + * @returns The best node for the target region, or null if no affinity data available + */ + private selectNodeByAffinity(targetRegion: string | null): QuaverNode | null { + if (!this.regionAffinity || this.affinityCache.size === 0 || !targetRegion) return null; + + const maxPingMs = settings.regionAffinity?.maxPingMs ?? 50; + + // Get list of node IDs that serve the target region + const targetNodeIds = this.regionMap.get(targetRegion); + if (!targetNodeIds || targetNodeIds.length === 0) return null; + + // Collect ready nodes with affinity data for the target region + // Track minimum ping for each node (not average) to avoid skewing from fallback usage + const nodeAffinityMap = new Map(); + + for (const affinityData of this.affinityCache.values()) { + const { nodeId, avgPing } = affinityData; + + // Only consider nodes that serve the target region + if (!targetNodeIds.includes(nodeId)) continue; + + const node = this.nodes.get(nodeId); + if (!node || !this.isNodeReady(node)) continue; + + // Track minimum ping for this node across all region prefixes + const existing = nodeAffinityMap.get(nodeId); + if (existing) { + existing.minPing = Math.min(existing.minPing, avgPing); + } else { + nodeAffinityMap.set(nodeId, { node, minPing: avgPing }); + } + } + + if (nodeAffinityMap.size === 0) return null; + + // Collect candidates with their minimum ping + const candidateNodes: Array<{ node: QuaverNode; nodeId: string; minPing: number }> = []; + for (const [nodeId, { node, minPing }] of nodeAffinityMap.entries()) { + candidateNodes.push({ + node, + nodeId, + minPing, + }); + } + + // Try to find nodes that meet the threshold + const suitableNodes = candidateNodes.filter(({ minPing }): boolean => minPing <= maxPingMs); + + let selectedNodes: typeof candidateNodes; + if (suitableNodes.length > 0) { + // Use nodes that meet the threshold + selectedNodes = suitableNodes; + } else { + // No nodes meet threshold, use all candidates (will pick lowest ping) + selectedNodes = candidateNodes; + } + + // Find the lowest ping + const lowestPing = Math.min(...selectedNodes.map(({ minPing }): number => minPing)); + + // Get all nodes with the lowest ping + const bestNodes = selectedNodes.filter(({ minPing }): boolean => minPing === lowestPing); + + // If multiple nodes have the same ping, use penalty-based selection as tiebreaker + if (bestNodes.length > 1) { + const nodeArray = bestNodes.map(({ node }): QuaverNode => node); + return Penalties.findBestNode(nodeArray); + } + + return bestNodes[0]?.node ?? null; + } + + /** + * Check if a node is ready (WebSocket connected) + */ + private isNodeReady(node: QuaverNode): boolean { + // Access the ws.state from the node with optional chaining + // LavalinkWSClientState.Ready = 2 + return node.ws?.state === 2; + } + + /** + * Get the best available node using penalty-based load balancing. + * Considers CPU load, player count, and frame statistics to select optimal node. + */ + private getNextAvailableNode(): QuaverNode | undefined { + const nodeArray = Array.from(this.nodes.values()); + if (nodeArray.length === 0) return undefined; + + // Collect all ready nodes + const readyNodes: QuaverNode[] = []; + for (const node of nodeArray) { + if (this.isNodeReady(node)) { + readyNodes.push(node); + } + } + + // Use penalty-based selection to find the best node + if (readyNodes.length > 0) { + return Penalties.findBestNode(readyNodes); + } + + // No ready nodes found, return first node anyway + return nodeArray[0]; + } + + /** + * Check if any node is ready + */ + get ready(): boolean { + for (const node of this.nodes.values()) { + if (this.isNodeReady(node)) { + return true; + } + } + return false; + } + + /** + * Get the first ready node's WebSocket client (for compatibility) + */ + get ws(): QuaverNode['ws'] | undefined { + for (const node of this.nodes.values()) { + if (this.isNodeReady(node)) { + return node.ws; + } + } + // Return first node's ws even if not ready + return Array.from(this.nodes.values())[0]?.ws; + } + + /** + * Get the first ready node's REST client (for compatibility) + */ + get rest(): QuaverNode['rest'] | undefined { + for (const node of this.nodes.values()) { + if (this.isNodeReady(node)) { + return node.rest; + } + } + // Return first node's rest even if not ready + return Array.from(this.nodes.values())[0]?.rest; + } + + /** + * Get the first ready node's API client (for compatibility) + */ + get api(): QuaverNode['api'] | undefined { + for (const node of this.nodes.values()) { + if (this.isNodeReady(node)) { + return node.api; + } + } + // Return first node's api even if not ready + return Array.from(this.nodes.values())[0]?.api; + } + + /** + * Connect all nodes + */ + connect(): void { + for (const node of this.nodes.values()) { + node.connect(); + } + } + + /** + * Disconnect all nodes and clean up resources + */ + disconnect(): void { + // Clear pruning interval + if (this.pruneInterval) { + clearInterval(this.pruneInterval); + this.pruneInterval = undefined; + } + + for (const node of this.nodes.values()) { + node.disconnect(); + } + } +} diff --git a/src/lib/music/QuaverPlayerManager.ts b/src/lib/music/QuaverPlayerManager.ts index da280ca08..627dce50d 100644 --- a/src/lib/music/QuaverPlayerManager.ts +++ b/src/lib/music/QuaverPlayerManager.ts @@ -16,6 +16,14 @@ export class QuaverPlayerManager< return super.resolve(guild) as QuaverPlayer | undefined; } + /** + * Get the node ID for a guild (returns null for single-node setup) + */ + // eslint-disable-next-line @typescript-eslint/no-unused-vars + getNodeIdForGuild(_guildId: string): string | null { + return null; + } + create(guild: Guild): QuaverPlayer { if (this.has(guild.id)) { return this.resolve(guild.id)!; diff --git a/src/lib/music/RegionAffinity.ts b/src/lib/music/RegionAffinity.ts new file mode 100644 index 000000000..ae1dd97ac --- /dev/null +++ b/src/lib/music/RegionAffinity.ts @@ -0,0 +1,124 @@ +import KeyvSqlite from '@keyv/sqlite'; +import Keyv from 'keyv'; + +interface AffinityData { + regionPrefix: string; + avgPing: number; + lastUpdated: number; +} + +/** + * Manages ping-based region affinity data for Lavalink nodes. + * Uses Keyv with SQLite backend for persistence. + */ +export class RegionAffinity { + private keyv: Keyv; + + /** + * Creates a new RegionAffinity instance. + * @param databaseUri - The SQLite database URI (e.g., 'sqlite://path/to/db.sqlite') + */ + constructor(databaseUri: string) { + this.keyv = new Keyv({ + store: new KeyvSqlite({ + uri: databaseUri, + }), + namespace: 'region-affinity', + }); + } + + /** + * Updates or creates affinity data for a node using exponential moving average. + * @param nodeId - The unique identifier for the Lavalink node + * @param regionPrefix - The region prefix (e.g., 'c-sin', 'c-use') + * @param ping - The current ping measurement in milliseconds + */ + async upsertAffinity(nodeId: string, regionPrefix: string, ping: number): Promise { + // Use compound key: nodeId:regionPrefix to track affinity per node per region + const key = `${nodeId}:${regionPrefix}`; + const existing = await this.keyv.get(key); + + let avgPing: number; + if (existing) { + // Calculate exponential moving average (EMA) with α = 0.5 + avgPing = existing.avgPing * 0.5 + ping * 0.5; + } else { + // First measurement + avgPing = ping; + } + + const data: AffinityData = { + regionPrefix, + avgPing, + lastUpdated: Date.now(), + }; + + await this.keyv.set(key, data); + } + + /** + * Gets all non-stale affinity data entries. + * @param staleAfterMs - Time in milliseconds after which data is considered stale + * @returns Array of entries with nodeId, regionPrefix, and affinity data + */ + async getAllNodes(staleAfterMs: number): Promise> { + const now = Date.now(); + const result: Array<{nodeId: string, regionPrefix: string, data: AffinityData}> = []; + + // Keyv doesn't have a native "get all keys" method, so we need to iterate + // This is a limitation, but acceptable for a small number of nodes + const iterator = this.keyv.iterator!(); + + for await (const [key, data] of iterator) { + const affinityData = data as AffinityData; + + // Skip stale entries + if (now - affinityData.lastUpdated > staleAfterMs) { + continue; + } + + // Parse compound key: nodeId:regionPrefix + const keyStr = key as string; + const separatorIndex = keyStr.indexOf(':'); + if (separatorIndex === -1) { + // Skip malformed keys + continue; + } + + const nodeId = keyStr.substring(0, separatorIndex); + const regionPrefix = keyStr.substring(separatorIndex + 1); + + result.push({ + nodeId, + regionPrefix, + data: affinityData, + }); + } + + return result; + } + + /** + * Removes stale affinity entries from the database. + * @param staleAfterMs - Time in milliseconds after which data is considered stale + */ + async pruneStaleEntries(staleAfterMs: number): Promise { + const now = Date.now(); + const iterator = this.keyv.iterator!(); + + const keysToDelete: string[] = []; + + for await (const [key, data] of iterator) { + const affinityData = data as AffinityData; + + if (now - affinityData.lastUpdated > staleAfterMs) { + keysToDelete.push(key as string); + } + } + + // Delete all stale keys + for (const key of keysToDelete) { + await this.keyv.delete(key); + } + } +} diff --git a/src/lib/music/index.ts b/src/lib/music/index.ts index a137822e4..e3317edd5 100644 --- a/src/lib/music/index.ts +++ b/src/lib/music/index.ts @@ -1,7 +1,11 @@ export * from './ConnectionHealthMonitor'; export * from './PlayerStateManager'; +export * from './QuaverCluster'; +export * from './ClusterPlayerManager'; export * from './QuaverNode'; export * from './QuaverPlayer'; export * from './QuaverPlayerManager'; export * from './QuaverQueue'; +export * from './Penalties'; +export * from './RegionAffinity'; export * from './types'; diff --git a/src/schemas/Settings.ts b/src/schemas/Settings.ts index 07cb26f43..40dad9843 100644 --- a/src/schemas/Settings.ts +++ b/src/schemas/Settings.ts @@ -150,18 +150,44 @@ export const SettingsSchema = z.object({ protocol: z.literal('sqlite'), path: z.string().default('database.sqlite'), }), - lavalink: z.object({ - host: z.string(), - port: z.number().int(), - password: z.string(), - secure: z.boolean().default(false), - reconnect: z - .object({ - delay: z.number().default(3000), - tries: z.number().default(5), - }) - .optional(), - }), + // Support both legacy single-node and multi-node configurations + lavalink: z.union([ + // Legacy single node (no region required) + z.object({ + host: z.string(), + port: z.number().int(), + password: z.string(), + secure: z.boolean().default(false), + reconnect: z + .object({ + delay: z.number().default(3000), + tries: z.number().default(5), + }) + .optional(), + }), + // New multi-node configuration + z.object({ + nodes: z + .array( + z.object({ + host: z.string(), + port: z.number().int(), + password: z.string(), + secure: z.boolean().default(false), + // Discord voice region identifier matching client.fetchVoiceRegions() + // e.g., 'singapore', 'us-east', 'japan', etc. + region: z.string(), + reconnect: z + .object({ + delay: z.number().default(3000), + tries: z.number().default(5), + }) + .optional(), + }), + ) + .min(1, 'At least one Lavalink node is required'), + }), + ]), features: z.object({ autolyrics: genericPremiumFeatureSchema, stay: genericPremiumFeatureSchema, @@ -249,6 +275,13 @@ export const SettingsSchema = z.object({ checkTimeoutMs: z.number().int().positive().default(2000), }), }), + regionAffinity: z.object({ + enabled: z.boolean().default(true), + maxPingMs: z.number().int().positive().default(50), + refreshSeconds: z.number().int().positive().default(30), + // 5 minutes + staleAfterMs: z.number().int().positive().default(5 * 60 * 1000), + }), ads: z .object({ enabled: z.boolean().default(false),