R与Discord间WebSocket连接异常排查及重连实现咨询

问题背景 #

我正在开发一段R代码，通过WebSocket实现Discord消息实时推送至R（替代httr::GET轮询方案，避免Discord请求次数限制）。使用websocket和httpuv库开发时，连接会在数分钟到数小时内随机断开，出现Unhandled promise error: invalid state等错误，怀疑缺少重连逻辑，需要排查代码问题并解决。

原代码如下：

library(websocket)
library(jsonlite)
library(async)


Heartbeat =10 #container for heartbite

DiscordSignals <- function(DiscordToken){

  #payloads or messages that R has to send to Discord.
  Rpayload = list(
    op= 2,
    d= list(
      token= DiscordToken,
      intents= 513,
      properties=list(
        os= "windows",
        browser= "firefox",
        device= "firefox")
    )
  )

  KeepAlive = list(op= 1, d= "null" )


  Jsonpayload   = toJSON(Rpayload, auto_unbox = TRUE, pretty = TRUE)
  JsonKeepAlive = toJSON(KeepAlive, auto_unbox = TRUE, pretty = TRUE)

  ws <<- WebSocket$new("wss://gateway.discord.gg/?v=9&encording=json", errorLogChannels ="warn")

  ws$onOpen(function(event) {ws$send(Jsonpayload)}) #Handshake

  ws$onMessage(function(event) {
    d <- event$data
    json = fromJSON(d)
    Alert= json$d$content
    OP<<-as.numeric(json$op) #Type of message sent from Discord

    #https://discord.com/developers/docs/topics/gateway#resuming
    #Discord may request additional heartbeats from your app by sending a Heartbeat (opcode OP1) event. Upon receiving the event, .. immediately send back .. Heartbeat ...

    #reset heartbeat rate to whatever asked by Discord. Heartbite rate requested from Discord is sent in OPs 1 or 10.
    if (OP  %in% c(1,10)){Heartbeat <<-round(abs(((json$d$heartbeat_interval )/1000)-runif(1)-3),0)} 

    print (Alert)
  })

  #send heartbeat every given interval
  async({
    p=1
    while (p==1){
      await(delay(Heartbeat)) 
      ws$send(JsonKeepAlive)
  
    }
  })

}


DiscordSignals (BotToken)

代码问题分析 #

• 全局变量滥用：Heartbeat、OP、ws使用全局赋值（<<-），多实例或异步场景下会引发变量冲突，导致状态异常。
• 心跳逻辑缺陷：心跳间隔加入随机值，可能偏离Discord要求的标准，触发连接失效判定；且心跳循环未检查连接状态，断开后仍尝试发送心跳，直接抛出invalid state错误。
• 无错误处理与重连机制：未监听连接错误、关闭事件，断开后无法自动恢复；也未处理Discord主动发送的重连指令（op=7）。
• 未保存会话信息：未记录Discord网关返回的session_id和sequence字段，无法使用**恢复连接（Resume）**机制，只能重新握手，增加连接成本。

修复后的代码 #

library(websocket)
library(jsonlite)
library(async)

DiscordSignals <- function(DiscordToken) {
  # 封装会话状态，避免全局变量污染
  session_state <- list(
    heartbeat_interval = 10,
    session_id = NULL,
    sequence = NULL,
    ws = NULL,
    heartbeat_task = NULL,
    reconnect_attempts = 0,
    max_reconnect_attempts = 5
  )

  # 初始化/恢复连接函数
  init_connection <- function(resume = FALSE) {
    # 清理旧连接资源
    if (!is.null(session_state$ws)) {
      session_state$ws$close()
    }

    # 创建新WebSocket连接
    session_state$ws <- WebSocket$new(
      url = "wss://gateway.discord.gg/?v=9&encoding=json",
      errorLogChannels = "warn"
    )

    # 构造握手或恢复连接的payload
    if (resume && !is.null(session_state$session_id) && !is.null(session_state$sequence)) {
      payload <- list(
        op = 6,
        d = list(
          token = DiscordToken,
          session_id = session_state$session_id,
          seq = session_state$sequence
        )
      )
    } else {
      payload <- list(
        op = 2,
        d = list(
          token = DiscordToken,
          intents = 513,
          properties = list(
            os = "windows",
            browser = "firefox",
            device = "firefox"
          )
        )
      )
    }
    json_payload <- toJSON(payload, auto_unbox = TRUE)

    # 连接打开事件处理
    session_state$ws$onOpen(function(event) {
      session_state$reconnect_attempts <- 0 # 重置重连计数
      session_state$ws$send(json_payload)
      cat("连接成功\n")
    })

    # 消息接收事件处理
    session_state$ws$onMessage(function(event) {
      json_data <- fromJSON(event$data)
      session_state$sequence <- json_data$s # 更新sequence值

      op <- as.numeric(json_data$op)
      d <- json_data$d

      # 处理不同Opcode的网关事件
      switch(as.character(op),
        "1" = { # 心跳请求，立即回复
          session_state$ws$send(toJSON(list(op = 1, d = session_state$sequence), auto_unbox = TRUE))
        },
        "10" = { # 初始化心跳间隔与会话信息
          session_state$heartbeat_interval <- d$heartbeat_interval / 1000
          start_heartbeat() # 启动心跳任务
          if (!resume) session_state$session_id <- d$session_id # 保存会话ID
        },
        "7" = { # 收到重连指令，立即尝试恢复连接
          cat("收到重连指令，尝试恢复会话\n")
          init_connection(resume = TRUE)
        },
        "0" = { # 普通消息事件
          if (!is.null(d$content)) cat("收到消息：", d$content, "\n")
        }
      )
    })

    # 错误事件处理
    session_state$ws$onError(function(event) {
      cat("连接错误：", event$message, "\n")
      attempt_reconnect()
    })

    # 连接关闭事件处理
    session_state$ws$onClose(function(event) {
      cat("连接关闭，代码：", event$code, "，原因：", event$reason, "\n")
      attempt_reconnect()
    })
  }

  # 启动心跳任务
  start_heartbeat <- function() {
    # 停止旧心跳任务
    if (!is.null(session_state$heartbeat_task)) {
      session_state$heartbeat_task$cancel()
    }
    # 启动新心跳循环
    session_state$heartbeat_task <- async({
      while (TRUE) {
        await(delay(session_state$heartbeat_interval))
        # 仅在连接处于OPEN状态时发送心跳
        if (session_state$ws$readyState() == 1) {
          session_state$ws$send(toJSON(list(op = 1, d = session_state$sequence), auto_unbox = TRUE))
        } else {
          break # 连接已关闭，终止心跳循环
        }
      }
    })
  }

  # 指数退避重连逻辑
  attempt_reconnect <- function() {
    if (session_state$reconnect_attempts >= session_state$max_reconnect_attempts) {
      cat("达到最大重连次数，停止尝试\n")
      return()
    }
    session_state$reconnect_attempts <- session_state$reconnect_attempts + 1
    # 计算退避时间，最长不超过30秒
    delay_seconds <- min(2^session_state$reconnect_attempts, 30)
    cat(sprintf("第%d次重连尝试，等待%d秒\n", session_state$reconnect_attempts, delay_seconds))
    async({
      await(delay(delay_seconds))
      init_connection(resume = TRUE)
    })
  }

  # 首次初始化连接
  init_connection(resume = FALSE)
}

# 调用示例（替换为你的Bot令牌）
# DiscordSignals("YOUR_BOT_TOKEN")

修复说明 #

• 会话状态封装：用session_state列表管理所有连接相关变量，避免全局变量冲突。
• 规范心跳逻辑：严格使用Discord返回的heartbeat_interval，移除随机值；心跳循环检查连接状态，仅在正常连接时发送心跳，避免无效状态错误。
• 完善重连机制：监听错误与关闭事件，触发指数退避重连，避免短时间内频繁请求网关；支持使用op=6恢复会话，提升重连效率。
• 完整Opcode处理：覆盖Discord网关核心指令，符合官方网关交互规范，减少异常触发点。

内容的提问来源于stack exchange，提问作者Camilo